feat(telemetry): tag content-bearing trace and log keys with lk.pii. for redaction AGT-3074

.changeset/redact-pii-telemetry.md

@@ -0,0 +1,5 @@
+---
+'@livekit/agents': patch
+---
+
+Tag content-bearing telemetry with `lk.pii.*` and redact trace exception details when job redaction is enabled. Dashboards and queries using the previous sensitive trace keys must migrate to their `lk.pii.*` replacements.

agents/src/inference/interruption/ws_transport.ts

@@ -184,7 +184,10 @@ export function createWsTransport(
         message = wsMessageSchema.parse(JSON.parse(data.toString()));
       } catch (err) {
         logger.warn(
-          { data: data.toString(), err: err instanceof Error ? err.message : String(err) },
+          {
+            'lk.pii.data': data.toString(),
+            err: err instanceof Error ? err.message : String(err),
+          },
           'Failed to parse WebSocket message',
         );
         return;

agents/src/inference/stt.ts

@@ -887,7 +887,10 @@ export class SpeechStream<TModel extends STTModels> extends BaseSpeechStream {
             const parseResult = await sttServerEventSchema.safeParseAsync(result.value);
             if (!parseResult.success) {
               this.#logger.warn(
-                { error: parseResult.error, rawData: result.value },
+                {
+                  error: parseResult.error,
+                  'lk.pii.raw_data': result.value,
+                },
                 'Failed to parse STT server event',
               );
               continue;

agents/src/ipc/job_proc_lazy_main.ts

@@ -64,7 +64,9 @@ class InfClient implements InferenceExecutor {
           const fut = this.#requests[msg.value.requestId];
           delete this.#requests[msg.value.requestId];
           if (!fut) {
-            this.#logger.child({ resp: msg.value }).warn('received unexpected inference response');
+            this.#logger
+              .child({ 'lk.pii.response': msg.value })
+              .warn('received unexpected inference response');
             return;
           }
           fut.resolve(msg.value);

agents/src/llm/fallback_adapter.ts

@@ -364,7 +364,10 @@ class FallbackLLMStream extends LLMStream {
 
           // Check if we sent data before failing
           if (textSent || toolCallsSent.length > 0) {
-            const extra = { textSent, toolCallsSent };
+            const extra = {
+              'lk.pii.response.text': textSent,
+              'lk.pii.response.function_calls': toolCallsSent,
+            };
 
             if (!this.adapter.retryOnChunkSent) {
               this._log.error(

agents/src/log.test.ts

@@ -39,4 +39,82 @@ describe('OTEL logging', () => {
       expect(messages).toContain('log from fresh logger');
     });
   });
+
+  it('exports sensitive content as a redactable attribute instead of log body text', async () => {
+    initializeLogger({ pretty: false, level: 'info' });
+    const emitSpy = vi.spyOn(PinoCloudExporter.prototype, 'emit').mockImplementation(() => {});
+
+    initPinoCloudExporter({
+      cloudHostname: 'example.livekit.cloud',
+      roomId: 'RM_test',
+      jobId: 'AJ_test',
+    });
+    enableOtelLogging();
+
+    log().info({ 'lk.pii.user_input': 'secret transcript' }, 'received user input');
+
+    await vi.waitFor(() => {
+      const record = emitSpy.mock.calls
+        .map(([logObj]) => logObj)
+        .find((logObj) => {
+          return logObj.msg === 'received user input';
+        });
+      expect(record).toMatchObject({
+        msg: 'received user input',
+        'lk.pii.user_input': 'secret transcript',
+      });
+      expect(record?.msg).not.toContain('secret transcript');
+    });
+  });
+
+  it('keeps operational errors, reasons, framework URLs, and resource IDs in non-PII fields', async () => {
+    initializeLogger({ pretty: false, level: 'info' });
+    const emitSpy = vi.spyOn(PinoCloudExporter.prototype, 'emit').mockImplementation(() => {});
+
+    initPinoCloudExporter({
+      cloudHostname: 'example.livekit.cloud',
+      roomId: 'RM_test',
+      jobId: 'AJ_test',
+    });
+    enableOtelLogging();
+
+    log().error({ error: new Error('connection failed') }, 'provider failed');
+    log().warn({ reason: 'remote close' }, 'provider disconnected');
+    log().info({ baseUrl: 'wss://example.livekit.cloud' }, 'connecting to framework');
+    log().info({ avatarId: 'avatar-123' }, 'avatar session started');
+
+    await vi.waitFor(() => {
+      const records = emitSpy.mock.calls.map(([logObj]) => logObj);
+      const errorRecord = records.find((logObj) => logObj.msg === 'provider failed');
+      expect(errorRecord).toMatchObject({
+        msg: 'provider failed',
+        error: {
+          type: 'Error',
+          message: 'connection failed',
+        },
+      });
+      expect(errorRecord).not.toHaveProperty('lk.pii.error');
+
+      const reasonRecord = records.find((logObj) => logObj.msg === 'provider disconnected');
+      expect(reasonRecord).toMatchObject({
+        msg: 'provider disconnected',
+        reason: 'remote close',
+      });
+      expect(reasonRecord).not.toHaveProperty('lk.pii.reason');
+
+      const connectionRecord = records.find((logObj) => logObj.msg === 'connecting to framework');
+      expect(connectionRecord).toMatchObject({
+        msg: 'connecting to framework',
+        baseUrl: 'wss://example.livekit.cloud',
+      });
+      expect(connectionRecord).not.toHaveProperty('lk.pii.base_url');
+
+      const avatarRecord = records.find((logObj) => logObj.msg === 'avatar session started');
+      expect(avatarRecord).toMatchObject({
+        msg: 'avatar session started',
+        avatarId: 'avatar-123',
+      });
+      expect(avatarRecord).not.toHaveProperty('lk.pii.avatar_id');
+    });
+  });
 });

agents/src/telemetry/index.ts

@@ -29,4 +29,9 @@ export {
   type SpanProcessorLike,
   type StartSpanOptions,
 } from './traces.js';
-export { recordException, recordRealtimeMetrics } from './utils.js';
+export {
+  REDACTED_EXCEPTION_MESSAGE,
+  recordException,
+  recordRealtimeMetrics,
+  type RecordExceptionOptions,
+} from './utils.js';

agents/src/telemetry/trace_types.test.ts

@@ -0,0 +1,125 @@
+// SPDX-FileCopyrightText: 2026 LiveKit, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+import { describe, expect, it } from 'vitest';
+import * as traceTypes from './trace_types.js';
+
+const PII_SEGMENT_RE = /(^|\.)pii(\.|$)/i;
+
+// Keys that carry no conversational content, tool payloads, or other user data.
+const SAFE_KEYS = new Set([
+  // Correlation IDs and session metadata
+  'lk.speech_id',
+  'lk.agent_label',
+  'lk.start_time',
+  'lk.end_time',
+  'lk.retry_count',
+  'lk.provider_request_ids',
+  'lk.participant_id',
+  'lk.participant_identity',
+  'lk.participant_kind',
+  'lk.job_id',
+  'lk.agent_name',
+  'lk.room_name',
+  'lk.session_options',
+  'lk.generation_id',
+  'lk.parent_generation_id',
+  'lk.interrupted',
+  // LLM node metadata
+  'lk.function_tools',
+  'lk.provider_tools',
+  'lk.tool_sets',
+  'lk.response.ttft',
+  // Function tool metadata
+  'lk.function_tool.id',
+  'lk.function_tool.name',
+  'lk.function_tool.is_error',
+  // TTS metadata
+  'lk.tts.streaming',
+  'lk.tts.label',
+  'lk.response.ttfb',
+  // EOU detection
+  'lk.eou.probability',
+  'lk.eou.unlikely_threshold',
+  'lk.eou.endpointing_delay',
+  'lk.eou.language',
+  'lk.eou.source',
+  'lk.eou.from_cache',
+  'lk.eou.detection_delay',
+  'lk.transcript_confidence',
+  'lk.transcription_delay',
+  'lk.end_of_turn_delay',
+  // Metrics
+  'lk.llm_metrics',
+  'lk.tts_metrics',
+  'lk.realtime_model_metrics',
+  'lk.e2e_latency',
+  // OpenTelemetry GenAI attributes and event names
+  'gen_ai.operation.name',
+  'gen_ai.request.model',
+  'gen_ai.provider.name',
+  'gen_ai.usage.input_tokens',
+  'gen_ai.usage.output_tokens',
+  'gen_ai.usage.input_text_tokens',
+  'gen_ai.usage.input_audio_tokens',
+  'gen_ai.usage.input_cached_tokens',
+  'gen_ai.usage.output_text_tokens',
+  'gen_ai.usage.output_audio_tokens',
+  'gen_ai.system.message',
+  'gen_ai.user.message',
+  'gen_ai.assistant.message',
+  'gen_ai.tool.message',
+  'gen_ai.choice',
+  // OpenTelemetry exception attributes
+  'exception.stacktrace',
+  'exception.type',
+  'exception.message',
+  // Vendor metadata
+  'langfuse.observation.completion_start_time',
+  // Answering machine detection
+  'lk.amd.category',
+  'lk.amd.reason',
+  'lk.amd.is_machine',
+  'lk.amd.interrupt_on_machine',
+  'lk.amd.speech_duration',
+  'lk.amd.delay',
+  // Adaptive interruption
+  'lk.is_interruption',
+  'lk.interruption.probability',
+  'lk.interruption.total_duration',
+  'lk.interruption.prediction_duration',
+  'lk.interruption.detection_delay',
+]);
+
+function declaredKeys(): Record<string, string> {
+  return Object.fromEntries(
+    Object.entries(traceTypes).filter((entry): entry is [string, string] => {
+      return typeof entry[1] === 'string';
+    }),
+  );
+}
+
+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),
+      ),
+    );
+
+    expect(unclassified).toEqual({});
+  });
+
+  it('does not mark safe keys as PII-bearing', () => {
+    const conflicting = [...SAFE_KEYS].filter((key) => PII_SEGMENT_RE.test(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();
+
+    expect(stale).toEqual([]);
+  });
+});

