Hooks

Where it lives

• hooks/__init__.pyGitHub ↗
  Read the script

  # Claude Code hooks for event-driven automation
  # hook-version: 1.0.0
  

• hooks/adr-enforcement.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  # hook-version: 1.0.0
  """
  PostToolUse Hook: ADR Compliance Enforcement
  
  After every Write or Edit tool call on a pipeline component file, automatically
  run adr-compliance.py and inject violations as feedback into the next context.
  
  Design Principles:
  - Non-blocking (always exits 0)
  - Silent on non-pipeline files (no noise)
  - Graceful degradation when adr-compliance.py not yet deployed
  - Skips when no active ADR session (.adr-session.json absent)
  """
  
  import json
  import os
  import re
  import subprocess
  import sys
  from pathlib import Path
  
  # Add lib directory to path for imports
  sys.path.insert(0, str(Path(__file__).parent / "lib"))
  
  from hook_utils import context_output, empty_output, hook_error, log_warning
  from stdin_timeout import read_stdin
  
  _EVENT_NAME = "PostToolUse"
  _TRUSTED_ROOT = Path(__file__).resolve().parent.parent
  
  # Pipeline component files that trigger enforcement (matched against repo-relative paths)
  _PIPELINE_COMPONENT_PATTERNS = [
      r"^skills/(?:[^/]+/)?[^/]+/SKILL\.md$",
      r"^agents/[^/]+\.md$",
      r"^scripts/[^/]+\.py$",
      r"^hooks/[^/]+\.py$",
  ]
  
  # Repo-relative paths to exclude even if they match a component pattern
  _EXCLUDE_PATTERNS = [
      r"^agents/INDEX\.json",
      r"^hooks/lib/",
      r"^hooks/adr-enforcement\.py$",
      r"^scripts/tests/",
      r"^scripts/__pycache__/",
  ]
  
  # Reference files used by adr-compliance.py
  _STEP_MENU = "skills/workflow/references/pipeline-scaffolder/references/step-menu.md"
  _SPEC_FORMAT = "skills/workflow/references/pipeline-scaffolder/references/pipeline-spec-format.md"
  
  
  def _project_root(event: dict) -> Path:
      """Resolve the repository the hook event is operating on."""
      candidate = event.get("cwd") or os.environ.get("CLAUDE_PROJECT_DIR") or Path.cwd()
      return Path(candidate).resolve()
  
  
  def _project_relative_path(file_path: str, project_root: Path) -> Path | None:
      """Return ``file_path`` relative to the event project, or None if outside it."""
      try:
          path = Path(file_path)
          absolute = path.resolve() if path.is_absolute() else (project_root / path).resolve()
          return absolute.relative_to(project_root)
      except (OSError, ValueError):
          return None
  
  
  def is_pipeline_component(file_path: str, project_root: Path | None = None) -> bool:
      """Check if the file is a pipeline component that should be compliance-checked.
  
      Normalizes the path to be relative to the repo root before matching, so
      patterns like agents/ only match the top-level agents/ directory and not
      arbitrary path components.
      """
      root = project_root or Path(os.environ.get("CLAUDE_PROJECT_DIR", Path.cwd())).resolve()
      rel_path = _project_relative_path(file_path, root)
      if rel_path is None:
          return False
      rel_str = rel_path.as_posix()
  
      for exclude in _EXCLUDE_PATTERNS:
          if re.search(exclude, rel_str):
              return False
  
      return any(re.match(pattern, rel_str) for pattern in _PIPELINE_COMPONENT_PATTERNS)
  
  
  def load_session(cwd: str) -> dict | None:
      """Load .adr-session.json from cwd. Returns None if absent or invalid."""
      session_path = Path(cwd) / ".adr-session.json"
      if not session_path.exists():
          return None
      try:
          with open(session_path) as f:
              return json.load(f)
      except (json.JSONDecodeError, OSError):
          return None
  
  
  def run_compliance_check(
      compliance_script: Path,
      file_path: str,
      project_root: Path,
  ) -> dict | None:
      """
      Run adr-compliance.py check on the file.
  
      Returns parsed JSON output dict, or None on subprocess failure.
      """
      cmd = [
          sys.executable,
          str(compliance_script),
          "check",
          "--file",
          str(project_root / file_path),
          "--step-menu",
          str(project_root / _STEP_MENU),
          "--spec-format",
          str(project_root / _SPEC_FORMAT),
      ]
  
      try:
          result = subprocess.run(
              cmd,
              capture_output=True,
              text=True,
              timeout=10,
              cwd=str(_TRUSTED_ROOT),
          )
          output = result.stdout.strip()
          if not output:
              return None
          return json.loads(output)
      except subprocess.TimeoutExpired:
          log_warning("adr-compliance.py timed out after 10s")
          return None
      except (json.JSONDecodeError, OSError):
          return None
  
  
  def format_violations(file_path: str, check_result: dict) -> str:
      """Format violation output for context injection."""
      violations = check_result.get("violations", [])
      lines = []
  
      # Use a relative display path when possible
      display_path = file_path
      _chk = "COMPLIANCE CHECK"
      lines.append(f"[adr-enforcement] {_chk}: {display_path}")
  
      count = len(violations)
      _vf = "VIOLATIONS FOUND"
      lines.append(f"[adr-enforcement] {_vf} ({count}):")
  
      for v in violations:
          line_num = v.get("line", "?")
          v_type = v.get("type", "unknown")
          value = v.get("value", "")
          suggestion = v.get("suggestion", "")
          entry = f'  Line {line_num}: {v_type} "{value}"'
          if suggestion:
              entry += f" — {suggestion}"
          lines.append(f"[adr-enforcement] {entry}")
  
      lines.append("[adr-enforcement] FIX REQUIRED before proceeding:")
      lines.append(f"[adr-enforcement]   python3 scripts/adr-compliance.py check --file {display_path} \\")
      lines.append(f"[adr-enforcement]     --step-menu {_STEP_MENU} \\")
      lines.append(f"[adr-enforcement]     --spec-format {_SPEC_FORMAT}")
  
      return "\n".join(lines)
  
  
  def format_pass(file_path: str, check_result: dict) -> str:
      """Format PASS output for context injection."""
      display_path = file_path
      # Include grounding counts if available in result metadata
      meta = check_result.get("stats", {})
      step_count = meta.get("step_names_checked", 0)
      schema_count = meta.get("schema_types_checked", 0)
  
      if step_count or schema_count:
          detail = f"({step_count} step names, {sche

