mariusv · October 12, 2026 09:36
diff --git a/README.md b/README.md
diff --git a/comment-cop.py b/comment-cop.py
 #!/usr/bin/env python3
 """Enforce a why-only policy for comments added by coding agents."""

 from __future__ import annotations

 import collections
 import hashlib
 import itertools
 import json
 import os
 import re
 import subprocess
 import sys
 import tempfile
 from dataclasses import dataclass
 from pathlib import Path


 SLASH = {
    ".go", ".js", ".ts", ".tsx", ".jsx", ".mjs", ".cjs", ".java", ".rs",
    ".c", ".h", ".hpp", ".cpp", ".cc", ".proto", ".kt", ".swift", ".scala", ".dart",
 }
 HASH = {
    ".py", ".sh", ".bash", ".zsh", ".yaml", ".yml", ".tf", ".tfvars",
    ".toml", ".rb", ".mk", ".pl", ".ps1", ".conf", ".cfg",
 }
 DASH = {".sql", ".lua", ".hs"}
 HASH_NAMES = {"Makefile", "Dockerfile", "Justfile", "Vagrantfile"}

 POLICY = (
    "Comments explain only non-obvious decisions, constraints, safety invariants, or workarounds. "
    "Do not narrate what code does or record change history; git holds history. Keep comments to "
    "one or two tight lines unless a longer explanation is genuinely necessary."
 )

 CHANGE_HISTORY = re.compile(
    r"\b(previously|used to|no longer (?:needed|used|required|exists|applies)|"
    r"we (?:removed|added|changed|moved|renamed|dropped|switched)|"
    r"removed (?:in favor|because)|renamed (?:from|to)|replaced by|"
    r"this (?:was|used to)|has been (?:removed|moved|renamed|replaced))\b",
    re.I,
 )
 NARRATION = re.compile(
    r"^(?:(?:this|the|these|those)\s+.{0,60}\s+"
    r"(?:is|are|does|do|will|contains|returns|sets|gets|creates|updates|deletes|"
    r"loops|iterates|checks|verifies|loads|stores|writes|reads|calls|uses|handles|"
    r"initializes|converts|parses|builds|starts|stops|opens|closes)|"
    r"(?:set|get|create|update|delete|return|loop|iterate|check|verify|load|save|"
    r"store|call|add|remove|initialize|convert|parse|build|handle|run|start|stop|"
    r"open|close|write|read)\b)",
    re.I,
 )
 WHY = re.compile(
    r"\b(because|otherwise|unless|until|to (?:avoid|prevent|preserve|ensure|keep)|"
    r"must|cannot|can't|required|invariant|race|deadlock|data loss|durab(?:le|ility)|"
    r"compatibility|security|atomic(?:ity|ally)?|workaround)\b",
    re.I,
 )
 EXEMPT = re.compile(
    r"^(?:!|todo\b|fixme\b|hack\b|xxx\b|spdx-|copyright\b|license\b|"
    r"code generated\b|go:|nolint\b|nosec\b|noqa\b|type:\s*ignore\b|"
    r"eslint\b|prettier\b|shellcheck\b|yamllint\b|hadolint\b|pragma\b|"
    r"fmt:|line\b|sourceMappingURL=|region\b|endregion\b|cspell\b)",
    re.I,
 )
 COMMENTED_CODE = re.compile(r"^\s*(?:\S+\s*[:=]|-\s|[{}\[\]])")
 HUNK = re.compile(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,\d+)? @@")


 @dataclass(frozen=True)
 class AddedLine:
    path: str
    number: int
    text: str
    group: int


 @dataclass(frozen=True)
 class Finding:
    path: str
    number: int
    rule: str
    text: str

    def fingerprint(self) -> str:
        normalized = " ".join(self.text.split()).lower()
        return f"{self.path}\0{self.rule}\0{normalized}"


 def marker_for(path: str) -> str | None:
    base = os.path.basename(path)
    if base in HASH_NAMES:
        return "#"
    ext = os.path.splitext(base)[1].lower()
    if ext in SLASH:
        return "//"
    if ext in HASH:
        return "#"
    if ext in DASH:
        return "--"
    return None


 def comment_body(line: str, marker: str) -> str | None:
    stripped = line.strip()
    if stripped.startswith(marker):
        return stripped[len(marker):].strip()
    if marker == "//" and (stripped.startswith("/*") or stripped.startswith("*")):
        return stripped.lstrip("/* ").rstrip("*/ ").strip()
    return None


 def inline_body(line: str, marker: str) -> str | None:
    token = " " + marker
    pos = line.find(token)
    if pos < 0:
        return None
    return line[pos + len(token):].strip()


 def analyze(lines: list[AddedLine]) -> list[Finding]:
    findings: list[Finding] = []
    for path, path_lines_iter in _group_by(lines, lambda line: line.path):
        marker = marker_for(path)
        if not marker:
            continue
        path_lines = list(path_lines_iter)
        for _, group_iter in _group_by(path_lines, lambda line: line.group):
            group_lines = list(group_iter)
            comment_count = 0
            code_count = 0
            runs: list[list[AddedLine]] = []
            run: list[AddedLine] = []
            for line in group_lines:
                body = comment_body(line.text, marker)
                if body is None:
                    if run:
                        runs.append(run)
                        run = []
                    if line.text.strip():
                        code_count += 1
                    inline = inline_body(line.text, marker)
                    if inline and CHANGE_HISTORY.search(inline):
                        findings.append(Finding(line.path, line.number, "change-history", inline))
                    continue
                if not body or EXEMPT.search(body) or COMMENTED_CODE.match(body):
                    if run:
                        runs.append(run)
                        run = []
                    continue
                comment_count += 1
                if run and line.number != run[-1].number + 1:
                    runs.append(run)
                    run = []
                run.append(line)
                findings.extend(_prose_findings(line, body))
            if run:
                runs.append(run)
            for comment_run in runs:
                if len(comment_run) >= 3:
                    findings.append(Finding(
                        path,
                        comment_run[0].number,
                        "long-block",
                        f"{len(comment_run)} consecutive comment lines",
                    ))
            if comment_count >= 3 and comment_count > code_count:
                first = next((line for line in group_lines if comment_body(line.text, marker) is not None), group_lines[0])
                findings.append(Finding(
                    path,
                    first.number,
                    "comment-density",
                    f"{comment_count} comment lines for {code_count} code lines in one change",
                ))
    return _deduplicate(findings)


 def _prose_findings(line: AddedLine, body: str) -> list[Finding]:
    if CHANGE_HISTORY.search(body):
        return [Finding(line.path, line.number, "change-history", body)]
    if NARRATION.search(body) and not WHY.search(body):
        return [Finding(line.path, line.number, "code-narration", body)]
    return []


 def _group_by(items, key):
    return itertools.groupby(items, key)


 def _deduplicate(findings: list[Finding]) -> list[Finding]:
    seen: set[tuple[str, int, str]] = set()
    result: list[Finding] = []
    for finding in findings:
        key = (finding.path, finding.number, finding.rule)
        if key not in seen:
            seen.add(key)
            result.append(finding)
    return result


 def parse_apply_patch(patch: str) -> list[AddedLine]:
    lines: list[AddedLine] = []
    path = ""
    number = 1
    group = 0
    for raw in patch.splitlines():
        if raw.startswith("*** Update File: ") or raw.startswith("*** Add File: "):
            path = raw.split(": ", 1)[1]
            number = 1
            group += 1
            continue
        match = HUNK.match(raw)
        if match:
            number = int(match.group(1))
            group += 1
            continue
        if not path:
            continue
        if raw.startswith("+") and not raw.startswith("+++"):
            lines.append(AddedLine(path, number, raw[1:], group))
            number += 1
        elif raw.startswith(" "):
            number += 1
        elif raw.startswith("-"):
            continue
    return lines


 def parse_git_diff(diff: str) -> list[AddedLine]:
    lines: list[AddedLine] = []
    path = ""
    number = 1
    group = 0
    for raw in diff.splitlines():
        if raw.startswith("+++ "):
            path = raw[4:]
            if path == "/dev/null":
                path = ""
            continue
        match = HUNK.match(raw)
        if match:
            number = int(match.group(1))
            group += 1
            continue
        if not path:
            continue
        if raw.startswith("+") and not raw.startswith("+++"):
            lines.append(AddedLine(path, number, raw[1:], group))
            number += 1
        elif raw.startswith(" "):
            number += 1
        elif raw.startswith("-"):
            continue
    return lines


 def git(cwd: str, *args: str) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["git", "-c", "core.quotePath=false", *args],
        cwd=cwd,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        check=False,
    )


 def worktree_findings(cwd: str, base: str = "HEAD") -> list[Finding]:
    root_result = git(cwd, "rev-parse", "--show-toplevel")
    if root_result.returncode:
        return []
    root = root_result.stdout.strip()
    diff_result = git(root, "diff", "--no-ext-diff", "--no-textconv", "--unified=0", "--no-color", "--no-prefix", base, "--")
    if diff_result.returncode:
        raise RuntimeError("Cannot audit comments against the session baseline")
    lines = parse_git_diff(diff_result.stdout)
    untracked = git(root, "ls-files", "--others", "--exclude-standard", "-z")
    if untracked.returncode == 0:
        group = max((line.group for line in lines), default=0)
        for relative in filter(None, untracked.stdout.split("\0")):
            if marker_for(relative) is None:
                continue
            path = Path(root, relative)
            try:
                if path.stat().st_size > 2_000_000:
                    continue
                content = path.read_text(errors="replace").splitlines()
            except OSError:
                continue
            group += 1
            lines.extend(AddedLine(relative, number, text, group) for number, text in enumerate(content, 1))
    return analyze(lines)


 def state_path(session_id: str) -> Path:
    state_dir = Path(tempfile.gettempdir(), f"comment-cop-{os.getuid()}")
    state_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
    digest = hashlib.sha256(session_id.encode()).hexdigest()
    return state_dir / f"{digest}.json"


 def record_baseline(data: dict) -> None:
    session_id = str(data.get("session_id") or "")
    cwd = str(data.get("cwd") or os.getcwd())
    if not session_id:
        return
    root_result = git(cwd, "rev-parse", "--show-toplevel")
    if root_result.returncode:
        return
    root = root_result.stdout.strip()
    path = state_path(session_id)
    try:
        state = json.loads(path.read_text())
    except (OSError, ValueError):
        state = {}
    if state.get("version") != 2:
        state = {"version": 2, "repos": {}}
    if root in state["repos"]:
        return
    head = git(root, "rev-parse", "HEAD")
    if head.returncode:
        raise RuntimeError("Cannot record the session baseline commit")
    counts = collections.Counter(finding.fingerprint() for finding in worktree_findings(root))
    state["repos"][root] = {"head": head.stdout.strip(), "findings": counts}
    path.write_text(json.dumps(state, sort_keys=True))


 def new_findings(data: dict) -> list[Finding]:
    session_id = str(data.get("session_id") or "")
    cwd = str(data.get("cwd") or os.getcwd())
    if not session_id:
        return []
    path = state_path(session_id)
    if not path.exists():
        return []
    try:
        state = json.loads(path.read_text())
    except (OSError, ValueError):
        return []
    root_result = git(cwd, "rev-parse", "--show-toplevel")
    if root_result.returncode:
        return []
    baseline_state = state.get("repos", {}).get(root_result.stdout.strip()) if state.get("version") == 2 else None
    if not baseline_state:
        return []
    baseline = collections.Counter(baseline_state["findings"])
    current = worktree_findings(cwd, baseline_state["head"])
    remaining = baseline.copy()
    result: list[Finding] = []
    for finding in current:
        fingerprint = finding.fingerprint()
        if remaining[fingerprint]:
            remaining[fingerprint] -= 1
        else:
            result.append(finding)
    return result


 def direct_edit_findings(data: dict) -> list[Finding]:
    tool_input = data.get("tool_input") or {}
    tool_name = data.get("tool_name") or ""
    if tool_name == "apply_patch":
        return analyze(parse_apply_patch(str(tool_input.get("command") or "")))
    path = str(tool_input.get("file_path") or "")
    marker = marker_for(path)
    if not marker:
        return []
    if tool_name == "Edit":
        text = str(tool_input.get("new_string") or "")
        old = {line.strip() for line in str(tool_input.get("old_string") or "").splitlines()}
    else:
        text = str(tool_input.get("content") or "")
        old = set()
        try:
            old = {line.strip() for line in Path(path).read_text(errors="replace").splitlines()}
        except OSError:
            pass
    lines = [
        AddedLine(path, number, line, 1)
        for number, line in enumerate(text.splitlines(), 1)
        if line.strip() not in old
    ]
    return analyze(lines)


 def format_findings(findings: list[Finding], limit: int = 8) -> str:
    shown = findings[:limit]
    details = "; ".join(
        f"{finding.path}:{finding.number} [{finding.rule}] {finding.text!r}"
        for finding in shown
    )
    if len(findings) > limit:
        details += f"; and {len(findings) - limit} more"
    return f"comment-cop: {details}. {POLICY} Remove or tighten these comments, then retry."


 def policy_context() -> dict:
    return {
        "hookSpecificOutput": {
            "hookEventName": "UserPromptSubmit",
            "additionalContext": "Code comment policy (hook-enforced): " + POLICY,
        }
    }


 def run_event(data: dict) -> dict | None:
    event = data.get("hook_event_name")
    if event == "SessionStart":
        record_baseline(data)
        return None
    if event == "UserPromptSubmit":
        record_baseline(data)
        return policy_context()
    if event == "PreToolUse":
        findings = direct_edit_findings(data)
        if findings:
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": format_findings(findings),
                }
            }
        return None
    if event == "Stop":
        if data.get("stop_hook_active"):
            return {}
        findings = new_findings(data)
        if findings:
            return {"decision": "block", "reason": format_findings(findings)}
        return {}
    return None


 def self_test() -> None:
    cases = [
        (
            "long yaml",
            ["# The migrate role reports like any other.", "# It inherits no envFrom.", "# The SDK falls back.", "envFrom:"],
            True,
        ),
        (
            "four-line go",
            ["// The restored base may contain a later row.", "// Never overwrite it.", "// The conflict handler covers constraints.", "// The final comparison is authoritative.", "insert()"],
            True,
        ),
        ("narration", ["// Loop through all rows.", "for rows.Next() {}"], True),
        ("history", ["// We removed the fallback because it was unsafe.", "run()"], True),
        ("why", ["// Keep the predecessor until the handoff is durable.", "drop()"], False),
        ("directive", ["//nolint:gosec", "run()"], False),
        ("commented config", ["# timeout: 10s", "enabled: true"], False),
    ]
    for name, content, expected in cases:
        lines = [AddedLine("test.go" if name not in {"long yaml", "commented config"} else "test.yaml", i, line, 1) for i, line in enumerate(content, 1)]
        actual = bool(analyze(lines))
        if actual != expected:
            raise AssertionError(f"{name}: expected {expected}, got {actual}")
    patch = "*** Begin Patch\n*** Update File: x.go\n@@ -1 +1,2 @@\n run()\n+// Set the value.\n*** End Patch"
    if not analyze(parse_apply_patch(patch)):
        raise AssertionError("apply_patch parsing did not find narration")
    with tempfile.TemporaryDirectory() as directory:
        subprocess.run(["git", "init", "-q"], cwd=directory, check=True)
        subprocess.run(["git", "config", "user.email", "comment-cop@example.invalid"], cwd=directory, check=True)
        subprocess.run(["git", "config", "user.name", "comment-cop"], cwd=directory, check=True)
        source = Path(directory, "main.go")
        source.write_text("package main\n")
        subprocess.run(["git", "add", "main.go"], cwd=directory, check=True)
        subprocess.run(["git", "commit", "-qm", "base"], cwd=directory, check=True)
        source.write_text("package main\n// Loop through rows.\n")
        event = {"session_id": "self-test", "cwd": directory}
        record_baseline(event)
        source.write_text("package main\n// Loop through rows.\n// Set the timeout.\n")
        findings = new_findings(event)
        if len(findings) != 1 or findings[0].text != "Set the timeout.":
            raise AssertionError(f"baseline isolation failed: {findings!r}")
    print("comment-cop self-test passed")


 def main() -> None:
    if len(sys.argv) > 1 and sys.argv[1] == "--self-test":
        self_test()
        return
    if len(sys.argv) > 1 and sys.argv[1] == "--check-diff":
        findings = worktree_findings(os.getcwd())
        if findings:
            print(format_findings(findings))
            raise SystemExit(1)
        return
    data = json.load(sys.stdin)
    output = run_event(data)
    if output is not None:
        print(json.dumps(output))


 if __name__ == "__main__":
    try:
        main()
    except Exception as error:
        if "--self-test" in sys.argv:
            raise
        print(json.dumps({"systemMessage": f"comment-cop failed open: {error}"}))
