From e91a42c51359f8fb175967cd7f20bec39adff70e Mon Sep 17 00:00:00 2001 From: Jeremy Schoemaker Date: Sun, 23 Aug 2026 15:40:47 -0500 Subject: [PATCH 1/2] =?UTF-8?q?=F0=9F=90=9B=20fix:=20max=5Fturns=20no=20lo?= =?UTF-8?q?nger=20clobbers=20a=20tripped=20input=20guardrail=20in=20stream?= =?UTF-8?q?ing=20runs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RunResultStreaming._check_errors() re-creates MaxTurnsExceeded and overwrites self._stored_exception on every call once current_turn > max_turns, because _max_turns_handled is only ever set True by the opt-in max_turns error-handler path in run_internal/run_loop.py. In the default (no custom handler) path it stays False forever, so _check_errors() re-fires on each of its three call sites in stream_events(), including the unconditional call in the finally block. Input guardrails default to run_in_parallel=True, so a tripped guardrail can already be drained into _stored_exception as InputGuardrailTripwireTriggered by an earlier _check_errors() call, then get silently overwritten by a freshly-minted MaxTurnsExceeded on a later one. The documented `except InputGuardrailTripwireTriggered` pattern never sees the tripwire. Set _max_turns_handled = True in the default branch too, mirroring what the handler path already does. Added a regression test in test_stream_input_guardrail_timing.py that trips an input guardrail on a run that also exceeds max_turns=1 and asserts the caller gets InputGuardrailTripwireTriggered, not MaxTurnsExceeded. --- src/agents/result.py | 1 + tests/test_stream_input_guardrail_timing.py | 74 ++++++++++++++++++++- 2 files changed, 73 insertions(+), 2 deletions(-) diff --git a/src/agents/result.py b/src/agents/result.py index 0979631200..f88819df54 100644 --- a/src/agents/result.py +++ b/src/agents/result.py @@ -1053,6 +1053,7 @@ def _check_errors(self): max_turns_exc = MaxTurnsExceeded(f"Max turns ({self.max_turns}) exceeded") max_turns_exc.run_data = self._create_error_details() self._stored_exception = max_turns_exc + self._max_turns_handled = True # Fetch all the completed guardrail results from the queue and raise if needed while not self._input_guardrail_queue.empty(): diff --git a/tests/test_stream_input_guardrail_timing.py b/tests/test_stream_input_guardrail_timing.py index 5d8bd676f7..68252a2f52 100644 --- a/tests/test_stream_input_guardrail_timing.py +++ b/tests/test_stream_input_guardrail_timing.py @@ -1,17 +1,25 @@ from __future__ import annotations import asyncio +import json from datetime import datetime from typing import Any import pytest from openai.types.responses import ResponseCompletedEvent -from agents import Agent, GuardrailFunctionOutput, InputGuardrail, RunContextWrapper, Runner +from agents import ( + Agent, + GuardrailFunctionOutput, + InputGuardrail, + MaxTurnsExceeded, + RunContextWrapper, + Runner, +) from agents.exceptions import InputGuardrailTripwireTriggered from agents.items import TResponseInputItem from agents.testing import ScriptedModel -from tests.test_responses import get_text_message +from tests.test_responses import get_function_tool, get_function_tool_call, get_text_message from tests.testing_processor import fetch_events, fetch_ordered_spans FAST_GUARDRAIL_DELAY = 0.005 @@ -173,6 +181,68 @@ async def test_run_streamed_input_guardrail_tripwire_raises(guardrail_delay: flo ) +@pytest.mark.asyncio +async def test_max_turns_does_not_clobber_input_guardrail_tripwire(): + """A guardrail tripwire recorded before max_turns fires must win over MaxTurnsExceeded. + + Regression test: RunResultStreaming._check_errors() re-creates a fresh + MaxTurnsExceeded and overwrites self._stored_exception on *every* call once + current_turn > max_turns, because self._max_turns_handled is only ever set + True by the max_turns error-handler path -- never in the default (no + handler) path. stream_events() calls _check_errors() again unconditionally + in its `finally` block, so a guardrail trip that was already captured as + InputGuardrailTripwireTriggered got silently replaced with MaxTurnsExceeded + by that final call. Callers using the documented + `except InputGuardrailTripwireTriggered` pattern never saw the tripwire. + """ + + async def tripping_guardrail( + ctx: RunContextWrapper[Any], agent: Agent[Any], input: str | list[TResponseInputItem] + ) -> GuardrailFunctionOutput: + # A real moderation/safety guardrail call takes tens to hundreds of ms, + # i.e. longer than these fast scripted model turns need to blow + # through max_turns=1. This timing is the default, not an edge case. + await asyncio.sleep(0.05) + return GuardrailFunctionOutput(output_info={"reason": "blocked"}, tripwire_triggered=True) + + model = ScriptedModel() + func_output = json.dumps({"a": "b"}) + model.extend( + [ + [ + get_text_message(str(i)), + get_function_tool_call("some_function", func_output, str(i)), + ] + for i in range(1, 10) + ] + ) + + agent = Agent( + name="MaxTurnsGuardrailAgent", + model=model, + tools=[get_function_tool("some_function", "result")], + # run_in_parallel defaults to True -- this is the default configuration, + # not an opt-in one. + input_guardrails=[InputGuardrail(guardrail_function=tripping_guardrail, name="trip")], + ) + + result = Runner.run_streamed(agent, input="user_message", max_turns=1) + + raised: BaseException | None = None + try: + async for _ in result.stream_events(): + pass + except BaseException as exc: # noqa: BLE001 - we need to inspect the exact type raised + raised = exc + + assert isinstance(raised, InputGuardrailTripwireTriggered), ( + f"Expected InputGuardrailTripwireTriggered, got " + f"{type(raised).__name__ if raised else None}. The tripped guardrail " + "result was silently clobbered by a freshly-minted MaxTurnsExceeded." + ) + assert not isinstance(raised, MaxTurnsExceeded) + + class SlowCompleteScriptedModel(ScriptedModel): """A ScriptedModel that delays just before emitting ResponseCompletedEvent in streaming.""" From 0ca6de54015f8d5e8b683c576d43190bedac3e38 Mon Sep 17 00:00:00 2001 From: Jeremy Schoemaker Date: Sun, 23 Aug 2026 16:35:35 -0500 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=94=A7=20fix:=20replace=20sleep-based?= =?UTF-8?q?=20race=20ordering=20with=20a=20deterministic=20event?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit test_max_turns_does_not_clobber_input_guardrail_tripwire ordered its race with asyncio.sleep(0.05) inside the tripping guardrail. Under CI load or slower model instrumentation, the guardrail could finish before current_turn > max_turns was ever reached, in which case the test would observe the correct exception even against the unpatched implementation and silently stop being a regression test. Replace the sleep with an asyncio.Event set by a max_turns error handler that declines (returns None, so it falls through to the exact same default raise path as no handler at all). The handler fires at the exact moment the run loop establishes current_turn > max_turns, and the guardrail now awaits that event before returning its tripwire, so it can only ever resolve after the max-turns condition genuinely holds. Verified 20/20 pass with the fix in place and 20/20 fail (on the correct assertion) with src/agents/result.py's fix reverted. --- tests/test_stream_input_guardrail_timing.py | 32 +++++++++++++++++---- 1 file changed, 27 insertions(+), 5 deletions(-) diff --git a/tests/test_stream_input_guardrail_timing.py b/tests/test_stream_input_guardrail_timing.py index 68252a2f52..0ed8cea262 100644 --- a/tests/test_stream_input_guardrail_timing.py +++ b/tests/test_stream_input_guardrail_timing.py @@ -194,15 +194,29 @@ async def test_max_turns_does_not_clobber_input_guardrail_tripwire(): InputGuardrailTripwireTriggered got silently replaced with MaxTurnsExceeded by that final call. Callers using the documented `except InputGuardrailTripwireTriggered` pattern never saw the tripwire. + + This race must not be ordered with a real-time sleep: a fixed delay only + approximates "the guardrail finishes after max_turns is exceeded", and + under CI load, tracing overhead, or slower model instrumentation the + guardrail can instead finish *before* current_turn > max_turns is ever + reached, in which case the test would observe the correct exception even + against the unpatched (buggy) implementation and silently stop being a + regression test. Instead, an `error_handlers={"max_turns": ...}` hook + (returning None, so it falls through to the exact same default raise path + as if no handler were registered) sets an `asyncio.Event` at the precise + moment the run loop establishes current_turn > max_turns. The guardrail + awaits that event before returning its tripwire, so it can only ever + resolve *after* the max-turns condition genuinely holds. """ + max_turns_reached = asyncio.Event() + async def tripping_guardrail( ctx: RunContextWrapper[Any], agent: Agent[Any], input: str | list[TResponseInputItem] ) -> GuardrailFunctionOutput: - # A real moderation/safety guardrail call takes tens to hundreds of ms, - # i.e. longer than these fast scripted model turns need to blow - # through max_turns=1. This timing is the default, not an edge case. - await asyncio.sleep(0.05) + # Wait for the run loop to have actually established current_turn > + # max_turns, rather than guessing at a delay long enough to outlast it. + await max_turns_reached.wait() return GuardrailFunctionOutput(output_info={"reason": "blocked"}, tripwire_triggered=True) model = ScriptedModel() @@ -226,7 +240,15 @@ async def tripping_guardrail( input_guardrails=[InputGuardrail(guardrail_function=tripping_guardrail, name="trip")], ) - result = Runner.run_streamed(agent, input="user_message", max_turns=1) + result = Runner.run_streamed( + agent, + input="user_message", + max_turns=1, + # Declining (returning None) preserves the exact default max_turns + # behavior; the handler exists purely to signal, deterministically, + # the moment current_turn > max_turns is established. + error_handlers={"max_turns": lambda data: max_turns_reached.set()}, + ) raised: BaseException | None = None try: