Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
30 kB · 571 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572import 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;}