diff --git a/.changeset/eager-tool-output-commit.md b/.changeset/eager-tool-output-commit.md new file mode 100644 index 000000000..b8eadaf76 --- /dev/null +++ b/.changeset/eager-tool-output-commit.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Commit completed tool outputs before starting the post-tool reply so overlapping turns cannot reuse stale preemptive generations. Preserve tool completion timestamps, pair tool-error outputs with their calls, and normalize unparseable call arguments before saving them. diff --git a/agents/src/inference/interruption/interruption_stream.ts b/agents/src/inference/interruption/interruption_stream.ts index efb02819f..8b086ceba 100644 --- a/agents/src/inference/interruption/interruption_stream.ts +++ b/agents/src/inference/interruption/interruption_stream.ts @@ -272,8 +272,8 @@ export class InterruptionStreamBase { } cache.clear(); } else if (chunk.type === 'overlap-speech-ended') { - this.logger.debug('overlap speech ended'); if (overlapSpeechStarted) { + this.logger.debug('overlap speech ended'); this.userSpeakingSpan = undefined; let latestEntry = cache.pop( (entry) => entry.totalDurationInS !== undefined && entry.totalDurationInS > 0, diff --git a/agents/src/voice/agent_activity.test.ts b/agents/src/voice/agent_activity.test.ts index 6fb51260f..4e9e57ae6 100644 --- a/agents/src/voice/agent_activity.test.ts +++ b/agents/src/voice/agent_activity.test.ts @@ -1013,6 +1013,9 @@ describe('AgentActivity - interrupted tool completion', () => { output: 'charged', isError: false, }); + call.createdAt = 200; + output.createdAt = 300; + chatCtx.insert(call); const toolOutput = { output: [ ToolExecutionOutput.create({ @@ -1030,12 +1033,13 @@ describe('AgentActivity - interrupted tool completion', () => { _commitInterruptedToolOutputs: ( toolOutput: typeof toolOutput, speechHandle: SpeechHandle, - createdAt: number, ) => void; } - )._commitInterruptedToolOutputs(toolOutput, SpeechHandle.create(), 123); + )._commitInterruptedToolOutputs(toolOutput, SpeechHandle.create()); expect(chatCtx.items).toContain(output); + expect(output.createdAt).toBe(300); + expect(chatCtx.items).toEqual([call, output]); expect(toolItemsAdded).toHaveBeenCalledWith([output]); expect(generateReply).not.toHaveBeenCalled(); }); @@ -1086,10 +1090,9 @@ describe('AgentActivity - interrupted tool completion', () => { _commitInterruptedToolOutputs: ( toolOutput: typeof toolOutput, speechHandle: SpeechHandle, - createdAt: number, ) => void; } - )._commitInterruptedToolOutputs(toolOutput, SpeechHandle.create(), 123); + )._commitInterruptedToolOutputs(toolOutput, SpeechHandle.create()); expect(chatCtx.items).not.toContain(call); expect(chatCtx.items).not.toContain(output); @@ -1157,10 +1160,9 @@ describe('AgentActivity - interrupted tool completion', () => { _commitInterruptedToolOutputs: ( toolOutput: typeof toolOutput, speechHandle: SpeechHandle, - createdAt: number, ) => void; } - )._commitInterruptedToolOutputs(toolOutput, SpeechHandle.create(), 123); + )._commitInterruptedToolOutputs(toolOutput, SpeechHandle.create()); expect(chatCtx.items).toHaveLength(2); expect(chatCtx.items).toEqual(expect.arrayContaining([completedCall, completedOutput])); @@ -1306,7 +1308,6 @@ describe('AgentActivity - interruption while waiting for tools', () => { }; toolOutput: ReturnType; speechHandle: SpeechHandle; - createdAt: number; }) => Promise; function buildActivity() { @@ -1334,12 +1335,11 @@ describe('AgentActivity - interruption while waiting for tools', () => { executeToolsTask: { result: Promise.resolve(), cancelAndWait }, toolOutput, speechHandle, - createdAt: 123, }); expect(shouldContinue).toBe(false); expect(cancelAndWait).toHaveBeenCalledOnce(); - expect(commitInterruptedToolOutputs).toHaveBeenCalledWith(toolOutput, speechHandle, 123); + expect(commitInterruptedToolOutputs).toHaveBeenCalledWith(toolOutput, speechHandle); expect(activity['_backgroundSpeeches']).not.toContain(speechHandle); }); @@ -1356,7 +1356,6 @@ describe('AgentActivity - interruption while waiting for tools', () => { }, toolOutput, speechHandle, - createdAt: 456, }); expect(activity['_backgroundSpeeches']).toContain(speechHandle); @@ -1364,7 +1363,7 @@ describe('AgentActivity - interruption while waiting for tools', () => { executionFinished.resolve(); await expect(waiting).resolves.toBe(false); - expect(commitInterruptedToolOutputs).toHaveBeenCalledWith(toolOutput, speechHandle, 456); + expect(commitInterruptedToolOutputs).toHaveBeenCalledWith(toolOutput, speechHandle); expect(activity['_backgroundSpeeches']).not.toContain(speechHandle); }); @@ -1386,11 +1385,10 @@ describe('AgentActivity - interruption while waiting for tools', () => { }, toolOutput, speechHandle, - createdAt: 789, }); expect(shouldContinue).toBe(false); - expect(commitInterruptedToolOutputs).toHaveBeenCalledWith(toolOutput, speechHandle, 789); + expect(commitInterruptedToolOutputs).toHaveBeenCalledWith(toolOutput, speechHandle); }); }); diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 3a1ff84cf..e2f4e22e0 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -2780,7 +2780,6 @@ export class AgentActivity implements RecognitionHooks { replyAbortController, instructions, newMessage, - toolsMessages, span, _previousUserMetrics, }: { @@ -2791,7 +2790,6 @@ export class AgentActivity implements RecognitionHooks { replyAbortController: AbortController; instructions?: string | Instructions; newMessage?: ChatMessage; - toolsMessages?: ChatItem[]; span: Span; _previousUserMetrics?: MetricsReport; }): Promise => { @@ -3213,24 +3211,6 @@ export class AgentActivity implements RecognitionHooks { span.setAttribute(traceTypes.ATTR_SPEECH_INTERRUPTED, speechHandle.interrupted); let hasSpeechMessage = false; - // add the tools messages that triggers this reply to the chat context - if (toolsMessages) { - for (const msg of toolsMessages) { - msg.createdAt = replyStartedAt; - } - // Only insert FunctionCallOutput items into agent._chatCtx since FunctionCall items - // were already added by onToolExecutionStarted when the tool execution began. - // Inserting function_calls again would create duplicates that break provider APIs - // (e.g. Google's "function response parts != function call parts" error). - const toolCallOutputs = toolsMessages.filter( - (m): m is FunctionCallOutput => m.type === 'function_call_output', - ); - if (toolCallOutputs.length > 0) { - this.agent._chatCtx.insert(toolCallOutputs); - this.agentSession._toolItemsAdded(toolCallOutputs); - } - } - if (speechHandle.interrupted) { this.logger.debug( { speech_id: speechHandle.id }, @@ -3280,7 +3260,7 @@ export class AgentActivity implements RecognitionHooks { speechHandle._markGenerationDone(); } await this.cancelToolExecutions(executeToolsTask, speechHandle, toolOutput); - this._commitInterruptedToolOutputs(toolOutput, speechHandle, replyStartedAt); + this._commitInterruptedToolOutputs(toolOutput, speechHandle); return; } @@ -3330,7 +3310,6 @@ export class AgentActivity implements RecognitionHooks { executeToolsTask, toolOutput, speechHandle, - createdAt: replyStartedAt, }); if (!toolExecutionCompleted) return; @@ -3364,6 +3343,15 @@ export class AgentActivity implements RecognitionHooks { ...functionToolsExecutedEvent.functionCalls, ...functionToolsExecutedEvent.functionCallOutputs, ] as ChatItem[]; + + // Function calls were committed when execution started. Commit their outputs before + // scheduling a reply so overlapping turns observe the completed tool context. + const toolCallOutputs = functionToolsExecutedEvent.functionCallOutputs; + if (toolCallOutputs.length > 0) { + this.agent._chatCtx.insert(toolCallOutputs); + this.agentSession._toolItemsAdded(toolCallOutputs); + } + if (shouldGenerateToolReply) { _stripRunningToolCalls(chatCtx); chatCtx.insert(toolMessages); @@ -3389,7 +3377,6 @@ export class AgentActivity implements RecognitionHooks { replyAbortController, instructions, undefined, - toolMessages, hasSpeechMessage ? undefined : userMetrics, ), ownedSpeechHandle: speechHandle, @@ -3399,19 +3386,6 @@ export class AgentActivity implements RecognitionHooks { toolResponseTask.result.finally(() => this.onPipelineReplyDone()); this.scheduleSpeech(speechHandle, SpeechHandle.SPEECH_PRIORITY_NORMAL, true); - } else if (functionToolsExecutedEvent.functionCallOutputs.length > 0) { - for (const msg of toolMessages) { - msg.createdAt = replyStartedAt; - } - - const toolCallOutputs = toolMessages.filter( - (m): m is FunctionCallOutput => m.type === 'function_call_output', - ); - - if (toolCallOutputs.length > 0) { - this.agent._chatCtx.insert(toolCallOutputs); - this.agentSession._toolItemsAdded(toolCallOutputs); - } } }; @@ -3423,7 +3397,6 @@ export class AgentActivity implements RecognitionHooks { replyAbortController: AbortController, instructions?: string | Instructions, newMessage?: ChatMessage, - toolsMessages?: ChatItem[], _previousUserMetrics?: MetricsReport, ): Promise => tracer.startActiveSpan( @@ -3436,7 +3409,6 @@ export class AgentActivity implements RecognitionHooks { replyAbortController, instructions, newMessage, - toolsMessages, span, _previousUserMetrics, }), @@ -4086,16 +4058,14 @@ export class AgentActivity implements RecognitionHooks { executeToolsTask, toolOutput, speechHandle, - createdAt, }: { executeToolsTask: Pick, 'result' | 'cancelAndWait'>; toolOutput: ToolOutput; speechHandle: SpeechHandle; - createdAt: number; }): Promise { if (speechHandle.interrupted) { await this.cancelToolExecutions(executeToolsTask, speechHandle, toolOutput); - this._commitInterruptedToolOutputs(toolOutput, speechHandle, createdAt); + this._commitInterruptedToolOutputs(toolOutput, speechHandle); return false; } @@ -4107,18 +4077,14 @@ export class AgentActivity implements RecognitionHooks { } if (speechHandle.interrupted) { - this._commitInterruptedToolOutputs(toolOutput, speechHandle, createdAt); + this._commitInterruptedToolOutputs(toolOutput, speechHandle); return false; } return true; } /** @internal */ - _commitInterruptedToolOutputs( - toolOutput: ToolOutput, - speechHandle: SpeechHandle, - createdAt: number, - ): void { + _commitInterruptedToolOutputs(toolOutput: ToolOutput, speechHandle: SpeechHandle): void { const interruptedHandoffCallIds = toolOutput.output .filter((output) => output.agentTask !== undefined) .map((output) => output.toolCall.callId); @@ -4141,9 +4107,6 @@ export class AgentActivity implements RecognitionHooks { functionToolsExecutedEvent, ); const outputs = functionToolsExecutedEvent.functionCallOutputs; - for (const output of outputs) { - output.createdAt = createdAt; - } if (outputs.length > 0) { this.agent._chatCtx.insert(outputs); this.agentSession._toolItemsAdded(outputs); diff --git a/agents/src/voice/agent_activity_tool_output_commit.test.ts b/agents/src/voice/agent_activity_tool_output_commit.test.ts new file mode 100644 index 000000000..d35cf167e --- /dev/null +++ b/agents/src/voice/agent_activity_tool_output_commit.test.ts @@ -0,0 +1,148 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { describe, expect, it, vi } from 'vitest'; +import type { ChatContext } from '../llm/chat_context.js'; +import { tool } from '../llm/tool_context.js'; +import { Future } from '../utils.js'; +import { Agent } from './agent.js'; +import { AgentSession } from './agent_session.js'; +import { RUNNING_TOOL_PLACEHOLDER } from './generation.js'; +import { FakeLLM } from './testing/fake_llm.js'; + +type PostToolContextObservation = { + canonicalHasToolOutput: boolean; + preToolSnapshotIsEquivalent: boolean; +}; + +class ContextInspectingLLM extends FakeLLM { + agent?: Agent; + preToolContext?: ChatContext; + readonly mhmmContexts: ChatContext[] = []; + readonly postToolInference = new Future(); + + override chat(options: Parameters[0]) { + if ( + options.chatCtx.items.some( + (item) => item.type === 'message' && item.role === 'user' && item.textContent === 'Mhmm', + ) + ) { + this.mhmmContexts.push(options.chatCtx.copy()); + } + + const toolOutput = options.chatCtx.items.find( + (item) => item.type === 'function_call_output' && item.output !== RUNNING_TOOL_PLACEHOLDER, + ); + if (toolOutput && !this.postToolInference.done) { + if (!this.agent || !this.preToolContext) { + throw new Error('tool context snapshots were not initialized'); + } + const canonicalContext = this.agent.chatCtx; + this.postToolInference.resolve({ + canonicalHasToolOutput: canonicalContext.items.some( + (item) => item.type === 'function_call_output' && item.callId === toolOutput.callId, + ), + preToolSnapshotIsEquivalent: this.preToolContext.isEquivalent(canonicalContext), + }); + } + return super.chat(options); + } +} + +describe('AgentActivity tool output commit ordering', () => { + it('invalidates a stale preemptive generation when late EOU interrupts the post-tool reply', async () => { + const llm = new ContextInspectingLLM([ + { + input: 'find loads', + toolCalls: [{ name: 'lookup_loads', args: {} }], + }, + { input: 'Mhmm', content: 'Let me check.' }, + { input: '"no loads"', content: 'No loads were found.', duration: 1_000 }, + ]); + const toolStarted = new Future(); + const releaseTool = new Future(); + const agent: Agent = new Agent({ + instructions: 'test', + tools: { + lookup_loads: tool({ + description: 'Look up available loads', + execute: async () => { + llm.preToolContext = agent.chatCtx.copy(); + toolStarted.resolve(); + await releaseTool.await; + return 'no loads'; + }, + }), + }, + }); + llm.agent = agent; + + const session = new AgentSession({ llm }); + session.output.setAudioEnabled(false); + session.output.setTranscriptionEnabled(false); + + await session.start({ agent }); + try { + const speech = session.generateReply({ userInput: 'find loads' }); + await toolStarted.await; + + const activity = session._activity as unknown as { + _currentSpeech?: unknown; + onPreemptiveGeneration: (info: { + newTranscript: string; + transcriptConfidence: number; + startedSpeakingAt?: number; + }) => void; + userTurnCompleted: (info: { + newTranscript: string; + transcriptConfidence: number; + skipReply: boolean; + }) => Promise; + }; + await vi.waitFor(() => expect(activity._currentSpeech).toBeUndefined()); + activity.onPreemptiveGeneration({ + newTranscript: 'Mhmm', + transcriptConfidence: 0.9, + }); + await vi.waitFor(() => expect(llm.mhmmContexts).toHaveLength(1)); + + releaseTool.resolve(); + const observation = await llm.postToolInference.await; + + expect(llm.preToolContext?.items.some((item) => item.type === 'function_call')).toBe(true); + expect(observation.canonicalHasToolOutput).toBe(true); + expect(observation.preToolSnapshotIsEquivalent).toBe(false); + + await vi.waitFor(() => expect(activity._currentSpeech).toBeDefined()); + await activity.userTurnCompleted({ + newTranscript: 'Mhmm', + transcriptConfidence: 0.9, + skipReply: false, + }); + await vi.waitFor(() => expect(llm.mhmmContexts).toHaveLength(2)); + + const staleOutput = llm.mhmmContexts[0]!.items.find( + (item) => item.type === 'function_call_output', + ); + const regeneratedOutput = llm.mhmmContexts[1]!.items.find( + (item) => item.type === 'function_call_output', + ); + expect(staleOutput?.output).toBe(RUNNING_TOOL_PLACEHOLDER); + expect(regeneratedOutput?.output).toBe('"no loads"'); + + await speech.waitForPlayout(); + + const toolCall = agent.chatCtx.items.find((item) => item.type === 'function_call'); + const toolOutputs = agent.chatCtx.items.filter( + (item) => item.type === 'function_call_output' && item.callId === toolCall?.callId, + ); + expect(toolCall).toBeDefined(); + expect(toolOutputs).toHaveLength(1); + expect(agent.chatCtx.items.indexOf(toolOutputs[0]!)).toBeGreaterThan( + agent.chatCtx.items.indexOf(toolCall!), + ); + } finally { + await session.close(); + } + }); +}); diff --git a/agents/src/voice/generation.ts b/agents/src/voice/generation.ts index 5e951dd8c..9df7b8276 100644 --- a/agents/src/voice/generation.ts +++ b/agents/src/voice/generation.ts @@ -1126,6 +1126,12 @@ export function performToolExecutions({ onToolExecutionCompleted(out); toolOutput.output.push(out); }; + const toolStarted = (toolCall: FunctionCall) => { + if (!toolOutput.firstToolStartedFuture.done) { + toolOutput.firstToolStartedFuture.resolve(); + } + onToolExecutionStarted(toolCall); + }; const executeToolsTask = async (controller: AbortController) => { const signal = controller.signal; @@ -1174,6 +1180,12 @@ export function performToolExecutions({ }, `unknown AI function ${toolCall.name}`, ); + try { + toolCall.args = JSON.stringify(parseFunctionArguments(toolCall.args || '{}')); + } catch { + toolCall.args = '{}'; + } + toolStarted(toolCall); toolCompleted( createToolOutput({ toolCall, @@ -1195,6 +1207,7 @@ export function performToolExecutions({ } let parsedArgs: object | undefined; + let argumentsParsed = false; // Ensure valid arguments try { @@ -1204,6 +1217,7 @@ export function performToolExecutions({ if (canonicalArgs !== rawArgs) { toolCall.args = canonicalArgs; } + argumentsParsed = true; if (isZodSchema(tool.parameters)) { const result = await parseZodSchema(tool.parameters, jsonArgs); @@ -1226,9 +1240,13 @@ export function performToolExecutions({ }, `tried to call AI function ${toolCall.name} with invalid arguments`, ); + if (!argumentsParsed) { + toolCall.args = '{}'; + } // Surface argument-validation errors to the LLM via ToolError so it can correct // its arguments instead of looping on the same invalid call. The argument schema // and the validator's error message do not contain server-side internals. + toolStarted(toolCall); toolCompleted( createToolOutput({ toolCall, @@ -1241,11 +1259,7 @@ export function performToolExecutions({ // Resolve right after argument parsing and before execution (including the // executor's duplicate-check). This ensures a tool that gets duplicate-rejected // by the executor doesn't leave callers awaiting `firstToolStartedFuture` hanging forever. - if (!toolOutput.firstToolStartedFuture.done) { - toolOutput.firstToolStartedFuture.resolve(); - } - - onToolExecutionStarted(toolCall); + toolStarted(toolCall); logger.info( { diff --git a/agents/src/voice/generation_tools.test.ts b/agents/src/voice/generation_tools.test.ts index c4e312edf..22865dc23 100644 --- a/agents/src/voice/generation_tools.test.ts +++ b/agents/src/voice/generation_tools.test.ts @@ -382,6 +382,34 @@ describe('Generation + Tool Execution', () => { await task1.result; }, 20_000); + it('normalizes and commits an unknown tool call before its error output', async () => { + const events: string[] = []; + const fc = FunctionCall.create({ + callId: 'call_unknown', + name: 'missing', + args: 'definitely not json', + }); + const [execTask, toolOutput] = performToolExecutions({ + session: {} as AgentSession, + speechHandle: { id: 'speech_unknown', _itemAdded: () => {} } as unknown as SpeechHandle, + toolCtx: ToolContext.empty(), + toolCallStream: createFunctionCallStream(fc), + controller: new AbortController(), + onToolExecutionStarted: (call) => events.push(`call:${call.callId}`), + onToolExecutionCompleted: (output) => events.push(`output:${output.toolCall.callId}`), + }); + + await execTask.result; + + expect(events).toEqual(['call:call_unknown', 'output:call_unknown']); + expect(fc.args).toBe('{}'); + expect(toolOutput.firstToolStartedFuture.done).toBe(true); + expect(toolOutput.output[0]?.toolCallOutput).toMatchObject({ + callId: 'call_unknown', + isError: true, + }); + }); + it('should surface zod validation errors to the LLM with field-level detail', async () => { const replyAbortController = new AbortController(); @@ -399,6 +427,7 @@ describe('Generation + Tool Execution', () => { args: JSON.stringify({ msg: 123 }), }); const toolCallStream = createFunctionCallStream(fc); + const onToolExecutionStarted = vi.fn(); const [execTask, toolOutput] = performToolExecutions({ session: {} as any, @@ -406,9 +435,12 @@ describe('Generation + Tool Execution', () => { toolCtx: new ToolContext([echo]) as any, toolCallStream, controller: replyAbortController, + onToolExecutionStarted, }); await execTask.result; + expect(onToolExecutionStarted).toHaveBeenCalledWith(fc); + expect(JSON.parse(fc.args)).toEqual({ msg: 123 }); expect(toolOutput.output.length).toBe(1); const out = toolOutput.output[0]; expect(out?.toolCallOutput?.isError).toBe(true); @@ -448,6 +480,7 @@ describe('Generation + Tool Execution', () => { }); await execTask.result; + expect(fc.args).toBe('{}'); expect(toolOutput.output.length).toBe(1); const out = toolOutput.output[0]; expect(out?.toolCallOutput?.isError).toBe(true);