feat(telemetry): trace RPC calls made and handled

livekit-agents/livekit/agents/job.py

@@ -41,6 +41,7 @@
 from .telemetry import (
     _upload_session_report,
     otel_metrics,
+    rpc as rpc_tracing,
     session_context,
     trace_types,
     utils as telemetry_utils,
@@ -719,6 +720,7 @@ async def connect(
                         ),
                     }
                 )
+            rpc_tracing.install(self._room.local_participant)
             self._on_connect()
 
             # Always registered: the callback ignores participants without the

livekit-agents/livekit/agents/telemetry/__init__.py

@@ -5,6 +5,7 @@
     metrics,
     otel_metrics,
     pii,
+    rpc,
     session_context,
     trace_types,
     utils,
@@ -28,6 +29,7 @@
     "http_server",
     "loop_monitor",
     "session_context",
+    "rpc",
     "set_tracer_provider",
     "utils",
     "_setup_cloud_tracer",

livekit-agents/livekit/agents/telemetry/rpc.py

@@ -0,0 +1,158 @@
+"""Trace RPCs the agent performs and handles.
+
+The room SDK exposes an ``RpcInterceptor`` hook (``livekit-rtc`` >= 1.1.18) that wraps every
+call made through ``LocalParticipant.perform_rpc`` and every invocation dispatched to a
+registered handler. This module installs one interceptor per local participant that turns
+each call into a span following the OpenTelemetry RPC semantic conventions:
+
+* ``rpc_call`` (``SpanKind.CLIENT``) for outgoing calls, under whatever span is current where
+  the call is made (an RPC issued from a tool nests under ``function_tool``);
+* ``rpc_handler`` (``SpanKind.SERVER``) for incoming invocations, under the primary agent
+  session's root span.
+
+Payloads are recorded truncated under ``lk.pii`` keys. Participant identities are application
+identifiers, not end-user data, and are recorded as is. On an SDK without the hook,
+``install`` is a no-op.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from opentelemetry import trace
+
+from livekit import rtc
+
+from ..log import logger
+from . import session_context, trace_types
+from .traces import tracer
+
+MAX_PAYLOAD_ATTR_LEN = 1024
+"""Request and response payloads longer than this many characters are truncated in span
+attributes."""
+
+_RpcInterceptorBase: type = getattr(rtc, "RpcInterceptor", object)
+_warned_unsupported = False
+
+
+def _truncate(payload: str) -> str:
+    return payload[:MAX_PAYLOAD_ATTR_LEN]
+
+
+def _payload_attributes(payload: str) -> dict[str, Any]:
+    attrs: dict[str, Any] = {trace_types.ATTR_RPC_PAYLOAD_SIZE: len(payload.encode("utf-8"))}
+    if payload:
+        attrs[trace_types.ATTR_RPC_PAYLOAD] = _truncate(payload)
+    return attrs
+
+
+def _response_attributes(response: str | None) -> dict[str, Any]:
+    response = response or ""
+    attrs: dict[str, Any] = {trace_types.ATTR_RPC_RESPONSE_SIZE: len(response.encode("utf-8"))}
+    if response:
+        attrs[trace_types.ATTR_RPC_RESPONSE] = _truncate(response)
+    return attrs
+
+
+_CANCEL_DESCRIPTIONS = {
+    "RESPONSE_TIMEOUT": "response timeout",
+    "RECIPIENT_DISCONNECTED": "caller disconnected",
+    "APPLICATION_ERROR": "handler cancelled",
+}
+
+
+def _cancellation_outcome(invocation: Any) -> tuple[Any, str]:
+    """The RPC error code the caller receives for a cancelled handler chain, and a status text.
+
+    ``cancel_reason`` is set by the SDK before it cancels the chain; ``None`` there means the
+    cancellation came from inside the chain, which the SDK answers as ``APPLICATION_ERROR``.
+    An SDK without the field gives no code."""
+    if not hasattr(invocation, "cancel_reason"):
+        return None, "cancelled"
+    code = invocation.cancel_reason
+    if code is None:
+        code = rtc.RpcError.ErrorCode.APPLICATION_ERROR
+    return code, _CANCEL_DESCRIPTIONS.get(getattr(code, "name", ""), "cancelled")
+
+
+class TracingRpcInterceptor(_RpcInterceptorBase):  # type: ignore[misc]
+    """An ``rtc.RpcInterceptor`` emitting ``rpc_call`` / ``rpc_handler`` spans."""
+
+    async def intercept_outgoing(self, call: Any, next: Callable[[Any], Awaitable[str]]) -> str:
+        attributes: dict[str, Any] = {
+            trace_types.ATTR_RPC_METHOD: call.method,
+            trace_types.ATTR_RPC_DESTINATION_IDENTITY: call.destination_identity,
+            **_payload_attributes(call.payload),
+        }
+        if call.response_timeout is not None:
+            attributes[trace_types.ATTR_RPC_RESPONSE_TIMEOUT] = call.response_timeout
+
+        with tracer.start_as_current_span(
+            "rpc_call", kind=trace.SpanKind.CLIENT, attributes=attributes
+        ) as span:
+            try:
+                response = await next(call)
+            except rtc.RpcError as e:
+                span.set_attribute(trace_types.ATTR_RPC_ERROR_CODE, int(e.code))
+                raise
+            span.set_attributes(_response_attributes(response))
+            return response
+
+    async def intercept_incoming(
+        self, invocation: Any, next: Callable[[Any], Awaitable[str | None]]
+    ) -> str | None:
+        attributes: dict[str, Any] = {
+            trace_types.ATTR_RPC_METHOD: getattr(invocation, "method", ""),
+            trace_types.ATTR_RPC_REQUEST_ID: invocation.request_id,
+            trace_types.ATTR_RPC_CALLER_IDENTITY: invocation.caller_identity,
+            trace_types.ATTR_RPC_RESPONSE_TIMEOUT: invocation.response_timeout,
+            trace_types.ATTR_RPC_HANDLER_REGISTERED: True,
+            **_payload_attributes(invocation.payload),
+        }
+        with tracer.start_as_current_span(
+            "rpc_handler",
+            context=session_context.session_root_context(),
+            kind=trace.SpanKind.SERVER,
+            attributes=attributes,
+        ) as span:
+            try:
+                response = await next(invocation)
+            except rtc.RpcError as e:
+                span.set_attribute(trace_types.ATTR_RPC_ERROR_CODE, int(e.code))
+                if e.code == rtc.RpcError.ErrorCode.UNSUPPORTED_METHOD:
+                    # a client called a method this agent never registered
+                    span.set_attribute(trace_types.ATTR_RPC_HANDLER_REGISTERED, False)
+                raise
+            except asyncio.CancelledError:
+                # the SDK maps a cancellation to an RpcError only after this interceptor has
+                # unwound, and a CancelledError is not an Exception, so the span would end
+                # UNSET; invocation.cancel_reason says what the caller gets (livekit>=1.1.19)
+                code, description = _cancellation_outcome(invocation)
+                if code is not None:
+                    span.set_attribute(trace_types.ATTR_RPC_ERROR_CODE, int(code))
+                span.set_status(trace.Status(trace.StatusCode.ERROR, description))
+                raise
+            span.set_attributes(_response_attributes(response))
+            return response
+
+
+_interceptor = TracingRpcInterceptor()
+
+
+def install(local_participant: Any) -> bool:
+    """Trace RPCs on ``local_participant``. Idempotent. Returns False when the installed
+    ``livekit-rtc`` has no interceptor support."""
+    global _warned_unsupported
+    add = getattr(local_participant, "add_rpc_interceptor", None)
+    if add is None:
+        if not _warned_unsupported:
+            _warned_unsupported = True
+            logger.debug(
+                "livekit-rtc has no RpcInterceptor support; RPC calls will not be traced "
+                "(requires livekit>=1.1.18)"
+            )
+        return False
+    add(_interceptor)
+    return True

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

@@ -74,6 +74,22 @@
 ATTR_OLD_STATE = "lk.old_state"
 ATTR_NEW_STATE = "lk.new_state"
 
+# rpc (``rpc.method`` from the OpenTelemetry RPC semantic conventions, plus lk.rpc.* details)
+ATTR_RPC_METHOD = "rpc.method"
+ATTR_RPC_REQUEST_ID = "lk.rpc.request_id"
+ATTR_RPC_CALLER_IDENTITY = "lk.rpc.caller_identity"
+ATTR_RPC_DESTINATION_IDENTITY = "lk.rpc.destination_identity"
+ATTR_RPC_PAYLOAD = "lk.pii.rpc.payload"
+"""Request payload, truncated to ``telemetry.rpc.MAX_PAYLOAD_ATTR_LEN`` characters."""
+ATTR_RPC_PAYLOAD_SIZE = "lk.rpc.payload_size"
+ATTR_RPC_RESPONSE = "lk.pii.rpc.response"
+"""Response payload, truncated like the request."""
+ATTR_RPC_RESPONSE_SIZE = "lk.rpc.response_size"
+ATTR_RPC_RESPONSE_TIMEOUT = "lk.rpc.response_timeout"
+ATTR_RPC_ERROR_CODE = "lk.rpc.error_code"
+ATTR_RPC_HANDLER_REGISTERED = "lk.rpc.handler_registered"
+"""False when a caller invoked a method this participant never registered."""
+
 # session close / job shutdown
 ATTR_CLOSE_REASON = "lk.close_reason"
 ATTR_CLOSE_DRAIN = "lk.close.drain"

livekit-agents/livekit/agents/voice/room_io/room_io.py

@@ -10,7 +10,7 @@
 from ... import utils
 from ...job import get_job_context
 from ...log import logger
-from ...telemetry import trace_types, tracer, utils as telemetry_utils
+from ...telemetry import rpc as rpc_tracing, trace_types, tracer, utils as telemetry_utils
 from ...types import (
     ATTRIBUTE_AGENT_STATE,
     ATTRIBUTE_PUBLISH_ON_BEHALF,
@@ -414,8 +414,12 @@ def _on_connection_state_changed(self, state: rtc.ConnectionState.ValueType) ->
             "connection_state_changed",
             {trace_types.ATTR_CONNECTION_STATE: rtc.ConnectionState.Name(state)},
         )
-        if self._room.isconnected() and not self._room_connected_fut.done():
-            self._room_connected_fut.set_result(None)
+        if self._room.isconnected():
+            # on every connect and reconnect; install is idempotent (one interceptor
+            # instance, deduped by the SDK), so JobContext.connect() installing too is fine
+            rpc_tracing.install(self._room.local_participant)
+            if not self._room_connected_fut.done():
+                self._room_connected_fut.set_result(None)
 
     def _on_participant_connected(self, participant: rtc.RemoteParticipant) -> None:
         if self._participant_available_fut.done():

tests/test_room_io.py

@@ -40,6 +40,7 @@ def __init__(self) -> None:
         self.name = "test-room"
         self._token = "test-token"
         self._server_url = "wss://test.livekit.cloud"
+        self.connected = True
 
     def on(self, event: str, callback: object) -> None:
         self._events[event].append(callback)
@@ -54,7 +55,7 @@ def listener_count(self, event: str) -> int:
         return len(self._events.get(event, []))
 
     def isconnected(self) -> bool:
-        return True
+        return self.connected
 
     def register_text_stream_handler(self, topic: str, callback: object) -> None:
         self.on(f"text:{topic}", callback)
@@ -270,6 +271,39 @@ async def test_transcription_output_strips_markup_but_keeps_links() -> None:
 
 
 @pytest.mark.asyncio
+async def test_rpc_tracing_is_installed_when_the_room_connects(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A session may start on a room that connects later (ctx.connect() after session.start(),
+    or a room the user connects). Tracing goes in on the connected transition, not only when
+    the room is already up at start(); install is idempotent so a reconnect is harmless."""
+    from livekit.agents.voice.room_io import room_io as room_io_mod
+
+    install = MagicMock(return_value=True)
+    monkeypatch.setattr(room_io_mod.rpc_tracing, "install", install)
+
+    room = _FakeRoom()
+    room.connected = False
+    agent_session = SimpleNamespace(
+        off=MagicMock(),
+        input=SimpleNamespace(audio=None, video=None),
+        output=SimpleNamespace(audio=None, transcription=None),
+    )
+    room_io = RoomIO(agent_session, room)
+
+    room_io._on_connection_state_changed(rtc.ConnectionState.CONN_DISCONNECTED)
+    install.assert_not_called()
+    assert not room_io._room_connected_fut.done()
+
+    room.connected = True
+    room_io._on_connection_state_changed(rtc.ConnectionState.CONN_CONNECTED)
+    install.assert_called_once_with(room.local_participant)
+    assert room_io._room_connected_fut.done()
+
+    room_io._on_connection_state_changed(rtc.ConnectionState.CONN_CONNECTED)  # reconnected
+    assert install.call_count == 2  # same singleton each time; the SDK dedups by identity
+
+
 async def test_roomio_aclose_unregisters_disconnect_and_closes_transcription_outputs() -> None:
     room = _FakeRoom()
     agent_session = SimpleNamespace(