Phase 4: corpus and retrieval — chunks, Qdrant, FTS5, freshness audit

Satisfies docs/phase-4-corpus.md. content/ stays the single source of truth:
one pass over it writes both indexes, so the static pages and the retrieval
corpus cannot drift.

- corpus/: stdlib-only pipeline. validate -> chunk -> enrich -> embed -> upsert.
  Chunking is block-based, so a boundary cannot fall inside a fenced block, a
  table or front-matter (verified structurally, not by inspection).
- Deterministic ids: sha256(source_path + chunk_index + content_hash), rendered
  as a UUID. Re-running updates points; re-chunking deletes the old ones by
  source filter, so duplicates and orphans are both impossible.
- Incremental by content hash against a SQLite manifest. A provider, dimension
  or chunker change is refused instead of mixed: it needs a rebuild.
- Payloads per section 4.2 plus provenance, validated before any vector is
  written; a missing field fails the run and names the document and chunk.
- SQLite FTS5 over chunks (the keyword half of hybrid retrieval, decision D10)
  with area/confidence filters and a `terms` column so front-matter tech and
  tags are findable.
- Freshness audit: stale documents, due soon, index drift, contract drift;
  optional Apprise alert; exit code for Prefect.
- Embeddings behind one interface: omniroute (OpenAI-compatible, retries, key
  from the environment only) and local-hashing (offline placeholder). The
  provider decision is still Sam's; the pipeline does not depend on it.
- deploy/qdrant.yml: mem_limit = memswap_limit = 400m, loopback only.
- docs/phase-4-notes.md: operator manual, DoD evidence, measured memory.
- 76 unit tests, no container and no network.

Measured, Qdrant 1.20.0 under a 400 m cap at 1024 dimensions: 500 / 5,000 /
20,000 chunks settle at 128 / 138 / 151 MiB, peaking at 149 / 305 / 292 MiB
during ingest. Qdrant's default settings are OOM-killed between 5,000 and
20,000, so the collection is created with HNSW and payloads on disk; recorded in
reference/stack.md with the ceiling and the escalation order.

Note: the `make corpus*` targets were swept into the Phase 3 deploy commit
24590cc by concurrent work in this tree; the Makefile content is correct, and
this commit contains the pipeline they call.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

corpus/__init__.py

@@ -0,0 +1,22 @@
+"""Phase 4 — the retrievable corpus.
+
+`content/` is the source of truth. This package turns it into the two indexes
+the AI console reads:
+
+* **vectors** in Qdrant, one point per chunk, with the payload that lets the
+  query path filter *before* searching (`area`, `confidence`);
+* **keywords** in SQLite FTS5, one row per chunk, so exact tokens (repo names,
+  `LoRaWAN`, `MQTT`, dates) are not lost the way they are with vectors alone
+  (decision D10, hybrid retrieval).
+
+Everything here is driven by `python3 -m corpus.cli`; `docs/phase-4-notes.md`
+is the operator manual and records what was measured. The pipeline is
+Provider-agnostic by design: `corpus/embeddings.py` is the only place that knows
+how a vector is made, and the provider + model + dimension are recorded per
+collection so a provider change is a governed migration, not silent corruption.
+
+Standard library only, plus the repository's own validator (`tools/`). No new
+dependency, no Node, no framework (AGENTS.md section 4.1).
+"""
+
+__all__ = ["config", "loader", "chunker", "payloads", "embeddings", "qdrant", "store", "pipeline", "freshness"]

corpus/__main__.py

@@ -0,0 +1,8 @@
+"""`python3 -m corpus` is the same as `python3 -m corpus.cli`."""
+
+import sys
+
+from .cli import main
+
+if __name__ == "__main__":
+    sys.exit(main())

corpus/chunker.py

@@ -0,0 +1,287 @@
+"""Code-aware chunking.
+
+The rules come from `docs/phase-4-corpus.md` §4.1:
+
+* 400–800 tokens per chunk, about 12% overlap;
+* **never split inside a fenced block, a table or a front-matter block.**
+
+The last rule is structural here, not a post-hoc check: the body is parsed into
+*blocks* first and chunks are packed out of whole blocks, so a boundary can only
+ever fall between blocks. Fenced code, tables, headings, list items and
+paragraphs are all blocks; the large ones cannot be subdivided.
+
+Front-matter is removed by `corpus.loader` before this module sees the body, and
+parsing a body that still has front-matter is an error, so it cannot be split
+either.
+
+`tokens` are estimated with a word-and-punctuation regex. A real BPE tokenizer
+counts 15–30% more for prose and more again for code, so a chunk that measures
+800 here is a conservative upper bound.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+# Packing targets (docs/phase-4-corpus.md §4.1).
+TOKEN_TARGET = 600
+TOKEN_MAX = 800
+TOKEN_MIN = 400
+OVERLAP_RATIO = 0.12
+
+FENCE_RE = re.compile(r"^ {0,3}(`{3,}|~{3,})")
+HEADING_RE = re.compile(r"^ {0,3}#{1,6}\s")
+LIST_RE = re.compile(r"^(\s*)([-*+]|\d{1,3}[.)])\s+")
+RULE_RE = re.compile(r"^ {0,3}([-*_])(\s*\1){2,}\s*$")
+TOKEN_RE = re.compile(r"\w+|[^\w\s]", re.UNICODE)
+
+PROTECTED_KINDS = ("fence", "table")
+
+
+class ChunkError(RuntimeError):
+    """The body could not be chunked without breaking the rules."""
+
+
+@dataclass(frozen=True)
+class Block:
+    """An indivisible unit of Markdown. Never split, only packed."""
+
+    kind: str  # fence | table | heading | list | quote | para | rule
+    text: str
+    tokens: int
+    line_start: int  # 1-based, relative to the body this block came from
+    line_end: int
+
+    @property
+    def protected(self) -> bool:
+        return self.kind in PROTECTED_KINDS
+
+
+@dataclass(frozen=True)
+class Chunk:
+    index: int
+    text: str
+    tokens: int
+    line_start: int
+    line_end: int
+    blocks: tuple
+    oversize: bool = False
+
+    @property
+    def kinds(self) -> tuple:
+        return tuple(block.kind for block in self.blocks)
+
+
+def estimate_tokens(text: str) -> int:
+    """Word-plus-punctuation token estimate. Deterministic and cheap."""
+    return len(TOKEN_RE.findall(text))
+
+
+def parse_blocks(body: str) -> list:
+    """Split a Markdown body into atomic blocks.
+
+    Raises ChunkError if the body still carries a front-matter block: that is a
+    loader bug, and silently chunking it would split something that must not be
+    split.
+    """
+    if body.lstrip().startswith("---"):
+        raise ChunkError(
+            "body still starts with a front-matter block; strip front-matter "
+            "before chunking (corpus.loader does this)"
+        )
+
+    lines = body.split("\n")
+    blocks = []
+    i = 0
+    total = len(lines)
+    while i < total:
+        raw = lines[i]
+        stripped = raw.strip()
+        if not stripped:
+            i += 1
+            continue
+        start = i
+        fence = FENCE_RE.match(raw)
+        if fence:
+            i = _read_fence(lines, i, fence.group(1))
+            kind = "fence"
+        elif stripped.startswith("|"):
+            while i < total and lines[i].strip().startswith("|"):
+                i += 1
+            kind = "table"
+        elif HEADING_RE.match(raw):
+            i += 1
+            kind = "heading"
+        elif RULE_RE.match(raw):
+            i += 1
+            kind = "rule"
+        elif LIST_RE.match(raw):
+            i = _read_list_item(lines, i)
+            kind = "list"
+        elif stripped.startswith(">"):
+            while i < total and lines[i].strip().startswith(">"):
+                i += 1
+            kind = "quote"
+        else:
+            i = _read_paragraph(lines, i)
+            kind = "para"
+        text = "\n".join(lines[start:i])
+        blocks.append(
+            Block(kind=kind, text=text, tokens=estimate_tokens(text), line_start=start + 1, line_end=i)
+        )
+    return blocks
+
+
+def _read_fence(lines: list, i: int, marker: str) -> int:
+    """Return the index just past a fenced block. Unterminated runs to EOF."""
+    char = marker[0]
+    width = len(marker)
+    i += 1
+    while i < len(lines):
+        candidate = FENCE_RE.match(lines[i])
+        if candidate and candidate.group(1)[0] == char and len(candidate.group(1)) >= width:
+            return i + 1
+        i += 1
+    return len(lines)
+
+
+def _read_list_item(lines: list, i: int) -> int:
+    """Return the index just past one top-level list item, continuations included."""
+    base_indent = len(LIST_RE.match(lines[i]).group(1))
+    i += 1
+    while i < len(lines):
+        raw = lines[i]
+        if not raw.strip():
+            # A blank line continues the item only if an indented line follows.
+            if i + 1 < len(lines) and lines[i + 1].startswith(" " * (base_indent + 1)):
+                i += 1
+                continue
+            return i
+        indent = len(raw) - len(raw.lstrip())
+        if indent > base_indent or LIST_RE.match(raw) and len(LIST_RE.match(raw).group(1)) > base_indent:
+            i += 1
+            continue
+        if LIST_RE.match(raw):
+            return i
+        if indent <= base_indent:
+            return i
+        i += 1
+    return i
+
+
+def _read_paragraph(lines: list, i: int) -> int:
+    i += 1
+    while i < len(lines):
+        raw = lines[i]
+        stripped = raw.strip()
+        if not stripped:
+            return i
+        if FENCE_RE.match(raw) or HEADING_RE.match(raw) or RULE_RE.match(raw):
+            return i
+        if stripped.startswith("|") or stripped.startswith(">") or LIST_RE.match(raw):
+            return i
+        i += 1
+    return i
+
+
+def chunk_blocks(blocks: list) -> list:
+    """Pack blocks into windows of at most TOKEN_MAX, with ~12% overlap."""
+    chunks = []
+    total = len(blocks)
+    if total == 0:
+        return chunks
+    i = 0
+    while i < total:
+        start = i
+        tokens = 0
+        end = i
+        while end < total:
+            block = blocks[end]
+            if end > start:
+                if tokens + block.tokens > TOKEN_MAX:
+                    break
+                # Prefer breaking in front of a heading.
+                if block.kind == "heading" and tokens >= int(TOKEN_TARGET * 0.75):
+                    break
+            tokens += block.tokens
+            end += 1
+            if tokens >= TOKEN_TARGET and end < total and blocks[end].kind == "heading":
+                break
+        window = blocks[start:end]
+        chunks.append(_make_chunk(len(chunks), window))
+        if end >= total:
+            break
+        # Back up over trailing blocks until roughly OVERLAP_RATIO of the window
+        # repeats, leaving at least one block behind so the walk always advances.
+        budget = max(1, int(round(TOKEN_TARGET * OVERLAP_RATIO)))
+        carried = 0
+        back = end
+        while back - 1 > start and carried + blocks[back - 1].tokens <= budget * 2:
+            carried += blocks[back - 1].tokens
+            back -= 1
+            if carried >= budget:
+                break
+        i = max(back, start + 1)
+    return chunks
+
+
+def _make_chunk(index: int, window: list) -> Chunk:
+    text = "\n\n".join(block.text for block in window)
+    return Chunk(
+        index=index,
+        text=text,
+        tokens=sum(block.tokens for block in window),
+        line_start=window[0].line_start,
+        line_end=window[-1].line_end,
+        blocks=tuple(window),
+        oversize=len(window) == 1 and window[0].tokens > TOKEN_MAX,
+    )
+
+
+def chunk_document(body: str) -> list:
+    """Chunk one document body. The pipeline calls this and nothing else."""
+    chunks = chunk_blocks(parse_blocks(body))
+    verify_no_split(chunks, protected_regions(body))
+    return chunks
+
+
+def protected_regions(body: str) -> list:
+    """Line ranges (1-based, inclusive) that a chunk boundary must not cross.
+
+    Recomputed from the body text alone, independently of the packing code, so
+    the check is not the implementation grading its own homework.
+    """
+    regions = []
+    lines = body.split("\n")
+    i = 0
+    while i < len(lines):
+        raw = lines[i]
+        fence = FENCE_RE.match(raw)
+        if fence:
+            end = _read_fence(lines, i, fence.group(1))
+            regions.append((i + 1, end))
+            i = end
+            continue
+        if raw.strip().startswith("|"):
+            start = i
+            while i < len(lines) and lines[i].strip().startswith("|"):
+                i += 1
+            regions.append((start + 1, i))
+            continue
+        i += 1
+    return regions
+
+
+def verify_no_split(chunks: list, regions: list) -> None:
+    """Raise ChunkError if any chunk boundary cuts a protected region."""
+    for chunk in chunks:
+        for first, last in regions:
+            spans_start = chunk.line_start <= first <= chunk.line_end
+            spans_end = chunk.line_start <= last <= chunk.line_end
+            covers = chunk.line_start <= first and chunk.line_end >= last
+            if (spans_start or spans_end) and not covers:
+                raise ChunkError(
+                    f"chunk {chunk.index} splits protected lines {first}-{last} "
+                    f"(chunk covers {chunk.line_start}-{chunk.line_end})"
+                )

corpus/cli.py

@@ -0,0 +1,257 @@
+"""Command line entry point: `python3 -m corpus.cli <command>`.
+
+Commands
+
+    sync          incremental update of the corpus (the normal command)
+    rebuild       drop the collection and rebuild it from content/ (recovery)
+    audit         freshness audit; reports stale documents, index drift
+    verify        does the index agree with content/ and with Qdrant?
+    demo-filter   the same query with and without an `area` filter
+    hybrid        vectors + FTS5 fused with reciprocal rank fusion
+    measure       Qdrant memory at 500 / 5,000 / 20,000 chunks
+    info          what this build thinks it is configured to do
+
+Exit codes: 0 success, 1 findings or verification problems, 2 configuration or
+usage error. Every command accepts `--json` where a machine-readable result is
+useful (Prefect reads JSON, a human reads the table).
+
+Configuration is environment-first (`corpus/config.py`); the flags exist so a
+one-off run can point at a scratch Qdrant without exporting variables.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime
+import json
+import os
+import sys
+
+from . import config as config_module
+from . import freshness, pipeline, store as store_module
+from .config import Config, ConfigError
+from .qdrant import Qdrant, QdrantError
+
+
+def main(argv=None) -> int:
+    parser = _build_parser()
+    args = parser.parse_args(argv)
+    if args.command is None:
+        parser.print_help()
+        return 2
+
+    overrides = {
+        "root": args.root,
+        "qdrant_url": args.qdrant_url,
+        "collection": args.collection,
+        "provider": args.provider,
+        "model": args.model,
+        "dim": args.dim,
+        "db_path": args.db,
+        "data_dir": args.data,
+    }
+    try:
+        cfg = Config.from_env(**overrides)
+    except ConfigError as exc:
+        print(f"configuration error: {exc}", file=sys.stderr)
+        return 2
+
+    log = (lambda _msg: None) if args.quiet else (lambda msg: print(msg, file=sys.stderr))
+
+    try:
+        handler = {
+            "sync": _cmd_sync,
+            "rebuild": _cmd_rebuild,
+            "audit": _cmd_audit,
+            "verify": _cmd_verify,
+            "demo-filter": _cmd_demo_filter,
+            "hybrid": _cmd_hybrid,
+            "measure": _cmd_measure,
+            "info": _cmd_info,
+        }[args.command]
+        return handler(cfg, args, log)
+    except (pipeline.PipelineError, ConfigError) as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 2
+    except QdrantError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 2
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="python3 -m corpus.cli",
+        description="Build and audit the Ai Resume retrieval corpus (docs/phase-4-corpus.md).",
+    )
+    parser.add_argument("--root", help="repository root (default: parent of corpus/)")
+    parser.add_argument("--qdrant-url", help="Qdrant REST URL (default http://127.0.0.1:6333)")
+    parser.add_argument("--collection", help="Qdrant collection name (default resume_corpus)")
+    parser.add_argument("--provider", choices=sorted(config_module.PROVIDERS), help="embedding provider")
+    parser.add_argument("--model", help="embedding model name")
+    parser.add_argument("--dim", type=int, help="embedding dimension")
+    parser.add_argument("--db", help="SQLite corpus database path")
+    parser.add_argument("--data", help="corpus data directory")
+    parser.add_argument("--quiet", action="store_true", help="progress on stderr only becomes silent")
+
+    sub = parser.add_subparsers(dest="command")
+
+    sync = sub.add_parser("sync", help="incremental update")
+    sync.add_argument("--json", action="store_true")
+
+    rebuild = sub.add_parser("rebuild", help="drop the collection and rebuild from content/")
+    rebuild.add_argument("--json", action="store_true")
+
+    audit = sub.add_parser("audit", help="freshness and drift audit")
+    audit.add_argument("--json", action="store_true")
+    audit.add_argument("--today", help="audit as of this date (YYYY-MM-DD), for tests")
+    audit.add_argument("--fail-on-findings", action="store_true", help="exit 1 when anything is found")
+    audit.add_argument("--alert", action="store_true", help="POST the report to RESUME_APPRISE_URL")
+
+    verify = sub.add_parser("verify", help="check the index against content/ and Qdrant")
+    verify.add_argument("--json", action="store_true")
+
+    demo = sub.add_parser("demo-filter", help="same query with and without an area filter")
+    demo.add_argument("--query", required=True)
+    demo.add_argument("--area", help="area to filter on (ai, platform, edge, products, history, about)")
+    demo.add_argument("--limit", type=int, default=5)
+    demo.add_argument("--json", action="store_true")
+
+    hybrid = sub.add_parser("hybrid", help="vectors + FTS5 fused with RRF")
+    hybrid.add_argument("--query", required=True)
+    hybrid.add_argument("--area")
+    hybrid.add_argument("--limit", type=int, default=5)
+    hybrid.add_argument("--json", action="store_true")
+
+    measure = sub.add_parser("measure", help="Qdrant memory at three corpus sizes")
+    measure.add_argument("--sizes", default="500,5000,20000")
+    measure.add_argument("--container", default="", help="Qdrant container name for docker stats")
+    measure.add_argument("--keep", action="store_true", help="keep the measurement collection")
+    measure.add_argument("--json", action="store_true")
+
+    sub.add_parser("info", help="print the resolved configuration")
+    return parser
+
+
+# -- commands --------------------------------------------------------------
+
+
+def _cmd_sync(cfg, args, log) -> int:
+    stats = pipeline.run(cfg, mode="sync", log=log)
+    if args.json:
+        print(json.dumps(stats.to_dict(), indent=2, sort_keys=True))
+    else:
+        print(stats.summary())
+    return 0
+
+
+def _cmd_rebuild(cfg, args, log) -> int:
+    stats = pipeline.run(cfg, mode="rebuild", log=log)
+    if args.json:
+        print(json.dumps(stats.to_dict(), indent=2, sort_keys=True))
+    else:
+        print(stats.summary())
+    return 0
+
+
+def _cmd_audit(cfg, args, log) -> int:
+    today = datetime.date.fromisoformat(args.today) if args.today else None
+    report = freshness.audit(cfg, today=today, log=log)
+    if args.json:
+        print(json.dumps(report, indent=2, sort_keys=True))
+    else:
+        print(freshness.render_text(report))
+    if args.alert:
+        try:
+            sent = freshness.alert(report, cfg.apprise_url)
+            if not sent:
+                log("alert: RESUME_APPRISE_URL is not set; nothing sent")
+        except Exception as exc:  # noqa: BLE001 - an alert failure must not hide the audit
+            print(f"warning: could not send the Apprise alert: {exc}", file=sys.stderr)
+    if args.fail_on_findings and not report["ok"]:
+        return 1
+    return 0
+
+
+def _cmd_verify(cfg, args, log) -> int:
+    problems = pipeline.verify(cfg, log=log)
+    if args.json:
+        print(json.dumps({"ok": not problems, "problems": problems}, indent=2))
+    else:
+        if problems:
+            print(f"FAIL: {len(problems)} problem(s)")
+            for problem in problems:
+                print(f"  - {problem}")
+        else:
+            print("OK: the index agrees with content/ and Qdrant")
+    return 1 if problems else 0
+
+
+def _cmd_demo_filter(cfg, args, log) -> int:
+    result = pipeline.demo_area_filter(cfg, args.query, area=args.area, limit=args.limit)
+    if args.json:
+        print(json.dumps(result, indent=2, sort_keys=True))
+        return 0
+    print(f"query: {result['query']!r}")
+    print("")
+    print(f"without a filter ({len(result['unfiltered'])} result(s)):")
+    for row in result["unfiltered"]:
+        print(f"  {row['score']:.3f}  [{row['area']}/{row['confidence']}] {row['title']}  {row['url']}")
+    print("")
+    if not args.area:
+        print("no --area given, so the filtered set is empty; re-run with e.g. --area edge")
+        return 0
+    print(f"with area={args.area!r} ({len(result['filtered'])} result(s)):")
+    for row in result["filtered"]:
+        print(f"  {row['score']:.3f}  [{row['area']}/{row['confidence']}] {row['title']}  {row['url']}")
+    wrong = [row for row in result["filtered"] if row["area"] != args.area]
+    if wrong:
+        print(f"FAIL: {len(wrong)} filtered result(s) are not in area {args.area!r}")
+        return 1
+    return 0
+
+
+def _cmd_hybrid(cfg, args, log) -> int:
+    rows = pipeline.hybrid_search(cfg, args.query, area=args.area, limit=args.limit)
+    if args.json:
+        print(json.dumps(rows, indent=2, sort_keys=True, default=str))
+        return 0
+    print(f"hybrid (RRF) query={args.query!r} area={args.area!r}: {len(rows)} result(s)")
+    for row in rows:
+        print(f"  {row['score']:.4f}  {row.get('title')}  {row.get('url')}  [{row['id'][:8]}]")
+    return 0
+
+
+def _cmd_measure(cfg, args, log) -> int:
+    from . import measure as measure_module
+
+    sizes = [int(part) for part in args.sizes.split(",") if part.strip()]
+    report = measure_module.measure(
+        cfg, sizes=sizes, container=args.container, keep=args.keep, log=log
+    )
+    if args.json:
+        print(json.dumps(report, indent=2, sort_keys=True))
+        return 0
+    print(measure_module.render_text(report))
+    return 0
+
+
+def _cmd_info(cfg, args, log) -> int:
+    print(cfg.describe())
+    qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout)
+    print("")
+    print(f"qdrant: {'answering, version ' + qdrant.version() if qdrant.health() else 'NOT answering'}")
+    if qdrant.collection_exists(cfg.collection):
+        print(f"collection {cfg.collection}: {qdrant.point_count(cfg.collection)} point(s)")
+    else:
+        print(f"collection {cfg.collection}: does not exist")
+    if os.path.exists(cfg.db_path):
+        with store_module.Store(cfg.db_path) as store:
+            print(f"store: {store.chunk_count()} chunk(s) across {store.source_count()} source(s)")
+            print(f"contract: {json.dumps(store.get_meta('collection'), sort_keys=True)}")
+    else:
+        print(f"store: {cfg.db_path} does not exist")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

