@@ -7,12 +7,13 @@
 import json
 import logging
 import os
+import threading
 import uuid
 from collections.abc import Mapping
 from dataclasses import dataclass, field, fields
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Protocol, TypeAlias
+from typing import TYPE_CHECKING, Any, ClassVar, Protocol, TypeAlias
 
 from ..exceptions import WorkflowCheckpointException
 
@@ -272,6 +273,11 @@ class FileCheckpointStorage:
         )
     """
 
+    _DELETE_LOCK_STRIPE_COUNT: ClassVar[int] = 64
+    _DELETE_LOCKS: ClassVar[tuple[threading.Lock, ...]] = tuple(
+        threading.Lock() for _ in range(_DELETE_LOCK_STRIPE_COUNT)
+    )
+
     def __init__(
         self,
         storage_path: str | Path,
@@ -312,6 +318,11 @@ def _validate_file_path(self, checkpoint_id: CheckpointID) -> Path:
             raise WorkflowCheckpointException(f"Invalid checkpoint ID: {checkpoint_id}")
         return file_path
 
+    @classmethod
+    def _delete_lock(cls, file_path: Path) -> threading.Lock:
+        """Return the process-local deletion lock for a checkpoint file."""
+        return cls._DELETE_LOCKS[hash(file_path) % cls._DELETE_LOCK_STRIPE_COUNT]
+
     async def save(self, checkpoint: WorkflowCheckpoint) -> CheckpointID:
         """Save a checkpoint and return its ID.
 
@@ -412,12 +423,14 @@ async def delete(self, checkpoint_id: CheckpointID) -> bool:
             True if the checkpoint was successfully deleted, False if no checkpoint with the given ID exists.
         """
         file_path = self._validate_file_path(checkpoint_id)
+        file_lock = self._delete_lock(file_path)
 
         def _delete() -> bool:
-            try:
-                file_path.unlink()
-            except FileNotFoundError:
-                return False
+            with file_lock:
+                try:
+                    file_path.unlink()
+                except FileNotFoundError:
+                    return False
             logger.info(f"Deleted checkpoint {checkpoint_id} from {file_path}")
             return True
 
 
@@ -1262,8 +1262,10 @@ async def test_file_checkpoint_storage_delete():
 
 
 async def test_file_checkpoint_storage_concurrent_delete():
+    """Serialize deletes when overlapping unlink calls could both report success."""
     with tempfile.TemporaryDirectory() as temp_dir:
         storage = FileCheckpointStorage(temp_dir)
+        other_storage = FileCheckpointStorage(temp_dir)
         checkpoint = WorkflowCheckpoint(
             workflow_name="test-workflow",
             graph_signature_hash="test-hash",
@@ -1272,19 +1274,44 @@ async def test_file_checkpoint_storage_concurrent_delete():
         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):
+        original_to_thread = asyncio.to_thread
+        original_unlink = Path.unlink
+        worker_barrier = threading.Barrier(2)
+        unlink_barrier = threading.Barrier(2)
+        unlink_guard = threading.Lock()
+        overlapping_delete_completed = False
+
+        async def synchronized_to_thread(function: Any, /, *args: Any, **kwargs: Any) -> Any:
+            def synchronized_call() -> Any:
+                worker_barrier.wait(timeout=5)
+                return function(*args, **kwargs)
+
+            return await original_to_thread(synchronized_call)
+
+        def macos_style_unlink(path: Path, missing_ok: bool = False) -> None:
+            nonlocal overlapping_delete_completed
+            if path.resolve() != file_path:
+                original_unlink(path, missing_ok=missing_ok)
+                return
+
+            try:
+                unlink_barrier.wait(timeout=0.5)
+            except threading.BrokenBarrierError:
+                original_unlink(path, missing_ok=missing_ok)
+                return
+
+            with unlink_guard:
+                if not overlapping_delete_completed:
+                    original_unlink(path, missing_ok=missing_ok)
+                    overlapping_delete_completed = True
+
+        with (
+            patch.object(asyncio, "to_thread", synchronized_to_thread),
+            patch.object(Path, "unlink", macos_style_unlink),
+        ):
             results = await asyncio.gather(
                 storage.delete(checkpoint.checkpoint_id),
-                storage.delete(checkpoint.checkpoint_id),
+                other_storage.delete(checkpoint.checkpoint_id),
             )
 
         assert sorted(results) == [False, True]