Python: [Bug]: Functional workflow raises a failure that asyncio.gather discarded when the workflow catches the gather's failure and then pauses for request_info #9612

Observed Behavior

A functional workflow runs two @step calls with asyncio.gather. Both fail, the second one later. The workflow catches the gather's failure (the first Boom) with try/except and asks for input with ctx.request_info.

When the first failure reaches the gather, the second step is still running. Python lets it run on, and when it fails, gather discards that exception. Nothing in the workflow awaits it.

On 1.22.0 and 1.23.0, the first run() raises the second step's Boom from _settle_workflow_tasks instead of returning the pending request. The workflow cannot be paused or resumed, although its own code handled the gather's failure.

Expected Behavior

The first run returns the pending request ask. Answering it with 1 completes the workflow with the output [1], as on 1.20.0 and 1.21.0.

Steps to Reproduce

1. pip install agent-framework-core==1.23.0
2. Run the script below.
3. The first app.run(0) raises Boom: boom after 0.01s.

Minimal Reproduction

import asyncio
from importlib.metadata import version

from agent_framework import RunContext, step, workflow


class Boom(Exception):
    pass


@step
async def boom(delay: float) -> int:
    await asyncio.sleep(delay)
    raise Boom(f"boom after {delay}s")


@workflow
async def flow(x: int, ctx: RunContext) -> int:
    try:
        await asyncio.gather(boom(0), boom(0.01))
    except Boom:  # Handled: the first failure. The second step is still running.
        return await ctx.request_info("answer?", int, request_id="ask")
    return 0


async def main():
    app = flow.build()
    result = await app.run(0)
    while pending := result.get_request_info_events():
        result = await app.run(responses={event.request_id: 1 for event in pending})
    print(f"agent-framework-core {version('agent-framework-core')}: outputs={result.get_outputs()}")


asyncio.run(main())

On 1.21.0 this prints:

agent-framework-core 1.21.0: outputs=[1]

Error Messages and Stack Traces

Traceback (most recent call last):
  File "repro.py", line 35, in main
  File "agent_framework/_types.py", line 4873, in get_final_response
  File "agent_framework/_types.py", line 4762, in __anext__
  File "agent_framework/_types.py", line 4532, in _pull_next_update
  File "agent_framework/_workflows/_functional.py", line 2176, in _run_core
  File "agent_framework/_workflows/_functional.py", line 2357, in _execute
  File "agent_framework/_workflows/_functional.py", line 1175, in _settle_workflow_tasks
  File "agent_framework/_workflows/_functional.py", line 1170, in _raise_completed_failure
  File "agent_framework/_workflows/_functional.py", line 125, in _tracked_coro
  File "agent_framework/_workflows/_functional.py", line 1574, in __call__
  File "repro.py", line 21, in boom
Boom: boom after 0.01s

Package Versions

agent-framework-core: 1.23.0 (also 1.22.0, and main at a8e284c)

Python Version

Python 3.13.9

Operating System

macOS

Regression

Yes

Additional Context

• At the pause, request_info registers every task that is inside a @step as a pause failure boundary (self._pause_failure_boundaries.update(self._active_step_tasks), _functional.py line 808). The task of boom(0.01) is still inside its step, so it becomes a boundary. Its gather has already finished, so _capture_waiter_tree records no waiter for it.
• _settle_workflow_tasks waits for that task to finish. Then _pause_boundary_was_handled (lines 1004-1009) returns False because the boundary has no waiter, and _raise_completed_failure raises its exception (line 1170). The pause did not hide this failure: the gather had already finished with the first Boom before the workflow paused.
• The same workflow without the pause (return -1 in the except) completes with [-1]. With a pause, the run succeeds if the second step fails before the pause, or if it is a plain coroutine instead of a @step. A gather with return_exceptions=True also pauses and resumes normally.
• Related to Python: [Bug]: Functional workflow re-raises a step failure that the workflow already handled, when a sibling branch pauses for request_info #9549 but a different path. There, a boundary whose waiter is a nested gather that failed is judged unhandled. Here, the boundary has no waiter. The fix proposed in Python: fix(workflow): keep a sibling pause when a nested step failure was handled #9571 resolves Python: [Bug]: Functional workflow re-raises a step failure that the workflow already handled, when a sibling branch pauses for request_info #9549's reproduction but not this one.
• Possible direction: if a boundary has no waiter and its exception was already retrieved (gather retrieves the exceptions of children that finish after it), treat it like the non-boundary tasks that _raise_completed_failure skips when their exception was retrieved (_log_traceback). Locally, with that change, this reproduction and its variants pause and resume, a failure that nobody retrieved still surfaces, and tests/workflow/test_functional_workflow.py passes (210 tests).

Acknowledgements

• I searched existing issues and did not find a duplicate.
• I personally verified this behavior and the reproduction details are authentic.
• I will wait for explicit maintainer agreement before starting implementation of a non-trivial change.