fix: preserve completed tool outputs when a sibling fails

src/agents/exceptions.py

@@ -29,6 +29,7 @@
 
 from .util._pretty_print import pretty_print_run_error_details
 
+_TOOL_LOCAL_CANCELLATION_ATTR = "_agents_tool_local_cancellation"
 _DRAIN_STREAM_EVENTS_ATTR = "_agents_drain_queued_stream_events"
 _DATA_REDACTED_ATTR = "_agents_data_redacted"
 _DATA_REDACTED_ERROR_MESSAGE = "Error details are redacted."
@@ -41,6 +42,16 @@ class _RedactedExceptionCancellationError(asyncio.CancelledError, Exception):
     """Payload-free cancellation that remains catchable as an Exception."""
 
 
+def _mark_tool_local_cancellation(error: asyncio.CancelledError) -> None:
+    setattr(error, _TOOL_LOCAL_CANCELLATION_ATTR, True)
+
+
+def _is_tool_local_cancellation(error: BaseException) -> bool:
+    return isinstance(error, asyncio.CancelledError) and bool(
+        getattr(error, _TOOL_LOCAL_CANCELLATION_ATTR, False)
+    )
+
+
 def _mark_error_to_drain_stream_events(error: BaseException) -> None:
     setattr(error, _DRAIN_STREAM_EVENTS_ATTR, True)
 

src/agents/items.py

@@ -448,7 +448,7 @@ class ToolCallOutputItem(RunItemBase[Any]):
     """SDK-only custom data attached to this tool output.
 
     This data is not part of ``raw_item`` and is not sent back to the model when the output item is
-    replayed as input.
+    replayed as input. On a failed run, unfinished custom-data extraction may leave this unset.
     """
 
     @property

src/agents/result.py

