import fs from "node:fs/promises";
import path from "node:path";
import YAML from "yaml";
import { z } from "zod";
import { canonicalJson, sha256, type JsonObject } from "../core/json.js";
import {
  DEFAULT_ALERT_INCIDENT_CATEGORIES,
  OPERATIONAL_INCIDENT_CATEGORIES,
} from "../incidents/types.js";
import {
  xActivityDirectionSchema,
  xActivityEventTypeSchema,
  xNumericIdSchema,
  xPublicPostEventTypeSchema,
  xSourceLaneSchema,
  xUsernameSchema,
} from "../connectors/x-contract.js";

const privacySchema = z.enum(["public-source", "private", "sensitive"]);
const idSchema = z.string().min(1).max(200).regex(/^[a-z0-9][a-z0-9._:-]*$/);

const sourceBase = {
  id: idSchema,
  enabled: z.boolean().default(false),
};

const filesystemSourceSchema = z.object({
  ...sourceBase,
  kind: z.literal("filesystem"),
  root: z.string().min(1),
  include: z.array(z.string().min(1)).min(1).max(100).optional(),
  ignore: z.array(z.string().min(1)).max(100).optional(),
  extensions: z.array(z.string().regex(/^\.[a-z0-9]+$/i)).min(1).max(50).optional(),
  debounceMs: z.number().int().min(0).max(60_000).default(250),
  maxFileBytes: z.number().int().positive().max(64 * 1024 * 1024).optional(),
  maxDiffChars: z.number().int().positive().max(1_000_000).optional(),
  privacy: privacySchema.default("sensitive"),
  storeContent: z.boolean().default(true),
}).strict();

const rssSourceSchema = z.object({
  ...sourceBase,
  kind: z.literal("rss"),
  url: z.url().refine((value) => ["http:", "https:"].includes(new URL(value).protocol), "RSS URL must use HTTP or HTTPS"),
  intervalMs: z.number().int().min(1_000).max(7 * 24 * 60 * 60 * 1_000),
  timeoutMs: z.number().int().min(100).max(10 * 60_000).default(15_000),
  maxItems: z.number().int().positive().max(10_000).default(200),
  privacy: privacySchema.default("public-source"),
  pollOnStart: z.boolean().default(true),
}).strict();

const fastmailJmapSourceSchema = z.object({
  ...sourceBase,
  kind: z.literal("fastmail-jmap"),
  tokenEnv: z.literal("FASTMAIL_API_KEY"),
  intervalMs: z.number().int().min(5_000).max(24 * 60 * 60 * 1_000).default(60_000),
  timeoutMs: z.number().int().min(1_000).max(120_000).default(15_000),
  maxResponseBytes: z.number().int().min(1_024).max(64 * 1024 * 1024).default(4 * 1024 * 1024),
  maxChanges: z.number().int().positive().max(1_000).default(100),
  maxPages: z.number().int().positive().max(100).default(10),
  resnapshotLimit: z.number().int().positive().max(1_000).default(200),
  credentialCustody: z.enum(["unprovisioned", "dedicated-mail-ingress", "shared-operator-accepted"]).default("unprovisioned"),
  pollOnStart: z.boolean().default(true),
  replay: z.literal("now").default("now"),
}).strict().superRefine((source, context) => {
  if (source.enabled && source.credentialCustody === "unprovisioned") {
    context.addIssue({
      code: "custom",
      path: ["credentialCustody"],
      message: "Enabled Fastmail sources require explicit credential custody",
    });
  }
});

const telegramSpoolSourceSchema = z.object({
  ...sourceBase,
  kind: z.literal("telegram-spool"),
  file: z.string().min(1),
  intervalMs: z.number().int().min(100).max(24 * 60 * 60 * 1_000).default(1_000),
  maxRecords: z.number().int().positive().max(100_000).default(1_000),
  maxReadBytes: z.number().int().positive().max(64 * 1024 * 1024).default(8 * 1024 * 1024),
  pollOnStart: z.boolean().default(true),
}).strict();

const telegramBootMessageSchema = z.object({
  enabled: z.boolean().default(false),
  text: z.string().min(1).max(4_096),
}).strict();

const telegramNotificationSchema = z.object({
  enabled: z.boolean().default(false),
  includeNormal: z.boolean().default(true),
  runStatuses: z.array(z.enum(["completed", "failed"])).min(1).max(2).default(["completed"]),
  allowedSources: z.array(idSchema).min(1).max(100),
  allowedActors: z.array(z.string().min(1).max(500)).max(100).default([]),
  directReplyAgentIds: z.array(idSchema).max(20).default([]),
  directReplySources: z.array(idSchema).max(100).default([]),
  notificationProposalAgentIds: z.array(idSchema).max(20).default([]),
  notificationProposalSources: z.array(idSchema).max(100).default([]),
  maxMessagesPerWindow: z.number().int().positive().max(100).default(3),
  windowMs: z.number().int().min(1_000).max(24 * 60 * 60 * 1_000).default(60_000),
  likeDigestDelayMs: z.number().int().nonnegative().max(24 * 60 * 60 * 1_000).default(60_000),
  maxLikesPerDigest: z.number().int().positive().max(100).default(10),
}).strict();

const telegramReactionFeedbackSchema = z.object({
  enabled: z.boolean().default(false),
  allowedUserIds: z.array(z.string().regex(/^[0-9]+$/, "Telegram user id must be a positive integer string")).min(1).max(100),
}).strict();

const telegramBotChannelSchema = z.object({
  id: z.string().regex(/^-?[0-9]+$/, "Telegram chat id must be an integer string"),
  enabled: z.boolean().default(true),
  bootMessage: telegramBootMessageSchema.optional(),
  notifications: telegramNotificationSchema.optional(),
  reactionFeedback: telegramReactionFeedbackSchema.optional(),
}).strict();

