ratatoskr/scripts/contract_drift_check.py

#!/usr/bin/env python3
"""Drift check: verify each .contract.md with a `prd:` block still matches its source.

Reads every `docs/contracts/**/*.contract.md` that has a `prd:` frontmatter block,
fetches the live issue body and (if specified) the lock-in comment from Gitea,
re-hashes them, and compares to the recorded `body_sha256_16` / `lock_in_sha256_16`.

Exits 0 if all pinned contracts are clean. Exits 1 with a diff report if any contract
has drifted. Skips contracts without a `prd:` block.

Usage:
    python scripts/contract_drift_check.py
    python scripts/contract_drift_check.py --contract docs/contracts/issues/138.contract.md
    python scripts/contract_drift_check.py --json

Requires GITEA_TOKEN in environment. Owner/repo are derived from the `origin` git remote by
default (override with GITEA_OWNER / GITEA_REPO; GITEA_URL defaults to the Gitea host).
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
from pathlib import Path
from typing import Any

import httpx
import yaml

PROJECT_ROOT = Path(__file__).parent.parent
CONTRACTS_GLOB = "docs/contracts/**/*.contract.md"


def _owner_repo_from_git_remote() -> tuple[str, str] | None:
    """Derive (owner, repo) from the `origin` git remote so the drift check targets THIS repo
    by default — instead of a hardcoded repo name that silently checks the WRONG repo for every
    other consumer. Supports ssh (git@host:owner/repo.git) and https (https://host/owner/repo.git)
    Gitea remotes; returns None if it can't resolve."""
    try:
        url = subprocess.run(
            ["git", "-C", str(PROJECT_ROOT), "remote", "get-url", "origin"],
            capture_output=True, text=True, check=True,
        ).stdout.strip()
    except (OSError, subprocess.SubprocessError):
        return None
    m = re.search(r"[:/]([^/:]+)/([^/]+?)(?:\.git)?/?$", url)
    return (m.group(1), m.group(2)) if m else None


def sha16(s: str) -> str:
    return hashlib.sha256(s.encode("utf-8")).hexdigest()[:16]


def load_frontmatter(text: str) -> dict[str, Any] | None:
    if not text.startswith("---\n"):
        return None
    end = text.find("\n---\n", 4)
    if end == -1:
        return None
    return yaml.safe_load(text[4:end])


def fetch_issue_state(token: str, base_url: str, owner: str, repo: str, n: int, comment_id: int | None) -> dict[str, Any]:
    hdr = {"Authorization": f"token {token}"}
    issue_url = f"{base_url}/api/v1/repos/{owner}/{repo}/issues/{n}"
    issue = httpx.get(issue_url, headers=hdr, timeout=15)
    issue.raise_for_status()
    body = issue.json()["body"]
    out: dict[str, Any] = {"body_sha256_16": sha16(body)}
    if comment_id:
        comment = httpx.get(
            f"{base_url}/api/v1/repos/{owner}/{repo}/issues/comments/{comment_id}",
            headers=hdr, timeout=15,
        )
        comment.raise_for_status()
        out["lock_in_sha256_16"] = sha16(comment.json()["body"])
    return out


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--contract", help="check a single contract file (default: all)")
    ap.add_argument("--json", action="store_true", help="emit machine-readable report")
    args = ap.parse_args()

    token = os.environ.get("GITEA_TOKEN", "")
    base_url = os.environ.get("GITEA_URL", "https://gitea.phasefinal.com")
    # Owner/repo default to the `origin` remote so the check targets THIS repo; GITEA_OWNER /
    # GITEA_REPO override when set. (Previously repo defaulted to a hardcoded "Worldtree", which
    # silently checked the WRONG repo for every other consumer unless GITEA_REPO was set in env —
    # a false-drift footgun. Derive it, and fail loud rather than guess.)
    git_remote = _owner_repo_from_git_remote()
    owner = os.environ.get("GITEA_OWNER") or (git_remote[0] if git_remote else None)
    repo = os.environ.get("GITEA_REPO") or (git_remote[1] if git_remote else None)
    if not token:
        print("error: GITEA_TOKEN not set", file=sys.stderr)
        return 2
    if not owner or not repo:
        print("error: could not resolve owner/repo — set GITEA_OWNER/GITEA_REPO or run inside a repo with an 'origin' remote", file=sys.stderr)
        return 2

    if args.contract:
        files = [Path(args.contract).resolve()]
    else:
        files = sorted(PROJECT_ROOT.glob(CONTRACTS_GLOB))

    findings: list[dict[str, Any]] = []
    for path in files:
        text = path.read_text()
        fm = load_frontmatter(text)
        if not fm or "prd" not in fm:
            continue
        prd = fm["prd"]
        n = prd["issue"]
        live = fetch_issue_state(token, base_url, owner, repo, n, prd.get("lock_in_comment_id"))
        body_drift = live["body_sha256_16"] != prd["body_sha256_16"]
        lock_drift = (
            prd.get("lock_in_comment_id") is not None
            and live.get("lock_in_sha256_16") != prd.get("lock_in_sha256_16")
        )
        status = "drift" if body_drift or lock_drift else "clean"
        findings.append({
            "contract": str(path.relative_to(PROJECT_ROOT)),
            "issue": n,
            "status": status,
            "body_drift": body_drift,
            "lock_in_drift": lock_drift,
            "pinned_at": prd.get("pinned_at"),
            "live": live,
            "pinned": {
                "body_sha256_16": prd["body_sha256_16"],
                "lock_in_sha256_16": prd.get("lock_in_sha256_16"),
            },
        })

    if args.json:
        print(json.dumps({"findings": findings}, indent=2))
    else:
        clean = [f for f in findings if f["status"] == "clean"]
        drift = [f for f in findings if f["status"] == "drift"]
        for f in findings:
            mark = "OK" if f["status"] == "clean" else "DRIFT"
            print(f"  [{mark}] {f['contract']} (issue #{f['issue']})")
            if f["body_drift"]:
                print(f"      body: pinned={f['pinned']['body_sha256_16']} live={f['live']['body_sha256_16']}")
            if f["lock_in_drift"]:
                print(f"      lock-in: pinned={f['pinned']['lock_in_sha256_16']} live={f['live'].get('lock_in_sha256_16')}")
        print()
        print(f"{len(clean)} clean, {len(drift)} drifted")

    return 1 if any(f["status"] == "drift" for f in findings) else 0


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