corpus/config.py

@@ -0,0 +1,160 @@
+"""Configuration for the corpus pipeline. Environment only.
+
+Nothing in this file holds a credential. The OmniRoute key is read from the
+process environment, which systemd populates from
+`~/.config/environment.d/10-secrets.conf` (AGENTS.md section 4.4). `describe()`
+is the only way the pipeline prints its configuration, and it redacts the key.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+# Bumped whenever chunk boundaries or payload shape change. A stored corpus
+# carrying a different version is stale by definition and gets rebuilt.
+CHUNKER_VERSION = "blocks-v1"
+
+# Provider defaults. `local-hashing` is a deterministic, offline provider used
+# by the tests, by `make corpus-test`, and by any rebuild on a host with no
+# model gateway. It is NOT a semantic embedding model: it exists so the whole
+# pipeline can be exercised without a provider decision, and so a rebuild never
+# depends on the network. The production provider is `omniroute`.
+PROVIDERS = {
+    "local-hashing": ("local-hashing-v1", 256),
+    "omniroute": ("text-embedding-3-small", 1536),
+}
+
+DEFAULT_QDRANT_URL = "http://127.0.0.1:6333"
+DEFAULT_COLLECTION = "resume_corpus"
+DEFAULT_PROVIDER = "local-hashing"
+DEFAULT_OMNIROUTE_URL = "https://omniroute.lab.audasmedia.com.au"
+
+# Freshness horizon: a document whose `last_verified` is older than this is
+# reported stale by the nightly audit. docs/phase-4-notes.md section 5.
+DEFAULT_MAX_AGE_DAYS = 180
+DEFAULT_WARN_DAYS = 30
+
+
+class ConfigError(RuntimeError):
+    """A configuration value is missing or unusable."""
+
+
+@dataclass(frozen=True)
+class Config:
+    """Everything the pipeline needs, resolved once."""
+
+    root: str
+    content_dir: str
+    data_dir: str
+    db_path: str
+    qdrant_url: str
+    collection: str
+    provider: str
+    model: str
+    dim: int
+    api_url: str
+    api_key: str
+    batch_size: int
+    timeout: float
+    max_age_days: int
+    warn_days: int
+    apprise_url: str
+    vectors_on_disk: bool = False
+
+    # -- construction -----------------------------------------------------
+
+    @classmethod
+    def from_env(cls, root: str | None = None, **overrides) -> "Config":
+        """Build a Config from the environment, then apply explicit overrides.
+
+        Overrides exist so the CLI can accept flags without a second code path;
+        they are applied last so a flag always wins over the environment.
+        """
+        root = os.path.abspath(root or os.environ.get("RESUME_ROOT") or _default_root())
+        content_dir = os.environ.get("RESUME_CONTENT") or os.path.join(root, "content")
+        data_dir = os.environ.get("RESUME_CORPUS_DATA") or os.path.join(root, "data", "corpus")
+        provider = os.environ.get("RESUME_EMBED_PROVIDER") or DEFAULT_PROVIDER
+        if provider not in PROVIDERS:
+            raise ConfigError(
+                f"RESUME_EMBED_PROVIDER={provider!r} is not one of {sorted(PROVIDERS)}"
+            )
+        default_model, default_dim = PROVIDERS[provider]
+
+        values = dict(
+            root=root,
+            content_dir=os.path.abspath(content_dir),
+            data_dir=os.path.abspath(data_dir),
+            db_path=os.path.abspath(
+                os.environ.get("RESUME_CORPUS_DB") or os.path.join(data_dir, "corpus.db")
+            ),
+            qdrant_url=(os.environ.get("RESUME_QDRANT_URL") or DEFAULT_QDRANT_URL).rstrip("/"),
+            collection=os.environ.get("RESUME_COLLECTION") or DEFAULT_COLLECTION,
+            provider=provider,
+            model=os.environ.get("RESUME_EMBED_MODEL") or default_model,
+            dim=int(os.environ.get("RESUME_EMBED_DIM") or default_dim),
+            api_url=(os.environ.get("OMNIROUTE_URL") or DEFAULT_OMNIROUTE_URL).rstrip("/"),
+            api_key=os.environ.get("OMNIROUTE_API_KEY") or "",
+            batch_size=int(os.environ.get("RESUME_EMBED_BATCH") or 32),
+            timeout=float(os.environ.get("RESUME_QDRANT_TIMEOUT") or 30.0),
+            max_age_days=int(
+                os.environ.get("RESUME_MAX_AGE_DAYS") or DEFAULT_MAX_AGE_DAYS
+            ),
+            warn_days=int(os.environ.get("RESUME_WARN_DAYS") or DEFAULT_WARN_DAYS),
+            apprise_url=os.environ.get("RESUME_APPRISE_URL") or "",
+            vectors_on_disk=(os.environ.get("RESUME_QDRANT_VECTORS_ON_DISK") or "").lower()
+            in ("1", "true", "yes"),
+        )
+        values.update({k: v for k, v in overrides.items() if v not in (None, "")})
+        cfg = cls(**values)
+        cfg.check()
+        return cfg
+
+    # -- validation -------------------------------------------------------
+
+    def check(self) -> None:
+        if self.dim <= 0:
+            raise ConfigError(f"embedding dimension must be positive, got {self.dim}")
+        if self.batch_size <= 0:
+            raise ConfigError(f"batch size must be positive, got {self.batch_size}")
+        if self.provider == "omniroute" and not self.api_key:
+            raise ConfigError(
+                "provider 'omniroute' needs OMNIROUTE_API_KEY in the environment "
+                "(~/.config/environment.d/10-secrets.conf). It is never read from this repo."
+            )
+        if not os.path.isdir(self.content_dir):
+            raise ConfigError(f"content directory not found: {self.content_dir}")
+
+    # -- presentation -----------------------------------------------------
+
+    def describe(self) -> str:
+        """Human-readable configuration. Never includes the API key value."""
+        key = "set" if self.api_key else "unset"
+        return (
+            f"root={self.root}\n"
+            f"content={self.content_dir}\n"
+            f"data={self.data_dir}\n"
+            f"db={self.db_path}\n"
+            f"qdrant={self.qdrant_url} collection={self.collection}\n"
+            f"provider={self.provider} model={self.model} dim={self.dim} "
+            f"batch={self.batch_size}\n"
+            f"api_url={self.api_url} api_key={key}\n"
+            f"freshness=max_age_days:{self.max_age_days} warn_days:{self.warn_days}\n"
+            f"qdrant vectors_on_disk={self.vectors_on_disk}\n"
+            f"chunker={CHUNKER_VERSION}"
+        )
+
+    # -- derived paths ----------------------------------------------------
+
+    @property
+    def chunker_version(self) -> str:
+        return CHUNKER_VERSION
+
+    @property
+    def meta_path(self) -> str:
+        return os.path.join(self.data_dir, "collection.json")
+
+
+def _default_root() -> str:
+    """The repository root: the parent of this package."""
+    return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

corpus/embeddings.py

@@ -0,0 +1,241 @@
+"""The one place a vector is made.
+
+`docs/phase-4-corpus.md` §4.3: *"Record the embedding model and dimension per
+collection. A provider change is a governed migration, not a silent corruption.
+Provider decision pending Sam — build provider-agnostic behind one interface."*
+
+So: one interface (`Provider`), two implementations, and the provider name,
+model and dimension are written to the collection metadata by the pipeline. A
+later run with a different model refuses to mix vectors into the same
+collection; it reports a governed migration instead.
+
+**local-hashing** is deterministic, offline, and *not semantic*. It hashes word
+tokens into a fixed-width vector. It exists so that the pipeline, its tests and
+a from-scratch rebuild never depend on a network or on an unmade decision. It is
+the default for `make corpus-test` and for the Qdrant memory measurement.
+
+**omniroute** is the production path: an OpenAI-compatible `/v1/embeddings`
+call through OmniRoute, which is the only permitted model gateway
+(AGENTS.md §4.5). No provider SDK, no direct vendor call, key from the
+environment only.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import time
+import urllib.error
+import urllib.request
+
+
+class EmbeddingError(RuntimeError):
+    """Embedding failed. The pipeline aborts rather than writing partial vectors."""
+
+
+class Provider:
+    """Interface every embedding provider implements."""
+
+    name = "provider"
+    model = "model"
+    dim = 0
+
+    def embed(self, texts: list) -> list:
+        raise NotImplementedError
+
+    def stats(self) -> dict:
+        """Call counters. The DoD proves incremental mode with these numbers."""
+        return {"calls": 0, "texts": 0}
+
+    def reset_stats(self) -> None:
+        pass
+
+    def describe(self) -> dict:
+        return {"provider": self.name, "model": self.model, "dim": self.dim}
+
+
+class LocalHashingProvider(Provider):
+    """Deterministic feature hashing. Offline, fast, not semantic.
+
+    Every token is hashed with sha256 (never Python's salted `hash()`), so the
+    same text produces the same vector on every host and every run. Vectors are
+    L2-normalised, which suits cosine distance.
+    """
+
+    name = "local-hashing"
+
+    def __init__(self, model: str = "local-hashing-v1", dim: int = 256) -> None:
+        if dim <= 0:
+            raise EmbeddingError(f"local-hashing needs a positive dimension, got {dim}")
+        self.model = model
+        self.dim = dim
+        self._calls = 0
+        self._texts = 0
+
+    def embed(self, texts: list) -> list:
+        self._calls += 1
+        self._texts += len(texts)
+        return [self._vector(text) for text in texts]
+
+    def _vector(self, text: str) -> list:
+        vector = [0.0] * self.dim
+        for token in _tokens(text):
+            digest = hashlib.sha256(token.encode("utf-8")).digest()
+            index = int.from_bytes(digest[:8], "big") % self.dim
+            sign = 1.0 if digest[8] & 1 else -1.0
+            vector[index] += sign
+        norm = math.sqrt(sum(value * value for value in vector))
+        if norm == 0.0:
+            # An empty text still needs a unit vector of the right width.
+            vector[0] = 1.0
+            return vector
+        return [value / norm for value in vector]
+
+    def stats(self) -> dict:
+        return {"calls": self._calls, "texts": self._texts}
+
+    def reset_stats(self) -> None:
+        self._calls = 0
+        self._texts = 0
+
+
+class OmniRouteProvider(Provider):
+    """OpenAI-compatible `/v1/embeddings` through OmniRoute.
+
+    Retries on 429 and 5xx with exponential backoff (the risk table in
+    `docs/phase-4-corpus.md` calls this out for bulk ingest). The key is never
+    logged, never put in an error message, and never read from this repository.
+    """
+
+    name = "omniroute"
+
+    def __init__(
+        self,
+        api_url: str,
+        api_key: str,
+        model: str,
+        dim: int,
+        batch_size: int = 32,
+        timeout: float = 30.0,
+        attempts: int = 4,
+        backoff: float = 1.5,
+    ) -> None:
+        if not api_key:
+            raise EmbeddingError(
+                "OmniRoute needs OMNIROUTE_API_KEY in the environment "
+                "(~/.config/environment.d/10-secrets.conf)"
+            )
+        self.api_url = api_url.rstrip("/")
+        self._api_key = api_key
+        self.model = model
+        self.dim = dim
+        self.batch_size = max(1, batch_size)
+        self.timeout = timeout
+        self.attempts = max(1, attempts)
+        self.backoff = backoff
+        self._calls = 0
+        self._texts = 0
+
+    def embed(self, texts: list) -> list:
+        out = []
+        for start in range(0, len(texts), self.batch_size):
+            batch = texts[start : start + self.batch_size]
+            out.extend(self._embed_batch(batch))
+        return out
+
+    def _embed_batch(self, batch: list) -> list:
+        body = json.dumps({"model": self.model, "input": batch}).encode("utf-8")
+        last_error = None
+        for attempt in range(self.attempts):
+            request = urllib.request.Request(
+                f"{self.api_url}/v1/embeddings",
+                data=body,
+                method="POST",
+                headers={
+                    "Content-Type": "application/json",
+                    "Authorization": f"Bearer {self._api_key}",
+                },
+            )
+            self._calls += 1
+            try:
+                with urllib.request.urlopen(request, timeout=self.timeout) as response:
+                    payload = json.loads(response.read().decode("utf-8"))
+                vectors = _vectors_from_response(payload, self.model, self.dim)
+                self._texts += len(batch)
+                return vectors
+            except urllib.error.HTTPError as exc:
+                last_error = f"HTTP {exc.code} from {self.api_url}/v1/embeddings"
+                exc.close()
+                # 4xx other than 429 is our fault: do not hammer the gateway.
+                if exc.code != 429 and exc.code < 500:
+                    raise EmbeddingError(f"{last_error} (not retryable)") from None
+            except (urllib.error.URLError, TimeoutError) as exc:
+                last_error = f"cannot reach {self.api_url}: {exc.reason}"
+            except EmbeddingError:
+                raise
+            if attempt + 1 < self.attempts:
+                time.sleep(self.backoff ** attempt)
+        raise EmbeddingError(f"{last_error} after {self.attempts} attempts")
+
+    def stats(self) -> dict:
+        return {"calls": self._calls, "texts": self._texts}
+
+    def reset_stats(self) -> None:
+        self._calls = 0
+        self._texts = 0
+
+
+def _vectors_from_response(payload: dict, model: str, dim: int) -> list:
+    data = payload.get("data")
+    if not isinstance(data, list) or not data:
+        raise EmbeddingError("embedding response has no 'data' array")
+    ordered = sorted(data, key=lambda item: item.get("index", 0))
+    vectors = []
+    for item in ordered:
+        vector = item.get("embedding")
+        if not isinstance(vector, list) or not vector:
+            raise EmbeddingError("embedding response item has no 'embedding' array")
+        if dim and len(vector) != dim:
+            raise EmbeddingError(
+                f"provider returned dimension {len(vector)} but the collection is "
+                f"configured for {dim}; refusing to mix dimensions"
+            )
+        vectors.append([float(value) for value in vector])
+    returned_model = payload.get("model")
+    if returned_model and model and returned_model != model:
+        raise EmbeddingError(
+            f"provider answered as model {returned_model!r}, not {model!r}; "
+            "a model change must be a governed migration"
+        )
+    return vectors
+
+
+def build_provider(cfg) -> Provider:
+    """Construct the provider named by the configuration."""
+    if cfg.provider == "local-hashing":
+        return LocalHashingProvider(model=cfg.model, dim=cfg.dim)
+    if cfg.provider == "omniroute":
+        return OmniRouteProvider(
+            api_url=cfg.api_url,
+            api_key=cfg.api_key,
+            model=cfg.model,
+            dim=cfg.dim,
+            batch_size=cfg.batch_size,
+            timeout=cfg.timeout,
+        )
+    raise EmbeddingError(f"unknown embedding provider {cfg.provider!r}")
+
+
+def _tokens(text: str) -> list:
+    out = []
+    current = []
+    for char in text.lower():
+        if char.isalnum():
+            current.append(char)
+        elif current:
+            out.append("".join(current))
+            current = []
+    if current:
+        out.append("".join(current))
+    return out

