Index locked passages so context queries skip unpacking archives

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/<artifact-sha256>.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
 

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,
@@ -113,9 +117,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)
@@ -128,32 +149,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("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)
+        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(
@@ -201,56 +376,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] = {

src/docpull/project.py

@@ -3002,6 +3002,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