Something went wrong. Try again.
A local-first event pipeline for independent agents, built on Jazz.
Something went wrong. Try again.
40 kB · 1028 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029import { z } from "zod";import { canonicalJson, sha256, type JsonObject } from "../core/json.js";import { newId } from "../core/ids.js";import type { EventCandidate, ThoughtEvent } from "../events/types.js";import type { JazzThoughtStore } from "../jazz/store.js";import type { SourceCursor } from "../store/types.js";import { projectTelegramFeedbackJudgments } from "../training/telegram-reactions.js";import { downloadTelegramImage, imageAttachmentMetadata, isImageDocument, selectPhoto, TelegramImagePermanentError,} from "./telegram-images.js";
const TELEGRAM_WEBHOOK_REVISION = "telegram-bot-api-webhook-v1";const COMPATIBLE_TELEGRAM_REVISIONS = new Set([ "telegram-bot-api-v1", "telegram-bot-api-v2", TELEGRAM_WEBHOOK_REVISION,]);export const TELEGRAM_WEBHOOK_ALLOWED_UPDATES = ["message", "edited_message", "message_reaction"] as const;const TELEGRAM_CORRECTION_PREFIX = "/correct ";const TELEGRAM_CORRECTION_MAX_CHARS = 2_000;
const userSchema = z.object({ id: z.number().int(), is_bot: z.boolean(), first_name: z.string(), last_name: z.string().optional(), username: z.string().optional(),}).passthrough();
const chatSchema = z.object({ id: z.number().int(), type: z.enum(["private", "group", "supergroup", "channel"]), title: z.string().optional(), username: z.string().optional(), first_name: z.string().optional(), last_name: z.string().optional(),}).passthrough();
const photoSizeSchema = z.object({ file_id: z.string(), file_unique_id: z.string(), width: z.number().int(), height: z.number().int(), file_size: z.number().int().nonnegative().optional(),}).passthrough();
const fileSchema = z.object({ file_id: z.string(), file_unique_id: z.string(), file_name: z.string().optional(), mime_type: z.string().optional(), file_size: z.number().int().nonnegative().optional(),}).passthrough();
const messageSchema = z.object({ message_id: z.number().int(), message_thread_id: z.number().int().optional(), from: userSchema.optional(), date: z.number().int().nonnegative(), edit_date: z.number().int().nonnegative().optional(), chat: chatSchema, text: z.string().optional(), caption: z.string().optional(), reply_to_message: z.object({ message_id: z.number().int() }).passthrough().optional(), photo: z.array(photoSizeSchema).optional(), document: fileSchema.optional(), audio: fileSchema.optional(), voice: fileSchema.optional(), video: fileSchema.optional(),}).passthrough();
const reactionTypeSchema = z.object({ type: z.string().min(1), emoji: z.string().optional(), custom_emoji_id: z.string().optional(),}).passthrough();
const messageReactionSchema = z.object({ chat: chatSchema, message_id: z.number().int(), user: userSchema.optional(), actor_chat: chatSchema.optional(), date: z.number().int().nonnegative(), old_reaction: z.array(reactionTypeSchema), new_reaction: z.array(reactionTypeSchema),}).passthrough();
const updateSchema = z.object({ update_id: z.number().int().nonnegative(), message: messageSchema.optional(), edited_message: messageSchema.optional(), message_reaction: messageReactionSchema.optional(),}).passthrough();
const envelopeSchema = z.object({ ok: z.boolean(), result: z.unknown().optional(), error_code: z.number().int().optional(), description: z.string().optional(), parameters: z.object({ retry_after: z.number().int().positive().optional() }).passthrough().optional(),}).passthrough();
const webhookInfoSchema = z.object({ url: z.string(), has_custom_certificate: z.boolean(), pending_update_count: z.number().int().nonnegative(), ip_address: z.string().optional(), last_error_date: z.number().int().nonnegative().optional(), last_error_message: z.string().optional(), last_synchronization_error_date: z.number().int().nonnegative().optional(), max_connections: z.number().int().positive().optional(), allowed_updates: z.array(z.string()).optional(),}).passthrough();
const botCommandSchema = z.object({ command: z.string().regex(/^[a-z0-9_]{1,32}$/), description: z.string().min(1).max(256),});const botCommandsSchema = z.array(botCommandSchema).max(100);const menuButtonSchema = z.object({ type: z.string().min(1) }).passthrough();const botNameSchema = z.object({ name: z.string().min(1).max(64) }).passthrough();
export type TelegramBotUser = z.infer<typeof userSchema>;export type TelegramBotUpdate = z.infer<typeof updateSchema>;export type TelegramWebhookInfo = z.infer<typeof webhookInfoSchema>;export type TelegramBotCommand = z.infer<typeof botCommandSchema>;
export interface TelegramBotClientOptions { token: string; baseUrl?: string; requestTimeoutMs?: number;}
export interface TelegramSendResult { messageId: string; chatId: string; date: string;}
export class TelegramBotApiError extends Error { constructor( message: string, readonly errorCode?: number, readonly retryAfterSeconds?: number, ) { super(message); this.name = "TelegramBotApiError"; }}
export class TelegramBotClient { private readonly token: string; private readonly baseUrl: string; private readonly requestTimeoutMs: number; private identityValue?: TelegramBotUser;
constructor(options: TelegramBotClientOptions) { this.token = required(options.token, "Telegram bot token"); this.baseUrl = (options.baseUrl ?? "https://api.telegram.org").replace(/\/$/, ""); this.requestTimeoutMs = boundedPositiveInteger(options.requestTimeoutMs ?? 35_000, "requestTimeoutMs", 120_000); }
async identity(signal?: AbortSignal): Promise<TelegramBotUser> { this.identityValue ??= await this.call("getMe", {}, userSchema, signal); return this.identityValue; }
async setWebhook(options: { url: string; secretToken: string; dropPendingUpdates?: boolean; signal?: AbortSignal; }): Promise<void> { await this.call("setWebhook", { url: required(options.url, "Telegram webhook URL"), secret_token: webhookSecret(options.secretToken), max_connections: 1, allowed_updates: [...TELEGRAM_WEBHOOK_ALLOWED_UPDATES], drop_pending_updates: options.dropPendingUpdates ?? false, }, z.literal(true), options.signal); }
async deleteWebhook(options: { dropPendingUpdates?: boolean; signal?: AbortSignal } = {}): Promise<void> { await this.call("deleteWebhook", { drop_pending_updates: options.dropPendingUpdates ?? false, }, z.literal(true), options.signal); }
async getWebhookInfo(signal?: AbortSignal): Promise<TelegramWebhookInfo> { return this.call("getWebhookInfo", {}, webhookInfoSchema, signal); }
async setCommands(commands: TelegramBotCommand[], signal?: AbortSignal): Promise<void> { const validated = botCommandsSchema.parse(commands); await this.call("setMyCommands", { commands: validated }, z.literal(true), signal); }
async getCommands(signal?: AbortSignal): Promise<TelegramBotCommand[]> { return this.call("getMyCommands", {}, botCommandsSchema, signal); }
async setName(name: string, signal?: AbortSignal): Promise<void> { await this.call("setMyName", { name: required(name, "Telegram bot name") }, z.literal(true), signal); }
async getName(signal?: AbortSignal): Promise<string> { return (await this.call("getMyName", {}, botNameSchema, signal)).name; }
async setCommandsMenuButton(chatId: string, signal?: AbortSignal): Promise<void> { await this.call("setChatMenuButton", { chat_id: required(chatId, "Telegram chat id"), menu_button: { type: "commands" }, }, z.literal(true), signal); }
async getMenuButton(chatId: string, signal?: AbortSignal): Promise<string> { const result = await this.call("getChatMenuButton", { chat_id: required(chatId, "Telegram chat id"), }, menuButtonSchema, signal); return result.type; }
async sendMessage(chatId: string, text: string, signal?: AbortSignal): Promise<TelegramSendResult> { const message = await this.call("sendMessage", { chat_id: required(chatId, "Telegram chat id"), text: required(text, "Telegram message text"), link_preview_options: { is_disabled: true }, }, messageSchema, signal); return { messageId: String(message.message_id), chatId: String(message.chat.id), date: new Date(message.date * 1_000).toISOString(), }; }
async sendTyping(chatId: string, signal?: AbortSignal): Promise<void> { const timeout = AbortSignal.timeout(3_000); const boundedSignal = signal ? AbortSignal.any([signal, timeout]) : timeout; await this.call("sendChatAction", { chat_id: required(chatId, "Telegram chat id"), action: "typing", }, z.literal(true), boundedSignal); }
/** * Expose the base URL and token for image extraction in the webhook process. * Used by TelegramBotConnector to download admitted images via getFile. */ get imageExtractionConfig(): { baseUrl: string; token: string } { return { baseUrl: this.baseUrl, token: this.token }; }
private async call<T>( method: string, body: Record<string, unknown>, resultSchema: z.ZodType<T>, signal?: AbortSignal, ): Promise<T> { const timeout = AbortSignal.timeout(this.requestTimeoutMs); const combinedSignal = signal ? AbortSignal.any([signal, timeout]) : timeout; let response: Response; try { response = await fetch(`${this.baseUrl}/bot${this.token}/${method}`, { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body), signal: combinedSignal, }); } catch (error) { const reason = error instanceof Error && error.name === "AbortError" ? "timed out or was cancelled" : "failed"; throw new TelegramBotApiError(`Telegram Bot API ${method} request ${reason}`); } let raw: unknown; try { raw = await response.json(); } catch { throw new TelegramBotApiError(`Telegram Bot API ${method} returned non-JSON HTTP ${response.status}`); } const envelope = envelopeSchema.parse(raw); if (!response.ok || !envelope.ok) { throw new TelegramBotApiError( envelope.description ?? `Telegram Bot API ${method} failed with HTTP ${response.status}`, envelope.error_code ?? response.status, envelope.parameters?.retry_after, ); } return resultSchema.parse(envelope.result); }}
export interface TelegramReactionFeedbackRoute { chatId: string; allowedUserIds: string[];}
export interface TelegramBotConnectorOptions { id: string; bot: TelegramBotUser; allowedChatIds: string[]; reactionFeedback?: TelegramReactionFeedbackRoute[]; feedbackProjector?: typeof projectTelegramFeedbackJudgments; /** * Optional image extraction configuration. When provided, the connector * downloads at most one admitted PNG/JPEG image per message via getFile, * validates magic bytes, and stores it content-addressed under the artifact root. * The event carries only an opaque relative reference, never bytes or tokens. */ imageExtraction?: { client: TelegramBotClient; artifactRoot: string; };}
export interface TelegramWebhookIngestResult { status: "updated" | "unchanged"; accepted: number; ignored: number; inserted: number; unchanged: number; events: ThoughtEvent[]; cursor: SourceCursor;}
export class TelegramBotConnector { readonly kind = "telegram" as const; readonly id: string; private readonly bot: TelegramBotUser; private readonly allowedChatIds: Set<string>; private readonly reactionUsersByChat = new Map<string, Set<string>>(); private readonly imageExtraction?: { client: TelegramBotClient; artifactRoot: string } | undefined; private readonly feedbackProjector: typeof projectTelegramFeedbackJudgments;
constructor(options: TelegramBotConnectorOptions) { this.id = required(options.id, "Telegram bot connector id"); this.bot = options.bot; this.allowedChatIds = new Set(options.allowedChatIds.map((value) => required(value, "Allowed Telegram chat id"))); if (this.allowedChatIds.size === 0) throw new Error("Telegram bot connector requires at least one allowed chat id"); for (const route of options.reactionFeedback ?? []) { const chatId = required(route.chatId, "Reaction feedback chat id"); if (!this.allowedChatIds.has(chatId)) throw new Error(`Reaction feedback chat is not an allowed Telegram chat: ${chatId}`); const userIds = new Set(route.allowedUserIds.map((value) => required(value, "Reaction feedback user id"))); if (userIds.size === 0) throw new Error(`Reaction feedback chat requires at least one allowed user: ${chatId}`); this.reactionUsersByChat.set(chatId, userIds); } this.imageExtraction = options.imageExtraction; this.feedbackProjector = options.feedbackProjector ?? projectTelegramFeedbackJudgments; }
describe(): JsonObject { return { id: this.id, kind: this.kind, transport: "telegram-bot-api-webhook", revision: TELEGRAM_WEBHOOK_REVISION, botIdHash: sha256(String(this.bot.id)), allowedChatIdsHash: sha256([...this.allowedChatIds].sort().join("\n")), reactionFeedbackRoutesHash: sha256([...this.reactionUsersByChat.entries()] .sort(([left], [right]) => left.localeCompare(right)) .map(([chatId, userIds]) => `${chatId}:${[...userIds].sort().join(",")}`) .join("\n")), authority: "read-private-chat-updates-and-allowlisted-feedback", }; }
async ingest(store: JazzThoughtStore, update: TelegramBotUpdate): Promise<TelegramWebhookIngestResult> { const cursorId = `cursor:${this.id}`; const prior = await store.getSourceCursor(cursorId); const correlationId = `${this.id}:webhook-update:${update.update_id}`; const attemptId = newId("telegram_webhook"); const startedAt = new Date().toISOString(); await store.appendEvent(this.connectorEvent("started", attemptId, startedAt, { status: "started", transport: "telegram-bot-api-webhook", updateId: String(update.update_id), })); try { this.assertCursorCompatible(prior, String(this.bot.id)); const accepted = this.accepts(update); const feedbackUpdate = accepted && isFeedbackUpdate(update); const candidate = accepted ? await this.eventCandidate(store, update, this.bot, correlationId) : undefined; const candidates = candidate ? [candidate] : []; const completedAt = new Date().toISOString(); const cursor: SourceCursor = { id: cursorId, source: this.id, cursor: { revision: TELEGRAM_WEBHOOK_REVISION, botId: String(this.bot.id), highestUpdateId: Math.max(priorHighestUpdateId(prior), update.update_id), }, lastSuccessAt: completedAt, ...(prior?.lastFailureAt ? { lastFailureAt: prior.lastFailureAt } : {}), updatedAt: completedAt, }; const sourceCount = candidates.length; candidates.push(this.cursorEvent(attemptId, completedAt, cursor)); if (prior?.lastFailureAt && (!prior.lastSuccessAt || prior.lastFailureAt > prior.lastSuccessAt)) { candidates.push(this.connectorEvent("recovered", attemptId, completedAt, { status: "recovered", previousFailureAt: prior.lastFailureAt, })); } candidates.push(this.connectorEvent("completed", attemptId, completedAt, { status: accepted ? "updated" : "unchanged", updateId: String(update.update_id), accepted: accepted ? 1 : 0, ignored: accepted ? 0 : 1, })); const batch = await store.appendProducerBatch(candidates, cursor); const offeredEvents = batch.events.slice(0, sourceCount); const insertedIds = new Set(batch.inserted.map((event) => event.id)); const events = offeredEvents.filter((event) => insertedIds.has(event.id)); if (feedbackUpdate) await this.feedbackProjector(store, this.id); return { status: accepted ? "updated" : "unchanged", accepted: accepted ? 1 : 0, ignored: accepted ? 0 : 1, inserted: events.length, unchanged: sourceCount - events.length, events, cursor, }; } catch (error) { const failedAt = new Date().toISOString(); const message = error instanceof Error ? error.message : String(error); const durableCursor = await store.getSourceCursor(cursorId) ?? prior; const failedCursor: SourceCursor = { id: cursorId, source: this.id, cursor: durableCursor?.cursor ?? { revision: TELEGRAM_WEBHOOK_REVISION, botId: String(this.bot.id), highestUpdateId: -1, }, ...(durableCursor?.lastSuccessAt ? { lastSuccessAt: durableCursor.lastSuccessAt } : {}), lastFailureAt: failedAt, lastError: message, updatedAt: failedAt, }; await store.appendProducerBatch([this.connectorEvent("failed", attemptId, failedAt, { status: "failed", error: message, phase: "telegram-webhook-ingest", updateId: String(update.update_id), })], failedCursor); throw error; } }
private accepts(update: TelegramBotUpdate): boolean { const message = update.edited_message ?? update.message; if (message) { if (!this.allowedChatIds.has(String(message.chat.id))) return false; if (!telegramCorrectionCommand(message.text)) return true; return message.chat.type === "private" && Boolean(message.from) && this.reactionUsersByChat.get(String(message.chat.id))?.has(String(message.from!.id)) === true; } const reaction = update.message_reaction; if (!reaction || reaction.chat.type !== "private" || !reaction.user) return false; return this.reactionUsersByChat.get(String(reaction.chat.id))?.has(String(reaction.user.id)) === true; }
private async eventCandidate( store: JazzThoughtStore, update: TelegramBotUpdate, bot: TelegramBotUser, correlationId: string, ): Promise<EventCandidate | undefined> { const message = update.edited_message ?? update.message; if (!message) { return update.message_reaction ? this.reactionEventCandidate(store, update, bot, correlationId) : undefined; } const correction = telegramCorrectionCommand(message.text); if (correction) { return this.correctionEventCandidate(store, update, message, bot, correlationId, correction); } const edited = update.edited_message !== undefined; const occurredAt = new Date((message.edit_date ?? message.date) * 1_000).toISOString(); const senderId = message.from ? String(message.from.id) : String(message.chat.id); const senderName = message.from ? message.from.username ?? [message.from.first_name, message.from.last_name].filter(Boolean).join(" ") : message.chat.username ?? message.chat.title ?? "unknown"; const text = message.text ?? message.caption ?? ""; const attachments = await this.resolveAttachments(message); const payload = compact({ accountId: String(bot.id), accountUsername: bot.username, updateId: String(update.update_id), chatId: String(message.chat.id), messageId: String(message.message_id), senderId, senderName, chatType: message.chat.type === "private" ? "direct" : "channel", occurredAt: new Date(message.date * 1_000).toISOString(), editedAt: edited ? occurredAt : undefined, text, threadId: message.message_thread_id === undefined ? undefined : String(message.message_thread_id), replyToMessageId: message.reply_to_message?.message_id === undefined ? undefined : String(message.reply_to_message.message_id), attachments, transport: "telegram-bot-api-webhook", }); const conversationEligible = text.trim().length > 0 || attachments?.some((attachment) => ( attachment.kind === "image" && attachment.status === "stored" )) === true; return { type: conversationEligible ? "stream.thought.source.telegram.message" : "stream.thought.source.telegram.nonconversation", schemaVersion: 1, source: this.id, sourceKind: this.kind, externalId: `${bot.id}:${message.chat.id}:${message.message_id}`, idempotencyKey: sha256(canonicalJson({ botId: bot.id, chatId: message.chat.id, messageId: message.message_id, revision: message.edit_date ?? "original", })), occurredAt, actor: senderId, correlationId, privacy: "sensitive", payload, }; }
private async correctionEventCandidate( store: JazzThoughtStore, update: TelegramBotUpdate, message: z.infer<typeof messageSchema>, bot: TelegramBotUser, correlationId: string, command: TelegramCorrectionCommand, ): Promise<EventCandidate> { const chatId = String(message.chat.id); const messageId = String(message.message_id); const senderId = String(message.from!.id); const replyToMessageId = message.reply_to_message?.message_id === undefined ? undefined : String(message.reply_to_message.message_id); const resolution: CorrectionDeliveryResolution = command.status !== "valid" ? { status: command.status === "invalid" ? "invalid-command" : "replacement-too-long" } : !replyToMessageId ? { status: "missing-reply-target" } : await this.resolveCorrectionDelivery(store, chatId, replyToMessageId); const occurredAt = new Date((message.edit_date ?? message.date) * 1_000).toISOString(); const payload = compact({ accountId: String(bot.id), accountUsername: bot.username, updateId: String(update.update_id), chatId, messageId, senderId, chatType: "direct", occurredAt: new Date(message.date * 1_000).toISOString(), commandVersion: "telegram-correct-v1", replacementText: command.status === "valid" ? command.replacementText : undefined, replacementChars: command.replacementChars, replacementSha256: command.replacementSha256, replyToMessageId, resolutionStatus: resolution.status, deliveryReceiptEventId: resolution.deliveryReceiptEventId, runId: resolution.runId, outputEventId: resolution.outputEventId, sourceRootEventId: resolution.sourceRootEventId, transport: "telegram-bot-api-webhook", }); return { type: "stream.thought.source.telegram.correction", schemaVersion: 1, source: this.id, sourceKind: this.kind, externalId: `${bot.id}:${message.chat.id}:${message.message_id}:correction`, idempotencyKey: sha256(canonicalJson({ botId: bot.id, chatId: message.chat.id, messageId: message.message_id, revision: message.edit_date ?? "original", kind: "correction", })), occurredAt, actor: senderId, ...(resolution.sourceRootEventId ? { rootEventId: resolution.sourceRootEventId } : {}), ...(resolution.deliveryReceiptEventId ? { parentEventId: resolution.deliveryReceiptEventId } : {}), correlationId: resolution.deliveryReceiptExternalId ?? correlationId, privacy: "sensitive", payload, }; }
/** * Resolve message attachments. When image extraction is configured, admitted * PNG/JPEG images (photos or image documents) are downloaded via getFile, * validated, content-addressed, and stored under the artifact root. * The attachment metadata carries only an opaque relative path reference, * never raw bytes, base64, the bot token, file URLs, or absolute paths. */ private async resolveAttachments(message: z.infer<typeof messageSchema>): Promise<JsonObject[] | undefined> { if (!this.imageExtraction) return attachmentsFor(message); const { client, artifactRoot } = this.imageExtraction; const { baseUrl, token } = client.imageExtractionConfig; const attachments: JsonObject[] = []; // Attempt to extract at most one admitted image from photo array if (message.photo && message.photo.length > 0) { const photo = selectPhoto(message.photo); if (photo) { try { const artifact = await downloadTelegramImage(photo.file_id, { baseUrl, token, artifactRoot, }); attachments.push({ ...imageAttachmentMetadata(artifact), id: photo.file_unique_id, reference: `telegram-file:${photo.file_id}`, width: photo.width, height: photo.height, }); } catch (error) { if (!(error instanceof TelegramImagePermanentError)) throw error; attachments.push(compact({ id: photo.file_unique_id, kind: "image", sizeBytes: photo.file_size, reference: `telegram-file:${photo.file_id}`, status: "rejected", reason: error.code, })); } } else { const smallest = [...message.photo].sort((left, right) => (left.width * left.height) - (right.width * right.height))[0]!; attachments.push(compact({ id: smallest.file_unique_id, kind: "image", sizeBytes: smallest.file_size, reference: `telegram-file:${smallest.file_id}`, status: "rejected", reason: "size-limit", })); } } // A Telegram photo array and image document are mutually exclusive in valid // updates. If a malformed update supplies both, the photo path owns the one // permitted image slot and the document is ignored. if ((!message.photo || message.photo.length === 0) && message.document && isImageDocument(message.document)) { try { const artifact = await downloadTelegramImage(message.document.file_id, { baseUrl, token, artifactRoot, }); attachments.push(compact({ ...imageAttachmentMetadata(artifact), id: message.document.file_unique_id, name: message.document.file_name, reference: `telegram-file:${message.document.file_id}`, })); } catch (error) { if (!(error instanceof TelegramImagePermanentError)) throw error; attachments.push(compact({ id: message.document.file_unique_id, name: message.document.file_name, kind: "image", sizeBytes: message.document.file_size, reference: `telegram-file:${message.document.file_id}`, status: "rejected", reason: error.code, })); } } // Add non-image attachments without extraction. for (const [kind, file] of [ ["file", message.document && !isImageDocument(message.document) ? message.document : undefined], ["audio", message.audio], ["audio", message.voice], ["video", message.video], ] as const) { if (!file) continue; attachments.push(compact({ id: file.file_unique_id, name: file.file_name, mimeType: file.mime_type, sizeBytes: file.file_size, kind, reference: `telegram-file:${file.file_id}`, })); } return attachments.length > 0 ? attachments : undefined; }
private async reactionEventCandidate( store: JazzThoughtStore, update: TelegramBotUpdate, bot: TelegramBotUser, correlationId: string, ): Promise<EventCandidate> { const reaction = update.message_reaction!; const chatId = String(reaction.chat.id); const messageId = String(reaction.message_id); const senderId = String(reaction.user!.id); const resolution = await this.resolveReactionDelivery(store, chatId, messageId); const oldLabel = reactionLabel(reaction.old_reaction); const newLabel = reactionLabel(reaction.new_reaction); const feedbackAction = resolution.status === "resolved" ? newLabel && newLabel !== oldLabel ? "set" : oldLabel && !newLabel ? "retract" : "none" : "none"; const payload = compact({ accountId: String(bot.id), accountUsername: bot.username, updateId: String(update.update_id), chatId, messageId, senderId, chatType: "direct", occurredAt: new Date(reaction.date * 1_000).toISOString(), oldReactions: reaction.old_reaction.map(reactionDescriptor), newReactions: reaction.new_reaction.map(reactionDescriptor), mappingVersion: "telegram-thumbs-v1", resolutionStatus: resolution.status, feedbackAction, feedbackLabel: resolution.status === "resolved" ? newLabel : undefined, deliveryReceiptEventId: resolution.deliveryReceiptEventId, runId: resolution.runId, outputEventId: resolution.outputEventId, sourceRootEventId: resolution.sourceRootEventId, transport: "telegram-bot-api-webhook", }); return { type: "stream.thought.source.telegram.reaction", schemaVersion: 1, source: this.id, sourceKind: this.kind, externalId: `${bot.id}:reaction:${update.update_id}`, idempotencyKey: sha256(canonicalJson({ botId: bot.id, updateId: update.update_id, kind: "message_reaction" })), occurredAt: new Date(reaction.date * 1_000).toISOString(), actor: senderId, ...(resolution.sourceRootEventId ? { rootEventId: resolution.sourceRootEventId } : {}), ...(resolution.deliveryReceiptEventId ? { parentEventId: resolution.deliveryReceiptEventId } : {}), correlationId: resolution.deliveryReceiptExternalId ?? correlationId, privacy: "sensitive", payload, }; }
private async resolveCorrectionDelivery( store: JazzThoughtStore, chatId: string, messageId: string, ): Promise<CorrectionDeliveryResolution> { const resolution = await this.resolveReactionDelivery(store, chatId, messageId); if (resolution.status !== "resolved") return resolution; const run = await store.getRun(resolution.runId!); if (!run || run.status !== "completed" || !run.result) return { status: "incomplete-run" }; if (run.outputEventIds.length === 0) return { status: "missing-output" }; if (run.outputEventIds.length !== 1) return { status: "ambiguous-output" }; const [output, trigger, delivery] = await Promise.all([ store.getEvent(run.outputEventIds[0]!), store.getEvent(run.triggerEventId), store.getEvent(resolution.deliveryReceiptEventId!), ]); if (!output) return { status: "missing-output" }; if ( !trigger || trigger.source !== this.id || trigger.payload.chatId !== chatId || output.type !== "stream.thought.derived.message.observation" || output.source !== `agent:${run.agentId}` || output.sourceKind !== "agent" || output.actor !== run.agentId || output.parentEventId !== trigger.id || output.rootEventId !== trigger.rootEventId || output.payload.runId !== run.id || output.payload.inputEventId !== trigger.id || !delivery || delivery.type !== "stream.thought.action.telegram.send.delivered" || delivery.source !== `telegram-dispatcher:${this.id}:${chatId}` || delivery.sourceKind !== "system" || delivery.actor !== delivery.source || delivery.parentEventId !== output.id || delivery.rootEventId !== trigger.rootEventId || delivery.payload.chatId !== chatId || !Array.isArray(delivery.payload.runIds) || delivery.payload.runIds.length !== 1 || delivery.payload.runIds[0] !== run.id || resolution.sourceRootEventId !== trigger.rootEventId ) return { status: "invalid-lineage" }; return { ...resolution, outputEventId: output.id, }; }
private async resolveReactionDelivery( store: JazzThoughtStore, chatId: string, messageId: string, ): Promise<ReactionDeliveryResolution> { const matches = (await store.listEvents({ types: ["stream.thought.action.telegram.send.delivered"] })) .filter((event) => event.payload.chatId === chatId && event.payload.messageId === messageId); if (matches.length === 0) return { status: "unknown-delivery" }; if (matches.length !== 1) return { status: "ambiguous-delivery" }; const receipt = matches[0]!; const runIds = Array.isArray(receipt.payload.runIds) ? receipt.payload.runIds.filter((value): value is string => typeof value === "string") : []; if (runIds.length !== 1) return { status: "ambiguous-run" }; const run = await store.getRun(runIds[0]!); if (!run) return { status: "missing-run" }; const trigger = await store.getEvent(run.triggerEventId); if (!trigger || trigger.source !== this.id || trigger.payload.chatId !== chatId || receipt.source !== `telegram-dispatcher:${this.id}:${chatId}` || receipt.sourceKind !== "system" || receipt.actor !== receipt.source || receipt.rootEventId !== trigger.rootEventId) return { status: "invalid-lineage" }; return { status: "resolved", deliveryReceiptEventId: receipt.id, deliveryReceiptExternalId: receipt.externalId, runId: run.id, ...(run.outputEventIds[0] ? { outputEventId: run.outputEventIds[0] } : {}), sourceRootEventId: trigger.rootEventId, }; }
private assertCursorCompatible(cursor: SourceCursor | undefined, botId: string): void { const revision = cursor?.cursor.revision; if (revision !== undefined && (typeof revision !== "string" || !COMPATIBLE_TELEGRAM_REVISIONS.has(revision))) { throw new Error("Telegram bot cursor revision does not match connector configuration"); } const priorBotId = cursor?.cursor.botId; if (priorBotId !== undefined && priorBotId !== botId) throw new Error("Telegram bot cursor belongs to a different bot identity"); priorHighestUpdateId(cursor); }
private cursorEvent(correlationId: string, at: string, cursor: SourceCursor): EventCandidate { return { type: "stream.thought.connector.cursor.advanced", schemaVersion: 1, source: this.id, sourceKind: this.kind, externalId: cursor.id, idempotencyKey: `${correlationId}:cursor`, occurredAt: at, actor: this.id, correlationId, privacy: "sensitive", payload: { source: this.id, cursorId: cursor.id, cursor: cursor.cursor }, }; }
private connectorEvent( phase: "started" | "completed" | "failed" | "recovered", correlationId: string, at: string, payload: JsonObject, ): EventCandidate { const type = phase === "failed" ? "stream.thought.connector.failed" : phase === "recovered" ? "stream.thought.connector.recovered" : `stream.thought.connector.ingest.${phase}`; return { type, schemaVersion: 1, source: this.id, sourceKind: this.kind, externalId: correlationId, idempotencyKey: `${correlationId}:${phase}`, occurredAt: at, actor: this.id, correlationId, privacy: "sensitive", payload, }; }}
function isFeedbackUpdate(update: TelegramBotUpdate): boolean { if (update.message_reaction) return true; const message = update.edited_message ?? update.message; return Boolean(message && telegramCorrectionCommand(message.text));}
interface ReactionDeliveryResolution { status: "resolved" | "unknown-delivery" | "ambiguous-delivery" | "ambiguous-run" | "missing-run" | "invalid-lineage"; deliveryReceiptEventId?: string; deliveryReceiptExternalId?: string; runId?: string; outputEventId?: string; sourceRootEventId?: string;}
interface CorrectionDeliveryResolution { status: ReactionDeliveryResolution["status"] | "incomplete-run" | "ambiguous-output" | "missing-output" | "invalid-command" | "replacement-too-long" | "missing-reply-target"; deliveryReceiptEventId?: string; deliveryReceiptExternalId?: string; runId?: string; outputEventId?: string; sourceRootEventId?: string;}
interface TelegramCorrectionCommand { status: "valid" | "invalid" | "too-long"; replacementText?: string; replacementChars: number; replacementSha256: string;}
function telegramCorrectionCommand(text: string | undefined): TelegramCorrectionCommand | undefined { if (text === "/correct") { return { status: "invalid", replacementChars: 0, replacementSha256: sha256("") }; } if (!text?.startsWith(TELEGRAM_CORRECTION_PREFIX)) return undefined; const replacementText = text.slice(TELEGRAM_CORRECTION_PREFIX.length); if (replacementText.length === 0) { return { status: "invalid", replacementChars: 0, replacementSha256: sha256("") }; } if (replacementText.length > TELEGRAM_CORRECTION_MAX_CHARS) { return { status: "too-long", replacementChars: replacementText.length, replacementSha256: sha256(replacementText), }; } return { status: "valid", replacementText, replacementChars: replacementText.length, replacementSha256: sha256(replacementText), };}
function reactionLabel(reactions: Array<z.infer<typeof reactionTypeSchema>>): "positive" | "negative" | undefined { if (reactions.length !== 1 || reactions[0]?.type !== "emoji") return undefined; if (reactions[0].emoji === "👍") return "positive"; if (reactions[0].emoji === "👎") return "negative"; return undefined;}
function reactionDescriptor(reaction: z.infer<typeof reactionTypeSchema>): JsonObject { return compact({ type: reaction.type, emoji: reaction.emoji, customEmojiId: reaction.custom_emoji_id, });}
function attachmentsFor(message: z.infer<typeof messageSchema>): JsonObject[] | undefined { const attachments: JsonObject[] = []; const photo = message.photo?.at(-1); if (photo) attachments.push(compact({ id: photo.file_unique_id, kind: "image", sizeBytes: photo.file_size, reference: `telegram-file:${photo.file_id}`, })); for (const [kind, file] of [ ["file", message.document], ["audio", message.audio], ["audio", message.voice], ["video", message.video], ] as const) { if (!file) continue; attachments.push(compact({ id: file.file_unique_id, name: file.file_name, mimeType: file.mime_type, sizeBytes: file.file_size, kind, reference: `telegram-file:${file.file_id}`, })); } return attachments.length > 0 ? attachments : undefined;}
function priorHighestUpdateId(cursor: SourceCursor | undefined): number { const revision = cursor?.cursor.revision; const value = revision === "telegram-bot-api-v1" || revision === "telegram-bot-api-v2" ? cursor?.cursor.updateOffset : cursor?.cursor.highestUpdateId; if (value === undefined) return -1; if (typeof value !== "number" || !Number.isSafeInteger(value) || value < (revision === TELEGRAM_WEBHOOK_REVISION ? -1 : 0)) { throw new Error("Telegram bot cursor update high-water mark must be a safe integer"); } return revision === "telegram-bot-api-v1" || revision === "telegram-bot-api-v2" ? Math.max(-1, value - 1) : value;}
export function parseTelegramBotUpdate(value: unknown): TelegramBotUpdate { return updateSchema.parse(value);}
function compact(value: Record<string, unknown>): JsonObject { return Object.fromEntries(Object.entries(value).filter(([, child]) => child !== undefined)) as JsonObject;}
function boundedPositiveInteger(value: number, label: string, maximum: number): number { if (!Number.isSafeInteger(value) || value <= 0 || value > maximum) throw new Error(`${label} must be a positive integer <= ${maximum}`); return value;}
function required(value: string, label: string): string { const normalized = value.trim(); if (!normalized) throw new Error(`${label} is required`); return normalized;}
function webhookSecret(value: string): string { const secret = required(value, "Telegram webhook secret"); if (!/^[A-Za-z0-9_-]{1,256}$/.test(secret)) { throw new Error("Telegram webhook secret must use 1-256 ASCII letters, digits, underscores, or hyphens"); } return secret;}