diff --git a/test_comment_cop.py b/test_comment_cop.py
 import runpy
 import subprocess
 import sys
 import unittest
 from pathlib import Path

 sys.path.insert(0, str(Path(__file__).parent))
 from test_xreview import RepoCase, git


 class CommentAudit(RepoCase):
    def setUp(self):
        super().setUp()
        self.cop = runpy.run_path(str(Path(__file__).with_name("comment-cop.py")))
        self.state = Path(self.tmp.name, "session.json")
        self.cop["record_baseline"].__globals__["state_path"] = lambda session_id: self.state
        self.data = {"session_id": "test", "cwd": str(self.repo)}
        self.source = self.repo / "code.go"
        self.source.write_text("package main\n")
        git(self.repo, "add", "code.go")
        git(self.repo, "commit", "-qm", "source")

    def record(self):
        self.cop["record_baseline"](self.data)

    def findings(self):
        return self.cop["new_findings"](self.data)

    def add_comment(self):
        self.source.write_text("package main\n// Loop through all rows.\n")

    def test_committed_comment_is_audited(self):
        self.record()
        self.add_comment()
        git(self.repo, "commit", "-qam", "comment")
        self.assertEqual([f.rule for f in self.findings()], ["code-narration"])

    def test_new_file_committed_comment_is_audited(self):
        self.record()
        Path(self.repo, "new.go").write_text("package main\n// Set the value.\n")
        git(self.repo, "add", "new.go")
        git(self.repo, "commit", "-qm", "new source")
        self.assertEqual([f.rule for f in self.findings()], ["code-narration"])

    def test_prompts_and_resume_preserve_session_anchor(self):
        self.record()
        self.add_comment()
        git(self.repo, "commit", "-qam", "comment")
        for event in ("UserPromptSubmit", "SessionStart"):
            self.cop["run_event"]({**self.data, "hook_event_name": event})
            self.assertEqual(len(self.findings()), 1)

    def test_preexisting_comment_is_excluded(self):
        self.add_comment()
        self.record()
        git(self.repo, "commit", "-qam", "preexisting comment")
        self.assertEqual(self.findings(), [])

    def test_removed_comment_clears_finding(self):
        self.record()
        self.add_comment()
        git(self.repo, "commit", "-qam", "comment")
        self.source.write_text("package main\n")
        self.assertEqual(self.findings(), [])

    def test_stop_blocks_committed_comment(self):
        self.record()
        self.add_comment()
        git(self.repo, "commit", "-qam", "comment")
        result = self.cop["run_event"]({**self.data, "hook_event_name": "Stop"})
        self.assertEqual(result["decision"], "block")

    def test_multiple_checkouts_keep_separate_anchors(self):
        self.record()
        self.add_comment()
        git(self.repo, "commit", "-qam", "comment")
        other = Path(self.tmp.name, "other")
        subprocess.run(["git", "clone", "-q", str(self.bare), str(other)], check=True)
        self.cop["record_baseline"]({**self.data, "cwd": str(other)})
        self.assertEqual(len(self.findings()), 1)
        self.assertEqual(self.cop["new_findings"]({**self.data, "cwd": str(other)}), [])


 if __name__ == "__main__":
    unittest.main()
