Python: [Bug]: Functional workflow re-executes a @step that was still running when a sibling asyncio.gather branch paused for request_info #9268

Observed Behavior

In a functional workflow, asyncio.gather runs a @step beside a branch that calls ctx.request_info. When the request pauses the run, the sibling step is still in flight. run() returns with the pending request, but the sibling step is neither awaited nor cancelled. It finishes in the background after the run has returned, so its result is missing from the captured step cache. On run(responses=...) the step executes a second time, and the workflow output comes from that second execution.

Any side effect of the step therefore happens twice (a chat-model call, a tool call, a write), and the first result is silently discarded.

Expected Behavior

A step should execute at most once per logical invocation across pause and resume, as it does when no sibling branch is in flight. Either the run should wait for in-flight sibling steps and cache their results before suspending, or it should cancel them before returning, so that no step runs after the run has returned.

Steps to Reproduce

1. Define a functional workflow that asyncio.gathers a step calling ctx.request_info and a step that awaits I/O.
2. Run it, then resume with the response.
3. Observe that the I/O step executed twice and that the output carries the second execution's result.

Minimal Reproduction

import asyncio

from agent_framework import RunContext, step, workflow

model_calls = []


@step
async def ask(ctx: RunContext) -> int:
    return await ctx.request_info("y?", int, request_id="y")


@step
async def model(prompt: str) -> str:
    model_calls.append(prompt)
    await asyncio.sleep(0.01)  # e.g. awaiting a chat-model response
    return f"answer #{len(model_calls)}"


@workflow
async def pipeline(x: int, ctx: RunContext) -> list:
    return list(await asyncio.gather(ask(), model("hello")))


async def main():
    wf = pipeline.build()
    first = await wf.run(1)
    print("paused on:", [e.request_id for e in first.get_request_info_events()])
    await asyncio.sleep(0.1)
    final = await wf.run(responses={"y": 5})
    print("output:", final.get_outputs())
    print("model step executions:", len(model_calls))


asyncio.run(main())

Output:

paused on: ['y']
output: [[5, 'answer #2']]
model step executions: 2

Expected output: [[5, 'answer #1']] and one execution.

Error Messages and Stack Traces

No error is raised; the output is silently computed from the second execution.

With asyncio.TaskGroup instead of asyncio.gather:
BaseExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)

Package Versions

agent-framework-core: 1.20.0; also reproduced on main (1.21.0, fd52de7)

Python Version

Python 3.13.9 on macOS 15.7 (arm64); also reproduced with Python 3.12.13 on Ubuntu 22.04 (x86_64)

Operating System

macOS

Regression

Unknown

Additional Context

The cause appears to be in _functional.py. Tasks created during a run are recorded in ctx._workflow_tasks (line 112 on main), but nothing awaits or cancels them. When WorkflowInterrupted is caught (line 1674), _capture_replay_state (line 1676) snapshots the step cache while the sibling task is still running, so the task's later completion is lost.

The structured-concurrency alternative doesn't help: with asyncio.TaskGroup in place of gather, the pause raises a BaseExceptionGroup out of run(), because the internal WorkflowInterrupted arrives wrapped in a group and isn't recognized.

Found with a differential workflow fuzzer: the same generated program run on a reference interpreter consumed one scripted model reply per call, while the functional workflow consumed two.

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.