Require non-cancelled nested run for recovery

Treat streamed nested-agent recovery as valid only when the nested run loop finished without cancellation, and update the agent-tool tests to cover both the completed and cancelled run-loop cases.

Co-authored-by: Codex <noreply@openai.com>

Author: 2 people

src/agents/agent.py

@@ -153,7 +153,7 @@ def _can_recover_cancelled_streamed_agent_tool(run_result: Any) -> bool:
     """Recover text only after the nested streamed run has already reached local terminal state."""
     run_loop_task = getattr(run_result, "run_loop_task", None)
     if isinstance(run_loop_task, asyncio.Task):
-        return run_loop_task.done()
+        return run_loop_task.done() and not run_loop_task.cancelled()
 
     return False
 

tests/test_agent_as_tool.py

@@ -1694,12 +1694,15 @@ def __init__(self) -> None:
                     response_id="resp_nested",
                 )
             ]
-            self.is_complete = True
+            self.run_loop_task = asyncio.create_task(asyncio.sleep(0))
 
         async def stream_events(self):
             yield stream_event
             raise asyncio.CancelledError("stream-cancelled")
 
+    streaming_result = DummyStreamingResult()
+    await streaming_result.run_loop_task
+
     def fake_run_streamed(
         cls,
         starting_agent,
@@ -1714,7 +1717,7 @@ def fake_run_streamed(
         conversation_id,
         session,
     ):
-        return DummyStreamingResult()
+        return streaming_result
 
     async def unexpected_run(*args: Any, **kwargs: Any) -> None:
         raise AssertionError("Runner.run should not be called when on_stream is provided.")
@@ -1838,6 +1841,86 @@ async def on_stream(payload: AgentToolStreamEvent) -> None:
             await streaming_result.run_loop_task
 
 
+@pytest.mark.asyncio
+async def test_agent_as_tool_streaming_reraises_caller_cancellation_with_cancelled_run_loop(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    agent = Agent(name="streamer")
+    stream_event = RawResponsesStreamEvent(data=cast(Any, {"type": "response_started"}))
+
+    class DummyStreamingResult:
+        def __init__(self) -> None:
+            self.final_output = ""
+            self.current_agent = agent
+            self.new_items: list[Any] = []
+            self.raw_responses = [
+                ModelResponse(
+                    output=[get_text_message("Recovered nested summary")],
+                    usage=Usage(),
+                    response_id="resp_nested",
+                )
+            ]
+            self.run_loop_task = asyncio.create_task(asyncio.sleep(60))
+
+        async def stream_events(self):
+            yield stream_event
+            raise asyncio.CancelledError("stream-cancelled")
+
+    streaming_result = DummyStreamingResult()
+    streaming_result.run_loop_task.cancel()
+    with contextlib.suppress(asyncio.CancelledError):
+        await streaming_result.run_loop_task
+
+    def fake_run_streamed(
+        cls,
+        starting_agent,
+        input,
+        *,
+        context,
+        max_turns,
+        hooks,
+        run_config,
+        previous_response_id,
+        auto_previous_response_id=False,
+        conversation_id,
+        session,
+    ):
+        return streaming_result
+
+    async def unexpected_run(*args: Any, **kwargs: Any) -> None:
+        raise AssertionError("Runner.run should not be called when on_stream is provided.")
+
+    monkeypatch.setattr(Runner, "run_streamed", classmethod(fake_run_streamed))
+    monkeypatch.setattr(Runner, "run", classmethod(unexpected_run))
+
+    async def on_stream(payload: AgentToolStreamEvent) -> None:
+        del payload
+
+    tool_call = ResponseFunctionToolCall(
+        id="call_cancelled_done",
+        arguments='{"input": "recover"}',
+        call_id="call-cancelled-done",
+        name="stream_tool",
+        type="function_call",
+    )
+
+    tool = agent.as_tool(
+        tool_name="stream_tool",
+        tool_description="Streams events",
+        on_stream=on_stream,
+    )
+
+    tool_context = ToolContext(
+        context=None,
+        tool_name="stream_tool",
+        tool_call_id=tool_call.call_id,
+        tool_arguments=tool_call.arguments,
+        tool_call=tool_call,
+    )
+    with pytest.raises(asyncio.CancelledError, match="stream-cancelled"):
+        await tool.on_invoke_tool(tool_context, '{"input": "recover"}')
+
+
 @pytest.mark.asyncio
 async def test_agent_as_tool_streaming_extractor_can_access_agent_tool_invocation(
     monkeypatch: pytest.MonkeyPatch,