corpus/freshness.py

@@ -0,0 +1,197 @@
+"""Freshness audit — the thing that stops the corpus rotting silently.
+
+The risk table in `docs/phase-4-corpus.md` names it: *"Silent staleness… a
+stale index is otherwise an invisible quality failure."* So the audit is a first
+class command, it runs nightly from Prefect, and it alerts through the existing
+Apprise service when it finds something.
+
+It reports four kinds of finding, and they are different problems:
+
+* **stale** — a document whose `last_verified` is older than the horizon. The
+  claim on the page may no longer be true.
+* **due soon** — inside the warning window. Re-verify before it becomes stale.
+* **index drift** — the source changed (or disappeared, or was never indexed)
+  since it was embedded. Retrieval is answering from an older world.
+* **contract drift** — the collection was built with a different model,
+  dimension or chunker than the running configuration.
+
+Exit code is 0 for a clean audit and 1 when findings exist and `--fail-on-findings`
+was passed, so the Prefect deployment can alert on the result without parsing
+text.
+"""
+
+from __future__ import annotations
+
+import datetime
+import json
+import urllib.request
+
+from . import loader, store as store_module
+from .config import CHUNKER_VERSION
+from .pipeline import _contract
+from .qdrant import Qdrant
+
+
+def audit(cfg, today: datetime.date = None, log=None) -> dict:
+    """Return the audit report as a plain dict (JSON-serialisable)."""
+    log = log or (lambda _msg: None)
+    today = today or datetime.date.today()
+    documents = loader.load_documents(cfg)
+
+    findings = {"stale": [], "due_soon": [], "index_drift": [], "contract_drift": []}
+
+    for doc in documents:
+        try:
+            verified = datetime.date.fromisoformat(doc.last_verified)
+        except ValueError:
+            findings["index_drift"].append(
+                {"source_path": doc.path, "problem": f"invalid last_verified {doc.last_verified!r}"}
+            )
+            continue
+        age = (today - verified).days
+        entry = {
+            "source_path": doc.path,
+            "title": doc.title,
+            "area": doc.area,
+            "confidence": doc.confidence,
+            "last_verified": doc.last_verified,
+            "age_days": age,
+        }
+        if age > cfg.max_age_days:
+            findings["stale"].append(entry)
+        elif age > cfg.max_age_days - cfg.warn_days:
+            findings["due_soon"].append(entry)
+
+    with store_module.Store(cfg.db_path) as store:
+        manifest = store.manifest()
+        on_disk = {doc.path: doc for doc in documents}
+
+        for doc in documents:
+            row = manifest.get(doc.path)
+            if row is None:
+                findings["index_drift"].append(
+                    {"source_path": doc.path, "problem": "not indexed (run sync)"}
+                )
+                continue
+            if row["content_hash"] != doc.digest:
+                findings["index_drift"].append(
+                    {"source_path": doc.path, "problem": "source changed since it was embedded"}
+                )
+            if row["chunker_version"] != CHUNKER_VERSION:
+                findings["index_drift"].append(
+                    {
+                        "source_path": doc.path,
+                        "problem": f"chunked with {row['chunker_version']}, current {CHUNKER_VERSION}",
+                    }
+                )
+
+        for source in sorted(set(manifest) - set(on_disk)):
+            findings["index_drift"].append(
+                {"source_path": source, "problem": "indexed but no longer in content/"}
+            )
+
+        stored_contract = store.get_meta("collection")
+        if stored_contract != _contract(cfg):
+            findings["contract_drift"].append(
+                {"problem": "collection contract does not match this configuration",
+                 "stored": stored_contract, "configured": _contract(cfg)}
+            )
+
+        qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout)
+        reachable = qdrant.health()
+        points = None
+        if reachable and qdrant.collection_exists(cfg.collection):
+            points = qdrant.point_count(cfg.collection)
+        elif not reachable:
+            findings["contract_drift"].append(
+                {"problem": f"Qdrant is not answering at {cfg.qdrant_url}"}
+            )
+
+        report = {
+            "generated_at": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
+            "today": today.isoformat(),
+            "horizon": {
+                "max_age_days": cfg.max_age_days,
+                "warn_days": cfg.warn_days,
+            },
+            "corpus": {
+                "documents": len(documents),
+                "chunks": store.chunk_count(),
+                "sources": store.source_count(),
+                "collection": cfg.collection,
+                "collection_points": points,
+                "provider": cfg.provider,
+                "model": cfg.model,
+                "dim": cfg.dim,
+                "chunker_version": CHUNKER_VERSION,
+            },
+            "findings": findings,
+            "summary": {
+                "stale": len(findings["stale"]),
+                "due_soon": len(findings["due_soon"]),
+                "index_drift": len(findings["index_drift"]),
+                "contract_drift": len(findings["contract_drift"]),
+            },
+            # `due_soon` is a warning, not a failure: the document is still
+            # inside its verification horizon. Stale, drift and contract
+            # problems are failures.
+            "ok": not (
+                findings["stale"] or findings["index_drift"] or findings["contract_drift"]
+            ),
+        }
+    log(f"audit: {json.dumps(report['summary'], sort_keys=True)}")
+    return report
+
+
+def render_text(report: dict) -> str:
+    """Human-readable audit, for a terminal or an Apprise body."""
+    lines = []
+    corpus = report["corpus"]
+    lines.append(
+        f"corpus: {corpus['documents']} document(s), {corpus['chunks']} chunk(s), "
+        f"{corpus['collection_points']} vector(s) in {corpus['collection']}"
+    )
+    lines.append(
+        f"horizon: last_verified within {report['horizon']['max_age_days']} days "
+        f"(warning at {report['horizon']['max_age_days'] - report['horizon']['warn_days']} days)"
+    )
+    findings = report["findings"]
+    for kind, label in (
+        ("stale", "STALE — past the verification horizon"),
+        ("due_soon", "DUE SOON — verify before it goes stale"),
+        ("index_drift", "INDEX DRIFT — the index does not match content/"),
+        ("contract_drift", "CONTRACT DRIFT — rebuild required"),
+    ):
+        rows = findings[kind]
+        if not rows:
+            continue
+        lines.append("")
+        lines.append(f"{label} ({len(rows)})")
+        for row in rows:
+            if "problem" in row:
+                lines.append(f"  - {row.get('source_path', '<collection>')}: {row['problem']}")
+            else:
+                lines.append(
+                    f"  - {row['source_path']} ({row['area']}, {row['confidence']}): "
+                    f"last_verified {row['last_verified']} ({row['age_days']} days ago)"
+                )
+    if report["ok"]:
+        lines.append("")
+        lines.append("OK: nothing stale, no drift." + (" (see warnings above)" if findings["due_soon"] else ""))
+    return "\n".join(lines)
+
+
+def alert(report: dict, apprise_url: str, timeout: float = 10.0) -> bool:
+    """POST the audit to the existing Apprise service. No-op without a URL."""
+    if not apprise_url:
+        return False
+    title = "ai-resume corpus audit: findings" if not report["ok"] else "ai-resume corpus audit: OK"
+    body = json.dumps({"title": title, "body": render_text(report), "type": "warning" if not report["ok"] else "info"})
+    request = urllib.request.Request(
+        apprise_url,
+        data=body.encode("utf-8"),
+        method="POST",
+        headers={"Content-Type": "application/json"},
+    )
+    with urllib.request.urlopen(request, timeout=timeout):
+        return True

corpus/loader.py

@@ -0,0 +1,162 @@
+"""Load `content/` into documents, reusing the Phase 1 validator's parser.
+
+There is exactly one front-matter dialect in this repository and exactly one
+parser for it: `tools/validate_content.py`. This module imports that parser
+rather than reimplementing it, so the corpus can never disagree with the gate
+about what a document says.
+
+The gate itself is `validate_tree` from the same file. The pipeline runs it
+before it chunks anything: an invalid content tree stops the run, it never
+reaches the vector store.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import os
+import sys
+from dataclasses import dataclass, field
+
+VALIDATOR_NAME = "validate_content"
+
+
+class ContentError(RuntimeError):
+    """The content tree is unreadable or invalid; the run must stop."""
+
+
+def load_validator(root: str):
+    """Import tools/validate_content.py as a module (cached per root)."""
+    cached = sys.modules.get(VALIDATOR_NAME)
+    if cached is not None and getattr(cached, "__validator_root__", None) == root:
+        return cached
+    path = os.path.join(root, "tools", "validate_content.py")
+    if not os.path.exists(path):
+        raise ContentError(f"validator not found at {path}")
+    spec = importlib.util.spec_from_file_location(VALIDATOR_NAME, path)
+    if spec is None or spec.loader is None:
+        raise ContentError(f"cannot load validator at {path}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    module.__validator_root__ = root  # type: ignore[attr-defined]
+    sys.modules[VALIDATOR_NAME] = module
+    return module
+
+
+def to_python(node):
+    """Convert a validator front-matter Node into plain Python values."""
+    if node is None or node.kind == "null":
+        return None
+    if node.kind == "scalar":
+        return node.value
+    if node.kind == "seq":
+        return [to_python(item) for item in node.value]
+    if node.kind == "map":
+        return {key: to_python(value) for key, value in node.value.items()}
+    raise ContentError(f"unexpected front-matter node kind {node.kind!r}")
+
+
+@dataclass
+class Document:
+    """One Markdown file plus the front-matter the schema guarantees."""
+
+    path: str  # repository-relative, posix, e.g. "content/edge/l3/edge-zigbee.md"
+    source_id: str
+    area: str
+    level: int
+    title: str
+    claim: str
+    confidence: str
+    period: str
+    tech: list
+    metrics: list
+    artefacts: list
+    public: bool
+    last_verified: str
+    tags: list
+    body: str
+    digest: str  # sha256 of the file bytes: the incremental-mode key
+    url: str  # canonical page URL, matching internal/content.PageURL
+
+    # Filled in by the pipeline as it goes; not part of the schema.
+    chunk_count: int = field(default=0)
+
+    @property
+    def artefact_urls(self) -> list:
+        return [str(a.get("url", "")) for a in self.artefacts if a.get("url")]
+
+
+def page_url(area: str, doc_id: str, level: int) -> str:
+    """Canonical URL for a document. Mirrors internal/content.PageURL exactly."""
+    if level == 1:
+        return f"/{area}/"
+    if level == 2:
+        return f"/{area}/{doc_id}/"
+    return f"/{area}/a/{doc_id}/"
+
+
+def digest_of(path: str) -> str:
+    """sha256 of the raw file bytes — the same digest for any platform."""
+    with open(path, "rb") as handle:
+        return hashlib.sha256(handle.read()).hexdigest()
+
+
+def load_documents(cfg) -> list:
+    """Load every document under content/, sorted by path.
+
+    Raises ContentError when the gate fails, so a broken content tree never
+    produces a half-correct index.
+    """
+    validator = load_validator(cfg.root)
+    violations, count = validator.validate_tree(cfg.root, cfg.content_dir, quiet=True)
+    if violations:
+        detail = "\n  ".join(violations[:20])
+        more = "" if len(violations) <= 20 else f"\n  … and {len(violations) - 20} more"
+        raise ContentError(
+            f"content gate failed with {len(violations)} violation(s) "
+            f"across {count} document(s). Run `make validate`:\n  {detail}{more}"
+        )
+
+    documents = []
+    for path in validator.iter_documents(cfg.content_dir):
+        documents.append(_load_one(cfg, validator, path))
+    documents.sort(key=lambda d: d.path)
+    if not documents:
+        raise ContentError(f"no documents found under {cfg.content_dir}")
+    return documents
+
+
+def _load_one(cfg, validator, path: str) -> Document:
+    rel = os.path.relpath(path, cfg.root).replace(os.sep, "/")
+    fm_lines, body, problem = validator.split_front_matter(path)
+    if problem:
+        raise ContentError(f"{rel}: {problem}")
+    try:
+        root_node = validator.parse_front_matter(fm_lines)
+    except validator.YamlError as exc:
+        raise ContentError(f"{rel}:{exc.line}: invalid YAML: {exc.msg}") from exc
+    fields = to_python(root_node)
+    if not isinstance(fields, dict):
+        raise ContentError(f"{rel}: front-matter must be a mapping")
+
+    level = int(fields["level"])
+    doc = Document(
+        path=rel,
+        source_id=str(fields["id"]),
+        area=str(fields["area"]),
+        level=level,
+        title=str(fields["title"]),
+        claim=str(fields["claim"]),
+        confidence=str(fields["confidence"]),
+        period=str(fields["period"]),
+        tech=list(fields.get("tech") or []),
+        metrics=list(fields.get("metrics") or []),
+        artefacts=list(fields.get("artefacts") or []),
+        public=bool(fields.get("public")),
+        last_verified=str(fields["last_verified"]),
+        tags=list(fields.get("tags") or []),
+        body=body,
+        digest=digest_of(path),
+        url=page_url(str(fields["area"]), str(fields["id"]), level),
+    )
+    return doc

corpus/measure.py

@@ -0,0 +1,242 @@
+"""Qdrant memory at 500 / 5,000 / 20,000 chunks.
+
+`docs/phase-4-corpus.md` requires the measurement and `reference/stack.md` is
+where the number is recorded. The corpus itself is small; the point of the
+measurement is to show the *shape* of the growth so the 400 m cap is a known
+quantity rather than a hope.
+
+Method:
+
+1. Create an empty collection and sample memory — the baseline.
+2. For each size: recreate the collection, upsert that many chunks with payloads
+   shaped exactly like production payloads (same fields, same text lengths),
+   wait for the write to settle, then sample.
+
+Memory is read from Qdrant's own Prometheus `app_process_resident_memory_bytes`,
+which is the process that would be capped, not the container sandbox. When
+`--container` is also given the container's counter is recorded next to it.
+
+The measurement collection is dropped afterwards unless `--keep` is passed, and
+it is never the production collection name.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import statistics
+import subprocess
+import time
+import uuid
+
+from . import chunker, embeddings, loader, payloads
+from .qdrant import Qdrant
+
+# Qdrant 1.20 exposes memory_resident_bytes; older builds used the Prometheus
+# default app_process_resident_memory_bytes. Accept either.
+RSS_METRIC = re.compile(
+    r"^(?:memory_resident_bytes|app_process_resident_memory_bytes)\s+([0-9.eE+]+)",
+    re.MULTILINE,
+)
+INGEST_BATCH = 250
+INGEST_SETTLE = 0.4
+MEMUSAGE = re.compile(r"([0-9.]+)\s*([KMG]i?B)", re.IGNORECASE)
+CAP_MIB = 400  # reference/stack.md memory budget for Qdrant
+
+
+def measure(cfg, sizes=(500, 5000, 20000), container: str = "", keep: bool = False, log=None) -> dict:
+    log = log or (lambda _msg: None)
+    collection = f"{cfg.collection}_measure"
+    provider = embeddings.build_provider(cfg)
+    qdrant = Qdrant(cfg.qdrant_url, timeout=max(cfg.timeout, 60.0))
+    if not qdrant.health():
+        raise RuntimeError(f"Qdrant is not answering at {cfg.qdrant_url}")
+
+    templates = _templates(cfg, provider)
+    log(f"measure: {len(templates)} payload template(s), dim={cfg.dim}, collection={collection}")
+
+    qdrant.drop_collection(collection)
+    _create(qdrant, cfg, collection)
+    time.sleep(1.0)
+    baseline = _sample(qdrant, container)
+
+    samples = []
+    for size in sorted(int(value) for value in sizes):
+        qdrant.drop_collection(collection)
+        _create(qdrant, cfg, collection)
+        points = _points(templates, size, provider)
+        started = time.time()
+        # Throttled ingest: the peak during a bulk write is what trips the
+        # cgroup limit, not the settled size. One batch at a time, sampling as we
+        # go, so the measurement reports the peak and not just the end state.
+        sent = 0
+        peak = baseline["rss_mib"]
+        for start in range(0, len(points), INGEST_BATCH):
+            sent += qdrant.upsert(collection, points[start : start + INGEST_BATCH], batch=INGEST_BATCH)
+            time.sleep(INGEST_SETTLE)
+            reading = _rss_mib(qdrant)
+            if reading is not None:
+                peak = max(peak, reading)
+        elapsed = round(time.time() - started, 2)
+        points_now = qdrant.point_count(collection)
+        time.sleep(2.0)
+        reading = _sample(qdrant, container)
+        growth = reading["rss_mib"] - baseline["rss_mib"]
+        samples.append(
+            {
+                "chunks_requested": size,
+                "points": points_now,
+                "sent": sent,
+                "upsert_seconds": elapsed,
+                "rss_mib": reading["rss_mib"],
+                "peak_rss_mib": round(peak, 1),
+                "container_mib": reading["container_mib"],
+                "growth_mib": round(growth, 1),
+                "bytes_per_chunk": int(growth * 1024 * 1024 / size) if size else None,
+            }
+        )
+        log(
+            f"measure: {size} chunk(s) -> {points_now} point(s), rss {reading['rss_mib']:.1f} MiB "
+            f"(+{growth:.1f} MiB), upsert {elapsed}s"
+        )
+
+    if not keep:
+        qdrant.drop_collection(collection)
+
+    report = {
+        "collection": collection,
+        "dim": cfg.dim,
+        "distance": "Cosine",
+        "provider": cfg.provider,
+        "model": cfg.model,
+        "qdrant_version": qdrant.version(),
+        "cap_mib": CAP_MIB,
+        "baseline": baseline,
+        "samples": samples,
+        "within_cap": all(
+            max(sample["rss_mib"], sample["peak_rss_mib"]) <= CAP_MIB for sample in samples
+        ),
+        "measured_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+        "kept": keep,
+    }
+    return report
+
+
+def _create(qdrant: Qdrant, cfg, collection: str) -> None:
+    """Same collection settings as production, so the memory reading means something."""
+    qdrant.create_collection(
+        collection, cfg.dim, vectors_on_disk=cfg.vectors_on_disk
+    )
+
+
+def _templates(cfg, provider) -> list:
+    """Real payloads from content/, so field count and text length are realistic."""
+    documents = loader.load_documents(cfg)
+    out = []
+    for doc in documents:
+        for chunk in chunker.chunk_document(doc.body):
+            payload = payloads.build_payload(doc, chunk, cfg)
+            payload["id"] = "template"
+            out.append(payload)
+    if not out:
+        raise RuntimeError("no content to build a measurement from")
+    return out
+
+
+def _points(templates: list, size: int, provider) -> list:
+    """`size` points with unique ids, unique source paths and real payload shape."""
+    points = []
+    texts = []
+    for index in range(size):
+        template = templates[index % len(templates)]
+        payload = dict(template)
+        payload["id"] = str(uuid.UUID(hex=hashlib.sha256(f"measure:{index}".encode()).hexdigest()[:32]))
+        payload["source_path"] = f"measure/{template['area']}/{index}.md"
+        payload["chunk_index"] = index
+        points.append({"id": payload["id"], "payload": payload})
+        texts.append(f"{index} {payload['text']}")
+    vectors = provider.embed(texts)
+    for point, vector in zip(points, vectors):
+        point["vector"] = vector
+    return points
+
+
+def _sample(qdrant: Qdrant, container: str) -> dict:
+    """Three readings two seconds apart; report the median so a GC pause is not the number."""
+    readings = []
+    for attempt in range(3):
+        readings.append(_rss_mib(qdrant))
+        if attempt < 2:
+            time.sleep(2.0)
+    readings = [value for value in readings if value is not None]
+    rss = round(statistics.median(readings), 1) if readings else 0.0
+    return {"rss_mib": rss, "container_mib": _container_mib(container) if container else None}
+
+
+def _rss_mib(qdrant: Qdrant):
+    text = qdrant.metrics()
+    if not text:
+        return None
+    match = RSS_METRIC.search(text)
+    if not match:
+        return None
+    return float(match.group(1)) / (1024 * 1024)
+
+
+def _container_mib(container: str):
+    """Container memory from `docker stats`. None when docker or the container is absent."""
+    try:
+        completed = subprocess.run(
+            ["docker", "stats", "--no-stream", "--format", "{{.MemUsage}}", container],
+            capture_output=True,
+            text=True,
+            timeout=20,
+            check=False,
+        )
+    except (OSError, subprocess.SubprocessError):
+        return None
+    if completed.returncode != 0:
+        return None
+    match = MEMUSAGE.search(completed.stdout)
+    if not match:
+        return None
+    value = float(match.group(1))
+    unit = match.group(2).upper()
+    if unit.startswith("G"):
+        value *= 1024
+    elif unit.startswith("K"):
+        value /= 1024
+    return round(value, 1)
+
+
+def render_text(report: dict) -> str:
+    lines = [
+        f"Qdrant {report['qdrant_version']} · collection {report['collection']} · "
+        f"dim {report['dim']} cosine · provider {report['provider']}/{report['model']}",
+        f"baseline (empty collection): {report['baseline']['rss_mib']:.1f} MiB",
+        "",
+        f"{'chunks':>8}  {'points':>8}  {'rss MiB':>8}  {'peak':>7}  {'growth':>7}  {'B/chunk':>8}  {'upsert s':>8}",
+    ]
+    for sample in report["samples"]:
+        per_chunk = sample["bytes_per_chunk"] or 0
+        lines.append(
+            f"{sample['chunks_requested']:>8}  {sample['points']:>8}  {sample['rss_mib']:>8.1f}  "
+            f"{sample['peak_rss_mib']:>7.1f}  {sample['growth_mib']:>+7.1f}  {per_chunk:>8}  "
+            f"{sample['upsert_seconds']:>8}"
+        )
+    lines.append("")
+    verdict = "within" if report["within_cap"] else "OVER"
+    lines.append(f"cap {report['cap_mib']} MiB: {verdict} cap at every measured size")
+    if report["kept"]:
+        lines.append(f"collection {report['collection']} was kept for inspection")
+    return "\n".join(lines)
+
+
+if __name__ == "__main__":
+    import sys
+
+    from .config import Config
+
+    cfg = Config.from_env(dim=int(sys.argv[1]) if len(sys.argv) > 1 else None)
+    print(render_text(measure(cfg, container="", log=print)))

