From 020dc8261ce4ff1e49301ac3210815402c1af1b8 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Fri, 3 Jul 2026 00:05:41 +0800 Subject: [PATCH 1/5] fix: deliver streamed Telegram content exactly once and in order MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Telegram frontend double-delivered every streamed assistant/thinking block: the final session.message rebroadcast sent the full content but never dropped the delta buffer, so the idle flush re-sent it. Worse, any unrelated session.message (e.g. a tool_call mid-stream) flush-deleted the LIVE buffer, dropping all later deltas and duplicating the flushed prefix when the final broadcast arrived. Long messages also chunked in parallel (out-of-order >4000-char output, "Done." overtaking content), and thinking flushes died silently on Telegram markdown parse errors. Extract the buffering into StreamRelay (stream.ts): - final full broadcast is authoritative: send the unflushed tail, drop the buffer - interleaving messages partial-flush live buffers via a per-buffer flushed offset instead of deleting them — later deltas keep appending - all sends go through one promise chain: chunks sequential, line-boundary splits, "Done." queued after the flush - thinking sends retry as plain text when Markdown parsing fails Tests exercise the real StreamRelay through the RelayApi seam. Co-Authored-By: Claude Fable 5 --- src/frontends/telegram/index.ts | 133 +++-------- src/frontends/telegram/stream.ts | 248 ++++++++++++++++++++ src/tests/telegram-stream.test.ts | 369 ++++++++++++++++++++++++++++++ 3 files changed, 651 insertions(+), 99 deletions(-) create mode 100644 src/frontends/telegram/stream.ts create mode 100644 src/tests/telegram-stream.test.ts diff --git a/src/frontends/telegram/index.ts b/src/frontends/telegram/index.ts index a2bd5c4..3237c21 100644 --- a/src/frontends/telegram/index.ts +++ b/src/frontends/telegram/index.ts @@ -34,6 +34,7 @@ import type { ToolInfo, } from "../../protocol/types.js"; import type { AttachedClient } from "../../daemon/session.js"; +import { StreamRelay, type RelayApi } from "./stream.js"; /** One AskUserQuestion question as it arrives in the tool input. */ interface AskQuestion { @@ -65,11 +66,11 @@ interface UserState { attachedSessionName: string | null; clientId: string; /** - * Per-messageId accumulator for streaming content. Telegram's rate limits - * make per-token streaming infeasible, so we buffer deltas and flush - * whenever Claude moves to a new message OR the session goes idle. + * Per-user stream relay. Buffers streaming deltas (Telegram's rate limits + * make per-token streaming infeasible) and delivers content exactly once, + * in order — see stream.ts for the flush rules. */ - streaming: Map; + relay: StreamRelay; /** * message_id of the live "⏹ Stop" control shown while a turn runs, so we * show exactly one per turn and remove it when the turn ends. Null when no @@ -89,6 +90,11 @@ export class TelegramFrontend implements Frontend { #users = new Map(); /** Short-token → pending approval, for inline-keyboard tool approvals. */ #approvals = new Map(); + /** Send surface handed to each user's StreamRelay. */ + #relayApi: RelayApi = { + sendMessage: (chatId, text, opts) => + this.#bot.api.sendMessage(chatId, text, opts as never), + }; constructor(botToken: string, allowedUserIds: number[]) { this.#bot = new Bot(botToken); @@ -329,7 +335,7 @@ export class TelegramFrontend implements Frontend { this.#manager.disconnectClient(state.clientId); state.attachedSessionId = null; state.attachedSessionName = null; - state.streaming.clear(); + state.relay.clear(); // Remove the ⏹ Stop button for the old session. The old session's // idle status_change (which normally deletes it) won't arrive after // we disconnected, so it would otherwise linger in the chat. @@ -372,7 +378,7 @@ export class TelegramFrontend implements Frontend { const name = state.attachedSessionName; state.attachedSessionId = null; state.attachedSessionName = null; - state.streaming.clear(); + state.relay.clear(); if (state.stopMessageId !== null) { this.#bot.api.deleteMessage(chatId, state.stopMessageId).catch(() => {}); state.stopMessageId = null; @@ -496,7 +502,7 @@ export class TelegramFrontend implements Frontend { } else { // Fall back to plain text for very long results const plain = lines.join("\n").replace(/\\([_*[\]()~`>#+\-=|{}.!\\])/g, "$1"); - this.#sendChunked(ctx.chat!.id, plain); + state.relay.sendChunked(ctx.chat!.id, plain); } } @@ -715,67 +721,30 @@ export class TelegramFrontend implements Frontend { #forwardToChat(chatId: number, userId: number, msg: DaemonMessage): void { const state = this.#users.get(userId); - const send = (text: string, opts?: { parse_mode?: string }) => - this.#bot.api.sendMessage(chatId, text, opts as Record).catch(() => {}); + if (!state) return; + const relay = state.relay; switch (msg.type) { case "session.message": { const m = msg; - // Flush any buffered streams that are stale (different messageId). - if (state) this.#flushStale(chatId, state, m.messageId); - - switch (m.role) { - case "user": - // Echo of our own send — no need to replay. - break; - case "assistant": - if (m.content) { - this.#sendChunked(chatId, m.content); - } else if (state) { - // Empty assistant = start of a streaming block. - state.streaming.set(m.messageId, { role: "assistant", content: "" }); - } - break; - case "thinking": - if (m.content) { - send(`💭 _thinking_\n${m.content.slice(0, 800)}`, { parse_mode: "Markdown" }); - } else if (state) { - state.streaming.set(m.messageId, { role: "thinking", content: "" }); - } - break; - case "tool_call": { - if (!m.tool) break; - const phase = m.tool.state.phase; - if (phase === "waiting_confirmation" && "approvalId" in m.tool.state) { - // Inline Approve/Deny (or AskUserQuestion option buttons) keyed - // to the exact approvalId — handles concurrent approvals. - this.#sendApproval(chatId, userId, m.tool); - } else if (phase === "executing") { - send(`⚡ ${m.tool.name}`); - } else if (phase === "completed") { - send(`✓ ${m.tool.name}`); - } else if (phase === "cancelled") { - send(`✗ ${m.tool.name} cancelled`); - } - break; - } - case "system": - if (m.content) send(`⚠️ ${m.content}`); - break; - case "info": - // Quiet by default on mobile. - break; + // Content buffering / exactly-once flushing lives in the relay. + relay.handleMessage(chatId, m); + if ( + m.role === "tool_call" && + m.tool && + m.tool.state.phase === "waiting_confirmation" && + "approvalId" in m.tool.state + ) { + // Inline Approve/Deny (or AskUserQuestion option buttons) keyed + // to the exact approvalId — handles concurrent approvals. + this.#sendApproval(chatId, userId, m.tool); } break; } case "session.message.delta": { // Buffer; don't stream to Telegram per-token (rate limits). - if (!state) break; - const buf = state.streaming.get(msg.messageId); - if (buf && msg.contentAppend) { - buf.content += msg.contentAppend; - } + relay.handleDelta(chatId, msg); break; } @@ -785,7 +754,7 @@ export class TelegramFrontend implements Frontend { if (active) { // Turn started — show a one-tap ⏹ Stop control (once per turn). // Mobile parity with Esc on desktop: no need to type /interrupt. - if (state?.attachedSessionId && state.stopMessageId === null) { + if (state.attachedSessionId && state.stopMessageId === null) { const sid = state.attachedSessionId; const kb = new InlineKeyboard().text("⏹ Stop", `stop:${sid}`); this.#bot.api @@ -799,21 +768,16 @@ export class TelegramFrontend implements Frontend { } } else if (msg.status === "idle" || msg.status === "error") { // Turn ended — remove the Stop control. - if (state && state.stopMessageId !== null) { + if (state.stopMessageId !== null) { this.#bot.api.deleteMessage(chatId, state.stopMessageId).catch(() => {}); state.stopMessageId = null; } if (msg.status === "idle") { - // Flush any remaining buffered streams. - if (state) { - for (const [, buf] of state.streaming) { - this.#flushBuffer(chatId, buf); - } - state.streaming.clear(); - } - send("✅ Done."); + // Flush any remaining buffered streams, then confirm — the relay + // queues "✅ Done." after the content so it can't overtake it. + relay.flushIdle(chatId); } else { - send("❌ Error."); + relay.send(chatId, "❌ Error."); } } break; @@ -998,35 +962,6 @@ export class TelegramFrontend implements Frontend { } } - /** Flush buffered streams whose messageId is no longer current. */ - #flushStale(chatId: number, state: UserState, currentMessageId: string): void { - for (const [id, buf] of state.streaming) { - if (id !== currentMessageId) { - this.#flushBuffer(chatId, buf); - state.streaming.delete(id); - } - } - } - - #flushBuffer(chatId: number, buf: { role: string; content: string }): void { - if (!buf.content) return; - if (buf.role === "thinking") { - this.#bot.api - .sendMessage(chatId, `💭 _thinking_\n${buf.content.slice(0, 1500)}`, { - parse_mode: "Markdown", - } as Record) - .catch(() => {}); - } else { - this.#sendChunked(chatId, buf.content); - } - } - - #sendChunked(chatId: number, text: string): void { - for (let i = 0; i < text.length; i += 4000) { - this.#bot.api.sendMessage(chatId, text.slice(i, i + 4000)).catch(() => {}); - } - } - // ── Helpers ─────────────────────────────────────────────────────────── #getOrCreate(userId: number): UserState { @@ -1037,7 +972,7 @@ export class TelegramFrontend implements Frontend { attachedSessionId: null, attachedSessionName: null, clientId: `telegram:${userId}`, - streaming: new Map(), + relay: new StreamRelay(this.#relayApi), stopMessageId: null, }; this.#users.set(userId, state); diff --git a/src/frontends/telegram/stream.ts b/src/frontends/telegram/stream.ts new file mode 100644 index 0000000..9a5c4ae --- /dev/null +++ b/src/frontends/telegram/stream.ts @@ -0,0 +1,248 @@ +/** + * Telegram streaming relay — buffers daemon stream deltas and delivers them + * to a chat exactly once, in order. + * + * Telegram's rate limits make per-token streaming infeasible, so deltas are + * accumulated per messageId and flushed at well-defined points: + * + * - when the message's own final full broadcast arrives (authoritative — + * the buffer is dropped so idle can't re-send it), + * - when an unrelated message (e.g. a tool_call) interleaves mid-stream — + * only the *unflushed tail* is sent and the buffer stays live, so later + * deltas keep appending (per-buffer flushed offset), + * - when the session goes idle, + * - when the user detaches/switches sessions (so buffered content is never + * silently discarded). + * + * All sends go through a single per-relay promise chain, so chunks of long + * messages, tool lines, and the final "✅ Done." marker arrive in order. + */ + +import type { + SessionMessage, + SessionMessageDelta, + ToolState, +} from "../../protocol/types.js"; + +/** Minimal Telegram API surface the relay needs (test-injectable). */ +export interface RelayApi { + sendMessage( + chatId: number, + text: string, + opts?: Record, + ): Promise; +} + +interface StreamBuf { + role: string; + content: string; + /** How much of `content` has already been sent to the chat. */ + flushed: number; +} + +/** Telegram message size limit is 4096; leave headroom. */ +const CHUNK_LIMIT = 4000; + +/** + * Split text into Telegram-sized chunks, preferring line boundaries so a + * message isn't cut mid-line when a newline exists within the window. + */ +export function chunkText(text: string, limit = CHUNK_LIMIT): string[] { + if (text.length <= limit) return text.length > 0 ? [text] : []; + const chunks: string[] = []; + let rest = text; + while (rest.length > limit) { + const nl = rest.lastIndexOf("\n", limit); + if (nl > 0) { + chunks.push(rest.slice(0, nl)); + rest = rest.slice(nl + 1); // drop the boundary newline + } else { + chunks.push(rest.slice(0, limit)); + rest = rest.slice(limit); + } + } + if (rest.length > 0) chunks.push(rest); + return chunks; +} + +export class StreamRelay { + #api: RelayApi; + /** Per-messageId accumulator for streaming assistant/thinking content. */ + #buffers = new Map(); + /** + * All sends are chained so they reach Telegram in the order they were + * produced — long messages chunk sequentially and "✅ Done." can never + * overtake content. + */ + #chain: Promise = Promise.resolve(); + + constructor(api: RelayApi) { + this.#api = api; + } + + /** Number of live streaming buffers (exposed for tests/invariants). */ + get bufferCount(): number { + return this.#buffers.size; + } + + /** Run `task` after every previously enqueued send has settled. */ + enqueue(task: () => Promise): Promise { + const result = this.#chain.then(task); + this.#chain = result.then( + () => {}, + () => {}, + ); + return result; + } + + /** Queue a plain message send (order-preserving, errors swallowed). */ + send( + chatId: number, + text: string, + opts?: Record, + ): Promise { + return this.enqueue(() => this.#api.sendMessage(chatId, text, opts)).then( + () => {}, + () => {}, + ); + } + + /** Queue a long message, split into ordered, sequentially-awaited chunks. */ + sendChunked(chatId: number, text: string): Promise { + const chunks = chunkText(text); + if (chunks.length === 0) return Promise.resolve(); + return this.enqueue(async () => { + for (const chunk of chunks) { + // Await each chunk before sending the next — parallel fire-and-forget + // delivers chunks out of order. + await this.#api.sendMessage(chatId, chunk).catch(() => {}); + } + }).then( + () => {}, + () => {}, + ); + } + + /** + * Send a thinking block. Model text is not valid Telegram Markdown, so an + * unbalanced `*`/`_`/backtick can 400 — retry as plain text rather than + * dropping the thought. + */ + sendThinking(chatId: number, text: string): Promise { + const body = `💭 _thinking_\n${text.slice(0, 1500)}`; + return this.enqueue(async () => { + try { + await this.#api.sendMessage(chatId, body, { parse_mode: "Markdown" }); + } catch { + await this.#api.sendMessage(chatId, body).catch(() => {}); + } + }).then( + () => {}, + () => {}, + ); + } + + /** Handle a full `session.message` broadcast. */ + handleMessage(chatId: number, m: SessionMessage): void { + // An interleaving message means earlier streamed content should render + // before it — partial-flush other live buffers (they stay live; later + // deltas keep appending after the flushed offset). + this.#flushOthers(chatId, m.messageId); + + switch (m.role) { + case "assistant": + case "thinking": { + if (m.content) { + // Final authoritative broadcast for this messageId: send whatever + // hasn't been partial-flushed yet and DROP the buffer, so the idle + // flush can't deliver the same content a second time. + const buf = this.#buffers.get(m.messageId); + const pending = buf ? m.content.slice(buf.flushed) : m.content; + this.#buffers.delete(m.messageId); + if (pending) { + if (m.role === "thinking") this.sendThinking(chatId, pending); + else this.sendChunked(chatId, pending); + } + } else if (!this.#buffers.has(m.messageId)) { + // Empty content = start of a streaming block. + this.#buffers.set(m.messageId, { + role: m.role, + content: "", + flushed: 0, + }); + } + break; + } + case "tool_call": { + if (!m.tool) break; + const line = toolLine(m.tool.name, m.tool.state); + if (line) this.send(chatId, line); + break; + } + case "system": + if (m.content) this.send(chatId, `⚠️ ${m.content}`); + break; + // user (echo of our own send) and info: quiet on mobile. + } + } + + /** Handle a `session.message.delta` broadcast. */ + handleDelta(_chatId: number, d: SessionMessageDelta): void { + if (d.contentAppend) { + const buf = this.#buffers.get(d.messageId); + if (buf) buf.content += d.contentAppend; + } + } + + /** Turn ended: flush every remaining buffer, clear, then confirm. */ + flushIdle(chatId: number): void { + for (const buf of this.#buffers.values()) this.#flushPartial(chatId, buf); + this.#buffers.clear(); + // Queued after the flushes, so "Done" always follows the content. + this.send(chatId, "✅ Done."); + } + + /** Drop all buffered state without sending (e.g. session switch). */ + clear(): void { + this.#buffers.clear(); + } + + /** Send the unflushed tail of every buffer except `exceptId` (kept live). */ + #flushOthers(chatId: number, exceptId: string): void { + for (const [id, buf] of this.#buffers) { + if (id !== exceptId) this.#flushPartial(chatId, buf); + } + } + + /** Send a buffer's unflushed tail and advance its flushed offset. */ + #flushPartial(chatId: number, buf: StreamBuf): void { + const pending = buf.content.slice(buf.flushed); + if (!pending) return; + buf.flushed = buf.content.length; + if (buf.role === "thinking") this.sendThinking(chatId, pending); + else this.sendChunked(chatId, pending); + } + + /** Resolves once everything queued so far has been sent (test helper). */ + settle(): Promise { + return this.#chain.then( + () => {}, + () => {}, + ); + } +} + +/** Render a tool lifecycle state as a one-line chat notification. */ +export function toolLine(name: string, state: ToolState): string | null { + switch (state.phase) { + case "executing": + return `⚡ ${name}`; + case "completed": + return state.success === false ? `✗ ${name} failed` : `✓ ${name}`; + case "cancelled": + return `✗ ${name} cancelled`; + default: + // streaming / waiting_confirmation — rendered elsewhere (approval UI). + return null; + } +} diff --git a/src/tests/telegram-stream.test.ts b/src/tests/telegram-stream.test.ts new file mode 100644 index 0000000..aa42dbf --- /dev/null +++ b/src/tests/telegram-stream.test.ts @@ -0,0 +1,369 @@ +/** + * Telegram StreamRelay tests — exactly-once, in-order delivery of streamed + * content. + * + * Unlike the older telegram tests that mirror module-private logic, these + * exercise the REAL StreamRelay from src/frontends/telegram/stream.ts with a + * fake Telegram API injected through the RelayApi seam (same spirit: offline, + * no Bot, side effects captured via callbacks). + * + * What we verify: + * 1. A streamed turn (empty start → deltas → final full rebroadcast → idle) + * delivers the content EXACTLY ONCE, with "✅ Done." after it. + * 2. Thinking blocks are delivered exactly once. + * 3. A mid-stream tool_call broadcast neither drops nor duplicates streamed + * text, and content arrives in order around the tool line. + * 4. Thinking flushes that fail Markdown parsing retry as plain text. + * 5. sendChunked emits >4000-char content as ordered, sequential chunks + * split on line boundaries. + * 6. chunkText edge cases. + */ + +import { describe, it, expect } from "bun:test"; +import { + StreamRelay, + chunkText, + type RelayApi, +} from "../frontends/telegram/stream.js"; +import type { + SessionMessage, + SessionMessageDelta, +} from "../protocol/types.js"; + +const CHAT = 999_001; + +// ── Fake Telegram API ───────────────────────────────────────────────────────── + +interface SentMessage { + chatId: number; + text: string; + opts?: Record; +} + +function makeApi(options?: { + /** Reject the send when this returns true (simulates a Telegram 400). */ + failWhen?: (text: string, opts?: Record) => boolean; + /** Per-send artificial delay in ms (to catch out-of-order parallel sends). */ + delayMs?: (callIndex: number) => number; +}) { + const sent: SentMessage[] = []; + let inFlight = 0; + let maxInFlight = 0; + let calls = 0; + const api: RelayApi = { + async sendMessage(chatId, text, opts) { + const idx = calls++; + inFlight++; + maxInFlight = Math.max(maxInFlight, inFlight); + const delay = options?.delayMs?.(idx) ?? 0; + if (delay > 0) await new Promise((r) => setTimeout(r, delay)); + inFlight--; + if (options?.failWhen?.(text, opts)) { + throw new Error("400: Bad Request: can't parse entities"); + } + sent.push({ chatId, text, opts }); + return { message_id: sent.length }; + }, + }; + return { + api, + sent, + texts: () => sent.map((m) => m.text), + get maxInFlight() { + return maxInFlight; + }, + }; +} + +/** Minimal full session.message (fields the relay doesn't read are stubbed). */ +function full( + messageId: string, + role: SessionMessage["role"], + content: string, + tool?: SessionMessage["tool"], +): SessionMessage { + return { + type: "session.message", + sessionId: "sess-1", + messageId, + role, + content, + tool, + identity: { sub: "agent:test", type: "agent" }, + timestamp: new Date().toISOString(), + } as SessionMessage; +} + +function delta(messageId: string, contentAppend: string): SessionMessageDelta { + return { + type: "session.message.delta", + sessionId: "sess-1", + messageId, + contentAppend, + timestamp: new Date().toISOString(), + }; +} + +/** Count non-overlapping occurrences of `needle` across all sent texts. */ +function countOccurrences(haystacks: string[], needle: string): number { + return haystacks.reduce((n, t) => n + t.split(needle).length - 1, 0); +} + +// ── Exactly-once delivery ───────────────────────────────────────────────────── + +describe("StreamRelay — exactly-once streamed delivery", () => { + it("streamed assistant turn (start, deltas, final rebroadcast, idle) delivers content exactly once", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + // Daemon flow: empty assistant message opens the stream… + relay.handleMessage(CHAT, full("m1", "assistant", "")); + // …deltas accumulate… + relay.handleDelta(CHAT, delta("m1", "Hello ")); + relay.handleDelta(CHAT, delta("m1", "world")); + // …the daemon re-broadcasts the finished message with full content… + relay.handleMessage(CHAT, full("m1", "assistant", "Hello world")); + // …then the session goes idle (this used to re-send the buffer). + relay.flushIdle(CHAT); + await relay.settle(); + + expect(countOccurrences(texts(), "Hello world")).toBe(1); + expect(texts()).toEqual(["Hello world", "✅ Done."]); + expect(relay.bufferCount).toBe(0); + }); + + it("thinking block is delivered exactly once", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleMessage(CHAT, full("t1", "thinking", "")); + relay.handleDelta(CHAT, delta("t1", "pondering deeply")); + relay.handleMessage(CHAT, full("t1", "thinking", "pondering deeply")); + relay.flushIdle(CHAT); + await relay.settle(); + + expect(countOccurrences(texts(), "pondering deeply")).toBe(1); + const thinkingSends = texts().filter((t) => t.startsWith("💭")); + expect(thinkingSends).toHaveLength(1); + // "Done" arrives last. + expect(texts().at(-1)).toBe("✅ Done."); + }); + + it("assistant content that never streamed (direct full message) is sent once", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleMessage(CHAT, full("m1", "assistant", "one-shot answer")); + relay.flushIdle(CHAT); + await relay.settle(); + + expect(texts()).toEqual(["one-shot answer", "✅ Done."]); + }); + + it("turn that ends without a final rebroadcast still flushes the buffer once on idle", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleMessage(CHAT, full("m1", "assistant", "")); + relay.handleDelta(CHAT, delta("m1", "partial answer")); + relay.flushIdle(CHAT); + await relay.settle(); + + expect(texts()).toEqual(["partial answer", "✅ Done."]); + }); + + it("duplicate empty start does not reset an accumulating buffer", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleMessage(CHAT, full("m1", "assistant", "")); + relay.handleDelta(CHAT, delta("m1", "abc")); + relay.handleMessage(CHAT, full("m1", "assistant", "")); // dup start + relay.handleDelta(CHAT, delta("m1", "def")); + relay.flushIdle(CHAT); + await relay.settle(); + + expect(texts()).toEqual(["abcdef", "✅ Done."]); + }); +}); + +// ── Interleaved tool calls ──────────────────────────────────────────────────── + +describe("StreamRelay — mid-stream tool_call interleaving", () => { + it("tool_call broadcast mid-stream flushes the prefix, keeps the buffer live, no dup, in order", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleMessage(CHAT, full("m1", "assistant", "")); + relay.handleDelta(CHAT, delta("m1", "Let me check. ")); + + // A tool_call broadcast with a DIFFERENT messageId arrives mid-stream. + relay.handleMessage( + CHAT, + full("tc1", "tool_call", "", { + toolId: "tu-1", + name: "Bash", + state: { phase: "executing" }, + }), + ); + + // Later deltas for m1 must NOT be dropped (buffer stayed live). + relay.handleDelta(CHAT, delta("m1", "Found it.")); + // Final rebroadcast carries the FULL content; only the tail may be sent. + relay.handleMessage(CHAT, full("m1", "assistant", "Let me check. Found it.")); + relay.flushIdle(CHAT); + await relay.settle(); + + expect(texts()).toEqual([ + "Let me check. ", + "⚡ Bash", + "Found it.", + "✅ Done.", + ]); + // Exactly once overall. + expect(countOccurrences(texts(), "Let me check. ")).toBe(1); + expect(countOccurrences(texts(), "Found it.")).toBe(1); + }); + + it("multiple interleavings never duplicate or reorder streamed text", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleMessage(CHAT, full("m1", "assistant", "")); + relay.handleDelta(CHAT, delta("m1", "A")); + relay.handleMessage( + CHAT, + full("tc1", "tool_call", "", { toolId: "t1", name: "Read", state: { phase: "executing" } }), + ); + relay.handleDelta(CHAT, delta("m1", "B")); + relay.handleMessage( + CHAT, + full("tc2", "tool_call", "", { toolId: "t2", name: "Grep", state: { phase: "executing" } }), + ); + relay.handleDelta(CHAT, delta("m1", "C")); + relay.handleMessage(CHAT, full("m1", "assistant", "ABC")); + relay.flushIdle(CHAT); + await relay.settle(); + + expect(texts()).toEqual(["A", "⚡ Read", "B", "⚡ Grep", "C", "✅ Done."]); + }); +}); + +// ── Thinking Markdown fallback ──────────────────────────────────────────────── + +describe("StreamRelay — thinking parse-failure fallback", () => { + it("retries as plain text when Markdown parse fails instead of dropping the thought", async () => { + // Fail any send that carries a parse_mode (simulates Telegram 400 on + // unbalanced markdown in raw model text). + const { api, sent } = makeApi({ + failWhen: (_text, opts) => opts?.parse_mode !== undefined, + }); + const relay = new StreamRelay(api); + + relay.handleMessage(CHAT, full("t1", "thinking", "unbalanced *bold _and `tick")); + await relay.settle(); + + expect(sent).toHaveLength(1); + expect(sent[0]!.opts?.parse_mode).toBeUndefined(); + expect(sent[0]!.text).toContain("unbalanced *bold _and `tick"); + }); + + it("keeps parse_mode Markdown when it succeeds", async () => { + const { api, sent } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleMessage(CHAT, full("t1", "thinking", "clean thought")); + await relay.settle(); + + expect(sent).toHaveLength(1); + expect(sent[0]!.opts?.parse_mode).toBe("Markdown"); + }); +}); + +// ── Chunking ────────────────────────────────────────────────────────────────── + +describe("chunkText", () => { + it("returns [] for empty text and [text] under the limit", () => { + expect(chunkText("")).toEqual([]); + expect(chunkText("short")).toEqual(["short"]); + expect(chunkText("x".repeat(4000))).toEqual(["x".repeat(4000)]); + }); + + it("splits on line boundaries when a newline exists in the window", () => { + const line = "y".repeat(3000); + const text = `${line}\n${line}`; + const chunks = chunkText(text); + expect(chunks).toEqual([line, line]); + }); + + it("hard-cuts a single line longer than the limit", () => { + const text = "z".repeat(9000); + const chunks = chunkText(text); + expect(chunks.map((c) => c.length)).toEqual([4000, 4000, 1000]); + expect(chunks.join("")).toBe(text); + }); + + it("every chunk fits the limit and content is preserved modulo boundary newlines", () => { + const text = Array.from({ length: 200 }, (_, i) => `line ${i} ${"a".repeat(40)}`).join("\n"); + const chunks = chunkText(text); + for (const c of chunks) expect(c.length).toBeLessThanOrEqual(4000); + expect(chunks.join("\n")).toBe(text); + }); +}); + +describe("StreamRelay — sequential chunked sends", () => { + it("emits chunks in order even when individual sends are slow", async () => { + // First send is slow; if chunks were fired in parallel, chunk 2 would + // land before chunk 1. + const fake = makeApi({ + delayMs: (i) => (i === 0 ? 30 : 0), + }); + const relay = new StreamRelay(fake.api); + + const first = "a".repeat(3999); + const second = "b".repeat(3999); + const third = "c".repeat(100); + relay.sendChunked(CHAT, `${first}\n${second}\n${third}`); + await relay.settle(); + + expect(fake.texts()).toEqual([first, second, third]); + // Sequential: never more than one send in flight. + expect(fake.maxInFlight).toBe(1); + }); + + it("'✅ Done.' never overtakes a long flush", async () => { + const { api, texts } = makeApi({ delayMs: (i) => (i === 0 ? 20 : 0) }); + const relay = new StreamRelay(api); + + relay.handleMessage(CHAT, full("m1", "assistant", "")); + relay.handleDelta(CHAT, delta("m1", "x".repeat(8100))); + relay.flushIdle(CHAT); + await relay.settle(); + + const all = texts(); + expect(all.at(-1)).toBe("✅ Done."); + expect(all.slice(0, -1).join("")).toBe("x".repeat(8100)); + }); + + it("a failed chunk does not abort or reorder the remaining chunks", async () => { + let failed = false; + const { api, texts } = makeApi({ + failWhen: (text) => { + if (!failed && text.startsWith("b")) { + failed = true; + return true; + } + return false; + }, + }); + const relay = new StreamRelay(api); + relay.sendChunked(CHAT, `${"a".repeat(10)}\n${"b".repeat(10)}\n${"c".repeat(10)}`); + // Under 4000 chars total → single chunk; force multi-chunk instead: + relay.sendChunked(CHAT, `${"b".repeat(4000)}${"c".repeat(4000)}`); + await relay.settle(); + + // First call: single chunk "aaa…\nbbb…\nccc…" (no failure — starts with a). + // Second call: chunk1 ("b"*4000) fails once and is skipped, chunk2 still lands. + expect(texts().at(-1)).toBe("c".repeat(4000)); + }); +}); From c3a4de056ee783ebf14c16bbd82b3b16de7350dd Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Fri, 3 Jul 2026 00:13:41 +0800 Subject: [PATCH 2/5] fix: keep Telegram polling alive on handler errors, escape /ls output, honor 429s MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Any thrown handler error permanently stopped long polling — grammy's default error handler stops the bot and re-throws. A deterministic trigger existed in /ls, which rendered session status and workdir unescaped in MarkdownV2: a tool_running status underscore is a Telegram 400, which killed the bot for good. - install bot.catch to log and keep processing updates - escape /ls status with escMd and workdir with escCode (inside a MarkdownV2 code span only backtick and backslash are special; escMd there would render literal backslashes) - install @grammyjs/auto-retry as an API transformer so 429s wait for retry_after and retry instead of being swallowed by .catch(() => {}) - consolidate the duplicate esc/escMd helpers into stream.ts Tests drive a real grammy Bot with a stubbed API transformer through handleUpdates (the polling entry point): a 400 reply no longer stops update processing, and a flood-limited send is retried once. Co-Authored-By: Claude Fable 5 --- bun.lock | 3 + package.json | 1 + src/frontends/telegram/index.ts | 51 ++++---- src/frontends/telegram/stream.ts | 33 +++++ src/tests/telegram-bot.test.ts | 203 +++++++++++++++++++++++++++++++ 5 files changed, 266 insertions(+), 25 deletions(-) create mode 100644 src/tests/telegram-bot.test.ts diff --git a/bun.lock b/bun.lock index 9f6dc6a..8b96776 100644 --- a/bun.lock +++ b/bun.lock @@ -7,6 +7,7 @@ "dependencies": { "@anthropic-ai/claude-agent-sdk": "^0.3.175", "@google/generative-ai": "^0.24.1", + "@grammyjs/auto-retry": "^2.0.2", "@highflame/sdk": "^0.3.17", "@xenova/transformers": "^2.17.2", "commander": "^13.0.0", @@ -71,6 +72,8 @@ "@google/generative-ai": ["@google/generative-ai@0.24.1", "", {}, "sha512-MqO+MLfM6kjxcKoy0p1wRzG3b4ZZXtPI+z2IE26UogS2Cm/XHO+7gGRBh6gcJsOiIVoH93UwKvW4HdgiOZCy9Q=="], + "@grammyjs/auto-retry": ["@grammyjs/auto-retry@2.0.2", "", { "dependencies": { "debug": "^4.3.4" }, "peerDependencies": { "grammy": "^1.10.0" } }, "sha512-b4A4p5jlYDiQtW0c0FXYe11WMkYoiW+5rvOaDMCOk1h7Pu2SDl7B7gmFF8cthWCx2+M2nWLOsBxAgbBG4kKWYg=="], + "@grammyjs/types": ["@grammyjs/types@3.25.0", "", {}, "sha512-iN9i5p+8ZOu9OMxWNcguojQfz4K/PDyMPOnL7PPCON+SoA/F8OKMH3uR7CVUkYfdNe0GCz8QOzAWrnqusQYFOg=="], "@highflame/sdk": ["@highflame/sdk@0.3.17", "", { "peerDependencies": { "@aws/bedrock-ai-agents-strands": ">=1.0.0", "@azure/ai-projects": ">=2.0.0", "@langchain/core": ">=0.3.0", "@langchain/langgraph": ">=1.0.0", "@opentelemetry/api": "^1.4.0", "@strands-agents/sdk": ">=1.0.0" }, "optionalPeers": ["@aws/bedrock-ai-agents-strands", "@azure/ai-projects", "@langchain/core", "@langchain/langgraph", "@opentelemetry/api", "@strands-agents/sdk"] }, "sha512-yeoTCHMWm4qR4JlScSlHnsQYOjIDlQNR5ZosDq70Qj4hz0DuGwVNbjRA/k0IUgHnYnf/EWH11Qlsz9vCn91LPg=="], diff --git a/package.json b/package.json index 70db750..1b73888 100644 --- a/package.json +++ b/package.json @@ -56,6 +56,7 @@ "dependencies": { "@anthropic-ai/claude-agent-sdk": "^0.3.175", "@google/generative-ai": "^0.24.1", + "@grammyjs/auto-retry": "^2.0.2", "@highflame/sdk": "^0.3.17", "@xenova/transformers": "^2.17.2", "commander": "^13.0.0", diff --git a/src/frontends/telegram/index.ts b/src/frontends/telegram/index.ts index 3237c21..467ad9b 100644 --- a/src/frontends/telegram/index.ts +++ b/src/frontends/telegram/index.ts @@ -17,6 +17,7 @@ */ import { Bot, type Context, InlineKeyboard } from "grammy"; +import { autoRetry } from "@grammyjs/auto-retry"; import { randomUUID } from "node:crypto"; import { verifyToken } from "../../daemon/auth.js"; import { ALL_SCOPES_STRING } from "../../protocol/scopes.js"; @@ -34,7 +35,12 @@ import type { ToolInfo, } from "../../protocol/types.js"; import type { AttachedClient } from "../../daemon/session.js"; -import { StreamRelay, type RelayApi } from "./stream.js"; +import { + StreamRelay, + escMd, + formatSessionLine, + type RelayApi, +} from "./stream.js"; /** One AskUserQuestion question as it arrives in the tool input. */ interface AskQuestion { @@ -96,8 +102,12 @@ export class TelegramFrontend implements Frontend { this.#bot.api.sendMessage(chatId, text, opts as never), }; - constructor(botToken: string, allowedUserIds: number[]) { - this.#bot = new Bot(botToken); + constructor(botToken: string, allowedUserIds: number[], bot?: Bot) { + // `bot` is injectable for tests (a Bot with a stubbed API transformer). + this.#bot = bot ?? new Bot(botToken); + // Honor Telegram 429s: wait for retry_after and retry instead of letting + // the flood error be swallowed by the `.catch(() => {})` on each send. + this.#bot.api.config.use(autoRetry()); this.#allowedUserIds = new Set(allowedUserIds); } @@ -143,6 +153,13 @@ export class TelegramFrontend implements Frontend { // ── Handlers ────────────────────────────────────────────────────────── #setupHandlers(): void { + // Never let a thrown handler error kill long polling — grammy's default + // error handler re-throws, which permanently stops the bot. Log and keep + // processing updates. + this.#bot.catch((err) => { + console.error("[codeoid:telegram] handler error:", err.error ?? err); + }); + // Gate: only allowed Telegram user IDs this.#bot.use(async (ctx, next) => { const userId = ctx.from?.id; @@ -270,15 +287,9 @@ export class TelegramFrontend implements Frontend { await ctx.reply("No active sessions."); return; } - const lines = resp.sessions.map((s) => { - const icon = - s.status === "idle" - ? "🟢" - : s.status === "thinking" || s.status === "tool_running" - ? "🟡" - : "🔴"; - return `${icon} *${esc(s.name)}* — ${s.status}\n \`${s.workdir}\``; - }); + // Status and workdir are escaped — a `tool_running` underscore or a + // path with markdown specials must not 400 the reply. + const lines = resp.sessions.map((s) => formatSessionLine(s)); await ctx.reply(lines.join("\n\n"), { parse_mode: "MarkdownV2" }); } } @@ -302,7 +313,7 @@ export class TelegramFrontend implements Frontend { ); if (resp.type === "response.ok") { - await ctx.reply(`Session *${esc(name)}* created\\.`, { parse_mode: "MarkdownV2" }); + await ctx.reply(`Session *${escMd(name)}* created\\.`, { parse_mode: "MarkdownV2" }); } else if (resp.type === "response.error") { await ctx.reply(`Error: ${resp.error}`); } @@ -360,7 +371,7 @@ export class TelegramFrontend implements Frontend { if (resp.type === "response.ok") { state.attachedSessionId = session.id; state.attachedSessionName = name; - await ctx.reply(`Attached to *${esc(name)}*\\. Send messages here\\.`, { parse_mode: "MarkdownV2" }); + await ctx.reply(`Attached to *${escMd(name)}*\\. Send messages here\\.`, { parse_mode: "MarkdownV2" }); } else if (resp.type === "response.error") { await ctx.reply(`Error: ${resp.error}`); } @@ -427,7 +438,7 @@ export class TelegramFrontend implements Frontend { state.attachedSessionId = null; state.attachedSessionName = null; } - await ctx.reply(`Session *${esc(name)}* destroyed\\.`, { parse_mode: "MarkdownV2" }); + await ctx.reply(`Session *${escMd(name)}* destroyed\\.`, { parse_mode: "MarkdownV2" }); } // ── Search ────────────────────────────────────────────────────────────── @@ -1000,16 +1011,6 @@ export class TelegramFrontend implements Frontend { } } -/** Escape MarkdownV2 special characters. */ -function escMd(text: string): string { - return text.replace(/[_*[\]()~`>#+\-=|{}.!\\]/g, (m) => `\\${m}`); -} - -/** @deprecated — legacy esc used elsewhere in this file. */ -function esc(text: string): string { - return text.replace(/[_*[\]()~`>#+\-=|{}.!\\]/g, "\\$&"); -} - /** Relative time string from a unix-ms timestamp. */ function formatAgo(when: number): string { const dt = Math.max(0, Date.now() - when); diff --git a/src/frontends/telegram/stream.ts b/src/frontends/telegram/stream.ts index 9a5c4ae..875eb97 100644 --- a/src/frontends/telegram/stream.ts +++ b/src/frontends/telegram/stream.ts @@ -232,6 +232,39 @@ export class StreamRelay { } } +/** Escape MarkdownV2 special characters (for regular text context). */ +export function escMd(text: string): string { + return text.replace(/[_*[\]()~`>#+\-=|{}.!\\]/g, (m) => `\\${m}`); +} + +/** + * Escape for a MarkdownV2 inline-code span. Inside code entities only + * backtick and backslash are special — escaping anything else (as escMd + * does) would render literal backslashes. + */ +export function escCode(text: string): string { + return text.replace(/[`\\]/g, "\\$&"); +} + +/** + * One /ls line. Status and workdir are runtime values and must be escaped — + * a `tool_running` status underscore or a backtick in a path is otherwise a + * MarkdownV2 parse error (Telegram 400). + */ +export function formatSessionLine(s: { + name: string; + status: string; + workdir: string; +}): string { + const icon = + s.status === "idle" + ? "🟢" + : s.status === "thinking" || s.status === "tool_running" + ? "🟡" + : "🔴"; + return `${icon} *${escMd(s.name)}* — ${escMd(s.status)}\n \`${escCode(s.workdir)}\``; +} + /** Render a tool lifecycle state as a one-line chat notification. */ export function toolLine(name: string, state: ToolState): string | null { switch (state.phase) { diff --git a/src/tests/telegram-bot.test.ts b/src/tests/telegram-bot.test.ts new file mode 100644 index 0000000..3bb6448 --- /dev/null +++ b/src/tests/telegram-bot.test.ts @@ -0,0 +1,203 @@ +/** + * Telegram frontend — bot-level integration tests (offline). + * + * Uses the TelegramFrontend's injectable Bot seam: a real grammy Bot with a + * pre-set botInfo (skips getMe) and an API transformer that intercepts every + * outgoing Telegram call, so no network is touched. getUpdates is parked on a + * never-resolving promise so long polling stays inert. + * + * What we verify: + * 1. bot.catch is installed — a handler error (Telegram 400 on a reply) + * does NOT reject handleUpdate / stop update processing (grammy's + * default error handler re-throws, which kills long polling). + * 2. The auto-retry transformer honors 429 retry_after and retries the + * call instead of failing it. + * 3. formatSessionLine escapes MarkdownV2 in /ls lines (unit-level, real + * helper): tool_running status + workdir with `_` and backticks. + */ + +import { describe, it, expect } from "bun:test"; +import { Bot } from "grammy"; +import type { UserFromGetMe } from "grammy/types"; +import { TelegramFrontend } from "../frontends/telegram/index.js"; +import type { FrontendContext } from "../frontends/types.js"; +import { formatSessionLine, escMd, escCode } from "../frontends/telegram/stream.js"; + +const ALLOWED_USER = 111; +const CHAT_ID = 555; + +const botInfo: UserFromGetMe = { + id: 42, + is_bot: true, + first_name: "codeoid-test", + username: "codeoid_test_bot", + can_join_groups: true, + can_read_all_group_messages: false, + supports_inline_queries: false, + can_connect_to_business: false, + has_main_web_app: false, + has_topics_enabled: false, + allows_users_to_create_topics: false, +}; + +interface ApiCall { + method: string; + payload: any; +} + +/** + * Build a Bot whose API layer is fully stubbed. `respond` decides the raw + * Telegram API response per call; getUpdates never resolves (polling parks). + * Installed BEFORE TelegramFrontend's own transformers (auto-retry), so + * auto-retry wraps this stub exactly like it wraps the real HTTP layer. + */ +function makeStubbedBot( + respond: (method: string, payload: any, calls: ApiCall[]) => any, +): { bot: Bot; calls: ApiCall[] } { + const bot = new Bot("42:TEST_TOKEN", { botInfo }); + const calls: ApiCall[] = []; + bot.api.config.use(async (_prev, method, payload, signal) => { + if (method === "getUpdates") { + // Park polling; reject on abort so bot.stop() (e.g. grammy's default + // error handler stopping the bot) fails fast instead of hanging. + return new Promise((_resolve, reject) => { + signal?.addEventListener("abort", () => reject(new Error("aborted"))); + }); + } + calls.push({ method, payload }); + return respond(method, payload, calls); + }); + return { bot, calls }; +} + +function fakeContext(): FrontendContext { + return { + manager: {} as never, + store: { audit() {} } as never, + auth: { baseUrl: "http://localhost:0" }, + httpServer: {} as never, + host: "localhost", + port: 0, + }; +} + +function textUpdate(updateId: number, text: string) { + return { + update_id: updateId, + message: { + message_id: updateId + 1000, + date: Math.floor(Date.now() / 1000), + chat: { id: CHAT_ID, type: "private" as const, first_name: "u" }, + from: { id: ALLOWED_USER, is_bot: false, first_name: "u" }, + text, + entities: text.startsWith("/") + ? [{ type: "bot_command" as const, offset: 0, length: text.split(" ")[0]!.length }] + : undefined, + }, + }; +} + +const OK = (method: string) => + ({ + ok: true, + result: method === "sendMessage" ? { message_id: 1 } : true, + }) as any; + +describe("Telegram bot — error handling keeps polling alive", () => { + it("a handler error (400 reply) is caught by bot.catch and does not stop update processing", async () => { + let fail = false; + const { bot, calls } = makeStubbedBot((method) => { + if (method === "sendMessage" && fail) { + fail = false; + return { + ok: false, + error_code: 400, + description: "Bad Request: can't parse entities", + } as any; + } + return OK(method); + }); + + const fe = new TelegramFrontend("42:TEST_TOKEN", [ALLOWED_USER], bot); + await fe.start(fakeContext()); + + // /help replies with MarkdownV2; make Telegram reject it. Drive the + // update through handleUpdates — the exact path grammy's polling loop + // uses, where the installed error handler runs. With grammy's DEFAULT + // handler this re-throws and long polling stops permanently. + fail = true; + // handleUpdates is TS-private but is the real polling entry point. + const drive = (u: unknown) => (bot as any).handleUpdates([u]) as Promise; + await expect(drive(textUpdate(1, "/help"))).resolves.toBeUndefined(); + + // The bot still processes the next update. + const before = calls.filter((c) => c.method === "sendMessage").length; + await drive(textUpdate(2, "/help")); + const after = calls.filter((c) => c.method === "sendMessage").length; + expect(after).toBe(before + 1); + }); +}); + +describe("Telegram bot — 429 auto-retry", () => { + it("retries a flood-limited sendMessage after retry_after instead of failing it", async () => { + let sendAttempts = 0; + const { bot, calls } = makeStubbedBot((method) => { + if (method === "sendMessage") { + sendAttempts++; + if (sendAttempts === 1) { + return { + ok: false, + error_code: 429, + description: "Too Many Requests: retry after 0", + parameters: { retry_after: 0 }, + } as any; + } + } + return OK(method); + }); + + const fe = new TelegramFrontend("42:TEST_TOKEN", [ALLOWED_USER], bot); + await fe.start(fakeContext()); + + await bot.handleUpdate(textUpdate(1, "/help")); + + // First attempt hit 429, auto-retry re-sent it, second attempt succeeded. + expect(sendAttempts).toBe(2); + expect(calls.filter((c) => c.method === "sendMessage")).toHaveLength(2); + }); +}); + +describe("/ls line rendering — MarkdownV2 escaping of runtime values", () => { + it("escapes a tool_running status (underscore would otherwise 400 and kill polling)", () => { + const line = formatSessionLine({ + name: "my-session", + status: "tool_running", + workdir: "/home/user/work", + }); + expect(line).toContain("tool\\_running"); + expect(line).toContain("*my\\-session*"); + }); + + it("escapes workdir containing _ and ` inside the code span", () => { + const line = formatSessionLine({ + name: "s", + status: "idle", + workdir: "/tmp/my_dir/weird`path", + }); + // Inside a MarkdownV2 code span only ` and \ are special: the backtick + // must be escaped (or it terminates the span); the underscore must NOT + // be escaped (the backslash would render literally). + expect(line).toContain("`/tmp/my_dir/weird\\`path`"); + }); + + it("status icons stay intact for idle/thinking/other", () => { + expect(formatSessionLine({ name: "a", status: "idle", workdir: "/w" })).toStartWith("🟢"); + expect(formatSessionLine({ name: "a", status: "thinking", workdir: "/w" })).toStartWith("🟡"); + expect(formatSessionLine({ name: "a", status: "error", workdir: "/w" })).toStartWith("🔴"); + }); + + it("escMd escapes every MarkdownV2 special; escCode escapes only ` and \\", () => { + expect(escMd("a_b*c[d]e")).toBe("a\\_b\\*c\\[d\\]e"); + expect(escCode("a_b`c\\d")).toBe("a_b\\`c\\\\d"); + }); +}); From 20420301db2126bc37b0cab55d6df824c65b49ce Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Fri, 3 Jul 2026 00:17:49 +0800 Subject: [PATCH 3/5] fix: render Telegram tool completion deltas and flush buffers on detach/switch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tool completed/cancelled states never rendered in Telegram: the daemon broadcasts them as session.message.delta with toolStateUpdate, but the delta handler only read contentAppend — tools showed ⚡ forever. And session switch/detach cleared the streaming buffers outright, silently discarding any streamed-but-undelivered content. - remember messageId → tool name from the tool_call broadcast and render ✓ name / ✗ name failed / ✗ name cancelled when a toolStateUpdate delta arrives, flushing streamed text first so lines land in order - replace clear() with flushAndClear() on session switch and /detach: buffered content is delivered with a "✂️ detached mid-stream" marker instead of vanishing - update the session-switch mirror tests to the flush-before-clear contract Co-Authored-By: Claude Fable 5 --- src/frontends/telegram/index.ts | 8 +- src/frontends/telegram/stream.ts | 41 +++++- src/tests/telegram-session-switch.test.ts | 41 ++++-- src/tests/telegram-stream.test.ts | 152 ++++++++++++++++++++++ 4 files changed, 228 insertions(+), 14 deletions(-) diff --git a/src/frontends/telegram/index.ts b/src/frontends/telegram/index.ts index 467ad9b..69cce5a 100644 --- a/src/frontends/telegram/index.ts +++ b/src/frontends/telegram/index.ts @@ -346,7 +346,9 @@ export class TelegramFrontend implements Frontend { this.#manager.disconnectClient(state.clientId); state.attachedSessionId = null; state.attachedSessionName = null; - state.relay.clear(); + // Deliver anything still buffered from the old session (with an + // interruption marker) rather than discarding it invisibly. + state.relay.flushAndClear(chatId); // Remove the ⏹ Stop button for the old session. The old session's // idle status_change (which normally deletes it) won't arrive after // we disconnected, so it would otherwise linger in the chat. @@ -389,7 +391,9 @@ export class TelegramFrontend implements Frontend { const name = state.attachedSessionName; state.attachedSessionId = null; state.attachedSessionName = null; - state.relay.clear(); + // Deliver anything still buffered before dropping state — detaching must + // not silently swallow streamed-but-unflushed content. + state.relay.flushAndClear(chatId); if (state.stopMessageId !== null) { this.#bot.api.deleteMessage(chatId, state.stopMessageId).catch(() => {}); state.stopMessageId = null; diff --git a/src/frontends/telegram/stream.ts b/src/frontends/telegram/stream.ts index 875eb97..22f36b4 100644 --- a/src/frontends/telegram/stream.ts +++ b/src/frontends/telegram/stream.ts @@ -69,6 +69,12 @@ export class StreamRelay { #api: RelayApi; /** Per-messageId accumulator for streaming assistant/thinking content. */ #buffers = new Map(); + /** + * messageId → tool name. Tool completion/cancellation arrives as a + * `session.message.delta` carrying only `toolStateUpdate` — the name lives + * on the original tool_call broadcast, so remember it here. + */ + #toolNames = new Map(); /** * All sends are chained so they reach Telegram in the order they were * produced — long messages chunk sequentially and "✅ Done." can never @@ -175,6 +181,9 @@ export class StreamRelay { } case "tool_call": { if (!m.tool) break; + // Remember the name — completion/cancellation arrives as a bare + // toolStateUpdate delta referencing this messageId. + this.#toolNames.set(m.messageId, m.tool.name); const line = toolLine(m.tool.name, m.tool.state); if (line) this.send(chatId, line); break; @@ -187,24 +196,50 @@ export class StreamRelay { } /** Handle a `session.message.delta` broadcast. */ - handleDelta(_chatId: number, d: SessionMessageDelta): void { + handleDelta(chatId: number, d: SessionMessageDelta): void { if (d.contentAppend) { const buf = this.#buffers.get(d.messageId); if (buf) buf.content += d.contentAppend; } + if (d.toolStateUpdate) { + // The daemon broadcasts tool completed/cancelled as a delta — render + // ✓/✗ here or the user never sees tools finish. Flush streamed text + // first so the tool line lands in order. + this.#flushOthers(chatId, d.messageId); + const name = this.#toolNames.get(d.messageId) ?? "tool"; + const line = toolLine(name, d.toolStateUpdate); + if (line) this.send(chatId, line); + const phase = d.toolStateUpdate.phase; + if (phase === "completed" || phase === "cancelled") { + this.#toolNames.delete(d.messageId); + } + } } /** Turn ended: flush every remaining buffer, clear, then confirm. */ flushIdle(chatId: number): void { for (const buf of this.#buffers.values()) this.#flushPartial(chatId, buf); this.#buffers.clear(); + this.#toolNames.clear(); // Queued after the flushes, so "Done" always follows the content. this.send(chatId, "✅ Done."); } - /** Drop all buffered state without sending (e.g. session switch). */ - clear(): void { + /** + * Session detach/switch: deliver whatever is buffered (never discard + * content invisibly), mark the stream as cut short, and reset state. + */ + flushAndClear(chatId: number): void { + let interrupted = false; + for (const buf of this.#buffers.values()) { + if (buf.content.length > buf.flushed) interrupted = true; + this.#flushPartial(chatId, buf); + } this.#buffers.clear(); + this.#toolNames.clear(); + if (interrupted) { + this.send(chatId, "✂️ Detached mid-stream — output above may be incomplete."); + } } /** Send the unflushed tail of every buffer except `exceptId` (kept live). */ diff --git a/src/tests/telegram-session-switch.test.ts b/src/tests/telegram-session-switch.test.ts index 1b712bb..3bccf7c 100644 --- a/src/tests/telegram-session-switch.test.ts +++ b/src/tests/telegram-session-switch.test.ts @@ -11,8 +11,10 @@ * the streaming buffer, and removes the stop-button reference. * 2. Re-attaching to the same session is a no-op — no disconnect, no clear. * 3. Attaching for the first time (no prior session) never calls disconnect. - * 4. Switching with an active streaming buffer discards partial content so it - * cannot bleed into the new session's output. + * 4. Switching with an active streaming buffer FLUSHES buffered content to + * the chat first (StreamRelay.flushAndClear) — undelivered content is + * never silently discarded — and the cleared buffer cannot bleed into + * the new session's output. * 5. Switching with an active stop-message id marks the message for deletion. * 6. #handleDetach clears streaming + stop-button state regardless of * whether there was an active turn. @@ -58,6 +60,11 @@ function makeUserState(clientId = "telegram:123"): UserState { interface SwitchDeps { disconnectClient: (clientId: string) => void; deleteMessage: (chatId: number, messageId: number) => void; + /** + * Mirror of StreamRelay.flushAndClear: deliver buffered undelivered + * content to the chat, then reset the buffer map. + */ + flushAndClear: (state: UserState) => void; } /** @@ -76,7 +83,7 @@ function performSwitch( deps.disconnectClient(state.clientId); state.attachedSessionId = null; state.attachedSessionName = null; - state.streaming.clear(); + deps.flushAndClear(state); if (state.stopMessageId !== null) { deps.deleteMessage(chatId, state.stopMessageId); state.stopMessageId = null; @@ -103,7 +110,7 @@ function performDetach( deps.disconnectClient(state.clientId); state.attachedSessionId = null; state.attachedSessionName = null; - state.streaming.clear(); + deps.flushAndClear(state); if (state.stopMessageId !== null) { deps.deleteMessage(chatId, state.stopMessageId); state.stopMessageId = null; @@ -111,14 +118,26 @@ function performDetach( } /** Stub deps that record calls for assertion. */ -function makeDeps(): SwitchDeps & { disconnected: string[]; deleted: [number, number][] } { +function makeDeps(): SwitchDeps & { + disconnected: string[]; + deleted: [number, number][]; + flushed: string[]; +} { const disconnected: string[] = []; const deleted: [number, number][] = []; + const flushed: string[] = []; return { disconnectClient: (id) => disconnected.push(id), deleteMessage: (chatId, msgId) => deleted.push([chatId, msgId]), + flushAndClear: (state) => { + for (const buf of state.streaming.values()) { + if (buf.content) flushed.push(buf.content); + } + state.streaming.clear(); + }, disconnected, deleted, + flushed, }; } @@ -158,7 +177,7 @@ describe("session switch — #handleAttach state transitions", () => { expect(deps.disconnected).toHaveLength(0); }); - it("streaming buffer is cleared on session switch", () => { + it("streaming buffer is flushed to the chat, then cleared, on session switch", () => { const state = makeUserState(); const deps = makeDeps(); @@ -170,8 +189,10 @@ describe("session switch — #handleAttach state transitions", () => { performSwitch(state, "sess-B", "session-B", CHAT_ID, deps); - // Buffer must be empty — no content from session A bleeds into session B. + // Buffer must be empty — no content from session A bleeds into session B — + // but the buffered content was delivered, not silently discarded. expect(state.streaming.size).toBe(0); + expect(deps.flushed).toEqual(["partial...", "reasoning..."]); }); it("streaming buffer is NOT cleared when re-attaching to the same session", () => { @@ -259,14 +280,16 @@ describe("detach — #handleDetach state transitions", () => { expect(deps.disconnected).toEqual(["telegram:77"]); }); - it("clears the streaming buffer", () => { + it("flushes then clears the streaming buffer", () => { const state = makeUserState(); state.attachedSessionId = "sess-A"; state.streaming.set("m1", { role: "assistant", content: "partial" }); + const deps = makeDeps(); - performDetach(state, CHAT_ID, makeDeps()); + performDetach(state, CHAT_ID, deps); expect(state.streaming.size).toBe(0); + expect(deps.flushed).toEqual(["partial"]); }); it("requests stop-message deletion when one is active", () => { diff --git a/src/tests/telegram-stream.test.ts b/src/tests/telegram-stream.test.ts index aa42dbf..463a055 100644 --- a/src/tests/telegram-stream.test.ts +++ b/src/tests/telegram-stream.test.ts @@ -28,6 +28,7 @@ import { import type { SessionMessage, SessionMessageDelta, + ToolState, } from "../protocol/types.js"; const CHAT = 999_001; @@ -367,3 +368,154 @@ describe("StreamRelay — sequential chunked sends", () => { expect(texts().at(-1)).toBe("c".repeat(4000)); }); }); + +// ── toolStateUpdate deltas (tool completed / cancelled) ─────────────────────── + +describe("StreamRelay — toolStateUpdate deltas render tool completion", () => { + function toolDelta( + messageId: string, + toolStateUpdate: ToolState, + ): SessionMessageDelta { + return { + type: "session.message.delta", + sessionId: "sess-1", + messageId, + toolStateUpdate, + timestamp: new Date().toISOString(), + }; + } + + it("renders ✓ with the remembered tool name when a completed delta arrives", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleMessage( + CHAT, + full("tc1", "tool_call", "", { toolId: "t1", name: "Bash", state: { phase: "executing" } }), + ); + relay.handleDelta(CHAT, toolDelta("tc1", { phase: "completed", success: true, output: "ok" })); + await relay.settle(); + + expect(texts()).toEqual(["⚡ Bash", "✓ Bash"]); + }); + + it("renders ✗ failed when the tool completed unsuccessfully", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleMessage( + CHAT, + full("tc1", "tool_call", "", { toolId: "t1", name: "Edit", state: { phase: "executing" } }), + ); + relay.handleDelta(CHAT, toolDelta("tc1", { phase: "completed", success: false })); + await relay.settle(); + + expect(texts()).toEqual(["⚡ Edit", "✗ Edit failed"]); + }); + + it("renders ✗ cancelled for a cancelled delta (denied approval)", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + // waiting_confirmation renders no line here (approval keyboard is the + // frontend's job) but must still register the tool name. + relay.handleMessage( + CHAT, + full("tc1", "tool_call", "", { + toolId: "t1", + name: "Write", + state: { phase: "waiting_confirmation", input: {}, description: "Write(file)", approvalId: "ap-1" }, + }), + ); + relay.handleDelta(CHAT, toolDelta("tc1", { phase: "cancelled", reason: "denied" })); + await relay.settle(); + + expect(texts()).toEqual(["✗ Write cancelled"]); + }); + + it("tool completion mid-stream flushes streamed text first (in order, no dup)", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleMessage(CHAT, full("m1", "assistant", "")); + relay.handleDelta(CHAT, delta("m1", "Running the build. ")); + relay.handleMessage( + CHAT, + full("tc1", "tool_call", "", { toolId: "t1", name: "Bash", state: { phase: "executing" } }), + ); + relay.handleDelta(CHAT, delta("m1", "It passed.")); + relay.handleDelta(CHAT, toolDelta("tc1", { phase: "completed", success: true })); + relay.handleMessage(CHAT, full("m1", "assistant", "Running the build. It passed.")); + relay.flushIdle(CHAT); + await relay.settle(); + + expect(texts()).toEqual([ + "Running the build. ", + "⚡ Bash", + "It passed.", + "✓ Bash", + "✅ Done.", + ]); + }); + + it("falls back to 'tool' when the name was never seen", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleDelta(CHAT, toolDelta("unknown-id", { phase: "completed", success: true })); + await relay.settle(); + + expect(texts()).toEqual(["✓ tool"]); + }); +}); + +// ── flush-on-detach / session switch ────────────────────────────────────────── + +describe("StreamRelay — flushAndClear (detach / session switch)", () => { + it("delivers buffered undelivered content with an interruption marker", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleMessage(CHAT, full("m1", "assistant", "")); + relay.handleDelta(CHAT, delta("m1", "half-finished answ")); + relay.flushAndClear(CHAT); + await relay.settle(); + + expect(texts()[0]).toBe("half-finished answ"); + expect(texts()[1]).toContain("✂️"); + expect(relay.bufferCount).toBe(0); + }); + + it("no marker when nothing was pending", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleMessage(CHAT, full("m1", "assistant", "already delivered")); + await relay.settle(); + relay.flushAndClear(CHAT); + await relay.settle(); + + expect(texts()).toEqual(["already delivered"]); + }); + + it("post-clear deltas for old messageIds are dropped, fresh streams work", async () => { + const { api, texts } = makeApi(); + const relay = new StreamRelay(api); + + relay.handleMessage(CHAT, full("m1", "assistant", "")); + relay.handleDelta(CHAT, delta("m1", "old session text")); + relay.flushAndClear(CHAT); + + // Stale delta from the old session after the switch — ignored. + relay.handleDelta(CHAT, delta("m1", " ghost")); + // New session streams normally. + relay.handleMessage(CHAT, full("m2", "assistant", "")); + relay.handleDelta(CHAT, delta("m2", "new session text")); + relay.flushIdle(CHAT); + await relay.settle(); + + expect(countOccurrences(texts(), "ghost")).toBe(0); + expect(countOccurrences(texts(), "new session text")).toBe(1); + expect(countOccurrences(texts(), "old session text")).toBe(1); + }); +}); From 45f0ca5e4167dfa5bb29262f909a58f72e8b155d Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Fri, 3 Jul 2026 00:40:33 +0800 Subject: [PATCH 4/5] fix: settle flushed output before switch/detach confirmations, drop stale session broadcasts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address CodeRabbit review on PR #76: - flushAndClear only queues sends on the relay chain, so the "Attached to …" / "Detached from …" confirmations (sent outside the relay) could overtake the flushed buffer tail and interruption marker. Await relay.settle() after flushing in both handlers. - #forwardToChat forwarded any broadcast as long as user state existed; an in-flight callback from the old session arriving after a detach or switch reached the current relay/chat — a stale idle status would even flush the new session's buffers and print a bogus "✅ Done.". Gate session-scoped messages on msg.sessionId === attachedSessionId via the exported isStaleBroadcast helper (messages without a sessionId are never dropped). - test hygiene (review nitpick): stop the stubbed bot in a finally block so no parked getUpdates polling handle leaks; the stub answers grammy stop()'s signal-less offset-save getUpdates immediately. Co-Authored-By: Claude Fable 5 --- src/frontends/telegram/index.ts | 43 ++++++++- src/tests/telegram-bot.test.ts | 110 +++++++++++++++++----- src/tests/telegram-session-switch.test.ts | 57 ++++++++++- src/tests/telegram-stream.test.ts | 23 +++++ 4 files changed, 201 insertions(+), 32 deletions(-) diff --git a/src/frontends/telegram/index.ts b/src/frontends/telegram/index.ts index 69cce5a..267fadd 100644 --- a/src/frontends/telegram/index.ts +++ b/src/frontends/telegram/index.ts @@ -347,8 +347,10 @@ export class TelegramFrontend implements Frontend { state.attachedSessionId = null; state.attachedSessionName = null; // Deliver anything still buffered from the old session (with an - // interruption marker) rather than discarding it invisibly. + // interruption marker) rather than discarding it invisibly — and wait + // for it to land so the "Attached to …" confirmation can't overtake it. state.relay.flushAndClear(chatId); + await state.relay.settle(); // Remove the ⏹ Stop button for the old session. The old session's // idle status_change (which normally deletes it) won't arrive after // we disconnected, so it would otherwise linger in the chat. @@ -364,6 +366,15 @@ export class TelegramFrontend implements Frontend { send: (msg: DaemonMessage) => this.#forwardToChat(chatId, userId, msg), }; + // Mark attached BEFORE the attach call: live broadcasts can start the + // moment the daemon registers the client, and the stale-session gate in + // #forwardToChat would otherwise drop them. Restored on failure (a + // failed same-session re-attach must not fake a detach). + const prevSessionId = state.attachedSessionId; + const prevSessionName = state.attachedSessionName; + state.attachedSessionId = session.id; + state.attachedSessionName = name; + const resp = await this.#manager.handle( { type: "session.attach", id: randomUUID(), sessionId: session.id }, state.auth!, @@ -371,10 +382,10 @@ export class TelegramFrontend implements Frontend { ); if (resp.type === "response.ok") { - state.attachedSessionId = session.id; - state.attachedSessionName = name; await ctx.reply(`Attached to *${escMd(name)}*\\. Send messages here\\.`, { parse_mode: "MarkdownV2" }); } else if (resp.type === "response.error") { + state.attachedSessionId = prevSessionId; + state.attachedSessionName = prevSessionName; await ctx.reply(`Error: ${resp.error}`); } } @@ -392,8 +403,10 @@ export class TelegramFrontend implements Frontend { state.attachedSessionId = null; state.attachedSessionName = null; // Deliver anything still buffered before dropping state — detaching must - // not silently swallow streamed-but-unflushed content. + // not silently swallow streamed-but-unflushed content. Wait for it to + // land so the "Detached from …" confirmation can't overtake it. state.relay.flushAndClear(chatId); + await state.relay.settle(); if (state.stopMessageId !== null) { this.#bot.api.deleteMessage(chatId, state.stopMessageId).catch(() => {}); state.stopMessageId = null; @@ -736,7 +749,11 @@ export class TelegramFrontend implements Frontend { #forwardToChat(chatId: number, userId: number, msg: DaemonMessage): void { const state = this.#users.get(userId); - if (!state) return; + // Drop broadcasts from sessions the user is no longer attached to — an + // in-flight callback from the old session can still fire after a detach + // or switch, and must not reach the current relay/chat (a stale idle + // would flush the new session's buffers and print a bogus "✅ Done."). + if (!state || isStaleBroadcast(msg, state.attachedSessionId)) return; const relay = state.relay; switch (msg.type) { @@ -1015,6 +1032,22 @@ export class TelegramFrontend implements Frontend { } } +/** + * True when a session-scoped daemon broadcast belongs to a session the user + * is not (or no longer) attached to. Messages without a sessionId (e.g. + * direct responses) are never considered stale. Exported for tests. + */ +export function isStaleBroadcast( + msg: DaemonMessage, + attachedSessionId: string | null, +): boolean { + const sid = + "sessionId" in msg && typeof msg.sessionId === "string" + ? msg.sessionId + : null; + return sid !== null && sid !== attachedSessionId; +} + /** Relative time string from a unix-ms timestamp. */ function formatAgo(when: number): string { const dt = Math.max(0, Date.now() - when); diff --git a/src/tests/telegram-bot.test.ts b/src/tests/telegram-bot.test.ts index 3bb6448..f012509 100644 --- a/src/tests/telegram-bot.test.ts +++ b/src/tests/telegram-bot.test.ts @@ -19,8 +19,9 @@ import { describe, it, expect } from "bun:test"; import { Bot } from "grammy"; import type { UserFromGetMe } from "grammy/types"; -import { TelegramFrontend } from "../frontends/telegram/index.js"; +import { TelegramFrontend, isStaleBroadcast } from "../frontends/telegram/index.js"; import type { FrontendContext } from "../frontends/types.js"; +import type { DaemonMessage } from "../protocol/types.js"; import { formatSessionLine, escMd, escCode } from "../frontends/telegram/stream.js"; const ALLOWED_USER = 111; @@ -58,10 +59,17 @@ function makeStubbedBot( const calls: ApiCall[] = []; bot.api.config.use(async (_prev, method, payload, signal) => { if (method === "getUpdates") { - // Park polling; reject on abort so bot.stop() (e.g. grammy's default - // error handler stopping the bot) fails fast instead of hanging. + // The poll loop passes the pollingAbortController signal — park it and + // reject on abort. bot.stop() additionally issues a signal-less + // getUpdates({limit: 1}) to save the offset — answer that immediately + // or stop() would hang forever. + if (!signal) return { ok: true, result: [] } as any; return new Promise((_resolve, reject) => { - signal?.addEventListener("abort", () => reject(new Error("aborted"))); + signal.addEventListener("abort", () => { + const err = new Error("aborted"); + err.name = "AbortError"; + reject(err); + }); }); } calls.push({ method, payload }); @@ -119,22 +127,27 @@ describe("Telegram bot — error handling keeps polling alive", () => { }); const fe = new TelegramFrontend("42:TEST_TOKEN", [ALLOWED_USER], bot); - await fe.start(fakeContext()); - - // /help replies with MarkdownV2; make Telegram reject it. Drive the - // update through handleUpdates — the exact path grammy's polling loop - // uses, where the installed error handler runs. With grammy's DEFAULT - // handler this re-throws and long polling stops permanently. - fail = true; - // handleUpdates is TS-private but is the real polling entry point. - const drive = (u: unknown) => (bot as any).handleUpdates([u]) as Promise; - await expect(drive(textUpdate(1, "/help"))).resolves.toBeUndefined(); - - // The bot still processes the next update. - const before = calls.filter((c) => c.method === "sendMessage").length; - await drive(textUpdate(2, "/help")); - const after = calls.filter((c) => c.method === "sendMessage").length; - expect(after).toBe(before + 1); + try { + await fe.start(fakeContext()); + + // /help replies with MarkdownV2; make Telegram reject it. Drive the + // update through handleUpdates — the exact path grammy's polling loop + // uses, where the installed error handler runs. With grammy's DEFAULT + // handler this re-throws and long polling stops permanently. + fail = true; + // handleUpdates is TS-private but is the real polling entry point. + const drive = (u: unknown) => (bot as any).handleUpdates([u]) as Promise; + await expect(drive(textUpdate(1, "/help"))).resolves.toBeUndefined(); + + // The bot still processes the next update. + const before = calls.filter((c) => c.method === "sendMessage").length; + await drive(textUpdate(2, "/help")); + const after = calls.filter((c) => c.method === "sendMessage").length; + expect(after).toBe(before + 1); + } finally { + // Unpark the getUpdates promise so no polling handle leaks. + await bot.stop().catch(() => {}); + } }); }); @@ -157,13 +170,18 @@ describe("Telegram bot — 429 auto-retry", () => { }); const fe = new TelegramFrontend("42:TEST_TOKEN", [ALLOWED_USER], bot); - await fe.start(fakeContext()); + try { + await fe.start(fakeContext()); - await bot.handleUpdate(textUpdate(1, "/help")); + await bot.handleUpdate(textUpdate(1, "/help")); - // First attempt hit 429, auto-retry re-sent it, second attempt succeeded. - expect(sendAttempts).toBe(2); - expect(calls.filter((c) => c.method === "sendMessage")).toHaveLength(2); + // First attempt hit 429, auto-retry re-sent it, second attempt succeeded. + expect(sendAttempts).toBe(2); + expect(calls.filter((c) => c.method === "sendMessage")).toHaveLength(2); + } finally { + // Unpark the getUpdates promise so no polling handle leaks. + await bot.stop().catch(() => {}); + } }); }); @@ -201,3 +219,45 @@ describe("/ls line rendering — MarkdownV2 escaping of runtime values", () => { expect(escCode("a_b`c\\d")).toBe("a_b\\`c\\\\d"); }); }); + +// ── Stale-session broadcast gating (#forwardToChat) ─────────────────────────── + +describe("isStaleBroadcast — drop daemon messages from unattached sessions", () => { + const sessionMsg = (sessionId: string): DaemonMessage => + ({ + type: "session.message", + sessionId, + messageId: "m1", + role: "assistant", + content: "hello", + identity: { sub: "agent:x", type: "agent" }, + timestamp: new Date().toISOString(), + }) as DaemonMessage; + + it("keeps messages for the currently attached session", () => { + expect(isStaleBroadcast(sessionMsg("sess-A"), "sess-A")).toBe(false); + }); + + it("drops in-flight messages from the old session after a switch", () => { + expect(isStaleBroadcast(sessionMsg("sess-OLD"), "sess-NEW")).toBe(true); + }); + + it("drops session-scoped messages after detach (attachedSessionId null)", () => { + expect(isStaleBroadcast(sessionMsg("sess-A"), null)).toBe(true); + }); + + it("drops a stale status_change (would otherwise flush + print Done in the new session)", () => { + const statusMsg = { + type: "session.status_change", + sessionId: "sess-OLD", + status: "idle", + } as unknown as DaemonMessage; + expect(isStaleBroadcast(statusMsg, "sess-NEW")).toBe(true); + }); + + it("never treats messages without a sessionId as stale", () => { + const pong = { type: "response.ok", requestId: "r1" } as unknown as DaemonMessage; + expect(isStaleBroadcast(pong, "sess-A")).toBe(false); + expect(isStaleBroadcast(pong, null)).toBe(false); + }); +}); diff --git a/src/tests/telegram-session-switch.test.ts b/src/tests/telegram-session-switch.test.ts index 3bccf7c..a83ad4d 100644 --- a/src/tests/telegram-session-switch.test.ts +++ b/src/tests/telegram-session-switch.test.ts @@ -65,6 +65,12 @@ interface SwitchDeps { * content to the chat, then reset the buffer map. */ flushAndClear: (state: UserState) => void; + /** + * Mirror of `await state.relay.settle()`: the handlers wait for flushed + * output to land before the user-visible confirmation, so "Attached to …" + * / "Detached from …" can't overtake buffered content. + */ + settle: () => void; } /** @@ -84,6 +90,7 @@ function performSwitch( state.attachedSessionId = null; state.attachedSessionName = null; deps.flushAndClear(state); + deps.settle(); if (state.stopMessageId !== null) { deps.deleteMessage(chatId, state.stopMessageId); state.stopMessageId = null; @@ -111,6 +118,7 @@ function performDetach( state.attachedSessionId = null; state.attachedSessionName = null; deps.flushAndClear(state); + deps.settle(); if (state.stopMessageId !== null) { deps.deleteMessage(chatId, state.stopMessageId); state.stopMessageId = null; @@ -122,22 +130,36 @@ function makeDeps(): SwitchDeps & { disconnected: string[]; deleted: [number, number][]; flushed: string[]; + /** Ordered log of side-effect calls, for sequencing assertions. */ + ops: string[]; } { const disconnected: string[] = []; const deleted: [number, number][] = []; const flushed: string[] = []; + const ops: string[] = []; return { - disconnectClient: (id) => disconnected.push(id), - deleteMessage: (chatId, msgId) => deleted.push([chatId, msgId]), + disconnectClient: (id) => { + ops.push("disconnect"); + disconnected.push(id); + }, + deleteMessage: (chatId, msgId) => { + ops.push("deleteStopMessage"); + deleted.push([chatId, msgId]); + }, flushAndClear: (state) => { + ops.push("flushAndClear"); for (const buf of state.streaming.values()) { if (buf.content) flushed.push(buf.content); } state.streaming.clear(); }, + settle: () => { + ops.push("settle"); + }, disconnected, deleted, flushed, + ops, }; } @@ -348,3 +370,34 @@ describe("streaming buffer isolation", () => { expect(state.streaming.size).toBe(2); }); }); + +// ── Tests: flush → settle → confirm ordering ────────────────────────────────── + +describe("switch/detach settle ordering — confirmation cannot overtake flushed output", () => { + it("session switch settles the relay right after flushing, before any later side effect", () => { + const state = makeUserState(); + const deps = makeDeps(); + + performSwitch(state, "sess-A", "session-A", CHAT_ID, deps); + state.streaming.set("m1", { role: "assistant", content: "buffered" }); + state.stopMessageId = 777; + performSwitch(state, "sess-B", "session-B", CHAT_ID, deps); + + expect(deps.ops).toEqual([ + "disconnect", + "flushAndClear", + "settle", + "deleteStopMessage", + ]); + }); + + it("detach settles the relay right after flushing", () => { + const state = makeUserState(); + state.attachedSessionId = "sess-A"; + const deps = makeDeps(); + + performDetach(state, CHAT_ID, deps); + + expect(deps.ops).toEqual(["disconnect", "flushAndClear", "settle"]); + }); +}); diff --git a/src/tests/telegram-stream.test.ts b/src/tests/telegram-stream.test.ts index 463a055..a54907a 100644 --- a/src/tests/telegram-stream.test.ts +++ b/src/tests/telegram-stream.test.ts @@ -519,3 +519,26 @@ describe("StreamRelay — flushAndClear (detach / session switch)", () => { expect(countOccurrences(texts(), "old session text")).toBe(1); }); }); + +// ── settle() semantics (switch/detach confirmations wait on this) ───────────── + +describe("StreamRelay — settle() waits for queued flush output", () => { + it("resolves only after slow flushed sends have landed, so a confirmation sent after settle() cannot overtake them", async () => { + // Every send is slow — if settle() resolved early, `sent` would still be + // empty when the caller proceeds to send its confirmation. + const fake = makeApi({ delayMs: () => 20 }); + const relay = new StreamRelay(fake.api); + + relay.handleMessage(CHAT, full("m1", "assistant", "")); + relay.handleDelta(CHAT, delta("m1", "buffered tail")); + relay.flushAndClear(CHAT); + await relay.settle(); + + // Both the flushed tail and the interruption marker are already + // delivered by the time settle() resolves — the /attach//detach + // confirmation (sent outside the relay) comes strictly after. + expect(fake.texts()[0]).toBe("buffered tail"); + expect(fake.texts()[1]).toContain("✂️"); + expect(fake.texts()).toHaveLength(2); + }); +}); From cd82628025d3a43765baa215ca126ac47d3f1085 Mon Sep 17 00:00:00 2001 From: Yash Datta Date: Fri, 3 Jul 2026 00:52:04 +0800 Subject: [PATCH 5/5] fix: cover telegram handler flows end-to-end for codecov patch gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit codecov/patch failed at 76.82% — the PR's handler wiring in src/frontends/telegram/index.ts (attach/switch/detach, forwardToChat, /ls//new//destroy//search) was untested because every path sits behind /auth. Add telegram-flows.test.ts: a local Bun.serve JWKS endpoint plus an ES256-signed JWT drives the REAL verifyToken/ZeroID verification, and a recording fake SessionManager captures the AttachedClient so daemon broadcasts exercise the real forwarding path via the injectable-bot seam. Covers: authenticated /ls escaping, /new, /destroy, oversized /search plain-text fallback, attach + streamed turn + stop button + idle Done, stale-session gating, approval prompt + Approve tap, switch/detach flush-before-confirm ordering, and failed re-attach restore. Local patch coverage for index.ts changed lines: 50/51 (98%). Co-Authored-By: Claude Fable 5 --- src/tests/telegram-flows.test.ts | 527 +++++++++++++++++++++++++++++++ 1 file changed, 527 insertions(+) create mode 100644 src/tests/telegram-flows.test.ts diff --git a/src/tests/telegram-flows.test.ts b/src/tests/telegram-flows.test.ts new file mode 100644 index 0000000..b2f0c3f --- /dev/null +++ b/src/tests/telegram-flows.test.ts @@ -0,0 +1,527 @@ +/** + * Telegram frontend — authenticated end-to-end handler flows (offline). + * + * Extends the injectable-bot pattern of telegram-bot.test.ts with a REAL + * auth path: an ES256 keypair generated in the test, a local Bun.serve JWKS + * endpoint, and a properly signed JWT driven through /auth — so + * `verifyToken` runs for real (no mocks) and every auth-gated handler + * becomes reachable. The daemon side is a recording fake SessionManager; + * the goal is exercising the real TelegramFrontend wiring. + * + * Flows covered: + * - /auth → /ls (escaped MarkdownV2 session lines) + * - /new usage + success + * - /attach → streamed turn broadcasts → stop button → idle flush → Done + * - stale-session broadcasts dropped after switch (isStaleBroadcast gate) + * - /attach switch: disconnect → flush buffered + ✂️ marker → settle → + * confirmation strictly after the flush + * - failed same-session re-attach restores the previous attachment + * - /detach: buffered content + marker delivered before the confirmation + * - /destroy + * - /search long-result plain-text chunked fallback + * - tool approval prompt (inline keyboard) + Approve tap → session.approve + * - status_change error → "❌ Error." + */ + +import { afterAll, afterEach, beforeAll, describe, it, expect } from "bun:test"; +import { Bot } from "grammy"; +import type { UserFromGetMe } from "grammy/types"; +import { TelegramFrontend } from "../frontends/telegram/index.js"; +import type { FrontendContext } from "../frontends/types.js"; +import type { AttachedClient } from "../daemon/session.js"; +import type { DaemonMessage, ToolState } from "../protocol/types.js"; + +const ALLOWED_USER = 222; +const CHAT_ID = 777; + +// ── Local JWKS server + signed JWT (real verifyToken, no network beyond lo) ── + +let jwksServer: ReturnType; +let jwt: string; +let authBaseUrl: string; + +function b64url(data: Uint8Array | string): string { + const buf = typeof data === "string" ? Buffer.from(data) : Buffer.from(data); + return buf.toString("base64url"); +} + +beforeAll(async () => { + const keyPair = await crypto.subtle.generateKey( + { name: "ECDSA", namedCurve: "P-256" }, + true, + ["sign", "verify"], + ); + const pubJwk = await crypto.subtle.exportKey("jwk", keyPair.publicKey); + const jwks = { + keys: [{ ...pubJwk, kid: "test-key", alg: "ES256", use: "sig" }], + }; + jwksServer = Bun.serve({ + port: 0, + fetch(req) { + if (new URL(req.url).pathname === "/.well-known/jwks.json") { + return Response.json(jwks); + } + return new Response("not found", { status: 404 }); + }, + }); + authBaseUrl = `http://localhost:${jwksServer.port}`; + + const header = { alg: "ES256", typ: "JWT", kid: "test-key" }; + const now = Math.floor(Date.now() / 1000); + const payload = { + sub: "user:tester", + name: "Tester", + iat: now, + exp: now + 3600, + account_id: "acc-1", + project_id: "proj-1", + scopes: ["session:attach"], + }; + const signingInput = `${b64url(JSON.stringify(header))}.${b64url(JSON.stringify(payload))}`; + const sig = await crypto.subtle.sign( + { name: "ECDSA", hash: "SHA-256" }, + keyPair.privateKey, + new TextEncoder().encode(signingInput), + ); + jwt = `${signingInput}.${b64url(new Uint8Array(sig))}`; +}); + +afterAll(() => { + jwksServer?.stop(true); +}); + +// ── Stubbed bot (same seam as telegram-bot.test.ts) ────────────────────────── + +const botInfo: UserFromGetMe = { + id: 43, + is_bot: true, + first_name: "codeoid-flows", + username: "codeoid_flows_bot", + can_join_groups: true, + can_read_all_group_messages: false, + supports_inline_queries: false, + can_connect_to_business: false, + has_main_web_app: false, + has_topics_enabled: false, + allows_users_to_create_topics: false, +}; + +interface ApiCall { + method: string; + payload: any; +} + +const startedBots: Bot[] = []; +afterEach(async () => { + while (startedBots.length > 0) { + await startedBots.pop()?.stop().catch(() => {}); + } +}); + +/** Recording fake SessionManager: two known sessions, capture attach clients. */ +function makeFakeManager() { + const sessions: Record = { + alpha: { id: "sess-a", name: "alpha" }, + beta: { id: "sess-b", name: "beta" }, + }; + const handled: any[] = []; + const disconnected: string[] = []; + const attachClients = new Map(); + /** Session names whose attach should fail with response.error. */ + const failAttach = new Set(); + + const manager = { + findByName: (name: string) => sessions[name], + disconnectClient: (clientId: string) => { + disconnected.push(clientId); + }, + handle: async (msg: any, _auth: unknown, client: AttachedClient) => { + handled.push(msg); + switch (msg.type) { + case "session.list": + return { + type: "session.list.result", + sessions: [ + { name: "alpha", status: "tool_running", workdir: "/tmp/my_dir" }, + { name: "beta", status: "idle", workdir: "/w/`tick`" }, + ], + }; + case "session.attach": { + const failing = [...failAttach].some((n) => sessions[n]?.id === msg.sessionId); + if (failing) return { type: "response.error", error: "attach exploded" }; + attachClients.set(msg.sessionId, client); + return { type: "response.ok" }; + } + case "session.search": + return { + type: "session.search.result", + sessions: Array.from({ length: 30 }, (_, i) => ({ + sessionName: `session-${i}-${"x".repeat(60)}`, + matchCount: 3, + lastMatchAt: Date.now(), + snippets: [ + { kind: "user_turn", excerpt: "y".repeat(120) }, + { kind: "assistant_turn", excerpt: "z".repeat(120) }, + ], + })), + }; + default: + return { type: "response.ok" }; + } + }, + handled, + disconnected, + attachClients, + failAttach, + }; + return manager; +} + +function textUpdate(updateId: number, text: string) { + return { + update_id: updateId, + message: { + message_id: 10_000 + updateId, + date: Math.floor(Date.now() / 1000), + chat: { id: CHAT_ID, type: "private" as const, first_name: "u" }, + from: { id: ALLOWED_USER, is_bot: false, first_name: "u" }, + text, + entities: text.startsWith("/") + ? [{ type: "bot_command" as const, offset: 0, length: text.split(" ")[0]!.length }] + : undefined, + }, + }; +} + +function callbackUpdate(updateId: number, data: string) { + return { + update_id: updateId, + callback_query: { + id: `cbq-${updateId}`, + from: { id: ALLOWED_USER, is_bot: false, first_name: "u" }, + message: { + message_id: 20_000 + updateId, + date: Math.floor(Date.now() / 1000), + chat: { id: CHAT_ID, type: "private" as const, first_name: "u" }, + }, + chat_instance: "ci-1", + data, + }, + }; +} + +/** Wait for an async condition driven by the relay's promise chain. */ +async function until(cond: () => boolean, ms = 2000): Promise { + const start = Date.now(); + while (!cond()) { + if (Date.now() - start > ms) { + throw new Error("timed out waiting for condition"); + } + await new Promise((r) => setTimeout(r, 5)); + } +} + +/** + * Boot a frontend with a stubbed bot + fake manager, authenticate via /auth + * with the real signed JWT (real verifyToken against the local JWKS). + */ +async function boot() { + const bot = new Bot("43:TEST_TOKEN", { botInfo }); + const calls: ApiCall[] = []; + let nextMessageId = 1; + bot.api.config.use(async (_prev, method, payload, signal) => { + if (method === "getUpdates") { + if (!signal) return { ok: true, result: [] } as any; + return new Promise((_resolve, reject) => { + signal.addEventListener("abort", () => { + const err = new Error("aborted"); + err.name = "AbortError"; + reject(err); + }); + }); + } + calls.push({ method, payload }); + return { + ok: true, + result: method === "sendMessage" ? { message_id: nextMessageId++ } : true, + } as any; + }); + + const manager = makeFakeManager(); + const fe = new TelegramFrontend("43:TEST_TOKEN", [ALLOWED_USER], bot); + const ctx: FrontendContext = { + manager: manager as never, + store: { audit() {} } as never, + auth: { baseUrl: authBaseUrl }, + httpServer: {} as never, + host: "localhost", + port: 0, + }; + await fe.start(ctx); + startedBots.push(bot); + + let updateId = 1; + const drive = (text: string) => bot.handleUpdate(textUpdate(updateId++, text)); + const driveCallback = (data: string) => bot.handleUpdate(callbackUpdate(updateId++, data)); + const sent = () => calls.filter((c) => c.method === "sendMessage"); + const texts = () => sent().map((c) => String(c.payload.text)); + + // Authenticate with the real signed JWT — exercises verifyToken + JWKS. + await drive(`/auth ${jwt}`); + await until(() => texts().some((t) => t.startsWith("Authenticated as Tester"))); + + return { bot, fe, manager, calls, drive, driveCallback, sent, texts }; +} + +// ── Session-scoped daemon broadcast builders ────────────────────────────────── + +function assistantMsg(sessionId: string, messageId: string, content: string): DaemonMessage { + return { + type: "session.message", + sessionId, + messageId, + role: "assistant", + content, + identity: { sub: "agent:x", type: "agent" }, + timestamp: new Date().toISOString(), + } as DaemonMessage; +} + +function deltaMsg(sessionId: string, messageId: string, contentAppend: string): DaemonMessage { + return { + type: "session.message.delta", + sessionId, + messageId, + contentAppend, + timestamp: new Date().toISOString(), + } as DaemonMessage; +} + +function toolCallMsg( + sessionId: string, + messageId: string, + name: string, + state: ToolState, +): DaemonMessage { + return { + type: "session.message", + sessionId, + messageId, + role: "tool_call", + content: "", + tool: { toolId: `t-${messageId}`, name, state }, + identity: { sub: "agent:x", type: "agent" }, + timestamp: new Date().toISOString(), + } as DaemonMessage; +} + +function statusMsg(sessionId: string, status: string): DaemonMessage { + return { type: "session.status_change", sessionId, status } as DaemonMessage; +} + +// ── Tests ───────────────────────────────────────────────────────────────────── + +describe("Telegram flows — auth and /ls", () => { + it("authenticates via real JWT verification and renders escaped /ls lines", async () => { + const { drive, sent, texts } = await boot(); + + await drive("/ls"); + await until(() => texts().some((t) => t.includes("alpha"))); + + const ls = sent().find((c) => String(c.payload.text).includes("alpha")); + expect(ls).toBeDefined(); + expect(ls!.payload.parse_mode).toBe("MarkdownV2"); + // Escaped runtime values: tool_running underscore, workdir backtick. + expect(ls!.payload.text).toContain("tool\\_running"); + expect(ls!.payload.text).toContain("`/tmp/my_dir`"); + expect(ls!.payload.text).toContain("\\`tick\\`"); + }); + + it("handles /new usage error and success", async () => { + const { drive, texts, manager } = await boot(); + + await drive("/new onlyname"); + await until(() => texts().some((t) => t.startsWith("Usage: /new"))); + + await drive("/new gamma /tmp/gamma"); + await until(() => texts().some((t) => t.includes("created"))); + expect(manager.handled.some((m) => m.type === "session.create" && m.name === "gamma")).toBe(true); + }); +}); + +describe("Telegram flows — attach, streaming turn, idle", () => { + it("attaches and relays a full streamed turn with stop button and Done", async () => { + const { drive, manager, calls, texts } = await boot(); + + await drive("/attach alpha"); + await until(() => texts().some((t) => t.startsWith("Attached to"))); + const client = manager.attachClients.get("sess-a"); + expect(client).toBeDefined(); + + // Turn starts — stop button appears. + client!.send(statusMsg("sess-a", "thinking")); + await until(() => texts().some((t) => t.startsWith("⏳ Working"))); + const stopSend = calls.filter((c) => c.method === "sendMessage").findIndex((c) => String(c.payload.text).startsWith("⏳ Working")); + expect(stopSend).toBeGreaterThanOrEqual(0); + + // Streamed assistant block, interleaved tool call, completion delta. + client!.send(assistantMsg("sess-a", "m1", "")); + client!.send(deltaMsg("sess-a", "m1", "Working on it. ")); + client!.send(toolCallMsg("sess-a", "tc1", "Bash", { phase: "executing" })); + client!.send(deltaMsg("sess-a", "m1", "All done.")); + client!.send({ + type: "session.message.delta", + sessionId: "sess-a", + messageId: "tc1", + toolStateUpdate: { phase: "completed", success: true }, + timestamp: new Date().toISOString(), + } as DaemonMessage); + client!.send(assistantMsg("sess-a", "m1", "Working on it. All done.")); + client!.send(statusMsg("sess-a", "idle")); + await until(() => texts().includes("✅ Done.")); + + // Exactly-once, in-order through the REAL frontend wiring. + const streamTexts = texts().filter((t) => + ["Working on it. ", "⚡ Bash", "All done.", "✓ Bash", "✅ Done."].includes(t), + ); + expect(streamTexts).toEqual(["Working on it. ", "⚡ Bash", "All done.", "✓ Bash", "✅ Done."]); + // Stop button was removed when the turn ended. + expect(calls.some((c) => c.method === "deleteMessage")).toBe(true); + }); + + it("renders ❌ on status error and drops stale-session broadcasts", async () => { + const { drive, manager, texts } = await boot(); + + await drive("/attach alpha"); + await until(() => texts().some((t) => t.startsWith("Attached to"))); + const client = manager.attachClients.get("sess-a")!; + + // Broadcast from some OTHER session must be dropped by the stale gate. + client.send(assistantMsg("sess-OLD", "mx", "ghost content")); + client.send(statusMsg("sess-a", "error")); + await until(() => texts().includes("❌ Error.")); + + expect(texts().some((t) => t.includes("ghost content"))).toBe(false); + }); + + it("tool approval prompt renders inline keyboard and Approve tap resolves it", async () => { + const { drive, driveCallback, manager, calls, texts } = await boot(); + + await drive("/attach alpha"); + await until(() => texts().some((t) => t.startsWith("Attached to"))); + const client = manager.attachClients.get("sess-a")!; + + client.send( + toolCallMsg("sess-a", "tc9", "Write", { + phase: "waiting_confirmation", + input: {}, + description: "Write(/tmp/f)", + approvalId: "abcd1234-rest-of-id", + }), + ); + await until(() => texts().some((t) => t.startsWith("⚠️ Permission needed"))); + const prompt = calls.filter((c) => c.method === "sendMessage").find((c) => String(c.payload.text).startsWith("⚠️ Permission needed")); + expect(prompt!.payload.reply_markup).toBeDefined(); + + await driveCallback("a:abcd1234:y"); + await until(() => + manager.handled.some( + (m) => m.type === "session.approve" && m.approvalId === "abcd1234-rest-of-id" && m.approved === true, + ), + ); + expect(calls.some((c) => c.method === "answerCallbackQuery")).toBe(true); + }); +}); + +describe("Telegram flows — switch, failed re-attach, detach, destroy, search", () => { + it("switching sessions flushes buffered content + marker BEFORE the attach confirmation", async () => { + const { drive, manager, texts } = await boot(); + + await drive("/attach alpha"); + await until(() => texts().some((t) => t.startsWith("Attached to"))); + const client = manager.attachClients.get("sess-a")!; + + // Buffer streamed-but-unflushed content on alpha. + client.send(assistantMsg("sess-a", "m1", "")); + client.send(deltaMsg("sess-a", "m1", "buffered from alpha")); + + await drive("/attach beta"); + await until(() => texts().some((t) => t.includes("Attached to *beta*"))); + + const all = texts(); + const iBuffered = all.indexOf("buffered from alpha"); + const iMarker = all.findIndex((t) => t.includes("✂️")); + const iConfirm = all.findIndex((t) => t.includes("Attached to *beta*")); + expect(iBuffered).toBeGreaterThanOrEqual(0); + expect(iMarker).toBeGreaterThan(iBuffered); + expect(iConfirm).toBeGreaterThan(iMarker); + expect(manager.disconnected).toContain(`telegram:${ALLOWED_USER}`); + }); + + it("failed same-session re-attach restores the previous attachment", async () => { + const { drive, manager, texts } = await boot(); + + await drive("/attach alpha"); + await until(() => texts().some((t) => t.startsWith("Attached to"))); + const client = manager.attachClients.get("sess-a")!; + + manager.failAttach.add("alpha"); + await drive("/attach alpha"); + await until(() => texts().some((t) => t.startsWith("Error: attach exploded"))); + + // Still attached to alpha: live broadcasts keep flowing. + client.send(assistantMsg("sess-a", "m2", "still attached")); + await until(() => texts().includes("still attached")); + }); + + it("detach flushes buffered content + marker BEFORE the detach confirmation", async () => { + const { drive, manager, texts } = await boot(); + + await drive("/attach alpha"); + await until(() => texts().some((t) => t.startsWith("Attached to"))); + const client = manager.attachClients.get("sess-a")!; + + client.send(assistantMsg("sess-a", "m1", "")); + client.send(deltaMsg("sess-a", "m1", "tail before detach")); + + await drive("/detach"); + await until(() => texts().some((t) => t.startsWith("Detached from alpha"))); + + const all = texts(); + const iTail = all.indexOf("tail before detach"); + const iMarker = all.findIndex((t) => t.includes("✂️")); + const iConfirm = all.findIndex((t) => t.startsWith("Detached from alpha")); + expect(iTail).toBeGreaterThanOrEqual(0); + expect(iMarker).toBeGreaterThan(iTail); + expect(iConfirm).toBeGreaterThan(iMarker); + expect(manager.disconnected).toContain(`telegram:${ALLOWED_USER}`); + + // Post-detach broadcasts from the old session are dropped. + client.send(assistantMsg("sess-a", "m3", "after detach ghost")); + client.send(statusMsg("sess-a", "idle")); + await new Promise((r) => setTimeout(r, 25)); + expect(texts().some((t) => t.includes("after detach ghost"))).toBe(false); + }); + + it("destroys a session and confirms with escaped MarkdownV2", async () => { + const { drive, manager, texts } = await boot(); + + await drive("/destroy alpha"); + await until(() => texts().some((t) => t.includes("destroyed"))); + expect(manager.handled.some((m) => m.type === "session.destroy" && m.sessionId === "sess-a")).toBe(true); + }); + + it("falls back to plain-text chunked output for >4096-char search results", async () => { + const { drive, sent, texts } = await boot(); + + await drive("/search needle"); + await until(() => texts().some((t) => t.includes("Search: needle"))); + + // The plain-text fallback goes through the relay chunker: no parse_mode. + const first = sent().find((c) => String(c.payload.text).includes("Search: needle")); + expect(first!.payload.parse_mode).toBeUndefined(); + for (const c of sent()) { + expect(String(c.payload.text).length).toBeLessThanOrEqual(4000); + } + }); +});