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..0ed8cea262 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,90 @@ 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. + + 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: + # 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() + 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, + # 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: + 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."""