fix(sessions): recover failed resumed Session writes before model calls

src/agents/memory/openai_responses_compaction_session.py

@@ -411,7 +411,14 @@ def _clear_deferred_compaction(self) -> None:
 
     async def add_items(self, items: list[TResponseInputItem]) -> None:
         async with self._mutation_lock:
-            await self.underlying_session.add_items(items)
+            try:
+                await self.underlying_session.add_items(items)
+            except (Exception, asyncio.CancelledError):
+                # The backend may have committed before acknowledgement failed. Re-read its
+                # authoritative history before compaction instead of retaining a stale cache.
+                self._compaction_candidate_items = None
+                self._session_items = None
+                raise
             if self._compaction_candidate_items is not None:
                 new_items = _normalize_compaction_session_items(items)
                 new_candidates = select_compaction_candidate_items(new_items)

src/agents/result.py

@@ -148,6 +148,7 @@ def _populate_state_from_result(
     if isinstance(source_state, RunState):
         state._generated_prompt_cache_key = source_state._generated_prompt_cache_key
         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
     else:
         state._generated_prompt_cache_key = getattr(result, "_generated_prompt_cache_key", None)

src/agents/run.py

@@ -139,6 +139,7 @@
     persist_session_items_for_guardrail_trip,
     prepare_input_with_session,
     reconcile_nested_history_owned_session_item_refs,
+    resume_pending_session_write,
     resumed_turn_items,
     save_result_to_session,
     save_resumed_turn_items,
@@ -634,6 +635,7 @@ async def _run_impl(
             )
             context = context_wrapper.context
 
+            await resume_pending_session_write(run_state, session, wrapper=context_wrapper)
             max_turns = run_state._max_turns
         else:
             raw_input = cast(str | list[TResponseInputItem], input)
@@ -1149,6 +1151,7 @@ def _mark_response_hooks_started() -> None:
                             ):
                                 run_state._current_turn_persisted_item_count = (
                                     await save_resumed_turn_items(
+                                        run_state=run_state,
                                         session=session,
                                         items=turn_session_items,
                                         persisted_count=(

src/agents/run_internal/run_loop.py

@@ -177,6 +177,7 @@
     persist_session_items_for_guardrail_trip,
     prepare_input_with_session,
     reconcile_nested_history_owned_session_item_refs,
+    resume_pending_session_write,
     resumed_turn_items,
     rewind_session_items,
     save_result_to_session,
@@ -392,6 +393,7 @@ async def _save_resumed_stream_items(
     ):
         return
     streamed_result._current_turn_persisted_item_count = await save_resumed_turn_items(
+        run_state=run_state,
         session=session,
         items=items,
         persisted_count=streamed_result._current_turn_persisted_item_count,
@@ -920,6 +922,12 @@ async def start_streaming(
             run_state._reasoning_item_id_policy = resolved_reasoning_item_id_policy
         streamed_result._reasoning_item_id_policy = resolved_reasoning_item_id_policy
 
+        if is_resumed_state and run_state is not None:
+            await resume_pending_session_write(run_state, session, wrapper=context_wrapper)
+            streamed_result._current_turn_persisted_item_count = (
+                run_state._current_turn_persisted_item_count
+            )
+
         if (
             conversation_id is not None
             or previous_response_id is not None

src/agents/run_internal/session_persistence.py

@@ -7,6 +7,7 @@
 
 import asyncio
 import copy
+import hashlib
 import inspect
 import json
 from collections import deque
@@ -60,7 +61,7 @@
     strip_internal_input_item_metadata,
 )
 from .oai_conversation import OpenAIServerConversationTracker
-from .run_steps import NextStepInterruption, ProcessedResponse, SingleStepResult
+from .run_steps import NextStepInterruption, NextStepRunAgain, ProcessedResponse, SingleStepResult
 
 __all__ = [
     "admit_pending_input",
@@ -73,6 +74,7 @@
     "resumed_turn_items",
     "save_result_to_session",
     "save_resumed_turn_items",
+    "resume_pending_session_write",
     "update_run_state_after_resume",
     "rewind_session_items",
     "wait_for_session_cleanup",
@@ -552,6 +554,7 @@ async def save_result_to_session(
     reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
     store: bool | None = None,
     wrapper: RunContextWrapper[Any] | None = None,
+    resumed_write_state: RunState | None = None,
 ) -> int:
     """
     Persist a turn to the session store, keeping track of what was already saved so retries
@@ -648,7 +651,20 @@ async def save_result_to_session(
             run_state._current_turn_persisted_item_count = already_persisted + saved_run_items_count
         return saved_run_items_count
 
-    await _session_add_items(session, items_to_save, wrapper=wrapper)
+    if resumed_write_state is not None:
+        if resumed_write_state._pending_session_write is not None:
+            raise UserError("Resolve the pending Session write before saving another batch")
+        resumed_write_state._pending_session_write = {
+            "session_id": session.session_id,
+            "items": copy.deepcopy(items_to_save),
+            "before": None,
+            "persisted_count": (
+                resumed_write_state._current_turn_persisted_item_count + saved_run_items_count
+            ),
+        }
+        await resume_pending_session_write(resumed_write_state, session, wrapper=wrapper)
+    else:
+        await _session_add_items(session, items_to_save, wrapper=wrapper)
 
     if run_state is not None:
         run_state._current_turn_persisted_item_count = already_persisted + saved_run_items_count
@@ -707,6 +723,7 @@ async def save_resumed_turn_items(
     reasoning_item_id_policy: ReasoningItemIdPolicy | None = None,
     store: bool | None = None,
     wrapper: RunContextWrapper[Any] | None = None,
+    run_state: RunState | None = None,
 ) -> int:
     """Persist resumed turn items and return the updated persisted count."""
     if session is None or not items:
@@ -720,10 +737,76 @@ async def save_resumed_turn_items(
         reasoning_item_id_policy=reasoning_item_id_policy,
         store=store,
         wrapper=wrapper,
+        resumed_write_state=(
+            run_state
+            if run_state is not None and isinstance(run_state._current_step, NextStepRunAgain)
+            else None
+        ),
     )
     return persisted_count + saved_count
 
 
+async def resume_pending_session_write(
+    run_state: RunState,
+    session: Session | None,
+    *,
+    wrapper: RunContextWrapper[Any] | None = None,
+) -> None:
+    """Settle a resumed output batch before allowing further model work.
+
+    The application must supply the original backend and serialize access to its history,
+    including independently restored RunState copies. Session has no distributed compare-and-swap
+    or backend identity contract. A changed tail is not repaired or searched for similar items.
+    """
+    pending = run_state._pending_session_write
+    if pending is None:
+        return
+    if run_state._session_write_in_progress:
+        raise UserError("The pending Session write is already in progress for this RunState")
+    if session is None or session.session_id != pending["session_id"]:
+        raise UserError("Resume the pending Session write with the original Session and session ID")
+
+    def digests(items: Sequence[TResponseInputItem]) -> list[str]:
+        return [
+            hashlib.sha256(
+                _fingerprint_or_repr(
+                    item, ignore_ids_for_matching=_ignore_ids_for_matching(session)
+                ).encode("utf-8")
+            ).hexdigest()
+            for item in items
+        ]
+
+    run_state._session_write_in_progress = True
+    try:
+        before = pending["before"]
+        if before is None:
+            # No append has started. Retain the batch even if this first read fails.
+            tail = await _session_get_items(
+                session, limit=len(pending["items"]) + 1, wrapper=wrapper
+            )
+            pending["before"] = digests(tail)
+            append = True
+        else:
+            expected = before + digests(pending["items"])
+            tail = await _session_get_items(session, limit=len(expected), wrapper=wrapper)
+            observed = digests(tail)
+            committed = observed == expected
+            unchanged = observed[-len(before) :] == before if before else not observed
+            if committed == unchanged:
+                raise UserError(
+                    "Cannot reconcile the pending Session write: history changed or is ambiguous. "
+                    "Repair the original Session before resuming; do not rerun the completed tool."
+                )
+            append = unchanged
+        if append:
+            # Backends may retain or transform their input; the durable checkpoint stays detached.
+            await _session_add_items(session, copy.deepcopy(pending["items"]), wrapper=wrapper)
+        run_state._current_turn_persisted_item_count = pending["persisted_count"]
+        run_state._pending_session_write = None
+    finally:
+        run_state._session_write_in_progress = False
+
+
 async def rewind_session_items(
     session: Session | None,
     items: Sequence[TResponseInputItem],

src/agents/run_state.py

@@ -165,6 +165,15 @@
 ]
 
 
+class _PendingSessionWrite(TypedDict):
+    """One canonical resumed-output append awaiting acknowledgement."""
+
+    session_id: str
+    items: list[TResponseInputItem]
+    before: list[str] | None
+    persisted_count: int
+
+
 def _default_run_state_validation_error(
     message: str,
     error_type: RunStateValidationErrorType,
@@ -216,7 +225,7 @@ def _default_run_state_validation_error(
     ),
     "1.17": (
         "Persists Docker container labels and current-response generated-item ownership across "
-        "resume flows."
+        "resume flows, including pending resumed Session writes."
     ),
 }
 SUPPORTED_SCHEMA_VERSIONS = frozenset(SCHEMA_VERSION_SUMMARIES)
@@ -757,6 +766,13 @@ class RunState(Generic[TContext, TAgent]):
     enough information to continue an interrupted run, including model responses, generated
     items, approval state, and optional server-managed conversation identifiers.
 
+    A failed Session append after resumed tool work that continues to another model call remains
+    pending across serialization.
+    Resume with the original Session backend and session ID, with exclusive access to that history.
+    Runner reconciles the exact pending batch before the next model call without rerunning the tool.
+    Changed or ambiguous history requires application repair. Independently restored snapshots must
+    not be resumed concurrently against the same Session.
+
     Context serialization is intentionally conservative:
 
     - Mapping contexts round-trip directly.
@@ -854,6 +870,12 @@ class RunState(Generic[TContext, TAgent]):
     _schema_version: str = field(default=CURRENT_SCHEMA_VERSION, repr=False)
     """Schema version the snapshot was loaded from for schema-gated resume compatibility."""
 
+    _pending_session_write: _PendingSessionWrite | None = field(default=None, repr=False)
+    """Canonical Session append that must settle before another model call."""
+
+    _session_write_in_progress: bool = field(default=False, repr=False)
+    """Live ownership guard; independent serialized copies require caller serialization."""
+
     def __init__(
         self,
         context: RunContextWrapper[TContext],
@@ -894,13 +916,17 @@ def __init__(
         self._trace_state = None
         self._sandbox = None
         self._schema_version = CURRENT_SCHEMA_VERSION
+        self._pending_session_write = None
+        self._session_write_in_progress = False
         from .agent_tool_state import get_agent_tool_state_scope
 
         self._agent_tool_state_scope_id = get_agent_tool_state_scope(context)
 
     def _copy_for_result_checkpoint(self) -> RunState[TContext, TAgent]:
         """Copy SDK-owned decision state when nesting this checkpoint in a result snapshot."""
         copied = copy.copy(self)
+        copied._pending_session_write = copy.deepcopy(self._pending_session_write)
+        copied._session_write_in_progress = False
         if self._context is None:
             return copied
         copied._context = self._context._copy_for_run_state()
@@ -1879,6 +1905,8 @@ def to_json(
             else None
         )
         result["current_turn_persisted_item_count"] = self._current_turn_persisted_item_count
+        if self._pending_session_write is not None:
+            result["pending_session_write"] = copy.deepcopy(self._pending_session_write)
         result["trace"] = self._serialize_trace_data(
             include_tracing_api_key=include_tracing_api_key
         )
@@ -4328,6 +4356,31 @@ async def _build_run_state_from_json(
     state._current_turn_persisted_item_count = state_json.get(
         "current_turn_persisted_item_count", 0
     )
+    pending_write = state_json.get("pending_session_write")
+    if pending_write is not None:
+        from .run_internal.run_steps import NextStepRunAgain
+
+        if (
+            (schema_major, schema_minor) < (1, 17)
+            or not isinstance(state._current_step, NextStepRunAgain)
+            or not isinstance(pending_write, dict)
+            or set(pending_write) != {"session_id", "items", "before", "persisted_count"}
+            or not isinstance(pending_write.get("session_id"), str)
+            or not isinstance(pending_write.get("items"), list)
+            or not pending_write["items"]
+            or not all(isinstance(item, dict) for item in pending_write["items"])
+            or (
+                pending_write.get("before") is not None
+                and (
+                    not isinstance(pending_write["before"], list)
+                    or not all(isinstance(item, str) for item in pending_write["before"])
+                )
+            )
+            or type(pending_write.get("persisted_count")) is not int
+            or pending_write["persisted_count"] < 0
+        ):
+            raise validation_error_factory("Run state pending Session write is invalid", UserError)
+        state._pending_session_write = copy.deepcopy(cast(_PendingSessionWrite, pending_write))
     serialized_policy = state_json.get("reasoning_item_id_policy")
     if serialized_policy in {"preserve", "omit"}:
         state._reasoning_item_id_policy = cast(Literal["preserve", "omit"], serialized_policy)
@@ -5591,6 +5644,7 @@ def _clone_original_input(original_input: str | list[Any]) -> str | list[Any]:
         ),
         "Run state agent not found in agent map",
         "Run state pending_input must be a list",
+        "Run state pending Session write is invalid",
         "Run state references an agent identity that is not present in the restored graph",
         (
             "RunState context was serialized from a custom type; provide context_deserializer "

tests/test_agent_runner_streamed.py

@@ -4317,6 +4317,7 @@ async def save_wrapper(
         reasoning_item_id_policy: str | None = None,
         store: bool | None = None,
         wrapper: RunContextWrapper[Any] | None = None,
+        run_state: RunState | None = None,
     ) -> int:
         observed_counts.append(persisted_count)
         result = await real_save_resumed(
@@ -4327,6 +4328,7 @@ async def save_wrapper(
             reasoning_item_id_policy=reasoning_item_id_policy,
             store=store,
             wrapper=wrapper,
+            run_state=run_state,
         )
         return int(result)