Skip to content

fix(interruption): keep the overlap armed when agent speech restarts mid-interrupt #2117

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
toubatbrian wants to merge 1 commit into from
Closed

fix(interruption): keep the overlap armed when agent speech restarts mid-interrupt #2117

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions .changeset/adaptive-interruption-overlap-state.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.
211 changes: 188 additions & 23 deletions agents/src/inference/interruption/interruption_pipeline.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,11 @@ interface Harness {
close: () => Promise<void>;
}

/** 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;
Expand All @@ -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.
Expand Down Expand Up @@ -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<MockWebSocket, number>([[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);
}
Expand All @@ -190,7 +210,7 @@ async function createHarness({
hooks,
ws,
events,
requestCount: () => answered,
requestCount: () => [...answered.values()].reduce((a, b) => a + b, 0),
bargeinOnNextRequest: () => {
bargeinPending = true;
},
Expand Down Expand Up @@ -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)
// ---------------------------------------------------------------------------
Expand Down
Loading
Loading