fix(folder): stable root ids in source_ref; migrate legacy rows in place

src/partial_recall/corpus/adapters/folder.py

@@ -27,9 +27,13 @@
 from collections.abc import Iterator
 from datetime import datetime
 from pathlib import Path
+from typing import TYPE_CHECKING
 
 import structlog
 
+if TYPE_CHECKING:
+    from partial_recall.store.vector_store import VectorStore
+
 from partial_recall.corpus.adapters._shared import matches_any, read_ignorefile, stable_item_key
 from partial_recall.corpus.types import Item, ItemKind, Source
 from partial_recall.errors import CorpusUnavailableError, PartialRecallError
@@ -51,6 +55,17 @@ class FolderAdapterError(PartialRecallError):
     """FolderAdapter-specific failure."""
 
 
+def _root_id(root: Path) -> str:
+    """Stable identifier for a configured root: survives reordering.
+
+    'r' + first 10 hex of SHA-256 of the resolved path. The 'r' marker
+    cannot collide with the legacy numeric root-index prefix or with an
+    absolute path, so every source_ref format stays distinguishable.
+    """
+    digest = hashlib.sha256(str(root.resolve()).encode("utf-8")).hexdigest()
+    return f"r{digest[:10]}"
+
+
 
 
 class FolderAdapter:
@@ -78,6 +93,13 @@ def __init__(
             if not r.is_dir():
                 raise CorpusUnavailableError(f"folder root is not a directory: {r}")
         self.recursive = recursive
+        # Stable per-root identifier: source_ref must not depend on the
+        # ORDER of `roots` in config — reordering used to orphan every
+        # chunk and re-embed the corpus. The "r" marker keeps the id
+        # distinguishable from the legacy numeric root-index prefix.
+        self._roots_by_id: dict[str, Path] = {
+            _root_id(r): r for r in self.roots
+        }
         # Default to text-only + pdf so v0.2.0 doesn't promise extractors
         # it doesn't have. EPUB/DOCX configured by the user still resolve
         # to "skip" until their extractors ship.
@@ -127,18 +149,17 @@ def get_sources(self, item: Item) -> Iterator[Source]:
         if not item.corpus_ref:
             return
         abs_path = Path(item.corpus_ref)
-        # Emit "{root_idx}:{rel_posix}" so source_ref is portable and
-        # unambiguous even with multiple roots that share filenames.
-        # The root_idx prefix encodes which configured root owns this file,
-        # avoiding the wrong-root ambiguity flagged in issue #23.
-        # Fall back to the absolute path only for files that fall outside
-        # every configured root (edge case; shouldn't happen in normal use).
-        for idx, root in enumerate(self.roots):
+        # Emit "{root_id}:{rel_posix}" so source_ref is portable,
+        # unambiguous with multiple roots that share filenames (issue
+        # #23), and stable when config roots are reordered. Fall back to
+        # the absolute path only for files that fall outside every
+        # configured root (edge case; shouldn't happen in normal use).
+        for root in self.roots:
             try:
                 rel = abs_path.relative_to(root).as_posix()
                 yield Source(
                     source_type="file",
-                    source_ref=f"{idx}:{rel}",
+                    source_ref=f"{_root_id(root)}:{rel}",
                     kind=ItemKind.TEXT,
                 )
                 return
@@ -153,20 +174,24 @@ def get_sources(self, item: Item) -> Iterator[Source]:
     def _resolve_source_ref(self, source_ref: str) -> Path | None:
         """Return an absolute Path for source_ref.
 
-        Accepts three formats:
-        - "{root_idx}:{rel_posix}"  — new portable format (issue #23)
-        - absolute path             — legacy rows indexed before this fix
+        Accepts every format ever written to a store:
+        - "{root_id}:{rel_posix}"   — current stable format
+        - "{root_idx}:{rel_posix}"  — legacy positional format (pre-stable-ids)
+        - absolute path             — legacy rows indexed before issue #23
         - bare relative path        — intermediate format (should not occur
                                       in production, but handled defensively)
         """
         p = Path(source_ref)
         if p.is_absolute():
             return p if p.exists() else None
-        # Try new "{idx}:{rel}" format.
         if ":" in source_ref:
-            idx_str, _, rel = source_ref.partition(":")
-            if idx_str.isdigit():
-                idx = int(idx_str)
+            prefix, _, rel = source_ref.partition(":")
+            root = self._roots_by_id.get(prefix)
+            if root is not None:
+                candidate = root / rel
+                return candidate if candidate.exists() else None
+            if prefix.isdigit():
+                idx = int(prefix)
                 if idx < len(self.roots):
                     candidate = self.roots[idx] / rel
                     return candidate if candidate.exists() else None
@@ -177,6 +202,80 @@ def _resolve_source_ref(self, source_ref: str) -> Path | None:
                 return candidate
         return None
 
+    def migrate_source_refs(self, store: VectorStore) -> dict[str, int]:
+        """Rewrite legacy source_refs to the stable root-id format.
+
+        Optional adapter hook, called by run_indexing before walking.
+        Handles both legacy formats: positional "{idx}:{rel}" (mapped
+        through the CURRENT root order — the same mapping the old lookup
+        used) and absolute paths that fall under a configured root.
+        Rows whose rewrite target already exists are merged by the store
+        (past drift created those duplicates); their vectors survive on
+        the merged row. Idempotent: current-format refs are untouched,
+        and unmappable refs (absolute paths outside every configured
+        root) are left alone and counted as skipped.
+        """
+        counts = {"rewritten": 0, "merged": 0, "skipped": 0}
+        for row in store.iter_chunk_refs(corpus=self.name):
+            new_ref = self._legacy_ref_to_stable(
+                row["source_ref"], row["item_corpus_ref"]
+            )
+            if new_ref == row["source_ref"]:
+                continue
+            if new_ref is None:
+                counts["skipped"] += 1
+                continue
+            outcome = store.rewrite_chunk_source_ref(
+                chunk_id=row["chunk_id"],
+                corpus=self.name,
+                new_source_ref=new_ref,
+            )
+            counts[outcome] += 1
+        if counts["rewritten"] or counts["merged"] or counts["skipped"]:
+            log.info("folder.adapter.source_refs_migrated", **counts)
+        return counts
+
+    def _legacy_ref_to_stable(
+        self, source_ref: str, item_corpus_ref: str | None
+    ) -> str | None:
+        """Map a legacy source_ref to the stable format; None if unmappable.
+
+        The item's corpus_ref (the file's absolute path, the same value
+        item_key is hashed from) is the authority for which root owns the
+        chunk: a positional ref like "0:x.md" cannot be trusted against
+        the CURRENT root order, because the roots may have been reordered
+        since the row was written — that reordering is this bug.
+        """
+        if any(source_ref.startswith(f"{rid}:") for rid in self._roots_by_id):
+            return source_ref  # already stable
+        if item_corpus_ref:
+            item_path = Path(item_corpus_ref)
+            if item_path.is_absolute():
+                for root in self.roots:
+                    try:
+                        rel = item_path.relative_to(root).as_posix()
+                    except ValueError:
+                        continue
+                    return f"{_root_id(root)}:{rel}"
+        p = Path(source_ref)
+        if p.is_absolute():
+            for root in self.roots:
+                try:
+                    rel = p.relative_to(root).as_posix()
+                except ValueError:
+                    continue
+                return f"{_root_id(root)}:{rel}"
+            return None  # outside every configured root — leave untouched
+        if ":" in source_ref:
+            prefix, _, rel = source_ref.partition(":")
+            if prefix.isdigit() and int(prefix) < len(self.roots):
+                # No corpus_ref to consult: positional mapping through the
+                # current order is only safe if the file is really there.
+                root = self.roots[int(prefix)]
+                if (root / rel).exists():
+                    return f"{_root_id(root)}:{rel}"
+        return None
+
     def get_text(self, item: Item, source: Source) -> str | None:
         if source.source_type != "file" or not source.source_ref:
             return None

src/partial_recall/corpus/protocol.py

@@ -15,6 +15,12 @@ class CorpusAdapter(Protocol):
 
     v0.0.1: ZoteroAdapter only.
     v0.1.0: + FolderAdapter.
+
+    Optional hook (NOT part of this protocol, discovered by duck-typing
+    so existing adapters keep validating): `migrate_source_refs(store)`.
+    If present, run_indexing calls it before walking items; adapters use
+    it to rewrite legacy source_ref formats in place so old rows keep
+    matching instead of being re-created and re-embedded.
     """
 
     @property

src/partial_recall/index/pipeline.py

@@ -209,6 +209,14 @@ def _run_indexing(
         )
         log.info("indexing.run.start", run_id=run_id, provider=meta.provider, model=meta.model_name)
 
+    # Optional adapter hook: heal legacy source_ref formats before the
+    # walk, so find_chunk_id matches existing rows instead of re-creating
+    # (and re-embedding) them. Duck-typed rather than part of the
+    # CorpusAdapter protocol — existing external adapters must not break.
+    migrate_refs = getattr(adapter, "migrate_source_refs", None)
+    if callable(migrate_refs):
+        migrate_refs(store)
+
     item_count = 0
     chunk_count = 0
     new_vector_count = 0

src/partial_recall/store/vector_store.py

@@ -6,6 +6,7 @@
 from pathlib import Path
 from typing import Any
 
+from partial_recall.errors import VectorStoreError
 from partial_recall.store.connection import connect
 
 
@@ -508,6 +509,87 @@ def find_chunk_id(
         ).fetchone()
         return None if row is None else (int(row["chunk_id"]), str(row["text_hash"]))
 
+    def iter_chunk_refs(self, *, corpus: str) -> list[dict[str, Any]]:
+        """All chunk identity rows for one corpus, for source_ref migration.
+
+        Includes the owning item's corpus_ref: for path-backed corpora it
+        holds the file's absolute path — the order-independent authority
+        for which root owns a legacy positional ref.
+        """
+        rows = self._conn.execute(
+            """
+            SELECT c.chunk_id, c.item_key, c.source_type, c.source_ref,
+                   c.chunk_index, c.chunker_version,
+                   i.corpus_ref AS item_corpus_ref
+            FROM chunks c
+            LEFT JOIN items i
+                ON i.owner = c.owner
+               AND i.corpus = c.corpus
+               AND i.item_key = c.item_key
+            WHERE c.owner = 'local' AND c.corpus = ?
+            """,
+            (corpus,),
+        ).fetchall()
+        return [dict(r) for r in rows]
+
+    def rewrite_chunk_source_ref(
+        self, *, chunk_id: int, corpus: str, new_source_ref: str
+    ) -> str:
+        """Rewrite one chunk's source_ref to its stable replacement.
+
+        When a row with the target identity already exists (a duplicate
+        created by past source_ref drift), the rows are merged instead:
+        the legacy row's vectors move to the survivor wherever the
+        (chunk_id, run_id) slot is free, and the legacy row is deleted
+        (cascade drops any vector whose run slot the survivor already
+        fills). Returns 'rewritten' or 'merged'.
+        """
+        row = self._conn.execute(
+            """
+            SELECT item_key, source_type, chunk_index, chunker_version
+            FROM chunks WHERE chunk_id = ?
+            """,
+            (chunk_id,),
+        ).fetchone()
+        if row is None:
+            raise VectorStoreError(f"chunk_id {chunk_id} not found")
+        twin = self._conn.execute(
+            """
+            SELECT chunk_id FROM chunks
+            WHERE owner = 'local'
+              AND corpus = ?
+              AND item_key = ?
+              AND source_type = ?
+              AND source_ref = ?
+              AND chunk_index = ?
+              AND chunker_version = ?
+              AND chunk_id != ?
+            """,
+            (
+                corpus, row["item_key"], row["source_type"], new_source_ref,
+                row["chunk_index"], row["chunker_version"], chunk_id,
+            ),
+        ).fetchone()
+        if twin is None:
+            self._conn.execute(
+                "UPDATE chunks SET source_ref = ? WHERE chunk_id = ?",
+                (new_source_ref, chunk_id),
+            )
+            return "rewritten"
+        twin_id = int(twin["chunk_id"])
+        self._conn.execute("BEGIN IMMEDIATE")
+        try:
+            self._conn.execute(
+                "UPDATE OR IGNORE vectors SET chunk_id = ? WHERE chunk_id = ?",
+                (twin_id, chunk_id),
+            )
+            self._conn.execute("DELETE FROM chunks WHERE chunk_id = ?", (chunk_id,))
+            self._conn.execute("COMMIT")
+        except BaseException:
+            self._conn.execute("ROLLBACK")
+            raise
+        return "merged"
+
     def update_chunk_content(
         self,
         *,