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

livekit-agents/livekit/agents/beta/workflows/dtmf_inputs.py

@@ -108,7 +108,10 @@ def _on_sip_dtmf_received(ev: rtc.SipDTMF) -> None:
                 return
 
             self._curr_dtmf_inputs.append(DtmfEvent(ev.digit))
-            logger.info(f"DTMF inputs: {format_dtmf(self._curr_dtmf_inputs)}")
+            logger.info(
+                "DTMF inputs received",
+                extra={"lk.pii.dtmf_inputs": format_dtmf(self._curr_dtmf_inputs)},
+            )
             self._generate_dtmf_reply.schedule()
 
         @debounced(delay=dtmf_input_timeout)
@@ -119,7 +122,7 @@ async def _generate_dtmf_reply() -> None:
                 self.session.interrupt()
 
                 dmtf_str = format_dtmf(self._curr_dtmf_inputs)
-                logger.debug(f"Generating DTMF reply, current inputs: {dmtf_str}")
+                logger.debug("Generating DTMF reply", extra={"lk.pii.dtmf_inputs": dmtf_str})
 
                 # if input not fully received (i.e. timeout), return None
                 if len(self._curr_dtmf_inputs) != num_digits:

livekit-agents/livekit/agents/llm/utils.py

@@ -495,7 +495,7 @@ def prepare_function_arguments(
         except ValueError as e:
             logger.error(
                 f"error parsing arguments for `{fnc.info.name}`",
-                extra={"function": fnc.info.name, "arguments": json_arguments},
+                extra={"function": fnc.info.name, "lk.pii.arguments": json_arguments},
             )
             raise ToolError(f"Error parsing arguments for `{fnc.info.name}`: {e}") from e
 
@@ -514,13 +514,13 @@ def prepare_function_arguments(
     except (pydantic.ValidationError, ValueError, TypeError) as e:
         logger.error(
             f"error parsing arguments for `{fnc.info.name}`",
-            extra={"function": fnc.info.name, "arguments": json_arguments},
+            extra={"function": fnc.info.name, "lk.pii.arguments": json_arguments},
         )
         raise ToolError(f"Error parsing arguments for `{fnc.info.name}`: {e}") from e
     except Exception:
         logger.exception(
             f"error parsing arguments for `{fnc.info.name}`",
-            extra={"function": fnc.info.name, "arguments": json_arguments},
+            extra={"function": fnc.info.name, "lk.pii.arguments": json_arguments},
         )
         raise
 
@@ -784,7 +784,7 @@ async def execute_function_call(
         if not isinstance(e, ToolError):
             logger.exception(
                 f"exception executing AI function `{tool_call.name}`",
-                extra={"call_id": tool_call.call_id, "arguments": tool_call.arguments},
+                extra={"call_id": tool_call.call_id, "lk.pii.arguments": tool_call.arguments},
             )
         out = make_function_call_output(fnc_call=fnc_call, output=None, exception=e)
 

livekit-agents/livekit/agents/telemetry/trace_types.py

@@ -1,3 +1,12 @@
+"""Span attribute and event name constants for LiveKit Agents telemetry.
+
+Attributes carrying conversational content, tool payloads, or other user data
+must include a dot-delimited ``pii`` segment (``lk.pii.<name>``): PII-enabled
+projects have these attributes stripped at the LiveKit Cloud collector, and the
+segment is the only marker it honors. Attributes must never embed such content
+in span names, event names, or log message bodies — those are not redactable.
+"""
+
 ATTR_SPEECH_ID = "lk.speech_id"
 ATTR_AGENT_LABEL = "lk.agent_label"
 ATTR_START_TIME = "lk.start_time"
@@ -24,28 +33,28 @@
 # agent turn
 ATTR_AGENT_TURN_ID = "lk.generation_id"
 ATTR_AGENT_PARENT_TURN_ID = "lk.parent_generation_id"
-ATTR_USER_INPUT = "lk.user_input"
-ATTR_INSTRUCTIONS = "lk.instructions"
+ATTR_USER_INPUT = "lk.pii.user_input"
+ATTR_INSTRUCTIONS = "lk.pii.instructions"
 ATTR_SPEECH_INTERRUPTED = "lk.interrupted"
 
 # llm node
-ATTR_CHAT_CTX = "lk.chat_ctx"
+ATTR_CHAT_CTX = "lk.pii.chat_ctx"
 ATTR_FUNCTION_TOOLS = "lk.function_tools"
 ATTR_PROVIDER_TOOLS = "lk.provider_tools"
 ATTR_TOOL_SETS = "lk.tool_sets"
-ATTR_RESPONSE_TEXT = "lk.response.text"
-ATTR_RESPONSE_FUNCTION_CALLS = "lk.response.function_calls"
+ATTR_RESPONSE_TEXT = "lk.pii.response.text"
+ATTR_RESPONSE_FUNCTION_CALLS = "lk.pii.response.function_calls"
 ATTR_RESPONSE_TTFT = "lk.response.ttft"
 
 # function tool
 ATTR_FUNCTION_TOOL_ID = "lk.function_tool.id"
 ATTR_FUNCTION_TOOL_NAME = "lk.function_tool.name"
-ATTR_FUNCTION_TOOL_ARGS = "lk.function_tool.arguments"
+ATTR_FUNCTION_TOOL_ARGS = "lk.pii.function_tool.arguments"
 ATTR_FUNCTION_TOOL_IS_ERROR = "lk.function_tool.is_error"
-ATTR_FUNCTION_TOOL_OUTPUT = "lk.function_tool.output"
+ATTR_FUNCTION_TOOL_OUTPUT = "lk.pii.function_tool.output"
 
 # tts node
-ATTR_TTS_INPUT_TEXT = "lk.input_text"
+ATTR_TTS_INPUT_TEXT = "lk.pii.input_text"
 ATTR_TTS_STREAMING = "lk.tts.streaming"
 ATTR_TTS_LABEL = "lk.tts.label"
 ATTR_RESPONSE_TTFB = "lk.response.ttfb"
@@ -55,7 +64,7 @@
 ATTR_EOU_UNLIKELY_THRESHOLD = "lk.eou.unlikely_threshold"
 ATTR_EOU_DELAY = "lk.eou.endpointing_delay"
 ATTR_EOU_LANGUAGE = "lk.eou.language"
-ATTR_USER_TRANSCRIPT = "lk.user_transcript"
+ATTR_USER_TRANSCRIPT = "lk.pii.user_transcript"
 ATTR_TRANSCRIPT_CONFIDENCE = "lk.transcript_confidence"
 ATTR_TRANSCRIPTION_DELAY = "lk.transcription_delay"
 ATTR_END_OF_TURN_DELAY = "lk.end_of_turn_delay"
@@ -108,7 +117,7 @@
 ATTR_AMD_REASON = "lk.amd.reason"
 ATTR_AMD_SPEECH_DURATION = "lk.amd.speech_duration"
 ATTR_AMD_DELAY = "lk.amd.delay"
-ATTR_AMD_TRANSCRIPT = "lk.amd.transcript"
+ATTR_AMD_TRANSCRIPT = "lk.pii.amd.transcript"
 
 # Adaptive Interruption attributes
 ATTR_IS_INTERRUPTION = "lk.is_interruption"

livekit-agents/livekit/agents/tts/stream_pacer.py

@@ -148,7 +148,7 @@ async def _send_task(self) -> None:
                     self._event_ch.send_nowait(TokenData(token=text))
                     logger.debug(
                         "sent text to tts",
-                        extra={"text": text, "remaining_audio": remaining_audio},
+                        extra={"lk.pii.text": text, "remaining_audio": remaining_audio},
                     )
                     generation_started = False
                     generation_stopped = False

livekit-agents/livekit/agents/voice/agent_activity.py

@@ -2137,7 +2137,7 @@ def on_end_of_turn(self, info: _EndOfTurnInfo) -> bool:
             self._cancel_preemptive_generation()
             logger.warning(
                 "skipping user input, speech scheduling is paused",
-                extra={"user_input": info.new_transcript},
+                extra={"lk.pii.user_input": info.new_transcript},
             )
 
             if self._session._closing:
@@ -2232,7 +2232,7 @@ async def _user_turn_completed_task(
             if not current_speech.allow_interruptions:
                 logger.warning(
                     "skipping reply to user input, current speech generation cannot be interrupted",
-                    extra={"user_input": info.new_transcript},
+                    extra={"lk.pii.user_input": info.new_transcript},
                 )
                 return
             await self._cancel_speech_pause(self._cancel_speech_pause_task)
@@ -2245,7 +2245,7 @@ async def _user_turn_completed_task(
         if self._scheduling_paused or self._new_turns_blocked:
             logger.warning(
                 "skipping on_user_turn_completed, speech scheduling is paused",
-                extra={"user_input": info.new_transcript},
+                extra={"lk.pii.user_input": info.new_transcript},
             )
             if self._session._closing:
                 self._agent._chat_ctx.items.append(user_message)
@@ -2279,7 +2279,7 @@ async def _user_turn_completed_task(
         if self._scheduling_paused or self._new_turns_blocked:
             logger.warning(
                 "skipping reply to user input, speech scheduling is paused",
-                extra={"user_input": info.new_transcript},
+                extra={"lk.pii.user_input": info.new_transcript},
             )
             if user_message and self._session._closing:
                 self._agent._chat_ctx.items.append(user_message)

livekit-agents/livekit/agents/voice/agent_session.py

@@ -1223,7 +1223,7 @@ async def _start_ivr_detection(self, transcript: str | None = None) -> None:
         if transcript is not None:
             logger.debug(
                 "IVR detection started with transcript",
-                extra={"transcript": transcript},
+                extra={"lk.pii.transcript": transcript},
             )
             self._ivr_activity._on_user_input_transcribed(
                 UserInputTranscribedEvent(transcript=transcript, is_final=True)
@@ -1799,7 +1799,7 @@ def _conversation_item_added(self, message: llm.ChatMessage) -> None:
         if text := message.text_content:
             logger.debug(
                 "conversation_item_added",
-                extra={"role": message.role, "text": text},
+                extra={"role": message.role, "lk.pii.text": text},
             )
         self.emit("conversation_item_added", ConversationItemAddedEvent(item=message))
 

livekit-agents/livekit/agents/voice/audio_recognition.py

@@ -1107,7 +1107,10 @@ async def _on_stt_event(self, ev: stt.SpeechEvent) -> None:
             if self._session.amd is not None:
                 self._session.amd._on_transcript(transcript)
 
-            extra: dict[str, Any] = {"user_transcript": transcript, "language": self._last_language}
+            extra: dict[str, Any] = {
+                "lk.pii.user_transcript": transcript,
+                "language": self._last_language,
+            }
             if self._last_speaking_time:
                 extra["transcript_delay"] = time.time() - self._last_speaking_time
             logger.debug("received user transcript", extra=extra)
@@ -1171,7 +1174,7 @@ async def _on_stt_event(self, ev: stt.SpeechEvent) -> None:
 
             logger.debug(
                 "received user preflight transcript",
-                extra={"user_transcript": transcript, "language": self._last_language},
+                extra={"lk.pii.user_transcript": transcript, "language": self._last_language},
             )
 
             # still need to increment it as it's used for turn detection,

livekit-plugins/livekit-plugins-asyncai/livekit/plugins/asyncai/tts.py

@@ -312,10 +312,10 @@ async def _recv_task(ws: aiohttp.ClientWebSocketResponse, asyncai_context_id: st
                         b64data = base64.b64decode(audio)
                         output_emitter.push(b64data)
                     except Exception:
-                        logger.warning("invalid audio payload %s", data)
+                        logger.warning("invalid audio payload", extra={"lk.pii.data": data})
                         continue
                 else:
-                    logger.warning("unexpected message %s", data)
+                    logger.warning("unexpected message", extra={"lk.pii.data": data})
 
         async_context_id = str(uuid.uuid4())
         try:

livekit-plugins/livekit-plugins-aws/livekit/plugins/aws/experimental/realtime/pretty_printer.py

@@ -40,8 +40,12 @@ class AnsiColors:
 def log_event_data(event_data: dict) -> None:
     event_type = SonicEventBuilder.get_event_type(event_data)
     color = EVENT_COLOR_MAP[event_type]
+    # The event payload may embed conversational content (transcripts, tool
+    # args, ...), so it must go through a redactable `lk.pii.*` extra instead
+    # of the log message body.
     logger.debug(
-        f"{color}{event_type.upper()}: {json.dumps(event_data, indent=2)}{AnsiColors.ENDC}"
+        f"{color}{event_type.upper()}{AnsiColors.ENDC}",
+        extra={"lk.pii.event_data": json.dumps(event_data, indent=2)},
     )
 
 

livekit-plugins/livekit-plugins-aws/livekit/plugins/aws/experimental/realtime/realtime_model.py

@@ -782,7 +782,7 @@ async def _send_raw_event(self, event_json: str) -> None:
 
         # Log the full JSON being sent (skip audio events to avoid log spam)
         if '"audioInput"' not in event_json:
-            logger.debug(f"[SEND] {event_json}")
+            logger.debug("[SEND] sending event", extra={"lk.pii.event_json": event_json})
 
         event = InvokeModelWithBidirectionalStreamInputChunk(
             value=BidirectionalInputPayloadPart(bytes_=event_json.encode("utf-8"))
@@ -966,8 +966,8 @@ async def initialize_streams(self, is_restart: bool = False) -> None:
                     restart_ctx = restart_ctx.copy()
                     restart_ctx.items.pop()
                     logger.debug(
-                        f"[SESSION] Popped last user message for interactive send: "
-                        f"{interactive_user_text[:60]}..."
+                        "[SESSION] Popped last user message for interactive send",
+                        extra={"lk.pii.user_text": interactive_user_text[:60]},
                     )
 
             init_events, history_events = self._event_builder.create_prompt_start_block(
@@ -1016,7 +1016,8 @@ async def initialize_streams(self, is_restart: bool = False) -> None:
             if interactive_user_text:
                 await self._stream_ready.wait()
                 logger.debug(
-                    f"[SESSION] Sending interactive user text: {interactive_user_text[:60]}..."
+                    "[SESSION] Sending interactive user text",
+                    extra={"lk.pii.user_text": interactive_user_text[:60]},
                 )
                 await self._send_text_message(interactive_user_text, interactive=True)
 
@@ -1219,7 +1220,7 @@ async def _handle_text_output_content_event(self, event_data: dict) -> None:
         content_type = self._current_generation.content_id_map.get(content_id)
 
         if content_type == "USER_ASR":
-            logger.debug(f"INPUT TRANSCRIPTION UPDATED: {text_content}")
+            logger.debug("input transcription updated", extra={"lk.pii.text_content": text_content})
             self._update_chat_ctx(role="user", text_content=text_content, content_id=content_id)
 
         elif content_type == "ASSISTANT_TEXT":
@@ -1240,7 +1241,9 @@ def _update_chat_ctx(
             a) 40 total messages limit
             b) 1kB message size limit
         """
-        logger.debug(f"Updating chat context with role: {role} and text_content: {text_content}")
+        logger.debug(
+            "Updating chat context", extra={"role": role, "lk.pii.text_content": text_content}
+        )
 
         # Start a new message when the user contentId changes (new utterance)
         force_new = False
@@ -1518,7 +1521,10 @@ async def _process_responses(self) -> None:
                             # logger.debug(f"Received event: {json_data}")
                             await self._handle_event(json_data)
                         except json.JSONDecodeError:
-                            logger.warning(f"JSON decode error: {response_data}")
+                            logger.warning(
+                                "JSON decode error",
+                                extra={"lk.pii.response_data": response_data},
+                            )
                     else:
                         logger.warning("No response received")
                 except concurrent.futures.InvalidStateError:
@@ -1713,7 +1719,7 @@ async def update_instructions(self, instructions: str) -> None:
             self._instructions_ready = asyncio.get_running_loop().create_future()
         if not self._instructions_ready.done():
             self._instructions_ready.set_result(True)
-        logger.debug(f"Instructions updated: {instructions}")
+        logger.debug("Instructions updated", extra={"lk.pii.instructions": instructions})
 
     async def update_chat_ctx(self, chat_ctx: llm.ChatContext) -> None:
         """Inject chat history and handle incremental user messages."""
@@ -1747,7 +1753,9 @@ async def update_chat_ctx(self, chat_ctx: llm.ChatContext) -> None:
             for item in chat_ctx.items:
                 if item.type == "message":
                     self._sent_message_ids.add(item.id)
-            logger.debug(f"Chat context updated: {self._chat_ctx.items}")
+            logger.debug(
+                "Chat context updated", extra={"lk.pii.chat_ctx_items": self._chat_ctx.items}
+            )
             self._chat_ctx_ready.set_result(True)
 
         # Process items in context
@@ -1757,7 +1765,7 @@ async def update_chat_ctx(self, chat_ctx: llm.ChatContext) -> None:
                 if item.call_id not in self._pending_tools:
                     continue
 
-                logger.debug(f"function call output: {item}")
+                logger.debug("function call output received", extra={"lk.pii.item": item})
                 self._pending_tools.discard(item.call_id)
 
                 # Format tool result as proper JSON
@@ -1785,7 +1793,8 @@ async def update_chat_ctx(self, chat_ctx: llm.ChatContext) -> None:
                 if item.id not in self._audio_message_ids:
                     if item.text_content and item.text_content.strip():
                         logger.debug(
-                            f"Sending user message as interactive text: {item.text_content}"
+                            "Sending user message as interactive text",
+                            extra={"lk.pii.text_content": item.text_content},
                         )
                         # Send interactive text to Nova Sonic (triggers generation)
                         # This is the flow for generate_reply(user_input=...) from the framework
@@ -1813,7 +1822,8 @@ async def _send_user_text(
                     self._chat_ctx.items.append(item)
                 else:
                     logger.debug(
-                        f"Skipping user message (already in context from audio): {item.text_content}"
+                        "Skipping user message (already in context from audio)",
+                        extra={"lk.pii.text_content": item.text_content},
                     )
                     self._sent_message_ids.add(item.id)
 
@@ -1983,7 +1993,9 @@ async def _process_audio_input(self) -> None:
                                     except Exception:
                                         logger.exception("Failed to parse tool result")
 
-                            logger.debug(f"Sending tool result: {tool_result}")
+                            logger.debug(
+                                "Sending tool result", extra={"lk.pii.tool_result": tool_result}
+                            )
                             await self._send_tool_events(tool_use_id, tool_result)
                             # Create new task for next tool result
                             tool_task = asyncio.create_task(self._tool_results_ch.recv())
@@ -2109,7 +2121,10 @@ async def _empty_function_stream() -> AsyncIterator[llm.FunctionCall]:
 
         # Nova 2.0: Only send if instructions provided
         if is_given(instructions):
-            logger.info(f"generate_reply: sending instructions='{instructions}'")
+            logger.info(
+                "generate_reply: sending instructions",
+                extra={"lk.pii.instructions": instructions},
+            )
 
             # Create future that will be resolved when generation starts
             fut = asyncio.Future[llm.GenerationCreatedEvent]()
@@ -2196,7 +2211,8 @@ async def _send_text_message(self, text: str, interactive: bool = True) -> None:
         await asyncio.sleep(0.01)
         await self._send_raw_event(self._event_builder.create_content_end_event(content_name))
         logger.info(
-            f"Sent text message (interactive={interactive}): {text[:50]}{'...' if len(text) > 50 else ''}"
+            "Sent text message",
+            extra={"interactive": interactive, "lk.pii.text": text[:50]},
         )
 
     def commit_audio(self) -> None:
@@ -2298,5 +2314,7 @@ async def aclose(self) -> None:
             tasks.append(self._main_atask)
 
         await asyncio.gather(*tasks, return_exceptions=True)
-        logger.debug(f"CHAT CONTEXT: {self._chat_ctx.items}")
+        logger.debug(
+            "chat context at session end", extra={"lk.pii.chat_ctx_items": self._chat_ctx.items}
+        )
         logger.info("Session end")

livekit-plugins/livekit-plugins-aws/livekit/plugins/aws/experimental/realtime/turn_tracker.py

@@ -87,7 +87,7 @@ def feed(self, event: dict) -> None:
                 return
 
         elif kind == "BARGE_IN":
-            logger.debug(f"BARGE-IN DETECTED IN TURN TRACKER: {turn}")
+            logger.debug("barge-in detected in turn tracker", extra={"lk.pii.turn": turn})
             # start new turn immediately to make interruptions snappier
             self._emit("input_speech_started", llm.InputSpeechStartedEvent())
             turn.phase = _Phase.DONE