From 35dc9e0d93eaba0f71a3f4b593484b4e0f84b1ce Mon Sep 17 00:00:00 2001 From: Toubat Date: Mon, 27 Jul 2026 19:25:58 -0700 Subject: [PATCH] fix(room_io): scope the audio pause gate to the segment being captured MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit clearBuffer() resolved an interruptedFuture that was only reset by the *next* segment's flush(), so for the whole of the following reply the signal still described an interruption that was already over. Any frame parking at the pause gate in that window was discarded — frames of a reply that had nothing to do with that interruption. Snapshot the interrupt count when a segment opens and bail only for interruptions raised at or after that point. Consult the snapshot on every frame, not only on frames that parked at a closed gate. cancelSpeechPause un-gates the sink as soon as the handle is interrupted, but the interrupted reply's forwardAudio loop runs for another event loop turn with seconds of TTS audio still to drain. Those frames found the gate open and took the unchecked path to the wire. Co-authored-by: Cursor --- .changeset/stale-interruption-pause-gate.md | 27 ++ .../false_interruption_audio_loss.test.ts | 258 ++++++++++++++++++ .../src/voice/recorder_io/recorder_io.test.ts | 4 + agents/src/voice/room_io/_output.test.ts | 69 ++++- agents/src/voice/room_io/_output.ts | 47 +++- .../_output_interrupted_segment.test.ts | 104 +++++++ 6 files changed, 504 insertions(+), 5 deletions(-) create mode 100644 .changeset/stale-interruption-pause-gate.md create mode 100644 agents/src/voice/false_interruption_audio_loss.test.ts create mode 100644 agents/src/voice/room_io/_output_interrupted_segment.test.ts diff --git a/.changeset/stale-interruption-pause-gate.md b/.changeset/stale-interruption-pause-gate.md new file mode 100644 index 000000000..93ea1b07b --- /dev/null +++ b/.changeset/stale-interruption-pause-gate.md @@ -0,0 +1,27 @@ +--- +'@livekit/agents': patch +--- + +Scope the audio pause gate to the segment being captured + +`ParticipantAudioOutput.clearBuffer()` resolves an `interruptedFuture` that frames parked at +the pause gate consult to decide whether to bail. Nothing reset that signal until the _next_ +segment's `flush()`, which only runs after that segment's frames have all been captured — so +for the whole of the following reply the signal still described an interruption that was +already over. If the output was paused mid-reply during that window (an ordinary +false-interruption pause, on by default), every remaining frame bailed at the gate and never +reached the wire. The session still reported the reply as fully spoken and committed it to the +chat context, so the loss was silent: in a reproduction with a real `AgentSession` and a real +`ParticipantAudioOutput`, 18 of 20 frames — 360ms of a 400ms reply — were dropped. + +The gate is now scoped to the segment being captured: a frame bails only for an interruption +raised at or after its own segment began. Frames parked at the gate are also woken by a +per-frame signal, so a `flush()` that replaces `interruptedFuture` can no longer strand one +there. + +The same snapshot is now consulted on every frame rather than only on frames that parked at a +closed gate. `cancelSpeechPause` un-gates the sink to admit the next reply as soon as the +handle is interrupted, but the interrupted reply's `forwardAudio` loop keeps running until its +abort signal fires an event loop turn later, and real TTS providers hand it several seconds of +audio ahead of realtime to drain in the meantime. Those frames found the gate already open and +took the unchecked path to the wire. diff --git a/agents/src/voice/false_interruption_audio_loss.test.ts b/agents/src/voice/false_interruption_audio_loss.test.ts new file mode 100644 index 000000000..239fbf45f --- /dev/null +++ b/agents/src/voice/false_interruption_audio_loss.test.ts @@ -0,0 +1,258 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { AudioFrame, type Room, TrackPublishOptions, TrackSource } from '@livekit/rtc-node'; +import { ReadableStream } from 'node:stream/web'; +import { describe, expect, it } from 'vitest'; +import { initializeLogger } from '../log.js'; +import { VADEventType } from '../vad.js'; +import { Agent } from './agent.js'; +import { AgentSession } from './agent_session.js'; +import { AgentSessionEventTypes } from './events.js'; +import { ParticipantAudioOutput } from './room_io/_output.js'; +import { FakeLLM } from './testing/fake_llm.js'; + +const SAMPLE_RATE = 24000; +const FRAME_MS = 20; +const FRAMES_PER_REPLY = 20; +const FALSE_INTERRUPTION_TIMEOUT = 400; + +function frame(): AudioFrame { + const samples = (SAMPLE_RATE * FRAME_MS) / 1000; + return new AudioFrame(new Int16Array(samples), SAMPLE_RATE, 1, samples); +} + +function vadEvent(type: VADEventType, speechDuration = 0, silenceDuration = 0) { + return { + type, + samplesIndex: 0, + timestamp: Date.now(), + speechDuration, + silenceDuration, + frames: [], + probability: 1, + inferenceDuration: 0, + speaking: type === VADEventType.START_OF_SPEECH, + rawAccumulatedSilence: 0, + rawAccumulatedSpeech: 0, + } as never; +} + +function sttFinal(text: string) { + return { + type: 'final_transcript', + alternatives: [{ text, language: 'en', startTime: 0, endTime: 0, confidence: 1 }], + } as never; +} + +/** The real ParticipantAudioOutput; only track publishing is skipped (no LiveKit server here). */ +class TestParticipantAudioOutput extends ParticipantAudioOutput { + constructor() { + super({} as Room, { + sampleRate: SAMPLE_RATE, + numChannels: 1, + trackPublishOptions: new TrackPublishOptions({ source: TrackSource.SOURCE_MICROPHONE }), + queueSizeMs: 100_000, + }); + (this as unknown as { startedFuture: { resolve: () => void } }).startedFuture.resolve(); + } + + override async start(): Promise {} +} + +/** Emits `FRAMES_PER_REPLY` frames spaced out in time so a turn can be interleaved with VAD events. */ +class FrameAgent extends Agent { + produced = 0; + onFrame?: (produced: number) => void; + + constructor() { + super({ instructions: 'test' }); + } + + async ttsNode(): Promise> { + let emitted = 0; + const emit = () => { + this.produced++; + this.onFrame?.(this.produced); + }; + return new ReadableStream({ + pull: async (controller) => { + if (emitted >= FRAMES_PER_REPLY) { + controller.close(); + return; + } + if (emitted > 0) await new Promise((resolve) => setTimeout(resolve, 15)); + emitted++; + controller.enqueue(frame()); + emit(); + }, + }); + } +} + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +async function makeHarness() { + const session = new AgentSession({ + llm: new FakeLLM([ + { input: 'one', content: 'first reply' }, + { input: 'two', content: 'second reply' }, + ]), + aecWarmupDuration: null, + turnHandling: { interruption: { falseInterruptionTimeout: FALSE_INTERRUPTION_TIMEOUT } }, + }); + + const out = new TestParticipantAudioOutput(); + session.output.audio = out; + + // Frame accounting at the boundary the bug lives on: how many frames actually reached + // the LiveKit AudioSource, i.e. the wire. + const source = ( + out as unknown as { audioSource: { captureFrame: (f: AudioFrame) => Promise } } + ).audioSource; + const captureFrame = source.captureFrame.bind(source); + let delivered = 0; + source.captureFrame = async (f: AudioFrame) => { + delivered++; + return captureFrame(f); + }; + + const falseInterruptions: boolean[] = []; + session.on(AgentSessionEventTypes.AgentFalseInterruption, (ev) => + falseInterruptions.push(ev.resumed), + ); + + const agent = new FrameAgent(); + await session.start({ agent }); + + const gate = out as unknown as { playbackEnabledFuture: { done: boolean } }; + + const waitForProduced = (n: number) => + new Promise((resolve) => { + agent.onFrame = (produced) => { + if (produced >= n) resolve(); + }; + }); + + return { + session, + out, + agent, + falseInterruptions, + delivered: () => delivered, + paused: () => !gate.playbackEnabledFuture.done, + waitForProduced, + activity: () => session._activity!, + async close() { + await session.close(); + await out.close(); + }, + }; +} + +type Harness = Awaited>; + +/** + * One reply that a backchannel pauses mid-flight and the false-interruption timer resumes. + * Every frame the TTS produced must reach the wire: the ones captured while the output was + * paused are held at the gate and pushed on resume. + */ +async function falseInterruptedReply(h: Harness, userInput: string) { + const producedBefore = h.agent.produced; + const deliveredBefore = h.delivered(); + + const handle = h.session.generateReply({ userInput }); + await h.waitForProduced(producedBefore + 3); + + h.activity().onStartOfSpeech(vadEvent(VADEventType.START_OF_SPEECH)); + h.activity().onVADInferenceDone(vadEvent(VADEventType.INFERENCE_DONE, 600)); + const pausedMidReply = h.paused(); + // The user only said "mm-hmm": no final transcript follows, so the false-interruption + // timer is what resumes playback. + h.activity().onEndOfSpeech(vadEvent(VADEventType.END_OF_SPEECH, 0, 100)); + + await handle.waitForPlayout(); + await sleep(FALSE_INTERRUPTION_TIMEOUT + 300); + + return { + pausedMidReply, + produced: h.agent.produced - producedBefore, + delivered: h.delivered() - deliveredBefore, + }; +} + +describe('false interruption after a previous interruption', () => { + initializeLogger({ pretty: false, level: 'silent' }); + + it('delivers the whole reply when nothing was interrupted before', async () => { + const h = await makeHarness(); + try { + const reply = await falseInterruptedReply(h, 'one'); + + expect(reply.pausedMidReply).toBe(true); + expect(reply.produced).toBe(FRAMES_PER_REPLY); + expect(reply.delivered).toBe(reply.produced); + expect(h.falseInterruptions).toEqual([true]); + } finally { + await h.close(); + } + }, 30000); + + it('delivers the whole reply when the previous turn ended in a barge-in', async () => { + const h = await makeHarness(); + try { + // A genuine barge-in: VAD pauses the output, the final transcript confirms the + // interruption, and the reply task calls clearBuffer() on the way out. That leaves + // interruptedFuture resolved, and nothing resets it until the *next* segment's flush — + // which only happens after that segment's frames have all been captured. + const bargedInto = h.session.generateReply({ userInput: 'one' }); + await h.waitForProduced(3); + h.activity().onStartOfSpeech(vadEvent(VADEventType.START_OF_SPEECH)); + h.activity().onVADInferenceDone(vadEvent(VADEventType.INFERENCE_DONE, 600)); + h.activity().onFinalTranscript(sttFinal('stop please'), false); + await bargedInto.waitForPlayout(); + await sleep(150); + + // An ordinary false interruption on the next reply. Every frame captured during the + // pause used to bail at the gate on the stale signal, losing the rest of the reply + // while the session still recorded it as fully spoken. + const reply = await falseInterruptedReply(h, 'two'); + + expect(reply.pausedMidReply).toBe(true); + expect(reply.produced).toBe(FRAMES_PER_REPLY); + expect(reply.delivered).toBe(reply.produced); + expect(h.falseInterruptions).toEqual([true]); + } finally { + await h.close(); + } + }, 30000); + + it('delivers the whole reply when a paused reply completed and was then cancelled', async () => { + const h = await makeHarness(); + try { + // The agent finishes its sentence while the user talks over the tail: all frames are + // captured, then the output is paused and the segment drains to a clean finish. + const completed = h.session.generateReply({ userInput: 'one' }); + await h.waitForProduced(FRAMES_PER_REPLY); + for (let i = 0; i < 200 && h.delivered() < FRAMES_PER_REPLY; i++) await sleep(5); + await sleep(10); + h.activity().onStartOfSpeech(vadEvent(VADEventType.START_OF_SPEECH)); + h.activity().onVADInferenceDone(vadEvent(VADEventType.INFERENCE_DONE, 600)); + await completed.waitForPlayout(); + await sleep(100); + expect(h.delivered()).toBe(FRAMES_PER_REPLY); + + // The user's transcript finalizes and cancelSpeechPause un-gates the output. + h.activity().onFinalTranscript(sttFinal('okay thanks'), false); + await sleep(150); + + const reply = await falseInterruptedReply(h, 'two'); + + expect(reply.pausedMidReply).toBe(true); + expect(reply.produced).toBe(FRAMES_PER_REPLY); + expect(reply.delivered).toBe(reply.produced); + } finally { + await h.close(); + } + }, 30000); +}); diff --git a/agents/src/voice/recorder_io/recorder_io.test.ts b/agents/src/voice/recorder_io/recorder_io.test.ts index c6f61aa42..df409ebf3 100644 --- a/agents/src/voice/recorder_io/recorder_io.test.ts +++ b/agents/src/voice/recorder_io/recorder_io.test.ts @@ -1032,6 +1032,10 @@ describe('RecorderAudioOutput in front of a real ParticipantAudioOutput', () => expect(await settleOrStall(output.waitForPlayout(), 2000)).not.toBe('did not settle'); output.resume(); + // The flush `forwardAudio` runs in its `finally` is what ends the interrupted segment. Without + // it the sink cannot tell a follow-on turn apart from the interrupted turn's TTS backlog, which + // keeps arriving after the resume that admits the next reply. + output.flush(); await output.captureFrame(makeFrame(100, 24000)); output.flush(); diff --git a/agents/src/voice/room_io/_output.test.ts b/agents/src/voice/room_io/_output.test.ts index 49a74c119..95d6e07ce 100644 --- a/agents/src/voice/room_io/_output.test.ts +++ b/agents/src/voice/room_io/_output.test.ts @@ -164,13 +164,20 @@ describe('ParticipantAudioOutput captureFrame segment accounting', () => { firstFrameEmitted: boolean; pushedDuration: number; _capturing: boolean; + interruptCount: number; + segmentInterruptCount: number; + segmentOpen: boolean; + gatedFrames: Set>; playbackSegmentsCount: number; playbackFinishedCount: number; playbackFinishedFuture: Future; onPlaybackStarted: (createdAt: number) => void; + logger: { error: () => void }; audioSource: { clearQueue: () => void; captureFrame: (frame: CaptureFrameArg) => Promise; + waitForPlayout: () => Promise; + queuedDuration: number; }; }; @@ -184,21 +191,39 @@ describe('ParticipantAudioOutput captureFrame segment accounting', () => { output.firstFrameEmitted = false; output.pushedDuration = 0; output._capturing = false; + output.interruptCount = 0; + output.segmentInterruptCount = 0; + output.segmentOpen = false; + output.gatedFrames = new Set(); output.playbackSegmentsCount = 0; output.playbackFinishedCount = 0; output.playbackFinishedFuture = new Future(); output.onPlaybackStarted = vi.fn(); - output.audioSource = { clearQueue: vi.fn(), captureFrame: vi.fn(async () => {}) }; + output.logger = { error: vi.fn() }; + output.audioSource = { + clearQueue: vi.fn(), + captureFrame: vi.fn(async () => {}), + // Playout never drains on its own, so a flush task stays pending like a real + // segment still on the wire. + waitForPlayout: () => new Promise(() => {}), + queuedDuration: 0, + }; return output; }; const frame = () => ({ samplesPerChannel: 480, sampleRate: 24000 }) as unknown as CaptureFrameArg; + const settledWithin = async (promise: Promise, ms: number) => + Promise.race([ + promise.then(() => 'settled' as const), + new Promise<'pending'>((resolve) => setTimeout(() => resolve('pending'), ms)), + ]); + it('does not strand the segment counter when a frame is interrupted while paused', async () => { const output = makeOutput({ paused: true }); const capture = output.captureFrame(frame()); - output.interruptedFuture.resolve(); + output.clearBuffer(); await capture; expect(output.playbackSegmentsCount).toBe(0); @@ -211,6 +236,46 @@ describe('ParticipantAudioOutput captureFrame segment accounting', () => { expect(result).toBe('resolved'); }); + it('does not drop a new segment at the pause gate after the previous one was interrupted', async () => { + const output = makeOutput({ paused: false }); + + // Segment 1 plays, is flushed and then interrupted — the state a barge-in leaves behind. + await output.captureFrame(frame()); + output.flush(); + output.clearBuffer(); + await nextTick(); + + // Segment 2 is a fresh reply that gets paused mid-flight by a new overlap. Its frames + // belong to a speech the earlier interruption knows nothing about, so they must wait + // for the resume rather than be discarded. + output.pause(); + const capture = output.captureFrame(frame()); + expect(await settledWithin(capture, 50)).toBe('pending'); + + output.resume(); + await capture; + + expect(output.audioSource.captureFrame).toHaveBeenCalledTimes(2); + expect(output.playbackSegmentsCount).toBe(2); + }); + + it('releases a parked frame when a concurrent flush has replaced the interruption signal', async () => { + const output = makeOutput({ paused: false }); + + await output.captureFrame(frame()); + output.pause(); + const capture = output.captureFrame(frame()); + await nextTick(); + + // An overlapping segment's flush swaps interruptedFuture out from under the parked + // frame; the interruption that follows must still reach it. + output.flush(); + output.clearBuffer(); + + expect(await settledWithin(capture, 500)).toBe('settled'); + expect(output.audioSource.captureFrame).toHaveBeenCalledTimes(1); + }); + it('registers a segment on the normal non-paused path', async () => { const output = makeOutput({ paused: false }); diff --git a/agents/src/voice/room_io/_output.ts b/agents/src/voice/room_io/_output.ts index a78caa9fe..0c4d02c6d 100644 --- a/agents/src/voice/room_io/_output.ts +++ b/agents/src/voice/room_io/_output.ts @@ -384,6 +384,14 @@ export class ParticipantAudioOutput extends AudioOutput { private firstFrameEmitted: boolean = false; /** Gate held closed while the output is paused; frame forwarding awaits it. */ private playbackEnabledFuture: Future = new Future(); + /** Monotonic count of interruptions signalled through clearBuffer(). */ + private interruptCount: number = 0; + /** interruptCount as of the start of the segment currently being captured. */ + private segmentInterruptCount: number = 0; + /** Whether a frame has been captured since the last flush boundary. */ + private segmentOpen: boolean = false; + /** Wake-up futures for frames parked at the pause gate, all resolved by clearBuffer(). */ + private gatedFrames: Set> = new Set(); constructor(room: Room, options: AudioOutputOptions) { super(options.sampleRate, undefined, { pause: true }); @@ -423,17 +431,43 @@ export class ParticipantAudioOutput extends AudioOutput { } async captureFrame(frame: AudioFrame): Promise { + if (!this.segmentOpen) { + this.segmentOpen = true; + // An interruption raised before this segment began belongs to a speech that is already + // over. interruptedFuture stays resolved until the next flush, so without this snapshot + // every frame of the new speech bails at the gate below and the reply is lost. + this.segmentInterruptCount = this.interruptCount; + } + await this.startedFuture.await; if (!this.playbackEnabledFuture.done) { this.audioSource.clearQueue(); // Race against interruption so a cancel-while-paused can't deadlock an in-flight frame. - await Promise.race([this.playbackEnabledFuture.await, this.interruptedFuture.await]); - if (this.interruptedFuture.done) { - return; + // The wake-up is per frame rather than the shared interruptedFuture, which + // waitForPlayoutTask may replace while this frame is parked. + if (this.interruptCount === this.segmentInterruptCount) { + const gate = new Future(); + this.gatedFrames.add(gate); + try { + await Promise.race([this.playbackEnabledFuture.await, gate.await]); + } finally { + this.gatedFrames.delete(gate); + } } } + // Tested on every frame, not only the ones that parked at the gate. `cancelSpeechPause` + // un-gates the sink to admit the next reply as soon as the handle is interrupted, but the + // interrupted reply's `forwardAudio` loop only stops an event loop turn later, when its + // abort signal fires — and real TTS hands it seconds of audio ahead of realtime to drain in + // the meantime. Those frames find the gate already open, so without this they reach the wire + // while the next reply's transcript is streaming. `forwardAudio` always flushes in its + // `finally`, which is what opens a fresh segment for the next reply. + if (this.interruptCount > this.segmentInterruptCount) { + return; + } + // Count the playback segment only after the pause/interrupt gate above. super.captureFrame // bumps playbackSegmentsCount; if a frame interrupted-while-paused bailed at the gate after // that bump, the count would strand ahead of playbackFinishedCount and the next @@ -495,6 +529,7 @@ export class ParticipantAudioOutput extends AudioOutput { */ flush(): void { super.flush(); + this.segmentOpen = false; if (!this.pushedDuration) { return; @@ -522,10 +557,16 @@ export class ParticipantAudioOutput extends AudioOutput { } clearBuffer(): void { + this.interruptCount++; // Signal interruption even if no frame has been pushed yet, so a gated captureFrame can bail. if (!this.interruptedFuture.done) { this.interruptedFuture.resolve(); } + for (const gate of this.gatedFrames) { + if (!gate.done) { + gate.resolve(); + } + } } private async publishTrack(signal: AbortSignal) { diff --git a/agents/src/voice/room_io/_output_interrupted_segment.test.ts b/agents/src/voice/room_io/_output_interrupted_segment.test.ts new file mode 100644 index 000000000..c18ff40f0 --- /dev/null +++ b/agents/src/voice/room_io/_output_interrupted_segment.test.ts @@ -0,0 +1,104 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { AudioFrame, type Room, TrackPublishOptions, TrackSource } from '@livekit/rtc-node'; +import { describe, expect, it } from 'vitest'; +import { ParticipantAudioOutput } from './_output.js'; + +const SAMPLE_RATE = 24000; +const FRAME_MS = 20; + +/** `marker` identifies which reply produced the frame once it reaches the wire. */ +function frame(marker: number): AudioFrame { + const samples = (SAMPLE_RATE * FRAME_MS) / 1000; + const data = new Int16Array(samples); + data[0] = marker; + return new AudioFrame(data, SAMPLE_RATE, 1, samples); +} + +/** The real output; only track publishing is skipped (no LiveKit server in a unit test). */ +class TestOutput extends ParticipantAudioOutput { + constructor() { + super({} as Room, { + sampleRate: SAMPLE_RATE, + numChannels: 1, + trackPublishOptions: new TrackPublishOptions({ source: TrackSource.SOURCE_MICROPHONE }), + queueSizeMs: 100_000, + }); + (this as unknown as { startedFuture: { resolve: () => void } }).startedFuture.resolve(); + } + + override async start(): Promise {} +} + +/** Records the marker of every frame that reaches the wire. */ +function wireMarkers(out: ParticipantAudioOutput): number[] { + const source = ( + out as unknown as { audioSource: { captureFrame: (f: AudioFrame) => Promise } } + ).audioSource; + const original = source.captureFrame.bind(source); + const markers: number[] = []; + source.captureFrame = async (f: AudioFrame) => { + markers.push(f.data[0]!); + return original(f); + }; + return markers; +} + +describe('a committed barge-in with TTS audio still in flight', () => { + /** + * Real TTS delivers several seconds of audio ahead of realtime, so at the moment of a barge-in + * the interrupted reply's `forwardAudio` loop is still holding a backlog of already-synthesized + * frames. It stays alive for the whole of the reply task's + * `cancelAndWait(forwardTasks, REPLY_TASK_CANCEL_TIMEOUT)`, and only flushes in its `finally` + * after that — well after `cancelSpeechPause` has un-gated the sink for the next reply. + * + * `captureFrame` only consults `interruptCount` while the pause gate is closed, so every one of + * those backlog frames takes the open-gate path and reaches the wire while the next reply's + * transcript is already streaming. That is the audio/transcript desync. + */ + it('does not put the interrupted reply’s backlog on the wire', async () => { + const out = new TestOutput(); + const markers = wireMarkers(out); + + // Reply A is playing. + await out.captureFrame(frame(1)); + + // The user barges in: VAD pauses the sink, then the commit clears and un-gates it. + out.pause(); + out.clearBuffer(); + out.resume(); + + // Reply A's forwarding loop has not noticed the abort yet and drains its backlog. + await out.captureFrame(frame(1)); + await out.captureFrame(frame(1)); + + expect(markers.filter((m) => m === 1)).toHaveLength(1); + + await out.close(); + }); + + /** Control: once reply A's forwarding loop unwinds and flushes, reply B must be audible. */ + it('plays the next reply after the interrupted one flushes', async () => { + const out = new TestOutput(); + const markers = wireMarkers(out); + + await out.captureFrame(frame(1)); + + out.pause(); + out.clearBuffer(); + out.resume(); + + await out.captureFrame(frame(1)); + + // `forwardAudio` flushes in a `finally`, closing the interrupted segment. + out.flush(); + + await out.captureFrame(frame(2)); + await out.captureFrame(frame(2)); + + expect(markers.filter((m) => m === 2)).toHaveLength(2); + + await out.close(); + }); +});