fix(sessions): checkpoint a resumed handoff before the Session append

src/agents/run.py

@@ -1135,7 +1135,12 @@ def _mark_response_hooks_started() -> None:
                                     generated_items=generated_items,
                                     session_items=session_items,
                                 )
-                                if isinstance(turn_result.next_step, NextStepInterruption):
+                                if isinstance(
+                                    turn_result.next_step,
+                                    NextStepInterruption | NextStepHandoff,
+                                ):
+                                    # Publish before the fallible append so a retry does not
+                                    # lose guardrail results for work that already ran.
                                     run_state._tool_input_guardrail_results = [
                                         *tool_input_guardrail_results,
                                         *turn_result.tool_input_guardrail_results,

src/agents/run_internal/run_loop.py

@@ -1405,6 +1405,15 @@ async def _save_max_turns_items(
                         break
 
                     if isinstance(turn_result.next_step, NextStepHandoff):
+                        if run_state is not None:
+                            # `_accumulate_tool_guardrail_results` already folded this turn's
+                            # results in; publish them before the fallible append.
+                            run_state._tool_input_guardrail_results = list(
+                                accepted_tool_input_guardrail_results
+                            )
+                            run_state._tool_output_guardrail_results = list(
+                                accepted_tool_output_guardrail_results
+                            )
                         await _save_resumed_items(
                             list(turn_session_items),
                             turn_result.model_response.response_id,
@@ -1421,7 +1430,6 @@ async def _save_max_turns_items(
                         streamed_result._event_queue.put_nowait(
                             AgentUpdatedStreamEvent(new_agent=current_agent)
                         )
-                        run_state._current_step = NextStepRunAgain()
                         if await _wait_for_streamed_turn_events_and_stop_if_cancelled(
                             streamed_result
                         ):

src/agents/run_internal/session_persistence.py

@@ -61,7 +61,13 @@
     strip_internal_input_item_metadata,
 )
 from .oai_conversation import OpenAIServerConversationTracker
-from .run_steps import NextStepInterruption, NextStepRunAgain, ProcessedResponse, SingleStepResult
+from .run_steps import (
+    NextStepHandoff,
+    NextStepInterruption,
+    NextStepRunAgain,
+    ProcessedResponse,
+    SingleStepResult,
+)
 
 __all__ = [
     "admit_pending_input",
@@ -541,7 +547,13 @@ def update_run_state_after_resume(
     run_state._generated_items = generated_items
     if session_items is not None:
         run_state._session_items = list(session_items)
-    run_state._current_step = turn_result.next_step  # type: ignore[assignment]
+    next_step = turn_result.next_step
+    if isinstance(next_step, NextStepHandoff):
+        # The target agent is already committed, so the rest of the turn is an ordinary
+        # "run again". Normalizing here, before the fallible Session append, lets the existing
+        # pending-write checkpoint carry the handoff batch without a new persisted step type.
+        next_step = NextStepRunAgain()
+    run_state._current_step = next_step  # type: ignore[assignment]
 
 
 async def save_result_to_session(

src/agents/tracing/processors.py

@@ -85,6 +85,8 @@ def __init__(
         self.base_delay = base_delay
         self.max_delay = max_delay
         self._shutdown_event = threading.Event()
+        self._shutdown_lock = threading.Lock()
+        self._shutdown_requests = 0
 
         # Keep a client open for connection pooling across multiple export calls
         self._client = httpx2.Client(timeout=httpx2.Timeout(timeout=60, connect=5.0))
@@ -535,7 +537,24 @@ def close(self):
         self._client.close()
 
     def _request_shutdown(self) -> None:
-        self._shutdown_event.set()
+        with self._shutdown_lock:
+            self._shutdown_requests += 1
+            self._shutdown_event.set()
+
+    def _reset_shutdown(self) -> None:
+        """Drop one shutdown request, restoring retries once the last one is released.
+
+        Processors share an exporter -- `default_exporter()` hands the same instance to
+        every one of them -- so more than one can have a shutdown in flight. Retries come
+        back only when every request has been released; clearing on the first release would
+        resurrect the retry backoff of a worker still exporting under a shutdown of its own.
+        """
+        with self._shutdown_lock:
+            if self._shutdown_requests == 0:
+                return
+            self._shutdown_requests -= 1
+            if self._shutdown_requests == 0:
+                self._shutdown_event.clear()
 
 
 class BatchTraceProcessor(TracingProcessor):
@@ -580,6 +599,8 @@ def __init__(
         self._thread_start_lock = threading.Lock()
         self._export_lock = threading.Lock()
         self._shutdown_deadline: float | None = None
+        self._requested_exporter_shutdown = False
+        self._exporter_shutdown_lock = threading.Lock()
 
     def _ensure_thread_started(self) -> None:
         # Fast path without holding the lock
@@ -624,11 +645,10 @@ def shutdown(self, timeout: float | None = None):
         """
         Called when the application stops. We signal our thread to stop, then join it.
         """
-        self._shutdown_event.set()
         if timeout is not None:
-            request_exporter_shutdown = getattr(self._exporter, "_request_shutdown", None)
-            if callable(request_exporter_shutdown):
-                request_exporter_shutdown()
+            self._request_exporter_shutdown()
+
+        self._shutdown_event.set()
 
         deadline = None if timeout is None else time.monotonic() + timeout
         self._shutdown_deadline = deadline
@@ -637,12 +657,57 @@ def shutdown(self, timeout: float | None = None):
         if self._worker_thread and self._worker_thread.is_alive():
             self._worker_thread.join(timeout=timeout)
             if self._worker_thread.is_alive():
+                # The worker outlived this shutdown, so it keeps the request until it stops.
                 logger.warning(
                     "[non-fatal] Tracing: shutdown timeout reached; dropping queued traces."
                 )
+            else:
+                # The worker released the request as it exited, unless it had already
+                # exited when we made it -- in which case this is the only release.
+                self._release_exporter_shutdown()
         else:
             # No background thread: process any remaining items synchronously.
             self._export_batches(deadline=deadline)
+            self._release_exporter_shutdown()
+
+    def _request_exporter_shutdown(self) -> None:
+        """Ask the exporter to abandon its retry backoff, at most once for this processor.
+
+        Under the lock so that concurrent `shutdown` calls make a single request between
+        them: the exporter counts requests, and this processor has exactly one release to
+        balance it with.
+        """
+        with self._exporter_shutdown_lock:
+            if self._requested_exporter_shutdown:
+                return
+            request_exporter_shutdown = getattr(self._exporter, "_request_shutdown", None)
+            if not callable(request_exporter_shutdown):
+                return
+            # Flagged before the request so a worker exiting concurrently either waits here
+            # and then releases what we asked for, or finds nothing to release yet and
+            # leaves it to the caller below, which releases once the worker has stopped.
+            self._requested_exporter_shutdown = True
+            request_exporter_shutdown()
+
+    def _release_exporter_shutdown(self) -> None:
+        """Give a shutdown request we made back to the exporter, now that our worker is done.
+
+        `default_exporter()` hands the same exporter to every processor, so leaving the
+        request outstanding makes the next processor give up on the first transient failure
+        instead of retrying -- blaming a shutdown that is long over. It is released only by
+        the processor that made it, and only once that processor's worker has stopped: while
+        the worker is still exporting it needs the cancellation to keep abandoning its
+        retries, including when `shutdown` timed out and returned without it. Other
+        processors' requests are counted separately by the exporter, so releasing ours never
+        takes the cancellation away from theirs.
+        """
+        with self._exporter_shutdown_lock:
+            if not self._requested_exporter_shutdown:
+                return
+            self._requested_exporter_shutdown = False
+            reset_exporter_shutdown = getattr(self._exporter, "_reset_shutdown", None)
+            if callable(reset_exporter_shutdown):
+                reset_exporter_shutdown()
 
     def force_flush(self):
         """
@@ -667,6 +732,9 @@ def _run(self):
         # Final drain after shutdown
         self._export_batches(deadline=self._shutdown_deadline)
 
+        # This worker is done exporting, so it no longer needs the exporter cancelled.
+        self._release_exporter_shutdown()
+
     def _export_batches(self, deadline: float | None = None):
         """Drains the queue and exports in batches of up to `max_batch_size` until the queue
         is completely empty.

tests/test_run_impl_resume_paths.py

@@ -1049,3 +1049,73 @@ async def needs_ok(text: str) -> str:
             isinstance(item, ToolCallOutputItem) and item.output == "one"
             for item in result.new_step_items
         )
+
+
+async def _approved_handoff_session_state(streamed: bool):
+    """Pause on an approval-gated call that shares its response with a handoff."""
+    effects: list[int] = []
+
+    @tool(needs_approval=True)
+    async def charge(amount: int) -> str:
+        effects.append(amount)
+        return "receipt-7"
+
+    model = ScriptedModel(
+        [
+            [
+                get_function_tool_call("charge", '{"amount":7}', call_id="charge-1"),
+                get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff-1"),
+            ],
+            [get_text_message("done")],
+            [get_text_message("fresh")],
+        ]
+    )
+    delegate = Agent(name="delegate", model=model)
+    agent = Agent(name="triage", model=model, tools=[charge], handoffs=[delegate])
+    session = _FailingResumeSession()
+    paused = await _run_session_resume(agent, "charge 7 then hand off", session, streamed)
+    state = paused.to_state()
+    state.approve(state.get_interruptions()[0])
+    return agent, model, session, state, effects
+
+
+def _call_pair(items: list[TResponseInputItem], call_id: str) -> list[str]:
+    return [
+        str(item.get("type"))
+        for item in items
+        if isinstance(item, dict) and item.get("call_id") == call_id
+    ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "failing_streamed,retry_streamed", [(False, False), (False, True), (True, False), (True, True)]
+)
+@pytest.mark.parametrize("round_trip", [False, True], ids=["live", "json"])
+@pytest.mark.parametrize("failure", ["before", "after"], ids=["atomic-failure", "lost-ack"])
+async def test_resumed_handoff_session_append_is_recovered_before_next_model(
+    failing_streamed: bool, retry_streamed: bool, round_trip: bool, failure: str
+) -> None:
+    agent, model, session, state, effects = await _approved_handoff_session_state(failing_streamed)
+    session.failure = failure
+    with pytest.raises(RuntimeError) as error:
+        await _run_session_resume(agent, state, session, failing_streamed)
+    assert error.value is session.error
+    assert effects == [7]
+    assert len(model.calls) == 1
+    if round_trip:
+        state = await RunState.from_json(agent, state.to_json())
+    assert state._current_agent is not None and state._current_agent.name == "delegate"
+
+    result = await _run_session_resume(agent, state, session, retry_streamed)
+    assert result.final_output == "done"
+    assert result.last_agent.name == "delegate"
+    assert effects == [7]
+    assert len(model.calls) == 2
+    expected_pair = ["function_call", "function_call_output"]
+    stored = await session.get_items()
+    assert _call_pair(stored, "charge-1") == expected_pair
+    assert _call_pair(stored, "handoff-1") == expected_pair
+    assert _call_pair(result.to_input_list(), "charge-1") == expected_pair
+    assert _call_pair(result.to_input_list(), "handoff-1") == expected_pair
+    assert "pending_session_write" not in result.to_state().to_json()