Agentic Governance Review

Event

Stop

Trigger

always

Language

python

Side effects

• calls an external LLM API (Anthropic / OpenAI / Google) — incurs inference cost
• emits GOVERNANCE-VIOLATION sentinel lines to stderr

Platforms

linuxmacoswindows

Notes

Requires at least one of: ANTHROPIC_API_KEY, OPENAI_API_KEY, AZURE_OPENAI_API_KEY, GOOGLE_API_KEY. Uses only Python stdlib for HTTP — no pip dependencies. Config override: ~/.config/ai/agentic-review.json. Silently skips if no API key is found or diff is empty.

Depends on

hook/lib

Script · agentic-review.py

#!/usr/bin/env python3
"""hooks/agentic-review.py — LLM-based governance review of the latest commit diff.

Fires on Stop. Collects `git diff HEAD~1 HEAD` (the just-committed work),
sends it to the cheapest available model, and emits GOVERNANCE-VIOLATION
sentinel lines for any findings. Non-blocking.

Provider auto-detection order:
  1. ANTHROPIC_API_KEY  → claude-haiku-4-5-20251001
  2. OPENAI_API_KEY     → gpt-4o-mini
  3. AZURE_OPENAI_API_KEY → gpt-4o-mini (requires AZURE_OPENAI_ENDPOINT)
  4. GOOGLE_API_KEY     → gemini-2.0-flash

Config override at ~/.config/ai/agentic-review.json:
  {
    "provider": "anthropic",
    "model": "claude-haiku-4-5-20251001",
    "max_diff_lines": 300,
    "enabled": true
  }

Self-check:
  --self-check  exits 0 (validates config load only; does not call API).
"""
from __future__ import annotations

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

sys.path.insert(0, str(Path(__file__).resolve().parent))
import _lib  # noqa: E402


# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------

_CONFIG_PATH = Path.home() / ".config" / "ai" / "agentic-review.json"

_PROVIDER_DEFAULTS: dict[str, str] = {
    "anthropic": "claude-haiku-4-5-20251001",
    "openai": "gpt-4o-mini",
    "azure-openai": "gpt-4o-mini",
    "google": "gemini-2.0-flash",
}


def _load_config() -> dict:
    defaults = {"enabled": True, "max_diff_lines": 300}
    try:
        raw = json.loads(_CONFIG_PATH.read_text(encoding="utf-8"))
        if isinstance(raw, dict):
            defaults.update(raw)
    except (OSError, json.JSONDecodeError):
        pass
    return defaults


def _detect_provider(cfg: dict) -> tuple[str | None, str | None, str | None]:
    """Return (provider, model, api_key). Provider/model may come from config."""
    explicit_provider = cfg.get("provider", "").lower()
    explicit_model = cfg.get("model", "")

    candidates = [
        ("anthropic", os.environ.get("ANTHROPIC_API_KEY", "")),
        ("openai", os.environ.get("OPENAI_API_KEY", "")),
        ("azure-openai", os.environ.get("AZURE_OPENAI_API_KEY", "")),
        ("google", os.environ.get("GOOGLE_API_KEY", "")),
    ]

    if explicit_provider:
        for name, key in candidates:
            if name == explicit_provider and key:
                model = explicit_model or _PROVIDER_DEFAULTS.get(name, "")
                return name, model, key
        return None, None, None  # configured provider but no key

    for name, key in candidates:
        if key:
            model = explicit_model or _PROVIDER_DEFAULTS[name]
            return name, model, key

    return None, None, None


# ---------------------------------------------------------------------------
# Diff collection
# ---------------------------------------------------------------------------

_REVIEW_PROMPT = """You are a code governance reviewer. Review the git diff below for violations of these rules. Be precise — only flag clear violations, not style preferences.

Rules:
- §4.1.1: Names must reveal intent. Flag symbols named: util, helper, data, temp, mgr, handler, manager, info, obj, val (unless genuinely the most precise word).
- §4.1.3: Comments must explain WHY, not WHAT. Flag comments that describe what the code does rather than why a decision was made.
- §4.1.6: Every catch/except/recover must contain a deliberate decision. Flag: empty blocks, bare pass/continue, or logging-only handlers with no recovery logic.
- §4.1.7: Functions should isolate side effects. Flag: functions that mix pure business logic with I/O, DB calls, or global mutations.
- §4.1.8: Inputs from external sources must be validated at boundaries. Flag: missing input validation on parameters received from user input, network, or files.
- §4.1.9: Magic values must be named constants. Flag: unnamed numeric literals (other than -1, 0, 1, 2) and unexplained magic strings.
- §4.1.10: Public functions must have type signatures. Flag: exported/public functions without type annotations in Python, TypeScript, Go, Java, or Kotlin.
- §4.3.3: Test names must describe behavior, not implementation. Flag: names like test_calls_save(), test_method_1(), test_function_works().
- §4.5.1: No bare print() in non-test production code. Flag: print() / console.log() / fmt.Println() outside test files.
- §5.1.3: No AI tells in prose or comments. Flag: em-dash overload, 'Let's dive in', 'In today's world', 'It's not just X it's Y', generic summary paragraphs.

Return ONLY valid JSON — no markdown, no explanation:
{"violations": [{"rule": "§X.X.X", "file": "path/file.py", "line": 42, "detail": "one sentence"}]}
If no violations: {"violations": []}

Diff:
"""


