fix: #2929 surface run-loop exceptions after stream_events() completes (#2931)

Author: nileshpatil6

src/agents/result.py

@@ -619,6 +619,31 @@ async def _await_run_and_cleanup() -> Any:
 
         self.run_loop_task = asyncio.create_task(_await_run_and_cleanup())
 
+    @property
+    def run_loop_exception(self) -> BaseException | None:
+        """The exception raised by the background run loop, if any.
+
+        When the run loop fails before producing stream events (for example during early
+        sandbox initialisation), the exception may not be re-raised through
+        :meth:`stream_events`. This property gives callers a reliable way to check for
+        silent failures after consuming the stream:
+
+        .. code-block:: python
+
+            result = Runner.run_streamed(agent, "hello")
+            async for event in result.stream_events():
+                pass
+            if result.run_loop_exception:
+                raise result.run_loop_exception
+
+        Returns ``None`` if the run loop completed without error, has not yet finished,
+        or was cancelled.
+        """
+        task = self.run_loop_task
+        if task is None or not task.done() or task.cancelled():
+            return None
+        return task.exception()
+
     def cancel(self, mode: Literal["immediate", "after_turn"] = "immediate") -> None:
         """Cancel the streaming run.
 
@@ -725,6 +750,11 @@ async def stream_events(self) -> AsyncIterator[StreamEvent]:
                     # Ensure main execution completes before cleanup to avoid race conditions
                     # with session operations.
                     await self._await_task_safely(self.run_loop_task)
+                    # Re-check for exceptions now that the run loop has fully settled.
+                    # _await_task_safely swallows exceptions; without this call, a run-loop
+                    # failure that races past the sentinel (e.g. early sandbox failures) would
+                    # be silently lost instead of surfaced via _stored_exception.
+                    self._check_errors()
                     # Safely terminate all background tasks after main execution has finished.
                     self._cleanup_tasks()
 

tests/test_cancel_streaming.py

@@ -230,3 +230,42 @@ async def consume_events():
     assert len(events) <= 1
     assert not block_event.is_set()
     assert result.is_complete
+
+
+@pytest.mark.asyncio
+async def test_run_loop_exception_property_is_none_on_success():
+    """run_loop_exception is None when the stream completes without error."""
+    model = FakeModel()
+    model.set_next_output([get_text_message("hello")])
+    agent = Agent(name="A", model=model)
+
+    result = Runner.run_streamed(agent, input="hi")
+    async for _ in result.stream_events():
+        pass
+
+    assert result.run_loop_exception is None
+
+
+@pytest.mark.asyncio
+async def test_run_loop_exception_surfaced_after_stream():
+    """run_loop_exception is set when the run loop raises before yielding events."""
+
+    class BoomModel(FakeModel):
+        async def get_response(self, *args, **kwargs):
+            raise RuntimeError("run loop boom")
+
+        async def stream_response(self, *args, **kwargs):
+            raise RuntimeError("run loop boom")
+            yield  # make this an async generator
+
+    agent = Agent(name="A", model=BoomModel())
+
+    result = Runner.run_streamed(agent, input="hi")
+    with pytest.raises(RuntimeError, match="run loop boom"):
+        async for _ in result.stream_events():
+            pass
+
+    # Property must also expose the exception for callers who want to inspect it directly.
+    assert result.run_loop_exception is not None
+    assert isinstance(result.run_loop_exception, RuntimeError)
+    assert "run loop boom" in str(result.run_loop_exception)