From bf6b671fab57900f4b9f7e92b85c967431094fa5 Mon Sep 17 00:00:00 2001 From: Toubat Date: Sun, 26 Jul 2026 13:23:04 -0700 Subject: [PATCH 1/2] test(interruption): cover the adaptive-interruption pipeline end to end Adds the first coverage of the audio-to-verdict path as AudioRecognition wires it: room audio -> interruption stream channel -> audio transformer -> WS transport, with a mock gateway standing in for the service. Regression coverage for "every overlap is classified as backchannel": the classifier can only return `isInterruption: true` from the inference-response path, which requires audio to actually reach the transport while the overlap is open. Also exports `OverlappingSpeechEvent` by name from `voice/events.ts` so `overlapping_speech` handlers can be typed without reaching into `inference/interruption/types.js`. Co-authored-by: Cursor --- .../interruption_pipeline.test.ts | 266 ++++++++++++++++++ agents/src/voice/events.ts | 6 + 2 files changed, 272 insertions(+) create mode 100644 agents/src/inference/interruption/interruption_pipeline.test.ts diff --git a/agents/src/inference/interruption/interruption_pipeline.test.ts b/agents/src/inference/interruption/interruption_pipeline.test.ts new file mode 100644 index 000000000..4005d6cbd --- /dev/null +++ b/agents/src/inference/interruption/interruption_pipeline.test.ts @@ -0,0 +1,266 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +// +// End-to-end (transport-mocked) test of the adaptive-interruption pipeline as it is wired by +// AudioRecognition: room audio -> interruption stream channel -> audio transformer -> WS transport. +// +// Regression coverage for "every overlap is classified as backchannel": the classifier can only +// return `isInterruption: true` from the inference-response path, which requires audio to actually +// reach the transport while the overlap is open. +import { AudioFrame } from '@livekit/rtc-node'; +import { ReadableStream } from 'node:stream/web'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { ChatContext } from '../../llm/chat_context.js'; +import { initializeLogger } from '../../log.js'; +import { AudioRecognition, type RecognitionHooks } from '../../voice/audio_recognition.js'; +import { createEndpointing } from '../../voice/turn_config/endpointing.js'; +import { MockWebSocket } from './_mock_ws.js'; +import { AdaptiveInterruptionDetector } from './interruption_detector.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 waitFor(predicate: () => boolean, timeoutMs = 3000): Promise { + const start = performance.now(); + while (!predicate()) { + if (performance.now() - start > timeoutMs) { + throw new Error('condition not met within timeout'); + } + await sleep(5); + } +} + +function createHooks(): RecognitionHooks { + return { + onInterruption: vi.fn(), + onBackchannelConfirmed: vi.fn(), + onStartOfSpeech: vi.fn(), + onVADInferenceDone: vi.fn(), + onEndOfSpeech: vi.fn(), + onInterimTranscript: vi.fn(), + onFinalTranscript: vi.fn(), + onPreemptiveGeneration: vi.fn(), + onAgentBackchannelOpportunity: vi.fn(), + retrieveChatCtx: () => ChatContext.empty(), + onEndOfTurn: vi.fn(async () => true), + } as unknown as RecognitionHooks; +} + +/** 10ms of non-silent mono PCM, as a room track would deliver it. */ +function makeFrame(sampleRate: number): AudioFrame { + const samples = Math.floor(sampleRate / 100); + const data = new Int16Array(samples); + for (let i = 0; i < samples; i++) { + data[i] = Math.round(8000 * Math.sin((2 * Math.PI * 220 * i) / sampleRate)); + } + return new AudioFrame(data, sampleRate, 1, samples); +} + +/** Number of binary (audio) frames the transport pushed onto the socket. */ +function audioSendCount(ws: MockWebSocket): number { + return ws.sent.filter((s) => s instanceof Uint8Array).length; +} + +/** `created_at` header the transport stamped on a binary frame. */ +function createdAtOf(buf: Uint8Array): number { + const view = new DataView(buf.buffer, buf.byteOffset, buf.byteLength); + return view.getUint32(4, true) * 0x100000000 + view.getUint32(0, true); +} + +interface Harness { + recognition: AudioRecognition; + detector: AdaptiveInterruptionDetector; + hooks: RecognitionHooks; + ws: MockWebSocket; + events: OverlapEvent[]; + /** Requests the mock gateway has answered so far. */ + requestCount: () => number; + /** Answer the next request with a bargein verdict instead of a plain inference_done. */ + bargeinOnNextRequest: () => void; + close: () => Promise; +} + +interface OverlapEvent { + isInterruption: boolean; + numRequests: number; + probability: number; +} + +async function createHarness({ + sampleRate = 48000, +}: { sampleRate?: number } = {}): Promise { + const detector = new AdaptiveInterruptionDetector({ + baseUrl: 'http://localhost:9999', + apiKey: 'test-key', + apiSecret: 'test-secret', + }); + const hooks = createHooks(); + const recognition = new AudioRecognition({ + recognitionHooks: hooks, + interruptionDetection: detector, + endpointing: createEndpointing({ mode: 'dynamic', minDelay: 500, maxDelay: 3000, alpha: 0.9 }), + }); + // The pipeline wiring keys off `interruptionDetection`; the enabled flag additionally requires a + // VAD, which this test substitutes for by driving the overlap callbacks directly. + (recognition as unknown as { isInterruptionEnabled: boolean }).isInterruptionEnabled = true; + + const events: OverlapEvent[] = []; + detector.on('overlapping_speech', (ev) => events.push(ev)); + + // 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. + let pumping = true; + let pushFrame!: (frame: AudioFrame) => void; + const audioStream = new ReadableStream({ + start(controller) { + pushFrame = (frame) => controller.enqueue(frame); + }, + }); + recognition.setInputAudioStream(audioStream); + const pump = (async () => { + while (pumping) { + pushFrame(makeFrame(sampleRate)); + await sleep(10); + } + })(); + + const ac = new AbortController(); + const task = ( + recognition as unknown as { + createInterruptionTask: ( + d: AdaptiveInterruptionDetector, + signal: AbortSignal, + ) => Promise; + } + ).createInterruptionTask(detector, ac.signal); + + await waitFor(() => MockWebSocket.instances.length > 0); + const ws = MockWebSocket.instances[MockWebSocket.instances.length - 1]!; + ws.simulateOpen(); + await waitFor(() => ws.sent.length > 0); // session.create + ws.simulateMessage({ type: 'session.created', default_threshold: 0.5 }); + 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; + 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, + }); + } + } + await sleep(5); + } + })(); + + return { + recognition, + detector, + hooks, + ws, + events, + requestCount: () => answered, + bargeinOnNextRequest: () => { + bargeinPending = true; + }, + close: async () => { + pumping = false; + await pump; + await responder; + ac.abort(); + await task.catch(() => {}); + await recognition.close().catch(() => {}); + }, + }; +} + +describe('adaptive interruption pipeline', () => { + beforeEach(() => { + MockWebSocket.instances.length = 0; + }); + + afterEach(() => { + vi.clearAllMocks(); + }); + + it('sends inference requests for user audio that overlaps agent speech', async () => { + const h = await createHarness(); + + await h.recognition.onStartOfAgentSpeech(Date.now()); + await sleep(200); + await h.recognition.onStartOfOverlapSpeech(200, Date.now()); + await sleep(600); + + expect(audioSendCount(h.ws)).toBeGreaterThan(0); + + await h.close(); + }); + + it('classifies a server bargein verdict as an interruption', async () => { + const h = await createHarness(); + + await h.recognition.onStartOfAgentSpeech(Date.now()); + await sleep(200); + h.bargeinOnNextRequest(); + await h.recognition.onStartOfOverlapSpeech(200, Date.now()); + await sleep(600); + + expect(audioSendCount(h.ws)).toBeGreaterThan(0); + await waitFor(() => h.events.length > 0); + + expect(h.events[0]!.isInterruption).toBe(true); + await waitFor(() => vi.mocked(h.hooks.onInterruption).mock.calls.length > 0); + + await h.close(); + }); + + it('keeps making requests across repeated short overlaps in one agent turn', async () => { + const h = await createHarness(); + + await h.recognition.onStartOfAgentSpeech(Date.now()); + await sleep(300); + + const perOverlapRequests: number[] = []; + for (let i = 0; i < 3; i++) { + const before = audioSendCount(h.ws); + await h.recognition.onStartOfOverlapSpeech(200, Date.now()); + await sleep(1200); // the user's overlaps were 1-2s each + await h.recognition.onEndOfOverlapSpeech(Date.now()); + await sleep(50); + perOverlapRequests.push(audioSendCount(h.ws) - before); + } + + expect(h.events).toHaveLength(3); + expect(perOverlapRequests.every((n) => n > 0)).toBe(true); + expect(h.events.map((e) => e.numRequests).every((n) => n > 0)).toBe(true); + + await h.close(); + }); +}); diff --git a/agents/src/voice/events.ts b/agents/src/voice/events.ts index 3a3050f4b..45d19489b 100644 --- a/agents/src/voice/events.ts +++ b/agents/src/voice/events.ts @@ -436,6 +436,12 @@ export const createAgentFalseInterruptionEvent = ({ createdAt, }); +/** + * Payload of {@link AgentSessionEventTypes.OverlappingSpeech}. Re-exported here so handlers can be + * typed by name; it carries `probability`, `probabilities` and `numRequests` alongside the verdict. + */ +export type { OverlappingSpeechEvent }; + export type AgentEvent = | UserInputTranscribedEvent | UserStateChangedEvent From f05882005c2b38ab6d5549c6d43c2dceed4429b5 Mon Sep 17 00:00:00 2001 From: Toubat Date: Sat, 25 Jul 2026 17:10:30 -0700 Subject: [PATCH 2/2] fix(interruption): keep the overlap armed when agent speech restarts mid-interrupt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `overlapSpeechStarted` is the gate that lets a user's overlapping audio reach the interruption model, and the only thing that raises it is a VAD start-of-speech. Two events cleared it while the user was still talking. Because VAD never re-announces speech 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 rebuilds InterruptionStreamBase from scratch — all of its state lives in a setupTransform() closure, unlike Python where the equivalent flags are instance attributes a reconnect leaves alone. Replaying `agent-speech-started` restored only half of it. Hand the in-progress overlap to the replacement stream instead, via a distinct `agent-speech-resumed` sentinel so the real one keeps meaning "new turn, reset everything". A second speech segment in one turn (a queued SpeechHandle, or the reply after a tool call) raises `agent-speech-started` again with no `agent-speech-ended` in between, because onPipelineReplyDone only reports the end once the speech queue drains. Preserve an open overlap across that; a genuine new turn still resets the overlap, audio buffer, cache and counters. Also marks the forwarding task's rejection handled at creation: it rejects as soon as the stream is torn down, but the `finally` only attaches a handler after the retry backoff, so the rejection surfaced as an unhandled rejection. Co-authored-by: Cursor --- .../adaptive-interruption-overlap-state.md | 22 ++ .../interruption_pipeline.test.ts | 211 ++++++++++++++++-- .../interruption/interruption_stream.ts | 77 ++++++- agents/src/inference/interruption/types.ts | 13 ++ agents/src/voice/audio_recognition.ts | 24 +- 5 files changed, 311 insertions(+), 36 deletions(-) create mode 100644 .changeset/adaptive-interruption-overlap-state.md 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 4005d6cbd..65a7fb64d 100644 --- a/agents/src/inference/interruption/interruption_pipeline.test.ts +++ b/agents/src/inference/interruption/interruption_pipeline.test.ts @@ -87,6 +87,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; @@ -113,6 +118,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. @@ -149,32 +157,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); } @@ -186,7 +206,7 @@ async function createHarness({ hooks, ws, events, - requestCount: () => answered, + requestCount: () => [...answered.values()].reduce((a, b) => a + b, 0), bargeinOnNextRequest: () => { bargeinPending = true; }, @@ -264,3 +284,148 @@ describe('adaptive interruption pipeline', () => { await h.close(); }); }); + +// --------------------------------------------------------------------------- +// 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); +}); diff --git a/agents/src/inference/interruption/interruption_stream.ts b/agents/src/inference/interruption/interruption_stream.ts index efb02819f..ee41c3ef9 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, @@ -29,6 +30,7 @@ import { createWsTransport } from './ws_transport.js'; // Re-export sentinel types for backwards compatibility export type { AgentSpeechEnded, + AgentSpeechResumed, AgentSpeechStarted, ApiConnectOptions, Flush, @@ -37,6 +39,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' }; @@ -46,6 +54,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, @@ -100,6 +116,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; @@ -139,6 +158,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. */ @@ -184,6 +218,7 @@ export class InterruptionStreamBase { overlapSpeechStarted = partial.overlapSpeechStarted; } }; + this.readOverlapSpeechStarted = () => overlapSpeechStarted; const handleSpanUpdate = (entry: InterruptionCacheEntry) => { if (this.userSpeakingSpan) { updateUserSpeakingSpan(this.userSpeakingSpan, entry); @@ -236,15 +271,41 @@ export class InterruptionStreamBase { controller.enqueue(audioSlice); } } 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 d062eb739..38ec96a23 100644 --- a/agents/src/inference/interruption/types.ts +++ b/agents/src/inference/interruption/types.ts @@ -59,6 +59,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). */ @@ -86,6 +98,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');