Add spec for merging forwarded audio with its comment
Fixes the debouncer race where a fast comment message flushes on its own before slow audio transcription completes, splitting a single user turn into two.
This commit is contained in:
EugeneTes
@@ -0,0 +1,174 @@
|
||||
# Merging forwarded audio with its comment: fixing the debounce race
|
||||
|
||||
## Goal
|
||||
|
||||
When the owner forwards an audio/voice message with a comment, Telegram delivers the two as separate updates: usually the text comment first, then the forwarded audio a moment later. `normalizeMessage` transcribes audio inline, which takes several seconds. The current `InputDebouncer` (600 ms window) flushes the comment on its own before the transcription completes, so the agent sees two turns — a bare comment, then a bare `[Forwarded from …]\n<transcript>` — instead of one merged turn.
|
||||
|
||||
Same shape as a text-plus-text forward, but the async transcription breaks the timing assumption the debouncer was built on.
|
||||
|
||||
The fix keeps the debouncer's shape and just teaches it that some inputs are still being prepared, so it should not flush until they land.
|
||||
|
||||
## Non-goals
|
||||
|
||||
- **Reordering.** We do not sort the merged input by "logical" order (e.g. comment last). The user sends the messages in whatever order Telegram delivers them; the merged text reflects that arrival order, same as today for text-plus-text.
|
||||
- **Batching the "🎤 <transcript>" echo.** The visible bot echo in the Telegram chat is untouched. It is a UI receipt for the owner, not agent input.
|
||||
- **Rewriting `normalizeMessage`.** The forward/reply/photo/document logic is unchanged. This is purely a timing fix in the debouncer and its caller.
|
||||
- **Cross-message deduplication.** If the transcript happens to contain the same text as the comment (unlikely), we do not dedupe. The merged input carries both, joined by `\n\n`.
|
||||
|
||||
## Architecture
|
||||
|
||||
Two files change: `src/telegram/debouncer.ts` and `src/telegram/bridge.ts`. Tests updated in both.
|
||||
|
||||
### `InputDebouncer` — pending-token API
|
||||
|
||||
Add two methods and one counter:
|
||||
|
||||
```ts
|
||||
class InputDebouncer {
|
||||
private buffer: AgentInput[] = [];
|
||||
private timer: NodeJS.Timeout | undefined;
|
||||
private pending = 0; // NEW
|
||||
|
||||
add(input: AgentInput): void {
|
||||
this.buffer.push(input);
|
||||
if (this.pending === 0) this.armTimer(); // arm only if nobody is still processing
|
||||
}
|
||||
|
||||
/** Call before an async step that will (probably) end with an `add`. Cancels any pending flush. */
|
||||
beginPending(): void { // NEW
|
||||
this.pending++;
|
||||
if (this.timer) { clearTimeout(this.timer); this.timer = undefined; }
|
||||
}
|
||||
|
||||
/** Call in a `finally` after the async step, whether or not `add` was called. Arms the timer when the last token releases and the buffer is non-empty. */
|
||||
endPending(): void { // NEW
|
||||
if (this.pending === 0) return; // defensive; never expected
|
||||
this.pending--;
|
||||
if (this.pending === 0 && this.buffer.length > 0) this.armTimer();
|
||||
}
|
||||
|
||||
flushNow(): void { /* unchanged */ }
|
||||
discard(): void { /* unchanged; leaves `pending` alone — the outstanding tokens will still call endPending, which is a no-op on empty buffer */ }
|
||||
}
|
||||
```
|
||||
|
||||
`armTimer()` is a small refactor of the current inline `setTimeout(...)` in `add()` so it can be reused from `endPending()`.
|
||||
|
||||
Invariant: while `pending > 0`, the timer is not armed and cannot fire. As long as every `beginPending` is paired with an `endPending` (guarded by `try/finally`), the debouncer always eventually flushes.
|
||||
|
||||
### `bridge.ts` — bracket the message handler
|
||||
|
||||
Wrap the whole per-message processing block:
|
||||
|
||||
```ts
|
||||
bot.on("message", async (ctx) => {
|
||||
const isCommand = /* unchanged */;
|
||||
if (isCommand) { await ctx.reply("Unknown command."); return; }
|
||||
|
||||
debouncer.beginPending(); // NEW: before awaiting normalizeMessage
|
||||
try {
|
||||
const result = await normalizeMessage(ctx.message, { /* unchanged */ });
|
||||
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(); // NEW: always paired
|
||||
}
|
||||
});
|
||||
```
|
||||
|
||||
The token is acquired before `normalizeMessage` starts so that any peer message-handler running concurrently (fast comment, slow audio) sees `pending > 0` and knows not to flush yet. The `finally` guarantees release even if `normalizeMessage` throws or a downstream `ctx.reply` fails.
|
||||
|
||||
### Concurrency
|
||||
|
||||
Grammy's default middleware runs each update on its own promise, so two updates arriving 50 ms apart are handled concurrently. That is what creates the race today and it is also what the new mechanism relies on: both handlers hold a token at the same time, so neither can fire the timer until both have finished (or discarded their message).
|
||||
|
||||
We do not need to serialize handlers per chat; the pending-counter is enough. It also generalises: any future async step upstream of `debouncer.add` (image OCR, document scanning) gets the same protection for free.
|
||||
|
||||
### Interaction with existing methods
|
||||
|
||||
- `flushNow()` still forces an immediate flush regardless of `pending`. Only used on… actually it is not called from anywhere in the current codebase; it stays as a general-purpose safety valve.
|
||||
- `discard()` still empties the buffer. If it is called while tokens are outstanding, those handlers will call `endPending` afterwards; with an empty buffer, `armTimer` is not called and the counter simply returns to zero. Correct.
|
||||
|
||||
## Data flow
|
||||
|
||||
Before, timing exaggerated:
|
||||
|
||||
```
|
||||
T=0.000 comment message arrives → normalizeMessage returns fast → debouncer.add(comment)
|
||||
timer armed for T+0.600
|
||||
T=0.100 audio message arrives → normalizeMessage awaits transcription…
|
||||
T=0.600 timer fires → flush → agent sees turn 1: comment alone
|
||||
T=3.500 transcription resolves → debouncer.add([Forwarded…]\ntranscript)
|
||||
timer armed for T+4.100
|
||||
T=4.100 timer fires → flush → agent sees turn 2: transcript alone
|
||||
```
|
||||
|
||||
After:
|
||||
|
||||
```
|
||||
T=0.000 comment message arrives → beginPending() pending=1
|
||||
T=0.010 normalizeMessage returns → debouncer.add(comment) buffer=[comment], pending>0 → NO timer
|
||||
T=0.020 endPending() pending=0, buffer non-empty → timer armed for T+0.620
|
||||
T=0.100 audio message arrives → beginPending() pending=1, timer cancelled
|
||||
T=0.110 normalizeMessage awaits transcription…
|
||||
T=3.500 transcription resolves → debouncer.add([Forwarded…]\ntranscript) buffer=[comment, transcript], pending>0 → NO timer
|
||||
T=3.510 endPending() pending=0, buffer non-empty → timer armed for T+4.110
|
||||
T=4.110 timer fires → flush → agent sees ONE turn:
|
||||
"<comment>\n\n[Forwarded from …]\n<transcript>"
|
||||
```
|
||||
|
||||
Arrival order (comment then audio) is preserved, matching how text-plus-text forwards already merge.
|
||||
|
||||
## Testing
|
||||
|
||||
### `src/telegram/debouncer.test.ts` — new cases
|
||||
|
||||
Uses `vi.useFakeTimers()` as the existing suite does.
|
||||
|
||||
1. `beginPending` after an `add` cancels the armed timer; the timer does not fire until `endPending` runs.
|
||||
2. Calling `add` while `pending > 0` buffers the input but does not arm the timer.
|
||||
3. Nested / overlapping tokens: two `beginPending` calls require two `endPending` calls before the timer arms.
|
||||
4. `beginPending`/`endPending` with no intervening `add` (handler produced no agent input) leaves the buffer alone and does not fire.
|
||||
5. `endPending` with an empty buffer does not arm the timer (nothing to send).
|
||||
6. Race scenario: `beginPending` A, `add` A, `endPending` A (timer armed), then `beginPending` B before the timer fires (timer cancelled), then `add` B, `endPending` B → one flush with both inputs merged.
|
||||
7. `discard()` while a token is outstanding: buffer clears; when the outstanding `endPending` fires it does not arm the timer.
|
||||
|
||||
### `src/telegram/bridge.test.ts` — one end-to-end race case
|
||||
|
||||
Simulate the actual scenario: two message events dispatched 50 ms apart. The first (comment) has no audio; the second has voice with a `transcriber` whose `transcribe` returns a promise that resolves after 3 s. Advance timers past the 3 s + 600 ms window and assert:
|
||||
|
||||
- `host.send` was called exactly once.
|
||||
- The input's `text` starts with the comment and ends with `[Forwarded from …]\n<transcript>` (or in whatever order the two events were dispatched, but as a single string).
|
||||
|
||||
If the existing `bridge.test.ts` does not already have infrastructure for fake timers around `bot.on("message")`, add the minimum needed rather than reworking the file.
|
||||
|
||||
### Existing tests
|
||||
|
||||
Should keep passing. The debouncer's public behaviour for a single `add` or two consecutive `add`s within the window is unchanged when no `beginPending` is called.
|
||||
|
||||
## Error handling
|
||||
|
||||
- If `normalizeMessage` throws, the `finally` releases the token; the buffer may still hold prior inputs and will flush after the window.
|
||||
- If `debouncer.add` is somehow not reached (early `return`, no `result.input`), the `finally` still releases the token. The buffer may still hold inputs from the other concurrent handler.
|
||||
- If `endPending` is called more times than `beginPending` (bug), the defensive `if (this.pending === 0) return;` prevents the counter from going negative and re-triggering a flush.
|
||||
|
||||
## README updates
|
||||
|
||||
None. This is invisible to the user beyond the improved behaviour.
|
||||
|
||||
## Manual E2E (add to README's manual test list)
|
||||
|
||||
> Forward a voice message from another chat with a written comment. The bot should reply once, referencing both your comment and what the voice message said, not twice with each in isolation.
|
||||
|
||||
## Rejected alternatives
|
||||
|
||||
- **Serialize `bot.on("message")` handlers with a per-chat promise chain.** Blocks all subsequent input while a slow transcription runs. Feels wrong — a fast text message shouldn't be held up by an earlier slow one — and it still needs the timer-suppression logic on top, since the debouncer resets its window on each `add`. The pending-token approach subsumes this.
|
||||
- **Bump the debounce window to 10 s when audio is present.** Heuristic; slower transcriptions still split, and every audio-only message pays an extra multi-second latency for no reason.
|
||||
- **Reserve a placeholder in the buffer at `beginPending` time and replace it on `add`.** More moving parts (needs a handle returned from `beginPending`) with no benefit — the ordering is already preserved by the arrival-order of `add` calls, and a handler that produces no input needs to leave nothing behind.
|
||||
- **Track in-flight normalisations inside `normalizeMessage` itself.** Wrong layer — `normalizeMessage` is per-message and doesn't know about the debouncer. Keeping the coupling in the bridge is cleaner.
|
||||
Reference in New Issue
Block a user