fix: commit tool outputs before post-tool replies

.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.

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,

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<typeof buildToolOutput>;
     speechHandle: SpeechHandle;
-    createdAt: number;
   }) => Promise<boolean>;
 
   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,15 +1356,14 @@ describe('AgentActivity - interruption while waiting for tools', () => {
       },
       toolOutput,
       speechHandle,
-      createdAt: 456,
     });
     expect(activity['_backgroundSpeeches']).toContain(speechHandle);
 
     speechHandle.interrupt();
     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);
   });
 });
 

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<void> => {
@@ -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<void> =>
     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<Task<void>, 'result' | 'cancelAndWait'>;
     toolOutput: ToolOutput;
     speechHandle: SpeechHandle;
-    createdAt: number;
   }): Promise<boolean> {
     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);