sam/ai-resume
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP Phase 4 ingestion: platform guides, exclusion gate, gitea/vault/graph harvest, payload provenance

Browse Source
This commit is contained in:
Paperclip Builder
2026-10-26 22:26:45 +11:00
parent 5c0622c1b1
commit 1e46ec009a
19 changed files with 2952 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

362
corpus/pipeline.py
View File

@@ -1,20 +1,32 @@
"""The pipeline: content/ -> validate -> chunk -> enrich -> embed -> upsert.
"""The ingestion pipeline: estate -> normalise -> 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.
* **rebuild** — drop the collections, empty the store, embed everything. This is
the "one command rebuilds the pool 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.
embedding dimension or chunker version changed are re-chunked and re-embedded.
Everything else is left alone, including the graph when graph.json is byte-for-byte
unchanged.
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.
The order matters. Every payload for every changed document is built and
validated *before* a single vector is written, so an invalid payload stops the
run with an empty diff rather than leaving half an index behind. Embedding is one
flat pass over the changed chunks, batched, so the provider call count is a
direct measure of how much work incremental mode actually did.
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.
Two Qdrant collections, one store:
* `resume_corpus` — one point per chunk of a document, from every platform;
* `resume_graph` — one point per graph node and one per graph edge, joined to the
note it came from by `source_note`.
Neither is ever read by the public site (acceptance test A1): the site serves
pre-rendered files and does not open a socket to Qdrant.
"""
from __future__ import annotations
@@ -25,9 +37,15 @@ import time
import uuid
from dataclasses import asdict, dataclass, field
from . import chunker, embeddings, loader, payloads, store as store_module
from . import chunker, embeddings, exclusions, payloads, store as store_module
from .config import CHUNKER_VERSION
from .qdrant import Qdrant, QdrantError
from .sources import estate as estate_module
from .sources import graph as graph_module
# How many texts one embed call carries. The provider may batch smaller
# internally (OmniRoute does); this bounds memory on the ingest host.
EMBED_BATCH = 128
class PipelineError(RuntimeError):
@@ -61,12 +79,22 @@ class RunStats:
points_sent: int = 0
collection_points: int = 0
store_chunks: int = 0
graph_nodes: int = 0
graph_edges: int = 0
graph_points: int = 0
graph_rebuilt: bool = False
rejections: int = 0
images: int = 0
duration_s: float = 0.0
collection: str = ""
graph_collection: str = ""
provider: str = ""
model: str = ""
dim: int = 0
chunker_version: str = CHUNKER_VERSION
platforms: dict = field(default_factory=dict)
rejection_summary: dict = field(default_factory=dict)
warnings: list = field(default_factory=list)
notes: list = field(default_factory=list)
def summary(self) -> str:
@@ -76,10 +104,18 @@ class RunStats:
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"graph nodes={self.graph_nodes} edges={self.graph_edges} points={self.graph_points} "
f"rebuilt={self.graph_rebuilt}",
f"rejections={self.rejections} image references={self.images}",
f"platforms={json.dumps(self.platforms, sort_keys=True)}",
f"provider={self.provider} model={self.model} dim={self.dim} chunker={self.chunker_version}",
f"collection={self.collection} duration={self.duration_s:.2f}s",
f"collection={self.collection} graph_collection={self.graph_collection} "
f"duration={self.duration_s:.2f}s",
]
if self.rejection_summary:
rows.append(f"rejected: {json.dumps(self.rejection_summary, sort_keys=True)}")
rows.extend(self.notes)
rows.extend(f"warning: {warning}" for warning in self.warnings)
return "\n".join(rows)
def to_dict(self) -> dict:
@@ -87,7 +123,7 @@ class RunStats:
def run(cfg, mode: str = "sync", log=None) -> RunStats:
"""Build or refresh the corpus. Returns what happened; raises on any error."""
"""Build or refresh the pool. 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()
@@ -95,15 +131,26 @@ def run(cfg, mode: str = "sync", log=None) -> RunStats:
stats = RunStats(
mode=mode,
collection=cfg.collection,
graph_collection=cfg.graph_collection,
provider=cfg.provider,
model=cfg.model,
dim=cfg.dim,
chunker_version=cfg.chunker_version,
)
documents = loader.load_documents(cfg)
# 1. Read the estate. Curated content is validated first (loader does it),
# so a broken content tree stops the run before Gitea is touched.
harvest = estate_module.harvest(cfg, log=log)
documents = harvest.documents
stats.documents = len(documents)
log(f"validated {len(documents)} document(s)")
stats.rejections = len(harvest.rejections)
stats.images = len(harvest.images)
stats.rejection_summary = harvest.rejection_summary()
stats.warnings = list(harvest.warnings)
stats.platforms = {
name: {"documents": row.get("documents", 0)} for name, row in harvest.sources.items()
}
log(f"estate: {len(documents)} document(s), {stats.rejections} rejection(s), {stats.images} image reference(s)")
provider = embeddings.build_provider(cfg)
qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout)
@@ -116,8 +163,7 @@ def run(cfg, mode: str = "sync", log=None) -> RunStats:
with store_module.Store(cfg.db_path) as store:
contract = _contract(cfg)
mode = _check_contract(store, qdrant, cfg, contract, mode)
_ensure_collection(qdrant, cfg)
_ensure_collection(qdrant, cfg.collection, cfg)
manifest = store.manifest()
changed, removed = _classify(documents, manifest, cfg, mode)
@@ -133,33 +179,46 @@ def run(cfg, mode: str = "sync", log=None) -> RunStats:
qdrant.delete_by_source(cfg.collection, path)
log(f"removed {path}")
# 2. Build and validate every payload before any write.
prepared = [] # [(doc, payloads)]
for doc in changed:
payloads_for_doc = _prepare(doc, cfg, manifest)
stats.chunks_written += len(payloads_for_doc)
prepared.append((doc, _prepare(doc, cfg)))
all_payloads = [payload for _, rows in prepared for payload in rows]
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"
)
# 3. Embed in one flat, batched pass. This is the number the DoD uses to
# prove incremental mode: it is proportional to changed chunks only.
vectors = _embed_all(provider, [payload["text"] for payload in all_payloads])
if len(vectors) != len(all_payloads):
raise PipelineError(
f"provider returned {len(vectors)} vectors for {len(all_payloads)} chunks"
)
cursor = 0
platform_changes = stats.platforms
for doc, rows in prepared:
points = []
for payload, vector in zip(payloads_for_doc, vectors):
for payload in rows:
vector = vectors[cursor]
cursor += 1
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.
# Remove what this document used to have, 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)
store.write_document(doc, rows)
stats.chunks_written += len(rows)
bucket = platform_changes.setdefault(doc.platform, {"documents": 0})
bucket["added" if doc.path in manifest else "updated"] = (
bucket.get("added" if doc.path in manifest else "updated", 0) + 1
)
if doc.path in manifest:
stats.updated += 1
log(f"updated {doc.path} ({len(points)} chunk(s))")
@@ -170,24 +229,47 @@ def run(cfg, mode: str = "sync", log=None) -> RunStats:
provider_stats = provider.stats()
stats.embed_calls = provider_stats["calls"]
stats.texts_embedded = provider_stats["texts"]
# 4. Graph: its own collection, rebuilt only when graph.json changed.
graph_stats = _sync_graph(qdrant, store, cfg, harvest, contract, mode, provider, log)
stats.graph_nodes = graph_stats["nodes"]
stats.graph_edges = graph_stats["edges"]
stats.graph_points = graph_stats["points"]
stats.graph_rebuilt = graph_stats["rebuilt"]
stats.notes.extend(graph_stats["notes"])
# 5. Consistency and the credential scan. Both are checks, not hopes.
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`."
)
_scan_pool(store)
_scan_graph(store)
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)
store.set_meta("sources", harvest.sources)
store.set_meta("rejections", harvest.rejections[:200])
store.set_meta("images", harvest.images[:500])
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, harvest)
stats.duration_s = round(time.time() - started, 3)
return stats
# -- planning -------------------------------------------------------------
def _contract(cfg) -> dict:
"""What this collection *is*. Recorded so a change is a migration, not drift."""
return {
@@ -200,13 +282,24 @@ def _contract(cfg) -> dict:
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()
qdrant.drop_collection(cfg.graph_collection)
if store.schema_stale:
store.rebuild_schema()
else:
store.reset()
store.set_meta("schema_version", store_module.SCHEMA_VERSION)
return mode
if store.schema_stale:
raise PipelineError(
"the corpus database was written by an older payload shape and has no "
"provenance columns. It is derived data: run `python3 -m corpus.cli rebuild`."
)
stored = store.get_meta("collection")
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 "
@@ -217,7 +310,7 @@ def _check_contract(store, qdrant, cfg, contract: dict, mode: str) -> str:
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"
"mix vectors in one collection (docs/phase-4-corpus.md §4).\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 "
@@ -226,20 +319,18 @@ def _check_contract(store, qdrant, cfg, contract: dict, mode: str) -> str:
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
)
def _ensure_collection(qdrant: Qdrant, name: str, cfg) -> None:
if not qdrant.collection_exists(name):
qdrant.create_collection(name, cfg.dim, vectors_on_disk=cfg.vectors_on_disk)
return
info = qdrant.collection_info(cfg.collection)
info = qdrant.collection_info(name)
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"collection {name!r} has vectors of size {size} but this run "
f"is configured for {cfg.dim}. That is a governed migration: rebuild."
)
@@ -268,10 +359,13 @@ def _classify(documents: list, manifest: dict, cfg, mode: str):
return changed, removed
def _prepare(doc, cfg, manifest: dict) -> list:
def _prepare(doc, cfg) -> list:
"""Chunk one document and build validated payloads. No network, no writes."""
try:
chunks = chunker.chunk_document(doc.body)
if getattr(doc, "kind", "markdown") == "code":
chunks = chunker.chunk_code(doc.body)
else:
chunks = chunker.chunk_document(doc.body)
except chunker.ChunkError as exc:
raise PipelineError(f"{doc.path}: {exc}") from exc
if not chunks:
@@ -290,14 +384,118 @@ def _prepare(doc, cfg, manifest: dict) -> list:
return out
def _write_collection_file(cfg, store, stats: RunStats) -> None:
"""Mirror the contract to data/corpus/collection.json for the operator."""
def _embed_all(provider, texts: list) -> list:
out = []
for start in range(0, len(texts), EMBED_BATCH):
out.extend(provider.embed(texts[start : start + EMBED_BATCH]))
return out
# -- graph ----------------------------------------------------------------
def _sync_graph(qdrant, store, cfg, harvest, contract, mode, provider, log) -> dict:
graph = harvest.graph
stored = store.get_meta("graph") or {}
notes = []
if graph is None:
if mode == "rebuild":
qdrant.drop_collection(cfg.graph_collection)
store.write_graph([], [])
store.set_meta("graph", {})
notes.append("graph: no graph.json was available; the graph collection is empty")
return {"nodes": 0, "edges": 0, "points": 0, "rebuilt": True, "notes": notes}
counts = store.graph_counts()
notes.append(
f"graph: graph.json was not available this run; kept the previous graph "
f"({counts['nodes']} nodes, {counts['edges']} edges)"
)
return {**counts, "points": counts["nodes"] + counts["edges"], "rebuilt": False, "notes": notes}
graph_contract = {
"digest": graph.digest,
"source_path": graph.source_path,
"provider": contract["provider"],
"model": contract["model"],
"dim": contract["dim"],
}
rebuild = mode == "rebuild" or stored != graph_contract
if not rebuild:
counts = store.graph_counts()
return {
**counts,
"points": counts["nodes"] + counts["edges"],
"rebuilt": False,
"notes": [f"graph: unchanged ({graph.digest[:12]}), not re-embedded"],
}
if not qdrant.collection_exists(cfg.graph_collection):
_ensure_collection(qdrant, cfg.graph_collection, cfg)
else:
qdrant.drop_collection(cfg.graph_collection)
qdrant.create_collection(cfg.graph_collection, cfg.dim, vectors_on_disk=cfg.vectors_on_disk)
rows = [("node", row) for row in graph.nodes] + [("edge", row) for row in graph.edges]
payloads_for_graph = [graph_module.point_payload(row, kind) for kind, row in rows]
vectors = _embed_all(provider, [payload["text"] for payload in payloads_for_graph])
points = [
{"id": row["point_id"], "vector": vector, "payload": payload}
for (_, row), vector, payload in zip(rows, vectors, payloads_for_graph)
]
sent = qdrant.upsert(cfg.graph_collection, points)
store.write_graph(graph.nodes, graph.edges)
store.set_meta("graph", graph_contract)
log(f"graph: rebuilt {len(graph.nodes)} node(s) and {len(graph.edges)} edge(s), {sent} point(s)")
return {
"nodes": len(graph.nodes),
"edges": len(graph.edges),
"points": sent,
"rebuilt": True,
"notes": notes,
}
# -- checks ---------------------------------------------------------------
def _scan_pool(store) -> None:
"""Search the finished pool for credential-shaped content. Fail, do not warn."""
hits = []
for source_path, chunk_index, text in store.pool_texts():
for kind, line, _excerpt in exclusions.scan_text(text):
hits.append((source_path, chunk_index, kind, line))
if hits:
detail = ", ".join(f"{path}#chunk-{index} ({kind} line {line})" for path, index, kind, line in hits[:10])
raise PipelineError(
f"credential-shaped content reached the pool in {len(hits)} place(s): {detail}. "
"The run is failed rather than published; fix the source or the exclusion rule."
)
def _scan_graph(store) -> None:
rows = store.db.execute("SELECT id, text FROM graph_nodes UNION ALL SELECT id, text FROM graph_edges")
hits = []
for row in rows:
for kind, line, _excerpt in exclusions.scan_text(row["text"]):
hits.append((row["id"], kind, line))
if hits:
detail = ", ".join(f"{node} ({kind} line {line})" for node, kind, line in hits[:10])
raise PipelineError(f"credential-shaped content reached the graph in {len(hits)} place(s): {detail}")
def _write_collection_file(cfg, store, stats: RunStats, harvest) -> None:
"""Mirror the contract and the last run to data/corpus/collection.json."""
body = {
"collection": cfg.collection,
"contract": _contract(cfg),
"qdrant_url": cfg.qdrant_url,
"chunks": store.chunk_count(),
"sources": store.source_count(),
"graph_collection": cfg.graph_collection,
"graph": store.graph_counts(),
"platforms": harvest.sources,
"rejections": harvest.rejection_summary(),
"images": len(harvest.images),
"last_run": store.get_meta("last_run"),
}
with open(cfg.meta_path, "w", encoding="utf-8") as handle:
@@ -309,10 +507,11 @@ def _write_collection_file(cfg, store, stats: RunStats) -> None:
def verify(cfg, log=None) -> list:
"""Return a list of problems. Empty means the index agrees with content/."""
"""Return a list of problems. Empty means the index agrees with the sources."""
log = log or (lambda _msg: None)
problems = []
documents = loader.load_documents(cfg)
harvest = estate_module.harvest(cfg)
documents = harvest.documents
qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout)
if not qdrant.health():
return [f"Qdrant is not answering at {cfg.qdrant_url}"]
@@ -342,6 +541,7 @@ def verify(cfg, log=None) -> list:
except payloads.PayloadError as exc:
problems.append(str(exc))
on_disk = {doc.path: doc for doc in documents}
for doc in documents:
payload_rows = by_source.get(doc.path, [])
if not payload_rows:
@@ -356,18 +556,48 @@ def verify(cfg, log=None) -> list:
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)")
for source in sorted(set(by_source) - set(on_disk)):
problems.append(f"{source}: indexed but no longer in content/ or the estate (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"
)
# Graph: every row must resolve to a note, and Qdrant must agree.
counts = store.graph_counts()
graph_meta = store.get_meta("graph") or {}
if harvest.graph is not None:
if not graph_meta:
problems.append("graph: graph.json is available but the graph was never indexed")
if qdrant.collection_exists(cfg.graph_collection):
graph_points = qdrant.point_count(cfg.graph_collection)
if graph_points != counts["nodes"] + counts["edges"]:
problems.append(
f"graph: {graph_points} point(s) in Qdrant but "
f"{counts['nodes'] + counts['edges']} row(s) in SQLite"
)
elif counts["nodes"] or counts["edges"]:
problems.append(f"graph: collection {cfg.graph_collection!r} does not exist")
for row in store.db.execute("SELECT id, source_note, origin_url FROM graph_nodes LIMIT 5"):
if not row["source_note"] or not row["origin_url"]:
problems.append(f"graph node {row['id']}: no source note or origin")
missing_notes = [
row["id"]
for row in store.db.execute(
"SELECT id FROM graph_nodes WHERE source_note = '' OR origin_url = '' LIMIT 10"
)
]
if missing_notes:
problems.append(f"graph: {len(missing_notes)} node(s) carry no source note")
log(
f"verify: {store.chunk_count()} chunk(s) across {store.source_count()} source(s), "
f"{total_points} vector(s) in Qdrant"
f"{total_points} vector(s) in Qdrant, {counts['nodes']} node(s), {counts['edges']} edge(s)"
)
for warning in harvest.warnings:
problems.append(f"estate warning: {warning}")
return problems
@@ -391,6 +621,8 @@ def _hit(row: dict) -> dict:
return {
"score": round(row["score"], 4),
"area": payload.get("area"),
"platform": payload.get("platform"),
"tier": payload.get("tier"),
"confidence": payload.get("confidence"),
"title": payload.get("title"),
"url": payload.get("url"),
@@ -400,7 +632,7 @@ def _hit(row: dict) -> dict:
}
def hybrid_search(cfg, query: str, area: str = None, limit: int = 5) -> list:
def hybrid_search(cfg, query: str, area: str = None, limit: int = 5, platform: str = None, tier: int = None) -> 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
@@ -410,11 +642,15 @@ def hybrid_search(cfg, query: str, area: str = None, limit: int = 5) -> list:
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)):
for rank, hit in enumerate(
qdrant.search(cfg.collection, vector, limit=limit * 4, area=area, platform=platform, tier=tier)
):
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)):
for rank, hit in enumerate(
store.keyword_search(query, area=area, platform=platform, tier=tier, 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)
@@ -424,3 +660,21 @@ def hybrid_search(cfg, query: str, area: str = None, limit: int = 5) -> list:
row["title"] = payload.get("title")
row["url"] = payload.get("url")
return ordered
def scan_pool(cfg) -> dict:
"""Run the credential scan without writing anything. Used by `make corpus-scan`."""
with store_module.Store(cfg.db_path) as store:
hits = []
texts = store.pool_texts()
for source_path, chunk_index, text in texts:
for kind, line, _excerpt in exclusions.scan_text(text):
hits.append(
{"source_path": source_path, "chunk": chunk_index, "kind": kind, "line": line}
)
try:
_scan_graph(store)
except PipelineError as exc:
hits.append({"source_path": "graph", "kind": "credential-shaped content", "detail": str(exc)})
chunks = len(texts)
return {"ok": not hits, "hits": hits, "chunks": chunks}