corpus/payloads.py

@@ -0,0 +1,155 @@
+"""Chunk payloads: the thing the retrieval filter runs against.
+
+`docs/phase-4-corpus.md` §4.2 fixes the payload. `area` and `confidence` are the
+two that matter: they let the query path narrow *before* it searches, and they
+let an answer qualify unfinished work instead of presenting it as delivered.
+
+**A missing or invalid field fails the run.** Nothing in this module supplies a
+default value: `validate_payload` raises on the first field that is absent,
+empty or out of enum, and the pipeline reports which document and chunk it was.
+That is the DoD item "a missing field fails the run rather than defaulting".
+"""
+
+from __future__ import annotations
+
+import datetime
+
+AREAS = ("ai", "platform", "edge", "products", "history", "about")
+LEVELS = (1, 2, 3)
+SOURCE_TYPES = ("doc", "code", "adr", "metric")
+CONFIDENCE = ("delivered", "running", "in-progress", "abandoned", "historic")
+
+# Fields every payload must carry (docs/phase-4-corpus.md §4.2), plus the ones
+# the index needs to answer "where did this come from?" without a second lookup.
+REQUIRED_FIELDS = (
+    "area",
+    "level",
+    "source_type",
+    "confidence",
+    "period",
+    "artefacts",
+    "last_verified",
+    "text",
+)
+PROVENANCE_FIELDS = (
+    "id",
+    "source_path",
+    "source_id",
+    "chunk_index",
+    "title",
+    "claim",
+    "url",
+    "content_hash",
+    "embedding_model",
+    "embedding_dim",
+    "chunker_version",
+)
+
+
+class PayloadError(RuntimeError):
+    """A payload is incomplete or out of enum. The run must stop."""
+
+
+def source_type_for(doc) -> str:
+    """Classify a document. Everything in content/ is a doc unless it says so.
+
+    The schema has no `source_type` field and the schema is frozen, so the
+    classification is derived from tags the schema does allow.
+    """
+    tags = {str(tag).lower() for tag in doc.tags}
+    if "adr" in tags:
+        return "adr"
+    if "metric" in tags:
+        return "metric"
+    if "code" in tags:
+        return "code"
+    return "doc"
+
+
+def build_payload(doc, chunk, cfg) -> dict:
+    """Build one chunk payload from a document and a chunk.
+
+    Values are copied, never guessed: if the document does not carry a field,
+    the payload does not gain one.
+    """
+    payload = {
+        # -- the retrieval filter (docs/phase-4-corpus.md §4.2) ------------
+        "area": doc.area,
+        "level": doc.level,
+        "source_type": source_type_for(doc),
+        "confidence": doc.confidence,
+        "period": doc.period,
+        "artefacts": [str(a.get("url", "")) for a in doc.artefacts],
+        "last_verified": doc.last_verified,
+        "text": chunk.text,
+        # -- provenance ----------------------------------------------------
+        "id": None,  # filled by the pipeline once the id is computed
+        "source_path": doc.path,
+        "source_id": doc.source_id,
+        "chunk_index": chunk.index,
+        "title": doc.title,
+        "claim": doc.claim,
+        "url": doc.url,
+        "content_hash": doc.digest,
+        "embedding_model": cfg.model,
+        "embedding_dim": cfg.dim,
+        "chunker_version": cfg.chunker_version,
+        # -- supporting evidence, used by the answer to cite ---------------
+        "public": bool(doc.public),
+        "tech": list(doc.tech),
+        "tags": list(doc.tags),
+        "chunk_tokens": chunk.tokens,
+    }
+    return payload
+
+
+def validate_payload(payload: dict, where: str = "chunk") -> None:
+    """Raise PayloadError if a required field is missing, empty or out of enum."""
+    for field in REQUIRED_FIELDS + PROVENANCE_FIELDS:
+        if field == "id" and payload.get("id") is None:
+            raise PayloadError(f"{where}: payload field 'id' was not set")
+        if field not in payload:
+            raise PayloadError(f"{where}: payload is missing required field {field!r}")
+
+    area = payload["area"]
+    if area not in AREAS:
+        raise PayloadError(f"{where}: area {area!r} is not one of {list(AREAS)}")
+    if payload["level"] not in LEVELS:
+        raise PayloadError(f"{where}: level {payload['level']!r} is not one of {list(LEVELS)}")
+    if payload["source_type"] not in SOURCE_TYPES:
+        raise PayloadError(
+            f"{where}: source_type {payload['source_type']!r} is not one of {list(SOURCE_TYPES)}"
+        )
+    if payload["confidence"] not in CONFIDENCE:
+        raise PayloadError(
+            f"{where}: confidence {payload['confidence']!r} is not one of {list(CONFIDENCE)}"
+        )
+
+    for field in ("period", "title", "claim", "url", "source_path", "source_id", "embedding_model"):
+        value = payload[field]
+        if not isinstance(value, str) or not value.strip():
+            raise PayloadError(f"{where}: field {field!r} must be a non-empty string")
+
+    text = payload["text"]
+    if not isinstance(text, str) or not text.strip():
+        raise PayloadError(f"{where}: field 'text' must be non-empty")
+
+    if not isinstance(payload["artefacts"], list):
+        raise PayloadError(f"{where}: field 'artefacts' must be a list (empty is allowed)")
+
+    _check_date(payload["last_verified"], where)
+    if not isinstance(payload["embedding_dim"], int) or payload["embedding_dim"] <= 0:
+        raise PayloadError(f"{where}: field 'embedding_dim' must be a positive integer")
+    if not isinstance(payload["chunk_index"], int) or payload["chunk_index"] < 0:
+        raise PayloadError(f"{where}: field 'chunk_index' must be a non-negative integer")
+    if not isinstance(payload["content_hash"], str) or len(payload["content_hash"]) != 64:
+        raise PayloadError(f"{where}: field 'content_hash' must be a sha256 hex digest")
+
+
+def _check_date(value: str, where: str) -> None:
+    try:
+        datetime.date.fromisoformat(str(value))
+    except ValueError as exc:
+        raise PayloadError(
+            f"{where}: field 'last_verified' must be an ISO date (YYYY-MM-DD), got {value!r}"
+        ) from exc

corpus/pipeline.py

@@ -0,0 +1,426 @@
+"""The pipeline: content/ -> validate -> chunk -> enrich -> embed -> upsert.
+
+One entry point, `run`, with two modes:
+
+* **rebuild** — drop the collection, empty the store, embed everything. This is
+  the "one command rebuilds the corpus from an empty volume" path.
+* **sync** — the default. Only documents whose content hash, embedding model,
+  embedding dimension or chunker version changed are re-chunked and
+  re-embedded. Everything else is left alone.
+
+The order matters. Payloads are built and validated for *every* changed document
+before a single vector is written, so an invalid payload stops the run with an
+empty diff rather than leaving half an index behind.
+
+Deterministic ids (`sha256(source_path + chunk_index + content_hash)`) mean
+re-running updates points instead of duplicating them, and the id is a valid
+UUID so Qdrant accepts it as a point id without a mapping table.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import time
+import uuid
+from dataclasses import asdict, dataclass, field
+
+from . import chunker, embeddings, loader, payloads, store as store_module
+from .config import CHUNKER_VERSION
+from .qdrant import Qdrant, QdrantError
+
+
+class PipelineError(RuntimeError):
+    """The run cannot continue without corrupting the index."""
+
+
+def chunk_id(source_path: str, chunk_index: int, content_hash: str) -> str:
+    """Deterministic point id: sha256(source_path + chunk_index + content_hash).
+
+    Rendered as a UUID because Qdrant only accepts unsigned integers or UUIDs as
+    point ids; the first 128 bits of the digest keep collisions improbable.
+    """
+    digest = hashlib.sha256(
+        f"{source_path}\x00{chunk_index}\x00{content_hash}".encode("utf-8")
+    ).hexdigest()
+    return str(uuid.UUID(hex=digest[:32]))
+
+
+@dataclass
+class RunStats:
+    mode: str = "sync"
+    documents: int = 0
+    added: int = 0
+    updated: int = 0
+    unchanged: int = 0
+    removed: int = 0
+    chunks_written: int = 0
+    chunks_deleted: int = 0
+    embed_calls: int = 0
+    texts_embedded: int = 0
+    points_sent: int = 0
+    collection_points: int = 0
+    store_chunks: int = 0
+    duration_s: float = 0.0
+    collection: str = ""
+    provider: str = ""
+    model: str = ""
+    dim: int = 0
+    chunker_version: str = CHUNKER_VERSION
+    notes: list = field(default_factory=list)
+
+    def summary(self) -> str:
+        rows = [
+            f"mode={self.mode} documents={self.documents} chunks={self.store_chunks}",
+            f"documents added={self.added} updated={self.updated} unchanged={self.unchanged} removed={self.removed}",
+            f"chunks written={self.chunks_written} deleted={self.chunks_deleted}",
+            f"embedding calls={self.embed_calls} texts={self.texts_embedded}",
+            f"points sent={self.points_sent} collection points={self.collection_points}",
+            f"provider={self.provider} model={self.model} dim={self.dim} chunker={self.chunker_version}",
+            f"collection={self.collection} duration={self.duration_s:.2f}s",
+        ]
+        rows.extend(self.notes)
+        return "\n".join(rows)
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+
+def run(cfg, mode: str = "sync", log=None) -> RunStats:
+    """Build or refresh the corpus. Returns what happened; raises on any error."""
+    if mode not in ("sync", "rebuild"):
+        raise PipelineError(f"unknown mode {mode!r}; expected 'sync' or 'rebuild'")
+    started = time.time()
+    log = log or (lambda _msg: None)
+    stats = RunStats(
+        mode=mode,
+        collection=cfg.collection,
+        provider=cfg.provider,
+        model=cfg.model,
+        dim=cfg.dim,
+        chunker_version=cfg.chunker_version,
+    )
+
+    documents = loader.load_documents(cfg)
+    stats.documents = len(documents)
+    log(f"validated {len(documents)} document(s)")
+
+    provider = embeddings.build_provider(cfg)
+    qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout)
+    if not qdrant.health():
+        raise PipelineError(
+            f"Qdrant is not answering at {cfg.qdrant_url}. Start it "
+            "(deploy/qdrant.yml) or point RESUME_QDRANT_URL at a running instance."
+        )
+
+    with store_module.Store(cfg.db_path) as store:
+        contract = _contract(cfg)
+        mode = _check_contract(store, qdrant, cfg, contract, mode)
+
+        _ensure_collection(qdrant, cfg)
+
+        manifest = store.manifest()
+        changed, removed = _classify(documents, manifest, cfg, mode)
+        stats.unchanged = len(documents) - len(changed)
+        stats.removed = len(removed)
+        log(
+            f"incremental plan: {len(changed)} to embed, "
+            f"{stats.unchanged} unchanged, {len(removed)} removed"
+        )
+
+        for path in removed:
+            stats.chunks_deleted += store.delete_document(path)
+            qdrant.delete_by_source(cfg.collection, path)
+            log(f"removed {path}")
+
+        for doc in changed:
+            payloads_for_doc = _prepare(doc, cfg, manifest)
+            stats.chunks_written += len(payloads_for_doc)
+
+            vectors = provider.embed([p["text"] for p in payloads_for_doc])
+            if len(vectors) != len(payloads_for_doc):
+                raise PipelineError(
+                    f"{doc.path}: provider returned {len(vectors)} vectors for "
+                    f"{len(payloads_for_doc)} chunks"
+                )
+
+            points = []
+            for payload, vector in zip(payloads_for_doc, vectors):
+                if len(vector) != cfg.dim:
+                    raise PipelineError(
+                        f"{doc.path}: vector dimension {len(vector)} != configured {cfg.dim}"
+                    )
+                points.append({"id": payload["id"], "vector": vector, "payload": payload})
+
+            # Order: remove what this document used to have (chunk counts change),
+            # then write the new rows and points. A crash between the two leaves an
+            # index that reports as incomplete, never one that claims to be whole.
+            stats.chunks_deleted += store.delete_document(doc.path)
+            qdrant.delete_by_source(cfg.collection, doc.path)
+            stats.points_sent += qdrant.upsert(cfg.collection, points)
+            store.write_document(doc, payloads_for_doc)
+
+            if doc.path in manifest:
+                stats.updated += 1
+                log(f"updated {doc.path} ({len(points)} chunk(s))")
+            else:
+                stats.added += 1
+                log(f"added {doc.path} ({len(points)} chunk(s))")
+
+        provider_stats = provider.stats()
+        stats.embed_calls = provider_stats["calls"]
+        stats.texts_embedded = provider_stats["texts"]
+        stats.store_chunks = store.chunk_count()
+        stats.collection_points = qdrant.point_count(cfg.collection)
+
+        if stats.collection_points != stats.store_chunks:
+            raise PipelineError(
+                "index is inconsistent after the run: "
+                f"{stats.collection_points} point(s) in Qdrant but "
+                f"{stats.store_chunks} chunk(s) in SQLite. Run `python3 -m corpus.cli verify`."
+            )
+
+        store.set_meta("collection", contract)
+        store.set_meta("last_run", {**stats.to_dict(), "finished_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())})
+        _write_collection_file(cfg, store, stats)
+
+    stats.duration_s = round(time.time() - started, 3)
+    return stats
+
+
+def _contract(cfg) -> dict:
+    """What this collection *is*. Recorded so a change is a migration, not drift."""
+    return {
+        "provider": cfg.provider,
+        "model": cfg.model,
+        "dim": cfg.dim,
+        "chunker_version": cfg.chunker_version,
+        "distance": "Cosine",
+    }
+
+
+def _check_contract(store, qdrant, cfg, contract: dict, mode: str) -> str:
+    stored = store.get_meta("collection")
+    if mode == "rebuild":
+        qdrant.drop_collection(cfg.collection)
+        store.reset()
+        return mode
+    if stored is None:
+        # No contract yet: either a fresh store, or one whose database was lost.
+        if qdrant.collection_exists(cfg.collection) and qdrant.point_count(cfg.collection) > 0:
+            raise PipelineError(
+                f"collection {cfg.collection!r} holds points but this store has no "
+                "collection contract (was data/corpus/corpus.db deleted?). "
+                "Confirm you want to overwrite it: python3 -m corpus.cli rebuild"
+            )
+        return mode
+    if stored != contract:
+        raise PipelineError(
+            "the collection contract does not match this configuration — refusing to "
+            "mix vectors in one collection (docs/phase-4-corpus.md §4.3).\n"
+            f"  stored: {json.dumps(stored, sort_keys=True)}\n"
+            f"  config: {json.dumps(contract, sort_keys=True)}\n"
+            "A provider or chunker change is a governed migration: run "
+            "`python3 -m corpus.cli rebuild` once Sam has approved the change."
+        )
+    return mode
+
+
+def _ensure_collection(qdrant: Qdrant, cfg) -> None:
+    if not qdrant.collection_exists(cfg.collection):
+        qdrant.create_collection(
+            cfg.collection, cfg.dim, vectors_on_disk=cfg.vectors_on_disk
+        )
+        return
+    info = qdrant.collection_info(cfg.collection)
+    vectors = ((info.get("config") or {}).get("params") or {}).get("vectors")
+    size = None
+    if isinstance(vectors, dict):
+        size = vectors.get("size")
+    if size is not None and int(size) != int(cfg.dim):
+        raise PipelineError(
+            f"collection {cfg.collection!r} has vectors of size {size} but this run "
+            f"is configured for {cfg.dim}. That is a governed migration: rebuild."
+        )
+
+
+def _classify(documents: list, manifest: dict, cfg, mode: str):
+    """Which documents need work, and which are gone."""
+    changed = []
+    on_disk = set()
+    for doc in documents:
+        on_disk.add(doc.path)
+        if mode == "rebuild":
+            changed.append(doc)
+            continue
+        entry = manifest.get(doc.path)
+        if entry is None:
+            changed.append(doc)
+            continue
+        if (
+            entry["content_hash"] != doc.digest
+            or entry["embedding_model"] != cfg.model
+            or int(entry["embedding_dim"]) != int(cfg.dim)
+            or entry["chunker_version"] != cfg.chunker_version
+        ):
+            changed.append(doc)
+    removed = sorted(path for path in manifest if path not in on_disk)
+    return changed, removed
+
+
+def _prepare(doc, cfg, manifest: dict) -> list:
+    """Chunk one document and build validated payloads. No network, no writes."""
+    try:
+        chunks = chunker.chunk_document(doc.body)
+    except chunker.ChunkError as exc:
+        raise PipelineError(f"{doc.path}: {exc}") from exc
+    if not chunks:
+        raise PipelineError(f"{doc.path}: produced no chunks (empty body?)")
+    doc.chunk_count = len(chunks)
+
+    out = []
+    for chunk in chunks:
+        payload = payloads.build_payload(doc, chunk, cfg)
+        payload["id"] = chunk_id(doc.path, chunk.index, doc.digest)
+        try:
+            payloads.validate_payload(payload, where=f"{doc.path}#chunk-{chunk.index}")
+        except payloads.PayloadError as exc:
+            raise PipelineError(str(exc)) from exc
+        out.append(payload)
+    return out
+
+
+def _write_collection_file(cfg, store, stats: RunStats) -> None:
+    """Mirror the contract to data/corpus/collection.json for the operator."""
+    body = {
+        "collection": cfg.collection,
+        "contract": _contract(cfg),
+        "qdrant_url": cfg.qdrant_url,
+        "chunks": store.chunk_count(),
+        "sources": store.source_count(),
+        "last_run": store.get_meta("last_run"),
+    }
+    with open(cfg.meta_path, "w", encoding="utf-8") as handle:
+        json.dump(body, handle, indent=2, sort_keys=True)
+        handle.write("\n")
+
+
+# -- verification and demonstration ---------------------------------------
+
+
+def verify(cfg, log=None) -> list:
+    """Return a list of problems. Empty means the index agrees with content/."""
+    log = log or (lambda _msg: None)
+    problems = []
+    documents = loader.load_documents(cfg)
+    qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout)
+    if not qdrant.health():
+        return [f"Qdrant is not answering at {cfg.qdrant_url}"]
+
+    contract = _contract(cfg)
+    with store_module.Store(cfg.db_path) as store:
+        stored = store.get_meta("collection")
+        if stored != contract:
+            problems.append(f"collection contract mismatch: stored={stored} config={contract}")
+
+        if not qdrant.collection_exists(cfg.collection):
+            problems.append(f"collection {cfg.collection!r} does not exist")
+            return problems
+
+        info = qdrant.collection_info(cfg.collection)
+        size = (((info.get("config") or {}).get("params") or {}).get("vectors") or {}).get("size")
+        if size is not None and int(size) != cfg.dim:
+            problems.append(f"collection vector size {size} != configured {cfg.dim}")
+
+        points_by_source = qdrant.scroll_sources(cfg.collection)
+        rows = store.chunk_payloads()
+        by_source = {}
+        for row in rows:
+            by_source.setdefault(row["source_path"], []).append(row)
+            try:
+                payloads.validate_payload(row, where=f"{row['source_path']}#chunk-{row['chunk_index']}")
+            except payloads.PayloadError as exc:
+                problems.append(str(exc))
+
+        for doc in documents:
+            payload_rows = by_source.get(doc.path, [])
+            if not payload_rows:
+                problems.append(f"{doc.path}: no chunks in the index (run sync)")
+                continue
+            if payload_rows[0]["content_hash"] != doc.digest:
+                problems.append(f"{doc.path}: source changed since it was embedded")
+            expected = len(payload_rows)
+            actual = points_by_source.get(doc.path, 0)
+            if actual != expected:
+                problems.append(
+                    f"{doc.path}: {expected} chunk row(s) but {actual} vector(s) in Qdrant"
+                )
+
+        for source in sorted(set(by_source) - {doc.path for doc in documents}):
+            problems.append(f"{source}: indexed but no longer in content/ (run sync)")
+
+        total_points = sum(points_by_source.values())
+        if total_points != store.chunk_count():
+            problems.append(
+                f"{total_points} vector(s) in Qdrant but {store.chunk_count()} chunk(s) in SQLite"
+            )
+        log(
+            f"verify: {store.chunk_count()} chunk(s) across {store.source_count()} source(s), "
+            f"{total_points} vector(s) in Qdrant"
+        )
+    return problems
+
+
+def demo_area_filter(cfg, query: str, area: str = None, limit: int = 5) -> dict:
+    """Same query with and without an `area` filter — the DoD's filter evidence."""
+    provider = embeddings.build_provider(cfg)
+    qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout)
+    vector = provider.embed([query])[0]
+    unfiltered = qdrant.search(cfg.collection, vector, limit=limit)
+    filtered = qdrant.search(cfg.collection, vector, limit=limit, area=area) if area else []
+    return {
+        "query": query,
+        "area": area,
+        "unfiltered": [_hit(row) for row in unfiltered],
+        "filtered": [_hit(row) for row in filtered],
+    }
+
+
+def _hit(row: dict) -> dict:
+    payload = row.get("payload") or {}
+    return {
+        "score": round(row["score"], 4),
+        "area": payload.get("area"),
+        "confidence": payload.get("confidence"),
+        "title": payload.get("title"),
+        "url": payload.get("url"),
+        "source_path": payload.get("source_path"),
+        "chunk_index": payload.get("chunk_index"),
+        "preview": (payload.get("text") or "").replace("\n", " ")[:120],
+    }
+
+
+def hybrid_search(cfg, query: str, area: str = None, limit: int = 5) -> list:
+    """Vectors + FTS5 fused with reciprocal rank fusion (decision D10).
+
+    Phase 5 owns the query graph; this is here because Phase 4 is the phase that
+    has to prove the two halves line up — same ids, same metadata, one ranking.
+    """
+    provider = embeddings.build_provider(cfg)
+    qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout)
+    vector = provider.embed([query])[0]
+    ranked = {}
+    for rank, hit in enumerate(qdrant.search(cfg.collection, vector, limit=limit * 4, area=area)):
+        entry = ranked.setdefault(hit["id"], {"id": hit["id"], "payload": hit["payload"], "score": 0.0})
+        entry["score"] += 1.0 / (60 + rank)
+    with store_module.Store(cfg.db_path) as store:
+        for rank, hit in enumerate(store.keyword_search(query, area=area, limit=limit * 4)):
+            entry = ranked.setdefault(hit["id"], {"id": hit["id"], "payload": {}, "score": 0.0})
+            entry.setdefault("keyword", hit["snippet"])
+            entry["score"] += 1.0 / (60 + rank)
+    ordered = sorted(ranked.values(), key=lambda row: row["score"], reverse=True)[:limit]
+    for row in ordered:
+        payload = row.get("payload") or {}
+        row["title"] = payload.get("title")
+        row["url"] = payload.get("url")
+    return ordered

