From 8a0dedb17d2cf2bd2ca99d83fcd42b3a1fc1f4a4 Mon Sep 17 00:00:00 2001 From: Madhu Ramasubramanian Date: Wed, 23 Sep 2026 21:14:53 -0400 Subject: [PATCH 1/2] VAPI-3990: return a bot's mark only after its queued audio has played out In Twilio's Media Streams protocol a bot sends a mark after queuing audio and is told, via the mark echoed back, when that audio has finished playing. Bots gate their next turn on it. The bridge echoed the mark the instant it arrived, so a mark-gated bot was told its utterance had finished while the audio was still queued and started its next turn over its own speech. Bandwidth's StartStream protocol sends no playback-complete signal and BwStreamSource has nothing to wait on, so the bridge now tracks playout by duration: - Each outbound media frame extends a playout clock by its mulaw duration (8 kHz mono, 8 bytes per ms). - A mark becomes due at the clock's value when it arrived, so audio queued after a mark does not delay it. Marks return in arrival order through a single timer. Nothing queued means the mark returns at once. - clear empties the queue and returns every outstanding mark immediately, per Twilio's docs ("empties all buffered audio and causes any mark messages to be sent back"). The ticket text said to discard them; the documented behavior is followed instead so a mark-gated bot is never left waiting. - stop and close() drop pending marks, since nothing queued will play. - Marks sent back now carry a sequenceNumber, as Twilio's do. Expose pendingPlayoutMs() for tests and diagnostics, and export mulawPayloadDurationMs(). Update AGENTS.md and replace the test that pinned the synchronous echo with cases for the clock math, immediate echo with nothing queued, holding a mark, ordering, non-delay by later audio, stop, and clear. Validating the clock against real Bandwidth playback needs the live source from VAPI-3991. --- AGENTS.md | 6 +- src/streams/bridge.ts | 87 +++++++++++++++++++-- test/streams-wire.test.ts | 158 ++++++++++++++++++++++++++++++++++---- 3 files changed, 227 insertions(+), 24 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index fa4bbf5..ef95a94 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -117,7 +117,11 @@ Translation is a fixed rulebook (`src/matrix/twilio-voice.json`), not a guess. children map to nested `` elements in order (Bandwidth allows at most 12; extras are dropped with a warning). Bandwidth echoes them in its `start` event as `streamParams`, and the bridge forwards them to the bot as - Twilio `customParameters`. + Twilio `customParameters`. Bandwidth sends no playback-complete signal, so + the bridge tracks playout by duration: a bot's `mark` is returned once the + mulaw audio queued before it has had time to play (8 bytes per ms), and a + `clear` empties the queue and returns every outstanding mark at once, as + Twilio does. - `Conference` — basic named conferences work, but `waitUrl` hold music has no Bandwidth equivalent, `beep` is only partially supported, and `startConferenceOnEnter`/`endConferenceOnExit`/`maxParticipants` have no diff --git a/src/streams/bridge.ts b/src/streams/bridge.ts index e56d178..56eee87 100644 --- a/src/streams/bridge.ts +++ b/src/streams/bridge.ts @@ -9,7 +9,19 @@ import type { EventEmitter } from "node:events"; * * flush() is called when the bot sends a Twilio "clear" event, signalling that * buffered audio on the playout queue should be discarded. + * + * There is deliberately no playback-complete signal here: Bandwidth's + * StartStream protocol does not send one. The bridge tracks playout by + * duration instead (see TwilioStreamBridge.pendingPlayoutMs). */ + +/** Bytes of 8 kHz mono mulaw per millisecond of audio: 8000 samples/s, 1 byte each. */ +const MULAW_BYTES_PER_MS = 8; + +/** Milliseconds of playback represented by a base64-encoded mulaw payload. */ +export function mulawPayloadDurationMs(payloadB64: string): number { + return Buffer.from(payloadB64, "base64").length / MULAW_BYTES_PER_MS; +} export interface BwStreamSource extends EventEmitter { sendMedia(payloadB64: string): void; flush(): void; @@ -62,6 +74,18 @@ export class TwilioStreamBridge { private chunkSeq = 0; private readyPromise: Promise; + // ── Playout clock ──────────────────────────────────────────────────────── + // Twilio returns a bot's "mark" only once every media frame queued before it + // has finished playing on the call. Bandwidth gives us no playback signal, so + // we model the playout queue as a clock: each outbound frame extends + // `playoutEndAt` by its mulaw duration, and a mark becomes due at whatever + // `playoutEndAt` was when the mark arrived. Audio queued after a mark does not + // delay it. Marks are due in arrival order because the clock only moves + // forward between clears. + private playoutEndAt = 0; + private pendingMarks: { mark: unknown; dueAt: number }[] = []; + private markTimer: NodeJS.Timeout | undefined; + constructor(private opts: BridgeOpts) { this.streamSid = "MZ" + randomBytes(16).toString("hex"); this.ws = new WebSocket(opts.botUrl); @@ -117,8 +141,9 @@ export class TwilioStreamBridge { }); }); - // BW network → bot: stream ended + // BW network → bot: stream ended. Anything still queued will never play. opts.source.on("stop", () => { + this.dropPendingMarks(); this.send({ event: "stop", sequenceNumber: String(++this.seq), @@ -144,21 +169,31 @@ export class TwilioStreamBridge { switch ((msg as any).event) { case "media": - // Bot is sending audio to be played out on the call + // Bot is sending audio to be played out on the call. Extend the + // playout clock by the frame's duration before handing it on. if (msg.media && typeof (msg.media as any).payload === "string") { - this.opts.source.sendMedia((msg.media as any).payload as string); + const payload = (msg.media as any).payload as string; + const now = Date.now(); + this.playoutEndAt = Math.max(now, this.playoutEndAt) + mulawPayloadDurationMs(payload); + this.opts.source.sendMedia(payload); } break; case "mark": - // Echo the mark back to acknowledge playback completion. - // Real playout tracking comes with the live BW binding. - this.send({ event: "mark", streamSid: this.streamSid, mark: (msg as any).mark }); + // Per Twilio: the mark comes back when the audio queued before it has + // finished playing. Nothing queued means it comes back at once. + this.pendingMarks.push({ mark: (msg as any).mark, dueAt: this.playoutEndAt }); + this.flushDueMarks(); break; case "clear": - // Flush buffered audio on the BW source's playout queue + // Per Twilio: "empties all buffered audio and causes any mark messages + // to be sent back". Discard the queue, then return every outstanding + // mark immediately so a mark-gated bot is not left waiting forever. this.opts.source.flush(); + this.playoutEndAt = 0; + for (const p of this.pendingMarks) p.dueAt = 0; + this.flushDueMarks(); break; } }); @@ -168,11 +203,49 @@ export class TwilioStreamBridge { return this.readyPromise; } + /** Milliseconds of bot audio still to play on the call, by the playout clock. */ + pendingPlayoutMs(): number { + return Math.max(0, this.playoutEndAt - Date.now()); + } + close(): void { + this.dropPendingMarks(); this.opts.source.close(); this.ws.close(); } + /** Echo every mark whose audio has played out, then arm one timer for the next. */ + private flushDueMarks(): void { + if (this.markTimer) { + clearTimeout(this.markTimer); + this.markTimer = undefined; + } + const now = Date.now(); + while (this.pendingMarks.length && this.pendingMarks[0].dueAt <= now) { + const { mark } = this.pendingMarks.shift()!; + this.send({ + event: "mark", + sequenceNumber: String(++this.seq), + streamSid: this.streamSid, + mark, + }); + } + if (this.pendingMarks.length) { + const wait = this.pendingMarks[0].dueAt - now; + this.markTimer = setTimeout(() => this.flushDueMarks(), wait); + this.markTimer.unref?.(); + } + } + + private dropPendingMarks(): void { + if (this.markTimer) { + clearTimeout(this.markTimer); + this.markTimer = undefined; + } + this.pendingMarks = []; + this.playoutEndAt = 0; + } + private send(obj: unknown): void { if (this.ws.readyState === WebSocket.OPEN) this.ws.send(JSON.stringify(obj)); } diff --git a/test/streams-wire.test.ts b/test/streams-wire.test.ts index c2aa1e6..c8d8d03 100644 --- a/test/streams-wire.test.ts +++ b/test/streams-wire.test.ts @@ -13,9 +13,13 @@ import { EventEmitter } from "node:events"; import { TwilioStreamBridge, customParametersFromBwStart, + mulawPayloadDurationMs, type BwStreamSource, } from "../src/streams/bridge.js"; +/** Base64 mulaw silence of the given duration at 8 kHz (8 bytes per ms). */ +const mulawMs = (ms: number) => Buffer.alloc(ms * 8, 0xff).toString("base64"); + // ─── helpers ──────────────────────────────────────────────────────────────── /** FakeBwSource that records outbound payloads and exposes flush tracking. */ @@ -325,9 +329,13 @@ describe("dtmf", () => { }); // ─── mark ─────────────────────────────────────────────────────────────────── +// +// VAPI-3990: the bridge used to echo a mark the instant it arrived, so a +// mark-gated bot was told its utterance had finished while the audio was still +// queued. Bandwidth sends no playback signal, so playout is tracked by duration. describe("mark", () => { - it("echoes mark back to the bot preserving streamSid and mark.name", async () => { + async function openBridge() { const port = nextPort(); const { messages, socket, close } = await botServer(port); const source = new FakeBwSource(); @@ -339,31 +347,149 @@ describe("mark", () => { }); await bridge.ready(); const bot = await socket; + const send = (obj: object) => bot.send(JSON.stringify({ streamSid: bridge.streamSid, ...obj })); + const marks = () => messages.filter((m: any) => m.event === "mark") as any[]; + return { bridge, source, messages, send, marks, close: () => (bridge.close(), close()) }; + } - bot.send( - JSON.stringify({ - event: "mark", - streamSid: bridge.streamSid, - mark: { name: "playback-done" }, - }) - ); + it("mulawPayloadDurationMs: 8 kHz mono mulaw is 8 bytes per millisecond", () => { + expect(mulawPayloadDurationMs(mulawMs(20))).toBe(20); + expect(mulawPayloadDurationMs(mulawMs(1000))).toBe(1000); + expect(mulawPayloadDurationMs("")).toBe(0); + }); - await waitFor( - () => (messages.filter((m: any) => m.event === "mark") as any[]).length >= 1 - ); - bridge.close(); - close(); + it("echoes a mark at once when no audio is queued, preserving streamSid and mark.name", async () => { + const t = await openBridge(); + t.send({ event: "mark", mark: { name: "playback-done" } }); + await waitFor(() => t.marks().length >= 1); + t.close(); - const markMsg = messages.find((m: any) => m.event === "mark") as any; - expect(markMsg.event).toBe("mark"); - expect(markMsg.streamSid).toBe(bridge.streamSid); + const markMsg = t.marks()[0]; + expect(markMsg.streamSid).toBe(t.bridge.streamSid); expect(markMsg.mark).toEqual({ name: "playback-done" }); + // Twilio numbers the marks it sends back like every other outbound message. + expect(markMsg.sequenceNumber).toBeDefined(); + }); + + it("holds a mark until the audio queued before it has played out", async () => { + const t = await openBridge(); + t.send({ event: "media", media: { payload: mulawMs(300) } }); + const sentAt = Date.now(); + t.send({ event: "mark", mark: { name: "turn-1" } }); + + // The bytes reach the Bandwidth side immediately, but the mark must not. + await waitFor(() => t.source.sent.length === 1); + await new Promise((r) => setTimeout(r, 100)); + expect(t.marks()).toHaveLength(0); + expect(t.bridge.pendingPlayoutMs()).toBeGreaterThan(100); + + await waitFor(() => t.marks().length >= 1, 1000); + const ackAfterMs = Date.now() - sentAt; + t.close(); + + expect(t.marks()[0].mark).toEqual({ name: "turn-1" }); + expect(ackAfterMs).toBeGreaterThanOrEqual(280); + expect(t.bridge.pendingPlayoutMs()).toBe(0); + }); + + it("pendingPlayoutMs reflects the bytes queued, and frames accumulate", async () => { + const t = await openBridge(); + t.send({ event: "media", media: { payload: mulawMs(500) } }); + t.send({ event: "media", media: { payload: mulawMs(500) } }); + await waitFor(() => t.source.sent.length === 2); + const pending = t.bridge.pendingPlayoutMs(); + t.close(); + expect(pending).toBeGreaterThan(900); + expect(pending).toBeLessThanOrEqual(1000); + }); + + it("returns marks in order, each after its own preceding audio", async () => { + const t = await openBridge(); + t.send({ event: "media", media: { payload: mulawMs(150) } }); + t.send({ event: "mark", mark: { name: "a" } }); + t.send({ event: "media", media: { payload: mulawMs(150) } }); + t.send({ event: "mark", mark: { name: "b" } }); + + const seen: { name: string; at: number }[] = []; + await waitFor(() => { + for (const m of t.marks().slice(seen.length)) seen.push({ name: m.mark.name, at: Date.now() }); + return seen.length >= 2; + }, 1500); + t.close(); + + expect(seen.map((s) => s.name)).toEqual(["a", "b"]); + expect(seen[1].at - seen[0].at).toBeGreaterThanOrEqual(100); + }); + + it("audio queued after a mark does not delay that mark", async () => { + const t = await openBridge(); + t.send({ event: "media", media: { payload: mulawMs(100) } }); + const sentAt = Date.now(); + t.send({ event: "mark", mark: { name: "early" } }); + t.send({ event: "media", media: { payload: mulawMs(5000) } }); + + await waitFor(() => t.marks().length >= 1, 1000); + const ackAfterMs = Date.now() - sentAt; + t.close(); + + expect(t.marks()[0].mark).toEqual({ name: "early" }); + expect(ackAfterMs).toBeLessThan(800); + }); + + it("drops pending marks when the stream stops", async () => { + const t = await openBridge(); + t.send({ event: "media", media: { payload: mulawMs(5000) } }); + t.send({ event: "mark", mark: { name: "never" } }); + await waitFor(() => t.source.sent.length === 1); + + t.source.emit("stop"); + await waitFor(() => t.messages.some((m: any) => m.event === "stop")); + await new Promise((r) => setTimeout(r, 50)); + t.close(); + + expect(t.marks()).toHaveLength(0); + expect(t.bridge.pendingPlayoutMs()).toBe(0); }); }); // ─── clear ────────────────────────────────────────────────────────────────── describe("clear", () => { + it("empties the playout queue and returns every outstanding mark at once (Twilio semantics)", async () => { + const port = nextPort(); + const { messages, socket, close } = await botServer(port); + const source = new FakeBwSource(); + const bridge = new TwilioStreamBridge({ + botUrl: `ws://127.0.0.1:${port}`, + callSid: "CAiii", + accountSid: "ACjjj", + source, + }); + await bridge.ready(); + const bot = await socket; + const send = (obj: object) => bot.send(JSON.stringify({ streamSid: bridge.streamSid, ...obj })); + + // Five seconds queued, two marks behind it. Without the clear they would + // come back after ~5 s; the bot interrupting (barge-in) must not wait that long. + send({ event: "media", media: { payload: mulawMs(5000) } }); + send({ event: "mark", mark: { name: "m1" } }); + send({ event: "mark", mark: { name: "m2" } }); + await waitFor(() => source.sent.length === 1); + expect(messages.filter((m: any) => m.event === "mark")).toHaveLength(0); + + const clearedAt = Date.now(); + send({ event: "clear" }); + await waitFor(() => messages.filter((m: any) => m.event === "mark").length >= 2, 1000); + const elapsed = Date.now() - clearedAt; + bridge.close(); + close(); + + expect(source.flushed).toBe(1); + expect(elapsed).toBeLessThan(500); + expect((messages.filter((m: any) => m.event === "mark") as any[]).map((m) => m.mark.name)).toEqual(["m1", "m2"]); + expect(bridge.pendingPlayoutMs()).toBe(0); + }); + it("calls source.flush() when the bot sends a clear event", async () => { const port = nextPort(); const { socket, close } = await botServer(port); From 7857fa49caac6ed097146ce1b49bb2079efad4bb Mon Sep 17 00:00:00 2001 From: Madhu Ramasubramanian Date: Wed, 23 Sep 2026 21:22:28 -0400 Subject: [PATCH 2/2] VAPI-3990: add playout latency pad, close test gaps, avoid per-frame decode Review follow-ups: - Marks return systematically early on the live path: the playout clock starts when a frame reaches the bridge, but the caller hears it one-way network latency plus Bandwidth's jitter buffer later. Add BridgeOpts.playoutLatencyPadMs, added to the due time of any mark with audio ahead of it (idle marks are unaffected). Default 0 until VAPI-3991 yields real measurements. - mulawPayloadDurationMs derives the decoded length from the base64 string instead of allocating a Buffer per frame. A test checks it against a real decode for every padding shape. - Tests: the playout clock resets after clear (audio queued afterwards is timed from scratch, ~100 ms not ~1100 ms); close() drops pending marks and their timer; the latency pad. - clear now hands the outstanding marks to a direct echo loop instead of zeroing their due times, and mark sending is factored into sendMark(). --- src/streams/bridge.ts | 44 +++++++++++++------- test/streams-wire.test.ts | 88 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 117 insertions(+), 15 deletions(-) diff --git a/src/streams/bridge.ts b/src/streams/bridge.ts index 56eee87..2b48a82 100644 --- a/src/streams/bridge.ts +++ b/src/streams/bridge.ts @@ -18,10 +18,16 @@ import type { EventEmitter } from "node:events"; /** Bytes of 8 kHz mono mulaw per millisecond of audio: 8000 samples/s, 1 byte each. */ const MULAW_BYTES_PER_MS = 8; -/** Milliseconds of playback represented by a base64-encoded mulaw payload. */ +/** Milliseconds of playback represented by a base64-encoded mulaw payload. + * The decoded length is derived from the string so no per-frame Buffer is + * allocated: 4 base64 chars encode 3 bytes, less one byte per '=' pad. */ export function mulawPayloadDurationMs(payloadB64: string): number { - return Buffer.from(payloadB64, "base64").length / MULAW_BYTES_PER_MS; + const len = payloadB64.length; + if (len === 0) return 0; + const pad = payloadB64.endsWith("==") ? 2 : payloadB64.endsWith("=") ? 1 : 0; + return (Math.floor((len * 3) / 4) - pad) / MULAW_BYTES_PER_MS; } + export interface BwStreamSource extends EventEmitter { sendMedia(payloadB64: string): void; flush(): void; @@ -38,6 +44,12 @@ export interface BridgeOpts { * "start" event as `streamParams`; build them with customParametersFromBwStart. */ customParameters?: Record; source: BwStreamSource; + /** Extra delay, in ms, added to every mark's due time. The playout clock + * starts when a frame reaches the bridge, but the caller hears it one-way + * network latency plus Bandwidth's jitter buffer later, so marks otherwise + * return slightly early in the same direction as the original bug. Default 0 + * until real measurements from VAPI-3991 give a value worth setting. */ + playoutLatencyPadMs?: number; } /** @@ -179,22 +191,26 @@ export class TwilioStreamBridge { } break; - case "mark": + case "mark": { // Per Twilio: the mark comes back when the audio queued before it has // finished playing. Nothing queued means it comes back at once. - this.pendingMarks.push({ mark: (msg as any).mark, dueAt: this.playoutEndAt }); + const pad = this.opts.playoutLatencyPadMs ?? 0; + const dueAt = this.playoutEndAt > Date.now() ? this.playoutEndAt + pad : this.playoutEndAt; + this.pendingMarks.push({ mark: (msg as any).mark, dueAt }); this.flushDueMarks(); break; + } - case "clear": + case "clear": { // Per Twilio: "empties all buffered audio and causes any mark messages // to be sent back". Discard the queue, then return every outstanding // mark immediately so a mark-gated bot is not left waiting forever. this.opts.source.flush(); - this.playoutEndAt = 0; - for (const p of this.pendingMarks) p.dueAt = 0; - this.flushDueMarks(); + const outstanding = this.pendingMarks; + this.dropPendingMarks(); + for (const { mark } of outstanding) this.sendMark(mark); break; + } } }); } @@ -222,13 +238,7 @@ export class TwilioStreamBridge { } const now = Date.now(); while (this.pendingMarks.length && this.pendingMarks[0].dueAt <= now) { - const { mark } = this.pendingMarks.shift()!; - this.send({ - event: "mark", - sequenceNumber: String(++this.seq), - streamSid: this.streamSid, - mark, - }); + this.sendMark(this.pendingMarks.shift()!.mark); } if (this.pendingMarks.length) { const wait = this.pendingMarks[0].dueAt - now; @@ -237,6 +247,10 @@ export class TwilioStreamBridge { } } + private sendMark(mark: unknown): void { + this.send({ event: "mark", sequenceNumber: String(++this.seq), streamSid: this.streamSid, mark }); + } + private dropPendingMarks(): void { if (this.markTimer) { clearTimeout(this.markTimer); diff --git a/test/streams-wire.test.ts b/test/streams-wire.test.ts index c8d8d03..4f0b821 100644 --- a/test/streams-wire.test.ts +++ b/test/streams-wire.test.ts @@ -358,6 +358,48 @@ describe("mark", () => { expect(mulawPayloadDurationMs("")).toBe(0); }); + it("mulawPayloadDurationMs matches a real decode for every base64 padding case", () => { + // Byte counts 0..9 exercise all three padding shapes (none, "=", "==") without + // allocating a Buffer on the hot path. + for (let bytes = 0; bytes <= 9; bytes++) { + const b64 = Buffer.alloc(bytes, 0x7f).toString("base64"); + expect(mulawPayloadDurationMs(b64)).toBe(Buffer.from(b64, "base64").length / 8); + } + }); + + it("playoutLatencyPadMs delays marks that have audio ahead of them, not idle ones", async () => { + const port = nextPort(); + const { messages, socket, close } = await botServer(port); + const source = new FakeBwSource(); + const bridge = new TwilioStreamBridge({ + botUrl: `ws://127.0.0.1:${port}`, + callSid: "CApad", + accountSid: "ACpad", + source, + playoutLatencyPadMs: 200, + }); + await bridge.ready(); + const bot = await socket; + const send = (obj: object) => bot.send(JSON.stringify({ streamSid: bridge.streamSid, ...obj })); + const marks = () => messages.filter((m: any) => m.event === "mark") as any[]; + + // Idle: no audio queued, so no pad either. + const idleAt = Date.now(); + send({ event: "mark", mark: { name: "idle" } }); + await waitFor(() => marks().length >= 1); + expect(Date.now() - idleAt).toBeLessThan(150); + + // 100 ms of audio + 200 ms pad: not before ~300 ms. + send({ event: "media", media: { payload: mulawMs(100) } }); + const sentAt = Date.now(); + send({ event: "mark", mark: { name: "padded" } }); + await waitFor(() => marks().length >= 2, 1000); + const ackAfterMs = Date.now() - sentAt; + bridge.close(); + close(); + expect(ackAfterMs).toBeGreaterThanOrEqual(280); + }); + it("echoes a mark at once when no audio is queued, preserving streamSid and mark.name", async () => { const t = await openBridge(); t.send({ event: "mark", mark: { name: "playback-done" } }); @@ -436,6 +478,19 @@ describe("mark", () => { expect(ackAfterMs).toBeLessThan(800); }); + it("close() drops pending marks and their timer", async () => { + const t = await openBridge(); + t.send({ event: "media", media: { payload: mulawMs(300) } }); + t.send({ event: "mark", mark: { name: "never" } }); + await waitFor(() => t.source.sent.length === 1); + expect(t.bridge.pendingPlayoutMs()).toBeGreaterThan(0); + + t.close(); + expect(t.bridge.pendingPlayoutMs()).toBe(0); + await new Promise((r) => setTimeout(r, 400)); + expect(t.marks()).toHaveLength(0); + }); + it("drops pending marks when the stream stops", async () => { const t = await openBridge(); t.send({ event: "media", media: { payload: mulawMs(5000) } }); @@ -490,6 +545,39 @@ describe("clear", () => { expect(bridge.pendingPlayoutMs()).toBe(0); }); + it("resets the playout clock, so audio queued after a clear is timed from scratch", async () => { + const port = nextPort(); + const { messages, socket, close } = await botServer(port); + const source = new FakeBwSource(); + const bridge = new TwilioStreamBridge({ + botUrl: `ws://127.0.0.1:${port}`, + callSid: "CAreset", + accountSid: "ACreset", + source, + }); + await bridge.ready(); + const bot = await socket; + const send = (obj: object) => bot.send(JSON.stringify({ streamSid: bridge.streamSid, ...obj })); + + // A stale second of audio, then the bot barges in and speaks 100 ms more. + send({ event: "media", media: { payload: mulawMs(1000) } }); + send({ event: "clear" }); + await waitFor(() => source.flushed === 1); + expect(bridge.pendingPlayoutMs()).toBe(0); + + send({ event: "media", media: { payload: mulawMs(100) } }); + const sentAt = Date.now(); + send({ event: "mark", mark: { name: "after-clear" } }); + await waitFor(() => messages.some((m: any) => m.event === "mark"), 1000); + const ackAfterMs = Date.now() - sentAt; + bridge.close(); + close(); + + // ~100 ms, not ~1100 ms: the cleared second must not count. + expect(ackAfterMs).toBeGreaterThanOrEqual(80); + expect(ackAfterMs).toBeLessThan(600); + }); + it("calls source.flush() when the bot sends a clear event", async () => { const port = nextPort(); const { socket, close } = await botServer(port);