agents/src/telemetry/trace_types.ts

@@ -2,6 +2,16 @@
 //
 // SPDX-License-Identifier: Apache-2.0
 
+/**
+ * Span attribute and event name constants for LiveKit Agents telemetry.
+ *
+ * Attributes carrying conversational content, tool payloads, or other user data must include a
+ * dot-delimited `pii` segment (`lk.pii.<name>`). PII-enabled projects have these attributes
+ * stripped at the LiveKit Cloud collector, and the segment is the marker it honors. Such content
+ * must not be embedded in span names, event names, or log message bodies because those are not
+ * redactable.
+ */
+
 // LiveKit custom attributes
 export const ATTR_SPEECH_ID = 'lk.speech_id';
 export const ATTR_AGENT_LABEL = 'lk.agent_label';
@@ -32,29 +42,29 @@ export const ATTR_SESSION_OPTIONS = 'lk.session_options';
 // assistant turn
 export const ATTR_AGENT_TURN_ID = 'lk.generation_id';
 export const ATTR_AGENT_PARENT_TURN_ID = 'lk.parent_generation_id';
-export const ATTR_USER_INPUT = 'lk.user_input';
-export const ATTR_INSTRUCTIONS = 'lk.instructions';
+export const ATTR_USER_INPUT = 'lk.pii.user_input';
+export const ATTR_INSTRUCTIONS = 'lk.pii.instructions';
 export const ATTR_SPEECH_INTERRUPTED = 'lk.interrupted';
 
 // llm node