const telegramWebhookSourceSchema = z.object({
  ...sourceBase,
  kind: z.literal("telegram-webhook"),
  tokenEnv: z.string().regex(/^[A-Z_][A-Z0-9_]*$/, "tokenEnv must be an environment variable name"),
  webhookSecretEnv: z.string().regex(/^[A-Z_][A-Z0-9_]*$/, "webhookSecretEnv must be an environment variable name"),
  webhookUrl: z.url().refine((value) => new URL(value).protocol === "https:", "Telegram webhook URL must use HTTPS"),
  webhookPath: z.string().min(2).max(200).regex(/^\/[A-Za-z0-9/_-]+$/, "Telegram webhook path must be an absolute URL path"),
  listenHost: z.enum(["127.0.0.1", "::1", "localhost"]).default("127.0.0.1"),
  listenPort: z.number().int().min(1).max(65_535).default(4_318),
  maxBodyBytes: z.number().int().min(1_024).max(8 * 1024 * 1024).default(1_048_576),
  requestTimeoutMs: z.number().int().min(1_000).max(120_000).default(35_000),
  dispatchIntervalMs: z.number().int().min(100).max(60_000).default(1_000),
  channels: z.array(telegramBotChannelSchema).min(1).max(100),
}).strict().superRefine((value, context) => {
  const webhookUrl = new URL(value.webhookUrl);
  if (webhookUrl.username || webhookUrl.password || webhookUrl.search || webhookUrl.hash) {
    context.addIssue({ code: "custom", path: ["webhookUrl"], message: "Telegram webhook URL may not contain credentials, query parameters, or a fragment" });
  }
  if (webhookUrl.pathname !== value.webhookPath) {
    context.addIssue({ code: "custom", path: ["webhookPath"], message: "Telegram webhook path must exactly match the public webhook URL path" });
  }
  if (webhookUrl.port && !["80", "88", "443", "8443"].includes(webhookUrl.port)) {
    context.addIssue({ code: "custom", path: ["webhookUrl"], message: "Telegram webhook URL must use port 443, 80, 88, or 8443" });
  }
  if (value.webhookPath.includes("//") || value.webhookPath.split("/").includes("..")) {
    context.addIssue({ code: "custom", path: ["webhookPath"], message: "Telegram webhook path must be normalized" });
  }
  if (value.tokenEnv === value.webhookSecretEnv) {
    context.addIssue({ code: "custom", path: ["webhookSecretEnv"], message: "Telegram bot token and webhook secret must use different environment variables" });
  }
  const seen = new Set<string>();
  for (const [index, channel] of value.channels.entries()) {
    if (seen.has(channel.id)) {
      context.addIssue({ code: "custom", path: ["channels", index, "id"], message: `Duplicate Telegram channel id: ${channel.id}` });
    }
    seen.add(channel.id);
    if (channel.notifications?.enabled && channel.notifications.allowedSources.length === 0) {
      context.addIssue({ code: "custom", path: ["channels", index, "notifications", "allowedSources"], message: "Enabled notifications require an allowed source" });
    }
    if (channel.notifications) {
      const notifications = channel.notifications;
      if (notifications.notificationProposalAgentIds.length > 0 && notifications.notificationProposalSources.length === 0) {
        context.addIssue({ code: "custom", path: ["channels", index, "notifications", "notificationProposalSources"], message: "Notification proposal agent ids require explicit proposal sources" });
      }
      for (const source of [...notifications.directReplySources, ...notifications.notificationProposalSources]) {
        if (!notifications.allowedSources.includes(source)) {
          context.addIssue({ code: "custom", path: ["channels", index, "notifications", "allowedSources"], message: `Telegram route source is not allowed: ${source}` });
        }
      }
      const sharedAgent = notifications.directReplyAgentIds.some((id) => notifications.notificationProposalAgentIds.includes(id));
      const sharedSource = notifications.directReplySources.some((source) => notifications.notificationProposalSources.includes(source));
      if (sharedAgent && sharedSource) {
        context.addIssue({ code: "custom", path: ["channels", index, "notifications", "notificationProposalSources"], message: "Direct-reply and notification-proposal route tuples must be disjoint" });
      }
    }
    if (channel.reactionFeedback?.enabled && !channel.enabled) {
      context.addIssue({ code: "custom", path: ["channels", index, "reactionFeedback", "enabled"], message: "Reaction feedback requires an enabled Telegram channel" });
    }
  }
  if (!value.channels.some((channel) => channel.enabled)) {
    context.addIssue({ code: "custom", path: ["channels"], message: "Telegram bot source requires at least one enabled channel" });
  }
});

const xExpectedSubscriptionSchema = z.object({
  eventType: xActivityEventTypeSchema,
  userId: xNumericIdSchema,
  direction: xActivityDirectionSchema.optional(),
  tag: z.string().min(1).max(200),
}).strict();

const xSubscriptionFileSchema = z.object({
  version: z.literal(1),
  source: idSchema,
  eventTypes: z.array(xPublicPostEventTypeSchema).min(1).max(2),
  accounts: z.array(z.object({
    handle: xUsernameSchema,
    userId: xNumericIdSchema,
  }).strict()).min(1).max(750),
}).strict().superRefine((value, context) => {
  const eventTypes = new Set<string>();
  for (const [index, eventType] of value.eventTypes.entries()) {
    if (eventTypes.has(eventType)) {
      context.addIssue({ code: "custom", path: ["eventTypes", index], message: `Duplicate X watch event type: ${eventType}` });
    }
    eventTypes.add(eventType);
  }
  const handles = new Set<string>();
  const userIds = new Set<string>();
  for (const [index, account] of value.accounts.entries()) {
    const handle = account.handle.toLowerCase();
    if (handles.has(handle)) {
      context.addIssue({ code: "custom", path: ["accounts", index, "handle"], message: `Duplicate X watch handle: ${account.handle}` });
    }
    handles.add(handle);
    if (userIds.has(account.userId)) {
      context.addIssue({ code: "custom", path: ["accounts", index, "userId"], message: `Duplicate X watch user id: ${account.userId}` });
    }
    userIds.add(account.userId);
  }
});

