Agent.as_tool(on_stream=...) deadlocks forever when the handler raises a BaseException

[abhay-codes07]

Please read this first

• Have you read the docs? Yes — Tools (agents as tools) and the Codex tool docs.
• Have you searched for related issues? Yes. Searched open and closed issues and PRs for deadlock, hang, on_stream, event_queue, task_done, and queue join. The nearest prior work is fix(voice): finish STT event handling after listener errors #4170, which fixed the same failure shape in the STT websocket listener; nothing covers the on_stream dispatchers.

Describe the bug

Agent.as_tool(on_stream=...) hangs forever — no exception, no timeout — when the on_stream handler raises a BaseException. asyncio.CancelledError is the realistic trigger: any asyncio.timeout(), wait_for, or task cancellation inside the callback produces one.

The dispatcher pairs a producer with a background consumer:

async def _run_handler(payload): 
    try:
        ...
    except Exception as exc:      # BaseException escapes
        log_model_and_tool_action_error(...)

async def dispatch_stream_events():
    while True:
        payload = await event_queue.get()
        try:
            if payload is not None:
                await _run_handler(payload)
        finally:
            event_queue.task_done()
        if payload is None:
            break
...
finally:
    await event_queue.put(None)
    await event_queue.join()      # <-- waits for a consumer that is already dead
    await dispatch_task

_run_handler catches only Exception, so a BaseException propagates out of dispatch_stream_events and the task ends. The producer then reaches event_queue.join(), which waits for task_done() calls that can never arrive. The run stops permanently.

The same copy of this pattern exists in two more places:

• src/agents/extensions/experimental/codex/codex_tool.py::_consume_events — hangs identically, and because the finally block finishes active tracing spans after the join, those spans are also stranded open.
• src/agents/sandbox/memory/manager.py::flush — the worker catches only Exception, so a BaseException from _process_rollout_file ends it and the await self._queue.join() before phase-two consolidation blocks forever.

Note the join() is redundant in all three: the sentinel is queued last and the consumer only returns after consuming it, so awaiting the consumer task already implies every event was handled.

Debug information

• Agents SDK version: 0.19.3 (reproduced on main at 19e364c1)
• Related library versions: n/a
• Python version: 3.12
• Operating system: Windows 11 (not platform specific — this is pure asyncio)
• Model and model provider: none needed; reproduced with tests/fake_model.py
• Does the issue reproduce with the latest Agents SDK release? Yes.
• Does the issue occur consistently or intermittently? Consistently and deterministically.

No exception is raised and no traceback is produced — the hang is the symptom.

Repro steps

import asyncio
import sys

sys.path.insert(0, "tests")  # run from the repo root

from agents import Agent, Runner
from fake_model import FakeModel
from test_responses import get_function_tool_call, get_text_message


async def main() -> None:
    inner_model = FakeModel()
    inner = Agent(name="Inner", model=inner_model)
    inner_model.add_multiple_turn_outputs([[get_text_message("inner done")]])

    async def on_stream(payload):
        # Any cancellation inside a handler -- an asyncio.timeout(), a cancelled
        # task -- surfaces here as CancelledError, which is a BaseException.
        raise asyncio.CancelledError()

    tool = inner.as_tool(
        tool_name="inner_tool", tool_description="run inner", on_stream=on_stream
    )

    outer_model = FakeModel()
    outer = Agent(name="Outer", model=outer_model, tools=[tool])
    outer_model.add_multiple_turn_outputs([
        [get_function_tool_call("inner_tool", '{"input": "hi"}', call_id="c1")],
        [get_text_message("outer done")],
    ])

    try:
        result = await asyncio.wait_for(Runner.run(outer, "go"), timeout=10)
        print("completed:", result.final_output)
    except asyncio.TimeoutError:
        print("DEADLOCK: the run never finished")


asyncio.run(main())

Actual behavior

DEADLOCK: the run never finished

The wait_for in the repro is only there to bound the script. Without it the coroutine never completes. Python also reports the dead consumer separately:

Task exception was never retrieved
future: <Task finished coro=<...dispatch_stream_events() done> exception=CancelledError()>

Expected behavior

The tool invocation settles. Either the handler's BaseException propagates to the caller or the run continues, but it must not block forever waiting on a consumer that has already exited.

Root-cause hypothesis

(hypothesis) Waiting on asyncio.Queue.join() assumes the consumer stays alive to call task_done(). Because the handler wrapper only catches Exception, that assumption breaks for any BaseException. Awaiting the consumer task instead of the queue gives the same completion guarantee — the sentinel is the last item queued and the consumer returns only after consuming it — while surfacing a dead consumer rather than waiting on it.

Proposed scope

Replace the join() wait with a wait on the consumer task in the three dispatchers above. No public API change and no change to handler error semantics for ordinary Exceptions, which stay logged and swallowed.

I have a fix with regression tests ready and will open a PR referencing this issue.