corpus/qdrant.py

@@ -0,0 +1,269 @@
+"""Qdrant REST client — standard library only.
+
+No `qdrant-client` dependency: the four calls this pipeline needs are HTTP
+requests, and AGENTS.md §4.1 freezes the stack. `_request` is the single
+transport hook, which is what the tests replace to run the whole pipeline
+without a container.
+
+Qdrant is bound to `127.0.0.1:6333` on the runtime host (`.13`) and is never
+publicly reachable; Caddy is the only public listener.
+"""
+
+from __future__ import annotations
+
+import http.client
+import json
+import time
+import urllib.error
+import urllib.request
+
+DISTANCE_COSINE = "Cosine"
+
+# The runtime host (.13) is memory-bound: 2.3 GiB free and 8 GiB of swap already
+# in use, so Qdrant gets a 400 m cap and must live inside it
+# (reference/stack.md). Measured on 2026-10-26: the *default* HNSW and payload
+# settings are OOM-killed at the cap between 5,000 and 20,000 chunks of 1024
+# dimensions. Keeping the HNSW graph and the payloads on disk, and using a
+# smaller graph (m=8, ef_construct=64), fits the cap because the kernel can
+# reclaim those page-cache pages instead of killing the process. Vectors stay in
+# RAM because they are the part search reads on every query.
+#
+# The trade is build time and a little search latency; the corpus is ~50 chunks
+# today and the public site has no query burst. See docs/phase-4-notes.md §7.
+MEMORY_FRUGAL_DEFAULTS = {
+    "on_disk_payload": True,
+    "hnsw_m": 8,
+    "hnsw_ef_construct": 64,
+    "vectors_on_disk": False,
+}
+
+# Segment behaviour under a cap. `memmap_threshold` makes Qdrant mmap segments
+# this size and larger (page cache the kernel can reclaim instead of OOM), and
+# `indexing_threshold` is raised so a bulk ingest does not build the HNSW graph
+# while it is still writing segments. Measured: 20,000 chunks at 1024
+# dimensions peaked at 296 MiB during ingest and settled to 96 MiB.
+MEMORY_FRUGAL_OPTIMIZERS = {
+    "default_segment_number": 2,
+    "memmap_threshold": 5000,
+    "indexing_threshold": 20000,
+}
+
+
+class QdrantError(RuntimeError):
+    """A Qdrant call failed, or Qdrant is not reachable."""
+
+
+class Qdrant:
+    def __init__(self, url: str = "http://127.0.0.1:6333", timeout: float = 30.0, attempts: int = 3) -> None:
+        self.url = url.rstrip("/")
+        self.timeout = timeout
+        self.attempts = max(1, attempts)
+
+    # -- transport --------------------------------------------------------
+
+    def _request(self, method: str, path: str, body=None) -> dict:
+        """One HTTP call with retry/backoff. The only place that touches the network."""
+        data = None if body is None else json.dumps(body).encode("utf-8")
+        last_error = None
+        for attempt in range(self.attempts):
+            request = urllib.request.Request(
+                f"{self.url}{path}",
+                data=data,
+                method=method,
+                headers={"Content-Type": "application/json"} if data else {},
+            )
+            try:
+                with urllib.request.urlopen(request, timeout=self.timeout) as response:
+                    raw = response.read().decode("utf-8")
+                return json.loads(raw) if raw.strip() else {}
+            except urllib.error.HTTPError as exc:
+                detail = exc.read().decode("utf-8", "replace")[:400]
+                exc.close()
+                if exc.code == 404:
+                    raise QdrantNotFound(f"{method} {path}: {detail}") from None
+                if exc.code != 429 and exc.code < 500:
+                    raise QdrantError(f"{method} {path} failed: HTTP {exc.code} {detail}") from None
+                last_error = f"HTTP {exc.code} {detail}"
+            except (urllib.error.URLError, TimeoutError, http.client.HTTPException, ConnectionError) as exc:
+                # RemoteDisconnected and friends arrive here when the server dies
+                # mid-request (an OOM kill looks exactly like this).
+                last_error = f"cannot reach {self.url}: {getattr(exc, 'reason', exc)}"
+            if attempt + 1 < self.attempts:
+                time.sleep(0.5 * (2 ** attempt))
+        raise QdrantError(f"{method} {path} failed after {self.attempts} attempts: {last_error}")
+
+    # -- cluster ----------------------------------------------------------
+
+    def health(self) -> bool:
+        """True when Qdrant answers. Never raises."""
+        try:
+            self._request("GET", "/")
+            return True
+        except (QdrantError, QdrantNotFound):
+            return False
+
+    def version(self) -> str:
+        response = self._request("GET", "/")
+        return str(response.get("version", "unknown"))
+
+    # -- collections ------------------------------------------------------
+
+    def collection_exists(self, name: str) -> bool:
+        try:
+            self._request("GET", f"/collections/{name}")
+            return True
+        except QdrantNotFound:
+            return False
+
+    def create_collection(
+        self,
+        name: str,
+        dim: int,
+        distance: str = DISTANCE_COSINE,
+        on_disk_payload: bool = None,
+        hnsw_m: int = None,
+        hnsw_ef_construct: int = None,
+        vectors_on_disk: bool = None,
+    ) -> None:
+        """Create a collection sized for a memory-bound host (see MEMORY_FRUGAL_DEFAULTS)."""
+        defaults = MEMORY_FRUGAL_DEFAULTS
+        body = {
+            "vectors": {
+                "size": int(dim),
+                "distance": distance,
+                "on_disk": bool(
+                    defaults["vectors_on_disk"] if vectors_on_disk is None else vectors_on_disk
+                ),
+            },
+            "hnsw_config": {
+                "m": int(defaults["hnsw_m"] if hnsw_m is None else hnsw_m),
+                "ef_construct": int(
+                    defaults["hnsw_ef_construct"] if hnsw_ef_construct is None else hnsw_ef_construct
+                ),
+                "on_disk": True,
+            },
+            "on_disk_payload": bool(
+                defaults["on_disk_payload"] if on_disk_payload is None else on_disk_payload
+            ),
+            "optimizers_config": dict(MEMORY_FRUGAL_OPTIMIZERS),
+        }
+        self._request("PUT", f"/collections/{name}", body)
+
+    def drop_collection(self, name: str) -> None:
+        try:
+            self._request("DELETE", f"/collections/{name}")
+        except QdrantNotFound:
+            pass
+
+    def collection_info(self, name: str) -> dict:
+        response = self._request("GET", f"/collections/{name}")
+        return dict(response.get("result") or {})
+
+    def point_count(self, name: str) -> int:
+        response = self._request("POST", f"/collections/{name}/points/count", {"exact": True})
+        return int((response.get("result") or {}).get("count", 0))
+
+    # -- points -----------------------------------------------------------
+
+    def upsert(self, name: str, points: list, batch: int = 256) -> int:
+        """Upsert points in batches. Returns the number of points sent."""
+        sent = 0
+        for start in range(0, len(points), max(1, batch)):
+            chunk = points[start : start + max(1, batch)]
+            self._request(
+                "PUT",
+                f"/collections/{name}/points?wait=true",
+                {"points": chunk},
+            )
+            sent += len(chunk)
+        return sent
+
+    def delete_by_source(self, name: str, source_path: str) -> None:
+        """Delete every point that came from one document.
+
+        Re-chunking a document can produce fewer chunks than before, so the old
+        points are removed rather than overwritten. Deterministic ids make
+        duplicates impossible; this makes orphans impossible.
+        """
+        self._request(
+            "POST",
+            f"/collections/{name}/points/delete?wait=true",
+            {"filter": {"must": [{"key": "source_path", "match": {"value": source_path}}]}},
+        )
+
+    def search(
+        self,
+        name: str,
+        vector: list,
+        limit: int = 5,
+        area: str = None,
+        confidence: str = None,
+        with_payload: bool = True,
+    ) -> list:
+        """Nearest neighbours, optionally filtered *before* the search."""
+        body = {
+            "query": [float(value) for value in vector],
+            "limit": int(limit),
+            "with_payload": bool(with_payload),
+        }
+        filters = _filter_conditions(area, confidence)
+        if filters:
+            body["filter"] = {"must": filters}
+        try:
+            response = self._request("POST", f"/collections/{name}/points/query", body)
+            points = ((response.get("result") or {}).get("points")) or []
+        except QdrantNotFound:
+            # Older Qdrant exposes the same operation as /points/search.
+            response = self._request("POST", f"/collections/{name}/points/search", body)
+            points = response.get("result") or []
+        return [
+            {
+                "id": point.get("id"),
+                "score": float(point.get("score") or 0.0),
+                "payload": point.get("payload") or {},
+            }
+            for point in points
+        ]
+
+    def scroll_sources(self, name: str, limit: int = 1000) -> dict:
+        """Map source_path -> number of vectors, for consistency checks."""
+        counts = {}
+        offset = None
+        while True:
+            body = {"limit": limit, "with_payload": ["source_path"], "with_vector": False}
+            if offset is not None:
+                body["offset"] = offset
+            response = self._request("POST", f"/collections/{name}/points/scroll", body)
+            result = response.get("result") or {}
+            points = result.get("points") or []
+            for point in points:
+                source = (point.get("payload") or {}).get("source_path", "<missing>")
+                counts[source] = counts.get(source, 0) + 1
+            offset = result.get("next_page_offset")
+            if not points or offset is None:
+                return counts
+
+    def metrics(self) -> str:
+        """Prometheus text from Qdrant's /metrics route (used by the memory measurement).
+
+        Returns "" when the route is not exposed, so callers can fall back to
+        reading the container's own memory counters.
+        """
+        try:
+            with urllib.request.urlopen(f"{self.url}/metrics", timeout=self.timeout) as response:
+                return response.read().decode("utf-8", "replace")
+        except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError):
+            return ""
+
+
+class QdrantNotFound(QdrantError):
+    """A 404 from Qdrant: the collection or route does not exist."""
+
+
+def _filter_conditions(area: str = None, confidence: str = None) -> list:
+    conditions = []
+    if area:
+        conditions.append({"key": "area", "match": {"value": area}})
+    if confidence:
+        conditions.append({"key": "confidence", "match": {"value": confidence}})
+    return conditions

corpus/store.py

