fix: expose speech generation errors via SpeechHandle.exception()

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

@@ -2757,14 +2757,13 @@ async def _pipeline_reply_task_impl(
 
         def _on_llm_task_done(task: asyncio.Task[bool]) -> None:
             # Surface a genuine LLM failure (not interruption/cancellation) so it
-            # propagates through the SpeechHandle to RunResult (i.e. session.run()).
-            # RunResult._mark_done() raises ``_maybe_run_final_output`` when it is a
-            # BaseException; this also retrieves the task exception (no "never
+            # propagates through SpeechHandle.exception() and RunResult (i.e.
+            # session.run()); this also retrieves the task exception (no "never
             # retrieved" warning).
             if task.cancelled():
                 return
             if (exc := task.exception()) is not None:
-                speech_handle._maybe_run_final_output = exc
+                speech_handle._error = exc
 
         llm_task.add_done_callback(_on_llm_task_done)
 
@@ -3247,6 +3246,7 @@ async def _realtime_reply_task(
                 generation_ev = await self._rt_session.say(text)
             except llm.RealtimeError as e:
                 logger.error("failed to say text: %s", str(e))
+                speech_handle._mark_done(error=e)
                 return
 
             await self._realtime_generation_task(
@@ -3306,6 +3306,7 @@ async def _realtime_reply_task(
                     " after tool execution" if tool_reply else "",
                     str(e),
                 )
+                speech_handle._mark_done(error=e)
                 self._session._update_agent_state("listening")
                 return
 

livekit-agents/livekit/agents/voice/run_result.py

@@ -203,12 +203,10 @@ def _mark_done(self) -> None:
                 self._done_fut.set_result(None)
                 return
 
-            # propagate speech handle errors (e.g. LLM failures)
-            if self.__last_speech_handle._done_fut.done():
-                exc = self.__last_speech_handle._done_fut.exception()
-                if exc is not None:
-                    self._done_fut.set_exception(exc)
-                    return
+            # propagate speech handle errors (e.g. LLM or realtime failures)
+            if self.__last_speech_handle._error is not None:
+                self._done_fut.set_exception(self.__last_speech_handle._error)
+                return
 
             final_output = self.__last_speech_handle._maybe_run_final_output
             if not isinstance(final_output, BaseException):

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

@@ -64,6 +64,7 @@ def _on_done(_: asyncio.Future[None]) -> None:
 
         self._done_fut.add_done_callback(_on_done)
         self._maybe_run_final_output: Any = None  # kept private
+        self._error: BaseException | None = None
 
     @staticmethod
     def create(
@@ -138,6 +139,24 @@ def chat_items(self) -> list[llm.ChatItem]:
     def done(self) -> bool:
         return self._done_fut.done()
 
+    def exception(self) -> BaseException | None:
+        """Return the error that caused this speech to fail, if any.
+
+        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).
+
+        Raises:
+            asyncio.InvalidStateError: If the speech is not done yet.
+
+        Returns:
+            BaseException | None: The error the generation failed with, or None.
+        """
+        if not self._done_fut.done():
+            raise asyncio.InvalidStateError("SpeechHandle is not done yet")
+
+        return self._error
+
     def interrupt(self, *, force: bool = False) -> SpeechHandle:
         """Interrupt the current speech generation.
 
@@ -274,11 +293,12 @@ def _mark_generation_done(self) -> None:
             self._generations[-1].set_result(None)
 
     def _mark_done(self, error: BaseException | None = None) -> None:
-        with contextlib.suppress(asyncio.InvalidStateError):
+        # the error is kept out of _done_fut so awaiting the handle never raises
+        # (most handles are never awaited); it is exposed via exception() instead
+        if not self._done_fut.done():
             if error is not None:
-                self._done_fut.set_exception(error)
-            else:
-                self._done_fut.set_result(None)
+                self._error = error
+            self._done_fut.set_result(None)
 
         if self._generations:
             self._mark_generation_done()

tests/test_speech_handle_exception.py

@@ -0,0 +1,75 @@
+"""
+Tests for SpeechHandle error reporting.
+
+When a generation fails (e.g. a realtime ``generate_reply`` timeout, see
+https://github.com/livekit/agents/issues/6224), the error is recorded on the
+SpeechHandle instead of being set on its done future: awaiting a handle never
+raises (most handles are never awaited, so a stored exception would trigger
+"Future exception was never retrieved" warnings). Users inspect failures with
+``SpeechHandle.exception()`` after the handle is done, and ``session.run()``
+still raises through RunResult.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from livekit.agents.llm import RealtimeError
+from livekit.agents.voice.run_result import RunResult
+from livekit.agents.voice.speech_handle import SpeechHandle
+
+pytestmark = pytest.mark.unit
+
+
+async def test_await_does_not_raise_on_error() -> None:
+    handle = SpeechHandle.create()
+    handle._mark_done(error=RealtimeError("generate_reply timed out."))
+
+    result = await handle
+    assert result is handle
+
+    await handle.wait_for_playout()
+
+    exc = handle.exception()
+    assert isinstance(exc, RealtimeError)
+    assert str(exc) == "generate_reply timed out."
+
+
+async def test_exception_is_none_without_error() -> None:
+    handle = SpeechHandle.create()
+    handle._mark_done()
+
+    await handle
+    assert handle.exception() is None
+
+
+async def test_exception_raises_if_not_done() -> None:
+    handle = SpeechHandle.create()
+
+    with pytest.raises(asyncio.InvalidStateError):
+        handle.exception()
+
+
+async def test_error_ignored_after_done() -> None:
+    handle = SpeechHandle.create()
+    handle._mark_done(error=RealtimeError("first"))
+    # e.g. the task-level done callback marking the handle again
+    handle._mark_done()
+    handle._mark_done(error=RealtimeError("second"))
+
+    exc = handle.exception()
+    assert isinstance(exc, RealtimeError)
+    assert str(exc) == "first"
+
+
+async def test_run_result_propagates_speech_handle_error() -> None:
+    run_result = RunResult[None](output_type=None)
+    handle = SpeechHandle.create()
+    run_result._watch_handle(handle)
+
+    handle._mark_done(error=RealtimeError("generate_reply timed out."))
+
+    with pytest.raises(RealtimeError, match="generate_reply timed out"):
+        await run_result