fix: expose OpenAI Realtime request IDs

livekit-agents/livekit/agents/llm/realtime.py

@@ -4,7 +4,7 @@
 import time
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterable, Awaitable
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from types import TracebackType
 from typing import Generic, Literal, TypeVar
 
@@ -45,6 +45,8 @@ class GenerationCreatedEvent:
     """True if the message was generated by the user using generate_reply()"""
     response_id: str | None = None
     """The response ID associated with this generation, used for metrics attribution"""
+    provider_request_ids: list[str] = field(default_factory=list)
+    """Additional provider correlation IDs associated with this generation."""
 
 
 class RealtimeModelError(BaseModel):
@@ -54,6 +56,7 @@ class RealtimeModelError(BaseModel):
     label: str
     error: Exception = Field(..., exclude=True)
     recoverable: bool
+    provider_request_ids: list[str] = Field(default_factory=list)
 
 
 @dataclass

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

@@ -3830,6 +3830,14 @@ async def _realtime_generation_task_impl(
                 trace_types.ATTR_GEN_AI_REQUEST_MODEL: self.llm.model,
             }
         )
+        provider_request_ids: list[str] = []
+        if generation_ev.response_id:
+            provider_request_ids.append(generation_ev.response_id)
+        for request_id in generation_ev.provider_request_ids:
+            if request_id and request_id not in provider_request_ids:
+                provider_request_ids.append(request_id)
+        if provider_request_ids:
+            current_span.set_attribute(trace_types.ATTR_PROVIDER_REQUEST_IDS, provider_request_ids)
         if self._realtime_spans is not None and generation_ev.response_id:
             self._realtime_spans[generation_ev.response_id] = current_span
 
@@ -4068,8 +4076,8 @@ def _create_assistant_message(
         ) -> llm.ChatMessage:
             assistant_metrics: llm.MetricsReport = {}
 
-            if generation_ev.response_id:
-                assistant_metrics["provider_request_ids"] = [generation_ev.response_id]
+            if provider_request_ids:
+                assistant_metrics["provider_request_ids"] = provider_request_ids
 
             if stopped_speaking_at and started_speaking_at:
                 assistant_metrics["started_speaking_at"] = started_speaking_at

livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/realtime/realtime_model.py

@@ -7,6 +7,7 @@
 import json
 import os
 import time
+import uuid
 import weakref
 from collections.abc import Iterator
 from dataclasses import dataclass, replace
