diff --git a/src/agents/result.py b/src/agents/result.py index 70d48fe3ef..1499a1edd5 100644 --- a/src/agents/result.py +++ b/src/agents/result.py @@ -46,7 +46,7 @@ ProcessedResponse, QueueCompleteSentinel, ) -from .run_state import RunState +from .run_state import RunState, _PendingSessionWrite from .stream_events import StreamEvent from .tool_guardrails import ToolInputGuardrailResult, ToolOutputGuardrailResult from .tracing import Trace @@ -156,6 +156,9 @@ def _populate_state_from_result( else: state._generated_prompt_cache_key = getattr(result, "_generated_prompt_cache_key", None) state._pending_input = copy.deepcopy(getattr(result, "_pending_input_for_state", [])) + state._pending_session_write = copy.deepcopy( + getattr(result, "_pending_session_write", None) + ) state._current_step = getattr(result, "_current_step_for_state", None) state._reasoning_item_id_policy = getattr(result, "_reasoning_item_id_policy", None) @@ -367,6 +370,10 @@ class RunResultBase(abc.ABC): default_factory=list, init=False, repr=False ) """Pending input preserved when a non-streaming result is converted back to RunState.""" + _pending_session_write: _PendingSessionWrite | None = field( + default=None, init=False, repr=False + ) + """Held pending Session write preserved when a non-streaming result becomes a RunState.""" _current_step_for_state: Any = field(default=None, init=False, repr=False) """Current step preserved when a non-streaming result is converted back to RunState.""" diff --git a/src/agents/run.py b/src/agents/run.py index 297895a347..cf0191353e 100644 --- a/src/agents/run.py +++ b/src/agents/run.py @@ -140,6 +140,8 @@ _session_get_items, admit_pending_input, commit_server_pending_input, + defer_interrupted_session_write, + extend_held_session_write, persist_session_items_for_guardrail_trip, prepare_input_with_session, reconcile_nested_history_owned_session_item_refs, @@ -148,6 +150,8 @@ save_result_to_session, save_resumed_turn_items, session_items_for_turn, + settle_held_batch_for_emptied_turn, + take_held_session_write, update_run_state_after_resume, ) from .run_internal.tool_use_tracker import ( @@ -1163,34 +1167,79 @@ def _mark_response_hooks_started() -> None: ] if ( - session_persistence_enabled - and turn_session_items - and run_state is not None + run_state is not None + # A final output is persisted by the final-turn sweep + # below, which claims the held batch itself. and not isinstance(turn_result.next_step, NextStepFinalOutput) - and not ( - isinstance(turn_result.next_step, NextStepInterruption) - and _should_defer_interrupted_session_items( - current_agent, - run_config, - ) - ) ): - run_state._current_turn_persisted_item_count = ( - await save_resumed_turn_items( - run_state=run_state, - session=session, - items=turn_session_items, - persisted_count=( - run_state._current_turn_persisted_item_count + if not session_persistence_enabled: + # A detached resume's save is a no-op, so the resolved + # items fold into the standing held batch and settle + # together at the reattach. + extend_held_session_write( + run_state, + run_items=turn_session_items, + reasoning_item_id_policy=( + run_state._reasoning_item_id_policy ), - response_id=turn_result.model_response.response_id, + ) + elif isinstance( + turn_result.next_step, NextStepInterruption + ) and _should_defer_interrupted_session_items( + current_agent, + run_config, + ): + # The re-park keeps deferring: the resolved items join + # the held batch instead of reaching the Session. + defer_interrupted_session_write( + run_state, + session, + run_items=turn_session_items, reasoning_item_id_policy=( run_state._reasoning_item_id_policy ), + response_id=turn_result.model_response.response_id, store=store_setting, - wrapper=context_wrapper, ) - ) + elif turn_session_items: + run_state._current_turn_persisted_item_count = ( + await save_resumed_turn_items( + run_state=run_state, + session=session, + items=turn_session_items, + held_input=take_held_session_write(run_state), + persisted_count=( + run_state._current_turn_persisted_item_count + ), + response_id=turn_result.model_response.response_id, + reasoning_item_id_policy=( + run_state._reasoning_item_id_policy + ), + store=store_setting, + wrapper=context_wrapper, + ) + ) + else: + # An emptied resolved turn (a handoff input_filter + # can drop every item) still settles the pairs the + # batch already holds: the approved tool ran, and + # dropping its output with the filtered items would + # lose the Session's only record of that. + run_state._current_turn_persisted_item_count = ( + await settle_held_batch_for_emptied_turn( + run_state, + session, + persisted_count=( + run_state._current_turn_persisted_item_count + ), + response_id=(turn_result.model_response.response_id), + reasoning_item_id_policy=( + run_state._reasoning_item_id_policy + ), + store=store_setting, + wrapper=context_wrapper, + ) + ) # After the resumed turn, treat subsequent turns as fresh so # counters and input saving behave normally. @@ -1323,6 +1372,12 @@ def _mark_response_hooks_started() -> None: blocked_message=blocked_message, ) list.extend(session_items, retained_items) + # The redaction derives the sanitized response from the + # run-state boundary, so the raw held batch must not be + # fed into this save: it could resurrect preambles the + # redaction dropped. The declaration is discarded once + # the blocked outcome is decided. + take_held_session_write(run_state) try: await save_final_turn_items_after_guardrails( session=session, @@ -1366,6 +1421,7 @@ def _mark_response_hooks_started() -> None: _attempt_input_guardrail_results() ), items=final_turn_items, + held_input=take_held_session_write(run_state), response_id=turn_result.model_response.response_id, store=store_setting, wrapper=context_wrapper, @@ -1373,7 +1429,7 @@ def _mark_response_hooks_started() -> None: raise final_turn_items = _final_turn_items_for_persistence( - turn_session_items, + list(turn_session_items), current_processed_response, run_state, current_agent, @@ -1390,6 +1446,10 @@ def _mark_response_hooks_started() -> None: session_persistence_enabled=session_persistence_enabled, input_guardrail_results=_attempt_input_guardrail_results(), items=final_turn_items, + # Safe even when the guardrail rebuild above already + # recovered the parked response: the save deduplicates + # the combined batch. + held_input=take_held_session_write(run_state), response_id=turn_result.model_response.response_id, store=store_setting, wrapper=context_wrapper, @@ -1569,6 +1629,7 @@ async def _save_max_turns_handler_output( output_guardrail_results=output_guardrail_results, save_items_after_guardrails=_save_max_turns_handler_output, include_in_history=include_in_history, + run_state=run_state, ) if include_in_history and not handler_output_recorded: # Only reachable once the handler output cleared its guardrails and @@ -2039,6 +2100,16 @@ async def _save_max_turns_handler_output( if run_state is not None: run_state._terminal_unrecoverable = False + if session is None and run_state is not None: + # A detached completion has no Session to settle against + # and the run ends here, so the batch is discarded + # rather than left to invalidate the completed run's + # checkpoint. Only here, though: the guardrails and the + # final save above can raise, and a run that raises may + # still be retried or reattached, with the executed + # tool's call and output reachable only through it. + take_held_session_write(run_state) + # Ensure starting_input is not None and not RunState final_output_result_input: str | list[TResponseInputItem] = ( normalized_starting_input @@ -2072,21 +2143,51 @@ async def _save_max_turns_handler_output( run_state._current_step = None return _finalize_result(result) elif isinstance(turn_result.next_step, NextStepInterruption): - if session_persistence_enabled and not ( - _should_defer_interrupted_session_items( - current_agent, - run_config, - ) + if session_persistence_enabled and not input_guardrails_triggered( + _attempt_input_guardrail_results() ): - if not input_guardrails_triggered( - _attempt_input_guardrail_results() + # Persist session items but skip approval placeholders. + input_items_for_save_interruption: list[TResponseInputItem] = ( + session_input_items_for_persistence + if session_input_items_for_persistence is not None + else [] + ) + if run_state is not None and ( + _should_defer_interrupted_session_items( + current_agent, + run_config, + ) ): - # Persist session items but skip approval placeholders. - input_items_for_save_interruption: list[TResponseInputItem] = ( - session_input_items_for_persistence - if session_input_items_for_persistence is not None - else [] + # The gate withholds the interrupted response, not + # the user's accepted input: any input still + # unsaved (the sandbox runtime defers the pre-turn + # save) persists here exactly as the non-deferred + # arm would, so the held batch never carries the + # Session's only copy of the input. Declaring the + # response batch on the checkpoint lets a resume + # settle it at a gate-legal exit instead of losing + # it. This runner always builds a RunState, so the + # narrowing never skips a real park. + if input_items_for_save_interruption: + await save_result_to_session( + session, + input_items_for_save_interruption, + [], + run_state, + store=store_setting, + wrapper=context_wrapper, + ) + defer_interrupted_session_write( + run_state, + session, + run_items=session_items_for_turn(turn_result), + reasoning_item_id_policy=( + run_state._reasoning_item_id_policy + ), + response_id=turn_result.model_response.response_id, + store=store_setting, ) + else: await save_result_to_session( session, input_items_for_save_interruption, @@ -2096,6 +2197,16 @@ async def _save_max_turns_handler_output( store=store_setting, wrapper=context_wrapper, ) + elif session is None and run_state is not None: + # A fresh park during a detached resume cannot write, + # but a standing held declaration carries the session + # identity: the new parked call folds into it so the + # reattach does not settle its output orphaned. + extend_held_session_write( + run_state, + run_items=session_items_for_turn(turn_result), + reasoning_item_id_policy=(run_state._reasoning_item_id_policy), + ) append_model_response_if_new( model_responses, turn_result.model_response ) diff --git a/src/agents/run_internal/agent_runner_helpers.py b/src/agents/run_internal/agent_runner_helpers.py index 6662f71e26..2bd03322bd 100644 --- a/src/agents/run_internal/agent_runner_helpers.py +++ b/src/agents/run_internal/agent_runner_helpers.py @@ -2,7 +2,8 @@ from __future__ import annotations -from collections.abc import Mapping +import copy +from collections.abc import Mapping, Sequence from typing import Any, cast from openai.types.responses.response_usage import OutputTokensDetails @@ -41,7 +42,13 @@ NextStepRunAgain, ProcessedResponse, ) -from .session_persistence import save_result_to_session, save_resumed_turn_items +from .session_persistence import ( + _held_items_safe_to_settle, + _pending_approval_call_ids, + final_items_cover_held_batch, + save_result_to_session, + save_resumed_turn_items, +) from .tool_use_tracker import AgentToolUseTracker, serialize_tool_use_tracker from .turn_preparation import get_model @@ -488,6 +495,10 @@ def build_interruption_result( if run_state is not None: result._current_turn_persisted_item_count = run_state._current_turn_persisted_item_count result._trace_state = run_state._trace_state + # The held pending write must survive the result checkpoint: a non-streamed + # caller serializes ``result.to_state()``, which has no live ``RunState`` to + # read the declaration from. + result._pending_session_write = copy.deepcopy(run_state._pending_session_write) result._original_input = copy_input_items(original_input) return result @@ -582,13 +593,25 @@ async def save_final_turn_items_after_guardrails( reasoning_item_id_policy: ReasoningItemIdPolicy | None = None, store: bool | None = None, wrapper: RunContextWrapper[Any] | None = None, + held_input: Sequence[TResponseInputItem] | None = None, ) -> int: - """Persist deferred final-turn items without skipping a partially persisted resumed turn.""" - if not session_persistence_enabled or not items: + """Persist deferred final-turn items without skipping a partially persisted resumed turn. + + ``held_input`` is a claimed held batch that must land ahead of the final items in + the same append. It is safe to pass even when the rebuilt final items already + contain the parked response: the save deduplicates the combined batch. + """ + if not session_persistence_enabled or (not items and not held_input): return 0 if input_guardrails_triggered(input_guardrail_results): return 0 + # Whether a held batch is being claimed at all, captured before any dedup empties + # it: the recovery registration below must stay armed even when the guardrail + # rebuild already carries the batch. + settling_held = bool(held_input) if run_state is not None and run_state._current_turn_persisted_item_count > 0: + # save_resumed_turn_items owns the dedup, pairing, and recovery arming; the raw + # held batch rides in so it can arm from its own pre-dedup view. run_state._current_turn_persisted_item_count = await save_resumed_turn_items( session=session, items=items, @@ -597,17 +620,35 @@ async def save_final_turn_items_after_guardrails( reasoning_item_id_policy=run_state._reasoning_item_id_policy, store=store, wrapper=wrapper, + run_state=run_state, + held_input=held_input, ) return run_state._current_turn_persisted_item_count + if held_input and final_items_cover_held_batch(items, held_input, reasoning_item_id_policy): + # The guardrail rebuild re-derived the whole current response, held requests + # included; feeding the batch again would duplicate its unkeyed companions. + held_input = None + if held_input: + held_input = _held_items_safe_to_settle( + held_input, + items, + reasoning_item_id_policy, + pending_call_ids=_pending_approval_call_ids(run_state), + ) return await save_result_to_session( session, - [], + list(held_input) if held_input else [], list(items), run_state, response_id=response_id, reasoning_item_id_policy=reasoning_item_id_policy, store=store, wrapper=wrapper, + # A settling held batch always registers, so a crash inside this append fails + # closed with the batch recorded instead of silently losing it, even when the + # payload was deduplicated from the append. + resumed_write_state=run_state if settling_held else None, + settling_held_batch=settling_held, ) diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py index 9871a54041..d812b30d7f 100644 --- a/src/agents/run_internal/run_loop.py +++ b/src/agents/run_internal/run_loop.py @@ -175,6 +175,8 @@ _session_get_items, admit_pending_input, commit_server_pending_input, + defer_interrupted_session_write, + extend_held_session_write, persist_session_items_for_guardrail_trip, prepare_input_with_session, reconcile_nested_history_owned_session_item_refs, @@ -184,6 +186,8 @@ save_result_to_session, save_resumed_turn_items, session_items_for_turn, + settle_held_batch_for_emptied_turn, + take_held_session_write, update_run_state_after_resume, ) from .streaming import stream_step_items_to_queue, stream_step_result_to_queue @@ -392,11 +396,19 @@ async def _save_resumed_stream_items( server_conversation_tracker=server_conversation_tracker, streamed_result=streamed_result, ): + if session is not None: + # Nothing of this run may persist (an input guardrail tripped), so the + # held batch must not outlive the run either; a detached run keeps it + # riding for the reattach instead. + take_held_session_write(run_state) return streamed_result._current_turn_persisted_item_count = await save_resumed_turn_items( run_state=run_state, session=session, items=items, + # An exit that saves nothing is not settling; the batch keeps riding (a + # re-park) or is discarded explicitly at the exit that owns that decision. + held_input=take_held_session_write(run_state) if items else None, persisted_count=streamed_result._current_turn_persisted_item_count, response_id=response_id, reasoning_item_id_policy=streamed_result._reasoning_item_id_policy, @@ -559,6 +571,11 @@ async def _finalize_streamed_final_output( owner_starts=owner_starts, blocked_message=blocked_message, ) + # The redaction derives the sanitized response from the run-state boundary, so + # the raw held batch must not be fed into this save: it could resurrect + # preambles the redaction dropped. The declaration is discarded once the + # blocked outcome is decided. + take_held_session_write(streamed_result._state) if retained_items: try: await save_items(retained_items, response_id, store_setting) @@ -760,8 +777,19 @@ async def finalize_max_turns_handler_output( output_guardrail_results: list[OutputGuardrailResult], save_items_after_guardrails: Callable[[list[RunItem]], Awaitable[None]], include_in_history: bool, + run_state: RunState | None = None, ) -> tuple[Any, RunItem]: - """Validate and finalize one synthesized max-turn handler output.""" + """Validate and finalize one synthesized max-turn handler output. + + A max-turn handler ends the run, so a held Session write still standing here has + no later gate-legal exit to settle it: it is discarded, exactly as a detached + completion discards it, so the finished run's checkpoint stays loadable and both + runners report the same terminal state. The discard waits for the run to actually + end, which is either a completed finalization or a decided blocked outcome: + validation, the final-output hooks and the guardrails can all raise, and a run that + raises may still be retried or reattached, with the executed tool's call and output + reachable only through this batch. + """ validated_output = validate_handler_final_output(agent, output) output_text = format_final_output_text(agent, validated_output) synthesized_item = create_message_output_item(agent, output_text) @@ -778,6 +806,9 @@ async def finalize_max_turns_handler_output( output_guardrail_results, ) except OutputGuardrailTripwireTriggered: + # A blocked outcome is decided and nothing of the withheld batch may reach the + # Session, exactly as every other tripwire path disposes of it. + take_held_session_write(run_state) raise except Exception as guardrail_error: guardrail_error_is_redacted = _is_error_data_redacted(guardrail_error) @@ -794,6 +825,7 @@ async def finalize_max_turns_handler_output( if redacted_persistence_error is not None: raise redacted_persistence_error from None + take_held_session_write(run_state) return validated_output, synthesized_item @@ -1396,17 +1428,64 @@ async def _save_max_turns_items( *accepted_tool_output_guardrail_results, *turn_result.tool_output_guardrail_results, ] + # A resume can interrupt again (a partial approval of a + # multi-approval response). If the gate still defers, the + # resolved items join the held batch; a detached re-park folds + # them the same way. An emptied resolved turn discards the + # batch instead: a call written without its output poisons the + # Session exactly as the orphaned output does. Mirrors the + # non-streaming path. + if session is None: + extend_held_session_write( + run_state, + run_items=turn_session_items, + reasoning_item_id_policy=( + streamed_result._reasoning_item_id_policy + ), + ) + reinterruption_items: list[RunItem] = [] + elif _should_defer_interrupted_session_items( + current_agent, + run_config, + ): + defer_interrupted_session_write( + run_state, + session, + run_items=turn_session_items, + reasoning_item_id_policy=( + streamed_result._reasoning_item_id_policy + ), + response_id=turn_result.model_response.response_id, + store=store_setting, + ) + reinterruption_items = [] + elif turn_session_items: + reinterruption_items = list(turn_session_items) + else: + # An emptied resolved turn still settles the pairs the + # batch already holds: the approved tool ran, and dropping + # its output with the filtered items would lose the + # Session's only record of that. + streamed_result._current_turn_persisted_item_count = ( + await settle_held_batch_for_emptied_turn( + run_state, + session, + persisted_count=( + streamed_result._current_turn_persisted_item_count + ), + response_id=turn_result.model_response.response_id, + reasoning_item_id_policy=( + streamed_result._reasoning_item_id_policy + ), + store=store_setting, + wrapper=streamed_result.context_wrapper, + ) + ) + reinterruption_items = [] await _finalize_streamed_interruption( streamed_result=streamed_result, save_items=_save_resumed_items, - items=( - [] - if _should_defer_interrupted_session_items( - current_agent, - run_config, - ) - else list(turn_session_items) - ), + items=reinterruption_items, response_id=turn_result.model_response.response_id, store_setting=store_setting, interruptions=approvals_from_step(turn_result.next_step), @@ -1428,8 +1507,39 @@ async def _save_max_turns_items( if run_state is not None: run_state._current_agent = current_agent _publish_streamed_result_agent(streamed_result, current_agent) + # A detached exit folds the resolved items into the held batch; + # an emptied resolved turn (a handoff input_filter can drop + # every item) discards the batch instead: a call written + # without its output poisons the Session just as the orphaned + # output does. + if session is None: + extend_held_session_write( + run_state, + run_items=turn_session_items, + reasoning_item_id_policy=( + streamed_result._reasoning_item_id_policy + ), + ) + elif not turn_session_items: + # An emptied resolved turn still settles the pairs the + # batch already holds; only the unpaired requests drop. + streamed_result._current_turn_persisted_item_count = ( + await settle_held_batch_for_emptied_turn( + run_state, + session, + persisted_count=( + streamed_result._current_turn_persisted_item_count + ), + response_id=turn_result.model_response.response_id, + reasoning_item_id_policy=( + streamed_result._reasoning_item_id_policy + ), + store=store_setting, + wrapper=streamed_result.context_wrapper, + ) + ) await _save_resumed_items( - list(turn_session_items), + list(turn_session_items) if turn_session_items else [], turn_result.model_response.response_id, store_setting, ) @@ -1467,12 +1577,53 @@ async def _save_max_turns_items( ) if streamed_result._stored_exception is not None: break + if session is None: + # A detached final output has no Session to settle against + # and the run ends here, so the batch is discarded rather + # than left to invalidate the completed run's checkpoint. + # Only here, though: the finalization above runs the hooks, + # the guardrails and the final save, any of which can raise, + # and a run that raises may still be retried or reattached + # with the executed tool's call and output reachable only + # through this batch. + take_held_session_write(run_state) run_state._current_step = None break if isinstance(turn_result.next_step, NextStepRunAgain): + # A detached exit folds the resolved items into the held batch; + # an emptied resolved turn (a handoff input_filter can drop + # every item) discards the batch instead: a call written + # without its output poisons the Session just as the orphaned + # output does. + if session is None: + extend_held_session_write( + run_state, + run_items=turn_session_items, + reasoning_item_id_policy=( + streamed_result._reasoning_item_id_policy + ), + ) + elif not turn_session_items: + # An emptied resolved turn still settles the pairs the + # batch already holds; only the unpaired requests drop. + streamed_result._current_turn_persisted_item_count = ( + await settle_held_batch_for_emptied_turn( + run_state, + session, + persisted_count=( + streamed_result._current_turn_persisted_item_count + ), + response_id=turn_result.model_response.response_id, + reasoning_item_id_policy=( + streamed_result._reasoning_item_id_policy + ), + store=store_setting, + wrapper=streamed_result.context_wrapper, + ) + ) await _save_resumed_items( - list(turn_session_items), + list(turn_session_items) if turn_session_items else [], turn_result.model_response.response_id, store_setting, ) @@ -1685,6 +1836,11 @@ def _record_max_turns_handler_output( break streamed_result._max_turns_handled = True streamed_result.current_turn = max_turns + # A max-turn handler ends the run, so a held Session write still + # standing has no later gate-legal exit to settle it. Discarding it + # keeps the finished run's checkpoint loadable and matches the + # non-streaming runner, which reports the same terminal state. + take_held_session_write(run_state) if run_state is not None and not is_resumed_state: run_state._current_turn = max_turns run_state._current_step = None @@ -1926,6 +2082,15 @@ def _record_max_turns_handler_output( ) if streamed_result._stored_exception is not None: break + if session is None: + # A detached completion has no Session to settle against and + # the run ends here, so the batch is discarded rather than + # left to invalidate the completed run's checkpoint. Only here, + # though: the finalization above runs the hooks, the guardrails + # and the final save, any of which can raise, and a run that + # raises may still be retried or reattached with the executed + # tool's call and output reachable only through this batch. + take_held_session_write(run_state) if run_state is not None: run_state._current_step = None break @@ -1944,17 +2109,40 @@ def _record_max_turns_handler_output( run_state._current_turn_persisted_item_count = ( streamed_result._current_turn_persisted_item_count ) + parked_items_deferred = _should_defer_interrupted_session_items( + current_agent, + run_config, + ) + if session is None: + # A fresh park during a detached resume cannot write, but a + # standing held declaration carries the session identity: the + # new parked call folds into it so the reattach does not + # settle its output orphaned. + extend_held_session_write( + run_state, + run_items=turn_session_items, + reasoning_item_id_policy=(streamed_result._reasoning_item_id_policy), + ) + elif parked_items_deferred and await _should_persist_stream_items( + session=session, + server_conversation_tracker=server_conversation_tracker, + streamed_result=streamed_result, + ): + # The gate withholds this write until the output guardrails + # decide; declaring the batch on the checkpoint lets a resume + # settle it at a gate-legal exit instead of losing it. + defer_interrupted_session_write( + run_state, + session, + run_items=turn_session_items, + reasoning_item_id_policy=(streamed_result._reasoning_item_id_policy), + response_id=turn_result.model_response.response_id, + store=store_setting, + ) await _finalize_streamed_interruption( streamed_result=streamed_result, save_items=_save_stream_items_with_count, - items=( - [] - if _should_defer_interrupted_session_items( - current_agent, - run_config, - ) - else turn_session_items - ), + items=([] if parked_items_deferred else turn_session_items), response_id=turn_result.model_response.response_id, store_setting=store_setting, interruptions=approvals_from_step(turn_result.next_step), diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py index f8d4f83b3c..efee745c03 100644 --- a/src/agents/run_internal/session_persistence.py +++ b/src/agents/run_internal/session_persistence.py @@ -15,11 +15,13 @@ from typing import Any, cast from .. import _debug +from .._tool_identity import get_hosted_mcp_approval_request_identity from ..exceptions import UserError from ..items import ( HandoffOutputItem, InputItem, ItemHelpers, + MCPApprovalResponseItem, ModelResponse, RunItem, ToolCallOutputItem, @@ -41,8 +43,9 @@ from ..memory.session import _call_session_method, _get_session_wrapper from ..models.fake_id import FAKE_RESPONSES_ID from ..run_context import RunContextWrapper -from ..run_state import RunState +from ..run_state import RunState, _PendingSessionWrite from .items import ( + _TOOL_CALL_TO_OUTPUT_TYPE, NestedHistoryOwnedItem, NestedHistoryOwnedItemRef, ReasoningItemIdPolicy, @@ -80,6 +83,10 @@ "resumed_turn_items", "save_result_to_session", "save_resumed_turn_items", + "defer_interrupted_session_write", + "extend_held_session_write", + "take_held_session_write", + "settle_held_batch_for_emptied_turn", "resume_pending_session_write", "update_run_state_after_resume", "rewind_session_items", @@ -89,6 +96,15 @@ _SESSION_LIMIT_UNSET = object() +# Serialized item types produced locally as the continuation of a model response: the +# output kinds of the canonical call-to-output map, plus the hosted MCP approval +# response, which is the locally produced half of its approval pair. Compaction for +# the response that carried the request must be deferred while any of these still +# needs to be associated with that response chain. +_LOCAL_CONTINUATION_OUTPUT_TYPES = frozenset(_TOOL_CALL_TO_OUTPUT_TYPE.values()) | { + "mcp_approval_response" +} + async def admit_pending_input( *, @@ -194,6 +210,29 @@ def retain_accepted_admissions(items: list[RunItem]) -> None: return True +def _session_method_accepts_limit(method: Any) -> bool: + """Return whether a ``get_items`` implementation can be passed ``limit``. + + A structural ``Session`` written against a pre-``limit`` release may declare + ``get_items(self)`` alone; passing ``limit`` to it raises ``TypeError`` and turns + every internal tail read into a hard failure. When the signature cannot be + inspected, assume the released shape. + """ + try: + parameters = inspect.signature(method).parameters.values() + except Exception: + return True + return any( + ( + parameter.name == "limit" + and parameter.kind + in (inspect.Parameter.POSITIONAL_OR_KEYWORD, inspect.Parameter.KEYWORD_ONLY) + ) + or parameter.kind is inspect.Parameter.VAR_KEYWORD + for parameter in parameters + ) + + async def _session_get_items( session: Session, limit: int | None | object = _SESSION_LIMIT_UNSET, @@ -207,6 +246,12 @@ async def _session_get_items( async def read_items() -> list[TResponseInputItem]: if limit is _SESSION_LIMIT_UNSET: result = await _call_session_method(session.get_items, wrapper=session_wrapper) + elif not _session_method_accepts_limit(session.get_items): + # Fall back to a full read and apply the released ``limit`` semantics + # locally: the latest ``limit`` items in chronological order. + result = await _call_session_method(session.get_items, wrapper=session_wrapper) + if isinstance(limit, int): + result = list(result)[-limit:] if limit > 0 else [] else: result = await _call_session_method( session.get_items, limit=limit, wrapper=session_wrapper @@ -598,13 +643,22 @@ async def save_result_to_session( store: bool | None = None, wrapper: RunContextWrapper[Any] | None = None, resumed_write_state: RunState | None = None, + settling_held_batch: bool = False, ) -> int: """ Persist a turn to the session store, keeping track of what was already saved so retries during streaming do not duplicate tool outputs or inputs. + ``settling_held_batch`` marks the calls that carry a withheld batch through + ``original_input``; only those look for local tool outputs in that slot, because on + an ordinary save the same slot holds the caller's own input. Those calls also count + the batch items this append actually wrote, which is not the batch's raw length: a + resolved turn re-delivers the outputs the batch already folded in and they dedup + away here. + Returns: - The number of new run items persisted for this call. + The number of new run items persisted for this call, plus the settled batch + items when ``settling_held_batch`` is set. """ already_persisted = run_state._current_turn_persisted_item_count if run_state is not None else 0 @@ -716,9 +770,26 @@ async def save_result_to_session( if run_state is not None: run_state._current_turn_persisted_item_count = already_persisted + saved_run_items_count + # The append wrote every deduplicated item; the ones that are not surviving run + # items are the settled batch's own, counted here because only this scope knows + # what the dedup kept. + settled_batch_items = len(items_to_save) - saved_run_items_count if settling_held_batch else 0 + if response_id and is_openai_responses_compaction_aware_session(session): + # A settling held batch carries its tool outputs as already-converted input + # items through ``original_input``, so looking only at ``new_items`` would + # report no local tool output and compact the very response whose outputs just + # landed. Only a settle reads that slot: on an ordinary save it holds the + # caller's input, whose earlier outputs say nothing about this response. has_local_tool_outputs = any( - isinstance(item, ToolCallOutputItem | HandoffOutputItem) for item in new_items + isinstance(item, ToolCallOutputItem | HandoffOutputItem | MCPApprovalResponseItem) + for item in new_items + ) or ( + settling_held_batch + and any( + isinstance(item, dict) and item.get("type") in _LOCAL_CONTINUATION_OUTPUT_TYPES + for item in items_to_save + ) ) if has_local_tool_outputs: defer_compaction = getattr(session, "_defer_compaction", None) @@ -733,7 +804,7 @@ async def save_result_to_session( "skip: deferring compaction for response %s due to local tool outputs", response_id, ) - return saved_run_items_count + return saved_run_items_count + settled_batch_items deferred_response_id = None get_deferred = getattr(session, "_get_deferred_compaction_response_id", None) @@ -758,7 +829,7 @@ async def save_result_to_session( wrapper=compaction_wrapper, ) - return saved_run_items_count + return saved_run_items_count + settled_batch_items async def save_resumed_turn_items( @@ -771,29 +842,358 @@ async def save_resumed_turn_items( store: bool | None = None, wrapper: RunContextWrapper[Any] | None = None, run_state: RunState | None = None, + held_input: Sequence[TResponseInputItem] | None = None, ) -> int: - """Persist resumed turn items and return the updated persisted count.""" - if session is None or not items: + """Persist resumed turn items and return the updated persisted count. + + ``held_input`` carries a claimed held batch (see ``take_held_session_write``) into + the same append as the resolved turn's items, ahead of them. One ordered write + keeps the interrupted ``function_call`` before its output and lets the whole batch + register as the one pending append with digest-based crash recovery; settling the + batch separately would either trip the single-slot rule or advance the persisted + count and slice the resolved items out of their own save. + """ + if session is None or (not items and not held_input): return persisted_count + # Whether this settle is claiming a held batch at all, captured before the dedup + # below can empty it: the recovery registration must stay armed even when the + # guardrail rebuild already carries the batch, because the append still lands the + # approved call and output and a crash inside it must reconcile on retry. + settling_held = bool(held_input) + if held_input and final_items_cover_held_batch(items, held_input, reasoning_item_id_policy): + # The guardrail rebuild re-derived the whole current response, held requests + # included; feeding the batch again would duplicate its unkeyed companions. + # Resolved-turn saves never carry request kinds, so this only fires on the + # final sweep. + held_input = None + if held_input: + held_input = _held_items_safe_to_settle( + held_input, + items, + reasoning_item_id_policy, + pending_call_ids=_pending_approval_call_ids(run_state), + ) saved_count = await save_result_to_session( session, - [], + list(held_input) if held_input else [], list(items), None, response_id=response_id, reasoning_item_id_policy=reasoning_item_id_policy, store=store, wrapper=wrapper, + settling_held_batch=settling_held, resumed_write_state=( run_state if run_state is not None - and isinstance(run_state._current_step, NextStepRunAgain | NextStepInterruption) + and ( + isinstance(run_state._current_step, NextStepRunAgain | NextStepInterruption) + # A settling held batch always registers, so a crash inside the append + # fails closed with the batch recorded instead of silently losing the + # only copy of an approved tool's call and output. + or settling_held + ) else None ), ) + # Settled held items are this turn's persisted items too. Leaving them uncounted + # would let a later gate-enabled resume pass the resumed-safety validation with a + # zero count and re-append the stored calls through the final sweep; counting them + # makes that resume fail fast on the existing persisted-items refusal instead. The + # append reports them itself, because the raw batch length overcounts whatever the + # dedup dropped and this count slices a later save of the same turn. return persisted_count + saved_count +async def settle_held_batch_for_emptied_turn( + run_state: RunState | None, + session: Session | None, + *, + persisted_count: int, + response_id: str | None, + reasoning_item_id_policy: ReasoningItemIdPolicy | None = None, + store: bool | None = None, + wrapper: RunContextWrapper[Any] | None = None, +) -> int: + """Settle the paired part of a held batch whose resolved turn came back empty. + + A handoff ``input_filter`` can drop every resolved item, but an approved tool has + already run by then and its output was folded into the batch. Emptiness of the turn + is therefore the wrong predicate: pairing is. The executed call and output settle + together and the unpaired requests drop, exactly as every other settle decides it, + so the Session keeps the only record that the tool ran and the next run does not + re-issue its side effect. + """ + return await save_resumed_turn_items( + run_state=run_state, + session=session, + items=[], + held_input=take_held_session_write(run_state), + persisted_count=persisted_count, + response_id=response_id, + reasoning_item_id_policy=reasoning_item_id_policy, + store=store, + wrapper=wrapper, + ) + + +def _held_pair_identity(item: TResponseInputItem | None) -> tuple[str, str] | None: + """Return the pairing key a held request kind must find an output for. + + The tool-call families come from the canonical ``_TOOL_CALL_TO_OUTPUT_TYPE`` map + in ``run_internal.items``, which owns the call-to-output pairing rule; hosted MCP + approvals pair outside that map, keyed by the canonical request identity with the + response pointing back via ``approval_request_id``. + """ + if not isinstance(item, dict): + return None + item_type = item.get("type") + if item_type == "mcp_approval_request": + identity = get_hosted_mcp_approval_request_identity(item) + request_id = identity.request_id if identity is not None else None + return (item_type, request_id) if request_id else None + if item_type not in _TOOL_CALL_TO_OUTPUT_TYPE: + return None + call_id = item.get("call_id") + return (item_type, call_id) if isinstance(call_id, str) and call_id else None + + +def _pending_approval_call_ids(run_state: RunState | None) -> set[str]: + """Return the ids still awaiting approval on the state's current step.""" + if run_state is None or not isinstance(run_state._current_step, NextStepInterruption): + return set() + ids: set[str] = set() + for approval in run_state._current_step.interruptions: + raw = getattr(approval, "raw_item", None) + for field in ("call_id", "id"): + value = raw.get(field) if isinstance(raw, dict) else getattr(raw, field, None) + if isinstance(value, str) and value: + ids.add(value) + return ids + + +def _held_items_safe_to_settle( + held_items: Sequence[TResponseInputItem], + run_items: Sequence[RunItem], + reasoning_item_id_policy: ReasoningItemIdPolicy | None, + *, + pending_call_ids: set[str] | None = None, +) -> list[TResponseInputItem]: + """Drop held calls whose outputs did not survive into the settling batch. + + A handoff ``input_filter`` may drop some resolved outputs while keeping others, so + the settling batch being non-empty does not make it safe: a held ``function_call`` + settled without its output poisons the Session exactly as the orphaned output does. + Pairing is the safety predicate. The paired part of the batch still settles, which + honors the filter's decision symmetrically: a dropped output takes its call with it, + and a kept output keeps its call. + + ``pending_call_ids`` names calls whose approvals are still open on the current + step: their outputs are missing because they have not run yet, not because a + filter removed them, so they settle now and pair up at a later exit, exactly as a + non-deferred park persists a call before its output exists. + """ + pending_call_ids = pending_call_ids or set() + working: list[TResponseInputItem] = list(held_items) + # A call whose approval is still open is exempt from the orphan prune; the prune + # only understands outputs, so the exemption rides in as a placeholder output that + # is discarded with the rest of the context below. + for item in held_items: + key = _held_pair_identity(item) + if key is not None and key[1] in pending_call_ids and key[0] != "mcp_approval_request": + working.append( + cast( + TResponseInputItem, + {"type": _TOOL_CALL_TO_OUTPUT_TYPE[key[0]], "call_id": key[1]}, + ) + ) + for run_item in run_items: + converted = run_item_to_input_item(run_item, reasoning_item_id_policy) + if converted is not None: + working.append(converted) + pruned = drop_orphan_function_calls(working) + surviving = {id(item) for item in pruned} + + # Hosted MCP approvals pair outside the canonical map: a request settles only with + # its response present or its approval still open. + mcp_response_ids = { + item.get("approval_request_id") + for item in working + if isinstance(item, dict) and item.get("type") == "mcp_approval_response" + } | pending_call_ids + + kept: list[TResponseInputItem] = [] + for item in held_items: + if isinstance(item, dict) and item.get("type") == "mcp_approval_request": + key = _held_pair_identity(item) + if key is not None and key[1] not in mcp_response_ids: + continue + kept.append(item) + elif id(item) in surviving: + kept.append(item) + return kept + + +def final_items_cover_held_batch( + items: Sequence[RunItem], + held_input: Sequence[TResponseInputItem], + reasoning_item_id_policy: ReasoningItemIdPolicy | None, +) -> bool: + """Return whether the final batch already carries the held batch's requests. + + With output guardrails the final sweep rebuilds the whole current response, held + requests included, and the deduplication cannot key the batch's unkeyed companions + (an assistant preamble, an id-less reasoning item), so feeding the batch again + would duplicate them. When every held request already appears in the final items + the whole batch is redundant; without guardrails the sweep returns the resolved + items verbatim, no held request appears there, and the batch must ride in. + """ + final_ids: set[str] = set() + for run_item in items: + converted = run_item_to_input_item(run_item, reasoning_item_id_policy) + key = _held_pair_identity(converted) if converted is not None else None + if key is not None: + final_ids.add(key[1]) + for item in held_input: + key = _held_pair_identity(item) + if key is not None and key[1] not in final_ids: + return False + return True + + +def defer_interrupted_session_write( + run_state: RunState, + session: Session | None, + *, + run_items: Sequence[RunItem], + reasoning_item_id_policy: ReasoningItemIdPolicy | None = None, + response_id: str | None = None, + store: bool | None = None, +) -> None: + """Register the interruption's withheld batch as a held pending Session write. + + Registering is not writing: this touches only the checkpoint, never the Session, + so the output-guardrail persistence gate stays intact. The batch settles, extends + or is discarded only at a gate-legal point of a later resume. A standing held + record is replaced by the superset of both batches, so a repeated park (a partial + approval interrupting again) keeps one canonical batch. A standing record that is + not held means a resumed append is mid-flight, which the existing single-slot rule + treats as a caller bug. + + Items are converted and deduplicated with the same helpers the real save uses, and + the count is taken over the converted items: approval placeholders drop out in + conversion, so counting the raw run items would corrupt the persisted count. A + detached re-park has no Session and takes its ``session_id`` from the standing + declaration. + + The batch carries only the withheld response: run input is never registered here, + because the gate withholds model output, not the user's accepted input, and a + tripwire discards the batch without inspecting it. + """ + pending = run_state._pending_session_write + if pending is not None and not pending.get("held"): + raise UserError("Resolve the pending Session write before saving another batch") + + # The normal persistence path forces the reasoning-id policy to ``None`` for a + # Conversations backend so a server-identified reasoning item stays persistable; + # the registration conversion must match or the sanitization later drops it. A + # standing record owns the policy its items were converted under, so a re-park + # folds new items under the same conversion instead of the resuming run's own. + if isinstance(session, OpenAIConversationsSession): + reasoning_item_id_policy = None + if pending is not None and "reasoning_item_id_policy" in pending: + reasoning_item_id_policy = pending["reasoning_item_id_policy"] + converted_run_items: list[TResponseInputItem] = [] + for run_item in run_items: + as_input = run_item_to_input_item(run_item, reasoning_item_id_policy) + if as_input is None: + continue + converted_run_items.append(ensure_input_item_format(as_input)) + + base_items = list(pending["items"]) if pending is not None else [] + items = deduplicate_input_items_preferring_latest(base_items + converted_run_items) + if isinstance(session, OpenAIConversationsSession): + items = [_sanitize_openai_conversation_item(item) for item in items] + items = [item for item in items if not _is_unpersistable_for_openai_conversation(item)] + if not items: + return + + session_id = ( + session.session_id + if session is not None + else (pending["session_id"] if pending is not None else None) + ) + if session_id is None: + return + record: _PendingSessionWrite = { + "session_id": session_id, + "items": copy.deepcopy(items), + "before": None, + "persisted_count": ( + run_state._current_turn_persisted_item_count + len(converted_run_items) + ), + "held": True, + # The response the withheld batch belongs to, so the settle can run the same + # compaction bookkeeping the ordinary persistence path runs for it. An extend + # keeps the original response: the batch is that response's write, and the + # settle resolves its compaction mode from that response's own storage setting. + # Presence decides, not truthiness: a park under the ordinary ``store=None`` + # records a real value, and letting a re-interruption's setting overwrite it + # would resolve the original response's compaction mode from the wrong turn. + "response_id": pending["response_id"] + if (pending is not None and "response_id" in pending) + else response_id, + "store": pending["store"] if (pending is not None and "store" in pending) else store, + "reasoning_item_id_policy": reasoning_item_id_policy, + } + run_state._pending_session_write = record + + +def extend_held_session_write( + run_state: RunState | None, + *, + run_items: Sequence[RunItem], + reasoning_item_id_policy: ReasoningItemIdPolicy | None = None, +) -> None: + """Fold a detached exit's resolved items into the standing held batch. + + With no Session attached the resolved turn's save is a no-op, so the executed + tool output exists only in this process; folding it into the held batch lets the + reattaching resume settle call and output together. Does nothing when no held + batch stands. The fold converts under the batch's registration policy, not the + caller's: a detached run cannot see the original backend, and a server reasoning + id stripped here could not be restored at the settle. + """ + if run_state is None or run_state._pending_session_write is None: + return + if not run_state._pending_session_write.get("held"): + return + defer_interrupted_session_write( + run_state, + None, + run_items=run_items, + reasoning_item_id_policy=reasoning_item_id_policy, + ) + + +def take_held_session_write(run_state: RunState | None) -> list[TResponseInputItem]: + """Claim the standing held batch for a settling write and free the slot. + + The caller must hand the returned items to a Session save in the same exit + (``held_input`` on ``save_resumed_turn_items``, or the input positional of + ``save_result_to_session``), or drop them deliberately when the exit's contract is + to discard the batch. The slot is freed first so the settling write can register + itself as the one pending append and inherit the digest-based crash recovery. + """ + if run_state is None: + return [] + pending = run_state._pending_session_write + if pending is None or not pending.get("held"): + return [] + run_state._pending_session_write = None + return list(pending["items"]) + + async def resume_pending_session_write( run_state: RunState, session: Session | None, @@ -809,6 +1209,53 @@ async def resume_pending_session_write( pending = run_state._pending_session_write if pending is None: return + if pending.get("held"): + # A held batch is the write the interruption park withheld under the + # output-guardrail gate, and resume entry is not a gate-legal settle point, so + # the declaration rides the checkpoint untouched; in particular a detached + # resume must not fail the boot over a batch it cannot settle. An attached + # Session must still be the declared one, and must fail here at boot: letting + # the run proceed would execute the approved tool and settle the batch into + # the wrong conversation. The exception to riding is a run-again checkpoint: + # the parked response's outputs already went back to the model, which only + # happens after the gate stopped applying to that response, so the batch + # settles here, before the next model call. This is also the only settle + # point such a checkpoint will ever reach, because the run-again turn's saves + # never arm ``resumed_write_state``. + if session is not None and session.session_id != pending["session_id"]: + raise UserError( + "Resume the pending Session write with the original Session and session ID" + ) + if session is None or not isinstance(run_state._current_step, NextStepRunAgain): + return + # The entry settle offers the batch with no accompanying resolved items, so the + # pairing contract applies against the batch alone: a call whose output a + # detached handoff filter dropped must not land dangling here either. + settling = _held_items_safe_to_settle(pending["items"], [], None) + response_id = pending.get("response_id") + settle_store = pending.get("store") + run_state._pending_session_write = None + if not settling: + return + # Settle through the canonical persistence path rather than appending behind + # its back: it owns the Conversations sanitization, the ordered dedup, the + # pending-write registration that makes a failed append recoverable, and the + # compaction bookkeeping for the response this batch belongs to. The slot is + # released first, so the re-entry this causes (``save_result_to_session`` + # registers the batch and calls back into here) sees an ordinary pending write + # and takes the append-and-reconcile path below, never this branch again. + await save_result_to_session( + session, + settling, + [], + run_state, + response_id=response_id, + store=settle_store, + wrapper=wrapper, + settling_held_batch=True, + resumed_write_state=run_state, + ) + return if run_state._session_write_in_progress: raise UserError("The pending Session write is already in progress for this RunState") if session is None or session.session_id != pending["session_id"]: diff --git a/src/agents/run_internal/turn_resolution.py b/src/agents/run_internal/turn_resolution.py index ae9fec7619..7bdc39e237 100644 --- a/src/agents/run_internal/turn_resolution.py +++ b/src/agents/run_internal/turn_resolution.py @@ -152,6 +152,7 @@ ToolRunMCPApprovalRequest, ToolRunShellCall, ) +from .session_persistence import extend_held_session_write from .tool_caller import ensure_programmatic_tool_call_parent, ensure_tool_caller_allowed from .tool_execution import ( build_litellm_json_tool_call, @@ -2457,6 +2458,17 @@ def _commit_tool_output(item: RunItem) -> None: ) if run_state is not None: run_state._generated_items = [*original_pre_step_items, *committed_tool_outputs] + # The approved tool's side effect is done and its output is committed, so + # the withheld batch takes it at this boundary rather than at the turn + # exit. A post-output callback that raises (``custom_data_extractor``, + # ``on_tool_end``) leaves a retry that skips the completed invocation and + # produces no new session items, and the batch would otherwise settle, or + # be discarded as an emptied turn, without the output the tool produced. + extend_held_session_write( + run_state, + run_items=[item], + reasoning_item_id_policy=run_state._reasoning_item_id_policy, + ) _register_tool_call_items(context_wrapper, [item]) ( diff --git a/src/agents/run_state.py b/src/agents/run_state.py index d79e0781a4..f8564e63d5 100644 --- a/src/agents/run_state.py +++ b/src/agents/run_state.py @@ -36,7 +36,7 @@ ProgramOutput, ) from pydantic import BaseModel, StringConstraints, TypeAdapter, ValidationError -from typing_extensions import TypedDict, TypeVar +from typing_extensions import NotRequired, TypedDict, TypeVar from ._run_state_agent_identity import ( _build_agent_identity_keys_by_id, @@ -109,6 +109,7 @@ from .run_context import RunContextWrapper from .run_internal.items import ( NestedHistoryOwnedItemRef, + ReasoningItemIdPolicy, digest_input_item, ensure_nested_history_run_item_occurrence_key, nested_history_run_item_occurrence_key, @@ -168,12 +169,34 @@ class _PendingSessionWrite(TypedDict): - """One canonical resumed-output append awaiting acknowledgement.""" + """One canonical resumed-output append awaiting acknowledgement. + + ``held`` marks a batch the interruption park withheld because the agent's output + guardrails had not approved the turn yet. A held batch was never offered to the + Session, so ``before`` stays ``None`` until a gate-legal exit starts settling it; + from that point it is an ordinary pending write and the digest reconciliation + recovers a half-acknowledged append. Absent or ``False`` keeps the released + meaning: an append already approved for eager settlement on resume entry. + + ``response_id`` records the model response the withheld batch belongs to, and + ``store`` the store setting that response was produced under, so the settle runs + the same compaction bookkeeping the ordinary persistence path would have run for + it instead of appending behind its back. + + ``reasoning_item_id_policy`` records how the batch's items were converted, so a + detached re-park folds new items under the same conversion: a Conversations-origin + batch preserves server reasoning ids even when the resuming run's own policy would + omit them, and an id stripped at registration cannot be restored at the settle. + """ session_id: str items: list[TResponseInputItem] before: list[str] | None persisted_count: int + held: NotRequired[bool] + response_id: NotRequired[str | None] + store: NotRequired[bool | None] + reasoning_item_id_policy: NotRequired[ReasoningItemIdPolicy | None] def _default_run_state_validation_error( @@ -190,10 +213,11 @@ def _default_run_state_validation_error( # 3. to_json() always emits CURRENT_SCHEMA_VERSION. # 4. Forward compatibility is intentionally fail-fast (older SDKs reject newer or unsupported # versions). -CURRENT_SCHEMA_VERSION = "1.17" +CURRENT_SCHEMA_VERSION = "1.18" _PROGRAMMATIC_TOOL_CALLING_MIN_SCHEMA_VERSION = "1.13" _HOSTED_MCP_APPROVALS_MIN_SCHEMA_VERSION = "1.14" _CURRENT_RESPONSE_OWNERSHIP_MIN_SCHEMA_VERSION = "1.17" +_HELD_PENDING_SESSION_WRITE_MIN_SCHEMA_VERSION = "1.18" # 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.", @@ -229,6 +253,11 @@ def _default_run_state_validation_error( "Persists Docker container labels and current-response generated-item ownership across " "resume flows, including pending resumed Session writes and terminal-unrecoverable runs." ), + "1.18": ( + "Persists the interrupted turn's withheld Session write, including the response it " + "belongs to and the conversion policy its items were registered under, so an " + "approval resume can settle it under the output-guardrail gate." + ), } SUPPORTED_SCHEMA_VERSIONS = frozenset(SCHEMA_VERSION_SUMMARIES) @@ -4372,11 +4401,52 @@ async def _build_run_state_from_json( if pending_write is not None: from .run_internal.run_steps import NextStepInterruption, NextStepRunAgain + # 1.17 defines this object as exactly four keys and its readers are already on + # main, so writing the held variant under that label would emit checkpoints + # those readers reject. The held keys are therefore gated to the version that + # introduced them, and a 1.17 payload keeps the four keys it defined and + # settles eagerly as it always did. + held_keys_allowed = (schema_major, schema_minor) >= tuple( + int(part) + for part in _HELD_PENDING_SESSION_WRITE_MIN_SCHEMA_VERSION.split(".", maxsplit=1) + ) + base_keys = {"session_id", "items", "before", "persisted_count"} + held_keys = ( + {"held", "response_id", "store", "reasoning_item_id_policy"} + if held_keys_allowed + else set() + ) if ( (schema_major, schema_minor) < (1, 17) or not isinstance(state._current_step, NextStepRunAgain | NextStepInterruption) or not isinstance(pending_write, dict) - or set(pending_write) != {"session_id", "items", "before", "persisted_count"} + or set(pending_write) - held_keys != base_keys + or ("held" in pending_write and type(pending_write["held"]) is not bool) + or (pending_write.get("held") is True and pending_write.get("before") is not None) + or ( + "response_id" in pending_write + and not isinstance(pending_write["response_id"], str | type(None)) + ) + or ( + "store" in pending_write + and pending_write["store"] is not None + and type(pending_write["store"]) is not bool + ) + or ( + "reasoning_item_id_policy" in pending_write + and pending_write["reasoning_item_id_policy"] not in (None, "preserve", "omit") + ) + # These keys describe the withheld batch, so they are meaningless on an + # ordinary pending write and are refused there rather than restored as + # state nothing consumes. + or ( + not pending_write.get("held") + and ( + "response_id" in pending_write + or "store" in pending_write + or "reasoning_item_id_policy" in pending_write + ) + ) or not isinstance(pending_write.get("session_id"), str) or not isinstance(pending_write.get("items"), list) or not pending_write["items"] diff --git a/tests/fixtures/run_state/README.md b/tests/fixtures/run_state/README.md index 82836b310d..85655acf48 100644 --- a/tests/fixtures/run_state/README.md +++ b/tests/fixtures/run_state/README.md @@ -1,6 +1,6 @@ # RunState compatibility corpus -The `minimal/` fixtures cover every schema version accepted by the current reader. The `features/` fixtures cover the schema-bearing behavior introduced in versions 1.2 through 1.17. The `resume/` fixture records an actual pending function-tool approval emitted by the v0.19.4 writer, and the `security/` fixture records that writer's credential-bearing sandbox state. `sources.json` records the source commit and provenance for every fixture. +The `minimal/` fixtures cover every schema version accepted by the current reader. The `features/` fixtures cover the schema-bearing behavior introduced in versions 1.2 through 1.18. The `resume/` fixture records an actual pending function-tool approval emitted by the v0.19.4 writer, and the `security/` fixture records that writer's credential-bearing sandbox state. `sources.json` records the source commit and provenance for every fixture. Regenerate the feature corpus from the recorded historical source trees with: @@ -10,6 +10,6 @@ UV_DEFAULT_INDEX=https://pypi.org/simple uv run python tests/fixtures/run_state/ The generator extracts each recorded commit with `git archive` and runs that commit's writer in a fresh locked environment. It does not import the current checkout. -Versions 1.7, 1.8, 1.16, and 1.17 are explicit exceptions. Release-boundary schema renumbering assigned duplicate-agent/sandbox state to 1.7 and prompt-cache state to 1.8 without any writer commit that emitted those final version numbers. The 1.16 schema transition likewise has no retained writer commit, but the retained 1.15 writer produces the same minimal and per-call-override payloads. The 1.17 labels transition uses the labels-capable 1.16 writer and changes only the schema label. These fixtures are therefore marked `canonical_compatibility`: the recorded writer produces the payload, and the generator changes only the schema label so the corresponding reader branch remains covered. They must not be represented as historical-writer output. +Versions 1.7, 1.8, 1.16, 1.17, and 1.18 are explicit exceptions. Release-boundary schema renumbering assigned duplicate-agent/sandbox state to 1.7 and prompt-cache state to 1.8 without any writer commit that emitted those final version numbers. The 1.16 schema transition likewise has no retained writer commit, but the retained 1.15 writer produces the same minimal and per-call-override payloads. The 1.17 labels transition uses the labels-capable 1.16 writer and changes only the schema label. The 1.18 held-pending-write transition uses the held-capable 1.17 writer and changes only the schema label; the 1.17 reader intentionally rejects that batch shape under its own label. These fixtures are therefore marked `canonical_compatibility`: the recorded writer produces the payload, and the generator changes only the schema label so the corresponding reader branch remains covered. They must not be represented as historical-writer output. Ordinary tests never run the generator. They read the frozen payloads, compare every durable field emitted by the historical writer across the upgrade, rewrite to the current schema, and verify that the rewritten form is idempotent. They also approve and reject the historical pending interruption through actual `Runner` resumes. The schema version itself is the only normalization for the ordinary corpus; fields added by newer writers may be absent from an older payload, but every field present in that payload must survive. The security fixture has one explicit migration normalization: persisted mount credentials and opaque driver options are removed and the trusted-rebind marker is added. All non-authority topology remains part of the comparison. diff --git a/tests/fixtures/run_state/features/v1_18_held_pending_session_write.json b/tests/fixtures/run_state/features/v1_18_held_pending_session_write.json new file mode 100644 index 0000000000..c0487e666a --- /dev/null +++ b/tests/fixtures/run_state/features/v1_18_held_pending_session_write.json @@ -0,0 +1,81 @@ +{ + "$schemaVersion": "1.18", + "auto_previous_response_id": false, + "context": { + "approvals": {}, + "context": {}, + "context_meta": { + "omitted": false, + "original_type": "mapping", + "requires_deserializer": false, + "serialized_via": "mapping" + }, + "tool_invocations": {}, + "usage": { + "input_tokens": 0, + "input_tokens_details": [ + { + "cache_write_tokens": 0, + "cached_tokens": 0 + } + ], + "output_tokens": 0, + "output_tokens_details": [ + { + "reasoning_tokens": 0 + } + ], + "request_usage_entries": [], + "requests": 0, + "total_tokens": 0 + } + }, + "conversation_id": null, + "current_agent": { + "name": "compat-agent" + }, + "current_step": { + "type": "next_step_run_again" + }, + "current_turn": 0, + "current_turn_persisted_item_count": 0, + "generated_items": [], + "generated_prompt_cache_key": null, + "generated_session_item_indexes": [], + "input_guardrail_results": [], + "last_model_response": null, + "last_processed_response": null, + "max_turns": 10, + "model_responses": [], + "nested_history_owned_session_item_refs": [], + "no_active_agent_run": true, + "original_input": "historical input", + "output_guardrail_results": [], + "pending_input": [], + "pending_session_write": { + "before": null, + "held": true, + "items": [ + { + "arguments": "{}", + "call_id": "call_held_1", + "name": "write_thing", + "type": "function_call" + }, + { + "call_id": "call_held_1", + "output": "wrote", + "type": "function_call_output" + } + ], + "persisted_count": 2, + "session_id": "session-118" + }, + "previous_response_id": null, + "reasoning_item_id_policy": null, + "session_items": [], + "tool_input_guardrail_results": [], + "tool_output_guardrail_results": [], + "tool_use_tracker": {}, + "trace": null +} diff --git a/tests/fixtures/run_state/generate_corpus.py b/tests/fixtures/run_state/generate_corpus.py index ef7a33ee2e..687e1357ba 100644 --- a/tests/fixtures/run_state/generate_corpus.py +++ b/tests/fixtures/run_state/generate_corpus.py @@ -460,6 +460,38 @@ def approval(call_id): "reader while preserving the Docker session payload." ), ), + Scenario( + "1.18", + "821afdc3f709f409a307c93f42a603704be63033", + "held_pending_session_write", + """ +from agents.run_internal.run_steps import NextStepRunAgain + +state._current_step = NextStepRunAgain() +state._pending_session_write = { + "session_id": "session-118", + "items": [ + { + "type": "function_call", + "call_id": "call_held_1", + "name": "write_thing", + "arguments": "{}", + }, + {"type": "function_call_output", "call_id": "call_held_1", "output": "wrote"}, + ], + "before": None, + "persisted_count": 2, + "held": True, +} +""", + provenance="canonical_compatibility", + emitted_version="1.17", + note=( + "The held pending Session write was first emitted with the unreleased 1.17 writer. " + "The fixture changes only the schema label to exercise the 1.18 compatibility " + "reader while preserving the withheld batch payload." + ), + ), ) @@ -490,6 +522,19 @@ def approval(call_id): "reader while preserving older payload compatibility." ), ), + Scenario( + "1.18", + "821afdc3f709f409a307c93f42a603704be63033", + "minimal", + "", + provenance="canonical_compatibility", + emitted_version="1.17", + note=( + "The held pending Session write was first emitted with the unreleased 1.17 writer. " + "The fixture changes only the schema label to exercise the 1.18 compatibility " + "reader while preserving older payload compatibility." + ), + ), ) diff --git a/tests/fixtures/run_state/minimal/v1_18.json b/tests/fixtures/run_state/minimal/v1_18.json new file mode 100644 index 0000000000..62255e7502 --- /dev/null +++ b/tests/fixtures/run_state/minimal/v1_18.json @@ -0,0 +1,60 @@ +{ + "$schemaVersion": "1.18", + "auto_previous_response_id": false, + "context": { + "approvals": {}, + "context": {}, + "context_meta": { + "omitted": false, + "original_type": "mapping", + "requires_deserializer": false, + "serialized_via": "mapping" + }, + "tool_invocations": {}, + "usage": { + "input_tokens": 0, + "input_tokens_details": [ + { + "cache_write_tokens": 0, + "cached_tokens": 0 + } + ], + "output_tokens": 0, + "output_tokens_details": [ + { + "reasoning_tokens": 0 + } + ], + "request_usage_entries": [], + "requests": 0, + "total_tokens": 0 + } + }, + "conversation_id": null, + "current_agent": { + "name": "compat-agent" + }, + "current_step": null, + "current_turn": 0, + "current_turn_persisted_item_count": 0, + "generated_items": [], + "generated_prompt_cache_key": null, + "generated_session_item_indexes": [], + "input_guardrail_results": [], + "last_model_response": null, + "last_processed_response": null, + "max_turns": 10, + "model_responses": [], + "nested_history_owned_session_item_refs": [], + "no_active_agent_run": true, + "original_input": "historical input", + "output_guardrail_results": [], + "pending_input": [], + "previous_response_id": null, + "reasoning_item_id_policy": null, + "session_items": [], + "tool_input_guardrail_results": [], + "tool_output_guardrail_results": [], + "tool_use_tracker": {}, + "trace": null +} diff --git a/tests/fixtures/run_state/sources.json b/tests/fixtures/run_state/sources.json index fc7be3fb55..67d9e16a3f 100644 --- a/tests/fixtures/run_state/sources.json +++ b/tests/fixtures/run_state/sources.json @@ -127,6 +127,15 @@ "note": "The labels implementation was first emitted with the unreleased 1.16 writer. The fixture changes only the schema label to exercise the 1.17 compatibility reader while preserving the Docker session payload.", "provenance": "canonical_compatibility", "version": "1.17" + }, + { + "commit": "821afdc3f709f409a307c93f42a603704be63033", + "emitted_version": "1.17", + "feature": "held_pending_session_write", + "fixture": "features/v1_18_held_pending_session_write.json", + "note": "The held pending Session write was first emitted with the unreleased 1.17 writer. The fixture changes only the schema label to exercise the 1.18 compatibility reader while preserving the withheld batch payload.", + "provenance": "canonical_compatibility", + "version": "1.18" } ], "resume": { @@ -195,6 +204,13 @@ "note": "The labels implementation was first emitted with the unreleased 1.16 writer. The fixture changes only the schema label to exercise the 1.17 compatibility reader while preserving older payload compatibility.", "provenance": "canonical_compatibility" }, + "1.18": { + "commit": "821afdc3f709f409a307c93f42a603704be63033", + "emitted_version": "1.17", + "fixture": "minimal/v1_18.json", + "note": "The held pending Session write was first emitted with the unreleased 1.17 writer. The fixture changes only the schema label to exercise the 1.18 compatibility reader while preserving older payload compatibility.", + "provenance": "canonical_compatibility" + }, "1.2": { "commit": "74e8c1e22d7441bd42c58bcd4270937ccc2dca8c", "fixture": "minimal/v1_2.json" diff --git a/tests/sandbox/test_docker.py b/tests/sandbox/test_docker.py index e4c7cc812f..1f999e26b2 100644 --- a/tests/sandbox/test_docker.py +++ b/tests/sandbox/test_docker.py @@ -1926,7 +1926,7 @@ async def test_docker_labels_roundtrip_through_run_state() -> None: serialized = run_state.to_json() restored = await RunState.from_json(agent, serialized) - assert serialized["$schemaVersion"] == CURRENT_SCHEMA_VERSION == "1.17" + assert serialized["$schemaVersion"] == CURRENT_SCHEMA_VERSION == "1.18" assert restored._sandbox is not None restored_session_state = restored._sandbox["session_state"] assert isinstance(restored_session_state, dict) diff --git a/tests/test_agent_runner_streamed.py b/tests/test_agent_runner_streamed.py index 2923286924..8348b3ce1e 100644 --- a/tests/test_agent_runner_streamed.py +++ b/tests/test_agent_runner_streamed.py @@ -4318,6 +4318,7 @@ async def save_wrapper( store: bool | None = None, wrapper: RunContextWrapper[Any] | None = None, run_state: RunState | None = None, + held_input: Any = None, ) -> int: observed_counts.append(persisted_count) result = await real_save_resumed( @@ -4329,6 +4330,7 @@ async def save_wrapper( store=store, wrapper=wrapper, run_state=run_state, + held_input=held_input, ) return int(result) diff --git a/tests/test_deferred_interrupted_session_write.py b/tests/test_deferred_interrupted_session_write.py new file mode 100644 index 0000000000..9dc2056040 --- /dev/null +++ b/tests/test_deferred_interrupted_session_write.py @@ -0,0 +1,2054 @@ +from __future__ import annotations + +import json +from collections.abc import Callable +from dataclasses import replace +from typing import Any, Literal, cast + +import pytest + +from agents import ( + Agent, + GuardrailFunctionOutput, + RunContextWrapper, + Runner, + RunResult, + RunResultStreaming, + RunState, + StopAtTools, + function_tool, + output_guardrail, +) +from agents.agent import Agent as AgentType +from agents.exceptions import OutputGuardrailTripwireTriggered +from agents.items import TResponseInputItem +from agents.lifecycle import RunHooks +from agents.memory.openai_conversations_session import OpenAIConversationsSession +from agents.run import RunConfig +from agents.testing import ModelStep, ScriptedModel, assistant_message, function_call +from tests.utils.simple_session import SimpleListSession + + +@function_tool(name_override="write_thing", needs_approval=True) +def write_thing(query: str) -> str: + return f"wrote:{query}" + + +@function_tool(name_override="write_other", needs_approval=True) +def write_other(query: str) -> str: + return f"other:{query}" + + +@function_tool(name_override="look_up", needs_approval=False) +def look_up(query: str) -> str: + return f"schema for {query}" + + +@output_guardrail +async def always_fine( + ctx: RunContextWrapper[object], agent: AgentType[object], output: object +) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=False) + + +@output_guardrail +async def always_trips( + ctx: RunContextWrapper[object], agent: AgentType[object], output: object +) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) + + +@output_guardrail +async def always_crashes( + ctx: RunContextWrapper[object], agent: AgentType[object], output: object +) -> GuardrailFunctionOutput: + raise RuntimeError("guardrail crashed") + + +# The two conditions that open ``_should_defer_interrupted_session_items``: output +# guardrails and a non-default ``tool_use_behavior``. The approved tool is not in the +# stop list, so the resume resolves into a run-again step rather than a terminal tool +# output. +_DEFERRING_BEHAVIOR = StopAtTools(stop_at_tool_names=["finish"]) + + +def _make_deferring_agent( + tool_use_behavior: StopAtTools | Literal["run_llm_again"] = _DEFERRING_BEHAVIOR, +) -> Agent: + """A gated write on the second model turn, so the resumed boundary has a prefix.""" + return Agent( + name="deferred repro", + instructions="Always call write_thing.", + model=ScriptedModel( + [ + ModelStep(output=[function_call("look_up", {"query": "x"}, call_id="call_LOOKUP")]), + ModelStep( + output=[function_call("write_thing", {"query": "x"}, call_id="call_PARKED")] + ), + ModelStep(output=[assistant_message("done")]), + ] + ), + tools=[look_up, write_thing], + output_guardrails=[always_fine], + tool_use_behavior=tool_use_behavior, + ) + + +def _make_multi_approval_agent( + tool_use_behavior: StopAtTools | Literal["run_llm_again"] = _DEFERRING_BEHAVIOR, +) -> Agent: + """One deferred model response carrying two approval-required calls.""" + return Agent( + name="deferred repro (multi)", + instructions="Call both tools.", + model=ScriptedModel( + [ + ModelStep( + output=[ + function_call("write_thing", {"query": "x"}, call_id="call_PARKED"), + function_call("write_other", {"query": "x"}, call_id="call_PARKED_2"), + ] + ), + ModelStep(output=[assistant_message("done")]), + ] + ), + tools=[write_thing, write_other], + output_guardrails=[always_fine], + tool_use_behavior=tool_use_behavior, + ) + + +_PREAMBLE_TEXT = "About to write the thing." + + +def _make_terminal_tool_agent( + *, + with_guardrails: bool = True, + tripping: bool = False, + crashing: bool = False, + with_preamble: bool = False, +) -> Agent: + """The approved tool is terminal, so the resume ends in a final output.""" + guardrails = [always_fine] + if tripping: + guardrails = [always_trips] + if crashing: + guardrails = [always_crashes] + parked_response = [function_call("write_thing", {"query": "x"}, call_id="call_PARKED")] + if with_preamble: + parked_response = [assistant_message(_PREAMBLE_TEXT), *parked_response] + return Agent( + name="deferred repro (terminal)", + instructions="Always call write_thing.", + model=ScriptedModel( + [ + ModelStep(output=[function_call("look_up", {"query": "x"}, call_id="call_LOOKUP")]), + ModelStep(output=parked_response), + ] + ), + tools=[look_up, write_thing], + output_guardrails=guardrails if with_guardrails else [], + tool_use_behavior=StopAtTools(stop_at_tool_names=["write_thing"]), + ) + + +def _make_emptying_handoff_agent() -> Agent: + """The gated call rides one response with a handoff whose filter empties the turn.""" + from agents import HandoffInputData, handoff + + def empties(data: HandoffInputData) -> HandoffInputData: + return HandoffInputData( + input_history=data.input_history, pre_handoff_items=(), new_items=() + ) + + target = Agent( + name="target", + instructions="x", + model=ScriptedModel( + [ + ModelStep(output=[assistant_message("done")]), + ModelStep(output=[assistant_message("done")]), + ] + ), + ) + return Agent( + name="deferred repro (emptied turn)", + instructions="x", + model=ScriptedModel( + [ + ModelStep( + output=[ + function_call("write_thing", {"query": "x"}, call_id="call_PARKED"), + function_call("transfer_to_target", {}, call_id="call_HANDOFF"), + ] + ), + ModelStep(output=[assistant_message("done")]), + ] + ), + tools=[write_thing], + handoffs=[handoff(target, input_filter=empties)], + output_guardrails=[always_fine], + tool_use_behavior=_DEFERRING_BEHAVIOR, + ) + + +def _make_partial_filter_handoff_agent() -> Agent: + """Two gated calls plus a handoff whose filter drops exactly one resolved output. + + An ``input_filter`` is an arbitrary caller callable, so dropping a subset of the + resolved outputs is a legitimate shape; the held batch must not settle a call whose + output the filter took away. + """ + from agents import HandoffInputData, handoff + + def drops_one_output(data: HandoffInputData) -> HandoffInputData: + def keep(items: tuple) -> tuple: + kept = [] + for item in items: + raw = getattr(item, "raw_item", None) + call_id = ( + raw.get("call_id") if isinstance(raw, dict) else getattr(raw, "call_id", None) + ) + if call_id == "call_PARKED_2" and item.type == "tool_call_output_item": + continue + kept.append(item) + return tuple(kept) + + return HandoffInputData( + input_history=data.input_history, + pre_handoff_items=keep(data.pre_handoff_items), + new_items=keep(data.new_items), + ) + + target = Agent( + name="target", + instructions="x", + model=ScriptedModel([ModelStep(output=[assistant_message("done")])]), + ) + return Agent( + name="deferred repro (partial filter)", + instructions="x", + model=ScriptedModel( + [ + ModelStep( + output=[ + function_call("write_thing", {"query": "x"}, call_id="call_PARKED"), + function_call("write_other", {"query": "x"}, call_id="call_PARKED_2"), + function_call("transfer_to_target", {}, call_id="call_HANDOFF"), + ] + ), + ModelStep(output=[assistant_message("done")]), + ] + ), + tools=[write_thing, write_other], + handoffs=[handoff(target, input_filter=drops_one_output)], + output_guardrails=[always_fine], + tool_use_behavior=_DEFERRING_BEHAVIOR, + ) + + +class _ContextRequiringSession(SimpleListSession): + """Track whether internal reads and writes carry the run's context wrapper.""" + + def __init__(self) -> None: + super().__init__() + self.wrapperless_operations = 0 + + async def get_items( + self, limit: int | None = None, *, wrapper: RunContextWrapper[Any] | None = None + ) -> list[TResponseInputItem]: + if limit is not None and wrapper is None: + self.wrapperless_operations += 1 + return await super().get_items(limit) + + async def add_items( + self, items: list[TResponseInputItem], *, wrapper: RunContextWrapper[Any] | None = None + ) -> None: + if wrapper is None: + self.wrapperless_operations += 1 + await super().add_items(items) + + async def pop_item( + self, *, wrapper: RunContextWrapper[Any] | None = None + ) -> TResponseInputItem | None: + return await super().pop_item() + + async def clear_session(self, *, wrapper: RunContextWrapper[Any] | None = None) -> None: + await super().clear_session() + + +class _LegacyGetItemsSession(SimpleListSession): + """A pre-limit Session whose ``get_items`` takes no arguments at all.""" + + async def get_items(self) -> list[TResponseInputItem]: # type: ignore[override] + return await super().get_items() + + +class _AppendRecordingSession(SimpleListSession): + """Record each ``add_items`` batch to observe write ordering and granularity.""" + + def __init__(self) -> None: + super().__init__() + self.batches: list[list[TResponseInputItem]] = [] + + async def add_items(self, items: list[TResponseInputItem]) -> None: + self.batches.append(list(items)) + await super().add_items(items) + + +class _FailingResumeSession(SimpleListSession): + """Control append acknowledgement at the public Session boundary.""" + + def __init__(self) -> None: + super().__init__() + self.failure: str | None = None + self.error = RuntimeError("session append failed") + + async def add_items(self, items: list[TResponseInputItem]) -> None: + failure, self.failure = self.failure, None + if failure == "before": + raise self.error + await super().add_items(items) + if failure == "after": + raise self.error + + +async def _run( + agent: Agent, run_input: Any, session: Any, *, streamed: bool +) -> RunResult | RunResultStreaming: + if streamed: + result = Runner.run_streamed(agent, run_input, session=session) + async for _ in result.stream_events(): + pass + return result + return await Runner.run(agent, run_input, session=session) + + +async def _serialized_round_trip(result: RunResult | RunResultStreaming, agent: Agent) -> RunState: + return await RunState.from_json(agent, json.loads(json.dumps(result.to_state().to_json()))) + + +def _call_ids(items: list[TResponseInputItem]) -> list[Any]: + return [item.get("call_id") for item in items if item.get("type") == "function_call"] + + +def _orphaned_outputs(items: list[TResponseInputItem]) -> list[Any]: + calls = set(_call_ids(items)) + return [ + item.get("call_id") + for item in items + if item.get("type") == "function_call_output" and item.get("call_id") not in calls + ] + + +def _parked_pair(items: list[TResponseInputItem]) -> list[str]: + return [ + str(item.get("type")) + for item in items + if isinstance(item, dict) and item.get("call_id") == "call_PARKED" + ] + + +async def _parked_and_approved( + agent: Agent, session: Any, *, streamed: bool, resume_agent: Agent | None = None +) -> RunState: + first = await _run(agent, "do the thing", session, streamed=streamed) + assert len(first.interruptions) == 1 + state = await _serialized_round_trip(first, resume_agent or agent) + state.approve(state.get_interruptions()[0]) + return state + + +_EXPECTED_PAIR = ["function_call", "function_call_output"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_deferred_parked_call_is_persisted_when_the_resume_runs_again( + streamed: bool, +) -> None: + session = SimpleListSession() + agent = _make_deferring_agent() + state = await _parked_and_approved(agent, session, streamed=streamed) + + resumed = await _run(agent, state, session, streamed=streamed) + assert resumed.final_output == "done" + + items = await session.get_items() + assert _orphaned_outputs(items) == [] + assert _parked_pair(items) == _EXPECTED_PAIR + assert "pending_session_write" not in resumed.to_state().to_json() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_park_time_deferral_survives_a_tool_use_behavior_change_on_resume( + streamed: bool, +) -> None: + # The deferral decision is the checkpoint's, not the resuming configuration's: the + # caller resumes with the default behavior, and deriving the decision from the live + # gate would drop the parked call again. + session = SimpleListSession() + resume_agent = _make_deferring_agent(tool_use_behavior="run_llm_again") + state = await _parked_and_approved( + _make_deferring_agent(), session, streamed=streamed, resume_agent=resume_agent + ) + + await _run(resume_agent, state, session, streamed=streamed) + + items = await session.get_items() + assert _orphaned_outputs(items) == [] + assert _parked_pair(items) == _EXPECTED_PAIR + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_non_deferred_park_is_not_double_written_on_resume(streamed: bool) -> None: + # The other direction: with the default behavior throughout, the interruption-time + # write runs, so the resume must not write the parked call a second time. + session = SimpleListSession() + agent = _make_deferring_agent(tool_use_behavior="run_llm_again") + state = await _parked_and_approved(agent, session, streamed=streamed) + + await _run(agent, state, session, streamed=streamed) + + assert _parked_pair(await session.get_items()) == _EXPECTED_PAIR + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_partial_approval_reinterruption_keeps_one_canonical_batch( + streamed: bool, +) -> None: + # Two approval-required calls in one deferred response; approving only one resolves + # into a second interruption. The held batch must absorb the resolved output and + # keep riding as one record, then land exactly once when the run finally continues. + session = SimpleListSession() + agent = _make_multi_approval_agent() + + first = await _run(agent, "go", session, streamed=streamed) + assert len(first.interruptions) == 2 + state = await _serialized_round_trip(first, agent) + state.approve( + next( + interruption + for interruption in state.get_interruptions() + if getattr(interruption.raw_item, "call_id", None) == "call_PARKED" + ) + ) + + second = await _run(agent, state, session, streamed=streamed) + assert len(second.interruptions) == 1 + second_checkpoint = second.to_state().to_json() + pending = second_checkpoint.get("pending_session_write") + assert pending is not None and pending.get("held") is True + assert {item.get("call_id") for item in pending["items"]} == { + "call_PARKED", + "call_PARKED_2", + } + + state = await RunState.from_json(agent, json.loads(json.dumps(second_checkpoint))) + for interruption in state.get_interruptions(): + state.approve(interruption) + final = await _run(agent, state, session, streamed=streamed) + assert final.final_output == "done" + + items = await session.get_items() + assert _orphaned_outputs(items) == [] + assert _call_ids(items).count("call_PARKED") == 1 + assert _call_ids(items).count("call_PARKED_2") == 1 + # Every call must also keep its output: losing the first approval's output while + # the batch rides the second park is the symmetric corruption. + outputs = {item.get("call_id") for item in items if item.get("type") == "function_call_output"} + assert set(_call_ids(items)) == outputs + + +@pytest.mark.asyncio +@pytest.mark.parametrize("resume_with_guardrails", [True, False]) +@pytest.mark.parametrize("streamed", [False, True]) +async def test_deferred_prefix_reaches_a_resume_that_ends_in_final_output( + resume_with_guardrails: bool, streamed: bool +) -> None: + # A resume may legitimately run without the guardrails the park had; the + # final-output exit must land the held batch either way. + session = SimpleListSession() + resume_agent = _make_terminal_tool_agent(with_guardrails=resume_with_guardrails) + state = await _parked_and_approved( + _make_terminal_tool_agent(), session, streamed=streamed, resume_agent=resume_agent + ) + + await _run(resume_agent, state, session, streamed=streamed) + + items = await session.get_items() + assert _orphaned_outputs(items) == [] + assert _parked_pair(items) == _EXPECTED_PAIR + + +@pytest.mark.asyncio +async def test_a_detached_resume_does_not_make_the_next_one_rewrite_the_session() -> None: + # A non-deferred park persists the interrupted turn's items; the resumed-safety + # validation then zeroes the counter for a detached resume. A later resume that + # reconnects the original Session must not rewrite items it already holds. + session = SimpleListSession() + + parked = await _run( + _make_multi_approval_agent(tool_use_behavior="run_llm_again"), + "go", + session, + streamed=True, + ) + assert len(parked.interruptions) == 2 + assert "call_PARKED" in _call_ids(await session.get_items()) + + deferring_agent = _make_multi_approval_agent() + state = await _serialized_round_trip(parked, deferring_agent) + state.approve( + next( + interruption + for interruption in state.get_interruptions() + if getattr(interruption.raw_item, "call_id", None) == "call_PARKED" + ) + ) + detached = await _run(deferring_agent, state, None, streamed=True) + + state = await _serialized_round_trip(detached, deferring_agent) + for interruption in state.get_interruptions(): + state.approve(interruption) + await _run(deferring_agent, state, session, streamed=True) + + call_ids = _call_ids(await session.get_items()) + assert call_ids.count("call_PARKED") == 1 + assert call_ids.count("call_PARKED_2") == 1 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_an_emptied_resolved_turn_settles_the_paired_part_of_the_held_batch( + streamed: bool, +) -> None: + # A handoff input_filter empties the resolved turn, but the approved tool already + # ran and its output was folded into the held batch: pairing is the predicate, so + # the executed pair settles and only the unpaired call drops. Discarding the whole + # batch would lose the Session's only record that the tool ran, and the next run + # would re-issue its side effect. + session = SimpleListSession() + agent = _make_emptying_handoff_agent() + state = await _parked_and_approved(agent, session, streamed=streamed) + resumed = await _run(agent, state, session, streamed=streamed) + + items = await session.get_items() + calls = set(_call_ids(items)) + outputs = {item.get("call_id") for item in items if item.get("type") == "function_call_output"} + assert calls - outputs == set(), f"dangling calls: {sorted(map(str, calls - outputs))}" + assert outputs - calls == set(), f"orphaned outputs: {sorted(map(str, outputs - calls))}" + assert "call_PARKED" in calls, "the executed pair must survive the emptied turn" + assert "call_HANDOFF" not in calls, "the unpaired call must not be written" + assert "pending_session_write" not in resumed.to_state().to_json() + # The discard must reach the live state too: a stale held record would invalidate + # any checkpoint later taken from this completed run. + assert state._pending_session_write is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_filter_that_drops_one_output_takes_its_held_call_with_it( + streamed: bool, +) -> None: + # The resolved turn is non-empty (one output survived the filter), so batch + # emptiness is the wrong safety predicate: settling the whole held batch would land + # the filtered call dangling, and discarding the whole batch would orphan the + # output the filter kept. Pairing is the contract, per call. + session = SimpleListSession() + agent = _make_partial_filter_handoff_agent() + first = await _run(agent, "go", session, streamed=streamed) + state = await _serialized_round_trip(first, agent) + for interruption in state.get_interruptions(): + state.approve(interruption) + resumed = await _run(agent, state, session, streamed=streamed) + + items = await session.get_items() + calls = set(_call_ids(items)) + outputs = {item.get("call_id") for item in items if item.get("type") == "function_call_output"} + assert calls - outputs == set(), f"dangling calls: {sorted(map(str, calls - outputs))}" + assert outputs - calls == set(), f"orphaned outputs: {sorted(map(str, outputs - calls))}" + assert "call_PARKED" in calls + assert "pending_session_write" not in resumed.to_state().to_json() + assert state._pending_session_write is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_held_resume_with_a_different_session_is_refused(streamed: bool) -> None: + # The held entry skip must not bypass the same-session contract: resuming the + # approval checkpoint against another Session would execute the tool and settle the + # withheld batch into the wrong conversation. + from agents.exceptions import UserError + + session = SimpleListSession() + agent = _make_deferring_agent() + state = await _parked_and_approved(agent, session, streamed=streamed) + + other_session = SimpleListSession("other") + with pytest.raises(UserError, match="pending Session write"): + await _run(agent, state, other_session, streamed=streamed) + + assert await other_session.get_items() == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_settle_reaches_a_context_aware_session_through_the_wrapper( + streamed: bool, +) -> None: + session = _ContextRequiringSession() + agent = _make_deferring_agent() + state = await _parked_and_approved(agent, session, streamed=streamed) + + await _run(agent, state, session, streamed=streamed) + + assert session.wrapperless_operations == 0 + assert _parked_pair(await session.get_items()) == _EXPECTED_PAIR + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_session_without_optional_kwargs_survives_a_deferred_resume( + streamed: bool, +) -> None: + session = _LegacyGetItemsSession() + agent = _make_deferring_agent() + state = await _parked_and_approved(agent, session, streamed=streamed) + + resumed = await _run(agent, state, session, streamed=streamed) + assert resumed.final_output == "done" + + items = await session.get_items() + assert _orphaned_outputs(items) == [] + assert _parked_pair(items) == _EXPECTED_PAIR + + +@pytest.mark.asyncio +async def test_after_turn_cancel_keeps_the_held_batch_for_the_next_attach() -> None: + # The detached carry: a detached resume executes the approved tool, an after-turn + # cancel flips the checkpoint to a run-again step, and the batch must bring the + # executed output to the reattaching resume. Cancellation only exists on the + # streaming runner, so this scenario has no non-streamed axis. + session = SimpleListSession() + agent = _make_deferring_agent() + state = await _parked_and_approved(agent, session, streamed=True) + + detached = Runner.run_streamed(agent, state, session=None) + detached.cancel(mode="after_turn") + async for _ in detached.stream_events(): + pass + + checkpoint = detached.to_state().to_json() + pending = checkpoint.get("pending_session_write") + assert pending is not None and pending.get("held") is True + assert {item.get("call_id") for item in pending["items"]} >= {"call_PARKED"} + + state = await RunState.from_json(agent, json.loads(json.dumps(checkpoint))) + reattached = Runner.run_streamed(agent, state, session=session) + async for _ in reattached.stream_events(): + pass + + items = await session.get_items() + assert _orphaned_outputs(items) == [] + assert _parked_pair(items) == _EXPECTED_PAIR + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_reject_persists_the_parked_call_with_its_rejection_output( + streamed: bool, +) -> None: + session = SimpleListSession() + agent = _make_deferring_agent() + first = await _run(agent, "do the thing", session, streamed=streamed) + assert len(first.interruptions) == 1 + state = await _serialized_round_trip(first, agent) + state.reject(state.get_interruptions()[0]) + + resumed = await _run(agent, state, session, streamed=streamed) + assert resumed.final_output == "done" + + items = await session.get_items() + assert _orphaned_outputs(items) == [] + assert _parked_pair(items) == _EXPECTED_PAIR + + +@pytest.mark.asyncio +async def test_the_held_batch_rides_a_non_streamed_result_into_its_checkpoint() -> None: + # The non-streamed runner has no live RunState on a fresh park, so the declaration + # must ride the result into ``to_state``; dropping it there is the one silent way + # to lose the batch. + session = SimpleListSession() + agent = _make_deferring_agent() + + first = await Runner.run(agent, "do the thing", session=session) + assert len(first.interruptions) == 1 + + checkpoint = first.to_state().to_json() + pending = checkpoint.get("pending_session_write") + assert pending is not None and pending.get("held") is True + assert "call_PARKED" in {item.get("call_id") for item in pending["items"]} + assert pending.get("before") is None + # The batch carries only the withheld response: the accepted input persists + # eagerly even at a deferred park, so a tripwire discard can never take the + # Session's only copy of the input with it. + assert not any(item.get("role") == "user" for item in pending["items"]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_the_settled_batch_and_the_resolved_turn_land_as_one_ordered_write( + streamed: bool, +) -> None: + # Settling separately from the resolved turn's save would either trip the + # single-slot rule or advance the persisted count and slice the resolved items out + # of their own save, so the pair must land in one append, call before output. + session = _AppendRecordingSession() + agent = _make_deferring_agent() + state = await _parked_and_approved(agent, session, streamed=streamed) + batches_before_resume = len(session.batches) + + await _run(agent, state, session, streamed=streamed) + + resume_batches = session.batches[batches_before_resume:] + settling_batches = [ + batch for batch in resume_batches if "call_PARKED" in {i.get("call_id") for i in batch} + ] + assert len(settling_batches) == 1 + assert _parked_pair(settling_batches[0]) == _EXPECTED_PAIR + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_tripwire_after_approval_keeps_the_sanitized_pair(streamed: bool) -> None: + session = SimpleListSession() + resume_agent = _make_terminal_tool_agent(tripping=True, with_preamble=True) + state = await _parked_and_approved( + _make_terminal_tool_agent(with_preamble=True), + session, + streamed=streamed, + resume_agent=resume_agent, + ) + + if streamed: + resumed = Runner.run_streamed(resume_agent, state, session=session) + with pytest.raises(OutputGuardrailTripwireTriggered): + async for _ in resumed.stream_events(): + pass + # The declaration is discarded when the blocked outcome is decided; a record + # that outlives the tripwire would invalidate the run's checkpoint. + assert "pending_session_write" not in resumed.to_state().to_json() + else: + with pytest.raises(OutputGuardrailTripwireTriggered): + await Runner.run(resume_agent, state, session=session) + + items = await session.get_items() + assert _orphaned_outputs(items) == [] + assert _parked_pair(items) == _EXPECTED_PAIR + # The redaction drops the blocked response's preamble; feeding the raw held batch + # into the blocked save would resurrect it. + assert not any(_PREAMBLE_TEXT in json.dumps(item) for item in items) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_guardrail_crash_still_persists_the_parked_call(streamed: bool) -> None: + session = SimpleListSession() + resume_agent = _make_terminal_tool_agent(crashing=True) + state = await _parked_and_approved( + _make_terminal_tool_agent(), session, streamed=streamed, resume_agent=resume_agent + ) + + if streamed: + resumed = Runner.run_streamed(resume_agent, state, session=session) + with pytest.raises(RuntimeError, match="guardrail crashed"): + async for _ in resumed.stream_events(): + pass + assert "pending_session_write" not in resumed.to_state().to_json() + else: + with pytest.raises(RuntimeError, match="guardrail crashed"): + await Runner.run(resume_agent, state, session=session) + # The crash-path save claims the batch, so no stale record survives on the state. + assert state._pending_session_write is None + + items = await session.get_items() + assert _orphaned_outputs(items) == [] + assert _parked_pair(items) == _EXPECTED_PAIR + + +@pytest.mark.asyncio +@pytest.mark.parametrize("retry_streamed", [False, True]) +@pytest.mark.parametrize("streamed", [False, 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_a_failed_settle_of_the_held_batch_is_recovered_on_the_next_resume( + retry_streamed: bool, streamed: bool, round_trip: bool, failure: str +) -> None: + session = _FailingResumeSession() + agent = _make_deferring_agent() + state = await _parked_and_approved(agent, session, streamed=streamed) + + session.failure = failure + with pytest.raises(RuntimeError) as error: + await _run(agent, state, session, streamed=streamed) + assert error.value is session.error + if round_trip: + state = await RunState.from_json(agent, state.to_json()) + + result = await _run(agent, state, session, streamed=retry_streamed) + assert result.final_output == "done" + + items = await session.get_items() + assert _orphaned_outputs(items) == [] + assert _parked_pair(items) == _EXPECTED_PAIR + assert "pending_session_write" not in result.to_state().to_json() + + +def _make_two_park_agent() -> Agent: + """Two approval-required calls on consecutive turns, so a resume can park again.""" + return Agent( + name="deferred repro (two parks)", + instructions="x", + model=ScriptedModel( + [ + ModelStep(output=[function_call("write_thing", {"query": "a"}, call_id="call_A")]), + ModelStep(output=[function_call("write_other", {"query": "b"}, call_id="call_B")]), + ModelStep(output=[assistant_message("done")]), + ] + ), + tools=[write_thing, write_other], + output_guardrails=[always_fine], + tool_use_behavior=_DEFERRING_BEHAVIOR, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_new_park_during_a_detached_resume_joins_the_held_batch( + streamed: bool, +) -> None: + # A detached resume resolves the first approval and parks a second call on the next + # turn. That fresh park cannot write anything, but the standing declaration carries + # the session identity, so the new call must fold into the held batch or the + # reattach settles its output orphaned. + session = SimpleListSession() + agent = _make_two_park_agent() + state = await _parked_and_approved(agent, session, streamed=streamed) + + detached = await _run(agent, state, None, streamed=streamed) + assert len(detached.interruptions) == 1 + state = await _serialized_round_trip(detached, agent) + state.approve(state.get_interruptions()[0]) + + reattached = await _run(agent, state, session, streamed=streamed) + assert reattached.final_output == "done" + + items = await session.get_items() + assert _orphaned_outputs(items) == [] + calls = set(_call_ids(items)) + outputs = {item.get("call_id") for item in items if item.get("type") == "function_call_output"} + assert calls == outputs + assert {"call_A", "call_B"} <= calls + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_gate_off_reinterruption_keeps_the_still_pending_call(streamed: bool) -> None: + # Approving one of two held calls and resuming with the default behavior turns the + # gate off, so the re-interruption exit settles the batch mid-run. The unapproved + # call's output does not exist yet because it is still pending, not because a + # filter removed it; dropping it there orphans its output on the final resume. + session = SimpleListSession() + resume_agent = _make_multi_approval_agent(tool_use_behavior="run_llm_again") + + first = await _run(_make_multi_approval_agent(), "go", session, streamed=streamed) + assert len(first.interruptions) == 2 + state = await _serialized_round_trip(first, resume_agent) + state.approve( + next( + interruption + for interruption in state.get_interruptions() + if getattr(interruption.raw_item, "call_id", None) == "call_PARKED" + ) + ) + + second = await _run(resume_agent, state, session, streamed=streamed) + assert len(second.interruptions) == 1 + state = await _serialized_round_trip(second, resume_agent) + for interruption in state.get_interruptions(): + state.approve(interruption) + final = await _run(resume_agent, state, session, streamed=streamed) + assert final.final_output == "done" + + items = await session.get_items() + assert _orphaned_outputs(items) == [] + calls = set(_call_ids(items)) + outputs = {item.get("call_id") for item in items if item.get("type") == "function_call_output"} + assert calls == outputs + assert {"call_PARKED", "call_PARKED_2"} <= calls + + +@pytest.mark.asyncio +async def test_entry_settle_drops_a_held_call_the_filter_unpaired() -> None: + # A detached resume of the partial-filter handoff folds the post-filter items into + # the batch, the handoff normalizes the checkpoint to run-again, and an after-turn + # cancellation stops the run there. The reattach settles at entry, where the same + # pairing contract applies: the filtered call must not land dangling. Cancellation + # only exists on the streaming runner, and this checkpoint shape resumes from the + # live state. + session = SimpleListSession() + agent = _make_partial_filter_handoff_agent() + first = await _run(agent, "go", session, streamed=True) + state = await _serialized_round_trip(first, agent) + for interruption in state.get_interruptions(): + state.approve(interruption) + + detached = Runner.run_streamed(agent, state, session=None) + detached.cancel(mode="after_turn") + async for _ in detached.stream_events(): + pass + + reattached = Runner.run_streamed(agent, detached.to_state(), session=session) + async for _ in reattached.stream_events(): + pass + + items = await session.get_items() + calls = set(_call_ids(items)) + outputs = {item.get("call_id") for item in items if item.get("type") == "function_call_output"} + assert calls - outputs == set(), f"dangling calls: {sorted(map(str, calls - outputs))}" + assert outputs - calls == set(), f"orphaned outputs: {sorted(map(str, outputs - calls))}" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_detached_completion_clears_the_held_record(streamed: bool) -> None: + # A detached resume that runs to completion has no Session to settle against and + # the fresh final exit ends the run; a held record left standing would invalidate + # the completed run's checkpoint and diverge between the runners. + session = SimpleListSession() + agent = _make_deferring_agent() + state = await _parked_and_approved(agent, session, streamed=streamed) + + detached = await _run(agent, state, None, streamed=streamed) + assert detached.final_output == "done" + assert "pending_session_write" not in detached.to_state().to_json() + assert state._pending_session_write is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_failed_final_settle_fails_closed_with_the_batch_recorded( + streamed: bool, +) -> None: + # The final-output settle registers the claimed batch before appending, so a crash + # inside that append leaves the batch recorded on the state instead of silently + # losing the only copy of the approved call and its output. The resulting + # checkpoint is rejected on load on purpose: the run ended mid-settle, and failing + # closed beats replaying an approved side effect as if nothing happened. + session = _FailingResumeSession() + # A guardrail-less resume: the final sweep returns the resolved items verbatim, so + # the held batch itself rides the append that fails. + resume_agent = _make_terminal_tool_agent(with_guardrails=False) + state = await _parked_and_approved( + _make_terminal_tool_agent(), session, streamed=streamed, resume_agent=resume_agent + ) + + session.failure = "before" + with pytest.raises(RuntimeError, match="session append failed"): + await _run(resume_agent, state, session, streamed=streamed) + + pending = state._pending_session_write + assert pending is not None + recorded = {item.get("call_id") for item in pending["items"]} + assert "call_PARKED" in recorded + with pytest.raises(Exception, match="pending Session write"): + await RunState.from_json(resume_agent, state.to_json()) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_failed_guarded_final_settle_fails_closed(streamed: bool) -> None: + # With output guardrails the final sweep rebuilds the response and the held batch + # is deduplicated out of the append, but the append still lands the approved call + # and output, so the recovery registration must stay armed: a crash inside it must + # leave the batch recorded, not silently lost. Guards the interaction between the + # dedup and the crash-safe registration. + session = _FailingResumeSession() + resume_agent = _make_terminal_tool_agent(with_preamble=True) + state = await _parked_and_approved( + _make_terminal_tool_agent(with_preamble=True), + session, + streamed=streamed, + resume_agent=resume_agent, + ) + + session.failure = "before" + with pytest.raises(RuntimeError, match="session append failed"): + await _run(resume_agent, state, session, streamed=streamed) + + pending = state._pending_session_write + assert pending is not None + assert "call_PARKED" in {item.get("call_id") for item in pending["items"]} + + +class _RecordingConversationsSession(OpenAIConversationsSession): + """Stand-in carrying the Conversations class identity the settle checks. + + The real backend talks to the Conversations API; the settle only asks whether the + session is one of these to decide that the batch needs the Conversations + sanitization, so this records what would be sent instead of sending it. + """ + + def __init__(self) -> None: + self.session_id = "conv-1" + self.added: list[TResponseInputItem] = [] + + async def get_items(self, limit: int | None = None) -> list[TResponseInputItem]: + return [] + + async def add_items(self, items: list[TResponseInputItem]) -> None: + self.added.extend(items) + + async def pop_item(self) -> TResponseInputItem | None: + return None + + async def clear_session(self) -> None: + return None + + +@pytest.mark.asyncio +async def test_entry_settle_restores_the_conversations_sanitization() -> None: + # A batch extended while detached missed the Conversations-specific sanitization; + # the attached entry settle must restore it or the create-items request rejects + # stale provider ids the normal persistence path strips. + from agents.run_internal.run_steps import NextStepRunAgain + from agents.run_internal.session_persistence import resume_pending_session_write + + session = _RecordingConversationsSession() + state = RunState( + context=None, + original_input="go", + starting_agent=_make_deferring_agent(), + max_turns=5, + ) + state._current_step = NextStepRunAgain() + state._pending_session_write = { + "session_id": "conv-1", + "items": [ + { + "type": "function_call", + "call_id": "call_PARKED", + "name": "write_thing", + "arguments": "{}", + "id": "__fake_id__", + }, + { + "type": "function_call_output", + "call_id": "call_PARKED", + "output": "wrote:x", + "id": "__fake_id__", + }, + ], + "before": None, + "persisted_count": 2, + "held": True, + } + + await resume_pending_session_write(state, session) # type: ignore[arg-type] + + assert state._pending_session_write is None + assert [item.get("call_id") for item in session.added] == ["call_PARKED", "call_PARKED"] + assert all("id" not in item for item in session.added) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_terminal_resume_with_a_preamble_lands_it_once(streamed: bool) -> None: + # With output guardrails the final sweep rebuilds the whole current response, held + # batch included; the deduplication cannot key the assistant preamble, so feeding + # the batch again used to land the preamble twice. + session = SimpleListSession() + agent = _make_terminal_tool_agent(with_preamble=True) + state = await _parked_and_approved(agent, session, streamed=streamed) + await _run(agent, state, session, streamed=streamed) + + items = await session.get_items() + assert _orphaned_outputs(items) == [] + assert _parked_pair(items) == _EXPECTED_PAIR + preambles = [item for item in items if _PREAMBLE_TEXT in json.dumps(item)] + assert len(preambles) == 1 + + +def test_the_pairing_guard_speaks_every_approval_identity() -> None: + # A hosted MCP approval request identifies itself with ``id`` and its response + # points back with ``approval_request_id``; custom calls pair by ``call_id``. A + # request kind the guard cannot key would settle alone and poison the Session the + # same way an unpaired function call does. + from agents.run_internal.session_persistence import _held_items_safe_to_settle + + unpaired_mcp: TResponseInputItem = { + "type": "mcp_approval_request", + "id": "mcpr_1", + "name": "do_it", + "server_label": "srv", + "arguments": "{}", + } + paired_mcp: TResponseInputItem = { + "type": "mcp_approval_request", + "id": "mcpr_2", + "name": "do_it", + "server_label": "srv", + "arguments": "{}", + } + mcp_response: TResponseInputItem = { + "type": "mcp_approval_response", + "approval_request_id": "mcpr_2", + "approve": True, + } + unpaired_custom: TResponseInputItem = { + "type": "custom_tool_call", + "call_id": "cust_1", + "name": "custom", + "input": "", + } + preamble: TResponseInputItem = {"role": "assistant", "content": "hi", "type": "message"} + + kept = _held_items_safe_to_settle( + [unpaired_mcp, paired_mcp, mcp_response, unpaired_custom, preamble], [], None + ) + assert kept == [paired_mcp, mcp_response, preamble] + + still_pending = _held_items_safe_to_settle( + [unpaired_mcp], [], None, pending_call_ids={"mcpr_1"} + ) + assert still_pending == [unpaired_mcp] + + +def test_the_pairing_guard_prunes_with_the_canonical_rule() -> None: + # The prune delegates to drop_orphan_function_calls, so every family that map + # owns pairs correctly (a shell call included) and a reasoning item riding + # immediately before a dropped call goes with it: the Responses API rejects + # reasoning without its required following item. + from agents.run_internal.session_persistence import _held_items_safe_to_settle + + reasoning = cast("TResponseInputItem", {"type": "reasoning", "id": "rs_1", "summary": []}) + unpaired_shell = cast( + "TResponseInputItem", + { + "type": "shell_call", + "call_id": "sh_1", + "id": "sh_item_1", + "status": "completed", + "action": {"type": "exec", "command": "ls"}, + }, + ) + paired_call: TResponseInputItem = { + "type": "function_call", + "call_id": "fn_1", + "name": "write_thing", + "arguments": "{}", + } + paired_output: TResponseInputItem = { + "type": "function_call_output", + "call_id": "fn_1", + "output": "ok", + } + + kept = _held_items_safe_to_settle( + [reasoning, unpaired_shell, paired_call, paired_output], [], None + ) + assert kept == [paired_call, paired_output] + + still_pending = _held_items_safe_to_settle( + [reasoning, unpaired_shell], [], None, pending_call_ids={"sh_1"} + ) + assert still_pending == [reasoning, unpaired_shell] + + +@pytest.mark.asyncio +async def test_settled_held_items_count_toward_the_turn_persisted_count() -> None: + # A held batch can settle with no accompanying run items (an approval-only turn + # converts to nothing persistable), so it lands through the original_input slot and + # save_result_to_session returns zero new items. The settled calls are still this + # turn's persisted items: leaving them uncounted would let a later gate-enabled + # resume pass the resumed-safety validation with a zero count and re-append them. + from agents.run_internal.run_steps import NextStepRunAgain + from agents.run_internal.session_persistence import save_resumed_turn_items + + session = SimpleListSession() + state = RunState( + context=None, + original_input="go", + starting_agent=_make_deferring_agent(), + max_turns=5, + ) + state._current_step = NextStepRunAgain() + held = [ + { + "type": "function_call", + "call_id": "call_PARKED", + "name": "write_thing", + "arguments": "{}", + }, + {"type": "function_call_output", "call_id": "call_PARKED", "output": "wrote:x"}, + ] + + count = await save_resumed_turn_items( + session=session, + items=[], + held_input=held, # type: ignore[arg-type] + persisted_count=0, + response_id=None, + run_state=state, + ) + + assert count == 2 + assert _parked_pair(await session.get_items()) == _EXPECTED_PAIR + + +@pytest.mark.asyncio +async def test_registration_forces_the_conversations_reasoning_policy() -> None: + # A Conversations backend keeps a server-identified reasoning item persistable by + # forcing the reasoning-id policy to None, exactly as the normal save path does; a + # deferred registration under "omit" must match or the sanitization drops it. + from openai.types.responses import ResponseReasoningItem + from openai.types.responses.response_reasoning_item import Summary + + from agents.items import ReasoningItem + from agents.run_internal.session_persistence import defer_interrupted_session_write + + agent = _make_deferring_agent() + state = RunState( + context=None, + original_input="go", + starting_agent=agent, + max_turns=5, + ) + state._reasoning_item_id_policy = "omit" + reasoning = ReasoningItem( + agent=agent, + raw_item=ResponseReasoningItem( + id="rs_server_1", + summary=[Summary(text="because", type="summary_text")], + type="reasoning", + ), + ) + + defer_interrupted_session_write( + state, + _RecordingConversationsSession(), # type: ignore[arg-type] + run_items=[reasoning], + reasoning_item_id_policy="omit", + ) + + pending = state._pending_session_write + assert pending is not None + reasoning_items = [item for item in pending["items"] if item.get("type") == "reasoning"] + assert reasoning_items and reasoning_items[0].get("id") == "rs_server_1" + + +@pytest.mark.asyncio +async def test_zero_count_final_save_arms_recovery_even_when_deduplicated() -> None: + # The zero-count branch of the final save: with guardrails the rebuilt items carry + # the held batch, so it deduplicates out of the append, yet the append still lands + # the approved call and output. The recovery registration must stay armed off the + # claimed-batch flag, not the emptied payload, or a failing append loses the batch + # with no pending record to reconcile. + from openai.types.responses import ResponseFunctionToolCall + + from agents.items import ToolCallItem, ToolCallOutputItem + from agents.run_internal.agent_runner_helpers import save_final_turn_items_after_guardrails + + session = _FailingResumeSession() + agent = _make_deferring_agent() + state = RunState(context=None, original_input="go", starting_agent=agent, max_turns=5) + state._current_turn_persisted_item_count = 0 + + call = ResponseFunctionToolCall( + call_id="call_PARKED", name="write_thing", arguments="{}", type="function_call" + ) + held = [ + { + "type": "function_call", + "call_id": "call_PARKED", + "name": "write_thing", + "arguments": "{}", + }, + {"type": "function_call_output", "call_id": "call_PARKED", "output": "wrote:x"}, + ] + # The rebuilt final items already contain the held batch (guardrail rebuild), so the + # held payload deduplicates out of the append. + final_items = [ + ToolCallItem(agent=agent, raw_item=call), + ToolCallOutputItem( + agent=agent, + raw_item={ + "type": "function_call_output", + "call_id": "call_PARKED", + "output": "wrote:x", + }, + output="wrote:x", + ), + ] + + session.failure = "before" + with pytest.raises(RuntimeError, match="session append failed"): + await save_final_turn_items_after_guardrails( + session=session, + run_state=state, + session_persistence_enabled=True, + input_guardrail_results=[], + items=final_items, + response_id=None, + held_input=held, # type: ignore[arg-type] + ) + + # The append was registered before it ran, so the batch is recorded to reconcile. + assert state._pending_session_write is not None + assert "call_PARKED" in {i.get("call_id") for i in state._pending_session_write["items"]} + + +@pytest.mark.asyncio +async def test_the_settled_count_matches_what_the_append_actually_wrote() -> None: + # The resolved turn re-delivers the very output the batch already folded in, so it + # dedups away inside the append. Counting the batch by its raw length would report + # more persisted items than exist, and the count slices the next save of this turn + # positionally: an inflated count drops resolved items out of their own write. + from agents.items import ToolCallOutputItem + from agents.run_internal.session_persistence import save_resumed_turn_items + + agent = _make_deferring_agent() + call: TResponseInputItem = { + "type": "function_call", + "call_id": "call_PARKED", + "name": "write_thing", + "arguments": "{}", + } + output: TResponseInputItem = { + "type": "function_call_output", + "call_id": "call_PARKED", + "output": "wrote:x", + } + session = SimpleListSession() + + count = await save_resumed_turn_items( + run_state=None, + session=session, + items=[ToolCallOutputItem(agent=agent, raw_item=output, output="wrote:x")], + held_input=[call, output], + persisted_count=0, + response_id=None, + reasoning_item_id_policy=None, + ) + + assert count == len(await session.get_items()) + + +class _FinalOutputHookFailure(RunHooks[Any]): + """Fail the run at the final-output hook, after the terminal step is decided.""" + + async def on_agent_end(self, context: Any, agent: Any, output: Any) -> None: + raise RuntimeError("final output hook failed") + + +@pytest.mark.asyncio +async def test_a_failed_max_turns_finalization_keeps_the_held_record() -> None: + # The batch is disposed of when the run actually ends, not when the terminal step + # is chosen. Validation, the final-output hooks and the output guardrails all run + # after that choice and all can raise, and a run that raises may still be retried + # or reattached with the executed tool's call and output reachable only here. + from agents.run_internal.run_loop import finalize_max_turns_handler_output + + session = SimpleListSession() + agent = _make_deferring_agent() + state = await _parked_and_approved(agent, session, streamed=False) + assert state._pending_session_write is not None + + async def _no_save(items: list[Any]) -> None: + return None + + with pytest.raises(RuntimeError): + await finalize_max_turns_handler_output( + agent=agent, + hooks=_FinalOutputHookFailure(), + run_config=RunConfig(tracing_disabled=True), + output="stopped at max turns", + context_wrapper=RunContextWrapper(context=None), + output_guardrail_results=[], + save_items_after_guardrails=_no_save, + include_in_history=False, + run_state=state, + ) + + assert state._pending_session_write is not None + + +def _make_deferring_agent_with_a_turn_after_the_resume() -> Agent: + """A gated write whose resume runs one more model turn before finishing. + + The extra turn moves the final output past the resumed boundary and onto the main + loop, which owns its own detached-completion disposal. + """ + return Agent( + name="deferred repro (turn after resume)", + instructions="Always call write_thing.", + model=ScriptedModel( + [ + ModelStep(output=[function_call("look_up", {"query": "x"}, call_id="call_LOOKUP")]), + ModelStep( + output=[function_call("write_thing", {"query": "x"}, call_id="call_PARKED")] + ), + ModelStep(output=[function_call("look_up", {"query": "y"}, call_id="call_AFTER")]), + ModelStep(output=[assistant_message("done")]), + ] + ), + tools=[look_up, write_thing], + output_guardrails=[always_fine], + tool_use_behavior=_DEFERRING_BEHAVIOR, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +@pytest.mark.parametrize( + "make_agent", + [_make_deferring_agent, _make_deferring_agent_with_a_turn_after_the_resume], + ids=["final-on-the-resumed-turn", "final-on-a-later-turn"], +) +async def test_a_failed_detached_completion_keeps_the_held_record( + streamed: bool, make_agent: Callable[[], Agent] +) -> None: + # A detached completion discards the batch because the run ends there, but only + # once it has ended: the guardrails and the final save run after the terminal step + # is chosen, and a failure there leaves a checkpoint whose reattach is the batch's + # only remaining way into the Session. + from agents import output_guardrail + + @output_guardrail + async def _fails(ctx: Any, agent: Agent, output: Any) -> GuardrailFunctionOutput: + raise RuntimeError("output guardrail failed") + + session = SimpleListSession() + agent = make_agent() + state = await _parked_and_approved(agent, session, streamed=streamed) + assert state._pending_session_write is not None + agent.output_guardrails = [*agent.output_guardrails, _fails] + + with pytest.raises(RuntimeError): + await _run(agent, state, None, streamed=streamed) + + assert state._pending_session_write is not None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("parked_store", [None, False, True]) +async def test_a_re_park_keeps_the_storage_setting_the_response_was_produced_under( + parked_store: bool | None, +) -> None: + # The batch belongs to the parked response, and the settle resolves that + # response's compaction mode from this value. Presence decides, not truthiness: a + # park under the ordinary ``store=None`` records a real setting, and a + # re-interruption under a different one must not overwrite it. + from agents.run_internal.session_persistence import defer_interrupted_session_write + + class _Session: + session_id = "s1" + + state = object.__new__(RunState) + state._pending_session_write = { + "session_id": "s1", + "items": [ + {"type": "function_call", "call_id": "call_PARKED", "name": "t", "arguments": "{}"} + ], + "before": None, + "persisted_count": 1, + "held": True, + "response_id": "resp_parked", + "store": parked_store, + } + state._current_turn_persisted_item_count = 0 + state._reasoning_item_id_policy = None + + defer_interrupted_session_write( + state, + _Session(), # type: ignore[arg-type] + run_items=[], + reasoning_item_id_policy=None, + response_id="resp_reinterrupted", + store=not parked_store, + ) + + assert state._pending_session_write is not None + assert state._pending_session_write["store"] is parked_store + assert state._pending_session_write["response_id"] == "resp_parked" + + +@pytest.mark.asyncio +async def test_a_detached_re_park_folds_under_the_batch_registration_policy() -> None: + # A Conversations-origin batch was converted preserving server reasoning ids. The + # detached re-park cannot see the backend, so it must fold under the policy the + # record carries rather than the resuming run's own: an id stripped here is + # unrecoverable and the reattach would drop the reasoning item as unpersistable. + from agents.items import ReasoningItem + from agents.run_internal.session_persistence import extend_held_session_write + + agent = _make_deferring_agent() + state = object.__new__(RunState) + state._pending_session_write = { + "session_id": "conv_abc", + "items": [ + {"type": "function_call", "call_id": "call_PARKED", "name": "t", "arguments": "{}"} + ], + "before": None, + "persisted_count": 1, + "held": True, + "response_id": "resp_parked", + "store": None, + "reasoning_item_id_policy": None, + } + state._current_turn_persisted_item_count = 0 + reasoning = ReasoningItem( + agent=agent, + raw_item={"id": "rs_SERVER_ID", "type": "reasoning", "summary": [], "content": []}, + ) + + extend_held_session_write(state, run_items=[reasoning], reasoning_item_id_policy="omit") + + items = state._pending_session_write["items"] + reasoning_ids = [i.get("id") for i in items if i.get("type") == "reasoning"] + assert reasoning_ids == ["rs_SERVER_ID"] + assert state._pending_session_write["reasoning_item_id_policy"] is None + + +@pytest.mark.asyncio +async def test_the_park_records_the_conversion_policy_it_used() -> None: + # The record owns how its items were converted. A Conversations park forces the + # preserving policy regardless of the run's own setting, and the recorded value is + # what a later detached fold must reuse. + from agents.items import ToolCallItem + from agents.memory.openai_conversations_session import OpenAIConversationsSession + from agents.run_internal.session_persistence import defer_interrupted_session_write + + session = object.__new__(OpenAIConversationsSession) + session._session_id = "conv_abc" + state = object.__new__(RunState) + state._pending_session_write = None + state._current_turn_persisted_item_count = 0 + call = ToolCallItem( + agent=_make_deferring_agent(), + raw_item={ + "type": "function_call", + "call_id": "call_PARKED", + "name": "t", + "arguments": "{}", + }, + ) + + defer_interrupted_session_write( + state, + session, + run_items=[call], + reasoning_item_id_policy="omit", + response_id="resp_parked", + store=None, + ) + + assert state._pending_session_write is not None + assert state._pending_session_write["reasoning_item_id_policy"] is None + + +class _CompactionRecordingSession(SimpleListSession): + """Record the compaction bookkeeping a compaction-aware backend expects.""" + + def __init__(self) -> None: + super().__init__() + self.compactions: list[dict[str, Any]] = [] + + async def _defer_compaction(self, response_id: str, store: bool | None = None) -> None: + self.compactions.append({"deferred": response_id, "store": store}) + + def _get_deferred_compaction_response_id(self) -> str | None: + return None + + async def run_compaction(self, args: Any = None) -> None: + self.compactions.append(dict(args or {})) + + +@pytest.mark.asyncio +async def test_the_compaction_deferral_reads_the_settling_batch_not_the_callers_input() -> None: + # The batch settles through ``original_input``, so the deferral has to look there; + # but that slot also carries the caller's own turn input on every ordinary + # interruption save. Reading the whole slot would defer compaction for a response + # that produced no local tool output, purely because the caller resumed with an + # earlier one in its input. + from agents.run_internal.session_persistence import save_result_to_session + + session = _CompactionRecordingSession() + caller_input: list[TResponseInputItem] = [ + {"type": "function_call", "call_id": "call_EARLIER", "name": "t", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "call_EARLIER", "output": "old"}, + {"role": "user", "content": "go"}, + ] + + await save_result_to_session(session, caller_input, [], None, response_id="resp_fresh") + + assert [entry for entry in session.compactions if "deferred" in entry] == [] + + +@pytest.mark.asyncio +async def test_the_settled_count_survives_the_compaction_deferral_branch() -> None: + # The deferral branch is the one every held settle with outputs takes on a + # compaction-aware backend, so returning the run-item count alone there reports a + # turn that persisted less than it wrote. That count gates the final sweep's + # re-append protection on a later gate-enabled resume. + from agents.run_internal.session_persistence import save_result_to_session + + session = _CompactionRecordingSession() + held: list[TResponseInputItem] = [ + {"type": "function_call", "call_id": "call_PARKED", "name": "t", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "call_PARKED", "output": "ok"}, + ] + + count = await save_result_to_session( + session, held, [], None, response_id="resp_parked", settling_held_batch=True + ) + + assert [entry for entry in session.compactions if "deferred" in entry] == [ + {"deferred": "resp_parked", "store": None} + ] + assert count == len(await session.get_items()) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_partial_settle_on_a_compaction_session_still_fails_the_gated_resume_fast( + streamed: bool, +) -> None: + # Park two calls, approve one, and let the gate lapse for that resume so the batch + # settles into a compaction-aware session mid-run. Re-enable the gate and approve + # the rest: the settled turn's count must cover what the settle wrote, or the + # final sweep treats the turn as unpersisted and appends the stored items again. + from agents.exceptions import UserError + + session = _CompactionRecordingSession() + agent = _make_multi_approval_agent() + + first = await _run(agent, "go", session, streamed=streamed) + state = await _serialized_round_trip(first, agent) + state.approve( + next( + interruption + for interruption in state.get_interruptions() + if getattr(interruption.raw_item, "call_id", None) == "call_PARKED" + ) + ) + gate = agent.output_guardrails + agent.output_guardrails = [] + second = await _run(agent, state, session, streamed=streamed) + assert len(second.interruptions) == 1 + agent.output_guardrails = gate + + state = await _serialized_round_trip(second, agent) + for interruption in state.get_interruptions(): + state.approve(interruption) + # The settled turn persisted items, so the re-enabled gate must refuse the resume + # outright; an undercounted turn is what would let it proceed and re-append the + # stored items through the final sweep. + with pytest.raises(UserError, match="output guardrails after current-turn items"): + await _run(agent, state, session, streamed=streamed) + + items = await session.get_items() + assert _call_ids(items).count("call_PARKED") == 1 + assert _call_ids(items).count("call_PARKED_2") == 1 + outputs = {item.get("call_id") for item in items if item.get("type") == "function_call_output"} + # Only the second call may still be awaiting its output; nothing is duplicated. + assert outputs == {"call_PARKED"} + + +@pytest.mark.asyncio +async def test_a_held_mcp_approval_pair_defers_compaction_when_it_settles() -> None: + # The approval response is the locally produced half of its pair and must stay + # associated with the response chain that carried the request; compacting that + # response before the model consumes the approval drops it in + # ``previous_response_id`` mode. + from agents.run_internal.session_persistence import save_result_to_session + + session = _CompactionRecordingSession() + held: list[TResponseInputItem] = [ + { + "type": "mcp_approval_request", + "id": "mcpr_1", + "server_label": "srv", + "name": "do_it", + "arguments": "{}", + }, + {"type": "mcp_approval_response", "approval_request_id": "mcpr_1", "approve": True}, + ] + + count = await save_result_to_session( + session, held, [], None, response_id="resp_parked", settling_held_batch=True + ) + + assert [entry for entry in session.compactions if "deferred" in entry] == [ + {"deferred": "resp_parked", "store": None} + ] + assert [entry for entry in session.compactions if "response_id" in entry] == [] + assert count == len(await session.get_items()) + + +@pytest.mark.asyncio +async def test_an_ordinary_mcp_approval_response_defers_compaction_too() -> None: + # The non-deferred resume commits the approval response as a run item, and the + # classification must treat both carriers alike: deferring for the settled dict + # but not for the run item would leave the same response compacted or not + # depending on which path persisted it. + from agents.items import MCPApprovalResponseItem + from agents.run_internal.session_persistence import save_result_to_session + + session = _CompactionRecordingSession() + agent = _make_deferring_agent() + response_item = MCPApprovalResponseItem( + agent=agent, + raw_item={ + "type": "mcp_approval_response", + "approval_request_id": "mcpr_1", + "approve": True, + }, + ) + + await save_result_to_session(session, [], [response_item], None, response_id="resp_live") + + assert [entry for entry in session.compactions if "deferred" in entry] == [ + {"deferred": "resp_live", "store": None} + ] + + +@pytest.mark.asyncio +async def test_the_final_sweep_settle_defers_compaction_and_counts_what_it_wrote() -> None: + # A reattached detached carry can reach the final exit with a zero persisted + # count, so the batch settles through the final sweep's direct save. That save + # must speak the same settle dialect as every other one: the deferral must see + # the batch's outputs even when the final turn carries none of its own, and the + # returned count must cover what the append actually wrote. + from agents.items import MessageOutputItem + from agents.run_internal.agent_runner_helpers import save_final_turn_items_after_guardrails + from agents.testing.model import assistant_message + + session = _CompactionRecordingSession() + agent = _make_deferring_agent() + state = object.__new__(RunState) + state._pending_session_write = None + state._current_turn_persisted_item_count = 0 + state._reasoning_item_id_policy = None + state._current_step = None + held: list[TResponseInputItem] = [ + {"type": "function_call", "call_id": "call_PARKED", "name": "t", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "call_PARKED", "output": "ok"}, + ] + + count = await save_final_turn_items_after_guardrails( + session=session, + run_state=state, + session_persistence_enabled=True, + input_guardrail_results=[], + items=[MessageOutputItem(agent=agent, raw_item=assistant_message("done"))], + response_id="resp_final", + held_input=held, + ) + + assert [entry for entry in session.compactions if "deferred" in entry] == [ + {"deferred": "resp_final", "store": None} + ] + assert count == len(await session.get_items()) + + +@pytest.mark.asyncio +async def test_the_entry_settle_runs_the_compaction_bookkeeping() -> None: + # The entry settle goes through the canonical persistence path, so a + # compaction-aware backend still gets the bookkeeping for the response the held + # batch belongs to. Appending behind that path would silently skip a supported + # compaction hook for the interrupted response. + from agents.run_internal.run_steps import NextStepRunAgain + from agents.run_internal.session_persistence import resume_pending_session_write + + session = _CompactionRecordingSession() + state = RunState( + context=None, + original_input="go", + starting_agent=_make_deferring_agent(), + max_turns=5, + ) + state._current_step = NextStepRunAgain() + state._pending_session_write = { + "session_id": "test", + "items": [ + { + "type": "function_call", + "call_id": "call_PARKED", + "name": "write_thing", + "arguments": "{}", + }, + {"type": "function_call_output", "call_id": "call_PARKED", "output": "wrote:x"}, + ], + "before": None, + "persisted_count": 2, + "held": True, + "response_id": "resp_parked", + } + + await resume_pending_session_write(state, session) # type: ignore[arg-type] + + assert state._pending_session_write is None + assert _parked_pair(await session.get_items()) == _EXPECTED_PAIR + # The batch carries the approved tool's output, so this response's compaction must + # be DEFERRED, not run: compacting it here would discard the very output that just + # landed. Asserting the specific hook is the point; "some hook fired" would pass + # either way. + assert session.compactions == [{"deferred": "resp_parked", "store": None}], ( + f"expected a deferred compaction for the parked response, got {session.compactions}" + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_the_park_records_the_response_the_batch_belongs_to(streamed: bool) -> None: + # The settle runs the compaction bookkeeping for the response the withheld batch + # came from, so the park has to record which response that was. + session = SimpleListSession() + agent = _make_deferring_agent() + + first = await _run(agent, "do the thing", session, streamed=streamed) + assert len(first.interruptions) == 1 + + pending = first.to_state().to_json()["pending_session_write"] + assert pending["held"] is True + assert pending["response_id"] == first.raw_responses[-1].response_id + + +@pytest.mark.asyncio +@pytest.mark.filterwarnings("ignore:Pydantic serializer warnings:UserWarning") +async def test_a_rejected_max_turns_handler_output_keeps_the_held_record() -> None: + # The discard belongs to a handler that actually ends the run. Validation rejects a + # wrongly typed handler output by raising, and the streamed runner discards only + # after its finalization completes, so discarding ahead of the raise would leave + # the caller's live RunState without a batch its streamed twin still holds. + from agents.exceptions import UserError + from agents.run_internal.run_loop import finalize_max_turns_handler_output + + session = SimpleListSession() + agent = _make_deferring_agent() + agent.output_type = int + state = await _parked_and_approved(agent, session, streamed=False) + assert state._pending_session_write is not None + + async def _no_save(items: list[Any]) -> None: + return None + + with pytest.raises(UserError): + await finalize_max_turns_handler_output( + agent=agent, + hooks=RunHooks(), + run_config=RunConfig(tracing_disabled=True), + output="not an int", + context_wrapper=RunContextWrapper(context=None), + output_guardrail_results=[], + save_items_after_guardrails=_no_save, + include_in_history=False, + run_state=state, + ) + + assert state._pending_session_write is not None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_max_turns_handler_completion_clears_the_held_record(streamed: bool) -> None: + # A max-turn handler ends the run, so a held batch still standing has no later + # gate-legal exit to settle it: both runners must report the same terminal state, + # with no pending write left to invalidate the finished run's checkpoint. + from agents.run_internal.run_loop import finalize_max_turns_handler_output + + session = SimpleListSession() + agent = _make_deferring_agent() + state = await _parked_and_approved(agent, session, streamed=streamed) + assert state._pending_session_write is not None + + async def _no_save(items: list[Any]) -> None: + return None + + await finalize_max_turns_handler_output( + agent=agent, + hooks=RunHooks(), + run_config=RunConfig(tracing_disabled=True), + output="stopped at max turns", + context_wrapper=RunContextWrapper(context=None), + output_guardrail_results=[], + save_items_after_guardrails=_no_save, + include_in_history=False, + run_state=state, + ) + + assert state._pending_session_write is None + + +def _boom_custom_data_extractor(ctx: Any) -> dict[str, Any]: + raise RuntimeError("extractor boom") + + +@function_tool( + name_override="write_thing", + needs_approval=True, + custom_data_extractor=_boom_custom_data_extractor, +) +def write_thing_with_failing_extractor(query: str) -> str: + return f"wrote:{query}" + + +def _make_failing_extractor_agent() -> Agent: + """The approved tool succeeds, then its post-output callback raises.""" + return Agent( + name="deferred repro (failing extractor)", + instructions="x", + model=ScriptedModel( + [ + ModelStep(output=[function_call("look_up", {"query": "x"}, call_id="call_LOOKUP")]), + ModelStep( + output=[function_call("write_thing", {"query": "x"}, call_id="call_PARKED")] + ), + ModelStep(output=[assistant_message("done")]), + ] + ), + tools=[look_up, write_thing_with_failing_extractor], + output_guardrails=[always_fine], + tool_use_behavior=_DEFERRING_BEHAVIOR, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_a_post_output_callback_failure_keeps_the_executed_output(streamed: bool) -> None: + # The approved tool ran and its output was committed when the post-output callback + # raised. A retry skips the completed invocation and produces no new session items, + # so the batch has to carry that output from the commit boundary or the executed + # call and its result vanish from history. + session = SimpleListSession() + agent = _make_failing_extractor_agent() + state = await _parked_and_approved(agent, session, streamed=streamed) + + with pytest.raises(Exception, match="extractor boom"): + await _run(agent, state, session, streamed=streamed) + + pending = state._pending_session_write + assert pending is not None + assert _parked_pair(pending["items"]) == _EXPECTED_PAIR + + +def _make_never_finishing_agent() -> Agent: + """Parks on turn two, then keeps calling tools so max turns is what ends the run.""" + steps = [ + ModelStep(output=[function_call("look_up", {"query": "a"}, call_id="call_LOOKUP")]), + ModelStep(output=[function_call("write_thing", {"query": "x"}, call_id="call_PARKED")]), + ] + steps += [ + ModelStep(output=[function_call("look_up", {"query": f"q{i}"}, call_id=f"call_L{i}")]) + for i in range(8) + ] + return Agent( + name="deferred repro (never finishing)", + instructions="x", + model=ScriptedModel(steps), + tools=[look_up, write_thing], + output_guardrails=[always_fine], + tool_use_behavior=_DEFERRING_BEHAVIOR, + ) + + +@pytest.mark.asyncio +async def test_a_streamed_max_turns_completion_clears_the_held_record() -> None: + # The streaming runner reaches its max-turn handler through its own terminal path, + # not the shared helper, so it needs its own coverage: a detached resume that runs + # out of turns must not report terminal handler output while carrying a resumable + # pending write the non-streaming runner had already dropped. + session = SimpleListSession() + agent = _make_never_finishing_agent() + first = await _run(agent, "go", session, streamed=True) + assert len(first.interruptions) == 1 + state = await _serialized_round_trip(first, agent) + state.approve(state.get_interruptions()[0]) + assert state._pending_session_write is not None + + resumed = Runner.run_streamed( + agent, + state, + session=None, + max_turns=3, + error_handlers={"max_turns": lambda data: "stopped at max turns"}, + ) + async for _ in resumed.stream_events(): + pass + + assert resumed.final_output == "stopped at max turns" + assert "pending_session_write" not in resumed.to_state().to_json() + assert state._pending_session_write is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_the_park_records_the_store_the_response_was_produced_under( + streamed: bool, +) -> None: + # The settle defers compaction for the parked response, and the deferral resolves a + # compaction mode from the store setting. That setting belongs to the turn the + # batch was withheld in, not to the resume, so the park records it. + session = SimpleListSession() + agent = _make_deferring_agent() + agent.model_settings = replace(agent.model_settings, store=True) + + first = await _run(agent, "do the thing", session, streamed=streamed) + assert len(first.interruptions) == 1 + + pending = first.to_state().to_json()["pending_session_write"] + assert pending["held"] is True + assert pending["store"] is True + + +@pytest.mark.asyncio +async def test_the_entry_settle_defers_with_the_recorded_store() -> None: + # The recorded store reaches the deferral, so the hook resolves the same compaction + # mode the ordinary persistence path would have resolved for that response. + from agents.run_internal.run_steps import NextStepRunAgain + from agents.run_internal.session_persistence import resume_pending_session_write + + session = _CompactionRecordingSession() + state = RunState( + context=None, + original_input="go", + starting_agent=_make_deferring_agent(), + max_turns=5, + ) + state._current_step = NextStepRunAgain() + state._pending_session_write = { + "session_id": "test", + "items": [ + { + "type": "function_call", + "call_id": "call_PARKED", + "name": "write_thing", + "arguments": "{}", + }, + {"type": "function_call_output", "call_id": "call_PARKED", "output": "wrote:x"}, + ], + "before": None, + "persisted_count": 2, + "held": True, + "response_id": "resp_parked", + "store": True, + } + + await resume_pending_session_write(state, session) # type: ignore[arg-type] + + assert session.compactions == [{"deferred": "resp_parked", "store": True}] diff --git a/tests/test_run_impl_resume_paths.py b/tests/test_run_impl_resume_paths.py index 482a13edd8..6b3084af0f 100644 --- a/tests/test_run_impl_resume_paths.py +++ b/tests/test_run_impl_resume_paths.py @@ -499,7 +499,18 @@ async def test_failed_streamed_result_checkpoint_retains_detached_pending_write( @pytest.mark.asyncio -@pytest.mark.parametrize("invalid", ["old-schema", "batch-shape"]) +@pytest.mark.parametrize( + "invalid", + [ + "old-schema", + "batch-shape", + "held-shape", + "held-with-before", + "held-under-1-17", + "held-keys-without-held", + "policy-shape", + ], +) async def test_pending_session_write_rejects_invalid_serialized_checkpoint(invalid: str) -> None: agent, _, session, state, _ = await _approved_session_state(False) session.failure = "before" @@ -508,12 +519,54 @@ async def test_pending_session_write_rejects_invalid_serialized_checkpoint(inval payload = state.to_json() if invalid == "old-schema": payload["$schemaVersion"] = "1.16" - else: + elif invalid == "batch-shape": payload["pending_session_write"]["items"] = "not an item batch" + elif invalid == "held-shape": + payload["pending_session_write"]["held"] = "yes" + elif invalid == "held-keys-without-held": + # response_id and store describe the withheld response, so they are refused on + # an ordinary pending write where nothing consumes them. + payload["pending_session_write"]["response_id"] = "resp_1" + elif invalid == "policy-shape": + # The conversion-policy key only speaks the two policy literals or None; any + # other value would silently change how a fold converts the batch's items. + payload["pending_session_write"]["held"] = True + payload["pending_session_write"]["before"] = None + payload["pending_session_write"]["reasoning_item_id_policy"] = "banana" + elif invalid == "held-under-1-17": + # 1.17 defined the pending write as exactly four keys, so the held variant is + # only readable under the version that introduced it. + payload["$schemaVersion"] = "1.17" + payload["pending_session_write"]["held"] = True + payload["pending_session_write"]["before"] = None + else: + # A held batch was never offered to the Session, so recorded digests and the + # held marker cannot coexist on one record. + payload["pending_session_write"]["held"] = True with pytest.raises(UserError, match="pending Session write is invalid"): await RunState.from_json(agent, payload) +@pytest.mark.asyncio +async def test_pending_session_write_without_the_held_key_keeps_its_meaning() -> None: + # The four-key form 1.17 defined still settles eagerly on resume entry under its + # own label, unchanged by the held variant that 1.18 introduced. + agent, model, session, state, effects = await _approved_session_state(False) + session.failure = "before" + with pytest.raises(RuntimeError): + await _run_session_resume(agent, state, session, False) + payload = state.to_json() + assert "held" not in payload["pending_session_write"] + payload["$schemaVersion"] = "1.17" + payload["pending_session_write"].pop("response_id", None) + restored = await RunState.from_json(agent, payload) + + result = await _run_session_resume(agent, restored, session, False) + assert result.final_output == "done" + assert effects == [7] + assert _charge_pair(await session.get_items()) == ["function_call", "function_call_output"] + + @pytest.mark.asyncio async def test_resumed_session_append_partial_commit_fails_closed() -> None: agent, model, session, state, effects = await _approved_session_state(False) diff --git a/tests/test_run_state.py b/tests/test_run_state.py index cd2daa51b6..4505ac8939 100644 --- a/tests/test_run_state.py +++ b/tests/test_run_state.py @@ -8611,6 +8611,7 @@ def test_supported_schema_versions_match_released_boundary(self): "1.14", "1.15", "1.16", + "1.17", CURRENT_SCHEMA_VERSION, } )