Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
5 changes: 5 additions & 0 deletions .changeset/adaptive-interruption-tool-calls.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
'@livekit/agents': patch
---

Preserve adaptive interruption boundaries when agent playout pauses or enters a tool-call thinking gap.
44 changes: 44 additions & 0 deletions agents/src/inference/interruption/interruption_stream.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
//
// SPDX-License-Identifier: Apache-2.0
import { describe, expect, it, vi } from 'vitest';
import type { InterruptionMetrics } from '../../metrics/base.js';
import { MockWebSocket } from './_mock_ws.js';
import { AdaptiveInterruptionDetector } from './interruption_detector.js';
import { InterruptionStreamBase, InterruptionStreamSentinel } from './interruption_stream.js';

vi.mock('ws', async () => {
const { MockWebSocket } = await import('./_mock_ws.js');
return { default: MockWebSocket, WebSocket: MockWebSocket };
});

describe('InterruptionStreamBase metrics', () => {
it('does not count agent-ended overlap as a backchannel', async () => {
MockWebSocket.instances.length = 0;
const detector = new AdaptiveInterruptionDetector({
apiKey: 'test-key',
apiSecret: 'test-secret',
baseUrl: 'http://localhost:9999',
});
const stream = new InterruptionStreamBase(detector, {});
const metrics: InterruptionMetrics[] = [];
detector.on('metrics_collected', (event) => metrics.push(event));
const reader = stream.stream().getReader();

try {
await vi.waitFor(() => expect(MockWebSocket.instances).toHaveLength(1));
MockWebSocket.instances[0]!.simulateOpen();
const event = reader.read();
await stream.pushFrame(InterruptionStreamSentinel.agentSpeechStarted());
await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechStarted(0, Date.now()));
await stream.pushFrame(InterruptionStreamSentinel.overlapSpeechEnded(Date.now(), true));
await event;

expect(metrics).toHaveLength(1);
expect(metrics[0]).toMatchObject({ numInterruptions: 0, numBackchannels: 0 });
} finally {
await reader.cancel();
await stream.close();
}
});
});
2 changes: 1 addition & 1 deletion agents/src/inference/interruption/interruption_stream.ts
Original file line number Diff line number Diff line change
Expand Up @@ -337,7 +337,7 @@ export class InterruptionStreamBase {
predictionDuration: chunk.predictionDurationInS * 1000,
detectionDelay: chunk.detectionDelayInS * 1000,
numInterruptions: chunk.isInterruption ? 1 : 0,
numBackchannels: chunk.isInterruption ? 0 : 1,
numBackchannels: !chunk.isInterruption && !chunk.agentEnded ? 1 : 0,
numRequests: chunk.numRequests,
metadata: {
modelProvider: this.model.provider,
Expand Down
4 changes: 2 additions & 2 deletions agents/src/voice/agent_activity.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -452,7 +452,7 @@ describe('AgentActivity - mainTask', () => {
}
});

it('does not deadlock cancelling a paused speech whose generation never finishes', async () => {
it('does not end paused agent speech twice when cancelling it', async () => {
const handle = SpeechHandle.create({ allowInterruptions: true });
handle._authorizeGeneration();

Expand Down Expand Up @@ -501,7 +501,7 @@ describe('AgentActivity - mainTask', () => {
expect(result).toBe('resolved');
expect(handle.interrupted).toBe(true);
expect(fakeActivity.pausedSpeech).toBeUndefined();
expect(fakeActivity.audioRecognition.onEndOfAgentSpeech).toHaveBeenCalledOnce();
expect(fakeActivity.audioRecognition.onEndOfAgentSpeech).not.toHaveBeenCalled();
});
});

Expand Down
29 changes: 14 additions & 15 deletions agents/src/voice/agent_activity.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1695,7 +1695,6 @@ export class AgentActivity implements RecognitionHooks {
if (this.audioRecognition) {
this.audioRecognition.onEndOfAgentSpeech(
options?.ignoreUserTranscriptUntil ?? Date.now(),
{ paused: true },
);
}
Comment on lines 1696 to 1699

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Pausing the agent's voice now reports a bogus "user talked over the agent" event every time

An overlap is opened and then immediately closed as agent-ended (onStartOfOverlapSpeech at agents/src/voice/agent_activity.ts:1684-1689 followed by onEndOfAgentSpeech at agents/src/voice/agent_activity.ts:1696) whenever agent playout is paused, so listeners receive an extra overlapping-speech notification and a zero-valued measurement for a pause that produced no real judgement.
Impact: Applications and dashboards subscribed to overlapping-speech events see spurious entries (and zeroed interruption metrics) each time the agent's voice is paused.

Mechanism: overlap opened right before the boundary that closes it

Before this PR the pause path passed { paused: true }, so no agent-speech-ended sentinel was written and the overlap opened at agents/src/voice/agent_activity.ts:1684 stayed open across the pause (an event was produced only on a real verdict or user-speech end).

Now onEndOfAgentSpeech (agents/src/voice/audio_recognition.ts:784-831) always closes any open overlap via closeOverlap(now, undefined, true) and then sends agent-speech-ended. Since the pause path opens the overlap synchronously one statement earlier, the queued sentinel pair is overlap-speech-started + overlap-speech-ended(agentEnded: true), and the transform at agents/src/inference/interruption/interruption_stream.ts:274-300 enqueues an OverlappingSpeechEvent built from InterruptionCacheEntry.default() (all zeros, isInterruption: false). That event is emitted publicly (this.model.emit('overlapping_speech', chunk) at agents/src/inference/interruption/interruption_stream.ts:331) and forwarded to the session, plus an interruption_metrics event with numInterruptions: 0 / numBackchannels: 0.

If the intent is that pausing is a plain speech boundary, the overlap open at agents/src/voice/agent_activity.ts:1683-1689 is now redundant and should be dropped (the overlap is re-opened by onStartOfAgentSpeech when playout resumes while the user is still speaking, agents/src/voice/audio_recognition.ts:770-775).

Prompt for agents
In AgentActivity.interruptByAudioActivity (agents/src/voice/agent_activity.ts, pause branch), an overlap inference is started via audioRecognition.onStartOfOverlapSpeech(...) immediately before audioRecognition.onEndOfAgentSpeech(...) is called for the same pause. With the new semantics in AudioRecognition.onEndOfAgentSpeech (which now always closes an open overlap with agentEnded=true and then resets the detector), this pair produces an immediate, meaningless overlapping_speech event with all-zero timings plus a zeroed interruption_metrics event on every pause. Decide whether the overlap should still be opened at pause time at all: since AudioRecognition.onStartOfAgentSpeech now re-opens the overlap when playout resumes while the user is still speaking, the pause-time open appears redundant. Verify against the upstream Python implementation and remove the redundant open (or suppress emitting an event for an overlap that is closed with no inference requests).
Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

if (this.isInterruptionDetectionEnabled) {
Expand All @@ -1719,9 +1718,7 @@ export class AgentActivity implements RecognitionHooks {
ignoreUserTranscriptUntil: ev.overlapStartedAt || ev.detectedAt,
});
if (this.audioRecognition) {
this.audioRecognition.onEndOfAgentSpeech(ev.overlapStartedAt || ev.detectedAt, {
paused: this.pausedSpeech !== undefined,
});
this.audioRecognition.onEndOfAgentSpeech(ev.overlapStartedAt || ev.detectedAt);
}
}

Expand Down Expand Up @@ -3485,6 +3482,12 @@ export class AgentActivity implements RecognitionHooks {

if (!speechHandle.interrupted && toolOutput.output.length > 0) {
this.agentSession._updateAgentState('thinking');
if (this.audioRecognition) {
this.audioRecognition.onEndOfAgentSpeech(Date.now());
}
if (this.isInterruptionDetectionEnabled) {
this.restoreInterruptionByAudioActivity();
}
} else if (this.agentSession.agentState === 'speaking') {
this.agentSession._updateAgentState('listening');
if (this.audioRecognition) {
Expand Down Expand Up @@ -4069,6 +4072,12 @@ export class AgentActivity implements RecognitionHooks {

if (toolOutput.output.length > 0) {
this.agentSession._updateAgentState('thinking');
if (this.audioRecognition) {
this.audioRecognition.onEndOfAgentSpeech(Date.now());
}
if (this.isInterruptionDetectionEnabled) {
this.restoreInterruptionByAudioActivity();
}
} else if (this.agentSession.agentState === 'speaking') {
this.agentSession._updateAgentState('listening');
if (this.audioRecognition) {
Expand Down Expand Up @@ -4850,7 +4859,7 @@ export class AgentActivity implements RecognitionHooks {
otelContext: this.pausedSpeech.handle._agentTurnContext,
});
if (this.audioRecognition && this.pausedSpeech.agentState === 'speaking') {
this.audioRecognition.onStartOfAgentSpeech(Date.now(), { resumed: true });
this.audioRecognition.onStartOfAgentSpeech(Date.now());
}
if (this.isInterruptionDetectionEnabled) {
this.disableVadInterruptionSoon();
Expand Down Expand Up @@ -4932,16 +4941,6 @@ export class AgentActivity implements RecognitionHooks {
return;
}

// The pause withheld end-of-agent-speech for a resume. Interrupting ends the turn instead;
// audio stopped when it was paused, so no playout is left to wait for.
if (interrupt && this.audioRecognition) {
void this.audioRecognition
.onEndOfAgentSpeech(Date.now())
.catch((error) =>
this.logger.warn({ error }, 'failed to report end of agent speech on pause cancel'),
);
}

if (
interrupt &&
!this.pausedSpeech.handle.interrupted &&
Expand Down
88 changes: 88 additions & 0 deletions agents/src/voice/agent_activity_tool_output_commit.test.ts
Original file line number Diff line number Diff line change
@@ -1,13 +1,18 @@
// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
//
// SPDX-License-Identifier: Apache-2.0
import { AudioFrame } from '@livekit/rtc-node';
import { ReadableStream } from 'node:stream/web';
import { describe, expect, it, vi } from 'vitest';
import type { ChatContext } from '../llm/chat_context.js';
import { tool } from '../llm/tool_context.js';
import { FakeSTT } from '../stt/testing/fake_stt.js';
import { Future } from '../utils.js';
import { Agent } from './agent.js';
import { AgentSession } from './agent_session.js';
import { AudioRecognition } from './audio_recognition.js';
import { RUNNING_TOOL_PLACEHOLDER } from './generation.js';
import { AudioOutput } from './io.js';
import { FakeLLM } from './testing/fake_llm.js';

type PostToolContextObservation = {
Expand Down Expand Up @@ -49,7 +54,90 @@ class ContextInspectingLLM extends FakeLLM {
}
}

class ImmediateOutput extends AudioOutput {
constructor() {
super(24_000);
}

override async captureFrame(frame: AudioFrame): Promise<void> {
const segmentCount = this.capturedPlayoutSegments;
await super.captureFrame(frame);
if (this.capturedPlayoutSegments > segmentCount) {
this.onPlaybackStarted(Date.now());
}
}

override flush(): void {
super.flush();
if (this.pendingPlayoutSegments > 0) {
this.onPlaybackFinished({ playbackPosition: 0.02, interrupted: false });
}
}

override clearBuffer(): void {
if (this.pendingPlayoutSegments > 0) {
this.onPlaybackFinished({ playbackPosition: 0, interrupted: true });
}
}
}

class FrameAgent extends Agent {
constructor() {
super({
instructions: 'test',
tools: {
lookup: tool({
description: 'Look up a value',
execute: async () => 'forecast',
}),
},
});
}

override async ttsNode(): Promise<ReadableStream<AudioFrame>> {
return new ReadableStream<AudioFrame>({
start(controller) {
controller.enqueue(new AudioFrame(new Int16Array(480), 24_000, 1, 480));
controller.close();
},
});
}
}

describe('AgentActivity tool output commit ordering', () => {
it('ends active speech after entering the tool-call thinking state', async () => {
const llm = new FakeLLM([
{
input: 'look it up',
content: 'Let me check.',
toolCalls: [{ name: 'lookup', args: {} }],
},
{ input: '"forecast"', content: 'The forecast is clear.' },
]);
const session = new AgentSession({ llm, stt: new FakeSTT() });
session.output.audio = new ImmediateOutput();
const speechEndStates: string[] = [];
const onEndOfAgentSpeech = AudioRecognition.prototype.onEndOfAgentSpeech;
const speechEndSpy = vi
.spyOn(AudioRecognition.prototype, 'onEndOfAgentSpeech')
.mockImplementation(async function (this: AudioRecognition, ignoreUntil: number) {
speechEndStates.push(session.agentState);
await onEndOfAgentSpeech.call(this, ignoreUntil);
});

await session.start({ agent: new FrameAgent() });
try {
const speech = session.generateReply({ userInput: 'look it up' });
await speech.waitForPlayout();

expect(speechEndStates[0]).toBe('thinking');
expect(speechEndStates.slice(1).every((state) => state === 'listening')).toBe(true);
} finally {
speechEndSpy.mockRestore();
await session.close();
}
});

it('invalidates a stale preemptive generation when late EOU interrupts the post-tool reply', async () => {
const llm = new ContextInspectingLLM([
{
Expand Down
Loading
Loading