Hooks

Where it lives

• hooks/_common.pyGitHub ↗
  Read the script

  """Shared helpers for the human-review-skill hooks.
  
  All three hook scripts (snapshot, track_edits, review_gate) coordinate through a
  session-scoped state directory rooted at the project's current working directory:
  
      <cwd>/.claude/human-review/<session_id>/
          snapshots/<sha1-of-abs-path>.json
          edit_log.jsonl
          review_shown.flag
  
  Snapshots are the only reliable way to undo edits in arbitrary directories
  (the target project may not be a git repo).
  """
  
  from __future__ import annotations
  
  import hashlib
  import json
  import os
  import time
  from pathlib import Path
  from typing import Any
  
  # 1 MiB cap: above this, a file is considered too large to snapshot cheaply.
  MAX_SNAPSHOT_BYTES = 1_048_576
  
  # Purge session state dirs older than this many seconds (30 days).
  STATE_TTL_SECONDS = 30 * 24 * 60 * 60
  
  # Sentinel written instead of file content when snapshot is skipped.
  SKIP_BINARY_OR_LARGE = "binary-or-large"
  
  
  def review_root() -> Path:
      """Root dir holding all per-session state, anchored at the current project."""
      return Path.cwd() / ".claude" / "human-review"
  
  
  def state_dir(session_id: str, create: bool = True) -> Path:
      """Return the per-session state directory. Creates it (and `snapshots/`) if requested."""
      d = review_root() / session_id
      if create:
          (d / "snapshots").mkdir(parents=True, exist_ok=True)
      return d
  
  
  def snapshot_file(session_id: str, file_path: str) -> Path:
      """Stable, collision-safe snapshot filename for a given absolute path."""
      abs_path = os.path.abspath(file_path)
      h = hashlib.sha1(abs_path.encode("utf-8")).hexdigest()
      return state_dir(session_id) / "snapshots" / f"{h}.json"
  
  
  def edit_log(session_id: str) -> Path:
      return state_dir(session_id) / "edit_log.jsonl"
  
  
  def review_flag(session_id: str) -> Path:
      return state_dir(session_id) / "review_shown.flag"
  
  
  def append_log(session_id: str, entry: dict[str, Any]) -> None:
      with edit_log(session_id).open("a", encoding="utf-8") as f:
          f.write(json.dumps(entry) + "\n")
  
  
  def read_log(session_id: str) -> list[dict[str, Any]]:
      p = edit_log(session_id)
      if not p.exists():
          return []
      out: list[dict[str, Any]] = []
      with p.open("r", encoding="utf-8") as f:
          for line in f:
              line = line.strip()
              if not line:
                  continue
              try:
                  out.append(json.loads(line))
              except json.JSONDecodeError:
                  # Skip malformed lines rather than crash the hook.
                  continue
      return out
  
  
  def is_snapshotable(path: str) -> tuple[bool, str | None]:
      """Return (ok, reason_if_not). Binary or too-large files get skipped."""
      try:
          size = os.path.getsize(path)
      except OSError:
          return True, None  # Treat unreadable-stat as OK; caller will handle missing file.
      if size > MAX_SNAPSHOT_BYTES:
          return False, SKIP_BINARY_OR_LARGE
      try:
          with open(path, "rb") as f:
              chunk = f.read(8192)
      except OSError:
          return True, None
      # Simple binary heuristic: NUL byte presence.
      if b"\x00" in chunk:
          return False, SKIP_BINARY_OR_LARGE
      return True, None
  
  
  def read_text_best_effort(path: str) -> str:
      """Read a file as text. Falls back to latin-1 so any byte sequence round-trips."""
      try:
          with open(path, "r", encoding="utf-8") as f:
              return f.read()
      except UnicodeDecodeError:
          with open(path, "r", encoding="latin-1") as f:
              return f.read()
  
  
  def cleanup_stale_sessions() -> None:
      """Delete state dirs older than STATE_TTL_SECONDS. Best-effort; never raises."""
      root = review_root()
      if not root.exists():
          return
      cutoff = time.time() - STATE_TTL_SECONDS
      try:
          for child in root.iterdir():
              if not child.is_dir():
                  continue
              try:
                  if child.stat().st_mtime < cutoff:
                      _rmtree(child)
              except OSError:
                  pass
      except OSError:
          pass
  
  
  def _rmtree(path: Path) -> None:
      """Minimal rmtree so we avoid importing shutil in a hot hook path."""
      for sub in path.rglob("*"):
          if sub.is_file() or sub.is_symlink():
              try:
                  sub.unlink()
              except OSError:
                  pass
      # Remove directories depth-first.
      for sub in sorted((p for p in path.rglob("*") if p.is_dir()), key=lambda p: -len(p.parts)):
          try:
              sub.rmdir()
          except OSError:
              pass
      try:
          path.rmdir()
      except OSError:
          pass
  
  
  def load_hook_stdin() -> dict[str, Any]:
      """Parse the hook JSON envelope from stdin. Returns {} on any failure."""
      import sys
      try:
          raw = sys.stdin.read()
          if not raw.strip():
              return {}
          return json.loads(raw)
      except (json.JSONDecodeError, OSError):
          return {}
  
  
  def clear_session_state(session_id: str) -> None:
      """Wipe a session's state directory. Used by implicit-approval cleanup."""
      d = state_dir(session_id, create=False)
      if d.exists():
          _rmtree(d)
  

