fix(sync): re-read file at batch reconcile to avoid reverting newer accepted content

Batch note_content reconciliation reconciled each entity using
`indexed.markdown_content`, a scan-time snapshot. Between scan and reconcile a
note materialization can rewrite the file to a newer accepted db_version; the
batch then promoted the stale snapshot and reverted the newer write. The
db_version compare-and-set guard does not help: the stale snapshot promotes
cleanly to a fresh version.

Add an optional `file_reader` (NoteContentReconcileFileReader) threaded through
`reconcile_indexed_note_content_batch`, `IndexedNoteContentReconciliationTask`,
`IndexBatchRuntime`, and `build_default_index_batch_runtime`. When set, the
reconcile task re-reads the current file: missing content (file gone) skips
reconciliation, otherwise the fresh content and its mtime drive reconciliation
instead of the scan snapshot. Default None preserves existing behavior so cloud
compiles unchanged.

Local indexing wires a filesystem-backed reader
(FileServiceNoteContentReconcileFileReader) built on FileService. Cloud passes
None for now to avoid doubling S3 reads and can opt in later via its S3 reader.

Adds regression coverage: a stale scan snapshot no longer reverts a newer
accepted note_content version once the reader re-reads the current file, plus a
companion test documenting the revert that occurs without the reader.

🐛 Fixes a concurrency defect the version guard cannot catch.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

src/basic_memory/index/local_dependencies.py

@@ -4,6 +4,7 @@
 
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
+from datetime import datetime
 from pathlib import Path
 from typing import Any, Protocol
 
@@ -68,6 +69,42 @@
 from basic_memory.services.search_service import SearchService
 
 
