Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/bright-spans-correlate.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
'@livekit/agents': patch
---

Correlate telemetry spans from the same agent conversation with the LiveKit room SID.
16 changes: 15 additions & 1 deletion agents/src/ipc/job_trace.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down Expand Up @@ -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
Expand All @@ -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',
Expand All @@ -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);
Expand Down
1 change: 1 addition & 0 deletions agents/src/ipc/job_trace.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 ?? '',
Expand Down
142 changes: 142 additions & 0 deletions agents/src/telemetry/conversation_id.test.ts
Original file line number Diff line number Diff line change
@@ -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<T>(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<void> {
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,
);
},
);
});
8 changes: 0 additions & 8 deletions agents/src/telemetry/gen_ai.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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);
}

Expand All @@ -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);
}
20 changes: 16 additions & 4 deletions agents/src/telemetry/traces.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand All @@ -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';

Expand Down Expand Up @@ -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 ?? {}) };
}
Comment thread
devin-ai-integration[bot] marked this conversation as resolved.

/**
* A dynamic tracer that allows the tracer provider to be changed at runtime.
*/
Expand Down Expand Up @@ -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,
},
Expand All @@ -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,
};
Expand Down Expand Up @@ -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,
};
Expand Down
9 changes: 3 additions & 6 deletions agents/src/telemetry/utils.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
});
}
}
Loading