• hooks/review_gate.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Stop hook: if any edits occurred this turn, block the stop and inject a
  prompt telling Claude to load the human-review skill and produce a review.
  
  Contract:
  - Exit 0 with no stdout → Claude stops normally (no edits this turn, or review
    already shown).
  - Exit 0 with a JSON `{"decision": "block", "reason": "..."}` on stdout →
    Claude cannot stop yet; the `reason` is injected as a continuation prompt.
  
  The flag file `review_shown.flag` prevents an infinite loop: once we've
  injected the review prompt, we won't inject again this session until
  approve/undo clears state (or the dev sends a fresh prompt, which triggers
  implicit-approval cleanup in snapshot.py).
  """
  
  from __future__ import annotations
  
  import json
  import os
  import sys
  
  sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
  
  from _common import edit_log, load_hook_stdin, review_flag, state_dir  # noqa: E402
  
  
  REVIEW_PROMPT_TEMPLATE = (
      "You just edited files this turn. Before stopping, invoke the "
      "**human-review** skill and produce a review for the developer, "
      "following that skill's instructions exactly.\n\n"
      "Edit log (one JSON object per line) is at:\n"
      "    {log_path}\n\n"
      "Session state root (for approve/undo helpers):\n"
      "    {state_root}\n"
      "Session id: {session_id}\n\n"
      "Your review MUST end with the two numbered options "
      "`**1. Approve**` and `**2. Undo**` exactly as the skill specifies. "
      "The developer will reply with 1/approve or 2/undo on their next turn."
  )
  
  
  def main() -> int:
      # Allow a disable escape hatch for advanced users.
      if os.environ.get("HUMAN_REVIEW_DISABLED") == "1":
          return 0
  
      data = load_hook_stdin()
      session_id = data.get("session_id") or "default"
  
      log_path = edit_log(session_id)
      if not log_path.exists() or log_path.stat().st_size == 0:
          return 0  # No edits happened; let Claude stop.
  
      flag = review_flag(session_id)
      if flag.exists():
          return 0  # Review already shown; don't loop.
  
      # Mark shown *before* emitting, so even if Claude is interrupted we won't
      # re-trigger on the next Stop until approve/undo clears state.
      try:
          state_dir(session_id).mkdir(parents=True, exist_ok=True)
          flag.write_text("1", encoding="utf-8")
      except OSError:
          # If we can't persist the flag, still block once — better a duplicate
          # review than no review at all.
          pass
  
      reason = REVIEW_PROMPT_TEMPLATE.format(
          log_path=str(log_path),
          state_root=str(state_dir(session_id, create=False)),
          session_id=session_id,
      )
      sys.stdout.write(json.dumps({"decision": "block", "reason": reason}))
      sys.stdout.flush()
      return 0
  
  
  if __name__ == "__main__":
      try:
          sys.exit(main())
      except Exception:
          # Never crash Claude's stop path.
          sys.exit(0)
  

• hooks/snapshot.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """PreToolUse hook: snapshot a file's original contents before Edit/Write/NotebookEdit.
  
  The snapshot lets `scripts/undo.sh` restore the file exactly as it was at the
  start of the current Claude turn, even when the project is not a git repo.
  
  Design notes:
  - Only the *first* edit of a file per session is snapshotted. Subsequent edits
    are relative to an already-captured baseline, so undo restores the true
    pre-turn state.
  - Binary and >1MB files are marked as skipped rather than stored; the review
    flags this and undo emits a warning.
  - Implicit-approval cleanup: if a prior turn left `review_shown.flag` behind
    (dev ignored the review and moved on), we treat that turn as approved,
    wipe its snapshots/log, then start fresh for this turn.
  - This hook must never block Claude; all errors fall through to exit 0.
  """
  
  from __future__ import annotations
  
  import json
  import os
  import sys
  
  # Allow running both as `python3 hooks/snapshot.py` and when CLAUDE_PLUGIN_ROOT
  # places this directory on the path.
  sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
  
  from _common import (  # noqa: E402
      SKIP_BINARY_OR_LARGE,
      clear_session_state,
      cleanup_stale_sessions,
      is_snapshotable,
      load_hook_stdin,
      read_text_best_effort,
      review_flag,
      snapshot_file,
      state_dir,
  )
  
  
  def main() -> int:
      data = load_hook_stdin()
      session_id = data.get("session_id") or "default"
      tool_input = data.get("tool_input") or {}
      file_path = tool_input.get("file_path") or tool_input.get("notebook_path")
      if not file_path:
          return 0  # Nothing to snapshot.
  
      # Implicit-approval cleanup: a previous turn's review was never answered.
      # Treat it as approved and wipe its state before snapshotting this turn's edits.
      if review_flag(session_id).exists():
          clear_session_state(session_id)
  
      # Occasional housekeeping (very cheap when root is empty).
      cleanup_stale_sessions()
  
      state_dir(session_id)  # ensure directory tree exists
      snap = snapshot_file(session_id, file_path)
      if snap.exists():
          return 0  # First-edit-wins: preserve the true pre-turn baseline.
  
      abs_path = os.path.abspath(file_path)
  
      if not os.path.exists(abs_path):
          # File doesn't exist yet → this edit is a create. Mark it so undo can delete.
          payload = {"path": abs_path, "existed": False}
      else:
          ok, reason = is_snapshotable(abs_path)
          if not ok:
              payload = {"path": abs_path, "existed": True, "skipped": reason or SKIP_BINARY_OR_LARGE}
          else:
              try:
                  content = read_text_best_effort(abs_path)
                  payload = {"path": abs_path, "existed": True, "content": content}
              except OSError:
                  # Couldn't read for some reason; record path so undo can warn.
                  payload = {"path": abs_path, "existed": True, "skipped": "unreadable"}
  
      try:
          snap.write_text(json.dumps(payload), encoding="utf-8")
      except OSError:
          pass  # Never block the tool on snapshot failure.
  
      return 0
  
  
  if __name__ == "__main__":
      try:
          sys.exit(main())
      except Exception:
          # Defensive: hooks must never crash Claude's tool flow.
          sys.exit(0)
  

• hooks/track_edits.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """PostToolUse hook: append one log line per successful file edit.
  
  The edit log is consumed by:
  - `review_gate.py` to decide whether a review is needed at end-of-turn
  - `scripts/undo.sh` to know which files to restore and whether each was a
    pre-existing modification or a fresh creation
  
  Failed tool calls produce no log entry — the PreToolUse snapshot for that call
  becomes an orphan and is cleaned up on approve/undo or TTL.
  """
  
  from __future__ import annotations
  
  import json
  import os
  import sys
  import time
  
  sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
  
  from _common import append_log, load_hook_stdin, snapshot_file  # noqa: E402
  
  
  def _was_successful(tool_response: object) -> bool:
      """Best-effort success check across the variants Claude Code emits."""
      if isinstance(tool_response, dict):
          # Explicit failure signal wins.
          if tool_response.get("success") is False:
              return False
          if tool_response.get("error"):
              return False
          return True
      # Non-dict responses (strings, etc.) — assume success if we got anything.
      return tool_response is not None
  
  
  def main() -> int:
      data = load_hook_stdin()
      session_id = data.get("session_id") or "default"
      tool_name = data.get("tool_name") or ""
      tool_input = data.get("tool_input") or {}
      tool_response = data.get("tool_response")
  
      file_path = tool_input.get("file_path") or tool_input.get("notebook_path")
      if not file_path:
          return 0
  
      if not _was_successful(tool_response):
          return 0
  
      abs_path = os.path.abspath(file_path)
  
      # Determine whether this was a creation (snapshot marked `existed: False`).
      # If the snapshot is missing (unexpected), default to "modify" so undo at
      # least has a record that something was touched.
      action = "modify"
      snap = snapshot_file(session_id, abs_path)
      if snap.exists():
          try:
              meta = json.loads(snap.read_text(encoding="utf-8"))
              if meta.get("existed") is False:
                  action = "create"
          except (OSError, json.JSONDecodeError):
              pass
  
      try:
          append_log(session_id, {
              "ts": time.time(),
              "tool": tool_name,
              "path": abs_path,
              "action": action,
          })
      except OSError:
          pass
  
      return 0
  
  
  if __name__ == "__main__":
      try:
          sys.exit(main())
      except Exception:
          sys.exit(0)
  

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.