Python: [Bug]: After a checkpoint restore, a sub-workflow re-emits a pending request while handling a sibling response, so later restores keep asking for an already-answered request #9270

Observed Behavior

A sub-workflow (WorkflowExecutor with propagate_request=True) fans out to two executors that each call ctx.request_info. Its parent pauses with both requests pending, is restored into a fresh workflow from the pause checkpoint (run(checkpoint_id=..., responses=...)), and both requests are answered in that call. The parent then continues to a later request c and pauses again.

The checkpoint written at that second pause still lists the sub-workflow's request b as pending, although b was answered. After restoring from it and answering c, the workflow yields its final output but still reports b as pending, so it ends idle with a pending request instead of completing. Answering b, as a client must, is ignored ("received a response for an unknown or already-handled request_id"). The same workflow resumed in place completes with no pending request.

Expected Behavior

Restoring a checkpoint and answering all pending requests should leave none of them pending in later checkpoints, exactly as when the workflow is resumed in place. After the last answer, the workflow should yield its output and complete with no pending request.

Steps to Reproduce

1. Build a parent workflow whose start executor is a WorkflowExecutor (propagate_request=True) wrapping a sub-workflow that fans out to two executors calling request_info and fans back in; follow it with an executor that calls request_info again, then an executor that yields the output.
2. Run it with a checkpoint storage; it pauses on the sub-workflow's two requests.
3. Resolve the pause checkpoint, build a fresh workflow and run(checkpoint_id=..., responses=...) answering both; it pauses on the later request.
4. Resolve the pause checkpoint again, build a fresh workflow and answer the later request.
5. Observe that the run yields the output but still reports an already-answered request as pending, and that answering it is ignored.

Minimal Reproduction

import asyncio
import sys
from dataclasses import dataclass

from agent_framework import (Executor, FunctionExecutor, InMemoryCheckpointStorage, WorkflowBuilder,
                             WorkflowContext, WorkflowExecutor, handler, response_handler)


@dataclass
class Ask:
    field: str
    data: dict


class Request(Executor):
    def __init__(self, field):
        super().__init__(id=f"ask_{field}")
        self.field = field

    @handler
    async def ask(self, data: dict, ctx: WorkflowContext[dict]) -> None:
        await ctx.request_info(Ask(self.field, data), int)

    @response_handler
    async def answer(self, request: Ask, response: int, ctx: WorkflowContext[dict]) -> None:
        await ctx.send_message({**request.data, request.field: response})


async def fork(data: dict, ctx: WorkflowContext[dict]) -> None:
    await ctx.send_message(data)


async def join(parts: list[dict], ctx: WorkflowContext[dict, dict]) -> None:
    await ctx.yield_output({k: v for part in parts for k, v in part.items()})


async def finish(data: dict, ctx: WorkflowContext[dict, dict]) -> None:
    await ctx.yield_output(data)


def build():
    start, a, b = FunctionExecutor(fork, id="fork"), Request("a"), Request("b")
    merge = FunctionExecutor(join, id="join")
    inner = (WorkflowBuilder(start_executor=start).add_fan_out_edges(start, [a, b])
             .add_fan_in_edges([a, b], merge).build())
    outer_start = WorkflowExecutor(inner, id="sub", propagate_request=True)
    after, sink = Request("c"), FunctionExecutor(finish, id="finish")
    return (WorkflowBuilder(start_executor=outer_start).add_edge(outer_start, after)
            .add_edge(after, sink).build())


async def main(restore: bool):
    storage = InMemoryCheckpointStorage()
    workflow = build()
    result = await workflow.run({}, checkpoint_storage=storage)
    while True:
        pending = {event.request_id: event.data.field for event in result.get_request_info_events()}
        print("outputs:", result.get_outputs(), "pending:", sorted(pending.values()))
        if not pending:
            break
        responses = {request_id: {"a": 1, "b": 2, "c": 3}[field] for request_id, field in pending.items()}
        if restore:  # Resume a fresh workflow from the pause checkpoint, as after a process restart.
            checkpoint_id = await workflow.resolve_pause_checkpoint_id(responses, checkpoint_storage=storage)
            workflow = build()
            result = await workflow.run(checkpoint_id=checkpoint_id, checkpoint_storage=storage, responses=responses)
        else:
            result = await workflow.run(responses=responses, checkpoint_storage=storage)


asyncio.run(main(sys.argv[1] == "restore"))

python repro.py live:

outputs: [] pending: ['a', 'b']
outputs: [] pending: ['c']
outputs: [{'a': 1, 'b': 2, 'c': 3}] pending: []

python repro.py restore:

outputs: [] pending: ['a', 'b']
outputs: [] pending: ['c']
outputs: [{'a': 1, 'b': 2, 'c': 3}] pending: ['b']
outputs: [] pending: []

Error Messages and Stack Traces

WorkflowExecutor sub received a response for an unknown or already-handled request_id: 5cb2afdf-d177-4a06-896b-d00f64d4b2b6. This response will be ignored.

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

WorkflowExecutor._handle_response forwards each response to the sub-workflow with workflow.run(responses={request_id: response}) and then processes the request_info events in the result. Instrumenting it on the repro shows the difference:

• Resumed in place, forwarding a returns no requests and forwarding b returns none.
• After the restore, forwarding a returns b again: b is still pending inside the sub-workflow but was already propagated before the checkpoint. The parent re-registers b while the same batch answers it, and every later checkpoint keeps it pending.

This looks related to #3255 (fixed in #3689), which stopped a restored sub-workflow from re-emitting a request at the start of a superstep. Here the re-emission happens in the first resumed run's result, when two requests are answered together.

Found with a differential workflow fuzzer: for the generated program sequence(subgraph(parallel(request, request)), request), resuming in place and every other configuration finished with one output and no pending request, but restoring each pause checkpoint into a fresh workflow left an answered request pending after the output.

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.