-export const ATTR_CHAT_CTX = 'lk.chat_ctx';
+export const ATTR_CHAT_CTX = 'lk.pii.chat_ctx';
 export const ATTR_FUNCTION_TOOLS = 'lk.function_tools';
 export const ATTR_PROVIDER_TOOLS = 'lk.provider_tools';
 export const ATTR_TOOL_SETS = 'lk.tool_sets';
-export const ATTR_RESPONSE_TEXT = 'lk.response.text';
-export const ATTR_RESPONSE_FUNCTION_CALLS = 'lk.response.function_calls';
+export const ATTR_RESPONSE_TEXT = 'lk.pii.response.text';
+export const ATTR_RESPONSE_FUNCTION_CALLS = 'lk.pii.response.function_calls';
 /** Time to first token in seconds. */
 export const ATTR_RESPONSE_TTFT = 'lk.response.ttft';
 
 // function tool
 export const ATTR_FUNCTION_TOOL_ID = 'lk.function_tool.id';
 export const ATTR_FUNCTION_TOOL_NAME = 'lk.function_tool.name';
-export const ATTR_FUNCTION_TOOL_ARGS = 'lk.function_tool.arguments';
+export const ATTR_FUNCTION_TOOL_ARGS = 'lk.pii.function_tool.arguments';
 export const ATTR_FUNCTION_TOOL_IS_ERROR = 'lk.function_tool.is_error';
