diff --git a/.agents/skills/implementation-final-review/scripts/test_skill_contract.py b/.agents/skills/implementation-final-review/scripts/test_skill_contract.py index be62e68543..c708f324a1 100644 --- a/.agents/skills/implementation-final-review/scripts/test_skill_contract.py +++ b/.agents/skills/implementation-final-review/scripts/test_skill_contract.py @@ -45,7 +45,10 @@ def test_quality_gates_cover_prior_failure_modes(self) -> None: "contract-surface inventory", "every consumer, forwarding branch, and adapter", "Search adjacent contract surfaces even when they are absent from the diff", - "do not add it to the current task manifest, report it as a current-pull-request finding, or let it block clean review", + ( + "do not add it to the current task manifest, report it as a " + "current-pull-request finding, or let it block clean review" + ), "await-boundary matrix", "a newer operation that starts and completes while suspended", "current active state is insufficient", diff --git a/src/agents/result.py b/src/agents/result.py index f88819df54..f27a3a533f 100644 --- a/src/agents/result.py +++ b/src/agents/result.py @@ -149,6 +149,15 @@ def _populate_state_from_result( state._generated_prompt_cache_key = source_state._generated_prompt_cache_key state._pending_input = copy.deepcopy(source_state._pending_input) state._current_step = source_state._current_step + if source_state._current_step is not None: + state._current_agent = source_state._current_agent + state._current_turn_persisted_item_count = ( + source_state._current_turn_persisted_item_count + ) + if source_state._pending_session_items: + state._pending_session_items = list(source_state._pending_session_items) + state._pending_session_id = source_state._pending_session_id + state._pending_session_store = source_state._pending_session_store else: state._generated_prompt_cache_key = getattr(result, "_generated_prompt_cache_key", None) state._pending_input = copy.deepcopy(getattr(result, "_pending_input_for_state", [])) diff --git a/src/agents/run.py b/src/agents/run.py index a782f80cce..7bece17ab6 100644 --- a/src/agents/run.py +++ b/src/agents/run.py @@ -140,6 +140,7 @@ prepare_input_with_session, reconcile_nested_history_owned_session_item_refs, resumed_turn_items, + retry_pending_resumed_turn_session_items, save_result_to_session, save_resumed_turn_items, session_items_for_turn, @@ -1160,6 +1161,14 @@ def _mark_response_hooks_started() -> None: ), store=store_setting, wrapper=context_wrapper, + run_state=( + run_state + if isinstance( + turn_result.next_step, + NextStepRunAgain | NextStepHandoff, + ) + else None + ), ) ) @@ -1293,6 +1302,8 @@ def _mark_response_hooks_started() -> None: owner_starts=blocked_output_owner_starts, blocked_message=blocked_message, ) + if run_state is not None: + run_state._current_step = NextStepRunAgain() list.extend(session_items, retained_items) try: await save_final_turn_items_after_guardrails( @@ -1308,6 +1319,7 @@ def _mark_response_hooks_started() -> None: response_id=turn_result.model_response.response_id, store=store_setting, wrapper=context_wrapper, + track_pending_write=True, ) except BaseException as persistence_error: raise _safe_redacted_persistence_error( @@ -1340,6 +1352,7 @@ def _mark_response_hooks_started() -> None: response_id=turn_result.model_response.response_id, store=store_setting, wrapper=context_wrapper, + track_pending_write=True, ) raise @@ -1350,6 +1363,13 @@ def _mark_response_hooks_started() -> None: current_agent, run_config, ) + if run_state is not None: + run_state._current_step = NextStepFinalOutput( + turn_result.next_step.output + ) + run_state._output_guardrail_results = list( + output_guardrail_results + ) await save_final_turn_items_after_guardrails( session=session, run_state=run_state, @@ -1359,6 +1379,7 @@ def _mark_response_hooks_started() -> None: response_id=turn_result.model_response.response_id, store=store_setting, wrapper=context_wrapper, + track_pending_write=True, ) current_step = getattr(run_state, "_current_step", None) approvals_from_state = approvals_from_step(current_step) @@ -1409,6 +1430,44 @@ def _mark_response_hooks_started() -> None: if run_state._current_step is None: run_state._current_step = NextStepRunAgain() + await retry_pending_resumed_turn_session_items( + session=session, + run_state=run_state, + response_id=( + run_state._model_responses[-1].response_id + if run_state._model_responses + else None + ), + wrapper=context_wrapper, + ) + + if isinstance(run_state._current_step, NextStepFinalOutput): + pending_final_output = run_state._current_step.output + run_state._current_step = None + result = RunResult( + input=original_input, + new_items=session_items, + raw_responses=model_responses, + final_output=pending_final_output, + _last_agent=current_agent, + input_guardrail_results=input_guardrail_results, + output_guardrail_results=output_guardrail_results, + tool_input_guardrail_results=tool_input_guardrail_results, + tool_output_guardrail_results=tool_output_guardrail_results, + context_wrapper=context_wrapper, + interruptions=[], + _tool_use_tracker_snapshot=_tool_use_tracker_snapshot(), + max_turns=max_turns, + ) + result._current_turn = current_turn + result._model_input_items = list(generated_items) + result._replay_from_model_input_items = list(generated_items) != list( + session_items + ) + result._trace_state = run_state._trace_state + result._original_input = copy_input_items(original_input) + return _finalize_result(result) + pending_input = run_state.pending_input if pending_input: pending_guardrails = current_agent.input_guardrails + ( diff --git a/src/agents/run_internal/agent_runner_helpers.py b/src/agents/run_internal/agent_runner_helpers.py index be1d976724..f9456a96f0 100644 --- a/src/agents/run_internal/agent_runner_helpers.py +++ b/src/agents/run_internal/agent_runner_helpers.py @@ -565,23 +565,32 @@ async def save_final_turn_items_after_guardrails( reasoning_item_id_policy: ReasoningItemIdPolicy | None = None, store: bool | None = None, wrapper: RunContextWrapper[Any] | None = None, + track_pending_write: bool = False, ) -> int: """Persist deferred final-turn items without skipping a partially persisted resumed turn.""" if not session_persistence_enabled or not items: return 0 if input_guardrails_triggered(input_guardrail_results): return 0 - if run_state is not None and run_state._current_turn_persisted_item_count > 0: - run_state._current_turn_persisted_item_count = await save_resumed_turn_items( + resumable_state = ( + run_state + if track_pending_write + and run_state is not None + and isinstance(run_state._current_step, NextStepRunAgain | NextStepFinalOutput) + else None + ) + if resumable_state is not None: + resumable_state._current_turn_persisted_item_count = await save_resumed_turn_items( session=session, items=items, - persisted_count=run_state._current_turn_persisted_item_count, + persisted_count=resumable_state._current_turn_persisted_item_count, response_id=response_id, - reasoning_item_id_policy=run_state._reasoning_item_id_policy, + reasoning_item_id_policy=resumable_state._reasoning_item_id_policy, store=store, wrapper=wrapper, + run_state=resumable_state, ) - return run_state._current_turn_persisted_item_count + return resumable_state._current_turn_persisted_item_count return await save_result_to_session( session, [], diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py index ba0c02f351..135c097a60 100644 --- a/src/agents/run_internal/run_loop.py +++ b/src/agents/run_internal/run_loop.py @@ -178,6 +178,7 @@ prepare_input_with_session, reconcile_nested_history_owned_session_item_refs, resumed_turn_items, + retry_pending_resumed_turn_session_items, rewind_session_items, save_result_to_session, save_resumed_turn_items, @@ -399,6 +400,12 @@ async def _save_resumed_stream_items( reasoning_item_id_policy=streamed_result._reasoning_item_id_policy, store=store, wrapper=streamed_result.context_wrapper, + run_state=( + run_state + if run_state is not None + and isinstance(run_state._current_step, NextStepRunAgain | NextStepFinalOutput) + else None + ), ) if run_state is not None: run_state._current_turn_persisted_item_count = ( @@ -501,6 +508,7 @@ async def _finalize_streamed_final_output( response_id: str | None, store_setting: bool | None, on_persisted_after_guardrails: Callable[[bool], None] | None = None, + track_pending_write: bool = False, ) -> None: output_guardrail_result_start = len(streamed_result.output_guardrail_results) redacted_persistence_error: BaseException | None = None @@ -556,6 +564,8 @@ async def _finalize_streamed_final_output( owner_starts=owner_starts, blocked_message=blocked_message, ) + if track_pending_write and streamed_result._state is not None: + streamed_result._state._current_step = NextStepRunAgain() if retained_items: try: await save_items(retained_items, response_id, store_setting) @@ -622,6 +632,11 @@ async def _finalize_streamed_final_output( agent, run_config, ) + if track_pending_write and streamed_result._state is not None: + streamed_result._state._current_step = NextStepFinalOutput(output) + streamed_result._state._output_guardrail_results = list( + streamed_result.output_guardrail_results + ) # 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 @@ -1429,6 +1444,7 @@ async def _save_max_turns_items( owner_starts=blocked_output_owner_starts, response_id=turn_result.model_response.response_id, store_setting=store_setting, + track_pending_write=True, ) if streamed_result._stored_exception is not None: break @@ -1458,6 +1474,27 @@ async def _save_max_turns_items( if streamed_result.is_complete: break + if run_state is not None: + streamed_result._current_turn_persisted_item_count = ( + await retry_pending_resumed_turn_session_items( + session=session, + run_state=run_state, + response_id=( + run_state._model_responses[-1].response_id + if run_state._model_responses + else None + ), + wrapper=streamed_result.context_wrapper, + ) + ) + + if isinstance(run_state._current_step, NextStepFinalOutput): + streamed_result.final_output = run_state._current_step.output + run_state._current_step = None + streamed_result.is_complete = True + streamed_result._event_queue.put_nowait(QueueCompleteSentinel()) + break + if run_state is not None and run_state._pending_input: if run_state._current_step is None: run_state._current_step = NextStepRunAgain() diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py index 8ebba55802..ac6829fd19 100644 --- a/src/agents/run_internal/session_persistence.py +++ b/src/agents/run_internal/session_persistence.py @@ -10,7 +10,7 @@ import inspect import json from collections import deque -from collections.abc import Sequence +from collections.abc import Callable, Sequence from typing import Any, cast from .. import _debug @@ -60,7 +60,14 @@ strip_internal_input_item_metadata, ) from .oai_conversation import OpenAIServerConversationTracker -from .run_steps import NextStepInterruption, ProcessedResponse, SingleStepResult +from .run_steps import ( + NextStepFinalOutput, + NextStepHandoff, + NextStepInterruption, + NextStepRunAgain, + ProcessedResponse, + SingleStepResult, +) __all__ = [ "admit_pending_input", @@ -72,6 +79,7 @@ "session_items_for_turn", "resumed_turn_items", "save_result_to_session", + "retry_pending_resumed_turn_session_items", "save_resumed_turn_items", "update_run_state_after_resume", "rewind_session_items", @@ -539,7 +547,17 @@ def update_run_state_after_resume( run_state._generated_items = generated_items if session_items is not None: run_state._session_items = list(session_items) - run_state._current_step = turn_result.next_step # type: ignore[assignment] + if isinstance(turn_result.next_step, NextStepHandoff): + # Advance the durable continuation before persisting the resumed outputs. If the + # atomic Session append fails, the RunState can retry that exact batch and then enter + # the target agent without re-running the approved tool or losing the handoff. + run_state._current_agent = turn_result.next_step.new_agent + run_state._current_step = NextStepRunAgain() + elif isinstance(turn_result.next_step, NextStepFinalOutput): + # Output guardrails still run before a terminal continuation becomes retryable. + run_state._current_step = NextStepRunAgain() + else: + run_state._current_step = turn_result.next_step async def save_result_to_session( @@ -552,6 +570,7 @@ async def save_result_to_session( reasoning_item_id_policy: ReasoningItemIdPolicy | None = None, store: bool | None = None, wrapper: RunContextWrapper[Any] | None = None, + on_session_items_persisted: Callable[[int], None] | None = None, ) -> int: """ Persist a turn to the session store, keeping track of what was already saved so retries @@ -646,12 +665,16 @@ async def save_result_to_session( if len(items_to_save) == 0: if run_state is not None: run_state._current_turn_persisted_item_count = already_persisted + saved_run_items_count + if on_session_items_persisted is not None: + on_session_items_persisted(saved_run_items_count) return saved_run_items_count 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 on_session_items_persisted is not None: + on_session_items_persisted(saved_run_items_count) if response_id and is_openai_responses_compaction_aware_session(session): has_local_tool_outputs = any( @@ -707,10 +730,26 @@ async def save_resumed_turn_items( reasoning_item_id_policy: ReasoningItemIdPolicy | None = None, store: bool | None = None, wrapper: RunContextWrapper[Any] | None = None, + run_state: RunState | None = None, ) -> int: """Persist resumed turn items and return the updated persisted count.""" if session is None or not items: return persisted_count + if type(persisted_count) is not int or persisted_count < 0: + raise UserError("Resumed Session persisted count must be a non-negative integer") + if run_state is not None: + run_state._pending_session_items = list(items) + run_state._pending_session_id = session.session_id + run_state._pending_session_store = store + + def acknowledge_session_write(saved_count: int) -> None: + if run_state is None: + return + run_state._current_turn_persisted_item_count = persisted_count + saved_count + run_state._pending_session_items = [] + run_state._pending_session_id = None + run_state._pending_session_store = None + saved_count = await save_result_to_session( session, [], @@ -720,8 +759,48 @@ async def save_resumed_turn_items( reasoning_item_id_policy=reasoning_item_id_policy, store=store, wrapper=wrapper, + on_session_items_persisted=acknowledge_session_write, + ) + updated_persisted_count = persisted_count + saved_count + return updated_persisted_count + + +async def retry_pending_resumed_turn_session_items( + *, + session: Session | None, + run_state: RunState, + response_id: str | None, + wrapper: RunContextWrapper[Any] | None = None, +) -> int: + """Retry an incomplete atomic resumed-turn append before the next model call.""" + if not run_state._pending_session_items: + return run_state._current_turn_persisted_item_count + if session is None: + raise UserError( + "Cannot resume a RunState with an incomplete Session write without the original " + "Session. Pass the same Session and retry." + ) + if session.session_id != run_state._pending_session_id: + raise UserError( + "Cannot resume a RunState with an incomplete Session write using a different " + "Session. Pass the Session with id " + f"{run_state._pending_session_id!r} and retry." + ) + if not isinstance(run_state._current_step, NextStepRunAgain | NextStepFinalOutput): + raise UserError( + "Cannot retry a pending Session write from a RunState without a resumable " + "continuation. Start a new run from safe input." + ) + return await save_resumed_turn_items( + session=session, + items=list(run_state._pending_session_items), + persisted_count=run_state._current_turn_persisted_item_count, + response_id=response_id, + reasoning_item_id_policy=run_state._reasoning_item_id_policy, + store=run_state._pending_session_store, + wrapper=wrapper, + run_state=run_state, ) - return persisted_count + saved_count async def rewind_session_items( diff --git a/src/agents/run_state.py b/src/agents/run_state.py index e00a73bdce..22a0c9f616 100644 --- a/src/agents/run_state.py +++ b/src/agents/run_state.py @@ -58,6 +58,7 @@ tool_output_identity, ) from .agent import Agent +from .agent_output import AgentOutputSchema, AgentOutputSchemaBase from .exceptions import ( ModelBehaviorError, UserError, @@ -146,6 +147,7 @@ from .guardrail import InputGuardrailResult, OutputGuardrailResult from .items import ModelResponse, RunItem from .run_internal.run_steps import ( + NextStepFinalOutput, NextStepInterruption, NextStepRunAgain, ProcessedResponse, @@ -183,6 +185,8 @@ def _default_run_state_validation_error( _PROGRAMMATIC_TOOL_CALLING_MIN_SCHEMA_VERSION = "1.13" _HOSTED_MCP_APPROVALS_MIN_SCHEMA_VERSION = "1.14" _CURRENT_RESPONSE_OWNERSHIP_MIN_SCHEMA_VERSION = "1.17" +_PENDING_SESSION_ITEMS_MIN_SCHEMA_VERSION = "1.17" +_PENDING_FINAL_OUTPUT_MIN_SCHEMA_VERSION = "1.17" # Keep this mapping in chronological order. Every schema bump must add a one-line summary here. SCHEMA_VERSION_SUMMARIES: dict[str, str] = { "1.0": "Initial RunState snapshot format for HITL pause/resume flows.", @@ -216,7 +220,8 @@ def _default_run_state_validation_error( ), "1.17": ( "Persists Docker container labels and current-response generated-item ownership across " - "resume flows." + "resume flows, plus incomplete atomic resumed-turn Session appends and terminal " + "continuations." ), } SUPPORTED_SCHEMA_VERSIONS = frozenset(SCHEMA_VERSION_SUMMARIES) @@ -827,7 +832,7 @@ class RunState(Generic[TContext, TAgent]): _tool_output_guardrail_results: list[ToolOutputGuardrailResult] = field(default_factory=list) """Results from tool output guardrails applied during the run.""" - _current_step: NextStepInterruption | NextStepRunAgain | None = None + _current_step: NextStepInterruption | NextStepRunAgain | NextStepFinalOutput | None = None """Current resumable step, or ``None`` when the state is terminal.""" _last_processed_response: ProcessedResponse | None = None @@ -839,6 +844,15 @@ class RunState(Generic[TContext, TAgent]): _current_turn_persisted_item_count: int = 0 """Tracks how many items from this turn were already written to the session.""" + _pending_session_items: list[RunItem] = field(default_factory=list, repr=False) + """Exact resumed-turn items whose atomic Session append has not completed.""" + + _pending_session_id: str | None = field(default=None, repr=False) + """Session identifier that owns the incomplete resumed-turn append.""" + + _pending_session_store: bool | None = field(default=None, repr=False) + """Resolved response storage mode that must be reused after an incomplete append.""" + _tool_use_tracker_snapshot: dict[str, list[str]] = field(default_factory=dict) """Serialized snapshot of the AgentToolUseTracker (agent name -> tools used).""" @@ -890,6 +904,9 @@ def __init__( self._last_processed_response = None self._generated_items_last_processed_marker = None self._current_turn_persisted_item_count = 0 + self._pending_session_items = [] + self._pending_session_id = None + self._pending_session_store = None self._tool_use_tracker_snapshot = {} self._trace_state = None self._sandbox = None @@ -949,7 +966,10 @@ def add_input(self, input: str | list[TResponseInputItem]) -> None: The input remains pending until its guardrails and conversation ownership boundary accept it. Terminal states reject new input before mutating the state. """ - from .run_internal.run_steps import NextStepInterruption, NextStepRunAgain + from .run_internal.run_steps import ( + NextStepInterruption, + NextStepRunAgain, + ) if not isinstance(self._current_step, NextStepInterruption | NextStepRunAgain): raise UserError("Cannot add input to a terminal RunState") @@ -1865,6 +1885,21 @@ def to_json( self._serialize_item(item, agent_identity_keys_by_id=agent_identity_keys_by_id) for item in list(self._session_items) ] + result["pending_session_write"] = ( + { + "session_id": self._pending_session_id, + "store": self._pending_session_store, + "items": [ + self._serialize_item( + item, + agent_identity_keys_by_id=agent_identity_keys_by_id, + ) + for item in self._pending_session_items + ], + } + if self._pending_session_items + else None + ) result["current_step"] = self._serialize_current_step() result["last_model_response"] = _serialize_last_model_response(model_responses) result["last_processed_response"] = ( @@ -1949,7 +1984,11 @@ def _serialize_processed_response( def _serialize_current_step(self) -> dict[str, Any] | None: """Serialize the current resumable step.""" # Import at runtime to avoid circular import - from .run_internal.run_steps import NextStepInterruption, NextStepRunAgain + from .run_internal.run_steps import ( + NextStepFinalOutput, + NextStepInterruption, + NextStepRunAgain, + ) agent_identity_keys_by_id = ( _build_agent_identity_keys_by_id(cast(Agent[Any], self._starting_agent)) @@ -1960,6 +1999,31 @@ def _serialize_current_step(self) -> dict[str, Any] | None: if isinstance(self._current_step, NextStepRunAgain): return {"type": "next_step_run_again"} + if isinstance(self._current_step, NextStepFinalOutput): + output_type = cast(Agent[Any], self._current_agent).output_type + if isinstance(output_type, AgentOutputSchemaBase): + if not isinstance(output_type, AgentOutputSchema): + raise UserError( + "Cannot serialize a pending final output that uses a custom output schema" + ) + output_type = output_type.output_type + if output_type is None or output_type is str: + if not isinstance(self._current_step.output, str): + raise UserError("Cannot serialize an invalid pending final output") + serialized_output = self._current_step.output + else: + try: + serialized_output = TypeAdapter(output_type).dump_python( + self._current_step.output, + mode="json", + ) + except Exception as exc: + raise UserError("Cannot serialize an invalid pending final output") from exc + return { + "type": "next_step_final_output", + "data": {"output": serialized_output}, + } + if self._current_step is None or not isinstance(self._current_step, NextStepInterruption): return None @@ -3951,6 +4015,34 @@ async def _build_run_state_from_json( UserError, ) + pending_session_major, pending_session_minor = ( + int(part) for part in _PENDING_SESSION_ITEMS_MIN_SCHEMA_VERSION.split(".", maxsplit=1) + ) + if state_json.get("pending_session_write") is not None and (schema_major, schema_minor) < ( + pending_session_major, + pending_session_minor, + ): + raise validation_error_factory( + "Run state pending_session_write requires schema version " + f"{_PENDING_SESSION_ITEMS_MIN_SCHEMA_VERSION} or newer", + UserError, + ) + + serialized_current_step = state_json.get("current_step") + if isinstance(serialized_current_step, Mapping): + pending_final_major, pending_final_minor = ( + int(part) for part in _PENDING_FINAL_OUTPUT_MIN_SCHEMA_VERSION.split(".", maxsplit=1) + ) + if serialized_current_step.get("type") == "next_step_final_output" and ( + schema_major, + schema_minor, + ) < (pending_final_major, pending_final_minor): + raise validation_error_factory( + "Run state pending final output requires schema version " + f"{_PENDING_FINAL_OUTPUT_MIN_SCHEMA_VERSION} or newer", + UserError, + ) + agent_identity_map = _build_agent_identity_map(initial_agent) agent_map = _build_agent_map(initial_agent) @@ -4132,6 +4224,57 @@ async def _build_run_state_from_json( serialized_session_items = [] state._session_items = state._merge_generated_items_with_processed() session_source_indexes = list(range(len(state._session_items))) + + serialized_pending_session_write = state_json.get("pending_session_write") + if serialized_pending_session_write is None: + serialized_pending_session_items: list[Any] = [] + pending_session_id = None + pending_session_store = None + elif not isinstance(serialized_pending_session_write, Mapping): + raise validation_error_factory( + "Run state pending_session_write must be an object or null", + UserError, + ) + else: + serialized_pending_session_items = serialized_pending_session_write.get("items", []) + pending_session_id = serialized_pending_session_write.get("session_id") + pending_session_store = serialized_pending_session_write.get("store") + if not isinstance(serialized_pending_session_items, list): + raise validation_error_factory( + "Run state pending_session_write items must be a list", + UserError, + ) + if serialized_pending_session_write is not None and not serialized_pending_session_items: + raise validation_error_factory( + "Run state pending_session_write items must not be empty", + UserError, + ) + if serialized_pending_session_write is not None and not isinstance(pending_session_id, str): + raise validation_error_factory( + "Run state pending_session_write session_id must be a string", + UserError, + ) + if pending_session_store is not None and not isinstance(pending_session_store, bool): + raise validation_error_factory( + "Run state pending_session_write store must be a boolean or null", + UserError, + ) + state._pending_session_items, pending_session_source_indexes = ( + _deserialize_items_with_source_indexes( + serialized_pending_session_items, + agent_map, + agent_identity_map=agent_identity_map, + validation_error_factory=validation_error_factory, + ) + ) + if pending_session_source_indexes != list(range(len(serialized_pending_session_items))): + raise validation_error_factory( + "Run state pending_session_write items must all be valid", + UserError, + ) + state._pending_session_id = pending_session_id if state._pending_session_items else None + state._pending_session_store = pending_session_store if state._pending_session_items else None + restored_session_indexes = { source_index: restored_index for restored_index, source_index in enumerate(session_source_indexes) @@ -4272,10 +4415,43 @@ async def _build_run_state_from_json( ) current_step_data = state_json.get("current_step") + if current_step_data is not None and not isinstance(current_step_data, Mapping): + raise validation_error_factory( + "Run state current_step must be an object or null", + UserError, + ) if current_step_data and current_step_data.get("type") == "next_step_run_again": from .run_internal.run_steps import NextStepRunAgain state._current_step = NextStepRunAgain() + elif current_step_data and current_step_data.get("type") == "next_step_final_output": + from .run_internal.run_steps import NextStepFinalOutput + + final_output_data = current_step_data.get("data") + if not isinstance(final_output_data, Mapping) or "output" not in final_output_data: + raise validation_error_factory( + "Run state pending final output data must contain output", + UserError, + ) + serialized_final_output = final_output_data["output"] + output_type = current_agent.output_type + try: + if isinstance(output_type, AgentOutputSchemaBase): + if not isinstance(output_type, AgentOutputSchema): + raise ValueError("custom output schema") + output_type = output_type.output_type + if output_type is None or output_type is str: + if not isinstance(serialized_final_output, str): + raise ValueError("plain-text final output must be a string") + final_output = serialized_final_output + else: + final_output = TypeAdapter(output_type).validate_python(serialized_final_output) + except Exception: + raise validation_error_factory( + "Run state pending final output is invalid", + UserError, + ) from None + state._current_step = NextStepFinalOutput(final_output) elif current_step_data and current_step_data.get("type") == "next_step_interruption": interruptions: list[ToolApprovalItem] = [] interruptions_data = current_step_data.get("data", {}).get( @@ -4325,9 +4501,24 @@ async def _build_run_state_from_json( for approval_item in state._current_step.interruptions: context._mark_restored_unbound_pending_approval(approval_item) - state._current_turn_persisted_item_count = state_json.get( - "current_turn_persisted_item_count", 0 - ) + from .run_internal.run_steps import NextStepFinalOutput, NextStepRunAgain + + if state._pending_session_items and not isinstance( + state._current_step, + NextStepRunAgain | NextStepFinalOutput, + ): + raise validation_error_factory( + "Run state pending_session_write requires a resumable continuation step", + UserError, + ) + + persisted_item_count = state_json.get("current_turn_persisted_item_count", 0) + if type(persisted_item_count) is not int or persisted_item_count < 0: + raise validation_error_factory( + "Run state current_turn_persisted_item_count must be a non-negative integer", + UserError, + ) + state._current_turn_persisted_item_count = persisted_item_count serialized_policy = state_json.get("reasoning_item_id_policy") if serialized_policy in {"preserve", "omit"}: state._reasoning_item_id_policy = cast(Literal["preserve", "omit"], serialized_policy) @@ -5591,6 +5782,25 @@ 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 requires schema version " + f"{_PENDING_SESSION_ITEMS_MIN_SCHEMA_VERSION} or newer" + ), + ( + "Run state pending final output requires schema version " + f"{_PENDING_FINAL_OUTPUT_MIN_SCHEMA_VERSION} or newer" + ), + "Run state current_step must be an object or null", + "Run state pending final output data must contain output", + "Run state pending_session_write must be an object or null", + "Run state pending_session_write items must be a list", + "Run state pending_session_write items must not be empty", + "Run state pending_session_write items must all be valid", + "Run state pending_session_write session_id must be a string", + "Run state pending_session_write store must be a boolean or null", + "Run state pending_session_write requires a resumable continuation step", + "Run state pending final output is invalid", + "Run state current_turn_persisted_item_count must be a non-negative integer", "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_agent_runner_streamed.py b/tests/test_agent_runner_streamed.py index 9664d65513..e4f66b8aad 100644 --- a/tests/test_agent_runner_streamed.py +++ b/tests/test_agent_runner_streamed.py @@ -4317,6 +4317,7 @@ async def save_wrapper( reasoning_item_id_policy: str | None = None, store: bool | None = None, wrapper: RunContextWrapper[Any] | None = None, + run_state: RunState[Any] | None = None, ) -> int: observed_counts.append(persisted_count) result = await real_save_resumed( @@ -4327,6 +4328,7 @@ async def save_wrapper( reasoning_item_id_policy=reasoning_item_id_policy, store=store, wrapper=wrapper, + run_state=run_state, ) return int(result) diff --git a/tests/test_run_impl_resume_paths.py b/tests/test_run_impl_resume_paths.py index c518c95ee0..d2f04218d5 100644 --- a/tests/test_run_impl_resume_paths.py +++ b/tests/test_run_impl_resume_paths.py @@ -6,15 +6,18 @@ from openai.types.responses import ResponseFunctionToolCall, ResponseOutputMessage import agents.run as run_module -from agents import Agent, Runner, function_tool +from agents import Agent, ModelSettings, Runner, function_tool from agents.agent import ToolsToFinalOutputResult from agents.agent_output import AgentOutputSchema +from agents.exceptions import UserError +from agents.guardrail import GuardrailFunctionOutput, OutputGuardrail from agents.items import ( MessageOutputItem, ModelResponse, ToolApprovalItem, ToolCallItem, ToolCallOutputItem, + TResponseInputItem, ) from agents.lifecycle import RunHooks from agents.run import RunConfig @@ -32,7 +35,7 @@ from agents.run_state import RunState from agents.testing import ScriptedModel from agents.usage import Usage -from tests.test_responses import get_function_tool_call, get_text_message +from tests.test_responses import get_function_tool_call, get_handoff_tool_call, get_text_message from tests.utils.hitl import ( make_agent, make_context_wrapper, @@ -42,6 +45,78 @@ from tests.utils.simple_session import SimpleListSession +class _FailNextAtomicAddSession(SimpleListSession): + """Fail selected atomic appends before mutating the in-memory history.""" + + def __init__(self) -> None: + super().__init__(session_id="fail-next-atomic-add") + self.fail_next_add = False + + async def add_items(self, items: list[TResponseInputItem]) -> None: + if self.fail_next_add: + self.fail_next_add = False + raise RuntimeError("injected atomic Session.add_items failure") + await super().add_items(items) + + +class _FailAfterAppendCompactionSession(_FailNextAtomicAddSession): + """Fail deferred compaction after the resumed output batch was appended.""" + + def __init__(self) -> None: + super().__init__() + self.fail_deferred_compaction = False + self.deferred_compaction_calls: list[tuple[str, bool | None]] = [] + + async def run_compaction(self, args: Any = None) -> None: + pass + + async def _defer_compaction(self, response_id: str, *, store: bool | None = None) -> None: + self.deferred_compaction_calls.append((response_id, store)) + if self.fail_deferred_compaction: + self.fail_deferred_compaction = False + raise RuntimeError("injected post-append compaction failure") + + +async def _run_in_mode( + mode: str, + agent: Agent[Any], + input_value: Any, + session: SimpleListSession, + run_config: RunConfig, +) -> Any: + if mode == "non_streamed": + return await Runner.run(agent, input_value, session=session, run_config=run_config) + result = Runner.run_streamed(agent, input_value, session=session, run_config=run_config) + async for _ in result.stream_events(): + pass + return result + + +async def _run_expecting_atomic_failure( + mode: str, + agent: Agent[Any], + state: RunState[Any], + session: _FailNextAtomicAddSession, + run_config: RunConfig, +) -> Any | None: + if mode == "non_streamed": + with pytest.raises(RuntimeError, match="injected atomic Session.add_items failure"): + await Runner.run(agent, state, session=session, run_config=run_config) + return None + result = Runner.run_streamed(agent, state, session=session, run_config=run_config) + with pytest.raises(RuntimeError, match="injected atomic Session.add_items failure"): + async for _ in result.stream_events(): + pass + return result + + +def _count_call_items(items: list[TResponseInputItem], item_type: str, call_id: str) -> int: + return sum( + isinstance(item, dict) and item.get("type") == item_type and item.get("call_id") == call_id + for item in items + ) + + @pytest.mark.asyncio async def test_resolve_interrupted_turn_final_output_short_circuit(monkeypatch) -> None: agent: Agent[dict[str, str]] = make_agent(model=ScriptedModel()) @@ -460,6 +535,525 @@ async def test_tool() -> str: assert output_count == 1 +@pytest.mark.parametrize( + ("pause_mode", "failing_mode", "retry_mode", "round_trip"), + [ + ("non_streamed", "non_streamed", "non_streamed", False), + ("non_streamed", "non_streamed", "streamed", True), + ("non_streamed", "streamed", "non_streamed", True), + ("non_streamed", "streamed", "streamed", False), + ("streamed", "non_streamed", "streamed", False), + ("streamed", "non_streamed", "non_streamed", True), + ("streamed", "streamed", "non_streamed", False), + ("streamed", "streamed", "streamed", True), + ], +) +@pytest.mark.asyncio +async def test_resumed_approval_retries_failed_atomic_session_write_before_model_call( + pause_mode: str, + failing_mode: str, + retry_mode: str, + round_trip: bool, +) -> None: + side_effects: list[int] = [] + + @function_tool(needs_approval=True) + async def charge(amount: int) -> str: + side_effects.append(amount) + return f"charged:{amount}" + + call_id = "call-charge" + model = ScriptedModel( + [ + [get_function_tool_call("charge", json.dumps({"amount": 7}), call_id=call_id)], + [get_text_message("done")], + ] + ) + agent = Agent(name="agent", model=model, tools=[charge]) + session = _FailNextAtomicAddSession() + run_config = RunConfig(tracing_disabled=True) + + paused = await _run_in_mode(pause_mode, agent, "charge 7", session, run_config) + state = paused.to_state() + state.approve(state.get_interruptions()[0]) + + session.fail_next_add = True + with pytest.raises(RuntimeError, match="injected atomic Session.add_items failure"): + await _run_in_mode(failing_mode, agent, state, session, run_config) + + assert side_effects == [7] + assert len(model.calls) == 1 + assert _count_call_items(await session.get_items(), "function_call", call_id) == 1 + assert _count_call_items(await session.get_items(), "function_call_output", call_id) == 0 + + with pytest.raises(UserError, match="without the original Session"): + await Runner.run(agent, state, run_config=run_config) + with pytest.raises(UserError, match="using a different Session"): + await Runner.run( + agent, + state, + session=SimpleListSession(session_id="different-session"), + run_config=run_config, + ) + assert len(model.calls) == 1 + + if round_trip: + serialized_state = state.to_json() + assert serialized_state["$schemaVersion"] == "1.17" + assert serialized_state["pending_session_write"]["session_id"] == session.session_id + assert serialized_state["pending_session_write"]["items"] + malformed_state = json.loads(json.dumps(serialized_state)) + malformed_state["pending_session_write"]["items"] = [{"type": "bogus"}] + with pytest.raises(UserError, match="pending_session_write items must all be valid"): + await RunState.from_json(agent, malformed_state) + empty_pending_state = json.loads(json.dumps(serialized_state)) + empty_pending_state["pending_session_write"]["items"] = [] + with pytest.raises(UserError, match="pending_session_write items must not be empty"): + await RunState.from_json(agent, empty_pending_state) + legacy_labeled_state = json.loads(json.dumps(serialized_state)) + legacy_labeled_state["$schemaVersion"] = "1.16" + with pytest.raises(UserError, match="pending_session_write requires schema version 1.17"): + await RunState.from_json(agent, legacy_labeled_state) + for invalid_count in ("0", True, -1): + invalid_count_state = json.loads(json.dumps(serialized_state)) + invalid_count_state["current_turn_persisted_item_count"] = invalid_count + with pytest.raises(UserError, match="must be a non-negative integer"): + await RunState.from_json(agent, invalid_count_state) + missing_count_state = json.loads(json.dumps(serialized_state)) + missing_count_state.pop("current_turn_persisted_item_count") + missing_count_restored = await RunState.from_json(agent, missing_count_state) + assert missing_count_restored._current_turn_persisted_item_count == 0 + state = await RunState.from_json(agent, serialized_state) + + session.fail_next_add = True + with pytest.raises(RuntimeError, match="injected atomic Session.add_items failure"): + await _run_in_mode(retry_mode, agent, state, session, run_config) + + assert side_effects == [7] + assert len(model.calls) == 1 + + resumed = await _run_in_mode(retry_mode, agent, state, session, run_config) + + assert resumed.final_output == "done" + assert side_effects == [7] + assert len(model.calls) == 2 + saved_items = await session.get_items() + replay_items = resumed.to_input_list() + assert _count_call_items(saved_items, "function_call", call_id) == 1 + assert _count_call_items(saved_items, "function_call_output", call_id) == 1 + assert _count_call_items(replay_items, "function_call", call_id) == 1 + assert _count_call_items(replay_items, "function_call_output", call_id) == 1 + + +@pytest.mark.parametrize( + ("failing_mode", "retry_mode", "round_trip"), + [ + ("non_streamed", "non_streamed", False), + ("non_streamed", "streamed", True), + ("streamed", "non_streamed", True), + ("streamed", "streamed", False), + ], +) +@pytest.mark.asyncio +async def test_resumed_approval_and_handoff_retries_session_write_before_target_model( + failing_mode: str, + retry_mode: str, + round_trip: bool, +) -> None: + side_effects: list[str] = [] + + @function_tool(needs_approval=True) + async def approved_tool() -> str: + side_effects.append("ran") + return "approved" + + source_model = ScriptedModel() + target_model = ScriptedModel([[get_text_message("done")]]) + target = Agent( + name="target", + model=target_model, + model_settings=ModelSettings(store=True), + ) + source = Agent( + name="source", + model=source_model, + model_settings=ModelSettings(store=False), + tools=[approved_tool], + handoffs=[target], + ) + approved_call_id = "call-approved-before-handoff" + handoff_call_id = "call-handoff-after-approval" + source_model.enqueue( + [ + get_function_tool_call("approved_tool", "{}", call_id=approved_call_id), + get_handoff_tool_call(target, call_id=handoff_call_id), + ] + ) + session = _FailAfterAppendCompactionSession() + run_config = RunConfig(tracing_disabled=True) + + paused = await Runner.run( + source, "approve and hand off", session=session, run_config=run_config + ) + state = paused.to_state() + state.approve(state.get_interruptions()[0]) + + session.fail_next_add = True + with pytest.raises(RuntimeError, match="injected atomic Session.add_items failure"): + await _run_in_mode(failing_mode, source, state, session, run_config) + + assert side_effects == ["ran"] + assert len(source_model.calls) == 1 + assert len(target_model.calls) == 0 + assert isinstance(state._current_step, NextStepRunAgain) + assert state._current_agent is target + assert ( + _count_call_items(await session.get_items(), "function_call_output", approved_call_id) == 0 + ) + assert ( + _count_call_items(await session.get_items(), "function_call_output", handoff_call_id) == 0 + ) + + if round_trip: + serialized_state = state.to_json() + assert serialized_state["pending_session_write"] is not None + assert serialized_state["pending_session_write"]["store"] is False + invalid_store_state = json.loads(json.dumps(serialized_state)) + invalid_store_state["pending_session_write"]["store"] = "false" + with pytest.raises(UserError, match="store must be a boolean or null"): + await RunState.from_json(source, invalid_store_state) + state = await RunState.from_json(source, serialized_state) + assert state._current_agent is target + + session.fail_next_add = True + with pytest.raises(RuntimeError, match="injected atomic Session.add_items failure"): + await _run_in_mode(retry_mode, source, state, session, run_config) + + assert side_effects == ["ran"] + assert len(source_model.calls) == 1 + assert len(target_model.calls) == 0 + + resumed = await _run_in_mode(retry_mode, source, state, session, run_config) + + assert resumed.final_output == "done" + assert resumed.last_agent is target + assert side_effects == ["ran"] + assert len(source_model.calls) == 1 + assert len(target_model.calls) == 1 + saved_items = await session.get_items() + assert _count_call_items(saved_items, "function_call_output", approved_call_id) == 1 + assert _count_call_items(saved_items, "function_call_output", handoff_call_id) == 1 + assert len(session.deferred_compaction_calls) == 1 + assert session.deferred_compaction_calls[0][1] is False + + +@pytest.mark.parametrize( + ("failing_mode", "retry_mode", "round_trip"), + [ + ("non_streamed", "non_streamed", True), + ("non_streamed", "streamed", False), + ("streamed", "non_streamed", False), + ("streamed", "streamed", True), + ], +) +@pytest.mark.asyncio +async def test_terminal_approval_retries_session_write_without_another_model_call( + failing_mode: str, + retry_mode: str, + round_trip: bool, +) -> None: + side_effects: list[str] = [] + + @function_tool(needs_approval=True) + async def terminal_tool() -> str: + side_effects.append("ran") + return "terminal-output" + + call_id = "call-terminal-after-approval" + model = ScriptedModel([[get_function_tool_call("terminal_tool", "{}", call_id=call_id)]]) + agent = Agent( + name="terminal-agent", + model=model, + tools=[terminal_tool], + tool_use_behavior="stop_on_first_tool", + ) + session = _FailNextAtomicAddSession() + run_config = RunConfig(tracing_disabled=True) + + paused = await Runner.run(agent, "run terminal tool", session=session, run_config=run_config) + state = paused.to_state() + state.approve(state.get_interruptions()[0]) + session.fail_next_add = True + failed_result = await _run_expecting_atomic_failure( + failing_mode, + agent, + state, + session, + run_config, + ) + + assert side_effects == ["ran"] + assert len(model.calls) == 1 + assert isinstance(state._current_step, NextStepFinalOutput) + assert state._pending_session_items + + if failed_result is not None: + state = failed_result.to_state() + assert isinstance(state._current_step, NextStepFinalOutput) + assert state._pending_session_items + + if round_trip: + payload = state.to_json() + assert payload["current_step"]["type"] == "next_step_final_output" + + legacy_final_payload = json.loads(json.dumps(payload)) + legacy_final_payload["$schemaVersion"] = "1.16" + legacy_final_payload["pending_session_write"] = None + with pytest.raises(UserError, match="pending final output requires schema version 1.17"): + await RunState.from_json(agent, legacy_final_payload) + + malformed_final_payload = json.loads(json.dumps(payload)) + malformed_final_payload["current_step"]["data"] = [] + with pytest.raises(UserError, match="pending final output data must contain output"): + await RunState.from_json(agent, malformed_final_payload) + + malformed_step_payload = json.loads(json.dumps(payload)) + malformed_step_payload["current_step"] = [] + with pytest.raises(UserError, match="current_step must be an object or null"): + await RunState.from_json(agent, malformed_step_payload) + + state = await RunState.from_json(agent, payload) + + resumed = await _run_in_mode(retry_mode, agent, state, session, run_config) + + assert resumed.final_output == "terminal-output" + assert side_effects == ["ran"] + assert len(model.calls) == 1 + assert _count_call_items(await session.get_items(), "function_call_output", call_id) == 1 + + +@pytest.mark.asyncio +async def test_non_streamed_terminal_guardrail_failure_retries_session_write_before_model() -> None: + side_effects: list[str] = [] + + @function_tool(needs_approval=True) + async def terminal_tool() -> str: + side_effects.append("ran") + return "terminal-output" + + def block_output( + _context: RunContextWrapper[Any], + _agent: Agent[Any], + _output: Any, + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) + + call_id = "call-terminal-guardrail-failure" + model = ScriptedModel([[get_function_tool_call("terminal_tool", "{}", call_id=call_id)]]) + agent = Agent( + name="terminal-guardrail-agent", + model=model, + tools=[terminal_tool], + tool_use_behavior="stop_on_first_tool", + output_guardrails=[OutputGuardrail(guardrail_function=block_output)], + ) + session = _FailNextAtomicAddSession() + run_config = RunConfig(tracing_disabled=True) + + paused = await Runner.run(agent, "run terminal tool", session=session, run_config=run_config) + state = paused.to_state() + state.approve(state.get_interruptions()[0]) + + session.fail_next_add = True + with pytest.raises(UserError): + await Runner.run(agent, state, session=session, run_config=run_config) + assert state._pending_session_items + assert len(model.calls) == 1 + assert side_effects == ["ran"] + + session.fail_next_add = True + with pytest.raises(RuntimeError, match="injected atomic Session.add_items failure"): + await Runner.run(agent, state, session=session, run_config=run_config) + assert state._pending_session_items + assert len(model.calls) == 1 + assert side_effects == ["ran"] + + agent.output_guardrails = [] + model.enqueue([get_text_message("continued")]) + resumed = await Runner.run(agent, state, session=session, run_config=run_config) + + assert resumed.final_output == "continued" + assert len(model.calls) == 2 + assert side_effects == ["ran"] + assert _count_call_items(await session.get_items(), "function_call_output", call_id) == 1 + + +@pytest.mark.parametrize("continuation", ["run_again", "handoff"]) +@pytest.mark.asyncio +async def test_failed_stream_to_state_preserves_pending_session_barrier( + continuation: str, +) -> None: + side_effects: list[str] = [] + + @function_tool(needs_approval=True) + async def approved_tool() -> str: + side_effects.append("ran") + return "approved" + + source_model = ScriptedModel() + target_model = ScriptedModel([[get_text_message("target-done")]]) + target = Agent(name="target", model=target_model) + source = Agent( + name="source", + model=source_model, + tools=[approved_tool], + handoffs=[target] if continuation == "handoff" else [], + ) + call_id = f"call-stream-state-{continuation}" + first_response = [get_function_tool_call("approved_tool", "{}", call_id=call_id)] + if continuation == "handoff": + first_response.append(get_handoff_tool_call(target, call_id="call-stream-handoff")) + source_model.enqueue(first_response) + expected_output = "target-done" + else: + source_model = ScriptedModel([first_response, [get_text_message("source-done")]]) + source.model = source_model + expected_output = "source-done" + + session = _FailNextAtomicAddSession() + run_config = RunConfig(tracing_disabled=True) + paused = await Runner.run(source, "approve", session=session, run_config=run_config) + state = paused.to_state() + state.approve(state.get_interruptions()[0]) + session.fail_next_add = True + + failed_result = await _run_expecting_atomic_failure( + "streamed", + source, + state, + session, + run_config, + ) + assert failed_result is not None + copied = failed_result.to_state() + + assert copied._pending_session_items + if continuation == "handoff": + assert copied._current_agent is target + + resumed = await Runner.run(source, copied, session=session, run_config=run_config) + assert resumed.final_output == expected_output + assert side_effects == ["ran"] + assert _count_call_items(await session.get_items(), "function_call_output", call_id) == 1 + + +@pytest.mark.asyncio +async def test_resumed_approval_does_not_retry_after_post_append_compaction_failure() -> None: + side_effects: list[int] = [] + + @function_tool(needs_approval=True) + async def charge(amount: int) -> str: + side_effects.append(amount) + return f"charged:{amount}" + + call_id = "call-charge-post-append" + model = ScriptedModel( + [ + [get_function_tool_call("charge", json.dumps({"amount": 9}), call_id=call_id)], + [get_text_message("done")], + ] + ) + agent = Agent(name="agent", model=model, tools=[charge]) + session = _FailAfterAppendCompactionSession() + run_config = RunConfig(tracing_disabled=True) + + paused = await Runner.run(agent, "charge 9", session=session, run_config=run_config) + state = paused.to_state() + state.approve(state.get_interruptions()[0]) + + session.fail_deferred_compaction = True + with pytest.raises(RuntimeError, match="injected post-append compaction failure"): + await Runner.run(agent, state, session=session, run_config=run_config) + + assert side_effects == [9] + assert len(model.calls) == 1 + assert state.to_json()["pending_session_write"] is None + assert _count_call_items(await session.get_items(), "function_call_output", call_id) == 1 + + resumed = await Runner.run(agent, state, session=session, run_config=run_config) + + assert resumed.final_output == "done" + assert side_effects == [9] + assert len(model.calls) == 2 + assert _count_call_items(await session.get_items(), "function_call_output", call_id) == 1 + + +@pytest.mark.asyncio +async def test_failed_stream_to_state_preserves_handoff_after_post_append_failure() -> None: + side_effects: list[str] = [] + + @function_tool(needs_approval=True) + async def approved_tool() -> str: + side_effects.append("ran") + return "approved" + + source_model = ScriptedModel() + target_model = ScriptedModel([[get_text_message("target-done")]]) + target = Agent(name="target", model=target_model) + source = Agent( + name="source", + model=source_model, + tools=[approved_tool], + handoffs=[target], + ) + approved_call_id = "call-post-append-approved" + source_model.enqueue( + [ + get_function_tool_call("approved_tool", "{}", call_id=approved_call_id), + get_handoff_tool_call(target, call_id="call-post-append-handoff"), + ] + ) + session = _FailAfterAppendCompactionSession() + run_config = RunConfig(tracing_disabled=True) + + paused = await Runner.run( + source, "approve and hand off", session=session, run_config=run_config + ) + state = paused.to_state() + state.approve(state.get_interruptions()[0]) + session.fail_deferred_compaction = True + + failed_result = Runner.run_streamed( + source, + state, + session=session, + run_config=run_config, + ) + with pytest.raises(RuntimeError, match="injected post-append compaction failure"): + async for _ in failed_result.stream_events(): + pass + + assert state._pending_session_items == [] + assert state._current_agent is target + assert isinstance(state._current_step, NextStepRunAgain) + + copied = failed_result.to_state() + assert copied._pending_session_items == [] + assert copied._current_agent is target + assert isinstance(copied._current_step, NextStepRunAgain) + + resumed = await Runner.run(source, copied, session=session, run_config=run_config) + + assert resumed.final_output == "target-done" + assert resumed.last_agent is target + assert side_effects == ["ran"] + assert len(source_model.calls) == 1 + assert len(target_model.calls) == 1 + assert ( + _count_call_items(await session.get_items(), "function_call_output", approved_call_id) == 1 + ) + + @pytest.mark.asyncio @pytest.mark.parametrize( ("schema_version", "expect_execution"),