diff --git a/test_xreview.py b/test_xreview.py
 import json
 import os
 import subprocess
 import sys
 import tempfile
 import unittest
 from dataclasses import replace
 from pathlib import Path
 from unittest.mock import patch

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


 def git(cwd, *args):
    return subprocess.run(["git", "-C", str(cwd), *args], check=True, capture_output=True, text=True).stdout.strip()


 class RepoCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        root = Path(self.tmp.name)
        self.bare = root / "origin.git"
        subprocess.run(["git", "init", "-q", "--bare", "-b", "main", str(self.bare)], check=True)
        self.repo = root / "work"
        subprocess.run(["git", "clone", "-q", str(self.bare), str(self.repo)], check=True)
        git(self.repo, "config", "user.email", "t@t")
        git(self.repo, "config", "user.name", "t")
        git(self.repo, "checkout", "-q", "-b", "main")
        (self.repo / "a.txt").write_text("one\n")
        git(self.repo, "add", "a.txt")
        git(self.repo, "commit", "-q", "-m", "init")
        git(self.repo, "push", "-q", "-u", "origin", "main")
        git(self.repo, "remote", "set-head", "origin", "main")
        git(self.repo, "checkout", "-q", "-b", "feature")
        (self.repo / "a.txt").write_text("two\n")
        git(self.repo, "commit", "-q", "-am", "change")
        self.cache = root / "cache"
        os.environ["XREVIEW_CACHE"] = str(self.cache)

    def tearDown(self):
        os.environ.pop("XREVIEW_CACHE", None)
        self.tmp.cleanup()


 class CommandClassification(unittest.TestCase):
    def test_pr_create_is_gated(self):
        self.assertTrue(xreview.is_gated_command("gh pr create --title x --body y"))
        self.assertTrue(xreview.is_gated_command("gh pr create --draft --fill"))
        self.assertTrue(xreview.is_gated_command("gh  pr   ready 123"))
        self.assertTrue(xreview.is_gated_command("git push -u origin HEAD && gh pr create --fill"))
        self.assertTrue(xreview.is_gated_command("gh pr --repo org/repo create --fill"))

    def test_other_commands_pass(self):
        for cmd in ["gh pr view 1", "gh pr list", "git push", "echo gh pr create", "gh api repos/x/pulls"]:
            self.assertFalse(xreview.is_gated_command(cmd), cmd)

    def test_skip_token(self):
        self.assertTrue(xreview.has_skip("XREVIEW_SKIP=1 gh pr create --fill"))
        self.assertFalse(xreview.has_skip("gh pr create --fill"))
        self.assertFalse(xreview.has_skip('gh pr create --body "XREVIEW_SKIP=1"'))
        self.assertFalse(xreview.has_skip("echo XREVIEW_SKIP=1; gh pr create"))
        self.assertTrue(xreview.has_skip("env XREVIEW_SKIP=1 gh pr create"))

    def test_trailing_newline_is_supported(self):
        calls = xreview.command_invocations("gh pr create --fill\n", "/tmp")
        self.assertEqual(len(calls), 1)

    def test_unsupported_wrappers_are_gated(self):
        for command in ["env -u GH_REPO gh pr create", "sudo gh pr create", "bash -c 'gh pr ready'", "echo $(gh pr create)", "gh pr create >out", "if true; then gh pr create; fi", "python3 -c 'import os; os.system(\"gh pr create\")'"]:
            self.assertTrue(xreview.is_gated_command(command), command)


 class DiffHash(RepoCase):
    def test_hash_is_stable_and_changes_with_commits(self):
        ctx1 = xreview.branch_context(self.repo)
        ctx2 = xreview.branch_context(self.repo)
        self.assertEqual(ctx1.diff_hash, ctx2.diff_hash)
        self.assertEqual(ctx1.base_ref, "origin/main")
        (self.repo / "a.txt").write_text("three\n")
        git(self.repo, "commit", "-q", "-am", "more")
        self.assertNotEqual(ctx1.diff_hash, xreview.branch_context(self.repo).diff_hash)

    def test_uncommitted_changes_do_not_affect_hash(self):
        before = xreview.branch_context(self.repo).diff_hash
        (self.repo / "a.txt").write_text("dirty\n")
        self.assertEqual(before, xreview.branch_context(self.repo).diff_hash)
        self.assertTrue(xreview.branch_context(self.repo).dirty)


 OK = {"verdict": "approve", "summary": "ok", "findings": [], "next_steps": []}
 FINDING = {"severity": "high", "title": "Bad thing", "body": "b", "file": "a.txt", "line_start": 1, "line_end": 1, "confidence": 0.9, "recommendation": "fix it"}
 BAD = {"verdict": "needs-attention", "summary": "no-ship", "findings": [FINDING], "next_steps": ["do x"]}


 class Gate(RepoCase):
    def setUp(self):
        super().setUp()
        self._which = xreview.shutil.which
        xreview.shutil.which = lambda name: None
        self.calls = []
        self.contexts = []
        self.addCleanup(patch.stopall)
        patch.object(xreview, "_gh", side_effect=self.gh_metadata).start()
        self._codex, self._claude = xreview.run_codex, xreview.run_claude

    def gh_metadata(self, cwd, args, env):
        if args[:2] == ["repo", "view"]:
            return {"nameWithOwner": "test/repo", "url": str(self.bare), "defaultBranchRef": {"name": "main"}}
        return {"headRefOid": git(self.repo, "rev-parse", "HEAD"), "headRefName": "feature", "baseRefName": "main"}

    def context(self, command="gh pr create"):
        return xreview.target_context(xreview.command_invocations(command, str(self.repo))[0])

    def tearDown(self):
        xreview.shutil.which = self._which
        xreview.run_codex, xreview.run_claude = self._codex, self._claude
        super().tearDown()

    def fake_reviewer(self, result):
        xreview.shutil.which = lambda name: "/usr/bin/" + name
        def runner(ctx, prompt):
            self.calls.append(prompt)
            self.contexts.append(ctx)
            return result
        xreview.run_codex = runner
        xreview.run_claude = runner

    def hook_input(self, command):
        return {"hook_event_name": "PreToolUse", "tool_name": "Bash",
                "tool_input": {"command": command}, "cwd": str(self.repo)}

    def test_denies_when_reviewer_cli_missing(self):
        out = xreview.gate(self.hook_input("gh pr create --fill"), me="claude")
        self.assertEqual(out["hookSpecificOutput"]["permissionDecision"], "deny")
        self.assertIn("codex", out["hookSpecificOutput"]["permissionDecisionReason"].lower())

    def test_runs_review_inline_and_allows_on_approve(self):
        self.fake_reviewer(OK)
        self.assertIsNone(xreview.gate(self.hook_input("gh pr create --fill"), me="claude"))
        self.assertEqual(len(self.calls), 1)
        self.assertIn("feature", self.calls[0])
        ctx = self.context()
        self.assertEqual(xreview.read_receipt(ctx, "codex")["verdict"], "approve")

    def test_runs_review_inline_and_denies_with_findings(self):
        self.fake_reviewer(BAD)
        out = xreview.gate(self.hook_input("gh pr create --fill"), me="claude")
        reason = out["hookSpecificOutput"]["permissionDecisionReason"]
        self.assertIn("Bad thing", reason)
        self.assertIn("fix it", reason)
        self.assertEqual(len(self.calls), 1)
        # Retrying unchanged means the author read the findings and chose to proceed.
        self.assertIsNone(xreview.gate(self.hook_input("gh pr create --fill"), me="claude"))
        self.assertEqual(len(self.calls), 1)

    def test_new_commit_after_findings_triggers_a_fresh_review(self):
        self.fake_reviewer(BAD)
        xreview.gate(self.hook_input("gh pr create"), me="claude")
        (self.repo / "a.txt").write_text("fixed\n")
        git(self.repo, "commit", "-q", "-am", "fix")
        self.fake_reviewer(OK)
        self.assertIsNone(xreview.gate(self.hook_input("gh pr create"), me="claude"))
        self.assertEqual(len(self.calls), 2)

    def test_dirty_tree_is_denied_before_review(self):
        self.fake_reviewer(OK)
        (self.repo / "a.txt").write_text("dirty\n")
        out = xreview.gate(self.hook_input("gh pr create"), me="claude")
        self.assertIn("uncommitted", out["hookSpecificOutput"]["permissionDecisionReason"].lower())
        self.assertEqual(self.calls, [])

    def test_reviewer_failure_denies(self):
        xreview.shutil.which = lambda name: "/usr/bin/" + name
        def boom(ctx, prompt):
            raise RuntimeError("codex exec failed (1): quota")
        xreview.run_codex = boom
        out = xreview.gate(self.hook_input("gh pr create"), me="claude")
        self.assertIn("quota", out["hookSpecificOutput"]["permissionDecisionReason"])

    def test_allows_with_matching_receipt(self):
        ctx = self.context()
        xreview.write_receipt(ctx, reviewer="codex", result={"verdict": "approve", "summary": "ok", "findings": [], "next_steps": []})
        self.assertIsNone(xreview.gate(self.hook_input("gh pr create --fill"), me="claude"))

    def test_both_reviewers_can_hold_receipts_for_one_diff(self):
        ctx = self.context()
        ok = {"verdict": "approve", "summary": "ok", "findings": [], "next_steps": []}
        xreview.write_receipt(ctx, reviewer="codex", result=ok)
        xreview.write_receipt(ctx, reviewer="claude", result=ok)
        self.assertIsNone(xreview.gate(self.hook_input("gh pr create"), me="claude"))
        self.assertIsNone(xreview.gate(self.hook_input("gh pr create"), me="codex"))

    def test_receipt_from_wrong_reviewer_is_rejected(self):
        ctx = self.context()
        xreview.write_receipt(ctx, reviewer="claude", result={"verdict": "approve", "summary": "ok", "findings": [], "next_steps": []})
        self.assertIsNotNone(xreview.gate(self.hook_input("gh pr create --fill"), me="claude"))
        self.assertIsNone(xreview.gate(self.hook_input("gh pr create --fill"), me="codex"))

    def test_receipt_invalidated_by_new_commit(self):
        ctx = self.context()
        xreview.write_receipt(ctx, reviewer="codex", result={"verdict": "needs-attention", "summary": "x", "findings": [], "next_steps": []})
        (self.repo / "a.txt").write_text("three\n")
        git(self.repo, "commit", "-q", "-am", "more")
        self.assertIsNotNone(xreview.gate(self.hook_input("gh pr ready"), me="claude"))

    def test_findings_do_not_block(self):
        ctx = self.context()
        finding = {"severity": "high", "title": "t", "body": "b", "file": "a.txt", "line_start": 1, "line_end": 1, "confidence": 0.9, "recommendation": "r"}
        xreview.write_receipt(ctx, reviewer="codex", result={"verdict": "needs-attention", "summary": "x", "findings": [finding], "next_steps": []})
        self.assertIsNone(xreview.gate(self.hook_input("gh pr create"), me="claude"))

    def test_non_gated_command_is_silent(self):
        self.assertIsNone(xreview.gate(self.hook_input("gh pr list"), me="claude"))

    def test_skip_allows(self):
        self.assertIsNone(xreview.gate(self.hook_input("XREVIEW_SKIP=1 gh pr create"), me="claude"))

    def test_branch_equal_to_base_allows(self):
        git(self.repo, "checkout", "-q", "main")
        self.assertIsNone(xreview.gate(self.hook_input("gh pr create"), me="claude"))

    def test_outside_git_repo_is_denied(self):
        inp = self.hook_input("gh pr create")
        inp["cwd"] = self.tmp.name
        self.assertIsNotNone(xreview.gate(inp, me="claude"))

    def test_command_variants_run_review(self):
        self.fake_reviewer(OK)
        for command in [" gh pr create --fill", "env GH_TOKEN=test gh pr create", "gh --repo test/repo pr create", "/usr/bin/gh pr create"]:
            with self.subTest(command=command):
                self.assertIsNone(xreview.gate(self.hook_input(command), "claude"))
        self.assertEqual(len(self.calls), 1)

    def test_compound_commands_are_denied_before_review(self):
        self.fake_reviewer(OK)
        for command in ["cd /other && gh pr create", "git commit -am x && gh pr create", "git push && gh pr create", "echo XREVIEW_SKIP=1; gh pr create", "gh pr create\ngh pr ready", "bash -c 'gh pr create'", "echo $(gh pr create)"]:
            with self.subTest(command=command):
                self.assertIsNotNone(xreview.gate(self.hook_input(command), "claude"))
        self.assertEqual(self.calls, [])

    def test_different_head_is_denied(self):
        self.fake_reviewer(OK)
        self.assertIsNotNone(xreview.gate(self.hook_input("gh pr create --head other"), "claude"))
        self.assertEqual(self.calls, [])

    def test_explicit_base_is_reviewed(self):
        git(self.repo, "push", "origin", "HEAD:release")
        (self.repo / "a.txt").write_text("three\n")
        git(self.repo, "commit", "-qam", "more")
        self.fake_reviewer(OK)
        self.assertIsNone(xreview.gate(self.hook_input("gh pr create --base release"), "claude"))
        self.assertEqual(self.contexts[0].base_ref, "release")
        self.assertNotIn("+two", self.contexts[0].diff)
        self.assertIn("+three", self.contexts[0].diff)

    def test_other_ready_pr_head_is_denied(self):
        self.fake_reviewer(OK)
        original = self.gh_metadata
        def metadata(cwd, args, env):
            result = original(cwd, args, env)
            if args[:2] == ["pr", "view"]:
                result["headRefOid"] = git(self.repo, "rev-parse", "main")
            return result
        xreview._gh.side_effect = metadata
        self.assertIsNotNone(xreview.gate(self.hook_input("gh pr ready 123"), "claude"))
        self.assertEqual(self.calls, [])

    def test_other_repository_is_denied(self):
        self.fake_reviewer(OK)
        original = self.gh_metadata
        def metadata(cwd, args, env):
            result = original(cwd, args, env)
            if "other/repo" in args:
                result["url"] = "https://github.com/other/repo"
            return result
        xreview._gh.side_effect = metadata
        self.assertIsNotNone(xreview.gate(self.hook_input("gh pr create --repo other/repo"), "claude"))
        self.assertEqual(self.calls, [])

    def test_target_resolution_failure_is_denied(self):
        xreview._gh.side_effect = RuntimeError("missing remote")
        self.assertIsNotNone(xreview.gate(self.hook_input("gh pr create"), "claude"))

    def test_ready_uses_remote_pr_base(self):
        git(self.repo, "push", "origin", "HEAD:release")
        (self.repo / "a.txt").write_text("ready change\n")
        git(self.repo, "commit", "-qam", "ready change")
        original = self.gh_metadata
        def metadata(cwd, args, env):
            result = original(cwd, args, env)
            if args[:2] == ["pr", "view"]:
                self.assertEqual(args[2], "123")
                result["baseRefName"] = "release"
            return result
        xreview._gh.side_effect = metadata
        self.fake_reviewer(OK)
        self.assertIsNone(xreview.gate(self.hook_input("gh pr ready 123"), "claude"))
        self.assertEqual(self.contexts[0].base_ref, "release")
        self.assertNotIn("+two", self.contexts[0].diff)

    def test_missing_remote_is_denied(self):
        git(self.repo, "remote", "remove", "origin")
        self.assertIsNotNone(xreview.gate(self.hook_input("gh pr create"), "claude"))

    def test_remote_head_mismatch_is_denied(self):
        git(self.repo, "push", "origin", "HEAD:feature")
        (self.repo / "a.txt").write_text("not pushed\n")
        git(self.repo, "commit", "-qam", "local change")
        self.fake_reviewer(OK)
        self.assertIsNotNone(xreview.gate(self.hook_input("gh pr create"), "claude"))
        self.assertEqual(self.calls, [])

    def test_explicit_matching_remote_head_is_reviewed(self):
        git(self.repo, "push", "origin", "HEAD:feature")
        self.fake_reviewer(OK)
        self.assertIsNone(xreview.gate(self.hook_input("gh pr create --head feature"), "claude"))
        self.assertEqual(len(self.calls), 1)

    def test_missing_explicit_remote_head_is_denied(self):
        self.assertIsNotNone(xreview.gate(self.hook_input("gh pr create --head feature"), "claude"))

    def test_dirty_tree_with_receipt_is_denied(self):
        xreview.write_receipt(self.context(), "codex", OK)
        (self.repo / "a.txt").write_text("dirty\n")
        self.assertIsNotNone(xreview.gate(self.hook_input("gh pr create"), "claude"))

    def test_invalid_review_does_not_authorize_retry(self):
        self.fake_reviewer({"type": "result", "is_error": True})
        for _ in range(2):
            self.assertIsNotNone(xreview.gate(self.hook_input("gh pr create"), "claude"))
        self.assertEqual(len(self.calls), 2)
        self.assertIsNone(xreview.read_receipt(self.context(), "codex"))

    def test_context_change_during_review_does_not_write_receipt(self):
        self.fake_reviewer(OK)
        original = xreview.run_codex
        def runner(ctx, prompt):
            result = original(ctx, prompt)
            (self.repo / "a.txt").write_text("changed during review\n")
            git(self.repo, "commit", "-qam", "concurrent change")
            return result
        xreview.run_codex = runner
        old = self.context()
        self.assertIsNotNone(xreview.gate(self.hook_input("gh pr create"), "claude"))
        self.assertIsNone(xreview.read_receipt(old, "codex"))


 class Receipts(RepoCase):
    def test_receipt_identity_is_bound_to_context(self):
        ctx = xreview.branch_context(self.repo)
        xreview.write_receipt(ctx, "codex", OK)
        for changes in [{"root": "/other/repo"}, {"branch": "other"}, {"base_ref": "release"}, {"base_head": "new-base"}, {"merge_base": "new-merge-base"}, {"head": "new-head"}, {"target_repo": "other/repo"}]:
            with self.subTest(changes=changes):
                self.assertIsNone(xreview.read_receipt(replace(ctx, **changes), "codex"))

    def test_tampered_receipt_is_rejected(self):
        ctx = xreview.branch_context(self.repo)
        path = xreview.write_receipt(ctx, "codex", OK)
        receipt = json.loads(path.read_text())
        receipt["head"] = "other"
        path.write_text(json.dumps(receipt))
        self.assertIsNone(xreview.read_receipt(ctx, "codex"))

    def test_empty_commit_invalidates_receipt(self):
        ctx = xreview.branch_context(self.repo)
        xreview.write_receipt(ctx, "codex", OK)
        git(self.repo, "commit", "--allow-empty", "-qm", "empty")
        self.assertIsNone(xreview.read_receipt(xreview.branch_context(self.repo), "codex"))


 class ReviewerResults(unittest.TestCase):
    def setUp(self):
        self.ctx = xreview.BranchContext("/tmp", "feature", "main", "base", "head", "diff", "hash", False)

    def test_claude_error_envelopes_are_rejected(self):
        for code, envelope in [(1, {"structured_output": OK}), (0, {"type": "result", "is_error": True}), (0, {"subtype": "error_max_turns", "structured_output": OK}), (0, {"type": "result"})]:
            with self.subTest(code=code, envelope=envelope), patch.object(xreview.subprocess, "run", return_value=subprocess.CompletedProcess([], code, json.dumps(envelope), "failure")):
                with self.assertRaises(RuntimeError):
                    xreview.run_claude(self.ctx, "review")

    def test_claude_success_is_validated(self):
        envelope = {"type": "result", "subtype": "success", "is_error": False, "structured_output": OK}
        with patch.object(xreview.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, json.dumps(envelope), "")):
            self.assertEqual(xreview.run_claude(self.ctx, "review"), OK)

    def test_codex_nonzero_exit_rejects_valid_output(self):
        def run(cmd, **kwargs):
            Path(cmd[cmd.index("-o") + 1]).write_text(json.dumps(OK))
            return subprocess.CompletedProcess(cmd, 1, "", "failure")
        with patch.object(xreview.subprocess, "run", side_effect=run), self.assertRaises(RuntimeError):
            xreview.run_codex(self.ctx, "review")

    def test_invalid_schema_results_are_rejected(self):
        for result in [[], {}, {**OK, "summary": ""}, {**OK, "findings": "bad"}, {**BAD, "findings": [{**FINDING, "confidence": 2}]}, {**OK, "findings": [FINDING]}, {**BAD, "findings": [{**FINDING, "line_end": 0}]}]:
            with self.subTest(result=result), self.assertRaises(RuntimeError):
                xreview.validate_result(result)

    def test_hook_errors_emit_a_deny_payload(self):
        from io import StringIO
        import argparse
        output = StringIO()
        with patch.object(xreview.sys, "stdin", StringIO('{"tool_input": []}')), patch.object(xreview.sys, "stdout", output), patch.object(xreview, "gate", side_effect=OSError("cache unavailable")):
            self.assertEqual(xreview.cmd_gate(argparse.Namespace(me="claude")), 0)
        self.assertEqual(json.loads(output.getvalue())["hookSpecificOutput"]["permissionDecision"], "deny")


 if __name__ == "__main__":
    unittest.main()
