From 472c7d6ed842430696ab1dd784ab2730de5bf7ab Mon Sep 17 00:00:00 2001 From: David Zhao Date: Mon, 14 Sep 2026 23:28:27 -0700 Subject: [PATCH] tests: trace-shape schema and checker Port of livekit/agents#7148. The nesting of spans is an emergent property of many call sites; a refactor can move a span under the wrong parent while every existing test passes, because each asserts one edge. agents/src/telemetry/testing/trace_schema.ts writes the rules down once: - SPAN_PARENTS: for every span the JS framework emits, the parents it may have (ROOT for none, ANY for spans that follow their caller). Unknown names are violations, so a new span must be registered. - MAY_OUTLIVE_PARENT: the child/parent edges where the child may end after its parent, each with its reason. Everything else must sit inside its parent, with 2 ms of slack. - checkTrace(): one trace id, every parent present and allowed, bounds, one agent_turn per lk.speech_id, lk.generation_count equal to the speech's own generation events, every eou_wait with an outcome. - fromReadableSpans / fromOtlpJson: an in-memory exporter or an export downloaded from LiveKit Cloud, same rules. As a CLI (`pnpm exec tsx agents/src/telemetry/testing/trace_schema.ts x.json`) it prints the span summary and the violations, tolerating orphans in a partial export. assertTraceWellFormed() now ends the full-session tests: the tool call and plain reply in agent_turn_span, the barge-in and handoff in coverage_spans, the hook and redaction sessions in eou_wait_span, the lifecycle and SIP sessions in session_lifecycle_span. The module is test support: imported by tests only, not exported from the package, and free of framework imports so the CLI runs on the source file. Schema differences from Python, all from what the JS code emits: answering_machine_detection is JS's name for `amd`; JS has no llm_fallback_adapter / tts_fallback_adapter / tts_stream_adapter spans (its adapters emit the plain request spans, which nest under the attempt's *_request_run), no wait_for_video_track, no judge_evaluation; rpc_handler may be a root, since without a session the SDK dispatches on a context carrying no span. Test hardening: the eou_wait full-session helper decided the turn 20 ms after the fake STT final was due, so a loaded host could open a second user turn (the intermittent failure seen in full-suite runs); the endpointing delay now leaves a 170 ms margin. Co-Authored-By: Claude Fable 5.1 --- .../telemetry/testing/trace_schema.test.ts | 417 +++++++++++++++++ agents/src/telemetry/testing/trace_schema.ts | 437 ++++++++++++++++++ agents/src/voice/agent_turn_span.test.ts | 5 + agents/src/voice/coverage_spans.test.ts | 5 + agents/src/voice/eou_wait_span.test.ts | 10 +- .../src/voice/session_lifecycle_span.test.ts | 5 + 6 files changed, 878 insertions(+), 1 deletion(-) create mode 100644 agents/src/telemetry/testing/trace_schema.test.ts create mode 100644 agents/src/telemetry/testing/trace_schema.ts diff --git a/agents/src/telemetry/testing/trace_schema.test.ts b/agents/src/telemetry/testing/trace_schema.test.ts new file mode 100644 index 0000000000..6cfdfd2f63 --- /dev/null +++ b/agents/src/telemetry/testing/trace_schema.test.ts @@ -0,0 +1,417 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +/** + * The trace-shape checker itself: the rules catch the mistakes they exist for, both span + * sources agree, and a full fake session passes clean. + */ +import { AudioFrame } from '@livekit/rtc-node'; +import { context as otelContext, trace } from '@opentelemetry/api'; +import { InMemorySpanExporter, 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 } from 'vitest'; +import { ChatContext } from '../../llm/chat_context.js'; +import { FallbackAdapter } from '../../llm/fallback_adapter.js'; +import { initializeLogger } from '../../log.js'; +import { FakeSTT } from '../../stt/testing/fake_stt.js'; +import { Agent } from '../../voice/agent.js'; +import { AgentSession } from '../../voice/agent_session.js'; +import { AudioOutput } from '../../voice/io.js'; +import { FakeLLM } from '../../voice/testing/fake_llm.js'; +import { setTracerProvider, tracer } from '../index.js'; +import { + ANY, + type EventRecord, + MAY_OUTLIVE_PARENT, + ROOT, + SPAN_PARENTS, + type SpanRecord, + assertTraceWellFormed, + checkTrace, + fromOtlpJson, + fromReadableSpans, +} from './trace_schema.js'; + +initializeLogger({ pretty: false, level: 'silent' }); + +function span( + name: string, + spanId: string, + parent: string | undefined, + start: number, + end: number, + attributes: Record = {}, + events: (string | EventRecord)[] = [], +): SpanRecord { + return { + name, + spanId, + parentId: parent, + traceId: 't1', + startMs: start * 1000, + endMs: end * 1000, + attributes: { ...attributes }, + events: events.map((e) => (typeof e === 'string' ? { name: e, attributes: {} } : e)), + }; +} + +function soundTrace(): SpanRecord[] { + return [ + span('job_entrypoint', 'j', undefined, 0.0, 30.0), + span('agent_session', 's', 'j', 1.0, 28.0), + span('user_turn', 'u', 's', 5.0, 7.0), + span('eou_wait', 'w', 'u', 6.5, 7.0, { 'lk.eou.outcome': 'committed' }), + span('eou_detection', 'd', 'w', 6.8, 6.9), + span('on_user_turn_completed', 'h', 'u', 6.95, 7.0), + span( + 'agent_turn', + 'a', + 's', + 7.0, + 12.0, + { 'lk.speech_id': 'speech_1', 'lk.generation_count': 2 }, + [ + { name: 'generation', attributes: { 'lk.generation_id': 'speech_1_1' } }, + { name: 'generation', attributes: { 'lk.generation_id': 'speech_1_2' } }, + ], + ), + span('llm_node', 'l', 'a', 7.0, 8.0), + span('llm_request', 'r', 'l', 7.0, 8.0), + span('function_tool', 'f', 'a', 8.0, 8.1), + span('job_shutdown', 'x', 'j', 28.0, 30.0), + ]; +} + +const TURN = 6; // index of the agent_turn in soundTrace() + +describe('trace schema rules', () => { + it('is self-consistent', () => { + // every parent named in the rules is itself a known span (or ROOT / ANY) + for (const [name, allowed] of SPAN_PARENTS) { + for (const parent of allowed) { + expect( + parent === ROOT || parent === ANY || SPAN_PARENTS.has(parent), + `${name}: unknown parent ${String(parent)}`, + ).toBe(true); + } + } + for (const key of MAY_OUTLIVE_PARENT.keys()) { + const [child, parent] = key.split(' -> ') as [string, string]; + expect(SPAN_PARENTS.has(child), key).toBe(true); + expect(parent === ANY || SPAN_PARENTS.has(parent), key).toBe(true); + } + }); + + it('accepts a sound trace', () => { + expect(checkTrace(soundTrace())).toEqual([]); + }); + + it('reports a wrong parent', () => { + // the keyterm-detection style mistake: an llm_request straight under agent_turn + let spans = soundTrace(); + spans.push(span('llm_request', 'k', 'a', 7.1, 7.9)); + const [violation, ...rest] = checkTrace(spans); + expect(rest).toEqual([]); + expect(violation).toMatch(/^llm_request: parent is agent_turn/); + + // eou_detection outside its wait + spans = soundTrace(); + spans.push(span('eou_detection', 'd2', 'u', 6.0, 6.1)); + expect(checkTrace(spans).some((v) => v.startsWith('eou_detection: parent is user_turn'))).toBe( + true, + ); + }); + + it('reports an unknown span and a missing parent', () => { + let spans = [...soundTrace(), span('mystery', 'm', 's', 2.0, 3.0)]; + expect(checkTrace(spans).some((v) => v.includes('mystery: unknown span'))).toBe(true); + + spans = [...soundTrace(), span('user_speaking', 'p', 'gone', 2.0, 3.0)]; + expect(checkTrace(spans).some((v) => v.includes('parent gone is not in the trace'))).toBe(true); + // a partial export (a view keyed to one span drops the ancestors): the orphan's edge is + // simply not checked + expect(checkTrace(spans, { allowMissingParents: true })).toEqual([]); + }); + + it('checks bounds except where deliberately allowed', () => { + let spans = soundTrace(); + spans.push(span('tts_node', 't', 'a', 11.0, 12.5)); // ends after agent_turn + expect( + checkTrace(spans).some((v) => + v.startsWith('tts_node: ends 500.0 ms after its parent agent_turn'), + ), + ).toBe(true); + + spans = soundTrace(); + spans.push(span('user_speaking', 'sp', 'u', 4.0, 6.0)); // starts before user_turn + expect( + checkTrace(spans).some((v) => v.startsWith('user_speaking: starts 1000.0 ms before')), + ).toBe(true); + + // session.start() returning before the participant is linked is a known shape + spans = soundTrace(); + spans.push(span('session_start', 'ss', 's', 1.0, 2.0)); + spans.push(span('wait_for_participant', 'wp', 'ss', 2.0, 4.0)); + expect(checkTrace(spans)).toEqual([]); + // and a stall's end is one tick late by construction, whatever it is under + spans.push(span('event_loop_blocked', 'b', 'f', 8.05, 8.15)); + expect(checkTrace(spans)).toEqual([]); + }); + + it('checks the per-turn invariants', () => { + let spans = soundTrace(); + spans.push( + span( + 'agent_turn', + 'a2', + 's', + 13.0, + 14.0, + { 'lk.speech_id': 'speech_1', 'lk.generation_count': 1 }, + [{ name: 'generation', attributes: { 'lk.generation_id': 'speech_1_1' } }], + ), + ); + expect(checkTrace(spans).some((v) => v.includes('speech speech_1 has 2 turns'))).toBe(true); + + spans = soundTrace(); + spans[TURN]!.attributes['lk.generation_count'] = 3; + expect( + checkTrace(spans).some((v) => v.includes('lk.generation_count=3 but 2 generation events')), + ).toBe(true); + + // a turn without its identity or its count is malformed, not exempt + spans = soundTrace(); + delete spans[TURN]!.attributes['lk.speech_id']; + expect(checkTrace(spans)).toContain('agent_turn: no lk.speech_id'); + spans = soundTrace(); + delete spans[TURN]!.attributes['lk.generation_count']; + expect(checkTrace(spans).some((v) => v.includes('lk.generation_count=undefined'))).toBe(true); + spans = soundTrace(); + spans[TURN]!.attributes['lk.generation_count'] = 'two'; + expect(checkTrace(spans).some((v) => v.includes('lk.generation_count="two"'))).toBe(true); + + // a discarded preemptive attempt's generation sits on the span and counts: the count is + // the turn's, not the finishing speech's + spans = soundTrace(); + spans[TURN]!.events = [ + { name: 'generation', attributes: { 'lk.generation_id': 'speech_0_1' } }, + { name: 'preemptive_generation_discarded', attributes: { 'lk.speech_id': 'speech_0' } }, + { name: 'generation', attributes: { 'lk.generation_id': 'speech_1_1' } }, + { name: 'generation', attributes: { 'lk.generation_id': 'speech_1_2' } }, + ]; + spans[TURN]!.attributes['lk.generation_count'] = 3; + expect(checkTrace(spans)).toEqual([]); + spans[TURN]!.attributes['lk.generation_count'] = 2; + expect( + checkTrace(spans).some((v) => v.includes('lk.generation_count=2 but 3 generation events')), + ).toBe(true); + // but a generation after the handoff must be the finishing speech's own + spans[TURN]!.attributes['lk.generation_count'] = 3; + spans[TURN]!.events[3] = { + name: 'generation', + attributes: { 'lk.generation_id': 'speech_9_1' }, + }; + expect( + checkTrace(spans).some((v) => + v.includes('generation speech_9_1 after the last handoff is not its own'), + ), + ).toBe(true); + + // a generation the framework emitted always carries its id: one without is malformed, + // a discarded attempt's before the handoff as much as the finishing speech's after it + spans = soundTrace(); + spans[TURN]!.events[1] = { name: 'generation', attributes: {} }; + expect(checkTrace(spans)).toContain('agent_turn speech_1: generation without lk.generation_id'); + spans = soundTrace(); + spans[TURN]!.events = [ + { name: 'generation', attributes: {} }, + { name: 'preemptive_generation_discarded', attributes: { 'lk.speech_id': 'speech_0' } }, + { name: 'generation', attributes: { 'lk.generation_id': 'speech_1_1' } }, + ]; + expect(checkTrace(spans)).toContain('agent_turn speech_1: generation without lk.generation_id'); + + // nothing exported is not a sound trace + expect(checkTrace([]).some((v) => v.includes('0 traces'))).toBe(true); + + spans = soundTrace(); + delete spans[3]!.attributes['lk.eou.outcome']; + expect(checkTrace(spans)).toContain('eou_wait: no lk.eou.outcome'); + + spans = soundTrace(); + spans[1]!.traceId = 't2'; + expect(checkTrace(spans).some((v) => v.includes('2 traces'))).toBe(true); + }); +}); + +/** 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 SilentAgent extends Agent { + constructor() { + super({ instructions: 'You are a helpful assistant.' }); + } + + override async ttsNode(): Promise> { + return new ReadableStream({ + start(controller) { + controller.enqueue(new AudioFrame(new Int16Array(480), 24_000, 1, 480)); + controller.close(); + }, + }); + } +} + +describe.sequential('trace schema on real spans', () => { + 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 () => { + setTracerProvider(originalProvider); + await provider.shutdown(); + trace.disable(); + otelContext.disable(); + }); + + it('reads OTLP/JSON and readable spans the same way', async () => { + let rootSpanId = ''; + await tracer.startActiveSpan( + async (root) => { + rootSpanId = root.spanContext().spanId; + await tracer.startActiveSpan( + async (turn) => { + turn.addEvent('generation', { 'lk.generation_id': 'x_1' }); + }, + { name: 'user_turn', attributes: { 'lk.speech_id': 'x' } }, + ); + }, + { name: 'agent_session' }, + ); + const readable = fromReadableSpans(exporter.getFinishedSpans()); + + const attrs = (values: Record) => + Object.entries(values).map(([key, value]) => ({ + key, + value: { stringValue: String(value) }, + })); + const document = { + resourceSpans: [ + { + scopeSpans: [ + { + spans: readable.map((s) => ({ + name: s.name, + spanId: s.spanId, + parentSpanId: s.parentId ?? '', + traceId: s.traceId, + startTimeUnixNano: String(BigInt(Math.round(s.startMs * 1e6))), + endTimeUnixNano: String(BigInt(Math.round(s.endMs * 1e6))), + attributes: attrs(s.attributes), + events: s.events.map((e) => ({ name: e.name, attributes: attrs(e.attributes) })), + })), + }, + ], + }, + ], + }; + const converted = fromOtlpJson(document); + expect(converted.map((s) => [s.name, s.parentId, s.events])).toEqual( + readable.map((s) => [s.name, s.parentId, s.events]), + ); + const turn = converted.find((s) => s.name === 'user_turn'); + expect(turn?.parentId).toBe(rootSpanId); + // nanosecond round trip keeps the millisecond timestamps + for (const [a, b] of converted.map((s, i) => [s, readable[i]!] as const)) { + expect(Math.abs(a.startMs - b.startMs)).toBeLessThan(1e-3); + expect(Math.abs(a.endMs - b.endMs)).toBeLessThan(1e-3); + } + expect(checkTrace(converted)).toEqual([]); + expect(checkTrace(readable)).toEqual([]); + }); + + it('finds a full fake session well-formed', async () => { + const llm = new FakeLLM([{ input: 'Hello there', content: 'Hi!' }]); + const session = new AgentSession({ llm, stt: new FakeSTT() }); + session.output.audio = new ImmediateOutput(); + await session.start({ agent: new SilentAgent() }); + try { + const speech = session.generateReply({ userInput: 'Hello there' }); + await speech.waitForPlayout(); + } finally { + await session.close(); + } + expect(exporter.getFinishedSpans().map((s) => s.name)).toContain('agent_turn'); + assertTraceWellFormed(exporter.getFinishedSpans()); + }); + + it('allows the fallback adapter request shapes', async () => { + // the adapter's request span stands in for the provider's; each attempt opens the wrapped + // stream inside its llm_request_run, so the provider's request span nests under the attempt + const adapter = new FallbackAdapter({ + llms: [new FakeLLM([{ input: 'hi', content: 'hello' }])], + attemptTimeout: 1, + }); + const chatCtx = ChatContext.empty(); + chatCtx.addMessage({ role: 'user', content: 'hi' }); + await tracer.startActiveSpan( + async () => + tracer.startActiveSpan( + async (turn) => { + turn.addEvent('generation', { 'lk.generation_id': 'sp_1' }); + await tracer.startActiveSpan( + async () => { + const stream = adapter.chat({ chatCtx }); + for await (const _chunk of stream) { + // drain + } + }, + { name: 'llm_node' }, + ); + }, + { + name: 'agent_turn', + attributes: { 'lk.speech_id': 'sp', 'lk.generation_count': 1 }, + }, + ), + { name: 'agent_session' }, + ); + const names = exporter.getFinishedSpans().map((s) => s.name); + expect(names.filter((n) => n === 'llm_request').length).toBeGreaterThanOrEqual(2); + assertTraceWellFormed(exporter.getFinishedSpans()); + }); +}); diff --git a/agents/src/telemetry/testing/trace_schema.ts b/agents/src/telemetry/testing/trace_schema.ts new file mode 100644 index 0000000000..d0646789c7 --- /dev/null +++ b/agents/src/telemetry/testing/trace_schema.ts @@ -0,0 +1,437 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +/** + * The shape every livekit-agents trace must have, and a checker that applies it. + * + * The nesting of spans is an emergent property of many call sites, and a refactor can move a + * span under the wrong parent while every existing test still passes, because each test only + * names the one edge it cares about. This module writes the rules down once: + * + * - {@link SPAN_PARENTS}: for each span name, the parents it may have (`ROOT` for none). + * - {@link MAY_OUTLIVE_PARENT}: the few child/parent edges where the child is allowed to end + * after its parent, each with the reason. Everything else must sit inside its parent. + * - {@link checkTrace}: applies those rules plus the per-turn invariants (one `agent_turn` per + * speech, its own generation events matching `lk.generation_count`) and returns the + * violations. + * + * It reads spans from an in-memory exporter (the fake-session tests) or from an OTLP/JSON export + * downloaded from LiveKit Cloud, so the same rules check a unit test and a real run: + * + * ``` + * pnpm exec tsx agents/src/telemetry/testing/trace_schema.ts path/to/traces.json + * ``` + * + * Adding a span means adding a row here. Moving one without updating the row fails every test + * that calls {@link assertTraceWellFormed}. + * + * Test support: imported by tests only, not exported from the package. Kept free of framework + * imports so the CLI runs on the source file directly. + */ +import { readFileSync } from 'node:fs'; +import { pathToFileURL } from 'node:url'; + +/** Allowed parent meaning "no parent at all". */ +export const ROOT = null; + +/** Allowed parent meaning "whatever was current": spans that follow their caller. */ +export const ANY = '*'; + +export type AllowedParent = string | typeof ROOT; + +/** 2 ms of slack for clocks read on either side of a span boundary. */ +export const TOLERANCE_MS = 2; + +const parents = (...names: AllowedParent[]): ReadonlySet => new Set(names); + +export const SPAN_PARENTS: ReadonlyMap> = new Map< + string, + ReadonlySet +>([ + // -- the job (ipc/job_proc_lazy_main, ipc/job_trace, job_lifecycle) + ['job_entrypoint', parents(ROOT)], + ['job_shutdown', parents('job_entrypoint')], + ['on_session_end', parents('job_shutdown')], + ['session_end_upload', parents('job_shutdown')], + ['room_disconnect', parents('job_shutdown')], + ['shutdown_callback', parents('job_shutdown')], + // -- the session (voice/agent_session); ROOT outside a job (tests, integrators) + ['agent_session', parents('job_entrypoint', ROOT)], + ['session_start', parents('agent_session')], + ['session_close', parents('agent_session')], + ['update_agent', parents('agent_session')], + // -- startup work: under session_start while the session starts, under the job before + ['room_connect', parents('session_start', 'job_entrypoint')], + ['wait_for_participant', parents('session_start', 'job_entrypoint')], + ['wait_for_audio_track', parents('session_start', 'agent_session')], + ['publish_audio_output', parents('session_start')], + // -- agent activity lifecycle + ['start_agent_activity', parents('session_start', 'update_agent')], + ['setup_toolsets', parents('start_agent_activity')], + ['on_enter', parents('start_agent_activity')], + ['pause_agent_activity', parents('update_agent')], + ['resume_agent_activity', parents('update_agent')], + ['drain_agent_activity', parents('update_agent', 'session_close', 'agent_session')], + ['on_exit', parents('drain_agent_activity')], + // -- the user's turn + ['user_turn', parents('agent_session')], + ['user_speaking', parents('user_turn', 'agent_session')], + ['eou_wait', parents('user_turn')], + ['eou_detection', parents('eou_wait')], + ['on_user_turn_completed', parents('user_turn', 'agent_session')], + // -- the agent's turn: one per speech handle, every generation inside it + ['agent_turn', parents('agent_session')], + ['llm_node', parents('agent_turn')], + ['tts_node', parents('agent_turn')], + ['function_tool', parents('agent_turn')], + ['agent_speaking', parents('agent_turn')], + ['realtime_inference', parents('agent_turn')], + ['realtime_metrics', parents('realtime_inference', 'agent_turn')], + // -- model requests: under the node that made them, or the feature that owns them. An + // adapter (LLM/TTS FallbackAdapter, TTS StreamAdapter) is itself an LLM/TTS, so its request + // span stands in for the provider's; each attempt (`*_request_run`) opens the wrapped stream, + // whose own request span nests inside it: + // llm_request (adapter) → llm_request_run → llm_request (provider) → llm_request_run + [ + 'llm_request', + parents('llm_node', 'llm_request_run', 'keyterm_detection', 'answering_machine_detection'), + ], + ['llm_request_run', parents('llm_request')], + ['tts_request', parents('tts_node', 'tts_request_run')], + ['tts_request_run', parents('tts_request')], + // -- session-scoped features + ['keyterm_detection', parents('agent_turn', 'agent_session')], + ['answering_machine_detection', parents('agent_session')], + // -- RPC: handlers are session events, calls follow their caller. Before or after the session + // a handler lands under the job; outside a job (an integrator's RoomIO) the SDK dispatches on + // a context carrying no span. + ['rpc_handler', parents('agent_session', 'job_entrypoint', ROOT)], + ['rpc_call', parents(ANY)], + // -- a stall lands under whatever was blocked (any span), or the session/job + ['event_loop_blocked', parents(ANY)], +]); + +/** + * Child/parent edges where the child may end after its parent, with the reason. Deliberate: + * each is a known property of the code, and a viewer draws them poking out of the parent. + */ +export const MAY_OUTLIVE_PARENT: ReadonlyMap = new Map([ + [ + edge('wait_for_participant', 'session_start'), + 'session.start() returns before a participant is linked', + ], + [edge('wait_for_audio_track', 'session_start'), 'session.start() returns before the first frame'], + [ + edge('publish_audio_output', 'session_start'), + 'session.start() returns before the track is published', + ], + [ + edge('on_enter', 'start_agent_activity'), + 'on_enter runs as a task the activity does not wait for', + ], + [ + edge('event_loop_blocked', ANY), + 'the heartbeat notices a stall one tick after the blocked call returned', + ], + [ + edge('keyterm_detection', 'agent_turn'), + 'the pass runs alongside the reply and can outlast a short or interrupted turn', + ], +]); + +/** The key of a child/parent edge in {@link MAY_OUTLIVE_PARENT}. */ +export function edge(child: string, parent: string): string { + return `${child} -> ${parent}`; +} + +export interface EventRecord { + name: string; + attributes: Record; +} + +/** The part of a span the rules look at, from either source. Times are epoch milliseconds. */ +export interface SpanRecord { + name: string; + spanId: string; + parentId: string | undefined; + traceId: string; + startMs: number; + endMs: number; + attributes: Record; + events: EventRecord[]; +} + +/** What the rules read off an OpenTelemetry SDK `ReadableSpan`, spelled structurally. */ +export interface ReadableSpanLike { + readonly name: string; + spanContext(): { spanId: string; traceId: string }; + readonly parentSpanContext?: { spanId: string }; + readonly startTime: [number, number]; + readonly endTime: [number, number]; + readonly attributes: Record; + readonly events: readonly { name: string; attributes?: Record }[]; +} + +function hrTimeMs(time: [number, number]): number { + return time[0] * 1000 + time[1] / 1e6; +} + +/** Records from OpenTelemetry SDK `ReadableSpan` objects (an in-memory exporter). */ +export function fromReadableSpans(spans: Iterable): SpanRecord[] { + const out: SpanRecord[] = []; + for (const span of spans) { + const ctx = span.spanContext(); + out.push({ + name: span.name, + spanId: ctx.spanId, + parentId: span.parentSpanContext?.spanId, + traceId: ctx.traceId, + startMs: hrTimeMs(span.startTime), + endMs: hrTimeMs(span.endTime), + attributes: { ...span.attributes }, + events: span.events.map((event) => ({ + name: event.name, + attributes: { ...(event.attributes ?? {}) }, + })), + }); + } + return out; +} + +type OtlpValue = Record; + +function otlpValue(value: OtlpValue): unknown { + if ('stringValue' in value) return value.stringValue; + if ('intValue' in value) return Number(value.intValue); + if ('doubleValue' in value) return Number(value.doubleValue); + if ('boolValue' in value) return Boolean(value.boolValue); + if ('arrayValue' in value) { + const values = (value.arrayValue as { values?: OtlpValue[] }).values ?? []; + return values.map(otlpValue); + } + return value; +} + +function otlpAttributes(item: { attributes?: { key: string; value: OtlpValue }[] }) { + const out: Record = {}; + for (const attr of item.attributes ?? []) out[attr.key] = otlpValue(attr.value); + return out; +} + +function unixNanoMs(value: unknown): number { + // int64 as a decimal string in OTLP/JSON; BigInt keeps the nanoseconds exact before the + // division, since epoch nanoseconds exceed the double's integer range + return Number(BigInt(String(value))) / 1e6; +} + +interface OtlpSpan { + name: string; + spanId: string; + parentSpanId?: string; + traceId: string; + startTimeUnixNano: string | number; + endTimeUnixNano: string | number; + attributes?: { key: string; value: OtlpValue }[]; + events?: { name: string; attributes?: { key: string; value: OtlpValue }[] }[]; +} + +interface OtlpDocument { + resourceSpans?: { scopeSpans?: { spans?: OtlpSpan[] }[] }[]; +} + +/** Records from an OTLP/JSON export (`resourceSpans` → `scopeSpans` → `spans`), or its path. */ +export function fromOtlpJson(document: OtlpDocument | string): SpanRecord[] { + const doc: OtlpDocument = + typeof document === 'string' + ? (JSON.parse(readFileSync(document, 'utf8')) as OtlpDocument) + : document; + const out: SpanRecord[] = []; + for (const resourceSpans of doc.resourceSpans ?? []) { + for (const scopeSpans of resourceSpans.scopeSpans ?? []) { + for (const span of scopeSpans.spans ?? []) { + out.push({ + name: span.name, + spanId: span.spanId, + parentId: span.parentSpanId || undefined, + traceId: span.traceId, + startMs: unixNanoMs(span.startTimeUnixNano), + endMs: unixNanoMs(span.endTimeUnixNano), + attributes: otlpAttributes(span), + events: (span.events ?? []).map((event) => ({ + name: event.name, + attributes: otlpAttributes(event), + })), + }); + } + } + } + return out; +} + +export interface CheckTraceOptions { + /** Clock slack, in milliseconds, for the bounds checks. */ + toleranceMs?: number; + /** + * For partial exports (a view keyed to one span drops the ancestors): a span whose parent is + * absent is then checked as if it were a root. + */ + allowMissingParents?: boolean; +} + +function showParent(parent: AllowedParent): string { + return parent === ROOT ? 'ROOT' : parent; +} + +/** Every way the spans break the rules, as one line each; empty when the trace is sound. */ +export function checkTrace( + spans: readonly SpanRecord[], + options: CheckTraceOptions = {}, +): string[] { + const { toleranceMs = TOLERANCE_MS, allowMissingParents = false } = options; + const violations: string[] = []; + const byId = new Map(spans.map((span) => [span.spanId, span])); + + const traceIds = new Set(spans.map((span) => span.traceId)); + if (traceIds.size !== 1) { + violations.push(`spans belong to ${traceIds.size} traces, expected one`); + } + + for (const span of spans) { + const allowed = SPAN_PARENTS.get(span.name); + if (allowed === undefined) { + violations.push(`${span.name}: unknown span, add it to trace_schema SPAN_PARENTS`); + continue; + } + const parent = span.parentId ? byId.get(span.parentId) : undefined; + if (span.parentId && parent === undefined) { + if (!allowMissingParents) { + violations.push(`${span.name}: parent ${span.parentId} is not in the trace`); + } + continue; // a partial export: nothing to check the edge against + } + const parentName: AllowedParent = parent ? parent.name : ROOT; + if (!allowed.has(ANY) && !allowed.has(parentName)) { + const shown = parentName === ROOT ? 'no parent' : parentName; + violations.push( + `${span.name}: parent is ${shown}, allowed: ${[...allowed].map(showParent).sort().join(', ')}`, + ); + } + if (parent) { + if (span.startMs + toleranceMs < parent.startMs) { + violations.push( + `${span.name}: starts ${(parent.startMs - span.startMs).toFixed(1)} ms before its ` + + `parent ${parent.name}`, + ); + } + const overrun = span.endMs - parent.endMs; + if ( + overrun > toleranceMs && + !MAY_OUTLIVE_PARENT.has(edge(span.name, parent.name)) && + !MAY_OUTLIVE_PARENT.has(edge(span.name, ANY)) + ) { + violations.push( + `${span.name}: ends ${overrun.toFixed(1)} ms after its parent ${parent.name}`, + ); + } + } + } + + // one agent_turn per speech handle, and its generations accounted for + const speechTurns = new Map(); + for (const span of spans) { + if (span.name !== 'agent_turn') continue; + const speechId = String(span.attributes['lk.speech_id'] ?? ''); + if (!speechId) { + violations.push('agent_turn: no lk.speech_id'); + continue; + } + speechTurns.set(speechId, [...(speechTurns.get(speechId) ?? []), span]); + // the count is of every generation on the turn, a discarded preemptive attempt's included + const generations = span.events.filter((event) => event.name === 'generation').length; + const count = span.attributes['lk.generation_count']; + const countOk = + (typeof count === 'number' || typeof count === 'string') && + Number.isInteger(Number(count)) && + Number(count) === generations; + if (!countOk) { + violations.push( + `agent_turn ${speechId}: lk.generation_count=${JSON.stringify(count)} but ` + + `${generations} generation events`, + ); + } + // generations before the last handoff belonged to the discarded attempts; the ones after + // it are the finishing speech's own (a realtime tool reply continues the tool call's id) + const lastHandoff = span.events + .map((event, index) => (event.name === 'preemptive_generation_discarded' ? index : -1)) + .reduce((a, b) => Math.max(a, b), -1); + for (const [index, event] of span.events.entries()) { + if (event.name !== 'generation') continue; + const id = event.attributes['lk.generation_id']; + // every generation the framework emitted carries its id, a discarded attempt's included + if (id === undefined) { + violations.push(`agent_turn ${speechId}: generation without lk.generation_id`); + } else if (index > lastHandoff && !String(id).startsWith(`${speechId}_`)) { + violations.push( + `agent_turn ${speechId}: generation ${String(id)} after the last handoff is not its own`, + ); + } + } + } + for (const [speechId, turns] of speechTurns) { + if (turns.length > 1) { + violations.push(`agent_turn: speech ${speechId} has ${turns.length} turns, expected one`); + } + } + + // a wait always ends with an outcome (eou_detection never running outside one is a parent rule) + for (const span of spans) { + if (span.name === 'eou_wait' && !('lk.eou.outcome' in span.attributes)) { + violations.push('eou_wait: no lk.eou.outcome'); + } + } + + return violations; +} + +/** For tests: `spans` are `ReadableSpan` objects from an in-memory exporter. */ +export function assertTraceWellFormed( + spans: Iterable, + options: CheckTraceOptions = {}, +): void { + const violations = checkTrace(fromReadableSpans(spans), options); + if (violations.length) { + throw new Error(`trace shape violations:\n ${violations.join('\n ')}`); + } +} + +function summary(spans: readonly SpanRecord[]): string { + const counts = new Map(); + for (const span of spans) counts.set(span.name, (counts.get(span.name) ?? 0) + 1); + return [...counts] + .sort((a, b) => b[1] - a[1] || a[0].localeCompare(b[0])) + .map(([name, count]) => `${name}×${count}`) + .join(', '); +} + +/** The CLI: print the span summary and the violations of an OTLP/JSON export. */ +export function main(argv: readonly string[]): number { + if (argv.length !== 1) { + console.error('usage: tsx agents/src/telemetry/testing/trace_schema.ts '); + return 2; + } + const records = fromOtlpJson(argv[0]!); + console.log(`${records.length} spans: ${summary(records)}`); + const violations = checkTrace(records, { allowMissingParents: true }); + if (!violations.length) { + console.log('trace shape OK'); + return 0; + } + console.log(`${violations.length} violation(s):`); + for (const violation of violations) console.log(` - ${violation}`); + return 1; +} + +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + process.exit(main(process.argv.slice(2))); +} diff --git a/agents/src/voice/agent_turn_span.test.ts b/agents/src/voice/agent_turn_span.test.ts index 08839437c8..fd6eb2e779 100644 --- a/agents/src/voice/agent_turn_span.test.ts +++ b/agents/src/voice/agent_turn_span.test.ts @@ -32,6 +32,7 @@ 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 { assertTraceWellFormed } from '../telemetry/testing/trace_schema.js'; import { Agent } from './agent.js'; import { continueDiscardedTurn, continueToolReplyTurn, withAgentTurn } from './agent_activity.js'; import { AgentSession } from './agent_session.js'; @@ -188,6 +189,8 @@ describe.sequential('agent_turn span', () => { expect(ms(turn!.startTime)).toBeLessThanOrEqual(ms(child.startTime) + 2); expect(ms(child.endTime)).toBeLessThanOrEqual(ms(turn!.endTime) + 2); } + // the whole tree, not just the edges this test names (telemetry/testing/trace_schema) + assertTraceWellFormed(exporter.getFinishedSpans()); }); it('a plain reply is one generation', async () => { @@ -201,6 +204,8 @@ describe.sequential('agent_turn span', () => { 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(); + // the whole tree, not just the edges this test names (telemetry/testing/trace_schema) + assertTraceWellFormed(exporter.getFinishedSpans()); }); it('a discarded preemptive generation hands its turn to the successor', async () => { diff --git a/agents/src/voice/coverage_spans.test.ts b/agents/src/voice/coverage_spans.test.ts index 13aca82ddd..c17fbd4011 100644 --- a/agents/src/voice/coverage_spans.test.ts +++ b/agents/src/voice/coverage_spans.test.ts @@ -24,6 +24,7 @@ import type { ToolChoice, ToolContextLike } 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 { assertTraceWellFormed } from '../telemetry/testing/trace_schema.js'; import { type APIConnectOptions, DEFAULT_API_CONNECT_OPTIONS } from '../types.js'; import { delay } from '../utils.js'; import { VAD, type VADEvent, VADEventType, VADStream } from '../vad.js'; @@ -315,6 +316,8 @@ describe.sequential('coverage spans', () => { if (other === turn) continue; expect(other.attributes[traceTypes.ATTR_INTERRUPTION_SOURCE]).toBeUndefined(); } + // the whole tree, not just the edges this test names (telemetry/testing/trace_schema) + assertTraceWellFormed(exporter.getFinishedSpans()); }); it('a committed user turn interrupts the queued replies for the same reason', async () => { @@ -547,6 +550,8 @@ describe.sequential('coverage spans', () => { // the initial start is not a handoff: it lives under session_start, not update_agent const sessionStart = only(exporter, 'session_start'); expect(childrenOf(exporter, 'start_agent_activity', sessionStart)).toHaveLength(1); + // the whole tree, not just the edges this test names (telemetry/testing/trace_schema) + assertTraceWellFormed(exporter.getFinishedSpans()); }); // -- fallback adapter attribution -- diff --git a/agents/src/voice/eou_wait_span.test.ts b/agents/src/voice/eou_wait_span.test.ts index a30d04c860..3d68437d9d 100644 --- a/agents/src/voice/eou_wait_span.test.ts +++ b/agents/src/voice/eou_wait_span.test.ts @@ -32,6 +32,7 @@ import { initializeLogger } from '../log.js'; import { FakeSTT } from '../stt/testing/fake_stt.js'; import { setTracerProvider, traceTypes, tracer } from '../telemetry/index.js'; import { REDACTED_EXCEPTION_MESSAGE } from '../telemetry/redaction.js'; +import { assertTraceWellFormed } from '../telemetry/testing/trace_schema.js'; import { Future, delay } from '../utils.js'; import { VAD, type VADEvent, VADEventType, VADStream } from '../vad.js'; import { Agent } from './agent.js'; @@ -561,7 +562,10 @@ describe.sequential('eou_wait span', () => { llm, turnHandling: { turnDetection: 'vad', - endpointing: { minDelay: 100, maxDelay: 100 }, + // the fake STT's final lands 100 ms after the speech ends; the decision must come + // after it by a margin a loaded CI host cannot eat, else a late final opens a + // second user turn + endpointing: { minDelay: 300, maxDelay: 300 }, }, }); const audioInput = new ScriptedAudioInput(); @@ -634,6 +638,8 @@ describe.sequential('eou_wait span', () => { expect(attrs[key], key).toBeTypeOf('number'); } expect(attrs[traceTypes.ATTR_ON_USER_TURN_COMPLETED_DELAY]).toBeGreaterThanOrEqual(0.045); + // the whole tree, not just the edges this test names (telemetry/testing/trace_schema) + assertTraceWellFormed(exporter.getFinishedSpans()); }); it('keeps the turn open for recognition when the hook returns at once', async () => { @@ -686,6 +692,8 @@ describe.sequential('eou_wait span', () => { JSON.stringify(hook.events.map((e) => [e.name, e.attributes ?? {}])); expect(rendered).not.toContain('Hello'); expect(rendered).not.toContain('lookup failed'); + // the whole tree, not just the edges this test names (telemetry/testing/trace_schema) + assertTraceWellFormed(exporter.getFinishedSpans()); }); }); }); diff --git a/agents/src/voice/session_lifecycle_span.test.ts b/agents/src/voice/session_lifecycle_span.test.ts index 2e47a67df6..bac58dd94f 100644 --- a/agents/src/voice/session_lifecycle_span.test.ts +++ b/agents/src/voice/session_lifecycle_span.test.ts @@ -21,6 +21,7 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { initializeLogger } from '../log.js'; import { FakeSTT } from '../stt/testing/fake_stt.js'; import { setTracerProvider, traceTypes, tracer } from '../telemetry/index.js'; +import { assertTraceWellFormed } from '../telemetry/testing/trace_schema.js'; import { delay } from '../utils.js'; import { VAD, type VADEvent, VADEventType, VADStream } from '../vad.js'; import { Agent } from './agent.js'; @@ -299,6 +300,8 @@ describe.sequential('session lifecycle spans', () => { expect(userStates.some((e) => e.attributes?.[traceTypes.ATTR_NEW_STATE] === 'speaking')).toBe( true, ); + // the whole tree, not just the edges this test names (telemetry/testing/trace_schema) + assertTraceWellFormed(exporter.getFinishedSpans()); }); it("copies a linked SIP participant's attributes with only the number tagged as PII", async () => { @@ -333,5 +336,7 @@ describe.sequential('session lifecycle spans', () => { const linked = root.events.filter((event) => event.name === 'participant_linked'); expect(linked).toHaveLength(1); expect(linked[0]!.attributes?.[traceTypes.ATTR_PARTICIPANT_KIND]).toBe('SIP'); + // the whole tree, not just the edges this test names (telemetry/testing/trace_schema) + assertTraceWellFormed(exporter.getFinishedSpans()); }); });