diff --git a/README.md b/README.md index af1507a..d2fa70e 100644 --- a/README.md +++ b/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 diff --git a/corpus/__init__.py b/corpus/__init__.py new file mode 100644 index 0000000..d47c4c5 --- /dev/null +++ b/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"] diff --git a/corpus/__main__.py b/corpus/__main__.py new file mode 100644 index 0000000..074b87a --- /dev/null +++ b/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()) diff --git a/corpus/chunker.py b/corpus/chunker.py new file mode 100644 index 0000000..f6ac30c --- /dev/null +++ b/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})" + ) diff --git a/corpus/cli.py b/corpus/cli.py new file mode 100644 index 0000000..22f70c5 --- /dev/null +++ b/corpus/cli.py @@ -0,0 +1,257 @@ +"""Command line entry point: `python3 -m corpus.cli `. + +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()) diff --git a/corpus/config.py b/corpus/config.py new file mode 100644 index 0000000..163305a --- /dev/null +++ b/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__))) diff --git a/corpus/embeddings.py b/corpus/embeddings.py new file mode 100644 index 0000000..2e16a92 --- /dev/null +++ b/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 diff --git a/corpus/freshness.py b/corpus/freshness.py new file mode 100644 index 0000000..682acad --- /dev/null +++ b/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', '')}: {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 diff --git a/corpus/loader.py b/corpus/loader.py new file mode 100644 index 0000000..a8b6a65 --- /dev/null +++ b/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 diff --git a/corpus/measure.py b/corpus/measure.py new file mode 100644 index 0000000..a07844e --- /dev/null +++ b/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))) diff --git a/corpus/payloads.py b/corpus/payloads.py new file mode 100644 index 0000000..8fe1aca --- /dev/null +++ b/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 diff --git a/corpus/pipeline.py b/corpus/pipeline.py new file mode 100644 index 0000000..ae79f46 --- /dev/null +++ b/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 diff --git a/corpus/qdrant.py b/corpus/qdrant.py new file mode 100644 index 0000000..31370eb --- /dev/null +++ b/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", "") + 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 diff --git a/corpus/store.py b/corpus/store.py new file mode 100644 index 0000000..c163792 --- /dev/null +++ b/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, '', '', '…', 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) diff --git a/corpus/tests/__init__.py b/corpus/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/corpus/tests/fake_qdrant.py b/corpus/tests/fake_qdrant.py new file mode 100644 index 0000000..92a2fd9 --- /dev/null +++ b/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 diff --git a/corpus/tests/support.py b/corpus/tests/support.py new file mode 100644 index 0000000..8fb41f6 --- /dev/null +++ b/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() diff --git a/corpus/tests/test_chunker.py b/corpus/tests/test_chunker.py new file mode 100644 index 0000000..ca9e3c8 --- /dev/null +++ b/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() diff --git a/corpus/tests/test_embeddings.py b/corpus/tests/test_embeddings.py new file mode 100644 index 0000000..5ea837d --- /dev/null +++ b/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() diff --git a/corpus/tests/test_freshness.py b/corpus/tests/test_freshness.py new file mode 100644 index 0000000..3c5bb92 --- /dev/null +++ b/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() diff --git a/corpus/tests/test_payloads.py b/corpus/tests/test_payloads.py new file mode 100644 index 0000000..2e86087 --- /dev/null +++ b/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() diff --git a/corpus/tests/test_pipeline.py b/corpus/tests/test_pipeline.py new file mode 100644 index 0000000..a0d6070 --- /dev/null +++ b/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() diff --git a/corpus/tests/test_qdrant_client.py b/corpus/tests/test_qdrant_client.py new file mode 100644 index 0000000..ffcb19e --- /dev/null +++ b/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() diff --git a/corpus/tests/test_store.py b/corpus/tests/test_store.py new file mode 100644 index 0000000..dab39cd --- /dev/null +++ b/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() diff --git a/deploy/qdrant.yml b/deploy/qdrant.yml new file mode 100644 index 0000000..dea7141 --- /dev/null +++ b/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 ' RESUME_EMBED_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". diff --git a/flows/corpus.py b/flows/corpus.py new file mode 100644 index 0000000..9884598 --- /dev/null +++ b/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()) diff --git a/reference/stack.md b/reference/stack.md index d7b7fb7..dc544de 100644 --- a/reference/stack.md +++ b/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 |