diff --git a/xreview-prompt.md b/xreview-prompt.md
diff --git a/xreview-schema.json b/xreview-schema.json
 {
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "additionalProperties": false,
  "required": [
    "verdict",
    "summary",
    "findings",
    "next_steps"
  ],
  "properties": {
    "verdict": {
      "type": "string",
      "enum": [
        "approve",
        "needs-attention"
      ]
    },
    "summary": {
      "type": "string",
      "minLength": 1
    },
    "findings": {
      "type": "array",
      "items": {
        "type": "object",
        "additionalProperties": false,
        "required": [
          "severity",
          "title",
          "body",
          "file",
          "line_start",
          "line_end",
          "confidence",
          "recommendation"
        ],
        "properties": {
          "severity": {
            "type": "string",
            "enum": [
              "critical",
              "high",
              "medium",
              "low"
            ]
          },
          "title": {
            "type": "string",
            "minLength": 1
          },
          "body": {
            "type": "string",
            "minLength": 1
          },
          "file": {
            "type": "string",
            "minLength": 1
          },
          "line_start": {
            "type": "integer",
            "minimum": 1
          },
          "line_end": {
            "type": "integer",
            "minimum": 1
          },
          "confidence": {
            "type": "number",
            "minimum": 0,
            "maximum": 1
          },
          "recommendation": {
            "type": "string"
          }
        }
      }
    },
    "next_steps": {
      "type": "array",
      "items": {
        "type": "string",
        "minLength": 1
      }
    }
  }
 }
