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 07:02:09 +00:00
parent 483df6f528
commit 76bbac2df9
1 changed files with 437 additions and 0 deletions
--- a/docs/superpowers/plans/2026-09-23-audio-comment-debounce-race.md
+++ 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<string>((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).