diff --git a/.changeset/genai-semconv-and-pii-stripping.md b/.changeset/genai-semconv-and-pii-stripping.md new file mode 100644 index 000000000..8dd3c4874 --- /dev/null +++ b/.changeset/genai-semconv-and-pii-stripping.md @@ -0,0 +1,35 @@ +--- +'@livekit/agents': minor +--- + +Emit the full OpenTelemetry GenAI semantic conventions on agent spans. + +Spans now carry the standard `gen_ai.*` attributes — operation, provider, request/response +model, per-modality token usage, cached tokens, finish reasons, time-to-first-chunk, tool +name/type/call id, and the `gen_ai.input.messages` / `gen_ai.output.messages` / +`gen_ai.system_instructions` / `gen_ai.tool.definitions` content payloads — so Datadog Agent +Observability, Langfuse and any other backend that reads the conventions (OTel v1.37+) +renders a LiveKit trace without a custom mapping. The session maps to `invoke_workflow`, an +agent turn to `invoke_agent`, inference to `chat` (realtime turns to `generate_content` on +their own nested span), and tool execution to `execute_tool`. + +`gen_ai.provider.name` now reports the registry spelling the convention requires. Plugins +expose `provider` either as a display name (`MistralAI`, `AWS Bedrock`, `Vertex AI`) or as the +client's base-URL host (`api.openai.com`, `api.anthropic.com`); both are normalized, so +`openai`, `anthropic`, `mistral_ai`, `aws.bedrock`, `gcp.vertex_ai`, `gcp.gemini`, `x_ai`, +`groq` and `perplexity` are recognized by GenAI backends. A provider outside the registry +keeps its own id, which the convention allows. + +Conversational content reaches every configured exporter, as before. To withhold it from a +third-party pipeline while LiveKit Cloud keeps receiving it, pass `allowPii: false`: + +```ts +telemetry.setTracerProvider(provider, { registerSpanProcessor, allowPii: false }); +``` + +or set `LIVEKIT_TELEMETRY_ALLOW_PII=0` when the framework adopts the ambient OpenTelemetry +provider and there is no call site. What LiveKit Cloud receives stays governed by the +project's PII setting in the dashboard; when that mandates redaction, PII is withheld from +every destination and `allowPii` does not weaken it. Content can be dropped entirely with +`telemetry.genAI.setCaptureContent(false)` or +`OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=false`. diff --git a/agents/etc/agents.api.md b/agents/etc/agents.api.md index fd7d36783..1bd81b498 100644 --- a/agents/etc/agents.api.md +++ b/agents/etc/agents.api.md @@ -532,7 +532,7 @@ export class AgentSession extends AgentSession_base force?: boolean; }): Future; // (undocumented) - get interruptionDetection(): "vad" | "adaptive" | undefined; + get interruptionDetection(): "adaptive" | "vad" | undefined; // @internal (undocumented) readonly _keytermDetector: KeytermDetector; get keyterms(): string[]; @@ -1207,6 +1207,11 @@ const ATTR_EOU_SOURCE = "lk.eou.source"; // @public (undocumented) const ATTR_EOU_UNLIKELY_THRESHOLD = "lk.eou.unlikely_threshold"; +// Warning: (ae-missing-release-tag) "ATTR_ERROR_TYPE" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_ERROR_TYPE = "error.type"; + // Warning: (ae-missing-release-tag) "ATTR_EXCEPTION_MESSAGE" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -1252,14 +1257,59 @@ const ATTR_FUNCTION_TOOL_OUTPUT = "lk.pii.function_tool.output"; // @public (undocumented) const ATTR_FUNCTION_TOOLS = "lk.function_tools"; +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_AGENT_NAME" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_AGENT_NAME = "gen_ai.agent.name"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_CONVERSATION_ID" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_CONVERSATION_ID = "gen_ai.conversation.id"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_EVALUATION_EXPLANATION" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_EVALUATION_EXPLANATION = "gen_ai.evaluation.explanation"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_INPUT_MESSAGES" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_INPUT_MESSAGES = "gen_ai.input.messages"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_MEMORY_QUERY_TEXT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_MEMORY_QUERY_TEXT = "gen_ai.memory.query.text"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_MEMORY_RECORDS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_MEMORY_RECORDS = "gen_ai.memory.records"; + // Warning: (ae-missing-release-tag) "ATTR_GEN_AI_OPERATION_NAME" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) const ATTR_GEN_AI_OPERATION_NAME = "gen_ai.operation.name"; -// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_PROVIDER_NAME" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_OUTPUT_MESSAGES" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_OUTPUT_MESSAGES = "gen_ai.output.messages"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_OUTPUT_TYPE" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_OUTPUT_TYPE = "gen_ai.output.type"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_PROMPT_VARIABLE" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public +const ATTR_GEN_AI_PROMPT_VARIABLE = "gen_ai.prompt.variable"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_PROVIDER_NAME" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) const ATTR_GEN_AI_PROVIDER_NAME = "gen_ai.provider.name"; // Warning: (ae-missing-release-tag) "ATTR_GEN_AI_REQUEST_MODEL" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -1267,6 +1317,101 @@ const ATTR_GEN_AI_PROVIDER_NAME = "gen_ai.provider.name"; // @public (undocumented) const ATTR_GEN_AI_REQUEST_MODEL = "gen_ai.request.model"; +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_REQUEST_STREAM" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_REQUEST_STREAM = "gen_ai.request.stream"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_RESPONSE_FINISH_REASONS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_RESPONSE_FINISH_REASONS = "gen_ai.response.finish_reasons"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_RESPONSE_ID" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_RESPONSE_ID = "gen_ai.response.id"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_RESPONSE_MODEL" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_RESPONSE_MODEL = "gen_ai.response.model"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +const ATTR_GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK = "gen_ai.response.time_to_first_chunk"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_RETRIEVAL_DOCUMENTS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_RETRIEVAL_DOCUMENTS = "gen_ai.retrieval.documents"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_RETRIEVAL_QUERY_TEXT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_RETRIEVAL_QUERY_TEXT = "gen_ai.retrieval.query.text"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_SYSTEM_INSTRUCTIONS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_SYSTEM_INSTRUCTIONS = "gen_ai.system_instructions"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_TOOL_CALL_ARGUMENTS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_TOOL_CALL_ARGUMENTS = "gen_ai.tool.call.arguments"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_TOOL_CALL_ID" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_TOOL_CALL_ID = "gen_ai.tool.call.id"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_TOOL_CALL_RESULT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_TOOL_CALL_RESULT = "gen_ai.tool.call.result"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_TOOL_DEFINITIONS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_TOOL_DEFINITIONS = "gen_ai.tool.definitions"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_TOOL_DESCRIPTION" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_TOOL_DESCRIPTION = "gen_ai.tool.description"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_TOOL_NAME" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_TOOL_NAME = "gen_ai.tool.name"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_TOOL_TYPE" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_TOOL_TYPE = "gen_ai.tool.type"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_USAGE_AUDIO_INPUT_TOKENS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_USAGE_AUDIO_INPUT_TOKENS = "gen_ai.usage.audio.input_tokens"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_USAGE_AUDIO_OUTPUT_TOKENS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_USAGE_AUDIO_OUTPUT_TOKENS = "gen_ai.usage.audio.output_tokens"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS = "gen_ai.usage.cache_read.input_tokens"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_USAGE_CACHE_WRITE_INPUT_TOKENS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_USAGE_CACHE_WRITE_INPUT_TOKENS = "gen_ai.usage.cache_write.input_tokens"; + // Warning: (ae-missing-release-tag) "ATTR_GEN_AI_USAGE_INPUT_AUDIO_TOKENS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -1302,6 +1447,31 @@ const ATTR_GEN_AI_USAGE_OUTPUT_TEXT_TOKENS = "gen_ai.usage.output_text_tokens"; // @public (undocumented) const ATTR_GEN_AI_USAGE_OUTPUT_TOKENS = "gen_ai.usage.output_tokens"; +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_USAGE_REASONING_OUTPUT_TOKENS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_USAGE_REASONING_OUTPUT_TOKENS = "gen_ai.usage.reasoning.output_tokens"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_USAGE_REASONING_TOKENS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_USAGE_REASONING_TOKENS = "gen_ai.usage.reasoning_tokens"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_USAGE_TEXT_INPUT_TOKENS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_USAGE_TEXT_INPUT_TOKENS = "gen_ai.usage.text.input_tokens"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_USAGE_TEXT_OUTPUT_TOKENS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_USAGE_TEXT_OUTPUT_TOKENS = "gen_ai.usage.text.output_tokens"; + +// Warning: (ae-missing-release-tag) "ATTR_GEN_AI_WORKFLOW_NAME" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const ATTR_GEN_AI_WORKFLOW_NAME = "gen_ai.workflow.name"; + // 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) @@ -2313,6 +2483,20 @@ interface ChatMessageEvent { type: 'message'; } +// Warning: (ae-missing-release-tag) "ChatMessagePayload" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +interface ChatMessagePayload { + // (undocumented) + [key: string]: unknown; + // (undocumented) + finishReason?: string; + // (undocumented) + parts: MessagePart[]; + // (undocumented) + role: string; +} + // Warning: (ae-missing-release-tag) "ChatRole" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -2573,6 +2757,11 @@ export interface ConnectionPoolOptions { maxSessionDuration?: number; } +// Warning: (ae-missing-release-tag) "conversationId" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function conversationId(): string | undefined; + // Warning: (ae-missing-release-tag) "ConversationItemAddedEvent" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -3194,6 +3383,11 @@ const EVENT_GEN_AI_ASSISTANT_MESSAGE = "gen_ai.assistant.message"; // @public (undocumented) const EVENT_GEN_AI_CHOICE = "gen_ai.choice"; +// Warning: (ae-missing-release-tag) "EVENT_GEN_AI_CLIENT_INFERENCE_OPERATION_DETAILS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const EVENT_GEN_AI_CLIENT_INFERENCE_OPERATION_DETAILS = "gen_ai.client.inference.operation.details"; + // Warning: (ae-missing-release-tag) "EVENT_GEN_AI_SYSTEM_MESSAGE" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -3633,6 +3827,14 @@ export class FinalizeSimulationError extends Error { readonly userVerdict: AgentSession_2.SessionResponse_FinalizeSimulationResponse_SimulationVerdict | undefined; } +// Warning: (ae-missing-release-tag) "finishReasonFor" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +function finishReasonFor(params: { + functionCalls?: readonly unknown[]; + interrupted?: boolean; +}): string; + // Warning: (ae-missing-release-tag) "FishAudioModels" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -3904,6 +4106,90 @@ interface GatewayOptions { apiSecret: string; } +// Warning: (ae-missing-release-tag) "GEN_AI_PROVIDER_NAMES" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +const GEN_AI_PROVIDER_NAMES: ReadonlySet; + +declare namespace genAI { + export { + setCaptureContent, + withInferenceTracking, + markInferenceSpanRecorded, + toSystemInstructions, + toInputMessages, + toOutputMessages, + toToolDefinitions, + finishReasonFor, + conversationId, + setContentAttributes, + setRequestAttributes, + setResponseAttributes, + setUsageAttributes, + realtimeUsageAttributes, + setToolAttributes, + setToolResult, + setErrorType, + setAgentAttributes, + setWorkflowAttributes, + MessagePart, + ChatMessagePayload, + InferenceMarker, + ToolCallLike + } +} + +// Warning: (ae-missing-release-tag) "GenAIFinishReason" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +const GenAIFinishReason: { + readonly STOP: "stop"; + readonly LENGTH: "length"; + readonly CONTENT_FILTER: "content_filter"; + readonly TOOL_CALL: "tool_call"; + readonly COMPACTION: "compaction"; + readonly ERROR: "error"; +}; + +// Warning: (ae-missing-release-tag) "GenAIOperationName" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +const GenAIOperationName: { + readonly CHAT: "chat"; + readonly GENERATE_CONTENT: "generate_content"; + readonly TEXT_COMPLETION: "text_completion"; + readonly EMBEDDINGS: "embeddings"; + readonly RETRIEVAL: "retrieval"; + readonly FETCH_RESPONSE: "fetch_response"; + readonly CREATE_AGENT: "create_agent"; + readonly INVOKE_AGENT: "invoke_agent"; + readonly EXECUTE_TOOL: "execute_tool"; + readonly INVOKE_WORKFLOW: "invoke_workflow"; + readonly PLAN: "plan"; + readonly SEARCH_MEMORY: "search_memory"; + readonly CREATE_MEMORY: "create_memory"; + readonly UPDATE_MEMORY: "update_memory"; + readonly UPSERT_MEMORY: "upsert_memory"; + readonly DELETE_MEMORY: "delete_memory"; + readonly CREATE_MEMORY_STORE: "create_memory_store"; + readonly DELETE_MEMORY_STORE: "delete_memory_store"; +}; + +// Warning: (ae-missing-release-tag) "GenAIOutputType" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +const GenAIOutputType: { + readonly TEXT: "text"; + readonly JSON: "json"; + readonly IMAGE: "image"; + readonly SPEECH: "speech"; +}; + +// Warning: (ae-missing-release-tag) "genAIProviderName" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function genAIProviderName(provider: string | undefined | null): string | undefined; + // Warning: (ae-missing-release-tag) "GenerationCreatedEvent" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -4125,6 +4411,14 @@ interface InferenceLLMOptions { strictToolSchema?: boolean; } +// Warning: (ae-missing-release-tag) "InferenceMarker" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +interface InferenceMarker { + // (undocumented) + recorded: boolean; +} + // Warning: (ae-internal-missing-underscore) The name "InferenceRunner" should be prefixed with an underscore because the declaration is marked as @internal // // @internal (undocumented) @@ -4492,6 +4786,7 @@ export class JobContext> { deleteRoom(roomName?: string): Promise; // (undocumented) get inferenceExecutor(): InferenceExecutor; + get inferenceHeaders(): Record; // (undocumented) get info(): RunningJobInfo; // Warning: (ae-forgotten-export) The symbol "ResolvedRecordingOptions" needs to be exported by the entry point index.d.ts @@ -5026,6 +5321,11 @@ export const logMetrics: (metrics: AgentMetrics) => void; // @public export function loopAudioFramesFromFile(filePath: string, options?: AudioDecodeOptions): AsyncGenerator; +// Warning: (ae-missing-release-tag) "markInferenceSpanRecorded" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function markInferenceSpanRecorded(): void; + // Warning: (ae-missing-release-tag) "MarkupInfo" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -5092,6 +5392,16 @@ export interface MessageGeneration { textStream: ReadableStream_2; } +// Warning: (ae-missing-release-tag) "MessagePart" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +interface MessagePart { + // (undocumented) + [key: string]: unknown; + // (undocumented) + type: string; +} + // Warning: (ae-missing-release-tag) "MetadataLogProcessor" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -5425,6 +5735,18 @@ export interface ParticipantTranscriptionOutputOptions extends TranscriptionOutp jsonFormat?: boolean; } +// Warning: (ae-missing-release-tag) "PIIFilteringLogProcessor" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +class PIIFilteringLogProcessor implements LogRecordProcessor { + // (undocumented) + forceFlush(): Promise; + // (undocumented) + onEmit(logRecord: SdkLogRecord): void; + // (undocumented) + shutdown(): Promise; +} + // Warning: (ae-missing-release-tag) "PinoCloudExporter" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -5571,6 +5893,13 @@ export enum PluginEventTypes { // @public (undocumented) export type ProviderFormat = 'openai' | 'openai.responses' | 'google' | 'mistralai'; +// @internal +const _providerTables: { + byHost: Record; + byHostSuffix: readonly [string, string][]; + byName: Record; +}; + // Warning: (ae-missing-release-tag) "ProviderTool" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -5779,6 +6108,11 @@ export abstract class RealtimeSession extends EventEmitter { export interface RealtimeSessionReconnectedEvent { } +// Warning: (ae-missing-release-tag) "realtimeUsageAttributes" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function realtimeUsageAttributes(metrics: RealtimeModelMetrics): Attributes; + // Warning: (ae-missing-release-tag) "RecognitionUsage" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -5822,6 +6156,11 @@ function recordRealtimeMetrics(span: Span, metrics: RealtimeModelMetrics): void; // @public (undocumented) const REDACTED_EXCEPTION_MESSAGE = "exception details redacted"; +// Warning: (ae-missing-release-tag) "redactionEnabled" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function redactionEnabled(spanAttributes?: Attributes): boolean; + // Warning: (ae-missing-release-tag) "rejectOnAbort" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -6228,7 +6567,7 @@ export type ScenarioUserdata = { // // @public (undocumented) const sendDtmfEvents: FunctionTool< { -events: ("0" | "1" | "2" | "3" | "4" | "5" | "6" | "7" | "#" | "*" | "8" | "9" | "A" | "B" | "C" | "D")[]; +events: ("0" | "1" | "2" | "3" | "4" | "5" | "6" | "7" | "8" | "9" | "#" | "*" | "A" | "B" | "C" | "D")[]; }, unknown, string>; // Warning: (ae-missing-release-tag) "SentenceStream" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -6500,6 +6839,74 @@ export type SessionUsageUpdatedEvent = { createdAt: number; }; +// Warning: (ae-missing-release-tag) "setAgentAttributes" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +function setAgentAttributes(span: Span, params: { + operation: string; + agentName: string; +}): void; + +// Warning: (ae-missing-release-tag) "setCaptureContent" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function setCaptureContent(enabled: boolean): void; + +// Warning: (ae-missing-release-tag) "setContentAttributes" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function setContentAttributes(span: Span, content: { + systemInstructions?: MessagePart[]; + inputMessages?: ChatMessagePayload[]; + outputMessages?: ChatMessagePayload[]; + toolDefinitions?: MessagePart[]; +}): void; + +// Warning: (ae-missing-release-tag) "setErrorType" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function setErrorType(span: Span, error: Error | string): void; + +// Warning: (ae-missing-release-tag) "setRequestAttributes" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function setRequestAttributes(span: Span, params: { + operation: string; + provider?: string; + model?: string; + stream?: boolean; + outputType?: string; +}): void; + +// Warning: (ae-missing-release-tag) "setResponseAttributes" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +function setResponseAttributes(span: Span, params: { + responseId?: string; + model?: string; + finishReasons?: string[]; + timeToFirstChunk?: number; +}): void; + +// Warning: (ae-missing-release-tag) "setToolAttributes" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function setToolAttributes(span: Span, params: { + name: string; + callId?: string; + toolType?: string; + description?: string; + args?: string; +}): void; + +// Warning: (ae-missing-release-tag) "setToolResult" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +function setToolResult(span: Span, params: { + result?: string; + isError: boolean; +}): void; + // Warning: (ae-missing-release-tag) "setTracerProvider" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // Warning: (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "FanoutSpanProcessor" // @@ -6510,6 +6917,7 @@ function setTracerProvider(provider: TracerProvider, options?: SetTracerProvider // // @public interface SetTracerProviderOptions { + allowPii?: boolean; createCloudSpanProcessor?: (options: CloudSpanProcessorOptions) => SpanProcessor; metadata?: Attributes; // Warning: (ae-forgotten-export) The symbol "SpanProcessorRegistrar" needs to be exported by the entry point index.d.ts @@ -6526,6 +6934,24 @@ function setupCloudTracer(options: ObservabilityEndpoint & { metadata?: Attributes; }): Promise; +// Warning: (ae-missing-release-tag) "setUsageAttributes" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function setUsageAttributes(span: Span, usage: { + promptTokens?: number; + completionTokens?: number; + promptCachedTokens?: number; + cacheCreationTokens?: number; + reasoningTokens?: number; +}): void; + +// Warning: (ae-missing-release-tag) "setWorkflowAttributes" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function setWorkflowAttributes(span: Span, params: { + name: string; +}): void; + // Warning: (ae-missing-release-tag) "shortuuid" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public @@ -7501,6 +7927,7 @@ declare namespace telemetry { export { ExtraDetailsProcessor, MetadataLogProcessor, + PIIFilteringLogProcessor, ObservabilityEndpoint, SimpleOTLPHttpLogExporter, SimpleLogRecord, @@ -7513,6 +7940,8 @@ declare namespace telemetry { PinoCloudExporterConfig, PinoCloudExporterUrlConfig, PinoLogObject, + genAI, + REDACTED_EXCEPTION_MESSAGE, traceTypes, FanoutSpanProcessor, flushOtelLogs, @@ -7524,9 +7953,9 @@ declare namespace telemetry { SetTracerProviderOptions, SpanProcessorLike, StartSpanOptions, - REDACTED_EXCEPTION_MESSAGE, recordException, recordRealtimeMetrics, + redactionEnabled, RecordExceptionOptions } } @@ -7708,6 +8137,11 @@ export interface TimedString { // @public export function toError(error: unknown): Error; +// Warning: (ae-missing-release-tag) "toInputMessages" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function toInputMessages(chatCtx: ChatContext): ChatMessagePayload[]; + // Warning: (ae-missing-release-tag) "toJsonSchema" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -7794,6 +8228,18 @@ export interface ToolCalledEvent { ctx: RunContext; } +// Warning: (ae-missing-release-tag) "ToolCallLike" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +interface ToolCallLike { + // (undocumented) + args: string; + // (undocumented) + callId: string; + // (undocumented) + name: string; +} + // Warning: (ae-missing-release-tag) "ToolChoice" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) @@ -7942,6 +8388,15 @@ export interface ToolsetCreateOptions { // @public (undocumented) export type ToolType = 'function' | 'provider'; +// Warning: (ae-missing-release-tag) "toOutputMessages" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function toOutputMessages(params: { + text?: string; + functionCalls?: readonly ToolCallLike[]; + finishReason?: string; +}): ChatMessagePayload[]; + // Warning: (ae-internal-missing-underscore) The name "toSnakeCaseDeep" should be prefixed with an underscore because the declaration is marked as @internal // // @internal @@ -7952,6 +8407,11 @@ export function toSnakeCaseDeep(value: unknown): unknown; // @public (undocumented) export function toStream(iterable: AsyncIterable): ReadableStream_2; +// Warning: (ae-missing-release-tag) "toSystemInstructions" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function toSystemInstructions(chatCtx: ChatContext): MessagePart[]; + // Warning: (ae-missing-release-tag) "toToolContext" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // Warning: (ae-missing-release-tag) "toToolContext" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // @@ -7961,6 +8421,11 @@ export function toToolContext(input: ToolContextLike // @public (undocumented) export function toToolContext(input: ToolContextLike | undefined): ToolContext | undefined; +// Warning: (ae-missing-release-tag) "toToolDefinitions" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function toToolDefinitions(tools: readonly unknown[] | Record): MessagePart[]; + // Warning: (ae-forgotten-export) The symbol "DynamicTracer" needs to be exported by the entry point index.d.ts // Warning: (ae-missing-release-tag) "tracer" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // @@ -7969,6 +8434,7 @@ const tracer: DynamicTracer; declare namespace traceTypes { export { + genAIProviderName, ATTR_SPEECH_ID, ATTR_AGENT_LABEL, ATTR_START_TIME, @@ -8033,20 +8499,60 @@ declare namespace traceTypes { ATTR_REALTIME_MODEL_METRICS, ATTR_E2E_LATENCY, ATTR_GEN_AI_OPERATION_NAME, - ATTR_GEN_AI_REQUEST_MODEL, ATTR_GEN_AI_PROVIDER_NAME, + ATTR_GEN_AI_REQUEST_MODEL, + ATTR_GEN_AI_REQUEST_STREAM, + ATTR_GEN_AI_RESPONSE_ID, + ATTR_GEN_AI_RESPONSE_MODEL, + ATTR_GEN_AI_RESPONSE_FINISH_REASONS, + ATTR_GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK, ATTR_GEN_AI_USAGE_INPUT_TOKENS, ATTR_GEN_AI_USAGE_OUTPUT_TOKENS, + ATTR_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS, + ATTR_GEN_AI_USAGE_CACHE_WRITE_INPUT_TOKENS, + ATTR_GEN_AI_USAGE_REASONING_OUTPUT_TOKENS, + ATTR_GEN_AI_USAGE_TEXT_INPUT_TOKENS, + ATTR_GEN_AI_USAGE_TEXT_OUTPUT_TOKENS, + ATTR_GEN_AI_USAGE_AUDIO_INPUT_TOKENS, + ATTR_GEN_AI_USAGE_AUDIO_OUTPUT_TOKENS, + ATTR_GEN_AI_CONVERSATION_ID, + ATTR_GEN_AI_AGENT_NAME, + ATTR_GEN_AI_TOOL_NAME, + ATTR_GEN_AI_TOOL_CALL_ID, + ATTR_GEN_AI_TOOL_DESCRIPTION, + ATTR_GEN_AI_TOOL_TYPE, + ATTR_GEN_AI_TOOL_CALL_ARGUMENTS, + ATTR_GEN_AI_TOOL_CALL_RESULT, + ATTR_GEN_AI_TOOL_DEFINITIONS, + ATTR_GEN_AI_SYSTEM_INSTRUCTIONS, + ATTR_GEN_AI_INPUT_MESSAGES, + ATTR_GEN_AI_OUTPUT_MESSAGES, + ATTR_GEN_AI_OUTPUT_TYPE, + ATTR_GEN_AI_RETRIEVAL_DOCUMENTS, + ATTR_GEN_AI_RETRIEVAL_QUERY_TEXT, + ATTR_GEN_AI_MEMORY_QUERY_TEXT, + ATTR_GEN_AI_MEMORY_RECORDS, + ATTR_GEN_AI_EVALUATION_EXPLANATION, + ATTR_GEN_AI_PROMPT_VARIABLE, + ATTR_GEN_AI_WORKFLOW_NAME, + ATTR_ERROR_TYPE, + GenAIOperationName, + GenAIOutputType, + GenAIFinishReason, + GEN_AI_PROVIDER_NAMES, + _providerTables, ATTR_GEN_AI_USAGE_INPUT_TEXT_TOKENS, ATTR_GEN_AI_USAGE_INPUT_AUDIO_TOKENS, ATTR_GEN_AI_USAGE_INPUT_CACHED_TOKENS, ATTR_GEN_AI_USAGE_OUTPUT_TEXT_TOKENS, ATTR_GEN_AI_USAGE_OUTPUT_AUDIO_TOKENS, + ATTR_GEN_AI_USAGE_REASONING_TOKENS, EVENT_GEN_AI_SYSTEM_MESSAGE, EVENT_GEN_AI_USER_MESSAGE, EVENT_GEN_AI_ASSISTANT_MESSAGE, EVENT_GEN_AI_TOOL_MESSAGE, EVENT_GEN_AI_CHOICE, + EVENT_GEN_AI_CLIENT_INFERENCE_OPERATION_DETAILS, ATTR_EXCEPTION_TRACE, ATTR_EXCEPTION_TYPE, ATTR_EXCEPTION_MESSAGE, @@ -9003,6 +9509,11 @@ interface WarmTransferTaskOptions { vad?: VAD | null; } +// Warning: (ae-missing-release-tag) "withInferenceTracking" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public +function withInferenceTracking(fn: (marker: InferenceMarker) => T): T; + // Warning: (ae-missing-release-tag) "withMockTools" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // Warning: (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "Disposable" // @@ -9177,9 +9688,9 @@ export const zipFunctionCallsAndOutputs: (event: FunctionToolsExecutedEvent) => // src/utils.ts:550:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "cancelled" // src/voice/agent_session.ts:380:3 - (ae-unresolved-link) The @link reference could not be resolved: This type of declaration is not supported yet by the resolver // src/voice/agent_session.ts:994:5 - (ae-forgotten-export) The symbol "RecordingOptions" needs to be exported by the entry point index.d.ts -// src/voice/agent_session.ts:1644:5 - (ae-forgotten-export) The symbol "STTError" needs to be exported by the entry point index.d.ts -// src/voice/agent_session.ts:1644:5 - (ae-forgotten-export) The symbol "TTSError" needs to be exported by the entry point index.d.ts -// src/voice/agent_session.ts:1644:5 - (ae-forgotten-export) The symbol "LLMError" needs to be exported by the entry point index.d.ts +// src/voice/agent_session.ts:1647:5 - (ae-forgotten-export) The symbol "STTError" needs to be exported by the entry point index.d.ts +// src/voice/agent_session.ts:1647:5 - (ae-forgotten-export) The symbol "TTSError" needs to be exported by the entry point index.d.ts +// src/voice/agent_session.ts:1647:5 - (ae-forgotten-export) The symbol "LLMError" needs to be exported by the entry point index.d.ts // src/voice/amd.ts:314:3 - (ae-unresolved-link) The @link reference could not be resolved: The reference is ambiguous because "waitForTrackPublication" has more than one declaration; you need to add a TSDoc member reference selector // src/voice/amd.ts:314:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "gateListening" // src/voice/amd.ts:322:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "aclose" diff --git a/agents/src/llm/llm.ts b/agents/src/llm/llm.ts index 6270a130f..ebf2a0d54 100644 --- a/agents/src/llm/llm.ts +++ b/agents/src/llm/llm.ts @@ -7,7 +7,7 @@ import { EventEmitter } from 'node:events'; import { APIConnectionError, APIError } from '../_exceptions.js'; import { log } from '../log.js'; import type { LLMMetrics } from '../metrics/base.js'; -import { recordException, traceTypes, tracer } from '../telemetry/index.js'; +import { genAI, recordException, traceTypes, tracer } from '../telemetry/index.js'; import { type APIConnectOptions, intervalForRetry } from '../types.js'; import { AsyncIterableQueue, Task, delay, startSoon, toError } from '../utils.js'; import { type ChatContext, type ChatRole, type FunctionCall } from './chat_context.js'; @@ -217,6 +217,11 @@ export abstract class LLMStream implements AsyncIterableIterator { this.closed = true; }); + // tells an enclosing `llm_node` span that this call is instrumented, so it does not + // record the convention's attributes a second time. Read here rather than in mainTask, + // which startSoon defers out of the node's context. + genAI.markInferenceSpanRecorded(); + // this is a hack to immitate asyncio.create_task so that mainTask // is run **after** the constructor has finished. Otherwise we get // runtime error when trying to access class variables in the @@ -232,9 +237,26 @@ export abstract class LLMStream implements AsyncIterableIterator { }); } + /** The GenAI inference span's request side, per the OTel GenAI conventions. */ + private recordGenAIRequest(span: Span) { + genAI.setRequestAttributes(span, { + operation: traceTypes.GenAIOperationName.CHAT, + provider: this.#llm.provider, + model: this.#llm.model, + stream: true, + outputType: traceTypes.GenAIOutputType.TEXT, + }); + genAI.setContentAttributes(span, { + systemInstructions: genAI.toSystemInstructions(this.#chatCtx), + inputMessages: genAI.toInputMessages(this.#chatCtx), + toolDefinitions: this.#toolCtx ? genAI.toToolDefinitions(this.#toolCtx.functionTools) : [], + }); + } + private _mainTaskImpl = async (span: Span) => { this.#llmRequestSpan = span; span.setAttribute(traceTypes.ATTR_GEN_AI_REQUEST_MODEL, this.#llm.model); + this.recordGenAIRequest(span); for (let i = 0; i < this._connOptions.maxRetry + 1; i++) { try { @@ -292,11 +314,12 @@ export abstract class LLMStream implements AsyncIterableIterator { } }; - private mainTask = async () => - tracer.startActiveSpan(async (span) => this._mainTaskImpl(span), { + private mainTask = async () => { + return tracer.startActiveSpan(async (span) => this._mainTaskImpl(span), { name: 'llm_request', endOnExit: false, }); + }; private emitError({ error, recoverable }: { error: Error; recoverable: boolean }) { this.#llm.emit('error', { @@ -314,6 +337,9 @@ export abstract class LLMStream implements AsyncIterableIterator { let requestId = ''; let usage: CompletionUsage | undefined; let completionStartTime: string | undefined; + // accumulated for `gen_ai.output.messages` + let responseContent = ''; + const toolCalls: FunctionCall[] = []; for await (const ev of this.queue) { if (this.abortController.signal.aborted) { @@ -330,6 +356,12 @@ export abstract class LLMStream implements AsyncIterableIterator { ttft = process.hrtime.bigint() - startTime; completionStartTime = new Date().toISOString(); } + if (ev.delta?.content) { + responseContent += ev.delta.content; + } + if (ev.delta?.toolCalls?.length) { + toolCalls.push(...ev.delta.toolCalls); + } if (ev.usage) { usage = ev.usage; } @@ -366,9 +398,25 @@ export abstract class LLMStream implements AsyncIterableIterator { if (this.#llmRequestSpan) { this.#llmRequestSpan.setAttribute(traceTypes.ATTR_LLM_METRICS, JSON.stringify(metrics)); - this.#llmRequestSpan.setAttributes({ - [traceTypes.ATTR_GEN_AI_USAGE_INPUT_TOKENS]: metrics.promptTokens, - [traceTypes.ATTR_GEN_AI_USAGE_OUTPUT_TOKENS]: metrics.completionTokens, + // the GenAI response side; the request side was recorded at span creation + genAI.setUsageAttributes(this.#llmRequestSpan, metrics); + + const finishReason = genAI.finishReasonFor({ + functionCalls: toolCalls, + interrupted: metrics.cancelled, + }); + genAI.setResponseAttributes(this.#llmRequestSpan, { + responseId: requestId || undefined, + model: this.#llm.model, + finishReasons: [finishReason], + timeToFirstChunk: metrics.ttftMs >= 0 ? metrics.ttftMs / 1000 : undefined, + }); + genAI.setContentAttributes(this.#llmRequestSpan, { + outputMessages: genAI.toOutputMessages({ + text: responseContent, + functionCalls: toolCalls, + finishReason, + }), }); if (completionStartTime) { diff --git a/agents/src/telemetry/gen_ai.test.ts b/agents/src/telemetry/gen_ai.test.ts new file mode 100644 index 000000000..40d8840cc --- /dev/null +++ b/agents/src/telemetry/gen_ai.test.ts @@ -0,0 +1,250 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { Span } from '@opentelemetry/api'; +import { + BasicTracerProvider, + InMemorySpanExporter, + SimpleSpanProcessor, +} from '@opentelemetry/sdk-trace-node'; +import { afterEach, describe, expect, it } from 'vitest'; +import { z } from 'zod'; +import { ChatContext, FunctionCall, FunctionCallOutput } from '../llm/chat_context.js'; +import { tool } from '../llm/tool_context.js'; +import * as genAI from './gen_ai.js'; +import * as traceTypes from './trace_types.js'; + +// The shapes asserted here follow the gen_ai.input.messages / output.messages / +// system_instructions / tool.definitions JSON schemas from +// https://github.com/open-telemetry/semantic-conventions-genai, so builder drift is +// caught here rather than by a backend that silently drops the span. + +function chatContext(): ChatContext { + const ctx = ChatContext.empty(); + ctx.addMessage({ role: 'system', content: 'You are a helpful agent.' }); + ctx.addMessage({ role: 'user', content: "What's the weather in Paris?" }); + ctx.insert(new FunctionCall({ callId: 'call_1', name: 'get_weather', args: '{"loc": "Paris"}' })); + ctx.insert( + new FunctionCallOutput({ + callId: 'call_1', + name: 'get_weather', + output: '{"temp": 14}', + isError: false, + }), + ); + ctx.addMessage({ role: 'assistant', content: "It's 14 degrees in Paris." }); + return ctx; +} + +function exportingSpan(name = 'llm_request'): { span: Span; exporter: InMemorySpanExporter } { + const exporter = new InMemorySpanExporter(); + const provider = new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] }); + return { span: provider.getTracer('test').startSpan(name), exporter }; +} + +afterEach(() => { + genAI.setCaptureContent(true); +}); + +describe('gen_ai builders', () => { + it("produces the convention's shapes", () => { + const ctx = chatContext(); + + // instructions are reported separately from history, not as a system message + expect(genAI.toSystemInstructions(ctx)).toEqual([ + { type: 'text', content: 'You are a helpful agent.' }, + ]); + + const messages = genAI.toInputMessages(ctx); + expect(messages.map((m) => m.role)).toEqual(['user', 'assistant', 'tool', 'assistant']); + expect(messages[1]!.parts[0]).toEqual({ + type: 'tool_call', + id: 'call_1', + name: 'get_weather', + arguments: { loc: 'Paris' }, + }); + // a serialized payload is deserialized, as the convention asks of instrumentations + expect(messages[2]!.parts[0]!.response).toEqual({ temp: 14 }); + + const output = genAI.toOutputMessages({ + text: 'one moment', + functionCalls: [{ callId: 'call_9', name: 'lookup', args: '{"q": "x"}' }], + }); + expect(output[0]!.role).toBe('assistant'); + expect(output[0]!.parts.map((p) => p.type)).toEqual(['text', 'tool_call']); + expect(genAI.toOutputMessages({ text: '' })).toEqual([]); + + const getWeather = tool({ + name: 'get_weather', + description: 'Get the current weather in a given location', + parameters: z.object({ location: z.string() }), + execute: async () => 'sunny', + }); + // `parameters` is omitted: the convention marks it NOT RECOMMENDED by default + expect(genAI.toToolDefinitions([getWeather])).toEqual([ + { + type: 'function', + name: 'get_weather', + description: 'Get the current weather in a given location', + }, + ]); + }); + + it("uses the convention's finish-reason values", () => { + expect(genAI.finishReasonFor({})).toBe('stop'); + expect(genAI.finishReasonFor({ interrupted: true })).toBe('error'); + expect(genAI.finishReasonFor({ functionCalls: [{}] })).toBe('tool_call'); + // a tool call emitted before the generation failed is not a successful handoff + expect(genAI.finishReasonFor({ functionCalls: [{}], interrupted: true })).toBe('error'); + }); +}); + +describe('gen_ai span attributes', () => { + it('uses the registry names on an inference span', () => { + const { span, exporter } = exportingSpan(); + genAI.setRequestAttributes(span, { + operation: traceTypes.GenAIOperationName.CHAT, + // a LiveKit plugin id, normalized to the registry spelling + provider: 'bedrock', + model: 'claude-sonnet-4', + stream: true, + }); + genAI.setResponseAttributes(span, { + responseId: 'resp_1', + finishReasons: ['stop'], + timeToFirstChunk: 0.4, + }); + genAI.setContentAttributes(span, { inputMessages: [{ role: 'user', parts: [] }] }); + span.end(); + + const attrs = exporter.getFinishedSpans()[0]!.attributes; + expect(attrs['gen_ai.operation.name']).toBe('chat'); + expect(attrs['gen_ai.provider.name']).toBe('aws.bedrock'); + expect(attrs['gen_ai.request.model']).toBe('claude-sonnet-4'); + expect(attrs['gen_ai.request.stream']).toBe(true); + expect(attrs['gen_ai.response.id']).toBe('resp_1'); + expect(attrs['gen_ai.response.finish_reasons']).toEqual(['stop']); + expect(attrs['gen_ai.response.time_to_first_chunk']).toBe(0.4); + // attributes cannot hold structured values, so content is a JSON string + expect(JSON.parse(attrs['gen_ai.input.messages'] as string)).toEqual([ + { role: 'user', parts: [] }, + ]); + }); + + it('omits content but keeps metadata when content capture is off', () => { + const { span, exporter } = exportingSpan(); + genAI.setCaptureContent(false); + genAI.setRequestAttributes(span, { + operation: traceTypes.GenAIOperationName.CHAT, + model: 'gpt-4o', + }); + genAI.setContentAttributes(span, { inputMessages: [{ role: 'user', parts: [] }] }); + span.end(); + + const attrs = exporter.getFinishedSpans()[0]!.attributes; + expect(attrs['gen_ai.input.messages']).toBeUndefined(); + expect(attrs['gen_ai.request.model']).toBe('gpt-4o'); + }); + + it('reports usage details alongside the totals, never added to them', () => { + const { span, exporter } = exportingSpan(); + genAI.setUsageAttributes(span, { + promptTokens: 300, + completionTokens: 180, + promptCachedTokens: 40, + cacheCreationTokens: 25, + reasoningTokens: 50, + }); + span.end(); + + const attrs = exporter.getFinishedSpans()[0]!.attributes; + expect(attrs['gen_ai.usage.input_tokens']).toBe(300); + expect(attrs['gen_ai.usage.output_tokens']).toBe(180); + expect(attrs['gen_ai.usage.cache_read.input_tokens']).toBe(40); + // both spellings, so the pipeline and realtime paths agree + expect(attrs['gen_ai.usage.input_cached_tokens']).toBe(40); + expect(attrs['gen_ai.usage.cache_write.input_tokens']).toBe(25); + expect(attrs['gen_ai.usage.reasoning.output_tokens']).toBe(50); + }); + + it('describes a tool execution as the execute_tool span', () => { + const { span, exporter } = exportingSpan('function_tool'); + genAI.setToolAttributes(span, { + name: 'get_weather', + callId: 'call_1', + description: 'Get the weather', + args: '{"location": "Paris"}', + }); + genAI.setToolResult(span, { result: '{"temp": 14}', isError: false }); + span.end(); + + const attrs = exporter.getFinishedSpans()[0]!.attributes; + expect(attrs['gen_ai.operation.name']).toBe('execute_tool'); + expect(attrs['gen_ai.tool.name']).toBe('get_weather'); + expect(attrs['gen_ai.tool.call.id']).toBe('call_1'); + expect(attrs['gen_ai.tool.type']).toBe('function'); + expect(JSON.parse(attrs['gen_ai.tool.call.arguments'] as string)).toEqual({ + location: 'Paris', + }); + expect(JSON.parse(attrs['gen_ai.tool.call.result'] as string)).toEqual({ temp: 14 }); + }); + + it('keeps error.type low-cardinality', () => { + const failed = exportingSpan('function_tool'); + genAI.setToolResult(failed.span, { result: 'boom', isError: true }); + failed.span.end(); + const toolAttrs = failed.exporter.getFinishedSpans()[0]!.attributes; + // "the result returned by the tool call (if any and if execution was successful)" + expect(toolAttrs['gen_ai.tool.call.result']).toBeUndefined(); + expect(toolAttrs['error.type']).toBe('tool_error'); + + const errored = exportingSpan(); + genAI.setErrorType(errored.span, Object.assign(new Error('rate limited'), { statusCode: 429 })); + errored.span.end(); + // a status code identifies the failure better than the error class + expect(errored.exporter.getFinishedSpans()[0]!.attributes['error.type']).toBe('429'); + }); +}); + +// Every `provider` value a plugin actually reports. The convention makes the registry +// spelling mandatory for a provider it enumerates, so a plugin returning a display name or a +// base-URL host must still land on it. +describe('provider normalization', () => { + it.each([ + // base-URL hosts, from the OpenAI-compatible clients + ['api.openai.com', 'openai'], + ['api.anthropic.com', 'anthropic'], + ['api.mistral.ai', 'mistral_ai'], + ['api.groq.com', 'groq'], + ['api.x.ai', 'x_ai'], + ['my-co.openai.azure.com', 'azure.ai.openai'], + ['bedrock-runtime.us-east-1.amazonaws.com', 'aws.bedrock'], + // display names + ['AWS Bedrock', 'aws.bedrock'], + ['Amazon', 'aws.bedrock'], + ['MistralAI', 'mistral_ai'], + ['Vertex AI', 'gcp.vertex_ai'], + ['Vertex AI Model Garden', 'gcp.vertex_ai'], + ['Gemini', 'gcp.gemini'], + ['google', 'gcp.gen_ai'], + ['xAI', 'x_ai'], + ['xai', 'x_ai'], + ['Perplexity', 'perplexity'], + // outside the registry: the convention allows a custom value, so it passes through + ['MiniMax', 'MiniMax'], + ['api.cerebras.ai', 'api.cerebras.ai'], + ])('resolves %s', (reported, expected) => { + expect(traceTypes.genAIProviderName(reported as string)).toBe(expected); + }); + + it('only targets registry values', () => { + // a typo in a mapping would emit a value no backend recognises + const { byHost, byHostSuffix, byName } = traceTypes._providerTables; + const targets = [ + ...Object.values(byHost), + ...Object.values(byName), + ...byHostSuffix.map(([, v]) => v), + ]; + expect(targets.filter((t) => !traceTypes.GEN_AI_PROVIDER_NAMES.has(t))).toEqual([]); + }); +}); diff --git a/agents/src/telemetry/gen_ai.ts b/agents/src/telemetry/gen_ai.ts new file mode 100644 index 000000000..66988ebe9 --- /dev/null +++ b/agents/src/telemetry/gen_ai.ts @@ -0,0 +1,538 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +/** + * Builders for the OpenTelemetry GenAI semantic conventions. + * + * Translates LiveKit's own types — {@link ChatContext}, {@link FunctionCall}, tool + * definitions, token usage — into the shapes the GenAI semantic conventions define, so a span + * LiveKit produces is understood by any backend that speaks the convention (Datadog Agent + * Observability, Langfuse, Braintrust, an OTLP collector) without a LiveKit-specific mapping. + * + * Message content follows the `gen_ai.input.messages` / `gen_ai.output.messages` / + * `gen_ai.system_instructions` JSON schemas from + * https://github.com/open-telemetry/semantic-conventions-genai. OpenTelemetry attributes + * cannot hold structured values, so the convention's fallback applies and each value is + * recorded as a JSON string. + * + * Content capture is on by default, matching the `lk.pii.*` content LiveKit already records, + * and can be turned off process-wide with {@link setCaptureContent} or the + * `OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT` environment variable. It is stripped + * before any exporter that is not LiveKit Cloud's regardless — see `telemetry/pii.ts`. + */ +import type { Attributes, Span } from '@opentelemetry/api'; +import { context as otelContext } from '@opentelemetry/api'; +import { getJobContext } from '../job.js'; +import type { ChatContext, ChatItem } from '../llm/chat_context.js'; +import type { RealtimeModelMetrics } from '../metrics/base.js'; +import * as traceTypes from './trace_types.js'; + +// Only type-only imports from `llm` here: a runtime import would pull the whole llm graph +// in behind the telemetry barrel, which llm itself imports. Chat items and tools are +// matched on their own `type` discriminants instead. + +const FALSY = new Set(['0', 'false', 'no', 'off']); + +// the env var name the GenAI conventions standardise for this opt-in +let captureContent = !FALSY.has( + (process.env.OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT ?? '').trim().toLowerCase(), +); + +/** + * Turn recording of GenAI message content on or off for the process. + * + * When off, spans still carry every non-content GenAI attribute (model, provider, token + * usage, finish reasons, tool names), but `gen_ai.input.messages`, `gen_ai.output.messages`, + * `gen_ai.system_instructions`, `gen_ai.tool.definitions` and + * `gen_ai.tool.call.{arguments,result}` are omitted. + */ +export function setCaptureContent(enabled: boolean): void { + captureContent = enabled; +} + +// --------------------------------------------------------------------------- +// message models +// --------------------------------------------------------------------------- + +export interface MessagePart { + type: string; + [key: string]: unknown; +} + +export interface ChatMessagePayload { + role: string; + parts: MessagePart[]; + finishReason?: string; + [key: string]: unknown; +} + +// A custom `llmNode` may do the inference itself — returning a plain string, streaming its +// own chunks, or calling a third-party engine — and never construct an LLMStream. Those paths +// have no nested `llm_request` span to carry the convention's attributes, so the node span +// records them instead. LLMStream marks the context when it does create one, which is what +// tells the two cases apart. +const INFERENCE_RECORDED = Symbol('lkInferenceRecorded'); + +export interface InferenceMarker { + recorded: boolean; +} + +/** Runs `fn` with a marker that fills in if an `llm_request` span is created inside it. */ +export function withInferenceTracking(fn: (marker: InferenceMarker) => T): T { + const marker: InferenceMarker = { recorded: false }; + return otelContext.with(otelContext.active().setValue(INFERENCE_RECORDED, marker), () => + fn(marker), + ); +} + +/** Called where an `llm_request` span is created, so the enclosing node stands down. */ +export function markInferenceSpanRecorded(): void { + const marker = otelContext.active().getValue(INFERENCE_RECORDED) as InferenceMarker | undefined; + if (marker) marker.recorded = true; +} + +function textPart(content: string): MessagePart { + return { type: 'text', content }; +} + +/** Best-effort deserialization, as the convention asks of instrumentations. */ +function maybeJson(raw: string): unknown { + if (!raw) return raw; + try { + return JSON.parse(raw); + } catch { + return raw; + } +} + +function messageParts(item: ChatItem): MessagePart[] { + const parts: MessagePart[] = []; + if (item.type === 'message') { + for (const content of item.content) { + if (typeof content === 'string') { + parts.push(textPart(content)); + } else if (content.type === 'instructions') { + parts.push(textPart(content.value)); + } else if (content.type === 'image_content') { + // a data: URL is inline bytes, which the convention models as a blob; recording the + // base64 payload on a span is never worth its size, so both forms are reported as a + // uri part without the payload + const image = content.image; + parts.push({ + type: 'uri', + modality: 'image', + mime_type: content.mimeType, + uri: typeof image === 'string' && !image.startsWith('data:') ? image : '', + }); + } else if (content.type === 'audio_content') { + const part: MessagePart = { type: 'blob', modality: 'audio', content: '' }; + if (content.transcript) { + // the transcript is the only audio content worth carrying; the frames are + // recorded separately by session recording, never on a span + part.transcript = content.transcript; + } + parts.push(part); + } + } + } else if (item.type === 'function_call') { + parts.push({ + type: 'tool_call', + id: item.callId, + name: item.name, + arguments: maybeJson(item.args), + }); + } else if (item.type === 'function_call_output') { + parts.push({ + type: 'tool_call_response', + id: item.callId, + response: maybeJson(item.output), + }); + } + return parts; +} + +/** + * The chat items to report, tolerating a context that carries none. + * + * Tracing must never be the thing that breaks an inference call, and a custom `llm_node` or a + * test double may hand us an object without an `items` list. + */ +function itemsOf(chatCtx: ChatContext | undefined): readonly ChatItem[] { + const items = chatCtx?.items; + return Array.isArray(items) ? items : []; +} + +/** + * `gen_ai.system_instructions` — the agent instructions, as text parts. + * + * LiveKit carries an agent's instructions as `system`/`developer` messages in the chat + * context, but they originate from the agent definition rather than from the conversation, + * so they are reported as instructions rather than history. + */ +export function toSystemInstructions(chatCtx: ChatContext): MessagePart[] { + const parts: MessagePart[] = []; + for (const item of itemsOf(chatCtx)) { + if (item.type === 'message' && (item.role === 'system' || item.role === 'developer')) { + const text = item.rawTextContent; + if (text !== undefined) parts.push(textPart(text)); + } + } + return parts; +} + +/** + * `gen_ai.input.messages` — the conversation history, in the order it was sent. + * + * `system`/`developer` messages are reported in `gen_ai.system_instructions` instead, and + * non-conversational items (agent handoffs, config updates) are skipped. + */ +export function toInputMessages(chatCtx: ChatContext): ChatMessagePayload[] { + const messages: ChatMessagePayload[] = []; + for (const item of itemsOf(chatCtx)) { + let role: string; + if (item.type === 'message') { + if (item.role === 'system' || item.role === 'developer') continue; + role = item.role; + } else if (item.type === 'function_call') { + role = 'assistant'; + } else if (item.type === 'function_call_output') { + role = 'tool'; + } else { + continue; + } + + const parts = messageParts(item); + if (!parts.length) continue; + + // consecutive tool calls from one assistant turn belong to a single message + const last = messages[messages.length - 1]; + if ( + last && + last.role === 'assistant' && + role === 'assistant' && + item.type === 'function_call' + ) { + last.parts.push(...parts); + continue; + } + + messages.push({ role, parts }); + } + return messages; +} + +export interface ToolCallLike { + callId: string; + name: string; + args: string; +} + +/** `gen_ai.output.messages` — the single assistant turn the model generated. */ +export function toOutputMessages(params: { + text?: string; + functionCalls?: readonly ToolCallLike[]; + finishReason?: string; +}): ChatMessagePayload[] { + const parts: MessagePart[] = []; + if (params.text) parts.push(textPart(params.text)); + for (const call of params.functionCalls ?? []) { + parts.push({ + type: 'tool_call', + id: call.callId, + name: call.name, + arguments: maybeJson(call.args), + }); + } + if (!parts.length) return []; + + const message: ChatMessagePayload = { role: 'assistant', parts }; + if (params.finishReason) message.finish_reason = params.finishReason; + return [message]; +} + +/** + * `parameters` is deliberately omitted: the convention marks it NOT RECOMMENDED by default + * because a schema is large, and building one per request would be pure overhead for + * telemetry. Tools are matched structurally for the layering reason above. + */ +export function toToolDefinitions( + tools: readonly unknown[] | Record, +): MessagePart[] { + const entries = Array.isArray(tools) ? tools : Object.values(tools ?? {}); + const definitions: MessagePart[] = []; + for (const entry of entries) { + const tool = entry as { type?: string; name?: string; description?: string; id?: string }; + if (tool?.type === 'function' && tool.name) { + const definition: MessagePart = { type: 'function', name: tool.name }; + if (tool.description) definition.description = tool.description; + definitions.push(definition); + } else if (tool?.type === 'provider' && tool.id) { + definitions.push({ type: tool.id, name: tool.id }); + } + } + return definitions; +} + +export function finishReasonFor(params: { + functionCalls?: readonly unknown[]; + interrupted?: boolean; +}): string { + // checked first: a generation that emitted a tool call and then failed ended abnormally, + // and the convention has no `cancelled` value for that + if (params.interrupted) return traceTypes.GenAIFinishReason.ERROR; + if (params.functionCalls?.length) return traceTypes.GenAIFinishReason.TOOL_CALL; + return traceTypes.GenAIFinishReason.STOP; +} + +// --------------------------------------------------------------------------- +// span attributes +// --------------------------------------------------------------------------- + +function toJson(value: unknown): string { + return JSON.stringify(value); +} + +/** + * `gen_ai.conversation.id` — the room a session runs in. + * + * The convention forbids fabricating one (no UUIDs, trace ids or content hashes), so this is + * the room id LiveKit already stamps, and `undefined` outside a job context. + */ +export function conversationId(): string | undefined { + return getJobContext(false)?.job?.room?.sid || undefined; +} + +/** The `create_agent` / `invoke_agent` span attributes. */ +/** + * Record the convention's content attributes, when content capture is enabled. + * + * Values are JSON strings: OpenTelemetry attributes cannot hold structured values, which the + * convention explicitly allows for spans. + */ +export function setContentAttributes( + span: Span, + content: { + systemInstructions?: MessagePart[]; + inputMessages?: ChatMessagePayload[]; + outputMessages?: ChatMessagePayload[]; + toolDefinitions?: MessagePart[]; + }, +): void { + if (!captureContent || !span.isRecording()) return; + + const attrs: Attributes = {}; + if (content.systemInstructions?.length) { + attrs[traceTypes.ATTR_GEN_AI_SYSTEM_INSTRUCTIONS] = toJson(content.systemInstructions); + } + if (content.inputMessages?.length) { + attrs[traceTypes.ATTR_GEN_AI_INPUT_MESSAGES] = toJson(content.inputMessages); + } + if (content.outputMessages?.length) { + attrs[traceTypes.ATTR_GEN_AI_OUTPUT_MESSAGES] = toJson(content.outputMessages); + } + if (content.toolDefinitions?.length) { + attrs[traceTypes.ATTR_GEN_AI_TOOL_DEFINITIONS] = toJson(content.toolDefinitions); + } + if (Object.keys(attrs).length) span.setAttributes(attrs); +} + +/** The attributes the convention asks for at span creation time. */ +export function setRequestAttributes( + span: Span, + params: { + operation: string; + provider?: string; + model?: string; + stream?: boolean; + outputType?: string; + }, +): void { + if (!span.isRecording()) return; + + const attrs: Attributes = { [traceTypes.ATTR_GEN_AI_OPERATION_NAME]: params.operation }; + const provider = traceTypes.genAIProviderName(params.provider); + if (provider) attrs[traceTypes.ATTR_GEN_AI_PROVIDER_NAME] = provider; + 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); +} + +export function setResponseAttributes( + span: Span, + params: { + responseId?: string; + model?: string; + finishReasons?: string[]; + /** Time to first chunk, in seconds. */ + timeToFirstChunk?: number; + }, +): void { + if (!span.isRecording()) return; + + const attrs: Attributes = {}; + if (params.responseId) attrs[traceTypes.ATTR_GEN_AI_RESPONSE_ID] = params.responseId; + if (params.model) attrs[traceTypes.ATTR_GEN_AI_RESPONSE_MODEL] = params.model; + if (params.finishReasons?.length) { + attrs[traceTypes.ATTR_GEN_AI_RESPONSE_FINISH_REASONS] = params.finishReasons; + } + if (params.timeToFirstChunk !== undefined && params.timeToFirstChunk >= 0) { + attrs[traceTypes.ATTR_GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK] = params.timeToFirstChunk; + } + if (Object.keys(attrs).length) span.setAttributes(attrs); +} + +/** + * Token usage in the convention's names. + * + * Per the convention the detailed counts are subsets of the totals, so cached and reasoning + * tokens are reported alongside — not added to — input and output tokens. + */ +export function setUsageAttributes( + span: Span, + usage: { + promptTokens?: number; + completionTokens?: number; + promptCachedTokens?: number; + cacheCreationTokens?: number; + reasoningTokens?: number; + }, +): void { + if (!span.isRecording()) return; + + const attrs: Attributes = { + [traceTypes.ATTR_GEN_AI_USAGE_INPUT_TOKENS]: usage.promptTokens ?? 0, + [traceTypes.ATTR_GEN_AI_USAGE_OUTPUT_TOKENS]: usage.completionTokens ?? 0, + }; + if (usage.promptCachedTokens) { + attrs[traceTypes.ATTR_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS] = usage.promptCachedTokens; + // the unofficial spelling is what Datadog's mapping table keys on, and the realtime path + // emits it; without this, cached tokens were attributed for realtime sessions and + // silently absent for pipeline ones + attrs[traceTypes.ATTR_GEN_AI_USAGE_INPUT_CACHED_TOKENS] = usage.promptCachedTokens; + } + if (usage.cacheCreationTokens) { + attrs[traceTypes.ATTR_GEN_AI_USAGE_CACHE_WRITE_INPUT_TOKENS] = usage.cacheCreationTokens; + } + if (usage.reasoningTokens) { + attrs[traceTypes.ATTR_GEN_AI_USAGE_REASONING_OUTPUT_TOKENS] = usage.reasoningTokens; + // unofficial spelling recognised by Langfuse, kept alongside the standard one + attrs[traceTypes.ATTR_GEN_AI_USAGE_REASONING_TOKENS] = usage.reasoningTokens; + } + span.setAttributes(attrs); +} + +/** Token usage for a realtime (speech-to-speech) turn, per modality. */ +export function realtimeUsageAttributes(metrics: RealtimeModelMetrics): Attributes { + const attrs: Attributes = { + [traceTypes.ATTR_GEN_AI_USAGE_INPUT_TOKENS]: metrics.inputTokens, + [traceTypes.ATTR_GEN_AI_USAGE_OUTPUT_TOKENS]: metrics.outputTokens, + [traceTypes.ATTR_GEN_AI_USAGE_TEXT_INPUT_TOKENS]: metrics.inputTokenDetails.textTokens, + [traceTypes.ATTR_GEN_AI_USAGE_AUDIO_INPUT_TOKENS]: metrics.inputTokenDetails.audioTokens, + [traceTypes.ATTR_GEN_AI_USAGE_TEXT_OUTPUT_TOKENS]: metrics.outputTokenDetails.textTokens, + [traceTypes.ATTR_GEN_AI_USAGE_AUDIO_OUTPUT_TOKENS]: metrics.outputTokenDetails.audioTokens, + [traceTypes.ATTR_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS]: metrics.inputTokenDetails.cachedTokens, + }; + return attrs; +} + +/** The `execute_tool` span's attributes, per the convention. */ +export function setToolAttributes( + span: Span, + params: { + name: string; + callId?: string; + toolType?: string; + description?: string; + /** Raw (typically JSON) arguments as produced by the model. */ + args?: string; + agentName?: string; + }, +): void { + if (!span.isRecording()) return; + + const attrs: Attributes = { + [traceTypes.ATTR_GEN_AI_OPERATION_NAME]: traceTypes.GenAIOperationName.EXECUTE_TOOL, + [traceTypes.ATTR_GEN_AI_TOOL_NAME]: params.name, + [traceTypes.ATTR_GEN_AI_TOOL_TYPE]: params.toolType ?? 'function', + }; + 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) { + attrs[traceTypes.ATTR_GEN_AI_TOOL_CALL_ARGUMENTS] = toJson(maybeJson(params.args)); + } + } + span.setAttributes(attrs); +} + +export function setToolResult(span: Span, params: { result?: string; isError: boolean }): void { + if (!span.isRecording()) return; + if (params.isError) { + // the convention records the result only on success + span.setAttribute(traceTypes.ATTR_ERROR_TYPE, 'tool_error'); + return; + } + if (captureContent && params.result !== undefined) { + span.setAttribute(traceTypes.ATTR_GEN_AI_TOOL_CALL_RESULT, toJson(maybeJson(params.result))); + } +} + +/** + * `error.type` — a low-cardinality identifier, per the convention. + * + * Never the error message: that is free-form and may carry user data. + */ +export function setErrorType(span: Span, error: Error | string): void { + if (!span.isRecording()) return; + if (typeof error === 'string') { + span.setAttribute(traceTypes.ATTR_ERROR_TYPE, error); + return; + } + const statusCode = (error as { statusCode?: unknown }).statusCode; + if (typeof statusCode === 'number') { + span.setAttribute(traceTypes.ATTR_ERROR_TYPE, String(statusCode)); + return; + } + span.setAttribute(traceTypes.ATTR_ERROR_TYPE, error.constructor.name); +} + +export function setAgentAttributes( + span: Span, + params: { operation: string; agentName: string; model?: string; provider?: string }, +): void { + if (!span.isRecording()) return; + + const attrs: Attributes = { + [traceTypes.ATTR_GEN_AI_OPERATION_NAME]: params.operation, + [traceTypes.ATTR_GEN_AI_AGENT_NAME]: params.agentName, + }; + // required on create_agent; the model is conditionally required and an agent is + // configured with exactly one, which is when the convention asks for it + 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); +} + +/** The `invoke_workflow` span attributes — LiveKit's session is the workflow. */ +export function setWorkflowAttributes(span: Span, params: { name: string }): void { + if (!span.isRecording()) return; + + const attrs: Attributes = { + [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/index.ts b/agents/src/telemetry/index.ts index 3e2fd70e8..40dabe4f8 100644 --- a/agents/src/telemetry/index.ts +++ b/agents/src/telemetry/index.ts @@ -2,7 +2,11 @@ // // SPDX-License-Identifier: Apache-2.0 -export { ExtraDetailsProcessor, MetadataLogProcessor } from './logging.js'; +export { + ExtraDetailsProcessor, + MetadataLogProcessor, + PIIFilteringLogProcessor, +} from './logging.js'; export type { ObservabilityEndpoint } from './observability_endpoint.js'; export { SimpleOTLPHttpLogExporter, @@ -19,6 +23,8 @@ export { type PinoCloudExporterUrlConfig, type PinoLogObject, } from './pino_otel_transport.js'; +export * as genAI from './gen_ai.js'; +export { REDACTED_EXCEPTION_MESSAGE } from './redaction.js'; export * as traceTypes from './trace_types.js'; export { FanoutSpanProcessor, @@ -33,8 +39,8 @@ export { type StartSpanOptions, } from './traces.js'; export { - REDACTED_EXCEPTION_MESSAGE, recordException, recordRealtimeMetrics, + redactionEnabled, type RecordExceptionOptions, } from './utils.js'; diff --git a/agents/src/telemetry/logging.ts b/agents/src/telemetry/logging.ts index 8af481c06..0a1b7b60e 100644 --- a/agents/src/telemetry/logging.ts +++ b/agents/src/telemetry/logging.ts @@ -4,6 +4,13 @@ import { ThrowsPromise } from '@livekit/throws-transformer/throws'; import type { Attributes } from '@opentelemetry/api'; import type { LogRecordProcessor, SdkLogRecord } from '@opentelemetry/sdk-logs'; +import { + REDACTED_EXCEPTION_ATTRIBUTES, + REDACTED_EXCEPTION_MESSAGE, + isPIIAttribute, + redactionEnabledFromAttributes as redactionEnabled, +} from './redaction.js'; +import * as traceTypes from './trace_types.js'; /** * Metadata log processor that injects metadata (room_id, job_id) into all log records. @@ -45,3 +52,33 @@ export class ExtraDetailsProcessor implements LogRecordProcessor { return ThrowsPromise.resolve(); } } + +/** + * Log counterpart of `PIIFilteringSpanProcessor`, for a logger provider the integrator + * builds themselves. + * + * Once the project mandates redaction the filtering applies to every destination, so the + * client never depends on a collector to strip a newly added key. Register it ahead of any + * exporting processor. + */ +export class PIIFilteringLogProcessor implements LogRecordProcessor { + onEmit(logRecord: SdkLogRecord): void { + if (!redactionEnabled(logRecord.attributes as Record)) return; + + for (const key of Object.keys(logRecord.attributes)) { + if (key === traceTypes.ATTR_EXCEPTION_MESSAGE) { + logRecord.setAttribute(key, REDACTED_EXCEPTION_MESSAGE); + } else if (isPIIAttribute(key) || REDACTED_EXCEPTION_ATTRIBUTES.has(key)) { + delete (logRecord.attributes as Record)[key]; + } + } + } + + shutdown(): Promise { + return ThrowsPromise.resolve(); + } + + forceFlush(): Promise { + return ThrowsPromise.resolve(); + } +} diff --git a/agents/src/telemetry/pii.test.ts b/agents/src/telemetry/pii.test.ts new file mode 100644 index 000000000..07f252ae3 --- /dev/null +++ b/agents/src/telemetry/pii.test.ts @@ -0,0 +1,189 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { Attributes } from '@opentelemetry/api'; +import { SpanStatusCode } from '@opentelemetry/api'; +import { + BasicTracerProvider, + InMemorySpanExporter, + SimpleSpanProcessor, +} from '@opentelemetry/sdk-trace-node'; +import { describe, expect, it } from 'vitest'; +import { ATTRIBUTE_REDACTION_ENABLED } from '../types.js'; +import { PIIFilteringLogProcessor } from './logging.js'; +import { PIIFilteringSpanProcessor } from './pii.js'; +import { REDACTED_EXCEPTION_MESSAGE, isPIIAttribute, restorePii } from './redaction.js'; +import * as traceTypes from './trace_types.js'; + +// Pins the SDK-side guarantee: PII never reaches an exporter that is not LiveKit Cloud's, +// whose own handling is the project's setting in the dashboard. `allowPii` lifts that per +// provider; the project flag overrides it and strips for every destination. + +const PII_ATTRS: Attributes = { + [traceTypes.ATTR_CHAT_CTX]: '{"items": []}', + [traceTypes.ATTR_USER_TRANSCRIPT]: 'my card number is 4111', + [traceTypes.ATTR_GEN_AI_INPUT_MESSAGES]: '[{"role": "user"}]', + [traceTypes.ATTR_GEN_AI_OUTPUT_MESSAGES]: '[{"role": "assistant"}]', + [traceTypes.ATTR_GEN_AI_SYSTEM_INSTRUCTIONS]: '[{"type": "text"}]', + [traceTypes.ATTR_GEN_AI_TOOL_CALL_ARGUMENTS]: '{"location": "Paris"}', + [traceTypes.ATTR_GEN_AI_TOOL_CALL_RESULT]: '{"temp": 14}', + // free-form, and recorded whenever the project allows it + [traceTypes.ATTR_EXCEPTION_TRACE]: 'Traceback: "my pin is 1234"', +}; +const SAFE_ATTRS: Attributes = { + [traceTypes.ATTR_GEN_AI_OPERATION_NAME]: 'chat', + [traceTypes.ATTR_GEN_AI_USAGE_INPUT_TOKENS]: 100, + [traceTypes.ATTR_SPEECH_ID]: 'speech_1', +}; + +function emit(options: { redaction?: boolean; exporterFirst?: boolean; allowPii?: boolean }) { + const exporter = new InMemorySpanExporter(); + const exportProcessor = new SimpleSpanProcessor(exporter); + const redactProcessor = new PIIFilteringSpanProcessor(options.allowPii ?? true); + const provider = new BasicTracerProvider({ + spanProcessors: options.exporterFirst + ? [exportProcessor, redactProcessor] + : [redactProcessor, exportProcessor], + }); + + const span = provider.getTracer('test').startSpan('llm_request'); + if (options.redaction) span.setAttribute(ATTRIBUTE_REDACTION_ENABLED, true); + span.setAttributes({ ...PII_ATTRS, ...SAFE_ATTRS }); + span.addEvent(traceTypes.EVENT_GEN_AI_USER_MESSAGE, { content: 'my pin is 1234' }); + span.addEvent('llm_started', { [traceTypes.ATTR_INSTRUCTIONS]: 'be brief', n: 1 }); + span.end(); + + return exporter.getFinishedSpans()[0]!; +} + +function leaked(attributes: Attributes): string[] { + return Object.keys(PII_ATTRS).filter((key) => key in attributes); +} + +describe('PIIFilteringSpanProcessor', () => { + it('withholds PII from third-party exporters on request', () => { + const span = emit({ allowPii: false }); + + expect(leaked(span.attributes)).toEqual([]); + for (const [key, value] of Object.entries(SAFE_ATTRS)) { + expect(span.attributes[key]).toEqual(value); + } + + const events = Object.fromEntries(span.events.map((e) => [e.name, e.attributes ?? {}])); + // the GenAI content events carry the body in a generic `content` attribute that + // cannot be marked, so the whole event goes + expect(events[traceTypes.EVENT_GEN_AI_USER_MESSAGE]).toBeUndefined(); + // a non-content event keeps its safe attributes and loses its PII ones + expect(events['llm_started']).toEqual({ n: 1 }); + }); + + it('lets exporters receive PII by default', () => { + // the GenAI conventions are only useful to a backend that can render the conversation + const span = emit({}); + + for (const [key, value] of Object.entries(PII_ATTRS)) { + expect(span.attributes[key]).toEqual(value); + } + expect(span.events.map((e) => e.name)).toContain(traceTypes.EVENT_GEN_AI_USER_MESSAGE); + }); + + it('still hands LiveKit Cloud the PII the project allows', () => { + const stripped = emit({ allowPii: false }); + const restored = restorePii(stripped); + + for (const [key, value] of Object.entries(PII_ATTRS)) { + expect(restored.attributes[key]).toEqual(value); + } + expect(restored.events.map((e) => e.name)).toContain(traceTypes.EVENT_GEN_AI_USER_MESSAGE); + // the view must not lose the getters ReadableSpan relies on + expect(restored.spanContext().spanId).toBe(stripped.spanContext().spanId); + expect(restored.name).toBe(stripped.name); + }); + + it('withholds PII from every destination when the project mandates redaction', () => { + // redaction mandated in the dashboard is not weakened by a local grant, and nothing is + // stashed for LiveKit Cloud to restore + const stripped = emit({ allowPii: true, redaction: true }); + + expect(leaked(stripped.attributes)).toEqual([]); + expect(restorePii(stripped)).toBe(stripped); + }); + + it('withholds exception details from third-party exporters', () => { + // recordException resolves the project's setting, so with redaction off it writes the + // real message onto the span, its `exception` event and the span status + const exporter = new InMemorySpanExporter(); + const provider = new BasicTracerProvider({ + spanProcessors: [new PIIFilteringSpanProcessor(false), new SimpleSpanProcessor(exporter)], + }); + const span = provider.getTracer('test').startSpan('llm_request'); + span.recordException(new Error('my pin is 1234')); + span.setStatus({ code: SpanStatusCode.ERROR, message: 'my pin is 1234' }); + span.end(); + + const exported = exporter.getFinishedSpans()[0]!; + const serialized = JSON.stringify([ + exported.attributes, + exported.events.map((e) => e.attributes), + exported.status.message, + ]); + expect(serialized).not.toContain('my pin is 1234'); + + // ... while LiveKit Cloud still receives all three, per the project's setting + // recordException records it on the `exception` event and the status, not the + // span attributes, which is why both are restored + const restored = restorePii(exported); + expect(restored.status.message).toBe('my pin is 1234'); + expect(JSON.stringify(restored.events)).toContain('my pin is 1234'); + }); + + it('protects an exporter registered before it', () => { + // onEnding runs for every processor before any onEnd, so ordering cannot leak PII + expect(leaked(emit({ allowPii: false, exporterFirst: true }).attributes)).toEqual([]); + }); + + it.each([ + ['lk.pii.chat_ctx', true], + ['gen_ai.input.messages', true], + ['gen_ai.prompt.variable.customer_name', true], + ['gen_ai.usage.input_tokens', false], + // a `pii` substring is not a `pii` segment + ['lk.piidata.x', false], + ])('classifies %s', (key, expected) => { + expect(isPIIAttribute(key as string)).toBe(expected); + }); +}); + +describe('PIIFilteringLogProcessor', () => { + it('filters PII and exception details once the project mandates redaction', () => { + const attributes: Record = { + [ATTRIBUTE_REDACTION_ENABLED]: true, + [traceTypes.ATTR_CHAT_CTX]: '{"items": []}', + [traceTypes.ATTR_GEN_AI_INPUT_MESSAGES]: '[{"role": "user"}]', + [traceTypes.ATTR_EXCEPTION_MESSAGE]: 'my pin is 1234', + [traceTypes.ATTR_EXCEPTION_TRACE]: 'Traceback: "my pin is 1234"', + function: 'get_weather', + }; + const record = { + attributes, + setAttribute(key: string, value: unknown) { + attributes[key] = value; + }, + }; + + new PIIFilteringLogProcessor().onEmit(record as never); + + expect(attributes).toEqual({ + [ATTRIBUTE_REDACTION_ENABLED]: true, + [traceTypes.ATTR_EXCEPTION_MESSAGE]: REDACTED_EXCEPTION_MESSAGE, + function: 'get_weather', + }); + }); + + it('leaves records alone when the project has not', () => { + const attributes: Record = { [traceTypes.ATTR_CHAT_CTX]: '{"items": []}' }; + new PIIFilteringLogProcessor().onEmit({ attributes } as never); + + expect(attributes).toEqual({ [traceTypes.ATTR_CHAT_CTX]: '{"items": []}' }); + }); +}); diff --git a/agents/src/telemetry/pii.ts b/agents/src/telemetry/pii.ts new file mode 100644 index 000000000..28bb693f4 --- /dev/null +++ b/agents/src/telemetry/pii.ts @@ -0,0 +1,110 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +/** + * In-process stripping of personally identifiable information from telemetry. + * + * Field-level filtering, not entity-level redaction: a matching attribute is dropped whole, + * never scanned and masked. "Redaction" in this codebase means the project setting (LiveKit + * Cloud dashboard, or `record: { redaction: true }`); this module is what the client does + * about it. + * + * LiveKit marks attributes carrying conversational content, tool payloads, or other user data + * with a dot-delimited `pii` segment (`lk.pii.`), and the GenAI content attributes carry + * the same kind of payload under names the semantic convention fixes, where the marker cannot + * be applied. Both are filtered before any exporter that is not LiveKit Cloud's — and before + * every exporter, Cloud included, once the project has enabled redaction, so the client never + * depends on a collector to strip a new key. {@link restorePii} puts the payload back on + * LiveKit Cloud's export path when the project still allows it. + */ +import type { Context } from '@opentelemetry/api'; +import { SpanStatusCode } from '@opentelemetry/api'; +import type { ReadableSpan, Span as SdkSpan, SpanProcessor } from '@opentelemetry/sdk-trace-node'; +import { + PII_EVENT_NAMES, + REDACTED_EXCEPTION_ATTRIBUTES, + REDACTED_EXCEPTION_MESSAGE, + isPIIAttribute, + stashPii, +} from './redaction.js'; +import * as traceTypes from './trace_types.js'; +import { redactionEnabled } from './utils.js'; + +/** + * Drops PII attributes so they never reach an exporter that is not LiveKit Cloud's. + * + * Runs in `onEnding`, which the SDK dispatches to every registered processor *before* any + * processor's `onEnd` and while the span is still mutable. Registration order therefore does + * not matter: an exporter the integrator attached before LiveKit's own still sees the + * redacted span. + * + * `allowPii` lifts the stripping for a provider whose exporters the integrator has explicitly + * granted PII (`setTracerProvider(provider, { allowPii: true })`). The project's redaction + * setting overrides that grant and strips for every destination, Cloud included. + */ +export class PIIFilteringSpanProcessor implements SpanProcessor { + constructor(private readonly allowPii: boolean = false) {} + + onStart(_span: SdkSpan, _parentContext: Context): void {} + + onEnding(span: SdkSpan): void { + const projectRedaction = redactionEnabled(span.attributes); + if (this.allowPii && !projectRedaction) return; + + const attributes = span.attributes as Record; + const events = span.events; + const piiKeys = Object.keys(attributes).filter( + (key) => isPIIAttribute(key) || REDACTED_EXCEPTION_ATTRIBUTES.has(key), + ); + const kept = events.filter((event) => !PII_EVENT_NAMES.has(event.name)); + const eventCarriesPii = kept.some((event) => + Object.keys(event.attributes ?? {}).some( + (key) => isPIIAttribute(key) || REDACTED_EXCEPTION_ATTRIBUTES.has(key), + ), + ); + if (!piiKeys.length && kept.length === events.length && !eventCarriesPii) { + if (span.status.code !== SpanStatusCode.ERROR || !span.status.message) return; + } + + if (!projectRedaction) { + // LiveKit Cloud still receives what the project allows + stashPii(span); + } + + for (const key of piiKeys) { + if (key === traceTypes.ATTR_EXCEPTION_MESSAGE) { + attributes[key] = REDACTED_EXCEPTION_MESSAGE; + } else { + delete attributes[key]; + } + } + + for (const event of kept) { + const attrs = event.attributes as Record | undefined; + for (const key of Object.keys(attrs ?? {})) { + if (key === traceTypes.ATTR_EXCEPTION_MESSAGE) { + attrs![key] = REDACTED_EXCEPTION_MESSAGE; + } else if (isPIIAttribute(key) || REDACTED_EXCEPTION_ATTRIBUTES.has(key)) { + delete attrs![key]; + } + } + } + if (kept.length !== events.length) { + events.length = 0; + events.push(...kept); + } + + // recordException also puts the message in the span status. setStatus still applies + // here: the SDK marks the span ended only after onEnding returns. + if (span.status.code === SpanStatusCode.ERROR && span.status.message) { + span.setStatus({ code: SpanStatusCode.ERROR, message: REDACTED_EXCEPTION_MESSAGE }); + } + } + + onEnd(_span: ReadableSpan): void {} + + async shutdown(): Promise {} + + async forceFlush(): Promise {} +} diff --git a/agents/src/telemetry/pino_otel_transport.ts b/agents/src/telemetry/pino_otel_transport.ts index 63b824458..b6907126d 100644 --- a/agents/src/telemetry/pino_otel_transport.ts +++ b/agents/src/telemetry/pino_otel_transport.ts @@ -12,7 +12,7 @@ import { SeverityNumber } from '@opentelemetry/api-logs'; import { AccessToken } from 'livekit-server-sdk'; import { ATTRIBUTE_REDACTION_ENABLED } from '../types.js'; import { resolveObservabilityUrl } from './observability_endpoint.js'; -import { REDACTED_EXCEPTION_MESSAGE } from './redaction.js'; +import { REDACTED_EXCEPTION_MESSAGE, isPIIAttribute } from './redaction.js'; import { fetchWithUploadGate, uploadGate } from './upload_gate.js'; export interface PinoLogObject { @@ -177,13 +177,15 @@ export class PinoCloudExporter { } for (const [key, value] of Object.entries(logObj)) { - if (!EXCLUDE_FIELDS.has(key)) { - const attributeValue = - redactionEnabled && (key === 'error' || key === 'err') - ? redactSerializedException(value) - : value; - attributes.push({ key, value: convertValue(attributeValue) }); - } + if (EXCLUDE_FIELDS.has(key)) continue; + // once the project mandates redaction the client filters the keys itself rather than + // relying on the collector to know them + if (redactionEnabled && isPIIAttribute(key)) continue; + const attributeValue = + redactionEnabled && (key === 'error' || key === 'err') + ? redactSerializedException(value) + : value; + attributes.push({ key, value: convertValue(attributeValue) }); } return { diff --git a/agents/src/telemetry/redaction.ts b/agents/src/telemetry/redaction.ts index 66de58a7c..c70f13b84 100644 --- a/agents/src/telemetry/redaction.ts +++ b/agents/src/telemetry/redaction.ts @@ -1,5 +1,151 @@ // SPDX-FileCopyrightText: 2026 LiveKit, Inc. // // SPDX-License-Identifier: Apache-2.0 +import type { Attributes, SpanStatus } from '@opentelemetry/api'; +import type { ReadableSpan, Span as SdkSpan, TimedEvent } from '@opentelemetry/sdk-trace-node'; +import { ATTRIBUTE_REDACTION_ENABLED } from '../types.js'; +import * as traceTypes from './trace_types.js'; + +// Type-only imports on purpose: this module is pulled in near the top of the telemetry +// barrel (via pino_otel_transport), which job.ts imports, so anything imported here starts +// evaluating that far earlier inside the job <-> telemetry cycle. export const REDACTED_EXCEPTION_MESSAGE = 'exception details redacted'; + +/** + * Mirrors the LiveKit Cloud collector's matcher: a whole dot-delimited `pii` segment, + * case-insensitive (`lk.chatpii` does not match, `lk.PII.x` does). + */ +const PII_SEGMENT_RE = /(^|\.)pii(\.|$)/i; + +/** + * GenAI attributes that carry content. Their names are fixed by the semantic convention, + * so they cannot carry the `lk.pii.` marker and are enumerated here instead. + */ +export const GEN_AI_PII_ATTRIBUTES: ReadonlySet = new Set([ + // flagged "likely to contain sensitive information including user/PII data" by the spec + traceTypes.ATTR_GEN_AI_INPUT_MESSAGES, + traceTypes.ATTR_GEN_AI_OUTPUT_MESSAGES, + traceTypes.ATTR_GEN_AI_SYSTEM_INSTRUCTIONS, + traceTypes.ATTR_GEN_AI_TOOL_CALL_ARGUMENTS, + traceTypes.ATTR_GEN_AI_TOOL_CALL_RESULT, + traceTypes.ATTR_GEN_AI_TOOL_DESCRIPTION, + traceTypes.ATTR_GEN_AI_TOOL_DEFINITIONS, + // free-form text the caller supplied or the model produced + traceTypes.ATTR_GEN_AI_RETRIEVAL_QUERY_TEXT, + traceTypes.ATTR_GEN_AI_RETRIEVAL_DOCUMENTS, + traceTypes.ATTR_GEN_AI_MEMORY_QUERY_TEXT, + traceTypes.ATTR_GEN_AI_MEMORY_RECORDS, + traceTypes.ATTR_GEN_AI_EVALUATION_EXPLANATION, +]); + +/** + * Events whose body rides on a generic attribute (`content`, `tool_calls`) that cannot be + * marked, so the whole event is dropped rather than filtered. + */ +export const PII_EVENT_NAMES: ReadonlySet = new Set([ + traceTypes.EVENT_GEN_AI_SYSTEM_MESSAGE, + traceTypes.EVENT_GEN_AI_USER_MESSAGE, + traceTypes.EVENT_GEN_AI_ASSISTANT_MESSAGE, + traceTypes.EVENT_GEN_AI_TOOL_MESSAGE, + traceTypes.EVENT_GEN_AI_CHOICE, + traceTypes.EVENT_GEN_AI_CLIENT_INFERENCE_OPERATION_DETAILS, +]); + +/** + * Exception details are recorded by `recordException`, which resolves the project's redaction + * setting; a third-party exporter must not see them either way. + */ +export const REDACTED_EXCEPTION_ATTRIBUTES: ReadonlySet = new Set([ + traceTypes.ATTR_EXCEPTION_MESSAGE, + traceTypes.ATTR_EXCEPTION_TRACE, +]); + +/** + * Whether `key` names an attribute that must be stripped: it carries a dot-delimited `pii` + * segment, or it is one of the GenAI content attributes. + */ +export function isPIIAttribute(key: string): boolean { + if (PII_SEGMENT_RE.test(key)) return true; + if (GEN_AI_PII_ATTRIBUTES.has(key)) return true; + // gen_ai.prompt.variable. holds the values interpolated into a prompt template + return key.startsWith(traceTypes.ATTR_GEN_AI_PROMPT_VARIABLE); +} + +const ALLOW_PII_ENV_VAR = 'LIVEKIT_TELEMETRY_ALLOW_PII'; +const FALSY = new Set(['0', 'false', 'no', 'off']); + +/** + * The `LIVEKIT_TELEMETRY_ALLOW_PII` setting, or `undefined` when unset. + * + * For integrators who let the framework adopt the ambient OpenTelemetry provider (a + * NodeSDK-style setup) and so have nowhere to pass `allowPii`. Set it to `0` to withhold + * conversational content from third-party exporters. + */ +/** + * Whether the record's own stamp says the project mandated redaction. + * + * Attribute-only on purpose: the modules that need this sit at the top of the telemetry + * barrel, which job.ts imports, so they cannot reach the ambient job context. The stamp is + * applied to every record by the metadata processor. + */ +export function redactionEnabledFromAttributes( + // loose on purpose: span attributes and log attributes are different OTel types + attributes: Record | undefined, +): boolean { + return Boolean(attributes?.[ATTRIBUTE_REDACTION_ENABLED]); +} + +export function allowPiiFromEnv(): boolean | undefined { + const raw = process.env[ALLOW_PII_ENV_VAR]; + if (raw === undefined) return undefined; + return !FALSY.has(raw.trim().toLowerCase()); +} + +const RAW_ATTRIBUTES = Symbol('lkRawAttributes'); +const RAW_EVENTS = Symbol('lkRawEvents'); +const RAW_STATUS = Symbol('lkRawStatus'); + +interface PiiStash { + [RAW_ATTRIBUTES]?: Attributes; + [RAW_EVENTS]?: TimedEvent[]; + [RAW_STATUS]?: SpanStatus; +} + +/** Keeps the pre-filter payload for {@link restorePii} to hand LiveKit Cloud. */ +export function stashPii(span: SdkSpan): void { + const stash = span as unknown as PiiStash; + stash[RAW_ATTRIBUTES] = { ...span.attributes }; + stash[RAW_EVENTS] = span.events.map((event) => ({ + ...event, + attributes: { ...event.attributes }, + })); + stash[RAW_STATUS] = { ...span.status }; +} + +/** + * The span as it was before PII was stripped for third-party exporters. + * + * Only LiveKit Cloud's own exporter calls this; every other destination sees the redacted + * span. Returns `span` unchanged when nothing was stashed. + * + * The view delegates to the span through its prototype so the getters `ReadableSpan` relies + * on keep working, with only `attributes` and `events` shadowed. + */ +export function restorePii(span: ReadableSpan): ReadableSpan { + const stash = span as unknown as PiiStash; + const attributes = stash[RAW_ATTRIBUTES]; + if (!attributes) return span; + + const view = Object.create(span) as ReadableSpan; + Object.defineProperty(view, 'attributes', { value: attributes, enumerable: true }); + Object.defineProperty(view, 'events', { + value: stash[RAW_EVENTS] ?? span.events, + enumerable: true, + }); + Object.defineProperty(view, 'status', { + value: stash[RAW_STATUS] ?? span.status, + enumerable: true, + }); + return view; +} diff --git a/agents/src/telemetry/trace_types.test.ts b/agents/src/telemetry/trace_types.test.ts index b3053132c..82a670bef 100644 --- a/agents/src/telemetry/trace_types.test.ts +++ b/agents/src/telemetry/trace_types.test.ts @@ -6,6 +6,7 @@ import { join, relative } from 'node:path'; import { fileURLToPath } from 'node:url'; import ts from 'typescript'; import { describe, expect, it } from 'vitest'; +import { GEN_AI_PII_ATTRIBUTES, isPIIAttribute } from './redaction.js'; import * as traceTypes from './trace_types.js'; const PII_SEGMENT_RE = /(^|\.)pii(\.|$)/i; @@ -197,6 +198,33 @@ const SAFE_KEYS = new Set([ 'lk.interruption.total_duration', 'lk.interruption.prediction_duration', 'lk.interruption.detection_delay', + // -- OTel GenAI semantic conventions -- + // identifiers, enums, counts and sampling settings; nothing free-form. The + // content-bearing gen_ai attributes live in GEN_AI_PII_ATTRIBUTES instead, because their + // names are fixed by the convention and cannot carry the `lk.pii.` marker. + 'error.type', + 'gen_ai.agent.name', + 'gen_ai.conversation.id', + 'gen_ai.output.type', + 'gen_ai.request.stream', + 'gen_ai.response.finish_reasons', + 'gen_ai.response.id', + 'gen_ai.response.model', + 'gen_ai.response.time_to_first_chunk', + 'gen_ai.tool.call.id', + 'gen_ai.tool.name', + 'gen_ai.tool.type', + 'gen_ai.usage.audio.input_tokens', + 'gen_ai.usage.audio.output_tokens', + 'gen_ai.usage.cache_read.input_tokens', + 'gen_ai.usage.cache_write.input_tokens', + 'gen_ai.usage.reasoning.output_tokens', + 'gen_ai.usage.reasoning_tokens', + 'gen_ai.usage.text.input_tokens', + 'gen_ai.usage.text.output_tokens', + 'gen_ai.workflow.name', + // GenAI event and metric names (not attribute keys) + 'gen_ai.client.inference.operation.details', ]); function declaredKeys(): Record { @@ -211,7 +239,8 @@ describe('telemetry key PII classification', () => { it('classifies every declared key as safe or PII-bearing', () => { const unclassified = Object.fromEntries( Object.entries(declaredKeys()).filter( - ([, value]) => !SAFE_KEYS.has(value) && !PII_SEGMENT_RE.test(value), + ([, value]) => + !SAFE_KEYS.has(value) && !PII_SEGMENT_RE.test(value) && !isPIIAttribute(value), ), ); @@ -219,16 +248,23 @@ describe('telemetry key PII classification', () => { }); it('does not mark safe keys as PII-bearing', () => { - const conflicting = [...SAFE_KEYS].filter((key) => PII_SEGMENT_RE.test(key)).sort(); + const conflicting = [...SAFE_KEYS] + .filter((key) => PII_SEGMENT_RE.test(key) || isPIIAttribute(key)) + .sort(); expect(conflicting).toEqual([]); }); - it('does not retain stale safe-list entries', () => { - const declared = new Set(Object.values(declaredKeys())); - const stale = [...SAFE_KEYS].filter((key) => !declared.has(key)).sort(); + it('agrees with the in-process stripper', () => { + // an integrator's own exporter never reaches the LiveKit Cloud collector, so every + // pii-marked key must be stripped in-process too + const marked = Object.values(declaredKeys()).filter((value) => PII_SEGMENT_RE.test(value)); + expect(marked.length).toBeGreaterThan(0); + expect(marked.filter((key) => !isPIIAttribute(key))).toEqual([]); - expect(stale).toEqual([]); + // a rename must not silently drop an attribute out of the stripped set + const declared = new Set(Object.values(declaredKeys())); + expect([...GEN_AI_PII_ATTRIBUTES].filter((key) => !declared.has(key)).sort()).toEqual([]); }); it('tags sensitive literal structured-log fields as PII', () => { diff --git a/agents/src/telemetry/trace_types.ts b/agents/src/telemetry/trace_types.ts index cb22e4b4a..22d0c1830 100644 --- a/agents/src/telemetry/trace_types.ts +++ b/agents/src/telemetry/trace_types.ts @@ -116,23 +116,216 @@ export const ATTR_REALTIME_MODEL_METRICS = 'lk.realtime_model_metrics'; /** End-to-end latency in seconds. */ export const ATTR_E2E_LATENCY = 'lk.e2e_latency'; -// OpenTelemetry GenAI attributes -// OpenTelemetry specification: https://opentelemetry.io/docs/specs/semconv/registry/attributes/gen-ai/ +// OpenTelemetry GenAI semantic conventions, mirroring the attribute registry of +// https://github.com/open-telemetry/semantic-conventions-genai. Backends ingest these +// directly, so the names must stay byte-for-byte identical to the registry. The ones the +// spec flags as sensitive are listed in GEN_AI_PII_ATTRIBUTES in ./pii.ts, since a +// standard name cannot carry the `lk.pii.` marker segment. + export const ATTR_GEN_AI_OPERATION_NAME = 'gen_ai.operation.name'; -export const ATTR_GEN_AI_REQUEST_MODEL = 'gen_ai.request.model'; -/** The provider name (e.g., 'openai', 'anthropic'). */ export const ATTR_GEN_AI_PROVIDER_NAME = 'gen_ai.provider.name'; + +export const ATTR_GEN_AI_REQUEST_MODEL = 'gen_ai.request.model'; +export const ATTR_GEN_AI_REQUEST_STREAM = 'gen_ai.request.stream'; + +export const ATTR_GEN_AI_RESPONSE_ID = 'gen_ai.response.id'; +export const ATTR_GEN_AI_RESPONSE_MODEL = 'gen_ai.response.model'; +export const ATTR_GEN_AI_RESPONSE_FINISH_REASONS = 'gen_ai.response.finish_reasons'; +/** Time to first chunk of a streaming response, in seconds. */ +export const ATTR_GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK = 'gen_ai.response.time_to_first_chunk'; + export const ATTR_GEN_AI_USAGE_INPUT_TOKENS = 'gen_ai.usage.input_tokens'; export const ATTR_GEN_AI_USAGE_OUTPUT_TOKENS = 'gen_ai.usage.output_tokens'; +export const ATTR_GEN_AI_USAGE_CACHE_READ_INPUT_TOKENS = 'gen_ai.usage.cache_read.input_tokens'; +export const ATTR_GEN_AI_USAGE_CACHE_WRITE_INPUT_TOKENS = 'gen_ai.usage.cache_write.input_tokens'; +export const ATTR_GEN_AI_USAGE_REASONING_OUTPUT_TOKENS = 'gen_ai.usage.reasoning.output_tokens'; +export const ATTR_GEN_AI_USAGE_TEXT_INPUT_TOKENS = 'gen_ai.usage.text.input_tokens'; +export const ATTR_GEN_AI_USAGE_TEXT_OUTPUT_TOKENS = 'gen_ai.usage.text.output_tokens'; +export const ATTR_GEN_AI_USAGE_AUDIO_INPUT_TOKENS = 'gen_ai.usage.audio.input_tokens'; +export const ATTR_GEN_AI_USAGE_AUDIO_OUTPUT_TOKENS = 'gen_ai.usage.audio.output_tokens'; + +export const ATTR_GEN_AI_CONVERSATION_ID = 'gen_ai.conversation.id'; + +export const ATTR_GEN_AI_AGENT_NAME = 'gen_ai.agent.name'; + +export const ATTR_GEN_AI_TOOL_NAME = 'gen_ai.tool.name'; +export const ATTR_GEN_AI_TOOL_CALL_ID = 'gen_ai.tool.call.id'; +export const ATTR_GEN_AI_TOOL_DESCRIPTION = 'gen_ai.tool.description'; +export const ATTR_GEN_AI_TOOL_TYPE = 'gen_ai.tool.type'; +export const ATTR_GEN_AI_TOOL_CALL_ARGUMENTS = 'gen_ai.tool.call.arguments'; +export const ATTR_GEN_AI_TOOL_CALL_RESULT = 'gen_ai.tool.call.result'; +export const ATTR_GEN_AI_TOOL_DEFINITIONS = 'gen_ai.tool.definitions'; + +export const ATTR_GEN_AI_SYSTEM_INSTRUCTIONS = 'gen_ai.system_instructions'; +export const ATTR_GEN_AI_INPUT_MESSAGES = 'gen_ai.input.messages'; +export const ATTR_GEN_AI_OUTPUT_MESSAGES = 'gen_ai.output.messages'; +export const ATTR_GEN_AI_OUTPUT_TYPE = 'gen_ai.output.type'; + +export const ATTR_GEN_AI_RETRIEVAL_DOCUMENTS = 'gen_ai.retrieval.documents'; +export const ATTR_GEN_AI_RETRIEVAL_QUERY_TEXT = 'gen_ai.retrieval.query.text'; +export const ATTR_GEN_AI_MEMORY_QUERY_TEXT = 'gen_ai.memory.query.text'; +export const ATTR_GEN_AI_MEMORY_RECORDS = 'gen_ai.memory.records'; +export const ATTR_GEN_AI_EVALUATION_EXPLANATION = 'gen_ai.evaluation.explanation'; +/** Template attribute: the concrete key is `gen_ai.prompt.variable.`. */ +export const ATTR_GEN_AI_PROMPT_VARIABLE = 'gen_ai.prompt.variable'; +export const ATTR_GEN_AI_WORKFLOW_NAME = 'gen_ai.workflow.name'; + +export const ATTR_ERROR_TYPE = 'error.type'; + +/** Well-known `gen_ai.operation.name` values. */ +export const GenAIOperationName = { + CHAT: 'chat', + GENERATE_CONTENT: 'generate_content', + TEXT_COMPLETION: 'text_completion', + EMBEDDINGS: 'embeddings', + RETRIEVAL: 'retrieval', + FETCH_RESPONSE: 'fetch_response', + CREATE_AGENT: 'create_agent', + INVOKE_AGENT: 'invoke_agent', + EXECUTE_TOOL: 'execute_tool', + INVOKE_WORKFLOW: 'invoke_workflow', + PLAN: 'plan', + SEARCH_MEMORY: 'search_memory', + CREATE_MEMORY: 'create_memory', + UPDATE_MEMORY: 'update_memory', + UPSERT_MEMORY: 'upsert_memory', + DELETE_MEMORY: 'delete_memory', + CREATE_MEMORY_STORE: 'create_memory_store', + DELETE_MEMORY_STORE: 'delete_memory_store', +} as const; + +/** Well-known `gen_ai.output.type` values. */ +export const GenAIOutputType = { + TEXT: 'text', + JSON: 'json', + IMAGE: 'image', + SPEECH: 'speech', +} as const; + +/** Well-known `gen_ai.response.finish_reasons` values. */ +export const GenAIFinishReason = { + STOP: 'stop', + LENGTH: 'length', + CONTENT_FILTER: 'content_filter', + TOOL_CALL: 'tool_call', + COMPACTION: 'compaction', + ERROR: 'error', +} as const; + +/** + * The `gen_ai.provider.name` values the registry enumerates. + * + * For a provider on this list the convention says the registry spelling MUST be used, since + * backends treat the attribute as the discriminator for provider-specific parsing. A provider + * that is not on it MAY report a custom value, so those pass through untouched. + */ +export const GEN_AI_PROVIDER_NAMES: ReadonlySet = new Set([ + 'openai', + 'gcp.gen_ai', + 'gcp.vertex_ai', + 'gcp.gemini', + 'anthropic', + 'cohere', + 'azure.ai.inference', + 'azure.ai.openai', + 'ibm.watsonx.ai', + 'aws.bedrock', + 'perplexity', + 'x_ai', + 'deepseek', + 'groq', + 'mistral_ai', + 'moonshot_ai', +]); + +// Plugins report `provider` either as a display name ('AWS Bedrock', 'MistralAI') or, for the +// OpenAI-compatible clients, as the base URL's host ('api.mistral.ai'). Both are mapped here: +// by host first, then by the display name reduced to lowercase alphanumerics, so +// 'AWS Bedrock' / 'aws_bedrock' / 'awsbedrock' all resolve alike. +const PROVIDER_BY_HOST: Record = { + 'api.anthropic.com': 'anthropic', + 'api.cohere.ai': 'cohere', + 'api.cohere.com': 'cohere', + 'api.deepseek.com': 'deepseek', + 'api.groq.com': 'groq', + 'api.mistral.ai': 'mistral_ai', + 'api.moonshot.ai': 'moonshot_ai', + 'api.moonshot.cn': 'moonshot_ai', + 'api.openai.com': 'openai', + 'api.perplexity.ai': 'perplexity', + 'api.x.ai': 'x_ai', + 'generativelanguage.googleapis.com': 'gcp.gemini', +}; + +const PROVIDER_BY_HOST_SUFFIX: readonly [string, string][] = [ + ['.openai.azure.com', 'azure.ai.openai'], + ['.services.ai.azure.com', 'azure.ai.inference'], + ['.aiplatform.googleapis.com', 'gcp.vertex_ai'], +]; + +const PROVIDER_BY_NAME: Record = { + amazon: 'aws.bedrock', + amazonbedrock: 'aws.bedrock', + anthropic: 'anthropic', + awsbedrock: 'aws.bedrock', + azureaiinference: 'azure.ai.inference', + azureopenai: 'azure.ai.openai', + bedrock: 'aws.bedrock', + cohere: 'cohere', + deepseek: 'deepseek', + gemini: 'gcp.gemini', + google: 'gcp.gen_ai', + googlecloudplatform: 'gcp.gen_ai', + googlegenai: 'gcp.gen_ai', + groq: 'groq', + ibmwatsonxai: 'ibm.watsonx.ai', + mistral: 'mistral_ai', + mistralai: 'mistral_ai', + moonshot: 'moonshot_ai', + moonshotai: 'moonshot_ai', + openai: 'openai', + perplexity: 'perplexity', + vertexai: 'gcp.vertex_ai', + vertexaimodelgarden: 'gcp.vertex_ai', + watsonx: 'ibm.watsonx.ai', + xai: 'x_ai', +}; + +/** Normalize a LiveKit plugin's `provider` to its GenAI registry spelling. */ +export function genAIProviderName(provider: string | undefined | null): string | undefined { + const value = provider?.trim(); + if (!value) return undefined; + + const host = value.toLowerCase(); + if (PROVIDER_BY_HOST[host]) return PROVIDER_BY_HOST[host]; + for (const [suffix, mapped] of PROVIDER_BY_HOST_SUFFIX) { + if (host.endsWith(suffix)) return mapped; + } + // only the Bedrock endpoints, not every AWS service that shares the domain + if (host.startsWith('bedrock') && host.endsWith('.amazonaws.com')) return 'aws.bedrock'; + + const canonical = host.replace(/[^a-z0-9]/g, ''); + // a provider outside the registry keeps its own id, which the convention allows + return PROVIDER_BY_NAME[canonical] ?? value; +} + +/** @internal Exposed for the guard test that walks the plugins' provider values. */ +export const _providerTables = { + byHost: PROVIDER_BY_HOST, + byHostSuffix: PROVIDER_BY_HOST_SUFFIX, + byName: PROVIDER_BY_NAME, +}; // Unofficial OpenTelemetry GenAI attributes, recognized by LangFuse // https://langfuse.com/integrations/native/opentelemetry#usage -// but not yet in the official OpenTelemetry specification. +// but not in the official OpenTelemetry specification. Emitted alongside the official +// `gen_ai.usage.*.{input,output}_tokens` names above. export const ATTR_GEN_AI_USAGE_INPUT_TEXT_TOKENS = 'gen_ai.usage.input_text_tokens'; export const ATTR_GEN_AI_USAGE_INPUT_AUDIO_TOKENS = 'gen_ai.usage.input_audio_tokens'; export const ATTR_GEN_AI_USAGE_INPUT_CACHED_TOKENS = 'gen_ai.usage.input_cached_tokens'; export const ATTR_GEN_AI_USAGE_OUTPUT_TEXT_TOKENS = 'gen_ai.usage.output_text_tokens'; export const ATTR_GEN_AI_USAGE_OUTPUT_AUDIO_TOKENS = 'gen_ai.usage.output_audio_tokens'; +export const ATTR_GEN_AI_USAGE_REASONING_TOKENS = 'gen_ai.usage.reasoning_tokens'; // OpenTelemetry GenAI event names (for structured logging) export const EVENT_GEN_AI_SYSTEM_MESSAGE = 'gen_ai.system.message'; @@ -140,6 +333,10 @@ export const EVENT_GEN_AI_USER_MESSAGE = 'gen_ai.user.message'; export const EVENT_GEN_AI_ASSISTANT_MESSAGE = 'gen_ai.assistant.message'; export const EVENT_GEN_AI_TOOL_MESSAGE = 'gen_ai.tool.message'; export const EVENT_GEN_AI_CHOICE = 'gen_ai.choice'; +export const EVENT_GEN_AI_CLIENT_INFERENCE_OPERATION_DETAILS = + 'gen_ai.client.inference.operation.details'; + +// OpenTelemetry GenAI metric names // Exception attributes export const ATTR_EXCEPTION_TRACE = 'exception.stacktrace'; diff --git a/agents/src/telemetry/traces.otel2.type.test.ts b/agents/src/telemetry/traces.otel2.type.test.ts index e92cb153f..2cae5c813 100644 --- a/agents/src/telemetry/traces.otel2.type.test.ts +++ b/agents/src/telemetry/traces.otel2.type.test.ts @@ -10,6 +10,7 @@ import { import { BatchSpanProcessor, NodeTracerProvider } from '@opentelemetry/sdk-trace-node'; import http from 'node:http'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { PIIFilteringSpanProcessor } from './pii.js'; import { type CloudSpanProcessorOptions, FanoutSpanProcessor, @@ -170,8 +171,9 @@ describe('setupCloudTracer with an OpenTelemetry SDK 2.x provider', () => { }); expect(tracer.getProvider()).toBe(provider); - // session metadata processor + built-in cloud span processor - expect(registered).toHaveLength(2); - expect(registered[1]).toBeInstanceOf(BatchSpanProcessor); + // PII stripper + session metadata processor + built-in cloud span processor + expect(registered).toHaveLength(3); + expect(registered[0]).toBeInstanceOf(PIIFilteringSpanProcessor); + expect(registered[2]).toBeInstanceOf(BatchSpanProcessor); }); }); diff --git a/agents/src/telemetry/traces.test.ts b/agents/src/telemetry/traces.test.ts index 554abbcb3..3df1a72be 100644 --- a/agents/src/telemetry/traces.test.ts +++ b/agents/src/telemetry/traces.test.ts @@ -22,6 +22,7 @@ import { log } from '../log.js'; import { version } from '../version.js'; import type { SessionReport } from '../voice/report.js'; import { SimpleOTLPHttpLogExporter } from './otel_http_exporter.js'; +import { PIIFilteringSpanProcessor } from './pii.js'; import { type CloudSpanProcessorOptions, setTracerProvider, @@ -59,6 +60,7 @@ describe('setupCloudTracer default provider resource', () => { exportedSpans.push(...spans); callback({ code: 0 }); }); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined); await setupCloudTracer({ roomId: 'room1', @@ -81,6 +83,9 @@ describe('setupCloudTracer default provider resource', () => { room_id: 'room1', job_id: 'job1', }); + // the framework-owned provider is built with the filtering processor attached, so the + // setTracerProvider call that follows must not report redaction as uninstallable + expect(warn.mock.calls.flat().join(' ')).not.toContain('PII redaction'); }); }); @@ -267,11 +272,13 @@ describe('setupCloudTracer with a user-configured provider', () => { // No span is created/ended here so the newly attached cloud BatchSpanProcessor has // nothing to flush over the network on shutdown. expect(tracer.getProvider()).toBe(userProvider); - // setTracerProvider registers the user metadata processor; setupCloudTracer registers the - // session metadata processor plus the built-in (SDK 2.x) cloud exporter. - expect(registeredProcessors).toHaveLength(3); + // setTracerProvider registers the user metadata processor and the in-process PII + // stripper; setupCloudTracer registers the session metadata processor plus the built-in + // (SDK 2.x) cloud exporter. + expect(registeredProcessors).toHaveLength(4); + expect(registeredProcessors[1]).toBeInstanceOf(PIIFilteringSpanProcessor); const setAttributes = vi.fn(); - registeredProcessors[1]!.onStart({ setAttributes } as never, otelContext.active()); + registeredProcessors[2]!.onStart({ setAttributes } as never, otelContext.active()); // agent_name rides the session metadata so spans (and logs) carry it even on // the custom-provider path, where the resource is left untouched. expect(setAttributes).toHaveBeenCalledWith({ @@ -279,7 +286,7 @@ describe('setupCloudTracer with a user-configured provider', () => { job_id: 'job1', 'lk.agent_name': 'my-agent', }); - expect(registeredProcessors[2]).toBeInstanceOf(BatchSpanProcessor); + expect(registeredProcessors[3]).toBeInstanceOf(BatchSpanProcessor); }); it('passes the gated exporter to a user-supplied cloud processor factory', async () => { @@ -303,8 +310,10 @@ describe('setupCloudTracer with a user-configured provider', () => { }); expect(createCloudSpanProcessor).toHaveBeenCalledOnce(); - expect(registeredProcessors).toHaveLength(2); - expect(registeredProcessors[1]).toBe(factoryProcessor); + // the PII stripper, the session metadata processor, then the factory's cloud processor + expect(registeredProcessors).toHaveLength(3); + expect(registeredProcessors[0]).toBeInstanceOf(PIIFilteringSpanProcessor); + expect(registeredProcessors[2]).toBe(factoryProcessor); }); it('requires registerSpanProcessor and never calls addSpanProcessor', async () => { diff --git a/agents/src/telemetry/traces.ts b/agents/src/telemetry/traces.ts index 1345c6310..4d5f65ff1 100644 --- a/agents/src/telemetry/traces.ts +++ b/agents/src/telemetry/traces.ts @@ -47,8 +47,10 @@ import { type SessionReport, sessionReportToJSON } from '../voice/report.js'; import type { ObservabilityEndpoint } from './observability_endpoint.js'; import { resolveObservabilityUrl } from './observability_endpoint.js'; import { type SimpleLogRecord, SimpleOTLPHttpLogExporter } from './otel_http_exporter.js'; +import { PIIFilteringSpanProcessor } from './pii.js'; import { flushPinoLogs, initPinoCloudExporter } from './pino_otel_transport.js'; import { uploadRecording } from './recording_upload.js'; +import { allowPiiFromEnv } from './redaction.js'; import { ATTR_AGENT_NAME, ATTR_CLOUD_AGENT_ID, ATTR_DEPLOYMENT_ID } from './trace_types.js'; import { UploadGateTraceExporter, uploadGate } from './upload_gate.js'; @@ -267,6 +269,40 @@ interface CustomProviderConfig { } const customProviderConfigs = new WeakMap(); +/** Providers that already carry the in-process PII stripper — installed at most once. */ +const piiRedactionInstalled = new WeakSet(); + +/** + * Installs {@link PIIFilteringSpanProcessor} on a provider LiveKit does not own. + * + * The processor strips PII in `onEnding`, which the SDK dispatches to every registered + * processor before any processor's `onEnd`, so it protects the integrator's exporters + * regardless of the order they were registered in. + */ +function installPIIRedaction( + provider: TracerProvider, + registerSpanProcessor: SpanProcessorRegistrar | undefined, + allowPii: boolean | undefined, +): void { + if (piiRedactionInstalled.has(provider)) return; + + if (!registerSpanProcessor) { + // never reached for a provider we own. addSpanProcessor is deliberately not used as a + // fallback: OpenTelemetry 2.x removed it, and attaching to an integrator's provider + // without being handed a registrar is not ours to do + console.warn( + 'Unable to install LiveKit PII redaction on the custom tracer provider, so its ' + + 'exporters may receive conversational content. Pass registerSpanProcessor to ' + + 'setTracerProvider, or set OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT=false.', + ); + return; + } + + piiRedactionInstalled.add(provider); + // PII flows to every exporter unless withheld: the GenAI conventions are only useful to a + // backend that can render the conversation + registerSpanProcessor(new PIIFilteringSpanProcessor(allowPii ?? allowPiiFromEnv() ?? true)); +} /** Options for configuring a custom tracer provider. */ export interface SetTracerProviderOptions { @@ -296,6 +332,16 @@ export interface SetTracerProviderOptions { * active. The returned processor must use OpenTelemetry SDK 2.x. */ createCloudSpanProcessor?: (options: CloudSpanProcessorOptions) => SpanProcessor; + /** + * Whether this provider's exporters may receive conversational content, tool payloads and + * other user data. + * + * Defaults to `true` (or `LIVEKIT_TELEMETRY_ALLOW_PII`, when set), since a GenAI backend + * can only render the conversation if it receives it. Pass `false` to strip PII in-process + * before every exporter but LiveKit Cloud's, leaving them the non-content attributes. + * Ignored when the project mandates redaction — that setting is not weakened from here. + */ + allowPii?: boolean; } /** @@ -349,6 +395,8 @@ export function setTracerProvider( ); } + installPIIRedaction(provider, registerSpanProcessor, options?.allowPii); + if (registerSpanProcessor) { customProviderConfigs.set(provider, { registerSpanProcessor, @@ -458,10 +506,16 @@ export async function setupCloudTracer( const tracerProvider = new NodeTracerProvider({ resource, spanProcessors: [ + // strips PII while the span is still mutable, ahead of every exporter's onEnd + new PIIFilteringSpanProcessor(allowPiiFromEnv() ?? true), new MetadataSpanProcessor(sessionMetadata), new BatchSpanProcessor(createCloudExporter()), ], }); + // the processor above is already attached, so record it: otherwise the + // setTracerProvider call below finds no registrar and warns that redaction could + // not be installed, on the default path where it demonstrably was + piiRedactionInstalled.add(tracerProvider); // register() installs an AsyncLocalStorageContextManager (needed for span nesting) // and sets the global tracer provider. Both use set-once semantics in the OTel API, // so if the user already called NodeSDK.start(), these are safe no-ops. @@ -490,6 +544,7 @@ export async function setupCloudTracer( // Resource shared by all exporters, so applying `resource` here would also relabel // the spans going to the user's own backend. room_id/job_id — the keys Cloud // correlates on — still ride along as span attributes via MetadataSpanProcessor. + installPIIRedaction(existingProvider, config.registerSpanProcessor, undefined); config.registerSpanProcessor(new MetadataSpanProcessor(sessionMetadata)); config.registerSpanProcessor(cloudSpanProcessor); } diff --git a/agents/src/telemetry/upload_gate.ts b/agents/src/telemetry/upload_gate.ts index 3c24c74be..d280224fc 100644 --- a/agents/src/telemetry/upload_gate.ts +++ b/agents/src/telemetry/upload_gate.ts @@ -6,6 +6,7 @@ import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-proto'; import { OTLPExporterError } from '@opentelemetry/otlp-exporter-base'; import type { ReadableSpan } from '@opentelemetry/sdk-trace-base'; import { log } from '../log_core.js'; +import { restorePii } from './redaction.js'; const DISABLED_MARKERS = ['data recording is disabled', 'disabled by owner']; @@ -106,7 +107,9 @@ export class UploadGateTraceExporter extends OTLPTraceExporter { return; } - super.export(items, (result) => { + // PII stripped for third-party exporters is put back here: what LiveKit Cloud may + // receive is the project's setting, applied at its collector + super.export(items.map(restorePii), (result) => { if (isDisabledTraceExport(result)) { uploadGate.disable(generation); resultCallback({ code: ExportResultCode.SUCCESS }); diff --git a/agents/src/telemetry/utils.test.ts b/agents/src/telemetry/utils.test.ts index 3c57583d9..e32d5f5ac 100644 --- a/agents/src/telemetry/utils.test.ts +++ b/agents/src/telemetry/utils.test.ts @@ -5,17 +5,16 @@ import type { Span } from '@opentelemetry/api'; import { SpanStatusCode } from '@opentelemetry/api'; import { beforeEach, describe, expect, it, vi } from 'vitest'; import { type JobContext, runWithJobContext } from '../job.js'; +import { REDACTED_EXCEPTION_MESSAGE } from './redaction.js'; import * as traceTypes from './trace_types.js'; -import { - REDACTED_EXCEPTION_MESSAGE, - type RecordExceptionOptions, - recordException, -} from './utils.js'; +import { type RecordExceptionOptions, recordException } from './utils.js'; function fakeSpan() { return { addEvent: vi.fn(), + isRecording: vi.fn(() => true), recordException: vi.fn(), + setAttribute: vi.fn(), setAttributes: vi.fn(), setStatus: vi.fn(), }; @@ -40,6 +39,8 @@ describe('recordException', () => { captureException(span, { redacted: false }); expect(span.recordException).toHaveBeenCalledOnce(); + // `error.type` is the GenAI/HTTP conventions' low-cardinality error identifier + expect(span.setAttribute).toHaveBeenCalledWith(traceTypes.ATTR_ERROR_TYPE, 'Error'); expect(span.setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.ERROR, message: 'secret transcript', @@ -67,10 +68,13 @@ describe('recordException', () => { message: REDACTED_EXCEPTION_MESSAGE, }); expect(span.setAttributes).toHaveBeenCalledWith(attrs); + // `error.type` names the error class, never its message, so it survives redaction + expect(span.setAttribute).toHaveBeenCalledWith(traceTypes.ATTR_ERROR_TYPE, 'Error'); expect( JSON.stringify([ span.addEvent.mock.calls, span.setStatus.mock.calls, + span.setAttribute.mock.calls, span.setAttributes.mock.calls, ]), ).not.toContain('secret transcript'); diff --git a/agents/src/telemetry/utils.ts b/agents/src/telemetry/utils.ts index 25bf54387..45edfd33c 100644 --- a/agents/src/telemetry/utils.ts +++ b/agents/src/telemetry/utils.ts @@ -1,13 +1,32 @@ // SPDX-FileCopyrightText: 2025 LiveKit, Inc. // // SPDX-License-Identifier: Apache-2.0 +import type { Attributes } from '@opentelemetry/api'; import { type Span, SpanStatusCode, context as otelContext, trace } from '@opentelemetry/api'; import { getJobContext } from '../job.js'; import type { RealtimeModelMetrics } from '../metrics/base.js'; +import { ATTRIBUTE_REDACTION_ENABLED } from '../types.js'; +import { realtimeUsageAttributes, setErrorType } from './gen_ai.js'; import { REDACTED_EXCEPTION_MESSAGE } from './redaction.js'; import * as traceTypes from './trace_types.js'; import { tracer } from './traces.js'; +/** + * Whether the project has mandated PII redaction. + * + * Set in the LiveKit Cloud dashboard (or per session with `record: { redaction: true }`) and + * never weakened from here — when it is on, PII is stripped for every destination, LiveKit + * Cloud included. `spanAttributes` lets a span ended outside the job's async context resolve + * from the flag stamped on it at span start. + * + * Stripping PII for *third-party* exporters is not this flag: that is the default, and is + * lifted per provider with `setTracerProvider(provider, { allowPii: true })`. + */ +export function redactionEnabled(spanAttributes?: Attributes): boolean { + if (spanAttributes?.[ATTRIBUTE_REDACTION_ENABLED]) return true; + return getJobContext(false)?._redactionEnabled ?? false; +} + export { REDACTED_EXCEPTION_MESSAGE } from './redaction.js'; export interface RecordExceptionOptions { @@ -23,7 +42,12 @@ export function recordException( error: Error, options: RecordExceptionOptions = {}, ): void { - const redacted = options.redacted ?? getJobContext(false)?._redactionEnabled ?? false; + const redacted = options.redacted ?? redactionEnabled(); + + // `error.type` is the GenAI/HTTP conventions' low-cardinality error identifier; unlike the + // message it never carries user data, so it is set either way + setErrorType(span, error); + if (redacted) { const attrs = { [traceTypes.ATTR_EXCEPTION_TYPE]: error.constructor.name, @@ -54,11 +78,13 @@ export function recordException( } export function recordRealtimeMetrics(span: Span, metrics: RealtimeModelMetrics): void { - const attrs: Record = { - [traceTypes.ATTR_GEN_AI_REQUEST_MODEL]: metrics.label || 'unknown', + const attrs: Record = { + // a realtime turn is a multimodal generation; `gen_ai.output.type` is set on the + // inference span, which knows whether this session outputs audio + [traceTypes.ATTR_GEN_AI_OPERATION_NAME]: traceTypes.GenAIOperationName.GENERATE_CONTENT, [traceTypes.ATTR_REALTIME_MODEL_METRICS]: JSON.stringify(metrics), - [traceTypes.ATTR_GEN_AI_USAGE_INPUT_TOKENS]: metrics.inputTokens, - [traceTypes.ATTR_GEN_AI_USAGE_OUTPUT_TOKENS]: metrics.outputTokens, + ...(realtimeUsageAttributes(metrics) as Record), + // unofficial spellings LangFuse reads, kept alongside the official ones [traceTypes.ATTR_GEN_AI_USAGE_INPUT_TEXT_TOKENS]: metrics.inputTokenDetails.textTokens, [traceTypes.ATTR_GEN_AI_USAGE_INPUT_AUDIO_TOKENS]: metrics.inputTokenDetails.audioTokens, [traceTypes.ATTR_GEN_AI_USAGE_INPUT_CACHED_TOKENS]: metrics.inputTokenDetails.cachedTokens, @@ -66,6 +92,19 @@ export function recordRealtimeMetrics(span: Span, metrics: RealtimeModelMetrics) [traceTypes.ATTR_GEN_AI_USAGE_OUTPUT_AUDIO_TOKENS]: metrics.outputTokenDetails.audioTokens, }; + // `metrics.label` names the plugin, not the model, so it is never reported as one + const modelName = metrics.metadata?.modelName; + if (modelName) { + attrs[traceTypes.ATTR_GEN_AI_REQUEST_MODEL] = modelName; + attrs[traceTypes.ATTR_GEN_AI_RESPONSE_MODEL] = modelName; + } + const provider = traceTypes.genAIProviderName(metrics.metadata?.modelProvider); + if (provider) attrs[traceTypes.ATTR_GEN_AI_PROVIDER_NAME] = provider; + if (metrics.requestId) attrs[traceTypes.ATTR_GEN_AI_RESPONSE_ID] = metrics.requestId; + if (metrics.ttftMs !== undefined && metrics.ttftMs >= 0) { + attrs[traceTypes.ATTR_GEN_AI_RESPONSE_TIME_TO_FIRST_CHUNK] = metrics.ttftMs / 1000; + } + // Add LangFuse-specific completion start time if TTFT is available if (metrics.ttftMs !== undefined && metrics.ttftMs !== -1) { const completionStartTime = metrics.timestamp + metrics.ttftMs; diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index bc20e9c90..152a8986f 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -69,7 +69,7 @@ import type { import { IdentityTransform } from '../stream/identity_transform.js'; import { MultiInputStream } from '../stream/multi_input_stream.js'; import { STT, type STTError, type SpeechEvent } from '../stt/stt.js'; -import { recordRealtimeMetrics, traceTypes, tracer } from '../telemetry/index.js'; +import { genAI, recordRealtimeMetrics, traceTypes, tracer } from '../telemetry/index.js'; import { splitWords } from '../tokenize/basic/word.js'; import { TTS, type TTSError } from '../tts/tts.js'; import { isFlushSentinel } from '../types.js'; @@ -597,7 +597,13 @@ export class AgentActivity implements RecognitionHooks { const startSpan = tracer.startSpan({ name: spanName, attributes: { [traceTypes.ATTR_AGENT_LABEL]: this.agent.id }, - context: ROOT_CONTEXT, + context: this.agentSession.rootSpanContext ?? ROOT_CONTEXT, + }); + genAI.setAgentAttributes(startSpan, { + operation: traceTypes.GenAIOperationName.CREATE_AGENT, + agentName: this.agent.id, + model: this.llm?.model, + provider: this.llm?.provider, }); this.agent._agentActivity = this; @@ -3699,6 +3705,7 @@ export class AgentActivity implements RecognitionHooks { }; const [executeToolsTask, toolOutput] = performToolExecutions({ + agentName: this.agent.id, session: this.agentSession, speechHandle, toolCtx, @@ -3942,6 +3949,17 @@ 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, @@ -3953,7 +3971,8 @@ export class AgentActivity implements RecognitionHooks { _previousUserMetrics?: MetricsReport, ): Promise => tracer.startActiveSpan( - async (span) => + async (span) => ( + this.recordAgentTurn(span), this._pipelineReplyTaskImpl({ stateLease, chatCtx, @@ -3964,7 +3983,8 @@ export class AgentActivity implements RecognitionHooks { newMessage, span, _previousUserMetrics, - }), + }) + ), { name: 'agent_turn', context: this.agentSession.rootSpanContext, @@ -3979,15 +3999,23 @@ export class AgentActivity implements RecognitionHooks { addToChatCtx: boolean = true, ): Promise { return tracer.startActiveSpan( - async (span) => - this._realtimeGenerationTaskImpl({ - stateLease, - ev, - modelSettings, - replyAbortController, - addToChatCtx, - span, - }), + async (span) => { + this.recordAgentTurn(span); + const inferenceSpan = tracer.startSpan({ name: 'realtime_inference' }); + try { + return await this._realtimeGenerationTaskImpl({ + stateLease, + ev, + modelSettings, + replyAbortController, + addToChatCtx, + span, + inferenceSpan, + }); + } finally { + inferenceSpan.end(); + } + }, { name: 'agent_turn', context: this.agentSession.rootSpanContext, @@ -4002,6 +4030,7 @@ export class AgentActivity implements RecognitionHooks { replyAbortController, addToChatCtx, span, + inferenceSpan, }: { stateLease: AgentStateLease; ev: GenerationCreatedEvent; @@ -4009,6 +4038,7 @@ export class AgentActivity implements RecognitionHooks { replyAbortController: AbortController; addToChatCtx: boolean; span: Span; + inferenceSpan: Span; }): Promise { const { speechHandle } = stateLease; speechHandle._agentTurnContext = otelContext.active(); @@ -4031,10 +4061,20 @@ export class AgentActivity implements RecognitionHooks { throw new Error('llm is not a realtime model'); } - // Store span for metrics recording when they arrive later - span.setAttribute(traceTypes.ATTR_GEN_AI_REQUEST_MODEL, realtimeModel.model); + genAI.setRequestAttributes(inferenceSpan, { + operation: traceTypes.GenAIOperationName.GENERATE_CONTENT, + provider: realtimeModel.provider, + model: realtimeModel.model, + stream: true, + // a realtime model can be configured text-only + outputType: this.agentSession.output.audioEnabled + ? traceTypes.GenAIOutputType.SPEECH + : traceTypes.GenAIOutputType.TEXT, + }); + // the provider metrics land here rather than on `agent_turn`; they can arrive after the + // turn ends, in which case recordRealtimeMetrics opens its own child span if (this.realtimeSpans && ev.responseId) { - this.realtimeSpans.set(ev.responseId, span); + this.realtimeSpans.set(ev.responseId, inferenceSpan); } this.logger.debug( @@ -4368,6 +4408,7 @@ export class AgentActivity implements RecognitionHooks { }; const [executeToolsTask, toolOutput] = performToolExecutions({ + agentName: this.agent.id, session: this.agentSession, speechHandle, toolCtx, @@ -5038,10 +5079,12 @@ export class AgentActivity implements RecognitionHooks { } async drain(options?: { newActivity?: AgentActivity }): Promise { - // Create drain_agent_activity as a ROOT span (new trace) to match Python behavior + // parented to the session rather than to whichever speech task is current, so the whole + // session stays one trace. Python reaches the same place by inheriting the session + // context that AgentSession attaches. return tracer.startActiveSpan(async (span) => this._drainImpl(span, options?.newActivity), { name: 'drain_agent_activity', - context: ROOT_CONTEXT, + context: this.agentSession.rootSpanContext ?? ROOT_CONTEXT, }); } diff --git a/agents/src/voice/agent_session.ts b/agents/src/voice/agent_session.ts index bf815df36..ceb55e156 100644 --- a/agents/src/voice/agent_session.ts +++ b/agents/src/voice/agent_session.ts @@ -55,7 +55,7 @@ import { type ModelUsage, ModelUsageCollector, filterZeroValues } from '../metri import { SimulationMode } from '../simulation.js'; import type { STT } from '../stt/index.js'; import type { STTError } from '../stt/stt.js'; -import { traceTypes, tracer } from '../telemetry/index.js'; +import { genAI, traceTypes, tracer } from '../telemetry/index.js'; import { DEFAULT_SPEECH_STEERING_OPTIONS, type SpeechSteeringOptions, @@ -1041,6 +1041,9 @@ export class AgentSession< this.sessionSpan = tracer.startSpan({ name: 'agent_session', }); + // the session is the convention's workflow: agent turns (`invoke_agent`), inference + // (`chat`) and tool spans (`execute_tool`) nest underneath it + genAI.setWorkflowAttributes(this.sessionSpan, { name: 'agent_session' }); this.rootSpanContext = trace.setSpan(otelContext.active(), this.sessionSpan); diff --git a/agents/src/voice/generation.ts b/agents/src/voice/generation.ts index f5432932b..54b9b7168 100644 --- a/agents/src/voice/generation.ts +++ b/agents/src/voice/generation.ts @@ -31,7 +31,7 @@ import { parseFunctionArguments } from '../llm/utils.js'; import { isZodSchema, parseZodSchema } from '../llm/zod-utils.js'; import { log } from '../log.js'; import { IdentityTransform } from '../stream/identity_transform.js'; -import { traceTypes, tracer } from '../telemetry/index.js'; +import { genAI, traceTypes, tracer } from '../telemetry/index.js'; import { stripAllMarkup } from '../tts/provider_format.js'; import { type FlushSentinel, @@ -631,7 +631,11 @@ export function performLLMInference( const toolCallWriter = toolCallStream.writable.getWriter(); const data = new _LLMGenerationData(textStream.readable, toolCallStream.readable); - const _performLLMInferenceImpl = async (signal: AbortSignal, span: Span) => { + const _performLLMInferenceImpl = async ( + signal: AbortSignal, + span: Span, + inference: genAI.InferenceMarker, + ) => { span.setAttribute( traceTypes.ATTR_CHAT_CTX, // snake_case wire shape, matching Python's `chat_ctx.to_dict()` for this span attribute @@ -640,12 +644,21 @@ export function performLLMInference( ); span.setAttribute(traceTypes.ATTR_FUNCTION_TOOLS, JSON.stringify(sortedToolNames(toolCtx))); - if (model) { - span.setAttribute(traceTypes.ATTR_GEN_AI_REQUEST_MODEL, model); - } - if (provider) { - span.setAttribute(traceTypes.ATTR_GEN_AI_PROVIDER_NAME, provider); - } + // the configured model and provider describe the inference only once it is known that + // this LLM served it; that is decided below, when the nested span is (or is not) there + const recordConfiguredModel = () => { + if (model) span.setAttribute(traceTypes.ATTR_GEN_AI_REQUEST_MODEL, model); + const normalizedProvider = traceTypes.genAIProviderName(provider); + if (normalizedProvider) { + span.setAttribute(traceTypes.ATTR_GEN_AI_PROVIDER_NAME, normalizedProvider); + } + }; + let nodeError: Error | string | undefined; + + // the GenAI inference attributes belong to the nested `llm_request` span, which is the + // provider call the convention describes. Setting them here as well made a backend + // summing gen_ai.usage.* over inference spans report twice the calls and tokens, and + // serialised the whole chat context onto both spans. let llmStreamReader: ReadableStreamDefaultReader | null = null; @@ -732,6 +745,9 @@ export function performLLMInference( // Python since chunk is defined in the type ChatChunk | string in TypeScript } } catch (error) { + // an Error is classified by setErrorType; anything else is never used verbatim, + // since error.type is low-cardinality and a thrown value can carry content + nodeError = error instanceof Error ? error : 'UnknownError'; if (error instanceof DOMException && error.name === 'AbortError') { // Abort signal was triggered, handle gracefully return; @@ -755,6 +771,38 @@ export function performLLMInference( if (data.ttft !== undefined) { span.setAttribute(traceTypes.ATTR_RESPONSE_TTFT, data.ttft); } + // a custom node may have generated this itself, with no nested `llm_request` span to + // carry the convention's attributes; when there was one, they are already recorded + if (inference.recorded) { + recordConfiguredModel(); + } else { + // a third-party engine served this, so the configured model and provider are left + // off rather than crediting it with a call it never made + genAI.setRequestAttributes(span, { + operation: traceTypes.GenAIOperationName.CHAT, + stream: true, + outputType: traceTypes.GenAIOutputType.TEXT, + }); + if (nodeError !== undefined) genAI.setErrorType(span, nodeError); + const finishReason = genAI.finishReasonFor({ + functionCalls: data.generatedToolCalls, + interrupted: nodeError !== undefined, + }); + genAI.setResponseAttributes(span, { + finishReasons: [finishReason], + timeToFirstChunk: data.ttft, + }); + genAI.setContentAttributes(span, { + systemInstructions: genAI.toSystemInstructions(chatCtx), + inputMessages: genAI.toInputMessages(chatCtx), + toolDefinitions: genAI.toToolDefinitions(toolCtx.functionTools), + outputMessages: genAI.toOutputMessages({ + text: data.generatedText, + functionCalls: data.generatedToolCalls, + finishReason, + }), + }); + } llmStreamReader?.releaseLock(); await llmStream?.cancel(); await textWriter.close(); @@ -766,10 +814,11 @@ export function performLLMInference( const currentContext = otelContext.active(); const inferenceTask = async (signal: AbortSignal) => - tracer.startActiveSpan(async (span) => _performLLMInferenceImpl(signal, span), { - name: 'llm_node', - context: currentContext, - }); + tracer.startActiveSpan( + async (span) => + genAI.withInferenceTracking((marker) => _performLLMInferenceImpl(signal, span, marker)), + { name: 'llm_node', context: currentContext }, + ); return [ Task.from((controller) => inferenceTask(controller.signal), controller, 'performLLMInference'), @@ -1199,6 +1248,7 @@ export function performToolExecutions({ toolCtx, toolChoice, toolCallStream, + agentName, onToolExecutionStarted = () => {}, onToolExecutionCompleted = () => {}, controller, @@ -1208,6 +1258,8 @@ export function performToolExecutions({ toolCtx: ToolContext; toolChoice?: ToolChoice; toolCallStream: ReadableStream; + /** The agent that invoked these tools — a handoff may already have swapped the session's. */ + agentName?: string; onToolExecutionStarted?: (toolCall: FunctionCall) => void; onToolExecutionCompleted?: (toolExecutionOutput: ToolExecutionOutput) => void; controller: AbortController; @@ -1368,6 +1420,13 @@ export function performToolExecutions({ const _tracableToolExecutionImpl = async (toolExecTask: Promise, span: Span) => { span.setAttribute(traceTypes.ATTR_FUNCTION_TOOL_NAME, toolCall.name); span.setAttribute(traceTypes.ATTR_FUNCTION_TOOL_ARGS, toolCall.args); + genAI.setToolAttributes(span, { + name: toolCall.name, + callId: toolCall.callId, + description: isFunctionTool(tool) ? tool.description : undefined, + args: toolCall.args, + agentName, + }); // Only completed executions produce tool output. An interrupted execution may still // finish in the background and must remain retryable rather than becoming a synthetic @@ -1390,6 +1449,10 @@ export function performToolExecutions({ traceTypes.ATTR_FUNCTION_TOOL_IS_ERROR, toolOutput.toolCallOutput.isError, ); + genAI.setToolResult(span, { + result: toolOutput.toolCallOutput.output, + isError: toolOutput.toolCallOutput.isError, + }); } } catch (rawError) { logger.error( @@ -1411,6 +1474,7 @@ export function performToolExecutions({ toolOutput.toolCallOutput.output, ); span.setAttribute(traceTypes.ATTR_FUNCTION_TOOL_IS_ERROR, true); + genAI.setToolResult(span, { isError: true }); } } finally { if (toolOutput) toolCompleted(toolOutput); diff --git a/agents/src/voice/generation_telemetry.test.ts b/agents/src/voice/generation_telemetry.test.ts index 4dd0c231d..33b1240f1 100644 --- a/agents/src/voice/generation_telemetry.test.ts +++ b/agents/src/voice/generation_telemetry.test.ts @@ -140,6 +140,16 @@ describe('performLLMInference response telemetry', () => { } expectFunctionCallTelemetry(span); + + // this node produced the response itself, without an LLMStream, so there is no nested + // `llm_request` span and the node span carries the convention's attributes + expect(span.attributes['gen_ai.operation.name']).toBe('chat'); + expect(span.attributes['gen_ai.response.finish_reasons']).toEqual(['tool_call']); + // the chat context is empty here, so only the output side has content to record + expect(JSON.parse(span.attributes['gen_ai.output.messages'] as string)[0].parts).toEqual([ + { type: 'text', content: 'partial response' }, + expect.objectContaining({ type: 'tool_call', name: 'lookup_weather' }), + ]); }); it('records accumulated response telemetry when the stream aborts', async () => { diff --git a/turbo.json b/turbo.json index c78e7c8fa..25d40cde1 100644 --- a/turbo.json +++ b/turbo.json @@ -57,6 +57,7 @@ "LOG_LEVEL", "OCTOAI_TOKEN", "OPENAI_API_KEY", + "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", "OTEL_RESOURCE_ATTRIBUTES", "OPENAI_API_VERSION", "OPENAI_BASE_URL", @@ -80,6 +81,7 @@ "LIVEKIT_SIP_NUMBER", "LIVEKIT_SIP_OUTBOUND_TRUNK", "LIVEKIT_SUPERVISOR_PHONE_NUMBER", + "LIVEKIT_TELEMETRY_ALLOW_PII", "GOOGLE_API_KEY", "GOOGLE_GENAI_API_KEY", "GOOGLE_GENAI_USE_VERTEXAI",