diff --git a/xreview.py b/xreview.py
 #!/usr/bin/env python3
 """Cross-model review gate: code opened as a PR from one harness must carry an
 adversarial review by the other model for the exact committed diff.

  xreview.py gate --me claude|codex      PreToolUse hook entry (JSON on stdin)
  xreview.py run codex|claude [--base REF] [focus ...]
  xreview.py status
 """

 from __future__ import annotations

 import argparse
 import hashlib
 import json
 import os
 import re
 import shlex
 import shutil
 import subprocess
 import sys
 import tempfile
 import time
 from dataclasses import dataclass
 from pathlib import Path
 from urllib.parse import urlsplit

 HERE = Path(__file__).resolve().parent
 PROMPT_FILE = HERE / "xreview-prompt.md"
 SCHEMA_FILE = HERE / "xreview-schema.json"
 OTHER = {"claude": "codex", "codex": "claude"}
 # Diffs beyond this are truncated in the prompt; the reviewer can still read files directly.
 MAX_DIFF_BYTES = 400_000
 GATED_RE = re.compile(r"\bgh\s+(?:[^\n;&|]*?\s+)?pr\s+(?:create|ready)\b")
 ASSIGNMENT_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*=")


 @dataclass
 class Invocation:
    action: str
    args: list[str]
    env: dict[str, str]
    cwd: str


 def command_invocations(command: str, cwd: str) -> list[Invocation]:
    lexer = shlex.shlex((command or "").strip(), posix=True, punctuation_chars=";&|()<>\n")
    lexer.whitespace = " \t\r"
    lexer.whitespace_split = True
    tokens = list(lexer)
    segments, segment = [], []
    for token in tokens:
        if token and all(c in ";&|()<>\n" for c in token):
            if segment:
                segments.append(segment)
                segment = []
            segments.append(token)
        else:
            segment.append(token)
    if segment:
        segments.append(segment)
    invocations = []
    for words in segments:
        if isinstance(words, str):
            continue
        i, env = 0, {}
        while i < len(words):
            word = words[i]
            if word in {"env", "command"} or word == "--":
                i += 1
            elif ASSIGNMENT_RE.match(word):
                key, value = word.split("=", 1)
                env[key] = value
                i += 1
            else:
                break
        if i == len(words) or Path(words[i]).name != "gh":
            continue
        i += 1
        global_args = []
        while i < len(words) and words[i].startswith("-"):
            word = words[i]
            global_args.append(word)
            i += 1
            if word in {"-R", "--repo", "--hostname"} and i < len(words):
                global_args.append(words[i])
                i += 1
        if i < len(words) and words[i] == "pr":
            i += 1
            while i < len(words) and words[i].startswith("-"):
                word = words[i]
                global_args.append(word)
                i += 1
                if word in {"-R", "--repo", "--hostname"} and i < len(words):
                    global_args.append(words[i])
                    i += 1
            if i < len(words) and words[i] in {"create", "ready"}:
                invocations.append(Invocation(words[i], global_args + words[i + 1:], env, cwd))
    if invocations:
        if len(segments) != 1 or any("$" in token or "`" in token for token in tokens):
            raise RuntimeError("Run gh pr create/ready as a separate command, without shell substitutions or redirections.")
        forbidden = set(invocations[0].env) - {"XREVIEW_SKIP", "GH_TOKEN", "GITHUB_TOKEN"}
        if forbidden:
            raise RuntimeError("Unsupported environment overrides for the PR command: " + ", ".join(sorted(forbidden)))
    elif GATED_RE.search(command or ""):
        inert = len(segments) == 1 and isinstance(segments[0], list) and segments[0][0] in {"echo", "printf"}
        if not inert or "$(" in command or "`" in command:
            raise RuntimeError("Run gh pr create/ready directly so its repository and target can be reviewed.")
    return invocations


 def cache_dir() -> Path:
    d = Path(os.environ.get("XREVIEW_CACHE") or Path.home() / ".cache" / "xreview")
    d.mkdir(parents=True, exist_ok=True)
    return d


 def is_gated_command(command: str) -> bool:
    try:
        return bool(command_invocations(command, os.getcwd()))
    except (ValueError, RuntimeError):
        return bool(GATED_RE.search(command or ""))


 def has_skip(command: str) -> bool:
    try:
        calls = command_invocations(command, os.getcwd())
        return len(calls) == 1 and calls[0].env.get("XREVIEW_SKIP") == "1"
    except (ValueError, RuntimeError):
        return False


 def _git(cwd, *args, check=True) -> str:
    r = subprocess.run(["git", "-C", str(cwd), *args], capture_output=True, text=True, timeout=120)
    if check and r.returncode != 0:
        raise RuntimeError(r.stderr.strip() or f"git {' '.join(args)} failed")
    return r.stdout.strip()


 @dataclass
 class BranchContext:
    root: str
    branch: str
    base_ref: str
    merge_base: str
    head: str
    diff: str
    diff_hash: str
    dirty: bool
    base_head: str = ""
    target_repo: str = ""

    @property
    def has_changes(self) -> bool:
        return bool(self.diff.strip())


 def default_base_ref(root) -> str:
    try:
        return _git(root, "symbolic-ref", "-q", "--short", "refs/remotes/origin/HEAD")
    except RuntimeError:
        return "origin/main"


 def branch_context(cwd, base_ref: str | None = None) -> BranchContext:
    root = _git(cwd, "rev-parse", "--show-toplevel")
    branch = _git(root, "rev-parse", "--abbrev-ref", "HEAD")
    base_ref = base_ref or default_base_ref(root)
    head = _git(root, "rev-parse", "HEAD")
    merge_base = _git(root, "merge-base", base_ref, "HEAD")
    base_head = _git(root, "rev-parse", "--verify", f"{base_ref}^{{commit}}")
    diff = _git(root, "diff", "--no-ext-diff", "--no-textconv", "--no-color", merge_base, head)
    dirty = bool(_git(root, "status", "--porcelain", "--untracked-files=no"))
    digest = hashlib.sha256(diff.encode("utf-8", "surrogateescape")).hexdigest()
    return BranchContext(root, branch, base_ref, merge_base, head, diff, digest, dirty, base_head)


 def _gh(cwd: str, args: list[str], env: dict[str, str]) -> dict:
    r = subprocess.run(["gh", *args], cwd=cwd, env={**os.environ, **env}, capture_output=True, text=True, timeout=120)
    if r.returncode:
        raise RuntimeError(f"Cannot resolve PR target: {r.stderr.strip()[-2000:]}")
    return _parse_json_object(r.stdout, "gh")


 def invocation_options(call: Invocation) -> tuple[dict[str, str], list[str]]:
    values = {"-R": "repo", "--repo": "repo", "--hostname": "hostname", "-B": "base", "--base": "base",
              "-H": "head", "--head": "head", "-t": "title", "--title": "title", "-b": "body", "--body": "body",
              "-F": "body-file", "--body-file": "body-file", "-a": "assignee", "--assignee": "assignee",
              "-l": "label", "--label": "label", "-m": "milestone", "--milestone": "milestone",
              "-p": "project", "--project": "project", "-r": "reviewer", "--reviewer": "reviewer", "-T": "template", "--template": "template"}
    switches = {"-d", "--draft", "--fill", "--fill-first", "--fill-verbose", "--no-maintainer-edit", "--undo"}
    options, positional = {}, []
    i = 0
    while i < len(call.args):
        arg = call.args[i]
        key, sep, value = arg.partition("=")
        if key in values:
            if not sep:
                i += 1
                if i == len(call.args):
                    raise RuntimeError(f"Missing value for {key}")
                value = call.args[i]
            options[values[key]] = value
        elif arg in switches:
            pass
        elif arg.startswith("-"):
            raise RuntimeError(f"Unsupported PR option {arg}; use explicit options without interactive target selection.")
        else:
            positional.append(arg)
        i += 1
    if "hostname" in options:
        raise RuntimeError("Use gh --repo HOST/OWNER/REPO to select a hostname explicitly.")
    if (call.action == "create" and positional) or len(positional) > 1:
        raise RuntimeError("Cannot resolve PR command arguments.")
    return options, positional


 def repository_key(url: str) -> str:
    if "://" in url:
        parsed = urlsplit(url)
        return f"{parsed.hostname}/{parsed.path.strip('/').removesuffix('.git')}".lower()
    if re.match(r"^[^/]+@[^/:]+:", url):
        host, path = url.split("@", 1)[1].split(":", 1)
        return f"{host}/{path.strip('/').removesuffix('.git')}".lower()
    return str(Path(url).resolve())


 def target_context(call: Invocation) -> BranchContext:
    options, positional = invocation_options(call)
    root = _git(call.cwd, "rev-parse", "--show-toplevel")
    branch = _git(root, "symbolic-ref", "--quiet", "--short", "HEAD")
    head = _git(root, "rev-parse", "HEAD")
    fields = "nameWithOwner,defaultBranchRef,url"
    repo = _gh(root, ["repo", "view", "--json", fields], call.env)
    if "repo" in options:
        selected = _gh(root, ["repo", "view", options["repo"], "--json", fields], call.env)
        if selected.get("url") != repo.get("url"):
            raise RuntimeError("The --repo target differs from this checkout. Run the PR command in the target repository.")
        repo = selected
    url = repo.get("url")
    if not isinstance(url, str) or not url or url.startswith("-"):
        raise RuntimeError("Cannot resolve the PR repository URL.")
    remotes = _git(root, "remote").splitlines()
    remote_urls = [_git(root, "remote", "get-url", remote) for remote in remotes]
    if not remote_urls or any(repository_key(remote) != repository_key(url) for remote in remote_urls):
        raise RuntimeError("The PR repository does not uniquely match this checkout's remotes. Use a checkout of the target repository.")
    base = options.get("base")
    if call.action == "ready":
        selector = positional or [branch]
        pr = _gh(root, ["pr", "view", *selector, "--repo", repo["url"], "--json", "headRefOid,headRefName,baseRefName"], call.env)
        if pr.get("headRefOid") != head or pr.get("headRefName") != branch:
            raise RuntimeError("The PR head differs from this checkout. Check out its exact branch and commit before marking it ready.")
        base = pr.get("baseRefName")
    else:
        if options.get("head", branch) != branch:
            raise RuntimeError("The --head target differs from this checkout. Check out the target branch before creating the PR.")
        remote_head = _git(root, "ls-remote", "--", url, f"refs/heads/{branch}")
        if remote_head and remote_head.split()[0] != head:
            raise RuntimeError("The remote PR branch differs from HEAD. Push the reviewed local commit first, then create the PR separately.")
        if "head" in options and not remote_head:
            raise RuntimeError("The explicit --head branch does not exist remotely. Push it first, then retry.")
        configured = _git(root, "config", "--get", f"branch.{branch}.gh-merge-base", check=False)
        base = base or configured or (repo.get("defaultBranchRef") or {}).get("name")
    if not isinstance(base, str) or not base:
        raise RuntimeError("Cannot resolve the PR base branch.")
    if subprocess.run(["git", "check-ref-format", f"refs/heads/{base}"], capture_output=True).returncode:
        raise RuntimeError("Invalid PR base branch.")
    _git(root, "fetch", "--no-tags", "--", url, f"refs/heads/{base}")
    base_head = _git(root, "rev-parse", "FETCH_HEAD")
    ctx = branch_context(root, base_head)
    if ctx.head != head or ctx.branch != branch:
        raise RuntimeError("The checked-out branch changed while resolving the PR target. Retry.")
    ctx.base_ref = base
    ctx.target_repo = url
    return ctx


 def receipt_identity(ctx: BranchContext, reviewer: str) -> dict:
    return {"version": 2, "repo": ctx.root, "branch": ctx.branch, "target_repo": ctx.target_repo,
            "base_ref": ctx.base_ref, "base_head": ctx.base_head, "merge_base": ctx.merge_base,
            "head": ctx.head, "diff_hash": ctx.diff_hash, "reviewer": reviewer}


 def receipt_path(ctx: BranchContext, reviewer: str) -> Path:
    digest = hashlib.sha256(json.dumps(receipt_identity(ctx, reviewer), sort_keys=True).encode()).hexdigest()
    return cache_dir() / f"{digest}.{reviewer}.json"


 def read_receipt(ctx: BranchContext, reviewer: str):
    p = receipt_path(ctx, reviewer)
    if not p.exists():
        return None
    try:
        receipt = json.loads(p.read_text())
        if not isinstance(receipt, dict) or any(receipt.get(k) != v for k, v in receipt_identity(ctx, reviewer).items()):
            return None
        validate_result({k: receipt.get(k) for k in ("verdict", "summary", "findings", "next_steps")})
        return receipt
    except (OSError, ValueError, RuntimeError):
        return None


 def read_any_receipt(ctx: BranchContext):
    for r in OTHER:
        rec = read_receipt(ctx, r)
        if rec:
            return rec
    return None


 def write_receipt(ctx: BranchContext, reviewer: str, result: dict) -> Path:
    validate_result(result)
    receipt = {
        **receipt_identity(ctx, reviewer),
        "verdict": result.get("verdict"),
        "summary": result.get("summary"),
        "findings": result.get("findings", []),
        "next_steps": result.get("next_steps", []),
        "reviewed_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
    }
    p = receipt_path(ctx, reviewer)
    with tempfile.NamedTemporaryFile("w", dir=p.parent, delete=False) as tmp:
        tmp.write(json.dumps(receipt, indent=1))
    os.replace(tmp.name, p)
    return p


 def log(line: str) -> None:
    with open(cache_dir() / "log", "a") as f:
        f.write(f"{time.strftime('%Y-%m-%dT%H:%M:%S%z')} {line}\n")


 def deny(reason: str) -> dict:
    return {"hookSpecificOutput": {"hookEventName": "PreToolUse",
                                   "permissionDecision": "deny",
                                   "permissionDecisionReason": reason}}


 def gate(hook_input: dict, me: str):
    """Return a deny payload, or None to let the command through."""
    if hook_input.get("tool_name", "Bash") != "Bash":
        return None
    command = (hook_input.get("tool_input") or {}).get("command", "")
    reviewer = OTHER[me]
    cwd = hook_input.get("cwd") or os.getcwd()
    try:
        calls = command_invocations(command, cwd)
    except (RuntimeError, ValueError) as e:
        return deny(f"Blocked: cannot safely resolve PR command. {e}")
    if not calls:
        return None
    call = calls[0]
    if call.env.get("XREVIEW_SKIP") == "1":
        log(f"SKIP me={me} cwd={cwd} action={call.action}")
        return None
    try:
        ctx = target_context(call)
    except (RuntimeError, OSError, ValueError, KeyError, subprocess.TimeoutExpired) as e:
        log(f"DENY me={me} cwd={cwd} target-error={e}")
        return deny(f"Blocked: cannot resolve the PR's exact repository, head and base. {e}\nFix the target or Git/GitHub access and retry.")
    if ctx.dirty:
        return deny(f"Blocked: '{ctx.branch}' has uncommitted tracked changes. Commit (or stash) first, then retry.")
    if not ctx.has_changes:
        return None
    receipt = read_receipt(ctx, reviewer)
    if receipt:
        log(f"ALLOW me={me} branch={ctx.branch} hash={ctx.diff_hash[:12]} verdict={receipt.get('verdict')}")
        return None
    if shutil.which(reviewer) is None:
        log(f"DENY me={me} branch={ctx.branch} reviewer-missing={reviewer}")
        return deny(f"Blocked: '{ctx.branch}' needs an adversarial review by {reviewer} before a PR is opened, and the "
                    f"'{reviewer}' CLI is not on PATH. Install it or ask Marius; XREVIEW_SKIP=1 only if he says to skip.")
    log(f"REVIEW-START me={me} by={reviewer} branch={ctx.branch} hash={ctx.diff_hash[:12]}")
    print(f"xreview: running {reviewer} adversarial review of {ctx.branch} against {ctx.base_ref}...", file=sys.stderr)
    try:
        prompt = build_prompt(ctx, reviewer, "")
        result = run_codex(ctx, prompt) if reviewer == "codex" else run_claude(ctx, prompt)
        validate_result(result)
        current = target_context(call)
        if current.dirty or receipt_identity(current, reviewer) != receipt_identity(ctx, reviewer):
            raise RuntimeError("The PR target or working tree changed during review. Retry against the current commits.")
        path = write_receipt(ctx, reviewer, result)
    except (RuntimeError, OSError, ValueError, KeyError, subprocess.TimeoutExpired) as e:
        log(f"DENY me={me} branch={ctx.branch} review-error={e}")
        return deny(f"Blocked: the {reviewer} adversarial review of '{ctx.branch}' failed, so the PR cannot be opened "
                    f"yet. Error:\n{e}\n\nFix the cause or rerun: python3 {HERE / 'xreview.py'} run {reviewer}")
    findings = result.get("findings") or []
    log(f"REVIEWED by={reviewer} branch={ctx.branch} hash={ctx.diff_hash[:12]} verdict={result.get('verdict')} findings={len(findings)}")
    if result.get("verdict") == "approve" and not findings:
        log(f"ALLOW me={me} branch={ctx.branch} hash={ctx.diff_hash[:12]} verdict=approve")
        return None
    return deny(
        f"Blocked: {reviewer} reviewed '{ctx.branch}' and found material issues. Fix the ones you agree with, commit, "
        f"and retry; the retry reviews the new diff. If you disagree with every finding, retry unchanged and say why "
        f"in the PR description.\n\n" + render(result, reviewer, path)
    )


 def _any_receipt_for_branch(ctx: BranchContext) -> bool:
    for p in cache_dir().glob("*.json"):
        try:
            r = json.loads(p.read_text())
        except json.JSONDecodeError:
            continue
        if r.get("repo") == ctx.root and r.get("branch") == ctx.branch:
            return True
    return False


 def build_prompt(ctx: BranchContext, reviewer: str, focus: str) -> str:
    stat = _git(ctx.root, "diff", "--stat", ctx.merge_base, ctx.head)
    diff = ctx.diff
    if len(diff) > MAX_DIFF_BYTES:
        diff = diff[:MAX_DIFF_BYTES] + "\n\n[diff truncated; read the repository for the rest]\n"
    review_input = (
        f"Repository: {ctx.root}\nBranch: {ctx.branch}\nBase: {ctx.base_ref} (merge-base {ctx.merge_base[:12]})\n"
        f"Head: {ctx.head[:12]}\n\nChanged files:\n{stat}\n\nDiff:\n{diff}"
    )
    tpl = PROMPT_FILE.read_text()
    return (tpl.replace("{{REVIEWER}}", reviewer.capitalize())
               .replace("{{TARGET_LABEL}}", f"branch {ctx.branch} against {ctx.base_ref}")
               .replace("{{USER_FOCUS}}", focus or "none")
               .replace("{{REVIEW_INPUT}}", review_input))


 def run_codex(ctx: BranchContext, prompt: str) -> dict:
    with tempfile.NamedTemporaryFile("r", suffix=".json", delete=False) as out:
        out_path = out.name
    cmd = ["codex", "exec", "--sandbox", "read-only", "--ephemeral", "--color", "never",
           "-C", ctx.root, "--output-schema", str(SCHEMA_FILE), "-o", out_path, "-"]
    try:
        r = subprocess.run(cmd, input=prompt, capture_output=True, text=True, timeout=720)
        text = Path(out_path).read_text().strip()
    finally:
        Path(out_path).unlink(missing_ok=True)
    if r.returncode != 0:
        raise RuntimeError(f"codex exec failed ({r.returncode}): {r.stderr.strip()[-2000:]}")
    return validate_result(_parse_json_object(text, "codex"))


 def run_claude(ctx: BranchContext, prompt: str) -> dict:
    # claude -p's validator does not resolve the 2020-12 meta-schema URI; codex exec needs the file as is.
    schema = json.loads(SCHEMA_FILE.read_text())
    schema.pop("$schema", None)
    # --restricted ignores the user's settings (auto mode, kubectl/aws allow rules, hooks), so the
    # reviewer can only read the repo and run the git subcommands listed here.
    cmd = ["claude", "-p", "--output-format", "json", "--json-schema", json.dumps(schema),
           "--no-session-persistence", "--restricted", "--strict-mcp-config", "--permission-mode", "dontAsk",
           "--tools", "Read,Grep,Glob,Bash",
           "--allowedTools", "Read", "Grep", "Glob", "Bash(git diff:*)", "Bash(git log:*)", "Bash(git show:*)",
           "Bash(git blame:*)", "--", prompt]
    r = subprocess.run(cmd, cwd=ctx.root, capture_output=True, text=True, timeout=720)
    if r.returncode != 0:
        raise RuntimeError(f"claude -p failed ({r.returncode}): {r.stderr.strip()[-2000:]}")
    envelope = _parse_json_object(r.stdout, "claude")
    if envelope.get("is_error") or ("subtype" in envelope and envelope["subtype"] != "success"):
        raise RuntimeError(f"claude review failed: {str(envelope)[-2000:]}")
    if isinstance(envelope.get("structured_output"), dict):
        return validate_result(envelope["structured_output"])
    result = envelope.get("result")
    if isinstance(result, dict):
        return validate_result(result)
    if isinstance(result, str):
        return validate_result(_parse_json_object(result, "claude result"))
    return validate_result(envelope)


 def validate_result(result: dict) -> dict:
    def validate(value, schema, path):
        kind = schema.get("type")
        types = {"object": dict, "array": list, "string": str, "integer": int, "number": (int, float)}
        if kind in types and (not isinstance(value, types[kind]) or (kind in {"number", "integer"} and isinstance(value, bool))):
            raise RuntimeError(f"Invalid review result: {path} must be {kind}")
        if "enum" in schema and value not in schema["enum"]:
            raise RuntimeError(f"Invalid review result: {path} has an unsupported value")
        if kind == "object":
            properties = schema.get("properties", {})
            if any(key not in value for key in schema.get("required", [])) or (schema.get("additionalProperties") is False and set(value) - set(properties)):
                raise RuntimeError(f"Invalid review result: {path} has missing or unknown fields")
            for key, item in value.items():
                if key in properties:
                    validate(item, properties[key], f"{path}.{key}")
        elif kind == "array":
            for i, item in enumerate(value):
                validate(item, schema["items"], f"{path}[{i}]")
        elif kind == "string" and len(value.strip()) < schema.get("minLength", 0):
            raise RuntimeError(f"Invalid review result: {path} is empty")
        elif kind in {"number", "integer"}:
            if not schema.get("minimum", -float("inf")) <= value <= schema.get("maximum", float("inf")):
                raise RuntimeError(f"Invalid review result: {path} is out of range")

    validate(result, json.loads(SCHEMA_FILE.read_text()), "result")
    if result["verdict"] == "approve" and result["findings"]:
        raise RuntimeError("Invalid review result: approval contains findings")
    if any(f["line_end"] < f["line_start"] for f in result["findings"]):
        raise RuntimeError("Invalid review result: reversed finding line range")
    return result


 def _parse_json_object(text: str, who: str) -> dict:
    text = text.strip()
    try:
        result = json.loads(text)
        if isinstance(result, dict):
            return result
    except json.JSONDecodeError:
        start, end = text.find("{"), text.rfind("}")
        if start >= 0 and end > start:
            try:
                result = json.loads(text[start:end + 1])
                if isinstance(result, dict):
                    return result
            except json.JSONDecodeError:
                pass
    raise RuntimeError(f"{who} returned no parseable JSON:\n{text[-2000:]}")


 def render(result: dict, reviewer: str, path: Path) -> str:
    lines = [f"{reviewer} adversarial review: {result.get('verdict', '?')}", "", result.get("summary", ""), ""]
    findings = result.get("findings") or []
    if findings:
        lines.append(f"Findings ({len(findings)}):")
        for i, f in enumerate(findings, 1):
            lines.append(f"\n{i}. [{f.get('severity')}] {f.get('title')}  ({f.get('file')}:{f.get('line_start')}-{f.get('line_end')}, confidence {f.get('confidence')})")
            lines.append(f"   {f.get('body')}")
            if f.get("recommendation"):
                lines.append(f"   Recommendation: {f['recommendation']}")
    else:
        lines.append("No findings.")
    steps = result.get("next_steps") or []
    if steps:
        lines += ["", "Next steps:"] + [f"- {s}" for s in steps]
    lines += ["", f"Receipt: {path} (any new commit invalidates it)"]
    return "\n".join(lines)


 def cmd_run(args) -> int:
    reviewer = args.reviewer
    call = Invocation("create", ["--base", args.base] if args.base else [], {}, os.getcwd())
    ctx = target_context(call)
    if ctx.dirty:
        print("Working tree has uncommitted tracked changes. Commit first so the review covers what the PR will contain.", file=sys.stderr)
        return 2
    if not ctx.has_changes:
        print(f"Nothing to review: {ctx.branch} has no committed diff against {ctx.base_ref}.", file=sys.stderr)
        return 2
    if shutil.which(reviewer) is None:
        print(f"'{reviewer}' CLI not found on PATH.", file=sys.stderr)
        return 2
    prompt = build_prompt(ctx, reviewer, " ".join(args.focus))
    print(f"Running {reviewer} adversarial review of {ctx.branch} against {ctx.base_ref} "
          f"({len(ctx.diff.splitlines())} diff lines)...", file=sys.stderr)
    result = run_codex(ctx, prompt) if reviewer == "codex" else run_claude(ctx, prompt)
    current = target_context(call)
    if current.dirty or receipt_identity(current, reviewer) != receipt_identity(ctx, reviewer):
        raise RuntimeError("The PR target or working tree changed during review. Retry against the current commits.")
    path = write_receipt(ctx, reviewer, result)
    log(f"REVIEWED by={reviewer} branch={ctx.branch} hash={ctx.diff_hash[:12]} verdict={result.get('verdict')} findings={len(result.get('findings') or [])}")
    print(render(result, reviewer, path))
    return 0


 def cmd_status(args) -> int:
    try:
        ctx = target_context(Invocation("create", ["--base", args.base] if args.base else [], {}, os.getcwd()))
    except RuntimeError as e:
        print(f"cannot resolve PR target: {e}", file=sys.stderr)
        return 2
    print(f"branch={ctx.branch} base={ctx.base_ref} hash={ctx.diff_hash[:12]} dirty={ctx.dirty} changes={ctx.has_changes}")
    found = False
    for reviewer in OTHER:
        r = read_receipt(ctx, reviewer)
        if r:
            found = True
            print(f"receipt: reviewer={reviewer} verdict={r['verdict']} findings={len(r.get('findings') or [])} at={r['reviewed_at']}")
    if not found:
        print("receipt: none for this diff")
    return 0


 def cmd_gate(args) -> int:
    try:
        raw = sys.stdin.read().strip()
        hook_input = json.loads(raw) if raw else {}
        out = gate(hook_input, args.me)
    except Exception as e:
        out = deny(f"Blocked: xreview could not complete the gate. {e}")
    if out:
        print(json.dumps(out))
    return 0


 def main(argv=None) -> int:
    p = argparse.ArgumentParser(prog="xreview")
    sub = p.add_subparsers(dest="cmd", required=True)
    g = sub.add_parser("gate")
    g.add_argument("--me", choices=OTHER, required=True)
    g.set_defaults(fn=cmd_gate)
    r = sub.add_parser("run")
    r.add_argument("reviewer", choices=OTHER)
    r.add_argument("--base")
    r.add_argument("focus", nargs="*")
    r.set_defaults(fn=cmd_run)
    s = sub.add_parser("status")
    s.add_argument("--base")
    s.set_defaults(fn=cmd_status)
    args = p.parse_args(argv)
    try:
        return args.fn(args)
    except RuntimeError as e:
        print(f"xreview: {e}", file=sys.stderr)
        return 1


 if __name__ == "__main__":
    sys.exit(main())
	#!/usr/bin/env python3
	"""Enforce a why-only policy for comments added by coding agents."""

	from __future__ import annotations

	import collections
	import hashlib
	import itertools
	import json
	import os
	import re
	import subprocess
	import sys
	import tempfile
	from dataclasses import dataclass
	from pathlib import Path


	SLASH = {
	".go", ".js", ".ts", ".tsx", ".jsx", ".mjs", ".cjs", ".java", ".rs",
	".c", ".h", ".hpp", ".cpp", ".cc", ".proto", ".kt", ".swift", ".scala", ".dart",
	}
	HASH = {
	".py", ".sh", ".bash", ".zsh", ".yaml", ".yml", ".tf", ".tfvars",
	".toml", ".rb", ".mk", ".pl", ".ps1", ".conf", ".cfg",
	}
	DASH = {".sql", ".lua", ".hs"}
	HASH_NAMES = {"Makefile", "Dockerfile", "Justfile", "Vagrantfile"}

	POLICY = (
	"Comments explain only non-obvious decisions, constraints, safety invariants, or workarounds. "
	"Do not narrate what code does or record change history; git holds history. Keep comments to "
	"one or two tight lines unless a longer explanation is genuinely necessary."
	)

	CHANGE_HISTORY = re.compile(
	r"\b(previously\|used to\|no longer (?:needed\|used\|required\|exists\|applies)\|"
	r"we (?:removed\|added\|changed\|moved\|renamed\|dropped\|switched)\|"
	r"removed (?:in favor\|because)\|renamed (?:from\|to)\|replaced by\|"
	r"this (?:was\|used to)\|has been (?:removed\|moved\|renamed\|replaced))\b",
	re.I,
	)
	NARRATION = re.compile(
	r"^(?:(?:this\|the\|these\|those)\s+.{0,60}\s+"
	r"(?:is\|are\|does\|do\|will\|contains\|returns\|sets\|gets\|creates\|updates\|deletes\|"
	r"loops\|iterates\|checks\|verifies\|loads\|stores\|writes\|reads\|calls\|uses\|handles\|"
	r"initializes\|converts\|parses\|builds\|starts\|stops\|opens\|closes)\|"
	r"(?:set\|get\|create\|update\|delete\|return\|loop\|iterate\|check\|verify\|load\|save\|"
	r"store\|call\|add\|remove\|initialize\|convert\|parse\|build\|handle\|run\|start\|stop\|"
	r"open\|close\|write\|read)\b)",
	re.I,
	)
	WHY = re.compile(
	r"\b(because\|otherwise\|unless\|until\|to (?:avoid\|prevent\|preserve\|ensure\|keep)\|"
	r"must\|cannot\|can't\|required\|invariant\|race\|deadlock\|data loss\|durab(?:le\|ility)\|"
	r"compatibility\|security\|atomic(?:ity\|ally)?\|workaround)\b",
	re.I,
	)
	EXEMPT = re.compile(
	r"^(?:!\|todo\b\|fixme\b\|hack\b\|xxx\b\|spdx-\|copyright\b\|license\b\|"
	r"code generated\b\|go:\|nolint\b\|nosec\b\|noqa\b\|type:\s*ignore\b\|"
	r"eslint\b\|prettier\b\|shellcheck\b\|yamllint\b\|hadolint\b\|pragma\b\|"
	r"fmt:\|line\b\|sourceMappingURL=\|region\b\|endregion\b\|cspell\b)",
	re.I,
	)
	COMMENTED_CODE = re.compile(r"^\s(?:\S+\s[:=]\|-\s\|[{}\[\]])")
	HUNK = re.compile(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,\d+)? @@")


	@dataclass(frozen=True)
	class AddedLine:
	path: str
	number: int
	text: str
	group: int


	@dataclass(frozen=True)
	class Finding:
	path: str
	number: int
	rule: str
	text: str

	def fingerprint(self) -> str:
	normalized = " ".join(self.text.split()).lower()
	return f"{self.path}\0{self.rule}\0{normalized}"


	def marker_for(path: str) -> str \| None:
	base = os.path.basename(path)
	if base in HASH_NAMES:
	return "#"
	ext = os.path.splitext(base)[1].lower()
	if ext in SLASH:
	return "//"
	if ext in HASH:
	return "#"
	if ext in DASH:
	return "--"
	return None


	def comment_body(line: str, marker: str) -> str \| None:
	stripped = line.strip()
	if stripped.startswith(marker):
	return stripped[len(marker):].strip()
	if marker == "//" and (stripped.startswith("/") or stripped.startswith("")):
	return stripped.lstrip("/* ").rstrip("*/ ").strip()
	return None


	def inline_body(line: str, marker: str) -> str \| None:
	token = " " + marker
	pos = line.find(token)
	if pos < 0:
	return None
	return line[pos + len(token):].strip()


	def analyze(lines: list[AddedLine]) -> list[Finding]:
	findings: list[Finding] = []
	for path, path_lines_iter in _group_by(lines, lambda line: line.path):
	marker = marker_for(path)
	if not marker:
	continue
	path_lines = list(path_lines_iter)
	for _, group_iter in _group_by(path_lines, lambda line: line.group):
	group_lines = list(group_iter)
	comment_count = 0
	code_count = 0
	runs: list[list[AddedLine]] = []
	run: list[AddedLine] = []
	for line in group_lines:
	body = comment_body(line.text, marker)
	if body is None:
	if run:
	runs.append(run)
	run = []
	if line.text.strip():
	code_count += 1
	inline = inline_body(line.text, marker)
	if inline and CHANGE_HISTORY.search(inline):
	findings.append(Finding(line.path, line.number, "change-history", inline))
	continue
	if not body or EXEMPT.search(body) or COMMENTED_CODE.match(body):
	if run:
	runs.append(run)
	run = []
	continue
	comment_count += 1
	if run and line.number != run[-1].number + 1:
	runs.append(run)
	run = []
	run.append(line)
	findings.extend(_prose_findings(line, body))
	if run:
	runs.append(run)
	for comment_run in runs:
	if len(comment_run) >= 3:
	findings.append(Finding(
	path,
	comment_run[0].number,
	"long-block",
	f"{len(comment_run)} consecutive comment lines",
	))
	if comment_count >= 3 and comment_count > code_count:
	first = next((line for line in group_lines if comment_body(line.text, marker) is not None), group_lines[0])
	findings.append(Finding(
	path,
	first.number,
	"comment-density",
	f"{comment_count} comment lines for {code_count} code lines in one change",
	))
	return _deduplicate(findings)


	def _prose_findings(line: AddedLine, body: str) -> list[Finding]:
	if CHANGE_HISTORY.search(body):
	return [Finding(line.path, line.number, "change-history", body)]
	if NARRATION.search(body) and not WHY.search(body):
	return [Finding(line.path, line.number, "code-narration", body)]
	return []


	def _group_by(items, key):
	return itertools.groupby(items, key)


	def _deduplicate(findings: list[Finding]) -> list[Finding]:
	seen: set[tuple[str, int, str]] = set()
	result: list[Finding] = []
	for finding in findings:
	key = (finding.path, finding.number, finding.rule)
	if key not in seen:
	seen.add(key)
	result.append(finding)
	return result


	def parse_apply_patch(patch: str) -> list[AddedLine]:
	lines: list[AddedLine] = []
	path = ""
	number = 1
	group = 0
	for raw in patch.splitlines():
	if raw.startswith("* Update File: ") or raw.startswith("* Add File: "):
	path = raw.split(": ", 1)[1]
	number = 1
	group += 1
	continue
	match = HUNK.match(raw)
	if match:
	number = int(match.group(1))
	group += 1
	continue
	if not path:
	continue
	if raw.startswith("+") and not raw.startswith("+++"):
	lines.append(AddedLine(path, number, raw[1:], group))
	number += 1
	elif raw.startswith(" "):
	number += 1
	elif raw.startswith("-"):
	continue
	return lines


	def parse_git_diff(diff: str) -> list[AddedLine]:
	lines: list[AddedLine] = []
	path = ""
	number = 1
	group = 0
	for raw in diff.splitlines():
	if raw.startswith("+++ "):
	path = raw[4:]
	if path == "/dev/null":
	path = ""
	continue
	match = HUNK.match(raw)
	if match:
	number = int(match.group(1))
	group += 1
	continue
	if not path:
	continue
	if raw.startswith("+") and not raw.startswith("+++"):
	lines.append(AddedLine(path, number, raw[1:], group))
	number += 1
	elif raw.startswith(" "):
	number += 1
	elif raw.startswith("-"):
	continue
	return lines


	def git(cwd: str, *args: str) -> subprocess.CompletedProcess[str]:
	return subprocess.run(
	["git", "-c", "core.quotePath=false", *args],
	cwd=cwd,
	text=True,
	stdout=subprocess.PIPE,
	stderr=subprocess.DEVNULL,
	check=False,
	)


	def worktree_findings(cwd: str, base: str = "HEAD") -> list[Finding]:
	root_result = git(cwd, "rev-parse", "--show-toplevel")
	if root_result.returncode:
	return []
	root = root_result.stdout.strip()
	diff_result = git(root, "diff", "--no-ext-diff", "--no-textconv", "--unified=0", "--no-color", "--no-prefix", base, "--")
	if diff_result.returncode:
	raise RuntimeError("Cannot audit comments against the session baseline")
	lines = parse_git_diff(diff_result.stdout)
	untracked = git(root, "ls-files", "--others", "--exclude-standard", "-z")
	if untracked.returncode == 0:
	group = max((line.group for line in lines), default=0)
	for relative in filter(None, untracked.stdout.split("\0")):
	if marker_for(relative) is None:
	continue
	path = Path(root, relative)
	try:
	if path.stat().st_size > 2_000_000:
	continue
	content = path.read_text(errors="replace").splitlines()
	except OSError:
	continue
	group += 1
	lines.extend(AddedLine(relative, number, text, group) for number, text in enumerate(content, 1))
	return analyze(lines)


	def state_path(session_id: str) -> Path:
	state_dir = Path(tempfile.gettempdir(), f"comment-cop-{os.getuid()}")
	state_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
	digest = hashlib.sha256(session_id.encode()).hexdigest()
	return state_dir / f"{digest}.json"


	def record_baseline(data: dict) -> None:
	session_id = str(data.get("session_id") or "")
	cwd = str(data.get("cwd") or os.getcwd())
	if not session_id:
	return
	root_result = git(cwd, "rev-parse", "--show-toplevel")
	if root_result.returncode:
	return
	root = root_result.stdout.strip()
	path = state_path(session_id)
	try:
	state = json.loads(path.read_text())
	except (OSError, ValueError):
	state = {}
	if state.get("version") != 2:
	state = {"version": 2, "repos": {}}
	if root in state["repos"]:
	return
	head = git(root, "rev-parse", "HEAD")
	if head.returncode:
	raise RuntimeError("Cannot record the session baseline commit")
	counts = collections.Counter(finding.fingerprint() for finding in worktree_findings(root))
	state["repos"][root] = {"head": head.stdout.strip(), "findings": counts}
	path.write_text(json.dumps(state, sort_keys=True))


	def new_findings(data: dict) -> list[Finding]:
	session_id = str(data.get("session_id") or "")
	cwd = str(data.get("cwd") or os.getcwd())
	if not session_id:
	return []
	path = state_path(session_id)
	if not path.exists():
	return []
	try:
	state = json.loads(path.read_text())
	except (OSError, ValueError):
	return []
	root_result = git(cwd, "rev-parse", "--show-toplevel")
	if root_result.returncode:
	return []
	baseline_state = state.get("repos", {}).get(root_result.stdout.strip()) if state.get("version") == 2 else None
	if not baseline_state:
	return []
	baseline = collections.Counter(baseline_state["findings"])
	current = worktree_findings(cwd, baseline_state["head"])
	remaining = baseline.copy()
	result: list[Finding] = []
	for finding in current:
	fingerprint = finding.fingerprint()
	if remaining[fingerprint]:
	remaining[fingerprint] -= 1
	else:
	result.append(finding)
	return result


	def direct_edit_findings(data: dict) -> list[Finding]:
	tool_input = data.get("tool_input") or {}
	tool_name = data.get("tool_name") or ""
	if tool_name == "apply_patch":
	return analyze(parse_apply_patch(str(tool_input.get("command") or "")))
	path = str(tool_input.get("file_path") or "")
	marker = marker_for(path)
	if not marker:
	return []
	if tool_name == "Edit":
	text = str(tool_input.get("new_string") or "")
	old = {line.strip() for line in str(tool_input.get("old_string") or "").splitlines()}
	else:
	text = str(tool_input.get("content") or "")
	old = set()
	try:
	old = {line.strip() for line in Path(path).read_text(errors="replace").splitlines()}
	except OSError:
	pass
	lines = [
	AddedLine(path, number, line, 1)
	for number, line in enumerate(text.splitlines(), 1)
	if line.strip() not in old
	]
	return analyze(lines)


	def format_findings(findings: list[Finding], limit: int = 8) -> str:
	shown = findings[:limit]
	details = "; ".join(
	f"{finding.path}:{finding.number} [{finding.rule}] {finding.text!r}"
	for finding in shown
	)
	if len(findings) > limit:
	details += f"; and {len(findings) - limit} more"
	return f"comment-cop: {details}. {POLICY} Remove or tighten these comments, then retry."


	def policy_context() -> dict:
	return {
	"hookSpecificOutput": {
	"hookEventName": "UserPromptSubmit",
	"additionalContext": "Code comment policy (hook-enforced): " + POLICY,
	}
	}


	def run_event(data: dict) -> dict \| None:
	event = data.get("hook_event_name")
	if event == "SessionStart":
	record_baseline(data)
	return None
	if event == "UserPromptSubmit":
	record_baseline(data)
	return policy_context()
	if event == "PreToolUse":
	findings = direct_edit_findings(data)
	if findings:
	return {
	"hookSpecificOutput": {
	"hookEventName": "PreToolUse",
	"permissionDecision": "deny",
	"permissionDecisionReason": format_findings(findings),
	}
	}
	return None
	if event == "Stop":
	if data.get("stop_hook_active"):
	return {}
	findings = new_findings(data)
	if findings:
	return {"decision": "block", "reason": format_findings(findings)}
	return {}
	return None


	def self_test() -> None:
	cases = [
	(
	"long yaml",
	["# The migrate role reports like any other.", "# It inherits no envFrom.", "# The SDK falls back.", "envFrom:"],
	True,
	),
	(
	"four-line go",
	["// The restored base may contain a later row.", "// Never overwrite it.", "// The conflict handler covers constraints.", "// The final comparison is authoritative.", "insert()"],
	True,
	),
	("narration", ["// Loop through all rows.", "for rows.Next() {}"], True),
	("history", ["// We removed the fallback because it was unsafe.", "run()"], True),
	("why", ["// Keep the predecessor until the handoff is durable.", "drop()"], False),
	("directive", ["//nolint:gosec", "run()"], False),
	("commented config", ["# timeout: 10s", "enabled: true"], False),
	]
	for name, content, expected in cases:
	lines = [AddedLine("test.go" if name not in {"long yaml", "commented config"} else "test.yaml", i, line, 1) for i, line in enumerate(content, 1)]
	actual = bool(analyze(lines))
	if actual != expected:
	raise AssertionError(f"{name}: expected {expected}, got {actual}")
	patch = "* Begin Patch\n* Update File: x.go\n@@ -1 +1,2 @@\n run()\n+// Set the value.\n*** End Patch"
	if not analyze(parse_apply_patch(patch)):
	raise AssertionError("apply_patch parsing did not find narration")
	with tempfile.TemporaryDirectory() as directory:
	subprocess.run(["git", "init", "-q"], cwd=directory, check=True)
	subprocess.run(["git", "config", "user.email", "comment-cop@example.invalid"], cwd=directory, check=True)
	subprocess.run(["git", "config", "user.name", "comment-cop"], cwd=directory, check=True)
	source = Path(directory, "main.go")
	source.write_text("package main\n")
	subprocess.run(["git", "add", "main.go"], cwd=directory, check=True)
	subprocess.run(["git", "commit", "-qm", "base"], cwd=directory, check=True)
	source.write_text("package main\n// Loop through rows.\n")
	event = {"session_id": "self-test", "cwd": directory}
	record_baseline(event)
	source.write_text("package main\n// Loop through rows.\n// Set the timeout.\n")
	findings = new_findings(event)
	if len(findings) != 1 or findings[0].text != "Set the timeout.":
	raise AssertionError(f"baseline isolation failed: {findings!r}")
	print("comment-cop self-test passed")


	def main() -> None:
	if len(sys.argv) > 1 and sys.argv[1] == "--self-test":
	self_test()
	return
	if len(sys.argv) > 1 and sys.argv[1] == "--check-diff":
	findings = worktree_findings(os.getcwd())
	if findings:
	print(format_findings(findings))
	raise SystemExit(1)
	return
	data = json.load(sys.stdin)
	output = run_event(data)
	if output is not None:
	print(json.dumps(output))


	if __name__ == "__main__":
	try:
	main()
	except Exception as error:
	if "--self-test" in sys.argv:
	raise
	print(json.dumps({"systemMessage": f"comment-cop failed open: {error}"}))
	import runpy
	import subprocess
	import sys
	import unittest
	from pathlib import Path

	sys.path.insert(0, str(Path(__file__).parent))
	from test_xreview import RepoCase, git


	class CommentAudit(RepoCase):
	def setUp(self):
	super().setUp()
	self.cop = runpy.run_path(str(Path(__file__).with_name("comment-cop.py")))
	self.state = Path(self.tmp.name, "session.json")
	self.cop["record_baseline"].__globals__["state_path"] = lambda session_id: self.state
	self.data = {"session_id": "test", "cwd": str(self.repo)}
	self.source = self.repo / "code.go"
	self.source.write_text("package main\n")
	git(self.repo, "add", "code.go")
	git(self.repo, "commit", "-qm", "source")

	def record(self):
	self.cop["record_baseline"](self.data)

	def findings(self):
	return self.cop["new_findings"](self.data)

	def add_comment(self):
	self.source.write_text("package main\n// Loop through all rows.\n")

	def test_committed_comment_is_audited(self):
	self.record()
	self.add_comment()
	git(self.repo, "commit", "-qam", "comment")
	self.assertEqual([f.rule for f in self.findings()], ["code-narration"])

	def test_new_file_committed_comment_is_audited(self):
	self.record()
	Path(self.repo, "new.go").write_text("package main\n// Set the value.\n")
	git(self.repo, "add", "new.go")
	git(self.repo, "commit", "-qm", "new source")
	self.assertEqual([f.rule for f in self.findings()], ["code-narration"])

	def test_prompts_and_resume_preserve_session_anchor(self):
	self.record()
	self.add_comment()
	git(self.repo, "commit", "-qam", "comment")
	for event in ("UserPromptSubmit", "SessionStart"):
	self.cop["run_event"]({**self.data, "hook_event_name": event})
	self.assertEqual(len(self.findings()), 1)

	def test_preexisting_comment_is_excluded(self):
	self.add_comment()
	self.record()
	git(self.repo, "commit", "-qam", "preexisting comment")
	self.assertEqual(self.findings(), [])

	def test_removed_comment_clears_finding(self):
	self.record()
	self.add_comment()
	git(self.repo, "commit", "-qam", "comment")
	self.source.write_text("package main\n")
	self.assertEqual(self.findings(), [])

	def test_stop_blocks_committed_comment(self):
	self.record()
	self.add_comment()
	git(self.repo, "commit", "-qam", "comment")
	result = self.cop["run_event"]({**self.data, "hook_event_name": "Stop"})
	self.assertEqual(result["decision"], "block")

	def test_multiple_checkouts_keep_separate_anchors(self):
	self.record()
	self.add_comment()
	git(self.repo, "commit", "-qam", "comment")
	other = Path(self.tmp.name, "other")
	subprocess.run(["git", "clone", "-q", str(self.bare), str(other)], check=True)
	self.cop["record_baseline"]({**self.data, "cwd": str(other)})
	self.assertEqual(len(self.findings()), 1)
	self.assertEqual(self.cop["new_findings"]({**self.data, "cwd": str(other)}), [])


	if __name__ == "__main__":
	unittest.main()
	{
	"$schema": "https://json-schema.org/draft/2020-12/schema",
	"type": "object",
	"additionalProperties": false,
	"required": [
	"verdict",
	"summary",
	"findings",
	"next_steps"
	],
	"properties": {
	"verdict": {
	"type": "string",
	"enum": [
	"approve",
	"needs-attention"
	]
	},
	"summary": {
	"type": "string",
	"minLength": 1
	},
	"findings": {
	"type": "array",
	"items": {
	"type": "object",
	"additionalProperties": false,
	"required": [
	"severity",
	"title",
	"body",
	"file",
	"line_start",
	"line_end",
	"confidence",
	"recommendation"
	],
	"properties": {
	"severity": {
	"type": "string",
	"enum": [
	"critical",
	"high",
	"medium",
	"low"
	]
	},
	"title": {
	"type": "string",
	"minLength": 1
	},
	"body": {
	"type": "string",
	"minLength": 1
	},
	"file": {
	"type": "string",
	"minLength": 1
	},
	"line_start": {
	"type": "integer",
	"minimum": 1
	},
	"line_end": {
	"type": "integer",
	"minimum": 1
	},
	"confidence": {
	"type": "number",
	"minimum": 0,
	"maximum": 1
	},
	"recommendation": {
	"type": "string"
	}
	}
	}
	},
	"next_steps": {
	"type": "array",
	"items": {
	"type": "string",
	"minLength": 1
	}
	}
	}
	}