Python: [Bug]: Parallel functional steps fork the checkpoint lineage

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.