From e88d3f30034ab6c80de88a263ec03375d3f0d280 Mon Sep 17 00:00:00 2001 From: Manas Srivastava Date: Wed, 19 Aug 2026 22:59:57 +0530 Subject: [PATCH 1/2] fix(interruption): bind inference responses to the overlap they were requested for MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `bargein_detected` / `inference_done` were matched to whichever overlap was open when they landed, not the one their request was cut for. A response arriving after its own overlap ended, while a later overlap was open, was accepted and attributed to that later overlap — emitting `isInterruption: true` for user audio the model never scored that way, and cutting the agent off. The cache clear at each overlap boundary was not a filter: `setOrUpdate()` recreates a missing entry from scratch, and `detectionDelayInS` was computed from the new overlap's start, so the result looked like a well-formed verdict. Plain response latency reaches this — the gap between two overlaps in one agent turn is often only a few hundred ms. Stamp the overlap generation onto each request in `sendAudioData()` and reject responses whose generation is no longer open. The generation ledger is kept outside `cache`, since `cache` is cleared at exactly the boundary where a late response still needs to be identifiable, and is bounded well above the request rate of a single overlap. The check fails open: a request whose generation is no longer on record counts as current, so losing the bookkeeping degrades to the previous behaviour rather than suppressing a genuine interruption. Fixes #2119 --- .../interruption_overlap_binding.test.ts | 187 ++++++++++++++++++ .../interruption/interruption_stream.ts | 9 + .../inference/interruption/ws_transport.ts | 27 +++ 3 files changed, 223 insertions(+) create mode 100644 agents/src/inference/interruption/interruption_overlap_binding.test.ts diff --git a/agents/src/inference/interruption/interruption_overlap_binding.test.ts b/agents/src/inference/interruption/interruption_overlap_binding.test.ts new file mode 100644 index 0000000000..ed951e3a6c --- /dev/null +++ b/agents/src/inference/interruption/interruption_overlap_binding.test.ts @@ -0,0 +1,187 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +// +// Regression tests for https://github.com/livekit/agents-js/issues/2119. +// +// A `bargein_detected` / `inference_done` response is matched to whichever overlap happens to be +// open when it lands, not to the overlap its request was cut for. A response that arrives after +// its own overlap ended, while a *later* overlap is open, used to be accepted and attributed to +// that later overlap — emitting `isInterruption: true` for user audio the model never scored. +// +// Plain response latency is enough to hit this: the gap between two overlaps inside one agent turn +// is often only a few hundred ms. +import { AudioFrame } from '@livekit/rtc-node'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { initializeLogger } from '../../log.js'; +import { MockWebSocket } from './_mock_ws.js'; +import { AdaptiveInterruptionDetector } from './interruption_detector.js'; +import { InterruptionStreamBase, InterruptionStreamSentinel } from './interruption_stream.js'; +import type { OverlappingSpeechEvent } from './types.js'; + +vi.mock('ws', async () => { + const { MockWebSocket } = await import('./_mock_ws.js'); + return { default: MockWebSocket, WebSocket: MockWebSocket }; +}); + +initializeLogger({ pretty: false, level: 'silent' }); + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +async function waitForInstance(timeoutMs = 2000): Promise { + const start = performance.now(); + while (MockWebSocket.instances.length === 0) { + if (performance.now() - start > timeoutMs) throw new Error('WebSocket was never constructed'); + await sleep(5); + } + return MockWebSocket.instances[MockWebSocket.instances.length - 1]!; +} + +function makeAudioFrame(numSamples = 1600, sampleRate = 16000): AudioFrame { + return new AudioFrame(new Int16Array(numSamples), sampleRate, 1, numSamples); +} + +function createDetector(): AdaptiveInterruptionDetector { + return new AdaptiveInterruptionDetector({ + baseUrl: 'http://localhost:9999', + apiKey: 'test-key', + apiSecret: 'test-secret', + }); +} + +/** Drain the event side into an array without blocking the test. */ +function collectEvents(stream: InterruptionStreamBase): OverlappingSpeechEvent[] { + const events: OverlappingSpeechEvent[] = []; + void (async () => { + const reader = stream.stream().getReader(); + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) return; + if (value) events.push(value); + } + } catch { + // stream errored or was torn down; the test asserts on what arrived before that + } + })(); + return events; +} + +/** The 8-byte little-endian `created_at` header the transport prefixes to each audio request. */ +function sentRequestIds(ws: MockWebSocket): number[] { + return ws.sent + .filter((f): f is Uint8Array => f instanceof Uint8Array) + .map((frame) => { + const view = new DataView(frame.buffer, frame.byteOffset, 8); + return view.getUint32(0, true) + view.getUint32(4, true) * 0x100000000; + }); +} + +/** Bring a stream up to an open socket with a server-provided threshold. */ +async function openStream(): Promise<{ stream: InterruptionStreamBase; ws: MockWebSocket }> { + const stream = new InterruptionStreamBase(createDetector(), {}); + const ws = await waitForInstance(); + ws.simulateOpen(); + await sleep(5); + ws.simulateMessage({ type: 'session.created', default_threshold: 0.5 }); + await sleep(5); + return { stream, ws }; +} + +beforeEach(() => { + MockWebSocket.instances.length = 0; +}); + +afterEach(() => { + vi.clearAllMocks(); +}); + +describe('interruption overlap binding (#2119)', () => { + it('does not attribute a late bargein from a closed overlap to a later overlap', async () => { + const { stream, ws } = await openStream(); + const events = collectEvents(stream); + + // Overlap A: send one inference request, then let the overlap close unanswered. + await stream.pushFrame(InterruptionStreamSentinel.agentSpeechStarted()); + await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechStarted(500, Date.now())); + await stream.pushFrame(makeAudioFrame()); + await sleep(20); + + const requestIds = sentRequestIds(ws); + expect(requestIds.length).toBeGreaterThan(0); + const overlapARequestId = requestIds[requestIds.length - 1]!; + + await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechEnded(Date.now())); + await sleep(20); + + // Overlap B opens a few hundred ms later, inside the same agent turn. + await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechStarted(500, Date.now())); + await sleep(20); + const beforeLateResponse = events.length; + + // Overlap A's response finally lands, while overlap B is open. + ws.simulateMessage({ + type: 'bargein_detected', + created_at: overlapARequestId, + probabilities: [0.99, 0.99, 0.99, 0.99], + prediction_duration: 0.05, + }); + await sleep(30); + + const late = events.slice(beforeLateResponse); + expect(late.filter((e) => e.isInterruption)).toEqual([]); + + await stream.close(); + }); + + it('still reports an interruption when the request generation is no longer on record', async () => { + // The generation ledger is bounded, so a very long overlap can evict a live request. Losing + // that bookkeeping must fail open — dropping the response instead would suppress a genuine + // interruption, which is worse than the misattribution this guard exists to prevent. + const { stream, ws } = await openStream(); + const events = collectEvents(stream); + + await stream.pushFrame(InterruptionStreamSentinel.agentSpeechStarted()); + await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechStarted(500, Date.now())); + await stream.pushFrame(makeAudioFrame()); + await sleep(20); + + // A `created_at` the transport never recorded, standing in for an evicted ledger entry. + ws.simulateMessage({ + type: 'bargein_detected', + created_at: Math.floor(performance.now()), + probabilities: [0.99, 0.99, 0.99, 0.99], + prediction_duration: 0.05, + }); + await sleep(30); + + expect(events.filter((e) => e.isInterruption).length).toBe(1); + + await stream.close(); + }); + + it('still reports an interruption for a response belonging to the open overlap', async () => { + const { stream, ws } = await openStream(); + const events = collectEvents(stream); + + await stream.pushFrame(InterruptionStreamSentinel.agentSpeechStarted()); + await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechStarted(500, Date.now())); + await stream.pushFrame(makeAudioFrame()); + await sleep(20); + + const requestIds = sentRequestIds(ws); + expect(requestIds.length).toBeGreaterThan(0); + + ws.simulateMessage({ + type: 'bargein_detected', + created_at: requestIds[requestIds.length - 1]!, + probabilities: [0.99, 0.99, 0.99, 0.99], + prediction_duration: 0.05, + }); + await sleep(30); + + expect(events.filter((e) => e.isInterruption).length).toBe(1); + + await stream.close(); + }); +}); diff --git a/agents/src/inference/interruption/interruption_stream.ts b/agents/src/inference/interruption/interruption_stream.ts index 8b086cebad..2c601f7caa 100644 --- a/agents/src/inference/interruption/interruption_stream.ts +++ b/agents/src/inference/interruption/interruption_stream.ts @@ -168,6 +168,12 @@ export class InterruptionStreamBase { let overlapSpeechStarted = false; let overlapCount = 0; const cache = new BoundedCache(10); + // Which overlap each in-flight request was cut for, keyed like `cache` by the request's + // `createdAt`. Deliberately kept outside `cache`: that one is cleared at every overlap + // boundary, which is exactly when a late response still needs to be identifiable. Bounded + // well above the ~10 requests/s one overlap produces, so a live request is not evicted + // mid-overlap. + const requestGenerations = new BoundedCache(256); const inferenceS16Data = new Int16Array( Math.ceil(this.options.maxAudioDurationInS * this.options.sampleRate), ).fill(0); @@ -178,6 +184,7 @@ export class InterruptionStreamBase { overlapSpeechStartedAt: this.overlapSpeechStartedAt, cache, overlapCount, + requestGenerations, }); const setState = (partial: { overlapSpeechStarted?: boolean }) => { if (partial.overlapSpeechStarted !== undefined) { @@ -245,6 +252,7 @@ export class InterruptionStreamBase { startIdx = 0; this.numRequests = 0; cache.clear(); + requestGenerations.clear(); } else if (chunk.type === 'agent-speech-ended') { this.logger.debug('agent speech ended'); agentSpeechStarted = false; @@ -255,6 +263,7 @@ export class InterruptionStreamBase { startIdx = 0; this.numRequests = 0; cache.clear(); + requestGenerations.clear(); } else if (chunk.type === 'overlap-speech-started' && agentSpeechStarted) { this.overlapSpeechStartedAt = chunk.startedAt; this.userSpeakingSpan = chunk.userSpeakingSpan; diff --git a/agents/src/inference/interruption/ws_transport.ts b/agents/src/inference/interruption/ws_transport.ts index a1b2526604..8e7e72a106 100644 --- a/agents/src/inference/interruption/ws_transport.ts +++ b/agents/src/inference/interruption/ws_transport.ts @@ -38,6 +38,10 @@ export interface WsTransportState { overlapSpeechStarted: boolean; overlapSpeechStartedAt: number | undefined; cache: BoundedCache; + /** Bumped on every `overlap-speech-started`; identifies the currently open overlap. */ + overlapCount: number; + /** Overlap generation each in-flight request was cut for, keyed by its `createdAt`. */ + requestGenerations: BoundedCache; } export const wsMessageSchema = z.discriminatedUnion('type', [ @@ -152,6 +156,18 @@ export interface WsTransportResult { close: () => void; } +/** + * Whether the response for `createdAt` belongs to an overlap that has since closed. + * + * Fails open: a request whose generation is no longer on record — evicted from the bounded + * ledger, or sent by an older transport — counts as current. Losing the bookkeeping therefore + * degrades to the previous behaviour rather than suppressing a genuine interruption. + */ +function isStaleResponse(state: WsTransportState, createdAt: number): boolean { + const recorded = state.requestGenerations.get(createdAt); + return recorded !== undefined && recorded.generation !== state.overlapCount; +} + /** * Creates a WebSocket transport TransformStream for interruption detection. * @@ -268,6 +284,10 @@ export function createWsTransport( case MSG_INTERRUPTION_DETECTED: { const createdAt = message.created_at; + if (isStaleResponse(state, createdAt)) { + logger.debug({ createdAt }, 'dropping bargein response from a closed overlap'); + break; + } const overlapSpeechStartedAt = state.overlapSpeechStartedAt; if (state.overlapSpeechStarted && overlapSpeechStartedAt !== undefined) { const existing = state.cache.get(createdAt); @@ -336,6 +356,12 @@ export function createWsTransport( case MSG_INFERENCE_DONE: { const createdAt = message.created_at; + if (isStaleResponse(state, createdAt)) { + // would otherwise poison the new overlap's cache entry, which + // `overlap-speech-ended` later pops to build its verdict + logger.debug({ createdAt }, 'dropping inference result from a closed overlap'); + break; + } const overlapSpeechStartedAt = state.overlapSpeechStartedAt; if (state.overlapSpeechStarted && overlapSpeechStartedAt !== undefined) { const existing = state.cache.get(createdAt); @@ -401,6 +427,7 @@ export function createWsTransport( speechInput: audioSlice, }), ); + state.requestGenerations.set(createdAt, { generation: state.overlapCount }); const header = new ArrayBuffer(8); const view = new DataView(header); From b078075509808bb7f5494e6631a0323e18c9aaac Mon Sep 17 00:00:00 2001 From: Manas Srivastava Date: Wed, 19 Aug 2026 23:10:34 +0530 Subject: [PATCH 2/2] fix(interruption): identify overlaps for the lifetime of the stream MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `overlapCount` restarts at 0 every agent turn, and the request ledger was cleared at that same boundary, so a response outliving its turn found no record, took the fail-open path, and was credited to whatever overlap was open in the next turn — the same misattribution this fixes within a turn. Track `overlapGeneration` separately and never reset or clear it. `overlapCount` keeps its per-turn meaning for the audio-prefix shift. Add the release-notes changeset. --- .changeset/quiet-moons-tap.md | 5 +++ .../interruption_overlap_binding.test.ts | 38 +++++++++++++++++++ .../interruption/interruption_stream.ts | 16 +++++--- .../inference/interruption/ws_transport.ts | 8 ++-- 4 files changed, 57 insertions(+), 10 deletions(-) create mode 100644 .changeset/quiet-moons-tap.md diff --git a/.changeset/quiet-moons-tap.md b/.changeset/quiet-moons-tap.md new file mode 100644 index 0000000000..70f36d00a2 --- /dev/null +++ b/.changeset/quiet-moons-tap.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Bind adaptive-interruption inference responses to the overlap they were requested for, so a late response can no longer be reported as an interruption of a later overlap. diff --git a/agents/src/inference/interruption/interruption_overlap_binding.test.ts b/agents/src/inference/interruption/interruption_overlap_binding.test.ts index ed951e3a6c..a4bfdc6c39 100644 --- a/agents/src/inference/interruption/interruption_overlap_binding.test.ts +++ b/agents/src/inference/interruption/interruption_overlap_binding.test.ts @@ -134,6 +134,44 @@ describe('interruption overlap binding (#2119)', () => { await stream.close(); }); + it('does not attribute a bargein from a previous agent turn to the current one', async () => { + // Overlap identity has to outlive the agent turn. A per-turn counter would restart at the + // same value next turn, and dropping the request's record at the boundary would let it + // through as "current" — either way the verdict lands on unrelated later speech. + const { stream, ws } = await openStream(); + const events = collectEvents(stream); + + await stream.pushFrame(InterruptionStreamSentinel.agentSpeechStarted()); + await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechStarted(500, Date.now())); + await stream.pushFrame(makeAudioFrame()); + await sleep(20); + + const requestIds = sentRequestIds(ws); + expect(requestIds.length).toBeGreaterThan(0); + const previousTurnRequestId = requestIds[requestIds.length - 1]!; + + await stream.pushFrame(InterruptionStreamSentinel.agentSpeechEnded()); + await sleep(20); + + // Next agent turn, first overlap — the same position the previous request was cut from. + await stream.pushFrame(InterruptionStreamSentinel.agentSpeechStarted()); + await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechStarted(500, Date.now())); + await sleep(20); + const beforeLateResponse = events.length; + + ws.simulateMessage({ + type: 'bargein_detected', + created_at: previousTurnRequestId, + probabilities: [0.99, 0.99, 0.99, 0.99], + prediction_duration: 0.05, + }); + await sleep(30); + + expect(events.slice(beforeLateResponse).filter((e) => e.isInterruption)).toEqual([]); + + await stream.close(); + }); + it('still reports an interruption when the request generation is no longer on record', async () => { // The generation ledger is bounded, so a very long overlap can evict a live request. Losing // that bookkeeping must fail open — dropping the response instead would suppress a genuine diff --git a/agents/src/inference/interruption/interruption_stream.ts b/agents/src/inference/interruption/interruption_stream.ts index 2c601f7caa..a6c9f62e45 100644 --- a/agents/src/inference/interruption/interruption_stream.ts +++ b/agents/src/inference/interruption/interruption_stream.ts @@ -167,12 +167,16 @@ export class InterruptionStreamBase { let accumulatedSamples = 0; let overlapSpeechStarted = false; let overlapCount = 0; + // Identifies an overlap for the lifetime of the stream. Distinct from `overlapCount`, which + // resets every agent turn and drives the audio-prefix shift: a response that outlives its + // turn must still be distinguishable, and per-turn counts collide across turns. + let overlapGeneration = 0; const cache = new BoundedCache(10); // Which overlap each in-flight request was cut for, keyed like `cache` by the request's - // `createdAt`. Deliberately kept outside `cache`: that one is cleared at every overlap - // boundary, which is exactly when a late response still needs to be identifiable. Bounded - // well above the ~10 requests/s one overlap produces, so a live request is not evicted - // mid-overlap. + // `createdAt`. Deliberately kept outside `cache`, and never cleared: `cache` is emptied at + // every overlap and agent-speech boundary, which is exactly when a late response still needs + // to be identifiable. Bounded well above the ~10 requests/s one overlap produces, so a live + // request is not evicted mid-overlap. const requestGenerations = new BoundedCache(256); const inferenceS16Data = new Int16Array( Math.ceil(this.options.maxAudioDurationInS * this.options.sampleRate), @@ -184,6 +188,7 @@ export class InterruptionStreamBase { overlapSpeechStartedAt: this.overlapSpeechStartedAt, cache, overlapCount, + overlapGeneration, requestGenerations, }); const setState = (partial: { overlapSpeechStarted?: boolean }) => { @@ -252,7 +257,6 @@ export class InterruptionStreamBase { startIdx = 0; this.numRequests = 0; cache.clear(); - requestGenerations.clear(); } else if (chunk.type === 'agent-speech-ended') { this.logger.debug('agent speech ended'); agentSpeechStarted = false; @@ -263,7 +267,6 @@ export class InterruptionStreamBase { startIdx = 0; this.numRequests = 0; cache.clear(); - requestGenerations.clear(); } else if (chunk.type === 'overlap-speech-started' && agentSpeechStarted) { this.overlapSpeechStartedAt = chunk.startedAt; this.userSpeakingSpan = chunk.userSpeakingSpan; @@ -271,6 +274,7 @@ export class InterruptionStreamBase { overlapSpeechStarted = true; accumulatedSamples = 0; overlapCount += 1; + overlapGeneration += 1; if (overlapCount <= 1) { const keepSize = Math.round((chunk.speechDuration / 1000) * this.options.sampleRate) + diff --git a/agents/src/inference/interruption/ws_transport.ts b/agents/src/inference/interruption/ws_transport.ts index 8e7e72a106..9855b85567 100644 --- a/agents/src/inference/interruption/ws_transport.ts +++ b/agents/src/inference/interruption/ws_transport.ts @@ -38,8 +38,8 @@ export interface WsTransportState { overlapSpeechStarted: boolean; overlapSpeechStartedAt: number | undefined; cache: BoundedCache; - /** Bumped on every `overlap-speech-started`; identifies the currently open overlap. */ - overlapCount: number; + /** Bumped on every `overlap-speech-started` and never reset; identifies the open overlap. */ + overlapGeneration: number; /** Overlap generation each in-flight request was cut for, keyed by its `createdAt`. */ requestGenerations: BoundedCache; } @@ -165,7 +165,7 @@ export interface WsTransportResult { */ function isStaleResponse(state: WsTransportState, createdAt: number): boolean { const recorded = state.requestGenerations.get(createdAt); - return recorded !== undefined && recorded.generation !== state.overlapCount; + return recorded !== undefined && recorded.generation !== state.overlapGeneration; } /** @@ -427,7 +427,7 @@ export function createWsTransport( speechInput: audioSlice, }), ); - state.requestGenerations.set(createdAt, { generation: state.overlapCount }); + state.requestGenerations.set(createdAt, { generation: state.overlapGeneration }); const header = new ArrayBuffer(8); const view = new DataView(header);