diff --git a/spec/architecture.md b/spec/architecture.md index fc235c2..8e563d3 100644 --- a/spec/architecture.md +++ b/spec/architecture.md @@ -50,6 +50,8 @@ Projectors are consumers with no special delivery path. They follow Jazz subscri External actions are owned by separate destination-specific dispatcher processes. A dispatcher reads completed candidate activity from Jazz, filters it against channel policy, accumulates and renders batches, applies destination velocity limits, performs the action, and appends started/delivered/failed evidence. Producers and consumers never wait on dispatcher policy or destination throughput. The first implementation supports Telegram delivery only. +The Telegram dispatcher also owns one ephemeral presence action for direct-reply agents: while a recent durable `running` run is rooted in an allowlisted Telegram message for that exact chat and agent, it refreshes Bot API `typing` before Telegram's five-second expiry. This action contains no model or source content, is bounded by the direct-reply allowlist, stops when no matching run remains, and is best-effort: failure cannot fail the run or message-delivery loop. Unlike human-visible message delivery, typing is repeatable transient UI state and creates no append-only delivery claim. + ### 8. Local interface and Review authority The interface is a loopback server and dense activity/Review page. Activity, execution, source-health, lineage, and adapter inventory are read projections. Review adds one separately configured mutation: an allowlisted OAuth browser may append a fixed decision after CSRF validation and a body-bound proxy-to-inspector capability check. Basic remains read-only. The route cannot create prompts, run models, export datasets, activate adapters, publish, or perform arbitrary Jazz mutation. Review evidence affects dataset projection only; it is not deployment authority. diff --git a/spec/recovery.md b/spec/recovery.md index 5b67dc1..259f3a9 100644 --- a/spec/recovery.md +++ b/spec/recovery.md @@ -67,6 +67,7 @@ Budget denial still follows the existing terminal semantics: the blocked run and - The dispatcher appends a `started` claim before external delivery. Only the process that inserts that claim may call the Bot API. - `delivered` records the Telegram message id. `failed` preserves the transport error. A crash after `started` is status-unknown and is not blindly retried because Telegram does not provide an idempotency key for `sendMessage`; replay must be explicit to avoid duplicate human-visible sends. - Velocity accounting reads recent durable `delivered` receipts, so restarting a dispatcher does not reset the channel's rate limit. +- Typing presence is rebuilt from recent durable `running` runs on every dispatcher cycle and kept only in process-local refresh state. Restart may repeat `sendChatAction` because it is transient UI state, not a human-visible message. Stale runs beyond the bounded presence age are ignored; Bot API failure is suppressed until the next refresh interval and never changes run or delivery evidence. ## Incident recovery diff --git a/spec/security.md b/spec/security.md index c9e1168..1ec7c85 100644 --- a/spec/security.md +++ b/spec/security.md @@ -10,7 +10,7 @@ thought stream has unusually broad read access. Its first security property is c - Model capabilities: receive a bounded context, produce typed proposals. - Action capabilities: send, publish, edit, delete, transact. -The system implements ingress and model capabilities plus one narrow action capability: Telegram delivery. Telegram ingress and egress are separate processes. The webhook receiver has no send or registration path. It binds to loopback, requires the exact configured secret header using constant-time comparison, rejects malformed or oversized bodies before persistence, and is exposed only through an operator-owned HTTPS reverse proxy. The dispatcher requires an enabled channel, source and actor allowlists, a destination velocity policy, and explicit started/delivered/failed receipt events. +The system implements ingress and model capabilities plus one narrow action capability: Telegram delivery. Telegram ingress and egress are separate processes. The webhook receiver has no send or registration path. It binds to loopback, requires the exact configured secret header using constant-time comparison, rejects malformed or oversized bodies before persistence, and is exposed only through an operator-owned HTTPS reverse proxy. The dispatcher requires an enabled channel, source and actor allowlists, a destination velocity policy, and explicit started/delivered/failed receipt events. Ephemeral Telegram typing uses the same egress-only credential and direct-reply source/actor/agent/chat allowlists. It sends only `{ chat_id, action: "typing" }`, never source or model content, and remains outside ingress and consumer authority. Action filtering happens at the egress boundary. Producers and consumers continue at source speed; the dispatcher alone decides which completed candidate activity may cross into a channel, how candidates are batched, and when destination capacity is available. Failed-run delivery is a separate allowlisted status and may include only a classified diagnostic. The dispatcher never reconstructs content from run traces and never renders `errorText` or arbitrary diagnostic strings. diff --git a/spec/testing.md b/spec/testing.md index 013f972..e0e11a8 100644 --- a/spec/testing.md +++ b/spec/testing.md @@ -125,7 +125,7 @@ Live Jetstream transport tests use an injected in-process WebSocket fixture, nev Telegram spool fixtures use normalized synthetic private messages, exact route identifiers, and opaque attachment references. Tests verify strict schema rejection, sensitive event classification, bounded restart-safe ingestion, incomplete trailing-line behavior, cursor advancement after durable records, and fail-closed detection when consumed spool content changes. They do not use a bot token, Telegram API, existing channel spool, or MessageChannel. -Telegram webhook tests use an in-process loopback receiver and synthetic Bot API updates. They verify exact path and method handling, constant-time secret-header admission, JSON/body bounds, allowlisted chat and reaction filtering, serial durable processing, deterministic replay, out-of-order lower update admission, migration from the former polling cursor revision, `2xx` only after durable settlement, `5xx` on store/projection failure, explicit registration with `max_connections=1`, and separation from dispatcher send authority. They never load a live token or webhook secret and never contact Telegram. +Telegram webhook and dispatcher tests use an in-process loopback receiver and synthetic Bot API updates. They verify exact path and method handling, constant-time secret-header admission, JSON/body bounds, allowlisted chat and reaction filtering, serial durable processing, deterministic replay, out-of-order lower update admission, migration from the former polling cursor revision, `2xx` only after durable settlement, `5xx` on store/projection failure, explicit registration with `max_connections=1`, and separation from dispatcher send authority. Dispatcher tests prove that only a recent running direct-reply run rooted in the exact allowlisted Telegram chat emits and refreshes one content-free `typing` action, terminal/stale/wrong-route runs do not, and chat-action failure does not block durable reply delivery. They never load a live token or webhook secret and never contact Telegram. Credential-compartment tests use generated synthetic values only. They prove webhook, consumer, dispatcher, and Jetstream outputs contain exactly their allowlisted variable names, no value is printed, malformed/duplicate assignments fail closed, Git/public destinations are refused, and generated directory/file modes are `0700`/`0600`. Live credential files are never read by the test or implementation gate. diff --git a/src/bridges/telegram-dispatcher.ts b/src/bridges/telegram-dispatcher.ts index 6fdbd65..3746829 100644 --- a/src/bridges/telegram-dispatcher.ts +++ b/src/bridges/telegram-dispatcher.ts @@ -10,6 +10,8 @@ const RECEIPT_TYPES = [ "stream.thought.action.telegram.send.delivered", "stream.thought.action.telegram.send.failed", ]; +const TELEGRAM_TYPING_REFRESH_MS = 4_000; +const TELEGRAM_TYPING_MAX_RUN_AGE_MS = 5 * 60_000; export interface TelegramChannelDispatcherOptions { id: string; @@ -50,6 +52,13 @@ export interface TelegramOperationalNoticeResult { error?: string; } +export interface TelegramTypingCycleResult { + activeRuns: number; + sent: number; + failed: number; + deferred: number; +} + interface NotificationCandidate { run: AgentRun; kind: "observation" | "failure"; @@ -72,6 +81,8 @@ export class TelegramChannelDispatcher { private readonly windowMs: number; private readonly maxLikesPerDigest: number; private readonly likeDigestDelayMs: number; + private lastTypingAttemptAt: number | undefined; + private typingFailureActive = false; constructor(options: TelegramChannelDispatcherOptions) { this.id = required(options.id, "Telegram channel dispatcher id"); @@ -206,6 +217,37 @@ export class TelegramChannelDispatcher { }; } + async refreshTyping( + store: JazzThoughtStore, + options: { now?: Date; signal?: AbortSignal } = {}, + ): Promise { + const now = options.now ?? new Date(); + const activeRuns = (await Promise.all((await store.listRuns({ status: "running" })) + .filter((run) => this.directReplyAgentIds.has(run.agentId)) + .filter((run) => recentRunningRun(run, now)) + .map(async (run) => ({ run, trigger: await store.getEvent(run.triggerEventId) })))) + .filter(({ run, trigger }) => trigger !== undefined && this.isDirectReplyRoute(run, trigger)); + if (activeRuns.length === 0) { + this.lastTypingAttemptAt = undefined; + return { activeRuns: 0, sent: 0, failed: 0, deferred: 0 }; + } + if (this.lastTypingAttemptAt !== undefined + && now.getTime() - this.lastTypingAttemptAt < TELEGRAM_TYPING_REFRESH_MS) { + return { activeRuns: activeRuns.length, sent: 0, failed: 0, deferred: 1 }; + } + this.lastTypingAttemptAt = now.getTime(); + try { + await this.client.sendTyping(this.chatId, options.signal); + if (this.typingFailureActive) console.warn("Telegram typing action recovered"); + this.typingFailureActive = false; + return { activeRuns: activeRuns.length, sent: 1, failed: 0, deferred: 0 }; + } catch { + if (!this.typingFailureActive) console.warn("Telegram typing action unavailable"); + this.typingFailureActive = true; + return { activeRuns: activeRuns.length, sent: 0, failed: 1, deferred: 0 }; + } + } + async sendOperationalNotice( store: JazzThoughtStore, text: string, @@ -343,6 +385,14 @@ export class TelegramChannelDispatcher { && group[0]!.trigger.type === "stream.thought.source.telegram.message" && this.directReplyAgentIds.has(group[0]!.run.agentId); } + + private isDirectReplyRoute(run: AgentRun, trigger: ThoughtEvent): boolean { + return trigger.type === "stream.thought.source.telegram.message" + && this.directReplyAgentIds.has(run.agentId) + && this.allowedSources.has(trigger.source) + && (this.allowedActors.size === 0 || this.allowedActors.has(trigger.actor)) + && trigger.payload.chatId === this.chatId; + } } function formatNotification(group: NotificationCandidate[], directReplyAgentIds: Set): string { @@ -423,6 +473,13 @@ function completedAt(run: AgentRun): string { return run.completedAt ?? run.updatedAt; } +function recentRunningRun(run: AgentRun, now: Date): boolean { + const startedAt = Date.parse(run.startedAt ?? run.createdAt); + return Number.isFinite(startedAt) + && startedAt <= now.getTime() + TELEGRAM_TYPING_REFRESH_MS + && startedAt > now.getTime() - TELEGRAM_TYPING_MAX_RUN_AGE_MS; +} + function notificationIdentity(run: AgentRun): string { return `${run.executionKey}:${run.status}`; } diff --git a/src/cli.ts b/src/cli.ts index 8c3158e..a279112 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -377,10 +377,13 @@ try { const notifications = []; for (const runtime of channelRuntimes) { if (!runtime.dispatcher || !runtime.notification?.enabled || !bootedChannels.has(runtime.channel.id)) continue; - const result = await runtime.dispatcher.sendPending(store, { - ...(process.argv.includes("--notify-existing") ? {} : { since: activeSince.get(runtime.channel.id)! }), - includeNormal: runtime.notification.includeNormal, - }); + const [, result] = await Promise.all([ + runtime.dispatcher.refreshTyping(store), + runtime.dispatcher.sendPending(store, { + ...(process.argv.includes("--notify-existing") ? {} : { since: activeSince.get(runtime.channel.id)! }), + includeNormal: runtime.notification.includeNormal, + }), + ]); notifications.push({ channelId: runtime.channel.id, ...result }); delivered += result.delivered; failed += result.failed; diff --git a/src/connectors/telegram-bot.ts b/src/connectors/telegram-bot.ts index 1b925d0..6b85152 100644 --- a/src/connectors/telegram-bot.ts +++ b/src/connectors/telegram-bot.ts @@ -189,6 +189,15 @@ export class TelegramBotClient { }; } + async sendTyping(chatId: string, signal?: AbortSignal): Promise { + 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); + } + private async call( method: string, body: Record, diff --git a/test/telegram-bot.test.ts b/test/telegram-bot.test.ts index d77bb34..e50bbd6 100644 --- a/test/telegram-bot.test.ts +++ b/test/telegram-bot.test.ts @@ -19,6 +19,7 @@ import { JetstreamConnector } from "../src/connectors/jetstream.js"; import type { JazzThoughtStore } from "../src/jazz/store.js"; import { describeRunResult } from "../src/projections/activity.js"; import { projectTrainingExamples } from "../src/training/judgments.js"; +import type { AgentRun } from "../src/store/types.js"; import { temporaryProject, testDeclarationEnvironment, testStore } from "./helpers.js"; const stores: JazzThoughtStore[] = []; @@ -27,6 +28,7 @@ const servers: Server[] = []; const execFileAsync = promisify(execFile); afterEach(async () => { + vi.restoreAllMocks(); await Promise.all(stores.splice(0).map((store) => store.close())); await Promise.all(roots.splice(0).map((root) => fs.rm(root, { recursive: true, force: true }))); await Promise.all(servers.splice(0).map((server) => new Promise((resolve) => server.close(() => resolve())))); @@ -413,6 +415,133 @@ describe("TelegramBotConnector", () => { }); describe("TelegramChannelDispatcher", () => { + test("refreshes typing only for a recent running direct-reply run in the exact chat", async () => { + const fixture = await telegramFixture(); + const project = await temporaryProject(); + roots.push(project); + const store = testStore(project); + stores.push(store); + const connector = new TelegramBotConnector({ + id: "telegram:thoughtstream-bot", + bot: telegramBotIdentity(), + allowedChatIds: ["123456789"], + }); + const trigger = (await connector.ingest(store, parseTelegramBotUpdate(fixture.updates[0]))).events[0]!; + const startedAt = "2026-07-28T05:00:00.000Z"; + const run = runningTelegramRun(trigger.id, "telegram-conversation", startedAt); + await store.upsertRun(run); + const dispatcher = new TelegramChannelDispatcher({ + id: "telegram-notifier:typing", + client: new TelegramBotClient({ token: "fixture-token", baseUrl: fixture.baseUrl }), + chatId: "123456789", + allowedSources: ["telegram:thoughtstream-bot"], + allowedActors: ["123456789"], + directReplyAgentIds: ["telegram-conversation"], + }); + + await expect(dispatcher.refreshTyping(store, { + now: new Date("2026-07-28T05:00:01.000Z"), + })).resolves.toEqual({ activeRuns: 1, sent: 1, failed: 0, deferred: 0 }); + await expect(dispatcher.refreshTyping(store, { + now: new Date("2026-07-28T05:00:04.000Z"), + })).resolves.toEqual({ activeRuns: 1, sent: 0, failed: 0, deferred: 1 }); + await expect(dispatcher.refreshTyping(store, { + now: new Date("2026-07-28T05:00:05.000Z"), + })).resolves.toEqual({ activeRuns: 1, sent: 1, failed: 0, deferred: 0 }); + expect(fixture.typingActions).toEqual([ + { chatId: "123456789", action: "typing" }, + { chatId: "123456789", action: "typing" }, + ]); + + await store.upsertRun({ + ...run, + status: "completed", + completedAt: "2026-07-28T05:00:06.000Z", + updatedAt: "2026-07-28T05:00:06.000Z", + result: { summary: "done", tags: ["conversation"], importance: "normal", confidence: 0.5 }, + }); + await store.upsertRun(runningTelegramRun( + trigger.id, + "other-agent", + "2026-07-28T05:00:06.000Z", + "run:wrong-agent", + )); + await store.upsertRun(runningTelegramRun( + trigger.id, + "telegram-conversation", + "2026-07-28T04:50:00.000Z", + "run:stale", + )); + const wrongChatTrigger = await store.appendEvent({ + type: "stream.thought.source.telegram.message", + schemaVersion: 1, + source: "telegram:thoughtstream-bot", + sourceKind: "telegram", + externalId: "wrong-chat", + idempotencyKey: "wrong-chat", + occurredAt: "2026-07-28T05:00:06.000Z", + actor: "123456789", + correlationId: "wrong-chat", + privacy: "sensitive", + payload: { ...trigger.payload, chatId: "999", messageId: "11", text: "wrong route" }, + }); + await store.upsertRun(runningTelegramRun( + wrongChatTrigger.event.id, + "telegram-conversation", + "2026-07-28T05:00:06.000Z", + "run:wrong-chat", + )); + await expect(dispatcher.refreshTyping(store, { + now: new Date("2026-07-28T05:00:10.000Z"), + })).resolves.toEqual({ activeRuns: 0, sent: 0, failed: 0, deferred: 0 }); + expect(fixture.typingActions).toHaveLength(2); + expect(await store.listEvents({ types: ["stream.thought.action.telegram.send.started"] })).toHaveLength(0); + }); + + test("typing failure is best-effort and cannot block the terminal reply", async () => { + const warning = vi.spyOn(console, "warn").mockImplementation(() => {}); + const fixture = await telegramFixture(); + const project = await temporaryProject(); + roots.push(project); + const store = testStore(project); + stores.push(store); + const connector = new TelegramBotConnector({ + id: "telegram:thoughtstream-bot", + bot: telegramBotIdentity(), + allowedChatIds: ["123456789"], + }); + const trigger = (await connector.ingest(store, parseTelegramBotUpdate(fixture.updates[0]))).events[0]!; + const run = runningTelegramRun(trigger.id, "telegram-conversation", "2026-07-28T05:00:00.000Z"); + await store.upsertRun(run); + const dispatcher = new TelegramChannelDispatcher({ + id: "telegram-notifier:typing-failure", + client: new TelegramBotClient({ token: "fixture-token", baseUrl: fixture.baseUrl }), + chatId: "123456789", + allowedSources: ["telegram:thoughtstream-bot"], + allowedActors: ["123456789"], + directReplyAgentIds: ["telegram-conversation"], + runStatuses: ["completed"], + }); + fixture.setTypingFailure(true); + await expect(dispatcher.refreshTyping(store, { + now: new Date("2026-07-28T05:00:01.000Z"), + })).resolves.toEqual({ activeRuns: 1, sent: 0, failed: 1, deferred: 0 }); + expect(warning).toHaveBeenCalledWith("Telegram typing action unavailable"); + + await store.upsertRun({ + ...run, + status: "completed", + completedAt: "2026-07-28T05:00:02.000Z", + updatedAt: "2026-07-28T05:00:02.000Z", + result: { summary: "The reply still arrives.", tags: ["conversation"], importance: "normal", confidence: 0.5 }, + }); + await expect(dispatcher.sendPending(store, { includeNormal: true })).resolves.toMatchObject({ + delivered: 1, + failed: 0, + }); + expect(fixture.sentMessages).toEqual([{ chatId: "123456789", text: "The reply still arrives." }]); + }); + test("carries an allowlisted Telegram blip through an agent run to a tied delivery receipt", async () => { const fixture = await telegramFixture(); const project = await temporaryProject(); @@ -708,18 +837,22 @@ describe("TelegramChannelDispatcher", () => { async function telegramFixture(): Promise<{ baseUrl: string; sentMessages: Array<{ chatId: string; text: string }>; + typingActions: Array<{ chatId: string; action: string }>; + setTypingFailure: (failed: boolean) => void; sentMessageIds: string[]; updates: Array>; webhookRegistrations: Array>; webhookDeletions: Array>; }> { const sentMessages: Array<{ chatId: string; text: string }> = []; + const typingActions: Array<{ chatId: string; action: string }> = []; const sentMessageIds: string[] = []; const webhookRegistrations: Array> = []; const webhookDeletions: Array> = []; let webhookUrl = ""; let webhookMaxConnections: number | undefined; let webhookAllowedUpdates: string[] | undefined; + let typingFailure = false; const updates: Array> = [ { update_id: 100, @@ -791,6 +924,13 @@ async function telegramFixture(): Promise<{ }, }); } + if (request.url?.endsWith("/sendChatAction")) { + typingActions.push({ chatId: String(body.chat_id), action: String(body.action) }); + if (typingFailure) { + return json(response, { ok: false, error_code: 503, description: "fixture unavailable" }, 503); + } + return json(response, { ok: true, result: true }); + } return json(response, { ok: false, error_code: 404, description: "not found" }, 404); }); servers.push(server); @@ -803,6 +943,8 @@ async function telegramFixture(): Promise<{ return { baseUrl: `http://127.0.0.1:${address.port}`, sentMessages, + typingActions, + setTypingFailure: (failed) => { typingFailure = failed; }, sentMessageIds, updates, webhookRegistrations, @@ -814,6 +956,32 @@ function telegramBotIdentity(): TelegramBotUser { return { id: 8765422491, is_bot: true, first_name: "ThoughtStream", username: "CameronStreamBot" }; } +function runningTelegramRun( + triggerEventId: string, + agentId: string, + startedAt: string, + id = "run:telegram-typing", +): AgentRun { + return { + id, + executionKey: `${agentId}:${triggerEventId}`, + triggerEventId, + agentId, + agentVersion: 4, + status: "running", + inputEventIds: [triggerEventId], + outputEventIds: [], + attempt: 1, + provider: "letta-agent-sdk", + model: "fixture", + promptHash: "fixture-prompt", + contextManifest: {}, + createdAt: startedAt, + startedAt, + updatedAt: startedAt, + }; +} + function telegramMessageUpdate(updateId: number, messageId: number, text: string): Record { return { update_id: updateId,