fix: preserve streamed cancellation and pending guardrail cleanup

src/agents/result.py

@@ -1045,25 +1045,25 @@ def register_current_consumer() -> None:
                 registered_consumer_task.remove_done_callback(consumer_task_done)
             unregister_consumer()
             try:
-                if cancelled:
-                    # Cancellation should return promptly, so avoid waiting on long-running tasks.
-                    # Tasks have already been cancelled above.
-                    self._cleanup_tasks()
-                else:
-                    # 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()
-
-                if not cancelled:
-                    await self._await_model_provider_cleanup()
-                    await self._run_sandbox_cleanup()
+                try:
+                    if cancelled:
+                        # Queue-wait cancellation should return without awaiting long-running tasks.
+                        # Tasks have already been cancelled above.
+                        self._cleanup_tasks()
+                    else:
+                        # 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 leaves errors to _check_errors(); without this call,
+                        # a run-loop failure racing past the sentinel would be silently lost.
+                        self._check_errors()
+                        # Safely terminate background tasks after main execution has finished.
+                        self._cleanup_tasks()
+                finally:
+                    if not cancelled:
+                        await self._await_model_provider_cleanup()
+                        await self._run_sandbox_cleanup()
             finally:
                 # Allow any pending callbacks (e.g., cancellation handlers) to enqueue their
                 # completion sentinels before we clear the queues for observability.
@@ -1166,20 +1166,20 @@ def __str__(self) -> str:
         return pretty_print_run_result_streaming(self)
 
     async def _await_task_safely(self, task: asyncio.Task[Any] | None) -> None:
-        """Await a task if present, ignoring cancellation and storing exceptions elsewhere.
+        """Wait for a task, leaving its cancellation and exceptions to _check_errors().
 
-        This ensures we do not lose late guardrail exceptions while not surfacing
-        CancelledError to callers of stream_events.
+        This ensures we do not lose late guardrail exceptions while not surfacing the task's
+        CancelledError to callers of stream_events. Consumer cancellation still propagates.
         """
         if task and not task.done():
             try:
-                await task
+                # Waiting directly would conflate child cancellation with consumer cancellation.
+                await asyncio.wait((task,))
             except asyncio.CancelledError:
-                # Task was cancelled (e.g., due to result.cancel()). Nothing to do here.
-                pass
-            except Exception:
-                # The exception will be surfaced via _check_errors() if needed.
-                pass
+                task.cancel()
+                # Preserve direct-await settlement before the consumer releases owned resources.
+                await asyncio.wait((task,))
+                raise
 
     def _drain_event_queue(self) -> None:
         """Remove any pending items from the event queue and mark them done."""

src/agents/run_internal/guardrails.py

@@ -226,14 +226,23 @@ def record(result: OutputGuardrailResult) -> None:
 
 async def input_guardrail_tripwire_triggered_for_stream(
     streamed_result: RunResultStreaming,
+    *,
+    ignore_cancelled: bool = False,
 ) -> bool:
-    """Return True if any input guardrail triggered during a streamed run."""
+    """Wait for input verdicts, optionally tolerating child cancellation during final cleanup."""
     task = streamed_result._input_guardrails_task
     if task is None:
         return False
 
     if not task.done():
-        await task
+        # A cancelled child must not abort the caller's remaining run cleanup.
+        try:
+            await asyncio.wait((task,))
+        except asyncio.CancelledError:
+            task.cancel()
+            raise
+        if not ignore_cancelled or not task.cancelled():
+            task.result()
 
     return any(
         guardrail_result.output.tripwire_triggered

src/agents/run_internal/run_loop.py

@@ -2047,7 +2047,9 @@ def _record_max_turns_handler_output(
         _sync_conversation_tracking_from_tracker()
         if streamed_result._input_guardrails_task:
             try:
-                triggered = await input_guardrail_tripwire_triggered_for_stream(streamed_result)
+                triggered = await input_guardrail_tripwire_triggered_for_stream(
+                    streamed_result, ignore_cancelled=True
+                )
                 if triggered:
                     first_trigger = next(
                         (

tests/test_agent_runner_streamed.py

@@ -1580,7 +1580,7 @@ async def safe_guardrail(
 
     error = RuntimeError("SECRET_STREAM_FINALIZER_ERROR")
 
-    async def fail_finalizer(_result: Any) -> bool:
+    async def fail_finalizer(_result: Any, *, ignore_cancelled: bool = False) -> bool:
         raise error
 
     monkeypatch.setattr(