fix(sync): re-verify planned deletes so concurrent writes survive scan reconciliation

Scan delete plans compare a storage snapshot against a later DB read
(deleted = all_db_paths - storage_paths). A note accepted and materialized
between snapshot capture and the DB read was planned as deleted, and the
delete batch destroyed its entity, search, and vector rows with no
re-verification.

apply_project_index_delete_batch now re-confirms each planned path through
a ProjectIndexDeletePathVerifier capability before touching the database.
The local runtime injects LocalProjectIndexDeletePathVerifier, which
probes the filesystem: a path present again (or whose probe fails) is
skipped and logged — the next scan picks the file up as modified. Cloud/S3
runtimes keep current behavior through the explicit
TrustPlannedProjectIndexDeleteVerifier pass-through default, since their
storage listing is authoritative and they have no cheap per-path probe at
apply time. Skipped paths are reported on the batch result and delete run.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/index/local_project.py

@@ -57,6 +57,7 @@
     run_project_index_coordinator,
 )
 from basic_memory.indexing.project_index_maintenance import (
+    ProjectIndexDeletePathVerifier,
     ProjectIndexMaintenanceRunner,
     ProjectIndexMovedEntitySearchRefresher,
     RepositoryProjectIndexMaintenanceStore,
@@ -392,6 +393,34 @@ def _reuse_indexed_checksum(
         return indexed.checksum
 
 
+@dataclass(frozen=True, slots=True)
+class LocalProjectIndexDeletePathVerifier(ProjectIndexDeletePathVerifier):
+    """Probe the local filesystem before applying scan-planned index deletes."""
+
+    file_service: FileService
+
+    async def confirm_deleted_paths(self, paths: Sequence[str]) -> frozenset[str]:
+        confirmed_paths: set[str] = set()
+        for path in paths:
+            # Trigger: the existence probe itself fails (permission/mount error).
+            # Why: without a positive confirmation of absence, deleting would
+            # turn a transient storage error into destroyed entity rows.
+            # Outcome: leave the path out of the confirmed set; the next scan
+            # reconciles it once the probe works again.
+            try:
+                still_absent = not await self.file_service.exists(path)
+            except FileOperationError as exc:
+                logger.warning(
+                    "Skipping planned index delete: existence probe failed",
+                    path=path,
+                    error=str(exc),
+                )
+                continue
+            if still_absent:
+                confirmed_paths.add(path)
+        return frozenset(confirmed_paths)
+
+
 @dataclass(frozen=True, slots=True)
 class LocalProjectIndexRuntime:
     """Dependencies for running project-wide local indexing through core fanout."""
@@ -558,6 +587,9 @@ def runtime_from_dependencies(
                 entity_service=dependencies.entity_service,
                 file_service=dependencies.file_service,
             ),
+            delete_path_verifier=LocalProjectIndexDeletePathVerifier(
+                file_service=dependencies.file_service,
+            ),
         )
         return LocalProjectIndexRuntime(
             observed_file_source=LocalProjectIndexObservedFileSource(

src/basic_memory/indexing/project_index_maintenance.py

@@ -59,6 +59,34 @@ async def refresh_moved_entities(self, entity_ids: Sequence[int]) -> None:
         """Refresh search rows for moved entity ids."""
 
 
+class ProjectIndexDeletePathVerifier(Protocol):
+    """Capability that re-confirms scan-planned delete paths at apply time.
+
+    Delete plans come from a storage snapshot that is stale by the time the
+    batch applies; a note accepted and materialized after the snapshot would
+    otherwise be destroyed. Implementations return only the paths whose
+    absence they can positively confirm right now.
+    """
+
+    async def confirm_deleted_paths(self, paths: Sequence[str]) -> frozenset[str]:
+        """Return the subset of paths confirmed absent from storage."""
+        ...
+
+
+@dataclass(frozen=True, slots=True)
+class TrustPlannedProjectIndexDeleteVerifier(ProjectIndexDeletePathVerifier):
+    """Confirm every planned delete without re-probing storage.
+
+    Cloud/S3 runtimes treat the scan's storage listing as authoritative and
+    have no cheap per-path existence probe at apply time, so they keep the
+    plan's verdict unchanged. Runtimes with a live filesystem (local) inject
+    a probing verifier instead.
+    """
+
+    async def confirm_deleted_paths(self, paths: Sequence[str]) -> frozenset[str]:
+        return frozenset(paths)
+
+
 @dataclass(frozen=True, slots=True)
 class ProjectIndexMovedFile:
     """One indexed file move that may need storage-backed metadata repair."""
@@ -214,6 +242,7 @@ class ProjectIndexDeleteBatchResult:
     deleted_entities: int
     relation_cleanup_entity_ids: frozenset[int] = frozenset()
     missing_paths: tuple[str, ...] = ()
+    skipped_paths: tuple[str, ...] = ()
 
 
 @dataclass(frozen=True, slots=True)
@@ -241,6 +270,13 @@ def missing_paths(self) -> tuple[str, ...]:
             missing_path for record in self.records for missing_path in record.result.missing_paths
         )
 
+    @property
+    def skipped_paths(self) -> tuple[str, ...]:
+        """Return every planned delete path skipped because it is present again."""
+        return tuple(
+            skipped_path for record in self.records for skipped_path in record.result.skipped_paths
+        )
+
 
 DELETE_PROJECT_INDEX_SEARCH_ROWS_SQL = text("""
     DELETE FROM search_index
@@ -414,6 +450,7 @@ class RepositoryProjectIndexMaintenanceStore:
     session_maker: async_sessionmaker[AsyncSession]
     project_id: ProjectId
     move_content_updater: ProjectIndexMoveContentUpdater | None = None
+    delete_path_verifier: ProjectIndexDeletePathVerifier = TrustPlannedProjectIndexDeleteVerifier()
 
     async def apply_project_index_move_batch(
         self,
@@ -599,19 +636,51 @@ async def apply_project_index_delete_batch(
         if not delete_batch.paths:
             return ProjectIndexDeleteBatchResult(deleted_entities=0)
 
+        # Trigger: a planned delete path exists in storage again at apply time.
+        # Why: the plan compares a storage snapshot against a later DB read, so
+        #      a note accepted and materialized in between is planned as deleted;
+        #      applying it would destroy the accepted entity, search, and vector
+        #      rows with no recovery.
+        # Outcome: only positively re-confirmed absences are deleted; skipped
+        #          paths are reported and the next scan picks the file up as
+        #          modified.
+        confirmed_paths = await self.delete_path_verifier.confirm_deleted_paths(
+            delete_batch.paths
+        )
+        skipped_paths = tuple(
+            deleted_path
+            for deleted_path in delete_batch.paths
+            if deleted_path not in confirmed_paths
+        )
+        if skipped_paths:
+            logger.warning(
+                "Skipping planned index deletes for paths present in storage again",
+                paths=skipped_paths,
+            )
+        if not confirmed_paths:
+            return ProjectIndexDeleteBatchResult(
+                deleted_entities=0,
+                skipped_paths=skipped_paths,
+            )
+
         async with db.scoped_session(self.session_maker) as session:
             target_result = await session.execute(
                 select(Entity.id, Entity.file_path).where(
                     Entity.project_id == self.project_id,
-                    Entity.file_path.in_(tuple(delete_batch.paths)),
+                    Entity.file_path.in_(tuple(confirmed_paths)),
                 )
             )
             target_rows = target_result.mappings().all()
 
             if not target_rows:
                 return ProjectIndexDeleteBatchResult(
                     deleted_entities=0,
-                    missing_paths=tuple(delete_batch.paths),
+                    missing_paths=tuple(
+                        deleted_path
+                        for deleted_path in delete_batch.paths
+                        if deleted_path in confirmed_paths
+                    ),
+                    skipped_paths=skipped_paths,
                 )
 
             deleted_entity_ids = tuple(int(row["id"]) for row in target_rows)
@@ -629,8 +698,9 @@ async def apply_project_index_delete_batch(
             missing_paths=tuple(
                 deleted_path
                 for deleted_path in delete_batch.paths
-                if deleted_path not in deleted_found_paths
+                if deleted_path in confirmed_paths and deleted_path not in deleted_found_paths
             ),
+            skipped_paths=skipped_paths,
         )
 
 

tests/index/test_local_project_index.py

@@ -18,6 +18,7 @@
 from basic_memory.index.local_project import (
     IndexedFileStat,
     LocalProjectIndexBatchEnqueuer,
+    LocalProjectIndexDeletePathVerifier,
     LocalProjectIndexObservedFileSource,
     LocalProjectIndexRuntime,
     LocalProjectIndexRuntimeFactory,
@@ -666,6 +667,104 @@ async def test_local_project_index_reads_current_file_when_file_changes_after_ob
     assert "This is the modified content" in note_content.markdown_content
 
 
+async def test_local_project_index_delete_path_verifier_confirms_only_probed_absent_paths(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    """Only a positive absence probe confirms a planned delete; probe failures skip."""
+    (tmp_path / "present.md").write_bytes(b"# Present\n")
+
+    file_service = FileService(tmp_path)
+    original_exists = file_service.exists
+
+    async def flaky_exists(path):
+        if str(path).endswith("flaky.md"):
+            raise FileOperationError("mount error")
+        return await original_exists(path)
+
+    monkeypatch.setattr(file_service, "exists", flaky_exists)
+
+    verifier = LocalProjectIndexDeletePathVerifier(file_service=file_service)
+    confirmed = await verifier.confirm_deleted_paths(("present.md", "absent.md", "flaky.md"))
+
+    assert confirmed == frozenset({"absent.md"})
+
+
+@dataclass(slots=True)
+class RecreatingObservedFileSource:
+    """Recreate a file after the scan snapshot, simulating a concurrent accepted write."""
+
+    source: ProjectIndexObservedFileSource
+    file_path: Path
+    content: bytes
+
+    async def list_observed_index_files(self) -> tuple[RuntimeObservedIndexFile, ...]:
+        observed = await self.source.list_observed_index_files()
+        self.file_path.write_bytes(self.content)
+        return observed
+
+
+async def test_local_project_index_delete_batch_spares_file_recreated_after_snapshot(
+    test_project: Project,
+    project_config,
+    entity_repository,
+    session_maker: async_sessionmaker[AsyncSession],
+    config_manager,
+    monkeypatch,
+) -> None:
+    """An entity whose file reappears after the scan snapshot must survive delete batches."""
+    del config_manager
+
+    note_path = project_config.home / "late.md"
+    note_content = b"# Late\n\nAccepted while the scan was running.\n"
+    note_path.write_bytes(note_content)
+
+    first = await run_local_project_index_for_project(
+        test_project,
+        runtime_factory=LocalProjectIndexRuntimeFactory(batch_size=10),
+        force_full=True,
+    )
+    assert first.enqueued_files == 1
+
+    async with db.scoped_session(session_maker) as session:
+        entity_before = await entity_repository.get_by_file_path(session, "late.md")
+    assert entity_before is not None
+
+    # The file is gone when the snapshot is captured, then written again before
+    # the delete batch applies — the write_note accept/materialize race.
+    note_path.unlink()
+    runtime_factory = LocalProjectIndexRuntimeFactory(batch_size=10)
+    runtime = await runtime_factory.runtime_for_project(test_project)
+    racing_runtime = LocalProjectIndexRuntime(
+        observed_file_source=RecreatingObservedFileSource(
+            source=runtime.observed_file_source,
+            file_path=note_path,
+            content=note_content,
+        ),
+        change_detector=runtime.change_detector,
+        maintenance_runner=runtime.maintenance_runner,
+        moved_entity_search_refresher=runtime.moved_entity_search_refresher,
+        batch_enqueuer=runtime.batch_enqueuer,
+        completion_relation_runtime=runtime.completion_relation_runtime,
+        batch_size=runtime.batch_size,
+    )
+
+    second = await run_local_project_index(
+        RuntimeProjectIndexJobRequest(
+            project=ProjectRuntimeReference.from_project(test_project),
+        ),
+        runtime=racing_runtime,
+    )
+
+    assert second.deleted_files == 0
+
+    async with db.scoped_session(session_maker) as session:
+        entity_after = await entity_repository.get_by_file_path(session, "late.md")
+
+    assert entity_after is not None
+    assert entity_after.id == entity_before.id
+
+
 @dataclass(slots=True)
 class RecordingObservedFileSource:
     observed_files: tuple[RuntimeObservedIndexFile, ...]