Satisfies docs/phase-4-corpus.md. content/ stays the single source of truth:
one pass over it writes both indexes, so the static pages and the retrieval
corpus cannot drift.
- corpus/: stdlib-only pipeline. validate -> chunk -> enrich -> embed -> upsert.
Chunking is block-based, so a boundary cannot fall inside a fenced block, a
table or front-matter (verified structurally, not by inspection).
- Deterministic ids: sha256(source_path + chunk_index + content_hash), rendered
as a UUID. Re-running updates points; re-chunking deletes the old ones by
source filter, so duplicates and orphans are both impossible.
- Incremental by content hash against a SQLite manifest. A provider, dimension
or chunker change is refused instead of mixed: it needs a rebuild.
- Payloads per section 4.2 plus provenance, validated before any vector is
written; a missing field fails the run and names the document and chunk.
- SQLite FTS5 over chunks (the keyword half of hybrid retrieval, decision D10)
with area/confidence filters and a `terms` column so front-matter tech and
tags are findable.
- Freshness audit: stale documents, due soon, index drift, contract drift;
optional Apprise alert; exit code for Prefect.
- Embeddings behind one interface: omniroute (OpenAI-compatible, retries, key
from the environment only) and local-hashing (offline placeholder). The
provider decision is still Sam's; the pipeline does not depend on it.
- deploy/qdrant.yml: mem_limit = memswap_limit = 400m, loopback only.
- docs/phase-4-notes.md: operator manual, DoD evidence, measured memory.
- 76 unit tests, no container and no network.
Measured, Qdrant 1.20.0 under a 400 m cap at 1024 dimensions: 500 / 5,000 /
20,000 chunks settle at 128 / 138 / 151 MiB, peaking at 149 / 305 / 292 MiB
during ingest. Qdrant's default settings are OOM-killed between 5,000 and
20,000, so the collection is created with HNSW and payloads on disk; recorded in
reference/stack.md with the ceiling and the escalation order.
Note: the `make corpus*` targets were swept into the Phase 3 deploy commit
24590cc by concurrent work in this tree; the Makefile content is correct, and
this commit contains the pipeline they call.
Co-Authored-By: Paperclip <noreply@paperclip.ing>
187 lines
5.8 KiB
Python
187 lines
5.8 KiB
Python
"""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()
|