@@ -0,0 +1,295 @@
+"""SQLite: the keyword index and the incremental state.
+
+Two jobs, one file (`RESUME_CORPUS_DB`, default `data/corpus/corpus.db`):
+
+* **`chunk_fts`** — an FTS5 index over chunk text with the same filter columns
+  the vectors carry. This is the keyword half of hybrid retrieval (decision
+  D10). Phase 2 indexes *pages* for the site's own search box; this indexes
+  *chunks*, because retrieval hands the model a passage, not a page.
+* **`manifest`** — one row per source document: its content hash, its chunk
+  count and the embedding model/dimension it was embedded with. Incremental mode
+  is a hash comparison against this table, and nothing else.
+
+`meta` records the collection contract (provider, model, dimension, chunker
+version) so a provider change is detected instead of silently corrupting the
+index.
+
+Standard library `sqlite3` only. The same file is opened read-only by the Phase
+5 retrieval service.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import sqlite3
+import time
+import unicodedata
+
+SCHEMA_VERSION = "1"
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS chunks (
+    id               TEXT PRIMARY KEY,
+    source_path      TEXT NOT NULL,
+    source_id        TEXT NOT NULL,
+    chunk_index      INTEGER NOT NULL,
+    area             TEXT NOT NULL,
+    level            INTEGER NOT NULL,
+    source_type      TEXT NOT NULL,
+    confidence       TEXT NOT NULL,
+    period           TEXT NOT NULL,
+    last_verified    TEXT NOT NULL,
+    title            TEXT NOT NULL,
+    claim            TEXT NOT NULL,
+    url              TEXT NOT NULL,
+    content_hash     TEXT NOT NULL,
+    embedding_model  TEXT NOT NULL,
+    embedding_dim    INTEGER NOT NULL,
+    chunker_version  TEXT NOT NULL,
+    chunk_tokens     INTEGER NOT NULL,
+    public           INTEGER NOT NULL,
+    artefacts        TEXT NOT NULL,
+    tech             TEXT NOT NULL,
+    tags             TEXT NOT NULL,
+    text             TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS chunks_source ON chunks(source_path);
+CREATE INDEX IF NOT EXISTS chunks_area ON chunks(area);
+
+CREATE VIRTUAL TABLE IF NOT EXISTS chunk_fts USING fts5(
+    id UNINDEXED,
+    source_path UNINDEXED,
+    title,
+    terms,
+    text,
+    area UNINDEXED,
+    confidence UNINDEXED,
+    tokenize = 'porter unicode61 remove_diacritics 2'
+);
+
+CREATE TABLE IF NOT EXISTS manifest (
+    source_path      TEXT PRIMARY KEY,
+    content_hash     TEXT NOT NULL,
+    chunk_count      INTEGER NOT NULL,
+    embedding_model  TEXT NOT NULL,
+    embedding_dim    INTEGER NOT NULL,
+    chunker_version  TEXT NOT NULL,
+    last_verified    TEXT NOT NULL,
+    updated_at       TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS meta (
+    key   TEXT PRIMARY KEY,
+    value TEXT NOT NULL
+);
+"""
+
+# Column order used by every INSERT in this module.
+_CHUNK_COLUMNS = (
+    "id", "source_path", "source_id", "chunk_index", "area", "level", "source_type",
+    "confidence", "period", "last_verified", "title", "claim", "url", "content_hash",
+    "embedding_model", "embedding_dim", "chunker_version", "chunk_tokens",
+    "public", "artefacts", "tech", "tags", "text",
+)
+
+_JSON_FIELDS = ("artefacts", "tech", "tags")
+
+
+class StoreError(RuntimeError):
+    """The corpus database is unreadable or inconsistent."""
+
+
+class Store:
+    def __init__(self, path: str) -> None:
+        self.path = path
+        os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True)
+        self.db = sqlite3.connect(path)
+        self.db.row_factory = sqlite3.Row
+        self.db.execute("PRAGMA journal_mode = WAL")
+        self.db.execute("PRAGMA synchronous = NORMAL")
+        self.db.executescript(SCHEMA)
+        self.set_meta("schema_version", SCHEMA_VERSION)
+
+    # -- lifecycle --------------------------------------------------------
+
+    def close(self) -> None:
+        self.db.close()
+
+    def __enter__(self) -> "Store":
+        return self
+
+    def __exit__(self, *_exc) -> None:
+        self.close()
+
+    def reset(self) -> None:
+        """Empty every table. Used by a from-scratch rebuild."""
+        for table in ("chunk_fts", "chunks", "manifest", "meta"):
+            self.db.execute(f"DELETE FROM {table}")
+        self.db.commit()
+
+    # -- writing ----------------------------------------------------------
+
+    def write_document(self, doc, payloads: list) -> None:
+        """Replace all rows for one document: chunks, FTS rows and manifest."""
+        self.delete_document(doc.path, keep_manifest=True)
+        for payload in payloads:
+            values = []
+            for column in _CHUNK_COLUMNS:
+                value = payload[column]
+                if column in _JSON_FIELDS:
+                    value = json.dumps(value, sort_keys=True)
+                values.append(value)
+            cursor = self.db.execute(
+                f"INSERT INTO chunks ({', '.join(_CHUNK_COLUMNS)}) "
+                f"VALUES ({', '.join('?' for _ in _CHUNK_COLUMNS)})",
+                values,
+            )
+            # `terms` is the keyword surface: the words a visitor or a recruiter
+            # would type that live in front-matter rather than the body — tech
+            # names, tags, the one-line claim. Without it, a search for "zigbee"
+            # misses a page whose body says "mesh". It is indexed, never retrieved.
+            terms = " ".join(
+                str(value)
+                for value in (
+                    payload["claim"],
+                    payload["title"],
+                    " ".join(payload["tech"]),
+                    " ".join(payload["tags"]),
+                )
+                if value
+            )
+            self.db.execute(
+                "INSERT INTO chunk_fts (rowid, id, source_path, title, terms, text, area, confidence) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+                (
+                    cursor.lastrowid,
+                    payload["id"],
+                    payload["source_path"],
+                    payload["title"],
+                    terms,
+                    payload["text"],
+                    payload["area"],
+                    payload["confidence"],
+                ),
+            )
+        self.db.execute(
+            "INSERT OR REPLACE INTO manifest "
+            "(source_path, content_hash, chunk_count, embedding_model, embedding_dim, "
+            " chunker_version, last_verified, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                doc.path,
+                doc.digest,
+                len(payloads),
+                payloads[0]["embedding_model"] if payloads else "",
+                payloads[0]["embedding_dim"] if payloads else 0,
+                payloads[0]["chunker_version"] if payloads else "",
+                doc.last_verified,
+                time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+            ),
+        )
+        self.db.commit()
+
+    def delete_document(self, source_path: str, keep_manifest: bool = False) -> int:
+        rowids = [
+            row["rowid"]
+            for row in self.db.execute("SELECT rowid FROM chunks WHERE source_path = ?", (source_path,))
+        ]
+        for rowid in rowids:
+            self.db.execute("DELETE FROM chunk_fts WHERE rowid = ?", (rowid,))
+        self.db.execute("DELETE FROM chunks WHERE source_path = ?", (source_path,))
+        if not keep_manifest:
+            self.db.execute("DELETE FROM manifest WHERE source_path = ?", (source_path,))
+        self.db.commit()
+        return len(rowids)
+
+    # -- reading ----------------------------------------------------------
+
+    def manifest(self) -> dict:
+        return {
+            row["source_path"]: dict(row)
+            for row in self.db.execute("SELECT * FROM manifest")
+        }
+
+    def chunk_count(self) -> int:
+        return int(self.db.execute("SELECT count(*) FROM chunks").fetchone()[0])
+
+    def source_count(self) -> int:
+        return int(self.db.execute("SELECT count(DISTINCT source_path) FROM chunks").fetchone()[0])
+
+    def chunk_ids(self) -> list:
+        return [row["id"] for row in self.db.execute("SELECT id FROM chunks ORDER BY id")]
+
+    def chunk_payloads(self, source_path: str = None) -> list:
+        """Every stored payload, optionally for one document. Used by verify()."""
+        sql = "SELECT * FROM chunks"
+        args = ()
+        if source_path:
+            sql += " WHERE source_path = ?"
+            args = (source_path,)
+        sql += " ORDER BY source_path, chunk_index"
+        out = []
+        for row in self.db.execute(sql, args):
+            payload = dict(row)
+            for field in _JSON_FIELDS:
+                payload[field] = json.loads(payload[field])
+            out.append(payload)
+        return out
+
+    def keyword_search(self, query: str, area: str = None, confidence: str = None, limit: int = 10) -> list:
+        """BM25 over chunk text, optionally filtered before ranking."""
+        match = match_expression(query)
+        if not match:
+            return []
+        sql = [
+            "SELECT c.id, c.url, c.title, c.area, c.confidence, c.source_path,",
+            "       c.chunk_index, snippet(chunk_fts, 4, '<mark>', '</mark>', '…', 18) AS snippet,",
+            "       bm25(chunk_fts) AS score",
+            "FROM chunk_fts JOIN chunks c ON c.rowid = chunk_fts.rowid",
+            "WHERE chunk_fts MATCH ?",
+        ]
+        args = [match]
+        if area:
+            sql.append("AND c.area = ?")
+            args.append(area)
+        if confidence:
+            sql.append("AND c.confidence = ?")
+            args.append(confidence)
+        sql.append("ORDER BY bm25(chunk_fts) LIMIT ?")
+        args.append(int(limit))
+        rows = self.db.execute(" ".join(sql), args)
+        return [dict(row) for row in rows]
+
+    # -- meta -------------------------------------------------------------
+
+    def set_meta(self, key: str, value) -> None:
+        self.db.execute(
+            "INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)",
+            (key, json.dumps(value, sort_keys=True)),
+        )
+        self.db.commit()
+
+    def get_meta(self, key: str, default=None):
+        row = self.db.execute("SELECT value FROM meta WHERE key = ?", (key,)).fetchone()
+        if row is None:
+            return default
+        return json.loads(row["value"])
+
+    def meta(self) -> dict:
+        return {row["key"]: json.loads(row["value"]) for row in self.db.execute("SELECT key, value FROM meta")}
+
+
+def match_expression(query: str) -> str:
+    """Free text -> a safe FTS5 expression: bare words, each a prefix, all required.
+
+    Mirrors `MatchExpression` in internal/search/search.go so the site's search
+    box and the retrieval path treat a visitor's words identically. FTS5
+    operators in the input are never honoured.
+    """
+    terms = []
+    for raw in re.split(r"[^0-9a-zA-Z\-]+", unicodedata.normalize("NFKC", query or "").lower()):
+        term = raw.strip("-")
+        if term:
+            terms.append('"' + term.replace('"', "") + '"*')
+    return " AND ".join(terms)

corpus/tests/fake_qdrant.py

@@ -0,0 +1,112 @@
+"""An in-memory stand-in for Qdrant, used by the pipeline tests.
+
+It implements the routes the pipeline actually calls and nothing else, and it is
+deliberately dumb: brute-force cosine over a dict. If the pipeline passes
+against this, the pipeline's own logic is right; the real server is exercised
+separately in `docs/phase-4-notes.md` (rebuild against a live container).
+"""
+
+from __future__ import annotations
+
+import math
+
+from corpus.qdrant import Qdrant, QdrantError, QdrantNotFound
+
+
+class FakeQdrant(Qdrant):
+    def __init__(self, dim: int):
+        super().__init__(url="fake://qdrant", timeout=1.0, attempts=1)
+        self.dim = dim
+        self.collections = {}  # name -> {id: {"vector": [...], "payload": {...}}}
+        self.calls = []
+
+    # -- transport --------------------------------------------------------
+
+    def _request(self, method: str, path: str, body=None) -> dict:
+        self.calls.append((method, path, body))
+        query = ""
+        if "?" in path:
+            path, query = path.split("?", 1)
+        parts = [part for part in path.split("/") if part]
+
+        if not parts:
+            return {"version": "fake-1.0"}
+
+        if parts[0] != "collections":
+            raise QdrantNotFound(f"no route {path}")
+        if len(parts) == 2:
+            name = parts[1]
+            if method == "PUT":
+                self.collections.setdefault(name, {})
+                return {"result": True}
+            if method == "DELETE":
+                if name not in self.collections:
+                    raise QdrantNotFound("collection not found")
+                del self.collections[name]
+                return {"result": True}
+            if name not in self.collections:
+                raise QdrantNotFound("collection not found")
+            return {
+                "result": {
+                    "points_count": len(self.collections[name]),
+                    "config": {"params": {"vectors": {"size": self.dim, "distance": "Cosine"}}},
+                }
+            }
+        name = parts[1]
+        if name not in self.collections:
+            raise QdrantNotFound("collection not found")
+        points = self.collections[name]
+
+        if parts[2:4] == ["points", "count"]:
+            return {"result": {"count": len(points)}}
+        if len(parts) == 3 and parts[2] == "points" and method == "PUT":
+            for point in (body or {}).get("points", []):
+                if len(point["vector"]) != self.dim:
+                    raise QdrantError("wrong vector dimension")
+                points[point["id"]] = {"vector": point["vector"], "payload": point["payload"]}
+            return {"result": {"status": "completed"}}
+        if parts[2:4] == ["points", "delete"]:
+            must = (((body or {}).get("filter") or {}).get("must")) or []
+            for point_id in list(points):
+                if _matches(points[point_id]["payload"], must):
+                    del points[point_id]
+            return {"result": {"status": "completed"}}
+        if parts[2:4] == ["points", "query"]:
+            vector = body["query"]
+            must = (((body or {}).get("filter") or {}).get("must")) or []
+            scored = []
+            for point_id, point in points.items():
+                if not _matches(point["payload"], must):
+                    continue
+                scored.append(
+                    {
+                        "id": point_id,
+                        "score": _cosine(vector, point["vector"]),
+                        "payload": point["payload"],
+                    }
+                )
+            scored.sort(key=lambda row: row["score"], reverse=True)
+            return {"result": {"points": scored[: int(body.get("limit", 10))]}}
+        if parts[2:4] == ["points", "scroll"]:
+            limit = int((body or {}).get("limit", 100))
+            rows = [
+                {"id": point_id, "payload": {"source_path": point["payload"].get("source_path")}}
+                for point_id, point in list(points.items())[:limit]
+            ]
+            return {"result": {"points": rows, "next_page_offset": None}}
+        raise QdrantNotFound(f"no route {path}")
+
+
+def _matches(payload: dict, conditions: list) -> bool:
+    for condition in conditions:
+        key = condition.get("key")
+        expected = (condition.get("match") or {}).get("value")
+        if payload.get(key) != expected:
+            return False
+    return True
+
+
+def _cosine(left: list, right: list) -> float:
+    dot = sum(a * b for a, b in zip(left, right))
+    norm = math.sqrt(sum(a * a for a in left)) * math.sqrt(sum(b * b for b in right))
+    return dot / norm if norm else 0.0

corpus/tests/support.py

@@ -0,0 +1,186 @@
+"""Shared fixtures for the corpus tests: a throwaway content tree and a fake Qdrant."""
+
+from __future__ import annotations
+
+import contextlib
+import os
+import shutil
+import tempfile
+from unittest import mock
+
+from corpus.config import Config
+from corpus.tests.fake_qdrant import FakeQdrant
+
+REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+DOC = """---
+id: {id}
+area: {area}
+level: {level}
+title: "{title}"
+claim: "{claim}"
+confidence: {confidence}
+period: "2023–"
+tech: [Go, SQLite]
+metrics: [{{label: endpoints, value: 8}}]
+artefacts: []
+public: true
+last_verified: {last_verified}
+tags: [{tags}]
+---
+
+# {title}
+
+{body}
+"""
+
+FENCE = """A paragraph before the code block explains the constraint.
+
+```go
+package main
+
+func main() {{
+	println("chunk boundary must never land inside this fence")
+}}
+```
+
+A paragraph after the code block.
+"""
+
+TABLE = """Measured results, one row per endpoint:
+
+| endpoint | latency | loss |
+|---|---|---|
+| alpha | 12ms | 0.0% |
+| beta | 14ms | 0.1% |
+| gamma | 9ms | 0.0% |
+
+And a closing paragraph.
+"""
+
+PROSE = "\n\n".join(
+    f"Paragraph {index} describes a delivery constraint in corporate language, "
+    "with enough words that the chunker has to think about where to break, and "
+    "never inside a fenced block, a table or a front-matter block."
+    for index in range(1, 40)
+)
+
+DEFAULT_DOCS = {
+    "ai/ai-overview.md": dict(
+        id="ai-overview", area="ai", level=1, title="Agent Systems",
+        claim="Harnesses, memory and routing.", confidence="running",
+        last_verified="2026-10-01", tags="agents, llm", body=PROSE,
+    ),
+    "edge/l3/edge-zigbee.md": dict(
+        id="edge-zigbee", area="edge", level=3, title="Low Power Mesh Sensing",
+        claim="A constrained mesh that survives power loss.", confidence="delivered",
+        last_verified="2026-09-15", tags="zigbee, mesh", body=FENCE,
+    ),
+    "platform/l3/platform-containers.md": dict(
+        id="platform-containers", area="platform", level=3, title="Container Estate",
+        claim="Fifty containers with measured resource limits.", confidence="in-progress",
+        last_verified="2026-08-01", tags="docker", body=TABLE,
+    ),
+}
+
+
+def temp_repo(docs: dict = None) -> str:
+    """Create a throwaway repository root with tools/ and content/."""
+    root = tempfile.mkdtemp(prefix="corpus-test-")
+    os.makedirs(os.path.join(root, "tools"), exist_ok=True)
+    shutil.copy(
+        os.path.join(REPO_ROOT, "tools", "validate_content.py"),
+        os.path.join(root, "tools", "validate_content.py"),
+    )
+    write_docs(root, docs or DEFAULT_DOCS)
+    return root
+
+
+def write_docs(root: str, docs: dict) -> None:
+    for rel, fields in docs.items():
+        path = os.path.join(root, "content", rel)
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "w", encoding="utf-8") as handle:
+            handle.write(DOC.format(**fields))
+
+
+def write_doc(root: str, rel: str, **fields) -> None:
+    path = os.path.join(root, "content", rel)
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "w", encoding="utf-8") as handle:
+        handle.write(DOC.format(**fields))
+    # A fresh interpreter's import cache is fine; the loader re-reads the file.
+
+
+def make_config(root: str, **overrides) -> Config:
+    values = dict(
+        provider="local-hashing",
+        model="local-hashing-v1",
+        dim=32,
+        qdrant_url="fake://qdrant",
+        collection="test_corpus",
+    )
+    values.update(overrides)
+    return Config.from_env(root=root, **values)
+
+
+@contextlib.contextmanager
+def fake_qdrant(cfg, fake: FakeQdrant = None):
+    """Point every module that builds a Qdrant client at one in-memory fake."""
+    fake = fake or FakeQdrant(cfg.dim)
+    with mock.patch("corpus.pipeline.Qdrant", return_value=fake), mock.patch(
+        "corpus.freshness.Qdrant", return_value=fake
+    ):
+        yield fake
+
+
+class StubServer:
+    """A throwaway HTTP server on a loopback ephemeral port, for provider tests.
+
+    `responder(method, path, body)` returns `(status, payload)`; every request is
+    recorded with its parsed body so a test can assert the wire shape without a
+    real provider or a real Qdrant.
+    """
+
+    def __init__(self, responder):
+        import json
+        import threading
+        from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+        self.requests = []
+        self._responder = responder
+        outer = self
+
+        class Handler(BaseHTTPRequestHandler):
+            def _handle(self):
+                length = int(self.headers.get("Content-Length") or 0)
+                raw = self.rfile.read(length).decode("utf-8") if length else ""
+                body = json.loads(raw) if raw.strip() else None
+                outer.requests.append(
+                    {"method": self.command, "path": self.path, "body": body,
+                     "headers": dict(self.headers)}
+                )
+                status, payload = outer._responder(self.command, self.path, body)
+                data = payload.encode("utf-8") if isinstance(payload, str) else json.dumps(payload).encode("utf-8")
+                self.send_response(status)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(data)))
+                self.end_headers()
+                self.wfile.write(data)
+
+            do_GET = do_POST = do_PUT = do_DELETE = _handle
+
+            def log_message(self, *_args):
+                pass
+
+        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        self.url = f"http://127.0.0.1:{self.server.server_address[1]}"
+        self._thread = threading.Thread(target=self.server.serve_forever, daemon=True)
+
+    def __enter__(self):
+        self._thread.start()
+        return self
+
+    def __exit__(self, *_exc):
+        self.server.shutdown()
+        self.server.server_close()

