Repository navigation
Python: [Bug]: Parallel functional steps fork the checkpoint lineage #7647
Description
Description
When two @step calls in a functional workflow with checkpoint storage run through asyncio.gather, both callbacks that run after a step completes can read the same previous_checkpoint_id before either callback updates the shared checkpoint chain. The two successful steps then create sibling root checkpoints. A later final or approval checkpoint continues from only one sibling, leaving the other checkpoint unreachable from the latest checkpoint's lineage.
I expected the checkpoint records created after each step to form one history chain, consistent with WorkflowCheckpoint.previous_checkpoint_id and the existing sequential test_per_step_checkpoint_chain contract. The final result itself remains correct in this reproduction, but checkpoint ordering and history traversal are no longer represented by one chain.
Related issue #4588 and merged PR #6695 established continuous previous_checkpoint_id ancestry during restoration from storage. This report is a different code path: concurrent saves made after steps complete in FunctionalWorkflow can fork the ancestry inside one run.
Minimal reproduction:
import asyncio
import tempfile
from pathlib import Path
from agent_framework import FileCheckpointStorage, step, workflow
@step
async def left(value: int) -> int:
return value + 1
@step
async def right(value: int) -> int:
return value + 2
async def main() -> None:
with tempfile.TemporaryDirectory() as directory:
storage = FileCheckpointStorage(Path(directory))
@workflow(checkpoint_storage=storage)
async def parallel(value: int) -> list[int]:
return await asyncio.gather(left(value), right(value))
await parallel.run(1)
checkpoints = await storage.list_checkpoints(workflow_name="parallel")
by_id = {checkpoint.checkpoint_id: checkpoint for checkpoint in checkpoints}
latest = await storage.get_latest(workflow_name="parallel")
reachable = set()
while latest is not None:
reachable.add(latest.checkpoint_id)
latest = by_id.get(latest.previous_checkpoint_id) if latest.previous_checkpoint_id else None
print("checkpoint count:", len(checkpoints))
print("root count:", sum(checkpoint.previous_checkpoint_id is None for checkpoint in checkpoints))
print("unreachable from latest:", len(checkpoints) - len(reachable))
asyncio.run(main())Output in 5/5 independent runs:
checkpoint count: 3
root count: 2
unreachable from latest: 1
The callback for each step currently updates a shared list containing one item without serializing the read, save, and update sequence:
async def _on_step_completed() -> None:
ckpt_chain[0] = await self._save_checkpoint(ctx, storage, ckpt_chain[0])Because #4588 and #6695 already establish continuous ancestry across restoration, does the same invariant intentionally apply to concurrent step checkpoints in FunctionalWorkflow? Serializing only the checkpoint chain update would preserve parallel step execution. Since this surface is experimental and open PR #7521 is changing how definitions are built, I would wait for confirmation before preparing the bounded change: one regression test plus internal serialization of the checkpoint read, save, and update sequence.
Code Sample
Error Messages / Stack Traces
Package Versions
agent-framework-core 1.13.0 from main at 7cfa905486acd325e8958a090b05af481285e525
Python Version
Python 3.13.7 on Apple silicon (arm64)
Additional Context
The focused tests for the functional surface, checkpoints, local evaluation, and observability pass unchanged: 233 tests passed. A broader external run also preserved restoration into a new object, approval request identity, one execution per completed side effect, exception chaining, and cancellation propagation; the isolated discrepancy is the parallel checkpoint lineage.
A controlled storage probe forced the first two saves made after steps completed to finish in reverse order, stopped the attempted final checkpoint, and restored from get_latest in a fresh FunctionalWorkflow object. Restore kept both results and did not execute either step again, so this report does not show output loss or duplicate side effects. The attempted final checkpoint nevertheless selected the earlier snapshot containing one completed step as its parent while the later snapshot containing both completed steps was its sibling, confirming that the chain can omit a strictly more advanced intermediate checkpoint.
Metadata
Metadata
Assignees
Labels
Type
Projects
- Status
Activity
github-actions commentedon Aug 13, 2026
🤖 Automated triage reproduction notes (agent-authored — trust but verify)
Agent analysis
Repro: The
_on_step_completedclosure inFunctionalWorkflow._run_impl()atpython/packages/core/agent_framework/_workflows/_functional.py:979-980races when concurrent steps (viaasyncio.gather) both readckpt_chain[0]before either updates it after_save_checkpointreturns. Minimal repro: define two@stepfunctions gathered in a@workflow(checkpoint_storage=<storage-with-yield>), run the workflow, and assert the checkpoint chain has exactly one root. Fix: serialize the read-save-update ofckpt_chain[0]with anasyncio.Lockaround the body of_on_step_completed.python/packages/core/tests/workflow/test_parallel_checkpoint_lineage.py1.13.01.13.0ZedingZhang commentedon Sep 21, 2026
I'd like to pick this up.
Plan, following Evan Mattson (@moonbox3)'s guidance on #7676: route all three checkpoint save paths (per-step, final, and the HITL/
request_info()interruption) through a single locked read/save/update helper, and add regression coverage for both the parallel-step and therequest_info()cases.Two things I confirmed by reproducing this on
669c8b9:ckpt_chain[0], which are exactly the paths KirschQAQ (@KirschBluteX) flagged as able to fork the lineage.request_info()case,asyncio.gatherdoes not cancel its remaining awaitables when one of them raises, so a sibling step really does keep running after the HITL checkpoint is written. It then links to the stale head and forks the chain. Gating the sibling step and releasing it after the run returns reproduces this deterministically: 2 roots and 1 checkpoint unreachable fromget_latest().PR to follow.
Python: retire functional workflow checkpoints once a run closes
Python: hold the run guard until in-flight checkpoint saves drain
Python: continue the checkpoint chain across a response-only HITL resume
Python: validate the paused checkpoint belongs to the target storage
Python: keep one checkpoint lineage in functional workflows
Python: retire functional workflow checkpoints once a run closes
Python: hold the run guard until in-flight checkpoint saves drain
Python: continue the checkpoint chain across a response-only HITL resume
Python: validate the paused checkpoint belongs to the target storage
eavanvalkenburg commentedon Oct 28, 2026
/plan Reassess the remaining functional-workflow checkpoint-lineage race against current main, reusing the design and review evidence in closed, unmerged #8569 rather than starting a competing implementation.
The current per-step callback still builds checkpoints from a shared
ckpt_chain[0]and advances that head after an awaited save, without serializing the read/save/update sequence. Concurrent completed steps can therefore select the same parent. The maintained per-step checkpoint-chain contract and prior maintainer review support addressing this, but a lock-only patch is not yet a complete plan.Material changes since the earlier proposal must be preserved: #9291 changed task settlement at human-input pauses, and merged #9644 fixed the separate save-interruption bug in #9615 by freezing a completed-step snapshot before yielding, tracking dedicated save tasks, shielding in-flight saves and draining them during cleanup. Do not copy the older lifecycle assumptions or regress that fix.
Please:
The report establishes an ancestry problem, not demonstrated output loss or an exactly-once side-effect guarantee. Keep that distinction and storage/process-failure limits explicit. Preserve the current issue owner; do not mark implementation-ready or reopen an old PR merely because planning was requested.