Python: handle concurrent FileCheckpointStorage deletion

python/packages/core/agent_framework/_workflows/_checkpoint.py

@@ -414,11 +414,12 @@ async def delete(self, checkpoint_id: CheckpointID) -> bool:
         file_path = self._validate_file_path(checkpoint_id)
 
         def _delete() -> bool:
-            if file_path.exists():
+            try:
                 file_path.unlink()
-                logger.info(f"Deleted checkpoint {checkpoint_id} from {file_path}")
-                return True
-            return False
+            except FileNotFoundError:
+                return False
+            logger.info(f"Deleted checkpoint {checkpoint_id} from {file_path}")
+            return True
 
         return await asyncio.to_thread(_delete)
 

python/packages/core/tests/workflow/test_checkpoint.py

@@ -1,11 +1,14 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+import asyncio
 import json
 import tempfile
+import threading
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
+from unittest.mock import patch
 
 import pytest
 
@@ -1258,6 +1261,35 @@ async def test_file_checkpoint_storage_delete():
         assert result is False
 
 
+async def test_file_checkpoint_storage_concurrent_delete():
+    with tempfile.TemporaryDirectory() as temp_dir:
+        storage = FileCheckpointStorage(temp_dir)
+        checkpoint = WorkflowCheckpoint(
+            workflow_name="test-workflow",
+            graph_signature_hash="test-hash",
+            checkpoint_id="same",
+        )
+        await storage.save(checkpoint)
+
+        file_path = (Path(temp_dir) / "same.json").resolve()
+        original_exists = Path.exists
+        barrier = threading.Barrier(2)
+
+        def synchronized_exists(path: Path) -> bool:
+            result = original_exists(path)
+            if path.resolve() == file_path:
+                barrier.wait(timeout=5)
+            return result
+
+        with patch.object(Path, "exists", synchronized_exists):
+            results = await asyncio.gather(
+                storage.delete(checkpoint.checkpoint_id),
+                storage.delete(checkpoint.checkpoint_id),
+            )
+
+        assert sorted(results) == [False, True]
+
+
 async def test_file_checkpoint_storage_directory_creation():
     with tempfile.TemporaryDirectory() as temp_dir:
         nested_path = Path(temp_dir) / "nested" / "checkpoint" / "storage"