fix(sync): defer moved-file frontmatter writes until the move batch commits

apply_project_index_move_batch called the move content updater inside the
open db.scoped_session, and the local updater wrote the moved file's
permalink frontmatter to disk via update_frontmatter_with_result before
the transaction committed. If the batch rolled back (e.g. an intra-batch
permalink collision), the database reverted but the on-disk rewrites
persisted, leaving files ahead of their indexed state.

The ProjectIndexMoveContentUpdater capability is now split into
plan_moved_file_content — which runs inside the transaction, resolves the
permalink, and builds the updated markdown in memory without touching
storage — and write_moved_file_content, which the store calls only after a
successful commit. The database rows are stamped from the planned content,
whose checksum matches the exact bytes the post-commit write persists, so
DB and file agree when the write lands; a failed write (or formatter or
platform newline divergence) is logged and surfaces on the next scan as a
checksum-mismatch modified file, which re-indexes from disk.

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

Author: phernandez

src/basic_memory/index/local_moves.py

@@ -2,15 +2,24 @@
 
 from __future__ import annotations
 
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Protocol
 
+import yaml
+from loguru import logger
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from basic_memory import db
 from basic_memory.config import BasicMemoryConfig
+from basic_memory.file_utils import (
+    ParseError,
+    compute_checksum,
+    has_frontmatter,
+    parse_frontmatter,
+    remove_frontmatter,
+)
 from basic_memory.markdown import EntityMarkdown
 from basic_memory.indexing.project_index_maintenance import (
     ProjectIndexMaintenanceRunner,
@@ -63,14 +72,41 @@ async def resolve_permalink(
     ) -> str: ...
 
 
+def merged_frontmatter_markdown(content: str, updates: Mapping[str, object]) -> str:
+    """Return markdown with frontmatter updates applied, without touching storage.
+
+    Mirrors FileService.update_frontmatter_with_result's merge — including its
+    tolerance for malformed YAML, which keeps the full content and prepends a
+    fresh frontmatter block — so planned bytes match what a direct frontmatter
+    rewrite would have produced.
+    """
+    current_frontmatter: dict[str, object] = {}
+    body = content
+    if has_frontmatter(content):
+        try:
+            current_frontmatter = dict(parse_frontmatter(content))
+            body = remove_frontmatter(content)
+        except ParseError as error:
+            logger.warning(
+                "Treating file with malformed frontmatter as plain markdown",
+                error=str(error),
+            )
+            current_frontmatter = {}
+            body = content
+
+    merged_frontmatter = {**current_frontmatter, **updates}
+    yaml_block = yaml.dump(merged_frontmatter, sort_keys=False, allow_unicode=True)
+    return f"---\n{yaml_block}---\n\n{body.strip()}"
+
+
 @dataclass(frozen=True, slots=True)
 class LocalProjectIndexMoveContentUpdater(ProjectIndexMoveContentUpdater):
     """Apply local markdown permalink policy for moved files."""
 
     entity_service: LocalMoveEntityService
     file_service: FileService
 
-    async def update_moved_file_content(
+    async def plan_moved_file_content(
         self,
         session: AsyncSession,
         moved_file: ProjectIndexMovedFile,
@@ -91,16 +127,30 @@ async def update_moved_file_content(
         if permalink == moved_file.old_permalink:
             return None
 
-        update = await self.file_service.update_frontmatter_with_result(
-            moved_file.new_path,
+        current_bytes = await self.file_service.read_file_bytes(moved_file.new_path)
+        planned_content = merged_frontmatter_markdown(
+            current_bytes.decode("utf-8"),
             {"permalink": permalink},
         )
+        # Invariant: the move batch stamps entity/note_content rows from this
+        # planned content while write_moved_file_content persists exactly these
+        # bytes after commit, so database and file checksums agree. Divergence
+        # (a failed write, an on-save formatter, platform newline translation)
+        # shows up on the next scan as a checksum-mismatch modified file and is
+        # re-indexed from disk.
         return ProjectIndexMovedFileContentUpdate(
             permalink=permalink,
-            checksum=update.checksum,
-            markdown_content=update.content,
+            checksum=await compute_checksum(planned_content),
+            markdown_content=planned_content,
         )
 
+    async def write_moved_file_content(
+        self,
+        moved_file: ProjectIndexMovedFile,
+        content_update: ProjectIndexMovedFileContentUpdate,
+    ) -> None:
+        await self.file_service.write_file(moved_file.new_path, content_update.markdown_content)
+
 
 @dataclass(frozen=True, slots=True)
 class LocalWatchMoveProcessingResult:

src/basic_memory/indexing/project_index_maintenance.py

@@ -99,22 +99,39 @@ class ProjectIndexMovedFile:
 
 @dataclass(frozen=True, slots=True)
 class ProjectIndexMovedFileContentUpdate:
-    """Storage result after rewriting a moved file's markdown metadata."""
+    """Planned markdown metadata rewrite for a moved file.
+
+    ``checksum`` is computed from ``markdown_content`` — the exact bytes the
+    post-commit write persists — so the database rows stamped during the batch
+    transaction agree with the file once the write lands.
+    """
 
     permalink: str
     checksum: str
     markdown_content: str
 
 
 class ProjectIndexMoveContentUpdater(Protocol):
-    """Capability that applies provider-specific moved-file content repair."""
+    """Capability that plans and persists provider-specific moved-file content repair.
+
+    Planning runs inside the move batch's database transaction and must not
+    mutate storage: the batch can still roll back (e.g. an intra-batch
+    permalink collision), and an already-rewritten file would survive that
+    rollback. The write runs only after the batch commits.
+    """
 
-    async def update_moved_file_content(
+    async def plan_moved_file_content(
         self,
         session: AsyncSession,
         moved_file: ProjectIndexMovedFile,
     ) -> ProjectIndexMovedFileContentUpdate | None: ...
 
+    async def write_moved_file_content(
+        self,
+        moved_file: ProjectIndexMovedFile,
+        content_update: ProjectIndexMovedFileContentUpdate,
+    ) -> None: ...
+
 
 @dataclass(frozen=True, slots=True)
 class ProjectIndexMoveTarget:
@@ -543,39 +560,39 @@ async def apply_project_index_move_batch(
                 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
-                        )
+                        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
+                        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
             }
+            planned_moved_files_by_entity_id: dict[int, ProjectIndexMovedFile] = {}
             if self.move_content_updater is not None:
                 for row in target_rows:
                     entity_id = int(row["id"])
                     old_path = str(row["file_path"])
-                    content_update = await self.move_content_updater.update_moved_file_content(
-                        session,
-                        ProjectIndexMovedFile(
-                            entity_id=entity_id,
-                            old_path=old_path,
-                            new_path=target_paths_by_old_path[old_path],
-                            old_permalink=(
-                                str(row["permalink"]) if row["permalink"] is not None else None
-                            ),
+                    moved_file = ProjectIndexMovedFile(
+                        entity_id=entity_id,
+                        old_path=old_path,
+                        new_path=target_paths_by_old_path[old_path],
+                        old_permalink=(
+                            str(row["permalink"]) if row["permalink"] is not None else None
                         ),
                     )
+                    content_update = await self.move_content_updater.plan_moved_file_content(
+                        session,
+                        moved_file,
+                    )
                     if content_update is not None:
                         content_updates_by_entity_id[entity_id] = content_update
+                        planned_moved_files_by_entity_id[entity_id] = moved_file
 
             if updated_old_paths:
                 replaced_entity_ids = frozenset(int(row["id"]) for row in replacement_rows)
@@ -686,6 +703,28 @@ async def apply_project_index_move_batch(
                         )
                     )
 
+        # Trigger: the batch committed with entity/note_content rows stamped from
+        # the planned markdown, and the files still hold their pre-move metadata.
+        # Why: writing files inside the transaction is not atomic with it — a
+        #      rollback would revert the database while the on-disk frontmatter
+        #      rewrites persisted, leaving files ahead of their indexed state.
+        # Outcome: writes happen only after a successful commit; a failed write
+        #          leaves the file with a checksum that no longer matches its
+        #          rows, which the next scan reconciles as a modified file.
+        if self.move_content_updater is not None:
+            for entity_id, content_update in content_updates_by_entity_id.items():
+                try:
+                    await self.move_content_updater.write_moved_file_content(
+                        planned_moved_files_by_entity_id[entity_id],
+                        content_update,
+                    )
+                except Exception as write_error:
+                    logger.error(
+                        "Failed to write moved file content after move batch commit",
+                        path=planned_moved_files_by_entity_id[entity_id].new_path,
+                        error=str(write_error),
+                    )
+
         missing_paths = tuple(
             move_target.old_path
             for move_target in move_batch.targets
@@ -716,9 +755,7 @@ async def apply_project_index_delete_batch(
         # 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
-        )
+        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