const xWebhookSourceSchema = z.object({
  ...sourceBase,
  kind: z.literal("x-webhook"),
  lane: xSourceLaneSchema,
  managementAuth: z.enum(["app-only", "user-context"]).default("app-only"),
  consumerSecretEnv: z.string().regex(/^[A-Z_][A-Z0-9_]*$/, "consumerSecretEnv must be an environment variable name"),
  managementBearerTokenEnv: z.string().regex(/^[A-Z_][A-Z0-9_]*$/, "managementBearerTokenEnv must be an environment variable name"),
  webhookUrl: z.url().refine((value) => new URL(value).protocol === "https:", "X webhook URL must use HTTPS"),
  webhookPath: z.string().min(2).max(200).regex(/^\/[A-Za-z0-9/_-]+$/, "X webhook path must be an absolute URL path"),
  listenHost: z.enum(["127.0.0.1", "::1", "localhost"]).default("127.0.0.1"),
  listenPort: z.number().int().min(1).max(65_535).default(4_319),
  maxBodyBytes: z.number().int().min(1_024).max(8 * 1024 * 1024).default(2 * 1024 * 1024),
  requestTimeoutMs: z.number().int().min(1_000).max(9_000).default(8_000),
  pendingRequestLimit: z.number().int().min(1).max(10_000).default(100),
  expectedSubscriptions: z.array(xExpectedSubscriptionSchema).min(1).max(1_500),
}).strict().superRefine((value, context) => {
  const webhookUrl = new URL(value.webhookUrl);
  if (webhookUrl.username || webhookUrl.password || webhookUrl.search || webhookUrl.hash || webhookUrl.port) {
    context.addIssue({ code: "custom", path: ["webhookUrl"], message: "X webhook URL may not contain credentials, a port, query parameters, or a fragment" });
  }
  if (webhookUrl.pathname !== value.webhookPath) {
    context.addIssue({ code: "custom", path: ["webhookPath"], message: "X webhook path must exactly match the public webhook URL path" });
  }
  if (value.webhookPath.includes("//") || value.webhookPath.split("/").includes("..")) {
    context.addIssue({ code: "custom", path: ["webhookPath"], message: "X webhook path must be normalized" });
  }
  if (value.consumerSecretEnv === value.managementBearerTokenEnv) {
    context.addIssue({ code: "custom", path: ["managementBearerTokenEnv"], message: "X webhook secret and management bearer token must use different environment variables" });
  }
  if (value.lane === "personal-private") {
    if (value.managementAuth !== "user-context") {
      context.addIssue({ code: "custom", path: ["managementAuth"], message: "X personal-private sources require user-context management authority" });
    }
  } else if (value.managementAuth !== "app-only") {
    context.addIssue({ code: "custom", path: ["managementAuth"], message: "X public sources require app-only management authority" });
  }
  const seenTags = new Set<string>();
  const seenSubscriptions = new Set<string>();
  const tagPrefix = `thoughtstream:${value.id}:`;
  for (const [index, subscription] of value.expectedSubscriptions.entries()) {
    if (!subscription.tag.startsWith(tagPrefix)) {
      context.addIssue({ code: "custom", path: ["expectedSubscriptions", index, "tag"], message: `X subscription tag must begin with ${tagPrefix}` });
    }
    if (seenTags.has(subscription.tag)) {
      context.addIssue({ code: "custom", path: ["expectedSubscriptions", index, "tag"], message: `Duplicate X subscription tag: ${subscription.tag}` });
    }
    seenTags.add(subscription.tag);
    if (value.lane === "personal-private") {
      if (subscription.eventType !== "like.create" || subscription.direction !== "outbound") {
        context.addIssue({ code: "custom", path: ["expectedSubscriptions", index], message: "X personal-private sources admit only outbound like.create" });
      }
    } else if (subscription.eventType === "like.create" || subscription.direction !== undefined) {
      context.addIssue({ code: "custom", path: ["expectedSubscriptions", index], message: "X public sources admit only directionless post events" });
    }
    const identity = `${subscription.eventType}\u0000${subscription.userId}\u0000${subscription.direction ?? ""}`;
    if (seenSubscriptions.has(identity)) {
      context.addIssue({ code: "custom", path: ["expectedSubscriptions", index], message: `Duplicate X subscription event/user/direction tuple: ${subscription.eventType}/${subscription.userId}/${subscription.direction ?? "none"}` });
    }
    seenSubscriptions.add(identity);
  }
});

