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,25 @@ def generate_reply(
 
         Returns:
             SpeechHandle: A handle to the generated reply.
+
+        Note:
+            ``await handle`` waits for the reply to finish and never raises; inspect
+            ``handle.exception()`` for the failure instead. With a realtime model, a reply
+            that races an already-active response (server-VAD created) fails fast with an
+            ``llm.RealtimeError`` whose ``code`` is ``conversation_already_has_active_response``
+            rather than stalling until a timeout. The retry policy is yours to choose::
+
+                handle = session.generate_reply(user_input="...")
+                await handle
+                err = handle.exception()
+                if isinstance(err, llm.RealtimeError) and (
+                    err.code == "conversation_already_has_active_response"
+                ):
+                    # let the in-flight response play out, then retry
+                    if session.current_speech is not None:
+                        await session.current_speech.wait_for_playout()
+                    handle = session.generate_reply(user_input="...")
+                    await handle
         """  # 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

@@ -2282,6 +2282,17 @@ def _handle_error(self, event: RealtimeErrorEvent) -> None:
             if not _is_fatal_error(event.error):
                 return
 
+        # a rejected response.create (e.g. the conversation already has an active response) draws
+        # an error, not a response.created, so nothing else settles the future generate_reply
+        # handed out. Fail it now with the provider code attached, instead of orphaning it until
+        # the 10s timeout turns it into a generic "generate_reply timed out". Fall through so the
+        # error still surfaces as an "error" event.
+        if (event_id := event.error.event_id) and (
+            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