From b50cc56b9fbb130cfadf55a623a56a3315e620ef Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 14 Sep 2026 23:16:07 -0700 Subject: [PATCH] feat(telemetry): one agent_turn span per speech handle Port of livekit/agents#7143. A reply that calls a tool runs two generations (LLM steps) in two tasks; they were two agent_turn spans linked only by lk.parent_generation_id, so one response rendered as two turns. A speech handle is now exactly one agent_turn. - The speech handle owns the span: the first reply task (pipeline, realtime, or say) opens agent_turn under agent_session; the follow-up generation after a tool call continues the open span. It ends with the speech in SpeechHandle._markDone, recording the speech's error redaction-aware and the gen_ai.invoke_agent.duration histogram for the whole turn (new in JS: otel_metrics.recordInvokeAgentDuration and trace_types.METRIC_GEN_AI_INVOKE_AGENT_DURATION; the Python metric already existed). - Each generation is a `generation` event with lk.generation_id and lk.parent_generation_id (SpeechHandle._generationId / _parentGenerationId, `_` like Python); the span carries the latest generation id and the new lk.generation_count. - A preemptive generation discarded for a successor answering the same user turn hands its open agent_turn over (preemptive_generation_discarded event, lk.speech_id follows the speech that answered), both on a newer attempt and on the real reply after onUserTurnCompleted invalidated it. The queue-wait and interruption helpers tolerate an ended span. - `say` gets an agent_turn too, as in Python's _tts_task; JS had none. Tests: agent_turn_span.test.ts (tool call is one turn with two generation events and every step nested; plain reply is one generation; discarded preemptive hand-off; LLM failure fails the turn; duration metric when sampled out; sampled-out hand-off). The preemptive-guard stand-in handle gained _takeAgentTurn; the PII key test skips METRIC_* names, which are not attribute keys. The handoff runs inside generateReply, before the reply task is created: Task starts its body synchronously and the task opens the turn in its first statements, so a handoff performed after generateReply() returned came too late, leaving the successor's own agent_turn unended and its llm_node, tts_node and function_tool dangling from a span that was never exported (seen in a cloud export as bare nodes and turns whose generation count did not match their events). A regression test drives the real reply task. Co-Authored-By: Claude Fable 5.1 --- .changeset/agent-turn-per-speech.md | 5 + agents/etc/agents.api.md | 47 ++- agents/src/telemetry/otel_metrics.ts | 63 +++- agents/src/telemetry/trace_types.test.ts | 4 +- agents/src/telemetry/trace_types.ts | 13 + agents/src/voice/agent_activity.test.ts | 7 +- agents/src/voice/agent_activity.ts | 193 ++++++++-- agents/src/voice/agent_turn_span.test.ts | 437 +++++++++++++++++++++++ agents/src/voice/speech_handle.ts | 145 +++++++- 9 files changed, 859 insertions(+), 55 deletions(-) create mode 100644 .changeset/agent-turn-per-speech.md create mode 100644 agents/src/voice/agent_turn_span.test.ts diff --git a/.changeset/agent-turn-per-speech.md b/.changeset/agent-turn-per-speech.md new file mode 100644 index 000000000..629ea0f62 --- /dev/null +++ b/.changeset/agent-turn-per-speech.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +One `agent_turn` span per speech handle: the follow-up generation after a tool call continues the open span instead of opening a second turn, each generation is a `generation` event with `lk.generation_count` on the span, the turn ends with the speech, and a discarded preemptive generation hands its turn to the reply that answered. diff --git a/agents/etc/agents.api.md b/agents/etc/agents.api.md index dff5eb956..c9627b165 100644 --- a/agents/etc/agents.api.md +++ b/agents/etc/agents.api.md @@ -1137,7 +1137,7 @@ const ATTR_AGENT_PARENT_TURN_ID = "lk.parent_generation_id"; // Warning: (ae-missing-release-tag) "ATTR_AGENT_TURN_ID" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // -// @public (undocumented) +// @public const ATTR_AGENT_TURN_ID = "lk.generation_id"; // Warning: (ae-missing-release-tag) "ATTR_AMD_CATEGORY" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -1640,6 +1640,11 @@ const ATTR_GEN_AI_USAGE_TEXT_OUTPUT_TOKENS = "gen_ai.usage.text.output_tokens"; // @public (undocumented) const ATTR_GEN_AI_WORKFLOW_NAME = "gen_ai.workflow.name"; +// Warning: (ae-missing-release-tag) "ATTR_GENERATION_COUNT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +const ATTR_GENERATION_COUNT = "lk.generation_count"; + // Warning: (ae-missing-release-tag) "ATTR_INSTRUCTIONS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -6131,6 +6136,11 @@ class MetadataLogProcessor implements LogRecordProcessor { shutdown(): Promise; } +// Warning: (ae-missing-release-tag) "METRIC_GEN_AI_INVOKE_AGENT_DURATION" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +const METRIC_GEN_AI_INVOKE_AGENT_DURATION = "gen_ai.invoke_agent.duration"; + declare namespace metrics { export { AgentMetrics, @@ -7912,7 +7922,15 @@ export class SpeechHandle { // @internal (undocumented) _addItemAddedCallback(callback: (item: ChatItem) => void): void; // @internal + _agentTurnAgentName?: string; + // @internal _agentTurnContext?: Context; + // @internal + _agentTurnGenerations: number; + // @internal + _agentTurnSpan?: Span; + // @internal + _agentTurnStartedAt?: number; // (undocumented) get allowInterruptions(): boolean; set allowInterruptions(value: boolean); @@ -7926,6 +7944,11 @@ export class SpeechHandle { get chatItems(): ChatItem[]; // @internal (undocumented) _clearAuthorization(): void; + // Warning: (ae-forgotten-export) The symbol "AgentTurnContinuation" needs to be exported by the entry point index.d.ts + // Warning: (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "_takeAgentTurn" + // + // @internal + _continueAgentTurn(carry: AgentTurnCarry, from: SpeechHandle, continuation?: AgentTurnContinuation): void; // (undocumented) static create(options?: { allowInterruptions?: boolean; @@ -7935,7 +7958,19 @@ export class SpeechHandle { }): SpeechHandle; // (undocumented) done(): boolean; + // @internal + _emittedGenerationStep?: number; + // @internal + _error: unknown; exception(): unknown; + // @internal + _generationBaseId?: string; + // @internal + get _generationId(): string; + // @internal + get _generationStep(): number; + // @internal (undocumented) + _generationStepBase: number; // @internal (undocumented) get _hasGenerations(): boolean; // @internal (undocumented) @@ -7966,6 +8001,8 @@ export class SpeechHandle { // (undocumented) readonly parent?: SpeechHandle | undefined; // @internal + get _parentGenerationId(): string | undefined; + // @internal _queueWait(): number | undefined; // @internal (undocumented) _releaseInterruptions(): void; @@ -7982,6 +8019,10 @@ export class SpeechHandle { static SPEECH_PRIORITY_NORMAL: number; // @internal (undocumented) _stepIndex: number; + // Warning: (ae-forgotten-export) The symbol "AgentTurnCarry" needs to be exported by the entry point index.d.ts + // + // @internal + _takeAgentTurn(): AgentTurnCarry | undefined; // @internal (undocumented) _tasks: Task[]; then(onFulfilled?: ((value: ResolvedSpeechHandle) => R1 | PromiseLike) | null, onRejected?: ((reason: unknown) => R2 | PromiseLike) | null): Promise; @@ -9316,6 +9357,7 @@ declare namespace traceTypes { ATTR_CALLBACK_NAME, ATTR_AGENT_TURN_ID, ATTR_AGENT_PARENT_TURN_ID, + ATTR_GENERATION_COUNT, ATTR_USER_INPUT, ATTR_INSTRUCTIONS, ATTR_SPEECH_INTERRUPTED, @@ -9445,7 +9487,8 @@ declare namespace traceTypes { ATTR_EXCEPTION_TRACE, ATTR_EXCEPTION_TYPE, ATTR_EXCEPTION_MESSAGE, - ATTR_LANGFUSE_COMPLETION_START_TIME + ATTR_LANGFUSE_COMPLETION_START_TIME, + METRIC_GEN_AI_INVOKE_AGENT_DURATION } } diff --git a/agents/src/telemetry/otel_metrics.ts b/agents/src/telemetry/otel_metrics.ts index ecf06d9ea..5ce82cd32 100644 --- a/agents/src/telemetry/otel_metrics.ts +++ b/agents/src/telemetry/otel_metrics.ts @@ -3,28 +3,41 @@ // SPDX-License-Identifier: Apache-2.0 import { type Attributes, type Histogram, type MeterProvider, metrics } from '@opentelemetry/api'; import { getJobContext } from '../job.js'; +import * as traceTypes from './trace_types.js'; +// Instruments are looked up per call: the global meter provider may be installed after this +// module loads (the cloud pipeline is set up when the job registers), and a histogram created +// on the no-op provider would stay a no-op. let meterProvider: MeterProvider | undefined; -let blockedDuration: Histogram | undefined; +const histograms = new Map(); -function eventLoopBlockedHistogram(): Histogram { +function histogram(name: string, description: string): Histogram { const currentProvider = metrics.getMeterProvider(); - if (currentProvider !== meterProvider || !blockedDuration) { + if (currentProvider !== meterProvider) { meterProvider = currentProvider; - blockedDuration = metrics - .getMeter('livekit-agents') - .createHistogram('lk.agents.event_loop.blocked_duration', { - unit: 's', - description: 'Duration of synchronous blocks detected on an agent event loop', - }); + histograms.clear(); } - return blockedDuration; + let instrument = histograms.get(name); + if (!instrument) { + instrument = metrics.getMeter('livekit-agents').createHistogram(name, { + unit: 's', + description, + }); + histograms.set(name, instrument); + } + return instrument; } -/** Record an event-loop stall in seconds, with the severity and what caused it. */ -export function recordEventLoopBlocked(duration: number, severity: string, cause: string): void { +/** + * Per-measurement job attribution. + * + * The meter provider has process lifetime (the OTel metrics global is set-once), so per-job + * fields cannot live on its resource. Each measurement carries the same per-job attributes that + * are stamped on spans and logs instead. Returns a fresh object; callers may add to it. + */ +function jobAttrs(): Attributes { const ctx = getJobContext(false); - const attributes: Attributes = { severity, cause }; + const attributes: Attributes = {}; if (ctx) { Object.assign(attributes, ctx._otelMetadata()); const roomId = ctx.job.room?.sid; @@ -32,5 +45,27 @@ export function recordEventLoopBlocked(duration: number, severity: string, cause if (ctx.job.id) attributes.job_id = ctx.job.id; if (ctx.job.agentName) attributes['lk.agent_name'] = ctx.job.agentName; } - eventLoopBlockedHistogram().record(duration, attributes); + return attributes; +} + +/** Record an event-loop stall in seconds, with the severity and what caused it. */ +export function recordEventLoopBlocked(duration: number, severity: string, cause: string): void { + const attributes = jobAttrs(); + attributes.severity = severity; + attributes.cause = cause; + histogram( + 'lk.agents.event_loop.blocked_duration', + 'Duration of synchronous blocks detected on an agent event loop', + ).record(duration, attributes); +} + +/** `gen_ai.invoke_agent.duration` for one agent turn, in seconds. */ +export function recordInvokeAgentDuration(duration: number, agentName: string): void { + const attributes = jobAttrs(); + attributes[traceTypes.ATTR_GEN_AI_OPERATION_NAME] = traceTypes.GenAIOperationName.INVOKE_AGENT; + attributes[traceTypes.ATTR_GEN_AI_AGENT_NAME] = agentName; + histogram(traceTypes.METRIC_GEN_AI_INVOKE_AGENT_DURATION, 'Agent invocation duration').record( + duration, + attributes, + ); } diff --git a/agents/src/telemetry/trace_types.test.ts b/agents/src/telemetry/trace_types.test.ts index 7e7b8b73f..b10c5ae3d 100644 --- a/agents/src/telemetry/trace_types.test.ts +++ b/agents/src/telemetry/trace_types.test.ts @@ -132,6 +132,7 @@ const SAFE_KEYS = new Set([ 'lk.deployment_id', 'lk.session_options', 'lk.generation_id', + 'lk.generation_count', 'lk.parent_generation_id', 'lk.interrupted', // LLM node metadata @@ -298,7 +299,8 @@ const SAFE_KEYS = new Set([ function declaredKeys(): Record { return Object.fromEntries( Object.entries(traceTypes).filter((entry): entry is [string, string] => { - return typeof entry[1] === 'string'; + // metric names are not attribute keys: they carry no values to classify + return typeof entry[1] === 'string' && !entry[0].startsWith('METRIC_'); }), ); } diff --git a/agents/src/telemetry/trace_types.ts b/agents/src/telemetry/trace_types.ts index 594056cfd..ca9a4957a 100644 --- a/agents/src/telemetry/trace_types.ts +++ b/agents/src/telemetry/trace_types.ts @@ -120,8 +120,17 @@ export const ATTR_SHUTDOWN_USER_INITIATED = 'lk.shutdown.user_initiated'; export const ATTR_CALLBACK_NAME = 'lk.callback.name'; // assistant turn +/** + * On `agent_turn`: the latest generation (LLM step) of the speech; each step is also a + * `generation` event carrying its own id. + */ export const ATTR_AGENT_TURN_ID = 'lk.generation_id'; export const ATTR_AGENT_PARENT_TURN_ID = 'lk.parent_generation_id'; +/** + * On `agent_turn`: how many generations (LLM steps) the speech took; more than one means tool + * calls were executed before the final reply. + */ +export const ATTR_GENERATION_COUNT = 'lk.generation_count'; export const ATTR_USER_INPUT = 'lk.pii.user_input'; export const ATTR_INSTRUCTIONS = 'lk.pii.instructions'; export const ATTR_SPEECH_INTERRUPTED = 'lk.interrupted'; @@ -486,3 +495,7 @@ export const ATTR_EXCEPTION_MESSAGE = 'exception.message'; // Platform-specific attributes export const ATTR_LANGFUSE_COMPLETION_START_TIME = 'langfuse.observation.completion_start_time'; + +// metric names (OpenTelemetry GenAI semantic conventions) +/** Histogram, seconds: one agent turn (`invoke_agent`), however many LLM steps it took. */ +export const METRIC_GEN_AI_INVOKE_AGENT_DURATION = 'gen_ai.invoke_agent.duration'; diff --git a/agents/src/voice/agent_activity.test.ts b/agents/src/voice/agent_activity.test.ts index 8f82706ee..95d53d614 100644 --- a/agents/src/voice/agent_activity.test.ts +++ b/agents/src/voice/agent_activity.test.ts @@ -905,7 +905,12 @@ function buildPreemptiveRunner(opts: Partial = {}) { }; const generateReply = vi.fn( - () => ({ id: 'speech_fake', _cancel: () => {} }) as unknown as SpeechHandle, + () => + ({ + id: 'speech_fake', + _cancel: () => {}, + _takeAgentTurn: () => undefined, + }) as unknown as SpeechHandle, ); const cancelPreemptiveGeneration = vi.fn(); diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index eece4edd6..27fcfc20f 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -371,9 +371,10 @@ function recordQueueWait(speechHandle: SpeechHandle): void { if (queueWait === undefined || speechHandle._agentTurnContext === undefined) { return; // no agent_turn span yet: never fall back to whatever span is current } - trace - .getSpan(speechHandle._agentTurnContext) - ?.setAttribute(traceTypes.ATTR_SPEECH_QUEUE_WAIT, queueWait / 1000); + const span = trace.getSpan(speechHandle._agentTurnContext); + if (span?.isRecording()) { + span.setAttribute(traceTypes.ATTR_SPEECH_QUEUE_WAIT, queueWait / 1000); + } } /** @@ -384,12 +385,96 @@ function recordInterruption(speechHandle: SpeechHandle): void { if (!speechHandle.interrupted || speechHandle._agentTurnContext === undefined) { return; } - trace.getSpan(speechHandle._agentTurnContext)?.setAttributes({ + const span = trace.getSpan(speechHandle._agentTurnContext); + if (!span?.isRecording()) return; // the turn may have ended with the speech + span.setAttributes({ [traceTypes.ATTR_SPEECH_INTERRUPTED]: true, [traceTypes.ATTR_INTERRUPTION_SOURCE]: speechHandle._interruptSource ?? 'programmatic', }); } +/** + * Run `fn` under the speech's `agent_turn` span, made current for one generation. + * + * One speech handle is one agent turn, however many LLM steps it takes: the follow-up + * generation after a tool call runs in a new task but continues the open span instead of + * opening a second turn. Each generation is a `generation` event on the span, whose + * `lk.generation_id` names the latest one and `lk.generation_count` how many there were. The + * span ends with the speech (`SpeechHandle._markDone`), not with the step. + * + * Module-level for the same reason as `recordQueueWait`. + * @internal + */ +export async function withAgentTurn( + speechHandle: SpeechHandle, + options: { rootContext: Context | undefined; agentLabel: string }, + fn: (span: Span) => Promise, +): Promise { + let span = speechHandle._agentTurnSpan; + if (span === undefined) { + span = tracer.startSpan({ + name: 'agent_turn', + context: options.rootContext, + attributes: { [traceTypes.ATTR_SPEECH_ID]: speechHandle.id }, + }); + // an agent turn is the convention's `invoke_agent`: the framework running the agent + // in-process, with the inference and tool spans nested underneath + genAI.setAgentAttributes(span, { + operation: traceTypes.GenAIOperationName.INVOKE_AGENT, + agentName: options.agentLabel, + }); + speechHandle._agentTurnSpan = span; + speechHandle._agentTurnContext = trace.setSpan(options.rootContext ?? ROOT_CONTEXT, span); + speechHandle._agentTurnStartedAt = performance.now(); + speechHandle._agentTurnAgentName = options.agentLabel; + } + + const generationAttrs: Record = { + [traceTypes.ATTR_AGENT_TURN_ID]: speechHandle._generationId, + }; + const parentId = speechHandle._parentGenerationId; + if (parentId) generationAttrs[traceTypes.ATTR_AGENT_PARENT_TURN_ID] = parentId; + span.addEvent('generation', generationAttrs); + // the count is of generation events on the span, whichever speech emitted them: a turn + // continued from a discarded preemptive attempt or a tool call keeps counting + speechHandle._agentTurnGenerations += 1; + speechHandle._emittedGenerationStep = speechHandle._generationStep; + span.setAttributes({ + [traceTypes.ATTR_AGENT_TURN_ID]: speechHandle._generationId, + [traceTypes.ATTR_GENERATION_COUNT]: speechHandle._agentTurnGenerations, + }); + const turnSpan = span; + return otelContext.with(speechHandle._agentTurnContext!, () => fn(turnSpan)); +} + +/** + * A preemptive generation discarded for `successor` (a newer attempt, or the real reply after + * the transcript changed) hands its open `agent_turn` over, so one turn shows the wasted + * generation and the one that answered. Module-level like `recordQueueWait`. + * @internal + */ +export function continueDiscardedTurn( + discarded: SpeechHandle | undefined, + successor: SpeechHandle, +): void { + if (discarded === undefined || discarded === successor) return; + const carry = discarded._takeAgentTurn(); + if (carry !== undefined) successor._continueAgentTurn(carry, discarded); +} + +/** + * A realtime tool reply runs on a new speech handle after the tool calls of `speech`; Python + * runs it on the same handle as its next step. The reply continues the tool call's open + * `agent_turn`, with its generation numbered after (and parented to) the tool call's, so the + * trace shows one turn either way. Module-level like `continueDiscardedTurn`. + * @internal + */ +export function continueToolReplyTurn(speech: SpeechHandle, reply: SpeechHandle): void { + if (speech === reply) return; + const carry = speech._takeAgentTurn(); + if (carry !== undefined) reply._continueAgentTurn(carry, speech, 'tool_reply'); +} + export class AgentActivity implements RecognitionHooks { agent: Agent; agentSession: AgentSession; @@ -2211,7 +2296,10 @@ export class AgentActivity implements RecognitionHooks { return; } - // more of the user's turn arrived: the attempt answered a transcript that is now stale + // a newer attempt supersedes the current one; if one is created below it continues the + // discarded attempt's agent_turn (the cancelled speech only ends once the loop runs). More + // of the user's turn arrived: the attempt answered a transcript that is now stale + const discarded = this._preemptiveGeneration?.speechHandle; this.cancelPreemptiveGeneration('user_turn'); if ( @@ -2246,6 +2334,8 @@ export class AgentActivity implements RecognitionHooks { chatCtx, scheduleSpeech: false, inputDetails: { modality: 'audio' }, + // a newer attempt supersedes the current one: it continues that attempt's agent_turn + continueTurnFrom: discarded, }); this._preemptiveGeneration = { @@ -2398,7 +2488,16 @@ export class AgentActivity implements RecognitionHooks { } if (ownedSpeechHandle) { - return speechHandleStorage.run(ownedSpeechHandle, () => taskFn(ctrl)); + return speechHandleStorage.run(ownedSpeechHandle, () => + taskFn(ctrl).catch((error: unknown) => { + // the first failure of an owned task fails the speech (its agent_turn ends with + // the error and exception() reports it); a cancellation is not a failure + if ((error as Error | undefined)?.name !== 'AbortError') { + ownedSpeechHandle._error ??= error; + } + throw error; + }), + ); } return taskFn(ctrl); }); @@ -2804,6 +2903,12 @@ export class AgentActivity implements RecognitionHooks { allowInterruptions?: boolean; scheduleSpeech?: boolean; inputDetails?: InputDetails; + /** + * A discarded preemptive attempt answering the same user turn: the new speech continues + * its open `agent_turn` (see {@link continueDiscardedTurn}). Handed over before the reply + * task starts, since the task opens the turn in its first synchronous statements. + */ + continueTurnFrom?: SpeechHandle; }): SpeechHandle { const { userMessage, @@ -2813,6 +2918,7 @@ export class AgentActivity implements RecognitionHooks { allowInterruptions: defaultAllowInterruptions, scheduleSpeech = true, inputDetails, + continueTurnFrom, } = options; let instructions: string | Instructions | undefined = defaultInstructions; @@ -2851,6 +2957,9 @@ export class AgentActivity implements RecognitionHooks { allowInterruptions: allowInterruptions ?? this.allowInterruptions, inputDetails, }); + // before the reply task below runs (Task starts its body synchronously): the task must find + // the adopted turn on the handle, or it opens a second one that nothing ever ends + continueDiscardedTurn(continueTurnFrom, handle); this.agentSession.emit( AgentSessionEventTypes.SpeechCreated, @@ -3189,6 +3298,7 @@ export class AgentActivity implements RecognitionHooks { } let speechHandle: SpeechHandle | undefined; + let discardedPreemptive: SpeechHandle | undefined; if (this._preemptiveGeneration !== undefined) { const preemptive = this._preemptiveGeneration; // make sure the onUserTurnCompleted didn't change some request parameters @@ -3219,6 +3329,7 @@ export class AgentActivity implements RecognitionHooks { this.logger.warn( 'preemptive generation invalidated after `onUserTurnCompleted` because the transcript, chat context, tools, or tool choice changed', ); + discardedPreemptive = preemptive.speechHandle; preemptive.speechHandle._cancel('user_turn'); } @@ -3232,6 +3343,8 @@ export class AgentActivity implements RecognitionHooks { userMessage, chatCtx, inputDetails: { modality: 'audio' }, + // the invalidated preemptive attempt answered this same turn: one agent_turn + continueTurnFrom: discardedPreemptive, }); } @@ -3258,9 +3371,31 @@ export class AgentActivity implements RecognitionHooks { modelSettings: ModelSettings, replyAbortController: AbortController, audio?: ReadableStream | null, + ): Promise { + return withAgentTurn( + stateLease.speechHandle, + { rootContext: this.agentSession.rootSpanContext, agentLabel: this.agent.id }, + () => + this.ttsTaskImpl( + stateLease, + text, + addToChatCtx, + modelSettings, + replyAbortController, + audio, + ), + ); + } + + private async ttsTaskImpl( + stateLease: AgentStateLease, + text: string | ReadableStream, + addToChatCtx: boolean, + modelSettings: ModelSettings, + replyAbortController: AbortController, + audio?: ReadableStream | null, ): Promise { const { speechHandle } = stateLease; - speechHandle._agentTurnContext = otelContext.active(); speechHandleStorage.enterWith(speechHandle); @@ -3486,7 +3621,6 @@ export class AgentActivity implements RecognitionHooks { _previousUserMetrics?: MetricsReport; }): Promise => { const { speechHandle } = stateLease; - speechHandle._agentTurnContext = otelContext.active(); span.setAttribute(traceTypes.ATTR_SPEECH_ID, speechHandle.id); if (instructions) { @@ -3577,6 +3711,12 @@ export class AgentActivity implements RecognitionHooks { this.llm?.provider, ); tasks.push(llmTask); + // as python's _on_llm_task_done: a genuine LLM failure (not a cancellation) fails the + // speech, through exception() and the agent_turn span. Nothing else awaits this task's + // rejection: the pipeline reads the node's streams, not its result + void llmTask.result.catch((error: unknown) => { + if ((error as Error | undefined)?.name !== 'AbortError') speechHandle._error ??= error; + }); interface SpeechSegment { textStream: ReadableStream; @@ -4134,17 +4274,6 @@ export class AgentActivity implements RecognitionHooks { } }; - /** - * An agent turn is the convention's `invoke_agent`: the framework running the agent - * in-process, with the inference (`chat`) and tool (`execute_tool`) spans nested underneath. - */ - private recordAgentTurn(span: Span): void { - genAI.setAgentAttributes(span, { - operation: traceTypes.GenAIOperationName.INVOKE_AGENT, - agentName: this.agent.id, - }); - } - private pipelineReplyTask = async ( stateLease: AgentStateLease, chatCtx: ChatContext, @@ -4155,9 +4284,10 @@ export class AgentActivity implements RecognitionHooks { newMessage?: ChatMessage, _previousUserMetrics?: MetricsReport, ): Promise => - tracer.startActiveSpan( - async (span) => ( - this.recordAgentTurn(span), + withAgentTurn( + stateLease.speechHandle, + { rootContext: this.agentSession.rootSpanContext, agentLabel: this.agent.id }, + (span) => this._pipelineReplyTaskImpl({ stateLease, chatCtx, @@ -4168,12 +4298,7 @@ export class AgentActivity implements RecognitionHooks { newMessage, span, _previousUserMetrics, - }) - ), - { - name: 'agent_turn', - context: this.agentSession.rootSpanContext, - }, + }), ); private async realtimeGenerationTask( @@ -4183,9 +4308,10 @@ export class AgentActivity implements RecognitionHooks { replyAbortController: AbortController, addToChatCtx: boolean = true, ): Promise { - return tracer.startActiveSpan( + return withAgentTurn( + stateLease.speechHandle, + { rootContext: this.agentSession.rootSpanContext, agentLabel: this.agent.id }, async (span) => { - this.recordAgentTurn(span); const inferenceSpan = tracer.startSpan({ name: 'realtime_inference' }); try { return await this._realtimeGenerationTaskImpl({ @@ -4201,10 +4327,6 @@ export class AgentActivity implements RecognitionHooks { inferenceSpan.end(); } }, - { - name: 'agent_turn', - context: this.agentSession.rootSpanContext, - }, ); } @@ -4226,7 +4348,6 @@ export class AgentActivity implements RecognitionHooks { inferenceSpan: Span; }): Promise { const { speechHandle } = stateLease; - speechHandle._agentTurnContext = otelContext.active(); span.setAttribute(traceTypes.ATTR_SPEECH_ID, speechHandle.id); @@ -4814,6 +4935,8 @@ export class AgentActivity implements RecognitionHooks { stepIndex: speechHandle.numSteps + 1, parent: speechHandle, }); + // one agent_turn for the tool call and its reply, as when they share a handle + continueToolReplyTurn(speechHandle, replySpeechHandle); this.agentSession.emit( AgentSessionEventTypes.SpeechCreated, createSpeechCreatedEvent({ diff --git a/agents/src/voice/agent_turn_span.test.ts b/agents/src/voice/agent_turn_span.test.ts new file mode 100644 index 000000000..08839437c --- /dev/null +++ b/agents/src/voice/agent_turn_span.test.ts @@ -0,0 +1,437 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +/** + * One `agent_turn` span per speech handle. + * + * A reply that calls a tool runs two generations (LLM steps) in two tasks; they used to be two + * `agent_turn` spans linked only by `lk.parent_generation_id`. The speech handle now owns a + * single span for its whole life: each generation is an event on it, tool and inference spans + * nest under it, and it ends with the speech. + */ +import { AudioFrame } from '@livekit/rtc-node'; +import { + INVALID_SPAN_CONTEXT, + ROOT_CONTEXT, + SpanStatusCode, + context as otelContext, + trace, +} from '@opentelemetry/api'; +import { + InMemorySpanExporter, + type ReadableSpan, + SimpleSpanProcessor, +} from '@opentelemetry/sdk-trace-base'; +import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node'; +import { ReadableStream } from 'node:stream/web'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { ChatMessage } from '../llm/chat_context.js'; +import { tool } from '../llm/tool_context.js'; +import { initializeLogger } from '../log.js'; +import { FakeSTT } from '../stt/testing/fake_stt.js'; +import { setTracerProvider, traceTypes, tracer } from '../telemetry/index.js'; +import * as otelMetrics from '../telemetry/otel_metrics.js'; +import { Agent } from './agent.js'; +import { continueDiscardedTurn, continueToolReplyTurn, withAgentTurn } from './agent_activity.js'; +import { AgentSession } from './agent_session.js'; +import { AudioOutput } from './io.js'; +import { SpeechHandle } from './speech_handle.js'; +import { FakeLLM } from './testing/fake_llm.js'; + +initializeLogger({ pretty: false, level: 'silent' }); + +function spansNamed(exporter: InMemorySpanExporter, name: string): ReadableSpan[] { + return exporter.getFinishedSpans().filter((span) => span.name === name); +} + +function childrenOf(exporter: InMemorySpanExporter, parent: ReadableSpan, name: string) { + return spansNamed(exporter, name).filter( + (span) => span.parentSpanContext?.spanId === parent.spanContext().spanId, + ); +} + +function ms(time: [number, number]): number { + return time[0] * 1000 + time[1] / 1e6; +} + +/** Reports playout as soon as frames arrive, so a reply "plays" instantly. */ +class ImmediateOutput extends AudioOutput { + constructor() { + super(24_000); + } + + override async captureFrame(frame: AudioFrame): Promise { + 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 WeatherAgent extends Agent { + constructor() { + super({ + instructions: 'You are a helpful assistant.', + tools: { + get_weather: tool({ + description: 'Look up the weather', + execute: async () => 'sunny in Tokyo', + }), + }, + }); + } + + override async ttsNode(): Promise> { + return new ReadableStream({ + start(controller) { + controller.enqueue(new AudioFrame(new Int16Array(480), 24_000, 1, 480)); + controller.close(); + }, + }); + } +} + +describe.sequential('agent_turn span', () => { + let exporter: InMemorySpanExporter; + let provider: NodeTracerProvider; + let originalProvider: ReturnType; + + beforeEach(() => { + originalProvider = tracer.getProvider(); + exporter = new InMemorySpanExporter(); + provider = new NodeTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] }); + provider.register(); + setTracerProvider(provider); + }); + + afterEach(async () => { + vi.restoreAllMocks(); + setTracerProvider(originalProvider); + await provider.shutdown(); + trace.disable(); + otelContext.disable(); + }); + + async function runReply(llm: FakeLLM, agent: Agent, userInput: string): Promise { + const session = new AgentSession({ llm, stt: new FakeSTT() }); + session.output.audio = new ImmediateOutput(); + await session.start({ agent }); + try { + const speech = session.generateReply({ userInput }); + await speech.waitForPlayout(); + } finally { + await session.close(); + } + } + + it('a tool call is one agent turn', async () => { + const llm = new FakeLLM([ + { + input: "What's the weather in Tokyo?", + content: '', + toolCalls: [{ name: 'get_weather', args: { location: 'Tokyo' } }], + }, + { input: '"sunny in Tokyo"', content: 'It is sunny in Tokyo.' }, + ]); + await runReply(llm, new WeatherAgent(), "What's the weather in Tokyo?"); + + const [root] = spansNamed(exporter, 'agent_session'); + const turns = spansNamed(exporter, 'agent_turn'); + expect( + turns.map((turn) => turn.attributes[traceTypes.ATTR_SPEECH_ID]), + 'one agent_turn per speech', + ).toHaveLength(1); + const [turn] = turns; + expect(turn!.parentSpanContext?.spanId).toBe(root!.spanContext().spanId); + + const attrs = turn!.attributes; + const speechId = attrs[traceTypes.ATTR_SPEECH_ID] as string; + expect(attrs[traceTypes.ATTR_GENERATION_COUNT]).toBe(2); + expect(attrs[traceTypes.ATTR_AGENT_TURN_ID]).toBe(`${speechId}_2`); + const generations = turn!.events.filter((event) => event.name === 'generation'); + expect(generations.map((event) => event.attributes?.[traceTypes.ATTR_AGENT_TURN_ID])).toEqual([ + `${speechId}_1`, + `${speechId}_2`, + ]); + expect(generations[0]!.attributes?.[traceTypes.ATTR_AGENT_PARENT_TURN_ID]).toBeUndefined(); + expect(generations[1]!.attributes?.[traceTypes.ATTR_AGENT_PARENT_TURN_ID]).toBe( + `${speechId}_1`, + ); + + // both generations' inference, the tool between them, and the speech all nest under it + expect(childrenOf(exporter, turn!, 'llm_node')).toHaveLength(2); + const [toolSpan] = childrenOf(exporter, turn!, 'function_tool'); + const [tts] = childrenOf(exporter, turn!, 'tts_node'); + const [speaking] = childrenOf(exporter, turn!, 'agent_speaking'); + expect(toolSpan).toBeDefined(); + expect(tts).toBeDefined(); + expect(speaking).toBeDefined(); + expect(ms(toolSpan!.startTime)).toBeLessThan(ms(tts!.startTime)); + // and the turn covers everything, ending with the speech rather than with the first step + // (2 ms of slack: the SDK anchors each span's clock at creation) + for (const child of [toolSpan!, tts!, speaking!]) { + expect(ms(turn!.startTime)).toBeLessThanOrEqual(ms(child.startTime) + 2); + expect(ms(child.endTime)).toBeLessThanOrEqual(ms(turn!.endTime) + 2); + } + }); + + it('a plain reply is one generation', async () => { + const llm = new FakeLLM([{ input: 'Hello', content: 'Hi there' }]); + await runReply(llm, new WeatherAgent(), 'Hello'); + + const [turn] = spansNamed(exporter, 'agent_turn'); + expect(spansNamed(exporter, 'agent_turn')).toHaveLength(1); + const attrs = turn!.attributes; + expect(attrs[traceTypes.ATTR_GENERATION_COUNT]).toBe(1); + expect(attrs[traceTypes.ATTR_AGENT_TURN_ID]).toBe(`${attrs[traceTypes.ATTR_SPEECH_ID]}_1`); + expect(turn!.events.filter((event) => event.name === 'generation')).toHaveLength(1); + expect(attrs[traceTypes.ATTR_AGENT_PARENT_TURN_ID]).toBeUndefined(); + }); + + it('a discarded preemptive generation hands its turn to the successor', async () => { + // a preemptive attempt discarded for the real reply (or a newer attempt) must not leave a + // second agent_turn behind: the successor continues the span, the discarded speech ends + // without touching it + const root = tracer.startSpan({ name: 'agent_session' }); + const rootCtx = trace.setSpan(ROOT_CONTEXT, root); + const attempt = SpeechHandle.create({ allowInterruptions: true }); + await withAgentTurn(attempt, { rootContext: rootCtx, agentLabel: 'a' }, async () => { + // the attempt's first generation ran here + }); + + const reply = SpeechHandle.create({ allowInterruptions: true }); + continueDiscardedTurn(attempt, reply); + attempt._markDone(); // the cancelled attempt finishes: the span must survive it + expect(spansNamed(exporter, 'agent_turn')).toEqual([]); + + await withAgentTurn(reply, { rootContext: rootCtx, agentLabel: 'a' }, async () => {}); + reply._markDone(); + root.end(); + + const [turn] = spansNamed(exporter, 'agent_turn'); + expect(spansNamed(exporter, 'agent_turn')).toHaveLength(1); + const attrs = turn!.attributes; + expect(attrs[traceTypes.ATTR_SPEECH_ID]).toBe(reply.id); + // the discarded attempt's generation and the reply's: the count is of the turn, not the handle + expect(attrs[traceTypes.ATTR_GENERATION_COUNT]).toBe(2); + expect(turn!.events.map((event) => event.name)).toEqual([ + 'generation', + 'preemptive_generation_discarded', + 'generation', + ]); + const discarded = turn!.events.find( + (event) => event.name === 'preemptive_generation_discarded', + ); + expect(discarded?.attributes?.[traceTypes.ATTR_SPEECH_ID]).toBe(attempt.id); + + // nothing to hand over: a plain successor is untouched + continueDiscardedTurn(undefined, reply); + continueDiscardedTurn(reply, reply); + }); + + it('hands the turn over before the successor task opens its own', async () => { + // Task runs its body synchronously, and the reply task opens agent_turn in its first + // statements: a handoff performed after generateReply() returned came too late, leaving the + // successor's own span unended and its llm_node / tts_node dangling from it. The handoff now + // happens inside generateReply, before the task starts. + const llm = new FakeLLM([{ input: 'Hello', content: 'Hi there' }]); + const session = new AgentSession({ llm, stt: new FakeSTT() }); + session.output.audio = new ImmediateOutput(); + const agent = new WeatherAgent(); + await session.start({ agent }); + try { + const activity = agent._agentActivity!; + const attempt = SpeechHandle.create({ allowInterruptions: true }); + await withAgentTurn( + attempt, + { rootContext: session.rootSpanContext, agentLabel: agent.id }, + async () => {}, + ); + const reply = activity.generateReply({ + userMessage: ChatMessage.create({ role: 'user', content: 'Hello' }), + inputDetails: { modality: 'audio' }, + continueTurnFrom: attempt, + }); + attempt._markDone(); + await reply.waitForPlayout(); + } finally { + await session.close(); + } + + const turns = spansNamed(exporter, 'agent_turn'); + expect(turns).toHaveLength(1); + const [turn] = turns; + expect(turn!.attributes[traceTypes.ATTR_GENERATION_COUNT]).toBe(2); + expect(turn!.events.map((event) => event.name)).toEqual([ + 'generation', + 'preemptive_generation_discarded', + 'generation', + ]); + // the successor's work nests under the adopted turn + expect(childrenOf(exporter, turn!, 'llm_node')).toHaveLength(1); + expect(childrenOf(exporter, turn!, 'tts_node')).toHaveLength(1); + expect(childrenOf(exporter, turn!, 'agent_speaking')).toHaveLength(1); + // and nothing dangles from a span that never ended + const exported = new Set(exporter.getFinishedSpans().map((span) => span.spanContext().spanId)); + const dangling = exporter + .getFinishedSpans() + .filter((span) => span.parentSpanContext && !exported.has(span.parentSpanContext.spanId)); + expect(dangling.map((span) => span.name)).toEqual([]); + }); + + it('keeps counting generations across repeated handoffs', async () => { + // attempt A is replaced by attempt B, which is replaced by the reply: three generations on + // one turn, and the count says so however many hands the turn went through + const root = tracer.startSpan({ name: 'agent_session' }); + const rootCtx = trace.setSpan(ROOT_CONTEXT, root); + const opts = { rootContext: rootCtx, agentLabel: 'a' }; + const a = SpeechHandle.create({ allowInterruptions: true }); + await withAgentTurn(a, opts, async () => {}); + const b = SpeechHandle.create({ allowInterruptions: true }); + continueDiscardedTurn(a, b); + a._markDone(); + await withAgentTurn(b, opts, async () => {}); + const reply = SpeechHandle.create({ allowInterruptions: true }); + continueDiscardedTurn(b, reply); + b._markDone(); + await withAgentTurn(reply, opts, async () => {}); + reply._markDone(); + root.end(); + + const [turn] = spansNamed(exporter, 'agent_turn'); + expect(spansNamed(exporter, 'agent_turn')).toHaveLength(1); + expect(turn!.attributes[traceTypes.ATTR_GENERATION_COUNT]).toBe(3); + expect(turn!.events.filter((event) => event.name === 'generation')).toHaveLength(3); + // each speech still numbers its own generations, as python does + expect(turn!.attributes[traceTypes.ATTR_AGENT_TURN_ID]).toBe(`${reply.id}_1`); + }); + + it('a realtime tool reply continues the tool call turn as its next generation', async () => { + // the framework runs the reply on a new handle; python runs it on the same one as step 2. + // Either way the trace is one turn: the reply's generation numbered after the tool call's + // and parented to it, under the tool call's speech id + const root = tracer.startSpan({ name: 'agent_session' }); + const rootCtx = trace.setSpan(ROOT_CONTEXT, root); + const opts = { rootContext: rootCtx, agentLabel: 'a' }; + const speech = SpeechHandle.create({ allowInterruptions: true }); + let reply: SpeechHandle | undefined; + await withAgentTurn(speech, opts, async () => { + // the tool calls ran; the framework creates the reply inside the tool call's turn + speech._numSteps += 1; // as the realtime path does before scheduling the reply + reply = SpeechHandle.create({ allowInterruptions: true, parent: speech }); + continueToolReplyTurn(speech, reply); + }); + speech._markDone(); // the tool call's own handle ends: the span must survive it + expect(spansNamed(exporter, 'agent_turn')).toEqual([]); + await withAgentTurn(reply!, opts, async () => {}); + reply!._markDone(); + root.end(); + + const turns = spansNamed(exporter, 'agent_turn'); + expect(turns).toHaveLength(1); + const [turn] = turns; + const attrs = turn!.attributes; + expect(attrs[traceTypes.ATTR_SPEECH_ID]).toBe(speech.id); + expect(attrs[traceTypes.ATTR_GENERATION_COUNT]).toBe(2); + expect(attrs[traceTypes.ATTR_AGENT_TURN_ID]).toBe(`${speech.id}_2`); + const generations = turn!.events.filter((event) => event.name === 'generation'); + expect(generations.map((event) => event.attributes?.[traceTypes.ATTR_AGENT_TURN_ID])).toEqual([ + `${speech.id}_1`, + `${speech.id}_2`, + ]); + expect(generations[1]!.attributes?.[traceTypes.ATTR_AGENT_PARENT_TURN_ID]).toBe( + `${speech.id}_1`, + ); + expect(turn!.events.some((event) => event.name === 'preemptive_generation_discarded')).toBe( + false, + ); + // no handoff to make: the same handle + continueToolReplyTurn(reply!, reply!); + }); + + it('a task failure fails the turn and surfaces on the handle', async () => { + // an LLM node that throws rejects the speech task; the turn ends with the error and the + // handle reports it, instead of an unremarkable success + class BrokenAgent extends WeatherAgent { + override async llmNode(): Promise { + throw new Error('provider unavailable'); + } + } + const llm = new FakeLLM([{ input: 'Hello', content: 'Hi there' }]); + const session = new AgentSession({ llm, stt: new FakeSTT() }); + session.output.audio = new ImmediateOutput(); + await session.start({ agent: new BrokenAgent() }); + let speech: SpeechHandle | undefined; + try { + speech = session.generateReply({ userInput: 'Hello' }); + await speech.waitForPlayout(); + } finally { + await session.close(); + } + + expect(speech!.exception()).toBeInstanceOf(Error); + expect((speech!.exception() as Error).message).toBe('provider unavailable'); + const [turn] = spansNamed(exporter, 'agent_turn'); + expect(turn!.status.code).toBe(SpanStatusCode.ERROR); + expect( + turn!.events.find((event) => event.name === 'exception')?.attributes?.['exception.message'], + ).toBe('provider unavailable'); + }); + + it('an LLM failure stored on the handle fails the turn', async () => { + // the pipeline stores the failure on the handle when it marks it done; the turn must end as + // failed whichever step it was on + const handle = SpeechHandle.create({ allowInterruptions: true }); + await withAgentTurn(handle, { rootContext: undefined, agentLabel: 'a' }, async () => {}); + handle._markDone(new Error('llm down')); + + const [turn] = spansNamed(exporter, 'agent_turn'); + expect(turn!.status.code).toBe(SpanStatusCode.ERROR); + expect(turn!.events.filter((event) => event.name === 'exception')).toHaveLength(1); + }); + + it('records the turn duration metric even when the span is sampled out', () => { + const record = vi.spyOn(otelMetrics, 'recordInvokeAgentDuration').mockImplementation(() => {}); + const handle = SpeechHandle.create({ allowInterruptions: true }); + handle._agentTurnSpan = trace.wrapSpanContext(INVALID_SPAN_CONTEXT); + handle._agentTurnStartedAt = performance.now() - 1_000; + handle._agentTurnAgentName = 'a'; + handle._markDone(); + expect(record).toHaveBeenCalledTimes(1); + const [duration, agentName] = record.mock.calls[0]!; + expect(agentName).toBe('a'); + expect(duration).toBeGreaterThanOrEqual(1); + expect(duration).toBeLessThan(5); + }); + + it('hands a sampled-out turn to the successor too', () => { + // the successor adopts a non-recording turn as well, so the duration metric keeps the + // discarded attempt's start time + const attempt = SpeechHandle.create({ allowInterruptions: true }); + attempt._agentTurnSpan = trace.wrapSpanContext(INVALID_SPAN_CONTEXT); + attempt._agentTurnStartedAt = 1; + attempt._agentTurnAgentName = 'a'; + + const reply = SpeechHandle.create({ allowInterruptions: true }); + continueDiscardedTurn(attempt, reply); + expect(attempt._agentTurnSpan).toBeUndefined(); + expect(reply._agentTurnSpan).toBeDefined(); + expect(reply._agentTurnStartedAt).toBe(1); + expect(reply._agentTurnAgentName).toBe('a'); + }); +}); diff --git a/agents/src/voice/speech_handle.ts b/agents/src/voice/speech_handle.ts index 37160a766..5bea164db 100644 --- a/agents/src/voice/speech_handle.ts +++ b/agents/src/voice/speech_handle.ts @@ -2,9 +2,12 @@ // // SPDX-License-Identifier: Apache-2.0 import { ThrowsPromise } from '@livekit/throws-transformer/throws'; -import type { Context } from '@opentelemetry/api'; +import { type Context, type Span, context as otelContext, trace } from '@opentelemetry/api'; import type { ChatItem } from '../llm/index.js'; import { log } from '../log.js'; +import { recordInvokeAgentDuration } from '../telemetry/otel_metrics.js'; +import * as traceTypes from '../telemetry/trace_types.js'; +import { recordException } from '../telemetry/utils.js'; import type { Task } from '../utils.js'; import { Event, Future, dedent, shortuuid } from '../utils.js'; import { functionCallStorage } from './agent.js'; @@ -96,6 +99,18 @@ export type ResolvedSpeechHandle = Omit; */ export type InterruptionSource = 'audio_activity' | 'user_turn' | 'programmatic'; +/** An open `agent_turn` handed from a discarded speech to its successor. @internal */ +export interface AgentTurnCarry { + span: Span; + startedAt: number | undefined; + agentName: string | undefined; + /** Generation events already on the span, so `lk.generation_count` keeps counting. */ + generations: number; +} + +/** How a speech came to continue another's `agent_turn` (see `SpeechHandle._continueAgentTurn`). */ +export type AgentTurnContinuation = 'preemptive_discarded' | 'tool_reply'; + export class SpeechHandleCircularWaitError extends Error { constructor(functionCallName: string) { super(dedent` @@ -139,7 +154,8 @@ export class SpeechHandle { private doneFut = new Future(); private generations: Future[] = []; private _chatItems: ChatItem[] = []; - private _error: unknown; + /** @internal The first failure of an owned task, or the one the pipeline stored; see exception(). */ + _error: unknown; private interruptionHolds = 0; private interruptionHoldsRestore: boolean; @@ -148,9 +164,30 @@ export class SpeechHandle { /** @internal */ _numSteps = 1; + /** + * @internal Generation ids continue another speech's numbering when this speech carries on + * its turn (a realtime tool reply, which the framework runs on a new handle): the base id and + * the step the other speech had reached. + */ + _generationBaseId?: string; + /** @internal */ + _generationStepBase = 0; + /** @internal The step of the last generation event this speech emitted on its turn. */ + _emittedGenerationStep?: number; + /** @internal Generation events on the turn this speech owns, carried over on a handoff. */ + _agentTurnGenerations = 0; + /** + * @internal One `agent_turn` span for the whole speech, however many generations (LLM steps) + * it takes; opened by the first reply task, ended with the speech in `_markDone`. + */ + _agentTurnSpan?: Span; /** @internal - OpenTelemetry context for the agent turn span */ _agentTurnContext?: Context; + /** @internal - when the turn opened (performance.now), for the duration metric */ + _agentTurnStartedAt?: number; + /** @internal - the agent the turn was opened for, for the duration metric */ + _agentTurnAgentName?: string; /** @internal - when the speech was scheduled, for the queue-wait attribute */ _scheduledAt?: number; @@ -224,6 +261,23 @@ export class SpeechHandle { return this._id; } + /** @internal The step of the current generation in the turn's numbering (see `_generationBaseId`). */ + get _generationStep(): number { + return this._generationStepBase + this._numSteps; + } + + /** @internal The id of the current generation (LLM step) of this speech. */ + get _generationId(): string { + return `${this._generationBaseId ?? this._id}_${this._generationStep}`; + } + + /** @internal The id of the generation before the current one; undefined on the first. */ + get _parentGenerationId(): string | undefined { + const step = this._generationStep; + if (step <= 1) return undefined; + return `${this._generationBaseId ?? this._id}_${step - 1}`; + } + get scheduled(): boolean { return this.scheduledFut.done; } @@ -528,6 +582,8 @@ export class SpeechHandle { } this.doneFut.resolve(); } + // a pipeline LLM failure is stored on the handle before the tasks finish + this.endAgentTurn(error !== undefined ? error : this._error); // Keep this outside the doneFut guard: if the handle is already done but a // generation future is still active, _waitForGeneration() must be released. @@ -538,6 +594,91 @@ export class SpeechHandle { this.clearInterruptTimeout(); } + /** + * Detach this speech's open `agent_turn` so a successor can continue it. + * + * Used when a preemptive generation is discarded for another speech answering the same user + * turn: the wasted generation stays visible under the one turn instead of becoming a turn of + * its own. After this the speech ends without touching the span. + * @internal + */ + _takeAgentTurn(): AgentTurnCarry | undefined { + const span = this._agentTurnSpan; + if (span === undefined) return undefined; + const carry: AgentTurnCarry = { + span, + startedAt: this._agentTurnStartedAt, + agentName: this._agentTurnAgentName, + generations: this._agentTurnGenerations, + }; + this._agentTurnSpan = undefined; + this._agentTurnContext = undefined; + this._agentTurnStartedAt = undefined; + this._agentTurnAgentName = undefined; + this._agentTurnGenerations = 0; + return carry; + } + + /** + * @internal Adopt the `agent_turn` taken from `from` (see {@link _takeAgentTurn}). + * + * - `preemptive_discarded` (the default): `from` was a preemptive attempt dropped for this + * speech; the span records that and takes this speech's id. Generation ids stay this + * speech's own, as in Python. + * - `tool_reply`: this speech is the realtime tool reply the framework runs on a new handle + * after `from`'s tool calls; Python runs it on the same handle as its next step. The turn + * keeps `from`'s speech id and this speech's generations continue `from`'s numbering, so the + * trace reads as Python's: one turn, the reply's generation parented to the tool call's. + */ + _continueAgentTurn( + carry: AgentTurnCarry, + from: SpeechHandle, + continuation: AgentTurnContinuation = 'preemptive_discarded', + ): void { + // adopted even when sampled out: the duration metric still needs the start time + const { span, startedAt, agentName } = carry; + const own = this._agentTurnSpan; + if (own !== undefined && own !== span) { + // this speech already opened a turn of its own (the handoff came after its task started): + // close it rather than leak an unended span that its children would dangle from + own.addEvent('superseded_by_adopted_turn', { [traceTypes.ATTR_SPEECH_ID]: from.id }); + if (own.isRecording()) own.end(); + } + if (continuation === 'preemptive_discarded') { + span.addEvent('preemptive_generation_discarded', { + [traceTypes.ATTR_SPEECH_ID]: from.id, + }); + span.setAttribute(traceTypes.ATTR_SPEECH_ID, this.id); + } else { + this._generationBaseId = from._generationBaseId ?? from.id; + this._generationStepBase = from._emittedGenerationStep ?? from._generationStep; + } + this._agentTurnSpan = span; + this._agentTurnContext = trace.setSpan(otelContext.active(), span); + this._agentTurnStartedAt = startedAt; + this._agentTurnAgentName = agentName; + this._agentTurnGenerations = carry.generations; + } + + /** Close the speech's `agent_turn` span: the speech is done, whatever step it was on. */ + private endAgentTurn(error: unknown): void { + const span = this._agentTurnSpan; + this._agentTurnSpan = undefined; + if (span === undefined) return; + // the duration metric does not depend on the span being sampled in + if (this._agentTurnStartedAt !== undefined && this._agentTurnAgentName !== undefined) { + recordInvokeAgentDuration( + (performance.now() - this._agentTurnStartedAt) / 1000, + this._agentTurnAgentName, + ); + } + if (!span.isRecording()) return; + if (error instanceof Error) { + recordException(span, error); + } + span.end(); + } + /** @internal */ _markScheduled(): void { this._scheduledAt ??= performance.now();