Add pending-token API to InputDebouncer

src/telegram/debouncer.test.ts

@@ -71,6 +71,99 @@ describe("InputDebouncer", () => {
     vi.advanceTimersByTime(1000);
     expect(flushed).toEqual([]);
   });
+
+  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 }]);
+  });
 });
 
 describe("merge", () => {

src/telegram/debouncer.ts

@@ -10,18 +10,37 @@ export interface InputDebouncerOptions {
  * 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.timer) clearTimeout(this.timer);
-    this.timer = setTimeout(() => this.flush(), this.opts.windowMs);
-    this.timer.unref?.();
+    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). */
@@ -38,6 +57,12 @@ export class InputDebouncer {
     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;