@@ -862,6 +863,9 @@ class RealtimeSession(
     - openai_client_event_queued: expose the raw client events sent to the OpenAI Realtime API
     """
 
+    _openai_request_id: str | None = None
+    _openai_client_request_id: str | None = None
+
     def __init__(
         self, realtime_model: RealtimeModel, *, turn_detection_disabled: bool = False
     ) -> None:
@@ -879,6 +883,8 @@ def __init__(
         self._instructions: str | None = None
         # set on aclose; trailing server events are ignored while it's set
         self._closing = False
+        self._openai_request_id = None
+        self._openai_client_request_id = None
         self._main_atask = asyncio.create_task(self._main_task(), name="RealtimeSession._main_task")
         self.send_event(self._create_session_update_event())
 
@@ -925,6 +931,23 @@ def _reset_input_turn_state(self) -> None:
         # value cannot, because a late transcript would consume the next turn's value.
         self._input_speech_started_at: dict[str, float] = {}
 
+    def _connection_request_ids(self) -> list[str]:
+        return [
+            request_id
+            for request_id in (self._openai_request_id, self._openai_client_request_id)
+            if request_id
+        ]
+
+    def _connection_log_fields(self) -> dict[str, str | list[str]]:
+        fields: dict[str, str | list[str]] = {
+            "provider_request_ids": self._connection_request_ids()
+        }
+        if self._openai_request_id:
+            fields["request_id"] = self._openai_request_id
+        if self._openai_client_request_id:
+            fields["client_request_id"] = self._openai_client_request_id
+        return fields
+
     @utils.log_exceptions(logger=logger)
     async def _main_task(self) -> None:
         num_retries: int = 0
@@ -933,7 +956,10 @@ async def _main_task(self) -> None:
         async def _reconnect() -> None:
             logger.debug(
                 f"reconnecting to {self._realtime_model._provider_label}",
-                extra={"max_session_duration": self._opts.max_session_duration},
+                extra={
+                    "max_session_duration": self._opts.max_session_duration,
+                    **self._connection_log_fields(),
+                },
             )
 
             events: list[RealtimeClientEvent | dict[str, Any]] = []
@@ -990,7 +1016,10 @@ async def _reconnect() -> None:
             self._discarded_event_ids.clear()
             self._close_current_generation("session reconnection")
 
-            logger.debug(f"reconnected to {self._realtime_model._provider_label}")
+            logger.debug(
+                f"reconnected to {self._realtime_model._provider_label}",
+                extra=self._connection_log_fields(),
+            )
             self.emit("session_reconnected", llm.RealtimeSessionReconnectedEvent())
 
         reconnecting = False
@@ -1019,7 +1048,11 @@ async def _reconnect() -> None:
                         logger.warning(
                             f"{self._realtime_model._provider_label} connection failed, retrying in {retry_interval}s",
                             exc_info=e,
-                            extra={"attempt": num_retries, "max_retries": max_retries},
+                            extra={
+                                "attempt": num_retries,
+                                "max_retries": max_retries,
+                                **self._connection_log_fields(),
+                            },
                         )
                         await asyncio.sleep(retry_interval)
                     num_retries += 1
@@ -1042,6 +1075,8 @@ async def _reconnect() -> None:
 
     async def _create_ws_conn(self) -> aiohttp.ClientWebSocketResponse:
         headers = {"User-Agent": "LiveKit Agents"}
+        self._openai_request_id = None
+        self._openai_client_request_id = None
         if self._opts.is_azure:
             if self._opts.entra_token:
                 headers["Authorization"] = f"Bearer {self._opts.entra_token}"
@@ -1050,6 +1085,8 @@ async def _create_ws_conn(self) -> aiohttp.ClientWebSocketResponse:
                 headers["api-key"] = self._opts.api_key
         else:
             headers["Authorization"] = f"Bearer {self._opts.api_key}"
+            self._openai_client_request_id = str(uuid.uuid4())
+            headers["X-Client-Request-Id"] = self._openai_client_request_id
 
         url = process_base_url(
             self._opts.base_url,
@@ -1068,9 +1105,15 @@ async def _create_ws_conn(self) -> aiohttp.ClientWebSocketResponse:
                 self._realtime_model._ensure_http_session().ws_connect(url=url, headers=headers),
                 self._opts.conn_options.timeout,
             )
+            # aiohttp does not expose WebSocket upgrade headers publicly.
+            response = getattr(ws, "_response", None)
+            if response is not None:
+                self._openai_request_id = response.headers.get("x-request-id")
             self._report_connection_acquired(time.perf_counter() - t0)
             return ws
         except aiohttp.ClientError as e:
+            if isinstance(e, aiohttp.ClientResponseError) and e.headers:
+                self._openai_request_id = e.headers.get("x-request-id")
             raise APIConnectionError(
                 f"{self._realtime_model._provider_label} client connection error"
             ) from e
@@ -1860,6 +1903,7 @@ def _handle_response_created(self, event: ResponseCreatedEvent) -> None:
             function_stream=self._current_generation.function_ch,
             user_initiated=False,
             response_id=event.response.id,
+            provider_request_ids=self._connection_request_ids(),
         )
 
         if client_event_id and (fut := self._response_created_futures.pop(client_event_id, None)):
@@ -2298,7 +2342,7 @@ def _handle_error(self, event: RealtimeErrorEvent) -> None:
         provider_label = self._realtime_model._provider_label
         logger.error(
             f"{provider_label} returned an error: {event.error}",
-            extra={"error": event.error},
+            extra={"error": event.error, **self._connection_log_fields()},
         )
         recoverable = not _is_fatal_error(event.error)
         error = APIError(
@@ -2324,5 +2368,6 @@ def _emit_error(self, error: Exception, recoverable: bool) -> None:
                 label=self._realtime_model._label,
                 error=error,
                 recoverable=recoverable,
+                provider_request_ids=self._connection_request_ids(),
             ),
         )

tests/test_realtime/test_openai_realtime_model.py

@@ -4,7 +4,9 @@
 import logging
 from types import SimpleNamespace
 from typing import cast
+from uuid import UUID
 
+import aiohttp
 import pytest
 from openai.types.beta.realtime.session import TurnDetection as BetaTurnDetection
 from openai.types.realtime import (
@@ -15,7 +17,7 @@
 from openai.types.realtime.realtime_audio_input_turn_detection import ServerVad
 
 from livekit.agents import llm
-from livekit.agents._exceptions import APIError
+from livekit.agents._exceptions import APIConnectionError, APIError
 from livekit.agents.llm.remote_chat_context import RemoteChatContext
 from livekit.agents.utils import is_given
 from livekit.plugins.openai.realtime.realtime_model import (
@@ -27,6 +29,72 @@
 pytestmark = pytest.mark.unit
 
 
+async def test_websocket_connection_exposes_openai_request_ids() -> None:
+    captured_headers: dict[str, str] = {}
+    ws = SimpleNamespace(_response=SimpleNamespace(headers={"x-request-id": "req_server"}))
+
+    class _HTTPSession:
+        async def ws_connect(self, *, url: str, headers: dict[str, str]) -> object:
+            captured_headers.update(headers)
+            return ws
+
+    model = RealtimeModel(api_key="fake")
+    model._http_session = cast("aiohttp.ClientSession", _HTTPSession())
+    session = RealtimeSession.__new__(RealtimeSession)
+    session._realtime_model = model
+    session._opts = model._opts
+    session._report_connection_acquired = lambda _: None  # type: ignore[method-assign]
+    session._openai_request_id = None
+    session._openai_client_request_id = None
+
+    assert await session._create_ws_conn() is ws
+
+    client_request_id = captured_headers["X-Client-Request-Id"]
+    assert str(UUID(client_request_id)) == client_request_id
+    assert session._connection_request_ids() == ["req_server", client_request_id]
+
+
+async def test_failed_websocket_handshake_exposes_openai_request_ids() -> None:
+    captured_headers: dict[str, str] = {}
+
+    class _HTTPSession:
+        async def ws_connect(self, *, url: str, headers: dict[str, str]) -> object:
+            captured_headers.update(headers)
+            raise aiohttp.WSServerHandshakeError(
+                None,
+                (),
+                status=500,
+                headers={"x-request-id": "req_failed"},
+            )
+
+    model = RealtimeModel(api_key="fake")
+    model._http_session = cast("aiohttp.ClientSession", _HTTPSession())
+    session = RealtimeSession.__new__(RealtimeSession)
+    session._realtime_model = model
+    session._opts = model._opts
+
+    with pytest.raises(APIConnectionError):
+        await session._create_ws_conn()
+
+    assert session._connection_request_ids() == [
+        "req_failed",
+        captured_headers["X-Client-Request-Id"],
+    ]
+
+
+def test_realtime_error_exposes_openai_request_ids() -> None:
+    captured: list[llm.RealtimeModelError] = []
+    session = RealtimeSession.__new__(RealtimeSession)
+    session._realtime_model = SimpleNamespace(_label="openai")  # type: ignore[assignment]
+    session._openai_request_id = "req_server"
+    session._openai_client_request_id = "req_client"
+    session.emit = lambda _, event: captured.append(event)  # type: ignore[method-assign]
+
+    RealtimeSession._emit_error(session, RuntimeError("disconnected"), recoverable=True)
+
+    assert captured[0].provider_request_ids == ["req_server", "req_client"]
+
+
 def test_update_options_only_propagates_given_turn_detection() -> None:
     # RealtimeModel.update_options must not force-sync turn_detection to sessions when the
     # caller didn't set it, else a session that opted out of server-side turn detection gets
@@ -171,6 +239,7 @@ def _handle_error_session(
             _opts=SimpleNamespace(turn_detection=turn_detection),
             _chat_ctx_event_futures={},
             _response_created_futures={},
+            _connection_log_fields=lambda: {},
             _emit_error=lambda error, recoverable: capture.update(recoverable=recoverable),
         ),
     )
@@ -336,6 +405,8 @@ async def test_an_error_outliving_its_update_is_still_reported() -> None:
     session._item_create_future = {}
     session._chat_ctx_event_futures = {}
     session._response_created_futures = {}
+    session._openai_request_id = None
+    session._openai_client_request_id = None
     sent: list[ConversationItemCreateEvent] = []
     session.send_event = sent.append  # type: ignore[method-assign,assignment]
     errors: list[llm.RealtimeModelError] = []

tests/test_realtime_message_metrics.py

@@ -1,18 +1,39 @@
 import asyncio
 import time
+from collections.abc import Iterator
 
 import pytest
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
 
 from livekit import rtc
 from livekit.agents import Agent, AgentSession, llm, utils
+from livekit.agents.telemetry import set_tracer_provider, tracer
 from livekit.agents.voice.events import ConversationItemAddedEvent
 
 from .fake_realtime import FakeRealtimeModel, fake_capabilities
 
-pytestmark = pytest.mark.unit
+pytestmark = [pytest.mark.unit, pytest.mark.no_concurrent]
 
 
-async def test_realtime_response_id_is_available_on_assistant_message() -> None:
+@pytest.fixture
+def span_exporter() -> Iterator[InMemorySpanExporter]:
+    original_provider = tracer._tracer_provider
+    provider = TracerProvider()
+    exporter = InMemorySpanExporter()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    set_tracer_provider(provider)
+    try:
+        yield exporter
+    finally:
+        set_tracer_provider(original_provider)
+        provider.shutdown()
+
+
+async def test_realtime_request_ids_are_available_on_message_and_span(
+    span_exporter: InMemorySpanExporter,
+) -> None:
     model = FakeRealtimeModel(capabilities=fake_capabilities(audio_output=False))
     conversation_events: list[ConversationItemAddedEvent] = []
 
@@ -51,6 +72,7 @@ async def test_realtime_response_id_is_available_on_assistant_message() -> None:
                 function_stream=function_ch,
                 user_initiated=True,
                 response_id="provider-response-id",
+                provider_request_ids=["provider-connection-id", "client-connection-id"],
             )
         )
         await speech_handle
@@ -61,7 +83,18 @@ async def test_realtime_response_id_is_available_on_assistant_message() -> None:
         if event.item.type == "message" and event.item.role == "assistant"
     ]
     assert len(assistant_messages) == 1
-    assert assistant_messages[0].metrics["provider_request_ids"] == ["provider-response-id"]
+    expected_request_ids = [
+        "provider-response-id",
+        "provider-connection-id",
+        "client-connection-id",
+    ]
+    assert assistant_messages[0].metrics["provider_request_ids"] == expected_request_ids
+
+    agent_turn_spans = [
+        span for span in span_exporter.get_finished_spans() if span.name == "agent_turn"
+    ]
+    assert len(agent_turn_spans) == 1
+    assert agent_turn_spans[0].attributes["lk.provider_request_ids"] == tuple(expected_request_ids)
 
 
 async def _transcribed_user_messages(