From 98e1ed2613fab23c729ac2aaef50e0762636f387 Mon Sep 17 00:00:00 2001 From: EugeneTes Date: Wed, 23 Sep 2026 07:05:09 +0000 Subject: [PATCH] Add pending-token API to InputDebouncer --- src/telegram/debouncer.test.ts | 93 ++++++++++++++++++++++++++++++++++ src/telegram/debouncer.ts | 31 ++++++++++-- 2 files changed, 121 insertions(+), 3 deletions(-) diff --git a/src/telegram/debouncer.test.ts b/src/telegram/debouncer.test.ts index d586a98..c5a4f8a 100644 --- a/src/telegram/debouncer.test.ts +++ b/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", () => { diff --git a/src/telegram/debouncer.ts b/src/telegram/debouncer.ts index 98ddace..4d30e50 100644 --- a/src/telegram/debouncer.ts +++ b/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;