fix(core): tolerate disappearing guarded files

src/basic_memory/index/local_notes.py

@@ -180,7 +180,14 @@ async def exists(self, path: RuntimeFilePath) -> bool:
         return await self.file_service.exists(path)
 
     async def compute_checksum(self, path: RuntimeFilePath) -> RuntimeFileChecksum:
-        return await self.file_service.compute_checksum(path)
+        try:
+            return await self.file_service.compute_checksum(path)
+        except FileError as exc:
+            if isinstance(exc.__context__, FileNotFoundError):
+                # Directory cleanup uses the portable guard, where deletion after
+                # the existence probe is the same absent-file outcome.
+                raise FileNotFoundError(path) from exc
+            raise
 
     async def delete_file_if_unchanged(
         self,
@@ -193,7 +200,12 @@ async def delete_file_if_unchanged(
         # guarantee portable: never delete an object that no longer matches (basic-memory-cloud#1618).
         if not await self.file_service.exists(path):
             return False
-        if await self.file_service.compute_checksum(path) != expected_checksum:
+        try:
+            actual_checksum = await self.compute_checksum(path)
+        except FileNotFoundError:
+            # Disappearance after the final existence probe is another safe no-delete outcome.
+            return False
+        if actual_checksum != expected_checksum:
             return False
         await self.file_service.delete_file(path)
         return True

src/basic_memory/index/note_content_materialization.py

@@ -532,7 +532,14 @@ async def exists(self, path: RuntimeFilePath) -> bool:
         return await self.file_service.exists(path)
 
     async def compute_checksum(self, path: RuntimeFilePath) -> RuntimeFileChecksum:
-        return await self.file_service.compute_checksum(path)
+        try:
+            return await self.file_service.compute_checksum(path)
+        except file_utils.FileError as exc:
+            if isinstance(exc.__context__, FileNotFoundError):
+                # The general FileService contract reports I/O failures as FileError,
+                # while runtime guards need concurrent deletion as an absent file.
+                raise FileNotFoundError(path) from exc
+            raise
 
     async def delete_file(self, path: RuntimeFilePath) -> None:
         await self.file_service.delete_file(path)
@@ -548,7 +555,12 @@ async def delete_file_if_unchanged(
         # atomic precondition; the race is negligible on a filesystem.
         if not await self.file_service.exists(path):
             return False
-        if await self.file_service.compute_checksum(path) != expected_checksum:
+        try:
+            actual_checksum = await self.compute_checksum(path)
+        except FileNotFoundError:
+            # Disappearance after the final existence probe is another safe no-delete outcome.
+            return False
+        if actual_checksum != expected_checksum:
             return False
         await self.file_service.delete_file(path)
         return True

src/basic_memory/runtime/note_file_guards.py

@@ -61,7 +61,12 @@ async def read_runtime_file_checksum(
     """Return the current runtime file checksum, or None when absent."""
     if not await reader.exists(file_path):
         return None
-    return await reader.compute_checksum(file_path)
+    try:
+        return await reader.compute_checksum(file_path)
+    except FileNotFoundError:
+        # Object stores cannot make exists-plus-read atomic. A concurrent delete
+        # after the probe has the same domain meaning as an initially absent file.
+        return None
 
 
 def runtime_file_conflict(

tests/runtime/test_note_file_guard_races.py

@@ -0,0 +1,115 @@
+"""Race regressions for portable note-file concurrency guards."""
+
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from basic_memory.file_utils import FileError
+from basic_memory.index.local_notes import LocalNoteFileDeleteStorage
+from basic_memory.index.note_content_materialization import LocalNoteContentStorage
+from basic_memory.indexing.note_file_delete_runner import run_note_file_delete
+from basic_memory.runtime.cleanup import RuntimeDeleteStatus, RuntimeNoteFileDeleteJobRequest
+from basic_memory.runtime.note_file_guards import read_runtime_file_checksum
+from basic_memory.services.file_service import FileService
+
+
+async def test_checksum_read_treats_post_probe_deletion_as_absent(tmp_path: Path) -> None:
+    """A disappearing object should not poison a durable materialization retry."""
+    file_service = FileService(tmp_path)
+    storage = LocalNoteContentStorage(file_service)
+
+    # The file disappears after storage reports it present but before checksum I/O.
+    with patch.object(file_service, "exists", AsyncMock(return_value=True)):
+        checksum = await read_runtime_file_checksum(storage, "notes/disappeared.md")
+
+    assert checksum is None
+
+
+async def test_directory_delete_converges_when_file_disappears_before_delete(
+    tmp_path: Path,
+) -> None:
+    """The final guarded checksum should treat a vanished target as a safe no-delete."""
+    file_service = FileService(tmp_path)
+    file_path = "notes/disappeared.md"
+    await file_service.write_file(file_path, "# Disappearing note\n")
+    accepted_checksum = await file_service.compute_checksum(file_path)
+    original_compute_checksum = file_service.compute_checksum
+    checksum_calls = 0
+
+    async def delete_before_final_checksum(path: str) -> str:
+        nonlocal checksum_calls
+        checksum_calls += 1
+        if checksum_calls == 2:
+            (tmp_path / path).unlink()
+        return await original_compute_checksum(path)
+
+    with patch.object(file_service, "compute_checksum", side_effect=delete_before_final_checksum):
+        result = await run_note_file_delete(
+            RuntimeNoteFileDeleteJobRequest(
+                project_id=101,
+                entity_id=42,
+                file_path=file_path,
+                file_checksum=accepted_checksum,
+            ),
+            storage=LocalNoteFileDeleteStorage(file_service),
+        )
+
+    assert result.status == RuntimeDeleteStatus.skipped
+    assert result.reason == f"file changed before delete: {file_path}"
+    assert not (tmp_path / file_path).exists()
+
+
+async def test_note_delete_converges_when_file_disappears_before_delete(
+    tmp_path: Path,
+) -> None:
+    """Ordinary note cleanup should share the safe final-checksum outcome."""
+    file_service = FileService(tmp_path)
+    file_path = "notes/disappeared.md"
+    await file_service.write_file(file_path, "# Disappearing note\n")
+    accepted_checksum = await file_service.compute_checksum(file_path)
+    original_compute_checksum = file_service.compute_checksum
+    checksum_calls = 0
+
+    async def delete_before_final_checksum(path: str) -> str:
+        nonlocal checksum_calls
+        checksum_calls += 1
+        if checksum_calls == 2:
+            (tmp_path / path).unlink()
+        return await original_compute_checksum(path)
+
+    with patch.object(file_service, "compute_checksum", side_effect=delete_before_final_checksum):
+        result = await run_note_file_delete(
+            RuntimeNoteFileDeleteJobRequest(
+                project_id=101,
+                entity_id=42,
+                file_path=file_path,
+                file_checksum=accepted_checksum,
+            ),
+            storage=LocalNoteContentStorage(file_service),
+        )
+
+    assert result.status == RuntimeDeleteStatus.skipped
+    assert result.reason == f"file changed before delete: {file_path}"
+    assert not (tmp_path / file_path).exists()
+
+
+async def test_direct_checksum_preserves_file_service_error_contract(tmp_path: Path) -> None:
+    """Direct callers still receive FileError when the checksum source is absent."""
+    file_service = FileService(tmp_path)
+
+    with pytest.raises(FileError):
+        await file_service.compute_checksum("notes/disappeared.md")
+
+
+async def test_directory_delete_checksum_treats_post_probe_deletion_as_absent(
+    tmp_path: Path,
+) -> None:
+    """Directory cleanup should converge when its target disappears before checksum I/O."""
+    file_service = FileService(tmp_path)
+    storage = LocalNoteFileDeleteStorage(file_service)
+
+    with patch.object(file_service, "exists", AsyncMock(return_value=True)):
+        checksum = await read_runtime_file_checksum(storage, "notes/disappeared.md")
+
+    assert checksum is None