• hooks/adr-lifecycle-on-merge.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  # hook-version: 1.0.0
  """PostToolUse Hook: ADR lifecycle checker on merge.
  
  Fires after a Bash tool use. Detects merge commands (gh pr merge / git merge)
  and scans the branch name and recent commit messages for ADR references.
  
  For each matched ADR it:
  1. Reads the ## Implementation section and extracts numbered steps.
  2. Diffs HEAD~1 to get changed files.
  3. Checks each step for keyword matches in the changed-file list.
  4. Reports a checklist with PARTIAL / COMPLETE status.
  5. If all steps matched, updates the ADR file status to "Completed" and moves
     it to adr/completed/ (shutil, not git mv — adr/ is gitignored).
  
  Always exits 0 (non-blocking, informational hook).
  Target execution time: <200ms (exits early when not a merge command).
  """
  
  import json
  import os
  import re
  import shutil
  import subprocess
  import sys
  from datetime import date
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent / "lib"))
  from hook_utils import context_output, empty_output, get_tool_input
  from stdin_timeout import read_stdin
  
  DEBUG_LOG = Path("/tmp/claude_hook_debug.log")
  EVENT_NAME = "PostToolUse"
  
  # Patterns to identify ADR references in branch names / commit messages
  ADR_PATTERNS = [
      re.compile(r"ADR-(\d+)", re.IGNORECASE),
      re.compile(r"adr/(\d+)-"),
      re.compile(r"adr-(\d+)", re.IGNORECASE),
  ]
  
  
  # ---------------------------------------------------------------------------
  # Logging
  # ---------------------------------------------------------------------------
  
  
  def debug_log(msg: str) -> None:
      try:
          with open(DEBUG_LOG, "a") as f:
              f.write(f"[adr-lifecycle] {msg}\n")
      except Exception:
          pass
  
  
  # ---------------------------------------------------------------------------
  # Helpers
  # ---------------------------------------------------------------------------
  
  
  def run_git(args: list[str], cwd: str | None = None, timeout: int = 5) -> str:
      """Run a git command and return stdout. Returns '' on failure."""
      try:
          result = subprocess.run(
              ["git"] + args,
              capture_output=True,
              text=True,
              timeout=timeout,
              cwd=cwd,
          )
          return result.stdout.strip() if result.returncode == 0 else ""
      except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
          return ""
  
  
  def extract_adr_numbers(text: str) -> list[str]:
      """Return deduplicated ADR numbers found in *text*."""
      seen: set[str] = set()
      numbers: list[str] = []
      for pattern in ADR_PATTERNS:
          for m in pattern.finditer(text):
              num = m.group(1).lstrip("0") or "0"
              if num not in seen:
                  seen.add(num)
                  numbers.append(num)
      return numbers
  
  
  def find_adr_file(adr_number: str) -> Path | None:
      """Locate adr/{NNN}-*.md, zero-padded searches 1-4 digits."""
      adr_dir = Path(os.environ.get("CLAUDE_PROJECT_DIR", ".")) / "adr"
      if not adr_dir.is_dir():
          return None
  
      # Try zero-padded variants: 1 → 001, 179 → 179, etc.
      num_int = int(adr_number)
      padded_variants = [
          str(num_int),
          f"{num_int:03d}",
          f"{num_int:04d}",
      ]
      for variant in padded_variants:
          matches = list(adr_dir.glob(f"{variant}-*.md"))
          if matches:
              return matches[0]
      return None
  
  
  def extract_implementation_steps(adr_content: str) -> list[str]:
      """Extract numbered steps from ## Implementation section."""
      steps: list[str] = []
      in_section = False
      for line in adr_content.splitlines():
          if re.match(r"^#{1,3}\s+Implementation", line, re.IGNORECASE):
              in_section = True
              continue
          if in_section:
              # Stop at next heading
              if re.match(r"^#{1,3}\s+", line):
                  break
              # Match numbered list items: "1. " or "1) "
              m = re.match(r"^\s*\d+[.)]\s+(.+)", line)
              if m:
                  steps.append(m.group(1).strip())
      return steps
  
  
  def get_changed_files() -> list[str]:
      """Return list of files changed in the last commit (HEAD~1 diff)."""
      output = run_git(["diff", "HEAD~1", "--name-only"])
      if not output:
          # Fallback: files in most recent commit
          output = run_git(["show", "--name-only", "--format=", "HEAD"])
      return [f.strip() for f in output.splitlines() if f.strip()]
  
  
  def keywords_from_step(step: str) -> list[str]:
      """Extract meaningful lowercase keywords (>3 chars) from a step description."""
      words = re.findall(r"[a-zA-Z0-9_\-/\.]+", step)
      return [w.lower() for w in words if len(w) > 3]
  
  
  def step_matches_files(step: str, changed_files: list[str]) -> tuple[bool, str]:
      """Return (matched, matching_file_or_empty)."""
      keywords = keywords_from_step(step)
      if not keywords:
          return False, ""
      lower_files = [f.lower() for f in changed_files]
      for keyword in keywords:
          for i, lf in enumerate(lower_files):
              if keyword in lf:
                  return True, changed_files[i]
      return False, ""
  
  
  def update_adr_status(adr_path: Path, adr_number: str) -> None:
      """Change ## Status to Completed and move file to adr/completed/."""
      try:
          content = adr_path.read_text()
          today = date.today().strftime("%Y-%m-%d")
          # Replace first occurrence of status proposed/accepted/approved
          updated = re.sub(
              r"(## Status\s*\n)(Proposed|Accepted|Approved|Active|Draft)",
              rf"\1Completed ({today})",
              content,
              flags=re.IGNORECASE,
          )
          if updated == content:
              # No status line found — append/replace generically
              updated = re.sub(
                  r"(## Status\s*\n)([^\n]*)",
                  rf"\1Completed ({today})",
                  content,
                  flags=re.IGNORECASE,
              )
  
          completed_dir = adr_path.parent / "completed"
          completed_dir.mkdir(exist_ok=True)
  
          # Write updated content to destin

