fix(realtime): surface pending-reply drop on reconnect via SpeechHandle.exception()

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

@@ -16,6 +16,7 @@
 from livekit.agents.metrics.base import Metadata
 
 from .. import inference, llm, stt, tts, utils, vad
+from .._exceptions import APIError
 from ..llm.chat_context import Instructions
 from ..llm.realtime_fallback_adapter import _FallbackRealtimeSession
 from ..llm.tool_context import (
@@ -3432,7 +3433,10 @@ async def _realtime_reply_task(
 
             try:
                 generation_ev = await generate_reply_fut
-            except llm.RealtimeError as e:
+            except (llm.RealtimeError, APIError) as e:
+                # RealtimeError: terminal (session closed / timeout); APIError: transient
+                # discard on reconnect (retryable=True). Both land on the SpeechHandle via
+                # exception() so the app can decide whether to re-prompt.
                 logger.error(
                     "failed to generate a reply%s: %s",
                     " after tool execution" if tool_reply else "",

livekit-agents/livekit/agents/voice/speech_handle.py

@@ -144,7 +144,9 @@ def exception(self) -> BaseException | None:
 
         Awaiting a SpeechHandle never raises; call this method after the handle
         is done to check whether the generation failed (e.g. ``llm.RealtimeError``
-        when a realtime reply timed out).
+        when the realtime session closed, or an ``APIError`` with ``retryable=True``
+        when a realtime reply was discarded on reconnect — a transient case where
+        re-prompting is sensible).
 
         Raises:
             asyncio.InvalidStateError: If the speech is not done yet.

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

@@ -883,6 +883,13 @@ def send_event(self, event: RealtimeClientEvent | dict[str, Any]) -> None:
         with contextlib.suppress(utils.aio.channel.ChanClosed):
             self._msg_ch.send_nowait(event)
 
+    def _fail_pending_response_futures(self, error: Exception) -> None:
+        """Fail and clear every pending generate_reply future with ``error``."""
+        for fut in self._response_created_futures.values():
+            if not fut.done():
+                fut.set_exception(error)
+        self._response_created_futures.clear()
+
     @utils.log_exceptions(logger=logger)
     async def _main_task(self) -> None:
         num_retries: int = 0
@@ -938,12 +945,14 @@ async def _reconnect() -> None:
                     ),
                 ) from e
 
-            for fut in self._response_created_futures.values():
-                if not fut.done():
-                    fut.set_exception(
-                        llm.RealtimeError("pending response discarded due to session reconnection")
-                    )
-            self._response_created_futures.clear()
+            # the socket dropped and reconnected; the pending reply was never created, but
+            # re-prompting is sensible once the new session is up, so surface it as a
+            # retryable APIConnectionError (retryable=True) rather than a bare RealtimeError
+            self._fail_pending_response_futures(
+                APIConnectionError(
+                    message="pending response discarded due to session reconnection",
+                )
+            )
             self._discarded_event_ids.clear()
             self._close_current_generation("session reconnection")
 
@@ -992,10 +1001,10 @@ async def _reconnect() -> None:
             # generate_reply futures so consumers don't hang and callers don't wait
             # out their timeout
             self._close_current_generation("session closed")
-            for fut in self._response_created_futures.values():
-                if not fut.done():
-                    fut.set_exception(llm.RealtimeError("realtime session closed"))
-            self._response_created_futures.clear()
+            # the session is gone for good (fatal error / retries exhausted / close); nothing
+            # to retry against, so this stays a terminal RealtimeError, not the retryable
+            # APIConnectionError used on reconnect
+            self._fail_pending_response_futures(llm.RealtimeError("realtime session closed"))
 
     async def _create_ws_conn(self) -> aiohttp.ClientWebSocketResponse:
         headers = {"User-Agent": "LiveKit Agents"}

tests/test_realtime/test_openai_realtime_model.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
+import asyncio
 from types import SimpleNamespace
 from typing import cast
 
 import pytest
 
 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.plugins.openai.realtime.realtime_model import RealtimeSession, _is_fatal_error
 
@@ -119,3 +120,39 @@ def test_response_done_failed_transient_stays_recoverable() -> None:
     )
     RealtimeSession._handle_response_done_but_not_complete(session, event)
     assert captured["recoverable"] is True
+
+
+# --------------------------------------------------------------------------- #
+# pending generate_reply futures: on a transient reconnect the plugin discards
+# them with a retryable APIConnectionError so the drop lands on the SpeechHandle
+# and the app can decide to re-prompt (isinstance(exc, APIError) and exc.retryable)
+# --------------------------------------------------------------------------- #
+
+
+async def test_fail_pending_response_futures_sets_error_and_clears() -> None:
+    pending = asyncio.get_event_loop().create_future()
+    already_done = asyncio.get_event_loop().create_future()
+    already_done.set_result(cast(llm.GenerationCreatedEvent, object()))
+
+    session = cast(
+        RealtimeSession,
+        SimpleNamespace(
+            _response_created_futures={"pending": pending, "done": already_done},
+        ),
+    )
+    error = APIConnectionError(message="pending response discarded due to session reconnection")
+    RealtimeSession._fail_pending_response_futures(session, error)
+
+    # pending future carries the error; already-done future is left untouched
+    assert pending.exception() is error
+    assert already_done.exception() is None
+    # dict is cleared so a subsequent reconnect starts fresh
+    assert session._response_created_futures == {}
+
+
+def test_reconnect_discard_error_is_retryable_api_error() -> None:
+    # the discard on reconnect must satisfy the documented app contract:
+    # isinstance(exc, APIError) and exc.retryable  ->  safe to re-prompt
+    error = APIConnectionError(message="pending response discarded due to session reconnection")
+    assert isinstance(error, APIError)
+    assert error.retryable is True

tests/test_speech_handle_exception.py

@@ -16,6 +16,7 @@
 
 import pytest
 
+from livekit.agents._exceptions import APIConnectionError, APIError
 from livekit.agents.llm import RealtimeError
 from livekit.agents.voice.run_result import RunResult
 from livekit.agents.voice.speech_handle import SpeechHandle
@@ -64,6 +65,22 @@ async def test_error_ignored_after_done() -> None:
     assert str(exc) == "first"
 
 
+async def test_retryable_api_error_round_trips_through_exception() -> None:
+    # a reply discarded on reconnect is reported as a retryable APIError;
+    # apps distinguish "safe to re-prompt" without string-matching the message
+    handle = SpeechHandle.create()
+    handle._mark_done(
+        error=APIConnectionError(message="pending response discarded due to session reconnection")
+    )
+
+    result = await handle
+    assert result is handle
+
+    exc = handle.exception()
+    assert isinstance(exc, APIError)
+    assert exc.retryable is True
+
+
 async def test_run_result_propagates_speech_handle_error() -> None:
     run_result = RunResult[None](output_type=None)
     handle = SpeechHandle.create()