From 09e928a53cb0f95c1cb69fb307971a35f28452ea Mon Sep 17 00:00:00 2001
From: Praveen Mittal <pkmittal28@gmail.com>
Date: Wed, 2 Sep 2026 19:30:26 +0200
Subject: [PATCH 1/5] fix(sessions): recover fresh streamed handoffs after
 session append failures

start_streaming()'s generic-loop NextStepHandoff branch awaited the fallible
session append (_save_stream_items_without_count) before updating
current_agent, run_state._current_agent, streamed_result.current_agent, and
run_state._current_step. If that append raised (e.g. a transient session
backend error), the run failed with those fields still pointing at the
pre-handoff agent, even though the handoff had already fully executed.
Resuming from result.to_state() after such a failure then re-invoked the
wrong agent with input that already contained its own handoff call/output.

This is the same defect PR #4725 fixed in the sibling is_resumed_state
branch (used when resuming an interrupted run) by moving the state updates
ahead of the fallible save. This applies the same reordering to the generic
branch, which every fresh streamed run's handoffs go through, not just
resumed ones.
---
 src/agents/run_internal/run_loop.py | 12 +++----
 tests/test_run_impl_resume_paths.py | 56 +++++++++++++++++++++++++++++
 2 files changed, 62 insertions(+), 6 deletions(-)

diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py
index 3c7d9ee586..ade09900d2 100644
--- a/src/agents/run_internal/run_loop.py
+++ b/src/agents/run_internal/run_loop.py
@@ -1879,23 +1879,23 @@ def _record_max_turns_handler_output(
                     server_conversation_tracker.track_server_items(turn_result.model_response)
 
                 if isinstance(turn_result.next_step, NextStepHandoff):
+                    current_agent = turn_result.next_step.new_agent
+                    if run_state is not None:
+                        run_state._current_agent = current_agent
+                    _publish_streamed_result_agent(streamed_result, current_agent)
+                    if streamed_result._state is not None:
+                        streamed_result._state._current_step = NextStepRunAgain()
                     await _save_stream_items_without_count(
                         turn_session_items,
                         turn_result.model_response.response_id,
                         store_setting,
                     )
-                    current_agent = turn_result.next_step.new_agent
-                    if run_state is not None:
-                        run_state._current_agent = current_agent
-                    _publish_streamed_result_agent(streamed_result, current_agent)
                     current_span.finish(reset_current=True)
                     current_span = None
                     should_run_agent_start_hooks = True
                     streamed_result._event_queue.put_nowait(
                         AgentUpdatedStreamEvent(new_agent=current_agent)
                     )
-                    if streamed_result._state is not None:
-                        streamed_result._state._current_step = NextStepRunAgain()
 
                     if await _wait_for_streamed_turn_events_and_stop_if_cancelled(streamed_result):
                         break
diff --git a/tests/test_run_impl_resume_paths.py b/tests/test_run_impl_resume_paths.py
index 8e6211ceab..1707779cad 100644
--- a/tests/test_run_impl_resume_paths.py
+++ b/tests/test_run_impl_resume_paths.py
@@ -82,6 +82,25 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
             raise self.error
 
 
+class _FailSecondAddItemsSession(SimpleListSession):
+    """Let the initial input-priming append succeed, then fail the next append.
+
+    Unlike ``_FailingResumeSession``, this targets a specific append by call order rather than
+    a resume-cycle phase, so it can isolate a fresh (non-resumed) run's first real turn save.
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.error = RuntimeError("session append failed")
+        self._call_count = 0
+
+    async def add_items(self, items: list[TResponseInputItem]) -> None:
+        self._call_count += 1
+        if self._call_count == 2:
+            raise self.error
+        await super().add_items(items)
+
+
 class _LostAckSQLiteSession(SQLiteSession):
     fail_after_commit = False
     error = RuntimeError("session append failed")
@@ -1221,3 +1240,40 @@ async def test_resumed_handoff_session_append_is_recovered_before_next_model(
     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()
+
+
+@pytest.mark.asyncio
+async def test_fresh_streamed_handoff_preserves_agent_after_session_append_failure() -> None:
+    """A fresh (non-resumed) streamed run's generic-loop handoff branch must publish the new
+    agent and next-step state before the fallible session append, mirroring the fix already
+    applied to the is_resumed_state branch covered by
+    test_resumed_handoff_session_append_is_recovered_before_next_model. Every fresh streamed
+    run passes through this branch, not just resumed ones.
+    """
+    model = ScriptedModel(
+        [
+            [get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff-1")],
+            [get_text_message("done")],
+        ]
+    )
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(name="triage", model=model, handoffs=[delegate])
+    session = _FailSecondAddItemsSession()
+
+    failed_result = Runner.run_streamed(
+        triage, "hello", session=session, run_config=RunConfig(tracing_disabled=True)
+    )
+    with pytest.raises(RuntimeError) as error:
+        async for _ in failed_result.stream_events():
+            pass
+    assert error.value is session.error
+
+    state = failed_result.to_state()
+    assert state._current_agent is not None
+    assert state._current_agent.name == "delegate"
+    assert failed_result.current_agent.name == "delegate"
+
+    result = await _run_session_resume(triage, state, session, False)
+    assert result.final_output == "done"
+    assert result.last_agent.name == "delegate"
+    assert len(model.calls) == 2

From 75fc64bae50bdcf90136fa066f6e78b263f0a110 Mon Sep 17 00:00:00 2001
From: Praveen Mittal <pkmittal28@gmail.com>
Date: Wed, 2 Sep 2026 23:14:39 +0200
Subject: [PATCH 2/5] address Codex review: checkpoint the handoff batch and
 reorder the agent-update event

Two issues from the automated Codex review on this PR, both real:

1. _save_stream_items_without_count() never registered a pending_session_write
   checkpoint for the handoff batch, unlike the sibling is_resumed_state branch's
   _save_resumed_stream_items(). A failed append followed by a successful resume
   invoked the correct (delegate) agent but permanently dropped the handoff's
   function_call/function_call_output pair from session history, since nothing
   recorded the batch for the existing resume_pending_session_write() recovery
   path to replay. Fixed by threading resumed_write_state through
   _save_stream_items into save_result_to_session, gated on the handoff branch
   already having set _current_step to NextStepRunAgain.

2. AgentUpdatedStreamEvent was still queued after the fallible session append,
   so a live stream_events() consumer would see handoff items followed directly
   by an error with no semantic agent-transition event, even though the result
   and resumed run both correctly identify the new agent. Moved the event queue
   call to sit with the other state-transition updates, before the append.

Both fixes are scoped to only the generic-loop branch this PR already touches;
the already-merged is_resumed_state branch (#4725) has the same pre-existing
event-ordering gap but is out of scope here.

Extended test_fresh_streamed_handoff_preserves_agent_after_session_append_failure
with a session-history assertion for issue 1, and added
test_fresh_streamed_handoff_publishes_agent_update_before_session_append_failure
for issue 2 (using a new session double that yields before failing, since a
purely synchronous raise never gives stream_events() a scheduling boundary to
prove event delivery either way).
---
 src/agents/run_internal/run_loop.py | 16 ++++++--
 tests/test_run_impl_resume_paths.py | 63 ++++++++++++++++++++++++++++-
 2 files changed, 75 insertions(+), 4 deletions(-)

diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py
index ade09900d2..f740570842 100644
--- a/src/agents/run_internal/run_loop.py
+++ b/src/agents/run_internal/run_loop.py
@@ -418,6 +418,7 @@ async def _save_stream_items(
     response_id: str | None,
     update_persisted_count: bool,
     store: bool | None = None,
+    resumed_write_state: RunState | None = None,
 ) -> None:
     if not await _should_persist_stream_items(
         session=session,
@@ -433,6 +434,7 @@ async def _save_stream_items(
         response_id=response_id,
         store=store,
         wrapper=streamed_result.context_wrapper,
+        resumed_write_state=resumed_write_state,
     )
     if update_persisted_count and streamed_result._state is not None:
         streamed_result._current_turn_persisted_item_count = (
@@ -1152,6 +1154,12 @@ async def _save_stream_items_without_count(
                 response_id=response_id,
                 update_persisted_count=False,
                 store=store_setting,
+                resumed_write_state=(
+                    run_state
+                    if run_state is not None
+                    and isinstance(run_state._current_step, NextStepRunAgain)
+                    else None
+                ),
             )
 
         async def _save_max_turns_items(
@@ -1885,6 +1893,11 @@ def _record_max_turns_handler_output(
                     _publish_streamed_result_agent(streamed_result, current_agent)
                     if streamed_result._state is not None:
                         streamed_result._state._current_step = NextStepRunAgain()
+                    # Queue the agent-transition event before the fallible session append so
+                    # stream consumers observe the transition even if the append later raises.
+                    streamed_result._event_queue.put_nowait(
+                        AgentUpdatedStreamEvent(new_agent=current_agent)
+                    )
                     await _save_stream_items_without_count(
                         turn_session_items,
                         turn_result.model_response.response_id,
@@ -1893,9 +1906,6 @@ def _record_max_turns_handler_output(
                     current_span.finish(reset_current=True)
                     current_span = None
                     should_run_agent_start_hooks = True
-                    streamed_result._event_queue.put_nowait(
-                        AgentUpdatedStreamEvent(new_agent=current_agent)
-                    )
 
                     if await _wait_for_streamed_turn_events_and_stop_if_cancelled(streamed_result):
                         break
diff --git a/tests/test_run_impl_resume_paths.py b/tests/test_run_impl_resume_paths.py
index 1707779cad..3f34741bf6 100644
--- a/tests/test_run_impl_resume_paths.py
+++ b/tests/test_run_impl_resume_paths.py
@@ -9,7 +9,7 @@
 from openai.types.responses import ResponseFunctionToolCall, ResponseOutputMessage
 
 import agents.run as run_module
-from agents import Agent, Runner, function_tool, handoff
+from agents import Agent, AgentUpdatedStreamEvent, Runner, function_tool, handoff
 from agents.agent import ToolsToFinalOutputResult
 from agents.agent_output import AgentOutputSchema
 from agents.decorators import tool, tool_input_guardrail, tool_output_guardrail
@@ -101,6 +101,20 @@ async def add_items(self, items: list[TResponseInputItem]) -> None:
         await super().add_items(items)
 
 
+class _FailSecondAddItemsSessionWithYield(_FailSecondAddItemsSession):
+    """Same failure shape as ``_FailSecondAddItemsSession``, but the failing call performs a
+    genuine ``await`` (a scheduler yield) before raising, like a real I/O-backed Session
+    (SQLite, network, etc.) would. A purely synchronous raise never yields control back to the
+    ``stream_events()`` consumer before the run-loop task finishes, so a test built on it cannot
+    observe whether an already-queued stream event was delivered before the error surfaced.
+    """
+
+    async def add_items(self, items: list[TResponseInputItem]) -> None:
+        if self._call_count == 1:
+            await asyncio.sleep(0)
+        await super().add_items(items)
+
+
 class _LostAckSQLiteSession(SQLiteSession):
     fail_after_commit = False
     error = RuntimeError("session append failed")
@@ -1277,3 +1291,50 @@ async def test_fresh_streamed_handoff_preserves_agent_after_session_append_failu
     assert result.final_output == "done"
     assert result.last_agent.name == "delegate"
     assert len(model.calls) == 2
+    expected_pair = ["function_call", "function_call_output"]
+    stored = await session.get_items()
+    assert _call_pair(stored, "handoff-1") == expected_pair
+    assert "pending_session_write" not in result.to_state().to_json()
+
+
+@pytest.mark.asyncio
+async def test_fresh_streamed_handoff_publishes_agent_update_before_session_append_failure() -> (
+    None
+):
+    """The generic-loop handoff branch must queue ``AgentUpdatedStreamEvent`` for the new agent
+    before the fallible session append, so ``stream_events()`` consumers observe the transition
+    even when the append later raises. Mirrors the already-merged ordering fix for the
+    ``is_resumed_state``-specific branch (lines ~1407-1437), which is out of scope here.
+
+    Uses a session whose failing append performs a genuine ``await`` before raising: a purely
+    synchronous raise (as in ``_FailSecondAddItemsSession``) never yields control back to this
+    consumer before the run-loop task finishes, so it cannot prove event delivery either way
+    (a separate, pre-existing gate: ``stream_events()`` only drains an already-queued event past
+    a terminal error when that error was marked via ``_mark_error_to_drain_stream_events()``,
+    which session-append failures never are).
+    """
+    model = ScriptedModel(
+        [
+            [get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff-1")],
+            [get_text_message("done")],
+        ]
+    )
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(name="triage", model=model, handoffs=[delegate])
+    session = _FailSecondAddItemsSessionWithYield()
+
+    failed_result = Runner.run_streamed(
+        triage, "hello", session=session, run_config=RunConfig(tracing_disabled=True)
+    )
+    collected_events: list[Any] = []
+    caught: RuntimeError | None = None
+    try:
+        async for event in failed_result.stream_events():
+            collected_events.append(event)
+    except RuntimeError as error:
+        caught = error
+    assert caught is session.error
+    assert any(
+        isinstance(event, AgentUpdatedStreamEvent) and event.new_agent.name == "delegate"
+        for event in collected_events
+    )

From 92891a6ea94b696c186f1f33dc3a1d169c519330 Mon Sep 17 00:00:00 2001
From: Praveen Mittal <pkmittal28@gmail.com>
Date: Thu, 3 Sep 2026 00:29:59 +0200
Subject: [PATCH 3/5] address 2nd Codex review round: guardrail race, event
 drain, deferred compaction

Three more issues from the automated Codex review on commit 75fc64b, all
verified with live reproduction before being fixed:

1. The handoff transition committed current_agent/run_state before a
   still-in-flight parallel input guardrail had resolved. A non-tripwire
   exception from that guardrail then left the resumable state pointing at
   the delegate agent, even though the starting agent's input guardrails
   never definitively cleared. Fixed by explicitly awaiting
   input_guardrail_tripwire_triggered_for_stream() as the first statement
   in the handoff branch, before any state mutation.

2. Queuing AgentUpdatedStreamEvent before the fallible session append
   doesn't guarantee delivery: stream_events() checks a stored exception
   before draining the queue, so a real (non-instant) consumer can lose an
   already-queued event to a task that raised without ever being marked for
   draining. Fixed by marking the session-persistence exception via
   _mark_error_to_drain_stream_events() before re-raising, the same pattern
   already used for model-behavior errors.

3. The pending_session_write checkpoint recovers the raw item append but
   never carried enough information (response_id, store, whether the batch
   had local tool outputs) for a later, separate resume to replay the same
   post-write Responses compaction decision save_result_to_session would
   have applied inline. Extracted the compaction decision into a shared
   _apply_post_write_compaction() helper, extended the checkpoint schema
   with those fields (optional, so an old-shaped serialized RunState still
   round-trips), and call the helper from resume_pending_session_write()
   once a checkpoint settles -- whether inline or on a separate resume --
   instead of duplicating the call at both sites.

Added 3 new regression tests to tests/test_run_impl_resume_paths.py
(72 total in the file, up from 69), each confirmed to fail against the
pre-fix code and pass after. Full verification stack clean: make format/
lint/typecheck, and the full suite (9372 passed, 33 skipped, 0 failed).
---
 src/agents/run_internal/run_loop.py           |  20 +-
 .../run_internal/session_persistence.py       | 135 ++++++++++----
 src/agents/run_state.py                       |  33 +++-
 tests/test_run_impl_resume_paths.py           | 174 ++++++++++++++++++
 4 files changed, 315 insertions(+), 47 deletions(-)

diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py
index f740570842..83f497db92 100644
--- a/src/agents/run_internal/run_loop.py
+++ b/src/agents/run_internal/run_loop.py
@@ -43,6 +43,7 @@
     _detach_data_redacted_error_traceback,
     _is_error_data_redacted,
     _mark_error_data_redacted,
+    _mark_error_to_drain_stream_events,
     _prepare_data_redacted_error,
 )
 from ..guardrail import OutputGuardrailResult
@@ -1887,6 +1888,11 @@ def _record_max_turns_handler_output(
                     server_conversation_tracker.track_server_items(turn_result.model_response)
 
                 if isinstance(turn_result.next_step, NextStepHandoff):
+                    # Resolve any still-in-flight parallel input guardrail before committing the
+                    # handoff transition, so a tripwire or guardrail exception is surfaced instead
+                    # of the state (current_agent, run_state, published events) racing ahead of an
+                    # input guardrail that was still validating the original input.
+                    await input_guardrail_tripwire_triggered_for_stream(streamed_result)
                     current_agent = turn_result.next_step.new_agent
                     if run_state is not None:
                         run_state._current_agent = current_agent
@@ -1898,11 +1904,15 @@ def _record_max_turns_handler_output(
                     streamed_result._event_queue.put_nowait(
                         AgentUpdatedStreamEvent(new_agent=current_agent)
                     )
-                    await _save_stream_items_without_count(
-                        turn_session_items,
-                        turn_result.model_response.response_id,
-                        store_setting,
-                    )
+                    try:
+                        await _save_stream_items_without_count(
+                            turn_session_items,
+                            turn_result.model_response.response_id,
+                            store_setting,
+                        )
+                    except BaseException as session_persistence_error:
+                        _mark_error_to_drain_stream_events(session_persistence_error)
+                        raise
                     current_span.finish(reset_current=True)
                     current_span = None
                     should_run_agent_start_hooks = True
diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py
index 2e53667e7e..e695a431c7 100644
--- a/src/agents/run_internal/session_persistence.py
+++ b/src/agents/run_internal/session_persistence.py
@@ -556,6 +556,65 @@ def update_run_state_after_resume(
     run_state._current_step = next_step  # type: ignore[assignment]
 
 
+async def _apply_post_write_compaction(
+    session: Session,
+    *,
+    response_id: str | None,
+    store: bool | None,
+    has_local_tool_outputs: bool,
+    wrapper: RunContextWrapper[Any] | None = None,
+) -> None:
+    """Evaluate deferred/forced Responses compaction for a settled session append.
+
+    Shared by the immediate-write path in ``save_result_to_session`` and the checkpoint
+    replay path in ``resume_pending_session_write``, so a batch that only settles later
+    (via a separate resume) still gets the same compaction decision it would have gotten
+    had the original append succeeded inline. ``wrapper`` is the caller's raw (pre-gating)
+    context wrapper; it is used as-is for ``run_compaction`` and re-gated here for
+    ``_defer_compaction``, mirroring the two call sites this helper replaces.
+    """
+    if not response_id or not is_openai_responses_compaction_aware_session(session):
+        return
+
+    if has_local_tool_outputs:
+        defer_compaction = getattr(session, "_defer_compaction", None)
+        if callable(defer_compaction):
+            await _call_session_method(
+                defer_compaction,
+                response_id,
+                store=store,
+                wrapper=_get_session_wrapper(session, wrapper),
+            )
+        logger.debug(
+            "skip: deferring compaction for response %s due to local tool outputs",
+            response_id,
+        )
+        return
+
+    deferred_response_id = None
+    get_deferred = getattr(session, "_get_deferred_compaction_response_id", None)
+    if callable(get_deferred):
+        deferred_response_id = get_deferred()
+    force_compaction = deferred_response_id is not None
+    if force_compaction:
+        logger.debug(
+            "compact: forcing for response %s after deferred %s",
+            response_id,
+            deferred_response_id,
+        )
+    compaction_args: OpenAIResponsesCompactionArgs = {
+        "response_id": response_id,
+        "force": force_compaction,
+    }
+    if store is not None:
+        compaction_args["store"] = store
+    await _call_session_method(
+        session.run_compaction,
+        compaction_args,
+        wrapper=wrapper,
+    )
+
+
 async def save_result_to_session(
     session: Session | None,
     original_input: str | list[TResponseInputItem],
@@ -663,6 +722,10 @@ async def save_result_to_session(
             run_state._current_turn_persisted_item_count = already_persisted + saved_run_items_count
         return saved_run_items_count
 
+    has_local_tool_outputs = any(
+        isinstance(item, ToolCallOutputItem | HandoffOutputItem) for item in new_items
+    )
+
     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")
@@ -673,53 +736,31 @@ async def save_result_to_session(
             "persisted_count": (
                 resumed_write_state._current_turn_persisted_item_count + saved_run_items_count
             ),
+            "response_id": response_id,
+            "store": store,
+            "has_local_tool_outputs": has_local_tool_outputs,
         }
-        await resume_pending_session_write(resumed_write_state, session, wrapper=wrapper)
+        # resume_pending_session_write() applies post-write compaction itself once the
+        # checkpoint settles, whether that happens inline below or on a later, separate
+        # resume -- so it is not repeated after this call returns.
+        await resume_pending_session_write(
+            resumed_write_state,
+            session,
+            wrapper=wrapper,
+            compaction_wrapper=compaction_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
 
-    if response_id and is_openai_responses_compaction_aware_session(session):
-        has_local_tool_outputs = any(
-            isinstance(item, ToolCallOutputItem | HandoffOutputItem) for item in new_items
-        )
-        if has_local_tool_outputs:
-            defer_compaction = getattr(session, "_defer_compaction", None)
-            if callable(defer_compaction):
-                await _call_session_method(
-                    defer_compaction,
-                    response_id,
-                    store=store,
-                    wrapper=wrapper,
-                )
-            logger.debug(
-                "skip: deferring compaction for response %s due to local tool outputs",
-                response_id,
-            )
-            return saved_run_items_count
-
-        deferred_response_id = None
-        get_deferred = getattr(session, "_get_deferred_compaction_response_id", None)
-        if callable(get_deferred):
-            deferred_response_id = get_deferred()
-        force_compaction = deferred_response_id is not None
-        if force_compaction:
-            logger.debug(
-                "compact: forcing for response %s after deferred %s",
-                response_id,
-                deferred_response_id,
-            )
-        compaction_args: OpenAIResponsesCompactionArgs = {
-            "response_id": response_id,
-            "force": force_compaction,
-        }
-        if store is not None:
-            compaction_args["store"] = store
-        await _call_session_method(
-            session.run_compaction,
-            compaction_args,
+    if resumed_write_state is None:
+        await _apply_post_write_compaction(
+            session,
+            response_id=response_id,
+            store=store,
+            has_local_tool_outputs=has_local_tool_outputs,
             wrapper=compaction_wrapper,
         )
 
@@ -764,12 +805,18 @@ async def resume_pending_session_write(
     session: Session | None,
     *,
     wrapper: RunContextWrapper[Any] | None = None,
+    compaction_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.
+
+    ``compaction_wrapper`` defaults to ``wrapper`` when omitted; ``save_result_to_session``
+    passes its own raw (pre-gating) wrapper explicitly so a batch that settles here -- either
+    inline or on a later, separate resume -- gets the exact same post-write Responses
+    compaction decision ``save_result_to_session`` would otherwise have applied itself.
     """
     pending = run_state._pending_session_write
     if pending is None:
@@ -819,6 +866,14 @@ def digests(items: Sequence[TResponseInputItem]) -> list[str]:
     finally:
         run_state._session_write_in_progress = False
 
+    await _apply_post_write_compaction(
+        session,
+        response_id=pending.get("response_id"),
+        store=pending.get("store"),
+        has_local_tool_outputs=pending.get("has_local_tool_outputs", False),
+        wrapper=compaction_wrapper if compaction_wrapper is not None else wrapper,
+    )
+
 
 async def rewind_session_items(
     session: Session | None,
diff --git a/src/agents/run_state.py b/src/agents/run_state.py
index 228dd574d8..819eee321d 100644
--- a/src/agents/run_state.py
+++ b/src/agents/run_state.py
@@ -38,7 +38,7 @@
     ProgramOutput,
 )
 from pydantic import BaseModel, StringConstraints, TypeAdapter, ValidationError
-from typing_extensions import TypedDict, TypeVar
+from typing_extensions import NotRequired, TypedDict, TypeVar
 
 from ._tool_identity import (
     FunctionToolLookupKey,
@@ -172,6 +172,14 @@ class _PendingSessionWrite(TypedDict):
     items: list[TResponseInputItem]
     before: list[str] | None
     persisted_count: int
+    # Compaction inputs for the batch this checkpoint is settling, so a later, separate
+    # resume_pending_session_write() call (not the original save_result_to_session() call)
+    # can still apply the same post-write Responses compaction decision. Optional so a
+    # RunState serialized before these fields existed degrades to "skip compaction" on
+    # read instead of raising KeyError.
+    response_id: NotRequired[str | None]
+    store: NotRequired[bool | None]
+    has_local_tool_outputs: NotRequired[bool]
 
 
 def _default_run_state_validation_error(
@@ -4360,11 +4368,18 @@ async def _build_run_state_from_json(
     if pending_write is not None:
         from .run_internal.run_steps import NextStepInterruption, NextStepRunAgain
 
+        required_pending_write_keys = {"session_id", "items", "before", "persisted_count"}
+        # response_id/store/has_local_tool_outputs carry the compaction inputs needed to replay
+        # deferred/forced Responses compaction on a later, separate resume; they are optional so
+        # a RunState serialized before these fields existed (same, unreleased schema version)
+        # still round-trips.
+        optional_pending_write_keys = {"response_id", "store", "has_local_tool_outputs"}
         if (
             (schema_major, schema_minor) < (1, 17)
             or not isinstance(state._current_step, NextStepRunAgain | NextStepInterruption)
             or not isinstance(pending_write, dict)
-            or set(pending_write) != {"session_id", "items", "before", "persisted_count"}
+            or not required_pending_write_keys <= set(pending_write)
+            or not set(pending_write) <= required_pending_write_keys | optional_pending_write_keys
             or not isinstance(pending_write.get("session_id"), str)
             or not isinstance(pending_write.get("items"), list)
             or not pending_write["items"]
@@ -4378,6 +4393,20 @@ async def _build_run_state_from_json(
             )
             or type(pending_write.get("persisted_count")) is not int
             or pending_write["persisted_count"] < 0
+            or (
+                "response_id" in pending_write
+                and pending_write["response_id"] is not None
+                and not isinstance(pending_write["response_id"], str)
+            )
+            or (
+                "store" in pending_write
+                and pending_write["store"] is not None
+                and not isinstance(pending_write["store"], bool)
+            )
+            or (
+                "has_local_tool_outputs" in pending_write
+                and not isinstance(pending_write["has_local_tool_outputs"], bool)
+            )
         ):
             raise validation_error_factory("Run state pending Session write is invalid", UserError)
         state._pending_session_write = copy.deepcopy(cast(_PendingSessionWrite, pending_write))
diff --git a/tests/test_run_impl_resume_paths.py b/tests/test_run_impl_resume_paths.py
index 3f34741bf6..8d7117e393 100644
--- a/tests/test_run_impl_resume_paths.py
+++ b/tests/test_run_impl_resume_paths.py
@@ -14,6 +14,7 @@
 from agents.agent_output import AgentOutputSchema
 from agents.decorators import tool, tool_input_guardrail, tool_output_guardrail
 from agents.exceptions import UserError
+from agents.guardrail import GuardrailFunctionOutput, input_guardrail
 from agents.items import (
     MessageOutputItem,
     ModelResponse,
@@ -1338,3 +1339,176 @@ async def test_fresh_streamed_handoff_publishes_agent_update_before_session_appe
         isinstance(event, AgentUpdatedStreamEvent) and event.new_agent.name == "delegate"
         for event in collected_events
     )
+
+
+@pytest.mark.asyncio
+async def test_fresh_streamed_handoff_drains_agent_update_event_for_slow_consumer() -> None:
+    """A session-append failure in the generic-loop handoff branch must mark itself for
+    stream-event draining, so a consumer that falls even slightly behind the producer (an
+    ordinary per-event delay, not a contrived zero-delay reader) still observes the
+    already-queued ``AgentUpdatedStreamEvent`` before the error surfaces.
+
+    test_fresh_streamed_handoff_publishes_agent_update_before_session_append_failure's
+    zero-delay consumer passes even without draining, since it never falls behind the
+    producer; this test exercises the actual drain guarantee stream_events() provides via
+    _mark_error_to_drain_stream_events()/_should_drain_stream_events_before_raising().
+    """
+    model = ScriptedModel(
+        [
+            [get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff-1")],
+            [get_text_message("done")],
+        ]
+    )
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(name="triage", model=model, handoffs=[delegate])
+    session = _FailSecondAddItemsSessionWithYield()
+
+    failed_result = Runner.run_streamed(
+        triage, "hello", session=session, run_config=RunConfig(tracing_disabled=True)
+    )
+    collected_events: list[Any] = []
+    caught: RuntimeError | None = None
+    try:
+        async for event in failed_result.stream_events():
+            # An ordinary bit of per-event consumer work, enough to fall behind the producer.
+            await asyncio.sleep(0.001)
+            collected_events.append(event)
+    except RuntimeError as error:
+        caught = error
+    assert caught is session.error
+    assert any(
+        isinstance(event, AgentUpdatedStreamEvent) and event.new_agent.name == "delegate"
+        for event in collected_events
+    )
+
+
+@pytest.mark.asyncio
+async def test_fresh_streamed_handoff_awaits_parallel_input_guardrail_before_transition() -> None:
+    """The generic-loop handoff branch must resolve an in-flight parallel input guardrail
+    before committing the handoff transition (current_agent, run_state, published events),
+    per the Guardrail Ordering contract in .agents/references/runner-lifecycle.md. Otherwise a
+    handoff on turn 1 can commit the transition while a still-running parallel input guardrail
+    that later raises has not yet been awaited.
+
+    Only one scripted turn is provided (the handoff itself), and the consumer adds a small
+    per-event delay: an in-process run with an instantly-draining consumer and a second
+    scripted turn can otherwise race straight through to completion before the guardrail's
+    sleep elapses, defeating the repro regardless of the fix. This mirrors an ordinary
+    consumer that does a bit of per-event work, not a contrived instant reader.
+    """
+    guardrail_error = RuntimeError("guardrail backend exploded")
+
+    @input_guardrail(run_in_parallel=True)
+    async def slow_failing_guardrail(
+        ctx: RunContextWrapper[Any],
+        agent: Agent[Any],
+        input: str | list[TResponseInputItem],
+    ) -> GuardrailFunctionOutput:
+        await asyncio.sleep(0.3)
+        raise guardrail_error
+
+    model = ScriptedModel(
+        [
+            [get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff-1")],
+        ]
+    )
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(
+        name="triage",
+        model=model,
+        handoffs=[delegate],
+        input_guardrails=[slow_failing_guardrail],
+    )
+
+    streamed_result = Runner.run_streamed(
+        triage, "hello", run_config=RunConfig(tracing_disabled=True)
+    )
+    caught: RuntimeError | None = None
+    try:
+        async for _ in streamed_result.stream_events():
+            await asyncio.sleep(0.05)
+    except RuntimeError as error:
+        caught = error
+    assert caught is guardrail_error
+    # The handoff transition must not have been committed: the guardrail task was still
+    # in flight (sleeping) when the model returned the handoff, and it raised a real error
+    # rather than a tripwire, so no part of the observable state should have moved past triage.
+    assert streamed_result.current_agent.name == "triage"
+    state = streamed_result.to_state()
+    assert state._current_agent is not None
+    assert state._current_agent.name == "triage"
+
+
+@pytest.mark.asyncio
+async def test_fresh_streamed_handoff_replays_deferred_compaction_after_resume() -> None:
+    """A checkpointed handoff batch that fails to append and later settles via a separate,
+    standalone resume_pending_session_write() call (the generic resume-startup path in
+    run.py/run_loop.py, not the original save_result_to_session() call) must still apply the
+    same post-write Responses compaction decision save_result_to_session would have applied
+    inline, instead of silently and permanently losing it. See
+    .agents/references/session-persistence.md.
+
+    Uses a should_trigger_compaction hook keyed on response_id (as a caller doing per-turn
+    compaction routing would) to make the loss observable: without the fix, the handoff's own
+    response_id is never evaluated by the hook at all, and the deferral it would have set is
+    never recorded, so the later forced compaction on the delegate's turn never happens either.
+    """
+    hook_calls: list[str | None] = []
+
+    def should_trigger_compaction(context: dict[str, Any]) -> bool:
+        hook_calls.append(context["response_id"])
+        return context["response_id"] == "resp-handoff"
+
+    compact_calls: list[list[TResponseInputItem]] = []
+
+    async def compact(**kwargs: Any) -> SimpleNamespace:
+        items = copy.deepcopy(kwargs["input"])
+        compact_calls.append(items)
+        return SimpleNamespace(output=items, usage=None)
+
+    backend = _FailSecondAddItemsSession()
+    session = OpenAIResponsesCompactionSession(
+        "compaction-handoff-test",
+        underlying_session=backend,
+        client=cast(Any, SimpleNamespace(responses=SimpleNamespace(compact=compact))),
+        compaction_mode="input",
+        should_trigger_compaction=should_trigger_compaction,
+    )
+
+    model = ScriptedModel(
+        [
+            {
+                "output": [
+                    get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff-1")
+                ],
+                "response_id": "resp-handoff",
+            },
+            {"output": [get_text_message("done")], "response_id": "resp-delegate"},
+        ]
+    )
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(name="triage", model=model, handoffs=[delegate])
+
+    failed_result = Runner.run_streamed(
+        triage, "hello", session=session, run_config=RunConfig(tracing_disabled=True)
+    )
+    with pytest.raises(RuntimeError) as error:
+        async for _ in failed_result.stream_events():
+            pass
+    assert error.value is backend.error
+    assert hook_calls == []
+    assert compact_calls == []
+    state = failed_result.to_state()
+    assert state._pending_session_write is not None
+    assert state._pending_session_write.get("response_id") == "resp-handoff"
+    assert state._pending_session_write.get("has_local_tool_outputs") is True
+
+    result = await _run_session_resume(triage, state, session, False)
+    assert result.final_output == "done"
+    # The handoff's own response_id must have been evaluated by the decision hook (and
+    # deferred), not skipped -- and, because force-compaction short-circuits the hook, it must
+    # be the only response_id the hook ever saw.
+    assert hook_calls == ["resp-handoff"]
+    # The deferred decision must actually have been forced through on the delegate's own save,
+    # i.e. the compact API was invoked at all -- not just checked and declined.
+    assert len(compact_calls) == 1

From 06e9e4008171fab5a4fa010018ecd8b7daf30944 Mon Sep 17 00:00:00 2001
From: Praveen Mittal <pkmittal28@gmail.com>
Date: Thu, 3 Sep 2026 09:44:17 +0200
Subject: [PATCH 4/5] address 3rd Codex review round: retain checkpoint until
 compaction settles

One more issue from the automated Codex review on commit 92891a6, verified
with live reproduction before being fixed:

resume_pending_session_write() cleared run_state._pending_session_write
before calling the newly-added _apply_post_write_compaction(), so if that
call raised or was cancelled, the checkpoint was already gone. A later
retry would then have nothing to redo the compaction step with, silently
and permanently losing the requested deferred/forced Responses compaction
even though the append itself had already succeeded.

Fixed by moving the compaction call inside the try block, before clearing
the checkpoint. The append reconciliation above already makes a retry
safe against duplicate appends (it detects an already-committed batch via
digest matching and skips re-appending), so this only changes when the
checkpoint is released, not the retry logic itself.

Added test_fresh_streamed_handoff_retains_checkpoint_when_post_write_compaction_fails
to tests/test_run_impl_resume_paths.py (73 total, up from 72), confirmed
to fail against the pre-fix code (checkpoint cleared despite the
compaction failure) and pass after. Full verification stack clean: make
format/lint/typecheck, and the full suite (9373 passed, 33 skipped, 0
failed).
---
 .../run_internal/session_persistence.py       | 19 +++--
 tests/test_run_impl_resume_paths.py           | 85 +++++++++++++++++++
 2 files changed, 96 insertions(+), 8 deletions(-)

diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py
index e695a431c7..ba386ac865 100644
--- a/src/agents/run_internal/session_persistence.py
+++ b/src/agents/run_internal/session_persistence.py
@@ -862,18 +862,21 @@ def digests(items: Sequence[TResponseInputItem]) -> list[str]:
             # 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"]
+        # Keep the checkpoint until compaction also settles: if _apply_post_write_compaction
+        # raises below, a later retry must still be able to redo just the compaction step
+        # instead of silently losing it. The append itself is retry-safe (the reconciliation
+        # above detects an already-committed batch and skips re-appending it).
+        await _apply_post_write_compaction(
+            session,
+            response_id=pending.get("response_id"),
+            store=pending.get("store"),
+            has_local_tool_outputs=pending.get("has_local_tool_outputs", False),
+            wrapper=compaction_wrapper if compaction_wrapper is not None else wrapper,
+        )
         run_state._pending_session_write = None
     finally:
         run_state._session_write_in_progress = False
 
-    await _apply_post_write_compaction(
-        session,
-        response_id=pending.get("response_id"),
-        store=pending.get("store"),
-        has_local_tool_outputs=pending.get("has_local_tool_outputs", False),
-        wrapper=compaction_wrapper if compaction_wrapper is not None else wrapper,
-    )
-
 
 async def rewind_session_items(
     session: Session | None,
diff --git a/tests/test_run_impl_resume_paths.py b/tests/test_run_impl_resume_paths.py
index 8d7117e393..21d06c3650 100644
--- a/tests/test_run_impl_resume_paths.py
+++ b/tests/test_run_impl_resume_paths.py
@@ -1512,3 +1512,88 @@ async def compact(**kwargs: Any) -> SimpleNamespace:
     # The deferred decision must actually have been forced through on the delegate's own save,
     # i.e. the compact API was invoked at all -- not just checked and declined.
     assert len(compact_calls) == 1
+
+
+@pytest.mark.asyncio
+async def test_fresh_streamed_handoff_retains_checkpoint_when_post_write_compaction_fails() -> None:
+    """If the post-write compaction decision raises after a checkpointed handoff batch's append
+    has already settled, the checkpoint (``_pending_session_write``) must survive so a later
+    retry can redo just the compaction step -- clearing it before the fallible compaction call
+    would silently and permanently lose the requested deferred/forced compaction with no way to
+    recover it. See .agents/references/session-persistence.md.
+    """
+    hook_calls: list[str | None] = []
+    compaction_error = RuntimeError("compaction decision hook exploded")
+    should_fail = True
+
+    def should_trigger_compaction(context: dict[str, Any]) -> bool:
+        hook_calls.append(context["response_id"])
+        if context["response_id"] == "resp-handoff" and should_fail:
+            raise compaction_error
+        return context["response_id"] == "resp-handoff"
+
+    compact_calls: list[list[TResponseInputItem]] = []
+
+    async def compact(**kwargs: Any) -> SimpleNamespace:
+        items = copy.deepcopy(kwargs["input"])
+        compact_calls.append(items)
+        return SimpleNamespace(output=items, usage=None)
+
+    backend = _FailSecondAddItemsSession()
+    session = OpenAIResponsesCompactionSession(
+        "compaction-handoff-failure-test",
+        underlying_session=backend,
+        client=cast(Any, SimpleNamespace(responses=SimpleNamespace(compact=compact))),
+        compaction_mode="input",
+        should_trigger_compaction=should_trigger_compaction,
+    )
+
+    model = ScriptedModel(
+        [
+            {
+                "output": [
+                    get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff-1")
+                ],
+                "response_id": "resp-handoff",
+            },
+            {"output": [get_text_message("done")], "response_id": "resp-delegate"},
+        ]
+    )
+    delegate = Agent(name="delegate", model=model)
+    triage = Agent(name="triage", model=model, handoffs=[delegate])
+
+    failed_result = Runner.run_streamed(
+        triage, "hello", session=session, run_config=RunConfig(tracing_disabled=True)
+    )
+    with pytest.raises(RuntimeError) as append_error:
+        async for _ in failed_result.stream_events():
+            pass
+    assert append_error.value is backend.error
+    state = failed_result.to_state()
+    assert state._pending_session_write is not None
+
+    # Resume: the append itself now succeeds (the backend's failure was one-shot), but the
+    # compaction decision hook raises for the handoff's own response_id.
+    with pytest.raises(RuntimeError) as compaction_error_info:
+        await _run_session_resume(triage, state, session, False)
+    assert compaction_error_info.value is compaction_error
+    # The checkpoint must still be present so a later retry can redo compaction alone, instead
+    # of the handoff's requested compaction being silently and permanently lost.
+    assert state._pending_session_write is not None
+    assert state._pending_session_write.get("response_id") == "resp-handoff"
+
+    # Retry: the hook no longer fails. The append must not be repeated (no duplicate items in
+    # session history), but compaction must actually run this time.
+    should_fail = False
+    hook_calls.clear()
+    result = await _run_session_resume(triage, state, session, False)
+    assert result.final_output == "done"
+    assert hook_calls == ["resp-handoff"]
+    assert len(compact_calls) == 1
+    stored = await session.get_items()
+    handoff_pair = [
+        str(item.get("type"))
+        for item in stored
+        if isinstance(item, dict) and item.get("call_id") == "handoff-1"
+    ]
+    assert handoff_pair == ["function_call", "function_call_output"]

From 20e7c3a23e9b081c4fa16d299dbaea4b0916bb4e Mon Sep 17 00:00:00 2001
From: Praveen Mittal <pkmittal28@gmail.com>
Date: Thu, 3 Sep 2026 14:30:05 +0200
Subject: [PATCH 5/5] address human review: clear the deferred-compaction
 marker only after success

One more issue, this time from a human reviewer (sylvesterkaczmarek) on
commit 06e9e40, verified with live reproduction before being fixed:

OpenAIResponsesCompactionSession.run_compaction() cleared
self._deferred_response_id before calling the fallible
client.responses.compact() API. If that call raised, the deferred marker
was already gone. The checkpoint-recovery code added in the last two
commits recomputes force=True purely from whether this marker is still
set, so on retry it silently recomputed force=False and could skip
compaction that was still owed -- even though the round-3 fix already let
the checkpoint itself survive for a retry.

Fixed by moving the clear to after compaction actually settles (after the
API call and the underlying session replacement both succeed), not before
attempting them. The digest-based retry-safety already added for the
append doesn't need any changes; this only moves when one session-internal
flag gets cleared.

Added test_run_compaction_retains_deferred_marker_when_api_call_fails to
tests/memory/test_openai_responses_compaction_session.py, confirmed to
fail against the pre-fix code (assert None == 'resp-handoff') and pass
after. Full verification stack clean: make format/lint/typecheck, and the
full suite (9374 passed, 33 skipped, 0 failed).
---
 .../openai_responses_compaction_session.py    |  7 ++-
 ...est_openai_responses_compaction_session.py | 52 +++++++++++++++++++
 2 files changed, 58 insertions(+), 1 deletion(-)

diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py
index f09c3a6edd..8a07e0718d 100644
--- a/src/agents/memory/openai_responses_compaction_session.py
+++ b/src/agents/memory/openai_responses_compaction_session.py
@@ -221,7 +221,6 @@ async def run_compaction(
             )
             return
 
-        self._deferred_response_id = None
         logger.debug(
             "compact: start for %s using %s (mode=%s)",
             self._response_id,
@@ -254,6 +253,12 @@ async def run_compaction(
             self._compaction_candidate_items = select_compaction_candidate_items(output_items)
             self._session_items = output_items
 
+        # Clear the deferred marker only now that compaction has actually settled. Clearing it
+        # before the fallible API call/replacement above would let a failed forced compaction
+        # silently lose its "this must be forced" signal: a later retry recomputes `force` from
+        # this marker, so an early clear makes the retry decline work that was still owed.
+        self._deferred_response_id = None
+
         logger.debug(
             "compact: done for %s (mode=%s, output=%s, candidates=%s)",
             self._response_id,
diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py
index 5519228ea6..e70ebf36f8 100644
--- a/tests/memory/test_openai_responses_compaction_session.py
+++ b/tests/memory/test_openai_responses_compaction_session.py
@@ -1208,6 +1208,58 @@ async def test_run_compaction_force_bypasses_threshold(self) -> None:
 
         mock_client.responses.compact.assert_called_once()
 
+    @pytest.mark.asyncio
+    async def test_run_compaction_retains_deferred_marker_when_api_call_fails(self) -> None:
+        """A forced compaction driven by a previously-deferred response must not lose that
+        "this must be forced" signal if the compact API call itself fails: a later retry needs
+        _deferred_response_id to still be set so it recomputes force=True, not force=False.
+        """
+        mock_session = self.create_mock_session()
+        mock_session.get_items.return_value = []
+
+        call_count = 0
+
+        async def compact(**kwargs: Any) -> MagicMock:
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                raise RuntimeError("compact API blew up")
+            response = MagicMock()
+            response.output = []
+            return response
+
+        mock_client = MagicMock()
+        mock_client.responses.compact = AsyncMock(side_effect=compact)
+
+        session = OpenAIResponsesCompactionSession(
+            session_id="test",
+            underlying_session=mock_session,
+            client=mock_client,
+            should_trigger_compaction=lambda _ctx: False,
+        )
+        # Simulate a prior turn (e.g. a handoff with local tool outputs) having deferred
+        # compaction for this response, the way _defer_compaction() would.
+        session._deferred_response_id = "resp-handoff"
+
+        with pytest.raises(RuntimeError, match="compact API blew up"):
+            await session.run_compaction(
+                {
+                    "response_id": "resp-delegate",
+                    "force": session._get_deferred_compaction_response_id() is not None,
+                }
+            )
+        assert session._get_deferred_compaction_response_id() == "resp-handoff"
+
+        # Retry, recomputing force the same way the checkpoint-recovery code does.
+        await session.run_compaction(
+            {
+                "response_id": "resp-delegate",
+                "force": session._get_deferred_compaction_response_id() is not None,
+            }
+        )
+        assert call_count == 2
+        assert session._get_deferred_compaction_response_id() is None
+
     @pytest.mark.asyncio
     async def test_run_compaction_suppresses_model_dump_warnings(self) -> None:
         mock_session = self.create_mock_session()