• hooks/afk-mode.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """SessionStart hook: inject autonomous behavioral context for AFK sessions.
  
  Detects SSH/tmux/screen/headless sessions and injects a posture block that
  instructs Claude to work proactively without requiring confirmation (ADR-143).
  
  Environment:
      CLAUDE_AFK_MODE  - Control override: 'always' (default), 'auto', or 'never'
      SSH_CONNECTION   - Set by SSH daemon when connected via SSH
      SSH_TTY          - Set when SSH allocates a TTY
      SSH_CLIENT       - Set by SSH daemon (older form)
      TMUX             - Set by tmux when inside a tmux session
      STY              - Set by GNU screen when inside a screen session
  
  Always exits 0 (advisory, never blocking).
  """
  
  import os
  import sys
  
  sys.path.insert(0, os.path.join(os.path.dirname(__file__), "lib"))
  
  from hook_utils import context_output, empty_output, hook_error
  
  EVENT_NAME = "SessionStart"
  
  AFK_CONTEXT = """<afk-mode>
  The terminal is unfocused — the user is not actively watching.
  Work proactively. Complete multi-step tasks without asking for confirmation.
  If you can determine the next logical step, take it.
  Produce concise task-completion summaries when finishing long-running work.
  </afk-mode>"""
  
  
  def is_afk() -> bool:
      """Return True if the session appears to be AFK (unattended).
  
      NOTE: Do NOT use sys.stdin.isatty() or sys.stdout.isatty() here.
      Claude Code pipes stdin (event JSON) and captures stdout (hook output),
      so both are always non-TTY in hook context — the check would always
      return True, activating AFK mode on every session regardless of type.
      """
      # SSH indicators
      if os.environ.get("SSH_CONNECTION"):
          return True
      if os.environ.get("SSH_TTY"):
          return True
      if os.environ.get("SSH_CLIENT"):
          return True
      # Multiplexer indicators
      if os.environ.get("TMUX"):
          return True
      if os.environ.get("STY"):
          return True
      return False
  
  
  def main() -> None:
      mode = os.environ.get("CLAUDE_AFK_MODE", "always").strip().lower()
  
      if mode == "always":
          context_output(EVENT_NAME, AFK_CONTEXT).print_and_exit()
  
      if mode == "never":
          empty_output(EVENT_NAME).print_and_exit()
  
      # Default: auto-detect
      if is_afk():
          context_output(EVENT_NAME, AFK_CONTEXT).print_and_exit()
      else:
          empty_output(EVENT_NAME).print_and_exit()
  
  
  if __name__ == "__main__":
      try:
          main()
      except Exception as e:
          hook_error("afk-mode", e)
      finally:
          sys.exit(0)
  

