Hold debouncer flush while a Telegram message handler is transcribing

src/telegram/bridge.test.ts

@@ -55,7 +55,7 @@ class FakeScheduler implements SchedulerLike {
   }
 }
 
-function harness(overrides: { debounceMs?: number } = {}) {
+function harness(overrides: { debounceMs?: number; transcriber?: import("../stt/openai.js").Transcriber } = {}) {
   const botInfo: UserFromGetMe = {
     id: 1, is_bot: true, first_name: "sec", username: "sec_bot", can_join_groups: false,
     can_read_all_group_messages: false, supports_inline_queries: false, can_connect_to_business: false, has_main_web_app: false,
@@ -82,6 +82,8 @@ function harness(overrides: { debounceMs?: number } = {}) {
     scheduler,
     outbound,
     logger,
+    transcriber: overrides.transcriber,
+    fetchFn: async () => new Response(new Uint8Array([0x4f, 0x67, 0x67, 0x53])),
     debounceMs: overrides.debounceMs ?? 5,
   });
   const texts = () => calls.filter((c) => c.method === "sendMessage").map((c) => c.payload.text as string);
@@ -175,6 +177,44 @@ describe("registerBridge", () => {
     );
   });
 
+  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 },
+      ]),
+    );
+  });
+
   it("drops buffered messages when /new starts a fresh session", async () => {
     const h = harness({ debounceMs: 100 });
     await h.bot.handleUpdate(textUpdate("hold this thought"));

src/telegram/bridge.ts

@@ -113,6 +113,8 @@ export function registerBridge(deps: BridgeDeps): void {
       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,
@@ -127,6 +129,9 @@ export function registerBridge(deps: BridgeDeps): void {
       }
       if (host.busy) await ctx.react("👀").catch(() => undefined);
       debouncer.add(result.input);
+    } finally {
+      debouncer.endPending();
+    }
   });
 
   let queue: Promise<void> = Promise.resolve();