@@ -152,6 +152,11 @@ def _populate_state_from_result(
         state._pending_input = copy.deepcopy(source_state._pending_input)
         state._pending_session_write = copy.deepcopy(source_state._pending_session_write)
         state._current_step = source_state._current_step
+        if isinstance(source_state._current_step, NextStepInterruption):
+            # Failure diagnostics may expose accepted outputs still awaiting their
+            # normal resume write. Keep the checkpoint's publication boundary so
+            # recovery does not mistake those outputs for already persisted work.
+            state._session_items = list(source_state._session_items)
         # A streamed result exists before its terminal append does, so a checkpoint taken from a
         # failed stream has to keep the fail-closed marker or it would look resumable.
         state._terminal_unrecoverable = source_state._terminal_unrecoverable
@@ -703,6 +708,8 @@ class RunResultStreaming(RunResultBase):
     _triggered_input_guardrail_result: InputGuardrailResult | None = field(default=None, repr=False)
     _output_guardrails_task: asyncio.Task[Any] | None = field(default=None, repr=False)
     _stored_exception: BaseException | None = field(default=None, repr=False)
+    _tool_error_selected: bool = field(default=False, init=False, repr=False)
+    """A selected tool error owns failure reporting while its input verdict settles."""
     _cancel_mode: Literal["none", "immediate", "after_turn"] = field(default="none", repr=False)
     _last_processed_response: ProcessedResponse | None = field(default=None, repr=False)
     """The last processed model response. This is needed for resuming from interruptions."""
@@ -1174,7 +1181,7 @@ def _check_errors(self):
         # Fetch all the completed guardrail results from the queue and raise if needed
         while not self._input_guardrail_queue.empty():
             guardrail_result = self._input_guardrail_queue.get_nowait()
-            if guardrail_result.output.tripwire_triggered:
+            if guardrail_result.output.tripwire_triggered and not self._tool_error_selected:
                 tripwire_exc = InputGuardrailTripwireTriggered(guardrail_result)
                 tripwire_exc.run_data = self._create_error_details()
                 self._stored_exception = tripwire_exc
@@ -1192,7 +1199,11 @@ def _check_errors(self):
                         run_impl_exc.run_data = self._create_error_details()
                     self._stored_exception = run_impl_exc
 
-        if self._input_guardrails_task and self._input_guardrails_task.done():
+        if (
+            not self._tool_error_selected
+            and self._input_guardrails_task
+            and self._input_guardrails_task.done()
+        ):
             if not self._input_guardrails_task.cancelled():
                 in_guard_exc = self._input_guardrails_task.exception()
                 if isinstance(in_guard_exc, Exception):

src/agents/run_internal/guardrails.py

@@ -105,6 +105,7 @@ async def run_input_guardrails_with_queue(
             isinstance(error, Exception)
             and asyncio.current_task() is streamed_result._input_guardrails_task
             and not streamed_result.is_complete
+            and not streamed_result._tool_error_selected
         ):
             if streamed_result.run_loop_task and not streamed_result.run_loop_task.done():
                 streamed_result.run_loop_task.cancel()

src/agents/run_internal/run_loop.py

@@ -42,6 +42,7 @@
     _copy_data_redacted_process_control_error,
     _detach_data_redacted_error_traceback,
     _is_error_data_redacted,
+    _is_tool_local_cancellation,
     _mark_error_data_redacted,
     _mark_error_to_drain_stream_events,
     _prepare_data_redacted_error,
@@ -188,6 +189,7 @@
     reconcile_nested_history_owned_session_item_refs,
     record_compaction_model_response,
     resume_pending_session_write,
+    resumed_failure_items,
     resumed_response_store,
     resumed_turn_items,
     rewind_session_items,
@@ -966,6 +968,23 @@ async def _finalize_streamed_interruption(
 T = TypeVar("T")
 
 
+class _ToolTaskCancellation(Exception):
+    """Carry selected tool cancellation across Task on Python versions that replace it."""
+
+    def __init__(self, error: asyncio.CancelledError) -> None:
+        super().__init__()
+        self.error = error
+
+
+async def preserve_tool_task_cancellation(awaitable: Awaitable[T]) -> T:
+    try:
+        return await awaitable
+    except asyncio.CancelledError as error:
+        if _is_tool_local_cancellation(error):
+            raise _ToolTaskCancellation(error) from None
+        raise
+
+
 async def start_streaming(
     starting_input: str | list[TResponseInputItem],
     streamed_result: RunResultStreaming,
@@ -1495,6 +1514,12 @@ async def _save_max_turns_items(
                             streamed_result._model_input_items != streamed_result.new_items
                         )
                         if not _is_error_data_redacted(exc):
+                            streamed_result.new_items = resumed_failure_items(
+                                run_state, current_agent, run_config
+                            )
+                            streamed_result._replay_from_model_input_items = (
+                                streamed_result._model_input_items != streamed_result.new_items
+                            )
                             streamed_result.tool_input_guardrail_results = list(
                                 run_state._tool_input_guardrail_results
                             )
@@ -1962,6 +1987,7 @@ def _record_max_turns_handler_output(
                 )
                 if current_turn_span is not None:
                     current_turn_span.start(mark_as_current=True)
+                partial_tool_results: list[SingleStepResult] = []
                 try:
                     if (
                         session is not None
@@ -1991,7 +2017,93 @@ def _record_max_turns_handler_output(
                         on_response_accepted=_commit_pending_server_response,
                         on_response_hooks_started=_mark_response_hooks_started,
                         run_state=run_state,
+                        on_tool_execution_error=partial_tool_results.append,
                     )
+                except (Exception, asyncio.CancelledError) as tool_error:
+                    if not partial_tool_results:
+                        raise
+                    input_task = streamed_result._input_guardrails_task
+                    if partial_tool_results and input_task is not None and not input_task.done():
+                        # The finalizer already waits for this verdict. Settle it before
+                        # publishing, without letting a late verdict replace the tool error.
+                        streamed_result._tool_error_selected = True
+                        try:
+                            await input_guardrail_tripwire_triggered_for_stream(
+                                streamed_result, ignore_cancelled=True
+                            )
+                        except Exception:
+                            pass
+                    input_accepted = input_task is None or (
+                        input_task.done()
+                        and not input_task.cancelled()
+                        and input_task.exception() is None
+                    )
+                    if (
+                        partial_tool_results
+                        and input_accepted
+                        and not any(
+                            result.output.tripwire_triggered
+                            for result in streamed_result.input_guardrail_results
+                        )
+                    ):
+                        partial_result = partial_tool_results[0]
+                        streamed_result._model_input_items.extend(partial_result.new_step_items)
+                        streamed_result.new_items.extend(partial_result.new_step_items)
+                        streamed_result.raw_responses.append(partial_result.model_response)
+                        _accumulate_tool_guardrail_results(
+                            streamed_result,
+                            partial_result,
+                            accepted_input_results=accepted_tool_input_guardrail_results,
+                            accepted_output_results=accepted_tool_output_guardrail_results,
+                        )
+                        streamed_result._tool_use_tracker_snapshot = serialize_tool_use_tracker(
+                            tool_use_tracker,
+                            starting_agent=(
+                                run_state._starting_agent
+                                if run_state is not None and run_state._starting_agent is not None
+                                else starting_agent
+                            ),
+                        )
+                        if run_state is not None:
+                            _synchronize_accepted_run_state(
+                                run_state,
+                                generated_items=streamed_result._model_input_items,
+                                session_items=streamed_result.new_items,
+                                model_responses=streamed_result.raw_responses,
+                                tool_input_guardrail_results=(
+                                    streamed_result.tool_input_guardrail_results
+                                ),
+                                tool_output_guardrail_results=(
+                                    streamed_result.tool_output_guardrail_results
+                                ),
+                                current_turn=current_turn,
+                            )
+                            run_state._current_step = NextStepRunAgain()
+                            run_state.set_tool_use_tracker_snapshot(
+                                streamed_result._tool_use_tracker_snapshot
+                            )
+                        stream_step_items_to_queue(
+                            [
+                                item
+                                for item in partial_result.new_step_items
+                                if item.type == "tool_call_output_item"
+                                # Model-provided outputs were already emitted before execution.
+                                and _stream_event_item_occurrence_key(item) is None
+                            ],
+                            streamed_result._event_queue,
+                        )
+                        _mark_error_to_drain_stream_events(tool_error)
+                        try:
+                            await _save_stream_items_without_count(
+                                partial_result.new_step_items,
+                                partial_result.model_response.response_id,
+                                current_agent.model_settings.resolve(
+                                    run_config.model_settings
+                                ).store,
+                            )
+                        except Exception:
+                            logger.warning("Failed to save completed tools after a tool error")
+                    raise
                 finally:
                     if current_turn_span is not None:
                         attach_usage_to_span(
@@ -2256,6 +2368,10 @@ def _record_max_turns_handler_output(
     except BaseException as error:
         if _is_error_data_redacted(error):
             _detach_data_redacted_error_traceback(error)
+        if _is_tool_local_cancellation(error) and streamed_result._tool_error_selected:
+            # Task.exception() cannot report cancellation. Keep the selected
+            # tool failure observable instead of ending the stream successfully.
+            streamed_result._stored_exception = error
         raise
     else:
         streamed_result.is_complete = True
@@ -2323,6 +2439,7 @@ async def run_single_turn_streamed(
     on_response_accepted: Callable[[ModelResponse, ProcessedResponse | None], bool] | None = None,
     on_response_hooks_started: Callable[[], None] | None = None,
     run_state: RunState[Any] | None = None,
+    on_tool_execution_error: Callable[[SingleStepResult], None] | None = None,
 ) -> SingleStepResult:
     """Run a single streamed turn and emit events as results arrive."""
     public_agent = bindings.public_agent
@@ -2630,6 +2747,10 @@ async def after_invocation_validation(
     async def check_input_guardrails_before_side_effects() -> None:
         await raise_if_input_guardrail_tripwire_known()
 
+    def on_tool_error_selected() -> None:
+        # The category has selected its failure, before draining native finalization.
+        streamed_result._tool_error_selected = True
+
     single_step_result = await get_single_step_result_from_response(
         bindings=bindings,
         original_input=streamed_result.input,
@@ -2647,6 +2768,8 @@ async def check_input_guardrails_before_side_effects() -> None:
         after_invocation_validation=after_invocation_validation,
         before_side_effects=check_input_guardrails_before_side_effects,
         run_state=run_state,
+        on_tool_execution_error=on_tool_execution_error,
+        on_tool_error_selected=on_tool_error_selected,
     )
 
     items_to_filter = session_items_for_turn(single_step_result)
@@ -2682,6 +2805,8 @@ async def run_single_turn(
     on_response_accepted: Callable[[ModelResponse, ProcessedResponse | None], bool] | None = None,
     on_response_hooks_started: Callable[[], None] | None = None,
     run_state: RunState[Any] | None = None,
+    on_tool_execution_error: Callable[[SingleStepResult], None] | None = None,
+    on_tool_error_selected: Callable[[], None] | None = None,
 ) -> SingleStepResult:
     """Run a single non-streaming turn of the agent loop."""
     public_agent = bindings.public_agent
@@ -2794,6 +2919,8 @@ async def after_invocation_validation(
         server_manages_conversation=server_conversation_tracker is not None,
         after_invocation_validation=after_invocation_validation,
         run_state=run_state,
+        on_tool_execution_error=on_tool_execution_error,
+        on_tool_error_selected=on_tool_error_selected,
     )
 
 

src/agents/run_internal/session_persistence.py

@@ -16,6 +16,8 @@
 
 from .. import _debug
 from .._tool_identity import get_hosted_mcp_approval_request_identity
+from .._tool_invocation import tool_output_identity
+from ..agent import Agent
 from ..exceptions import UserError
 from ..items import (
     HandoffOutputItem,
@@ -42,9 +44,14 @@
 from ..memory.openai_conversations_session import OpenAIConversationsSession
 from ..memory.session import _call_session_method, _get_session_wrapper
 from ..models.fake_id import FAKE_RESPONSES_ID
+from ..run_config import RunConfig
 from ..run_context import RunContextWrapper
 from ..run_state import RunState, _PendingSessionWrite
-from .blocked_output import _current_response_boundary, _identity_sequence_start
+from .blocked_output import (
+    _current_response_boundary,
+    _identity_sequence_start,
+    _should_defer_interrupted_session_items,
+)
 from .items import (
     _TOOL_CALL_TO_OUTPUT_TYPE,
     NestedHistoryOwnedItem,
@@ -622,6 +629,31 @@ def session_items_for_turn(turn_result: SingleStepResult) -> list[RunItem]:
     return list(items)
 
 
+def resumed_failure_items(
+    run_state: RunState, agent: Agent[Any], run_config: RunConfig
+) -> list[RunItem]:
+    """Expose accepted checkpoint outputs without marking them as already persisted."""
+    public_items = list(run_state._session_items)
+    if _should_defer_interrupted_session_items(agent, run_config):
+        return public_items
+    boundary = _current_response_boundary((), run_state._last_processed_response, run_state)
+    if not boundary.proven or boundary.generated_start is None:
+        return public_items
+    published_outputs = {
+        tool_output_identity(item.raw_item)
+        for item in public_items
+        if isinstance(item, ToolCallOutputItem)
+    }
+    return public_items + [
+        item
+        for item in run_state._generated_items[
+            boundary.generated_start + len(boundary.processed_items) :
+        ]
+        if isinstance(item, ToolCallOutputItem)
+        and tool_output_identity(item.raw_item) not in published_outputs
+    ]
+
+
 def resumed_turn_items(
     turn_result: SingleStepResult, session_items: list[RunItem], run_state: RunState
 ) -> tuple[list[RunItem], list[RunItem], list[RunItem]]: