fix(sync): stop scan move batches deleting concurrently created destination entities

Scan change planning only pairs a move with a destination that had no DB
row at snapshot time, so an entity found at the destination during apply
was created concurrently (e.g. an accepted write_note whose content may
not be materialized yet). apply_project_index_move_batch deleted that
entity unconditionally and re-pointed the stale source entity onto the
path, destroying the accepted content.

Scan runtimes now opt into verify_replaced_move_targets on the maintenance
store: a replacement row is only deleted when its checksum equals the
moved entity's indexed checksum — the checksum the move was planned
against — which proves it merely indexes the moved bytes (a racing event
index of the new path, a safe dedupe). Any other replacement drops that
move from the batch with a warning; nothing is deleted or re-pointed, and
the next scan reconciles the stale source path as deleted-or-present.
Dropped source paths are reported on the batch result and move run.

The unconditional-replace branch stays the default because the watcher
flow legitimately moves onto an existing indexed file (mv over an indexed
path) where the pre-existing destination row must be replaced; cloud scan
runtimes also keep current behavior.

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

@@ -590,6 +590,10 @@ def runtime_from_dependencies(
             delete_path_verifier=LocalProjectIndexDeletePathVerifier(
                 file_service=dependencies.file_service,
             ),
+            # Scan-planned moves guarantee the destination had no DB row at
+            # snapshot time, so a replacement row found at apply time is a
+            # concurrent creation that must be checksum-verified before deletion.
+            verify_replaced_move_targets=True,
         )
         return LocalProjectIndexRuntime(
             observed_file_source=LocalProjectIndexObservedFileSource(

src/basic_memory/indexing/project_index_maintenance.py

@@ -7,7 +7,7 @@
 from typing import Protocol
 
 from loguru import logger
-from sqlalchemy import bindparam, case, column, delete, select, table, text, update
+from sqlalchemy import RowMapping, bindparam, case, column, delete, select, table, text, update
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from basic_memory import db
@@ -169,6 +169,7 @@ class ProjectIndexMoveBatchResult:
     replaced_entity_ids: frozenset[int] = frozenset()
     relation_cleanup_entity_ids: frozenset[int] = frozenset()
     missing_paths: tuple[str, ...] = ()
+    dropped_move_paths: tuple[str, ...] = ()
 
 
 @dataclass(frozen=True, slots=True)
@@ -198,6 +199,15 @@ def missing_paths(self) -> tuple[str, ...]:
             missing_path for record in self.records for missing_path in record.result.missing_paths
         )
 
+    @property
+    def dropped_move_paths(self) -> tuple[str, ...]:
+        """Return every move source path dropped because its destination changed."""
+        return tuple(
+            dropped_path
+            for record in self.records
+            for dropped_path in record.result.dropped_move_paths
+        )
+
 
 @dataclass(frozen=True, slots=True)
 class ProjectIndexDeleteBatch:
@@ -451,6 +461,16 @@ class RepositoryProjectIndexMaintenanceStore:
     project_id: ProjectId
     move_content_updater: ProjectIndexMoveContentUpdater | None = None
     delete_path_verifier: ProjectIndexDeletePathVerifier = TrustPlannedProjectIndexDeleteVerifier()
+    # Trigger: an entity occupies a move destination at apply time.
+    # Why: scan change planning only pairs moves with paths that had no DB row
+    #      at snapshot time, so a row found there was created concurrently and
+    #      may carry accepted-but-unmaterialized content; the watcher flow, by
+    #      contrast, legitimately moves onto an existing indexed file and must
+    #      keep replacing it unconditionally.
+    # Outcome: scan runtimes set this True so a replacement is only deleted
+    #          when its checksum proves it indexes the moved bytes; mismatches
+    #          drop the move for the next scan to reconcile.
+    verify_replaced_move_targets: bool = False
 
     async def apply_project_index_move_batch(
         self,
@@ -466,20 +486,79 @@ async def apply_project_index_move_batch(
 
         async with db.scoped_session(self.session_maker) as session:
             existing_paths_result = await session.execute(
-                select(Entity.id, Entity.file_path, Entity.permalink).where(
+                select(Entity.id, Entity.file_path, Entity.permalink, Entity.checksum).where(
                     Entity.project_id == self.project_id,
                     Entity.file_path.in_(old_paths),
                 )
             )
-            target_rows = existing_paths_result.mappings().all()
+            target_rows = list(existing_paths_result.mappings().all())
+            replaced_entity_ids: frozenset[int] = frozenset()
+            relation_cleanup_entity_ids: frozenset[int] = frozenset()
+            dropped_move_paths: tuple[str, ...] = ()
+            content_updates_by_entity_id: dict[int, ProjectIndexMovedFileContentUpdate] = {}
+            replacement_rows: list[RowMapping] = []
+
+            if target_rows:
+                new_paths = tuple(
+                    sorted({target_paths_by_old_path[str(row["file_path"])] for row in target_rows})
+                )
+                replacement_result = await session.execute(
+                    select(Entity.id, Entity.file_path, Entity.checksum).where(
+                        Entity.project_id == self.project_id,
+                        Entity.file_path.in_(new_paths),
+                        Entity.id.not_in(tuple(int(row["id"]) for row in target_rows)),
+                    )
+                )
+                replacement_rows = list(replacement_result.mappings().all())
+
+            if self.verify_replaced_move_targets and replacement_rows:
+                old_path_by_new_path = {
+                    target_paths_by_old_path[str(row["file_path"])]: str(row["file_path"])
+                    for row in target_rows
+                }
+                # The move was planned by matching the destination file's checksum
+                # to the source entity's indexed checksum, so that checksum is the
+                # only content a replacement row may legitimately index.
+                expected_checksum_by_new_path = {
+                    target_paths_by_old_path[str(row["file_path"])]: row["checksum"]
+                    for row in target_rows
+                }
+                verified_replacement_rows: list[RowMapping] = []
+                dropped_new_paths: set[str] = set()
+                for replacement_row in replacement_rows:
+                    replacement_path = str(replacement_row["file_path"])
+                    expected_checksum = expected_checksum_by_new_path.get(replacement_path)
+                    if (
+                        expected_checksum is not None
+                        and replacement_row["checksum"] == expected_checksum
+                    ):
+                        verified_replacement_rows.append(replacement_row)
+                        continue
+                    dropped_new_paths.add(replacement_path)
+                    logger.warning(
+                        "Dropping planned move: destination holds a concurrently created entity",
+                        old_path=old_path_by_new_path.get(replacement_path),
+                        new_path=replacement_path,
+                    )
+                replacement_rows = verified_replacement_rows
+                if dropped_new_paths:
+                    dropped_move_paths = tuple(
+                        sorted(
+                            old_path_by_new_path[new_path] for new_path in dropped_new_paths
+                        )
+                    )
+                    target_rows = [
+                        row
+                        for row in target_rows
+                        if target_paths_by_old_path[str(row["file_path"])]
+                        not in dropped_new_paths
+                    ]
+
             updated_old_paths = frozenset(str(row["file_path"]) for row in target_rows)
             target_paths_by_entity_id = {
                 int(row["id"]): target_paths_by_old_path[str(row["file_path"])]
                 for row in target_rows
             }
-            replaced_entity_ids: frozenset[int] = frozenset()
-            relation_cleanup_entity_ids: frozenset[int] = frozenset()
-            content_updates_by_entity_id: dict[int, ProjectIndexMovedFileContentUpdate] = {}
             if self.move_content_updater is not None:
                 for row in target_rows:
                     entity_id = int(row["id"])
@@ -499,15 +578,6 @@ async def apply_project_index_move_batch(
                         content_updates_by_entity_id[entity_id] = content_update
 
             if updated_old_paths:
-                new_paths = tuple(sorted(set(target_paths_by_entity_id.values())))
-                replacement_result = await session.execute(
-                    select(Entity.id, Entity.file_path).where(
-                        Entity.project_id == self.project_id,
-                        Entity.file_path.in_(new_paths),
-                        Entity.id.not_in(tuple(target_paths_by_entity_id)),
-                    )
-                )
-                replacement_rows = replacement_result.mappings().all()
                 replaced_entity_ids = frozenset(int(row["id"]) for row in replacement_rows)
                 relation_cleanup_entity_ids = await delete_project_index_entities(
                     session,
@@ -620,13 +690,15 @@ async def apply_project_index_move_batch(
             move_target.old_path
             for move_target in move_batch.targets
             if move_target.old_path not in updated_old_paths
+            and move_target.old_path not in dropped_move_paths
         )
         return ProjectIndexMoveBatchResult(
             updated_files=len(updated_old_paths),
             moved_entity_ids=frozenset(target_paths_by_entity_id),
             replaced_entity_ids=replaced_entity_ids,
             relation_cleanup_entity_ids=relation_cleanup_entity_ids,
             missing_paths=missing_paths,
+            dropped_move_paths=dropped_move_paths,
         )
 
     async def apply_project_index_delete_batch(

tests/index/test_local_project_index.py

@@ -44,7 +44,10 @@
     IndexedEntity,
     IndexingBatchResult,
 )
-from basic_memory.indexing.project_index_coordinator import ProjectIndexObservedFileSource
+from basic_memory.indexing.project_index_coordinator import (
+    ProjectIndexChangeDetector,
+    ProjectIndexObservedFileSource,
+)
 from basic_memory.indexing.project_index_maintenance import (
     ProjectIndexDeleteRun,
     ProjectIndexMoveRun,
@@ -56,6 +59,7 @@
     UnresolvedRelation,
 )
 from basic_memory.models import Entity, Project
+from basic_memory.repository import EntityRepository
 from basic_memory.repository.note_content_repository import NoteContentRepository
 from basic_memory.runtime.jobs import (
     RuntimeIndexFileBatchJobRequest,
@@ -1407,6 +1411,134 @@ async def test_local_project_index_treats_rename_over_existing_path_as_modify_an
     assert results[0].file_path == "note.md"
 
 
+@dataclass(slots=True)
+class ConcurrentDestinationWriteChangeDetector:
+    """After planning, simulate an accepted write_note landing at a move destination."""
+
+    detector: ProjectIndexChangeDetector
+    session_maker: async_sessionmaker[AsyncSession]
+    entity_repository: EntityRepository
+    destination_file: Path
+    destination_relative: str
+    accepted_content: bytes
+    permalink: str
+
+    async def detect_all_changes(
+        self,
+        storage_files: Mapping[str, RuntimeObservedIndexFile],
+    ) -> ChangeReport:
+        report = await self.detector.detect_all_changes(storage_files)
+        self.destination_file.write_bytes(self.accepted_content)
+        entity = Entity(
+            permalink=self.permalink,
+            title="Accepted",
+            note_type="note",
+            file_path=self.destination_relative,
+            checksum=sha256(self.accepted_content).hexdigest(),
+            content_type="text/markdown",
+            created_at=datetime.now(timezone.utc),
+            updated_at=datetime.now(timezone.utc),
+        )
+        async with db.scoped_session(self.session_maker) as session:
+            await self.entity_repository.add(session, entity)
+        return report
+
+
+async def test_local_project_index_move_spares_concurrently_created_destination_entity(
+    test_project: Project,
+    project_config,
+    entity_repository,
+    session_maker: async_sessionmaker[AsyncSession],
+    config_manager,
+    monkeypatch,
+) -> None:
+    """A planned move must not destroy an entity created at its destination mid-run."""
+    del config_manager
+
+    source_path = project_config.home / "notes" / "move-race.md"
+    destination_path = project_config.home / "archive" / "move-race.md"
+    source_path.parent.mkdir(parents=True, exist_ok=True)
+    destination_path.parent.mkdir(parents=True, exist_ok=True)
+    moved_content = b"# Move Race\n\nOriginal content that moves.\n"
+    source_path.write_bytes(moved_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:
+        source_before = await entity_repository.get_by_file_path(session, "notes/move-race.md")
+    assert source_before is not None
+
+    source_path.rename(destination_path)
+
+    accepted_content = b"# Accepted\n\nConcurrently accepted note content.\n"
+    runtime_factory = LocalProjectIndexRuntimeFactory(batch_size=10)
+    runtime = await runtime_factory.runtime_for_project(test_project)
+    racing_runtime = LocalProjectIndexRuntime(
+        observed_file_source=runtime.observed_file_source,
+        change_detector=ConcurrentDestinationWriteChangeDetector(
+            detector=runtime.change_detector,
+            session_maker=session_maker,
+            entity_repository=entity_repository,
+            destination_file=destination_path,
+            destination_relative="archive/move-race.md",
+            accepted_content=accepted_content,
+            permalink=f"{test_project.permalink}/archive/move-race-accepted",
+        ),
+        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,
+    )
+
+    # The planned move was dropped: nothing moved and nothing was deleted.
+    assert second.moved_files == 0
+    assert second.deleted_files == 0
+
+    async with db.scoped_session(session_maker) as session:
+        destination_entity = await entity_repository.get_by_file_path(
+            session, "archive/move-race.md"
+        )
+        stale_source = await entity_repository.get_by_file_path(session, "notes/move-race.md")
+
+    assert destination_entity is not None
+    assert destination_entity.checksum == sha256(accepted_content).hexdigest()
+    assert destination_path.read_bytes() == accepted_content
+    # The stale source row is left for the next scan to reconcile as deleted.
+    assert stale_source is not None
+
+    third = await run_local_project_index_for_project(
+        test_project,
+        runtime_factory=LocalProjectIndexRuntimeFactory(batch_size=10),
+    )
+
+    assert third.deleted_files == 1
+
+    async with db.scoped_session(session_maker) as session:
+        reconciled_source = await entity_repository.get_by_file_path(
+            session, "notes/move-race.md"
+        )
+        surviving_destination = await entity_repository.get_by_file_path(
+            session, "archive/move-race.md"
+        )
+
+    assert reconciled_source is None
+    assert surviving_destination is not None
+    assert surviving_destination.id == destination_entity.id
+
+
 async def test_local_project_index_move_repairs_observation_search_permalinks(
     test_project: Project,
     project_config,