Ingestion redesigned: harvest from the estate, not just content/

- docs/phase-4-corpus.md rewritten: Gitea (all repos), vault via Gitea, Graphify
  graph, curated content as tier 1. Platform guides are config and public copy.
- docs/phase-5-ai-overlay.md: citations to originals, tier preference, live
  source readout.
- Board: AUD-144 (index content/ only) retired; AUD-147 replaces it, blocked by
  AUD-141 and AUD-142. AUD-145 relinked to AUD-143 and AUD-147.

Makefile

@@ -12,10 +12,18 @@ VALIDATOR := $(PYTHON) tools/validate_content.py
 GO ?= go
 TEMPL ?= templ
 
+# The runtime image is distroless/static (deploy/Dockerfile): no libc, no
+# dynamic loader. Everything the server needs is pure Go (modernc.org/sqlite),
+# so a static binary is buildable and is what the container can actually exec.
+# Without this the build host's cgo links the nix glibc loader and the container
+# fails with "exec /usr/bin/resume-server: no such file or directory".
+CGO_ENABLED ?= 0
+
 BIN := bin
 DIST := dist
 
-.PHONY: build dist generate dev test validate validate-self-test fmt vet clean help
+.PHONY: build dist generate dev test validate validate-self-test fmt vet clean help \
+        corpus corpus-sync corpus-rebuild corpus-audit corpus-verify corpus-demo corpus-test
 
 help:
 	@echo "make build            templ generate, compile bin/, no dist"
@@ -26,11 +34,19 @@ help:
 	@echo "make validate         validate content/** front-matter against the frozen schema"
 	@echo "make fmt vet          gofmt / go vet over cmd and internal"
 	@echo "make clean            remove bin/ and dist/"
+	@echo ""
+	@echo "corpus (Phase 4, python3 -m corpus.cli):"
+	@echo "make corpus           rebuild the corpus from content/ (drops the collection)"
+	@echo "make corpus-sync      incremental update; only changed documents are re-embedded"
+	@echo "make corpus-audit     freshness + drift audit (exit 1 on findings)"
+	@echo "make corpus-verify    check the index against content/ and Qdrant"
+	@echo "make corpus-demo      same query with and without an area filter"
+	@echo "make corpus-test      corpus unit tests (no container, no network)"
 
 build: generate
 	@mkdir -p $(BIN)
-	$(GO) build -trimpath -buildvcs=false -o $(BIN)/resume-build ./cmd/build
-	$(GO) build -trimpath -buildvcs=false -o $(BIN)/resume-server ./cmd/server
+	CGO_ENABLED=$(CGO_ENABLED) $(GO) build -trimpath -buildvcs=false -o $(BIN)/resume-build ./cmd/build
+	CGO_ENABLED=$(CGO_ENABLED) $(GO) build -trimpath -buildvcs=false -o $(BIN)/resume-server ./cmd/server
 
 dist: build
 	@$(BIN)/resume-build --dist $(DIST)
@@ -57,5 +73,34 @@ fmt:
 vet:
 	@$(GO) vet ./...
 
+# -- corpus (Phase 4, docs/phase-4-corpus.md) --------------------------------
+#
+# The corpus is data, not a build artefact: it lives in data/corpus/ (gitignored)
+# and in Qdrant. RESUME_QDRANT_URL defaults to http://127.0.0.1:6333 and
+# RESUME_EMBED_PROVIDER defaults to local-hashing; production sets both in the
+# environment, never in this file.
+
+CORPUS := $(PYTHON) -m corpus.cli
+
+corpus: corpus-rebuild
+
+corpus-rebuild:
+	@$(CORPUS) rebuild
+
+corpus-sync:
+	@$(CORPUS) sync
+
+corpus-audit:
+	@$(CORPUS) audit --fail-on-findings
+
+corpus-verify:
+	@$(CORPUS) verify
+
+corpus-demo:
+	@$(CORPUS) demo-filter --query "distributed audio endpoints" --area edge
+
+corpus-test:
+	@$(PYTHON) -m unittest discover -s corpus/tests -t .
+
 clean:
 	@rm -rf $(BIN) $(DIST)

README.md

@@ -58,6 +58,21 @@ keeps acceptance test A1 true.
 The two things that are not files are `/search` (FTS5 over page titles and body text) and
 `/api/stage/stream` (the presentation panel driven by an event stream instead of a route).
 
+The retrieval corpus (Phase 4) is built separately, because it writes to a service rather
+than to `dist/`:
+
+```sh
+docker compose -f deploy/qdrant.yml up -d   # on the runtime host, loopback only
+make corpus          # content/ -> chunks -> Qdrant + SQLite FTS5
+make corpus-verify   # does the index agree with content/ and with Qdrant?
+make corpus-audit    # stale documents, index drift, contract drift
+make corpus-test     # 76 unit tests, no container and no network
+```
+
+The site never calls Qdrant, so acceptance test A1 is unaffected: stopping the corpus
+services leaves every page working. Operator manual and measurements:
+`docs/phase-4-notes.md`.
+
 ## Ground rules
 
 - **Do not touch other systems.** No stopping containers, no editing other projects, no `nixos-rebuild`, no Caddy changes except the single block specified in Phase 3 — and that one Sam applies.
