diff --git a/src/agents/memory/openai_responses_compaction_session.py b/src/agents/memory/openai_responses_compaction_session.py index d118f110cc..2717c3cde5 100644 --- a/src/agents/memory/openai_responses_compaction_session.py +++ b/src/agents/memory/openai_responses_compaction_session.py @@ -431,7 +431,6 @@ async def _run_compaction_locked( if is_automatic and snapshot is None and self.max_rollback_items is not None: await self._get_all_underlying_session_items() - self._deferred_response_id = None logger.debug( "compact: start for %s using %s (mode=%s)", self._response_id, @@ -490,6 +489,12 @@ async def _run_compaction_locked( ) self._session_items = None if read_items is not None else 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/agent_runner_helpers.py b/src/agents/run_internal/agent_runner_helpers.py index 103d2af444..6c7e025765 100644 --- a/src/agents/run_internal/agent_runner_helpers.py +++ b/src/agents/run_internal/agent_runner_helpers.py @@ -505,18 +505,16 @@ def build_interruption_result( def reject_unrecoverable_terminal_state(run_state: RunState | None) -> None: - """Fail closed when a previous run already produced a final output that cannot be reproduced. + """Fail closed when a previous run reached a boundary that cannot safely be resumed. - The marker is set once that output, its guardrails, and its terminal hooks have completed, - and is cleared only once the turn is fully persisted. In between, the run owns a result no - resume can settle, so resuming would repeat the model call and the lifecycle hooks for an - output the caller already received. Raised before any Session, sandbox, model, tool, - guardrail, or hook work so the rejection has no side effects of its own. + This includes failed terminal persistence and a speculative handoff rejected by an input + guardrail. Resuming would repeat completed work or bypass the failed input check. Reject + before any Session, sandbox, model, tool, guardrail, or hook work. """ if run_state is not None and run_state._terminal_unrecoverable: raise UserError( - "This RunState already produced a final output whose Session write did not " - "complete, so it cannot be resumed. Start a new run instead." + "This RunState ended at an unrecoverable boundary and cannot be resumed. " + "Start a new run instead." ) diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py index e0a4011155..da527bb5f9 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 @@ -422,6 +423,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, @@ -437,6 +439,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 = ( @@ -1165,6 +1168,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( @@ -1886,23 +1895,42 @@ 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, - ) + # A failed input check makes this completed speculative turn non-resumable: + # replaying it would skip the starting agent's guardrails. Keep its executed + # call/output records coherent instead of partially rolling back history. + try: + if streamed_result._input_guardrails_task is not None: + await streamed_result._input_guardrails_task + for guardrail_result in streamed_result.input_guardrail_results: + if guardrail_result.output.tripwire_triggered: + raise InputGuardrailTripwireTriggered(guardrail_result) + except BaseException: + if run_state is not None: + run_state._terminal_unrecoverable = True + raise 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 c26c708648..22d251fa3c 100644 --- a/src/agents/run_internal/session_persistence.py +++ b/src/agents/run_internal/session_persistence.py @@ -631,6 +631,71 @@ 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 + if wrapper is not None: + wrapper._session_compaction_is_automatic = True # type: ignore[attr-defined] + try: + await _call_session_method( + session.run_compaction, + compaction_args, + wrapper=wrapper, + ) + finally: + if wrapper is not None: + wrapper._session_compaction_is_automatic = False # type: ignore[attr-defined] + + async def save_result_to_session( session: Session | None, original_input: str | list[TResponseInputItem], @@ -738,6 +803,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") @@ -748,7 +817,13 @@ 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, } + # 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, @@ -760,53 +835,14 @@ 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 - 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 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, ) - 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 - if compaction_wrapper is not None: - compaction_wrapper._session_compaction_is_automatic = True # type: ignore[attr-defined] - try: - await _call_session_method( - session.run_compaction, - compaction_args, - wrapper=compaction_wrapper, - ) - finally: - if compaction_wrapper is not None: - compaction_wrapper._session_compaction_is_automatic = False # type: ignore[attr-defined] return saved_run_items_count @@ -886,10 +922,10 @@ def digests(items: Sequence[TResponseInputItem]) -> list[str]: append = True else: expected = before + digests(pending["items"]) - committed_generation: int | None = None + observed_generation: int | None = None get_with_generation = getattr(session, "_get_items_with_generation", None) if wrapper is not None and callable(get_with_generation): - tail, committed_generation = await _call_session_method( + tail, observed_generation = await _call_session_method( get_with_generation, lambda: _session_get_items(session, limit=len(expected), wrapper=wrapper), ) @@ -904,12 +940,26 @@ def digests(items: Sequence[TResponseInputItem]) -> list[str]: "Repair the original Session before resuming; do not rerun the completed tool." ) append = unchanged - if committed and committed_generation is not None and wrapper is not None: - wrapper._session_compaction_generation = committed_generation # type: ignore[attr-defined] + # The original append can advance the wrapper generation even when it fails + # atomically. Reconciled unchanged history is also safe to append against; + # subsequent mutations still revoke ownership through the normal generation check. + if observed_generation is not None and wrapper is not None: + wrapper._session_compaction_generation = observed_generation # type: ignore[attr-defined] if append: # 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=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 982f6564fd..bf38f49cc7 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, @@ -175,6 +175,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( @@ -233,7 +241,8 @@ def _default_run_state_validation_error( "1.18": ( "Binds restored local MCP calls to their configured server and original tool name, " "preserves independent apply_patch approval scopes, and binds function-tool approval " - "decisions to their owning agent." + "decisions to their owning agent, and retains compaction metadata for pending " + "Session writes." ), } SUPPORTED_SCHEMA_VERSIONS = frozenset(SCHEMA_VERSION_SUMMARIES) @@ -4499,11 +4508,20 @@ 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"} + # Released schema 1.17 wrote exactly four keys. Compaction metadata belongs to 1.18; + # older checkpoints remain readable without claiming the newer recovery behavior. + optional_pending_write_keys = ( + {"response_id", "store", "has_local_tool_outputs"} + if (schema_major, schema_minor) >= (1, 18) + 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 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"] @@ -4517,6 +4535,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 be06ab8361..0104a98a2e 100644 --- a/tests/memory/test_openai_responses_compaction_session.py +++ b/tests/memory/test_openai_responses_compaction_session.py @@ -1674,6 +1674,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 05dacb0145..0db8f68abd 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, GuardrailFunctionOutput, Runner, function_tool, handoff, output_guardrail +from agents import Agent, AgentUpdatedStreamEvent, Runner, function_tool, handoff, output_guardrail 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.exceptions import InputGuardrailTripwireTriggered, UserError +from agents.guardrail import GuardrailFunctionOutput, input_guardrail from agents.items import ( MessageOutputItem, ModelResponse, @@ -89,6 +90,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") @@ -499,23 +533,46 @@ 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", "compaction-under-1.17"]) async def test_pending_session_write_rejects_invalid_serialized_checkpoint(invalid: str) -> None: agent, _, session, state, _ = await _approved_session_state(False) session.failure = "before" with pytest.raises(RuntimeError): await _run_session_resume(agent, state, session, False) payload = state.to_json() - if invalid == "old-schema": + if invalid in {"old-schema", "compaction-under-1.17"}: for entry in payload["context"].pop("function_tool_approvals", []): payload["context"]["approvals"][entry["tool_key"]] = entry["decision"] + if invalid == "old-schema": payload["$schemaVersion"] = "1.16" + elif invalid == "compaction-under-1.17": + payload["$schemaVersion"] = "1.17" else: payload["pending_session_write"]["items"] = "not an item batch" with pytest.raises(UserError, match="pending Session write is invalid"): await RunState.from_json(agent, payload) +@pytest.mark.asyncio +async def test_legacy_pending_session_write_still_resumes_without_compaction_metadata() -> None: + agent, model, session, state, effects = await _approved_session_state(False) + session.failure = "before" + with pytest.raises(RuntimeError, match="session append failed"): + await _run_session_resume(agent, state, session, False) + payload = state.to_json() + payload["$schemaVersion"] = "1.17" + for entry in payload["context"].pop("function_tool_approvals", []): + payload["context"]["approvals"][entry["tool_key"]] = entry["decision"] + for key in ("response_id", "store", "has_local_tool_outputs"): + payload["pending_session_write"].pop(key) + 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 len(model.calls) == 2 + 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) @@ -1416,6 +1473,345 @@ async def test_resumed_handoff_session_append_is_recovered_before_next_model( 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 +): + """A yielding Session failure still delivers the completed handoff's agent update.""" + 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 +@pytest.mark.parametrize("raises_error", [False, True], ids=["tripwire", "guardrail-error"]) +@pytest.mark.parametrize("retry_streamed", [False, True]) +async def test_fresh_streamed_handoff_failed_guardrail_state_cannot_resume( + raises_error: bool, + retry_streamed: bool, +) -> None: + handoff_observed = asyncio.Event() + guardrail_error = RuntimeError("guardrail backend failed") + + @input_guardrail(run_in_parallel=True) + async def pending_guardrail( + ctx: RunContextWrapper[Any], + agent: Agent[Any], + input: str | list[TResponseInputItem], + ) -> GuardrailFunctionOutput: + await asyncio.wait_for(handoff_observed.wait(), timeout=5) + if raises_error: + raise guardrail_error + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) + + 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=[pending_guardrail] + ) + session = SimpleListSession() + streamed_result = Runner.run_streamed( + triage, "hello", session=session, run_config=RunConfig(tracing_disabled=True) + ) + events: list[Any] = [] + expected_error = RuntimeError if raises_error else InputGuardrailTripwireTriggered + with pytest.raises(expected_error): + async for event in streamed_result.stream_events(): + events.append(event) + if getattr(event, "name", None) == "handoff_occured": + handoff_observed.set() + await asyncio.sleep(0) + assert handoff_observed.is_set() + assert streamed_result.current_agent is triage + assert not any( + isinstance(event, AgentUpdatedStreamEvent) and event.new_agent is delegate + for event in events + ) + state = streamed_result.to_state() + assert state.to_json()["terminal_unrecoverable"] is True + restored = await RunState.from_json(triage, state.to_json()) + assert restored._current_agent is triage + for checkpoint in (state, restored): + with pytest.raises(UserError, match="cannot be resumed"): + await _run_session_resume(triage, checkpoint, session, retry_streamed) + assert len(model.calls) == 1 + assert _call_pair(await session.get_items(), "handoff-1") == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("round_trip", [False, True], ids=["live", "json"]) +async def test_fresh_streamed_handoff_replays_deferred_compaction_after_resume( + round_trip: bool, +) -> 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 + + if round_trip: + payload = state.to_json() + assert payload["$schemaVersion"] == "1.18" + state = await RunState.from_json(triage, payload) + + 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"] + + class _TerminalLifecycleHooks(RunHooks[Any]): """Count the agent lifecycle hooks an application can attach its own effects to."""