"""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