ai-resume/corpus/freshness.py

"""Freshness audit — the thing that stops the corpus rotting silently.

The risk table in `docs/phase-4-corpus.md` names it: *"Silent staleness… a
stale index is otherwise an invisible quality failure."* So the audit is a first
class command, it runs nightly from Prefect, and it alerts through the existing
Apprise service when it finds something.

It reports four kinds of finding, and they are different problems:

* **stale** — a document whose `last_verified` is older than the horizon. The
  claim on the page may no longer be true.
* **due soon** — inside the warning window. Re-verify before it becomes stale.
* **index drift** — the source changed (or disappeared, or was never indexed)
  since it was embedded. Retrieval is answering from an older world.
* **contract drift** — the collection was built with a different model,
  dimension or chunker than the running configuration.

Exit code is 0 for a clean audit and 1 when findings exist and `--fail-on-findings`
was passed, so the Prefect deployment can alert on the result without parsing
text.
"""

from __future__ import annotations

import datetime
import json
import urllib.request

from . import loader, store as store_module
from .config import CHUNKER_VERSION
from .pipeline import _contract
from .qdrant import Qdrant


def audit(cfg, today: datetime.date = None, log=None) -> dict:
    """Return the audit report as a plain dict (JSON-serialisable)."""
    log = log or (lambda _msg: None)
    today = today or datetime.date.today()
    documents = loader.load_documents(cfg)

    findings = {"stale": [], "due_soon": [], "index_drift": [], "contract_drift": []}

    for doc in documents:
        try:
            verified = datetime.date.fromisoformat(doc.last_verified)
        except ValueError:
            findings["index_drift"].append(
                {"source_path": doc.path, "problem": f"invalid last_verified {doc.last_verified!r}"}
            )
            continue
        age = (today - verified).days
        entry = {
            "source_path": doc.path,
            "title": doc.title,
            "area": doc.area,
            "confidence": doc.confidence,
            "last_verified": doc.last_verified,
            "age_days": age,
        }
        if age > cfg.max_age_days:
            findings["stale"].append(entry)
        elif age > cfg.max_age_days - cfg.warn_days:
            findings["due_soon"].append(entry)

    with store_module.Store(cfg.db_path) as store:
        manifest = store.manifest()
        on_disk = {doc.path: doc for doc in documents}

        for doc in documents:
            row = manifest.get(doc.path)
            if row is None:
                findings["index_drift"].append(
                    {"source_path": doc.path, "problem": "not indexed (run sync)"}
                )
                continue
            if row["content_hash"] != doc.digest:
                findings["index_drift"].append(
                    {"source_path": doc.path, "problem": "source changed since it was embedded"}
                )
            if row["chunker_version"] != CHUNKER_VERSION:
                findings["index_drift"].append(
                    {
                        "source_path": doc.path,
                        "problem": f"chunked with {row['chunker_version']}, current {CHUNKER_VERSION}",
                    }
                )

        for source in sorted(set(manifest) - set(on_disk)):
            findings["index_drift"].append(
                {"source_path": source, "problem": "indexed but no longer in content/"}
            )

        stored_contract = store.get_meta("collection")
        if stored_contract != _contract(cfg):
            findings["contract_drift"].append(
                {"problem": "collection contract does not match this configuration",
                 "stored": stored_contract, "configured": _contract(cfg)}
            )

        qdrant = Qdrant(cfg.qdrant_url, timeout=cfg.timeout)
        reachable = qdrant.health()
        points = None
        if reachable and qdrant.collection_exists(cfg.collection):
            points = qdrant.point_count(cfg.collection)
        elif not reachable:
            findings["contract_drift"].append(
                {"problem": f"Qdrant is not answering at {cfg.qdrant_url}"}
            )

        report = {
            "generated_at": datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
            "today": today.isoformat(),
            "horizon": {
                "max_age_days": cfg.max_age_days,
                "warn_days": cfg.warn_days,
            },
            "corpus": {
                "documents": len(documents),
                "chunks": store.chunk_count(),
                "sources": store.source_count(),
                "collection": cfg.collection,
                "collection_points": points,
                "provider": cfg.provider,
                "model": cfg.model,
                "dim": cfg.dim,
                "chunker_version": CHUNKER_VERSION,
            },
            "findings": findings,
            "summary": {
                "stale": len(findings["stale"]),
                "due_soon": len(findings["due_soon"]),
                "index_drift": len(findings["index_drift"]),
                "contract_drift": len(findings["contract_drift"]),
            },
            # `due_soon` is a warning, not a failure: the document is still
            # inside its verification horizon. Stale, drift and contract
            # problems are failures.
            "ok": not (
                findings["stale"] or findings["index_drift"] or findings["contract_drift"]
            ),
        }
    log(f"audit: {json.dumps(report['summary'], sort_keys=True)}")
    return report


def render_text(report: dict) -> str:
    """Human-readable audit, for a terminal or an Apprise body."""
    lines = []
    corpus = report["corpus"]
    lines.append(
        f"corpus: {corpus['documents']} document(s), {corpus['chunks']} chunk(s), "
        f"{corpus['collection_points']} vector(s) in {corpus['collection']}"
    )
    lines.append(
        f"horizon: last_verified within {report['horizon']['max_age_days']} days "
        f"(warning at {report['horizon']['max_age_days'] - report['horizon']['warn_days']} days)"
    )
    findings = report["findings"]
    for kind, label in (
        ("stale", "STALE — past the verification horizon"),
        ("due_soon", "DUE SOON — verify before it goes stale"),
        ("index_drift", "INDEX DRIFT — the index does not match content/"),
        ("contract_drift", "CONTRACT DRIFT — rebuild required"),
    ):
        rows = findings[kind]
        if not rows:
            continue
        lines.append("")
        lines.append(f"{label} ({len(rows)})")
        for row in rows:
            if "problem" in row:
                lines.append(f"  - {row.get('source_path', '<collection>')}: {row['problem']}")
            else:
                lines.append(
                    f"  - {row['source_path']} ({row['area']}, {row['confidence']}): "
                    f"last_verified {row['last_verified']} ({row['age_days']} days ago)"
                )
    if report["ok"]:
        lines.append("")
        lines.append("OK: nothing stale, no drift." + (" (see warnings above)" if findings["due_soon"] else ""))
    return "\n".join(lines)


def alert(report: dict, apprise_url: str, timeout: float = 10.0) -> bool:
    """POST the audit to the existing Apprise service. No-op without a URL."""
    if not apprise_url:
        return False
    title = "ai-resume corpus audit: findings" if not report["ok"] else "ai-resume corpus audit: OK"
    body = json.dumps({"title": title, "body": render_text(report), "type": "warning" if not report["ok"] else "info"})
    request = urllib.request.Request(
        apprise_url,
        data=body.encode("utf-8"),
        method="POST",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(request, timeout=timeout):
        return True