diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py index f09c3a6edd..8a07e0718d 100644 --- a/src/agents/memory/openai_responses_compaction_session.py +++ b/src/agents/memory/openai_responses_compaction_session.py @@ -221,7 +221,6 @@ async def run_compaction( ) return - self._deferred_response_id = None logger.debug( "compact: start for %s using %s (mode=%s)", self._response_id, @@ -254,6 +253,12 @@ async def run_compaction( self._compaction_candidate_items = select_compaction_candidate_items(output_items) self._session_items = output_items + # Clear the deferred marker only now that compaction has actually settled. Clearing it + # before the fallible API call/replacement above would let a failed forced compaction + # silently lose its "this must be forced" signal: a later retry recomputes `force` from + # this marker, so an early clear makes the retry decline work that was still owed. + self._deferred_response_id = None + logger.debug( "compact: done for %s (mode=%s, output=%s, candidates=%s)", self._response_id, diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py index 3c7d9ee586..83f497db92 100644 --- a/src/agents/run_internal/run_loop.py +++ b/src/agents/run_internal/run_loop.py @@ -43,6 +43,7 @@ _detach_data_redacted_error_traceback, _is_error_data_redacted, _mark_error_data_redacted, + _mark_error_to_drain_stream_events, _prepare_data_redacted_error, ) from ..guardrail import OutputGuardrailResult @@ -418,6 +419,7 @@ async def _save_stream_items( response_id: str | None, update_persisted_count: bool, store: bool | None = None, + resumed_write_state: RunState | None = None, ) -> None: if not await _should_persist_stream_items( session=session, @@ -433,6 +435,7 @@ async def _save_stream_items( response_id=response_id, store=store, wrapper=streamed_result.context_wrapper, + resumed_write_state=resumed_write_state, ) if update_persisted_count and streamed_result._state is not None: streamed_result._current_turn_persisted_item_count = ( @@ -1152,6 +1155,12 @@ async def _save_stream_items_without_count( response_id=response_id, update_persisted_count=False, store=store_setting, + resumed_write_state=( + run_state + if run_state is not None + and isinstance(run_state._current_step, NextStepRunAgain) + else None + ), ) async def _save_max_turns_items( @@ -1879,23 +1888,34 @@ def _record_max_turns_handler_output( server_conversation_tracker.track_server_items(turn_result.model_response) if isinstance(turn_result.next_step, NextStepHandoff): - await _save_stream_items_without_count( - turn_session_items, - turn_result.model_response.response_id, - store_setting, - ) + # Resolve any still-in-flight parallel input guardrail before committing the + # handoff transition, so a tripwire or guardrail exception is surfaced instead + # of the state (current_agent, run_state, published events) racing ahead of an + # input guardrail that was still validating the original input. + await input_guardrail_tripwire_triggered_for_stream(streamed_result) current_agent = turn_result.next_step.new_agent if run_state is not None: run_state._current_agent = current_agent _publish_streamed_result_agent(streamed_result, current_agent) - current_span.finish(reset_current=True) - current_span = None - should_run_agent_start_hooks = True + if streamed_result._state is not None: + streamed_result._state._current_step = NextStepRunAgain() + # Queue the agent-transition event before the fallible session append so + # stream consumers observe the transition even if the append later raises. streamed_result._event_queue.put_nowait( AgentUpdatedStreamEvent(new_agent=current_agent) ) - if streamed_result._state is not None: - streamed_result._state._current_step = NextStepRunAgain() + try: + await _save_stream_items_without_count( + turn_session_items, + turn_result.model_response.response_id, + store_setting, + ) + except BaseException as session_persistence_error: + _mark_error_to_drain_stream_events(session_persistence_error) + raise + current_span.finish(reset_current=True) + current_span = None + should_run_agent_start_hooks = True if await _wait_for_streamed_turn_events_and_stop_if_cancelled(streamed_result): break diff --git a/src/agents/run_internal/session_persistence.py b/src/agents/run_internal/session_persistence.py index 2e53667e7e..ba386ac865 100644 --- a/src/agents/run_internal/session_persistence.py +++ b/src/agents/run_internal/session_persistence.py @@ -556,6 +556,65 @@ def update_run_state_after_resume( run_state._current_step = next_step # type: ignore[assignment] +async def _apply_post_write_compaction( + session: Session, + *, + response_id: str | None, + store: bool | None, + has_local_tool_outputs: bool, + wrapper: RunContextWrapper[Any] | None = None, +) -> None: + """Evaluate deferred/forced Responses compaction for a settled session append. + + Shared by the immediate-write path in ``save_result_to_session`` and the checkpoint + replay path in ``resume_pending_session_write``, so a batch that only settles later + (via a separate resume) still gets the same compaction decision it would have gotten + had the original append succeeded inline. ``wrapper`` is the caller's raw (pre-gating) + context wrapper; it is used as-is for ``run_compaction`` and re-gated here for + ``_defer_compaction``, mirroring the two call sites this helper replaces. + """ + if not response_id or not is_openai_responses_compaction_aware_session(session): + return + + if has_local_tool_outputs: + defer_compaction = getattr(session, "_defer_compaction", None) + if callable(defer_compaction): + await _call_session_method( + defer_compaction, + response_id, + store=store, + wrapper=_get_session_wrapper(session, wrapper), + ) + logger.debug( + "skip: deferring compaction for response %s due to local tool outputs", + response_id, + ) + return + + deferred_response_id = None + get_deferred = getattr(session, "_get_deferred_compaction_response_id", None) + if callable(get_deferred): + deferred_response_id = get_deferred() + force_compaction = deferred_response_id is not None + if force_compaction: + logger.debug( + "compact: forcing for response %s after deferred %s", + response_id, + deferred_response_id, + ) + compaction_args: OpenAIResponsesCompactionArgs = { + "response_id": response_id, + "force": force_compaction, + } + if store is not None: + compaction_args["store"] = store + await _call_session_method( + session.run_compaction, + compaction_args, + wrapper=wrapper, + ) + + async def save_result_to_session( session: Session | None, original_input: str | list[TResponseInputItem], @@ -663,6 +722,10 @@ async def save_result_to_session( run_state._current_turn_persisted_item_count = already_persisted + saved_run_items_count return saved_run_items_count + has_local_tool_outputs = any( + isinstance(item, ToolCallOutputItem | HandoffOutputItem) for item in new_items + ) + if resumed_write_state is not None: if resumed_write_state._pending_session_write is not None: raise UserError("Resolve the pending Session write before saving another batch") @@ -673,53 +736,31 @@ async def save_result_to_session( "persisted_count": ( resumed_write_state._current_turn_persisted_item_count + saved_run_items_count ), + "response_id": response_id, + "store": store, + "has_local_tool_outputs": has_local_tool_outputs, } - await resume_pending_session_write(resumed_write_state, session, wrapper=wrapper) + # resume_pending_session_write() applies post-write compaction itself once the + # checkpoint settles, whether that happens inline below or on a later, separate + # resume -- so it is not repeated after this call returns. + await resume_pending_session_write( + resumed_write_state, + session, + wrapper=wrapper, + compaction_wrapper=compaction_wrapper, + ) else: await _session_add_items(session, items_to_save, wrapper=wrapper) if run_state is not None: run_state._current_turn_persisted_item_count = already_persisted + saved_run_items_count - if response_id and is_openai_responses_compaction_aware_session(session): - has_local_tool_outputs = any( - isinstance(item, ToolCallOutputItem | HandoffOutputItem) for item in new_items - ) - if has_local_tool_outputs: - defer_compaction = getattr(session, "_defer_compaction", None) - if callable(defer_compaction): - await _call_session_method( - defer_compaction, - response_id, - store=store, - wrapper=wrapper, - ) - logger.debug( - "skip: deferring compaction for response %s due to local tool outputs", - response_id, - ) - return saved_run_items_count - - deferred_response_id = None - get_deferred = getattr(session, "_get_deferred_compaction_response_id", None) - if callable(get_deferred): - deferred_response_id = get_deferred() - force_compaction = deferred_response_id is not None - if force_compaction: - logger.debug( - "compact: forcing for response %s after deferred %s", - response_id, - deferred_response_id, - ) - compaction_args: OpenAIResponsesCompactionArgs = { - "response_id": response_id, - "force": force_compaction, - } - if store is not None: - compaction_args["store"] = store - await _call_session_method( - session.run_compaction, - compaction_args, + if resumed_write_state is None: + await _apply_post_write_compaction( + session, + response_id=response_id, + store=store, + has_local_tool_outputs=has_local_tool_outputs, wrapper=compaction_wrapper, ) @@ -764,12 +805,18 @@ async def resume_pending_session_write( session: Session | None, *, wrapper: RunContextWrapper[Any] | None = None, + compaction_wrapper: RunContextWrapper[Any] | None = None, ) -> None: """Settle a resumed output batch before allowing further model work. The application must supply the original backend and serialize access to its history, including independently restored RunState copies. Session has no distributed compare-and-swap or backend identity contract. A changed tail is not repaired or searched for similar items. + + ``compaction_wrapper`` defaults to ``wrapper`` when omitted; ``save_result_to_session`` + passes its own raw (pre-gating) wrapper explicitly so a batch that settles here -- either + inline or on a later, separate resume -- gets the exact same post-write Responses + compaction decision ``save_result_to_session`` would otherwise have applied itself. """ pending = run_state._pending_session_write if pending is None: @@ -815,6 +862,17 @@ def digests(items: Sequence[TResponseInputItem]) -> list[str]: # Backends may retain or transform their input; the durable checkpoint stays detached. await _session_add_items(session, copy.deepcopy(pending["items"]), wrapper=wrapper) run_state._current_turn_persisted_item_count = pending["persisted_count"] + # Keep the checkpoint until compaction also settles: if _apply_post_write_compaction + # raises below, a later retry must still be able to redo just the compaction step + # instead of silently losing it. The append itself is retry-safe (the reconciliation + # above detects an already-committed batch and skips re-appending it). + await _apply_post_write_compaction( + session, + response_id=pending.get("response_id"), + store=pending.get("store"), + has_local_tool_outputs=pending.get("has_local_tool_outputs", False), + wrapper=compaction_wrapper if compaction_wrapper is not None else wrapper, + ) run_state._pending_session_write = None finally: run_state._session_write_in_progress = False diff --git a/src/agents/run_state.py b/src/agents/run_state.py index 228dd574d8..819eee321d 100644 --- a/src/agents/run_state.py +++ b/src/agents/run_state.py @@ -38,7 +38,7 @@ ProgramOutput, ) from pydantic import BaseModel, StringConstraints, TypeAdapter, ValidationError -from typing_extensions import TypedDict, TypeVar +from typing_extensions import NotRequired, TypedDict, TypeVar from ._tool_identity import ( FunctionToolLookupKey, @@ -172,6 +172,14 @@ class _PendingSessionWrite(TypedDict): items: list[TResponseInputItem] before: list[str] | None persisted_count: int + # Compaction inputs for the batch this checkpoint is settling, so a later, separate + # resume_pending_session_write() call (not the original save_result_to_session() call) + # can still apply the same post-write Responses compaction decision. Optional so a + # RunState serialized before these fields existed degrades to "skip compaction" on + # read instead of raising KeyError. + response_id: NotRequired[str | None] + store: NotRequired[bool | None] + has_local_tool_outputs: NotRequired[bool] def _default_run_state_validation_error( @@ -4360,11 +4368,18 @@ async def _build_run_state_from_json( if pending_write is not None: from .run_internal.run_steps import NextStepInterruption, NextStepRunAgain + required_pending_write_keys = {"session_id", "items", "before", "persisted_count"} + # response_id/store/has_local_tool_outputs carry the compaction inputs needed to replay + # deferred/forced Responses compaction on a later, separate resume; they are optional so + # a RunState serialized before these fields existed (same, unreleased schema version) + # still round-trips. + optional_pending_write_keys = {"response_id", "store", "has_local_tool_outputs"} 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 not required_pending_write_keys <= set(pending_write) + or not set(pending_write) <= required_pending_write_keys | optional_pending_write_keys or not isinstance(pending_write.get("session_id"), str) or not isinstance(pending_write.get("items"), list) or not pending_write["items"] @@ -4378,6 +4393,20 @@ async def _build_run_state_from_json( ) or type(pending_write.get("persisted_count")) is not int or pending_write["persisted_count"] < 0 + or ( + "response_id" in pending_write + and pending_write["response_id"] is not None + and not isinstance(pending_write["response_id"], str) + ) + or ( + "store" in pending_write + and pending_write["store"] is not None + and not isinstance(pending_write["store"], bool) + ) + or ( + "has_local_tool_outputs" in pending_write + and not isinstance(pending_write["has_local_tool_outputs"], bool) + ) ): raise validation_error_factory("Run state pending Session write is invalid", UserError) state._pending_session_write = copy.deepcopy(cast(_PendingSessionWrite, pending_write)) diff --git a/tests/memory/test_openai_responses_compaction_session.py b/tests/memory/test_openai_responses_compaction_session.py index 5519228ea6..e70ebf36f8 100644 --- a/tests/memory/test_openai_responses_compaction_session.py +++ b/tests/memory/test_openai_responses_compaction_session.py @@ -1208,6 +1208,58 @@ async def test_run_compaction_force_bypasses_threshold(self) -> None: mock_client.responses.compact.assert_called_once() + @pytest.mark.asyncio + async def test_run_compaction_retains_deferred_marker_when_api_call_fails(self) -> None: + """A forced compaction driven by a previously-deferred response must not lose that + "this must be forced" signal if the compact API call itself fails: a later retry needs + _deferred_response_id to still be set so it recomputes force=True, not force=False. + """ + mock_session = self.create_mock_session() + mock_session.get_items.return_value = [] + + call_count = 0 + + async def compact(**kwargs: Any) -> MagicMock: + nonlocal call_count + call_count += 1 + if call_count == 1: + raise RuntimeError("compact API blew up") + response = MagicMock() + response.output = [] + return response + + mock_client = MagicMock() + mock_client.responses.compact = AsyncMock(side_effect=compact) + + session = OpenAIResponsesCompactionSession( + session_id="test", + underlying_session=mock_session, + client=mock_client, + should_trigger_compaction=lambda _ctx: False, + ) + # Simulate a prior turn (e.g. a handoff with local tool outputs) having deferred + # compaction for this response, the way _defer_compaction() would. + session._deferred_response_id = "resp-handoff" + + with pytest.raises(RuntimeError, match="compact API blew up"): + await session.run_compaction( + { + "response_id": "resp-delegate", + "force": session._get_deferred_compaction_response_id() is not None, + } + ) + assert session._get_deferred_compaction_response_id() == "resp-handoff" + + # Retry, recomputing force the same way the checkpoint-recovery code does. + await session.run_compaction( + { + "response_id": "resp-delegate", + "force": session._get_deferred_compaction_response_id() is not None, + } + ) + assert call_count == 2 + assert session._get_deferred_compaction_response_id() is None + @pytest.mark.asyncio async def test_run_compaction_suppresses_model_dump_warnings(self) -> None: mock_session = self.create_mock_session() diff --git a/tests/test_run_impl_resume_paths.py b/tests/test_run_impl_resume_paths.py index 8e6211ceab..21d06c3650 100644 --- a/tests/test_run_impl_resume_paths.py +++ b/tests/test_run_impl_resume_paths.py @@ -9,11 +9,12 @@ from openai.types.responses import ResponseFunctionToolCall, ResponseOutputMessage import agents.run as run_module -from agents import Agent, Runner, function_tool, handoff +from agents import Agent, AgentUpdatedStreamEvent, Runner, function_tool, handoff from agents.agent import ToolsToFinalOutputResult from agents.agent_output import AgentOutputSchema from agents.decorators import tool, tool_input_guardrail, tool_output_guardrail from agents.exceptions import UserError +from agents.guardrail import GuardrailFunctionOutput, input_guardrail from agents.items import ( MessageOutputItem, ModelResponse, @@ -82,6 +83,39 @@ async def add_items(self, items: list[TResponseInputItem]) -> None: raise self.error +class _FailSecondAddItemsSession(SimpleListSession): + """Let the initial input-priming append succeed, then fail the next append. + + Unlike ``_FailingResumeSession``, this targets a specific append by call order rather than + a resume-cycle phase, so it can isolate a fresh (non-resumed) run's first real turn save. + """ + + def __init__(self) -> None: + super().__init__() + self.error = RuntimeError("session append failed") + self._call_count = 0 + + async def add_items(self, items: list[TResponseInputItem]) -> None: + self._call_count += 1 + if self._call_count == 2: + raise self.error + await super().add_items(items) + + +class _FailSecondAddItemsSessionWithYield(_FailSecondAddItemsSession): + """Same failure shape as ``_FailSecondAddItemsSession``, but the failing call performs a + genuine ``await`` (a scheduler yield) before raising, like a real I/O-backed Session + (SQLite, network, etc.) would. A purely synchronous raise never yields control back to the + ``stream_events()`` consumer before the run-loop task finishes, so a test built on it cannot + observe whether an already-queued stream event was delivered before the error surfaced. + """ + + async def add_items(self, items: list[TResponseInputItem]) -> None: + if self._call_count == 1: + await asyncio.sleep(0) + await super().add_items(items) + + class _LostAckSQLiteSession(SQLiteSession): fail_after_commit = False error = RuntimeError("session append failed") @@ -1221,3 +1255,345 @@ async def test_resumed_handoff_session_append_is_recovered_before_next_model( assert _call_pair(result.to_input_list(), "charge-1") == expected_pair assert _call_pair(result.to_input_list(), "handoff-1") == expected_pair assert "pending_session_write" not in result.to_state().to_json() + + +@pytest.mark.asyncio +async def test_fresh_streamed_handoff_preserves_agent_after_session_append_failure() -> None: + """A fresh (non-resumed) streamed run's generic-loop handoff branch must publish the new + agent and next-step state before the fallible session append, mirroring the fix already + applied to the is_resumed_state branch covered by + test_resumed_handoff_session_append_is_recovered_before_next_model. Every fresh streamed + run passes through this branch, not just resumed ones. + """ + model = ScriptedModel( + [ + [get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff-1")], + [get_text_message("done")], + ] + ) + delegate = Agent(name="delegate", model=model) + triage = Agent(name="triage", model=model, handoffs=[delegate]) + session = _FailSecondAddItemsSession() + + failed_result = Runner.run_streamed( + triage, "hello", session=session, run_config=RunConfig(tracing_disabled=True) + ) + with pytest.raises(RuntimeError) as error: + async for _ in failed_result.stream_events(): + pass + assert error.value is session.error + + state = failed_result.to_state() + assert state._current_agent is not None + assert state._current_agent.name == "delegate" + assert failed_result.current_agent.name == "delegate" + + result = await _run_session_resume(triage, state, session, False) + assert result.final_output == "done" + assert result.last_agent.name == "delegate" + assert len(model.calls) == 2 + expected_pair = ["function_call", "function_call_output"] + stored = await session.get_items() + assert _call_pair(stored, "handoff-1") == expected_pair + assert "pending_session_write" not in result.to_state().to_json() + + +@pytest.mark.asyncio +async def test_fresh_streamed_handoff_publishes_agent_update_before_session_append_failure() -> ( + None +): + """The generic-loop handoff branch must queue ``AgentUpdatedStreamEvent`` for the new agent + before the fallible session append, so ``stream_events()`` consumers observe the transition + even when the append later raises. Mirrors the already-merged ordering fix for the + ``is_resumed_state``-specific branch (lines ~1407-1437), which is out of scope here. + + Uses a session whose failing append performs a genuine ``await`` before raising: a purely + synchronous raise (as in ``_FailSecondAddItemsSession``) never yields control back to this + consumer before the run-loop task finishes, so it cannot prove event delivery either way + (a separate, pre-existing gate: ``stream_events()`` only drains an already-queued event past + a terminal error when that error was marked via ``_mark_error_to_drain_stream_events()``, + which session-append failures never are). + """ + model = ScriptedModel( + [ + [get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff-1")], + [get_text_message("done")], + ] + ) + delegate = Agent(name="delegate", model=model) + triage = Agent(name="triage", model=model, handoffs=[delegate]) + session = _FailSecondAddItemsSessionWithYield() + + failed_result = Runner.run_streamed( + triage, "hello", session=session, run_config=RunConfig(tracing_disabled=True) + ) + collected_events: list[Any] = [] + caught: RuntimeError | None = None + try: + async for event in failed_result.stream_events(): + collected_events.append(event) + except RuntimeError as error: + caught = error + assert caught is session.error + assert any( + isinstance(event, AgentUpdatedStreamEvent) and event.new_agent.name == "delegate" + for event in collected_events + ) + + +@pytest.mark.asyncio +async def test_fresh_streamed_handoff_drains_agent_update_event_for_slow_consumer() -> None: + """A session-append failure in the generic-loop handoff branch must mark itself for + stream-event draining, so a consumer that falls even slightly behind the producer (an + ordinary per-event delay, not a contrived zero-delay reader) still observes the + already-queued ``AgentUpdatedStreamEvent`` before the error surfaces. + + test_fresh_streamed_handoff_publishes_agent_update_before_session_append_failure's + zero-delay consumer passes even without draining, since it never falls behind the + producer; this test exercises the actual drain guarantee stream_events() provides via + _mark_error_to_drain_stream_events()/_should_drain_stream_events_before_raising(). + """ + model = ScriptedModel( + [ + [get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff-1")], + [get_text_message("done")], + ] + ) + delegate = Agent(name="delegate", model=model) + triage = Agent(name="triage", model=model, handoffs=[delegate]) + session = _FailSecondAddItemsSessionWithYield() + + failed_result = Runner.run_streamed( + triage, "hello", session=session, run_config=RunConfig(tracing_disabled=True) + ) + collected_events: list[Any] = [] + caught: RuntimeError | None = None + try: + async for event in failed_result.stream_events(): + # An ordinary bit of per-event consumer work, enough to fall behind the producer. + await asyncio.sleep(0.001) + collected_events.append(event) + except RuntimeError as error: + caught = error + assert caught is session.error + assert any( + isinstance(event, AgentUpdatedStreamEvent) and event.new_agent.name == "delegate" + for event in collected_events + ) + + +@pytest.mark.asyncio +async def test_fresh_streamed_handoff_awaits_parallel_input_guardrail_before_transition() -> None: + """The generic-loop handoff branch must resolve an in-flight parallel input guardrail + before committing the handoff transition (current_agent, run_state, published events), + per the Guardrail Ordering contract in .agents/references/runner-lifecycle.md. Otherwise a + handoff on turn 1 can commit the transition while a still-running parallel input guardrail + that later raises has not yet been awaited. + + Only one scripted turn is provided (the handoff itself), and the consumer adds a small + per-event delay: an in-process run with an instantly-draining consumer and a second + scripted turn can otherwise race straight through to completion before the guardrail's + sleep elapses, defeating the repro regardless of the fix. This mirrors an ordinary + consumer that does a bit of per-event work, not a contrived instant reader. + """ + guardrail_error = RuntimeError("guardrail backend exploded") + + @input_guardrail(run_in_parallel=True) + async def slow_failing_guardrail( + ctx: RunContextWrapper[Any], + agent: Agent[Any], + input: str | list[TResponseInputItem], + ) -> GuardrailFunctionOutput: + await asyncio.sleep(0.3) + raise guardrail_error + + model = ScriptedModel( + [ + [get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff-1")], + ] + ) + delegate = Agent(name="delegate", model=model) + triage = Agent( + name="triage", + model=model, + handoffs=[delegate], + input_guardrails=[slow_failing_guardrail], + ) + + streamed_result = Runner.run_streamed( + triage, "hello", run_config=RunConfig(tracing_disabled=True) + ) + caught: RuntimeError | None = None + try: + async for _ in streamed_result.stream_events(): + await asyncio.sleep(0.05) + except RuntimeError as error: + caught = error + assert caught is guardrail_error + # The handoff transition must not have been committed: the guardrail task was still + # in flight (sleeping) when the model returned the handoff, and it raised a real error + # rather than a tripwire, so no part of the observable state should have moved past triage. + assert streamed_result.current_agent.name == "triage" + state = streamed_result.to_state() + assert state._current_agent is not None + assert state._current_agent.name == "triage" + + +@pytest.mark.asyncio +async def test_fresh_streamed_handoff_replays_deferred_compaction_after_resume() -> None: + """A checkpointed handoff batch that fails to append and later settles via a separate, + standalone resume_pending_session_write() call (the generic resume-startup path in + run.py/run_loop.py, not the original save_result_to_session() call) must still apply the + same post-write Responses compaction decision save_result_to_session would have applied + inline, instead of silently and permanently losing it. See + .agents/references/session-persistence.md. + + Uses a should_trigger_compaction hook keyed on response_id (as a caller doing per-turn + compaction routing would) to make the loss observable: without the fix, the handoff's own + response_id is never evaluated by the hook at all, and the deferral it would have set is + never recorded, so the later forced compaction on the delegate's turn never happens either. + """ + hook_calls: list[str | None] = [] + + def should_trigger_compaction(context: dict[str, Any]) -> bool: + hook_calls.append(context["response_id"]) + return context["response_id"] == "resp-handoff" + + compact_calls: list[list[TResponseInputItem]] = [] + + async def compact(**kwargs: Any) -> SimpleNamespace: + items = copy.deepcopy(kwargs["input"]) + compact_calls.append(items) + return SimpleNamespace(output=items, usage=None) + + backend = _FailSecondAddItemsSession() + session = OpenAIResponsesCompactionSession( + "compaction-handoff-test", + underlying_session=backend, + client=cast(Any, SimpleNamespace(responses=SimpleNamespace(compact=compact))), + compaction_mode="input", + should_trigger_compaction=should_trigger_compaction, + ) + + model = ScriptedModel( + [ + { + "output": [ + get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff-1") + ], + "response_id": "resp-handoff", + }, + {"output": [get_text_message("done")], "response_id": "resp-delegate"}, + ] + ) + delegate = Agent(name="delegate", model=model) + triage = Agent(name="triage", model=model, handoffs=[delegate]) + + failed_result = Runner.run_streamed( + triage, "hello", session=session, run_config=RunConfig(tracing_disabled=True) + ) + with pytest.raises(RuntimeError) as error: + async for _ in failed_result.stream_events(): + pass + assert error.value is backend.error + assert hook_calls == [] + assert compact_calls == [] + state = failed_result.to_state() + assert state._pending_session_write is not None + assert state._pending_session_write.get("response_id") == "resp-handoff" + assert state._pending_session_write.get("has_local_tool_outputs") is True + + result = await _run_session_resume(triage, state, session, False) + assert result.final_output == "done" + # The handoff's own response_id must have been evaluated by the decision hook (and + # deferred), not skipped -- and, because force-compaction short-circuits the hook, it must + # be the only response_id the hook ever saw. + assert hook_calls == ["resp-handoff"] + # The deferred decision must actually have been forced through on the delegate's own save, + # i.e. the compact API was invoked at all -- not just checked and declined. + assert len(compact_calls) == 1 + + +@pytest.mark.asyncio +async def test_fresh_streamed_handoff_retains_checkpoint_when_post_write_compaction_fails() -> None: + """If the post-write compaction decision raises after a checkpointed handoff batch's append + has already settled, the checkpoint (``_pending_session_write``) must survive so a later + retry can redo just the compaction step -- clearing it before the fallible compaction call + would silently and permanently lose the requested deferred/forced compaction with no way to + recover it. See .agents/references/session-persistence.md. + """ + hook_calls: list[str | None] = [] + compaction_error = RuntimeError("compaction decision hook exploded") + should_fail = True + + def should_trigger_compaction(context: dict[str, Any]) -> bool: + hook_calls.append(context["response_id"]) + if context["response_id"] == "resp-handoff" and should_fail: + raise compaction_error + return context["response_id"] == "resp-handoff" + + compact_calls: list[list[TResponseInputItem]] = [] + + async def compact(**kwargs: Any) -> SimpleNamespace: + items = copy.deepcopy(kwargs["input"]) + compact_calls.append(items) + return SimpleNamespace(output=items, usage=None) + + backend = _FailSecondAddItemsSession() + session = OpenAIResponsesCompactionSession( + "compaction-handoff-failure-test", + underlying_session=backend, + client=cast(Any, SimpleNamespace(responses=SimpleNamespace(compact=compact))), + compaction_mode="input", + should_trigger_compaction=should_trigger_compaction, + ) + + model = ScriptedModel( + [ + { + "output": [ + get_function_tool_call("transfer_to_delegate", "{}", call_id="handoff-1") + ], + "response_id": "resp-handoff", + }, + {"output": [get_text_message("done")], "response_id": "resp-delegate"}, + ] + ) + delegate = Agent(name="delegate", model=model) + triage = Agent(name="triage", model=model, handoffs=[delegate]) + + failed_result = Runner.run_streamed( + triage, "hello", session=session, run_config=RunConfig(tracing_disabled=True) + ) + with pytest.raises(RuntimeError) as append_error: + async for _ in failed_result.stream_events(): + pass + assert append_error.value is backend.error + state = failed_result.to_state() + assert state._pending_session_write is not None + + # Resume: the append itself now succeeds (the backend's failure was one-shot), but the + # compaction decision hook raises for the handoff's own response_id. + with pytest.raises(RuntimeError) as compaction_error_info: + await _run_session_resume(triage, state, session, False) + assert compaction_error_info.value is compaction_error + # The checkpoint must still be present so a later retry can redo compaction alone, instead + # of the handoff's requested compaction being silently and permanently lost. + assert state._pending_session_write is not None + assert state._pending_session_write.get("response_id") == "resp-handoff" + + # Retry: the hook no longer fails. The append must not be repeated (no duplicate items in + # session history), but compaction must actually run this time. + should_fail = False + hook_calls.clear() + result = await _run_session_resume(triage, state, session, False) + assert result.final_output == "done" + assert hook_calls == ["resp-handoff"] + assert len(compact_calls) == 1 + stored = await session.get_items() + handoff_pair = [ + str(item.get("type")) + for item in stored + if isinstance(item, dict) and item.get("call_id") == "handoff-1" + ] + assert handoff_pair == ["function_call", "function_call_output"]