From 76bbac2df955c574c95a7d7ca6c6edda334ce75e Mon Sep 17 00:00:00 2001 From: EugeneTes Date: Wed, 23 Sep 2026 07:02:09 +0000 Subject: [PATCH] Add implementation plan for forwarded audio + comment merge Three TDD tasks: pending-token API on the debouncer, bracket the bridge message handler, and one manual E2E line in the README. --- .../2026-09-23-audio-comment-debounce-race.md | 437 ++++++++++++++++++ 1 file changed, 437 insertions(+) create mode 100644 docs/superpowers/plans/2026-09-23-audio-comment-debounce-race.md diff --git a/docs/superpowers/plans/2026-09-23-audio-comment-debounce-race.md b/docs/superpowers/plans/2026-09-23-audio-comment-debounce-race.md new file mode 100644 index 0000000..02227fc --- /dev/null +++ b/docs/superpowers/plans/2026-09-23-audio-comment-debounce-race.md @@ -0,0 +1,437 @@ +# Audio-comment debounce race fix — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Make the Telegram bridge deliver a forwarded audio message and its accompanying text comment as a single agent turn instead of two, by teaching `InputDebouncer` to defer its flush timer while any message handler is still preparing its input. + +**Architecture:** Add a `pending` counter to `InputDebouncer` with `beginPending()` / `endPending()` methods that suppress and re-arm the flush timer. Bracket the entire per-message handler in `bridge.ts` with `beginPending()` … `endPending()` in a `try/finally`, so a fast comment handler cannot flush while a slow audio-transcription handler is still awaiting the transcriber. + +**Tech Stack:** TypeScript, Node.js, grammy, vitest (with `vi.useFakeTimers()` for debouncer unit tests; real timers + `vi.waitFor` in bridge tests). + +**Spec:** `docs/superpowers/specs/2026-09-23-audio-comment-debounce-race-design.md` + +--- + +## File Structure + +- **Modify** `src/telegram/debouncer.ts` — add `pending: number`, `beginPending()`, `endPending()`; factor timer arming into a private `armTimer()`; wire pending-awareness into `add()`. +- **Modify** `src/telegram/debouncer.test.ts` — add unit tests for the new pending-token behaviour. +- **Modify** `src/telegram/bridge.ts` — bracket the `bot.on("message")` handler body with `beginPending()` / `endPending()`. +- **Modify** `src/telegram/bridge.test.ts` — extend `harness()` to inject a stub `Transcriber`; add a concurrent-handling race test. +- **Modify** `README.md` — append one line to the manual end-to-end checklist. + +No new files. All changes are on `master` (no worktree needed for a change this small — the workflow rule to fetch is satisfied by the earlier `git pull`). + +--- + +## Task 1: Debouncer pending-token API (TDD) + +**Files:** +- Modify: `src/telegram/debouncer.ts` +- Modify: `src/telegram/debouncer.test.ts` + +- [ ] **Step 1: Write failing tests for the new behaviour** + +Append these tests inside the existing `describe("InputDebouncer", …)` block in `src/telegram/debouncer.test.ts`, after the existing `it("discard drops buffered items …")` and before the closing `});` of the block: + +```ts + it("beginPending after an add cancels the armed timer until endPending", () => { + const flushed: AgentInput[] = []; + const d = new InputDebouncer({ windowMs: 200, onFlush: (i) => flushed.push(i) }); + d.add({ text: "hi" }); + d.beginPending(); + vi.advanceTimersByTime(500); + expect(flushed).toEqual([]); + d.endPending(); + vi.advanceTimersByTime(199); + expect(flushed).toEqual([]); + vi.advanceTimersByTime(1); + expect(flushed).toEqual([{ text: "hi", images: undefined }]); + }); + + it("add during pending does not arm the timer; endPending arms it", () => { + const flushed: AgentInput[] = []; + const d = new InputDebouncer({ windowMs: 100, onFlush: (i) => flushed.push(i) }); + d.beginPending(); + d.add({ text: "buffered" }); + vi.advanceTimersByTime(500); + expect(flushed).toEqual([]); + d.endPending(); + vi.advanceTimersByTime(100); + expect(flushed).toEqual([{ text: "buffered", images: undefined }]); + }); + + it("requires one endPending per beginPending before the timer arms", () => { + const flushed: AgentInput[] = []; + const d = new InputDebouncer({ windowMs: 100, onFlush: (i) => flushed.push(i) }); + d.beginPending(); + d.beginPending(); + d.add({ text: "one" }); + d.add({ text: "two" }); + d.endPending(); + vi.advanceTimersByTime(500); + expect(flushed).toEqual([]); + d.endPending(); + vi.advanceTimersByTime(100); + expect(flushed).toEqual([{ text: "one\n\ntwo", images: undefined }]); + }); + + it("beginPending / endPending with no add does not fire", () => { + const flushed: AgentInput[] = []; + const d = new InputDebouncer({ windowMs: 50, onFlush: (i) => flushed.push(i) }); + d.beginPending(); + d.endPending(); + vi.advanceTimersByTime(1000); + expect(flushed).toEqual([]); + }); + + it("simulates the comment-then-audio race and delivers one merged flush", () => { + const flushed: AgentInput[] = []; + const d = new InputDebouncer({ windowMs: 100, onFlush: (i) => flushed.push(i) }); + // Comment handler starts. + d.beginPending(); + // Audio handler starts a beat later. + d.beginPending(); + // Comment handler finishes fast: adds, releases. + d.add({ text: "look at this" }); + d.endPending(); + // A long transcription window passes; still pending, so no flush. + vi.advanceTimersByTime(3000); + expect(flushed).toEqual([]); + // Audio handler finishes: adds, releases. + d.add({ text: "[Forwarded from @alice]\nhello everyone" }); + d.endPending(); + vi.advanceTimersByTime(100); + expect(flushed).toEqual([ + { text: "look at this\n\n[Forwarded from @alice]\nhello everyone", images: undefined }, + ]); + }); + + it("discard while a token is outstanding leaves the timer unarmed after release", () => { + const flushed: AgentInput[] = []; + const d = new InputDebouncer({ windowMs: 100, onFlush: (i) => flushed.push(i) }); + d.beginPending(); + d.add({ text: "gone" }); + d.discard(); + d.endPending(); + vi.advanceTimersByTime(1000); + expect(flushed).toEqual([]); + }); + + it("extra endPending calls are ignored (pending never goes negative)", () => { + const flushed: AgentInput[] = []; + const d = new InputDebouncer({ windowMs: 100, onFlush: (i) => flushed.push(i) }); + d.endPending(); + d.endPending(); + d.add({ text: "hi" }); + vi.advanceTimersByTime(100); + expect(flushed).toEqual([{ text: "hi", images: undefined }]); + }); +``` + +- [ ] **Step 2: Run the new tests and confirm they fail** + +Run: `npm test -- src/telegram/debouncer.test.ts` + +Expected: seven new failing tests (`d.beginPending is not a function` / `d.endPending is not a function`). Existing tests still pass. + +- [ ] **Step 3: Implement `beginPending` / `endPending` and pending-awareness in `add`** + +Replace the entire contents of `src/telegram/debouncer.ts` with: + +```ts +import type { AgentImage, AgentInput } from "../agent/events.js"; + +export interface InputDebouncerOptions { + windowMs: number; + onFlush: (input: AgentInput) => void; +} + +/** + * Buffers AgentInputs arriving in quick succession and flushes a single merged + * input once the window goes quiet. Telegram sends "forward with a comment" as + * two separate messages (the comment, then the forwarded content); this stitches + * them together so the agent sees one turn. + * + * The pending counter lets callers announce "I'm about to prepare an input" + * before an async step (e.g. audio transcription). While `pending > 0` the + * flush timer is never armed, so a fast handler cannot flush while a slow one + * is still awaiting. Pair every `beginPending()` with `endPending()` in a + * `try/finally`. + */ +export class InputDebouncer { + private buffer: AgentInput[] = []; + private timer: NodeJS.Timeout | undefined; + private pending = 0; + + constructor(private readonly opts: InputDebouncerOptions) {} + + add(input: AgentInput): void { + this.buffer.push(input); + if (this.pending === 0) this.armTimer(); + } + + beginPending(): void { + this.pending++; + if (this.timer) { + clearTimeout(this.timer); + this.timer = undefined; + } + } + + endPending(): void { + if (this.pending === 0) return; + this.pending--; + if (this.pending === 0 && this.buffer.length > 0) this.armTimer(); + } + + /** Flushes immediately (e.g. on shutdown or when the caller wants to force delivery). */ + flushNow(): void { + if (this.timer) clearTimeout(this.timer); + this.timer = undefined; + this.flush(); + } + + /** Drops the buffered items without delivering them. */ + discard(): void { + if (this.timer) clearTimeout(this.timer); + this.timer = undefined; + this.buffer = []; + } + + private armTimer(): void { + if (this.timer) clearTimeout(this.timer); + this.timer = setTimeout(() => this.flush(), this.opts.windowMs); + this.timer.unref?.(); + } + + private flush(): void { + this.timer = undefined; + if (this.buffer.length === 0) return; + const merged = merge(this.buffer); + this.buffer = []; + this.opts.onFlush(merged); + } +} + +export function merge(inputs: AgentInput[]): AgentInput { + const parts: string[] = []; + const images: AgentImage[] = []; + for (const input of inputs) { + if (input.text) parts.push(input.text); + if (input.images) images.push(...input.images); + } + return { text: parts.join("\n\n"), images: images.length > 0 ? images : undefined }; +} +``` + +- [ ] **Step 4: Run the debouncer tests and confirm they pass** + +Run: `npm test -- src/telegram/debouncer.test.ts` + +Expected: all tests pass (the seven new ones plus the pre-existing suite). + +- [ ] **Step 5: Run the full test suite and typecheck to confirm nothing else broke** + +Run: `npm test && npm run typecheck` + +Expected: green. `npm run typecheck` completes without errors. + +- [ ] **Step 6: Commit** + +```bash +git add src/telegram/debouncer.ts src/telegram/debouncer.test.ts +git commit -m "Add pending-token API to InputDebouncer" +``` + +--- + +## Task 2: Bracket the bridge message handler with pending tokens (TDD) + +**Files:** +- Modify: `src/telegram/bridge.ts` +- Modify: `src/telegram/bridge.test.ts` + +- [ ] **Step 1: Extend `harness()` to accept a stub transcriber** + +In `src/telegram/bridge.test.ts`, modify the `harness()` function signature and body to accept a `transcriber` option and pass it through to `registerBridge`. Replace the existing: + +```ts +function harness(overrides: { debounceMs?: number } = {}) { +``` + +with: + +```ts +function harness(overrides: { debounceMs?: number; transcriber?: import("../stt/openai.js").Transcriber } = {}) { +``` + +Then, inside the body, in the `registerBridge({ … })` call, add `transcriber: overrides.transcriber,` next to the other deps (right after `logger,` is a natural spot): + +```ts + registerBridge({ + bot, + config: { telegramBotToken: "1:TEST", telegramOwnerId: OWNER, workspaceDir: process.cwd(), timezone: "UTC" }, + host, + scheduler, + outbound, + logger, + transcriber: overrides.transcriber, + debounceMs: overrides.debounceMs ?? 5, + }); +``` + +- [ ] **Step 2: Also stub the `getFile` return for `.oga` voice in the harness (already present) and add a `getFile` return for the specific test voice** + +The harness already returns `voice/file_1.oga` for any `getFile` (see the existing `bot.api.config.use` block). No change needed here — leaving this step as a check: confirm `voice/file_1.oga` is what the harness's fake API returns and that the bridge's real `downloadFile` will then `fetch(...)` that URL. In tests we do not want a real HTTP fetch — we work around this by making the transcriber's `transcribe` method never depend on the download bytes (it just returns a preconfigured string). But `downloadFile` itself will still `fetch`, which in Node's test environment will try to hit `api.telegram.org`. Verify by looking at bridge.ts `downloadFile` — it accepts `fetchFn`. We therefore also need to inject `fetchFn`. + +In the `registerBridge` call inside `harness()`, add a stub `fetchFn` too: + +```ts + registerBridge({ + bot, + config: { telegramBotToken: "1:TEST", telegramOwnerId: OWNER, workspaceDir: process.cwd(), timezone: "UTC" }, + host, + scheduler, + outbound, + logger, + transcriber: overrides.transcriber, + fetchFn: async () => new Response(new Uint8Array([0x4f, 0x67, 0x67, 0x53])), // "OggS" + debounceMs: overrides.debounceMs ?? 5, + }); +``` + +- [ ] **Step 3: Write the failing race test** + +Add this test inside `describe("registerBridge", …)` in `src/telegram/bridge.test.ts`, immediately after the existing `it("merges a forward-with-comment pair into a single agent turn", …)`: + +```ts + it("merges a fast comment with a slow-transcribing forwarded voice into one agent turn", async () => { + let resolveTranscription!: (text: string) => void; + const transcription = new Promise((r) => { resolveTranscription = r; }); + const transcriber = { transcribe: () => transcription }; + const h = harness({ debounceMs: 30, transcriber }); + + // Comment arrives first, resolves fast. + const commentDone = h.bot.handleUpdate(textUpdate("about this voice note")); + + // Forwarded voice message arrives right after, transcription blocks. + const voiceUpdate = textUpdate(""); + delete (voiceUpdate.message as { text?: string }).text; + (voiceUpdate.message as { voice?: unknown }).voice = { file_id: "v1", file_unique_id: "u", duration: 3, mime_type: "audio/ogg" }; + (voiceUpdate.message as { forward_origin?: unknown }).forward_origin = { + type: "user", + date: 0, + sender_user: { id: 77, is_bot: false, first_name: "Alice", username: "alice" }, + }; + const voiceDone = h.bot.handleUpdate(voiceUpdate); + + // Comment handler completes; audio handler is still awaiting the transcriber. + await commentDone; + + // Wait past the debounce window to prove the flush is being suppressed. + await new Promise((r) => setTimeout(r, 100)); + expect(h.host.sent).toEqual([]); + + // Transcription resolves; audio handler unblocks and both flush together. + resolveTranscription("buy milk"); + await voiceDone; + + await vi.waitFor(() => + expect(h.host.sent).toEqual([ + { text: "about this voice note\n\n[Forwarded from @alice]\nbuy milk", images: undefined }, + ]), + ); + }); +``` + +- [ ] **Step 4: Run the new test and confirm it fails** + +Run: `npm test -- src/telegram/bridge.test.ts` + +Expected: the new test fails. Depending on timing you should see either two entries in `h.host.sent` (the comment flushed on its own, then the transcript flushed later) or the assertion at "expect(h.host.sent).toEqual([])" fails because the comment already flushed on its own. Either way it is red. + +- [ ] **Step 5: Bracket the bridge message handler with beginPending / endPending** + +In `src/telegram/bridge.ts`, find the `bot.on("message", async (ctx) => { … })` handler (currently around line 110–130) and replace its body so it becomes: + +```ts + bot.on("message", async (ctx) => { + const isCommand = ctx.message.entities?.some((e) => e.type === "bot_command" && e.offset === 0) ?? false; + if (isCommand) { + await ctx.reply("Unknown command."); + return; + } + debouncer.beginPending(); + try { + const result = await normalizeMessage(ctx.message, { + download: (fileId) => downloadFile(bot, config.telegramBotToken, fileId, deps.fetchFn), + transcriber: deps.transcriber, + workspaceDir: config.workspaceDir, + botUserId: bot.botInfo?.id, + }); + for (const note of result.notes) await ctx.reply(note); + if (result.transcript) await ctx.reply(`🎤 ${result.transcript}`); + if (!result.input) { + if (result.notes.length === 0) await ctx.reply("I can handle text, photos, voice notes, and files."); + return; + } + if (host.busy) await ctx.react("👀").catch(() => undefined); + debouncer.add(result.input); + } finally { + debouncer.endPending(); + } + }); +``` + +Note: the `beginPending()` call is **inside** the handler but **outside** the try, so a throw from `beginPending` itself (impossible in practice) would not leak a token, and the `endPending` is guaranteed by the `finally`. The command-early-return above `beginPending` is deliberate — commands do not go through the debouncer at all. + +- [ ] **Step 6: Run the bridge tests and confirm the new test passes** + +Run: `npm test -- src/telegram/bridge.test.ts` + +Expected: all tests pass, including the new race test and the existing "merges a forward-with-comment pair" test (which uses sequential `await`s and therefore was never subject to the race — it still works because `beginPending` is released before the next handler starts). + +- [ ] **Step 7: Run the full suite and typecheck** + +Run: `npm test && npm run typecheck` + +Expected: everything green. + +- [ ] **Step 8: Commit** + +```bash +git add src/telegram/bridge.ts src/telegram/bridge.test.ts +git commit -m "Hold debouncer flush while a Telegram message handler is transcribing" +``` + +--- + +## Task 3: Manual E2E entry in the README + +**Files:** +- Modify: `README.md` + +- [ ] **Step 1: Append item 20 to the manual end-to-end checklist** + +Open `README.md` and append a new numbered item after the existing item 19 (currently the last item in the "Manual end-to-end checklist" section, line 107): + +``` +20. Forward a voice message from another chat and add a text comment. The bot should reply once, referencing both your comment and what the voice said, not twice with each in isolation. +``` + +- [ ] **Step 2: Commit** + +```bash +git add README.md +git commit -m "Add manual E2E entry for forward-with-audio-comment merge" +``` + +--- + +## Self-Review Notes + +- **Spec coverage:** Task 1 covers the debouncer API (`InputDebouncer — pending-token API` section of the spec) and all seven debouncer unit test cases. Task 2 covers the `bridge.ts` wiring (`Bridge — bracket the message handler` section) and the end-to-end bridge test. Task 3 covers the README manual-E2E addition. +- **Placeholder scan:** no TBDs, no "similar to Task N", every code step has a complete code block. +- **Type consistency:** `beginPending()` / `endPending()` are the only new method names; both spellings are used identically in the debouncer, the bridge, and every test. +- **Discarded-during-pending path** is covered by both the debouncer unit test (Task 1 Step 1, sixth new test) and by the fact that `discard()` is only called from the `/new` command handler, which is guarded by the `isCommand` early return in the bridge (so it can never race with a `beginPending()` in the same request).