-export const ATTR_FUNCTION_TOOL_OUTPUT = 'lk.function_tool.output';
+export const ATTR_FUNCTION_TOOL_OUTPUT = 'lk.pii.function_tool.output';
 
 // tts node
-export const ATTR_TTS_INPUT_TEXT = 'lk.input_text';
+export const ATTR_TTS_INPUT_TEXT = 'lk.pii.input_text';
 export const ATTR_TTS_STREAMING = 'lk.tts.streaming';
 export const ATTR_TTS_LABEL = 'lk.tts.label';
 /** Time to first byte in seconds. */
@@ -72,7 +82,7 @@ export const ATTR_EOU_SOURCE = 'lk.eou.source';
 export const ATTR_EOU_FROM_CACHE = 'lk.eou.from_cache';
 /** Latest input-audio creation time → prediction receive time (ms). */
 export const ATTR_EOU_DETECTION_DELAY = 'lk.eou.detection_delay';
-export const ATTR_USER_TRANSCRIPT = 'lk.user_transcript';
+export const ATTR_USER_TRANSCRIPT = 'lk.pii.user_transcript';
 export const ATTR_TRANSCRIPT_CONFIDENCE = 'lk.transcript_confidence';
 export const ATTR_TRANSCRIPTION_DELAY = 'lk.transcription_delay';
 export const ATTR_END_OF_TURN_DELAY = 'lk.end_of_turn_delay';
@@ -86,7 +96,7 @@ export const ATTR_AMD_INTERRUPT_ON_MACHINE = 'lk.amd.interrupt_on_machine';
 export const ATTR_AMD_SPEECH_DURATION = 'lk.amd.speech_duration';
 /** Time between speech end and the AMD verdict emission (milliseconds). */
 export const ATTR_AMD_DELAY = 'lk.amd.delay';
-export const ATTR_AMD_TRANSCRIPT = 'lk.amd.transcript';
+export const ATTR_AMD_TRANSCRIPT = 'lk.pii.amd.transcript';
 
 // Adaptive Interruption attributes
 export const ATTR_IS_INTERRUPTION = 'lk.is_interruption';

agents/src/telemetry/traces.test.ts

@@ -432,6 +432,38 @@ describe('uploadSessionReport metadata', () => {
     expect(records[0]?.attributes).not.toHaveProperty('session.simulation');
   });
 
+  it('marks session keyterms as PII in exported session-report logs', async () => {
+    const exportSpy = vi
+      .spyOn(SimpleOTLPHttpLogExporter.prototype, 'export')
+      .mockResolvedValue(undefined);
+    const keytermsOptions = {
+      keyterms: ['Acme Corp'],
+      keytermDetection: { enabled: false },
+    };
+    const report = makeReport({
+      audio: false,
+      traces: true,
+      logs: false,
+      transcript: false,
+      redaction: false,
+    });
+    report.options = { keytermsOptions };
+
+    await uploadSessionReport({
+      agentName: 'agent',
+      cloudHostname: 'example.livekit.cloud',
+      report,
+    });
+
+    const records = exportSpy.mock.calls[0]?.[0] ?? [];
+    const serializedOptions = records[0]?.attributes['session.options'] as Record<string, unknown>;
+    const serializedKeytermsOptions = serializedOptions.keytermsOptions as Record<string, unknown>;
+    expect(serializedKeytermsOptions['lk.pii.keyterms']).toEqual(['Acme Corp']);
+    expect(serializedKeytermsOptions).not.toHaveProperty('keyterms');
+    expect(serializedKeytermsOptions.keytermDetection).toEqual({ enabled: false });
+    expect(keytermsOptions.keyterms).toEqual(['Acme Corp']);
+  });
+
   it('sets job, simulation, and redaction fields on the multipart recording header', async () => {
     vi.spyOn(SimpleOTLPHttpLogExporter.prototype, 'export').mockResolvedValue(undefined);
     const submitSpy = mockSuccessfulFormSubmit();