Python: [Bug]: Functional workflow raises a bare CancelledError when a step in a nested asyncio.gather starts after a sibling branch pauses for request_info #9516

Observed Behavior

A functional workflow runs two branches with asyncio.gather. One branch awaits a nested asyncio.gather of @step calls. The other is a @step that calls ctx.request_info.

On 1.22.0, the first run() raises a bare asyncio.CancelledError from _settle_workflow_tasks, instead of returning the pending request. The workflow therefore cannot be paused or resumed. The same workflow pauses and resumes correctly on 1.20.0 and 1.21.0.

Expected Behavior

The first run returns the pending request (auto::0). Answering it with 7 completes the workflow with the output [[2, 3], 7], as on 1.21.0.

Steps to Reproduce

1. pip install agent-framework-core==1.22.0
2. Run the script below.
3. The first instance.run(1) raises asyncio.exceptions.CancelledError.

Minimal Reproduction

import asyncio

from agent_framework import RunContext, step, workflow


@step
async def first(value: int) -> int:
    return value + 1


@step
async def second(value: int) -> int:
    return value + 2


@step
async def ask(value: int, ctx: RunContext) -> int:
    return await ctx.request_info({"ask": "a number"}, int)


@workflow
async def pipeline(value: int, ctx: RunContext) -> list:
    async def both():
        return await asyncio.gather(first(value), second(value))

    return await asyncio.gather(both(), ask(value))


async def main():
    instance = pipeline.build()
    result = await instance.run(1)
    requests = result.get_request_info_events()
    print("asked:", [event.request_id for event in requests])
    result = await instance.run(responses={event.request_id: 7 for event in requests})
    print("outputs:", result.get_outputs())  # Expected: [[[2, 3], 7]]


asyncio.run(main())

On 1.21.0 this prints:

asked: ['auto::0']
outputs: [[[2, 3], 7]]

Error Messages and Stack Traces

agent_framework._workflows._functional.WorkflowInterrupted: Workflow interrupted by request_info (request_id=auto::0)

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "repro.py", in main
    result = await instance.run(1)
  File "agent_framework/_types.py", line 4821, in get_final_response
  File "agent_framework/_types.py", line 4716, in __anext__
  File "agent_framework/_types.py", line 4486, in _pull_next_update
  File "agent_framework/_workflows/_functional.py", line 2176, in _run_core
    return_value = await self._execute(ctx, message)
  File "agent_framework/_workflows/_functional.py", line 2350, in _execute
    await ctx._settle_workflow_tasks(
  File "agent_framework/_workflows/_functional.py", line 1175, in _settle_workflow_tasks
    _raise_completed_failure()
  File "agent_framework/_workflows/_functional.py", line 1140, in _raise_completed_failure
    task.result()
asyncio.exceptions.CancelledError

Package Versions

agent-framework-core: 1.22.0 (also main at ebcf4db)

Python Version

Python 3.13.9 (also reproduced on Python 3.12 on Linux)

Operating System

macOS

Regression

Yes

Additional Context

The failure comes from the pause settlement added in #9291.

1. ask pauses first. first and second are then started by the nested gather, and stop at step entry by raising the run's WorkflowInterrupted (_raise_if_current_task_interrupted). On Python 3.11+ that exception is a CancelledError subclass, so their tasks end in the cancelled state.
2. asyncio keeps the exception that cancelled a task only for the first retrieval: Future._make_cancelled_error() returns it once and then clears it. The nested gather's done callback makes that first retrieval, when it propagates the child's cancellation to its outer future.
3. In _raise_completed_failure, task.result() on such a step task therefore raises a new bare CancelledError, not WorkflowInterrupted. The fallback accepts a cancelled task only if it is an ancestor of the pausing task (_get_new_descendant_interruption). A step under the sibling branch is not, so the CancelledError is re-raised and escapes run().

With a single, flat asyncio.gather(ask(value), first(value), second(value)), the gather's callback only consumes the exception of the first cancelled child, which is the pausing step itself, so the run pauses correctly. A nested gather of a single step is enough to fail, in either order of the outer gather's branches.

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.