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

Author: Ayushraj06-bit

src/agents/result.py

@@ -150,6 +150,9 @@ def _populate_state_from_result(
         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
+        # A streamed result exists before its terminal append does, so a checkpoint taken from a
+        # failed stream has to keep the fail-closed marker or it would look resumable.
+        state._terminal_unrecoverable = source_state._terminal_unrecoverable
     else:
         state._generated_prompt_cache_key = getattr(result, "_generated_prompt_cache_key", None)
         state._pending_input = copy.deepcopy(getattr(result, "_pending_input_for_state", []))

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(
@@ -1550,7 +1561,16 @@ async def _save_max_turns_handler_output(
                             include_in_history=include_in_history,
                         )
                         if include_in_history and not handler_output_recorded:
+                            # Only reachable once the handler output cleared its guardrails and
+                            # ran its end hooks, so this append carries an accepted result like
+                            # any other terminal one. The callback itself stays unmarked because
+                            # finalize_max_turns_handler_output() also drives it from its
+                            # guardrail-error path, where no output was ever accepted.
+                            if run_state is not None:
+                                run_state._terminal_unrecoverable = True
                             await _save_max_turns_handler_output([synthesized_item])
+                            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 +2014,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 +2026,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

@@ -227,7 +227,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)
@@ -878,6 +878,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],
@@ -920,6 +927,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)
@@ -1909,6 +1917,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
         )
@@ -4383,6 +4393,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)
@@ -5172,6 +5189,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 "