const jetstreamSourceSchema = z.object({
  ...sourceBase,
  kind: z.literal("jetstream"),
  endpoint: z.url().refine((value) => ["ws:", "wss:"].includes(new URL(value).protocol), "Jetstream endpoint must use WS or WSS").optional(),
  collections: z.array(z.string().min(1)).min(1).max(100),
  dids: z.array(z.string().min(1)).max(10_000).default([]),
  rewindUs: z.number().int().nonnegative().max(300_000_000).default(2_000_000),
  maxReconnects: z.number().int().nonnegative().max(100).default(8),
  restartBackoffMs: z.number().int().min(100).max(24 * 60 * 60 * 1_000).default(30_000),
  maxMessageSizeBytes: z.number().int().positive().max(10 * 1024 * 1024).default(1_048_576),
  pendingMessageLimit: z.number().int().positive().max(100_000).default(1_000),
}).strict();

const sourceSchema = z.discriminatedUnion("kind", [
  filesystemSourceSchema,
  rssSourceSchema,
  fastmailJmapSourceSchema,
  telegramSpoolSourceSchema,
  telegramWebhookSourceSchema,
  xWebhookSourceSchema,
  jetstreamSourceSchema,
]);

const batchDeclarationSchema = z.object({
  id: idSchema,
  version: z.number().int().positive(),
  enabled: z.boolean().default(false),
  input: z.object({
    eventTypes: z.array(z.string().min(1).max(300)).min(1).max(100),
    sourceIds: z.array(idSchema).min(1).max(100),
  }).strict(),
  output: z.object({
    sourceId: idSchema,
    eventType: z.string().min(1).max(300),
  }).strict(),
  quietWindowMs: z.number().int().min(100).max(24 * 60 * 60_000),
  maxAgeMs: z.number().int().min(100).max(7 * 24 * 60 * 60_000),
  maxItems: z.number().int().positive().max(1_000),
  privacy: z.enum(["preserve", "most-private"]),
  replay: z.enum(["beginning", "now"]),
  pollIntervalMs: z.number().int().min(100).max(60_000).default(1_000),
}).strict().superRefine((value, context) => {
  if (new Set(value.input.eventTypes).size !== value.input.eventTypes.length) {
    context.addIssue({ code: "custom", path: ["input", "eventTypes"], message: "Batch input event types must be unique" });
  }
  if (new Set(value.input.sourceIds).size !== value.input.sourceIds.length) {
    context.addIssue({ code: "custom", path: ["input", "sourceIds"], message: "Batch input source ids must be unique" });
  }
  if (value.input.sourceIds.includes(value.output.sourceId)) {
    context.addIssue({ code: "custom", path: ["output", "sourceId"], message: "Batch output source cannot feed the same declaration" });
  }
  if (value.maxAgeMs < value.quietWindowMs) {
    context.addIssue({ code: "custom", path: ["maxAgeMs"], message: "Batch maximum age must be at least the quiet window" });
  }
  if (value.privacy === "preserve" && value.input.sourceIds.length !== 1) {
    context.addIssue({ code: "custom", path: ["privacy"], message: "Preserved batch privacy requires exactly one input source" });
  }
});

