fix(sessions): declare the withheld interrupted write as a held pending Session write

src/agents/result.py

@@ -46,7 +46,7 @@
     ProcessedResponse,
     QueueCompleteSentinel,
 )
-from .run_state import RunState
+from .run_state import RunState, _PendingSessionWrite
 from .stream_events import StreamEvent
 from .tool_guardrails import ToolInputGuardrailResult, ToolOutputGuardrailResult
 from .tracing import Trace
@@ -156,6 +156,9 @@ def _populate_state_from_result(
     else:
         state._generated_prompt_cache_key = getattr(result, "_generated_prompt_cache_key", None)
         state._pending_input = copy.deepcopy(getattr(result, "_pending_input_for_state", []))
+        state._pending_session_write = copy.deepcopy(
+            getattr(result, "_pending_session_write", None)
+        )
         state._current_step = getattr(result, "_current_step_for_state", None)
     state._reasoning_item_id_policy = getattr(result, "_reasoning_item_id_policy", None)
 
@@ -367,6 +370,10 @@ class RunResultBase(abc.ABC):
         default_factory=list, init=False, repr=False
     )
     """Pending input preserved when a non-streaming result is converted back to RunState."""
+    _pending_session_write: _PendingSessionWrite | None = field(
+        default=None, init=False, repr=False
+    )
+    """Held pending Session write preserved when a non-streaming result becomes a RunState."""
     _current_step_for_state: Any = field(default=None, init=False, repr=False)
     """Current step preserved when a non-streaming result is converted back to RunState."""
 

src/agents/run_internal/agent_runner_helpers.py

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-from collections.abc import Mapping
+import copy
+from collections.abc import Mapping, Sequence
 from typing import Any, cast
 
 from openai.types.responses.response_usage import OutputTokensDetails
@@ -41,7 +42,13 @@
     NextStepRunAgain,
     ProcessedResponse,
 )
-from .session_persistence import save_result_to_session, save_resumed_turn_items
+from .session_persistence import (
+    _held_items_safe_to_settle,
+    _pending_approval_call_ids,
+    final_items_cover_held_batch,
+    save_result_to_session,
+    save_resumed_turn_items,
+)
 from .tool_use_tracker import AgentToolUseTracker, serialize_tool_use_tracker
 from .turn_preparation import get_model
 
@@ -488,6 +495,10 @@ def build_interruption_result(
     if run_state is not None:
         result._current_turn_persisted_item_count = run_state._current_turn_persisted_item_count
         result._trace_state = run_state._trace_state
+        # The held pending write must survive the result checkpoint: a non-streamed
+        # caller serializes ``result.to_state()``, which has no live ``RunState`` to
+        # read the declaration from.
+        result._pending_session_write = copy.deepcopy(run_state._pending_session_write)
     result._original_input = copy_input_items(original_input)
     return result
 
@@ -582,13 +593,25 @@ async def save_final_turn_items_after_guardrails(
     reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
     store: bool | None = None,
     wrapper: RunContextWrapper[Any] | None = None,
+    held_input: Sequence[TResponseInputItem] | None = None,
 ) -> int:
-    """Persist deferred final-turn items without skipping a partially persisted resumed turn."""
-    if not session_persistence_enabled or not items:
+    """Persist deferred final-turn items without skipping a partially persisted resumed turn.
+
+    ``held_input`` is a claimed held batch that must land ahead of the final items in
+    the same append. It is safe to pass even when the rebuilt final items already
+    contain the parked response: the save deduplicates the combined batch.
+    """
+    if not session_persistence_enabled or (not items and not held_input):
         return 0
     if input_guardrails_triggered(input_guardrail_results):
         return 0
+    # Whether a held batch is being claimed at all, captured before any dedup empties
+    # it: the recovery registration below must stay armed even when the guardrail
+    # rebuild already carries the batch.
+    settling_held = bool(held_input)
     if run_state is not None and run_state._current_turn_persisted_item_count > 0:
+        # save_resumed_turn_items owns the dedup, pairing, and recovery arming; the raw
+        # held batch rides in so it can arm from its own pre-dedup view.
         run_state._current_turn_persisted_item_count = await save_resumed_turn_items(
             session=session,
             items=items,
@@ -597,17 +620,35 @@ async def save_final_turn_items_after_guardrails(
             reasoning_item_id_policy=run_state._reasoning_item_id_policy,
             store=store,
             wrapper=wrapper,
+            run_state=run_state,
+            held_input=held_input,
         )
         return run_state._current_turn_persisted_item_count
+    if held_input and final_items_cover_held_batch(items, held_input, reasoning_item_id_policy):
+        # The guardrail rebuild re-derived the whole current response, held requests
+        # included; feeding the batch again would duplicate its unkeyed companions.
+        held_input = None
+    if held_input:
+        held_input = _held_items_safe_to_settle(
+            held_input,
+            items,
+            reasoning_item_id_policy,
+            pending_call_ids=_pending_approval_call_ids(run_state),
+        )
     return await save_result_to_session(
         session,
-        [],
+        list(held_input) if held_input else [],
         list(items),
         run_state,
         response_id=response_id,
         reasoning_item_id_policy=reasoning_item_id_policy,
         store=store,
         wrapper=wrapper,
+        # A settling held batch always registers, so a crash inside this append fails
+        # closed with the batch recorded instead of silently losing it, even when the
+        # payload was deduplicated from the append.
+        resumed_write_state=run_state if settling_held else None,
+        settling_held_batch=settling_held,
     )