scripts/codex_review.py

#!/usr/bin/env python3
"""stop hook: Codex-powered diff review when session changes exceed threshold.

Reads a baseline git tree snapshot (saved by codex_baseline.py at session
start), computes the diff, and runs ``codex exec review`` when changes are
substantial.  The review prompt references AGENTS.md and checks for banned
patterns, hacky workarounds, and code-quality issues.

Change tracking is independent of commits: the baseline is a git tree
object captured at session start, and the diff is computed against the
live working tree at stop time.
"""

import json
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

project_root = Path(__file__).resolve().parent.parent
state_dir = project_root / ".cursor" / "hooks" / "state"
baseline_path = state_dir / "codex_review_baseline"

CODEX_REVIEW_MIN_LINES = 50


def count_diff_lines(diff_text: str) -> int:
    """Count additions and deletions in a unified diff."""
    count = 0
    for line in diff_text.splitlines():
        is_add = line.startswith("+") and not line.startswith("+++")
        is_del = line.startswith("-") and not line.startswith("---")
        if is_add or is_del:
            count += 1
    return count


def compute_session_diff(baseline_sha: str) -> str:
    """Diff between the baseline snapshot and the current working tree.

    Includes untracked files by using intent-to-add (git add -N) so new
    files created during the session appear in the diff.
    """
    subprocess.run(
        ["/usr/bin/git", "add", "-N", "."],
        capture_output=True,
        text=True,
        timeout=10,
        check=False,
        cwd=str(project_root),
    )
    result = subprocess.run(
        ["/usr/bin/git", "diff", baseline_sha],
        capture_output=True,
        text=True,
        timeout=30,
        check=False,
        cwd=str(project_root),
    )
    return result.stdout


def build_review_prompt(lines_changed: int) -> str:
    """Construct the custom instructions for codex exec review."""
    kw_sep = "keyword" + "-only separator"
    return (
        f"This session changed {lines_changed} lines. "
        "Read AGENTS.md in the repo root for the full project rules.\n\n"
        "Key banned patterns to check for:\n"
        "- try/except without rationale, assert, nested functions, @dataclass\n"
        "- typing.Any, typing.cast, functools.cache\n"
        "- Magic numbers at module level, os.environ/os.getenv\n"
        f"- {kw_sep}, string-key membership checks on dicts\n"
        "- Defensive .get() after model_validate, isinstance after validation\n"
        "- Inline suppression comments without owner approval\n\n"
        "Also flag:\n"
        "1. Hacky workarounds, TODO/FIXME/HACK, band-aid fixes\n"
        "2. Loose dicts where typed Pydantic models belong\n"
        "3. God functions, excessive nesting, duplicated logic\n"
        "4. Unclear or misleading names\n"
        "5. Missing type annotations\n"
        "6. Hand-rolled code that a library already handles\n\n"
        "Be concise. Report actionable findings only.\n"
        "Format: file:line [category] description\n"
        "If clean: LGTM"
    )


def run_codex_review(lines_changed: int) -> str:
    """Run codex exec review --uncommitted with project-specific instructions."""
    codex_path = shutil.which("codex")
    if codex_path is None:
        return ""

    with tempfile.NamedTemporaryFile(
        mode="w",
        suffix=".txt",
        delete=False,
    ) as tmp:
        output_path = tmp.name

    cmd = [
        codex_path,
        "exec",
        "review",
        "--uncommitted",
        "--ephemeral",
        "-o",
        output_path,
        build_review_prompt(lines_changed),
    ]

    subprocess.run(
        cmd,
        capture_output=True,
        text=True,
        timeout=150,
        check=False,
        cwd=str(project_root),
    )

    result_file = Path(output_path)
    review_text = result_file.read_text().strip()
    result_file.unlink(missing_ok=True)
    return review_text


def main() -> None:
    """Run codex review if session changes exceed the configured threshold."""
    state_dir.mkdir(parents=True, exist_ok=True)
    parsed = json.loads(sys.stdin.read())

    status = parsed.get("status", "")
    loop_count = parsed.get("loop_count", 0)

    if status != "completed":
        baseline_path.unlink(missing_ok=True)
        print("{}")
        return

    if loop_count > 0:
        print("{}")
        return

    if not baseline_path.exists():
        print("{}")
        return

    baseline_sha = baseline_path.read_text().strip()
    baseline_path.unlink(missing_ok=True)

    if not baseline_sha:
        print("{}")
        return

    diff = compute_session_diff(baseline_sha)
    lines_changed = count_diff_lines(diff)

    if lines_changed < CODEX_REVIEW_MIN_LINES:
        print("{}")
        return

    review = run_codex_review(lines_changed)

    if not review:
        print("{}")
        return

    upper_review = review.strip().upper()
    if upper_review.startswith("LGTM"):
        print("{}")
        return

    msg = (
        f"Codex Review ({lines_changed} lines changed):\n\n"
        f"{review}\n\n"
        "Address these findings or explain why they are acceptable."
    )
    print(json.dumps({"followup_message": msg}))


if __name__ == "__main__":
    main()