const incidentRuntimeSchema = z.object({
  enabled: z.boolean().default(false),
  ledgerPath: z.string().min(1).max(500).refine((value) => {
    if (path.isAbsolute(value)) return false;
    const normalized = path.normalize(value);
    return normalized !== "." && !normalized.startsWith("..");
  }, "Incident ledger path must stay below the runtime root").default(".thoughtstream/error-ledger.jsonl"),
  intervalMs: z.number().int().min(100).max(60_000).default(1_000),
  telegramAlerts: z.object({
    enabled: z.boolean().default(false),
    sourceId: idSchema.optional(),
    channelIds: z.array(z.string().regex(/^-?[0-9]+$/, "Telegram chat id must be an integer string")).max(20).default([]),
    categories: z.array(z.enum(OPERATIONAL_INCIDENT_CATEGORIES)).min(1).max(OPERATIONAL_INCIDENT_CATEGORIES.length)
      .default([...DEFAULT_ALERT_INCIDENT_CATEGORIES]),
    cooldownMs: z.number().int().min(1_000).max(7 * 24 * 60 * 60_000).default(15 * 60_000),
    maxMessagesPerWindow: z.number().int().positive().max(100).default(3),
    windowMs: z.number().int().min(1_000).max(24 * 60 * 60_000).default(15 * 60_000),
  }).strict().default({
    enabled: false,
    channelIds: [],
    categories: [...DEFAULT_ALERT_INCIDENT_CATEGORIES],
    cooldownMs: 15 * 60_000,
    maxMessagesPerWindow: 3,
    windowMs: 15 * 60_000,
  }),
}).strict().default({
  enabled: false,
  ledgerPath: ".thoughtstream/error-ledger.jsonl",
  intervalMs: 1_000,
  telegramAlerts: {
    enabled: false,
    channelIds: [],
    categories: [...DEFAULT_ALERT_INCIDENT_CATEGORIES],
    cooldownMs: 15 * 60_000,
    maxMessagesPerWindow: 3,
    windowMs: 15 * 60_000,
  },
});

const manifestSchema = z.object({
  version: z.literal(1),
  runtime: z.object({
    revision: z.string().min(1).max(200).default("local-dev"),
    scheduler: z.object({
      maxConcurrentOperations: z.number().int().positive().max(64).default(4),
      reconcileIntervalMs: z.number().int().min(100).max(60_000).default(1_000),
    }).strict().default({ maxConcurrentOperations: 4, reconcileIntervalMs: 1_000 }),
  }).strict().default({ revision: "local-dev", scheduler: { maxConcurrentOperations: 4, reconcileIntervalMs: 1_000 } }),
  inspector: z.object({
    host: z.enum(["127.0.0.1", "::1", "localhost"]).default("127.0.0.1"),
    port: z.number().int().min(0).max(65_535).default(4_317),
  }).strict().default({ host: "127.0.0.1", port: 4_317 }),
  agents: z.object({
    directory: z.string().min(1).default("agents"),
  }).strict().default({ directory: "agents" }),
  incidents: incidentRuntimeSchema,
  sources: z.array(sourceSchema).max(1_000).default([]),
  batches: z.array(batchDeclarationSchema).max(1_000).default([]),
}).strict().superRefine((value, context) => {
  const seen = new Set<string>();
  const xWebhookUrls = new Set<string>();
  const xListenEndpoints = new Set<string>();
  for (const [index, source] of value.sources.entries()) {
    if (seen.has(source.id)) {
      context.addIssue({ code: "custom", path: ["sources", index, "id"], message: `Duplicate source id: ${source.id}` });
    }
    seen.add(source.id);
    if (source.kind === "x-webhook") {
      if (xWebhookUrls.has(source.webhookUrl)) {
        context.addIssue({ code: "custom", path: ["sources", index, "webhookUrl"], message: `Duplicate X webhook URL: ${source.webhookUrl}` });
      }
      xWebhookUrls.add(source.webhookUrl);
      const endpoint = `${source.listenHost}:${source.listenPort}`;
      if (xListenEndpoints.has(endpoint)) {
        context.addIssue({ code: "custom", path: ["sources", index, "listenPort"], message: `Duplicate X webhook listen endpoint: ${endpoint}` });
      }
      xListenEndpoints.add(endpoint);
    }
  }
  const batchIds = new Set<string>();
  const batchOutputs = new Set<string>();
  for (const [index, batch] of value.batches.entries()) {
    const identity = `${batch.id}@${batch.version}`;
    if (batchIds.has(identity)) {
      context.addIssue({ code: "custom", path: ["batches", index, "id"], message: `Duplicate batch declaration: ${identity}` });
    }
    batchIds.add(identity);
    if (batch.output.sourceId.includes("*")) {
      context.addIssue({ code: "custom", path: ["batches", index, "output", "sourceId"], message: "Batch output source must be exact" });
    }
    const output = `${batch.output.sourceId}\u0000${batch.output.eventType}`;
    if (batchOutputs.has(output)) {
      context.addIssue({ code: "custom", path: ["batches", index, "output"], message: "Batch declarations may not share an output source/type" });
    }
    batchOutputs.add(output);
  }
  const alerts = value.incidents.telegramAlerts;
  if (alerts.enabled) {
    if (!value.incidents.enabled) {
      context.addIssue({ code: "custom", path: ["incidents", "telegramAlerts", "enabled"], message: "Incident alerts require incident projection and ledger" });
    }
    const source = value.sources.find((candidate) => candidate.id === alerts.sourceId);
    if (!source || source.kind !== "telegram-webhook" || !source.enabled) {
      context.addIssue({ code: "custom", path: ["incidents", "telegramAlerts", "sourceId"], message: "Incident alerts require an enabled telegram-webhook source" });
    } else {
      const enabledChannels = new Set(source.channels.filter((channel) => channel.enabled).map((channel) => channel.id));
      if (alerts.channelIds.length === 0) {
        context.addIssue({ code: "custom", path: ["incidents", "telegramAlerts", "channelIds"], message: "Incident alerts require at least one channel" });
      }
      for (const [index, channelId] of alerts.channelIds.entries()) {
        if (!enabledChannels.has(channelId)) {
          context.addIssue({ code: "custom", path: ["incidents", "telegramAlerts", "channelIds", index], message: "Incident alert channel must be enabled on the selected Telegram source" });
        }
      }
    }
    if (alerts.categories.every((category) => category === "telegram-delivery-failed")) {
      context.addIssue({ code: "custom", path: ["incidents", "telegramAlerts", "categories"], message: "Telegram delivery failures cannot recursively alert through Telegram" });
    }
  }
});

