fix(sync): carry indexed files under unreadable directories through scan reconciliation

os.walk's onerror only aborted the scan when the project root itself was
unreadable; a deeper unreadable subdirectory was pruned from the observed
snapshot with a warning. Delete reconciliation plans
all_db_paths - storage_paths, so every indexed row under the vanished
subtree was destroyed even though the files still exist.

Following the unobservable-file carry-through pattern (46d0011), the walk
now reports unreadable non-root directories in a LocalProjectIndexScan
result, and the observed-file source carries every indexed path under them
through the snapshot as present-with-unknown-checksum
(RuntimeObservedIndexFile with checksum None). Those paths classify as
modified, never deleted, and the batch planner re-checks them next pass.
The indexed-path source is the already-loaded indexed-stat snapshot; the
bare observed source (no stat source) has nothing to carry.

A walk error without a directory attribution now fails the scan outright:
it cannot be carried, and finishing the walk could still plan a mass
delete for an unknown subtree.

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

@@ -142,30 +142,83 @@ async def load_indexed_file_stats(self) -> dict[str, IndexedFileStat]:
         }
 
 
+@dataclass(frozen=True, slots=True)
+class LocalProjectIndexScan:
+    """Walk result for one local project scan.
+
+    ``unreadable_directories`` names project-relative directories whose listing
+    failed mid-walk; their contents are absent from ``file_paths`` even though
+    the files may still exist, so callers must never treat that absence as a
+    delete signal.
+    """
+
+    file_paths: tuple[str, ...]
+    unreadable_directories: tuple[str, ...]
+
+
+def record_local_project_scan_walk_error(
+    error: OSError,
+    *,
+    project_root: Path,
+    unreadable_directories: list[str],
+) -> None:
+    """Classify one os.walk error raised during a local project scan."""
+    # Trigger: the walk error carries no directory attribution.
+    # Why: without a directory we cannot carry its indexed rows through the
+    # snapshot, so finishing the scan could still plan a mass delete for an
+    # unknown subtree.
+    # Outcome: fail the whole scan (RuntimeError bypasses the walk loop's
+    # partial-snapshot OSError handling); the next scan retries.
+    if error.filename is None:
+        raise RuntimeError("Local project scan failed without a directory attribution") from error
+    # Trigger: the project root itself is unreadable (missing/unmounted).
+    # Why: an empty snapshot would make the coordinator treat every indexed
+    # entity as deleted.
+    # Outcome: re-raise so the scan aborts instead of returning empty.
+    if Path(error.filename) == project_root:
+        raise error
+    unreadable_directory = Path(error.filename).relative_to(project_root).as_posix()
+    logger.warning(
+        "Recording unreadable directory during project scan",
+        path=unreadable_directory,
+        error=str(error),
+    )
+    unreadable_directories.append(unreadable_directory)
+
+
 def local_project_index_file_paths(
     project_root: Path,
     *,
     ignore_patterns: LocalProjectIndexIgnorePatterns | None = None,
 ) -> tuple[str, ...]:
     """Return sorted project-relative files eligible for local project indexing."""
+    return scan_local_project_index_files(project_root, ignore_patterns=ignore_patterns).file_paths
+
+
+def scan_local_project_index_files(
+    project_root: Path,
+    *,
+    ignore_patterns: LocalProjectIndexIgnorePatterns | None = None,
+) -> LocalProjectIndexScan:
+    """Walk one local project and report eligible files plus unreadable subtrees."""
     project_root = project_root.expanduser().resolve()
     active_ignore_patterns = (
         ignore_patterns if ignore_patterns is not None else load_gitignore_patterns(project_root)
     )
     file_paths: list[str] = []
+    unreadable_directories: list[str] = []
 
     # os.walk lets us prune ignored/hidden directories *before* descending —
     # rglob("*") would stat every file inside node_modules/.venv/.git first — and
     # does not follow symlinked directories (followlinks=False). Symlinked files
     # are skipped explicitly so the batch reader never reads outside the project
     # boundary. This restores the pre-refactor scanner's no-follow + dir-pruning.
     def _scan_error(error: OSError) -> None:
