feat(voice): stream tool call status events

.changeset/stream-tool-status-events.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Stream voice tool call status lifecycle events.

agents/src/voice/agent_activity.ts

@@ -108,6 +108,7 @@ import {
   createMetricsCollectedEvent,
   createSessionUsageUpdatedEvent,
   createSpeechCreatedEvent,
+  createToolExecutionUpdatedEvent,
   createUserInputTranscribedEvent,
 } from './events.js';
 import type { ToolExecutionOutput, ToolOutput, _TTSGenerationData } from './generation.js';
@@ -3069,6 +3070,11 @@ export class AgentActivity implements RecognitionHooks {
       const respondToolChoice =
         schedulingPaused || modelSettings.toolChoice === 'none' ? 'none' : 'auto';
 
+      const updateIds = functionToolsExecutedEvent.functionCallOutputs.map(
+        (output) => output.callId,
+      );
+      const initialChatItemCount = speechHandle.chatItems.length;
+
       // Reuse the same speechHandle for the tool response.
       const toolResponseTask = this.createSpeechTask({
         taskFn: () =>
@@ -3087,7 +3093,33 @@ export class AgentActivity implements RecognitionHooks {
         name: 'AgentActivity.pipelineReply',
       });
 
-      toolResponseTask.result.finally(() => this.onPipelineReplyDone());
+      this.agentSession.emit(
+        AgentSessionEventTypes.ToolExecutionUpdated,
+        createToolExecutionUpdatedEvent({
+          type: 'tool_reply_updated',
+          updateIds,
+          status: 'scheduled',
+          speechId: speechHandle.id,
+        }),
+      );
+
+      toolResponseTask.result.finally(() => {
+        const status = speechHandle.interrupted
+          ? 'interrupted'
+          : speechHandle.chatItems.length > initialChatItemCount
+            ? 'completed'
+            : 'skipped';
+        this.agentSession.emit(
+          AgentSessionEventTypes.ToolExecutionUpdated,
+          createToolExecutionUpdatedEvent({
+            type: 'tool_reply_updated',
+            updateIds,
+            status,
+            speechId: speechHandle.id,
+          }),
+        );
+        this.onPipelineReplyDone();
+      });
 
       this.scheduleSpeech(speechHandle, SpeechHandle.SPEECH_PRIORITY_NORMAL, true);
     } else if (functionToolsExecutedEvent.functionCallOutputs.length > 0) {

agents/src/voice/agent_session.ts

@@ -74,6 +74,7 @@ import {
   type SessionUsageUpdatedEvent,
   type ShutdownReason,
   type SpeechCreatedEvent,
+  type ToolExecutionUpdatedEvent,
   type UserInputTranscribedEvent,
   type UserState,
   type UserStateChangedEvent,
@@ -213,6 +214,7 @@ export type AgentSessionCallbacks = {
   [AgentSessionEventTypes.UserStateChanged]: (ev: UserStateChangedEvent) => void;
   [AgentSessionEventTypes.ConversationItemAdded]: (ev: ConversationItemAddedEvent) => void;
   [AgentSessionEventTypes.FunctionToolsExecuted]: (ev: FunctionToolsExecutedEvent) => void;
+  [AgentSessionEventTypes.ToolExecutionUpdated]: (ev: ToolExecutionUpdatedEvent) => void;
   [AgentSessionEventTypes.MetricsCollected]: (ev: MetricsCollectedEvent) => void;
   [AgentSessionEventTypes.SessionUsageUpdated]: (ev: SessionUsageUpdatedEvent) => void;
   [AgentSessionEventTypes.DebugMessage]: (ev: pb.DebugMessage) => void;

agents/src/voice/events.ts

@@ -28,6 +28,7 @@ export enum AgentSessionEventTypes {
   UserStateChanged = 'user_state_changed',
   ConversationItemAdded = 'conversation_item_added',
   FunctionToolsExecuted = 'function_tools_executed',
+  ToolExecutionUpdated = 'tool_execution_updated',
   MetricsCollected = 'metrics_collected',
   SessionUsageUpdated = 'session_usage_updated',
   DebugMessage = 'debug_message',
@@ -214,6 +215,59 @@ export const zipFunctionCallsAndOutputs = (
   return event.functionCalls.map((call, index) => [call, event.functionCallOutputs[index]!]);
 };
 
+export type ToolCallStarted = {
+  type: 'tool_call_started';
+  functionCall: FunctionCall;
+};
+
+export type ToolCallUpdated = {
+  type: 'tool_call_updated';
+  /** Entry id: `callId` inline, `${callId}_update_N` when deferred. */
+  id: string;
+  callId: string;
+  message: string;
+};
+
+export type ToolCallEnded = {
+  type: 'tool_call_ended';
+  /** Entry id: `callId` inline, `${callId}_final` when deferred. */
+  id: string;
+  callId: string;
+  /** Result or error text; null when there is nothing to voice. */
+  message: string | null;
+  status: 'done' | 'error' | 'cancelled';
+};
+
+export type ToolReplyUpdated = {
+  type: 'tool_reply_updated';
+  /** `ToolCallUpdated.id` / `ToolCallEnded.id` values this reply covers. */
+  updateIds: string[];
+  status: 'scheduled' | 'completed' | 'interrupted' | 'skipped';
+  /** Id of the reply speech; `speech_created` carries its handle. */
+  speechId: string;
+};
+
+export type ToolExecutionUpdate =
+  | ToolCallStarted
+  | ToolCallUpdated
+  | ToolCallEnded
+  | ToolReplyUpdated;
+
+export type ToolExecutionUpdatedEvent = {
+  type: 'tool_execution_updated';
+  update: ToolExecutionUpdate;
+  createdAt: number;
+};
+
+export const createToolExecutionUpdatedEvent = (
+  update: ToolExecutionUpdate,
+  createdAt: number = Date.now(),
+): ToolExecutionUpdatedEvent => ({
+  type: 'tool_execution_updated',
+  update,
+  createdAt,
+});
+
 export type SpeechCreatedEvent = {
   type: 'speech_created';
   /**
@@ -444,6 +498,7 @@ export type AgentEvent =
   | SessionUsageUpdatedEvent
   | ConversationItemAddedEvent
   | FunctionToolsExecutedEvent
+  | ToolExecutionUpdatedEvent
   | SpeechCreatedEvent
   | AgentFalseInterruptionEvent
   | OverlappingSpeechEvent

agents/src/voice/generation.ts

@@ -49,6 +49,7 @@ import {
 } from './agent.js';
 import type { ForwardOutput } from './agent_activity.ts';
 import type { AgentSession } from './agent_session.js';
+import { AgentSessionEventTypes, createToolExecutionUpdatedEvent } from './events.js';
 import {
   AudioOutput,
   type LLMNode,
@@ -978,9 +979,34 @@ export function performToolExecutions({
     firstToolStartedFuture: new Future(),
   };
 
+  const emitToolExecutionUpdate = (
+    update: Parameters<typeof createToolExecutionUpdatedEvent>[0],
+  ) => {
+    if (typeof session.emit !== 'function') return;
+    session.emit(
+      AgentSessionEventTypes.ToolExecutionUpdated,
+      createToolExecutionUpdatedEvent(update),
+    );
+  };
+
   const toolCompleted = (out: ToolExecutionOutput) => {
     onToolExecutionCompleted(out);
     toolOutput.output.push(out);
+
+    const status =
+      out.rawException?.message === 'tool call was aborted'
+        ? 'cancelled'
+        : out.rawException && !isStopResponse(out.rawException)
+          ? 'error'
+          : 'done';
+    const message = status === 'cancelled' ? null : out.toolCallOutput?.output ?? null;
+    emitToolExecutionUpdate({
+      type: 'tool_call_ended',
+      id: out.toolCall.callId,
+      callId: out.toolCall.callId,
+      message,
+      status,
+    });
   };
 
   const executeToolsTask = async (controller: AbortController) => {
@@ -1029,6 +1055,8 @@ export function performToolExecutions({
         continue;
       }
 
+      emitToolExecutionUpdate({ type: 'tool_call_started', functionCall: toolCall });
+
       let parsedArgs: object | undefined;
 
       // Ensure valid arguments

agents/src/voice/remote_session.ts

@@ -39,12 +39,15 @@ import {
   type ErrorEvent,
   type FunctionToolsExecutedEvent,
   type MetricsCollectedEvent,
+  type ToolExecutionUpdatedEvent,
   type UserInputTranscribedEvent,
   type UserState,
   type UserStateChangedEvent,
 } from './events.js';
 import type { RoomIO } from './room_io/room_io.js';
 
+type ProtoMessageConstructor = new (params: Record<string, unknown>) => unknown;
+
 // ===========================================================================
 // Shared types (TextInput, Client event types, wire format aliases)
 // ===========================================================================
@@ -64,6 +67,7 @@ export type RemoteSessionEventTypes =
   | 'conversation_item_added'
   | 'user_input_transcribed'
   | 'function_tools_executed'
+  | 'tool_execution_updated'
   | 'overlapping_speech'
   | 'amd_prediction'
   | 'eot_prediction'
@@ -78,6 +82,7 @@ export type RemoteSessionCallbacks = {
   conversation_item_added: (ev: pb.AgentSessionEvent_ConversationItemAdded) => void;
   user_input_transcribed: (ev: pb.AgentSessionEvent_UserInputTranscribed) => void;
   function_tools_executed: (ev: pb.AgentSessionEvent_FunctionToolsExecuted) => void;
+  tool_execution_updated: (ev: unknown) => void;
   overlapping_speech: (ev: pb.AgentSessionEvent_OverlappingSpeech) => void;
   amd_prediction: (ev: pb.AgentSessionEvent_AmdPrediction) => void;
   eot_prediction: (ev: pb.AgentSessionEvent_EotPrediction) => void;
@@ -666,6 +671,7 @@ export class SessionHost {
       session.on(AgentSessionEventTypes.ConversationItemAdded, this.onConversationItemAdded);
       session.on(AgentSessionEventTypes.UserInputTranscribed, this.onUserInputTranscribed);
       session.on(AgentSessionEventTypes.FunctionToolsExecuted, this.onFunctionToolsExecuted);
+      session.on(AgentSessionEventTypes.ToolExecutionUpdated, this.onToolExecutionUpdated);
       session.on(AgentSessionEventTypes.MetricsCollected, this.onMetricsCollected);
       session.on(AgentSessionEventTypes.OverlappingSpeech, this.onOverlappingSpeech);
       session.on(AgentSessionEventTypes.EotPrediction, this.onEotPrediction);
@@ -692,6 +698,7 @@ export class SessionHost {
       this.session.off(AgentSessionEventTypes.ConversationItemAdded, this.onConversationItemAdded);
       this.session.off(AgentSessionEventTypes.UserInputTranscribed, this.onUserInputTranscribed);
       this.session.off(AgentSessionEventTypes.FunctionToolsExecuted, this.onFunctionToolsExecuted);
+      this.session.off(AgentSessionEventTypes.ToolExecutionUpdated, this.onToolExecutionUpdated);
       this.session.off(AgentSessionEventTypes.MetricsCollected, this.onMetricsCollected);
       this.session.off(AgentSessionEventTypes.OverlappingSpeech, this.onOverlappingSpeech);
       this.session.off(AgentSessionEventTypes.EotPrediction, this.onEotPrediction);
@@ -833,6 +840,112 @@ export class SessionHost {
     );
   };
 
+  private onToolExecutionUpdated = (event: ToolExecutionUpdatedEvent): void => {
+    const protocol = pb as unknown as Record<string, unknown>;
+    const ToolExecutionUpdated = protocol.AgentSessionEvent_ToolExecutionUpdated as
+      | ProtoMessageConstructor
+      | undefined;
+    if (!ToolExecutionUpdated) return;
+
+    const update = event.update;
+    let value: unknown;
+    switch (update.type) {
+      case 'tool_call_started': {
+        const Started = protocol.AgentSessionEvent_ToolExecutionUpdated_Started as
+          | ProtoMessageConstructor
+          | undefined;
+        if (!Started) return;
+        value = new ToolExecutionUpdated({
+          update: {
+            case: 'started',
+            value: new Started({
+              functionCall: new pb.FunctionCall({
+                id: update.functionCall.id,
+                callId: update.functionCall.callId,
+                name: update.functionCall.name,
+                arguments: update.functionCall.args,
+              }),
+            }),
+          },
+        });
+        break;
+      }
+      case 'tool_call_updated': {
+        const CallUpdated = protocol.AgentSessionEvent_ToolExecutionUpdated_CallUpdated as
+          | ProtoMessageConstructor
+          | undefined;
+        if (!CallUpdated) return;
+        value = new ToolExecutionUpdated({
+          update: {
+            case: 'callUpdated',
+            value: new CallUpdated({
+              id: update.id,
+              callId: update.callId,
+              message: update.message,
+            }),
+          },
+        });
+        break;
+      }
+      case 'tool_call_ended': {
+        const Ended = protocol.AgentSessionEvent_ToolExecutionUpdated_Ended as
+          | ProtoMessageConstructor
+          | undefined;
+        const ToolCallStatus = protocol.ToolCallStatus as Record<string, unknown> | undefined;
+        if (!Ended || !ToolCallStatus) return;
+        value = new ToolExecutionUpdated({
+          update: {
+            case: 'ended',
+            value: new Ended({
+              id: update.id,
+              callId: update.callId,
+              message: update.message ?? undefined,
+              status:
+                update.status === 'done'
+                  ? ToolCallStatus.TC_DONE
+                  : update.status === 'error'
+                    ? ToolCallStatus.TC_ERROR
+                    : ToolCallStatus.TC_CANCELLED,
+            }),
+          },
+        });
+        break;
+      }
+      case 'tool_reply_updated': {
+        const ReplyUpdated = protocol.AgentSessionEvent_ToolExecutionUpdated_ReplyUpdated as
+          | ProtoMessageConstructor
+          | undefined;
+        const ToolReplyStatus = protocol.ToolReplyStatus as Record<string, unknown> | undefined;
+        if (!ReplyUpdated || !ToolReplyStatus) return;
+        value = new ToolExecutionUpdated({
+          update: {
+            case: 'replyUpdated',
+            value: new ReplyUpdated({
+              updateIds: update.updateIds,
+              speechId: update.speechId,
+              status:
+                update.status === 'scheduled'
+                  ? ToolReplyStatus.TR_SCHEDULED
+                  : update.status === 'completed'
+                    ? ToolReplyStatus.TR_COMPLETED
+                    : update.status === 'interrupted'
+                      ? ToolReplyStatus.TR_INTERRUPTED
+                      : ToolReplyStatus.TR_SKIPPED,
+            }),
+          },
+        });
+        break;
+      }
+    }
+
+    this.sendEvent(
+      new pb.AgentSessionEvent({
+        createdAt: msToTimestamp(event.createdAt),
+        event: { case: 'toolExecutionUpdated' as never, value } as never,
+      }),
+    );
+  };
+
   private onEotPrediction = (event: EotPredictionEvent): void => {
     this._onEotPrediction(event);
   };
@@ -1192,6 +1305,11 @@ export class RemoteSession extends (EventEmitter as new () => TypedEventEmitter<
 
   private dispatchEvent(event: pb.AgentSessionEvent): void {
     const ev = event.event;
+    if ((ev as unknown as { case?: string }).case === 'toolExecutionUpdated') {
+      this.emit('tool_execution_updated', (ev as unknown as { value: unknown }).value);
+      return;
+    }
+
     switch (ev.case) {
       case 'agentStateChanged':
         this.emit('agent_state_changed', ev.value);