export type ThoughtStreamManifest = z.infer<typeof manifestSchema>;
export type ThoughtStreamSource = ThoughtStreamManifest["sources"][number];
export type FilesystemSourceManifest = Extract<ThoughtStreamSource, { kind: "filesystem" }>;
export type RssSourceManifest = Extract<ThoughtStreamSource, { kind: "rss" }>;
export type FastmailJmapSourceManifest = Extract<ThoughtStreamSource, { kind: "fastmail-jmap" }>;
export type TelegramSpoolSourceManifest = Extract<ThoughtStreamSource, { kind: "telegram-spool" }>;
export type TelegramWebhookSourceManifest = Extract<ThoughtStreamSource, { kind: "telegram-webhook" }>;
export type XWebhookSourceManifest = Extract<ThoughtStreamSource, { kind: "x-webhook" }>;
export type JetstreamSourceManifest = Extract<ThoughtStreamSource, { kind: "jetstream" }>;
export type BatchDeclarationManifest = ThoughtStreamManifest["batches"][number];

export interface LoadedThoughtStreamManifest {
  path: string;
  manifest: ThoughtStreamManifest;
  hash: string;
}

export async function loadThoughtStreamManifest(
  projectRoot: string,
  requestedPath = "thoughtstream.yaml",
): Promise<LoadedThoughtStreamManifest> {
  const manifestPath = path.resolve(projectRoot, requestedPath);
  const stat = await fs.lstat(manifestPath).catch((error: NodeJS.ErrnoException) => {
    if (error.code === "ENOENT") {
      throw new Error(`thought stream manifest not found: ${manifestPath}`);
    }
    throw error;
  });
  if (stat.isSymbolicLink()) throw new Error(`thought stream manifest may not be a symlink: ${manifestPath}`);
  if (!stat.isFile()) throw new Error(`thought stream manifest is not a regular file: ${manifestPath}`);
  const raw = await fs.readFile(manifestPath, "utf8");
  const parsed = manifestPath.endsWith(".json") ? JSON.parse(raw) : YAML.parse(raw);
  const materialized = await materializeXSubscriptionFiles(projectRoot, parsed);
  const manifest = manifestSchema.parse(materialized);
  const agentDirectory = resolveProjectPath(projectRoot, manifest.agents.directory);
  if (!isWithin(projectRoot, agentDirectory)) {
    throw new Error(`Agent directory escapes project root: ${manifest.agents.directory}`);
  }
  return { path: manifestPath, manifest, hash: sha256(canonicalJson(asJsonObject(manifest))) };
}