-        # Abort only if the project root itself is unreadable (missing/unmounted):
-        # returning an empty snapshot would make the coordinator treat every indexed
-        # entity as deleted. A failure deeper in the tree just prunes that subtree.
-        if error.filename is not None and Path(error.filename) == project_root:
-            raise error
-        logger.warning("Skipping unreadable directory during project scan", path=error.filename)
+        record_local_project_scan_walk_error(
+            error,
+            project_root=project_root,
+            unreadable_directories=unreadable_directories,
+        )
 
     walker = os.walk(project_root, followlinks=False, onerror=_scan_error)
     while True:
@@ -204,7 +257,10 @@ def _scan_error(error: OSError) -> None:
                 continue
             file_paths.append(relative_path)
 
-    return tuple(sorted(file_paths))
+    return LocalProjectIndexScan(
+        file_paths=tuple(sorted(file_paths)),
+        unreadable_directories=tuple(unreadable_directories),
+    )
 
 
 @dataclass(frozen=True, slots=True)
@@ -216,11 +272,12 @@ class LocalProjectIndexObservedFileSource(ProjectIndexObservedFileSource):
     indexed_stat_source: LocalProjectIndexedFileStatSource | None = None
 
     async def list_observed_index_files(self) -> tuple[RuntimeObservedIndexFile, ...]:
-        file_paths = await asyncio.to_thread(
-            local_project_index_file_paths,
+        scan = await asyncio.to_thread(
+            scan_local_project_index_files,
             self.file_service.base_path,
             ignore_patterns=self.ignore_patterns,
         )
+        file_paths = scan.file_paths
         # Trigger: a stat source is wired (local runtime); it is absent in
         # cloud/tests that observe without a database.
         # Why: hashing every file on every startup is O(project bytes) and
@@ -267,6 +324,35 @@ async def list_observed_index_files(self) -> tuple[RuntimeObservedIndexFile, ...
                     size=metadata.size,
                 )
             )
+
+        # Trigger: the walk could not list a directory below the project root,
+        # so every file under it is missing from the scan snapshot.
+        # Why: delete reconciliation plans all_db_paths - storage_paths; letting
+        # an unreadable subtree fall out of the snapshot would destroy the
+        # entity/search rows for every indexed file it contains (same hazard as
+        # a single unobservable file, carried above). Only the DB-backed stat
+        # source can enumerate the affected rows; the bare observed source
+        # (no indexed_stat_source) has nothing to carry.
+        # Outcome: indexed paths under the unreadable directory re-enter the
+        # snapshot with an unknown checksum, classifying as modified — the
+        # batch planner re-checks them next pass instead of deleting them.
+        for unreadable_directory in scan.unreadable_directories:
+            directory_prefix = f"{unreadable_directory}/"
+            carried_paths = sorted(
+                indexed_path
+                for indexed_path in indexed_stats
+                if indexed_path.startswith(directory_prefix)
+            )
+            if not carried_paths:
+                continue
+            logger.warning(
+                "Carrying indexed files under unreadable directory through change detection",
+                directory=unreadable_directory,
+                file_count=len(carried_paths),
+            )
+            observed_files.extend(
+                RuntimeObservedIndexFile(path=carried_path) for carried_path in carried_paths
+            )
         return tuple(observed_files)
 
     async def _observed_file_confirmed_missing(self, file_path: RuntimeFilePath) -> bool:

tests/index/test_local_project_index.py

@@ -21,13 +21,14 @@
     LocalProjectIndexObservedFileSource,
     LocalProjectIndexRuntime,
     LocalProjectIndexRuntimeFactory,