corpus/tests/test_chunker.py

@@ -0,0 +1,145 @@
+"""Chunking rules: size, overlap, determinism, and the hard one — no split inside
+a fence, a table or front-matter (docs/phase-4-corpus.md DoD item 3)."""
+
+from __future__ import annotations
+
+import os
+import unittest
+
+from corpus import chunker, loader
+from corpus.config import Config
+from corpus.tests import support
+
+REPO_ROOT = support.REPO_ROOT
+
+
+def content_documents():
+    cfg = Config.from_env(root=REPO_ROOT, provider="local-hashing", dim=32)
+    return loader.load_documents(cfg)
+
+
+class BlockParsingTest(unittest.TestCase):
+    def test_fence_is_one_block(self):
+        blocks = chunker.parse_blocks(support.FENCE)
+        kinds = [block.kind for block in blocks]
+        self.assertEqual(kinds, ["para", "fence", "para"])
+        self.assertIn("chunk boundary must never land", blocks[1].text)
+
+    def test_table_is_one_block(self):
+        blocks = chunker.parse_blocks(support.TABLE)
+        table = [block for block in blocks if block.kind == "table"]
+        self.assertEqual(len(table), 1)
+        self.assertEqual(len(table[0].text.splitlines()), 5)  # header, rule, 3 rows
+
+    def test_front_matter_in_body_is_refused(self):
+        with self.assertRaises(chunker.ChunkError):
+            chunker.parse_blocks("---\nid: x\n---\n\n# Heading\n")
+
+    def test_unterminated_fence_runs_to_end(self):
+        blocks = chunker.parse_blocks("intro\n\n```go\nfunc main() {}\n")
+        self.assertEqual(blocks[-1].kind, "fence")
+        self.assertTrue(blocks[-1].text.rstrip().endswith("}"))
+
+    def test_headings_start_blocks(self):
+        blocks = chunker.parse_blocks("# One\n\ntext\n\n## Two\n\nmore\n")
+        self.assertEqual([block.kind for block in blocks], ["heading", "para", "heading", "para"])
+
+    def test_list_items_are_separate_blocks(self):
+        blocks = chunker.parse_blocks("- one\n- two\n  continued\n- three\n")
+        self.assertEqual([block.kind for block in blocks], ["list", "list", "list"])
+        self.assertIn("continued", blocks[1].text)
+
+
+class ChunkingTest(unittest.TestCase):
+    def test_small_body_is_one_chunk(self):
+        chunks = chunker.chunk_document("# Title\n\nA short body.\n")
+        self.assertEqual(len(chunks), 1)
+        self.assertEqual(chunks[0].index, 0)
+        self.assertFalse(chunks[0].oversize)
+
+    def test_long_body_respects_maximum_and_overlaps(self):
+        chunks = chunker.chunk_document(support.PROSE)
+        self.assertGreater(len(chunks), 1)
+        for chunk in chunks:
+            self.assertLessEqual(chunk.tokens, chunker.TOKEN_MAX, f"chunk {chunk.index} too large")
+        # Overlap: consecutive chunks share at least one block.
+        for previous, current in zip(chunks, chunks[1:]):
+            self.assertTrue(
+                set(id(block) for block in previous.blocks) & set(id(block) for block in current.blocks),
+                f"chunks {previous.index} and {current.index} do not overlap",
+            )
+
+    def test_chunk_indices_are_sequential(self):
+        chunks = chunker.chunk_document(support.PROSE)
+        self.assertEqual([chunk.index for chunk in chunks], list(range(len(chunks))))
+
+    def test_deterministic(self):
+        first = chunker.chunk_document(support.PROSE)
+        second = chunker.chunk_document(support.PROSE)
+        self.assertEqual([chunk.text for chunk in first], [chunk.text for chunk in second])
+
+    def test_code_fence_is_never_split(self):
+        body = support.PROSE + "\n\n" + support.FENCE + "\n\n" + support.PROSE
+        chunks = chunker.chunk_document(body)
+        fences = [block for block in chunker.parse_blocks(body) if block.kind == "fence"]
+        self.assertTrue(fences)
+        for fence in fences:
+            holders = [chunk for chunk in chunks if fence in chunk.blocks]
+            self.assertTrue(holders, "the fenced block disappeared from every chunk")
+            for holder in holders:
+                self.assertIn(fence.text, holder.text, "a chunk holds part of a fenced block")
+
+    def test_oversize_single_block_is_flagged_not_split(self):
+        huge = "```go\n" + "\n".join(f"line{i} := {i}" for i in range(1200)) + "\n```\n"
+        chunks = chunker.chunk_document(huge)
+        self.assertEqual(len(chunks), 1)
+        self.assertTrue(chunks[0].oversize)
+        self.assertGreater(chunks[0].tokens, chunker.TOKEN_MAX)
+
+
+class RealContentTest(unittest.TestCase):
+    """The same rules, over every document in content/."""
+
+    def test_no_boundary_splits_a_protected_region(self):
+        for doc in content_documents():
+            regions = chunker.protected_regions(doc.body)
+            chunks = chunker.chunk_document(doc.body)
+            chunker.verify_no_split(chunks, regions)  # raises on failure
+
+    def test_every_chunk_is_whole_blocks(self):
+        for doc in content_documents():
+            expected = [block.text for block in chunker.parse_blocks(doc.body)]
+            seen = []
+            for chunk in chunker.chunk_document(doc.body):
+                for block in chunk.blocks:
+                    self.assertIn(block.text, chunk.text)
+                    if not seen or seen[-1] != block.text:
+                        seen.append(block.text)
+            # Every block appears, in order, exactly as parsed (overlap repeats a
+            # block at the start of the next chunk, which is expected).
+            position = 0
+            for text in expected:
+                while position < len(seen) and seen[position] != text:
+                    position += 1
+                self.assertLess(position, len(seen), f"{doc.path}: block was dropped: {text[:40]!r}")
+                position += 1
+
+    def test_chunks_are_within_the_size_band(self):
+        for doc in content_documents():
+            for chunk in chunker.chunk_document(doc.body):
+                if chunk.oversize:
+                    self.assertEqual(len(chunk.blocks), 1)
+                    continue
+                self.assertLessEqual(chunk.tokens, chunker.TOKEN_MAX, f"{doc.path} chunk {chunk.index}")
+
+    def test_content_tree_is_chunked_completely(self):
+        total = 0
+        for doc in content_documents():
+            chunks = chunker.chunk_document(doc.body)
+            self.assertTrue(chunks, f"{doc.path}: no chunks")
+            total += len(chunks)
+        self.assertGreater(total, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()

corpus/tests/test_embeddings.py

@@ -0,0 +1,125 @@
+"""Providers: determinism offline, and an OpenAI-compatible wire shape through OmniRoute.
+
+There is no direct vendor SDK call anywhere (AGENTS.md §4.5), so the provider
+test asserts the HTTP request shape, not a client library's behaviour.
+"""
+
+from __future__ import annotations
+
+import math
+import unittest
+
+from corpus import embeddings
+from corpus.tests import support
+
+
+class LocalHashingTest(unittest.TestCase):
+    def test_deterministic_and_unit_length(self):
+        provider = embeddings.LocalHashingProvider(dim=64)
+        first = provider.embed(["zigbee mesh latency"])[0]
+        second = embeddings.LocalHashingProvider(dim=64).embed(["zigbee mesh latency"])[0]
+        self.assertEqual(first, second)
+        self.assertEqual(len(first), 64)
+        self.assertAlmostEqual(math.sqrt(sum(value * value for value in first)), 1.0, places=9)
+
+    def test_different_text_differs(self):
+        provider = embeddings.LocalHashingProvider(dim=64)
+        a, b = provider.embed(["zigbee", "lora"])
+        self.assertNotEqual(a, b)
+
+    def test_empty_text_still_returns_a_unit_vector(self):
+        provider = embeddings.LocalHashingProvider(dim=16)
+        vector = provider.embed([""])[0]
+        self.assertEqual(len(vector), 16)
+        self.assertAlmostEqual(math.sqrt(sum(value * value for value in vector)), 1.0, places=9)
+
+    def test_stats_count_calls_and_texts(self):
+        provider = embeddings.LocalHashingProvider(dim=16)
+        provider.embed(["a", "b", "c"])
+        provider.embed(["d"])
+        self.assertEqual(provider.stats(), {"calls": 2, "texts": 4})
+        provider.reset_stats()
+        self.assertEqual(provider.stats(), {"calls": 0, "texts": 0})
+
+    def test_rejects_a_zero_dimension(self):
+        with self.assertRaises(embeddings.EmbeddingError):
+            embeddings.LocalHashingProvider(dim=0)
+
+
+class OmniRouteTest(unittest.TestCase):
+    def responder_factory(self, dim=8, status_plan=None):
+        plan = list(status_plan or [])
+        state = {"n": 0}
+
+        def responder(method, path, body):
+            state["n"] += 1
+            if plan:
+                status = plan.pop(0)
+                if status != 200:
+                    return status, {"error": "planned failure"}
+            vectors = [
+                {"index": index, "embedding": [0.1] * dim}
+                for index, _text in enumerate(body.get("input") or [])
+            ]
+            return 200, {"model": body.get("model"), "data": vectors}
+
+        return responder
+
+    def test_posts_to_v1_embeddings_with_the_configured_model(self):
+        with support.StubServer(self.responder_factory()) as server:
+            provider = embeddings.OmniRouteProvider(
+                api_url=server.url, api_key="test-key", model="text-embedding-3-small", dim=8
+            )
+            vectors = provider.embed(["one", "two"])
+        self.assertEqual(len(vectors), 2)
+        request = server.requests[0]
+        self.assertEqual(request["method"], "POST")
+        self.assertEqual(request["path"], "/v1/embeddings")
+        self.assertEqual(request["body"]["model"], "text-embedding-3-small")
+        self.assertEqual(request["body"]["input"], ["one", "two"])
+        self.assertEqual(request["headers"]["Authorization"], "Bearer test-key")
+        self.assertEqual(provider.stats(), {"calls": 1, "texts": 2})
+
+    def test_batches_requests(self):
+        with support.StubServer(self.responder_factory()) as server:
+            provider = embeddings.OmniRouteProvider(
+                api_url=server.url, api_key="k", model="m", dim=8, batch_size=2
+            )
+            provider.embed(["a", "b", "c", "d", "e"])
+        self.assertEqual(len(server.requests), 3)
+        self.assertEqual(provider.stats(), {"calls": 3, "texts": 5})
+
+    def test_dimension_mismatch_is_refused(self):
+        with support.StubServer(self.responder_factory(dim=4)) as server:
+            provider = embeddings.OmniRouteProvider(
+                api_url=server.url, api_key="k", model="m", dim=8
+            )
+            with self.assertRaises(embeddings.EmbeddingError) as caught:
+                provider.embed(["one"])
+        self.assertIn("dimension", str(caught.exception))
+
+    def test_retries_then_succeeds(self):
+        with support.StubServer(self.responder_factory(status_plan=[429, 200])) as server:
+            provider = embeddings.OmniRouteProvider(
+                api_url=server.url, api_key="k", model="m", dim=8, backoff=0.01
+            )
+            vectors = provider.embed(["one"])
+        self.assertEqual(len(vectors), 1)
+        self.assertEqual(len(server.requests), 2)
+
+    def test_client_error_is_not_retried(self):
+        with support.StubServer(self.responder_factory(status_plan=[400])) as server:
+            provider = embeddings.OmniRouteProvider(
+                api_url=server.url, api_key="k", model="m", dim=8, backoff=0.01
+            )
+            with self.assertRaises(embeddings.EmbeddingError):
+                provider.embed(["one"])
+        self.assertEqual(len(server.requests), 1)
+
+    def test_missing_key_is_refused_before_any_call(self):
+        with self.assertRaises(embeddings.EmbeddingError):
+            embeddings.OmniRouteProvider(api_url="http://127.0.0.1:1", api_key="", model="m", dim=8)
+
+
+if __name__ == "__main__":
+    unittest.main()

corpus/tests/test_freshness.py

@@ -0,0 +1,110 @@
+"""Freshness audit: stale documents, index drift and contract drift."""
+
+from __future__ import annotations
+
+import datetime
+import unittest
+
+from corpus import freshness, pipeline, store as store_module
+from corpus.tests import support
+
+
+class AuditTest(unittest.TestCase):
+    def setUp(self):
+        self.root = support.temp_repo()
+        self.cfg = support.make_config(self.root)
+        self.addCleanup(_cleanup, self.root)
+        with support.fake_qdrant(self.cfg):
+            pipeline.run(self.cfg, mode="rebuild")
+
+    def audit(self, today=None, cfg=None):
+        with support.fake_qdrant(cfg or self.cfg):
+            return freshness.audit(cfg or self.cfg, today=today)
+
+    def test_clean_soon_after_rebuild(self):
+        report = self.audit(today=datetime.date(2026, 10, 5))
+        self.assertTrue(report["ok"], report["findings"])
+        self.assertEqual(report["corpus"]["documents"], 3)
+
+    def test_reports_stale_documents(self):
+        report = self.audit(today=datetime.date(2027, 10, 5))
+        stale = {row["source_path"] for row in report["findings"]["stale"]}
+        self.assertEqual(
+            stale,
+            {
+                "content/ai/ai-overview.md",
+                "content/edge/l3/edge-zigbee.md",
+                "content/platform/l3/platform-containers.md",
+            },
+        )
+        self.assertFalse(report["ok"])
+        text = freshness.render_text(report)
+        self.assertIn("STALE", text)
+        self.assertIn("platform-containers", text)
+
+    def test_warns_before_a_document_goes_stale(self):
+        # platform-containers was verified 2026-08-01. Horizon 180 days, warning
+        # 30, so on 2027-01-20 it is 172 days old: due soon, not yet stale.
+        report = self.audit(today=datetime.date(2027, 1, 20))
+        due = {row["source_path"] for row in report["findings"]["due_soon"]}
+        self.assertEqual(due, {"content/platform/l3/platform-containers.md"})
+        self.assertEqual(report["findings"]["stale"], [])
+        self.assertTrue(report["ok"])
+
+    def test_reports_index_drift_when_a_source_changes(self):
+        support.write_doc(
+            self.root,
+            "ai/ai-overview.md",
+            id="ai-overview", area="ai", level=1, title="Agent Systems",
+            claim="Edited but not re-indexed.", confidence="running",
+            last_verified="2026-10-05", tags="agents", body="New words for the corpus.",
+        )
+        report = self.audit(today=datetime.date(2026, 10, 6))
+        problems = [row["problem"] for row in report["findings"]["index_drift"]]
+        self.assertIn("source changed since it was embedded", problems)
+        self.assertFalse(report["ok"])
+
+    def test_reports_contract_drift_when_the_model_changes(self):
+        other = support.make_config(self.root, model="different-model")
+        report = self.audit(today=datetime.date(2026, 10, 5), cfg=other)
+        self.assertTrue(report["findings"]["contract_drift"])
+
+    def test_report_is_json_serialisable(self):
+        import json
+
+        report = self.audit(today=datetime.date(2026, 10, 5))
+        json.dumps(report)
+
+
+class AlertTest(unittest.TestCase):
+    def test_no_url_means_no_alert(self):
+        report = {"ok": True, "findings": {"stale": []}, "corpus": {}, "horizon": {}, "summary": {}}
+        self.assertFalse(freshness.alert(report, ""))
+
+    def test_posts_to_apprise(self):
+        report = {
+            "generated_at": "2026-10-26T00:00:00Z",
+            "today": "2026-10-26",
+            "horizon": {"max_age_days": 180, "warn_days": 30},
+            "corpus": {
+                "documents": 1, "chunks": 1, "sources": 1, "collection": "c",
+                "collection_points": 1, "provider": "p", "model": "m", "dim": 8,
+                "chunker_version": "v",
+            },
+            "findings": {"stale": [], "due_soon": [], "index_drift": [], "contract_drift": []},
+            "summary": {"stale": 0},
+            "ok": True,
+        }
+        with support.StubServer(lambda method, path, body: (200, "ok")) as server:
+            self.assertTrue(freshness.alert(report, server.url))
+        self.assertIn("title", server.requests[0]["body"])
+
+
+def _cleanup(root):
+    import shutil
+
+    shutil.rmtree(root, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    unittest.main()

corpus/tests/test_payloads.py

@@ -0,0 +1,100 @@
+"""Payload rules: every field present, enums enforced, missing field fails the run."""
+
+from __future__ import annotations
+
+import unittest
+
+from corpus import chunker, loader, payloads
+from corpus.config import Config
+from corpus.pipeline import chunk_id
+from corpus.tests import support
+
+
+def first_payload():
+    cfg = Config.from_env(root=support.REPO_ROOT, provider="local-hashing", dim=32)
+    doc = loader.load_documents(cfg)[0]
+    chunk = chunker.chunk_document(doc.body)[0]
+    payload = payloads.build_payload(doc, chunk, cfg)
+    payload["id"] = chunk_id(doc.path, chunk.index, doc.digest)
+    return cfg, doc, payload
+
+
+class BuildTest(unittest.TestCase):
+    def test_payload_has_every_required_field(self):
+        cfg, doc, payload = first_payload()
+        payloads.validate_payload(payload)
+        for field in payloads.REQUIRED_FIELDS:
+            self.assertIn(field, payload)
+        self.assertEqual(payload["area"], doc.area)
+        self.assertEqual(payload["confidence"], doc.confidence)
+        self.assertEqual(payload["last_verified"], doc.last_verified)
+        self.assertEqual(payload["embedding_model"], cfg.model)
+        self.assertEqual(payload["embedding_dim"], cfg.dim)
+
+    def test_every_real_document_produces_valid_payloads(self):
+        cfg = Config.from_env(root=support.REPO_ROOT, provider="local-hashing", dim=32)
+        for doc in loader.load_documents(cfg):
+            for chunk in chunker.chunk_document(doc.body):
+                payload = payloads.build_payload(doc, chunk, cfg)
+                payload["id"] = chunk_id(doc.path, chunk.index, doc.digest)
+                payloads.validate_payload(payload, where=doc.path)
+
+    def test_source_type_is_doc_by_default(self):
+        _cfg, doc, payload = first_payload()
+        self.assertEqual(payload["source_type"], "doc")
+
+    def test_source_type_from_tags(self):
+        doc = loader.Document(
+            path="content/ai/adr-x.md", source_id="adr-x", area="ai", level=3, title="ADR",
+            claim="c", confidence="delivered", period="2024", tech=[], metrics=[], artefacts=[],
+            public=True, last_verified="2026-01-01", tags=["adr"], body="", digest="0" * 64,
+            url="/ai/a/adr-x/",
+        )
+        self.assertEqual(payloads.source_type_for(doc), "adr")
+
+
+class ValidationTest(unittest.TestCase):
+    def test_missing_field_raises(self):
+        _cfg, _doc, payload = first_payload()
+        for field in ("area", "level", "source_type", "confidence", "period", "artefacts", "last_verified", "text"):
+            broken = dict(payload)
+            del broken[field]
+            with self.assertRaises(payloads.PayloadError, msg=f"missing {field} passed validation"):
+                payloads.validate_payload(broken)
+
+    def test_missing_provenance_field_raises(self):
+        _cfg, _doc, payload = first_payload()
+        for field in payloads.PROVENANCE_FIELDS:
+            broken = dict(payload)
+            del broken[field]
+            with self.assertRaises(payloads.PayloadError, msg=f"missing {field} passed validation"):
+                payloads.validate_payload(broken)
+
+    def test_bad_enums_raise(self):
+        _cfg, _doc, payload = first_payload()
+        for field, value in (
+            ("area", "sales"),
+            ("level", 4),
+            ("source_type", "blog"),
+            ("confidence", "vibes"),
+        ):
+            broken = dict(payload)
+            broken[field] = value
+            with self.assertRaises(payloads.PayloadError, msg=f"{field}={value!r} passed validation"):
+                payloads.validate_payload(broken)
+
+    def test_empty_text_raises(self):
+        _cfg, _doc, payload = first_payload()
+        payload["text"] = "   "
+        with self.assertRaises(payloads.PayloadError):
+            payloads.validate_payload(payload)
+
+    def test_bad_date_raises(self):
+        _cfg, _doc, payload = first_payload()
+        payload["last_verified"] = "last october"
+        with self.assertRaises(payloads.PayloadError):
+            payloads.validate_payload(payload)
+
+
+if __name__ == "__main__":
+    unittest.main()

corpus/tests/test_pipeline.py

@@ -0,0 +1,179 @@
+"""The pipeline: deterministic ids, no duplicate vectors on re-run, incremental mode,
+incomplete payloads never reach the store (docs/phase-4-corpus.md DoD 1-6)."""
+
+from __future__ import annotations
+
+import os
+import unittest
+from unittest import mock
+
+from corpus import chunker, loader, pipeline, store as store_module
+from corpus.tests import support
+
+
+class RebuildTest(unittest.TestCase):
+    def setUp(self):
+        self.root = support.temp_repo()
+        self.cfg = support.make_config(self.root)
+        self.addCleanup(_cleanup, self.root)
+        # One fake for the whole test: a second run must see the first run's points.
+        self.fake = support.FakeQdrant(self.cfg.dim)
+
+    def run_pipeline(self, mode="sync", cfg=None):
+        cfg = cfg or self.cfg
+        with support.fake_qdrant(cfg, self.fake):
+            return pipeline.run(cfg, mode=mode)
+
+    def test_rebuild_from_empty_volume(self):
+        stats = self.run_pipeline("rebuild")
+        self.assertEqual(stats.documents, 3)
+        self.assertEqual(stats.added, 3)
+        self.assertGreater(stats.store_chunks, 0)
+        self.assertEqual(stats.store_chunks, stats.collection_points)
+        self.assertEqual(stats.collection_points, len(self.fake.collections[self.cfg.collection]))
+        self.assertGreater(stats.embed_calls, 0)
+
+    def test_running_twice_does_not_duplicate_vectors(self):
+        first = self.run_pipeline("rebuild")
+        points_after_first = len(self.fake.collections[self.cfg.collection])
+        second = self.run_pipeline("sync")
+        points_after_second = len(self.fake.collections[self.cfg.collection])
+        self.assertEqual(points_after_first, points_after_second)
+        self.assertEqual(first.store_chunks, second.store_chunks)
+        self.assertEqual(second.added, 0)
+        self.assertEqual(second.updated, 0)
+        self.assertEqual(second.chunks_written, 0)
+
+    def test_incremental_reembeds_only_changed_documents(self):
+        self.run_pipeline("rebuild")
+
+        # Nothing changed: no embedding calls at all.
+        unchanged = self.run_pipeline("sync")
+        self.assertEqual(unchanged.embed_calls, 0)
+        self.assertEqual(unchanged.texts_embedded, 0)
+
+        # One document changes: exactly one embedding call, for that document.
+        support.write_doc(
+            self.root,
+            "edge/l3/edge-zigbee.md",
+            id="edge-zigbee", area="edge", level=3, title="Low Power Mesh Sensing",
+            claim="A constrained mesh that survives power loss.", confidence="delivered",
+            last_verified="2026-09-15", tags="zigbee, mesh",
+            body="The mesh now also reports battery voltage on every node.",
+        )
+        changed = self.run_pipeline("sync")
+        self.assertEqual(changed.added + changed.updated, 1)
+        self.assertEqual(changed.embed_calls, 1)
+        expected_chunks = len(
+            chunker.chunk_document(
+                loader.load_documents(support.make_config(self.root))[1].body
+            )
+        )
+        self.assertEqual(changed.texts_embedded, expected_chunks)
+        self.assertEqual(changed.unchanged, 2)
+
+    def test_removed_document_is_removed_from_both_indexes(self):
+        self.run_pipeline("rebuild")
+        before = self.run_pipeline("sync").store_chunks
+        os.remove(os.path.join(self.root, "content", "platform", "l3", "platform-containers.md"))
+        after = self.run_pipeline("sync")
+        self.assertEqual(after.removed, 1)
+        self.assertLess(after.store_chunks, before)
+        self.assertEqual(after.store_chunks, after.collection_points)
+        with store_module.Store(self.cfg.db_path) as store:
+            self.assertEqual(store.keyword_search("endpoint latency loss"), [])
+
+    def test_ids_are_deterministic(self):
+        first = pipeline.chunk_id("content/ai/ai-overview.md", 0, "a" * 64)
+        second = pipeline.chunk_id("content/ai/ai-overview.md", 0, "a" * 64)
+        other = pipeline.chunk_id("content/ai/ai-overview.md", 1, "a" * 64)
+        self.assertEqual(first, second)
+        self.assertNotEqual(first, other)
+        self.assertEqual(len(first), 36)
+
+    def test_invalid_payload_aborts_before_any_write(self):
+        real_build = pipeline.payloads.build_payload
+
+        def broken(doc, chunk, cfg):
+            payload = real_build(doc, chunk, cfg)
+            payload.pop("confidence")
+            return payload
+
+        with mock.patch("corpus.payloads.build_payload", side_effect=broken):
+            with self.assertRaises(pipeline.PipelineError):
+                self.run_pipeline("rebuild")
+        self.assertEqual(self.fake.collections.get(self.cfg.collection, {}), {})
+
+    def test_provider_change_is_refused_without_rebuild(self):
+        self.run_pipeline("rebuild")
+        other = support.make_config(self.root, model="local-hashing-v2")
+        with self.assertRaises(pipeline.PipelineError) as caught:
+            self.run_pipeline("sync", cfg=other)
+        self.assertIn("governed migration", str(caught.exception))
+
+    def test_contract_is_recorded(self):
+        self.run_pipeline("rebuild")
+        with store_module.Store(self.cfg.db_path) as store:
+            contract = store.get_meta("collection")
+        self.assertEqual(contract["model"], self.cfg.model)
+        self.assertEqual(contract["dim"], self.cfg.dim)
+        self.assertEqual(contract["chunker_version"], self.cfg.chunker_version)
+        self.assertTrue(os.path.exists(self.cfg.meta_path))
+
+
+class VerifyTest(unittest.TestCase):
+    def setUp(self):
+        self.root = support.temp_repo()
+        self.cfg = support.make_config(self.root)
+        self.addCleanup(_cleanup, self.root)
+
+    def test_clean_after_rebuild(self):
+        with support.fake_qdrant(self.cfg):
+            pipeline.run(self.cfg, mode="rebuild")
+            self.assertEqual(pipeline.verify(self.cfg), [])
+
+    def test_detects_source_changed_without_sync(self):
+        with support.fake_qdrant(self.cfg):
+            pipeline.run(self.cfg, mode="rebuild")
+            support.write_doc(
+                self.root,
+                "ai/ai-overview.md",
+                id="ai-overview", area="ai", level=1, title="Agent Systems",
+                claim="Changed.", confidence="running", last_verified="2026-10-02",
+                tags="agents", body="A different body entirely.",
+            )
+            problems = pipeline.verify(self.cfg)
+        self.assertTrue(any("changed since it was embedded" in problem for problem in problems), problems)
+
+    def test_detects_missing_document(self):
+        with support.fake_qdrant(self.cfg):
+            pipeline.run(self.cfg, mode="rebuild")
+            os.remove(os.path.join(self.root, "content", "ai", "ai-overview.md"))
+            problems = pipeline.verify(self.cfg)
+        self.assertTrue(any("no longer in content/" in problem for problem in problems), problems)
+
+
+class AreaFilterTest(unittest.TestCase):
+    def setUp(self):
+        self.root = support.temp_repo()
+        self.cfg = support.make_config(self.root)
+        self.addCleanup(_cleanup, self.root)
+
+    def test_filter_narrows_the_result_set(self):
+        with support.fake_qdrant(self.cfg):
+            pipeline.run(self.cfg, mode="rebuild")
+            result = pipeline.demo_area_filter(self.cfg, "container estate limits", area="edge")
+        self.assertTrue(result["unfiltered"])
+        self.assertTrue(result["filtered"])
+        self.assertEqual({row["area"] for row in result["filtered"]}, {"edge"})
+        self.assertGreater(len({row["area"] for row in result["unfiltered"]}), 1)
+
+
+def _cleanup(root):
+    import shutil
+
+    shutil.rmtree(root, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    unittest.main()

corpus/tests/test_qdrant_client.py

@@ -0,0 +1,104 @@
+"""Qdrant client: the wire shape the pipeline depends on, and the fallback route."""
+
+from __future__ import annotations
+
+import unittest
+
+from corpus import qdrant as qdrant_module
+from corpus.tests import support
+
+
+class ClientTest(unittest.TestCase):
+    def test_create_collection_sends_size_and_distance(self):
+        with support.StubServer(lambda method, path, body: (200, {"result": True})) as server:
+            client = qdrant_module.Qdrant(server.url, attempts=1)
+            client.create_collection("resume_corpus", 1536)
+        request = server.requests[0]
+        self.assertEqual(request["method"], "PUT")
+        self.assertEqual(request["path"], "/collections/resume_corpus")
+        body = request["body"]
+        self.assertEqual(body["vectors"]["size"], 1536)
+        self.assertEqual(body["vectors"]["distance"], "Cosine")
+        # A memory-bound host: payloads and the HNSW graph live on disk.
+        self.assertTrue(body["on_disk_payload"])
+        self.assertTrue(body["hnsw_config"]["on_disk"])
+        self.assertFalse(body["vectors"]["on_disk"])
+        self.assertGreater(body["optimizers_config"]["memmap_threshold"], 0)
+
+    def test_upsert_waits_and_batches(self):
+        with support.StubServer(lambda method, path, body: (200, {"result": {}})) as server:
+            client = qdrant_module.Qdrant(server.url, attempts=1)
+            points = [{"id": f"id-{index}", "vector": [0.0], "payload": {}} for index in range(5)]
+            sent = client.upsert("c", points, batch=2)
+        self.assertEqual(sent, 5)
+        self.assertEqual(len(server.requests), 3)
+        self.assertTrue(all("wait=true" in request["path"] for request in server.requests))
+        self.assertEqual(sum(len(r["body"]["points"]) for r in server.requests), 5)
+
+    def test_delete_by_source_sends_a_source_filter(self):
+        with support.StubServer(lambda method, path, body: (200, {"result": {}})) as server:
+            client = qdrant_module.Qdrant(server.url, attempts=1)
+            client.delete_by_source("c", "content/edge/l3/edge-zigbee.md")
+        request = server.requests[0]
+        self.assertEqual(request["method"], "POST")
+        self.assertIn("/collections/c/points/delete", request["path"])
+        self.assertEqual(
+            request["body"],
+            {"filter": {"must": [{"key": "source_path", "match": {"value": "content/edge/l3/edge-zigbee.md"}}]}},
+        )
+
+    def test_search_filters_before_searching(self):
+        responder = lambda method, path, body: (200, {"result": {"points": []}})  # noqa: E731
+        with support.StubServer(responder) as server:
+            client = qdrant_module.Qdrant(server.url, attempts=1)
+            client.search("c", [0.1, 0.2], limit=3, area="edge", confidence="delivered")
+        body = server.requests[0]["body"]
+        self.assertEqual(server.requests[0]["path"], "/collections/c/points/query")
+        self.assertEqual(body["limit"], 3)
+        self.assertEqual(
+            body["filter"],
+            {
+                "must": [
+                    {"key": "area", "match": {"value": "edge"}},
+                    {"key": "confidence", "match": {"value": "delivered"}},
+                ]
+            },
+        )
+
+    def test_search_falls_back_to_the_legacy_route(self):
+        def responder(method, path, body):
+            if path.endswith("/points/query"):
+                return 404, {"status": "not found"}
+            return 200, {"result": [{"id": "p1", "score": 0.9, "payload": {"area": "ai"}}]}
+
+        with support.StubServer(responder) as server:
+            client = qdrant_module.Qdrant(server.url, attempts=1)
+            hits = client.search("c", [0.1])
+        self.assertEqual([hit["id"] for hit in hits], ["p1"])
+        self.assertEqual([r["path"] for r in server.requests][-1], "/collections/c/points/search")
+
+    def test_point_count_is_exact(self):
+        responder = lambda method, path, body: (200, {"result": {"count": 42}})  # noqa: E731
+        with support.StubServer(responder) as server:
+            client = qdrant_module.Qdrant(server.url, attempts=1)
+            self.assertEqual(client.point_count("c"), 42)
+        self.assertEqual(server.requests[0]["body"], {"exact": True})
+
+    def test_health_is_false_when_unreachable(self):
+        client = qdrant_module.Qdrant("http://127.0.0.1:1", timeout=0.2, attempts=1)
+        self.assertFalse(client.health())
+
+    def test_upsert_retries_a_server_error(self):
+        plan = [500, 200]
+
+        def responder(method, path, body):
+            return plan.pop(0), {"result": {}}
+
+        with support.StubServer(responder) as server:
+            client = qdrant_module.Qdrant(server.url, attempts=2)
+            client.upsert("c", [{"id": "x", "vector": [0.0], "payload": {}}])
+        self.assertEqual(len(server.requests), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()

corpus/tests/test_store.py

@@ -0,0 +1,106 @@
+"""FTS5 chunk index, manifest and meta: the keyword half of hybrid retrieval."""
+
+from __future__ import annotations
+
+import unittest
+
+from corpus import chunker, loader, payloads, store as store_module
+from corpus.config import Config
+from corpus.pipeline import chunk_id
+from corpus.tests import support
+
+
+def build_store(path, cfg, documents):
+    store = store_module.Store(path)
+    for doc in documents:
+        payload_list = []
+        for chunk in chunker.chunk_document(doc.body):
+            payload = payloads.build_payload(doc, chunk, cfg)
+            payload["id"] = chunk_id(doc.path, chunk.index, doc.digest)
+            payload_list.append(payload)
+        store.write_document(doc, payload_list)
+    return store
+
+
+class StoreTest(unittest.TestCase):
+    def setUp(self):
+        self.root = support.temp_repo()
+        self.cfg = support.make_config(self.root)
+        self.addCleanup(_cleanup, self.root)
+        self.documents = loader.load_documents(self.cfg)
+        self.store = build_store(self.cfg.db_path, self.cfg, self.documents)
+        self.addCleanup(self.store.close)
+
+    def test_counts_match_the_documents(self):
+        self.assertEqual(self.store.source_count(), len(self.documents))
+        expected = sum(len(chunker.chunk_document(doc.body)) for doc in self.documents)
+        self.assertEqual(self.store.chunk_count(), expected)
+
+    def test_keyword_search_finds_exact_tokens(self):
+        hits = self.store.keyword_search("zigbee")
+        self.assertTrue(hits, "no hit for a token that is literally in the document")
+        self.assertEqual(hits[0]["area"], "edge")
+
+    def test_keyword_search_respects_area_filter(self):
+        all_hits = self.store.keyword_search("latency")
+        edge_hits = self.store.keyword_search("latency", area="edge")
+        self.assertTrue(all_hits)
+        for hit in edge_hits:
+            self.assertEqual(hit["area"], "edge")
+
+    def test_keyword_search_returns_snippets_and_scores(self):
+        hits = self.store.keyword_search("fence")
+        self.assertTrue(hits)
+        self.assertIn("score", hits[0])
+        self.assertIn("snippet", hits[0])
+
+    def test_operator_injection_is_not_honoured(self):
+        # FTS5 operators must be treated as words, never as query syntax.
+        hits = self.store.keyword_search('edge OR "NEAR"(a b)')
+        self.assertIsInstance(hits, list)
+
+    def test_manifest_records_the_content_hash(self):
+        manifest = self.store.manifest()
+        self.assertEqual(set(manifest), {doc.path for doc in self.documents})
+        for doc in self.documents:
+            self.assertEqual(manifest[doc.path]["content_hash"], doc.digest)
+            self.assertGreater(manifest[doc.path]["chunk_count"], 0)
+
+    def test_delete_removes_rows_from_both_tables(self):
+        target = self.documents[1]
+        self.store.delete_document(target.path)
+        self.assertNotIn(target.path, self.store.manifest())
+        for row in self.store.chunk_payloads():
+            self.assertNotEqual(row["source_path"], target.path)
+
+    def test_meta_roundtrip(self):
+        self.store.set_meta("collection", {"model": "x", "dim": 8})
+        self.assertEqual(self.store.get_meta("collection"), {"model": "x", "dim": 8})
+        self.assertIn("collection", self.store.meta())
+
+    def test_payloads_survive_the_roundtrip(self):
+        rows = self.store.chunk_payloads()
+        self.assertTrue(rows)
+        for row in rows:
+            payloads.validate_payload(row)
+
+
+class MatchExpressionTest(unittest.TestCase):
+    def test_terms_become_prefix_queries(self):
+        self.assertEqual(store_module.match_expression("Zigbee mesh"), '"zigbee"* AND "mesh"*')
+
+    def test_empty_query_is_empty(self):
+        self.assertEqual(store_module.match_expression("   "), "")
+
+    def test_punctuation_is_dropped(self):
+        self.assertEqual(store_module.match_expression("-edge- LoRaWAN, 2023"), '"edge"* AND "lorawan"* AND "2023"*')
+
+
+def _cleanup(root):
+    import shutil
+
+    shutil.rmtree(root, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    unittest.main()