diff --git a/.changeset/redact-pii-telemetry.md b/.changeset/redact-pii-telemetry.md new file mode 100644 index 000000000..157ccccc9 --- /dev/null +++ b/.changeset/redact-pii-telemetry.md @@ -0,0 +1,7 @@ +--- +'@livekit/agents': minor +'@livekit/agents-plugin-deepgram': minor +'@livekit/agents-plugin-minimax': minor +--- + +Tag content-bearing telemetry with `lk.pii.*` and redact trace exception details when project or session redaction is enabled. Redacted audio uploads now require transcript uploads, Deepgram parse failures retain readable details without logging connection query parameters, and MiniMax task failure payloads no longer appear in exception messages. Dashboards and queries using the previous sensitive trace keys must migrate to their `lk.pii.*` replacements. diff --git a/agents/etc/agents.api.md b/agents/etc/agents.api.md index e766c1bf2..085888701 100644 --- a/agents/etc/agents.api.md +++ b/agents/etc/agents.api.md @@ -1101,12 +1101,12 @@ const ATTR_AMD_SPEECH_DURATION = "lk.amd.speech_duration"; // Warning: (ae-missing-release-tag) "ATTR_AMD_TRANSCRIPT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) -const ATTR_AMD_TRANSCRIPT = "lk.amd.transcript"; +const ATTR_AMD_TRANSCRIPT = "lk.pii.amd.transcript"; // Warning: (ae-missing-release-tag) "ATTR_CHAT_CTX" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) -const ATTR_CHAT_CTX = "lk.chat_ctx"; +const ATTR_CHAT_CTX = "lk.pii.chat_ctx"; // Warning: (ae-missing-release-tag) "ATTR_CLOUD_AGENT_ID" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // @@ -1186,7 +1186,7 @@ const ATTR_EXCEPTION_TYPE = "exception.type"; // Warning: (ae-missing-release-tag) "ATTR_FUNCTION_TOOL_ARGS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) -const ATTR_FUNCTION_TOOL_ARGS = "lk.function_tool.arguments"; +const ATTR_FUNCTION_TOOL_ARGS = "lk.pii.function_tool.arguments"; // Warning: (ae-missing-release-tag) "ATTR_FUNCTION_TOOL_ID" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // @@ -1206,7 +1206,7 @@ const ATTR_FUNCTION_TOOL_NAME = "lk.function_tool.name"; // Warning: (ae-missing-release-tag) "ATTR_FUNCTION_TOOL_OUTPUT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) -const ATTR_FUNCTION_TOOL_OUTPUT = "lk.function_tool.output"; +const ATTR_FUNCTION_TOOL_OUTPUT = "lk.pii.function_tool.output"; // Warning: (ae-missing-release-tag) "ATTR_FUNCTION_TOOLS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // @@ -1266,7 +1266,7 @@ const ATTR_GEN_AI_USAGE_OUTPUT_TOKENS = "gen_ai.usage.output_tokens"; // 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) -const ATTR_INSTRUCTIONS = "lk.instructions"; +const ATTR_INSTRUCTIONS = "lk.pii.instructions"; // Warning: (ae-missing-release-tag) "ATTR_INTERRUPTION_DETECTION_DELAY" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // @@ -1316,7 +1316,7 @@ const ATTR_PARTICIPANT_ID = "lk.participant_id"; // Warning: (ae-missing-release-tag) "ATTR_PARTICIPANT_IDENTITY" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) -const ATTR_PARTICIPANT_IDENTITY = "lk.participant_identity"; +const ATTR_PARTICIPANT_IDENTITY = "lk.pii.participant_identity"; // Warning: (ae-missing-release-tag) "ATTR_PARTICIPANT_KIND" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // @@ -1341,12 +1341,12 @@ const ATTR_REALTIME_MODEL_METRICS = "lk.realtime_model_metrics"; // Warning: (ae-missing-release-tag) "ATTR_RESPONSE_FUNCTION_CALLS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) -const ATTR_RESPONSE_FUNCTION_CALLS = "lk.response.function_calls"; +const ATTR_RESPONSE_FUNCTION_CALLS = "lk.pii.response.function_calls"; // Warning: (ae-missing-release-tag) "ATTR_RESPONSE_TEXT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) -const ATTR_RESPONSE_TEXT = "lk.response.text"; +const ATTR_RESPONSE_TEXT = "lk.pii.response.text"; // Warning: (ae-missing-release-tag) "ATTR_RESPONSE_TTFB" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // @@ -1366,7 +1366,7 @@ const ATTR_RETRY_COUNT = "lk.retry_count"; // Warning: (ae-missing-release-tag) "ATTR_ROOM_NAME" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) -const ATTR_ROOM_NAME = "lk.room_name"; +const ATTR_ROOM_NAME = "lk.pii.room_name"; // Warning: (ae-missing-release-tag) "ATTR_SESSION_OPTIONS" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // @@ -1375,7 +1375,7 @@ const ATTR_SESSION_OPTIONS = "lk.session_options"; // Warning: (ae-missing-release-tag) "ATTR_SPEECH_ID" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // -// @public (undocumented) +// @public const ATTR_SPEECH_ID = "lk.speech_id"; // Warning: (ae-missing-release-tag) "ATTR_SPEECH_INTERRUPTED" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) @@ -1406,7 +1406,7 @@ const ATTR_TRANSCRIPTION_DELAY = "lk.transcription_delay"; // Warning: (ae-missing-release-tag) "ATTR_TTS_INPUT_TEXT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) -const ATTR_TTS_INPUT_TEXT = "lk.input_text"; +const ATTR_TTS_INPUT_TEXT = "lk.pii.input_text"; // Warning: (ae-missing-release-tag) "ATTR_TTS_LABEL" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // @@ -1426,12 +1426,12 @@ const ATTR_TTS_STREAMING = "lk.tts.streaming"; // Warning: (ae-missing-release-tag) "ATTR_USER_INPUT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) -const ATTR_USER_INPUT = "lk.user_input"; +const ATTR_USER_INPUT = "lk.pii.user_input"; // Warning: (ae-missing-release-tag) "ATTR_USER_TRANSCRIPT" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) -const ATTR_USER_TRANSCRIPT = "lk.user_transcript"; +const ATTR_USER_TRANSCRIPT = "lk.pii.user_transcript"; // Warning: (ae-missing-release-tag) "ATTRIBUTE_REDACTION_ENABLED" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // @@ -4485,6 +4485,8 @@ export class JobContext> { _primaryAgentSession?: AgentSession; // (undocumented) get proc(): JobProcess; + // @internal (undocumented) + _redactionEnabled: boolean; // (undocumented) get room(): Room; // (undocumented) @@ -5735,7 +5737,14 @@ class RecognizeSentinel { // Warning: (ae-missing-release-tag) "recordException" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) // // @public (undocumented) -function recordException(span: Span, error: Error): void; +function recordException(span: Span, error: Error, options?: RecordExceptionOptions): void; + +// Warning: (ae-missing-release-tag) "RecordExceptionOptions" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +interface RecordExceptionOptions { + redacted?: boolean; +} // Warning: (ae-internal-missing-underscore) The name "recordingEnabled" should be prefixed with an underscore because the declaration is marked as @internal // @@ -5747,6 +5756,11 @@ export function recordingEnabled(options: Record): boolean; // @public (undocumented) function recordRealtimeMetrics(span: Span, metrics: RealtimeModelMetrics): void; +// Warning: (ae-missing-release-tag) "REDACTED_EXCEPTION_MESSAGE" is part of the package's API, but it is missing a release tag (@alpha, @beta, @public, or @internal) +// +// @public (undocumented) +const REDACTED_EXCEPTION_MESSAGE = "exception details redacted"; + // 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) @@ -7340,8 +7354,10 @@ declare namespace telemetry { SetTracerProviderOptions, SpanProcessorLike, StartSpanOptions, + REDACTED_EXCEPTION_MESSAGE, recordException, - recordRealtimeMetrics + recordRealtimeMetrics, + RecordExceptionOptions } } @@ -8982,7 +8998,7 @@ export const zipFunctionCallsAndOutputs: (event: FunctionToolsExecutedEvent) => // src/voice/amd.ts:309: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:317:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "aclose" // src/voice/amd.ts:507: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:907:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "canEmit" +// src/voice/amd.ts:910:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "canEmit" // src/voice/events.ts:423:3 - (ae-forgotten-export) The symbol "InterruptionDetectionError" needs to be exported by the entry point index.d.ts // src/voice/room_io/_output.ts:178:3 - (ae-unresolved-link) The @link reference could not be resolved: The package "@livekit/agents" does not have an export "segmentTags" // src/voice/testing/run_result.ts:93:5 - (ae-forgotten-export) The symbol "OutputSchema" needs to be exported by the entry point index.d.ts diff --git a/agents/src/inference/avatar.test.ts b/agents/src/inference/avatar.test.ts index 88c99e75d..7a22a450d 100644 --- a/agents/src/inference/avatar.test.ts +++ b/agents/src/inference/avatar.test.ts @@ -404,8 +404,8 @@ it('start warns when the gateway minted a different avatar identity', async () = expect(warn).toHaveBeenCalledWith( expect.objectContaining({ - requestedIdentity: 'lemonslice-inference-avatar', - mintedIdentity: 'gateway-normalized-avatar', + 'lk.pii.requested_identity': 'lemonslice-inference-avatar', + 'lk.pii.minted_identity': 'gateway-normalized-avatar', }), expect.stringContaining('different identity than requested'), ); diff --git a/agents/src/inference/avatar.ts b/agents/src/inference/avatar.ts index d3717ddc9..2b43f63a0 100644 --- a/agents/src/inference/avatar.ts +++ b/agents/src/inference/avatar.ts @@ -275,8 +275,8 @@ export class AvatarSession extends BaseAvatarSession { { provider: this.providerName, sessionId: this._sessionId, - requestedIdentity: this.avatarParticipantIdentity, - mintedIdentity: createResp.avatar_identity, + 'lk.pii.requested_identity': this.avatarParticipantIdentity, + 'lk.pii.minted_identity': createResp.avatar_identity, }, 'avatar gateway minted the worker token for a different identity than requested; the avatar participant will join as the minted identity and this session will not reach it', ); diff --git a/agents/src/inference/interruption/ws_transport.ts b/agents/src/inference/interruption/ws_transport.ts index a1b252660..c419c85a0 100644 --- a/agents/src/inference/interruption/ws_transport.ts +++ b/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; diff --git a/agents/src/inference/stt.ts b/agents/src/inference/stt.ts index d6defd0a9..a47dd81e1 100644 --- a/agents/src/inference/stt.ts +++ b/agents/src/inference/stt.ts @@ -943,7 +943,10 @@ export class SpeechStream 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; diff --git a/agents/src/ipc/job_proc_lazy_main.ts b/agents/src/ipc/job_proc_lazy_main.ts index 5812c168f..24805d837 100644 --- a/agents/src/ipc/job_proc_lazy_main.ts +++ b/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); diff --git a/agents/src/job.test.ts b/agents/src/job.test.ts index 649bae444..23cae684a 100644 --- a/agents/src/job.test.ts +++ b/agents/src/job.test.ts @@ -251,6 +251,12 @@ describe('simulator participant lifecycle', () => { }); describe('JobContext telemetry metadata', () => { + it('includes redaction when the project enables it', () => { + const { ctx } = createJobContextWithRoom({}, { enableRedaction: true }); + + expect(ctx._otelMetadata()).toEqual({ 'lk.redaction.enabled': true }); + }); + it('includes redaction when the recording option enables it', () => { const ctx = createJobContext(); @@ -279,3 +285,43 @@ describe('JobContext telemetry metadata', () => { expect(ctx._otelMetadata()).toEqual({ 'lk.simulation.enabled': true }); }); }); + +describe('JobContext recording redaction', () => { + it('tracks project and session redaction in the resolved state', async () => { + const { ctx: projectRedacted } = createJobContextWithRoom({}, { enableRedaction: true }); + const { ctx: sessionRedacted } = createJobContextWithRoom(); + + expect(projectRedacted._redactionEnabled).toBe(true); + expect(sessionRedacted._redactionEnabled).toBe(false); + + await sessionRedacted.initRecording({ + audio: false, + traces: false, + logs: false, + transcript: false, + redaction: true, + }); + + expect(sessionRedacted._redactionEnabled).toBe(true); + }); + + it.each([ + { source: 'project', projectRedaction: true, sessionRedaction: false }, + { source: 'session', projectRedaction: false, sessionRedaction: true }, + ])( + 'rejects audio without transcripts under $source redaction', + async ({ projectRedaction, sessionRedaction }) => { + const { ctx } = createJobContextWithRoom({}, { enableRedaction: projectRedaction }); + + await expect( + ctx.initRecording({ + audio: true, + traces: false, + logs: false, + transcript: false, + redaction: sessionRedaction, + }), + ).rejects.toThrow('audio upload requires transcript upload when redaction is enabled'); + }, + ); +}); diff --git a/agents/src/job.ts b/agents/src/job.ts index 4ec3680c9..5bd4f2fa3 100644 --- a/agents/src/job.ts +++ b/agents/src/job.ts @@ -142,6 +142,9 @@ export class JobContext> { /** @internal */ _sessionDirectory: string; + /** @internal */ + _redactionEnabled: boolean; + // Lazily built from the job's simulation attributes; undefined when not // under a simulation. #simulationResolved guards the one-time parse. #simulationCtx?: SimulationContext; @@ -174,9 +177,10 @@ export class JobContext> { this.#room.on(RoomEvent.ParticipantDisconnected, this.onParticipantDisconnected); this.#logger = log().child({ jobId: this.#info.job.id, - roomName: this.#info.job.room?.name, + 'lk.pii.room_name': this.#info.job.room?.name, }); this.#inferenceExecutor = inferenceExecutor; + this._redactionEnabled = Boolean(info.job.enableRedaction); // In console mode, recordings land in a local user-visible directory // (mirrors python's AgentsConsole); real jobs use a temp dir. const agentsConsole = AgentsConsole.getInstance(); @@ -385,9 +389,9 @@ export class JobContext> { try { const client = new RoomServiceClient(this.#info.url, this.#info.apiKey, this.#info.apiSecret); await client.deleteRoom(targetRoomName); - this.#logger.info({ roomName: targetRoomName }, 'room deleted'); + this.#logger.info({ 'lk.pii.room_name': targetRoomName }, 'room deleted'); } catch (error) { - this.#logger.warn({ error, roomName: targetRoomName }, 'error while deleting room'); + this.#logger.warn({ error, 'lk.pii.room_name': targetRoomName }, 'error while deleting room'); } } @@ -551,6 +555,12 @@ export class JobContext> { return; } + const redactionEnabled = Boolean(this.job.enableRedaction || options.redaction); + if (redactionEnabled && options.audio && !options.transcript) { + throw new Error('audio upload requires transcript upload when redaction is enabled'); + } + this._redactionEnabled = redactionEnabled; + const url = new URL(this.#info.url); if (!isCloud(url)) { return; @@ -591,7 +601,7 @@ export class JobContext> { // Ignore malformed simulation dispatch metadata, matching Python's parse-failure behavior. } } - if (options?.redaction) { + if (this._redactionEnabled || options?.redaction) { metadata[ATTRIBUTE_REDACTION_ENABLED] = true; } return Object.keys(metadata).length > 0 ? metadata : undefined; diff --git a/agents/src/llm/fallback_adapter.ts b/agents/src/llm/fallback_adapter.ts index fdacac36b..19ca1fa8d 100644 --- a/agents/src/llm/fallback_adapter.ts +++ b/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( diff --git a/agents/src/log.test.ts b/agents/src/log.test.ts index 0acdcd81a..198e6bbd7 100644 --- a/agents/src/log.test.ts +++ b/agents/src/log.test.ts @@ -3,7 +3,12 @@ // SPDX-License-Identifier: Apache-2.0 import { afterEach, describe, expect, it, vi } from 'vitest'; import { enableOtelLogging, initializeLogger, log } from './log.js'; -import { PinoCloudExporter, initPinoCloudExporter } from './telemetry/pino_otel_transport.js'; +import { + PinoCloudExporter, + flushPinoLogs, + initPinoCloudExporter, +} from './telemetry/pino_otel_transport.js'; +import { REDACTED_EXCEPTION_MESSAGE } from './telemetry/utils.js'; const OTEL_ENABLED_KEY = Symbol.for('@livekit/agents:otelEnabled'); @@ -15,6 +20,8 @@ function resetOtelLoggingState() { describe('OTEL logging', () => { afterEach(() => { vi.restoreAllMocks(); + vi.unstubAllEnvs(); + vi.unstubAllGlobals(); resetOtelLoggingState(); }); @@ -39,4 +46,144 @@ 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'); + }); + }); + + it.each([ + ['error', false], + ['error', true], + ['err', false], + ['err', true], + ] as const)( + 'redacts serialized %s details when redaction is %s', + async (exceptionKey, redactionEnabled) => { + vi.stubEnv('LIVEKIT_API_KEY', 'devkey'); + vi.stubEnv('LIVEKIT_API_SECRET', 'secret'); + const fetchMock = vi.fn().mockResolvedValue({ ok: true }); + vi.stubGlobal('fetch', fetchMock); + initializeLogger({ pretty: false, level: 'info' }); + + initPinoCloudExporter({ + cloudHostname: 'example.livekit.cloud', + roomId: 'RM_test', + jobId: 'AJ_test', + metadata: { 'lk.redaction.enabled': redactionEnabled }, + }); + enableOtelLogging(); + + const exception = new Error('secret transcript', { + cause: new Error('secret nested cause'), + }) as Error & { body: unknown }; + exception.body = { transcript: 'secret provider payload' }; + log().error({ [exceptionKey]: exception }, 'provider failed'); + await flushPinoLogs(); + + expect(fetchMock).toHaveBeenCalledOnce(); + const requestBody = fetchMock.mock.calls[0]![1].body as string; + const payload = JSON.parse(requestBody) as { + resourceLogs: Array<{ + scopeLogs: Array<{ + logRecords: Array<{ + attributes: Array<{ key: string; value: { stringValue?: string } }>; + }>; + }>; + }>; + }; + const attributes = payload.resourceLogs[0]!.scopeLogs[0]!.logRecords[0]!.attributes; + const serializedError = attributes.find(({ key }) => key === exceptionKey)?.value.stringValue; + expect(serializedError).toBeDefined(); + const error = JSON.parse(serializedError!) as Record; + + if (redactionEnabled) { + expect(error).toEqual({ + type: 'Error', + message: REDACTED_EXCEPTION_MESSAGE, + }); + expect(requestBody).not.toContain('secret transcript'); + expect(requestBody).not.toContain('secret nested cause'); + expect(requestBody).not.toContain('secret provider payload'); + } else { + expect(error.message).toContain('secret transcript'); + expect(error.message).toContain('secret nested cause'); + expect(error.stack).toContain('secret transcript'); + expect(error.body).toEqual({ transcript: 'secret provider payload' }); + } + }, + ); }); diff --git a/agents/src/telemetry/index.ts b/agents/src/telemetry/index.ts index a69227e26..cb3e25383 100644 --- a/agents/src/telemetry/index.ts +++ b/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'; diff --git a/agents/src/telemetry/otel_http_exporter.test.ts b/agents/src/telemetry/otel_http_exporter.test.ts index c88eb2466..376e2ef64 100644 --- a/agents/src/telemetry/otel_http_exporter.test.ts +++ b/agents/src/telemetry/otel_http_exporter.test.ts @@ -62,4 +62,19 @@ describe('SimpleOTLPHttpLogExporter attribute conversion', () => { expect(payload).toContain('turn-detector-v1'); expect(payload).toContain('ws://gateway'); }); + + it('encodes nested PII markers as OTLP key-value keys', async () => { + const payload = await exportAttributes({ + 'session.options': { + keytermsOptions: { + 'lk.pii.keyterms': ['Acme Corp'], + keytermDetection: { enabled: false }, + }, + }, + }); + + expect(payload).toContain('"key":"session.options"'); + expect(payload).toContain('"key":"lk.pii.keyterms"'); + expect(payload).toContain('"kvlistValue"'); + }); }); diff --git a/agents/src/telemetry/pino_otel_transport.ts b/agents/src/telemetry/pino_otel_transport.ts index 98d56b1fb..f76178162 100644 --- a/agents/src/telemetry/pino_otel_transport.ts +++ b/agents/src/telemetry/pino_otel_transport.ts @@ -10,6 +10,8 @@ */ import { SeverityNumber } from '@opentelemetry/api-logs'; import { AccessToken } from 'livekit-server-sdk'; +import { ATTRIBUTE_REDACTION_ENABLED } from '../types.js'; +import { REDACTED_EXCEPTION_MESSAGE } from './redaction.js'; import { fetchWithUploadGate, uploadGate } from './upload_gate.js'; export interface PinoLogObject { @@ -71,6 +73,22 @@ function convertValue(value: unknown): unknown { return { stringValue: String(value) }; } +function redactSerializedException(value: unknown): unknown { + if (value === null || typeof value !== 'object' || Array.isArray(value)) { + return value; + } + + const exception = value as Record; + if (typeof exception.type !== 'string' || typeof exception.message !== 'string') { + return value; + } + + return { + type: exception.type, + message: REDACTED_EXCEPTION_MESSAGE, + }; +} + /** * Standalone Pino log exporter for LiveKit Cloud. * @@ -125,6 +143,7 @@ export class PinoCloudExporter { private convertToOtlpRecord(logObj: PinoLogObject): any { const { severityNumber, severityText } = mapPinoLevelToSeverity(logObj.level); + const redactionEnabled = this.config.metadata?.[ATTRIBUTE_REDACTION_ENABLED] === true; const attributes: any[] = [ { key: 'room_id', value: { stringValue: this.config.roomId } }, @@ -145,7 +164,11 @@ export class PinoCloudExporter { for (const [key, value] of Object.entries(logObj)) { if (!EXCLUDE_FIELDS.has(key)) { - attributes.push({ key, value: convertValue(value) }); + const attributeValue = + redactionEnabled && (key === 'error' || key === 'err') + ? redactSerializedException(value) + : value; + attributes.push({ key, value: convertValue(attributeValue) }); } } diff --git a/agents/src/telemetry/redaction.ts b/agents/src/telemetry/redaction.ts new file mode 100644 index 000000000..66de58a7c --- /dev/null +++ b/agents/src/telemetry/redaction.ts @@ -0,0 +1,5 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 + +export const REDACTED_EXCEPTION_MESSAGE = 'exception details redacted'; diff --git a/agents/src/telemetry/trace_types.test.ts b/agents/src/telemetry/trace_types.test.ts new file mode 100644 index 000000000..b3053132c --- /dev/null +++ b/agents/src/telemetry/trace_types.test.ts @@ -0,0 +1,237 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { existsSync, readFileSync, readdirSync } from 'node:fs'; +import { join, relative } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import ts from 'typescript'; +import { describe, expect, it } from 'vitest'; +import * as traceTypes from './trace_types.js'; + +const PII_SEGMENT_RE = /(^|\.)pii(\.|$)/i; +const LOG_METHODS = new Set(['child', 'debug', 'error', 'fatal', 'info', 'trace', 'warn']); +const SENSITIVE_LOG_FIELDS = new Set([ + 'arguments', + 'avatarIdentity', + 'avatarParticipantIdentity', + 'callerIdentity', + 'callerRoom', + 'chatCtx', + 'chat_ctx', + 'destinationIdentity', + 'expectedIdentities', + 'humanAgentIdentity', + 'humanAgentRoomName', + 'identity', + 'interimTranscript', + 'interim_transcript', + 'mintedIdentity', + 'participant', + 'participantIdentity', + 'participantValue', + 'participant_identity', + 'publishOnBehalf', + 'progressMessage', + 'progress_message', + 'rawArguments', + 'raw_arguments', + 'repaired', + 'requestedIdentity', + 'room', + 'roomName', + 'room_name', + 'senderIdentity', + 'transcript', +]); + +const REPO_ROOT = fileURLToPath(new URL('../../../', import.meta.url)); + +function sourceFiles(root: string): string[] { + return readdirSync(root, { withFileTypes: true }).flatMap((entry) => { + const path = join(root, entry.name); + if (entry.isDirectory()) { + return sourceFiles(path); + } + if (!entry.name.endsWith('.ts') || entry.name.endsWith('.test.ts')) { + return []; + } + return [path]; + }); +} + +function propertyName(name: ts.PropertyName): string | undefined { + if (ts.isIdentifier(name) || ts.isStringLiteralLike(name)) { + return name.text; + } + return undefined; +} + +function sensitiveLogFieldsWithoutPii(): string[] { + const pluginsRoot = join(REPO_ROOT, 'plugins'); + const roots = [ + join(REPO_ROOT, 'agents', 'src'), + ...readdirSync(pluginsRoot, { withFileTypes: true }) + .filter((entry) => entry.isDirectory()) + .map((entry) => join(pluginsRoot, entry.name, 'src')) + .filter(existsSync), + ]; + const untagged: string[] = []; + + for (const path of roots.flatMap(sourceFiles)) { + const source = readFileSync(path, 'utf8'); + const sourceFile = ts.createSourceFile(path, source, ts.ScriptTarget.Latest, true); + + function visit(node: ts.Node): void { + if ( + ts.isCallExpression(node) && + ts.isPropertyAccessExpression(node.expression) && + LOG_METHODS.has(node.expression.name.text) + ) { + for (const argument of node.arguments) { + if (!ts.isObjectLiteralExpression(argument)) continue; + for (const property of argument.properties) { + if (!ts.isPropertyAssignment(property) && !ts.isShorthandPropertyAssignment(property)) { + continue; + } + const key = propertyName(property.name); + const leaf = key?.split('.').at(-1); + if (!key || !leaf || !SENSITIVE_LOG_FIELDS.has(leaf) || PII_SEGMENT_RE.test(key)) { + continue; + } + const line = sourceFile.getLineAndCharacterOfPosition( + property.getStart(sourceFile), + ).line; + untagged.push(`${relative(REPO_ROOT, path)}:${line + 1}: ${key}`); + } + } + } + ts.forEachChild(node, visit); + } + + visit(sourceFile); + } + + return untagged.sort(); +} + +// 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_kind', + 'lk.job_id', + 'lk.agent_name', + 'lk.cloud_agent_id', + 'lk.deployment_id', + '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 { + 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([]); + }); + + it('tags sensitive literal structured-log fields as PII', () => { + expect(sensitiveLogFieldsWithoutPii()).toEqual([]); + }); +}); diff --git a/agents/src/telemetry/trace_types.ts b/agents/src/telemetry/trace_types.ts index a31b43d05..cb22e4b4a 100644 --- a/agents/src/telemetry/trace_types.ts +++ b/agents/src/telemetry/trace_types.ts @@ -2,6 +2,17 @@ // // 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.`). 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. For structured attributes, the collector applies the marker recursively to keys in + * nested OTLP key-value lists and arrays. + */ + // LiveKit custom attributes export const ATTR_SPEECH_ID = 'lk.speech_id'; export const ATTR_AGENT_LABEL = 'lk.agent_label'; @@ -20,7 +31,7 @@ export const ATTR_RETRY_COUNT = 'lk.retry_count'; export const ATTR_PROVIDER_REQUEST_IDS = 'lk.provider_request_ids'; export const ATTR_PARTICIPANT_ID = 'lk.participant_id'; -export const ATTR_PARTICIPANT_IDENTITY = 'lk.participant_identity'; +export const ATTR_PARTICIPANT_IDENTITY = 'lk.pii.participant_identity'; export const ATTR_PARTICIPANT_KIND = 'lk.participant_kind'; // session start @@ -28,35 +39,35 @@ export const ATTR_JOB_ID = 'lk.job_id'; export const ATTR_AGENT_NAME = 'lk.agent_name'; export const ATTR_CLOUD_AGENT_ID = 'lk.cloud_agent_id'; export const ATTR_DEPLOYMENT_ID = 'lk.deployment_id'; -export const ATTR_ROOM_NAME = 'lk.room_name'; +export const ATTR_ROOM_NAME = 'lk.pii.room_name'; 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. */ @@ -74,7 +85,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'; @@ -88,7 +99,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'; diff --git a/agents/src/telemetry/traces.test.ts b/agents/src/telemetry/traces.test.ts index 3bcd644a2..5e888dbaf 100644 --- a/agents/src/telemetry/traces.test.ts +++ b/agents/src/telemetry/traces.test.ts @@ -440,6 +440,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 = { ...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; + const serializedKeytermsOptions = serializedOptions.keytermsOptions as Record; + 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(); diff --git a/agents/src/telemetry/traces.ts b/agents/src/telemetry/traces.ts index 0bc76bb5b..57357df39 100644 --- a/agents/src/telemetry/traces.ts +++ b/agents/src/telemetry/traces.ts @@ -737,6 +737,28 @@ function toRFC3339(valueMs: number | Date): string { return truncated.toISOString(); } +const SESSION_OPTION_KEY_ALIASES: Record = { + keyterms: 'lk.pii.keyterms', +}; + +function serializeSessionOptions(options: SessionReport['options']): Record { + const serialize = (value: Record): Record => + Object.fromEntries( + Object.entries(value).map(([key, nestedValue]) => [ + SESSION_OPTION_KEY_ALIASES[key] ?? key, + nestedValue !== null && + typeof nestedValue === 'object' && + !Array.isArray(nestedValue) && + (Object.getPrototypeOf(nestedValue) === Object.prototype || + Object.getPrototypeOf(nestedValue) === null) + ? serialize(nestedValue as Record) + : nestedValue, + ]), + ); + + return serialize(options as unknown as Record); +} + /** * Upload session report to LiveKit Cloud observability. * @param options - Configuration with agentName, cloudHostname, and report @@ -791,7 +813,7 @@ export async function uploadSessionReport(options: { timestampMs: report.startedAt || report.timestamp || 0, attributes: { ...commonAttrs, - 'session.options': report.options || {}, + 'session.options': serializeSessionOptions(report.options), 'session.report_timestamp': report.timestamp, agent_name: agentName, usage, diff --git a/agents/src/telemetry/utils.test.ts b/agents/src/telemetry/utils.test.ts new file mode 100644 index 000000000..3c57583d9 --- /dev/null +++ b/agents/src/telemetry/utils.test.ts @@ -0,0 +1,117 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +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 * as traceTypes from './trace_types.js'; +import { + REDACTED_EXCEPTION_MESSAGE, + type RecordExceptionOptions, + recordException, +} from './utils.js'; + +function fakeSpan() { + return { + addEvent: vi.fn(), + recordException: vi.fn(), + setAttributes: vi.fn(), + setStatus: vi.fn(), + }; +} + +function captureException( + span: ReturnType, + options?: RecordExceptionOptions, +): void { + const error = new Error('secret transcript'); + recordException(span as unknown as Span, error, options); +} + +describe('recordException', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it('preserves exception details when redaction is explicitly disabled', () => { + const span = fakeSpan(); + + captureException(span, { redacted: false }); + + expect(span.recordException).toHaveBeenCalledOnce(); + expect(span.setStatus).toHaveBeenCalledWith({ + code: SpanStatusCode.ERROR, + message: 'secret transcript', + }); + expect(span.setAttributes).toHaveBeenCalledWith({ + [traceTypes.ATTR_EXCEPTION_TYPE]: 'Error', + [traceTypes.ATTR_EXCEPTION_MESSAGE]: 'secret transcript', + [traceTypes.ATTR_EXCEPTION_TRACE]: expect.stringContaining('secret transcript'), + }); + }); + + it('omits exception details when redaction is explicitly enabled', () => { + const span = fakeSpan(); + + captureException(span, { redacted: true }); + + const attrs = { + [traceTypes.ATTR_EXCEPTION_TYPE]: 'Error', + [traceTypes.ATTR_EXCEPTION_MESSAGE]: REDACTED_EXCEPTION_MESSAGE, + }; + expect(span.recordException).not.toHaveBeenCalled(); + expect(span.addEvent).toHaveBeenCalledWith('exception', attrs); + expect(span.setStatus).toHaveBeenCalledWith({ + code: SpanStatusCode.ERROR, + message: REDACTED_EXCEPTION_MESSAGE, + }); + expect(span.setAttributes).toHaveBeenCalledWith(attrs); + expect( + JSON.stringify([ + span.addEvent.mock.calls, + span.setStatus.mock.calls, + span.setAttributes.mock.calls, + ]), + ).not.toContain('secret transcript'); + }); + + it('uses the resolved session redaction setting by default', () => { + const span = fakeSpan(); + const context = { + job: { enableRedaction: false }, + _redactionEnabled: true, + } as unknown as JobContext; + + runWithJobContext(context, () => captureException(span)); + + expect(span.recordException).not.toHaveBeenCalled(); + expect(span.setAttributes).toHaveBeenCalledWith({ + [traceTypes.ATTR_EXCEPTION_TYPE]: 'Error', + [traceTypes.ATTR_EXCEPTION_MESSAGE]: REDACTED_EXCEPTION_MESSAGE, + }); + }); + + it('preserves exception details by default without a job context', () => { + const span = fakeSpan(); + + captureException(span); + + expect(span.recordException).toHaveBeenCalledOnce(); + }); + + it('allows an explicit false override for resolved redaction', () => { + const span = fakeSpan(); + const context = { + _redactionEnabled: true, + } as unknown as JobContext; + + runWithJobContext(context, () => captureException(span, { redacted: false })); + + expect(span.recordException).toHaveBeenCalledOnce(); + expect(span.setStatus).toHaveBeenCalledWith({ + code: SpanStatusCode.ERROR, + message: 'secret transcript', + }); + }); +}); diff --git a/agents/src/telemetry/utils.ts b/agents/src/telemetry/utils.ts index 4eba4f043..25bf54387 100644 --- a/agents/src/telemetry/utils.ts +++ b/agents/src/telemetry/utils.ts @@ -2,11 +2,42 @@ // // SPDX-License-Identifier: Apache-2.0 import { type Span, SpanStatusCode, context as otelContext, trace } from '@opentelemetry/api'; +import { getJobContext } from '../job.js'; import type { RealtimeModelMetrics } from '../metrics/base.js'; +import { REDACTED_EXCEPTION_MESSAGE } from './redaction.js'; import * as traceTypes from './trace_types.js'; import { tracer } from './traces.js'; -export function recordException(span: Span, error: Error): void { +export { REDACTED_EXCEPTION_MESSAGE } from './redaction.js'; + +export interface RecordExceptionOptions { + /** + * Whether to omit exception messages and stack traces from telemetry. Defaults to the resolved + * redaction setting for the current session. + */ + redacted?: boolean; +} + +export function recordException( + span: Span, + error: Error, + options: RecordExceptionOptions = {}, +): void { + const redacted = options.redacted ?? getJobContext(false)?._redactionEnabled ?? false; + if (redacted) { + const attrs = { + [traceTypes.ATTR_EXCEPTION_TYPE]: error.constructor.name, + [traceTypes.ATTR_EXCEPTION_MESSAGE]: REDACTED_EXCEPTION_MESSAGE, + }; + span.addEvent('exception', attrs); + span.setStatus({ + code: SpanStatusCode.ERROR, + message: REDACTED_EXCEPTION_MESSAGE, + }); + span.setAttributes(attrs); + return; + } + span.recordException(error); span.setStatus({ code: SpanStatusCode.ERROR, diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index e3e48e7e9..2b55a4704 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -1858,7 +1858,7 @@ export class AgentActivity implements RecognitionHooks { this.logger.info( { - newTranscript: info.newTranscript, + 'lk.pii.new_transcript': info.newTranscript, transcriptConfidence: info.transcriptConfidence, }, 'starting preemptive generation', @@ -2047,7 +2047,7 @@ export class AgentActivity implements RecognitionHooks { if (this.schedulingPaused || this.newTurnsBlocked) { this.cancelPreemptiveGeneration(); this.logger.warn( - { user_input: info.newTranscript }, + { 'lk.pii.user_input': info.newTranscript }, 'skipping user input, speech scheduling is paused', ); if (this.agentSession?._closing) { @@ -2596,7 +2596,7 @@ export class AgentActivity implements RecognitionHooks { if (currentSpeech) { if (!currentSpeech.allowInterruptions) { this.logger.warn( - { user_input: info.newTranscript }, + { 'lk.pii.user_input': info.newTranscript }, 'skipping user input, current speech generation cannot be interrupted', ); return; @@ -2621,7 +2621,7 @@ export class AgentActivity implements RecognitionHooks { if (this.schedulingPaused || this.newTurnsBlocked) { this.logger.warn( - { user_input: info.newTranscript }, + { 'lk.pii.user_input': info.newTranscript }, 'skipping onUserTurnCompleted, speech scheduling is paused', ); if (this.agentSession._closing) { @@ -2657,7 +2657,7 @@ export class AgentActivity implements RecognitionHooks { if (this.schedulingPaused || this.newTurnsBlocked) { this.logger.warn( - { user_input: info.newTranscript }, + { 'lk.pii.user_input': info.newTranscript }, 'skipping reply to user input, speech scheduling is paused', ); if (userMessage && this.agentSession._closing) { @@ -3462,7 +3462,7 @@ export class AgentActivity implements RecognitionHooks { } this.logger.info( - { speech_id: speechHandle.id, message: forwardedText }, + { speech_id: speechHandle.id, 'lk.pii.message': forwardedText }, 'playout completed with interrupt', ); if (speechHandle._hasGenerations) { @@ -3493,7 +3493,7 @@ export class AgentActivity implements RecognitionHooks { this.agentSession._conversationItemAdded(message); span.setAttribute(traceTypes.ATTR_RESPONSE_TEXT, forwardedText); this.logger.info( - { speech_id: speechHandle.id, message: forwardedText }, + { speech_id: speechHandle.id, 'lk.pii.message': forwardedText }, 'playout completed without interruption', ); } @@ -3990,7 +3990,10 @@ export class AgentActivity implements RecognitionHooks { const { done, value } = await reader.read(); if (done) break; - this.logger.debug({ tool_call: value }, 'received tool call from the realtime API'); + this.logger.debug( + { 'lk.pii.tool_call': value }, + 'received tool call from the realtime API', + ); toolCalls.push(value); } } finally { @@ -4358,8 +4361,8 @@ export class AgentActivity implements RecognitionHooks { { speechId: speechHandle.id, name: sanitizedOut.toolCall?.name, - args: sanitizedOut.toolCall.args, - output: sanitizedOut.toolCallOutput?.output, + 'lk.pii.arguments': sanitizedOut.toolCall.args, + 'lk.pii.output': sanitizedOut.toolCallOutput?.output, isError: sanitizedOut.toolCallOutput?.isError, }, 'Tool call execution finished', diff --git a/agents/src/voice/amd.ts b/agents/src/voice/amd.ts index 572ff9522..40e7aabf7 100644 --- a/agents/src/voice/amd.ts +++ b/agents/src/voice/amd.ts @@ -818,7 +818,7 @@ export class AMD extends (EventEmitter as new () => TypedEmitter) isMachine: result.isMachine, speechDurationMs: result.speechDurationMs, delayMs: result.delayMs, - transcript: result.transcript, + 'lk.pii.transcript': result.transcript, }, 'amd prediction', ); @@ -869,7 +869,10 @@ export class AMD extends (EventEmitter as new () => TypedEmitter) return false; } this._log.debug( - { category: this.verdictResult.category, transcript: info.newTranscript }, + { + category: this.verdictResult.category, + 'lk.pii.transcript': info.newTranscript, + }, 'skipping auto reply: AMD already returned a machine verdict', ); return true; diff --git a/agents/src/voice/audio_recognition.ts b/agents/src/voice/audio_recognition.ts index 501129488..3b049f949 100644 --- a/agents/src/voice/audio_recognition.ts +++ b/agents/src/voice/audio_recognition.ts @@ -64,7 +64,7 @@ import { setParticipantSpanAttributes, } from './utils.js'; -// Maximum number of chat items included in the `lk.chat_ctx` attribute of the +// Maximum number of chat items included in the `lk.pii.chat_ctx` attribute of the // `eou_detection` span (mirrors Python's `_EOU_MAX_HISTORY_TURNS`). const EOU_MAX_HISTORY_TURNS = 6; const MIN_LANGUAGE_DETECTION_LENGTH = 5; @@ -1170,7 +1170,7 @@ export class AudioRecognition { this.logger.debug( { - user_transcript: transcript, + 'lk.pii.user_transcript': transcript, language: this.lastLanguage, }, 'received user transcript', @@ -1194,7 +1194,7 @@ export class AudioRecognition { if (this.vadBaseTurnDetection || this.userTurnCommitted) { if (transcriptChanged) { this.logger.debug( - { transcript: this.audioTranscript }, + { 'lk.pii.transcript': this.audioTranscript }, 'triggering preemptive generation (FINAL_TRANSCRIPT)', ); this.hooks.onPreemptiveGeneration({ @@ -1232,7 +1232,7 @@ export class AudioRecognition { this.logger.debug( { - user_transcript: preflightTranscript, + 'lk.pii.user_transcript': preflightTranscript, language: this.lastLanguage, }, 'received user preflight transcript', @@ -1254,7 +1254,7 @@ export class AudioRecognition { const confidenceVals = [...this.finalTranscriptConfidence, preflightConfidence]; this.logger.debug( { - transcript: + 'lk.pii.transcript': this.audioPreflightTranscript.length > 100 ? this.audioPreflightTranscript.slice(0, 100) + '...' : this.audioPreflightTranscript, @@ -1272,7 +1272,10 @@ export class AudioRecognition { } break; case SpeechEventType.INTERIM_TRANSCRIPT: - this.logger.debug({ transcript: ev.alternatives?.[0]?.text }, 'interim transcript'); + this.logger.debug( + { 'lk.pii.transcript': ev.alternatives?.[0]?.text }, + 'interim transcript', + ); this.hooks.onInterimTranscript( ev, this.hasUserVad || this.turnDetectionMode === 'stt' ? this.speaking : undefined, @@ -1420,7 +1423,7 @@ export class AudioRecognition { this.logger.debug( { stt: this.stt, - audioTranscript: this.audioTranscript, + 'lk.pii.audio_transcript': this.audioTranscript, turnDetectionMode: this.turnDetectionMode, }, 'running EOU detection', @@ -1699,7 +1702,7 @@ export class AudioRecognition { return; } - this.logger.debug({ transcript: this.audioTranscript }, 'end of user turn'); + this.logger.debug({ 'lk.pii.transcript': this.audioTranscript }, 'end of user turn'); const confidenceAvg = this.finalTranscriptConfidence.length > 0 diff --git a/agents/src/voice/audio_recognition_span.test.ts b/agents/src/voice/audio_recognition_span.test.ts index 63356125a..c18974ae2 100644 --- a/agents/src/voice/audio_recognition_span.test.ts +++ b/agents/src/voice/audio_recognition_span.test.ts @@ -191,13 +191,13 @@ describe('AudioRecognition user_turn span', () => { // creation-time attributes expect(userTurn.attributes['lk.participant_id']).toBe('p1'); - expect(userTurn.attributes['lk.participant_identity']).toBe('bob'); + expect(userTurn.attributes['lk.pii.participant_identity']).toBe('bob'); expect(userTurn.attributes['lk.participant_kind']).toBe('AGENT'); expect(userTurn.attributes['gen_ai.request.model']).toBe('deepgram-nova2'); expect(userTurn.attributes['gen_ai.provider.name']).toBe('deepgram'); // end-of-turn attributes - expect(userTurn.attributes['lk.user_transcript']).toContain('hello'); + expect(userTurn.attributes['lk.pii.user_transcript']).toContain('hello'); expect(userTurn.attributes['lk.transcript_confidence']).toBeGreaterThan(0); }); @@ -275,7 +275,7 @@ describe('AudioRecognition user_turn span', () => { } // eslint-disable-next-line @typescript-eslint/no-explicit-any - const items = JSON.parse(String(eou.attributes['lk.chat_ctx'])).items as any[]; + const items = JSON.parse(String(eou.attributes['lk.pii.chat_ctx'])).items as any[]; expect(items.length).toBeLessThanOrEqual(6); // only non-empty, non-system messages survive; the pending user transcript is last expect(items.every((i) => i.type === 'message')).toBe(true); diff --git a/agents/src/voice/avatar/avatar_session.test.ts b/agents/src/voice/avatar/avatar_session.test.ts index a6c7512b8..b50fbeedc 100644 --- a/agents/src/voice/avatar/avatar_session.test.ts +++ b/agents/src/voice/avatar/avatar_session.test.ts @@ -224,7 +224,7 @@ describe('AvatarSession base', () => { expect(removeParticipantMock).toHaveBeenCalledWith('test-room', 'avatar-identity'); expect(debug).toHaveBeenCalledWith( - { identity: 'avatar-identity' }, + { 'lk.pii.avatar_identity': 'avatar-identity' }, 'avatar participant not in room, skipping removal', ); expect(warn).not.toHaveBeenCalled(); @@ -251,7 +251,7 @@ describe('AvatarSession base', () => { expect(removeParticipantMock).toHaveBeenCalledWith('test-room', 'avatar-identity'); expect(warn).toHaveBeenCalledWith( - { error, identity: 'avatar-identity' }, + { error, 'lk.pii.avatar_identity': 'avatar-identity' }, 'failed to remove avatar participant', ); expect(debug).not.toHaveBeenCalledWith( diff --git a/agents/src/voice/avatar/avatar_session.ts b/agents/src/voice/avatar/avatar_session.ts index 13c76ecf8..017dc873e 100644 --- a/agents/src/voice/avatar/avatar_session.ts +++ b/agents/src/voice/avatar/avatar_session.ts @@ -142,12 +142,12 @@ export class AvatarSession extends (EventEmitter as new () => TypedEmitter { - this.logger.debug({ participant: participant.identity }, 'onTrackSubscribed in _input'); + this.logger.debug( + { 'lk.pii.participant_identity': participant.identity }, + 'onTrackSubscribed in _input', + ); if ( this.participantIdentity !== participant.identity || publication.source !== TrackSource.SOURCE_MICROPHONE || diff --git a/agents/src/voice/room_io/room_io.ts b/agents/src/voice/room_io/room_io.ts index 34f9ce0b3..4f1c6be9b 100644 --- a/agents/src/voice/room_io/room_io.ts +++ b/agents/src/voice/room_io/room_io.ts @@ -276,7 +276,7 @@ export class RoomIO { ) { this.logger.info( { - participant: participant.identity, + 'lk.pii.participant_identity': participant.identity, reason: DisconnectReason[participant.disconnectReason], }, 'closing agent session due to participant disconnect ' + @@ -317,7 +317,7 @@ export class RoomIO { return; } this.logger.info( - { room: this.room.name }, + { 'lk.pii.room_name': this.room.name }, 'deleting room on agent session close ' + '(disable via `RoomInputOptions.deleteRoomOnClose=false`)', ); @@ -472,7 +472,10 @@ export class RoomIO { /** Switch to a different participant */ setParticipant(participantIdentity: string | null) { - this.logger.debug({ participantIdentity }, 'setting participant'); + this.logger.debug( + { 'lk.pii.participant_identity': participantIdentity }, + 'setting participant', + ); if (participantIdentity === null) { this.unsetParticipant(); return; @@ -643,7 +646,10 @@ export class RoomIO { if (!(error instanceof IdleTimeoutError)) { throw error; } - this.logger.warn({ room: this.room.name }, 'automatic room deletion timed out'); + this.logger.warn( + { 'lk.pii.room_name': this.room.name }, + 'automatic room deletion timed out', + ); } } } diff --git a/agents/src/worker.ts b/agents/src/worker.ts index 9f6020176..020bf9ee9 100644 --- a/agents/src/worker.ts +++ b/agents/src/worker.ts @@ -651,7 +651,10 @@ export class AgentServer { wsData = `${wsData.slice(0, 128)}...(+${wsData.length - 128} more)`; } const type = typeof event.data; - this.#logger.warn({ type, ws_data: wsData }, `unexpected message type: ${type}`); + this.#logger.warn( + { type, 'lk.pii.ws_data': wsData }, + 'received unexpected worker message type', + ); return; } diff --git a/agents/src/workflows/warm_transfer.ts b/agents/src/workflows/warm_transfer.ts index fbf870dfa..74b0e9234 100644 --- a/agents/src/workflows/warm_transfer.ts +++ b/agents/src/workflows/warm_transfer.ts @@ -279,7 +279,7 @@ export function createWarmTransferTask({ const cancelForCallerHangup = (participantIdentity?: string): void => { if (task.done) return; logger.info( - { participantIdentity }, + { 'lk.pii.participant_identity': participantIdentity }, 'caller hung up before the transfer completed, cancelling transfer', ); callerHangupFut.resolve(); @@ -319,7 +319,7 @@ export function createWarmTransferTask({ } logger.info( - { participantIdentity: participant.identity }, + { 'lk.pii.participant_identity': participant.identity }, 'participant disconnected from caller room, closing', ); @@ -362,7 +362,10 @@ export function createWarmTransferTask({ humanAgentRoom.off(RoomEvent.Disconnected, onHumanAgentRoomClose); logger.debug( - { humanAgentIdentity, callerRoom: callerRoom.name }, + { + 'lk.pii.participant_identity': humanAgentIdentity, + 'lk.pii.room_name': callerRoom.name, + }, 'moving human agent to caller room', ); @@ -415,7 +418,10 @@ export function createWarmTransferTask({ canSubscribe: true, } as VideoGrant); - logger.debug({ wsUrl: ctx.info.url, humanAgentRoomName }, 'connecting to human agent room'); + logger.debug( + { wsUrl: ctx.info.url, 'lk.pii.room_name': humanAgentRoomName }, + 'connecting to human agent room', + ); const jwt = await token.toJwt(); room.on(RoomEvent.Disconnected, onHumanAgentRoomClose); diff --git a/examples/src/frontdesk/calendar_api.test.ts b/examples/src/frontdesk/calendar_api.test.ts index 4e7a08fd2..a1585c0a2 100644 --- a/examples/src/frontdesk/calendar_api.test.ts +++ b/examples/src/frontdesk/calendar_api.test.ts @@ -10,6 +10,11 @@ import { getUniqueHash, } from './calendar_api.js'; +const infoMock = vi.hoisted(() => vi.fn()); +vi.mock('@livekit/agents', () => ({ + log: () => ({ child: () => ({ info: infoMock }) }), +})); + describe('Calendar API', () => { describe('createAvailableSlot', () => { it('should create a valid AvailableSlot', () => { @@ -167,6 +172,7 @@ describe('Calendar API', () => { let calendar: CalComCalendar; beforeEach(() => { + infoMock.mockClear(); calendar = new CalComCalendar({ apiKey: 'test-api-key', timezone: 'America/New_York', @@ -197,15 +203,14 @@ describe('Calendar API', () => { }), } as Response); - const consoleSpy = vi.spyOn(console, 'info').mockImplementation(() => {}); - await calendar.initialize(); expect(mockFetch).toHaveBeenCalledTimes(2); - expect(consoleSpy).toHaveBeenCalledWith('[cal.com] using cal.com username: testuser'); - expect(consoleSpy).toHaveBeenCalledWith('[cal.com] event type id: 123'); - - consoleSpy.mockRestore(); + expect(infoMock).toHaveBeenCalledWith( + { 'lk.pii.username': 'testuser' }, + 'using cal.com username', + ); + expect(infoMock).toHaveBeenCalledWith({ eventTypeId: 123 }, 'cal.com event type found'); }); it('should create new event type when not exists', async () => { @@ -232,18 +237,18 @@ describe('Calendar API', () => { }), } as Response); - const consoleSpy = vi.spyOn(console, 'info').mockImplementation(() => {}); - await calendar.initialize(); expect(mockFetch).toHaveBeenCalledTimes(3); - expect(consoleSpy).toHaveBeenCalledWith('[cal.com] using cal.com username: testuser'); - expect(consoleSpy).toHaveBeenCalledWith( - '[cal.com] successfully added livekit-front-desk event type', + expect(infoMock).toHaveBeenCalledWith( + { 'lk.pii.username': 'testuser' }, + 'using cal.com username', ); - expect(consoleSpy).toHaveBeenCalledWith('[cal.com] event type id: 456'); - - consoleSpy.mockRestore(); + expect(infoMock).toHaveBeenCalledWith( + { eventType: 'livekit-front-desk' }, + 'successfully added cal.com event type', + ); + expect(infoMock).toHaveBeenCalledWith({ eventTypeId: 456 }, 'cal.com event type found'); }); it('should handle API errors', async () => { diff --git a/examples/src/frontdesk/calendar_api.ts b/examples/src/frontdesk/calendar_api.ts index cd6a8a477..b2ce54bd5 100644 --- a/examples/src/frontdesk/calendar_api.ts +++ b/examples/src/frontdesk/calendar_api.ts @@ -1,6 +1,7 @@ // SPDX-FileCopyrightText: 2025 LiveKit, Inc. // // SPDX-License-Identifier: Apache-2.0 +import { log } from '@livekit/agents'; import { createHash } from 'crypto'; export class SlotUnavailableError extends Error { @@ -134,14 +135,11 @@ export class CalComCalendar implements Calendar { private tz: string; private _apiKey: string; private _lkEventId?: number; - private _logger: { info: (message: string) => void }; + private _logger = log().child({ provider: 'cal.com' }); constructor(options: { apiKey: string; timezone: string }) { this.tz = options.timezone; this._apiKey = options.apiKey; - this._logger = { - info: (message: string) => console.info(`[cal.com] ${message}`), - }; } async initialize(): Promise { @@ -155,7 +153,7 @@ export class CalComCalendar implements Calendar { const meData = await meResponse.json(); const username = meData.data.username; - this._logger.info(`using cal.com username: ${username}`); + this._logger.info({ 'lk.pii.username': username }, 'using cal.com username'); const params = new URLSearchParams({ username }); const eventTypesResponse = await fetch(`${_BASE_URL}event-types/?${params}`, { @@ -195,12 +193,15 @@ export class CalComCalendar implements Calendar { ); } - this._logger.info(`successfully added ${_CAL_COM_EVENT_TYPE} event type`); + this._logger.info( + { eventType: _CAL_COM_EVENT_TYPE }, + 'successfully added cal.com event type', + ); const createData = await createResponse.json(); this._lkEventId = createData.data.id; } - this._logger.info(`event type id: ${this._lkEventId}`); + this._logger.info({ eventTypeId: this._lkEventId }, 'cal.com event type found'); } async scheduleAppointment(options: { startTime: Date; attendeeEmail: string }): Promise { diff --git a/examples/src/instructions_per_modality.ts b/examples/src/instructions_per_modality.ts index f2ab6b910..b0744b591 100644 --- a/examples/src/instructions_per_modality.ts +++ b/examples/src/instructions_per_modality.ts @@ -64,7 +64,10 @@ class SchedulingAgent extends voice.Agent { time: z.string().describe('The time of the appointment in the format HH:MM'), }), execute: async ({ date, time }) => { - log().info(`booking appointment for ${date} at ${time}`); + log().info( + { 'lk.pii.appointment_date': date, 'lk.pii.appointment_time': time }, + 'booking appointment', + ); return `Appointment booked for ${date} at ${time}`; }, }), diff --git a/examples/src/inworld_tts.ts b/examples/src/inworld_tts.ts index ba9f6a888..112972d5f 100644 --- a/examples/src/inworld_tts.ts +++ b/examples/src/inworld_tts.ts @@ -77,13 +77,27 @@ export default defineAgent({ if (data.wordAlignment) { const { words, starts, ends } = data.wordAlignment; for (let i = 0; i < words.length; i++) { - console.log(`[Inworld TTS] Word: "${words[i]}", Start: ${starts[i]}, End: ${ends[i]}`); + logger.info( + { + 'lk.pii.word': words[i], + startTime: starts[i], + endTime: ends[i], + }, + 'Inworld TTS word alignment', + ); } } if (data.characterAlignment) { const { chars, starts, ends } = data.characterAlignment; for (let i = 0; i < chars.length; i++) { - console.log(`[Inworld TTS] Char: "${chars[i]}", Start: ${starts[i]}, End: ${ends[i]}`); + logger.info( + { + 'lk.pii.character': chars[i], + startTime: starts[i], + endTime: ends[i], + }, + 'Inworld TTS character alignment', + ); } } }); diff --git a/examples/src/liveavatar_avatar.ts b/examples/src/liveavatar_avatar.ts index bc428142d..f431b8c33 100644 --- a/examples/src/liveavatar_avatar.ts +++ b/examples/src/liveavatar_avatar.ts @@ -67,7 +67,7 @@ export default defineAgent({ session.on(voice.AgentSessionEventTypes.UserInputTranscribed, (ev) => { logger.info( - { final: ev.isFinal, transcript: ev.transcript, language: ev.language }, + { final: ev.isFinal, 'lk.pii.transcript': ev.transcript, language: ev.language }, 'User transcript received', ); }); diff --git a/examples/src/otel_trace.ts b/examples/src/otel_trace.ts index d8efa979b..6cc15bf20 100644 --- a/examples/src/otel_trace.ts +++ b/examples/src/otel_trace.ts @@ -80,7 +80,7 @@ const lookupWeather = llm.tool({ location: z.string().describe('The location they are asking for'), }), execute: async ({ location }) => { - logger().info({ location }, 'Looking up weather'); + logger().info({ 'lk.pii.location': location }, 'Looking up weather'); return 'sunny with a temperature of 70 degrees.'; }, }); diff --git a/examples/src/phonic_realtime_agent.ts b/examples/src/phonic_realtime_agent.ts index 174eae9ec..e4004eba4 100644 --- a/examples/src/phonic_realtime_agent.ts +++ b/examples/src/phonic_realtime_agent.ts @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: 2026 LiveKit, Inc. // // SPDX-License-Identifier: Apache-2.0 -import { type JobContext, ServerOptions, cli, defineAgent, llm, voice } from '@livekit/agents'; +import { type JobContext, ServerOptions, cli, defineAgent, llm, log, voice } from '@livekit/agents'; import * as phonic from '@livekit/agents-plugin-phonic'; import { fileURLToPath } from 'node:url'; import { z } from 'zod'; @@ -14,7 +14,7 @@ const toggleLight = llm.tool({ state: z.enum(['on', 'off']).describe('Whether to turn the light on or off'), }), execute: async ({ light_id, state }) => { - console.log(`Turning ${state} light ${light_id}`); + log().info({ state, 'lk.pii.light_id': light_id }, 'toggling light'); await new Promise((resolve) => setTimeout(resolve, 1_000)); return `Light ${light_id} turned ${state}`; }, diff --git a/examples/src/play_local_audio_file.ts b/examples/src/play_local_audio_file.ts index cf80ffca8..51c916309 100644 --- a/examples/src/play_local_audio_file.ts +++ b/examples/src/play_local_audio_file.ts @@ -40,7 +40,7 @@ export default defineAgent({ const resourcesPath = join(currentDir, '../../agents/resources'); const audioFile = join(resourcesPath, 'office-ambience.ogg'); - logger.info(`Playing: ${audioFile}`); + logger.info({ 'lk.pii.audio_file': audioFile }, 'Playing audio file'); const abortController = new AbortController(); diff --git a/examples/src/realtime_streaming_transcript.ts b/examples/src/realtime_streaming_transcript.ts index 00f525a31..0685a5f47 100644 --- a/examples/src/realtime_streaming_transcript.ts +++ b/examples/src/realtime_streaming_transcript.ts @@ -24,12 +24,12 @@ export default defineAgent({ let lastPartialLength = 0; session.on(voice.AgentSessionEventTypes.UserInputTranscribed, (ev) => { if (ev.isFinal) { - logger.info({ transcript: ev.transcript }, '[user transcript FINAL]'); + logger.info({ 'lk.pii.transcript': ev.transcript }, 'user transcript final'); lastPartialLength = 0; return; } if (ev.transcript.length - lastPartialLength >= 6) { - logger.info({ transcript: ev.transcript }, '[user transcript partial]'); + logger.info({ 'lk.pii.transcript': ev.transcript }, 'user transcript partial'); lastPartialLength = ev.transcript.length; } }); diff --git a/examples/src/realtime_with_tts.ts b/examples/src/realtime_with_tts.ts index d0e898630..d6f69b417 100644 --- a/examples/src/realtime_with_tts.ts +++ b/examples/src/realtime_with_tts.ts @@ -19,7 +19,7 @@ export default defineAgent({ location: z.string().describe('The location to get the weather for'), }), execute: async ({ location }) => { - logger.info(`getting weather for ${location}`); + logger.info({ 'lk.pii.location': location }, 'getting weather'); return `The weather in ${location} is sunny, and the temperature is 20 degrees Celsius.`; }, }); diff --git a/plugins/anam/src/avatar.ts b/plugins/anam/src/avatar.ts index 90e038d33..0cbd2b754 100644 --- a/plugins/anam/src/avatar.ts +++ b/plugins/anam/src/avatar.ts @@ -102,8 +102,9 @@ export class AvatarSession extends voice.AvatarSession { personaId: this.opts.personaConfig?.personaId, apiUrl: apiUrl ?? '(default https://api.anam.ai)', livekitUrl, - avatarParticipantIdentity: this.opts.avatarParticipantIdentity ?? 'anam-avatar-agent', - publishOnBehalf: localIdentity, + 'lk.pii.avatar_participant_identity': + this.opts.avatarParticipantIdentity ?? 'anam-avatar-agent', + 'lk.pii.publish_on_behalf': localIdentity, }, 'starting Anam avatar session', ); diff --git a/plugins/assemblyai/src/stt.ts b/plugins/assemblyai/src/stt.ts index 421116ef3..f2f0db277 100644 --- a/plugins/assemblyai/src/stt.ts +++ b/plugins/assemblyai/src/stt.ts @@ -517,7 +517,13 @@ export class SpeechStream extends stt.SpeechStream { resolve(); } } catch (err) { - this.#logger.error(`AssemblyAI: error processing message: ${msg}`); + this.#logger.error( + { + error: err, + 'lk.pii.message': msg.toString(), + }, + 'AssemblyAI failed to process message', + ); reject(err); } }; diff --git a/plugins/baseten/src/stt.ts b/plugins/baseten/src/stt.ts index 84dc7952c..723163fbb 100644 --- a/plugins/baseten/src/stt.ts +++ b/plugins/baseten/src/stt.ts @@ -242,7 +242,7 @@ export class SpeechStream extends stt.SpeechStream { // Skip if no transcript text if (!transcript) { - this.#logger.debug('Received non-transcript message:', msg); + this.#logger.debug({ 'lk.pii.message': msg }, 'Received non-transcript message'); return; } @@ -307,7 +307,13 @@ export class SpeechStream extends stt.SpeechStream { resolve(); } } catch (err) { - this.#logger.error(`STT: Error processing message: ${data}`); + this.#logger.error( + { + error: err, + 'lk.pii.message': String(data), + }, + 'Baseten STT failed to process message', + ); reject(err); } }); diff --git a/plugins/cartesia/src/stt.ts b/plugins/cartesia/src/stt.ts index 46b9fee7f..cf5f3041c 100644 --- a/plugins/cartesia/src/stt.ts +++ b/plugins/cartesia/src/stt.ts @@ -499,7 +499,7 @@ export class SpeechStream extends stt.SpeechStream { } try { if (msg.type === 'error') { - this.#logger.error('Cartesia sent an error', msg); + this.#logger.error({ 'lk.pii.message': msg }, 'Cartesia sent an error'); // do not close the websocket on bad requests since that may be caused by invalid messages if (msg.status_code === undefined || msg.status_code >= 500) { @@ -657,7 +657,7 @@ export class SpeechStream extends stt.SpeechStream { } default: - this.#logger.warn('received unexpected message from Cartesia STT', { data }); + this.#logger.warn({ 'lk.pii.data': data }, 'received unexpected message from Cartesia STT'); } } diff --git a/plugins/cartesia/src/tts.ts b/plugins/cartesia/src/tts.ts index 7755bac00..34d70ae25 100644 --- a/plugins/cartesia/src/tts.ts +++ b/plugins/cartesia/src/tts.ts @@ -635,7 +635,7 @@ export class SynthesizeStream extends tts.SynthesizeStream { } else if (!isFlushDoneMessage(serverMsg) && !isErrorMessage(serverMsg)) { // flush_done is an ack with nothing to do; error frames without // done:true were already logged above. - this.#logger.warn({ message: serverMsg }, 'Unknown Cartesia message'); + this.#logger.warn({ 'lk.pii.message': serverMsg }, 'Unknown Cartesia message'); } } diff --git a/plugins/deepgram/src/stt.ts b/plugins/deepgram/src/stt.ts index e17349f78..e560e91be 100644 --- a/plugins/deepgram/src/stt.ts +++ b/plugins/deepgram/src/stt.ts @@ -551,7 +551,9 @@ export class SpeechStream extends stt.SpeechStream { break; } default: { - this.#logger.child({ msg: json }).warn('received unexpected message from Deepgram'); + this.#logger + .child({ 'lk.pii.message': json }) + .warn('received unexpected message from Deepgram'); break; } } @@ -560,7 +562,10 @@ export class SpeechStream extends stt.SpeechStream { resolve(); } } catch (err) { - this.#logger.error(`STT: Error processing message: ${msg}`); + this.#logger.error( + { error: err, 'lk.pii.message': msg.toString() }, + 'Deepgram STT failed to process message', + ); reject(err); } }); diff --git a/plugins/deepgram/src/stt_logging.test.ts b/plugins/deepgram/src/stt_logging.test.ts new file mode 100644 index 000000000..9da180697 --- /dev/null +++ b/plugins/deepgram/src/stt_logging.test.ts @@ -0,0 +1,78 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { log } from '@livekit/agents'; +import { once } from 'node:events'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { type WebSocket, WebSocketServer } from 'ws'; +import { STT } from './stt.js'; +import { STTv2 } from './stt_v2.js'; + +const servers: WebSocketServer[] = []; +const streams: Array<{ close(): void }> = []; + +async function startServer(onConnection?: (socket: WebSocket) => void): Promise { + const server = new WebSocketServer({ host: '127.0.0.1', port: 0 }); + servers.push(server); + await once(server, 'listening'); + if (onConnection) server.on('connection', onConnection); + + const address = server.address(); + if (address === null || typeof address === 'string') { + throw new Error('expected a TCP address'); + } + return `ws://127.0.0.1:${address.port}`; +} + +afterEach(async () => { + for (const stream of streams.splice(0)) stream.close(); + await Promise.all( + servers.splice(0).map(async (server) => { + for (const client of server.clients) client.terminate(); + await new Promise((resolve) => server.close(() => resolve())); + }), + ); + vi.restoreAllMocks(); +}); + +describe('Deepgram STT logging', () => { + it('logs invalid messages as text with the parse error', async () => { + const errorSpy = vi.spyOn(log(), 'error'); + const baseUrl = await startServer((socket) => { + setTimeout(() => socket.send('not-json'), 10); + }); + const recognizer = new STT({ apiKey: 'test', baseUrl }); + const stream = recognizer.stream(); + streams.push(stream); + + await vi.waitFor(() => { + expect(errorSpy).toHaveBeenCalledWith( + { + error: expect.any(SyntaxError), + 'lk.pii.message': 'not-json', + }, + 'Deepgram STT failed to process message', + ); + }); + }); + + it('does not log the v2 connection URL', async () => { + const debugSpy = vi.spyOn(log(), 'debug'); + const baseUrl = await startServer(); + const recognizer = new STTv2({ + apiKey: 'test', + endpointUrl: `${baseUrl}/v2/listen`, + keyterms: ['private-keyterm'], + }); + const stream = recognizer.stream(); + streams.push(stream); + + await vi.waitFor(() => { + const connectionCalls = debugSpy.mock.calls.filter((call) => + call.includes('connecting to Deepgram'), + ); + expect(connectionCalls).toEqual([['connecting to Deepgram']]); + expect(JSON.stringify(connectionCalls)).not.toContain('private-keyterm'); + }); + }); +}); diff --git a/plugins/deepgram/src/stt_v2.ts b/plugins/deepgram/src/stt_v2.ts index f0ccc5a4d..da28c868b 100644 --- a/plugins/deepgram/src/stt_v2.ts +++ b/plugins/deepgram/src/stt_v2.ts @@ -314,7 +314,7 @@ class SpeechStreamv2 extends stt.SpeechStream { this.#reconnectEvent.clear(); const url = this.#getDeepgramUrl(); - this.#logger.debug(`Connecting to Deepgram: ${url}`); + this.#logger.debug('connecting to Deepgram'); this.#ws = new WebSocket(url, { headers: { Authorization: `Token ${this.#opts.apiKey}` }, @@ -506,7 +506,7 @@ class SpeechStreamv2 extends stt.SpeechStream { this.#onEndOfSpeech(); } } else if (data.type === 'Error') { - this.#logger.warn('deepgram sent an error', { data }); + this.#logger.warn({ 'lk.pii.data': data }, 'deepgram sent an error'); const desc = (data.description as string) || 'unknown error from deepgram'; throw new Error(`Deepgram API Error: ${desc}`); } diff --git a/plugins/deepgram/src/tts.ts b/plugins/deepgram/src/tts.ts index 56f7b651f..6a7f11092 100644 --- a/plugins/deepgram/src/tts.ts +++ b/plugins/deepgram/src/tts.ts @@ -422,7 +422,7 @@ export class SynthesizeStream extends tts.SynthesizeStream { reject(new APIError('Deepgram TTS returned error', { body: message })); return; } else if (message.type !== 'Metadata') { - this.#logger.warn({ message }, 'Unknown Deepgram message type'); + this.#logger.warn({ 'lk.pii.message': message }, 'Unknown Deepgram message type'); } resetMessageTimeout(reject); diff --git a/plugins/elevenlabs/src/stt.ts b/plugins/elevenlabs/src/stt.ts index ba18e4c58..66d2ed062 100644 --- a/plugins/elevenlabs/src/stt.ts +++ b/plugins/elevenlabs/src/stt.ts @@ -862,7 +862,7 @@ export class SpeechStream extends stt.SpeechStream { } if (messageType === 'partial_transcript') { - this.#logger.debug({ data }, 'Received message type partial_transcript'); + this.#logger.debug({ 'lk.pii.data': data }, 'Received message type partial_transcript'); if (text) { if (!this.#speaking) { this.queue.put({ type: stt.SpeechEventType.START_OF_SPEECH }); @@ -915,7 +915,13 @@ export class SpeechStream extends stt.SpeechStream { ) { return; } else { - this.#logger.warn(`ElevenLabs STT unknown message type: ${messageType}, data: ${data}`); + this.#logger.warn( + { + messageType, + 'lk.pii.data': data, + }, + 'ElevenLabs STT returned an unknown message type', + ); } } } diff --git a/plugins/elevenlabs/src/tts.ts b/plugins/elevenlabs/src/tts.ts index e0bc31c97..38902b614 100644 --- a/plugins/elevenlabs/src/tts.ts +++ b/plugins/elevenlabs/src/tts.ts @@ -479,7 +479,11 @@ class Connection { if (data.error) { this.#logger.error( - { context_id: contextId, error: data.error, data }, + { + context_id: contextId, + error: data.error, + 'lk.pii.data': data, + }, 'elevenlabs tts returned error', ); if (contextId) { @@ -494,13 +498,16 @@ class Connection { if (!ctx) { if (data.type === 'flush_done') { this.#logger.debug( - { context_id: contextId, data }, + { context_id: contextId, 'lk.pii.data': data }, 'ignoring elevenlabs flush_done message for inactive context', ); continue; } - this.#logger.warn({ data }, 'unexpected message received from elevenlabs tts'); + this.#logger.warn( + { 'lk.pii.data': data }, + 'unexpected message received from elevenlabs tts', + ); continue; } diff --git a/plugins/fishaudio/src/tts.ts b/plugins/fishaudio/src/tts.ts index 21d916091..c451f7923 100644 --- a/plugins/fishaudio/src/tts.ts +++ b/plugins/fishaudio/src/tts.ts @@ -646,7 +646,7 @@ export class SynthesizeStream extends tts.SynthesizeStream { } if (!finished.done) finished.resolve(); } else { - this.#logger.debug({ event }, 'unknown Fish Audio event'); + this.#logger.debug({ 'lk.pii.event': event }, 'unknown Fish Audio event'); } }; diff --git a/plugins/google/src/realtime/realtime_api.ts b/plugins/google/src/realtime/realtime_api.ts index cefd46906..aec19831a 100644 --- a/plugins/google/src/realtime/realtime_api.ts +++ b/plugins/google/src/realtime/realtime_api.ts @@ -1197,7 +1197,13 @@ export class RealtimeSession extends llm.RealtimeSession { case 'content': const { turns, turnComplete } = msg.value; if (LK_GOOGLE_DEBUG) { - this.#logger.debug(`(client) -> ${JSON.stringify(this.loggableClientEvent(msg))}`); + this.#logger.debug( + { + eventType: msg.type, + 'lk.pii.event': this.loggableClientEvent(msg), + }, + 'sent Gemini Live client event', + ); } await session.sendClientContent({ turns, @@ -1208,7 +1214,13 @@ export class RealtimeSession extends llm.RealtimeSession { const { functionResponses } = msg.value; if (functionResponses) { if (LK_GOOGLE_DEBUG) { - this.#logger.debug(`(client) -> ${JSON.stringify(this.loggableClientEvent(msg))}`); + this.#logger.debug( + { + eventType: msg.type, + 'lk.pii.event': this.loggableClientEvent(msg), + }, + 'sent Gemini Live client event', + ); } try { await session.sendToolResponse({ @@ -1269,9 +1281,15 @@ export class RealtimeSession extends llm.RealtimeSession { (part) => part.inlineData?.data, ); if (LK_GOOGLE_DEBUG) { - this.#logger.debug(`(server) <- ${JSON.stringify(this.loggableServerMessage(response))}`); + this.#logger.debug( + { 'lk.pii.response': this.loggableServerMessage(response) }, + 'received Gemini Live server event', + ); } else if (!hasAudioData) { - this.#logger.debug(`(server) <- ${JSON.stringify(this.loggableServerMessage(response))}`); + this.#logger.debug( + { 'lk.pii.response': this.loggableServerMessage(response) }, + 'received Gemini Live server event', + ); } const unlock = await this.sessionLock.lock(); diff --git a/plugins/inworld/src/tts.ts b/plugins/inworld/src/tts.ts index bbf23e0f5..db081bcd5 100644 --- a/plugins/inworld/src/tts.ts +++ b/plugins/inworld/src/tts.ts @@ -512,7 +512,13 @@ class ChunkedStream extends tts.ChunkedStream { throw new Error(data.error.message); } } catch (e) { - log().warn({ error: e, line }, 'Failed to parse Inworld chunk'); + log().warn( + { + error: e, + 'lk.pii.line': line, + }, + 'Failed to parse Inworld chunk', + ); } } } diff --git a/plugins/lemonslice/src/meeting/chat.ts b/plugins/lemonslice/src/meeting/chat.ts index 57fe28311..c5254b10e 100644 --- a/plugins/lemonslice/src/meeting/chat.ts +++ b/plugins/lemonslice/src/meeting/chat.ts @@ -100,7 +100,10 @@ export class MeetingChatRelay { return; } - this.#logger.info({ userInput: userInput.slice(0, 120) }, 'meeting chat relay: user_input'); + this.#logger.info( + { 'lk.pii.user_input': userInput.slice(0, 120) }, + 'meeting chat relay received user input', + ); try { this.session.interrupt(); this.session.generateReply({ userInput }); diff --git a/plugins/livekit/src/turn_detector/base.ts b/plugins/livekit/src/turn_detector/base.ts index d95621ea3..2e0094862 100644 --- a/plugins/livekit/src/turn_detector/base.ts +++ b/plugins/livekit/src/turn_detector/base.ts @@ -93,7 +93,13 @@ export abstract class EOURunnerBase extends InferenceRunner): Promise { + const server = new WebSocketServer({ host: '127.0.0.1', port: 0 }); + servers.push(server); + await once(server, 'listening'); + + server.on('connection', (socket: WebSocket) => { + socket.on('message', (raw) => { + const message = JSON.parse(raw.toString()) as { event?: string }; + if (message.event !== 'task_start') return; + + socket.send( + JSON.stringify({ event: 'task_started', session_id: 'session-123', trace_id: 'trace-123' }), + ); + socket.send(JSON.stringify({ event: 'task_failed', trace_id: 'trace-123', data: payload })); + }); + }); + + const address = server.address(); + if (address === null || typeof address === 'string') { + throw new Error('expected a TCP address'); + } + return `http://127.0.0.1:${address.port}`; +} + +async function consume(stream: AsyncIterable): Promise { + for await (const _event of stream) { + // Consume the complete stream. + } +} + +afterEach(async () => { + await Promise.all( + servers.splice(0).map(async (server) => { + for (const client of server.clients) client.terminate(); + await new Promise((resolve) => server.close(() => resolve())); + }), + ); +}); + +describe('MiniMax TTS redaction', () => { + it('keeps a failed task payload out of the emitted error', async () => { + const baseUrl = await startTaskFailureServer({ transcript: 'secret transcript' }); + const synthesizer = new TTS({ apiKey: 'test-key', baseUrl }); + const errorEvent = new Promise((resolve) => { + synthesizer.once('error', (event) => resolve(event.error)); + }); + const stream = synthesizer.stream({ + connOptions: { maxRetry: 0, retryIntervalMs: 0, timeoutMs: 1000 }, + }); + + stream.endInput(); + await consume(stream); + const error = await errorEvent; + + expect(error).toBeInstanceOf(APIError); + expect(error.message).toBe('MiniMax task failed (trace_id: trace-123)'); + expect(error.message).not.toContain('secret transcript'); + expect((error as APIError).retryable).toBe(false); + }); +}); + const hasMinimaxConfig = Boolean(process.env.MINIMAX_API_KEY); if (hasMinimaxConfig) { diff --git a/plugins/minimax/src/tts.ts b/plugins/minimax/src/tts.ts index 13e7e2cdb..2d7cad24a 100644 --- a/plugins/minimax/src/tts.ts +++ b/plugins/minimax/src/tts.ts @@ -374,7 +374,7 @@ export class ChunkedStream extends tts.ChunkedStream { buffer = buffer.slice(newlineIdx + 1); if (!line) continue; if (!line.startsWith('data:')) { - this.#logger.warn({ line }, 'unexpected MiniMax SSE line'); + this.#logger.warn({ 'lk.pii.line': line }, 'unexpected MiniMax SSE line'); continue; } @@ -520,7 +520,13 @@ export class SynthesizeStream extends tts.SynthesizeStream { try { data = JSON.parse(rawMsg); } catch (e) { - this.#logger.warn({ err: e, rawMsg }, 'failed to parse MiniMax WS message'); + this.#logger.warn( + { + err: e, + 'lk.pii.raw_message': rawMsg, + }, + 'failed to parse MiniMax WS message', + ); continue; } @@ -583,12 +589,11 @@ export class SynthesizeStream extends tts.SynthesizeStream { if (event === 'task_failed') { // task_failed denotes a permanent server-side failure (invalid // params, unsupported voice, etc.); don't retry. - throw new APIError( - `MiniMax task failed (trace_id: ${currentTraceId}): ${JSON.stringify(data)}`, - { retryable: false }, - ); + const errorMessage = `MiniMax task failed (trace_id: ${currentTraceId})`; + this.#logger.error({ 'lk.pii.data': data }, errorMessage); + throw new APIError(errorMessage, { retryable: false }); } - this.#logger.warn({ data }, 'unexpected MiniMax WS event'); + this.#logger.warn({ 'lk.pii.data': data }, 'unexpected MiniMax WS event'); } }; diff --git a/plugins/openai/src/realtime/realtime_model.ts b/plugins/openai/src/realtime/realtime_model.ts index 0be3cb55e..6f5a34d4e 100644 --- a/plugins/openai/src/realtime/realtime_model.ts +++ b/plugins/openai/src/realtime/realtime_model.ts @@ -1289,7 +1289,13 @@ export class RealtimeSession extends llm.RealtimeSession { } if (lkOaiDebug && event.type !== 'input_audio_buffer.append') { - this.#logger.debug(this.loggableEvent(event), `(client) -> ${event.type}`); + this.#logger.debug( + { + eventType: event.type, + 'lk.pii.event': this.loggableEvent(event), + }, + 'sent OpenAI Realtime client event', + ); } this.emit('openai_client_event_queued', event); @@ -1320,7 +1326,13 @@ export class RealtimeSession extends llm.RealtimeSession { this.emit('openai_server_event_received', event); if (lkOaiDebug) { - this.#logger.debug(this.loggableEvent(event), `(server) <- ${event.type}`); + this.#logger.debug( + { + eventType: event.type, + 'lk.pii.event': this.loggableEvent(event), + }, + 'received OpenAI Realtime server event', + ); } try { @@ -1404,7 +1416,13 @@ export class RealtimeSession extends llm.RealtimeSession { wsConn.close(); return; } - this.#logger.error({ error, event }, 'Failed to handle OpenAI Realtime API event'); + this.#logger.error( + { + error, + 'lk.pii.event': event, + }, + 'Failed to handle OpenAI Realtime API event', + ); } }; diff --git a/plugins/phonic/src/realtime/realtime_model.ts b/plugins/phonic/src/realtime/realtime_model.ts index 5a7aae2f8..790c1c7d9 100644 --- a/plugins/phonic/src/realtime/realtime_model.ts +++ b/plugins/phonic/src/realtime/realtime_model.ts @@ -497,7 +497,10 @@ export class RealtimeSession extends llm.RealtimeSession { } if (item?.type === 'message') { if ((item.role === 'system' || item.role === 'developer') && item.rawTextContent) { - this.#logger.debug(`Sending add system message: ${item.rawTextContent}`); + this.#logger.debug( + { 'lk.pii.system_message': item.rawTextContent }, + 'sending add system message', + ); this.socket?.sendAddSystemMessage({ type: 'add_system_message', system_message: item.rawTextContent, @@ -507,7 +510,7 @@ export class RealtimeSession extends llm.RealtimeSession { // Only treat a user message as text input when it's appended at the tail of the context. if (item.role === 'user' && itemId === lastItemId && item.rawTextContent) { - this.#logger.debug(`Received user text input: ${item.rawTextContent}`); + this.#logger.debug({ 'lk.pii.text': item.rawTextContent }, 'received user text input'); this.pendingUserText = item.rawTextContent; bufferedUserText = true; } diff --git a/plugins/sarvam/src/stt.ts b/plugins/sarvam/src/stt.ts index 48a6d7abc..6ca9dece6 100644 --- a/plugins/sarvam/src/stt.ts +++ b/plugins/sarvam/src/stt.ts @@ -759,7 +759,10 @@ export class SpeechStream extends stt.SpeechStream { ws.on('message', (msg: RawData) => { try { const raw = msg.toString(); - this.#logger.debug(`Sarvam STT raw WS message: ${raw.substring(0, 500)}`); + this.#logger.debug( + { 'lk.pii.raw_message': raw.substring(0, 500) }, + 'received Sarvam STT WebSocket message', + ); const json = JSON.parse(raw); const msgType: string = json['type'] ?? ''; @@ -835,7 +838,13 @@ export class SpeechStream extends stt.SpeechStream { resolve(); } } catch (err) { - this.#logger.error(`Error processing Sarvam STT message: ${msg}`); + this.#logger.error( + { + error: err, + 'lk.pii.message': msg.toString(), + }, + 'Error processing Sarvam STT message', + ); reject(err); } }); diff --git a/plugins/soniox/src/tts.ts b/plugins/soniox/src/tts.ts index fb820bcb9..ba745836c 100644 --- a/plugins/soniox/src/tts.ts +++ b/plugins/soniox/src/tts.ts @@ -640,7 +640,13 @@ class Connection { try { response = JSON.parse(raw) as Record; } catch (error) { - this.#logger.warn({ error, raw }, 'Failed to parse Soniox TTS response'); + this.#logger.warn( + { + error, + 'lk.pii.raw_response': raw, + }, + 'Failed to parse Soniox TTS response', + ); return; } diff --git a/plugins/xai/src/stt.ts b/plugins/xai/src/stt.ts index 3468c6e9c..c373a53c2 100644 --- a/plugins/xai/src/stt.ts +++ b/plugins/xai/src/stt.ts @@ -332,7 +332,13 @@ export class SpeechStream extends stt.SpeechStream { resolve(); } } catch (err) { - this.#logger.error(`xAI STT: error processing message: ${msg}`); + this.#logger.error( + { + error: err, + 'lk.pii.message': msg.toString(), + }, + 'xAI STT failed to process message', + ); reject(err); } }); diff --git a/plugins/xai/src/tts.ts b/plugins/xai/src/tts.ts index aacade64d..1617f1dbd 100644 --- a/plugins/xai/src/tts.ts +++ b/plugins/xai/src/tts.ts @@ -417,7 +417,13 @@ export class SynthesizeStream extends tts.SynthesizeStream { try { data = JSON.parse(event.data.toString()) as Record; } catch (e) { - this.#logger.warn({ err: e }, 'Unexpected xAI message'); + this.#logger.warn( + { + err: e, + 'lk.pii.data': event.data.toString(), + }, + 'Unexpected xAI message', + ); continue; } @@ -446,7 +452,7 @@ export class SynthesizeStream extends tts.SynthesizeStream { options: { body: data }, }); default: - this.#logger.warn({ data }, 'Unexpected xAI message'); + this.#logger.warn({ 'lk.pii.data': data }, 'Unexpected xAI message'); } } };