Python: fix: make FileCheckpointStorage tmp filename unique per save

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)
 

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