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/job.py

@@ -235,6 +235,7 @@ def __init__(
         self._lock = asyncio.Lock()
         self._tagger = Tagger()
         self._recording_initialized = False
+        self._redaction_enabled = info.job.enable_redaction
         self._early_log_handler: _BufferingHandler | None = None
 
     def _on_setup(self) -> None:
@@ -797,11 +798,16 @@ def add_participant_entrypoint(
         self._participant_entrypoints.append((entrypoint_fnc, kind))
 
     def init_recording(self, options: RecordingOptions) -> None:
+        redaction_enabled = self.job.enable_redaction or options.get("redaction", False)
+        if redaction_enabled and options.get("audio", True) and not options.get("transcript", True):
+            raise ValueError("audio upload requires transcript upload when redaction is enabled")
+
         if self._recording_initialized:
             self._stop_log_buffering()
             return
 
         self._recording_initialized = True
+        self._redaction_enabled = redaction_enabled
 
         needs_cloud = (
             options.get("traces", True)

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

@@ -673,7 +673,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
 
@@ -692,13 +692,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
 
@@ -956,7 +956,7 @@ def make_function_call_output(
     if not _is_valid_function_output(output):
         logger.error(
             f"AI function `{fnc_call.name}` returned an invalid output",
-            extra={"call_id": fnc_call.call_id, "output": output},
+            extra={"call_id": fnc_call.call_id, "lk.pii.output": output},
         )
         return FunctionCallResult(
             fnc_call=fnc_call,
@@ -1032,7 +1032,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"
@@ -12,42 +21,42 @@
 
 
 ATTR_PARTICIPANT_ID = "lk.participant_id"
-ATTR_PARTICIPANT_IDENTITY = "lk.participant_identity"
+ATTR_PARTICIPANT_IDENTITY = "lk.pii.participant_identity"
 ATTR_PARTICIPANT_KIND = "lk.participant_kind"
 
 # session start
 ATTR_JOB_ID = "lk.job_id"
 ATTR_AGENT_NAME = "lk.agent_name"
 ATTR_CLOUD_AGENT_ID = "lk.cloud_agent_id"
 ATTR_DEPLOYMENT_ID = "lk.deployment_id"
-ATTR_ROOM_NAME = "lk.room_name"
+ATTR_ROOM_NAME = "lk.pii.room_name"
 ATTR_SESSION_OPTIONS = "lk.session_options"
 
 # 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"
@@ -57,7 +66,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"
@@ -110,7 +119,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/telemetry/traces.py

@@ -55,14 +55,32 @@
     ATTRIBUTE_SIMULATION_ENABLED,
     recording_enabled,
 )
-from . import trace_types
+from . import trace_types, utils as telemetry_utils
 
 if TYPE_CHECKING:
     from ..llm import ChatContext, ChatItem
     from ..observability import Tagger
+    from ..voice.agent_session import AgentSessionOptions
     from ..voice.report import SessionReport
 
 
+_SESSION_OPTION_KEY_ALIASES = {
+    "keyterms": "lk.pii.keyterms",
+}
+
+
+def _serialize_session_options(options: AgentSessionOptions) -> dict[str, Any]:
+    def _serialize(value: dict[str, Any]) -> dict[str, Any]:
+        return {
+            _SESSION_OPTION_KEY_ALIASES.get(key, key): (
+                _serialize(nested_value) if isinstance(nested_value, dict) else nested_value
+            )
+            for key, nested_value in value.items()
+        }
+
+    return _serialize(vars(options))
+
+
 class _DynamicTracer(Tracer):
     def __init__(self, instrumenting_module_name: str) -> None:
         self._instrumenting_module_name = instrumenting_module_name
@@ -81,6 +99,12 @@ def start_span(self, *args: Any, **kwargs: Any) -> Span:
 
     @_agnosticcontextmanager
     def start_as_current_span(self, *args: Any, **kwargs: Any) -> Iterator[Span]:
+        if telemetry_utils._redaction_enabled():
+            kwargs = {
+                **kwargs,
+                "record_exception": False,
+                "set_status_on_exception": False,
+            }
         with self._tracer.start_as_current_span(*args, **kwargs) as span:
             yield span
 
@@ -575,7 +599,7 @@ def _log(
             body="session report",
             timestamp=int((report.started_at or report.timestamp or 0) * 1e9),
             attributes={
-                "session.options": vars(report.options),
+                "session.options": _serialize_session_options(report.options),
                 "session.report_timestamp": report.timestamp,
                 "session.tags": sorted(tagger.tags) if tagger.tags else None,
                 "agent_name": agent_name,

livekit-agents/livekit/agents/telemetry/utils.py

@@ -6,13 +6,41 @@
 
 from opentelemetry import trace
 
+from ..types import NOT_GIVEN, NotGivenOr
 from . import trace_types
 
 if TYPE_CHECKING:
     from ..metrics import RealtimeModelMetrics
 
 
-def record_exception(span: trace.Span, exception: Exception) -> None:
+REDACTED_EXCEPTION_MESSAGE = "exception details redacted"
+
+
+def _redaction_enabled() -> bool:
+    from ..job import get_job_context
+
+    job_ctx = get_job_context(required=False)
+    if job_ctx is None:
+        return False
+    return job_ctx._redaction_enabled
+
+
+def record_exception(
+    span: trace.Span, exception: Exception, *, redacted: NotGivenOr[bool] = NOT_GIVEN
+) -> None:
+    if redacted is NOT_GIVEN:
+        redacted = _redaction_enabled()
+
+    if redacted:
+        attrs = {
+            trace_types.ATTR_EXCEPTION_TYPE: exception.__class__.__name__,
+            trace_types.ATTR_EXCEPTION_MESSAGE: REDACTED_EXCEPTION_MESSAGE,
+        }
+        span.add_event("exception", attrs)
+        span.set_status(trace.Status(trace.StatusCode.ERROR, REDACTED_EXCEPTION_MESSAGE))
+        span.set_attributes(attrs)
+        return
+
     span.record_exception(exception)
     span.set_status(trace.Status(trace.StatusCode.ERROR, str(exception)))
     # set the exception in span attributes in case the exception event is not rendered

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/version.py

@@ -12,4 +12,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-__version__ = "1.6.10"
+__version__ = "1.6.11.rc1"

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

@@ -2338,7 +2338,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:
@@ -2456,7 +2456,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)
@@ -2469,7 +2469,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)
@@ -2503,7 +2503,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

@@ -1395,7 +1395,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)
@@ -2033,7 +2033,7 @@ def _conversation_item_added(self, message: llm.ChatMessage) -> None:
         if text := message.raw_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

@@ -1223,7 +1223,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)
@@ -1284,7 +1287,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-agents/livekit/agents/voice/generation.py

@@ -1003,7 +1003,7 @@ def make_tool_output(
         if len(agent_tasks) > 1:
             logger.error(
                 f"AI function `{fnc_call.name}` returned multiple AgentTask instances, ignoring the output",  # noqa: E501
-                extra={"call_id": fnc_call.call_id, "output": output},
+                extra={"call_id": fnc_call.call_id, "lk.pii.output": output},
             )
             return ToolExecutionOutput(
                 fnc_call=fnc_call.model_copy(),