feat(telemetry): one agent_turn span per speech handle

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

agents/etc/agents.api.md

@@ -1133,7 +1133,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)
@@ -1636,6 +1636,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)
@@ -6135,6 +6140,11 @@ class MetadataLogProcessor implements LogRecordProcessor {
     shutdown(): Promise<void>;
 }
 
+// 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,
@@ -7963,7 +7973,13 @@ export class SpeechHandle {
     // @internal (undocumented)
     _addItemAddedCallback(callback: (item: ChatItem) => void): void;
     // @internal
+    _agentTurnAgentName?: string;
+    // @internal
     _agentTurnContext?: Context;
+    // @internal
+    _agentTurnSpan?: Span;
+    // @internal
+    _agentTurnStartedAt?: number;
     // (undocumented)
     get allowInterruptions(): boolean;
     set allowInterruptions(value: boolean);
@@ -7977,6 +7993,10 @@ export class SpeechHandle {
     get chatItems(): ChatItem[];
     // @internal (undocumented)
     _clearAuthorization(): void;
+    // 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, discarded: SpeechHandle): void;
     // (undocumented)
     static create(options?: {
         allowInterruptions?: boolean;
@@ -7987,6 +8007,8 @@ export class SpeechHandle {
     // (undocumented)
     done(): boolean;
     exception(): unknown;
+    // @internal
+    get _generationId(): string;
     // @internal (undocumented)
     get _hasGenerations(): boolean;
     // @internal (undocumented)
@@ -8017,6 +8039,8 @@ export class SpeechHandle {
     // (undocumented)
     readonly parent?: SpeechHandle | undefined;
     // @internal
+    get _parentGenerationId(): string | undefined;
+    // @internal
     _queueWait(): number | undefined;
     // @internal (undocumented)
     _releaseInterruptions(): void;
@@ -8033,6 +8057,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<void>[];
     then<R1 = ResolvedSpeechHandle, R2 = never>(onFulfilled?: ((value: ResolvedSpeechHandle) => R1 | PromiseLike<R1>) | null, onRejected?: ((reason: unknown) => R2 | PromiseLike<R2>) | null): Promise<R1 | R2>;
@@ -9368,6 +9396,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,
@@ -9497,7 +9526,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
     }
 }
 

agents/src/telemetry/otel_metrics.ts

@@ -3,34 +3,69 @@
 // 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<string, Histogram>();
 
-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;
     if (roomId) attributes.room_id = roomId;
     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,
+  );
 }

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
@@ -299,7 +300,8 @@ const SAFE_KEYS = new Set([
 function declaredKeys(): Record<string, string> {
   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_');
     }),
   );
 }

agents/src/telemetry/trace_types.ts

@@ -122,8 +122,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';
@@ -488,3 +497,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';

agents/src/voice/agent_activity.test.ts

@@ -905,7 +905,12 @@ function buildPreemptiveRunner(opts: Partial<PreemptiveOpts> = {}) {
   };
 
   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();