diff --git a/.changeset/bright-spans-correlate.md b/.changeset/bright-spans-correlate.md new file mode 100644 index 0000000000..b2bcf1bb89 --- /dev/null +++ b/.changeset/bright-spans-correlate.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Correlate telemetry spans from the same agent conversation with the LiveKit room SID. diff --git a/agents/src/ipc/job_trace.test.ts b/agents/src/ipc/job_trace.test.ts index 6e8f909535..711dfc972a 100644 --- a/agents/src/ipc/job_trace.test.ts +++ b/agents/src/ipc/job_trace.test.ts @@ -12,7 +12,7 @@ import { } from '@opentelemetry/sdk-trace-base'; import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import { JobContext, JobProcess, type RunningJobInfo } from '../job.js'; +import { JobContext, JobProcess, type RunningJobInfo, getJobContext } from '../job.js'; import { initializeLogger } from '../log.js'; import { setTracerProvider, traceTypes, tracer } from '../telemetry/index.js'; import type { InferenceExecutor } from './inference_executor.js'; @@ -176,6 +176,7 @@ describe('job dispatch telemetry', () => { assignedAt: T0 + 500, launchedAt: T0 + 600, }); + expect(getJobContext(false)).toBeUndefined(); const span = startJobSpan(jobContext(running), T0 + 1000); expect(span.isRecording()).toBe(true); // a child created under it (the session, a connect) nests: the root is a real span @@ -189,6 +190,7 @@ describe('job dispatch telemetry', () => { expect(root!.attributes[traceTypes.ATTR_WORKER_ID]).toBe('W_1'); expect(root!.attributes[traceTypes.ATTR_JOB_AGENT_ID]).toBe('AG_1'); expect(root!.attributes[traceTypes.ATTR_ROOM_SID]).toBe('RM_1'); + expect(root!.attributes[traceTypes.ATTR_GEN_AI_CONVERSATION_ID]).toBe('RM_1'); expect(root!.attributes[traceTypes.ATTR_JOB_ACCEPT_LATENCY]).toBeCloseTo(0.2); expect(eventNames(root!).slice(0, 4)).toEqual([ 'job_received', @@ -200,6 +202,18 @@ describe('job dispatch telemetry', () => { expect(session!.parentSpanContext?.spanId).toBe(root!.spanContext().spanId); }); + it.each([undefined, new ProtoRoom({ sid: '' })])( + 'omits the conversation ID when the job room has no SID (%j)', + (room) => { + const running = info(); + running.job.room = room; + startJobSpan(jobContext(running)).end(); + + const [root] = spans(exporter, 'job_entrypoint'); + expect(root!.attributes).not.toHaveProperty(traceTypes.ATTR_GEN_AI_CONVERSATION_ID); + }, + ); + it('reads server timestamps in nanoseconds, milliseconds, or seconds', () => { expect(serverTimestampMs(1_700_000_000_123_456_789n)).toBeCloseTo(1_700_000_000_123.456, 2); expect(serverTimestampMs(1_700_000_000_123)).toBe(1_700_000_000_123); diff --git a/agents/src/ipc/job_trace.ts b/agents/src/ipc/job_trace.ts index 8b6c2a801e..1ced077561 100644 --- a/agents/src/ipc/job_trace.ts +++ b/agents/src/ipc/job_trace.ts @@ -25,6 +25,7 @@ export function startJobSpan( [traceTypes.ATTR_AGENT_NAME]: job.agentName, [traceTypes.ATTR_ROOM_NAME]: job.room?.name ?? '', [traceTypes.ATTR_ROOM_SID]: job.room?.sid ?? '', + [traceTypes.ATTR_GEN_AI_CONVERSATION_ID]: job.room?.sid || undefined, [traceTypes.ATTR_DISPATCH_ID]: job.dispatchId, [traceTypes.ATTR_WORKER_ID]: info.workerId, [traceTypes.ATTR_JOB_AGENT_ID]: job.state?.agentId ?? '', diff --git a/agents/src/telemetry/conversation_id.test.ts b/agents/src/telemetry/conversation_id.test.ts new file mode 100644 index 0000000000..2c413d2a98 --- /dev/null +++ b/agents/src/telemetry/conversation_id.test.ts @@ -0,0 +1,142 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { Attributes, Span } from '@opentelemetry/api'; +import { InMemorySpanExporter, SimpleSpanProcessor } from '@opentelemetry/sdk-trace-base'; +import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { type JobContext, runWithJobContext } from '../job.js'; +import * as genAI from './gen_ai.js'; +import { setTracerProvider, tracer } from './traces.js'; + +const APIs = ['startSpan', 'startActiveSpan', 'startActiveSpanSync', 'detachedSpan'] as const; +type TracerApi = (typeof APIs)[number]; + +function inConversation(conversationId: string | undefined, fn: () => T): T { + if (!conversationId) return fn(); + return runWithJobContext({ job: { room: { sid: conversationId } } } as unknown as JobContext, fn); +} + +async function withSpan( + api: TracerApi, + attributes: Attributes | undefined, + fn: (span: Span) => void, +): Promise { + const options = { name: 'test', attributes }; + if (api === 'startSpan') { + const span = tracer.startSpan(options); + try { + fn(span); + } finally { + span.end(); + } + } else if (api === 'startActiveSpan') { + await tracer.startActiveSpan(async (span) => fn(span), options); + } else if (api === 'startActiveSpanSync') { + tracer.startActiveSpanSync(fn, options); + } else { + await tracer.detachedSpan(async (span) => fn(span), options); + } +} + +describe.sequential('telemetry conversation id', () => { + let originalProvider = tracer.getProvider(); + let provider: NodeTracerProvider; + let exporter: InMemorySpanExporter; + + beforeEach(() => { + originalProvider = tracer.getProvider(); + exporter = new InMemorySpanExporter(); + provider = new NodeTracerProvider({ + spanProcessors: [new SimpleSpanProcessor(exporter)], + }); + setTracerProvider(provider); + }); + + afterEach(async () => { + setTracerProvider(originalProvider); + await provider.shutdown(); + }); + + it.each( + APIs.flatMap((api) => + [undefined, 'RM_test'].flatMap((conversationId) => + [undefined, 'explicit'].map((explicitId) => ({ api, conversationId, explicitId })), + ), + ), + )( + '$api preserves caller attributes with conversation=$conversationId explicit=$explicitId', + async ({ api, conversationId, explicitId }) => { + const attributes = Object.freeze({ + 'test.attribute': 'value', + ...(explicitId ? { 'gen_ai.conversation.id': explicitId } : {}), + }); + const original = { ...attributes }; + + await inConversation(conversationId, () => withSpan(api, attributes, () => undefined)); + + expect(attributes).toEqual(original); + expect(exporter.getFinishedSpans()[0]!.attributes).toEqual({ + ...original, + ...(!explicitId && conversationId ? { 'gen_ai.conversation.id': conversationId } : {}), + }); + }, + ); + + it.each(APIs)('%s adds the conversation id when attributes are absent', async (api) => { + await inConversation('RM_test', () => withSpan(api, undefined, () => undefined)); + + expect(exporter.getFinishedSpans()[0]!.attributes['gen_ai.conversation.id']).toBe('RM_test'); + }); + + const setters = [ + { + name: 'request', + set: (span: Span) => genAI.setRequestAttributes(span, { operation: 'chat' }), + }, + { + name: 'tool', + set: (span: Span) => genAI.setToolAttributes(span, { name: 'get_weather' }), + }, + { + name: 'agent', + set: (span: Span) => + genAI.setAgentAttributes(span, { operation: 'invoke_agent', agentName: 'agent' }), + }, + { + name: 'workflow', + set: (span: Span) => genAI.setWorkflowAttributes(span, { name: 'agent_session' }), + }, + ]; + + it.each( + APIs.flatMap((api) => + setters.flatMap((setter) => + [undefined, 'RM_test'].flatMap((conversationId) => + [undefined, 'explicit'].map((explicitId) => ({ + api, + setter, + conversationId, + explicitId, + })), + ), + ), + ), + )( + '$api $setter.name setter preserves conversation=$conversationId explicit=$explicitId', + async ({ api, setter, conversationId, explicitId }) => { + const attributes = explicitId ? { 'gen_ai.conversation.id': explicitId } : undefined; + + await inConversation(conversationId, () => + withSpan(api, attributes, (span) => { + setter.set(span); + inConversation('RM_other', () => setter.set(span)); + }), + ); + + expect(exporter.getFinishedSpans()[0]!.attributes['gen_ai.conversation.id']).toBe( + explicitId ?? conversationId, + ); + }, + ); +}); diff --git a/agents/src/telemetry/gen_ai.ts b/agents/src/telemetry/gen_ai.ts index 0a78f40b76..c9ffdb1f9e 100644 --- a/agents/src/telemetry/gen_ai.ts +++ b/agents/src/telemetry/gen_ai.ts @@ -371,8 +371,6 @@ export function setRequestAttributes( if (params.model) attrs[traceTypes.ATTR_GEN_AI_REQUEST_MODEL] = params.model; // "if and only if the request is streaming; if unset, assumed non-streaming" if (params.stream) attrs[traceTypes.ATTR_GEN_AI_REQUEST_STREAM] = true; - const conv = conversationId(); - if (conv) attrs[traceTypes.ATTR_GEN_AI_CONVERSATION_ID] = conv; if (params.outputType) attrs[traceTypes.ATTR_GEN_AI_OUTPUT_TYPE] = params.outputType; span.setAttributes(attrs); } @@ -482,8 +480,6 @@ export function setToolAttributes( }; if (params.callId) attrs[traceTypes.ATTR_GEN_AI_TOOL_CALL_ID] = params.callId; if (params.agentName) attrs[traceTypes.ATTR_GEN_AI_AGENT_NAME] = params.agentName; - const conv = conversationId(); - if (conv) attrs[traceTypes.ATTR_GEN_AI_CONVERSATION_ID] = conv; if (captureContent) { if (params.description) attrs[traceTypes.ATTR_GEN_AI_TOOL_DESCRIPTION] = params.description; if (params.args !== undefined) { @@ -539,8 +535,6 @@ export function setAgentAttributes( const normalizedProvider = traceTypes.genAIProviderName(params.provider); if (normalizedProvider) attrs[traceTypes.ATTR_GEN_AI_PROVIDER_NAME] = normalizedProvider; if (params.model) attrs[traceTypes.ATTR_GEN_AI_REQUEST_MODEL] = params.model; - const conv = conversationId(); - if (conv) attrs[traceTypes.ATTR_GEN_AI_CONVERSATION_ID] = conv; span.setAttributes(attrs); } @@ -552,7 +546,5 @@ export function setWorkflowAttributes(span: Span, params: { name: string }): voi [traceTypes.ATTR_GEN_AI_OPERATION_NAME]: traceTypes.GenAIOperationName.INVOKE_WORKFLOW, [traceTypes.ATTR_GEN_AI_WORKFLOW_NAME]: params.name, }; - const conv = conversationId(); - if (conv) attrs[traceTypes.ATTR_GEN_AI_CONVERSATION_ID] = conv; span.setAttributes(attrs); } diff --git a/agents/src/telemetry/traces.ts b/agents/src/telemetry/traces.ts index 2cba358a23..7459cebfc4 100644 --- a/agents/src/telemetry/traces.ts +++ b/agents/src/telemetry/traces.ts @@ -52,6 +52,7 @@ import { import { version } from '../version.js'; import { type SessionReport, sessionReportToJSON } from '../voice/report.js'; import { blockedSpanTracker } from './blocked_span_tracker.js'; +import { conversationId } from './gen_ai.js'; import type { ObservabilityEndpoint } from './observability_endpoint.js'; import { resolveObservabilityUrl } from './observability_endpoint.js'; import { type SimpleLogRecord, SimpleOTLPHttpLogExporter } from './otel_http_exporter.js'; @@ -60,7 +61,12 @@ import { flushPinoLogs, initPinoCloudExporter } from './pino_otel_transport.js'; import { uploadRecording } from './recording_upload.js'; import { allowPiiFromEnv, redactionEnabledFromAttributes } from './redaction.js'; import { JobSpanGateExporter } from './span_gate.js'; -import { ATTR_AGENT_NAME, ATTR_CLOUD_AGENT_ID, ATTR_DEPLOYMENT_ID } from './trace_types.js'; +import { + ATTR_AGENT_NAME, + ATTR_CLOUD_AGENT_ID, + ATTR_DEPLOYMENT_ID, + ATTR_GEN_AI_CONVERSATION_ID, +} from './trace_types.js'; import { UploadGateMetricExporter, UploadGateTraceExporter, uploadGate } from './upload_gate.js'; import { recordException } from './utils.js'; @@ -97,6 +103,12 @@ export interface DescribesOptions { /** @deprecated Use OpenTelemetry SDK 2.x's `SpanProcessor` type directly. */ export type SpanProcessorLike = SpanProcessor; +function withConversationId(attributes: Attributes | undefined): Attributes | undefined { + const id = conversationId(); + if (!id) return attributes; + return { [ATTR_GEN_AI_CONVERSATION_ID]: id, ...(attributes ?? {}) }; +} + /** * A dynamic tracer that allows the tracer provider to be changed at runtime. */ @@ -149,7 +161,7 @@ class DynamicTracer { const span = this.tracer.startSpan( options.name, { - attributes: options.attributes, + attributes: withConversationId(options.attributes), startTime: options.startTime, kind: options.kind, }, @@ -171,7 +183,7 @@ class DynamicTracer { const ctx = options.context || otelContext.active(); const endOnExit = options.endOnExit === undefined ? true : options.endOnExit; // default true const opts: SpanOptions = { - attributes: options.attributes, + attributes: withConversationId(options.attributes), startTime: options.startTime, kind: options.kind, }; @@ -220,7 +232,7 @@ class DynamicTracer { const ctx = options.context || otelContext.active(); const endOnExit = options.endOnExit === undefined ? true : options.endOnExit; // default true const opts: SpanOptions = { - attributes: options.attributes, + attributes: withConversationId(options.attributes), startTime: options.startTime, kind: options.kind, }; diff --git a/agents/src/telemetry/utils.ts b/agents/src/telemetry/utils.ts index 8cab106389..73e054b70d 100644 --- a/agents/src/telemetry/utils.ts +++ b/agents/src/telemetry/utils.ts @@ -153,12 +153,9 @@ export function recordRealtimeMetrics(span: Span, metrics: RealtimeModelMetrics) const spanContext = trace.setSpan(currentContext, span); // Create a dedicated child span for orphaned metrics - tracer.getTracer().startActiveSpan('realtime_metrics', {}, spanContext, (child) => { - try { - child.setAttributes(attrs); - } finally { - child.end(); - } + tracer.startActiveSpanSync((child) => child.setAttributes(attrs), { + name: 'realtime_metrics', + context: spanContext, }); } }