From c73e9c8027c8a760f5875b9e91b6d449e0216749 Mon Sep 17 00:00:00 2001 From: ww26 Date: Wed, 19 Aug 2026 13:53:46 -0400 Subject: [PATCH 1/7] Multi-invocation incremental builds: --sync-key, changes/verify subcommands, zero-chunk snapshot commit, LEANN_NO_REGISTER (#2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(cli): --sync-key stable snapshot identity Keyed builds use one global FileSynchronizer over the expanded file manifest (snapshot sync_key_{sha256(key)[:12]}.pickle) instead of per-docs-list hashed snapshots, so varying --docs lists across invocations keep merkle incrementality. Key lifecycle persisted in sync_roots.json (reuse stored key; rekey requires --force). load_snapshot() now raises SnapshotCorruptError on unreadable pickles (missing file still means first build); keyed builds fail loudly on corruption instead of warn-and-skip. Also: remove-only deltas on non-IVF backends now trigger the full rebuild instead of returning early (previously gated on IVF). * feat(cli): leann changes subcommand Non-mutating merkle diff against the stored snapshot: prints one JSON document {added, modified, removed} (sorted) on stdout, diagnostics on stderr. Scope comes from --docs or, when omitted, from the index's stored sync_roots.json. Snapshot errors are strict (nonzero exit, no false-clean report) instead of build's warn-and-skip. LeannCLI no longer eagerly creates .leann/indexes at construction, so read-only commands leave the working directory untouched. * feat(cli): leann verify subcommand Cross-artifact integrity check: passages.jsonl/.idx offset and id-set consistency for any backend, plus IVF invariants (id-map exact inverses, id-map values vs offset-map keys, FAISS ntotal vs id-map cardinality, next_id monotonicity, missing index file). Exit 0 clean, 1 with a findings report. Covers vector/id-map/passage artifacts only (not BM25 sqlite or ids.txt). * fix(cli): commit snapshot on safely-empty incremental deltas An add/modify-only delta that loads successfully but produces zero chunks now commits the snapshot and sync config, so the same no-op delta is not rediscovered on every subsequent build. Loader failures and removal-bearing deltas still leave the snapshot untouched. Snapshot pickles and sync_roots.json now publish via tmp + os.replace so an interrupted write cannot corrupt existing state. * feat(registry): LEANN_NO_REGISTER env switch LEANN_NO_REGISTER=1 makes register_project_directory a no-op before any path traversal or ~/.leann creation, for automation running builds in disposable directories. * fix(cli): verify reads real on-disk IVF artifact names E2E smoke against a real IVF build showed verify looked for documents.leann.{meta.json → wrong via with_suffix, ivf_id_map.json, index}; the backend actually writes documents.leann.meta.json, documents.ivf_id_map.json and documents.index (id map and FAISS file use the stem without .leann). Test fixtures now mirror the real layout. * fix(cli): harden incremental build/changes/verify error paths from PR #2 review - refuse empty-delta snapshot commit when document loads failed (masked data loss) - changes: exit 1 on missing index, empty scope, or wrong --sync-key - corrupt sync snapshot / sync_roots.json fail loud; --force resets - sync: unreadable existing file keeps previous hash instead of 'removed' - verify: no crash on non-numeric id-map keys; no misleading findings when idx unreadable - drop dead 'changes --json' flag; register project dir on empty-delta commit - changelog entry for the PR's features * fix: address Gemini review findings for PR #2 - abort incremental build before mutation on any document load failure - verify: snapshot presence/unpickle checks, IVF type + mapped-id invlists check, type validation for decodable-but-malformed artifacts - verify test fixture builds a real IndexIVFFlat (was IndexFlatL2) - sync: os.walk onerror fails loud instead of mass-removal on unreadable subtree - changes: index-existence check applies with --docs too - sync_roots.json wrong-JSON-type handled by the --force recovery path 7 issues fixed, 0 false positives, 0 skipped See .doc/pr-review-comments/PR-2-multi-invocation-incremental.md * fix: watch survives fail-loud errors; changes rejects key on unkeyed index * fix: CI lint/type errors in new tests --- docs/CHANGELOG.md | 20 + packages/leann-core/src/leann/cli.py | 452 +++++++++++++++++++++- packages/leann-core/src/leann/registry.py | 2 + packages/leann-core/src/leann/sync.py | 30 +- tests/test_cli_changes.py | 367 ++++++++++++++++++ tests/test_cli_verify.py | 318 +++++++++++++++ tests/test_no_register.py | 46 +++ tests/test_sync.py | 43 ++ tests/test_sync_key.py | 367 ++++++++++++++++++ tests/test_watch_sync_scope.py | 49 +++ tests/test_zero_chunk_commit.py | 270 +++++++++++++ 11 files changed, 1941 insertions(+), 23 deletions(-) create mode 100644 tests/test_cli_changes.py create mode 100644 tests/test_cli_verify.py create mode 100644 tests/test_no_register.py create mode 100644 tests/test_sync_key.py create mode 100644 tests/test_zero_chunk_commit.py diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 6e3ba60c..0810a0d8 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -66,3 +66,23 @@ fixes). Newest entries at the bottom. already-working Windows `cp314` installs. Completing the wheel matrix is the correct fix; the next 0.3.8 patch release will carry full `cp314` coverage once CI confirms the new platforms build cleanly. + +## 2026-08-19: Multi-invocation incremental builds (`--sync-key`, `changes`, `verify`) + review hardening + +- `leann build --sync-key `: stable snapshot identity shared across invocations with + different `--docs` lists (single keyed Merkle snapshot instead of per-root snapshots). + Mismatched keys on a keyed index are rejected unless `--force` rekeys. +- New `leann changes` subcommand: non-mutating JSON report of pending added/modified/removed + files vs the stored snapshot. Errors (exit 1) on a missing index, empty sync scope, wrong + sync key, or corrupt snapshot instead of reporting a false clean delta. +- New `leann verify` subcommand: cross-artifact integrity check (meta.json, passages.jsonl, + passages.idx offsets, IVF id map inversion, FAISS vector count). +- Safely-empty incremental deltas (zero new chunks, nothing modified/removed) now commit the + snapshot so re-runs report "up to date" — but only when no document failed to load; a + swallowed loader failure aborts the build without committing, so the failed files stay pending. +- Corrupt sync snapshots and unreadable `sync_roots.json` now fail loud on build (recover + with `--force`) instead of silently degrading to full-rediff or unkeyed identity. +- A transiently unreadable file keeps its previous hash instead of being classified as + removed (which deleted its chunks from the index). +- `LEANN_NO_REGISTER=1` env switch skips project-directory registration (for tests/CI). + diff --git a/packages/leann-core/src/leann/cli.py b/packages/leann-core/src/leann/cli.py index 4cacdee2..677fe346 100644 --- a/packages/leann-core/src/leann/cli.py +++ b/packages/leann-core/src/leann/cli.py @@ -30,7 +30,13 @@ resolve_openai_api_key, resolve_openai_base_url, ) -from .sync import DEFAULT_INDEX_EXTENSIONS, FileSynchronizer, parse_include_extensions +from .sync import ( + DEFAULT_INDEX_EXTENSIONS, + FileSynchronizer, + SnapshotCorruptError, + _iter_directory_files, + parse_include_extensions, +) def _non_negative_int(value: str) -> int: @@ -222,7 +228,7 @@ class LeannCLI: def __init__(self): # Always use project-local .leann directory (like .git) self.indexes_dir = Path.cwd() / ".leann" / "indexes" - self.indexes_dir.mkdir(parents=True, exist_ok=True) + self._load_errors = 0 # Default parser for documents self.node_parser = SentenceSplitter( @@ -362,6 +368,12 @@ def create_parser(self) -> argparse.ArgumentParser: action="store_true", help="Force full rebuild of existing index (without this, build does incremental update: add new files only)", ) + build_parser.add_argument( + "--sync-key", + type=str, + default=None, + help="Stable identity key for change tracking: one global snapshot keyed by this value, independent of the exact --docs invocation", + ) build_parser.add_argument( "--graph-degree", type=int, default=32, help="Graph degree (default: 32)" ) @@ -896,6 +908,33 @@ def _add_index_args(p, default_name): "--force", "-f", action="store_true", help="Force removal without confirmation" ) + # Changes command (non-mutating diff vs stored snapshot) + changes_parser = subparsers.add_parser( + "changes", help="Show pending file changes vs the stored index snapshot" + ) + changes_parser.add_argument("index_name", help="Index name") + changes_parser.add_argument( + "--docs", + type=str, + nargs="+", + default=None, + help="Scope to diff (default: stored sync_roots.json scope)", + ) + changes_parser.add_argument( + "--sync-key", type=str, default=None, help="Sync key (default: stored key)" + ) + changes_parser.add_argument( + "--file-types", type=str, default=None, help="Comma-separated extensions filter" + ) + changes_parser.add_argument( + "--include-hidden", action="store_true", help="Include hidden files" + ) + + verify_parser = subparsers.add_parser( + "verify", help="Verify cross-artifact integrity of an index" + ) + verify_parser.add_argument("index_name", help="Index name") + # Serve command (HTTP API server) serve_parser = subparsers.add_parser( "serve", help="Start HTTP API server for LEANN vector DB" @@ -1555,6 +1594,7 @@ def load_documents( include_hidden: bool = False, args: Optional[dict[str, Any]] = None, ): + self._load_errors = 0 # Handle both single path (string) and multiple paths (list) for backward compatibility if isinstance(docs_paths, str): docs_paths = [docs_paths] @@ -1574,6 +1614,7 @@ def load_documents( directories.append(str(path_obj)) else: print(f"⚠️ Warning: Path '{path}' does not exist, skipping...") + self._load_errors += 1 continue # Print summary of what we're processing @@ -1641,9 +1682,11 @@ def _path_has_hidden_segment(p: Path) -> bool: ) except Exception as e: print(f" ❌ Warning: Could not load files from {parent_dir}: {e}") + self._load_errors += 1 except Exception as e: print(f"❌ Error processing individual files: {e}") + self._load_errors += 1 # Define file extensions to process if custom_file_types: @@ -1719,6 +1762,7 @@ def _path_has_hidden_segment(p: Path) -> bool: documents.extend(default_docs) except Exception as e: print(f"Warning: Could not process {file_path}: {e}") + self._load_errors += 1 # Load other file types with default reader # Exclude PDFs from code_extensions if they were already processed separately @@ -2006,34 +2050,73 @@ def _create_synchronizers( explicit_files: list[str], include_extensions: list[str], include_hidden: bool = False, + sync_key: Optional[str] = None, + strict: bool = False, + reset_corrupt: bool = False, ) -> list[FileSynchronizer]: """Create FileSynchronizers with snapshots stored in the index dir. Shared by build and watch.""" + + def _init(**kw) -> FileSynchronizer: + try: + return FileSynchronizer(**kw) + except SnapshotCorruptError: + if not reset_corrupt: + raise + Path(kw["snapshot_path"]).unlink(missing_ok=True) + return FileSynchronizer(**kw) + + if sync_key: + manifest = list(explicit_files) + for root in directories: + manifest.extend(_iter_directory_files(root, include_extensions, include_hidden)) + tag = hashlib.sha256(sync_key.encode()).hexdigest()[:12] + # Keyed path fails loud (SnapshotCorruptError propagates) — no warn-and-skip. + return [ + _init( + explicit_files=sorted(set(manifest)), + include_extensions=include_extensions, + include_hidden=include_hidden, + snapshot_path=str(index_dir / f"sync_key_{tag}.pickle"), + ) + ] synchronizers: list[FileSynchronizer] = [] for root in directories: tag = hashlib.sha256(root.encode()).hexdigest()[:12] snapshot_path = str(index_dir / f"sync_{tag}.pickle") try: - fs = FileSynchronizer( + fs = _init( root_dir=root, include_extensions=include_extensions, include_hidden=include_hidden, snapshot_path=snapshot_path, ) synchronizers.append(fs) + except SnapshotCorruptError as exc: + raise SnapshotCorruptError( + f"{exc}. Re-run with --force to reset the snapshot." + ) from exc except Exception as exc: + if strict: + raise print(f"Warning: Failed to init synchronizer for {root}: {exc}") if explicit_files: tag = hashlib.sha256("|".join(explicit_files).encode()).hexdigest()[:12] snapshot_path = str(index_dir / f"sync_files_{tag}.pickle") try: - fs = FileSynchronizer( + fs = _init( explicit_files=explicit_files, include_extensions=include_extensions, include_hidden=include_hidden, snapshot_path=snapshot_path, ) synchronizers.append(fs) + except SnapshotCorruptError as exc: + raise SnapshotCorruptError( + f"{exc}. Re-run with --force to reset the snapshot." + ) from exc except Exception as exc: + if strict: + raise print(f"Warning: Failed to init synchronizer for explicit files: {exc}") return synchronizers @@ -2043,12 +2126,20 @@ def _build_synchronizers( index_dir: Path, file_types: Optional[str] = None, include_hidden: bool = False, + sync_key: Optional[str] = None, + reset_corrupt: bool = False, ) -> list[FileSynchronizer]: """Create FileSynchronizers for build from docs_paths.""" directories, files = self._resolve_sync_scope(docs_paths) include_extensions = self._parse_file_types(file_types) return self._create_synchronizers( - index_dir, directories, files, include_extensions, include_hidden + index_dir, + directories, + files, + include_extensions, + include_hidden, + sync_key=sync_key, + reset_corrupt=reset_corrupt, ) def _detect_build_changes( @@ -2329,6 +2420,7 @@ def _write_sync_config( include_extensions: list[str], include_hidden: bool, build_config: Optional[dict[str, Any]] = None, + sync_key: Optional[str] = None, ) -> None: sync_config_path = index_dir / "sync_roots.json" config = { @@ -2338,10 +2430,14 @@ def _write_sync_config( "include_extensions": include_extensions, "ignore_patterns": self._sync_ignore_patterns(include_hidden), } + if sync_key: + config["sync_key"] = sync_key if build_config is not None: config["build_config"] = build_config - with open(sync_config_path, "w", encoding="utf-8") as f: + tmp_path = sync_config_path.with_suffix(sync_config_path.suffix + ".tmp") + with open(tmp_path, "w", encoding="utf-8") as f: json.dump(config, f, indent=2) + os.replace(tmp_path, sync_config_path) def _write_sync_config_for_docs( self, @@ -2358,6 +2454,7 @@ def _write_sync_config_for_docs( self._parse_file_types(args.file_types), args.include_hidden, build_config, + sync_key=getattr(args, "sync_key", None), ) def _load_sync_scope(self, index_dir: Path) -> tuple[list[str], list[str], list[str], bool]: @@ -2370,12 +2467,29 @@ def _load_sync_scope(self, index_dir: Path) -> tuple[list[str], list[str], list[ config = json.load(f) except (json.JSONDecodeError, OSError): return [], [], list(DEFAULT_INDEX_EXTENSIONS), False + if not isinstance(config, dict): + return [], [], list(DEFAULT_INDEX_EXTENSIONS), False directories = config.get("directories") or config.get("roots") or [] files = config.get("files") or [] include_extensions = config.get("include_extensions") or list(DEFAULT_INDEX_EXTENSIONS) include_hidden = config.get("ignore_patterns") is None return directories, files, include_extensions, include_hidden + def _load_stored_sync_key(self, index_dir: Path) -> Optional[str]: + sync_config_path = index_dir / "sync_roots.json" + if not sync_config_path.exists(): + return None + try: + with open(sync_config_path, encoding="utf-8") as f: + config = json.load(f) + if not isinstance(config, dict): + raise json.JSONDecodeError("not a JSON object", "", 0) + return config.get("sync_key") + except (json.JSONDecodeError, OSError) as exc: + # Treating an unreadable config as "unkeyed" would silently bypass the + # sync-key mismatch guard and rekey the snapshot identity. + raise ValueError(f"sync_roots.json unreadable at {sync_config_path}: {exc}") from exc + def _load_sync_roots(self, index_dir: Path) -> list[str]: """Load directory + explicit file paths for chunk ID lookup.""" directories, files, _, _ = self._load_sync_scope(index_dir) @@ -2517,11 +2631,31 @@ async def build_index(self, args): ) # Detect changes first so we can skip load_documents for remove-only + try: + stored_key = self._load_stored_sync_key(index_dir) + except ValueError: + if not args.force: + raise + stored_key = None # --force rewrites sync_roots.json anyway + requested_key = getattr(args, "sync_key", None) + if requested_key is None: + args.sync_key = stored_key + elif stored_key is not None and requested_key != stored_key and not args.force: + raise ValueError( + f"Index '{index_name}' is keyed with sync key '{stored_key}'; " + f"got '{requested_key}'. Use --force to rekey." + ) index_dir.mkdir(parents=True, exist_ok=True) synchronizers = self._build_synchronizers( - docs_paths, index_dir, file_types=args.file_types, include_hidden=args.include_hidden + docs_paths, + index_dir, + file_types=args.file_types, + include_hidden=args.include_hidden, + sync_key=args.sync_key, + reset_corrupt=args.force, ) + all_texts: list[dict] | None = None if index_dir.exists() and not args.force and synchronizers: meta_path = index_dir / "documents.leann.meta.json" new_paths, removed_paths, modified_paths = self._detect_build_changes(synchronizers) @@ -2606,9 +2740,19 @@ def _normalize_model_name(name: str) -> str: include_hidden=args.include_hidden, args=args, ) + if self._load_errors: + # Proceeding would mutate the index (IVF removes old chunks before + # re-insert) and commit the failed files as indexed. + raise RuntimeError( + f"{self._load_errors} path(s) failed to load; incremental update " + f"aborted before modifying the index. Fix the inputs and re-run." + ) # Proceed even when all_texts is empty (e.g. file emptied): we still need to remove old chunks - if not all_texts and not (can_ivf_update and (modified_paths or removed_paths)): + if not all_texts and not (modified_paths or removed_paths): print("No documents found") + self._commit_synchronizers(synchronizers) + self._write_sync_config_for_docs(index_dir, docs_paths, args, build_config) + self.register_project_dir() return if can_ivf_update and (new_paths or modified_paths or removed_paths): @@ -2650,9 +2794,7 @@ def _normalize_model_name(name: str) -> str: ) # Full rebuild: load documents if not already loaded (first build or force) - try: - _ = all_texts - except NameError: + if all_texts is None: all_texts = self.load_documents( docs_paths, args.file_types, include_hidden=args.include_hidden, args=args ) @@ -2692,6 +2834,7 @@ def _normalize_model_name(name: str) -> str: target_index_dir, file_types=args.file_types, include_hidden=args.include_hidden, + sync_key=args.sync_key, ) if publish_from_staging else synchronizers @@ -2725,14 +2868,274 @@ def _watch_check_changes(self, index_name: str) -> tuple[set[str], set[str], set if not directories and not files: return set(), set(), set() - synchronizers = self._create_synchronizers( - index_dir, - directories, - files, - include_extensions, - include_hidden, + # Watch must survive transient failures (corrupt config/snapshot, unreadable + # subtree) that build/changes fail loud on: skip the tick, keep watching. + try: + synchronizers = self._create_synchronizers( + index_dir, + directories, + files, + include_extensions, + include_hidden, + sync_key=self._load_stored_sync_key(index_dir), + ) + return self._detect_build_changes(synchronizers) + except (ValueError, SnapshotCorruptError, OSError) as exc: + print(f"Warning: watch tick skipped: {exc}") + return set(), set(), set() + + def changes_command(self, args) -> int: + """Report pending file changes vs the stored snapshot without mutating anything.""" + index_dir = self.indexes_dir / args.index_name + if not index_dir.exists(): + print(f"Error: index '{args.index_name}' not found at {index_dir}", file=sys.stderr) + return 1 + if args.docs: + directories, files = self._resolve_sync_scope(args.docs) + include_extensions = self._parse_file_types(args.file_types) + include_hidden = args.include_hidden + else: + directories, files, include_extensions, include_hidden = self._load_sync_scope( + index_dir + ) + if not directories and not files: + print( + f"Error: no sync scope recorded for index '{args.index_name}' " + f"(missing or unreadable sync_roots.json); pass --docs", + file=sys.stderr, + ) + return 1 + try: + stored_key = self._load_stored_sync_key(index_dir) + except ValueError as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 + if args.sync_key and args.sync_key != stored_key: + # stored_key None included: a key against an unkeyed index would diff + # a never-written snapshot and report every file as added. + print( + f"Error: index '{args.index_name}' is keyed with sync key " + f"'{stored_key}'; got '{args.sync_key}'.", + file=sys.stderr, + ) + return 1 + sync_key = args.sync_key or stored_key + try: + synchronizers = self._create_synchronizers( + index_dir, + directories, + files, + include_extensions, + include_hidden, + sync_key=sync_key, + strict=True, + ) + added, removed, modified = self._detect_build_changes(synchronizers) + except SnapshotCorruptError as exc: + print(f"Error: {exc}", file=sys.stderr) + return 1 + print( + json.dumps( + { + "added": sorted(added), + "modified": sorted(modified), + "removed": sorted(removed), + } + ) ) - return self._detect_build_changes(synchronizers) + return 0 + + def verify_command(self, args) -> int: + """Verify cross-artifact integrity of an index; print findings, return 0 if healthy.""" + prefix = self.indexes_dir / args.index_name / "documents.leann" + findings: list[str] = [] + + meta = None + try: + meta = json.loads(Path(str(prefix) + ".meta.json").read_text(encoding="utf-8")) + except Exception as exc: + findings.append(f"meta.json unreadable: {exc}") + + jsonl_path = Path(str(prefix) + ".passages.jsonl") + jsonl_ids: list[str] = [] + try: + with open(jsonl_path, encoding="utf-8") as f: + for lineno, line in enumerate(f, 1): + try: + pid = json.loads(line)["id"] + except Exception as exc: + findings.append(f"passages.jsonl line {lineno} unparseable: {exc}") + continue + jsonl_ids.append(pid) + except Exception as exc: + findings.append(f"passages.jsonl unreadable: {exc}") + if len(set(jsonl_ids)) != len(jsonl_ids): + findings.append("passages.jsonl contains duplicate ids") + + offsets: dict[str, int] = {} + offsets_ok = False + try: + with open(str(prefix) + ".passages.idx", "rb") as f: + loaded_offsets = pickle.load(f) + if isinstance(loaded_offsets, dict): + offsets = loaded_offsets + offsets_ok = True + else: + findings.append(f"passages.idx is not a dict (got {type(loaded_offsets).__name__})") + except Exception as exc: + findings.append(f"passages.idx unreadable: {exc}") + + if offsets_ok and (offsets or jsonl_ids): + if len(offsets) != len(jsonl_ids): + findings.append( + f"passages.idx has {len(offsets)} entries but " + f"passages.jsonl has {len(jsonl_ids)} lines" + ) + if set(offsets) != set(jsonl_ids): + findings.append("passages.idx keys do not match passages.jsonl ids") + if jsonl_path.exists(): + with open(jsonl_path, "rb") as f: + for pid, offset in offsets.items(): + try: + f.seek(offset) + record = json.loads(f.readline().decode("utf-8")) + if record["id"] != pid: + findings.append( + f"offset for id {pid!r} points at id {record['id']!r}" + ) + except Exception as exc: + findings.append(f"offset for id {pid!r} invalid: {exc}") + + if meta and meta.get("backend_name") == "ivf": + findings.extend(self._verify_ivf(prefix, offsets if offsets_ok else None)) + + findings.extend(self._verify_snapshots(prefix.parent)) + + for finding in findings: + print(finding) + return 1 if findings else 0 + + def _verify_snapshots(self, index_dir: Path) -> list[str]: + """Check the sync snapshots the recorded scope implies exist and unpickle.""" + findings: list[str] = [] + if not (index_dir / "sync_roots.json").exists(): + return findings + try: + stored_key = self._load_stored_sync_key(index_dir) + except ValueError as exc: + return [str(exc)] + + snapshot_paths: list[Path] = [] + if stored_key: + tag = hashlib.sha256(stored_key.encode()).hexdigest()[:12] + snapshot_paths.append(index_dir / f"sync_key_{tag}.pickle") + else: + directories, files, _, _ = self._load_sync_scope(index_dir) + for root in directories: + tag = hashlib.sha256(root.encode()).hexdigest()[:12] + snapshot_paths.append(index_dir / f"sync_{tag}.pickle") + if files: + tag = hashlib.sha256("|".join(files).encode()).hexdigest()[:12] + snapshot_paths.append(index_dir / f"sync_files_{tag}.pickle") + + for snapshot_path in snapshot_paths: + if not snapshot_path.exists(): + findings.append( + f"sync snapshot missing: {snapshot_path.name} " + f"(build may have been interrupted before snapshot commit)" + ) + continue + try: + with open(snapshot_path, "rb") as f: + pickle.load(f) + except Exception as exc: + findings.append(f"sync snapshot {snapshot_path.name} corrupt: {exc}") + return findings + + def _verify_ivf(self, prefix: Path, offsets: Optional[dict[str, int]]) -> list[str]: + findings: list[str] = [] + # The IVF backend writes its artifacts against the stem without ".leann" + # (documents.ivf_id_map.json / documents.index), unlike the passage files. + stem = prefix.parent / prefix.name.removesuffix(".leann") + try: + id_map = json.loads(Path(str(stem) + ".ivf_id_map.json").read_text(encoding="utf-8")) + id_to_passage = id_map["id_to_passage"] + passage_to_id = id_map["passage_to_id"] + next_id = id_map["next_id"] + except Exception as exc: + return [f"ivf_id_map.json unreadable: {exc}"] + if not isinstance(id_to_passage, dict) or not isinstance(passage_to_id, dict): + return ["ivf_id_map.json: id_to_passage/passage_to_id are not dicts"] + if not isinstance(next_id, int): + findings.append(f"ivf_id_map.json: next_id is not an integer (got {next_id!r})") + next_id = None + + for key in id_to_passage: + try: + if int(key) < 0: + findings.append(f"id_to_passage key {key!r} is negative") + except ValueError: + findings.append(f"id_to_passage key {key!r} is not an integer") + if len(id_to_passage) != len(passage_to_id): + findings.append("id_to_passage and passage_to_id have different sizes") + for key, pid in id_to_passage.items(): + if str(passage_to_id.get(pid)) != key: + findings.append(f"id_to_passage[{key!r}]={pid!r} is not inverted in passage_to_id") + if offsets is not None and set(id_to_passage.values()) != set(offsets): + findings.append("id_to_passage values do not match passages.idx keys") + numeric_ids = [ + int(k) for k in id_to_passage if isinstance(k, str) and k.lstrip("-").isdigit() + ] + if numeric_ids and next_id is not None: + max_id = max(numeric_ids) + if next_id <= max_id: + findings.append(f"next_id {next_id} is not greater than max id {max_id}") + + index_path = Path(str(stem) + ".index") + if not index_path.exists(): + findings.append("missing .index file for ivf index") + return findings + try: + try: + import faiss + except ImportError: + from leann_backend_hnsw import faiss + index = faiss.read_index(str(index_path)) + if index.ntotal != len(id_to_passage): + findings.append( + f".index has {index.ntotal} vectors but id map has {len(id_to_passage)}" + ) + # extract_index_ivf instead of isinstance: the index may have been + # written by a different faiss build (SWIG classes don't compare). + try: + ivf_index = faiss.extract_index_ivf(index) + except Exception: + ivf_index = None + if ivf_index is None: + findings.append(f".index is not an IVF index (got {type(index).__name__})") + else: + # faiss does not persist the direct map, so enumerate the stored + # ids from the inverted lists (read-only) and check every mapped + # id actually exists in the index. + invlists = getattr(faiss.downcast_index(index), "invlists", None) + if invlists is not None: + stored_ids: set[int] = set() + for list_no in range(ivf_index.nlist): + list_size = invlists.list_size(list_no) + if list_size: + ids_ptr = invlists.get_ids(list_no) + stored_ids.update( + int(x) for x in faiss.rev_swig_ptr(ids_ptr, list_size) + ) + invlists.release_ids(list_no, ids_ptr) + missing = [i for i in numeric_ids if i not in stored_ids] + if missing: + findings.append( + f".index is missing {len(missing)} of {len(numeric_ids)} mapped ids" + ) + except Exception as exc: + findings.append(f".index unreadable: {exc}") + return findings def _watch_report_changes( self, @@ -2793,7 +3196,7 @@ def _reconstruct_build_args( return None with open(sync_config_path, encoding="utf-8") as f: config = json.load(f) - roots = config.get("roots") or [] + roots = (config.get("roots") or []) + (config.get("files") or []) if not roots: if verbose: print(f"Cannot rebuild '{index_name}': sync config has no document roots.") @@ -2869,6 +3272,9 @@ def config_or_backend(config_key: str, backend_key: str) -> Any: if include_hidden: build_args_list.append("--include-hidden") + if config.get("sync_key"): + build_args_list.extend(["--sync-key", config["sync_key"]]) + add_option("--doc-chunk-size", build_config.get("doc_chunk_size")) add_option("--doc-chunk-overlap", build_config.get("doc_chunk_overlap")) add_option("--code-chunk-size", build_config.get("code_chunk_size")) @@ -3808,6 +4214,14 @@ async def run(self, args=None): elif args.command == "build": with suppress_cpp_output(suppress): await self.build_index(args) + elif args.command == "changes": + rc = self.changes_command(args) + if rc: + sys.exit(rc) + elif args.command == "verify": + rc = self.verify_command(args) + if rc: + sys.exit(rc) elif args.command == "watch": await self.watch_index(args) elif args.command == "migrate-ids": diff --git a/packages/leann-core/src/leann/registry.py b/packages/leann-core/src/leann/registry.py index b950ee65..3a6f9aa0 100644 --- a/packages/leann-core/src/leann/registry.py +++ b/packages/leann-core/src/leann/registry.py @@ -94,6 +94,8 @@ def register_project_directory( max_depth: Maximum directory depth used when looking for App-format indexes. None preserves full-depth discovery for existing API callers. """ + if os.environ.get("LEANN_NO_REGISTER") == "1": + return if project_dir is None: project_dir = Path.cwd() else: diff --git a/packages/leann-core/src/leann/sync.py b/packages/leann-core/src/leann/sync.py index fa04bd1f..cdf07bef 100644 --- a/packages/leann-core/src/leann/sync.py +++ b/packages/leann-core/src/leann/sync.py @@ -60,6 +60,10 @@ ] +class SnapshotCorruptError(Exception): + """Raised when a sync snapshot exists but cannot be unpickled.""" + + def hash_data(data: str | bytes): if isinstance(data, str): data = data.encode() @@ -97,8 +101,13 @@ def _iter_directory_files( if not root.is_dir(): return [] + def _walk_error(exc: OSError) -> None: + # A silently skipped subtree would make its previously indexed files + # look removed, deleting their chunks. + raise exc + paths: list[str] = [] - for dirpath, dirnames, filenames in os.walk(root): + for dirpath, dirnames, filenames in os.walk(root, onerror=_walk_error): if not include_hidden: dirnames[:] = [d for d in dirnames if not d.startswith(".")] current = Path(dirpath) @@ -189,7 +198,7 @@ def __init__( self.explicit_files = ( [str(Path(path).resolve()) for path in explicit_files] if explicit_files else [] ) - if self.root_dir is None and not self.explicit_files: + if self.root_dir is None and not self.explicit_files and snapshot_path is None: raise ValueError("FileSynchronizer requires root_dir and/or explicit_files") if self.root_dir is not None and not os.path.isdir(self.root_dir): raise ValueError("This is not a valid directory") @@ -230,7 +239,16 @@ def generate_file_hashes(self): try: file_hashes[file_path] = _hash_file_bytes(Path(file_path)) except OSError: - logger.warning("Cannot hash file %s", file_path) + # Carry the old hash forward so a transiently unreadable file is + # not classified as removed (which would delete its chunks). + prev = ( + self.tree.root.children.get(file_path) if self.tree and self.tree.root else None + ) + if prev is not None: + logger.warning("Cannot hash file %s; keeping previous hash", file_path) + file_hashes[file_path] = prev.data + else: + logger.warning("Cannot hash new file %s; skipping", file_path) return file_hashes def build_merkle_tree(self, file_hashes): @@ -289,8 +307,10 @@ def snapshot_path(self): def save_snapshot(self): assert self.tree is not None - with open(self.snapshot_path, "wb") as f: + tmp_path = f"{self.snapshot_path}.tmp" + with open(tmp_path, "wb") as f: pickle.dump(self.tree, f) + os.replace(tmp_path, self.snapshot_path) def load_snapshot(self): try: @@ -298,3 +318,5 @@ def load_snapshot(self): self.tree = pickle.load(f) except FileNotFoundError: self.tree = None + except Exception as exc: + raise SnapshotCorruptError(f"Corrupt sync snapshot at {self.snapshot_path}") from exc diff --git a/tests/test_cli_changes.py b/tests/test_cli_changes.py new file mode 100644 index 00000000..31804b91 --- /dev/null +++ b/tests/test_cli_changes.py @@ -0,0 +1,367 @@ +"""Failing tests for `leann changes` (non-mutating merkle diff vs stored snapshot).""" + +import asyncio +import hashlib +import json +from pathlib import Path + +from leann.cli import LeannCLI + + +def _keyed_snapshot_name(key: str) -> str: + return f"sync_key_{hashlib.sha256(key.encode()).hexdigest()[:12]}.pickle" + + +def _make_fake_builder(): + class FakeBuilder: + def __init__(self, **kwargs): + self.kwargs = kwargs + + def add_text(self, _text, metadata=None): + pass + + def build_index(self, index_path): + target_dir = Path(index_path).parent + target_dir.mkdir(parents=True, exist_ok=True) + (target_dir / "documents.leann.meta.json").write_text( + json.dumps( + { + "backend_name": "hnsw", + "embedding_model": self.kwargs.get("embedding_model", "m"), + "embedding_mode": self.kwargs.get( + "embedding_mode", "sentence-transformers" + ), + "passage_id_scheme": "sequential", + "backend_kwargs": { + "graph_degree": 32, + "complexity": 64, + "num_threads": 1, + "is_compact": False, + "is_recompute": True, + }, + } + ), + encoding="utf-8", + ) + (target_dir / "documents.leann.index").write_bytes(b"fake") + (target_dir / "documents.leann.passages.jsonl").write_text("", encoding="utf-8") + + def update_index(self, index_path): + pass + + return FakeBuilder + + +def _fake_load_documents(docs_paths, custom_file_types=None, include_hidden=False, args=None): + if isinstance(docs_paths, str): + docs_paths = [docs_paths] + files: list[Path] = [] + for p in docs_paths: + path = Path(p) + if path.is_dir(): + files.extend(sorted(path.rglob("*.txt"))) + elif path.is_file(): + files.append(path) + return [{"text": f.name, "metadata": {"file_path": str(f.resolve())}} for f in files] + + +def _wire_cli(monkeypatch) -> LeannCLI: + cli = LeannCLI() + monkeypatch.setattr(cli, "load_documents", _fake_load_documents) + monkeypatch.setattr(cli, "register_project_dir", lambda: None) + monkeypatch.setattr("leann.cli.LeannBuilder", _make_fake_builder()) + return cli + + +def _build_args(index_name: str, docs: list[str], extra: list[str] | None = None) -> list[str]: + return [ + "build", + index_name, + "--docs", + *docs, + "--backend-name", + "hnsw", + "--no-compact", + "--embedding-model", + "m", + "--embedding-mode", + "sentence-transformers", + *(extra or []), + ] + + +def _run_changes(cli: LeannCLI, argv: list[str]) -> int: + args = cli.create_parser().parse_args(argv) + handler = cli.changes_command + try: + result = handler(args) + if asyncio.iscoroutine(result): + result = asyncio.run(result) + return int(result or 0) + except SystemExit as e: + return int(e.code or 0) + + +def _hash_tree(root: Path) -> dict[str, str]: + return { + str(p.relative_to(root)): hashlib.sha256(p.read_bytes()).hexdigest() + for p in sorted(root.rglob("*")) + if p.is_file() + } + + +def test_changes_reports_modified_and_added_as_single_sorted_json_doc( + tmp_path, monkeypatch, capsys +): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + file_a = docs / "a.txt" + file_b = docs / "b.txt" + file_a.write_text("alpha", encoding="utf-8") + file_b.write_text("beta", encoding="utf-8") + cli = _wire_cli(monkeypatch) + asyncio.run( + cli.build_index( + cli.create_parser().parse_args( + _build_args("idx", [str(docs)], ["--sync-key", "corpus-v1"]) + ) + ) + ) + file_a.write_text("alpha modified", encoding="utf-8") + file_c = docs / "c.txt" + file_c.write_text("gamma", encoding="utf-8") + capsys.readouterr() + + # Act + rc = _run_changes(cli, ["changes", "idx", "--docs", str(docs), "--sync-key", "corpus-v1"]) + out = capsys.readouterr().out + + # Assert + assert rc == 0 + report = json.loads(out) + assert report["modified"] == [str(file_a.resolve())] + assert report["added"] == [str(file_c.resolve())] + assert report["removed"] == [] + assert report["added"] == sorted(report["added"]) + assert report["modified"] == sorted(report["modified"]) + + +def test_changes_is_non_mutating_and_never_commits_snapshot(tmp_path, monkeypatch, capsys): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch) + asyncio.run( + cli.build_index( + cli.create_parser().parse_args( + _build_args("idx", [str(docs)], ["--sync-key", "corpus-v1"]) + ) + ) + ) + (docs / "b.txt").write_text("beta", encoding="utf-8") + leann_dir = tmp_path / ".leann" + hashes_before = _hash_tree(leann_dir) + capsys.readouterr() + + # Act + argv = ["changes", "idx", "--docs", str(docs), "--sync-key", "corpus-v1"] + rc_first = _run_changes(cli, argv) + out_first = capsys.readouterr().out + rc_second = _run_changes(cli, argv) + out_second = capsys.readouterr().out + + # Assert + assert rc_first == 0 + assert rc_second == 0 + assert json.loads(out_first) == json.loads(out_second) + assert json.loads(out_first)["added"] == [str((docs / "b.txt").resolve())] + assert _hash_tree(leann_dir) == hashes_before + + +def test_changes_without_docs_uses_stored_sync_roots_scope(tmp_path, monkeypatch, capsys): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch) + asyncio.run(cli.build_index(cli.create_parser().parse_args(_build_args("idx", [str(docs)])))) + new_file = docs / "new.txt" + new_file.write_text("new", encoding="utf-8") + capsys.readouterr() + + # Act + rc = _run_changes(cli, ["changes", "idx"]) + report = json.loads(capsys.readouterr().out) + + # Assert + assert rc == 0 + assert report["added"] == [str(new_file.resolve())] + assert report["modified"] == [] + assert report["removed"] == [] + + +def test_changes_reports_empty_delta_immediately_after_build(tmp_path, monkeypatch, capsys): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch) + asyncio.run( + cli.build_index( + cli.create_parser().parse_args( + _build_args("idx", [str(docs)], ["--sync-key", "corpus-v1"]) + ) + ) + ) + capsys.readouterr() + + # Act + rc = _run_changes(cli, ["changes", "idx", "--docs", str(docs), "--sync-key", "corpus-v1"]) + + # Assert + assert rc == 0 + assert json.loads(capsys.readouterr().out) == {"added": [], "modified": [], "removed": []} + + +def test_changes_with_corrupt_snapshot_exits_nonzero_without_false_clean_report( + tmp_path, monkeypatch, capsys +): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch) + asyncio.run( + cli.build_index( + cli.create_parser().parse_args( + _build_args("idx", [str(docs)], ["--sync-key", "corpus-v1"]) + ) + ) + ) + snapshot = tmp_path / ".leann" / "indexes" / "idx" / _keyed_snapshot_name("corpus-v1") + assert snapshot.exists() + snapshot.write_bytes(b"not a pickle at all") + capsys.readouterr() + + # Act + rc = _run_changes(cli, ["changes", "idx", "--docs", str(docs), "--sync-key", "corpus-v1"]) + out = capsys.readouterr().out + + # Assert + assert rc != 0 + if out.strip(): + assert json.loads(out) != {"added": [], "modified": [], "removed": []} + + +def test_cli_construction_does_not_create_indexes_dir_in_cwd(tmp_path, monkeypatch): + # Arrange + fresh = tmp_path / "fresh" + fresh.mkdir() + monkeypatch.chdir(fresh) + + # Act + LeannCLI() + + # Assert + assert not (fresh / ".leann" / "indexes").exists() + + +def test_changes_on_missing_index_exits_nonzero(tmp_path, monkeypatch, capsys): + # Arrange + monkeypatch.chdir(tmp_path) + cli = _wire_cli(monkeypatch) + + # Act + rc = _run_changes(cli, ["changes", "no-such-index"]) + captured = capsys.readouterr() + + # Assert + assert rc != 0 + assert "not found" in captured.err + assert captured.out.strip() == "" + + +def test_changes_with_wrong_sync_key_exits_nonzero(tmp_path, monkeypatch, capsys): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch) + asyncio.run( + cli.build_index( + cli.create_parser().parse_args( + _build_args("idx", [str(docs)], ["--sync-key", "corpus-v1"]) + ) + ) + ) + capsys.readouterr() + + # Act + rc = _run_changes(cli, ["changes", "idx", "--docs", str(docs), "--sync-key", "typo-key"]) + captured = capsys.readouterr() + + # Assert + assert rc != 0 + assert "corpus-v1" in captured.err + assert captured.out.strip() == "" + + +def test_changes_with_empty_scope_and_no_docs_exits_nonzero(tmp_path, monkeypatch, capsys): + # Arrange: index dir exists but has no sync_roots.json + monkeypatch.chdir(tmp_path) + index_dir = tmp_path / ".leann" / "indexes" / "idx" + index_dir.mkdir(parents=True) + cli = _wire_cli(monkeypatch) + + # Act + rc = _run_changes(cli, ["changes", "idx"]) + captured = capsys.readouterr() + + # Assert + assert rc != 0 + assert "sync scope" in captured.err + assert captured.out.strip() == "" + + +def test_changes_on_missing_index_with_docs_exits_nonzero(tmp_path, monkeypatch, capsys): + # Arrange: --docs given but the index itself does not exist (likely a typo) + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch) + + # Act + rc = _run_changes(cli, ["changes", "no-such-index", "--docs", str(docs)]) + captured = capsys.readouterr() + + # Assert + assert rc != 0 + assert "not found" in captured.err + + +def test_changes_with_key_on_unkeyed_index_exits_nonzero(tmp_path, monkeypatch, capsys): + # Arrange: index built WITHOUT a sync key + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch) + asyncio.run(cli.build_index(cli.create_parser().parse_args(_build_args("idx", [str(docs)])))) + capsys.readouterr() + + # Act: a key against the unkeyed index would diff a never-written snapshot + rc = _run_changes(cli, ["changes", "idx", "--docs", str(docs), "--sync-key", "typo"]) + captured = capsys.readouterr() + + # Assert + assert rc != 0 + assert captured.out.strip() == "" diff --git a/tests/test_cli_verify.py b/tests/test_cli_verify.py new file mode 100644 index 00000000..3b4f39f4 --- /dev/null +++ b/tests/test_cli_verify.py @@ -0,0 +1,318 @@ +"""Failing tests for `leann verify` (cross-artifact index integrity check).""" + +import json +import pickle +from pathlib import Path + +import numpy as np +from leann.cli import LeannCLI + + +def _import_faiss(): + # Same import order as the IVF backend and verify, so writer and reader + # share one SWIG build (mixing builds breaks invlists access). + try: + import faiss + except ImportError: + from leann_backend_hnsw import faiss + return faiss + + +def _write_faiss_index(path: Path, num_vectors: int, dim: int = 4) -> None: + # Mirror the real IVF backend: IndexIVFFlat + DirectMap.Hashtable with + # explicit ids, so verify's type/direct-map checks run against the real shape. + faiss = _import_faiss() + + quantizer = faiss.IndexFlatL2(dim) + index = faiss.IndexIVFFlat(quantizer, dim, 1, faiss.METRIC_L2) + index.set_direct_map_type(faiss.DirectMap.Hashtable) + vectors = np.ascontiguousarray( + np.random.default_rng(0).random((max(num_vectors, 1), dim), dtype=np.float32) + ) + ids = np.arange(num_vectors, dtype=np.int64) + try: + index.train(vectors) + if num_vectors: + index.add_with_ids(vectors[:num_vectors], ids) + except TypeError: + index.train(vectors.shape[0], faiss.swig_ptr(vectors)) + if num_vectors: + index.add_with_ids(num_vectors, faiss.swig_ptr(vectors), faiss.swig_ptr(ids)) + faiss.write_index(index, str(path)) + + +def _make_ivf_index(tmp_path: Path, passage_ids: list[str]) -> Path: + index_dir = tmp_path / ".leann" / "indexes" / "idx" + index_dir.mkdir(parents=True) + prefix = index_dir / "documents.leann" + + Path(str(prefix) + ".meta.json").write_text( + json.dumps( + { + "backend_name": "ivf", + "embedding_model": "m", + "embedding_mode": "sentence-transformers", + "dimensions": 4, + "backend_kwargs": {}, + } + ), + encoding="utf-8", + ) + + offsets: dict[str, int] = {} + with open(str(prefix) + ".passages.jsonl", "wb") as f: + for pid in passage_ids: + offsets[pid] = f.tell() + line = json.dumps({"id": pid, "text": f"passage {pid}", "metadata": {}}) + "\n" + f.write(line.encode("utf-8")) + with open(str(prefix) + ".passages.idx", "wb") as f: + pickle.dump(offsets, f) + + id_map = { + "id_to_passage": {str(i): pid for i, pid in enumerate(passage_ids)}, + "passage_to_id": {pid: i for i, pid in enumerate(passage_ids)}, + "next_id": len(passage_ids), + } + Path(str(prefix).removesuffix(".leann") + ".ivf_id_map.json").write_text( + json.dumps(id_map), encoding="utf-8" + ) + + _write_faiss_index( + Path(str(prefix).removesuffix(".leann") + ".index"), num_vectors=len(passage_ids) + ) + return prefix + + +def _run_verify(index_name: str = "idx") -> int: + cli = LeannCLI() + args = cli.create_parser().parse_args(["verify", index_name]) + return int(cli.verify_command(args) or 0) + + +def test_verify_passes_on_healthy_ivf_index(tmp_path, monkeypatch): + # Arrange + monkeypatch.chdir(tmp_path) + _make_ivf_index(tmp_path, ["0", "1", "2"]) + + # Act + rc = _run_verify() + + # Assert + assert rc == 0 + + +def test_verify_fails_on_truncated_passages_jsonl(tmp_path, monkeypatch, capsys): + # Arrange + monkeypatch.chdir(tmp_path) + prefix = _make_ivf_index(tmp_path, ["0", "1", "2"]) + jsonl = Path(str(prefix) + ".passages.jsonl") + lines = jsonl.read_bytes().splitlines(keepends=True) + jsonl.write_bytes(b"".join(lines[:-1])) + + # Act + rc = _run_verify() + + # Assert + assert rc != 0 + captured = capsys.readouterr() + assert (captured.out + captured.err).strip() + + +def test_verify_fails_when_id_map_not_exact_inverse(tmp_path, monkeypatch): + # Arrange + monkeypatch.chdir(tmp_path) + prefix = _make_ivf_index(tmp_path, ["0", "1", "2"]) + id_map_path = Path(str(prefix).removesuffix(".leann") + ".ivf_id_map.json") + id_map = json.loads(id_map_path.read_text(encoding="utf-8")) + id_map["passage_to_id"]["2"] = 0 + id_map_path.write_text(json.dumps(id_map), encoding="utf-8") + + # Act + rc = _run_verify() + + # Assert + assert rc != 0 + + +def test_verify_fails_on_bad_offset_in_passages_idx(tmp_path, monkeypatch): + # Arrange + monkeypatch.chdir(tmp_path) + prefix = _make_ivf_index(tmp_path, ["0", "1", "2"]) + idx_path = Path(str(prefix) + ".passages.idx") + with open(idx_path, "rb") as f: + offsets = pickle.load(f) + offsets["1"] = offsets["1"] + 3 + with open(idx_path, "wb") as f: + pickle.dump(offsets, f) + + # Act + rc = _run_verify() + + # Assert + assert rc != 0 + + +def test_verify_fails_when_id_map_has_id_missing_from_passages(tmp_path, monkeypatch): + # Arrange + monkeypatch.chdir(tmp_path) + prefix = _make_ivf_index(tmp_path, ["0", "1", "2"]) + id_map_path = Path(str(prefix).removesuffix(".leann") + ".ivf_id_map.json") + id_map = json.loads(id_map_path.read_text(encoding="utf-8")) + id_map["id_to_passage"]["3"] = "orphan" + id_map["passage_to_id"]["orphan"] = 3 + id_map["next_id"] = 4 + id_map_path.write_text(json.dumps(id_map), encoding="utf-8") + _write_faiss_index(Path(str(prefix).removesuffix(".leann") + ".index"), num_vectors=4) + + # Act + rc = _run_verify() + + # Assert + assert rc != 0 + + +def test_verify_reports_finding_instead_of_crashing_on_non_numeric_id_keys( + tmp_path, monkeypatch, capsys +): + # Arrange: every id_to_passage key is non-numeric (the exact corruption verify flags) + prefix = _make_ivf_index(tmp_path, ["p0", "p1"]) + id_map_path = prefix.parent / "documents.ivf_id_map.json" + id_map = json.loads(id_map_path.read_text(encoding="utf-8")) + id_map["id_to_passage"] = {"abc": "p0", "xyz": "p1"} + id_map["passage_to_id"] = {"p0": "abc", "p1": "xyz"} + id_map_path.write_text(json.dumps(id_map), encoding="utf-8") + monkeypatch.chdir(tmp_path) + + # Act + rc = _run_verify() + out = capsys.readouterr().out + + # Assert: findings printed, no ValueError traceback + assert rc == 1 + assert "not an integer" in out + + +def test_verify_unreadable_idx_does_not_emit_misleading_cross_findings( + tmp_path, monkeypatch, capsys +): + # Arrange + prefix = _make_ivf_index(tmp_path, ["p0", "p1"]) + Path(str(prefix) + ".passages.idx").write_bytes(b"not a pickle") + monkeypatch.chdir(tmp_path) + + # Act + rc = _run_verify() + out = capsys.readouterr().out + + # Assert: the unreadable idx is the finding; no derived mismatch noise + assert rc == 1 + assert "passages.idx unreadable" in out + assert "do not match passages.idx" not in out + assert "passages.jsonl has" not in out + + +def test_verify_fails_when_index_is_not_ivf_type(tmp_path, monkeypatch, capsys): + # Arrange: replace the IVF index with a flat index (no direct map) + faiss = _import_faiss() + + prefix = _make_ivf_index(tmp_path, ["p0", "p1"]) + flat = faiss.IndexFlatL2(4) + vectors = np.ascontiguousarray(np.random.default_rng(0).random((2, 4), dtype=np.float32)) + try: + flat.add(vectors) + except TypeError: + flat.add(2, faiss.swig_ptr(vectors)) + faiss.write_index(flat, str(prefix.parent / "documents.index")) + monkeypatch.chdir(tmp_path) + + # Act + rc = _run_verify() + out = capsys.readouterr().out + + # Assert + assert rc == 1 + assert "not an IVF index" in out + + +def test_verify_reports_findings_on_malformed_artifact_types(tmp_path, monkeypatch, capsys): + # Arrange: decodable but wrong-shaped artifacts must not traceback + prefix = _make_ivf_index(tmp_path, ["p0", "p1"]) + with open(str(prefix) + ".passages.idx", "wb") as f: + pickle.dump(["not", "a", "dict"], f) + id_map_path = prefix.parent / "documents.ivf_id_map.json" + id_map = json.loads(id_map_path.read_text(encoding="utf-8")) + id_map["next_id"] = "two" + id_map_path.write_text(json.dumps(id_map), encoding="utf-8") + monkeypatch.chdir(tmp_path) + + # Act + rc = _run_verify() + out = capsys.readouterr().out + + # Assert + assert rc == 1 + assert "passages.idx is not a dict" in out + assert "next_id is not an integer" in out + + +def test_verify_flags_missing_snapshot_for_recorded_scope(tmp_path, monkeypatch, capsys): + # Arrange: sync_roots.json records a root but the snapshot pickle is absent + # (interrupted between index write and snapshot commit) + prefix = _make_ivf_index(tmp_path, ["p0", "p1"]) + docs = tmp_path / "docs" + docs.mkdir() + (prefix.parent / "sync_roots.json").write_text( + json.dumps({"directories": [str(docs)], "files": []}), encoding="utf-8" + ) + monkeypatch.chdir(tmp_path) + + # Act + rc = _run_verify() + out = capsys.readouterr().out + + # Assert + assert rc == 1 + assert "sync snapshot missing" in out + + +def test_verify_flags_corrupt_snapshot_for_recorded_scope(tmp_path, monkeypatch, capsys): + # Arrange + import hashlib as _hashlib + + prefix = _make_ivf_index(tmp_path, ["p0", "p1"]) + docs = tmp_path / "docs" + docs.mkdir() + (prefix.parent / "sync_roots.json").write_text( + json.dumps({"directories": [str(docs)], "files": []}), encoding="utf-8" + ) + tag = _hashlib.sha256(str(docs).encode()).hexdigest()[:12] + (prefix.parent / f"sync_{tag}.pickle").write_bytes(b"not a pickle") + monkeypatch.chdir(tmp_path) + + # Act + rc = _run_verify() + out = capsys.readouterr().out + + # Assert + assert rc == 1 + assert "corrupt" in out + + +def test_verify_flags_mapped_id_absent_from_index_vectors(tmp_path, monkeypatch, capsys): + # Arrange: id map references id 5 which the 2-vector index does not contain + prefix = _make_ivf_index(tmp_path, ["p0", "p1"]) + id_map_path = prefix.parent / "documents.ivf_id_map.json" + id_map = json.loads(id_map_path.read_text(encoding="utf-8")) + id_map["id_to_passage"] = {"0": "p0", "5": "p1"} + id_map["passage_to_id"] = {"p0": 0, "p1": 5} + id_map["next_id"] = 6 + id_map_path.write_text(json.dumps(id_map), encoding="utf-8") + monkeypatch.chdir(tmp_path) + + # Act + rc = _run_verify() + out = capsys.readouterr().out + + # Assert + assert rc == 1 + assert "missing 1 of 2 mapped ids" in out diff --git a/tests/test_no_register.py b/tests/test_no_register.py new file mode 100644 index 00000000..2d54f076 --- /dev/null +++ b/tests/test_no_register.py @@ -0,0 +1,46 @@ +import json + +import pytest +from leann.registry import register_project_directory + + +@pytest.fixture +def project_dir(tmp_path): + proj = tmp_path / "proj" + (proj / ".leann" / "indexes" / "dummy").mkdir(parents=True) + return proj + + +@pytest.fixture +def fake_home(tmp_path, monkeypatch): + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + return home + + +def test_no_register_env_set_skips_registration(project_dir, fake_home, monkeypatch): + # Arrange + monkeypatch.setenv("LEANN_NO_REGISTER", "1") + + # Act + register_project_directory(project_dir) + + # Assert + assert not (fake_home / ".leann" / "projects.json").exists() + assert not (fake_home / ".leann").exists() + + +def test_env_unset_registers_project(project_dir, fake_home, monkeypatch): + """Guard test: proves the fixture is registrable when the switch is off.""" + # Arrange + monkeypatch.delenv("LEANN_NO_REGISTER", raising=False) + + # Act + register_project_directory(project_dir) + + # Assert + registry_file = fake_home / ".leann" / "projects.json" + assert registry_file.exists() + projects = json.loads(registry_file.read_text()) + assert str(project_dir.resolve()) in projects diff --git a/tests/test_sync.py b/tests/test_sync.py index f57971ae..82abb65c 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -2,8 +2,10 @@ import tempfile import unittest from pathlib import Path +from unittest import mock from unittest.mock import Mock +import leann.sync as sync_module from leann.sync import FileSynchronizer, MerkleTree, hash_data @@ -153,3 +155,44 @@ def test_touch_no_false_positive(self): fs2 = FileSynchronizer(root_dir=str(docs), snapshot_path=snapshot) added, removed, modified = fs2.detect_changes() assert not added and not removed and not modified + + +class TestUnreadableFileHandling(unittest.TestCase): + def test_unreadable_existing_file_keeps_previous_hash(self): + # A transiently unreadable file must not be classified as removed + # (which would delete its chunks from the index). + with tempfile.TemporaryDirectory() as temp_dir: + file_path = Path(temp_dir) / "file.txt" + file_path.write_text("hello", encoding="utf-8") + fs = FileSynchronizer(root_dir=temp_dir, auto_load=False) + fs.tree = fs.build_merkle_tree(fs.generate_file_hashes()) + + def broken_hash(path: Path) -> str: + raise OSError("permission denied") + + with mock.patch.object(sync_module, "_hash_file_bytes", broken_hash): + added, removed, modified = fs.detect_changes() + + assert removed == [] + assert modified == [] + assert added == [] + + +class TestWalkErrorHandling(unittest.TestCase): + def test_unreadable_subtree_raises_instead_of_reporting_removals(self): + if os.geteuid() == 0: + self.skipTest("permission checks are bypassed as root") + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + sub = root / "sub" + sub.mkdir() + (sub / "file.txt").write_text("hello", encoding="utf-8") + fs = FileSynchronizer(root_dir=temp_dir, auto_load=False) + fs.tree = fs.build_merkle_tree(fs.generate_file_hashes()) + + sub.chmod(0o000) + try: + with self.assertRaises(OSError): + fs.detect_changes() + finally: + sub.chmod(0o755) diff --git a/tests/test_sync_key.py b/tests/test_sync_key.py new file mode 100644 index 00000000..0d2d9adb --- /dev/null +++ b/tests/test_sync_key.py @@ -0,0 +1,367 @@ +"""Failing tests for `leann build --sync-key` (stable global snapshot identity).""" + +import asyncio +import hashlib +import json +from pathlib import Path + +import pytest +from leann.cli import LeannCLI + + +def _keyed_snapshot_name(key: str) -> str: + return f"sync_key_{hashlib.sha256(key.encode()).hexdigest()[:12]}.pickle" + + +def _make_fake_builder(recorded_builds: list[list[str]]): + class FakeBuilder: + def __init__(self, **kwargs): + self.kwargs = kwargs + self.paths: list[str] = [] + recorded_builds.append(self.paths) + + def add_text(self, _text, metadata=None): + self.paths.append((metadata or {}).get("file_path", "")) + + def build_index(self, index_path): + target_dir = Path(index_path).parent + target_dir.mkdir(parents=True, exist_ok=True) + (target_dir / "documents.leann.meta.json").write_text( + json.dumps( + { + "backend_name": "hnsw", + "embedding_model": self.kwargs.get("embedding_model", "m"), + "embedding_mode": self.kwargs.get( + "embedding_mode", "sentence-transformers" + ), + "passage_id_scheme": "sequential", + "backend_kwargs": { + "graph_degree": 32, + "complexity": 64, + "num_threads": 1, + "is_compact": False, + "is_recompute": True, + }, + } + ), + encoding="utf-8", + ) + (target_dir / "documents.leann.index").write_bytes(b"fake") + (target_dir / "documents.leann.passages.jsonl").write_text("", encoding="utf-8") + + def update_index(self, index_path): + pass + + return FakeBuilder + + +def _fake_load_documents(loaded_calls: list[set[str]]): + def fake(docs_paths, custom_file_types=None, include_hidden=False, args=None): + if isinstance(docs_paths, str): + docs_paths = [docs_paths] + files: list[Path] = [] + for p in docs_paths: + path = Path(p) + if path.is_dir(): + files.extend(sorted(path.rglob("*.txt"))) + elif path.is_file(): + files.append(path) + resolved = {str(f.resolve()) for f in files} + loaded_calls.append(resolved) + return [{"text": f.name, "metadata": {"file_path": str(f.resolve())}} for f in files] + + return fake + + +def _wire_cli(monkeypatch, recorded_builds, loaded_calls): + cli = LeannCLI() + monkeypatch.setattr(cli, "load_documents", _fake_load_documents(loaded_calls)) + monkeypatch.setattr(cli, "register_project_dir", lambda: None) + monkeypatch.setattr("leann.cli.LeannBuilder", _make_fake_builder(recorded_builds)) + return cli + + +def _build_args(index_name: str, docs: list[str], extra: list[str] | None = None) -> list[str]: + return [ + "build", + index_name, + "--docs", + *docs, + "--backend-name", + "hnsw", + "--no-compact", + "--embedding-model", + "m", + "--embedding-mode", + "sentence-transformers", + *(extra or []), + ] + + +def test_same_key_different_docs_lists_share_snapshot_and_load_only_new_files( + tmp_path, monkeypatch +): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + file_a = docs / "a.txt" + file_b = docs / "b.txt" + file_c = docs / "c.txt" + file_a.write_text("alpha", encoding="utf-8") + file_b.write_text("beta", encoding="utf-8") + recorded_builds: list[list[str]] = [] + loaded_calls: list[set[str]] = [] + cli = _wire_cli(monkeypatch, recorded_builds, loaded_calls) + parser = cli.create_parser() + + # Act + asyncio.run( + cli.build_index( + parser.parse_args( + _build_args("keyed", [str(file_a), str(file_b)], ["--sync-key", "corpus-v1"]) + ) + ) + ) + index_dir = tmp_path / ".leann" / "indexes" / "keyed" + file_c.write_text("gamma", encoding="utf-8") + calls_before_second = len(loaded_calls) + asyncio.run( + cli.build_index( + parser.parse_args( + _build_args( + "keyed", + [str(file_a), str(file_b), str(file_c)], + ["--sync-key", "corpus-v1"], + ) + ) + ) + ) + + # Assert + assert (index_dir / _keyed_snapshot_name("corpus-v1")).exists() + second_build_loaded: set[str] = set() + for call in loaded_calls[calls_before_second:]: + second_build_loaded |= call + assert second_build_loaded == {str(file_c.resolve())} + + +def test_sync_key_persisted_in_sync_roots_json(tmp_path, monkeypatch): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch, [], []) + parser = cli.create_parser() + + # Act + asyncio.run( + cli.build_index( + parser.parse_args(_build_args("keyed", [str(docs)], ["--sync-key", "corpus-v1"])) + ) + ) + + # Assert + sync_config = json.loads( + (tmp_path / ".leann" / "indexes" / "keyed" / "sync_roots.json").read_text(encoding="utf-8") + ) + assert sync_config["sync_key"] == "corpus-v1" + + +def test_different_key_on_keyed_index_errors_without_force_and_succeeds_with_force( + tmp_path, monkeypatch +): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch, [], []) + parser = cli.create_parser() + asyncio.run( + cli.build_index( + parser.parse_args(_build_args("keyed", [str(docs)], ["--sync-key", "key-one"])) + ) + ) + index_dir = tmp_path / ".leann" / "indexes" / "keyed" + + # Act / Assert + with pytest.raises((SystemExit, RuntimeError, ValueError)): + asyncio.run( + cli.build_index( + parser.parse_args(_build_args("keyed", [str(docs)], ["--sync-key", "key-two"])) + ) + ) + + asyncio.run( + cli.build_index( + parser.parse_args( + _build_args("keyed", [str(docs)], ["--sync-key", "key-two", "--force"]) + ) + ) + ) + sync_config = json.loads((index_dir / "sync_roots.json").read_text(encoding="utf-8")) + assert sync_config["sync_key"] == "key-two" + + +def test_directory_dropped_from_docs_triggers_rebuild_with_remaining_files_only( + tmp_path, monkeypatch +): + # Arrange + monkeypatch.chdir(tmp_path) + dir_one = tmp_path / "d1" + dir_two = tmp_path / "d2" + dir_one.mkdir() + dir_two.mkdir() + kept = dir_one / "kept.txt" + kept.write_text("kept", encoding="utf-8") + (dir_two / "dropped.txt").write_text("dropped", encoding="utf-8") + recorded_builds: list[list[str]] = [] + cli = _wire_cli(monkeypatch, recorded_builds, []) + parser = cli.create_parser() + asyncio.run( + cli.build_index( + parser.parse_args( + _build_args("keyed", [str(dir_one), str(dir_two)], ["--sync-key", "corpus-v1"]) + ) + ) + ) + builds_before = len(recorded_builds) + + # Act + asyncio.run( + cli.build_index( + parser.parse_args(_build_args("keyed", [str(dir_one)], ["--sync-key", "corpus-v1"])) + ) + ) + + # Assert + rebuilt_paths: set[str] = set() + for build in recorded_builds[builds_before:]: + rebuilt_paths |= set(build) + assert rebuilt_paths == {str(kept.resolve())} + + +def test_load_snapshot_raises_snapshot_corrupt_error_on_corrupt_pickle(tmp_path): + from leann.sync import FileSynchronizer, SnapshotCorruptError + + # Arrange + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + corrupt_path = tmp_path / "snapshot.pickle" + corrupt_path.write_bytes(b"not a pickle at all") + fs = FileSynchronizer(root_dir=str(docs), snapshot_path=str(corrupt_path), auto_load=False) + + # Act / Assert + with pytest.raises(SnapshotCorruptError): + fs.load_snapshot() + + missing = FileSynchronizer( + root_dir=str(docs), snapshot_path=str(tmp_path / "absent.pickle"), auto_load=False + ) + missing.load_snapshot() + assert missing.tree is None + + +def test_unkeyed_build_fails_loud_on_corrupt_snapshot_and_force_resets(tmp_path, monkeypatch): + from leann.sync import SnapshotCorruptError + + # Arrange: unkeyed build, then corrupt the per-root snapshot + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + recorded_builds: list[list[str]] = [] + loaded_calls: list[set[str]] = [] + cli = _wire_cli(monkeypatch, recorded_builds, loaded_calls) + asyncio.run(cli.build_index(cli.create_parser().parse_args(_build_args("idx", [str(docs)])))) + index_dir = tmp_path / ".leann" / "indexes" / "idx" + snapshots = list(index_dir.glob("sync_*.pickle")) + assert snapshots + for snap in snapshots: + snap.write_bytes(b"not a pickle") + + # Act / Assert: without --force the corruption is a hard error + with pytest.raises(SnapshotCorruptError, match="--force"): + asyncio.run( + cli.build_index(cli.create_parser().parse_args(_build_args("idx", [str(docs)]))) + ) + + # Act: --force resets the corrupt snapshot and rebuilds + asyncio.run( + cli.build_index( + cli.create_parser().parse_args(_build_args("idx", [str(docs)], ["--force"])) + ) + ) + + # Assert: snapshot is valid again (a subsequent build sees no changes) + asyncio.run(cli.build_index(cli.create_parser().parse_args(_build_args("idx", [str(docs)])))) + + +def test_build_fails_loud_on_corrupt_sync_roots_json(tmp_path, monkeypatch): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + recorded_builds: list[list[str]] = [] + loaded_calls: list[set[str]] = [] + cli = _wire_cli(monkeypatch, recorded_builds, loaded_calls) + asyncio.run( + cli.build_index( + cli.create_parser().parse_args( + _build_args("idx", [str(docs)], ["--sync-key", "corpus-v1"]) + ) + ) + ) + sync_roots = tmp_path / ".leann" / "indexes" / "idx" / "sync_roots.json" + sync_roots.write_text("{not json", encoding="utf-8") + + # Act / Assert: an unreadable config must not silently unkey the index + with pytest.raises(ValueError, match=r"sync_roots\.json"): + asyncio.run( + cli.build_index(cli.create_parser().parse_args(_build_args("idx", [str(docs)]))) + ) + + # Act: --force ignores the corrupt config and rewrites it + asyncio.run( + cli.build_index( + cli.create_parser().parse_args(_build_args("idx", [str(docs)], ["--force"])) + ) + ) + assert json.loads(sync_roots.read_text(encoding="utf-8")) + + +def test_build_fails_loud_on_sync_roots_json_with_wrong_json_type(tmp_path, monkeypatch): + # Arrange: valid JSON that is not an object must not silently unkey the index + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + recorded_builds: list[list[str]] = [] + loaded_calls: list[set[str]] = [] + cli = _wire_cli(monkeypatch, recorded_builds, loaded_calls) + asyncio.run( + cli.build_index( + cli.create_parser().parse_args( + _build_args("idx", [str(docs)], ["--sync-key", "corpus-v1"]) + ) + ) + ) + sync_roots = tmp_path / ".leann" / "indexes" / "idx" / "sync_roots.json" + sync_roots.write_text("[]", encoding="utf-8") + + # Act / Assert + with pytest.raises(ValueError, match=r"sync_roots\.json"): + asyncio.run( + cli.build_index(cli.create_parser().parse_args(_build_args("idx", [str(docs)]))) + ) + + # --force recovers + asyncio.run( + cli.build_index( + cli.create_parser().parse_args(_build_args("idx", [str(docs)], ["--force"])) + ) + ) diff --git a/tests/test_watch_sync_scope.py b/tests/test_watch_sync_scope.py index 7c306024..bdc9ed80 100644 --- a/tests/test_watch_sync_scope.py +++ b/tests/test_watch_sync_scope.py @@ -81,3 +81,52 @@ def test_mixed_txt_and_bin_directory_skips_bin_without_crash(tmp_path): ) added, removed, modified = fs2.detect_changes() assert not added and not removed and not modified + + +def test_watch_tick_survives_corrupt_sync_roots(tmp_path, monkeypatch, capsys): + # Arrange: a registered index whose sync_roots.json is corrupt + monkeypatch.chdir(tmp_path) + from leann.cli import LeannCLI + + cli = LeannCLI() + index_dir = tmp_path / ".leann" / "indexes" / "idx" + index_dir.mkdir(parents=True) + docs = tmp_path / "docs" + docs.mkdir() + (index_dir / "sync_roots.json").write_text( + '{"directories": ["' + str(docs) + '"], "files": [], "sync_key": null', encoding="utf-8" + ) # truncated JSON + monkeypatch.setattr(cli, "_resolve_index_for_watch", lambda name: {"index_dir": index_dir}) + + # Act: must not raise — a watch tick skips, it doesn't kill the daemon + added, removed, modified = cli._watch_check_changes("idx") + + # Assert + assert (added, removed, modified) == (set(), set(), set()) + + +def test_watch_tick_survives_corrupt_snapshot(tmp_path, monkeypatch, capsys): + # Arrange: valid scope but corrupt snapshot pickle + monkeypatch.chdir(tmp_path) + import hashlib as _hashlib + import json as _json + + from leann.cli import LeannCLI + + cli = LeannCLI() + index_dir = tmp_path / ".leann" / "indexes" / "idx" + index_dir.mkdir(parents=True) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + (index_dir / "sync_roots.json").write_text( + _json.dumps({"directories": [str(docs)], "files": []}), encoding="utf-8" + ) + tag = _hashlib.sha256(str(docs).encode()).hexdigest()[:12] + (index_dir / f"sync_{tag}.pickle").write_bytes(b"not a pickle") + monkeypatch.setattr(cli, "_resolve_index_for_watch", lambda name: {"index_dir": index_dir}) + + # Act / Assert + added, removed, modified = cli._watch_check_changes("idx") + assert (added, removed, modified) == (set(), set(), set()) + assert "watch tick skipped" in capsys.readouterr().out diff --git a/tests/test_zero_chunk_commit.py b/tests/test_zero_chunk_commit.py new file mode 100644 index 00000000..8c9cced5 --- /dev/null +++ b/tests/test_zero_chunk_commit.py @@ -0,0 +1,270 @@ +"""Failing tests for committing sync snapshots on safely-empty (zero-chunk) build deltas.""" + +import asyncio +import hashlib +import json +import os +from pathlib import Path + +import pytest +from leann.cli import LeannCLI + + +def _make_fake_builder(): + class FakeBuilder: + def __init__(self, **kwargs): + self.kwargs = kwargs + + def add_text(self, _text, metadata=None): + pass + + def build_index(self, index_path): + target_dir = Path(index_path).parent + target_dir.mkdir(parents=True, exist_ok=True) + (target_dir / "documents.leann.meta.json").write_text( + json.dumps( + { + "backend_name": "hnsw", + "embedding_model": self.kwargs.get("embedding_model", "m"), + "embedding_mode": self.kwargs.get( + "embedding_mode", "sentence-transformers" + ), + "passage_id_scheme": "sequential", + "backend_kwargs": { + "graph_degree": 32, + "complexity": 64, + "num_threads": 1, + "is_compact": False, + "is_recompute": True, + }, + } + ), + encoding="utf-8", + ) + (target_dir / "documents.leann.index").write_bytes(b"fake") + (target_dir / "documents.leann.passages.jsonl").write_text("", encoding="utf-8") + + def update_index(self, index_path): + pass + + return FakeBuilder + + +def _loading_fake(docs_paths, custom_file_types=None, include_hidden=False, args=None): + if isinstance(docs_paths, str): + docs_paths = [docs_paths] + files: list[Path] = [] + for p in docs_paths: + path = Path(p) + if path.is_dir(): + files.extend(sorted(path.rglob("*.txt"))) + elif path.is_file(): + files.append(path) + return [{"text": f.name, "metadata": {"file_path": str(f.resolve())}} for f in files] + + +def _wire_cli(monkeypatch) -> LeannCLI: + cli = LeannCLI() + monkeypatch.setattr(cli, "load_documents", _loading_fake) + monkeypatch.setattr(cli, "register_project_dir", lambda: None) + monkeypatch.setattr("leann.cli.LeannBuilder", _make_fake_builder()) + return cli + + +def _build_args(index_name: str, docs: list[str], extra: list[str] | None = None) -> list[str]: + return [ + "build", + index_name, + "--docs", + *docs, + "--backend-name", + "hnsw", + "--no-compact", + "--embedding-model", + "m", + "--embedding-mode", + "sentence-transformers", + *(extra or []), + ] + + +def _run_build(cli: LeannCLI, argv: list[str]) -> None: + asyncio.run(cli.build_index(cli.create_parser().parse_args(argv))) + + +def _run_changes(cli: LeannCLI, argv: list[str]) -> int: + args = cli.create_parser().parse_args(argv) + result = cli.changes_command(args) + if asyncio.iscoroutine(result): + result = asyncio.run(result) + return int(result or 0) + + +def _index_artifact_hashes(index_dir: Path) -> dict[str, str]: + return { + p.name: hashlib.sha256(p.read_bytes()).hexdigest() + for p in sorted(index_dir.glob("documents.leann*")) + if p.is_file() + } + + +def test_add_only_zero_chunk_delta_commits_snapshot_and_leaves_index_untouched( + tmp_path, monkeypatch, capsys +): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch) + _run_build(cli, _build_args("idx", [str(docs)])) + (docs / "b.txt").write_text("beta", encoding="utf-8") + index_dir = tmp_path / ".leann" / "indexes" / "idx" + hashes_before = _index_artifact_hashes(index_dir) + zero_chunk_calls: list[list[str]] = [] + + def zero_chunk_load(docs_paths, custom_file_types=None, include_hidden=False, args=None): + zero_chunk_calls.append(list(docs_paths)) + return [] + + monkeypatch.setattr(cli, "load_documents", zero_chunk_load) + + # Act + _run_build(cli, _build_args("idx", [str(docs)])) + monkeypatch.setattr(cli, "load_documents", _loading_fake) + capsys.readouterr() + rc = _run_changes(cli, ["changes", "idx"]) + report = json.loads(capsys.readouterr().out) + + # Assert + assert zero_chunk_calls, "second build should have loaded the added file" + assert rc == 0 + assert report["added"] == [] + assert report["modified"] == [] + assert report["removed"] == [] + assert (index_dir / "sync_roots.json").exists() + assert _index_artifact_hashes(index_dir) == hashes_before + + +def test_loader_failure_does_not_advance_snapshot(tmp_path, monkeypatch, capsys): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch) + _run_build(cli, _build_args("idx", [str(docs)])) + new_file = docs / "b.txt" + new_file.write_text("beta", encoding="utf-8") + + def failing_load(docs_paths, custom_file_types=None, include_hidden=False, args=None): + raise RuntimeError("parser exploded") + + monkeypatch.setattr(cli, "load_documents", failing_load) + + # Act + with pytest.raises(RuntimeError, match="parser exploded"): + _run_build(cli, _build_args("idx", [str(docs)])) + monkeypatch.setattr(cli, "load_documents", _loading_fake) + capsys.readouterr() + rc = _run_changes(cli, ["changes", "idx"]) + report = json.loads(capsys.readouterr().out) + + # Assert + assert rc == 0 + assert report["added"] == [str(new_file.resolve())] + + +def test_snapshot_and_sync_config_writes_use_atomic_os_replace(tmp_path, monkeypatch): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch) + replaced_targets: list[str] = [] + real_replace = os.replace + + def spy_replace(src, dst, *args, **kwargs): + replaced_targets.append(str(dst)) + return real_replace(src, dst, *args, **kwargs) + + monkeypatch.setattr(os, "replace", spy_replace) + + # Act + _run_build(cli, _build_args("idx", [str(docs)])) + + # Assert + index_dir = tmp_path / ".leann" / "indexes" / "idx" + assert any(t.endswith(".pickle") for t in replaced_targets), ( + "save_snapshot should publish the snapshot via os.replace" + ) + assert any(t.endswith("sync_roots.json") for t in replaced_targets), ( + "_write_sync_config should publish sync_roots.json via os.replace" + ) + assert not list(index_dir.glob("*.tmp")) + + +def test_swallowed_loader_failure_blocks_snapshot_commit(tmp_path, monkeypatch, capsys): + # Arrange + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch) + _run_build(cli, _build_args("idx", [str(docs)])) + new_file = docs / "b.txt" + new_file.write_text("beta", encoding="utf-8") + + def swallowed_failure_load(docs_paths, custom_file_types=None, include_hidden=False, args=None): + cli._load_errors = 1 # simulate load_documents warn-and-continue on a broken file + return [] + + monkeypatch.setattr(cli, "load_documents", swallowed_failure_load) + + # Act + with pytest.raises(RuntimeError, match="failed to load"): + _run_build(cli, _build_args("idx", [str(docs)])) + monkeypatch.setattr(cli, "load_documents", _loading_fake) + capsys.readouterr() + rc = _run_changes(cli, ["changes", "idx"]) + report = json.loads(capsys.readouterr().out) + + # Assert: snapshot not committed, so the failed file is still pending + assert rc == 0 + assert report["added"] == [str(new_file.resolve())] + + +def test_partial_loader_failure_aborts_before_mutating_index(tmp_path, monkeypatch, capsys): + # Arrange: two new files, one loads and one fails — nothing may be committed + monkeypatch.chdir(tmp_path) + docs = tmp_path / "docs" + docs.mkdir() + (docs / "a.txt").write_text("alpha", encoding="utf-8") + cli = _wire_cli(monkeypatch) + _run_build(cli, _build_args("idx", [str(docs)])) + good = docs / "b.txt" + bad = docs / "c.txt" + good.write_text("beta", encoding="utf-8") + bad.write_text("gamma", encoding="utf-8") + index_dir = tmp_path / ".leann" / "indexes" / "idx" + hashes_before = _index_artifact_hashes(index_dir) + + def partial_failure_load(docs_paths, custom_file_types=None, include_hidden=False, args=None): + cli._load_errors = 1 # one file failed with a swallowed warning + return [{"text": "beta", "metadata": {"file_path": str(good.resolve())}}] + + monkeypatch.setattr(cli, "load_documents", partial_failure_load) + + # Act + with pytest.raises(RuntimeError, match="failed to load"): + _run_build(cli, _build_args("idx", [str(docs)])) + monkeypatch.setattr(cli, "load_documents", _loading_fake) + capsys.readouterr() + rc = _run_changes(cli, ["changes", "idx"]) + report = json.loads(capsys.readouterr().out) + + # Assert: index untouched, both files still pending + assert rc == 0 + assert sorted(report["added"]) == [str(good.resolve()), str(bad.resolve())] + assert _index_artifact_hashes(index_dir) == hashes_before From dfd7f848aa5701014658aed1e3dfbbc2d75eee49 Mon Sep 17 00:00:00 2001 From: ww2283 Date: Wed, 19 Aug 2026 14:11:42 -0400 Subject: [PATCH 2/7] chore: strip extra trailing newline in CHANGELOG --- docs/CHANGELOG.md | 1 - 1 file changed, 1 deletion(-) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 0810a0d8..416d9a00 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -85,4 +85,3 @@ fixes). Newest entries at the bottom. - A transiently unreadable file keeps its previous hash instead of being classified as removed (which deleted its chunks from the index). - `LEANN_NO_REGISTER=1` env switch skips project-directory registration (for tests/CI). - From 1c2f04b13d21a7468b90b5762b495ada64f858c0 Mon Sep 17 00:00:00 2001 From: ww2283 Date: Wed, 19 Aug 2026 15:21:00 -0400 Subject: [PATCH 3/7] fix(verify): enumerate IVF stored ids in a subprocess With two faiss SWIG builds loaded in one process (faiss-cpu + the hnsw bundle, as on CI), downcast_index silently loses invlists access and the mapped-id check was skipped, returning a false healthy verdict. A subprocess loads exactly one build. --- packages/leann-core/src/leann/cli.py | 56 +++++++++++++++++++++------- 1 file changed, 42 insertions(+), 14 deletions(-) diff --git a/packages/leann-core/src/leann/cli.py b/packages/leann-core/src/leann/cli.py index 677fe346..b2b696c0 100644 --- a/packages/leann-core/src/leann/cli.py +++ b/packages/leann-core/src/leann/cli.py @@ -6,6 +6,7 @@ import json import os import pickle +import subprocess import sys import time import uuid @@ -3114,20 +3115,8 @@ def _verify_ivf(self, prefix: Path, offsets: Optional[dict[str, int]]) -> list[s if ivf_index is None: findings.append(f".index is not an IVF index (got {type(index).__name__})") else: - # faiss does not persist the direct map, so enumerate the stored - # ids from the inverted lists (read-only) and check every mapped - # id actually exists in the index. - invlists = getattr(faiss.downcast_index(index), "invlists", None) - if invlists is not None: - stored_ids: set[int] = set() - for list_no in range(ivf_index.nlist): - list_size = invlists.list_size(list_no) - if list_size: - ids_ptr = invlists.get_ids(list_no) - stored_ids.update( - int(x) for x in faiss.rev_swig_ptr(ids_ptr, list_size) - ) - invlists.release_ids(list_no, ids_ptr) + stored_ids = self._ivf_stored_ids(index_path) + if stored_ids is not None: missing = [i for i in numeric_ids if i not in stored_ids] if missing: findings.append( @@ -3137,6 +3126,45 @@ def _verify_ivf(self, prefix: Path, offsets: Optional[dict[str, int]]) -> list[s findings.append(f".index unreadable: {exc}") return findings + @staticmethod + def _ivf_stored_ids(index_path: Path) -> Optional[set[int]]: + """Enumerate ids stored in an IVF index's inverted lists, or None if unavailable. + + Runs in a subprocess so exactly one faiss SWIG build is loaded: with two + builds in one process (e.g. faiss-cpu + leann_backend_hnsw) the type + tables clash and downcast_index silently loses invlists access. + """ + script = ( + "import json, sys\n" + "try:\n" + " import faiss\n" + "except ImportError:\n" + " from leann_backend_hnsw import faiss\n" + "index = faiss.read_index(sys.argv[1])\n" + "ivf = faiss.extract_index_ivf(index)\n" + "invlists = getattr(faiss.downcast_index(index), 'invlists', None) or ivf.invlists\n" + "stored = []\n" + "for list_no in range(ivf.nlist):\n" + " size = invlists.list_size(list_no)\n" + " if size:\n" + " ptr = invlists.get_ids(list_no)\n" + " stored.extend(int(x) for x in faiss.rev_swig_ptr(ptr, size))\n" + " invlists.release_ids(list_no, ptr)\n" + "print(json.dumps(stored))\n" + ) + try: + result = subprocess.run( + [sys.executable, "-c", script, str(index_path)], + capture_output=True, + text=True, + timeout=120, + ) + if result.returncode != 0: + return None + return set(json.loads(result.stdout)) + except Exception: + return None + def _watch_report_changes( self, index_name: str, From ec831637747f98ad6f0b45ce886820acf80d63ca Mon Sep 17 00:00:00 2001 From: ww2283 Date: Wed, 19 Aug 2026 15:31:35 -0400 Subject: [PATCH 4/7] fix(ci): pip-install patchelf>=0.14.5 for auditwheel repair ubuntu-22.04's apt patchelf (0.14.3) fell below current auditwheel's >=0.14.5 floor, failing every Linux wheel-repair job. --- .github/workflows/build-reusable.yml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/workflows/build-reusable.yml b/.github/workflows/build-reusable.yml index 32074708..d9d36ea9 100644 --- a/.github/workflows/build-reusable.yml +++ b/.github/workflows/build-reusable.yml @@ -310,7 +310,8 @@ jobs: fi retry 5 uv pip install --python "$BUILD_PY" scikit-build-core numpy swig Cython pybind11 if [[ "$RUNNER_OS" == "Linux" ]]; then - retry 5 uv pip install --python "$BUILD_PY" auditwheel + # pip patchelf: ubuntu-22.04's apt patchelf (0.14.3) is below auditwheel's >=0.14.5 floor + retry 5 uv pip install --python "$BUILD_PY" auditwheel 'patchelf>=0.14.5' elif [[ "$RUNNER_OS" == "macOS" ]]; then retry 5 uv pip install --python "$BUILD_PY" delocate else From 100c2c9523237fa35151245756b2556e444c1f8b Mon Sep 17 00:00:00 2001 From: ww2283 Date: Wed, 19 Aug 2026 16:32:39 -0400 Subject: [PATCH 5/7] fix(tests): skip Unix-permission walk test on non-POSIX platforms --- tests/test_sync.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_sync.py b/tests/test_sync.py index 82abb65c..7eb1ca95 100644 --- a/tests/test_sync.py +++ b/tests/test_sync.py @@ -178,6 +178,7 @@ def broken_hash(path: Path) -> str: assert added == [] +@unittest.skipUnless(os.name == "posix", "relies on Unix permission semantics") class TestWalkErrorHandling(unittest.TestCase): def test_unreadable_subtree_raises_instead_of_reporting_removals(self): if os.geteuid() == 0: From 076ea380ce51e4df48da6d7577e39c9315bece13 Mon Sep 17 00:00:00 2001 From: ww26 Date: Wed, 19 Aug 2026 17:40:14 -0400 Subject: [PATCH 6/7] fix(verify): allow duplicate content-hash passage ids (#5) (#6) * fix(verify): allow duplicate content-hash passage ids (#5) Content-hash ids are legitimately many-to-one for byte-identical chunks; verify's bijection invariants flagged healthy indexes. - jsonl dup ids allowed only when text is identical per id - idx cardinality compared against unique jsonl ids - IVF passage_to_id checked as last-wins partial inverse over deduped id_to_passage values * docs: changelog entry for verify duplicate-id fix (#5) * fix(verify): harden duplicate-id conflict check against mixed types Codex review of PR #6: type-prefixed text hash so 1 vs "1" counts as different text, and type-independent sort of conflicting ids so mixed int/str ids report findings instead of raising TypeError. --- docs/CHANGELOG.md | 13 +++ packages/leann-core/src/leann/cli.py | 35 +++++-- tests/test_cli_verify.py | 146 +++++++++++++++++++++++++++ 3 files changed, 184 insertions(+), 10 deletions(-) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 416d9a00..22e133ec 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -85,3 +85,16 @@ fixes). Newest entries at the bottom. - A transiently unreadable file keeps its previous hash instead of being classified as removed (which deleted its chunks from the index). - `LEANN_NO_REGISTER=1` env switch skips project-directory registration (for tests/CI). + +## 2026-08-19: verify accepts duplicate content-hash passage ids (issue #5) + +- `leann verify` no longer false-positives on healthy indexes built with + `--id-scheme=content-hash`, where byte-identical chunks legitimately share one passage id + (many-to-one by construction: jsonl keeps one line per chunk, the offset map and IVF + `passage_to_id` are last-wins one-entry-per-unique-id). +- Invariants rewritten to the actual contract: duplicate jsonl ids allowed only with + identical text per id; idx cardinality vs unique ids; IVF `passage_to_id` checked as a + partial inverse (each passage id maps back to one of its FAISS labels, key set equals + the deduped `id_to_passage` value set). +- Same id with different text, broken inversion, or a missing `passage_to_id` entry still + fail verify; tests now include a duplicate-content fixture. diff --git a/packages/leann-core/src/leann/cli.py b/packages/leann-core/src/leann/cli.py index b2b696c0..a1010e7d 100644 --- a/packages/leann-core/src/leann/cli.py +++ b/packages/leann-core/src/leann/cli.py @@ -2959,19 +2959,29 @@ def verify_command(self, args) -> int: jsonl_path = Path(str(prefix) + ".passages.jsonl") jsonl_ids: list[str] = [] + text_hash_by_id: dict[str, str] = {} + conflicting_ids: set[str] = set() try: with open(jsonl_path, encoding="utf-8") as f: for lineno, line in enumerate(f, 1): try: - pid = json.loads(line)["id"] + record = json.loads(line) + pid = record["id"] except Exception as exc: findings.append(f"passages.jsonl line {lineno} unparseable: {exc}") continue jsonl_ids.append(pid) + text = record.get("text", "") + text_hash = hashlib.sha256(f"{type(text).__name__}:{text}".encode()).hexdigest() + if pid in text_hash_by_id: + if text_hash_by_id[pid] != text_hash: + conflicting_ids.add(pid) + else: + text_hash_by_id[pid] = text_hash except Exception as exc: findings.append(f"passages.jsonl unreadable: {exc}") - if len(set(jsonl_ids)) != len(jsonl_ids): - findings.append("passages.jsonl contains duplicate ids") + for pid in sorted(conflicting_ids, key=str): + findings.append(f"passages.jsonl id {pid!r} appears with different text") offsets: dict[str, int] = {} offsets_ok = False @@ -2987,10 +2997,11 @@ def verify_command(self, args) -> int: findings.append(f"passages.idx unreadable: {exc}") if offsets_ok and (offsets or jsonl_ids): - if len(offsets) != len(jsonl_ids): + unique_jsonl_ids = set(jsonl_ids) + if len(offsets) != len(unique_jsonl_ids): findings.append( f"passages.idx has {len(offsets)} entries but " - f"passages.jsonl has {len(jsonl_ids)} lines" + f"passages.jsonl has {len(unique_jsonl_ids)} unique ids" ) if set(offsets) != set(jsonl_ids): findings.append("passages.idx keys do not match passages.jsonl ids") @@ -3077,11 +3088,15 @@ def _verify_ivf(self, prefix: Path, offsets: Optional[dict[str, int]]) -> list[s findings.append(f"id_to_passage key {key!r} is negative") except ValueError: findings.append(f"id_to_passage key {key!r} is not an integer") - if len(id_to_passage) != len(passage_to_id): - findings.append("id_to_passage and passage_to_id have different sizes") - for key, pid in id_to_passage.items(): - if str(passage_to_id.get(pid)) != key: - findings.append(f"id_to_passage[{key!r}]={pid!r} is not inverted in passage_to_id") + if len(passage_to_id) != len(set(id_to_passage.values())): + findings.append("passage_to_id size does not match unique id_to_passage values") + for pid, fid in passage_to_id.items(): + if id_to_passage.get(str(fid)) != pid: + findings.append( + f"passage_to_id[{pid!r}]={fid!r} does not map back in id_to_passage" + ) + if set(passage_to_id) != set(id_to_passage.values()): + findings.append("passage_to_id keys do not match id_to_passage values") if offsets is not None and set(id_to_passage.values()) != set(offsets): findings.append("id_to_passage values do not match passages.idx keys") numeric_ids = [ diff --git a/tests/test_cli_verify.py b/tests/test_cli_verify.py index 3b4f39f4..65a19798 100644 --- a/tests/test_cli_verify.py +++ b/tests/test_cli_verify.py @@ -3,6 +3,7 @@ import json import pickle from pathlib import Path +from typing import Any import numpy as np from leann.cli import LeannCLI @@ -83,6 +84,50 @@ def _make_ivf_index(tmp_path: Path, passage_ids: list[str]) -> Path: return prefix +def _make_ivf_index_with_entries(tmp_path: Path, entries: list[tuple[Any, Any]]) -> Path: + # Like _make_ivf_index but accepts explicit (pid, text) pairs so a pid may + # repeat (content-hash id scheme). idx and passage_to_id are last-wins. + index_dir = tmp_path / ".leann" / "indexes" / "idx" + index_dir.mkdir(parents=True) + prefix = index_dir / "documents.leann" + + Path(str(prefix) + ".meta.json").write_text( + json.dumps( + { + "backend_name": "ivf", + "embedding_model": "m", + "embedding_mode": "sentence-transformers", + "dimensions": 4, + "backend_kwargs": {}, + } + ), + encoding="utf-8", + ) + + offsets: dict[str, int] = {} + with open(str(prefix) + ".passages.jsonl", "wb") as f: + for pid, text in entries: + offsets[pid] = f.tell() + line = json.dumps({"id": pid, "text": text, "metadata": {}}) + "\n" + f.write(line.encode("utf-8")) + with open(str(prefix) + ".passages.idx", "wb") as f: + pickle.dump(offsets, f) + + id_map = { + "id_to_passage": {str(i): pid for i, (pid, _) in enumerate(entries)}, + "passage_to_id": {pid: i for i, (pid, _) in enumerate(entries)}, + "next_id": len(entries), + } + Path(str(prefix).removesuffix(".leann") + ".ivf_id_map.json").write_text( + json.dumps(id_map), encoding="utf-8" + ) + + _write_faiss_index( + Path(str(prefix).removesuffix(".leann") + ".index"), num_vectors=len(entries) + ) + return prefix + + def _run_verify(index_name: str = "idx") -> int: cli = LeannCLI() args = cli.create_parser().parse_args(["verify", index_name]) @@ -316,3 +361,104 @@ def test_verify_flags_mapped_id_absent_from_index_vectors(tmp_path, monkeypatch, # Assert assert rc == 1 assert "missing 1 of 2 mapped ids" in out + + +def test_verify_passes_on_healthy_index_with_duplicate_content_hash_ids( + tmp_path, monkeypatch, capsys +): + # Arrange: byte-identical chunks legitimately share one content-hash id + _make_ivf_index_with_entries( + tmp_path, [("dup", "same text"), ("dup", "same text"), ("p1", "other text")] + ) + monkeypatch.chdir(tmp_path) + + # Act + rc = _run_verify() + out = capsys.readouterr().out + + # Assert + assert rc == 0 + assert "duplicate" not in out + assert "do not match" not in out + + +def test_verify_fails_on_duplicate_id_with_different_text(tmp_path, monkeypatch, capsys): + # Arrange + _make_ivf_index_with_entries( + tmp_path, [("dup", "text one"), ("dup", "text two"), ("p1", "other text")] + ) + monkeypatch.chdir(tmp_path) + + # Act + rc = _run_verify() + out = capsys.readouterr().out + + # Assert + assert rc == 1 + assert out.strip() + + +def test_verify_fails_on_duplicate_id_with_type_coerced_text(tmp_path, monkeypatch, capsys): + # Arrange: same id, texts 1 (int) vs "1" (str) must count as different text + _make_ivf_index_with_entries(tmp_path, [("dup", 1), ("dup", "1")]) + monkeypatch.chdir(tmp_path) + + # Act + rc = _run_verify() + out = capsys.readouterr().out + + # Assert + assert rc == 1 + assert "different text" in out + + +def test_verify_reports_conflicts_with_heterogeneous_id_types(tmp_path, monkeypatch, capsys): + # Arrange: conflicting ids of mixed types (int and str) must not crash sorted() + _make_ivf_index_with_entries( + tmp_path, [(1, "text a"), (1, "text b"), ("x", "text c"), ("x", "text d")] + ) + monkeypatch.chdir(tmp_path) + + # Act + rc = _run_verify() + out = capsys.readouterr().out + + # Assert + assert rc == 1 + assert "different text" in out + + +def test_verify_fails_when_passage_to_id_label_maps_to_other_pid(tmp_path, monkeypatch): + # Arrange: passage_to_id["p1"] points at a faiss label owned by "dup" + prefix = _make_ivf_index_with_entries( + tmp_path, [("dup", "same text"), ("dup", "same text"), ("p1", "other text")] + ) + id_map_path = prefix.parent / "documents.ivf_id_map.json" + id_map = json.loads(id_map_path.read_text(encoding="utf-8")) + id_map["passage_to_id"]["p1"] = 0 + id_map_path.write_text(json.dumps(id_map), encoding="utf-8") + monkeypatch.chdir(tmp_path) + + # Act + rc = _run_verify() + + # Assert + assert rc != 0 + + +def test_verify_fails_when_pid_missing_from_passage_to_id(tmp_path, monkeypatch): + # Arrange + prefix = _make_ivf_index_with_entries( + tmp_path, [("dup", "same text"), ("dup", "same text"), ("p1", "other text")] + ) + id_map_path = prefix.parent / "documents.ivf_id_map.json" + id_map = json.loads(id_map_path.read_text(encoding="utf-8")) + del id_map["passage_to_id"]["p1"] + id_map_path.write_text(json.dumps(id_map), encoding="utf-8") + monkeypatch.chdir(tmp_path) + + # Act + rc = _run_verify() + + # Assert + assert rc != 0 From b853319cf15965ff0ac1b5424d50080041e2f31a Mon Sep 17 00:00:00 2001 From: ww26 Date: Wed, 19 Aug 2026 19:14:45 -0400 Subject: [PATCH 7/7] fix(cli): copy chunk metadata before id assignment to break dict aliasing (#7) (#8) --- packages/leann-core/src/leann/cli.py | 4 ++-- tests/test_incremental_build.py | 34 ++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/packages/leann-core/src/leann/cli.py b/packages/leann-core/src/leann/cli.py index a1010e7d..3954b525 100644 --- a/packages/leann-core/src/leann/cli.py +++ b/packages/leann-core/src/leann/cli.py @@ -2175,7 +2175,7 @@ def _assign_chunk_ids(chunks: list[dict]) -> None: for path_key, path_chunks in by_path.items(): for idx, c in enumerate(path_chunks): sid = hashlib.sha256(f"{path_key}:{idx}".encode()).hexdigest()[:16] - c.setdefault("metadata", {})["id"] = sid + c["metadata"] = {**c.get("metadata", {}), "id": sid} c["id"] = sid @staticmethod @@ -2183,7 +2183,7 @@ def _assign_unique_chunk_ids(chunks: list[dict]) -> None: """Assign unique IDs for incremental (avoids collision when path lookup misses some old ids).""" for c in chunks: sid = uuid.uuid4().hex[:16] - c.setdefault("metadata", {})["id"] = sid + c["metadata"] = {**c.get("metadata", {}), "id": sid} c["id"] = sid def _chunks_for_paths(self, all_texts: list[dict], paths: set[str]) -> list[dict]: diff --git a/tests/test_incremental_build.py b/tests/test_incremental_build.py index 63ec0ea2..e9cb366a 100644 --- a/tests/test_incremental_build.py +++ b/tests/test_incremental_build.py @@ -371,3 +371,37 @@ def test_ivf_multiple_incremental_no_duplicates(tmp_path): assert len(stale_ids) == 0, ( f"passages.jsonl has {len(stale_ids)} stale entries not in offset_map: {stale_ids[:5]}" ) + + +def test_assign_unique_chunk_ids_with_shared_metadata_dict(): + from leann.cli import LeannCLI + + # Arrange + shared = {"file_path": "a.md"} + chunks = [{"text": t, "metadata": shared} for t in ("one", "two", "three")] + + # Act + LeannCLI._assign_unique_chunk_ids(chunks) + + # Assert + metadata_ids = [c["metadata"]["id"] for c in chunks] + assert len(set(metadata_ids)) == len(chunks) + for c in chunks: + assert c["metadata"]["id"] == c["id"] + + +def test_assign_chunk_ids_with_shared_metadata_dict(): + from leann.cli import LeannCLI + + # Arrange + shared = {"file_path": "a.md"} + chunks = [{"text": t, "metadata": shared} for t in ("one", "two", "three")] + + # Act + LeannCLI._assign_chunk_ids(chunks) + + # Assert + metadata_ids = [c["metadata"]["id"] for c in chunks] + assert len(set(metadata_ids)) == len(chunks) + for c in chunks: + assert c["metadata"]["id"] == c["id"]