• hooks/ci-merge-gate.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  # hook-version: 1.0.0
  """PreToolUse hook: Block gh pr merge when CI checks haven't passed.
  
  Intercepts Bash tool calls containing 'gh pr merge' and checks GitHub
  Actions status before allowing the merge. Blocks if any checks are
  failing or still pending.
  """
  
  import json
  import os
  import subprocess
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent / "lib"))
  from hook_utils import deny_tool_use, record_governance
  from stdin_timeout import read_stdin
  
  
  def main() -> None:
      data = json.loads(read_stdin(timeout=2))
  
      # tool_name filter removed — matcher "Bash" in settings.json prevents
      # this hook from spawning for non-Bash tools.
  
      command = data.get("tool_input", {}).get("command", "")
  
      # Only intercept gh pr merge commands
      if "gh pr merge" not in command and "gh pr merge" not in command.replace("  ", " "):
          return
  
      parts = command.split()
  
      # --- Block --admin before any CI check ---
      if "--admin" in parts:
          if os.environ.get("ALLOW_ADMIN_MERGE") == "1":
              print("[ci-merge-gate] WARNING: --admin override allowed via ALLOW_ADMIN_MERGE=1", file=sys.stderr)
          else:
              print("[ci-merge-gate] BLOCKED: --admin bypasses branch protection", file=sys.stderr)
              deny_tool_use("PreToolUse", "Use of --admin bypasses CI checks. Remove --admin and wait for CI to pass.")
              sys.exit(0)
  
      # --- Block --force before any CI check ---
      if "--force" in parts:
          if os.environ.get("ALLOW_FORCE_MERGE") == "1":
              print("[ci-merge-gate] WARNING: --force override allowed via ALLOW_FORCE_MERGE=1", file=sys.stderr)
          else:
              print("[ci-merge-gate] BLOCKED: --force bypasses merge safeguards", file=sys.stderr)
              deny_tool_use("PreToolUse", "Use of --force bypasses merge safeguards. Remove --force and merge normally.")
              sys.exit(0)
  
      # Extract PR number from command
      # Patterns: gh pr merge 55, gh pr merge #55, gh pr merge --squash 55
      pr_number = None
      for i, part in enumerate(parts):
          if part.lstrip("#").isdigit() and i > 0 and parts[i - 1] != "--count":
              pr_number = part.lstrip("#")
              break
  
      if not pr_number:
          # No PR number found — might be merging current branch PR
          # Try to get it from current branch
          try:
              result = subprocess.run(
                  ["gh", "pr", "view", "--json", "number", "--jq", ".number"],
                  capture_output=True,
                  text=True,
                  timeout=10,
              )
              if result.returncode == 0 and result.stdout.strip().isdigit():
                  pr_number = result.stdout.strip()
          except (subprocess.TimeoutExpired, FileNotFoundError):
              pass
  
      if not pr_number:
          # Can't determine PR number — let it through with a warning
          print("[ci-merge-gate] WARNING: Could not determine PR number. Skipping CI check.", file=sys.stderr)
          return
  
      # Check CI status
      try:
          result = subprocess.run(
              ["gh", "pr", "checks", pr_number, "--json", "name,state,bucket"],
              capture_output=True,
              text=True,
              timeout=15,
          )
      except (subprocess.TimeoutExpired, FileNotFoundError):
          print(
              "[ci-merge-gate] WARNING: Could not check CI status (gh not available or timeout).",
              file=sys.stderr,
          )
          return
  
      if result.returncode != 0:
          # gh pr checks failed — might be no checks configured
          if "no checks" in result.stderr.lower():
              return
          print(
              f"[ci-merge-gate] WARNING: Could not fetch CI checks: {result.stderr.strip()}",
              file=sys.stderr,
          )
          return
  
      try:
          checks = json.loads(result.stdout)
      except json.JSONDecodeError:
          print("[ci-merge-gate] WARNING: Could not parse CI check results.", file=sys.stderr)
          return
  
      # bucket field values: pass, fail, pending, skipping, cancel
      failing = [c for c in checks if c.get("bucket") == "fail"]
      pending = [c for c in checks if c.get("bucket") == "pending"]
  
      if failing:
          names = ", ".join(c["name"] for c in failing)
          print(f"[ci-merge-gate] BLOCKED: CI checks failing: {names}", file=sys.stderr)
          print(f"[ci-merge-gate] Fix the failing checks before merging PR #{pr_number}.", file=sys.stderr)
          record_governance(
              "approval_requested",
              hook_name="ci-merge-gate",
              tool_name="Bash",
              hook_phase="pre",
              severity="medium",
              blocked=True,
              command=command,
          )
          deny_tool_use(
              "PreToolUse",
              f"CI checks are failing for PR #{pr_number}: {names}. Fix the failing checks before merging.",
          )
          sys.exit(0)
  
      if pending:
          names = ", ".join(c["name"] for c in pending)
          print(f"[ci-merge-gate] BLOCKED: CI checks still running: {names}", file=sys.stderr)
          print(f"[ci-merge-gate] Wait for checks to complete before merging PR #{pr_number}.", file=sys.stderr)
          record_governance(
              "approval_requested",
              hook_name="ci-merge-gate",
              tool_name="Bash",
              hook_phase="pre",
              severity="medium",
              blocked=True,
              command=command,
          )
          deny_tool_use(
              "PreToolUse",
              f"CI checks are still running for PR #{pr_number}: {names}. "
              "Wait for all checks to complete before merging.",
          )
          sys.exit(0)
  
      # All checks passed
      # PreToolUse stdout is protocol-only. Human-readable diagnostics must use
      # stderr or Codex's adapter will reject an otherwise allowed invocation.
      print(f"[ci-merge-gate] CI checks passed for PR #{pr_number}. Merge allowed.", file=sys.stderr)
  
  
  if __name__ == "__main__":
      try:
          main()
      except SystemEx

