fix(openai): fail generate_reply fast on conversation_already_has_act…

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

@@ -90,8 +90,18 @@ class RealtimeCapabilities:
 
 
 class RealtimeError(Exception):
-    def __init__(self, message: str) -> None:
+    """Error raised by a realtime session when a request fails.
+
+    Args:
+        message: Human-readable description of the failure.
+        code: Provider error code when the failure mirrors one (e.g. OpenAI's
+            ``conversation_already_has_active_response``), so callers can branch on it
+            programmatically; ``None`` when the error carries no provider code.
+    """
+
+    def __init__(self, message: str, *, code: str | None = None) -> None:
         super().__init__(message)
+        self.code = code
 
 
 class RealtimeModel:

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

@@ -1465,6 +1465,13 @@ def generate_reply(
 
         Returns:
             SpeechHandle: A handle to the generated reply.
+
+        Note:
+            ``await handle`` waits for the reply to finish and never raises; check
+            ``handle.exception()`` for the failure instead. With a realtime model, a reply
+            that races an already-active response fails fast with an ``llm.RealtimeError``
+            whose ``code`` is ``conversation_already_has_active_response``, so callers can
+            catch it and retry rather than waiting out a timeout.
         """  # noqa: E501
         if self._activity is None:
             raise RuntimeError("AgentSession isn't running")

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

@@ -2271,16 +2271,20 @@ def _handle_response_done_but_not_complete(self, event: ResponseDoneEvent) -> No
             logger.debug("Unknown response status: %s", event.response.status)
 
     def _handle_error(self, event: RealtimeErrorEvent) -> None:
-        # a rejected item event gets no deleted/added reply, so fail its future rather than
-        # leave update_chat_ctx to stall inside the speech that awaits it
-        if (event_id := event.error.event_id) and (
-            fut := self._chat_ctx_event_futures.pop(event_id, None)
-        ):
-            if not fut.done():
-                fut.set_exception(llm.RealtimeError(event.error.message))
-            # a terminal one still has to end the session, whatever it came in reply to
-            if not _is_fatal_error(event.error):
-                return
+        if event_id := event.error.event_id:
+            # a rejected item event gets no deleted/added reply, so fail its future rather than
+            # leave update_chat_ctx to stall inside the speech that awaits it
+            if fut := self._chat_ctx_event_futures.pop(event_id, None):
+                if not fut.done():
+                    fut.set_exception(llm.RealtimeError(event.error.message))
+                # a terminal one still has to end the session, whatever it came in reply to
+                if not _is_fatal_error(event.error):
+                    return
+            # a rejected response.create gets no response.created; fail its future now
+            # instead of orphaning it until the 10s timeout (still emitted/raised below)
+            elif fut := self._response_created_futures.pop(event_id, None):
+                if not fut.done():
+                    fut.set_exception(llm.RealtimeError(event.error.message, code=event.error.code))
 
         if event.error.message.startswith("Cancellation failed"):
             return

tests/test_realtime/test_openai_realtime_model.py

@@ -170,6 +170,7 @@ def _handle_error_session(
             _realtime_model=SimpleNamespace(_provider_label="openai"),
             _opts=SimpleNamespace(turn_detection=turn_detection),
             _chat_ctx_event_futures={},
+            _response_created_futures={},
             _emit_error=lambda error, recoverable: capture.update(recoverable=recoverable),
         ),
     )
@@ -334,6 +335,7 @@ async def test_an_error_outliving_its_update_is_still_reported() -> None:
     session._item_delete_future = {}
     session._item_create_future = {}
     session._chat_ctx_event_futures = {}
+    session._response_created_futures = {}
     sent: list[ConversationItemCreateEvent] = []
     session.send_event = sent.append  # type: ignore[method-assign,assignment]
     errors: list[llm.RealtimeModelError] = []
@@ -375,3 +377,58 @@ async def test_a_fatal_error_on_a_chat_ctx_event_still_ends_the_session() -> Non
 
     assert exc_info.value.retryable is False
     assert isinstance(waiter.exception(), llm.RealtimeError)
+
+
+# --------------------------------------------------------------------------- #
+# a response.create rejected before any response.created (the conversation already
+# has an active response) must fail its generate_reply future immediately with the
+# provider code, instead of orphaning it until the 10s timeout — while still emitting
+# the error event.
+# --------------------------------------------------------------------------- #
+
+
+def _active_response_rejection(event_id: str) -> RealtimeErrorEvent:
+    return RealtimeErrorEvent.construct(
+        type="error",
+        event_id=event_id,
+        error={
+            "message": "Conversation already has an active response",
+            "type": "invalid_request_error",
+            "code": "conversation_already_has_active_response",
+            "event_id": event_id,
+        },
+    )
+
+
+def test_active_response_rejection_fails_generate_reply_future_fast() -> None:
+    captured: dict[str, object] = {}
+    session = _handle_error_session(captured)
+    fut: asyncio.Future[llm.GenerationCreatedEvent] = asyncio.Future()
+    session._response_created_futures = {"response_create_1": fut}
+
+    RealtimeSession._handle_error(session, _active_response_rejection("response_create_1"))
+
+    # settled immediately (no 10s timeout), with the typed error and provider code
+    assert fut.done()
+    err = fut.exception()
+    assert isinstance(err, llm.RealtimeError)
+    assert err.code == "conversation_already_has_active_response"
+    # the future was consumed so nothing else touches it
+    assert "response_create_1" not in session._response_created_futures
+    # both surfaces: the error is still emitted as a recoverable "error" event
+    assert captured["recoverable"] is True
+
+
+def test_error_with_unknown_event_id_leaves_generate_reply_futures_untouched() -> None:
+    # an error naming an event_id we aren't tracking must not disturb a pending future
+    captured: dict[str, object] = {}
+    session = _handle_error_session(captured)
+    fut: asyncio.Future[llm.GenerationCreatedEvent] = asyncio.Future()
+    session._response_created_futures = {"response_create_1": fut}
+
+    RealtimeSession._handle_error(session, _active_response_rejection("response_create_other"))
+
+    assert not fut.done()
+    assert session._response_created_futures == {"response_create_1": fut}
+    # still reported down the ordinary path
+    assert captured["recoverable"] is True