diff --git a/src/partial_recall/corpus/adapters/folder.py b/src/partial_recall/corpus/adapters/folder.py index 33bb052..dd246c2 100644 --- a/src/partial_recall/corpus/adapters/folder.py +++ b/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 diff --git a/src/partial_recall/corpus/protocol.py b/src/partial_recall/corpus/protocol.py index 8d1faa9..95b585c 100644 --- a/src/partial_recall/corpus/protocol.py +++ b/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 diff --git a/src/partial_recall/index/pipeline.py b/src/partial_recall/index/pipeline.py index 6d6040f..d1c3532 100644 --- a/src/partial_recall/index/pipeline.py +++ b/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 diff --git a/src/partial_recall/store/vector_store.py b/src/partial_recall/store/vector_store.py index b37b0b0..98977c4 100644 --- a/src/partial_recall/store/vector_store.py +++ b/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, *, diff --git a/tests/test_corpus_folder.py b/tests/test_corpus_folder.py index a023548..78bdebb 100644 --- a/tests/test_corpus_folder.py +++ b/tests/test_corpus_folder.py @@ -251,3 +251,242 @@ def test_get_text_works_with_absolute_legacy_source_ref( text = adapter.get_text(item, legacy_source) assert text is not None assert "caste" in text.lower() + + +# --------------------------------------------------------------------------- +# Stable root ids + source_ref migration +# --------------------------------------------------------------------------- + + +def _two_roots(tmp_path: Path) -> tuple[Path, Path]: + a = tmp_path / "root_a" + b = tmp_path / "root_b" + for root in (a, b): + root.mkdir(parents=True) + (a / "alpha.md").write_text("alpha text", encoding="utf-8") + (b / "bravo.md").write_text("bravo text", encoding="utf-8") + return a, b + + +def _refs(adapter: FolderAdapter) -> dict[str, str]: + out: dict[str, str] = {} + for item in adapter.list_items(): + for source in adapter.get_sources(item): + out[item.title] = source.source_ref + return out + + +def test_source_refs_survive_root_reordering(tmp_path: Path) -> None: + """THE bug: source_ref used to embed the root's position in config, + so reordering roots orphaned every chunk and re-embedded the corpus.""" + a, b = _two_roots(tmp_path) + first = FolderAdapter(roots=[a, b]) + second = FolderAdapter(roots=[b, a]) + try: + assert _refs(first) == _refs(second) + finally: + first.close() + second.close() + + +def test_resolves_every_source_ref_format(tmp_path: Path) -> None: + a, b = _two_roots(tmp_path) + adapter = FolderAdapter(roots=[a, b]) + try: + stable = _refs(adapter)["alpha"] + expected = (a / "alpha.md").resolve() + assert adapter._resolve_source_ref(stable) == expected + # Legacy positional format still resolves through root order. + assert adapter._resolve_source_ref("0:alpha.md") == expected + # Legacy absolute path. + assert adapter._resolve_source_ref(str(expected)) == expected + # Bare relative fallback. + assert adapter._resolve_source_ref("alpha.md") == expected + finally: + adapter.close() + + +def test_migrate_source_refs_rewrites_and_merges(tmp_path: Path) -> None: + """Legacy rows (positional prefix + absolute path) are rewritten in + place; a legacy row whose target identity already exists is merged + with its vectors preserved on the survivor.""" + from datetime import UTC, datetime + + from partial_recall.store.vector_store import VectorStore + + a, b = _two_roots(tmp_path / "corpus") + adapter = FolderAdapter(roots=[a, b]) + store = VectorStore(tmp_path / "vectors.sqlite") + now = datetime.now(UTC).isoformat(timespec="seconds") + run_id = store.create_run( + provider="fake", model_name="fake", model_version="v1", + dimensions=4, quantization="int8", normalized=True, + distance_metric="cosine", chunker_name="c", chunker_version="v1", + started_at=now, + ) + try: + alpha_key = _stable_item_key(a / "alpha.md") + bravo_key = _stable_item_key(b / "bravo.md") + for key in (alpha_key, bravo_key): + store.upsert_item( + item_key=key, corpus="folder", item_type="file", title=key, + date=None, creators_json="[]", abstract=None, + metadata_hash=f"h-{key}", last_indexed_at=now, corpus_ref=None, + ) + # Legacy positional row (no stable twin) — should be rewritten. + store.insert_chunk( + item_key=alpha_key, corpus="folder", source_type="file", + source_ref="0:alpha.md", chunk_index=0, + char_offset_start=0, char_offset_end=10, text_hash="t1", + text_preview="alpha text", chunker_version="v1", indexed_at=now, + detected_locale=None, + ) + # Drift pair: a stable-format row (with the active vector) AND a + # legacy absolute-path row (with an older run's vector) for the + # same chunk — should merge, keeping both vectors. + stable_ref = _refs(adapter)["bravo"] + stable_row = store.insert_chunk( + item_key=bravo_key, corpus="folder", source_type="file", + source_ref=stable_ref, chunk_index=0, + char_offset_start=0, char_offset_end=10, text_hash="t2", + text_preview="bravo text", chunker_version="v1", indexed_at=now, + detected_locale=None, + ) + legacy_abs = store.insert_chunk( + item_key=bravo_key, corpus="folder", source_type="file", + source_ref=str((b / "bravo.md").resolve()), chunk_index=0, + char_offset_start=0, char_offset_end=10, text_hash="t2", + text_preview="bravo text", chunker_version="v1", indexed_at=now, + detected_locale=None, + ) + old_run = store.create_run( + provider="fake", model_name="fake", model_version="v1", + dimensions=4, quantization="int8", normalized=True, + distance_metric="cosine", chunker_name="c", chunker_version="v1", + started_at=now, + ) + store.insert_vector( + chunk_id=stable_row, run_id=run_id, + vector=b"\x7f\x00\x00\x00", norm=None, indexed_at=now, + ) + store.insert_vector( + chunk_id=legacy_abs, run_id=old_run, + vector=b"\x00\x7f\x00\x00", norm=None, indexed_at=now, + ) + # Unmappable row: absolute path outside every root — left alone. + outside = tmp_path / "outside.md" + outside.write_text("outside", encoding="utf-8") + outside_key = _stable_item_key(outside) + store.upsert_item( + item_key=outside_key, corpus="folder", item_type="file", + title="outside", date=None, creators_json="[]", abstract=None, + metadata_hash="h-out", last_indexed_at=now, corpus_ref=None, + ) + store.insert_chunk( + item_key=outside_key, corpus="folder", source_type="file", + source_ref=str(outside.resolve()), chunk_index=0, + char_offset_start=0, char_offset_end=7, text_hash="t3", + text_preview="outside", chunker_version="v1", indexed_at=now, + detected_locale=None, + ) + + counts = adapter.migrate_source_refs(store) + + assert counts == {"rewritten": 1, "merged": 1, "skipped": 1} + refs = { + r["source_ref"] + for r in store.iter_chunk_refs(corpus="folder") + } + assert _refs(adapter)["alpha"] in refs # positional → stable + assert "0:alpha.md" not in refs + assert str((b / "bravo.md").resolve()) not in refs # merged away + assert str(outside.resolve()) in refs # skipped, untouched + # Both vectors survive on the surviving bravo row. + rows = store._conn.execute( + "SELECT run_id FROM vectors WHERE chunk_id = ? ORDER BY run_id", + (stable_row,), + ).fetchall() + assert [r["run_id"] for r in rows] == [run_id, old_run] + # Idempotent: second run is a no-op. + assert adapter.migrate_source_refs(store) == { + "rewritten": 0, "merged": 0, "skipped": 1, + } + finally: + adapter.close() + store.close() + + +def test_reordered_roots_do_not_reembed(tmp_path: Path) -> None: + """End-to-end regression for the root-order footgun: index with roots + [A, B], reorder to [B, A], extend — nothing may be re-created or + re-embedded.""" + from partial_recall.index.pipeline import run_indexing + from partial_recall.store.vector_store import VectorStore + from tests.test_pipeline import FakeEmbeddingProvider + + a, b = _two_roots(tmp_path / "corpus") + store = VectorStore(tmp_path / "vectors.sqlite") + first_adapter = FolderAdapter(roots=[a, b]) + second_adapter = FolderAdapter(roots=[b, a]) + try: + first = run_indexing( + adapter=first_adapter, store=store, provider=FakeEmbeddingProvider(), + ) + assert first.chunk_count == 2 + result = run_indexing( + adapter=second_adapter, store=store, + provider=FakeEmbeddingProvider(), + extend_run_id=first.run_id, + ) + assert result.chunk_count == 0 + assert result.new_vector_count == 0 + assert result.skipped_chunk_count == 2 + finally: + first_adapter.close() + second_adapter.close() + store.close() + + +def test_migration_respects_original_root_after_reorder(tmp_path: Path) -> None: + """Codex review scenario: the DB was written with roots [A, B], the + user reorders config to [B, A], THEN upgrades. A legacy "0:alpha.md" + still belongs to A; mapping it through the current order would + assign it B's stable id. The item's corpus_ref (A's absolute path) + must win — even when B contains a same-named decoy file.""" + from datetime import UTC, datetime + + from partial_recall.store.vector_store import VectorStore + + a, b = _two_roots(tmp_path / "corpus") + (b / "alpha.md").write_text("decoy in B", encoding="utf-8") + reordered = FolderAdapter(roots=[b, a]) # B is now index 0 + store = VectorStore(tmp_path / "vectors.sqlite") + now = datetime.now(UTC).isoformat(timespec="seconds") + try: + alpha_path = (a / "alpha.md").resolve() + alpha_key = _stable_item_key(alpha_path) + store.upsert_item( + item_key=alpha_key, corpus="folder", item_type="file", + title="alpha", date=None, creators_json="[]", abstract=None, + metadata_hash="h", last_indexed_at=now, + corpus_ref=str(alpha_path), + ) + store.insert_chunk( + item_key=alpha_key, corpus="folder", source_type="file", + source_ref="0:alpha.md", chunk_index=0, + char_offset_start=0, char_offset_end=10, text_hash="t", + text_preview="alpha text", chunker_version="v1", indexed_at=now, + detected_locale=None, + ) + + counts = reordered.migrate_source_refs(store) + + assert counts == {"rewritten": 1, "merged": 0, "skipped": 0} + (migrated,) = store.iter_chunk_refs(corpus="folder") + assert migrated["source_ref"] == _refs(reordered)["alpha"] + # And the walk agrees: get_sources for A's alpha emits this exact ref. + item = next(i for i in reordered.list_items() if i.corpus_ref == str(alpha_path)) + assert next(reordered.get_sources(item)).source_ref == migrated["source_ref"] + finally: + reordered.close() + store.close()