Python: FileCheckpointStorage.save crashes on concurrent saves of the same checkpoint id: fixed .tmp filename race

[ptimizeroracle]

description

FileCheckpointStorage.save uses one fixed temporary filename per checkpoint id: file_path.with_suffix(".json.tmp") in _checkpoint.py (the _write_atomic helper). when two saves of the same checkpoint id run concurrently, both write the same .tmp path and both call os.replace(tmp, final). whichever replace runs first consumes the tmp file, and the other save then dies with a raw FileNotFoundError from os.replace, which is neither WorkflowCheckpointException nor any documented failure of save.

this matters because file storage is the cross-process durability story. two processes resuming the same workflow after a failover or a duplicate delivery can both checkpoint the same id; the storage should serialize or tolerate that, not crash. i could not find any test or doc that pins concurrent-save behavior (checked test_checkpoint.py, the storage conformance suite, and the tracker), so as far as i can tell this surface is simply unpinned.

for the record, the decode side is solid: i also probed the restricted unpickler with a class whose __reduce__ lies about its own reconstruction, and it correctly rejects it with a type-mismatch error. the save path is the weak end.

found via a differential/concurrency probe of the checkpoint storages. ai assistance disclosed. repro is standalone and offline.

reproduction steps

import asyncio, sys, tempfile, uuid
sys.path.insert(0, "python/packages/core")
from agent_framework._workflows._checkpoint import FileCheckpointStorage, WorkflowCheckpoint

async def main():
    with tempfile.TemporaryDirectory() as d:
        fcs = FileCheckpointStorage(d)
        cid = str(uuid.uuid4())
        cps = [
            WorkflowCheckpoint(
                workflow_name="conc", graph_signature_hash="s",
                checkpoint_id=cid, state={"i": i},
            )
            for i in range(20)
        ]
        results = await asyncio.gather(*[fcs.save(cp) for cp in cps], return_exceptions=True)
        n_err = sum(1 for r in results if isinstance(r, Exception))
        print(f"{n_err}/20 concurrent same-id saves failed")
        for r in results:
            if isinstance(r, Exception):
                print(type(r).__name__, str(r)[:90])
                break

asyncio.run(main())

typical output: several of the 20 saves raise FileNotFoundError: [Errno 2] No such file or directory: '.../<id>.json.tmp' -> '.../<id>.json' from os.replace.

suggested direction: give each writer a unique tmp name (pid + uuid suffix or tempfile.mkstemp in the storage dir) and optionally wrap replace failures in WorkflowCheckpointException. happy to implement if that direction works.

environment

• agent-framework-core @ main 8e803b9 (editable, python/packages/core)
• python 3.12, macOS, offline repro

[manjunathshiva]

Your diagnosis is right and it is current — _checkpoint.py:331 on main today is still tmp_path = file_path.with_suffix(".json.tmp"), so two concurrent saves of one id do collide on that path and the loser's os.replace raises a bare FileNotFoundError.

It is already tracked though: #7748 reports the same shared-temp-path race, and #7757 has a fix in review for it — per-save .maf-ckpt-<uuid>.tmp names in the destination directory, plus per-destination serialization so the replaces cannot interleave. That PR has been through a few review rounds, which is why the bug is still reachable on main and why you found it independently rather than from stale code.

Flagging for dedupe rather than because anything you wrote is wrong. Worth reading #7757 if you are interested in this area — the concurrency contract around it turned out to be the interesting part.

[github-actions]

🤖 Automated triage reproduction notes (agent-authored — trust but verify)

Agent analysis

The bug occurs in python/packages/core/agent_framework/_workflows/_checkpoint.py::FileCheckpointStorage.save, specifically _write_atomic at lines 330-334, when concurrent saves use the same checkpoint ID and therefore the same .json.tmp path. Minimal repro: synchronize two save calls immediately before os.replace; one raises FileNotFoundError after the other consumes the shared temporary file.

• Failing test: python/packages/core/tests/workflow/test_checkpoint.py::test_file_checkpoint_storage_concurrent_saves_same_id
• Files examined: python/packages/core/agent_framework/_workflows/_checkpoint.py, python/packages/core/tests/workflow/test_checkpoint.py, python/packages/core/tests/workflow/test_checkpoint_storage_conformance.py, python/packages/core/pyproject.toml
• Tests run: checkpoint test module baseline (46 passed), test_file_checkpoint_storage_concurrent_saves_same_id (failed), standalone 20-save probe (8/20 failed)
• Current version: 1.17.0

[ptimizeroracle]

bookkeeping: open PR #7757 (opened aug 19, predates this issue) already fixes this race with the same unique-temp-name approach, so my duplicate fix PR #8220 was closed in its favor. keeping the repro and the failure evidence here for reference, and the branch stays on my fork if any of the regression tests are useful.

[ptimizeroracle]

housekeeping note: #7757 (open since aug 19, under review) already covers this race with the same fix direction, so this issue should close with that PR rather than stay open.