• hooks/codex-auto-review.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  # hook-version: 1.0.1
  """
  UserPromptSubmit hook — auto-inject Codex cross-model review for PR review operations.
  
  Detects when the user invokes a review skill or phrases a review request,
  and injects a nudge to also run the pr-workflow skill's codex-review intent
  for cross-model second-opinion findings — but only when the codex CLI is
  actually installed.
  
  Detection triggers:
    Skill invocations: /systematic-code-review, /parallel-code-review,
                       /full-repo-review, /pr-review
    Intent phrases: "review this pr", "review my changes", "code review",
                    "review the pr", "pr review"
  
  Non-triggers (already covered):
    /pr-workflow   — umbrella skill handles Codex via codex-review intent
  
  Design decisions:
    - DON'T inject the raw codex command; tell Claude to invoke the skill
    - DON'T inject if the user already asked for /pr-workflow directly
    - DON'T block if codex is missing — silently skip
    - Keep injection short — it's a nudge, not a manual
  
  Note on UserPromptSubmit timing: this hook fires BEFORE /do selects an agent,
  so it only injects session-wide, agent-agnostic context. Agent-scoped context
  injection belongs at routing time, inside the skill.
  """
  
  import json
  import os
  import re
  import subprocess
  import sys
  import traceback
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent / "lib"))
  
  from hook_utils import context_output, empty_output, hook_error
  from stdin_timeout import read_stdin
  
  EVENT_NAME = "UserPromptSubmit"
  
  # Skill invocations that trigger cross-model review injection.
  # /pr-workflow is excluded — it already handles Codex via codex-review intent.
  REVIEW_SKILL_PATTERN = re.compile(
      r"/(?:systematic-code-review|parallel-code-review|full-repo-review|pr-review)\b",
      re.IGNORECASE,
  )
  
  # Natural-language intent phrases that indicate a review request.
  REVIEW_INTENT_PATTERN = re.compile(
      r"(?:review\s+(?:this\s+)?(?:pr|the\s+pr|my\s+changes)|code\s+review|pr\s+review)",
      re.IGNORECASE,
  )
  
  # Patterns that indicate the user is already requesting pr-workflow.
  # The umbrella takes care of Codex via its codex-review intent; no need to inject.
  ALREADY_HANDLED_PATTERN = re.compile(
      r"/pr-workflow\b",
      re.IGNORECASE,
  )
  
  CODEX_INJECTION = (
      "[codex-auto-review] Codex CLI detected. After completing the primary review, "
      "use the codex-review intent to get cross-model "
      "findings from OpenAI Codex (GPT-5.4 xhigh reasoning). If the codex-review intent "
      "has already been invoked or is part of the current pipeline, skip this step. "
      "Call the Skill tool with `pr-workflow`."
  )
  
  
  def extract_prompt(event: dict) -> str:
      """Extract the user's prompt text from the hook event.
  
      Claude Code UserPromptSubmit events deliver the text in tool_input.prompt.
      Returns empty string if the field is absent or not a string.
      """
      tool_input = event.get("tool_input", {})
      if not isinstance(tool_input, dict):
          return ""
      prompt = tool_input.get("prompt", "")
      return prompt if isinstance(prompt, str) else ""
  
  
  def is_review_request(prompt: str) -> bool:
      """Return True if the prompt contains a review trigger."""
      return bool(REVIEW_SKILL_PATTERN.search(prompt) or REVIEW_INTENT_PATTERN.search(prompt))
  
  
  def is_already_handled(prompt: str) -> bool:
      """Return True if the prompt already requests pr-workflow directly."""
      return bool(ALREADY_HANDLED_PATTERN.search(prompt))
  
  
  def codex_is_available() -> bool:
      """Return True if the codex CLI is installed and reachable.
  
      Uses `which codex` with a 2-second timeout to avoid blocking.
      """
      try:
          result = subprocess.run(
              ["which", "codex"],
              capture_output=True,
              timeout=2,
          )
          return result.returncode == 0
      except (OSError, subprocess.TimeoutExpired):
          return False
  
  
  def main() -> None:
      debug = os.environ.get("CLAUDE_HOOKS_DEBUG")
  
      raw = read_stdin(timeout=5)
      if not raw or not raw.strip():
          empty_output(EVENT_NAME).print_and_exit()
  
      try:
          event = json.loads(raw)
      except json.JSONDecodeError as exc:
          if debug:
              print(f"[codex-auto-review] JSON parse error: {exc}", file=sys.stderr)
          empty_output(EVENT_NAME).print_and_exit()
  
      prompt = extract_prompt(event)
  
      if not prompt:
          empty_output(EVENT_NAME).print_and_exit()
  
      if is_already_handled(prompt):
          if debug:
              print("[codex-auto-review] Already handled by pr-workflow — skipping", file=sys.stderr)
          empty_output(EVENT_NAME).print_and_exit()
  
      if not is_review_request(prompt):
          if debug:
              print("[codex-auto-review] No review trigger detected — skipping", file=sys.stderr)
          empty_output(EVENT_NAME).print_and_exit()
  
      if not codex_is_available():
          if debug:
              print("[codex-auto-review] codex CLI not found — skipping injection", file=sys.stderr)
          empty_output(EVENT_NAME).print_and_exit()
  
      if debug:
          print("[codex-auto-review] Review trigger detected and codex available — injecting", file=sys.stderr)
  
      context_output(EVENT_NAME, CODEX_INJECTION).print_and_exit()
  
  
  if __name__ == "__main__":
      try:
          main()
      except Exception as exc:
          hook_error("codex-auto-review", exc)
      finally:
          sys.exit(0)
  

