diff --git a/src/agents/run.py b/src/agents/run.py index 37dd66582e..b93e872944 100644 --- a/src/agents/run.py +++ b/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] = ( diff --git a/src/agents/run_internal/agent_runner_helpers.py b/src/agents/run_internal/agent_runner_helpers.py index be1d976724..6662f71e26 100644 --- a/src/agents/run_internal/agent_runner_helpers.py +++ b/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, diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py index 3c7d9ee586..9871a54041 100644 --- a/src/agents/run_internal/run_loop.py +++ b/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 diff --git a/src/agents/run_state.py b/src/agents/run_state.py index 228dd574d8..89839ace94 100644 --- a/src/agents/run_state.py +++ b/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 " diff --git a/tests/test_run_impl_resume_paths.py b/tests/test_run_impl_resume_paths.py index 8e6211ceab..391f785095 100644 --- a/tests/test_run_impl_resume_paths.py +++ b/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)