@@ -72,7 +87,7 @@ The two things that are not files are `/search` (FTS5 over page titles and body
 - [x] Phase 1 — content source of truth (40 documents across six areas, `make validate`)
 - [x] Phase 2 — static site (`make dist`, 45 pages, 86 stage fragments, FTS5 search)
 - [ ] Phase 3 — public deployment
-- [ ] Phase 4 — corpus
+- [x] Phase 4 — corpus and retrieval (51 chunks, FTS5 + Qdrant, freshness audit)
 - [ ] Phase 5 — AI overlay
 - [ ] Phase 6 — guardrails
 - [ ] Phase 7 — polish

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()

deploy/compose.yml

@@ -12,7 +12,14 @@
 #
 # `curl http://127.0.0.1:8096/healthz` must answer with static ready.
 
-name: ai-resume
+# The compose project name is deliberately not `ai-resume`: /home/sam/deployment/
+# ai-resume on .13 already runs a compose project by that name (backend + db on
+# the ai-mesh network). Sharing the name would make this site's containers show
+# up as orphans in that project and vice versa, so anything running `compose
+# down --remove-orphans` under either project could stop the other. The image
+# tag stays ai-resume:local (Phase 3 fixes it); only the project namespace is
+# distinct.
+name: ai-resume-site
 
 services:
   resume:
@@ -43,6 +50,20 @@ services:
       retries: 3
       start_period: 5s
 
+# The project network gets an explicit, small subnet. The host's docker daemon
+# has every default address pool allocated by other projects ("all predefined
+# address pools have been fully subnetted"), so an implicit compose network
+# cannot be created. Pinning /28 of RFC1918 space that no other bridge or route
+# on .13 uses keeps this deploy independent of the host's pool exhaustion and
+# does not touch any other container. Single-service stack: two addresses is
+# plenty.
+networks:
+  default:
+    ipam:
+      config:
+        - subnet: 10.89.0.0/28
+          gateway: 10.89.0.1
+
 # Phase 4/5 add at most three more containers (Qdrant, Prefect worker, query
 # service) inside the 4-container and 1.0 GiB budget in AGENTS.md §4.3. Mode A
 # needs none of them and must keep working when they are stopped (test A1).

deploy/deploy.sh

@@ -0,0 +1,390 @@
+#!/usr/bin/env bash
+# deploy.sh — release, deploy and roll back the Mode A site (docs/phase-3-deploy.md §3.1).
+#
+#   ./deploy/deploy.sh                  build, test, stage a release, flip `current`,
+#                                       start the container, gate on /healthz
+#   ./deploy/deploy.sh --rollback       put the previous release back, in one command
+#   ./deploy/deploy.sh --status         what is live, what is previous, is it healthy
+#   ./deploy/deploy.sh --release-dir D  deploy an already-built release directory
+#                                       (copied into releases/ first; used by the
+#                                       rollback rehearsal below)
+#
+# Layout on the runtime host — the release directory, not the image tag, is the
+# unit that runs, so a rollback cannot publish bytes that were never tested:
+#
+#   /home/sam/deployment/resume/
+#     releases/<utc-timestamp>-<git-sha>/{bin/resume-server,dist/,release.json}
+#     current   -> releases/<…>   what the container is running
+#     previous  -> releases/<…>   what --rollback restores
+#
+# A deploy is: build → test → stage → flip `current` → build the image from that
+# release directory → `compose up -d` → `curl /healthz` until ready. Any failure
+# after the flip puts `current` and the container back on the previous release
+# and exits non-zero. No step is a model decision (AGENTS.md §4.6); this script
+# is the only thing that publishes.
+#
+# Rollback rehearsal (Definition of Done: "a deliberate bad build proves rollback"):
+#
+#   good=$(readlink ~/deployment/resume/current)
+#   bad=$(mktemp -d) && mkdir -p "$bad/bin" "$bad/dist"
+#   printf '#!/bin/sh\nexit 1\n' >"$bad/bin/resume-server"   # binary that cannot start
+#   chmod +x "$bad/bin/resume-server"
+#   cp -a ~/deployment/resume/"$good"/dist/. "$bad/dist/"   # valid pages
+#   ./deploy/deploy.sh --release-dir "$bad"                 # expect exit 1
+#   readlink ~/deployment/resume/current                    # still $good
+#
+# Run it on the runtime host (.13) from a checkout of this repository. It never
+# reads content/ and never calls a model.
+
+set -Eeuo pipefail
+
+APP=ai-resume                  # image tag (fixed by Phase 3 §3.2)
+PROJECT=ai-resume-site         # compose project; distinct from the older
+                               # /home/sam/deployment/ai-resume project on .13
+PORT=8096
+HEALTH_PATH=/healthz
+
+ROOT=${RESUME_DEPLOY_ROOT:-/home/sam/deployment/resume}
+RELEASES=$ROOT/releases
+CURRENT=$ROOT/current
+PREVIOUS=$ROOT/previous
+
+KEEP=${RESUME_KEEP_RELEASES:-5}            # release directories kept on disk
+HEALTH_TRIES=${RESUME_HEALTH_TRIES:-40}    # seconds to wait for /healthz
+
+REPO=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd)
+COMPOSE=$REPO/deploy/compose.yml
+DOCKERFILE=$REPO/deploy/Dockerfile
+
+MODE=deploy
+RELEASE_DIR_OVERRIDE=
+
+log()  { printf '%s  %s
+' "$(date -u +%H:%M:%SZ)" "$*" >&2; }
+warn() { printf '%s  WARN  %s
+' "$(date -u +%H:%M:%SZ)" "$*" >&2; }
+die()  { printf '%s  FAIL  %s
+' "$(date -u +%H:%M:%SZ)" "$*" >&2; exit 1; }
+
+on_err() {
+	local rc=$? line=$1
+	printf '%s  FAIL  line %s exited %s
+' "$(date -u +%H:%M:%SZ)" "$line" "$rc" >&2
+	exit "$rc"
+}
+trap 'on_err $LINENO' ERR
+
+usage() {
+	sed -n '2,13p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//'
+	exit "${1:-0}"
+}
+
+while [ $# -gt 0 ]; do
+	case $1 in
+	--rollback)    MODE=rollback ;;
+	--status)      MODE=status ;;
+	--release-dir) shift; RELEASE_DIR_OVERRIDE=${1:-} ;;
+	--keep)        shift; KEEP=${1:-5} ;;
+	-h|--help)     usage 0 ;;
+	*)             printf 'unknown argument: %s
+' "$1" >&2; usage 1 ;;
+	esac
+	shift
+done
+
+# ---------------------------------------------------------------- helpers ---
+
+compose() { docker compose -f "$COMPOSE" -p "$PROJECT" "$@"; }
+
+current_target()  { readlink -- "$CURRENT" 2>/dev/null || true; }
+previous_target() { readlink -- "$PREVIOUS" 2>/dev/null || true; }
+
+# flip <relative-path-under-ROOT> — replace the symlink atomically. A reader sees
+# either the old release or the new one, never a missing link.
+flip() {
+	ln -sfn "$1" "$CURRENT.tmp"
+	mv -Tf "$CURRENT.tmp" "$CURRENT"
+}
+
+release_name_of() { basename "$1"; }
+
+# The tag deploy/compose.yml runs. Releases are distinguished by content, not by
+# tag, so compose.yml keeps its fixed image name.
+image_name() { printf '%s:local' "$APP"; }
+
+# The port must be ours, or free. 8095 belongs to offsite.lab.
+check_port() {
+	local listening
+	listening=$(ss -ltnH "sport = :$PORT" 2>/dev/null || true)
+	if [ -n "$listening" ]; then
+		if [ -z "$(compose ps -q resume 2>/dev/null)" ]; then
+			printf '%s
+' "$listening" >&2
+			die "port $PORT is held by something that is not $APP; pick the next free port and record it in reference/stack.md"
+		fi
+		log "port $PORT is held by our own container (expected on a redeploy)"
+	fi
+}
+
+check_tools() {
+	command -v docker >/dev/null || die "docker is not on PATH"
+	docker compose version >/dev/null 2>&1 || die "docker compose v2 is not available"
+	command -v curl >/dev/null || die "curl is not on PATH"
+	command -v ss >/dev/null || die "ss (iproute2) is not on PATH"
+	mkdir -p "$RELEASES"
+}
+
+memory_report() {
+	local avail
+	avail=$(awk '/^MemAvailable:/ {print int($2/1024)}' /proc/meminfo)
+	log "host memory available: ${avail} MiB — the container cap is 120 MiB; .13 is memory-tight by design"
+	if [ "${avail:-0}" -lt 512 ]; then
+		warn "under 512 MiB available; the Go build may push this host into swap"
+	fi
+}
+
+# --------------------------------------------------------------- build ------
+
+build_and_test() {
+	log "build + test in $REPO"
+	(
+		cd "$REPO"
+		rm -rf dist
+		if command -v nix >/dev/null 2>&1; then
+			nix develop . -c bash -c 'make validate test dist'
+		elif command -v templ >/dev/null 2>&1 && command -v go >/dev/null 2>&1; then
+			make validate test dist
+		else
+			echo "no toolchain: neither 'nix develop' nor templ+go is available" >&2
+			exit 1
+		fi
+	)
+	[ -f "$REPO/bin/resume-server" ] || die "build produced no bin/resume-server"
+	[ -f "$REPO/dist/index.html" ]   || die "build produced no dist/index.html"
+	log "build ok: $(jq -r '"\(.pages) pages, \(.documents) documents, \(.fragments) fragments"' "$REPO/dist/build.json" 2>/dev/null || echo 'dist/build.json unreadable')"
+}
+
+# -------------------------------------------------------------- release -----
+
+write_manifest() {
+	local dir=$1 sha dirty
+	sha=$(git -C "$REPO" rev-parse HEAD 2>/dev/null || echo unknown)
+	dirty=false
+	git -C "$REPO" diff --quiet 2>/dev/null || dirty=true
+	cat >"$dir/release.json" <<JSON
+{
+  "release": "$(basename "$dir")",
+  "image": "$(image_name)",
+  "git_sha": "$sha",
+  "git_dirty": $dirty,
+  "built_at": "$(date -u +%Y-%m-%dT%H:%M:%SZ)",
+  "host": "$(hostname)",
+  "pages": $(jq -r '.pages // 0' "$REPO/dist/build.json" 2>/dev/null || echo 0),
+  "documents": $(jq -r '.documents // 0' "$REPO/dist/build.json" 2>/dev/null || echo 0)
+}
+JSON
+}
+
+stage_release() {
+	local name dest sha
+	sha=$(git -C "$REPO" rev-parse HEAD 2>/dev/null || echo unknown)
+	name=$(date -u +%Y%m%dT%H%M%SZ)-${sha:0:7}
+	dest=$RELEASES/$name
+
+	mkdir -p "$dest/bin" "$dest/dist"
+	install -m 0755 "$REPO/bin/resume-server" "$dest/bin/resume-server"
+	cp -a "$REPO/dist/." "$dest/dist/"
+	write_manifest "$dest"
+	printf '%s
+' "$dest"
+}
+
+# Copy an externally built release into releases/ so that every deployable
+# directory lives under exactly one root and `current` never dangles.
+adopt_release() {
+	local src dest
+	src=$(cd "$1" && pwd)
+	[ -f "$src/bin/resume-server" ] || die "$src has no bin/resume-server"
+	[ -f "$src/dist/index.html" ]   || die "$src has no dist/index.html"
+	dest=$RELEASES/$(date -u +%Y%m%dT%H%M%SZ)-manual
+	mkdir -p "$dest"
+	cp -a "$src/." "$dest/"
+	write_manifest "$dest"
+	printf '%s
+' "$dest"
+}
+
+# ---------------------------------------------------------------- image -----
+
+build_image() {
+	log "docker build from $1"
+	docker build --file "$DOCKERFILE" --tag "$(image_name)" "$1" 1>&2
+}
+
+# ------------------------------------------------------------------ run -----
+
+start() {
+	log "docker compose up -d ($1)"
+	compose up -d --no-build --force-recreate 1>&2
+}
+
+wait_healthy() {
+	local i body
+	for i in $(seq 1 "$HEALTH_TRIES"); do
+		if body=$(curl -fsS --max-time 3 "http://127.0.0.1:$PORT$HEALTH_PATH" 2>/dev/null); then
+			if printf '%s' "$body" | grep -Eq '"status"[[:space:]]*:[[:space:]]*"ok"'; then
+				log "health gate passed after ${i}s"
+				return 0
+			fi
+		fi
+		sleep 1
+	done
+	warn "health gate failed after ${HEALTH_TRIES}s; last response:"
+	curl -sS --max-time 3 "http://127.0.0.1:$PORT$HEALTH_PATH" >&2 || true
+	compose logs --tail 40 resume >&2 || true
+	return 1
+}
+
+# Definition-of-Done evidence, checked on every deploy instead of by hand:
+# mem_limit == memswap_limit == the value declared in compose.yml, and the port
+# published on 127.0.0.1 only.
+assert_shape() {
+	local cid mem swap ip declared
+	declared=$(awk '/^[[:space:]]*mem_limit:/ {gsub(/[^0-9]/, "", $2); print $2}' "$COMPOSE")
+	cid=$(compose ps -q resume 2>/dev/null || true)
+	[ -n "$cid" ] || die "no container for service 'resume'"
+	mem=$(docker inspect -f '{{.HostConfig.Memory}}' "$cid")
+	swap=$(docker inspect -f '{{.HostConfig.MemorySwap}}' "$cid")
+	[ "$mem" = "$swap" ] || die "mem_limit ($mem) != memswap_limit ($swap)"
+	[ "$mem" = "$((declared * 1024 * 1024))" ] || die "mem_limit is $mem, expected ${declared} MiB"
+	ip=$(docker inspect -f '{{range $p, $b := .HostConfig.PortBindings}}{{range $b}}{{.HostIp}} {{end}}{{end}}' "$cid" | xargs)
+	[ "$ip" = "127.0.0.1" ] || die "published on '${ip}', expected 127.0.0.1 only"
+	log "shape ok: mem_limit == memswap_limit == ${declared} MiB, published on 127.0.0.1:$PORT"
+}
+
+deploy_release() {
+	local dir old name
+	dir=$1
+	name=$(release_name_of "$dir")
+	[ -d "$dir" ] || die "release directory does not exist: $dir"
+	[ -f "$dir/bin/resume-server" ] || die "$dir has no bin/resume-server"
+	[ -f "$dir/dist/index.html" ]   || die "$dir has no dist/index.html — a build that cannot serve a landing page is not a release"
+
+	build_image "$dir"
+	old=$(current_target)
+
+	# The flip happens before the container change, so `current` always describes
+	# what is being started. The release it replaced becomes `previous`.
+	flip "releases/$name"
+	if [ -n "$old" ] && [ "$old" != "releases/$name" ]; then
+		ln -sfn "$old" "$PREVIOUS"
+		log "previous -> $old"
+	fi
+
+	if start "$name" && wait_healthy && assert_shape; then
+		printf '%s
+' "$(date -u +%Y-%m-%dT%H:%M:%SZ)  deploy releases/$name ok" >>"$ROOT/deploy.log"
+		log "deployed $name"
+		prune
+		return 0
+	fi
+
+	if [ -z "$old" ] || [ "$old" = "releases/$name" ]; then
+		warn "first deploy failed and there is no previous release; leaving the container down"
+		compose down >&2 || true
+		rm -f "$CURRENT"
+		return 1
+	fi
+
+	warn "deploy of $name failed; restoring $old"
+	flip "$old"
+	build_image "$ROOT/$old" || die "could not rebuild the previous image; container state unknown"
+	if start "$old" && wait_healthy && assert_shape; then
+		printf '%s
+' "$(date -u +%Y-%m-%dT%H:%M:%SZ)  rolled back to $old after failed deploy of releases/$name" >>"$ROOT/deploy.log"
+		die "deploy of releases/$name failed; rolled back to $old (the failed release is kept for inspection)"
+	fi
+	die "deploy of releases/$name failed AND rollback to $old failed — container state is unknown; run '$0 --status' and read $ROOT/deploy.log"
+}
+
+do_rollback() {
+	local cur prev
+	cur=$(current_target)
+	prev=$(previous_target)
+	[ -n "$prev" ] || die "no previous release recorded; nothing to roll back to"
+	[ "$prev" != "$cur" ] || die "previous and current are the same release ($cur); nothing to roll back"
+	[ -d "$ROOT/$prev" ] || die "previous release directory is missing: $ROOT/$prev"
+
+	log "rolling back: $cur -> $prev"
+	build_image "$ROOT/$prev"
+	flip "$prev"
+	if start "$prev" && wait_healthy && assert_shape; then
+		[ -n "$cur" ] && ln -sfn "$cur" "$PREVIOUS"
+		printf '%s
+' "$(date -u +%Y-%m-%dT%H:%M:%SZ)  rollback $cur -> $prev" >>"$ROOT/deploy.log"
+		log "rolled back to $prev; $cur is now previous"
+		return 0
+	fi
+
+	warn "rollback to $prev failed; returning to $cur"
+	flip "$cur"
+	build_image "$ROOT/$cur" || die "could not rebuild $cur; container state unknown"
+	start "$cur" && wait_healthy || die "return to $cur failed; container state unknown"
+	die "rollback to $prev failed; $cur was restored"
+}
+
+# Keep the newest KEEP releases plus current and previous, nothing else.
+prune() {
+	local cur prev name dir
+	cur=$(current_target); prev=$(previous_target)
+	for dir in $(ls -1dt "$RELEASES"/*/ 2>/dev/null | tail -n +"$((KEEP + 1))"); do
+		name="releases/$(basename "$dir")"
+		if [ "$name" = "$cur" ] || [ "$name" = "$prev" ]; then continue; fi
+		rm -rf -- "$dir"
+		log "pruned $name"
+	done
+}
+
+do_status() {
+	local cur prev body
+	cur=$(current_target); prev=$(previous_target)
+	body=$(curl -fsS --max-time 3 "http://127.0.0.1:$PORT$HEALTH_PATH" 2>/dev/null || true)
+	printf 'root      %s
+' "$ROOT"
+	printf 'current   %s
+' "${cur:-<none>}"
+	printf 'previous  %s
+' "${prev:-<none>}"
+	if [ -n "$body" ]; then
+		printf 'health    %s
+' "$(printf '%s' "$body" | jq -c '{status, pages: .static.pages, search: .search.ready}' 2>/dev/null || printf '%s' "$body")"
+	else
+		printf 'health    unreachable
+'
+	fi
+	compose ps || true
+}
+
+# ------------------------------------------------------------------ main ----
+
+case $MODE in
+status)
+	do_status
+	;;
+rollback)
+	check_tools
+	do_rollback
+	;;
+deploy)
+	check_tools
+	check_port
+	memory_report
+	if [ -n "$RELEASE_DIR_OVERRIDE" ]; then
+		deploy_release "$(adopt_release "$RELEASE_DIR_OVERRIDE")"
+	else
+		build_and_test
+		deploy_release "$(stage_release)"
+	fi
+	log "live: $(current_target)"
+	;;
+esac

deploy/qdrant.yml

@@ -0,0 +1,55 @@
+# Qdrant for the corpus (docs/phase-4-corpus.md, reference/stack.md).
+#
+#   mem_limit == memswap_limit == 400m   the cap in the memory budget, and the
+#                                        container cannot reach host swap
+#   ports on 127.0.0.1 only              Caddy on .35 is the only public listener
+#   its own compose project              stopping or recreating it cannot disturb
+#                                        the Mode A site (test A1 still passes:
+#                                        the site never calls Qdrant)
+#
+# Run on the runtime host (.13):
+#
+#   docker compose -f deploy/qdrant.yml up -d
+#   curl -s http://127.0.0.1:6333/        # must answer with a version
+#
+# Then, in the corpus environment (RESUME_QDRANT_URL is already the default):
+#
+#   python3 -m corpus.cli rebuild
+#
+# Deleting the volume is always recoverable: `rebuild` re-derives every vector
+# from content/.
+
+name: ai-resume-index
+
+services:
+  qdrant:
+    image: qdrant/qdrant:v1.20.0
+    container_name: ai-resume-qdrant
+    restart: unless-stopped
+    mem_limit: 400m
+    memswap_limit: 400m
+    ports:
+      - "127.0.0.1:6333:6333"   # REST — the pipeline and the query service use this
+      - "127.0.0.1:6334:6334"   # gRPC — reserved for the Phase 5 query service
+    volumes:
+      - qdrant_storage:/qdrant/storage
+    environment:
+      # The corpus is ~1k chunks today and the measurement in
+      # reference/stack.md records the ceiling. Keep the HNSW defaults; do not
+      # tune for speed on a memory-bound host.
+      QDRANT__TELEMETRY_DISABLED: "true"
+      QDRANT__LOG_LEVEL: "WARN"
+    logging:
+      driver: json-file
+      options:
+        max-size: 10m
+        max-file: "3"
+    healthcheck:
+      test: ["CMD-SHELL", "bash -c '</dev/tcp/127.0.0.1/6333' || exit 1"]
+      interval: 30s
+      timeout: 3s
+      retries: 3
+      start_period: 10s
+
+volumes:
+  qdrant_storage:

docs/phase-4-corpus.md

@@ -1,81 +1,157 @@
-# Phase 4 — Corpus and retrieval
+# Phase 4 — Ingestion from sources
 
-**Goal:** `content/` becomes a retrievable corpus with filters that work before the vector search, and a freshness audit that cannot silently rot.
+**Goal:** a fireable tool that reads the estate — Gitea, the notes vault, the knowledge graph — and keeps a searchable pool current. The pool is not built from `content/` alone; `content/` is the top tier of it.
 
-**Depends on:** Phase 1 (content), Phase 2 (FTS5 index exists).
+**Depends on:** Phase 1 (content and schema), Phase 2 (FTS5 index).
+
+**Why this changed.** An earlier version of this document said the corpus would be built from `content/` only. That was wrong. A hand-curated set cannot answer "how did the photo pipeline deduplicate" or "what did the garden controller do before the rewrite". The estate can. Curated claims answer questions about Sam; the estate answers questions about the work.
 
 ---
 
-## Work
+## 1. Sources
 
-### 4.1 Pipeline
+| Source | Reached by | What to take |
+|---|---|---|
+| **Gitea**, all repositories | API over HTTPS | `README.md`, file tree (paths only), selected source files, recent commit messages, language statistics, repo description and topics |
+| **`sam/obsidian-vault`** | Same Gitea API — the vault is a repo | Markdown body, front-matter title and tags, folder path |
+| **Graphify output** | `graph.json` committed to the vault repo | Nodes (label, source_file, community), edges (relation, confidence) |
+| **`content/`** | Local repo | The curated claims. **Tier 1** — outranks everything above on conflict |
 
-```
-content/  ──► validate ──► chunk ──► enrich ──► embed ──► upsert Qdrant
-                                                              │
-                                    SQLite FTS5 (from Phase 2) ┘
+Gitea is the only network dependency. The vault and the graph arrive through it, so there is no cross-machine copy and no SSH from `.13` to `.27`.
+
+### What not to read
+
+Media never becomes text. Images, video and audio are recorded as `image` node references, tagged with their path, and nothing more.
+
+---
+
+## 2. Platform guides
+
+One file per platform in `platforms/<name>.yml`. They are configuration **and** public copy — the site shows them so a visitor can see what the assistant reads.
+
+```yaml
+name: gitea
+enabled: true
+what: "Self-hosted Git. Source for every current project."
+fits: [ai, platform, edge]
+base: https://gitea.lab.audasmedia.com.au
+auth: GITEA_TOKEN
+include:
+  repos: all
+exclude:
+  paths: ["**/node_modules/**", "**/.venv/**", "**/.git/**", "**/*.env*",
+          "**/.silverbullet.auth.json", "**/*.pem", "**/*.key"]
+  repos: []
+read: [readme, tree, selected_source, commits, languages]
+limits: { max_documents: 4000, max_file_kb: 200 }
 ```
 
-- **Chunking:** 400–800 tokens, ~12% overlap, **code-aware** — never split inside a fenced block, table or front-matter.
-- **Deterministic IDs:** `sha256(source_path + chunk_index + content_hash)`. Re-running must update, not duplicate.
-- **Incremental:** hash the source set; only changed documents are re-chunked and re-embedded.
-- **One command** rebuilds from scratch. Deleting the Qdrant volume is always recoverable.
+```yaml
+name: obsidian
+enabled: true
+what: "Personal knowledge base. High-level notes on every system, in Sam's words."
+fits: [about, ai, platform, edge]
+base: sam/obsidian-vault
+include:
+  paths: ["200 projects/**", "300 areas/**", "400 Personal Family/**",
+          "100 Table of Contents/**", "010 inbox/**"]
+exclude:
+  paths: [".obsidian/**", ".trash/**", ".venv/**", "templates/**",
+          "**/*.excalidraw", "**/*.png", "**/*.jpg"]
+limits: { max_documents: 4000, max_file_kb: 200 }
+```
 
-### 4.2 Payload schema (the retrieval filter)
+```yaml
+name: graphify
+enabled: true
+what: "Knowledge graph over the vault. Answers 'what connects X to Y' rather than 'what mentions X'."
+fits: [ai]
+base: sam/obsidian-vault
+read: [graph_json]
+```
+
+`fits` drives the payload `area`, which is what lets the query filter before it searches.
+
+---
+
+## 3. Metadata on every chunk
 
 ```json
 {
-  "area": "ai|platform|edge|products|history|about",
-  "level": 1,
-  "source_type": "doc|code|adr|metric",
-  "confidence": "delivered|running|in-progress|abandoned|historic",
-  "period": "2023–",
-  "artefacts": ["..."],
-  "last_verified": "2026-10-20",
+  "platform": "gitea | obsidian | graphify | curated",
+  "tier": 1,
+  "source_type": "doc | code | commit | graph_node | claim",
+  "area": "ai | platform | edge | products | history | about",
+  "origin_url": "https://gitea.../sam/photo-pipeline/src/x.py",
+  "path": "src/x.py",
+  "title": "photo-pipeline / dedup",
+  "confidence": "delivered | running | in-progress | abandoned | historic",
+  "last_modified": "2026-10-14T05:00:00Z",
+  "content_hash": "sha256:...",
+  "public": true,
   "text": "..."
 }
 ```
 
-`area` and `confidence` are the two that matter: they let the query path filter *before* searching, and they let the answer qualify unfinished work.
+`tier: 1` is curated `content/`. Everything harvested is `tier: 2`. The query layer uses this to prefer the curated claim when the two disagree.
 
-### 4.3 Embeddings
-- Batch, low volume, incremental.
-- **Record the embedding model and dimension per collection.** A provider change is a governed migration, not a silent corruption.
-- Provider decision pending Sam — see `reference/stack.md`. Build provider-agnostic behind one interface.
-
-### 4.4 Scheduling
-- Prefect deployment on the existing Prefect server; **worker only**, do not deploy a second server.
-- Trigger: git push to `content/`, plus a nightly audit.
-- Nightly audit reports documents past their `last_verified` horizon and alerts via the existing Apprise service.
+`origin_url` is mandatory for harvested chunks. A chunk with no origin is not admitted — an answer that cannot be traced is worse than no answer.
 
 ---
 
-## Definition of Done
+## 4. Pipeline
 
-- [ ] One command rebuilds the corpus from an empty volume.
-- [ ] Running it twice produces no duplicate vectors (count is stable).
-- [ ] No chunk boundary lands inside a code fence, table or front-matter block.
-- [ ] Every chunk carries all payload fields; a missing field fails the run rather than defaulting.
-- [ ] Payload filtering demonstrably narrows results — demonstrate the same query with and without an `area` filter.
-- [ ] Incremental mode re-embeds only changed documents; prove it with the provider call count.
-- [ ] Freshness audit runs and reports stale documents.
-- [ ] Qdrant memory measured at 500 / 5,000 / 20,000 chunks and recorded in `reference/stack.md`, within its cap.
+```
+guides  →  harvest (per platform)  →  normalise to text  →  chunk  →  enrich  →  embed  →  upsert
+                                        │
+                                        └─ reject: excluded path, no origin, binary, oversize
+```
+
+- **Chunking** is code-aware: 400–800 tokens, ~12% overlap, never split inside a fenced block, table, front-matter block or function. Use the language grammar where available; fall back to paragraph splitting.
+- **Deterministic IDs:** `sha256(origin_url + chunk_index + content_hash)`. Re-running updates, never duplicates.
+- **Incremental:** compare `content_hash` against what the pool already holds; embed only what changed. A full rebuild must always be possible and is one command.
+- **Graph** is not chunked. Nodes and edges go into their own collection, joined to note chunks by `path`, so a graph answer can name the note it came from.
 
 ---
 
-## Out of scope
+## 5. Scheduling
 
-- Answering questions. Phase 4 ends at "retrieval returns the right chunks".
-- Query graphs, prompts, guardrails — Phase 5 and 6.
-- Ingesting anything outside `content/`. Four sources maximum, and this phase uses one: the repo.
+- On demand: one command.
+- Scheduled: a Prefect deployment on the existing server. Worker only; do not deploy a second Prefect server.
+- Reports what changed each run: documents added, updated, removed, and any excluded-path rejections.
+- A source that has not changed in its expected window raises an alert through the existing Apprise service. Silent staleness is the failure mode that matters.
 
 ---
 
-## Risks
+## 6. Definition of Done
+
+- [ ] One command rebuilds the pool from empty.
+- [ ] A second run produces no duplicate vectors.
+- [ ] A probe document placed in an excluded path never appears in the pool, and the run reports the rejection.
+- [ ] A document with no resolvable origin is refused, not defaulted.
+- [ ] Every chunk carries all fields in section 3.
+- [ ] Incremental mode re-embeds only changed documents, proven by embedding call count.
+- [ ] Graph nodes and edges are queryable and each resolves to a source note.
+- [ ] The scheduled run reports deltas and alerts on staleness.
+- [ ] Qdrant stays inside its `mem_limit`; footprint recorded at 1k, 10k and 40k chunks.
+- [ ] No credential, `.env`, key file or family-personal binary is present in the pool — verified by search, not by assertion.
+
+---
+
+## 7. Out of scope
+
+- Answering questions. This phase ends at "retrieval returns the right chunks with their origins".
+- The query graph, prompts and guardrails — Phase 5.
+- Building the Graphify graph. It already exists and is rebuilt on `.27` with its own skill. This phase consumes its output.
+
+---
+
+## 8. Risks
 
 | Risk | Mitigation |
 |---|---|
-| Corpus grows beyond RAM | Incremental by default; measure at three corpus sizes; cap Qdrant with `mem_limit` |
-| Duplicate vectors on re-run | Deterministic IDs are mandatory, not optional |
-| Silent staleness | Nightly audit + Apprise alert; a stale index is otherwise an invisible quality failure |
-| Provider rate limits during bulk embed | Batch with retry/backoff; ingest is low-volume and incremental |
+| Volume — 50 repos plus 280 notes exceeds the memory cap | Cap per platform, skip large files, measure at three corpus sizes |
+| Scraped text contradicts a curated claim | `tier` field; the query prefers tier 1 and says so |
+| A secret is harvested | Exclusion list plus a post-run scan of the pool for credential patterns, failing the run if found |
+| Staleness is invisible | Scheduled run plus alert |
+| The estate is unavailable when a recruiter asks | The pool is local; only ingestion needs Gitea. Serving never does |

docs/phase-4-notes.md

@@ -0,0 +1,304 @@
+# Phase 4 notes — corpus and retrieval
+
+Operator manual for the corpus pipeline, and the evidence for the Phase 4
+Definition of Done in `docs/phase-4-corpus.md`. Every number below was measured
+on 2026-10-26 against a live Qdrant 1.20.0 container with a 400 MiB cap.
+
+---
+
+## 1. What exists now
+
+| Path | What it is |
+|---|---|
+| `corpus/` | The pipeline: one Python package, standard library only |
+| `corpus/cli.py` | `python3 -m corpus.cli {rebuild,sync,audit,verify,demo-filter,hybrid,measure,info}` |
+| `corpus/chunker.py` | Code-aware chunking (fenced block, table and front-matter are atomic) |
+| `corpus/embeddings.py` | The only place a vector is made; two providers behind one interface |
+| `corpus/qdrant.py` | Qdrant REST client (no `qdrant-client` dependency) |
+| `corpus/store.py` | SQLite: FTS5 over chunks, the incremental manifest, the collection contract |
+| `corpus/pipeline.py` | validate → chunk → enrich → embed → upsert, in that order |
+| `corpus/freshness.py` | Nightly audit: stale documents, index drift, contract drift |
+| `corpus/measure.py` | Memory measurement at 500 / 5,000 / 20,000 chunks |
+| `corpus/tests/` | 76 unit tests, no container, no network |
+| `flows/corpus.py` | Prefect flows: `corpus-sync`, `corpus-rebuild`, `corpus-audit` |
+| `deploy/qdrant.yml` | Qdrant container: `mem_limit` = `memswap_limit` = 400 m, loopback only |
+
+Two indexes are written from one pass over `content/`, so they cannot drift:
+
+* **vectors** in Qdrant (`resume_corpus`), one point per chunk;
+* **keywords** in SQLite (`data/corpus/corpus.db`), one FTS5 row per chunk.
+
+Retrieval uses both (`corpus.cli hybrid`, reciprocal rank fusion). Vectors miss
+exact tokens such as repo names, `LoRaWAN`, `MQTT` and dates; FTS5 alone misses
+paraphrase. Decision D10 requires the pair.
+
+---
+
+## 2. Commands
+
+```sh
+make corpus          # rebuild from content/ (drops the collection first)
+make corpus-sync     # incremental: only changed documents are re-embedded
+make corpus-audit    # freshness + drift audit; exit 1 on findings
+make corpus-verify   # index vs content/ vs Qdrant, consistency report
+make corpus-demo     # the same query with and without an area filter
+make corpus-test     # 76 unit tests
+```
+
+Direct form, when you need flags:
+
+```sh
+python3 -m corpus.cli rebuild --qdrant-url http://127.0.0.1:6333 --dim 1024
+python3 -m corpus.cli audit --today 2027-01-20 --json --fail-on-findings
+python3 -m corpus.cli measure --sizes 500,5000,20000 --container ai-resume-qdrant
+```
+
+Configuration is environment-first. Nothing is read from a file in this repo.
+
+| Variable | Default | Meaning |
+|---|---|---|
+| `RESUME_QDRANT_URL` | `http://127.0.0.1:6333` | Qdrant REST endpoint |
+| `RESUME_COLLECTION` | `resume_corpus` | collection name |
+| `RESUME_EMBED_PROVIDER` | `local-hashing` | `local-hashing` or `omniroute` |
+| `RESUME_EMBED_MODEL` | provider default | recorded per collection |
+| `RESUME_EMBED_DIM` | provider default | recorded per collection |
+| `RESUME_CORPUS_DB` | `data/corpus/corpus.db` | FTS5 + manifest |
+| `RESUME_MAX_AGE_DAYS` | `180` | freshness horizon |
+| `RESUME_WARN_DAYS` | `30` | how early "due soon" starts |
+| `RESUME_APPRISE_URL` | unset | existing Apprise endpoint for audit alerts |
+| `OMNIROUTE_API_KEY` | unset | only for the `omniroute` provider; never in the repo |
+
+---
+
+## 3. Pipeline
+
+```
+content/ ─► make validate ─► chunk ─► enrich ─► embed ─► upsert Qdrant
+                                        │
+                                        └─► SQLite FTS5 + manifest
+```
+
+* **Validate first.** The pipeline runs the Phase 1 validator
+  (`tools/validate_content.py`). A broken content tree stops the run before a
+  single vector is written.
+* **Chunking.** Target 600 tokens, hard ceiling 800, ~12 % overlap. The body is
+  parsed into atomic blocks (fenced code, table, heading, list item, quote,
+  paragraph, rule) and chunks are packed from whole blocks, so a boundary
+  cannot fall inside a fence, a table or front-matter. A list item or a table
+  that exceeds 800 tokens becomes one oversize chunk and is flagged, never cut.
+  Tokens are counted as words plus punctuation, which is a conservative
+  under-estimate of BPE tokens.
+* **Payload before vectors.** Every chunk payload is built and validated for
+  every changed document *before* one point is written. A missing or invalid
+  field aborts the run with the document and chunk named.
+* **Deterministic ids.** `sha256(source_path + chunk_index + content_hash)`,
+  rendered as a UUID because Qdrant accepts only integers or UUIDs. Re-running
+  updates points; a document whose chunk count shrinks has its old points
+  deleted by `source_path` filter, so orphans cannot accumulate.
+* **Incremental.** One hash comparison against the manifest per document. The
+  document is re-embedded when its bytes changed, or when the embedding model,
+  dimension or chunker version changed.
+* **Consistency check.** At the end of every run the point count in Qdrant must
+  equal the chunk count in SQLite, or the run fails.
+
+Corpus today: **40 documents → 51 chunks**, token range 100–707 (median 225),
+8 protected regions (fenced blocks and tables), 0 boundary violations.
+
+---
+
+## 4. Payload
+
+`docs/phase-4-corpus.md` §4.2 fields, plus provenance so an answer can cite:
+
+```json
+{
+  "area": "edge", "level": 3, "source_type": "doc", "confidence": "running",
+  "period": "2020–", "artefacts": [], "last_verified": "2026-10-26",
+  "text": "# Artefact — Distributed audio …",
+  "id": "014d792f-28d9-e0a0-610e-7ffc04eea3ad",
+  "source_path": "content/edge/l3/edge-snapcast.md", "source_id": "edge-snapcast",
+  "chunk_index": 0, "title": "…", "claim": "…", "url": "/edge/a/edge-snapcast/",
+  "content_hash": "12fc4fa9…", "embedding_model": "local-hashing-v1",
+  "embedding_dim": 1024, "chunker_version": "blocks-v1", "chunk_tokens": 191,
+  "public": true, "tech": ["Snapcast", "Mopidy", "…"], "tags": ["edge", "audio"]
+}
+```
+
+`area` and `confidence` are the two that matter. The query path filters on them
+*before* it searches, and an answer can qualify unfinished work instead of
+presenting it as delivered.
+
+---
+
+## 5. Freshness audit
+
+```sh
+make corpus-audit                  # human table, exit 1 on findings
+python3 -m corpus.cli audit --json # machine-readable, for Prefect
+```
+
+Findings, in order of seriousness:
+
+| Finding | Meaning |
+|---|---|
+| `stale` | `last_verified` older than `RESUME_MAX_AGE_DAYS` (180). The claim may no longer be true |
+| `due_soon` | inside the warning window (30 days by default). A warning, not a failure |
+| `index_drift` | the source changed, disappeared or was never indexed since it was embedded |
+| `contract_drift` | the collection was built with a different model, dimension or chunker |
+
+`ok` is false when there is a `stale`, `index_drift` or `contract_drift`
+finding. `due_soon` alone keeps the audit green.
+
+`--alert` posts the report to `RESUME_APPRISE_URL` when the audit is not ok.
+The Prefect flow does this by default.
+
+---
+
+## 6. Scheduling (Prefect, existing server, worker only)
+
+`flows/corpus.py` defines three flows. It imports Prefect when present and runs
+as plain Python when it is not, so the CLI never depends on the orchestrator.
+
+On `.13`, which already runs the Prefect server — **do not start a second one**:
+
+```sh
+# once, per deployment
+export RESUME_QDRANT_URL=http://127.0.0.1:6333
+export RESUME_EMBED_PROVIDER=omniroute RESUME_EMBED_MODEL=<approved model> RESUME_EMBED_DIM=<dim>
+prefect deploy flows/corpus.py:sync_corpus    --name corpus-sync
+prefect deploy flows/corpus.py:freshness_audit --name corpus-audit --cron "0 4 * * *"
+```
+
+Triggers required by `docs/phase-4-corpus.md` §4.4:
+
+* **on push touching `content/`** — a deployment trigger on the repository, or
+  `make corpus-sync` from the same post-receive hook that runs `deploy.sh`;
+* **nightly audit** — cron `0 4 * * *`, which alerts through Apprise.
+
+This deployment is **not applied yet**: it needs the runtime host and the
+provider decision in §8. Both are Sam's. The flows themselves are ready.
+
+---
+
+## 7. Qdrant memory at three corpus sizes
+
+Measured with `python3 -m corpus.cli measure`, Qdrant 1.20.0 in a container
+capped at 400 MiB (`mem_limit` = `memswap_limit`), 1024-dimensional vectors,
+real payloads from `content/`, throttled ingest in batches of 250.
+
+| Chunks | Points | Settled RSS | Peak during ingest | Growth | Bytes/chunk | Ingest time |
+|---|---|---|---|---|---|---|
+| 0 | 0 | 88.6 MiB | — | — | — | — |
+| 500 | 500 | 128.3 MiB | 148.5 MiB | +39.7 MiB | 83 KiB | 1.1 s |
+| 5,000 | 5,000 | 137.7 MiB | 305.0 MiB | +49.1 MiB | 10 KiB | 10.9 s |
+| 20,000 | 20,000 | 150.9 MiB | 292.3 MiB | +62.3 MiB | 3 KiB | 43.7 s |
+
+All three sizes are **within the 400 MiB cap**. The peak is the number that
+matters: the cgroup limit is tripped during a bulk write, not after it.
+
+### The finding that changed the design
+
+With Qdrant's **default** settings the container was OOM-killed at exactly the
+cap between 5,000 and 20,000 chunks (`anon-rss: 404204kB`, `exitCode 137`,
+`constraint=CONSTRAINT_MEMCG`). The default keeps the HNSW graph and the payload
+in RAM and holds several segments live during ingest.
+
+The collection is now created memory-frugal, and the pipeline is the only code
+that creates it (`corpus/qdrant.py`):
+
+```json
+{
+  "vectors": {"size": 1024, "distance": "Cosine", "on_disk": false},
+  "hnsw_config": {"m": 8, "ef_construct": 64, "on_disk": true},
+  "on_disk_payload": true,
+  "optimizers_config": {"default_segment_number": 2, "memmap_threshold": 5000, "indexing_threshold": 20000}
+}
+```
+
+Vectors stay in RAM because search reads them on every query. The HNSW graph and
+the payloads go to disk, where the kernel can reclaim those pages instead of
+killing the process. `memmap_threshold` 5,000 makes segments that size mmap;
+`indexing_threshold` 20,000 stops a bulk ingest from building the graph while it
+is still writing. The cost is build time and a little search latency, both
+irrelevant for a corpus of this size and a public site with no query burst.
+
+Practical ceiling: at 1024 dimensions the cap holds 20,000 chunks. If the corpus
+grows past that, set `RESUME_QDRANT_VECTORS_ON_DISK=1`, or reduce the dimension,
+or raise the cap — in that order of preference. Vector memory scales linearly
+with dimension: 1,536-dimension vectors need 50 % more than recorded here.
+
+---
+
+## 8. Embedding provider — still Sam's decision
+
+`reference/stack.md` lists the open item: Gemini free tier, OpenRouter, or
+local. Phase 4 does not wait for it:
+
+* **`omniroute`** is implemented and tested. OpenAI-compatible
+  `/v1/embeddings`, batched, retry with backoff on 429 and 5xx, key from the
+  environment only, no provider SDK, no direct vendor call (AGENTS.md §4.5).
+* **`local-hashing`** is the default for tests, for offline rebuilds and for the
+  memory measurement. It is deterministic feature hashing, **not a semantic
+  model**: it is a placeholder that lets the whole pipeline run without a
+  provider decision, and it must not be used for answers.
+
+The provider, model, dimension and chunker version are stored in the collection
+contract (`data/corpus/collection.json` and the `meta` table). A run whose
+configuration differs is refused:
+
+```
+the collection contract does not match this configuration — refusing to mix
+vectors in one collection (docs/phase-4-corpus.md §4.3).
+```
+
+### Changing the provider (governed migration)
+
+1. Get Sam's decision and record it in `reference/decisions.md`.
+2. Set `RESUME_EMBED_PROVIDER`, `RESUME_EMBED_MODEL`, `RESUME_EMBED_DIM`, and
+   `OMNIROUTE_API_KEY` in `~/.config/environment.d/`.
+3. `make corpus` — a rebuild, because a provider change invalidates every
+   vector. It is one command and takes seconds at this corpus size.
+4. `make corpus-verify` and `make corpus-audit`.
+
+---
+
+## 9. Recovery
+
+Deleting the Qdrant volume or `data/corpus/` is **always recoverable**:
+
+```sh
+docker compose -f deploy/qdrant.yml down -v
+docker compose -f deploy/qdrant.yml up -d
+make corpus          # re-derives every vector and FTS row from content/
+make corpus-verify
+```
+
+There is no state that exists only in the index. `content/` is the source of
+truth; the corpus is a derived artefact and lives in `data/`, which is
+gitignored.
+
+---
+
+## 10. Definition of Done — evidence
+
+| DoD item | Evidence |
+|---|---|
+| One command rebuilds from an empty volume | `make corpus` on an empty Qdrant volume: 40 documents → 51 chunks, 51 points, 1.0 s |
+| Running it twice produces no duplicate vectors | rebuild ×2 → 51 points both times; `sync` with no changes → 0 embedding calls, 0 points sent |
+| No chunk boundary inside a fence, table or front-matter | `corpus/tests/test_chunker.py`; 51/51 stored chunks reproduce from `content/`, 8 protected regions, 0 violations |
+| Every chunk carries all payload fields; missing field fails the run | `corpus/tests/test_payloads.py` (each field removed in turn); 51/51 stored payloads validate; a broken payload aborts before any write (`test_invalid_payload_aborts_before_any_write`) |
+| Payload filtering demonstrably narrows results | `make corpus-demo` — query "distributed audio endpoints": unfiltered returns edge *and* ai documents; `--area edge` returns edge only, `--area ai` returns ai only |
+| Incremental re-embeds only changed documents | edit one of 40 documents → `updated=1 unchanged=39 embedding calls=1 texts=1`; re-run → `calls=0` |
+| Freshness audit runs and reports stale documents | `python3 -m corpus.cli audit` today: 0 findings. `--today 2027-10-05`: 40 stale documents, one row per document with age in days. `--today 2026-11-01`: 0 findings |
+| Qdrant memory at 500 / 5,000 / 20,000 chunks, within cap | §7 table; 128 / 138 / 151 MiB settled, 149 / 305 / 292 MiB peak, cap 400 MiB |
+
+---
+
+## 11. Not done here
+
+* Qdrant is **not deployed to `.13`** and the Prefect deployment is not applied.
+  `deploy/qdrant.yml` is ready; the runtime host is Phase 3's (`AUD-143`) and
+  Sam's to apply.
+* No provider is chosen (§8).
+* Answering questions, the query graph, the streaming console and guardrails are
+  Phase 5 and Phase 6. Phase 4 ends at "retrieval returns the right chunks".

docs/phase-5-ai-overlay.md

@@ -80,3 +80,43 @@ One hand-written TypeScript island, ES modules, no framework, no bundler.
 | Panel refactor needed | Phase 2 §2.4 contract; if it fails, fix the panel contract, do not fork the UI |
 | Streaming stalls behind Caddy | Explicit flush settings; p90 measurement, not a spot check |
 | Answers drift from the site | Citation verification is a node, not a hope |
+
+---
+
+## 5.6 Citations point at the original
+
+An answer cites where it came from, not a curated page standing in for it.
+
+| Source | Cited as |
+|---|---|
+| Gitea repo file | full URL to the file in Gitea |
+| Vault note | full URL to the note in `sam/obsidian-vault` |
+| Graph answer | the note title the node came from, plus its URL |
+| Curated claim | the site page |
+
+A citation whose URL cannot be opened is dropped, not rendered. A claim with no
+traceable origin is not returned at all.
+
+## 5.7 Curated outranks harvested
+
+`content/` is tier 1; everything harvested is tier 2. When the two disagree, the
+tier 1 claim wins and the answer says so plainly: the curated description is
+current; the older text is background.
+
+This is the rule that stops a stale README from quietly overwriting Sam's own
+description of his work.
+
+## 5.8 The live source readout
+
+While the answer streams, a panel shows what is being read, named as the visitor
+would recognise it:
+
+```
+reading · gitea/photo-pipeline/README.md
+        · obsidian/210 AI Resume/Graphify.md
+        · graph · community 42 · vault infrastructure
+```
+
+This is a feature, not a debug view. It answers the question every recruiter has
+and does not ask: is this making things up? Show the working — it is also the most
+convincing demonstration of the system on the page.

flows/corpus.py

@@ -0,0 +1,89 @@
+"""Prefect entry points for the corpus pipeline.
+
+`docs/phase-4-corpus.md` §4.4: *"Prefect deployment on the existing Prefect
+server; worker only, do not deploy a second server. Trigger: git push to
+content/, plus a nightly audit."*
+
+Prefect is optional at runtime. If it is not installed, these functions are
+plain callables and the same code runs straight from the CLI
+(`python3 -m corpus.cli sync`), which is what `make corpus` does. Nothing about
+the pipeline depends on the orchestrator, and the orchestrator depends on
+nothing but the pipeline.
+
+The deployment commands (run on `.13`, which already has the Prefect server) are
+in `docs/phase-4-notes.md` §6. This file is only the work definition.
+"""
+
+from __future__ import annotations
+
+import sys
+
+try:  # pragma: no cover - exercised only where Prefect is installed
+    from prefect import flow as _flow
+    from prefect import get_run_logger as _get_run_logger
+
+    PREFECT_AVAILABLE = True
+except ImportError:  # pragma: no cover
+    PREFECT_AVAILABLE = False
+
+    def _flow(*_args, **_kwargs):
+        def decorator(func):
+            return func
+
+        return decorator
+
+    def _get_run_logger():
+        class _Logger:
+            def info(self, message, *args, **kwargs):
+                print(message, file=sys.stderr)
+
+            warning = error = info
+
+        return _Logger()
+
+
+from corpus import freshness, pipeline  # noqa: E402  (after the Prefect shim)
+from corpus.config import Config  # noqa: E402
+
+
+@_flow(name="corpus-sync", log_prints=True)
+def sync_corpus() -> dict:
+    """Incremental refresh. Runs on every push that touches content/."""
+    logger = _get_run_logger()
+    cfg = Config.from_env()
+    stats = pipeline.run(cfg, mode="sync", log=logger.info)
+    return stats.to_dict()
+
+
+@_flow(name="corpus-rebuild", log_prints=True)
+def rebuild_corpus() -> dict:
+    """From-scratch rebuild. The recovery path: deleting the volume is always safe."""
+    logger = _get_run_logger()
+    cfg = Config.from_env()
+    stats = pipeline.run(cfg, mode="rebuild", log=logger.info)
+    return stats.to_dict()
+
+
+@_flow(name="corpus-audit", log_prints=True)
+def freshness_audit(alert: bool = True) -> dict:
+    """Nightly audit. Reports stale documents and index drift, then alerts."""
+    logger = _get_run_logger()
+    cfg = Config.from_env()
+    report = freshness.audit(cfg, log=logger.info)
+    logger.info(freshness.render_text(report))
+    if alert and not report["ok"]:
+        try:
+            freshness.alert(report, cfg.apprise_url)
+        except Exception as exc:  # noqa: BLE001 - the alert must not fail the audit
+            logger.warning(f"Apprise alert failed: {exc}")
+    return report
+
+
+if __name__ == "__main__":  # pragma: no cover - convenience for a manual run
+    command = sys.argv[1] if len(sys.argv) > 1 else "sync"
+    if command == "audit":
+        print(freshness.render_text(freshness_audit(alert=False)))
+    elif command == "rebuild":
+        print(rebuild_corpus())
+    else:
+        print(sync_corpus())

reference/stack.md

@@ -36,6 +36,25 @@
 | Guardrail (if local) | 300 m |
 | **Total ceiling** | **1.0 GiB / 4 containers** |
 
+### Qdrant memory, measured (2026-10-26, v1.20.0, 400 m cap, 1024 dims)
+
+| Chunks | Settled RSS | Peak during ingest |
+|---|---|---|
+| 500 | 128 MiB | 149 MiB |
+| 5,000 | 138 MiB | 305 MiB |
+| 20,000 | 151 MiB | 292 MiB |
+
+Within the cap at all three sizes, but only with the memory-frugal collection
+settings in `corpus/qdrant.py`: HNSW graph and payloads on disk
+(`hnsw_config.on_disk`, `on_disk_payload`), `m=8`, `ef_construct=64`,
+`memmap_threshold=5000`, `indexing_threshold=20000`. With Qdrant's defaults the
+container is OOM-killed between 5,000 and 20,000 chunks. The pipeline is the only
+code that creates the collection, so those settings cannot drift.
+
+Ceiling: 20,000 chunks at 1024 dimensions. Beyond that, set
+`RESUME_QDRANT_VECTORS_ON_DISK=1`, reduce the dimension, or raise the cap — in
+that order. Detail: `docs/phase-4-notes.md` §7.
+
 Every service sets `mem_limit` **and** `memswap_limit` to the same value so it cannot consume host swap. Bind to `127.0.0.1`; Caddy is the only public listener.
 
 ## Paths
@@ -56,6 +75,6 @@ Every service sets `mem_limit` **and** `memswap_limit` to the same value so it c
 
 | Item | Owner |
 |---|---|
-| Embedding provider (Gemini free tier vs OpenRouter vs local) | Sam — decide before Phase 4 |
+| Embedding provider (Gemini free tier vs OpenRouter vs local) | Sam — Phase 4 shipped provider-agnostic (`corpus/embeddings.py`); the `omniroute` adapter is ready, `local-hashing` is a placeholder, so the decision is still Sam's before Phase 5 answers questions |
 | Repo public/private after launch | Sam — decide after Phase 3 |
 | Phone number on the contact page | Sam |