diff --git a/.changeset/adaptive-interruption-overlap-state.md b/.changeset/adaptive-interruption-overlap-state.md new file mode 100644 index 000000000..1fce58c3c --- /dev/null +++ b/.changeset/adaptive-interruption-overlap-state.md @@ -0,0 +1,22 @@ +--- +'@livekit/agents': patch +--- + +fix(interruption): keep adaptive interruption armed when agent speech restarts mid-overlap + +`overlapSpeechStarted` is the gate that lets a user's overlapping audio reach the interruption +model, and only a VAD start-of-speech raises it. Two events cleared it while the user was still +talking, and because VAD never re-announces speech that is already under way, nothing could re-arm +it for the rest of the agent turn: every remaining frame was dropped, no inference request was +made, and the agent talked straight through the interruption. + +- A transport failover rebuilt `InterruptionStreamBase` from scratch and replayed only + `agent-speech-started`, which marks the agent as speaking but leaves the overlap disarmed. The + in-progress overlap is now handed to the replacement stream. +- A second speech segment in the same turn (a queued `SpeechHandle`, or the reply after a tool + call) raises `agent-speech-started` again with no `agent-speech-ended` in between, and was + treated as a new turn. An open overlap is now preserved across it; a genuine new turn still + resets the overlap, audio buffer, cache and counters. + +Also stops a rejected forwarding task from surfacing as an unhandled rejection during the failover +backoff. diff --git a/agents/src/inference/interruption/interruption_pipeline.test.ts b/agents/src/inference/interruption/interruption_pipeline.test.ts index 5e7b41b89..79ee8f24f 100644 --- a/agents/src/inference/interruption/interruption_pipeline.test.ts +++ b/agents/src/inference/interruption/interruption_pipeline.test.ts @@ -91,6 +91,11 @@ interface Harness { close: () => Promise; } +/** Audio frames the transport pushed across every socket it has opened so far. */ +function totalAudioSendCount(): number { + return MockWebSocket.instances.reduce((total, ws) => total + audioSendCount(ws), 0); +} + interface OverlapEvent { isInterruption: boolean; numRequests: number; @@ -117,6 +122,9 @@ async function createHarness({ const events: OverlapEvent[] = []; detector.on('overlapping_speech', (ev) => events.push(ev)); + // A simulated transport failure is emitted as a recoverable detector error; without a listener + // the EventEmitter would rethrow it as an unhandled 'error'. + detector.on('error', () => {}); // Continuous room audio, pumped in the background for the whole test, exactly as a subscribed // track behaves — the interruption stream must pick it up on its own. @@ -153,32 +161,44 @@ async function createHarness({ await sleep(20); // Stand-in for the gateway: answer every audio frame promptly, as the real service does. Without - // this the transport's own 0.7s inference timeout tears the stream down. - let answered = 0; + // this the transport's own 0.7s inference timeout tears the stream down. Sockets opened later + // (an options reconnect, or the replacement stream built by a failover retry) are adopted too, + // so the gateway keeps behaving normally across a reconnect. + const answered = new Map([[ws, 0]]); let bargeinPending = false; const responder = (async () => { while (pumping) { - const binary = ws.sent.filter((s): s is Uint8Array => s instanceof Uint8Array); - while (answered < binary.length) { - const createdAt = createdAtOf(binary[answered]!); - answered++; - if (bargeinPending) { - bargeinPending = false; - ws.simulateMessage({ - type: 'bargein_detected', - created_at: createdAt, - probabilities: [0.91, 0.93, 0.95], - prediction_duration: 0.02, - }); - } else { - ws.simulateMessage({ - type: 'inference_done', - created_at: createdAt, - probabilities: [0.01, 0.02], - prediction_duration: 0.02, - is_bargein: false, - }); + for (const socket of MockWebSocket.instances) { + if (!answered.has(socket)) { + answered.set(socket, 0); + socket.simulateOpen(); + await sleep(1); + socket.simulateMessage({ type: 'session.created', default_threshold: 0.5 }); + } + const binary = socket.sent.filter((s): s is Uint8Array => s instanceof Uint8Array); + let seen = answered.get(socket)!; + while (seen < binary.length) { + const createdAt = createdAtOf(binary[seen]!); + seen++; + if (bargeinPending) { + bargeinPending = false; + socket.simulateMessage({ + type: 'bargein_detected', + created_at: createdAt, + probabilities: [0.91, 0.93, 0.95], + prediction_duration: 0.02, + }); + } else { + socket.simulateMessage({ + type: 'inference_done', + created_at: createdAt, + probabilities: [0.01, 0.02], + prediction_duration: 0.02, + is_bargein: false, + }); + } } + answered.set(socket, seen); } await sleep(5); } @@ -190,7 +210,7 @@ async function createHarness({ hooks, ws, events, - requestCount: () => answered, + requestCount: () => [...answered.values()].reduce((a, b) => a + b, 0), bargeinOnNextRequest: () => { bargeinPending = true; }, @@ -269,6 +289,151 @@ describe('adaptive interruption pipeline', () => { }); }); +// --------------------------------------------------------------------------- +// Overlap state must survive events that are not a new agent turn (regression) +// --------------------------------------------------------------------------- + +/** + * `overlapSpeechStarted` is the gate that lets user audio reach the gateway at all, and the only + * thing that ever raises it is an `overlap-speech-started` sentinel, which in turn only comes from + * a VAD start-of-speech. VAD does not re-announce speech that is already under way, so once the + * flag is cleared mid-overlap nothing can re-arm it: every remaining frame of the interruption is + * dropped, no inference request is made, and the agent talks straight through the user. + * + * Two things clear the flag without the user's turn having ended. + */ +describe('adaptive interruption overlap state retention', () => { + beforeEach(() => { + MockWebSocket.instances.length = 0; + }); + + /** + * A transient socket failure fails the stream over. JS rebuilds `InterruptionStreamBase` from + * scratch on retry — all of its state lives in a `setupTransform()` closure — whereas Python + * keeps `_agent_speech_started` / `_overlap_started` on `self` and only reconnects the socket. + * Replaying `agent-speech-started` alone restores half the state: the agent is known to be + * speaking again, but the in-flight overlap is gone. + */ + it('keeps sending user audio after the transport fails over mid-overlap', async () => { + const h = await createHarness(); + + await h.recognition.onStartOfAgentSpeech(Date.now()); + await sleep(200); + await h.recognition.onStartOfOverlapSpeech(200, Date.now()); + await sleep(400); + + expect(totalAudioSendCount()).toBeGreaterThan(0); + const socketsBefore = MockWebSocket.instances.length; + + // Transient socket failure while the user is mid-interrupt. + h.ws.emit('error', new Error('connection reset')); + + // The failover sleeps intervalForRetry(0) (2s + jitter) before rebuilding the stream. + await waitFor(() => MockWebSocket.instances.length > socketsBefore, 8000); + await sleep(200); + + // The user is still talking; from here on every frame must reach the new socket. + const sendsAfterFailover = totalAudioSendCount(); + await sleep(800); + + expect(totalAudioSendCount() - sendsAfterFailover).toBeGreaterThan(0); + + await h.recognition.onEndOfOverlapSpeech(Date.now()); + await waitFor(() => h.events.length > 0); + expect(h.events[h.events.length - 1]!.numRequests).toBeGreaterThan(0); + + await h.close(); + }, 30_000); + + /** + * One user-perceived agent turn can contain several speech segments — a tool call sandwiched + * between two replies, or a queued `say()`. `AgentActivity.onPipelineReplyDone` only reports + * `onEndOfAgentSpeech` once the speech queue has drained, so the second segment raises + * `agent-speech-started` again with no `agent-speech-ended` in between. Treating that as a new + * turn resets the overlap the user is in the middle of. + */ + it('keeps sending user audio when a second speech segment starts mid-overlap', async () => { + const h = await createHarness(); + + await h.recognition.onStartOfAgentSpeech(Date.now()); + await sleep(200); + await h.recognition.onStartOfOverlapSpeech(200, Date.now()); + await sleep(400); + + const before = totalAudioSendCount(); + expect(before).toBeGreaterThan(0); + + // Second segment of the same turn: no `onEndOfAgentSpeech` precedes it. + await h.recognition.onStartOfAgentSpeech(Date.now()); + await sleep(600); + + expect(totalAudioSendCount() - before).toBeGreaterThan(0); + + await h.recognition.onEndOfOverlapSpeech(Date.now()); + await waitFor(() => h.events.length > 0); + expect(h.events[h.events.length - 1]!.numRequests).toBeGreaterThan(0); + + await h.close(); + }, 30_000); + + /** + * The bargein verdict must still surface after a mid-overlap segment change — restoring the gate + * is only useful if the recovered audio can still produce `isInterruption: true`. + */ + it('still reports a bargein after a second speech segment starts mid-overlap', async () => { + const h = await createHarness(); + + await h.recognition.onStartOfAgentSpeech(Date.now()); + await sleep(200); + await h.recognition.onStartOfOverlapSpeech(200, Date.now()); + await sleep(400); + + await h.recognition.onStartOfAgentSpeech(Date.now()); + await sleep(200); + h.bargeinOnNextRequest(); + await waitFor(() => h.events.length > 0, 5000); + + expect(h.events[0]!.isInterruption).toBe(true); + expect(h.events[0]!.numRequests).toBeGreaterThan(0); + await waitFor(() => vi.mocked(h.hooks.onInterruption).mock.calls.length > 0); + + await h.close(); + }, 30_000); + + /** A genuine new turn must still wipe the overlap, the cache and the counters. */ + it('resets overlap state on a new agent turn', async () => { + const h = await createHarness(); + + await h.recognition.onStartOfAgentSpeech(Date.now()); + await sleep(200); + await h.recognition.onStartOfOverlapSpeech(200, Date.now()); + await sleep(400); + expect(totalAudioSendCount()).toBeGreaterThan(0); + + // The agent turn ends, then a new one begins. The user is no longer overlapping anything, + // so their audio must not be forwarded until a fresh overlap is announced. + await h.recognition.onEndOfAgentSpeech(Date.now()); + await sleep(20); + await h.recognition.onStartOfAgentSpeech(Date.now()); + + const afterNewTurn = totalAudioSendCount(); + await sleep(600); + expect(totalAudioSendCount()).toBe(afterNewTurn); + + // ...and the new turn's first overlap starts from a clean counter. + await h.recognition.onStartOfOverlapSpeech(200, Date.now()); + await sleep(400); + await h.recognition.onEndOfOverlapSpeech(Date.now()); + await waitFor(() => h.events.length > 0); + + const last = h.events[h.events.length - 1]!; + expect(last.numRequests).toBeGreaterThan(0); + expect(totalAudioSendCount()).toBeGreaterThan(afterNewTurn); + + await h.close(); + }, 30_000); +}); + // --------------------------------------------------------------------------- // Send-time overlap gate (regression) // --------------------------------------------------------------------------- diff --git a/agents/src/inference/interruption/interruption_stream.ts b/agents/src/inference/interruption/interruption_stream.ts index ffb4b2ca6..f6a4e01b3 100644 --- a/agents/src/inference/interruption/interruption_stream.ts +++ b/agents/src/inference/interruption/interruption_stream.ts @@ -14,6 +14,7 @@ import { InterruptionCacheEntry } from './interruption_cache_entry.js'; import type { AdaptiveInterruptionDetector } from './interruption_detector.js'; import { type AgentSpeechEnded, + type AgentSpeechResumed, type AgentSpeechStarted, type ApiConnectOptions, type Flush, @@ -30,6 +31,7 @@ import { createWsTransport } from './ws_transport.js'; // Re-export sentinel types for backwards compatibility export type { AgentSpeechEnded, + AgentSpeechResumed, AgentSpeechStarted, ApiConnectOptions, Flush, @@ -38,6 +40,12 @@ export type { OverlapSpeechStarted, }; +/** Snapshot of an overlap that is still being classified. */ +export interface ActiveOverlap { + startedAt: number; + userSpeakingSpan?: Span; +} + export class InterruptionStreamSentinel { static agentSpeechStarted(): AgentSpeechStarted { return { type: 'agent-speech-started' }; @@ -47,6 +55,14 @@ export class InterruptionStreamSentinel { return { type: 'agent-speech-ended' }; } + static agentSpeechResumed(overlap?: ActiveOverlap): AgentSpeechResumed { + return { + type: 'agent-speech-resumed', + overlapStartedAt: overlap?.startedAt, + userSpeakingSpan: overlap?.userSpeakingSpan, + }; + } + static overlapSpeechStarted( speechDuration: number, startedAt: number, @@ -101,6 +117,9 @@ export class InterruptionStreamBase { private wsClose?: () => void; + // The overlap flag lives in the setupTransform() closure; this exposes it for `activeOverlap`. + private readOverlapSpeechStarted?: () => boolean; + // Mutable transport options that can be updated via updateOptions() private transportOptions: { baseUrl: string; @@ -140,6 +159,21 @@ export class InterruptionStreamBase { this.eventStream = this.setupTransform(); } + /** + * The overlap this stream is currently classifying, if any. + * + * Unlike Python — where the equivalent flags are attributes on the stream instance and a + * reconnect leaves them alone — every retry here builds a new stream and drops the state held in + * `setupTransform()`. Callers that recreate the stream use this to hand the in-progress overlap + * to its replacement via {@link InterruptionStreamSentinel.agentSpeechResumed}. + */ + get activeOverlap(): ActiveOverlap | undefined { + if (!this.readOverlapSpeechStarted?.() || this.overlapSpeechStartedAt === undefined) { + return undefined; + } + return { startedAt: this.overlapSpeechStartedAt, userSpeakingSpan: this.userSpeakingSpan }; + } + /** * Update stream options. For WebSocket transport, this triggers a reconnection. */ @@ -188,6 +222,7 @@ export class InterruptionStreamBase { overlapSpeechStarted = partial.overlapSpeechStarted; } }; + this.readOverlapSpeechStarted = () => overlapSpeechStarted; const handleSpanUpdate = (entry: InterruptionCacheEntry) => { if (this.userSpeakingSpan) { updateUserSpeakingSpan(this.userSpeakingSpan, entry); @@ -248,15 +283,41 @@ export class InterruptionStreamBase { }); } } else if (chunk.type === 'agent-speech-started') { - this.logger.debug('agent speech started'); + // One agent turn can span several speech segments — a queued SpeechHandle, or the + // reply that follows a tool call — and `AgentActivity.onPipelineReplyDone` only + // reports `agent-speech-ended` once the speech queue has drained. The later segments + // therefore arrive here with no end in between. Resetting on those would strand an + // overlap the user is still in the middle of: `overlapSpeechStarted` is the gate that + // lets their audio reach the gateway at all, and only a VAD start-of-speech can raise + // it again — which never comes for speech that is already under way. + if (agentSpeechStarted && overlapSpeechStarted) { + this.logger.debug('agent speech continued into a new segment, keeping open overlap'); + } else { + this.logger.debug('agent speech started'); + agentSpeechStarted = true; + overlapSpeechStarted = false; + this.overlapSpeechStartedAt = undefined; + accumulatedSamples = 0; + overlapCount = 0; + startIdx = 0; + this.numRequests = 0; + cache.clear(); + } + } else if (chunk.type === 'agent-speech-resumed') { + // This stream replaces one the transport failover tore down. Adopt what the previous + // stream knew instead of treating it as a new turn; everything else is already at its + // freshly-constructed value. + this.logger.debug( + { overlapStartedAt: chunk.overlapStartedAt }, + 'resuming agent speech on a replacement interruption stream', + ); agentSpeechStarted = true; - overlapSpeechStarted = false; - this.overlapSpeechStartedAt = undefined; - accumulatedSamples = 0; - overlapCount = 0; - startIdx = 0; - this.numRequests = 0; - cache.clear(); + if (chunk.overlapStartedAt !== undefined) { + overlapSpeechStarted = true; + overlapCount = 1; + this.overlapSpeechStartedAt = chunk.overlapStartedAt; + this.userSpeakingSpan = chunk.userSpeakingSpan; + } } else if (chunk.type === 'agent-speech-ended') { this.logger.debug('agent speech ended'); agentSpeechStarted = false; diff --git a/agents/src/inference/interruption/types.ts b/agents/src/inference/interruption/types.ts index 32bfc7e97..3335de89e 100644 --- a/agents/src/inference/interruption/types.ts +++ b/agents/src/inference/interruption/types.ts @@ -72,6 +72,18 @@ export interface AgentSpeechEnded { type: 'agent-speech-ended'; } +/** + * Restores a replacement stream after the transport failed over, which builds a brand new + * `InterruptionStreamBase` with none of the previous one's state. Kept distinct from + * {@link AgentSpeechStarted} so that sentinel keeps meaning "a new agent turn began, reset". + */ +export interface AgentSpeechResumed { + type: 'agent-speech-resumed'; + /** Absolute timestamp (ms) the overlap started at, when one was still open at failover. */ + overlapStartedAt?: number; + userSpeakingSpan?: Span; +} + export interface OverlapSpeechStarted { type: 'overlap-speech-started'; /** Duration of the speech segment in milliseconds (matches VADEvent.speechDuration units). */ @@ -99,6 +111,7 @@ export interface Flush { export type InterruptionSentinel = | AgentSpeechStarted | AgentSpeechEnded + | AgentSpeechResumed | OverlapSpeechStarted | OverlapSpeechEnded | Flush; diff --git a/agents/src/voice/audio_recognition.ts b/agents/src/voice/audio_recognition.ts index d11ed4680..8f6fc157c 100644 --- a/agents/src/voice/audio_recognition.ts +++ b/agents/src/voice/audio_recognition.ts @@ -23,7 +23,10 @@ import { import { apiConnectDefaults, intervalForRetry } from '../inference/interruption/defaults.js'; import { InterruptionDetectionError } from '../inference/interruption/errors.js'; import type { AdaptiveInterruptionDetector } from '../inference/interruption/interruption_detector.js'; -import { InterruptionStreamSentinel } from '../inference/interruption/interruption_stream.js'; +import { + type ActiveOverlap, + InterruptionStreamSentinel, +} from '../inference/interruption/interruption_stream.js'; import { type InterruptionSentinel, type OverlappingSpeechEvent, @@ -1960,6 +1963,8 @@ export class AudioRecognition { let numRetries = 0; const maxRetries = apiConnectDefaults.maxRetries; + // Overlap the torn-down stream was still classifying, handed to its replacement below. + let resumeOverlap: ActiveOverlap | undefined; while (!signal.aborted) { const stream = interruptionDetection.createStream(); @@ -1980,11 +1985,14 @@ export class AudioRecognition { let forwardTask: Promise | undefined; try { - // Unlike Python where _agent_speech_started lives on `self` and survives retries, - // JS creates a fresh InterruptionStreamBase per retry with agentSpeechStarted = false. - // Re-inject the sentinel so the new stream knows the agent is mid-speech. + // Unlike Python where _agent_speech_started and _overlap_started live on `self` and are + // untouched by a reconnect, JS creates a fresh InterruptionStreamBase per retry and loses + // everything the previous one held. Hand it the whole picture, not just "the agent is + // speaking": an overlap that was open at failover can never be re-armed otherwise, since + // only a VAD start-of-speech raises the flag and VAD does not re-announce speech that is + // already under way. The rest of that turn's user audio would be dropped on the floor. if (numRetries > 0 && this.isAgentSpeaking) { - await stream.pushFrame(InterruptionStreamSentinel.agentSpeechStarted()); + await stream.pushFrame(InterruptionStreamSentinel.agentSpeechResumed(resumeOverlap)); } forwardTask = (async () => { @@ -2005,6 +2013,10 @@ export class AudioRecognition { inputReader.releaseLock(); } })(); + // The stream this task pushes into is torn down as soon as the read loop below fails, so + // it rejects long before the `finally` attaches a handler — which only runs after the + // retry backoff. Mark it handled now so the rejection is not reported as unhandled. + forwardTask.catch(() => {}); const abortPromise = waitForAbort(signal); @@ -2069,6 +2081,8 @@ export class AudioRecognition { break; } } finally { + // Read before cleanup() closes the stream, so a retry can restore the open overlap. + resumeOverlap = stream.activeOverlap; await cleanup(); await forwardTask?.catch((e) => { this.logger.debug({ err: e }, 'interruption task exited with error');