diff --git a/python/packages/core/agent_framework/_workflows/_checkpoint.py b/python/packages/core/agent_framework/_workflows/_checkpoint.py index d9b30d8391..9264380b0d 100644 --- a/python/packages/core/agent_framework/_workflows/_checkpoint.py +++ b/python/packages/core/agent_framework/_workflows/_checkpoint.py @@ -328,10 +328,20 @@ async def save(self, checkpoint: WorkflowCheckpoint) -> CheckpointID: encoded_checkpoint = encode_checkpoint_value(checkpoint_dict) def _write_atomic() -> None: - tmp_path = file_path.with_suffix(".json.tmp") - with open(tmp_path, "w") as f: - json.dump(encoded_checkpoint, f, indent=2, ensure_ascii=False) - os.replace(tmp_path, file_path) + # Short, id-independent temp name: embeds no checkpoint id, so a + # checkpoint id accepted by _validate_file_path can never push the + # temp name over the filesystem's filename-length limit. + tmp_path = file_path.with_name(f".maf-ckpt-{uuid.uuid4().hex}.tmp") + try: + with open(tmp_path, "w") as f: + json.dump(encoded_checkpoint, f, indent=2, ensure_ascii=False) + os.replace(tmp_path, file_path) + finally: + # best-effort cleanup if replace failed or the write raised + try: + tmp_path.unlink(missing_ok=True) + except OSError: + pass await asyncio.to_thread(_write_atomic) diff --git a/python/packages/core/tests/workflow/test_checkpoint.py b/python/packages/core/tests/workflow/test_checkpoint.py index be8de7c13b..b3bdb6e21c 100644 --- a/python/packages/core/tests/workflow/test_checkpoint.py +++ b/python/packages/core/tests/workflow/test_checkpoint.py @@ -1843,4 +1843,20 @@ async def test_file_checkpoint_storage_roundtrip_empty_collections(): assert loaded.pending_request_info_events == {} +async def test_file_checkpoint_storage_concurrent_saves_same_id(tmp_path): + """Concurrent saves of the same checkpoint id must not race on a shared tmp file.""" + import asyncio + + storage = FileCheckpointStorage(tmp_path) + checkpoint = WorkflowCheckpoint(workflow_name="test-workflow", graph_signature_hash="test-hash") + + await asyncio.gather(*(storage.save(checkpoint) for _ in range(8))) + + loaded = await storage.load(checkpoint.checkpoint_id) + assert loaded.checkpoint_id == checkpoint.checkpoint_id + assert loaded.workflow_name == checkpoint.workflow_name + assert loaded.state == checkpoint.state + assert list(Path(storage.storage_path).glob("*.tmp")) == [] + + # endregion