Python: [Bug]: An answered request_info is asked again under a new auto:: id when a sibling @step was running while it paused #9517

Observed Behavior

In a functional workflow, asyncio.gather runs a @step beside a step that calls ctx.request_info without a request_id. The request pauses the run as auto::0 while the sibling step is still in flight. Since 1.22.0 (#9291), the sibling step then completes and its result is cached.

When the run is resumed with responses={"auto::0": ...}, the sibling step is served from the cache, and the request is issued again as auto::1. The answer to auto::0 is never delivered, and the run pauses a second time for input it has already received. Only a second answer, to auto::1, completes the run.

A client that answers each request once therefore sees the workflow ask the same question twice, under different ids.

Expected Behavior

The resumed run delivers the response to auto::0 and completes, as it does when the sibling step finishes before the request is made, or when the request has an explicit request_id.

Steps to Reproduce

1. Define a functional workflow that asyncio.gathers a @step that awaits (here asyncio.sleep(0.01)) and a @step that calls ctx.request_info(...) without a request_id.
2. Run it. The run pauses with the pending request auto::0.
3. Resume with responses={"auto::0": 10}. The run pauses again, now with the pending request auto::1, and produces no output.

Minimal Reproduction

import asyncio

from agent_framework import RunContext, step, workflow


@step
async def work(x: int) -> int:
    await asyncio.sleep(0.01)  # Still running when its sibling asks for input.
    return x + 1


@step
async def ask(x: int, ctx: RunContext) -> int:
    return await ctx.request_info("a number?", int)  # No request_id: an automatic auto::<index> id.


@workflow
async def flow(x: int) -> int:
    done, answer = await asyncio.gather(work(x), ask(x))
    return done + answer


app = flow.build()


async def main():
    result = await app.run(1)
    asked = [event.request_id for event in result.get_request_info_events()]
    print("first run asks:", asked)
    result = await app.run(responses={asked[0]: 10})
    again = [event.request_id for event in result.get_request_info_events()]
    print("after answering it, the run asks:", again, "outputs:", result.get_outputs())
    assert not again and result.get_outputs() == [12], "an answered request was asked again under a new id"


asyncio.run(main())

Error Messages and Stack Traces

first run asks: ['auto::0']
after answering it, the run asks: ['auto::1'] outputs: []
AssertionError: an answered request was asked again under a new id


On 1.20.0 and 1.21.0 the same script prints:


first run asks: ['auto::0']
after answering it, the run asks: [] outputs: [12]

Package Versions

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

Python Version

Python 3.13.9

Operating System

macOS

Regression

Yes

Additional Context

• RunContext._auto_request_info_index is a single counter shared by all concurrent tasks of a run. StepWrapper records, for each completed step, how many automatic request ids the step used: the counter after the step's body minus the counter before it (_set_cached_step_auto_request_info_count). On a cache hit, _advance_auto_request_info_index_for_cached_step advances the counter by that number, so that later automatic ids keep their values.
• When a sibling calls request_info while the step's body is awaiting, the sibling's id falls inside that difference. The step is recorded as having used one automatic id that it never requested.
• On resume, the cached step advances the counter to 1 before the sibling replays its call, so the sibling's request becomes auto::1, which has no response.
• This accounting is the same in 1.20.0 and 1.21.0. There, a step still running when a sibling paused never reached the cache (Python: [Bug]: Functional workflow re-executes a @step that was still running when a sibling asyncio.gather branch paused for request_info #9268), so the inflated count was never recorded. Python: Preserve concurrent steps across workflow pause #9291 lets that step complete and be cached, which exposes it.
• The count would need to be kept per task (for example, the ids requested by the step's own task and its descendants) rather than taken from the shared counter.
• Not covered by Python: [Bug]: Functional workflow checkpoint replay can swap cached results between concurrent calls to the same @step #8292 or Python: [Bug]: Functional workflow checkpoint replay returns a wrong cached result when a @step calls another @step #9385, which concern which cached result a step call receives, not automatic request ids.
• An explicit request_id avoids the problem.

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.