From 63215787cd3555597aa2ce92fd79bf0a65da7d3e Mon Sep 17 00:00:00 2001 From: "rosetta-livekit-bot[bot]" <282703043+rosetta-livekit-bot[bot]@users.noreply.github.com> Date: Wed, 1 Jul 2026 10:40:29 +0000 Subject: [PATCH] feat(voice): stream tool call status events --- .changeset/stream-tool-status-events.md | 5 + agents/src/voice/agent_activity.ts | 34 ++++++- agents/src/voice/agent_session.ts | 2 + agents/src/voice/events.ts | 55 +++++++++++ agents/src/voice/generation.ts | 28 ++++++ agents/src/voice/remote_session.ts | 118 ++++++++++++++++++++++++ agents/src/voice/run_context.ts | 18 ++++ 7 files changed, 259 insertions(+), 1 deletion(-) create mode 100644 .changeset/stream-tool-status-events.md diff --git a/.changeset/stream-tool-status-events.md b/.changeset/stream-tool-status-events.md new file mode 100644 index 000000000..bcdc171ee --- /dev/null +++ b/.changeset/stream-tool-status-events.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Stream voice tool call status lifecycle events. diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 412a7aab0..c6b78e4e0 100644 --- a/agents/src/voice/agent_activity.ts +++ b/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) { diff --git a/agents/src/voice/agent_session.ts b/agents/src/voice/agent_session.ts index 7e2b4423b..0140007e8 100644 --- a/agents/src/voice/agent_session.ts +++ b/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; diff --git a/agents/src/voice/events.ts b/agents/src/voice/events.ts index 3a3050f4b..ea6312ebe 100644 --- a/agents/src/voice/events.ts +++ b/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 diff --git a/agents/src/voice/generation.ts b/agents/src/voice/generation.ts index e438b25ce..5add3e71c 100644 --- a/agents/src/voice/generation.ts +++ b/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[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 diff --git a/agents/src/voice/remote_session.ts b/agents/src/voice/remote_session.ts index ae2af7bbe..f06011fa6 100644 --- a/agents/src/voice/remote_session.ts +++ b/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) => 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; + 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 | 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 | 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); diff --git a/agents/src/voice/run_context.ts b/agents/src/voice/run_context.ts index df9994ba8..46607228d 100644 --- a/agents/src/voice/run_context.ts +++ b/agents/src/voice/run_context.ts @@ -3,12 +3,14 @@ // SPDX-License-Identifier: Apache-2.0 import type { FunctionCall } from '../llm/chat_context.js'; import type { AgentSession } from './agent_session.js'; +import { AgentSessionEventTypes, createToolExecutionUpdatedEvent } from './events.js'; import type { SpeechHandle } from './speech_handle.js'; export type UnknownUserData = unknown; export class RunContext { private readonly initialStepIdx: number; + private updateCount = 0; constructor( public readonly session: AgentSession, public readonly speechHandle: SpeechHandle, @@ -31,4 +33,20 @@ export class RunContext { async waitForPlayout() { return this.speechHandle._waitForGeneration(this.initialStepIdx); } + + update(message: unknown): void { + if (typeof this.session.emit !== 'function') return; + const rawMessage = typeof message === 'string' ? message : String(message); + const id = `${this.functionCall.callId}_update_${this.updateCount}`; + this.updateCount += 1; + this.session.emit( + AgentSessionEventTypes.ToolExecutionUpdated, + createToolExecutionUpdatedEvent({ + type: 'tool_call_updated', + id, + callId: this.functionCall.callId, + message: rawMessage, + }), + ); + } }