ratatoskr/scripts/canonical_drift.py

#!/usr/bin/env python3
"""
canonical_drift.py — CI gate / manual drift detection for pinned canonical
Corviduo specs.

Reads .corviduo-canonicals.toml at the repo root. For each pin, locates the
canonical at ~/development/<canonical_source>/<canonical_path>, computes the
current SHA-256 (first 16 hex chars), and compares against the pin's
pinned_sha256_16. Also verifies the consumer copy matches the pinned hash.

Reports any drift. Exits non-zero on staleness, divergence, or missing
canonical.

Usage:
  python scripts/canonical_drift.py [--manifest PATH] [--allow-warn]

Exit codes:
  0 — all pins current
  1 — one or more pins stale or consumer-copy diverged
  2 — manifest missing or invalid
  3 — canonical source path missing for one or more pins

Composes with `canonical_sync.py` (which fetches + bumps); this tool is the
read-only verifier suitable for CI gating.
"""
from __future__ import annotations

import argparse
import hashlib
import sys
import tomllib
from pathlib import Path

DEFAULT_MANIFEST = ".corviduo-canonicals.toml"
DEV_ROOT = Path.home() / "development"


def sha256_16(path: Path) -> str:
    """SHA-256 hash of file contents, first 16 hex chars."""
    return hashlib.sha256(path.read_bytes()).hexdigest()[:16]


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Detect drift between pinned canonicals and their sources.",
    )
    ap.add_argument("--manifest", type=Path, default=Path(DEFAULT_MANIFEST))
    ap.add_argument("--allow-warn", action="store_true",
                    help="Treat pins with tolerate_drift=true as warnings only")
    args = ap.parse_args()

    if not args.manifest.exists():
        print(f"error: manifest not found at {args.manifest}", file=sys.stderr)
        return 2

    with args.manifest.open("rb") as f:
        manifest = tomllib.load(f)

    pins = manifest.get("pins", [])
    if not pins:
        print("warning: no pins in manifest", file=sys.stderr)
        return 0

    ok: list[str] = []
    warn: list[str] = []
    stale: list[str] = []
    consumer_drift: list[str] = []
    missing: list[str] = []

    for pin in pins:
        pin_id = pin.get("id", "<unknown>")
        source = pin.get("canonical_source")
        canon_rel = pin.get("canonical_path")
        consumer_rel = pin.get("consumer_path")
        if not all((source, canon_rel, consumer_rel)):
            missing.append(f"{pin_id} (manifest entry incomplete)")
            continue

        canon_path = DEV_ROOT / source / canon_rel
        if not canon_path.exists():
            missing.append(f"{pin_id} (canonical missing: {canon_path})")
            continue

        current = sha256_16(canon_path)
        pinned = pin.get("pinned_sha256_16", "")

        consumer_path = Path(consumer_rel)
        consumer_matches_pin = (
            consumer_path.exists()
            and sha256_16(consumer_path) == pinned
        )

        tolerate = pin.get("tolerate_drift", False)

        if current == pinned and consumer_matches_pin:
            ok.append(pin_id)
        elif current != pinned and tolerate and args.allow_warn:
            warn.append(f"{pin_id} (canonical {pinned} -> {current})")
        elif current != pinned:
            stale.append(f"{pin_id} (canonical {pinned} -> {current})")
        else:
            consumer_drift.append(
                f"{pin_id} (consumer copy diverged from pinned hash)",
            )

    for pin_id in ok:
        print(f"OK         {pin_id}")
    for entry in warn:
        print(f"WARN       {entry}")
    for entry in stale:
        print(f"STALE      {entry}", file=sys.stderr)
    for entry in consumer_drift:
        print(f"DIVERGED   {entry}", file=sys.stderr)
    for entry in missing:
        print(f"MISSING    {entry}", file=sys.stderr)

    if missing:
        return 3
    if stale or consumer_drift:
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())