async function materializeXSubscriptionFiles(projectRoot: string, value: unknown): Promise<unknown> {
  if (!isRecord(value) || !Array.isArray(value.sources)) return value;
  const sources = await Promise.all(value.sources.map(async (source): Promise<unknown> => {
    if (!isRecord(source) || source.kind !== "x-webhook" || !("subscriptionFile" in source)) return source;
    if ("expectedSubscriptions" in source) {
      throw new Error("X webhook source may not declare both subscriptionFile and expectedSubscriptions");
    }
    if (typeof source.subscriptionFile !== "string" || !source.subscriptionFile.trim()) {
      throw new Error("X subscriptionFile must name one project-relative YAML file");
    }
    const requestedPath = source.subscriptionFile;
    if (path.isAbsolute(requestedPath) || !/\.ya?ml$/i.test(requestedPath)) {
      throw new Error("X subscriptionFile must be a project-relative YAML file");
    }
    const subscriptionPath = path.resolve(projectRoot, requestedPath);
    if (!isWithin(projectRoot, subscriptionPath)) {
      throw new Error(`X subscriptionFile escapes project root: ${requestedPath}`);
    }
    const stat = await fs.lstat(subscriptionPath).catch((error: NodeJS.ErrnoException) => {
      if (error.code === "ENOENT") throw new Error(`X subscriptionFile not found: ${requestedPath}`);
      throw error;
    });
    if (stat.isSymbolicLink() || !stat.isFile()) {
      throw new Error(`X subscriptionFile must be a regular nonsymlink file: ${requestedPath}`);
    }
    if (stat.size > 262_144) throw new Error(`X subscriptionFile exceeds 262144 bytes: ${requestedPath}`);
    const [realRoot, realSubscriptionPath] = await Promise.all([fs.realpath(projectRoot), fs.realpath(subscriptionPath)]);
    if (!isWithin(realRoot, realSubscriptionPath)) {
      throw new Error(`X subscriptionFile resolves outside project root: ${requestedPath}`);
    }
    const config = xSubscriptionFileSchema.parse(YAML.parse(await fs.readFile(subscriptionPath, "utf8")));
    if (typeof source.id !== "string" || config.source !== source.id) {
      throw new Error(`X subscriptionFile source does not match manifest source: ${requestedPath}`);
    }
    const expectedSubscriptions = config.accounts
      .slice()
      .sort((left, right) => left.handle.toLowerCase().localeCompare(right.handle.toLowerCase()) || left.userId.localeCompare(right.userId))
      .flatMap((account) => config.eventTypes.slice().sort().map((eventType) => ({
        eventType,
        userId: account.userId,
        tag: `thoughtstream:${config.source}:${account.handle.toLowerCase()}:${eventType.replaceAll(".", "-")}`,
      })));
    const { subscriptionFile: _subscriptionFile, ...rest } = source;
    return { ...rest, expectedSubscriptions };
  }));
  return { ...value, sources };
}

export function resolveProjectPath(projectRoot: string, value: string): string {
  return path.isAbsolute(value) ? path.normalize(value) : path.resolve(projectRoot, value);
}

function isWithin(root: string, candidate: string): boolean {
  const relative = path.relative(path.resolve(root), candidate);
  return relative === "" || (!relative.startsWith("..") && !path.isAbsolute(relative));
}

function isRecord(value: unknown): value is Record<string, unknown> {
  return Boolean(value && typeof value === "object" && !Array.isArray(value));
}

function asJsonObject(value: unknown): JsonObject {
  const normalized = JSON.parse(JSON.stringify(value)) as unknown;
  if (!normalized || typeof normalized !== "object" || Array.isArray(normalized)) {
    throw new Error("Expected a JSON object");
  }
  return normalized as JsonObject;
}