fix(sessions): reject resuming a run whose accepted terminal output was not persisted

src/agents/run.py

@@ -70,6 +70,7 @@
     finalize_conversation_tracking,
     get_unsent_tool_call_ids_for_interrupted_state,
     input_guardrails_triggered,
+    reject_unrecoverable_terminal_state,
     resolve_processed_response,
     resolve_resumed_context,
     resolve_trace_settings,
@@ -635,6 +636,7 @@ async def _run_impl(
             )
             context = context_wrapper.context
 
+            reject_unrecoverable_terminal_state(run_state)
             await resume_pending_session_write(run_state, session, wrapper=context_wrapper)
             max_turns = run_state._max_turns
         else:
@@ -1367,6 +1369,11 @@ def _mark_response_hooks_started() -> None:
                                     current_agent,
                                     run_config,
                                 )
+                                # The output, its guardrails, and its terminal hooks are all
+                                # complete, so from here until the turn is persisted this run owns
+                                # a result no resume can reproduce.
+                                if run_state is not None:
+                                    run_state._terminal_unrecoverable = True
                                 await save_final_turn_items_after_guardrails(
                                     session=session,
                                     run_state=run_state,
@@ -1377,6 +1384,10 @@ def _mark_response_hooks_started() -> None:
                                     store=store_setting,
                                     wrapper=context_wrapper,
                                 )
+                                # The append and any post-append maintenance both succeeded,
+                                # so the turn is durable and the state is open again.
+                                if run_state is not None:
+                                    run_state._terminal_unrecoverable = False
                                 current_step = getattr(run_state, "_current_step", None)
                                 approvals_from_state = approvals_from_step(current_step)
                                 result = RunResult(
@@ -1994,6 +2005,8 @@ async def _save_max_turns_handler_output(
                                 current_agent,
                                 run_config,
                             )
+                            if run_state is not None:
+                                run_state._terminal_unrecoverable = True
                             await save_final_turn_items_after_guardrails(
                                 session=session,
                                 run_state=run_state,
@@ -2004,6 +2017,8 @@ async def _save_max_turns_handler_output(
                                 store=store_setting,
                                 wrapper=context_wrapper,
                             )
+                            if run_state is not None:
+                                run_state._terminal_unrecoverable = False
 
                             # Ensure starting_input is not None and not RunState
                             final_output_result_input: str | list[TResponseInputItem] = (

src/agents/run_internal/agent_runner_helpers.py

@@ -53,6 +53,7 @@
     "build_interruption_result",
     "build_resumed_stream_debug_extra",
     "describe_run_state_step",
+    "reject_unrecoverable_terminal_state",
     "ensure_context_wrapper",
     "finalize_conversation_tracking",
     "get_unsent_tool_call_ids_for_interrupted_state",
@@ -491,6 +492,22 @@ def build_interruption_result(
     return result
 
 
+def reject_unrecoverable_terminal_state(run_state: RunState | None) -> None:
+    """Fail closed when a previous run already produced a final output that cannot be reproduced.
+
+    The marker is set once that output, its guardrails, and its terminal hooks have completed,
+    and is cleared only once the turn is fully persisted. In between, the run owns a result no
+    resume can settle, so resuming would repeat the model call and the lifecycle hooks for an
+    output the caller already received. Raised before any Session, sandbox, model, tool,
+    guardrail, or hook work so the rejection has no side effects of its own.
+    """
+    if run_state is not None and run_state._terminal_unrecoverable:
+        raise UserError(
+            "This RunState already produced a final output whose Session write did not "
+            "complete, so it cannot be resumed. Start a new run instead."
+        )
+
+
 def append_model_response_if_new(
     model_responses: list[ModelResponse],
     response: ModelResponse,

src/agents/run_internal/run_loop.py

@@ -102,6 +102,7 @@
     apply_resumed_conversation_settings,
     attach_usage_to_span,
     get_unsent_tool_call_ids_for_interrupted_state,
+    reject_unrecoverable_terminal_state,
     snapshot_usage,
     usage_delta,
     validate_output_guardrails_with_server_managed_conversation,
@@ -628,6 +629,10 @@ async def _finalize_streamed_final_output(
     # Saved as one ordered batch so the session mirrors the model response. Doing it in two
     # halves would both reorder the turn and, because the first save advances the turn's
     # persisted-item count, make the second one a no-op.
+    # The output, its guardrails, and its terminal hooks are all complete, so from here until
+    # the turn is persisted this run owns a result no resume can reproduce.
+    if streamed_result._state is not None:
+        streamed_result._state._terminal_unrecoverable = True
     if on_persisted_after_guardrails is None:
         await save_items(final_turn_items, response_id, store_setting)
     else:
@@ -640,6 +645,10 @@ async def _finalize_streamed_final_output(
             streamed_result.is_complete = True
             streamed_result._event_queue.put_nowait(QueueCompleteSentinel())
             return
+    # The append and any post-append maintenance both succeeded, so the turn is durable and the
+    # state is open again.
+    if streamed_result._state is not None:
+        streamed_result._state._terminal_unrecoverable = False
 
     streamed_result.final_output = output
     if on_persisted_after_guardrails is not None:
@@ -923,6 +932,7 @@ async def start_streaming(
         streamed_result._reasoning_item_id_policy = resolved_reasoning_item_id_policy
 
         if is_resumed_state and run_state is not None:
+            reject_unrecoverable_terminal_state(run_state)
             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

src/agents/run_state.py

@@ -225,7 +225,7 @@ def _default_run_state_validation_error(
     ),
     "1.17": (
         "Persists Docker container labels and current-response generated-item ownership across "
-        "resume flows, including pending resumed Session writes."
+        "resume flows, including pending resumed Session writes and terminal-unrecoverable runs."
     ),
 }
 SUPPORTED_SCHEMA_VERSIONS = frozenset(SCHEMA_VERSION_SUMMARIES)
@@ -876,6 +876,13 @@ class RunState(Generic[TContext, TAgent]):
     _session_write_in_progress: bool = field(default=False, repr=False)
     """Live ownership guard; independent serialized copies require caller serialization."""
 
+    _terminal_unrecoverable: bool = field(default=False, repr=False)
+    """Set once a final output, its guardrails, and its terminal hooks have all completed.
+
+    It closes the state for the window where the run owns an accepted result that no resume can
+    reproduce, and it is cleared only once that turn is fully persisted.
+    """
+
     def __init__(
         self,
         context: RunContextWrapper[TContext],
@@ -918,6 +925,7 @@ def __init__(
         self._schema_version = CURRENT_SCHEMA_VERSION
         self._pending_session_write = None
         self._session_write_in_progress = False
+        self._terminal_unrecoverable = False
         from .agent_tool_state import get_agent_tool_state_scope
 
         self._agent_tool_state_scope_id = get_agent_tool_state_scope(context)
@@ -1907,6 +1915,8 @@ def to_json(
         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)
+        if self._terminal_unrecoverable:
+            result["terminal_unrecoverable"] = True
         result["trace"] = self._serialize_trace_data(
             include_tracing_api_key=include_tracing_api_key
         )
@@ -4381,6 +4391,13 @@ async def _build_run_state_from_json(
         ):
             raise validation_error_factory("Run state pending Session write is invalid", UserError)
         state._pending_session_write = copy.deepcopy(cast(_PendingSessionWrite, pending_write))
+    terminal_unrecoverable = state_json.get("terminal_unrecoverable")
+    if terminal_unrecoverable is not None:
+        # An older label never wrote this marker, so honoring one would let a snapshot claim a
+        # resume boundary the schema it declares does not have.
+        if (schema_major, schema_minor) < (1, 17) or terminal_unrecoverable is not True:
+            raise validation_error_factory("Run state terminal marker is invalid", UserError)
+        state._terminal_unrecoverable = True
     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)
@@ -5645,6 +5662,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 terminal marker 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_run_impl_resume_paths.py

@@ -37,6 +37,7 @@
     SingleStepResult,
 )
 from agents.run_state import RunState
+from agents.sandbox.runtime import SandboxRuntime
 from agents.testing import ScriptedModel
 from agents.tool import Tool
 from agents.tool_guardrails import (
@@ -1221,3 +1222,142 @@ 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()
+
+
+class _TerminalLifecycleHooks(RunHooks[Any]):
+    """Count the agent lifecycle hooks an application can attach its own effects to."""
+
+    def __init__(self) -> None:
+        self.starts = 0
+        self.ends: list[str] = []
+
+    async def on_agent_start(self, context: Any, agent: Agent[Any]) -> None:
+        self.starts += 1
+
+    async def on_agent_end(self, context: Any, agent: Agent[Any], output: Any) -> None:
+        self.ends.append(str(output))
+
+
+async def _terminal_output_session_state(
+    streamed: bool,
+    session: Session | None = None,
+    hooks: RunHooks[Any] | None = None,
+):
+    """Pause on an approval whose tool output becomes the terminal agent output."""
+    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_text_message("retry-final")],
+        ]
+    )
+    agent = Agent(
+        name="payment",
+        model=model,
+        tools=[charge],
+        tool_use_behavior="stop_on_first_tool",
+    )
+    session = session if session is not None else _FailingResumeSession()
+    paused = await _run_session_resume(agent, "charge 7", session, streamed, hooks=hooks)
+    state = paused.to_state()
+    state.approve(state.get_interruptions()[0])
+    return agent, model, session, state, effects
+
+
+@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_terminal_session_append_failure_rejects_every_later_resume(
+    failing_streamed: bool, retry_streamed: bool, round_trip: bool, failure: str
+) -> None:
+    """An accepted terminal output whose append failed is not resumable, and never replayed."""
+    hooks = _TerminalLifecycleHooks()
+    agent, model, session, state, effects = await _terminal_output_session_state(
+        failing_streamed, hooks=hooks
+    )
+    session.failure = failure
+    with pytest.raises(RuntimeError) as error:
+        await _run_session_resume(agent, state, session, failing_streamed, hooks=hooks)
+    assert error.value is session.error
+
+    # The output, its guardrails, and its terminal hooks all completed exactly once.
+    assert effects == [7]
+    assert len(model.calls) == 1
+    assert hooks.ends == ["receipt-7"]
+    starts_after_failure = hooks.starts
+    assert state.to_json()["terminal_unrecoverable"] is True
+
+    if round_trip:
+        state = await RunState.from_json(agent, state.to_json())
+
+    # Every later resume fails closed, including a second one.
+    for _ in range(2):
+        with pytest.raises(UserError, match="cannot be resumed"):
+            await _run_session_resume(agent, state, session, retry_streamed, hooks=hooks)
+        assert len(model.calls) == 1
+        assert effects == [7]
+        assert hooks.starts == starts_after_failure
+        assert hooks.ends == ["receipt-7"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("streamed", [False, True])
+async def test_unrecoverable_terminal_state_rejects_before_any_resumed_work(
+    streamed: bool, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The rejection precedes Session reconciliation and sandbox preparation."""
+    agent, model, session, state, effects = await _terminal_output_session_state(streamed)
+    session.failure = "before"
+    with pytest.raises(RuntimeError, match="session append failed"):
+        await _run_session_resume(agent, state, session, streamed)
+
+    async def _fail_get_items(*args: Any, **kwargs: Any) -> list[TResponseInputItem]:
+        raise AssertionError("Session reconciliation must not run for a rejected terminal state")
+
+    async def _fail_prepare_agent(*args: Any, **kwargs: Any):
+        raise AssertionError("sandbox preparation must not run for a rejected terminal state")
+
+    monkeypatch.setattr(type(session), "get_items", _fail_get_items)
+    monkeypatch.setattr(SandboxRuntime, "prepare_agent", _fail_prepare_agent)
+
+    restored = await RunState.from_json(agent, state.to_json())
+    with pytest.raises(UserError, match="cannot be resumed"):
+        await _run_session_resume(agent, restored, session, not streamed)
+    assert len(model.calls) == 1
+    assert effects == [7]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("streamed", [False, True])
+async def test_terminal_marker_is_cleared_once_the_turn_is_persisted(streamed: bool) -> None:
+    """A terminal turn that persists cleanly leaves a normal, unmarked result."""
+    agent, model, session, state, effects = await _terminal_output_session_state(streamed)
+    result = await _run_session_resume(agent, state, session, streamed)
+
+    assert result.final_output == "receipt-7"
+    assert effects == [7]
+    assert "terminal_unrecoverable" not in result.to_state().to_json()
+    assert _charge_pair(await session.get_items()) == ["function_call", "function_call_output"]
+
+
+@pytest.mark.asyncio
+async def test_terminal_marker_rejects_an_older_schema_label() -> None:
+    """The marker is only honored on the schema boundary that introduced it."""
+    agent, _, session, state, _ = await _terminal_output_session_state(False)
+    session.failure = "before"
+    with pytest.raises(RuntimeError, match="session append failed"):
+        await _run_session_resume(agent, state, session, False)
+
+    payload = state.to_json()
+    payload["$schemaVersion"] = "1.16"
+    with pytest.raises(UserError, match="terminal marker is invalid"):
+        await RunState.from_json(agent, payload)