check_guardrail.py

#!/usr/bin/env python3
"""check_guardrail.py — pre-commit enforcement for the guardrail rule.

The guardrail rule (<workspace>/.claude/rules/guardrail.md) says: don't build
infrastructure that requires remembering to use it. This script enforces
that rule mechanically at commit time.

Three layers, in order:

  1. STRUCTURAL CHECKS (instant, free):
     - New .md rule files under .claude/rules/ MUST have a LOAD MARKER:
       line so they appear in the load banner. No silent rules.
     - New top-level .md files at vault root are blocked unless in the
       allowlist — stops the "let me just drop a doc at root" pattern.

  2. LLM JUDGE (the real teeth, ~2-10 sec per commit, ~1¢):
     The staged diff is sent to Claude Haiku 4.5 via OpenRouter. The
     model judges: does this diff introduce notebook-shaped infrastructure
     (discipline-dependent, requires remembering to use)? If yes → block.
     This is the layer that catches notebook patterns the structural
     checks can't see — natural language nuance, dashboards that need
     clicking, skills that require explicit invocation, etc.

Failure modes are explicit (changed 2026-07-01 after hook evaluation):
  - LLM unavailable / key missing / malformed response → ALLOW with a loud
    warning + a logged fail-open event. Rationale: fail-closed meant no wifi
    = no committing, and enforcement that holds commits hostage gets bypassed
    wholesale, which is worse than one unjudged commit. The CI backstop is
    the layer that catches what a fail-open lets through.

Every judge verdict (and every fail-open) is appended as one JSON line to
logs/guardrail-judge.verdicts.log so the judge's real catch-rate can be
evaluated on data instead of faith.

Exit 0: no violations.
Exit 1: violations; commit blocked. Override with `git commit --no-verify`
if you genuinely need to bypass — but each override is one more notebook.

CLI:
  python3 check_guardrail.py                          # staged diff (pre-commit)
  python3 check_guardrail.py --no-llm                 # skip LLM (debug / offline)
  python3 check_guardrail.py --diff-range A..B        # judge a commit range (CI backstop)
  python3 check_guardrail.py --diff-range A..B --fail-closed
      # CI mode: judge unavailability is a FAILURE, not a fail-open — a red
      # backstop that can't judge must be visible, it doesn't block commits.
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
import urllib.error
import urllib.request
from pathlib import Path

# Repo root derived from this script's location (os/_infra/cos/…) —
# works locally and on CI runners where the checkout path differs.
WORKSPACE = Path(__file__).resolve().parents[3]

# Files allowed at the vault root. Everything else is blocked from being
# added there — the agent must justify why a new top-level doc belongs.
ROOT_MD_ALLOWLIST = {
    "AGENTS.md",
    "CLAUDE.md",
    "Dashboard.md",
    "README.md",
    "LICENSE.md",
    "PROJECTS.md",
    # Example allowlist — front-door docs that legitimately describe process:
    "Operating Model.md",
    "Review Cadence.md",
}

# LLM judge configuration. Uses OpenRouter so we control cost and model
# selection. Claude Haiku 4.5 chosen for: low cost (~$0.001-0.002 per
# commit), strong instruction following, reliable JSON output.
LLM_MODEL = "anthropic/claude-haiku-4.5"
LLM_BASE_URL = "https://openrouter.ai/api/v1"
LLM_API_KEY_ENV = "OPENROUTER_API_KEY"
LLM_TIMEOUT_SECONDS = 60
# Cap the diff sent to the LLM so cost doesn't run away on massive renames.
MAX_DIFF_CHARS = 80000
# One JSON line per judge verdict / fail-open event. *.log is gitignored.
VERDICT_LOG = Path(__file__).resolve().parent / "logs" / "guardrail-judge.verdicts.log"


def _log_verdict(event: str, verdict: str, reason: str, staged: list[tuple[str, str]],
                 diff_chars: int) -> None:
    """Append one JSONL record. Never let logging break the hook."""
    try:
        from datetime import datetime, timezone
        VERDICT_LOG.parent.mkdir(parents=True, exist_ok=True)
        rec = {
            "stamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
            "event": event,  # "judged" | "fail-open"
            "verdict": verdict,
            "reason": reason[:300],
            "n_files": len(staged),
            "files": [p for _, p in staged][:20],
            "diff_chars": diff_chars,
        }
        with VERDICT_LOG.open("a", encoding="utf-8") as f:
            f.write(json.dumps(rec) + "\n")
    except Exception:
        pass


# CI mode state, set from CLI in main(). None = judge the staged diff.
_DIFF_RANGE: str | None = None
_FAIL_CLOSED = False


def _diff_cmd(extra: list[str]) -> list[str]:
    if _DIFF_RANGE:
        return ["git", "diff", *extra, _DIFF_RANGE]
    return ["git", "diff", "--cached", *extra]


def _git_status() -> list[tuple[str, str]]:
    """Return list of (status, path) for the judged diff (staged, or --diff-range).
    Status one of: A, M, R, D, C."""
    try:
        result = subprocess.run(
            _diff_cmd(["--name-status"]),
            capture_output=True, text=True, check=True, cwd=WORKSPACE,
        )
    except subprocess.CalledProcessError as e:
        print(f"git diff --cached --name-status failed: {e}", file=sys.stderr)
        sys.exit(1)
    out: list[tuple[str, str]] = []
    for line in result.stdout.splitlines():
        if not line.strip():
            continue
        parts = line.split("\t")
        status = parts[0][:1]  # 'A100' for renames, take first char
        path = parts[-1]
        out.append((status, path))
    return out


def _full_staged_diff() -> str:
    try:
        result = subprocess.run(
            _diff_cmd([]),
            capture_output=True, text=True, check=True, cwd=WORKSPACE,
        )
    except subprocess.CalledProcessError:
        return ""
    return result.stdout


# -- Structural checks --------------------------------------------------------

def check_new_rule_has_load_marker(staged: list[tuple[str, str]]) -> list[str]:
    violations: list[str] = []
    for status, path in staged:
        if status not in ("A", "R", "C"):
            continue
        if not path.startswith(".claude/rules/"):
            continue
        if not path.endswith(".md"):
            continue
        name = Path(path).name
        if name.startswith("_"):
            continue
        full = WORKSPACE / path
        if not full.exists():
            continue
        text = full.read_text(encoding="utf-8", errors="replace")
        if "LOAD MARKER:" not in text:
            violations.append(
                f"  - {path}\n"
                f"      New rule file is missing a 'LOAD MARKER:' line. Every rule\n"
                f"      must declare its emoji + name so it appears in the load banner."
            )
    return violations


def check_root_md_allowlist(staged: list[tuple[str, str]]) -> list[str]:
    violations: list[str] = []
    for status, path in staged:
        if status not in ("A",):
            continue
        if "/" in path:
            continue
        if not path.endswith(".md"):
            continue
        if path in ROOT_MD_ALLOWLIST:
            continue
        violations.append(
            f"  - {path}\n"
            f"      New top-level .md at vault root. Either move it into a\n"
            f"      lane (00_inbox, 01_business, 02_projects, etc.) or add it\n"
            f"      to ROOT_MD_ALLOWLIST in check_guardrail.py with a reason."
        )
    return violations


# -- LLM judge ----------------------------------------------------------------

def check_llm_judge(staged: list[tuple[str, str]]) -> list[str]:
    if not staged:
        return []
    diff_text = _full_staged_diff()
    if not diff_text.strip():
        return []
    truncated = False
    if len(diff_text) > MAX_DIFF_CHARS:
        diff_text = diff_text[:MAX_DIFF_CHARS]
        truncated = True

    api_key = os.environ.get(LLM_API_KEY_ENV)
    if not api_key:
        if _FAIL_CLOSED:
            return [
                f"  - guardrail AI judge: {LLM_API_KEY_ENV} not set (fail-closed mode)\n"
                f"      In CI this means the repo secret is missing or not exposed\n"
                f"      to this workflow. A backstop that can't judge must be red."
            ]
        _log_verdict("fail-open", "unjudged", f"{LLM_API_KEY_ENV} not set",
                     staged, len(diff_text))
        print(
            f"⚠  guardrail AI judge SKIPPED: {LLM_API_KEY_ENV} not set — commit "
            f"allowed UNJUDGED (logged). CI backstop is the safety net.",
            file=sys.stderr,
        )
        return []

    prompt = _judge_prompt(diff_text, truncated)

    try:
        verdict, reason = _call_llm_judge(prompt, api_key)
    except Exception as e:
        if _FAIL_CLOSED:
            return [
                f"  - guardrail AI judge: call failed ({e}) (fail-closed mode)\n"
                f"      Network/model problem in CI. A backstop that can't judge\n"
                f"      must be red — re-run the workflow once the cause is fixed."
            ]
        _log_verdict("fail-open", "unjudged", f"call failed: {e}", staged, len(diff_text))
        print(
            f"⚠  guardrail AI judge SKIPPED: call failed ({e}) — commit allowed "
            f"UNJUDGED (logged). CI backstop is the safety net.",
            file=sys.stderr,
        )
        return []

    _log_verdict("judged", verdict, reason, staged, len(diff_text))
    if verdict == "notebook":
        return [
            f"  - guardrail AI judge: this diff introduces notebook-shaped infra\n"
            f"      Reason: {reason}\n"
            f"      Spec: <workspace>/.claude/rules/guardrail.md\n"
            f"      Either restructure so it fires automatically, or remove the\n"
            f"      part the judge flagged. Override (rare): git commit --no-verify"
        ]
    if verdict == "uncertain":
        print(
            f"⚠  guardrail AI judge: uncertain verdict — '{reason}' (allowed)",
            file=sys.stderr,
        )
    return []


def _judge_prompt(diff_text: str, truncated: bool) -> str:
    trunc_note = "\n[Note: diff truncated for cost control]\n" if truncated else ""
    return (
        "You are evaluating a git diff against the GUARDRAIL rule of a private "
        "personal-workspace repo.\n\n"
        "THE RULE (verbatim): Don't build infrastructure that requires remembering "
        "to use it. Either make it fire automatically (guardrail), or don't "
        "build it.\n\n"
        "THE CORE TEST: does the text try to GOVERN FUTURE BEHAVIOR (instruct a "
        "future reader/agent to act a certain way, with nothing enforcing it), or "
        "does it RECORD past decisions, findings, or history? Governing without "
        "enforcement = notebook. Recording = allowed. Apply this test before "
        "anything else.\n\n"
        "Examples of NOTEBOOK-shaped infra (BLOCK these):\n"
        "  - A new prose instruction with ongoing operational force — 'remember "
        "to do X', 'always check Y', 'every new project must Z' — aimed at "
        "future readers/agents, with no enforcement mechanism behind it.\n"
        "  - A new rule file in .claude/rules/ that loads but has no enforcement "
        "mechanism backing it.\n"
        "  - A new skill or tool that requires explicit user invocation where the "
        "invocation could be forgotten and there's no fallback.\n"
        "  - A new dashboard or UI that requires manual clicks to surface info.\n"
        "  - A new 'guidelines' or 'best practices' doc with no automation behind "
        "it.\n"
        "  - A new convention added to CLAUDE.md / AGENTS.md that asks the reader "
        "to behave a certain way without a check enforcing it.\n\n"
        "Examples of GUARDRAIL-shaped infra or harmless content (ALLOW these):\n"
        "  - Records of decisions, findings, or history: planning docs, decision "
        "logs ('DECIDED: ...'), audits, post-mortems, changelogs, review "
        "findings, meeting notes, research notes, inventories, session handoffs. "
        "These document what WAS decided or found — history, not live "
        "instructions. A decision record saying an enforcement mechanism will be "
        "built later is still a record, not the mechanism — do not block the "
        "record for the mechanism not existing yet.\n"
        "  - A new pre-commit hook, cron, LaunchAgent, daemon.\n"
        "  - A new auto-loaded rule WITH a corresponding enforcement script.\n"
        "    IMPORTANT: a rule file (.md) and its enforcement script (.py, .sh,\n"
        "    hook, wrapper, validator) are often committed together in a single\n"
        "    diff. Before flagging a rule as notebook, scan the ENTIRE diff for\n"
        "    new files that look like enforcement (`check_*.py`, `*-wrap.py`,\n"
        "    `*-validator.*`, a pre-commit hook addition, a wrapper script\n"
        "    around an existing tool). If found, the rule IS backed by enforcement\n"
        "    and the verdict is guardrail.\n"
        "  - A new validated config file or schema.\n"
        "  - Generated artifacts (e.g. .github/copilot-instructions.md regenerated "
        "by sync_rules.py).\n"
        "  - Plain code changes (functions, classes, scripts) that don't add new "
        "prose conventions.\n"
        "  - Bug fixes, tests, documentation of existing code.\n"
        "  - Data, datasets, manifests, output files.\n\n"
        "Return exactly one minified JSON object and nothing else:\n"
        '  {"verdict": "guardrail" | "notebook" | "uncertain", "reason": "<one '
        'short sentence>"}\n\n'
        "Verdict guidance:\n"
        "  - 'notebook' if the diff clearly adds discipline-dependent "
        "infrastructure as defined above.\n"
        "  - 'guardrail' if the diff is fine OR is itself guardrail-shaped "
        "infrastructure.\n"
        "  - 'uncertain' only if you genuinely can't tell — this allows the "
        "commit through but logs a warning. Use sparingly.\n\n"
        f"DIFF TO JUDGE:{trunc_note}\n"
        f"{diff_text}\n"
    )


def _extract_json_object(content: str) -> dict | None:
    """Pull the first valid JSON object out of an LLM response. Tolerates
    preamble (e.g. an emitted LOAD MARKER banner from rule content), markdown
    code fences, or raw JSON. Returns None if nothing parseable is found.

    Same robust extraction used in gh-wrap.py — kept in sync via PR review."""
    import re

    text = content.strip()

    # Strategy 1: direct parse (response_format json_object honored).
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        pass

    # Strategy 2: extract first ```...``` block. Drop language tag if present.
    fence_match = re.search(r"```(?:json|JSON)?\s*\n?(.*?)\n?```", text, re.DOTALL)
    if fence_match:
        block = fence_match.group(1).strip()
        try:
            return json.loads(block)
        except json.JSONDecodeError:
            pass

    # Strategy 3: find first {...} substring that looks like our schema.
    obj_match = re.search(r'\{[^{}]*"verdict"[^{}]*\}', text, re.DOTALL)
    if obj_match:
        try:
            return json.loads(obj_match.group(0))
        except json.JSONDecodeError:
            pass

    return None


def _call_llm_judge(prompt: str, api_key: str) -> tuple[str, str]:
    payload = {
        "model": LLM_MODEL,
        "messages": [{"role": "user", "content": prompt}],
        "response_format": {"type": "json_object"},
        "temperature": 0,
    }
    body = json.dumps(payload).encode("utf-8")
    req = urllib.request.Request(
        f"{LLM_BASE_URL}/chat/completions",
        data=body,
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
            "HTTP-Referer": "https://github.com/tjp2021/cos",
            "X-Title": "cos guardrail",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=LLM_TIMEOUT_SECONDS) as resp:
            data = json.loads(resp.read())
    except urllib.error.HTTPError as e:
        detail = e.read().decode("utf-8", errors="replace")[:300]
        raise RuntimeError(f"HTTP {e.code}: {detail}")

    try:
        content = data["choices"][0]["message"]["content"]
    except (KeyError, IndexError, TypeError):
        raise RuntimeError(f"response missing content: {json.dumps(data)[:300]}")
    if not isinstance(content, str) or not content.strip():
        raise RuntimeError("response content was empty")

    obj = _extract_json_object(content)
    if obj is None:
        raise RuntimeError(f"could not extract JSON from response: {content[:200]!r}")

    verdict = obj.get("verdict")
    reason = obj.get("reason", "")
    if verdict not in {"guardrail", "notebook", "uncertain"}:
        raise RuntimeError(f"unexpected verdict value: {verdict!r}")
    return verdict, str(reason)[:300]


# -- Main ---------------------------------------------------------------------

def main() -> int:
    global _DIFF_RANGE, _FAIL_CLOSED
    argv = sys.argv[1:]
    skip_llm = "--no-llm" in argv
    _FAIL_CLOSED = "--fail-closed" in argv
    if "--diff-range" in argv:
        try:
            _DIFF_RANGE = argv[argv.index("--diff-range") + 1]
        except IndexError:
            print("--diff-range requires a value (e.g. abc123..HEAD)", file=sys.stderr)
            return 1
    staged = _git_status()
    if not staged:
        return 0

    checks = [
        ("Rule file missing LOAD MARKER", check_new_rule_has_load_marker),
        ("Stray top-level .md", check_root_md_allowlist),
    ]
    if not skip_llm:
        checks.append(("AI judge (Claude Haiku 4.5 via OpenRouter)", check_llm_judge))

    violations: list[tuple[str, list[str]]] = []
    for label, fn in checks:
        found = fn(staged)
        if found:
            violations.append((label, found))

    if not violations:
        return 0

    print(
        "\n❌ Guardrail check: commit refused — notebook-shaped infrastructure detected.\n",
        file=sys.stderr,
    )
    for label, items in violations:
        print(f"  [{label}] ({len(items)})", file=sys.stderr)
        for v in items:
            print(v, file=sys.stderr)
            print("", file=sys.stderr)
    print(
        "Spec: <workspace>/.claude/rules/guardrail.md\n"
        "\n"
        "Either fix the issues above (preferred) or override with:\n"
        "  git commit --no-verify\n"
        "\n"
        "Override should be rare. Each bypass is one more notebook in the tree.\n",
        file=sys.stderr,
    )
    return 1


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