+    LocalProjectIndexScan,
     RepositoryLocalProjectIndexedFileStatSource,
     local_project_index_file_paths,
     run_local_project_index,
     run_local_project_index_for_project,
 )
 from basic_memory.indexing.change_detector import ChangeDetector
-from basic_memory.indexing.change_planning import ChangeReport
+from basic_memory.indexing.change_planning import ChangeReport, plan_file_changes
 from basic_memory.indexing.embedding_index_planning import EmbeddingIndexTarget
 from basic_memory.indexing.file_batch_runner import IndexFileBatchReadResult
 from basic_memory.indexing.file_index_checking import IndexedFileChecksumRow
@@ -2381,6 +2382,63 @@ async def tracking_checksum(path):
     assert set(hashed_paths) == {"notes/remtimed.md", "notes/resized.md", "notes/new.md"}
 
 
+async def test_local_project_index_observed_source_carries_indexed_files_under_unreadable_directory(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    """Indexed rows under a directory the walk cannot read must not plan as deletes."""
+    keep = tmp_path / "keep.md"
+    keep_content = b"# Keep\n"
+    keep.write_bytes(keep_content)
+
+    indexed_stats = {
+        "keep.md": IndexedFileStat(mtime=None, size=None, checksum=None),
+        "locked/a.md": IndexedFileStat(mtime=1.0, size=10, checksum="indexed-a"),
+        "locked/deep/b.md": IndexedFileStat(mtime=1.0, size=11, checksum="indexed-b"),
+        # Indexed elsewhere and genuinely gone from storage — must not be carried.
+        "other/gone.md": IndexedFileStat(mtime=1.0, size=12, checksum="indexed-gone"),
+    }
+
+    def unreadable_scan(*args, **kwargs) -> LocalProjectIndexScan:
+        return LocalProjectIndexScan(
+            file_paths=("keep.md",),
+            unreadable_directories=("locked",),
+        )
+
+    monkeypatch.setattr(
+        "basic_memory.index.local_project.scan_local_project_index_files",
+        unreadable_scan,
+    )
+
+    observed = await LocalProjectIndexObservedFileSource(
+        FileService(tmp_path),
+        ignore_patterns=set(),
+        indexed_stat_source=_StaticIndexedFileStatSource(indexed_stats),
+    ).list_observed_index_files()
+
+    assert [target.path for target in observed] == ["keep.md", "locked/a.md", "locked/deep/b.md"]
+    carried = {target.path: target for target in observed[1:]}
+    assert carried["locked/a.md"].checksum is None
+    assert carried["locked/deep/b.md"].checksum is None
+
+    keep_checksum = sha256(keep_content).hexdigest()
+    report = plan_file_changes(
+        storage_checksum_by_path={target.path: target.checksum for target in observed},
+        db_checksum_by_path={
+            "keep.md": keep_checksum,
+            "locked/a.md": "indexed-a",
+            "locked/deep/b.md": "indexed-b",
+        },
+        all_db_paths=("keep.md", "locked/a.md", "locked/deep/b.md", "other/gone.md"),
+        move_candidates=(),
+    )
+
+    # The unreadable subtree re-enters change detection as modified, never deleted;
+    # only the path that is genuinely gone from storage is planned as a delete.
+    assert report.deleted_files == ["other/gone.md"]
+    assert sorted(report.modified_files) == ["locked/a.md", "locked/deep/b.md"]
+
+
 async def test_local_project_index_observed_source_hashes_without_stat_source(
     tmp_path: Path,
     monkeypatch,

tests/index/test_local_project_scan_parity.py

@@ -1,13 +1,17 @@
 """Local project scan parity with the legacy sync oracle."""
 
+import os
 from pathlib import Path
 
 import pytest
 
 import basic_memory.index.local_project as local_project
 from basic_memory.index.local_project import (
     LocalProjectIndexObservedFileSource,
+    LocalProjectIndexScan,
     local_project_index_file_paths,
+    record_local_project_scan_walk_error,
+    scan_local_project_index_files,
 )
 from basic_memory.services import FileService
 
@@ -83,6 +87,69 @@ def recording_is_file(path: Path) -> bool:
     assert not any(".hidden" in path for path in visited)
 
 
+def test_scan_local_project_index_files_records_unreadable_subdirectory(
+    monkeypatch,
+    tmp_path: Path,
+) -> None:
+    """An unreadable subdirectory is reported, not silently pruned from the scan."""
+    project_root = tmp_path.resolve()
+    (project_root / "keep.md").write_text("# keep\n", encoding="utf-8")
+    locked_dir = project_root / "locked"
+    locked_dir.mkdir()
+    (locked_dir / "note.md").write_text("# locked\n", encoding="utf-8")
+
+    real_walk = os.walk
+
+    def walk_with_locked_scandir_error(top, *, followlinks, onerror):
+        # Emulate exactly what os.walk does when scandir(locked) fails: report
+        # the directory through onerror and never yield its listing.
+        for dirpath, dirnames, filenames in real_walk(
+            top, followlinks=followlinks, onerror=onerror
+        ):
+            if Path(dirpath) == locked_dir:
+                onerror(PermissionError(13, "Permission denied", str(locked_dir)))
+                continue
+            yield dirpath, dirnames, filenames
+
+    monkeypatch.setattr(local_project.os, "walk", walk_with_locked_scandir_error)
+
+    scan = scan_local_project_index_files(project_root, ignore_patterns=set())
+
+    assert scan.file_paths == ("keep.md",)
+    assert scan.unreadable_directories == ("locked",)
+
+
+def test_record_local_project_scan_walk_error_classifies_root_none_and_subtree(
+    tmp_path: Path,
+) -> None:
+    """Root failures abort, unattributed failures fail the scan, subtrees are recorded."""
+    project_root = tmp_path.resolve()
+    unreadable_directories: list[str] = []
+
+    root_error = PermissionError(13, "Permission denied", str(project_root))
+    with pytest.raises(PermissionError):
+        record_local_project_scan_walk_error(
+            root_error,
+            project_root=project_root,
+            unreadable_directories=unreadable_directories,
+        )
+
+    with pytest.raises(RuntimeError, match="without a directory attribution"):
+        record_local_project_scan_walk_error(
+            PermissionError(13, "Permission denied"),
+            project_root=project_root,
+            unreadable_directories=unreadable_directories,
+        )
+    assert unreadable_directories == []
+
+    record_local_project_scan_walk_error(
+        PermissionError(13, "Permission denied", str(project_root / "locked" / "deep")),
+        project_root=project_root,
+        unreadable_directories=unreadable_directories,
+    )
+    assert unreadable_directories == ["locked/deep"]
+
+
 def test_local_project_index_file_paths_aborts_when_root_unreadable(tmp_path: Path) -> None:
     """A missing/unmounted project root must raise, not return an empty snapshot —
     the coordinator would otherwise classify every indexed entity as deleted."""
@@ -117,15 +184,18 @@ async def test_local_project_index_observed_file_source_skips_files_missing_afte
     keep_path.write_text("# Keep\n", encoding="utf-8")
     warnings: list[tuple[str, dict[str, object]]] = []
 
-    def discovered_paths(*args, **kwargs) -> tuple[str, ...]:
-        return ("missing.md", "keep.md")
+    def discovered_scan(*args, **kwargs) -> LocalProjectIndexScan:
+        return LocalProjectIndexScan(
+            file_paths=("missing.md", "keep.md"),
+            unreadable_directories=(),
+        )
 
     def record_warning(message: str, **kwargs: object) -> None:
         warnings.append((message, kwargs))
 
     monkeypatch.setattr(
-        "basic_memory.index.local_project.local_project_index_file_paths",
-        discovered_paths,
+        "basic_memory.index.local_project.scan_local_project_index_files",
+        discovered_scan,
     )
     monkeypatch.setattr(local_project.logger, "warning", record_warning)