• hooks/codex-hook-adapter.pyGitHub ↗
• hooks/creation-protocol-enforcer.pyGitHub ↗
• hooks/creation-request-enforcer-userprompt.pyGitHub ↗
• hooks/cross-repo-agents.pyGitHub ↗
• hooks/fish-shell-detector.pyGitHub ↗
• hooks/herdr-state-reporter.pyGitHub ↗
• hooks/hook-version-parity-check.pyGitHub ↗
• hooks/jev-route-injector-userprompt.pyGitHub ↗
• hooks/mcp-health-check.pyGitHub ↗
• hooks/operator-context-detector.pyGitHub ↗
• hooks/pipeline-context-detector.pyGitHub ↗
• hooks/pipeline-phase-gate.pyGitHub ↗
• hooks/postcompact-handler.pyGitHub ↗
• hooks/posttool-bash-injection-scan.pyGitHub ↗
• hooks/posttool-docs-drift-alert.pyGitHub ↗
• hooks/posttool-rename-sweep.pyGitHub ↗
• hooks/posttool-security-scan.pyGitHub ↗
• hooks/posttool-session-reads.pyGitHub ↗
• hooks/posttool-skill-frontmatter-check.pyGitHub ↗
• hooks/posttool-voice-quality-check.pyGitHub ↗
• hooks/posttooluse-joy-check-warn.pyGitHub ↗
• hooks/posttooluse-sync-agent-index.pyGitHub ↗
• hooks/posttooluse-sync-skill-index.pyGitHub ↗
• hooks/precompact-archive.pyGitHub ↗
• hooks/pretool-adr-creation-gate.pyGitHub ↗
• hooks/pretool-branch-safety.pyGitHub ↗
• hooks/pretool-config-protection.pyGitHub ↗
• hooks/pretool-dispatch-spec-gate.pyGitHub ↗
• hooks/pretool-file-backup.pyGitHub ↗
• hooks/pretool-plan-gate.pyGitHub ↗
• hooks/pretool-private-name-leak-gate.pyGitHub ↗
• hooks/pretool-prompt-injection-scanner.pyGitHub ↗
• hooks/pretool-ruff-format-gate.pyGitHub ↗
• hooks/pretool-section-integrity-validator.pyGitHub ↗
• hooks/pretool-subagent-warmstart.pyGitHub ↗
• hooks/pretool-synthesis-gate.pyGitHub ↗
• hooks/pretool-unified-gate.pyGitHub ↗
• hooks/pretool-voice-publish-gate.pyGitHub ↗
• hooks/pretool-worktree-edit-guard.pyGitHub ↗
• hooks/prompt-capture.pyGitHub ↗
• hooks/reference-loading-enforcer.pyGitHub ↗
• hooks/retro-knowledge-injector.pyGitHub ↗
• hooks/review-capture.pyGitHub ↗
• hooks/review-false-positive-capture.pyGitHub ↗
• hooks/routing-decision-recorder.pyGitHub ↗
• hooks/routing-outcome-finalizer.pyGitHub ↗
• hooks/routing-outcome-recorder.pyGitHub ↗
• hooks/routing-outcome-stop-fallback.pyGitHub ↗
• hooks/rules-distill-injector.pyGitHub ↗
• hooks/rules-distill-trigger.pyGitHub ↗
• hooks/sapcc-go-detector.pyGitHub ↗
• hooks/security-review-hook.pyGitHub ↗
• hooks/session-adr-health-check.pyGitHub ↗
• hooks/session-context.pyGitHub ↗
• hooks/session-github-briefing.pyGitHub ↗
• hooks/session-manifest-cache.pyGitHub ↗
• hooks/session-summary.pyGitHub ↗
• hooks/session-task-registry.pyGitHub ↗
• hooks/skill-evaluator.pyGitHub ↗
• hooks/stop-drift-guard.pyGitHub ↗
• hooks/stop-failure-handler.pyGitHub ↗
• hooks/subagent-completion-guard.pyGitHub ↗
• hooks/subagent-start-warmstart.pyGitHub ↗
• hooks/subagent-state-tracker.pyGitHub ↗
• hooks/suggest-compact.pyGitHub ↗
• hooks/sync-to-user-claude.pyGitHub ↗
• hooks/team-config-loader.pyGitHub ↗
• hooks/usage-tracker.pyGitHub ↗
• hooks/voice-output-gate.pyGitHub ↗
• hooks/voice-pipeline-tracker.pyGitHub ↗
• hooks/zsh-shell-detector.pyGitHub ↗

All 77 scripts are listed above. The source is inlined for 6 of them, starting with whatever hooks.json actually runs. See all of them in the repo.

Read the script before you install anything that runs on your machine. This is the one part of a plugin that acts without being asked.