From a7d2ff5967e1d59670363f78a1d42d0dd467be4a Mon Sep 17 00:00:00 2001 From: Gunjan Jaswal Date: Mon, 27 Jul 2026 19:00:37 +0530 Subject: [PATCH] fix(run): cancel the parallel input-guardrail task when the model turn fails In the parallel input-guardrail path, the guardrail coroutine and the model turn are awaited together with asyncio.gather without return_exceptions=True, and the only handler catches InputGuardrailTripwireTriggered. When the model turn (or a guardrail) raises any other exception, gather propagates it but does not cancel the sibling awaitable, so it is orphaned and keeps running network calls and side effects after the run has already failed and returned. An exception raised inside the guardrail task after gather resolves is also swallowed. Wrap the guardrail coroutine in an explicit task and add a symmetric cleanup branch: on any non-tripwire failure, cancel whichever of the guardrail/model tasks is still pending and drain both with return_exceptions=True before re-raising. The existing tripwire behavior and the flag-gated model-task cancellation are preserved. --- src/agents/run.py | 28 +++++++++--- tests/test_guardrails.py | 98 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 120 insertions(+), 6 deletions(-) diff --git a/src/agents/run.py b/src/agents/run.py index bb07bd2554..d0e3f9ee56 100644 --- a/src/agents/run.py +++ b/src/agents/run.py @@ -1266,14 +1266,17 @@ def _finalize_result(result: RunResult) -> RunResult: ) if parallel_guardrails: + guardrail_task = asyncio.create_task( + run_input_guardrails( + starting_agent, + parallel_guardrails, + copy_input_items(original_input), + context_wrapper, + ) + ) try: parallel_results, turn_result = await asyncio.gather( - run_input_guardrails( - starting_agent, - parallel_guardrails, - copy_input_items(original_input), - context_wrapper, - ), + guardrail_task, model_task, ) except InputGuardrailTripwireTriggered: @@ -1292,6 +1295,19 @@ def _finalize_result(result: RunResult) -> RunResult: ) ) raise + except BaseException: + # A non-tripwire failure (the model turn raising, or a + # guardrail raising a non-tripwire error) propagates from + # gather without cancelling the sibling task. Cancel and drain + # whichever side is still pending so it is not left running + # after the run has failed and its exception is not swallowed. + for pending_task in (guardrail_task, model_task): + if not pending_task.done(): + pending_task.cancel() + await asyncio.gather( + guardrail_task, model_task, return_exceptions=True + ) + raise else: turn_result = await model_task diff --git a/tests/test_guardrails.py b/tests/test_guardrails.py index 511c342ca5..08e57e67b1 100644 --- a/tests/test_guardrails.py +++ b/tests/test_guardrails.py @@ -627,6 +627,104 @@ async def slow_get_response(*args, **kwargs): assert model_cancelled.is_set() is False +@pytest.mark.asyncio +async def test_model_error_cancels_parallel_input_guardrail_task(): + """A non-tripwire model failure must cancel the still-running guardrail task. + + Without cancellation the guardrail task is orphaned and keeps running after + ``Runner.run`` has already raised. + """ + guardrail_started = asyncio.Event() + guardrail_cancelled = asyncio.Event() + guardrail_finished = asyncio.Event() + + @input_guardrail(run_in_parallel=True) + async def slow_parallel_check( + ctx: RunContextWrapper[Any], agent: Agent[Any], input: str | list[TResponseInputItem] + ) -> GuardrailFunctionOutput: + guardrail_started.set() + try: + await asyncio.sleep(LONG_DELAY) + guardrail_finished.set() + return GuardrailFunctionOutput( + output_info="parallel_ok", + tripwire_triggered=False, + ) + except asyncio.CancelledError: + guardrail_cancelled.set() + raise + + model = FakeModel() + + async def boom_get_response(*args, **kwargs): + # Only blow up once the guardrail is genuinely mid-flight. + await asyncio.wait_for(guardrail_started.wait(), timeout=1) + raise RuntimeError("model boom") + + agent = Agent( + name="model_error_agent", + input_guardrails=[slow_parallel_check], + model=model, + ) + + with patch.object(model, "get_response", side_effect=boom_get_response): + with pytest.raises(RuntimeError, match="model boom"): + await Runner.run(agent, "trigger guardrail") + + # By the time Runner.run returns, the guardrail task must already be + # cancelled rather than left running to completion in the background. + assert guardrail_started.is_set() is True + assert guardrail_cancelled.is_set() is True + assert guardrail_finished.is_set() is False + + +@pytest.mark.asyncio +async def test_parallel_guardrail_non_tripwire_error_not_swallowed(): + """A non-tripwire error raised inside a parallel guardrail must propagate. + + It should also cancel the in-flight model task rather than leave it running. + """ + model_started = asyncio.Event() + model_cancelled = asyncio.Event() + model_finished = asyncio.Event() + + @input_guardrail(run_in_parallel=True) + async def raising_parallel_check( + ctx: RunContextWrapper[Any], agent: Agent[Any], input: str | list[TResponseInputItem] + ) -> GuardrailFunctionOutput: + await asyncio.wait_for(model_started.wait(), timeout=1) + raise ValueError("guardrail boom") + + model = FakeModel() + original_get_response = model.get_response + + async def slow_get_response(*args, **kwargs): + model_started.set() + try: + await asyncio.sleep(LONG_DELAY) + return await original_get_response(*args, **kwargs) + except asyncio.CancelledError: + model_cancelled.set() + raise + finally: + model_finished.set() + + agent = Agent( + name="guardrail_error_agent", + input_guardrails=[raising_parallel_check], + model=model, + ) + model.set_next_output([get_text_message("should_not_finish")]) + + with patch.object(model, "get_response", side_effect=slow_get_response): + with pytest.raises(ValueError, match="guardrail boom"): + await Runner.run(agent, "trigger guardrail") + + await asyncio.wait_for(model_finished.wait(), timeout=1) + assert model_started.is_set() is True + assert model_cancelled.is_set() is True + + @pytest.mark.asyncio async def test_parallel_guardrail_may_not_prevent_tool_execution_streaming(): tool_was_executed = False