From aa7c6b0a05bb3bb785f9e182aeedef1453692878 Mon Sep 17 00:00:00 2001 From: Henry Su Date: Tue, 28 Jul 2026 10:20:33 -0500 Subject: [PATCH 1/3] fix: defer session save until after output guardrails Non-streamed runs were persisting final-turn assistant items before output guardrails ran, so a tripwire left rejected output in the session while streamed runs correctly deferred persistence. --- src/agents/run.py | 9 +- tests/test_agent_runner.py | 132 ++++++++++++++++++++++++++++ tests/test_agent_runner_streamed.py | 31 +++++++ 3 files changed, 170 insertions(+), 2 deletions(-) diff --git a/src/agents/run.py b/src/agents/run.py index d0e3f9ee56..a279f7256d 100644 --- a/src/agents/run.py +++ b/src/agents/run.py @@ -1414,7 +1414,12 @@ def _finalize_result(result: RunResult) -> RunResult: ) ): items_to_save_turn.append(item) - if items_to_save_turn: + # Defer final-output persistence until after output guardrails + # succeed so a tripwire does not leave rejected assistant output + # in the session (matches the streamed path). + if items_to_save_turn and not isinstance( + turn_result.next_step, NextStepFinalOutput + ): logger.debug( "Persisting turn items (types=%s)", [item.type for item in items_to_save_turn], @@ -1489,7 +1494,7 @@ def _finalize_result(result: RunResult) -> RunResult: run_state=run_state, session_persistence_enabled=session_persistence_enabled, input_guardrail_results=input_guardrail_results, - items=session_items_for_turn(turn_result), + items=items_to_save_turn, response_id=turn_result.model_response.response_id, store=store_setting, ) diff --git a/tests/test_agent_runner.py b/tests/test_agent_runner.py index ab6485045d..e3baaa2c5c 100644 --- a/tests/test_agent_runner.py +++ b/tests/test_agent_runner.py @@ -3382,6 +3382,9 @@ async def noop_initialize_computer_tools( monkeypatch.setattr( "agents.run_internal.session_persistence.save_result_to_session", save_wrapper ) + monkeypatch.setattr( + "agents.run_internal.agent_runner_helpers.save_result_to_session", save_wrapper + ) monkeypatch.setattr("agents.run.run_single_turn", fake_run_single_turn) monkeypatch.setattr("agents.run_internal.run_loop.run_single_turn", fake_run_single_turn) monkeypatch.setattr("agents.run.run_output_guardrails", fake_run_output_guardrails) @@ -3434,6 +3437,135 @@ def guardrail_function( await Runner.run(agent, input="user_message") +@pytest.mark.asyncio +async def test_output_guardrail_tripwire_does_not_save_assistant_message_to_session(): + def guardrail_function( + context: RunContextWrapper[Any], agent: Agent[Any], agent_output: Any + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput( + output_info=None, + tripwire_triggered=True, + ) + + session = SimpleListSession() + model = FakeModel() + model.set_next_output([get_text_message("should_not_be_saved")]) + agent = Agent( + name="test", + model=model, + output_guardrails=[OutputGuardrail(guardrail_function=guardrail_function)], + ) + + with pytest.raises(OutputGuardrailTripwireTriggered): + await Runner.run(agent, input="user_message", session=session) + + items = await session.get_items() + assert len(items) == 1 + first_item = cast(dict[str, Any], items[0]) + assert first_item["role"] == "user" + assert first_item["content"] == "user_message" + + +def test_output_guardrail_tripwire_does_not_save_assistant_message_to_session_sync(): + def guardrail_function( + context: RunContextWrapper[Any], agent: Agent[Any], agent_output: Any + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput( + output_info=None, + tripwire_triggered=True, + ) + + session = SimpleListSession() + model = FakeModel() + model.set_next_output([get_text_message("should_not_be_saved")]) + agent = Agent( + name="test", + model=model, + output_guardrails=[OutputGuardrail(guardrail_function=guardrail_function)], + ) + + with pytest.raises(OutputGuardrailTripwireTriggered): + Runner.run_sync(agent, input="user_message", session=session) + + items = asyncio.run(session.get_items()) + assert len(items) == 1 + first_item = cast(dict[str, Any], items[0]) + assert first_item["role"] == "user" + + +@pytest.mark.asyncio +async def test_output_guardrail_success_still_saves_assistant_message_to_session(): + def guardrail_function( + context: RunContextWrapper[Any], agent: Agent[Any], agent_output: Any + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput( + output_info=None, + tripwire_triggered=False, + ) + + session = SimpleListSession() + model = FakeModel() + model.set_next_output([get_text_message("safe_response")]) + agent = Agent( + name="test", + model=model, + output_guardrails=[OutputGuardrail(guardrail_function=guardrail_function)], + ) + + result = await Runner.run(agent, input="user_message", session=session) + assert result.final_output == "safe_response" + + items = await session.get_items() + assert len(items) == 2 + assert cast(dict[str, Any], items[0])["role"] == "user" + assert cast(dict[str, Any], items[1])["role"] == "assistant" + + +@pytest.mark.asyncio +async def test_output_guardrail_tripwire_keeps_prior_tool_turn_in_session(): + """Earlier completed turns remain persisted when a later final turn is rejected.""" + + def guardrail_function( + context: RunContextWrapper[Any], agent: Agent[Any], agent_output: Any + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput( + output_info=None, + tripwire_triggered=True, + ) + + @function_tool + def foo(a: str) -> str: + return f"result:{a}" + + session = SimpleListSession() + model = FakeModel() + model.add_multiple_turn_outputs( + [ + [get_function_tool_call("foo", json.dumps({"a": "b"}))], + [get_text_message("should_not_be_saved")], + ] + ) + agent = Agent( + name="test", + model=model, + tools=[foo], + output_guardrails=[OutputGuardrail(guardrail_function=guardrail_function)], + ) + + with pytest.raises(OutputGuardrailTripwireTriggered): + await Runner.run(agent, input="user_message", session=session) + + items = await session.get_items() + assert any(cast(dict[str, Any], item).get("role") == "user" for item in items) + assert any(cast(dict[str, Any], item).get("type") == "function_call" for item in items) + assert any(cast(dict[str, Any], item).get("type") == "function_call_output" for item in items) + assert not any( + cast(dict[str, Any], item).get("role") == "assistant" + and cast(dict[str, Any], item).get("type") == "message" + for item in items + ) + + @pytest.mark.asyncio async def test_input_guardrail_no_tripwire_continues_execution(): """Test input guardrail that doesn't trigger tripwire continues execution.""" diff --git a/tests/test_agent_runner_streamed.py b/tests/test_agent_runner_streamed.py index 2ba360d2ef..9d15e68a3e 100644 --- a/tests/test_agent_runner_streamed.py +++ b/tests/test_agent_runner_streamed.py @@ -1542,6 +1542,37 @@ def guardrail_function( pass +@pytest.mark.asyncio +async def test_output_guardrail_streamed_does_not_save_assistant_message_to_session(): + def guardrail_function( + context: RunContextWrapper[Any], agent: Agent[Any], agent_output: Any + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput( + output_info=None, + tripwire_triggered=True, + ) + + session = SimpleListSession() + model = FakeModel() + model.set_next_output([get_text_message("should_not_be_saved")]) + agent = Agent( + name="test", + model=model, + output_guardrails=[OutputGuardrail(guardrail_function=guardrail_function)], + ) + + with pytest.raises(OutputGuardrailTripwireTriggered): + result = Runner.run_streamed(agent, input="user_message", session=session) + async for _ in result.stream_events(): + pass + + items = await session.get_items() + assert len(items) == 1 + first_item = cast(dict[str, Any], items[0]) + assert first_item["role"] == "user" + assert first_item["content"] == "user_message" + + @pytest.mark.asyncio async def test_output_guardrail_tripwire_raises_from_run_loop_task_before_stream_consumption(): def guardrail_function( From 564e67d5be25ba8f85a7c8a5de36f2535727c70b Mon Sep 17 00:00:00 2001 From: Henry Su Date: Tue, 28 Jul 2026 18:26:11 -0500 Subject: [PATCH 2/3] fix: persist resumed finals after output guardrails When _current_turn_persisted_item_count is already positive, route the post-guardrail FinalOutput save through save_resumed_turn_items so accepted assistant items are appended instead of being skipped. --- src/agents/run.py | 49 +++++++++++++++++++++----- tests/test_agent_runner.py | 70 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 110 insertions(+), 9 deletions(-) diff --git a/src/agents/run.py b/src/agents/run.py index a279f7256d..199f46e4c2 100644 --- a/src/agents/run.py +++ b/src/agents/run.py @@ -1489,15 +1489,46 @@ def _finalize_result(result: RunResult) -> RunResult: result._current_turn_persisted_item_count = ( run_state._current_turn_persisted_item_count ) - await save_turn_items_if_needed( - session=session, - run_state=run_state, - session_persistence_enabled=session_persistence_enabled, - input_guardrail_results=input_guardrail_results, - items=items_to_save_turn, - response_id=turn_result.model_response.response_id, - store=store_setting, - ) + if ( + session_persistence_enabled + and items_to_save_turn + and not input_guardrails_triggered(input_guardrail_results) + ): + # When earlier items from this turn were already persisted + # (resume / soft-cancel), save_turn_items_if_needed would + # no-op. Use the resumed-turn path so accepted final items + # are still appended after output guardrails succeed. + if ( + run_state is not None + and run_state._current_turn_persisted_item_count > 0 + ): + run_state._current_turn_persisted_item_count = ( + await save_resumed_turn_items( + session=session, + items=items_to_save_turn, + 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, + ) + ) + result._current_turn_persisted_item_count = ( + run_state._current_turn_persisted_item_count + ) + else: + await save_turn_items_if_needed( + session=session, + run_state=run_state, + session_persistence_enabled=session_persistence_enabled, + input_guardrail_results=input_guardrail_results, + items=items_to_save_turn, + response_id=turn_result.model_response.response_id, + store=store_setting, + ) result._original_input = copy_input_items(original_input) return _finalize_result(result) elif isinstance(turn_result.next_step, NextStepInterruption): diff --git a/tests/test_agent_runner.py b/tests/test_agent_runner.py index e3baaa2c5c..0c2d2144e3 100644 --- a/tests/test_agent_runner.py +++ b/tests/test_agent_runner.py @@ -3566,6 +3566,76 @@ def foo(a: str) -> str: ) +@pytest.mark.asyncio +async def test_resumed_final_output_persists_after_passing_output_guardrail(): + """Resumed runs with a positive persisted count still save accepted finals.""" + + def guardrail_function( + context: RunContextWrapper[Any], agent: Agent[Any], agent_output: Any + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput( + output_info=None, + tripwire_triggered=False, + ) + + @function_tool + def foo(a: str) -> str: + return f"result:{a}" + + session = SimpleListSession() + model = FakeModel() + model.set_next_output([get_function_tool_call("foo", json.dumps({"a": "b"}))]) + agent = Agent( + name="test", + model=model, + tools=[foo], + output_guardrails=[OutputGuardrail(guardrail_function=guardrail_function)], + ) + + streamed = Runner.run_streamed(agent, input="user_message", session=session) + async for event in streamed.stream_events(): + if event.type == "run_item_stream_event" and event.name == "tool_output": + streamed.cancel(mode="after_turn") + + items_before_resume = await session.get_items() + assert any( + cast(dict[str, Any], item).get("type") == "function_call" for item in items_before_resume + ) + assert any( + cast(dict[str, Any], item).get("type") == "function_call_output" + for item in items_before_resume + ) + assert not any( + cast(dict[str, Any], item).get("role") == "assistant" + and cast(dict[str, Any], item).get("type") == "message" + for item in items_before_resume + ) + + # Soft-cancel saves the completed tool turn; recreate a resume boundary with a + # positive persisted count so deferred FinalOutput saves must use the + # resumed-turn path instead of save_turn_items_if_needed. + state = streamed.to_state() + state._current_turn_persisted_item_count = 2 + + model.set_next_output([get_text_message("accepted_final")]) + resumed = await Runner.run(agent, state, session=session) + assert resumed.final_output == "accepted_final" + + items_after_resume = await session.get_items() + assert items_after_resume[: len(items_before_resume)] == items_before_resume + + assistant_messages = [ + item + for item in items_after_resume + if cast(dict[str, Any], item).get("role") == "assistant" + and cast(dict[str, Any], item).get("type") == "message" + ] + assert len(assistant_messages) == 1 + content = cast(dict[str, Any], assistant_messages[0]).get("content") + assert isinstance(content, list) + assert any(isinstance(part, dict) and part.get("text") == "accepted_final" for part in content) + + @pytest.mark.asyncio async def test_input_guardrail_no_tripwire_continues_execution(): """Test input guardrail that doesn't trigger tripwire continues execution.""" From 449c6a7e3a0bdeeabdd4fdfccd097e0f2c43ee26 Mon Sep 17 00:00:00 2001 From: Henry Su Date: Mon, 3 Aug 2026 02:36:36 -0500 Subject: [PATCH 3/3] fix: keep executed tool items when a final turn is rejected A final-output turn produced by tool_use_behavior stop_on_first_tool, stop_at_tool_names or a custom callable carries the tool call and its output, so deferring the whole turn discarded the record of a side effect that had already run. Defer only the model's own output for that turn and persist everything else immediately, in the non-streamed, streamed and resumed paths. The streamed path had the same gap before this change, so matching it is no longer a justification for dropping tool items. --- src/agents/run.py | 60 ++++++++--- src/agents/run_internal/run_loop.py | 18 +++- tests/test_agent_runner.py | 150 ++++++++++++++++++++++++++++ 3 files changed, 213 insertions(+), 15 deletions(-) diff --git a/src/agents/run.py b/src/agents/run.py index 199f46e4c2..c5477b5550 100644 --- a/src/agents/run.py +++ b/src/agents/run.py @@ -446,6 +446,12 @@ def run_streamed( ) +# Only the model's own output for the final turn waits for the output-guardrail verdict. +# This is an allow-list on purpose: a future item type that records something the run +# already did defaults to "persist" rather than silently inheriting "discard". +_DEFERRED_FINAL_OUTPUT_ITEM_TYPES = frozenset({"message_output_item", "reasoning_item"}) + + class AgentRunner: """ WARNING: this class is experimental and not part of the public API @@ -919,15 +925,27 @@ def _finalize_result(result: RunResult) -> RunResult: session_items=session_items, ) + # A resumed turn that ends the run carries the same obligation as a + # fresh one: side-effect records persist now, model output waits for + # the guardrail verdict below. + resumed_items_now = ( + [ + item + for item in turn_session_items + if item.type not in _DEFERRED_FINAL_OUTPUT_ITEM_TYPES + ] + if isinstance(turn_result.next_step, NextStepFinalOutput) + else turn_session_items + ) if ( session_persistence_enabled - and turn_session_items + and resumed_items_now and run_state is not None ): run_state._current_turn_persisted_item_count = ( await save_resumed_turn_items( session=session, - items=turn_session_items, + items=resumed_items_now, persisted_count=( run_state._current_turn_persisted_item_count ), @@ -1414,21 +1432,35 @@ def _finalize_result(result: RunResult) -> RunResult: ) ): items_to_save_turn.append(item) - # Defer final-output persistence until after output guardrails - # succeed so a tripwire does not leave rejected assistant output - # in the session (matches the streamed path). - if items_to_save_turn and not isinstance( - turn_result.next_step, NextStepFinalOutput - ): + # Defer only the model's own output until after output guardrails + # succeed, so a tripwire does not leave rejected assistant output in + # the session. Tool calls and their outputs record side effects that + # already happened, so they persist now even on a final-output turn + # (tool_use_behavior="stop_on_first_tool" ends the run on such a turn). + if isinstance(turn_result.next_step, NextStepFinalOutput): + items_now_turn = [ + item + for item in items_to_save_turn + if item.type not in _DEFERRED_FINAL_OUTPUT_ITEM_TYPES + ] + deferred_items_turn = [ + item + for item in items_to_save_turn + if item.type in _DEFERRED_FINAL_OUTPUT_ITEM_TYPES + ] + else: + items_now_turn = items_to_save_turn + deferred_items_turn = [] + if items_now_turn: logger.debug( "Persisting turn items (types=%s)", - [item.type for item in items_to_save_turn], + [item.type for item in items_now_turn], ) if is_resumed_state and run_state is not None: saved_count = await save_result_to_session( session, [], - items_to_save_turn, + items_now_turn, None, response_id=turn_result.model_response.response_id, reasoning_item_id_policy=( @@ -1441,7 +1473,7 @@ def _finalize_result(result: RunResult) -> RunResult: await save_result_to_session( session, [], - items_to_save_turn, + items_now_turn, run_state, response_id=turn_result.model_response.response_id, store=store_setting, @@ -1491,7 +1523,7 @@ def _finalize_result(result: RunResult) -> RunResult: ) if ( session_persistence_enabled - and items_to_save_turn + and deferred_items_turn and not input_guardrails_triggered(input_guardrail_results) ): # When earlier items from this turn were already persisted @@ -1505,7 +1537,7 @@ def _finalize_result(result: RunResult) -> RunResult: run_state._current_turn_persisted_item_count = ( await save_resumed_turn_items( session=session, - items=items_to_save_turn, + items=deferred_items_turn, persisted_count=( run_state._current_turn_persisted_item_count ), @@ -1525,7 +1557,7 @@ def _finalize_result(result: RunResult) -> RunResult: run_state=run_state, session_persistence_enabled=session_persistence_enabled, input_guardrail_results=input_guardrail_results, - items=items_to_save_turn, + items=deferred_items_turn, response_id=turn_result.model_response.response_id, store=store_setting, ) diff --git a/src/agents/run_internal/run_loop.py b/src/agents/run_internal/run_loop.py index 168d646876..c8ad164303 100644 --- a/src/agents/run_internal/run_loop.py +++ b/src/agents/run_internal/run_loop.py @@ -414,6 +414,12 @@ async def _run_output_guardrails_for_stream( raise +# Only the model's own output for the final turn waits for the output-guardrail verdict. +# This is an allow-list on purpose: a future item type that records something the run +# already did defaults to "persist" rather than silently inheriting "discard". +_DEFERRED_FINAL_OUTPUT_ITEM_TYPES = frozenset({"message_output_item", "reasoning_item"}) + + async def _finalize_streamed_final_output( *, streamed_result: RunResultStreaming, @@ -426,6 +432,15 @@ async def _finalize_streamed_final_output( response_id: str | None, store_setting: bool | None, ) -> None: + # Tool calls and their outputs record side effects that already happened, so they persist + # before the guardrails can raise. Only the model's own output waits for a clean verdict. + side_effect_items = [ + item for item in items if item.type not in _DEFERRED_FINAL_OUTPUT_ITEM_TYPES + ] + deferred_items = [item for item in items if item.type in _DEFERRED_FINAL_OUTPUT_ITEM_TYPES] + if side_effect_items: + await save_items(side_effect_items, response_id, store_setting) + output_guardrail_results = await _run_output_guardrails_for_stream( agent=agent, run_config=run_config, @@ -437,7 +452,8 @@ async def _finalize_streamed_final_output( streamed_result.final_output = output streamed_result.is_complete = True - await save_items(items, response_id, store_setting) + if deferred_items: + await save_items(deferred_items, response_id, store_setting) streamed_result._event_queue.put_nowait(QueueCompleteSentinel()) diff --git a/tests/test_agent_runner.py b/tests/test_agent_runner.py index 0c2d2144e3..5c51f9288b 100644 --- a/tests/test_agent_runner.py +++ b/tests/test_agent_runner.py @@ -3566,6 +3566,156 @@ def foo(a: str) -> str: ) +def _tripwire_guardrail() -> OutputGuardrail[Any]: + def guardrail_function( + context: RunContextWrapper[Any], agent: Agent[Any], agent_output: Any + ) -> GuardrailFunctionOutput: + return GuardrailFunctionOutput(output_info=None, tripwire_triggered=True) + + return OutputGuardrail(guardrail_function=guardrail_function) + + +def _tool_final_output_agent( + model: FakeModel, + tool_use_behavior: Any, +) -> tuple[Agent[Any], list[str]]: + calls: list[str] = [] + + @function_tool + def charge(a: str) -> str: + calls.append(a) + return f"charged:{a}" + + agent = Agent( + name="test", + model=model, + tools=[charge], + tool_use_behavior=tool_use_behavior, + output_guardrails=[_tripwire_guardrail()], + ) + return agent, calls + + +def _stop_at_charge( + context: RunContextWrapper[Any], + results: list[Any], +) -> ToolsToFinalOutputResult: + return ToolsToFinalOutputResult(is_final_output=True, final_output=results[0].output) + + +TOOL_FINAL_OUTPUT_BEHAVIORS: list[Any] = [ + "stop_on_first_tool", + {"stop_at_tool_names": ["charge"]}, + _stop_at_charge, +] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("tool_use_behavior", TOOL_FINAL_OUTPUT_BEHAVIORS) +async def test_output_guardrail_tripwire_keeps_same_turn_tool_items(tool_use_behavior: Any): + """A tool that already ran stays in the session even when it ends the run and is rejected. + + ``tool_use_behavior`` settings that stop on a tool make the tool turn the final-output turn, + so deferring the whole turn would discard the record of a side effect that already happened. + """ + + session = SimpleListSession() + model = FakeModel() + model.set_next_output([get_function_tool_call("charge", json.dumps({"a": "b"}))]) + agent, calls = _tool_final_output_agent(model, tool_use_behavior) + + with pytest.raises(OutputGuardrailTripwireTriggered): + await Runner.run(agent, input="user_message", session=session) + + assert calls == ["b"], "the tool must have run for this scenario to be meaningful" + item_types = [cast(dict[str, Any], item).get("type") for item in await session.get_items()] + assert "function_call" in item_types + assert "function_call_output" in item_types + + +@pytest.mark.asyncio +@pytest.mark.parametrize("tool_use_behavior", TOOL_FINAL_OUTPUT_BEHAVIORS) +async def test_output_guardrail_tripwire_keeps_same_turn_tool_items_streamed( + tool_use_behavior: Any, +): + """The streamed path keeps the same-turn tool record, matching the non-streamed path.""" + + session = SimpleListSession() + model = FakeModel() + model.set_next_output([get_function_tool_call("charge", json.dumps({"a": "b"}))]) + agent, calls = _tool_final_output_agent(model, tool_use_behavior) + + with pytest.raises(OutputGuardrailTripwireTriggered): + streamed = Runner.run_streamed(agent, input="user_message", session=session) + async for _ in streamed.stream_events(): + pass + + assert calls == ["b"] + item_types = [cast(dict[str, Any], item).get("type") for item in await session.get_items()] + assert "function_call" in item_types + assert "function_call_output" in item_types + + +@pytest.mark.asyncio +@pytest.mark.parametrize("streamed", [False, True]) +async def test_output_guardrail_tripwire_splits_mixed_final_turn(streamed: bool): + """A final turn carrying both a message and a tool call is split, not kept or dropped whole. + + Deciding per turn instead of per item would either discard the executed tool or persist the + rejected message; only a per-item split satisfies both halves. + """ + + session = SimpleListSession() + model = FakeModel() + model.set_next_output( + [ + get_text_message("chatty_preamble"), + get_function_tool_call("charge", json.dumps({"a": "b"})), + ] + ) + agent, calls = _tool_final_output_agent(model, "stop_on_first_tool") + + with pytest.raises(OutputGuardrailTripwireTriggered): + if streamed: + streamed_result = Runner.run_streamed(agent, input="user_message", session=session) + async for _ in streamed_result.stream_events(): + pass + else: + await Runner.run(agent, input="user_message", session=session) + + assert calls == ["b"] + items = await session.get_items() + item_types = [cast(dict[str, Any], item).get("type") for item in items] + assert "function_call" in item_types + assert "function_call_output" in item_types + assert not any(cast(dict[str, Any], item).get("role") == "assistant" for item in items) + + +@pytest.mark.asyncio +async def test_output_guardrail_tripwire_still_drops_rejected_message_on_tool_final_turn(): + """Only side-effect records survive the tripwire; model output is still withheld.""" + + session = SimpleListSession() + model = FakeModel() + model.add_multiple_turn_outputs( + [ + [get_function_tool_call("charge", json.dumps({"a": "b"}))], + [get_text_message("should_not_be_saved")], + ] + ) + agent, _calls = _tool_final_output_agent(model, "run_llm_again") + + with pytest.raises(OutputGuardrailTripwireTriggered): + await Runner.run(agent, input="user_message", session=session) + + items = await session.get_items() + assert not any( + cast(dict[str, Any], item).get("role") == "assistant" + and cast(dict[str, Any], item).get("type") == "message" + for item in items + ) + + @pytest.mark.asyncio async def test_resumed_final_output_persists_after_passing_output_guardrail(): """Resumed runs with a positive persisted count still save accepted finals."""