def _get_diff(cwd: str, max_lines: int) -> str | None:
    """Return the latest commit diff, capped at max_lines. None if unavailable."""
    for cmd in (
        ["git", "diff", "HEAD~1", "HEAD"],
        ["git", "diff", "HEAD"],
    ):
        try:
            r = subprocess.run(
                cmd, capture_output=True, text=True, check=False, cwd=cwd
            )
            if r.returncode == 0 and r.stdout.strip():
                lines = r.stdout.splitlines()
                if len(lines) > max_lines:
                    lines = lines[:max_lines]
                    lines.append(f"... (truncated at {max_lines} lines)")
                return "\n".join(lines)
        except FileNotFoundError:
            return None
    return None


# ---------------------------------------------------------------------------
# API callers (stdlib urllib only — no pip deps)
# ---------------------------------------------------------------------------

def _call_anthropic(model: str, key: str, prompt: str) -> str:
    payload = json.dumps({
        "model": model,
        "max_tokens": 1024,
        "messages": [{"role": "user", "content": prompt}],
    }).encode()
    req = urllib.request.Request(
        "https://api.anthropic.com/v1/messages",
        data=payload,
        headers={
            "x-api-key": key,
            "anthropic-version": "2023-06-01",
            "content-type": "application/json",
        },
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.loads(resp.read())["content"][0]["text"]


def _call_openai(model: str, key: str, prompt: str, base_url: str = "https://api.openai.com/v1") -> str:
    payload = json.dumps({
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "max_tokens": 1024,
    }).encode()
    req = urllib.request.Request(
        f"{base_url}/chat/completions",
        data=payload,
        headers={"Authorization": f"Bearer {key}", "content-type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.loads(resp.read())["choices"][0]["message"]["content"]


def _call_google(model: str, key: str, prompt: str) -> str:
    payload = json.dumps({
        "contents": [{"parts": [{"text": prompt}]}],
        "generationConfig": {"maxOutputTokens": 1024},
    }).encode()
    url = f"https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={key}"
    req = urllib.request.Request(
        url, data=payload, headers={"content-type": "application/json"}
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        return json.loads(resp.read())["candidates"][0]["content"]["parts"][0]["text"]


def _call_api(provider: str, model: str, key: str, prompt: str) -> str:
    if provider == "anthropic":
        return _call_anthropic(model, key, prompt)
    if provider == "openai":
        return _call_openai(model, key, prompt)
    if provider == "azure-openai":
        endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT", "").rstrip("/")
        base = f"{endpoint}/openai/deployments/{model}"
        return _call_openai(model, key, prompt, base_url=base)
    if provider == "google":
        return _call_google(model, key, prompt)
    raise ValueError(f"unknown provider: {provider}")


# ---------------------------------------------------------------------------
# Response parsing
# ---------------------------------------------------------------------------

def _parse_violations(raw: str) -> list[dict]:
    """Extract the violations array from the model response."""
    # Strip markdown fences if the model wrapped the JSON
    text = raw.strip()
    if text.startswith("```"):
        lines = text.splitlines()
        text = "\n".join(l for l in lines if not l.startswith("```"))
    try:
        data = json.loads(text)
        return data.get("violations", []) if isinstance(data, dict) else []
    except json.JSONDecodeError:
        return []


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

def main() -> None:
    if "--self-check" in sys.argv:
        cfg = _load_config()
        _lib.log(f"self-check OK — config loaded, enabled={cfg.get('enabled', True)}")
        sys.exit(0)

    cfg = _load_config()
    if not cfg.get("enabled", True):
        sys.exit(0)

    try:
        event = json.load(sys.stdin)
    except (json.JSONDecodeError, EOFError):
        event = {}

    cwd = event.get("cwd") or event.get("workingDirectory") or os.getcwd()

    diff = _get_diff(cwd, int(cfg.get("max_diff_lines", 300)))
    if not diff:
        sys.exit(0)

    provider, model, api_key = _detect_provider(cfg)
    if not provider:
        sys.exit(0)  # no API key available — skip silently

    prompt = _REVIEW_PROMPT + diff

    try:
        raw = _call_api(provider, model, api_key, prompt)
    except (urllib.error.URLError, OSError, KeyError, IndexError) as e:
        _lib.log(f"agentic-review: API call failed ({provider}/{model}): {e}")
        sys.exit(0)

    violations = _parse_violations(raw)
    if not violations:
        _lib.log(f"agentic-review: no violations found ({provider}/{model})")
        sys.exit(0)

    for v in violations:
        rule = v.get("rule", "?")
        file_ = v.get("file", "?")
        line = v.get("line", "?")
        detail = v.get("detail", "?")
        _lib.log(f"GOVERNANCE-VIOLATION: {rule} — {detail} [{file_}:{line}]")


if __name__ == "__main__":
    main()