fix(core): max_turns no longer clobbers a tripped input guardrail exception in streaming

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():

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."""