From 1ed6c3ee4f7eae56da09a6264cef9a763b77eb98 Mon Sep 17 00:00:00 2001 From: ayaangazali Date: Mon, 24 Aug 2026 23:54:39 -0700 Subject: [PATCH 1/2] fix(core): keep completed tool guardrail results when a resumed append fails On a resumed turn that leaves another approval pending, the run publishes the turn's tool guardrail results only inside the NextStepInterruption branch, which sits after the resumed Session append. When that append raises, the approved tool has already executed and both guardrails have already completed, but their results never reach RunState, so they are absent from the checkpoint, from its serialization, and from every later result. Publish them alongside the other resumed state updates, before the append, in both the streaming and non-streaming paths. --- src/agents/run.py | 11 ++++ src/agents/run_internal/run_loop.py | 11 ++++ tests/test_run_impl_resume_paths.py | 78 +++++++++++++++++++++++++++++ 3 files changed, 100 insertions(+) diff --git a/src/agents/run.py b/src/agents/run.py index 629b5ff23f..a5a4a8eb69 100644 --- a/src/agents/run.py +++ b/src/agents/run.py @@ -1135,6 +1135,17 @@ def _mark_response_hooks_started() -> None: generated_items=generated_items, session_items=session_items, ) + # Publish this turn's tool guardrail results before the + # resumed Session append, which can raise and skip the + # interruption branch below. The tool already ran. + run_state._tool_input_guardrail_results = [ + *tool_input_guardrail_results, + *turn_result.tool_input_guardrail_results, + ] + run_state._tool_output_guardrail_results = [ + *tool_output_guardrail_results, + *turn_result.tool_output_guardrail_results, + ] if ( session_persistence_enabled diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py index 7c22fee317..98d6cf6dd1 100644 --- a/src/agents/run_internal/run_loop.py +++ b/src/agents/run_internal/run_loop.py @@ -1354,6 +1354,17 @@ async def _save_max_turns_items( generated_items=generated_items, session_items=streamed_result.new_items, ) + # Publish this turn's tool guardrail results before the resumed + # Session append, which can raise and skip the interruption + # branch below. The tool already ran. + run_state._tool_input_guardrail_results = [ + *accepted_tool_input_guardrail_results, + *turn_result.tool_input_guardrail_results, + ] + run_state._tool_output_guardrail_results = [ + *accepted_tool_output_guardrail_results, + *turn_result.tool_output_guardrail_results, + ] run_state._current_turn_persisted_item_count = ( streamed_result._current_turn_persisted_item_count ) diff --git a/tests/test_run_impl_resume_paths.py b/tests/test_run_impl_resume_paths.py index 9a4d88f061..4318ea3210 100644 --- a/tests/test_run_impl_resume_paths.py +++ b/tests/test_run_impl_resume_paths.py @@ -38,6 +38,11 @@ ) from agents.run_state import RunState from agents.testing import ScriptedModel +from agents.tool_guardrails import ( + ToolGuardrailFunctionOutput, + tool_input_guardrail, + tool_output_guardrail, +) from agents.usage import Usage from tests.test_responses import get_function_tool_call, get_text_message from tests.utils.hitl import ( @@ -984,3 +989,76 @@ async def needs_ok(text: str) -> str: isinstance(item, ToolCallOutputItem) and item.output == "one" for item in result.new_step_items ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +@pytest.mark.parametrize("round_trip", [False, True], ids=["live", "json"]) +async def test_failed_resumed_append_keeps_completed_tool_guardrail_results( + streamed: bool, round_trip: bool +) -> None: + """A failed resumed append must not discard guardrail results the tool already produced.""" + + @tool_input_guardrail + def allow_input(_data: Any) -> ToolGuardrailFunctionOutput: + return ToolGuardrailFunctionOutput.allow(output_info="input-ok") + + @tool_output_guardrail + def allow_output(_data: Any) -> ToolGuardrailFunctionOutput: + return ToolGuardrailFunctionOutput.allow(output_info="output-ok") + + @tool( + needs_approval=True, + tool_input_guardrails=[allow_input], + tool_output_guardrails=[allow_output], + ) + async def charge(amount: int) -> str: + return "receipt-7" + + @tool(needs_approval=True) + async def notify() -> str: + raise AssertionError("the unresolved approval must not execute") + + model = ScriptedModel( + [ + [ + get_function_tool_call("charge", '{"amount":7}', call_id="charge-1"), + get_function_tool_call("notify", "{}", call_id="notify-1"), + ], + [get_text_message("done")], + ] + ) + agent = Agent(name="payment", model=model, tools=[charge, notify]) + session = _FailingResumeSession() + + paused = await _run_session_resume(agent, "charge 7 and notify", session, streamed) + state = paused.to_state() + state.approve( + next(item for item in state.get_interruptions() if item.raw_item.call_id == "charge-1") + ) + + session.failure = "before" + with pytest.raises(RuntimeError) as error: + await _run_session_resume(agent, state, session, streamed) + assert error.value is session.error + + # The tool ran and both guardrails completed, so the checkpoint must retain their results. + assert len(state._tool_input_guardrail_results) == 1 + assert len(state._tool_output_guardrail_results) == 1 + serialized = state.to_json() + assert len(serialized["tool_input_guardrail_results"]) == 1 + assert len(serialized["tool_output_guardrail_results"]) == 1 + + if round_trip: + state = await RunState.from_json(agent, serialized) + + pending = await _run_session_resume(agent, state, session, streamed) + pending_state = pending.to_state() + remaining = pending_state.get_interruptions() + assert [item.raw_item.call_id for item in remaining] == ["notify-1"] + + pending_state.reject(remaining[0], rejection_message="declined") + result = await _run_session_resume(agent, pending_state, session, streamed) + assert result.final_output == "done" + assert len(result.tool_input_guardrail_results) == 1 + assert len(result.tool_output_guardrail_results) == 1 From 4e0309e53645b9a9886de4da8c6132643e80a3a3 Mon Sep 17 00:00:00 2001 From: ayaangazali Date: Tue, 25 Aug 2026 10:05:41 -0700 Subject: [PATCH 2/2] fix(core): limit resumed guardrail publication to the interruption transition The previous commit published the resumed turn's guardrail results on every resumed turn. On a sole pending approval the step is NextStepRunAgain, and the streamed recovery path rebuilds its checkpoint from RunResultStreaming, so the source RunState and the detached checkpoint disagreed after a failed append. Guard both publications on NextStepInterruption. Run-again accounting is left exactly as it was, and the renewed-interruption transition still matches the clean path. --- src/agents/run.py | 23 ++++++++++++----------- src/agents/run_internal/run_loop.py | 23 ++++++++++++----------- 2 files changed, 24 insertions(+), 22 deletions(-) diff --git a/src/agents/run.py b/src/agents/run.py index a5a4a8eb69..93bb68fda7 100644 --- a/src/agents/run.py +++ b/src/agents/run.py @@ -1135,17 +1135,18 @@ def _mark_response_hooks_started() -> None: generated_items=generated_items, session_items=session_items, ) - # Publish this turn's tool guardrail results before the - # resumed Session append, which can raise and skip the - # interruption branch below. The tool already ran. - run_state._tool_input_guardrail_results = [ - *tool_input_guardrail_results, - *turn_result.tool_input_guardrail_results, - ] - run_state._tool_output_guardrail_results = [ - *tool_output_guardrail_results, - *turn_result.tool_output_guardrail_results, - ] + if isinstance(turn_result.next_step, NextStepInterruption): + # Publish this turn's tool guardrail results before the + # resumed Session append, which can raise and skip the + # interruption branch below. The tool already ran. + run_state._tool_input_guardrail_results = [ + *tool_input_guardrail_results, + *turn_result.tool_input_guardrail_results, + ] + run_state._tool_output_guardrail_results = [ + *tool_output_guardrail_results, + *turn_result.tool_output_guardrail_results, + ] if ( session_persistence_enabled diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py index 98d6cf6dd1..3ce2e1a301 100644 --- a/src/agents/run_internal/run_loop.py +++ b/src/agents/run_internal/run_loop.py @@ -1354,17 +1354,18 @@ async def _save_max_turns_items( generated_items=generated_items, session_items=streamed_result.new_items, ) - # Publish this turn's tool guardrail results before the resumed - # Session append, which can raise and skip the interruption - # branch below. The tool already ran. - run_state._tool_input_guardrail_results = [ - *accepted_tool_input_guardrail_results, - *turn_result.tool_input_guardrail_results, - ] - run_state._tool_output_guardrail_results = [ - *accepted_tool_output_guardrail_results, - *turn_result.tool_output_guardrail_results, - ] + if isinstance(turn_result.next_step, NextStepInterruption): + # Publish this turn's tool guardrail results before the resumed + # Session append, which can raise and skip the interruption + # branch below. The tool already ran. + run_state._tool_input_guardrail_results = [ + *accepted_tool_input_guardrail_results, + *turn_result.tool_input_guardrail_results, + ] + run_state._tool_output_guardrail_results = [ + *accepted_tool_output_guardrail_results, + *turn_result.tool_output_guardrail_results, + ] run_state._current_turn_persisted_item_count = ( streamed_result._current_turn_persisted_item_count )