+@dataclass(frozen=True, slots=True)
+class FileServiceReconcileFile:
+    """Canonical file state re-read at batch reconcile time."""
+
+    content: bytes | None
+    last_modified: datetime | None
+
+
+@dataclass(frozen=True, slots=True)
+class FileServiceNoteContentReconcileFileReader:
+    """Filesystem-backed reader that re-reads canonical markdown at reconcile time.
+
+    Batch reconciliation indexes a scan-time snapshot of each file. A note
+    materialization can rewrite the same file to a newer accepted version between
+    the scan and reconcile; re-reading here lets reconciliation promote the
+    current file instead of reverting to the stale snapshot.
+    """
+
+    file_service: FileService
+
+    async def get_file(self, path: RuntimeFilePath) -> FileServiceReconcileFile:
+        """Return current file bytes and mtime, or None content when the file is gone."""
+        try:
+            content = await self.file_service.read_file_bytes(path)
+        except FileOperationError as exc:
+            # Trigger: the file was deleted between scan and reconcile.
+            # Why: a missing file has no content to promote; surface None so the
+            #   batch task skips reconciliation instead of failing the whole batch.
+            # Outcome: caller treats None content as "nothing to reconcile".
+            if isinstance(exc.__cause__, FileNotFoundError):
+                return FileServiceReconcileFile(content=None, last_modified=None)
+            raise
+        metadata = await self.file_service.get_file_metadata(path)
+        return FileServiceReconcileFile(content=content, last_modified=metadata.modified_at)
+
+
 class LocalIndexEntityRepository(
     IndexMarkdownEntityRepository,
     IndexedFileChecksumRepository,
@@ -598,6 +635,7 @@ async def build_local_index_project_dependencies(
         frontmatter_storage=file_service,
         content_type_provider=file_service,
         session_maker=session_maker,
+        file_reader=FileServiceNoteContentReconcileFileReader(file_service=file_service),
     )
     file_indexer = build_local_markdown_file_indexer(
         project_id=project.id,

src/basic_memory/indexing/index_batch_runtime.py

@@ -31,7 +31,10 @@
     IndexedNoteContentTimestampProvider,
     reconcile_indexed_note_content_batch,
 )
-from basic_memory.indexing.note_content_reconciler import NoteContentReconciler
+from basic_memory.indexing.note_content_reconciler import (
+    NoteContentReconcileFileReader,
+    NoteContentReconciler,
+)
 from basic_memory.models import Entity
 from basic_memory.repository import EntityRepository, NoteContentRepository, RelationRepository
 from basic_memory.runtime.storage import ProjectId
@@ -62,6 +65,12 @@ class IndexBatchRuntime[EntityT: IndexedNoteContentEntity, FileInfoT: LoadedInde
     note_content_reconciler: IndexedNoteContentReconciler[EntityT]
     timestamp_provider: IndexedNoteContentTimestampProvider[FileInfoT]
     note_content_source: str = "index"
+    # Optional canonical-file reader. When set, batch reconciliation re-reads each
+    # file at reconcile time instead of trusting the scan snapshot, so a note
+    # materialization that rewrote the file between scan and reconcile is not
+    # reverted. Local indexing passes a filesystem reader; cloud leaves this None
+    # to avoid doubling S3 reads (it can opt in later).
+    file_reader: NoteContentReconcileFileReader | None = None
 
     async def index_loaded_files(
         self,
@@ -90,6 +99,7 @@ async def index_loaded_files(
             timestamp_provider=self.timestamp_provider,
             max_concurrent=metadata_update_max_concurrent or max_concurrent,
             source=self.note_content_source,
+            file_reader=self.file_reader,
         )
         result.errors.extend(error.as_tuple() for error in note_content_errors)
         result.search_indexed = count_search_indexed_entities(result.indexed)
@@ -136,6 +146,7 @@ def build_default_index_batch_runtime[FileInfoT: LoadedIndexFile](
     frontmatter_storage: IndexFrontmatterStorage,
     content_type_provider: IndexContentTypeProvider,
     session_maker: async_sessionmaker[AsyncSession],
+    file_reader: NoteContentReconcileFileReader | None = None,
 ) -> DefaultIndexBatchRuntime[FileInfoT]:
     """Compose the default repository-backed batch index runtime.
 
@@ -170,6 +181,7 @@ def build_default_index_batch_runtime[FileInfoT: LoadedIndexFile](
             session_maker=session_maker,
             note_content_reconciler=note_content_reconciler,
             timestamp_provider=DefaultIndexedNoteContentTimestampProvider(),
+            file_reader=file_reader,
         ),
         note_content_reconciler=note_content_reconciler,
     )

src/basic_memory/indexing/note_content_batch_reconciliation.py

@@ -15,6 +15,7 @@
 from basic_memory import db
 from basic_memory.indexing.file_index_planning import FileIndexPath
 from basic_memory.indexing.models import IndexedEntity
+from basic_memory.indexing.note_content_reconciler import NoteContentReconcileFileReader
 
 
 class IndexedNoteContentEntity(Protocol):
@@ -175,6 +176,7 @@ class IndexedNoteContentReconciliationTask[
     note_content_reconciler: IndexedNoteContentReconciler[EntityT]
     timestamp_provider: IndexedNoteContentTimestampProvider[FileInfoT]
     source: str
+    file_reader: NoteContentReconcileFileReader | None = None
 
     async def run(self) -> IndexedNoteContentReconciliationError | None:
         if self.indexed.markdown_content is None:
@@ -187,14 +189,35 @@ async def run(self) -> IndexedNoteContentReconciliationError | None:
                 f"Entity {self.indexed.entity_id} not found after indexing",
             )
 
+        # --- Resolve the markdown version to reconcile ---
+        # indexed.markdown_content is a SCAN-TIME snapshot. Between the scan and
+        # this reconcile a note materialization can rewrite the file to a newer
+        # accepted db_version; promoting the snapshot would revert that write and
+        # the db_version compare-and-set guard cannot catch it (the stale snapshot
+        # still promotes cleanly to a fresh version).
+        if self.file_reader is not None:
+            # Trigger: a filesystem reader is configured (local indexing path).
+            # Why: re-read the file so we reconcile the CURRENT accepted content,
+            #   not the possibly-stale scan snapshot.
+            # Outcome: fresh content + its last_modified drive reconciliation.
+            fresh = await self.file_reader.get_file(self.indexed.path)
+            if fresh.content is None:
+                # File was removed between scan and reconcile; nothing to promote.
+                return None
+            markdown_content = fresh.content.decode("utf-8")
+            observed_at = fresh.last_modified
+        else:
+            markdown_content = self.indexed.markdown_content
+            observed_at = self.timestamp_provider.observed_at(
+                self.indexed,
+                self.file_infos.get(self.indexed.path),
+            )
+
         try:
             await self.note_content_reconciler.reconcile(
                 entity=entity,
-                markdown_content=self.indexed.markdown_content,
-                observed_at=self.timestamp_provider.observed_at(
-                    self.indexed,
-                    self.file_infos.get(self.indexed.path),
-                ),
+                markdown_content=markdown_content,
+                observed_at=observed_at,
                 source=self.source,
             )
             return None
@@ -218,6 +241,7 @@ async def reconcile_indexed_note_content_batch[
     timestamp_provider: IndexedNoteContentTimestampProvider[FileInfoT],
     max_concurrent: int,
     source: str = "index",
+    file_reader: NoteContentReconcileFileReader | None = None,
 ) -> tuple[IndexedNoteContentReconciliationError, ...]:
     """Hydrate note_content rows for indexed markdown entities after batch indexing."""
     markdown_entities = tuple(
@@ -242,6 +266,7 @@ async def reconcile_indexed_note_content_batch[
                 note_content_reconciler=note_content_reconciler,
                 timestamp_provider=timestamp_provider,
                 source=source,
+                file_reader=file_reader,
             )
             for indexed in markdown_entities
         ],

tests/index/test_local_reconcile_file_reader.py

@@ -0,0 +1,54 @@
+"""Tests for the local filesystem note_content reconcile file reader."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, cast
+
+import pytest
+
+from basic_memory.index.local_dependencies import (
+    FileServiceNoteContentReconcileFileReader,
+    FileServiceReconcileFile,
+)
+from basic_memory.services.exceptions import FileOperationError
+from basic_memory.services.file_service import FileService
+
+
+@pytest.mark.asyncio
+async def test_reader_returns_current_bytes_and_mtime(file_service: FileService) -> None:
+    """The reader returns the current file bytes and modification time."""
+    path = "note.md"
+    (Path(file_service.base_path) / path).write_text("# Fresh\n", encoding="utf-8")
+
+    reader = FileServiceNoteContentReconcileFileReader(file_service=file_service)
+    result = await reader.get_file(path)
+
+    assert result.content == b"# Fresh\n"
+    assert isinstance(result.last_modified, datetime)
+
+
+@pytest.mark.asyncio
+async def test_reader_returns_none_content_when_file_missing(file_service: FileService) -> None:
+    """A file removed between scan and reconcile yields None content, not an error."""
+    reader = FileServiceNoteContentReconcileFileReader(file_service=file_service)
+
+    result = await reader.get_file("gone.md")
+
+    assert result == FileServiceReconcileFile(content=None, last_modified=None)
+
+
+@pytest.mark.asyncio
+async def test_reader_reraises_non_missing_file_errors() -> None:
+    """Read failures other than a missing file must propagate."""
+
+    async def read_file_bytes(path: str) -> bytes:
+        raise FileOperationError("boom") from ValueError("disk error")
+
+    file_service = SimpleNamespace(read_file_bytes=read_file_bytes)
+    reader = FileServiceNoteContentReconcileFileReader(file_service=cast(Any, file_service))
+
+    with pytest.raises(FileOperationError):
+        await reader.get_file("broken.md")