From 2f56b3490f52168eb8df19d3c54b1de5f949c485 Mon Sep 17 00:00:00 2001 From: admin-raintree <277948009+admin-raintree@users.noreply.github.com> Date: Mon, 12 Oct 2026 13:15:52 -0700 Subject: [PATCH 1/2] Index locked passages so context queries skip unpacking archives Sync now writes a full-text index of the pinned artifact's passages to .docpull/search/.sqlite. Queries still hash the archive so missing or tampered artifacts fail closed, then read the index read-only instead of unpacking, parsing, and splitting every record. A missing, corrupt, or stale index falls back to the scan path with identical results, and queries never write project files. Co-Authored-By: Claude Opus 5.5 --- docs/context-dependencies.md | 12 +- src/docpull/context_query.py | 302 +++++++++++++++++++++++++++-------- src/docpull/project.py | 4 + tests/test_context_query.py | 34 +++- 4 files changed, 278 insertions(+), 74 deletions(-) diff --git a/docs/context-dependencies.md b/docs/context-dependencies.md index 5532232..641ade0 100644 --- a/docs/context-dependencies.md +++ b/docs/context-dependencies.md @@ -218,9 +218,15 @@ query without partial results. The CLI exits with code 1, the SDK raises `ContextQueryError`, and MCP returns a tool error. The project artifact directory is searched before additional directories; a corrupt archive is never silently replaced. Locks without an archive digest require a new sync or an export of -their retained run, as described under frozen installation. This initial query -path verifies and scans complete archives on each call; it does not use a -persistent search index or semantic embeddings. +their retained run, as described under frozen installation. + +Each query hashes the pinned archive. When the archive changes, sync writes a +full-text index of its passages to `.docpull/search/.sqlite`. +Queries read that index instead of unpacking and splitting the archive. Queries +only read the index, and only when its stored lock facts match the current lock. +A missing, corrupt, or stale index falls back to scanning the archive, with +identical results. Query never writes project files. Retrieval is lexical; it +does not use semantic embeddings. ## Context CI diff --git a/src/docpull/context_query.py b/src/docpull/context_query.py index b2ab3f6..58aefcc 100644 --- a/src/docpull/context_query.py +++ b/src/docpull/context_query.py @@ -5,14 +5,18 @@ import argparse import hashlib import json +import logging +import os import re +import sqlite3 import sys import tempfile from collections.abc import Iterator, Sequence +from contextlib import closing from pathlib import Path from typing import Any -from .frozen import FrozenArtifactError, stage_run_artifact +from .frozen import FrozenArtifactError, _digest, stage_run_artifact from .pack_tools import _search_score, _term_count from .project import ( ProjectError, @@ -104,9 +108,26 @@ def _filters(values: list[str] | None, name: str) -> set[str]: return {v.strip() for v in values} -def _locked_records( - root: Path, artifact_dirs: list[Path], pack_versions: set[str] -) -> Iterator[tuple[dict[str, Any], int, dict[str, Any]]]: +_INDEX_SCHEMA = 1 +_PASSAGE_KEYS = ( + "record_index", + "char_start", + "char_end", + "line_start", + "line_end", + "title", + "url", + "document_id", + "content_hash", + "source", + "version", + "text", +) +_log = logging.getLogger(__name__) + + +def _verified_lock(root: Path, pack_versions: set[str]) -> tuple[dict[str, Any], Path] | None: + """Return the validated lock and project state dir, or None when the run is filtered out.""" config = load_project_config(root) paths = project_paths(root) lock = _read_context_lock(paths.context_lock) @@ -119,32 +140,186 @@ def _locked_records( raise ContextQueryError(f"No run pinned in {root}; run `docpull sync` first") run_id = _safe_run_id(lock["run_id"]) if pack_versions and run_id not in pack_versions: - return + return None artifact = lock.get("artifact") if not isinstance(artifact, dict) or not re.fullmatch(r"[0-9a-f]{64}", str(artifact.get("sha256"))): raise ContextQueryError(f"No frozen artifact pinned in {root}; run `docpull sync` first") - stores = [paths.state / "artifacts", *artifact_dirs] - store = next((p for p in stores if (p / f"{artifact['sha256']}.zip").is_file()), stores[0]) + return lock, paths.state + + +def _artifact_records(lock: dict[str, Any], store: Path) -> list[tuple[int, dict[str, Any]]]: + """Unpack the pinned artifact and verify every record against the lock.""" with tempfile.TemporaryDirectory(prefix="docpull-query-") as temporary: staged = Path(temporary) - stage_run_artifact(artifact, store, staged) + stage_run_artifact(lock["artifact"], store, staged) run = _read_json(staged / "run.json") - if not isinstance(run, dict) or run.get("run_id") != run_id or run.get("project") != config.name: + if ( + not isinstance(run, dict) + or run.get("run_id") != lock["run_id"] + or run.get("project") != lock["project"] + ): raise ContextQueryError("Frozen artifact does not match the locked run") records = _read_jsonl(staged / "documents.ndjson") - if sorted(str(row["content_hash"]) for row in records if row.get("content_hash")) != lock.get( - "content_hashes" - ) or _content_hash_summary(records) != lock.get("content_hash_summary"): - raise ContextQueryError("Frozen artifact content does not match the lockfile") - for index, record in enumerate(records, 1): - if not isinstance(record.get("url"), str) or not record["url"].strip(): - raise ContextQueryError(f"Missing source URL in locked record {index}") - content = record.get("content") - if not isinstance(content, str) or hashlib.sha256( - content.encode("utf-8") - ).hexdigest() != record.get("content_hash"): - raise ContextQueryError(f"Invalid content hash in locked record {index}") - yield lock, index, record + if sorted(str(row["content_hash"]) for row in records if row.get("content_hash")) != lock.get( + "content_hashes" + ) or _content_hash_summary(records) != lock.get("content_hash_summary"): + raise ContextQueryError("Frozen artifact content does not match the lockfile") + for index, record in enumerate(records, 1): + if not isinstance(record.get("url"), str) or not record["url"].strip(): + raise ContextQueryError(f"Missing source URL in locked record {index}") + content = record.get("content") + if not isinstance(content, str) or hashlib.sha256(content.encode("utf-8")).hexdigest() != record.get( + "content_hash" + ): + raise ContextQueryError(f"Invalid content hash in locked record {index}") + return list(enumerate(records, 1)) + + +def _record_passages(record_index: int, record: dict[str, Any]) -> Iterator[dict[str, Any]]: + metadata = record.get("metadata") or {} + version = metadata.get("version") if isinstance(metadata, dict) else None + source = metadata.get("docpull_project_source") if isinstance(metadata, dict) else None + content = record["content"] + for start, end in _spans(content): + # Trim outside whitespace only; never rewrite captured passage text. + raw = content[start:end] + start += len(raw) - len(raw.lstrip()) + end -= len(raw) - len(raw.rstrip()) + text = content[start:end] + if not text: + continue + yield { + "record_index": record_index, + "char_start": start, + "char_end": end, + "line_start": content.count("\n", 0, start) + 1, + "line_end": content.count("\n", 0, end - 1) + 1, + "title": str(record.get("title") or ""), + "url": record["url"], + "document_id": record.get("document_id"), + "content_hash": record["content_hash"], + "source": source if isinstance(source, str) else None, + "version": version if isinstance(version, str) else None, + "text": text, + } + + +def _index_path(state: Path, digest: str) -> Path: + return state / "search" / f"{digest}.sqlite" + + +def _index_key(lock: dict[str, Any]) -> str: + """Lock facts the scan path verifies against the artifact; the index must match them too.""" + fields = ("artifact", "run_id", "project", "content_hashes", "content_hash_summary") + return json.dumps({field: lock.get(field) for field in fields}, sort_keys=True) + + +def build_query_index(state: Path, lock: dict[str, Any]) -> bool: + """Index a pinned artifact's passages for ``query_context``; best effort, never required. + + The index is content-addressed by artifact digest and built from the same verified + records and passages as the scan path, so queries return identical results. + """ + try: + target = _index_path(state, lock["artifact"]["sha256"]) + if target.is_file(): + return True + rows = [ + tuple(passage[key] for key in _PASSAGE_KEYS) + for index, record in _artifact_records(lock, state / "artifacts") + for passage in _record_passages(index, record) + ] + target.parent.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile(dir=target.parent, suffix=".tmp", delete=False) as handle: + temporary = Path(handle.name) + try: + with closing(sqlite3.connect(temporary)) as conn, conn: + conn.execute("CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)") + conn.execute(f"CREATE TABLE passages ({', '.join(_PASSAGE_KEYS)})") + conn.execute( + "CREATE VIRTUAL TABLE passages_fts USING fts5(text, content='passages', " + "content_rowid='rowid', tokenize='unicode61 remove_diacritics 2')" + ) + conn.execute("INSERT INTO meta VALUES ('lock', ?)", (_index_key(lock),)) + conn.executemany(f"INSERT INTO passages VALUES ({', '.join('?' * len(_PASSAGE_KEYS))})", rows) + conn.execute("INSERT INTO passages_fts(passages_fts) VALUES ('rebuild')") + conn.execute(f"PRAGMA user_version = {_INDEX_SCHEMA}") + os.replace(temporary, target) + finally: + temporary.unlink(missing_ok=True) + except ( + ContextQueryError, + FrozenArtifactError, + ProjectError, + KeyError, + OSError, + UnicodeError, + sqlite3.Error, + ) as err: + _log.warning("Skipped query index for %s: %s", state, err) + return False + return True + + +def _indexed_passages(path: Path, lock: dict[str, Any], terms: list[str]) -> list[dict[str, Any]] | None: + """Return passages matching any term, or None to fall back to scanning the artifact.""" + if not path.is_file(): + return None + # ponytail: FTS only prefilters candidates; query_context still applies exact \b + # matching and scoring. unicode61 splits on every non-alphanumeric and folds + # diacritics, so it over-matches the term regex. Rare Unicode combining-mark + # spellings could differ; delete the index to force the scan path if one does. + match = " OR ".join(f'"{term}"' for term in terms) + try: + # immutable=1 keeps queries read-only: no journal, lock, or WAL files. + with closing(sqlite3.connect(f"{path.as_uri()}?mode=ro&immutable=1", uri=True)) as conn: + conn.row_factory = sqlite3.Row + stored = conn.execute("SELECT value FROM meta WHERE key = 'lock'").fetchone() + if ( + conn.execute("PRAGMA user_version").fetchone()[0] != _INDEX_SCHEMA + or stored is None + or stored[0] != _index_key(lock) + ): + return None + rows = conn.execute( + "SELECT p.* FROM passages_fts JOIN passages p ON p.rowid = passages_fts.rowid " + "WHERE passages_fts MATCH ?", + (match,), + ).fetchall() + except sqlite3.Error: + return None + return [dict(row) for row in rows] + + +def _archive_matches(archive: Path, digest: str) -> bool: + try: + with archive.open("rb") as stream: + return _digest(stream) == digest + except OSError: + return False + + +def _passages( + root: Path, artifact_dirs: list[Path], pack_versions: set[str], terms: list[str] +) -> Iterator[tuple[dict[str, Any], dict[str, Any]]]: + found = _verified_lock(root, pack_versions) + if found is None: + return + lock, state = found + digest = lock["artifact"]["sha256"] + stores = [state / "artifacts", *artifact_dirs] + store = next((p for p in stores if (p / f"{digest}.zip").is_file()), stores[0]) + # Hash the pinned archive on every query so a missing or tampered artifact still + # fails closed; the index only saves unpacking, parsing, and passage splitting. + if _archive_matches(store / f"{digest}.zip", digest): + indexed = _indexed_passages(_index_path(state, digest), lock, terms) + if indexed is not None: + for passage in indexed: + yield lock, passage + return + for index, record in _artifact_records(lock, store): + for passage in _record_passages(index, record): + yield lock, passage def query_context( @@ -192,56 +367,43 @@ def query_context( raise ContextQueryError("project_roots must contain at least one project") stores = [Path(p).expanduser().resolve() for p in artifact_dirs or []] for root in sorted({Path(p).expanduser().resolve() for p in roots}): - for lock, record_index, record in _locked_records(root, stores, runs): - metadata = record.get("metadata") or {} - version = metadata.get("version") if isinstance(metadata, dict) else None - source = metadata.get("docpull_project_source") if isinstance(metadata, dict) else None - if not isinstance(version, str): - version = None - if not isinstance(source, str): - source = None - if source_versions and version not in source_versions: + for lock, passage in _passages(root, stores, runs, terms): + if source_versions and passage["version"] not in source_versions: + continue + if source_names and passage["source"] not in source_names: continue - if source_names and source not in source_names: + text = passage["text"] + if not any(_term_count(text, term) for term in terms): continue - content = record["content"] - for start, end in _spans(content): - # Trim outside whitespace only; never rewrite captured passage text. - raw = content[start:end] - start += len(raw) - len(raw.lstrip()) - end -= len(raw) - len(raw.rstrip()) - text = content[start:end] - if not text or not any(_term_count(text, term) for term in terms): - continue - score, _ = _search_score( - query_terms=terms, - phrase=query.strip().lower(), - title=str(record.get("title") or ""), - heading="", - url=str(record.get("url") or ""), - content=text, - ) - digest = lock["artifact"]["sha256"] - identity = f"{digest}:{record_index}:{start}:{end}" - citation = { - "id": "ctx_" + hashlib.sha256(identity.encode()).hexdigest()[:24], - "pack": lock["project"], - "pack_version": lock["run_id"], - "artifact_sha256": digest, - "source": source, - "version": version, - "url": record["url"], - "document_id": record.get("document_id"), - "record_index": record_index, - "content_hash": record["content_hash"], - "char_start": start, - "char_end": end, - "line_start": content.count("\n", 0, start) + 1, - "line_end": content.count("\n", 0, end - 1) + 1, - } - group = groups.setdefault(text, {"score": score, "text": text, "citations": {}}) - group["score"] = max(group["score"], score) - group["citations"][citation["id"]] = citation + score, _ = _search_score( + query_terms=terms, + phrase=query.strip().lower(), + title=passage["title"], + heading="", + url=passage["url"], + content=text, + ) + digest = lock["artifact"]["sha256"] + identity = f"{digest}:{passage['record_index']}:{passage['char_start']}:{passage['char_end']}" + citation = { + "id": "ctx_" + hashlib.sha256(identity.encode()).hexdigest()[:24], + "pack": lock["project"], + "pack_version": lock["run_id"], + "artifact_sha256": digest, + "source": passage["source"], + "version": passage["version"], + "url": passage["url"], + "document_id": passage["document_id"], + "record_index": passage["record_index"], + "content_hash": passage["content_hash"], + "char_start": passage["char_start"], + "char_end": passage["char_end"], + "line_start": passage["line_start"], + "line_end": passage["line_end"], + } + group = groups.setdefault(text, {"score": score, "text": text, "citations": {}}) + group["score"] = max(group["score"], score) + group["citations"][citation["id"]] = citation except (ProjectError, FrozenArtifactError, OSError, UnicodeError) as err: raise ContextQueryError(str(err)) from err payload: dict[str, Any] = { diff --git a/src/docpull/project.py b/src/docpull/project.py index 67521b9..04e1faf 100644 --- a/src/docpull/project.py +++ b/src/docpull/project.py @@ -2988,6 +2988,10 @@ def _write_context_lock( raise ProjectError(str(err)) from err payload["lock_hash"] = _context_lock_hash(payload) _write_json(paths.context_lock, payload) + if payload.get("artifact"): + from .context_query import build_query_index + + build_query_index(paths.state, payload) return payload diff --git a/tests/test_context_query.py b/tests/test_context_query.py index 84d5db5..8438c04 100644 --- a/tests/test_context_query.py +++ b/tests/test_context_query.py @@ -6,12 +6,13 @@ import json import shutil import socket +import sqlite3 from pathlib import Path from typing import Any import pytest -from docpull import ContextQueryError, query_context +from docpull import ContextQueryError, context_query, query_context from docpull import project as project_module from docpull.cli import main from docpull.context_query import serialize_query_result @@ -280,3 +281,34 @@ async def test_context_query_stdio_contract(tmp_path: Path, monkeypatch: pytest. "context_query", {"query": "retry", "project_roots": [str(tmp_path / "missing")]} ) assert missing.isError is True + + +@pytest.mark.parametrize("damage", ["none", "corrupt", "stale"]) +def test_sync_index_matches_scan_and_bad_indexes_fall_back( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, damage: str +) -> None: + content = "# Guide\n\nCafé retry-policy: retry_count 2.\n\nNo match here.\n\n```python\nretry()\n```\n" + root = _project(tmp_path, monkeypatch, "indexed", [(content, "1.0"), ("Retry later.", None)]) + index = next((root / ".docpull/search").glob("*.sqlite")) + query = "cafe retry-policy retry_count retry" + indexed = query_context(query, project_roots=[root], max_tokens=10000) + index.rename(index.with_suffix(".off")) + scanned = query_context(query, project_roots=[root], max_tokens=10000) + index.with_suffix(".off").rename(index) + assert indexed == scanned + assert indexed["status"] == "ok" and len(indexed["results"]) == 3 + + if damage == "corrupt": + index.write_bytes(b"not a database") + elif damage == "stale": + with sqlite3.connect(index) as conn: + conn.execute("UPDATE meta SET value = 'other' WHERE key = 'lock'") + else: + # A valid index must answer without unpacking the artifact. + def no_stage(*args: Any, **kwargs: Any) -> None: + pytest.fail("Indexed query unpacked the artifact") + + monkeypatch.setattr(context_query, "stage_run_artifact", no_stage) + before = _files(root) + assert query_context(query, project_roots=[root], max_tokens=10000) == scanned + assert _files(root) == before From 905b2f925313e02a26e740acbec4da2be70030e3 Mon Sep 17 00:00:00 2001 From: admin-raintree <277948009+admin-raintree@users.noreply.github.com> Date: Mon, 12 Oct 2026 13:20:35 -0700 Subject: [PATCH 2/2] Use a literal placeholder list for the query index insert Bandit B608 flagged the f-string even though every value is bound. Co-Authored-By: Claude Opus 5.5 --- src/docpull/context_query.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/docpull/context_query.py b/src/docpull/context_query.py index 58aefcc..2795ec5 100644 --- a/src/docpull/context_query.py +++ b/src/docpull/context_query.py @@ -241,7 +241,7 @@ def build_query_index(state: Path, lock: dict[str, Any]) -> bool: "content_rowid='rowid', tokenize='unicode61 remove_diacritics 2')" ) conn.execute("INSERT INTO meta VALUES ('lock', ?)", (_index_key(lock),)) - conn.executemany(f"INSERT INTO passages VALUES ({', '.join('?' * len(_PASSAGE_KEYS))})", rows) + conn.executemany("INSERT INTO passages VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", rows) conn.execute("INSERT INTO passages_fts(passages_fts) VALUES ('rebuild')") conn.execute(f"PRAGMA user_version = {_INDEX_SCHEMA}") os.replace(temporary, target)