Hooks

Where it lives

• hooks/backup_on_change.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  backup_on_change.py -- PostToolUse hook that snapshots on config changes.
  
  Designed to be registered in ``~/.claude/settings.json`` under:
  
      "hooks": {
        "PostToolUse": [
          {
            "matcher": "Edit|Write|MultiEdit",
            "hooks": [
              {
                "type": "command",
                "command": "python <repo>/hooks/backup_on_change.py"
              }
            ]
          }
        ]
      }
  
  Claude Code delivers each PostToolUse event as a JSON payload on stdin.
  This script:
  
    1. Parses the payload.
    2. Checks whether the tool edited a file that BackupConfig tracks
       (top_files, trees, or projects/*/memory when memory_glob is on).
    3. If so, shells out to ``python src/backup_mirror.py snapshot-if-changed
       --reason <tool>:<basename>`` so the snapshot name records what fired it.
    4. Never blocks the tool: any error is logged to stderr and the hook
       exits 0 so a bug here can't stall the user's session.
  
  Snapshots only happen when content *actually* changed (SHA diff against
  the last snapshot's manifest) — so a no-op Edit won't create a folder.
  """
  
  from __future__ import annotations
  
  import json
  import os
  import subprocess
  import sys
  from pathlib import Path
  from typing import Any
  
  
  REPO_ROOT = Path(__file__).resolve().parent.parent
  SRC = REPO_ROOT / "src"
  if str(SRC) not in sys.path:
      sys.path.insert(0, str(SRC))
  
  
  def _load_payload() -> dict[str, Any]:
      """Read the PostToolUse payload from stdin. Empty on error."""
      try:
          raw = sys.stdin.read()
          if not raw.strip():
              return {}
          data = json.loads(raw)
          return data if isinstance(data, dict) else {}
      except (json.JSONDecodeError, OSError):
          return {}
  
  
  def _extract_touched_path(payload: dict[str, Any]) -> Path | None:
      """Pull the file path out of an Edit / Write / MultiEdit payload."""
      tool_input = payload.get("tool_input") or {}
      if not isinstance(tool_input, dict):
          return None
      # Edit, Write, MultiEdit all use ``file_path``.
      candidate = tool_input.get("file_path")
      if isinstance(candidate, str) and candidate:
          try:
              return Path(candidate).expanduser().resolve(strict=False)
          except (OSError, ValueError):
              return None
      return None
  
  
  def _is_tracked(path: Path, claude_home: Path) -> bool:
      """True when ``path`` is one of the files BackupConfig mirrors."""
      # Lazy import: hook must still function even when the rest of the
      # repo's dependency graph is in a weird state (e.g. during install).
      try:
          from backup_config import from_ctx_config  # noqa: PLC0415
      except ImportError:
          return False
  
      cfg = from_ctx_config()
  
      try:
          path_resolved = path.resolve(strict=False)
          home_resolved = claude_home.resolve(strict=False)
      except OSError:
          return False
  
      try:
          rel = path_resolved.relative_to(home_resolved)
      except ValueError:
          return False
  
      rel_posix = rel.as_posix()
  
      # Top-level files: match by basename against cfg.top_files.
      if rel_posix in cfg.top_files:
          return True
  
      # Trees: match any file under a tracked tree's src prefix.
      for tree in cfg.trees:
          prefix = tree.src.rstrip("/") + "/"
          if rel_posix == tree.src or rel_posix.startswith(prefix):
              return True
  
      # Memory glob: projects/<slug>/memory/...
      if cfg.memory_glob:
          parts = rel.parts
          if len(parts) >= 3 and parts[0] == "projects" and parts[2] == "memory":
              return True
  
      return False
  
  
  def _invoke_snapshot(reason: str) -> int:
      """Shell out to snapshot-if-changed. Returns child exit code (or 0)."""
      mirror = SRC / "backup_mirror.py"
      if not mirror.is_file():
          print(f"[backup_on_change] missing {mirror}", file=sys.stderr)
          return 0
      try:
          result = subprocess.run(
              [sys.executable, str(mirror), "snapshot-if-changed", "--reason", reason],
              capture_output=True,
              text=True,
              timeout=60,
              check=False,
          )
          if result.stdout.strip():
              print(result.stdout.strip(), file=sys.stderr)
          if result.returncode != 0 and result.stderr.strip():
              print(result.stderr.strip(), file=sys.stderr)
          return result.returncode
      except (OSError, subprocess.TimeoutExpired) as exc:
          print(f"[backup_on_change] snapshot failed: {exc}", file=sys.stderr)
          return 0
  
  
  def main() -> int:
      payload = _load_payload()
      tool_name = str(payload.get("tool_name") or "unknown")
  
      touched = _extract_touched_path(payload)
      if touched is None:
          return 0
  
      claude_home = Path(os.path.expanduser("~/.claude"))
      if not _is_tracked(touched, claude_home):
          return 0
  
      reason = f"{tool_name}:{touched.name}"
      _invoke_snapshot(reason)
      # Always exit 0: hook failures must not block the user's tool.
      return 0
  
  
  if __name__ == "__main__":
      sys.exit(main())
  

• hooks/quality_on_session_end.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  quality_on_session_end.py -- Stop hook that recomputes quality for the slugs
  this session touched.
  
  Designed for ``~/.claude/settings.json``:
  
      "hooks": {
        "Stop": [
          {
            "hooks": [
              {
                "type": "command",
                "command": "python <repo>/hooks/quality_on_session_end.py"
              }
            ]
          }
        ]
      }
  
  Why incremental instead of ``recompute --all``:
  
    - Full recompute walks every installed skill + agent (2,000+ pages) and
      runs four signal extractors per page. That's ~30s on a warm cache and
      dominates the tail of every session.
    - The only signals that *changed* since last session are telemetry
      (we logged new loads) and maybe intake (if the user edited a skill
      file). Every other signal moves on a slower clock.
    - So we compute the set of slugs that showed up in the telemetry event
      stream since the last time this hook ran, and rescore just those.
  
  Always exits 0: a hook that blocks session shutdown is worse than a
  slightly stale quality score.
  """
  
  from __future__ import annotations
  
  import json
  import os
  import subprocess
  import sys
  from datetime import datetime, timedelta, timezone
  from pathlib import Path
  from typing import Any
  
  
  REPO_ROOT = Path(__file__).resolve().parent.parent
  SRC = REPO_ROOT / "src"
  if str(SRC) not in sys.path:
      sys.path.insert(0, str(SRC))
  
  
  # How far back to look for touched slugs if no marker file exists.
  # Matches ``recent_window_days`` default in ``QualityConfig`` so a
  # freshly-installed system scores every recently-loaded skill on first run.
  _DEFAULT_LOOKBACK_HOURS = 24
  
  # State file: stores the ISO timestamp of the last successful run. Lives
  # under ~/.claude so it persists across repo clones and venv moves.
  _STATE_PATH = Path(os.path.expanduser("~/.claude/skill-quality/.hook-state.json"))
  _EVENTS_PATH = Path(os.path.expanduser("~/.claude/skill-events.jsonl"))
  
  # Upper bound on how many slugs we'll hand to the recompute subcommand in
  # one invocation. Pathological: a user loads 500 distinct skills in one
  # session. We'd rather recompute the top 50 than stall on session-end.
  _MAX_SLUGS_PER_RUN = 50
  
  
  def _load_payload() -> dict[str, Any]:
      try:
          raw = sys.stdin.read()
          if not raw.strip():
              return {}
          data = json.loads(raw)
          return data if isinstance(data, dict) else {}
      except (json.JSONDecodeError, OSError):
          return {}
  
  
  def _read_cutoff() -> datetime:
      """Return the 'since' cutoff for scanning events."""
      if _STATE_PATH.is_file():
          try:
              data = json.loads(_STATE_PATH.read_text(encoding="utf-8"))
              ts = data.get("last_run_at")
              if isinstance(ts, str):
                  parsed = datetime.fromisoformat(ts)
                  if parsed.tzinfo is None:
                      parsed = parsed.replace(tzinfo=timezone.utc)
                  return parsed.astimezone(timezone.utc)
          except (json.JSONDecodeError, ValueError, OSError):
              pass
      return datetime.now(timezone.utc) - timedelta(hours=_DEFAULT_LOOKBACK_HOURS)
  
  
  def _write_state(now: datetime) -> None:
      try:
          _STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
          _STATE_PATH.write_text(
              json.dumps({"last_run_at": now.isoformat(timespec="seconds")}),
              encoding="utf-8",
          )
      except OSError as exc:
          print(f"[quality_on_session_end] could not write state: {exc}", file=sys.stderr)
  
  
  def _touched_slugs_since(cutoff: datetime, events_path: Path) -> list[str]:
      """Return a deduplicated list of skill slugs that appear after ``cutoff``."""
      if not events_path.is_file():
          return []
      seen: dict[str, None] = {}
      try:
          with events_path.open(encoding="utf-8") as fh:
              for raw in fh:
                  line = raw.strip()
                  if not line:
                      continue
                  try:
                      obj = json.loads(line)
                  except json.JSONDecodeError:
                      continue
                  if not isinstance(obj, dict):
                      continue
                  slug = obj.get("skill")
                  ts_raw = obj.get("timestamp")
                  if not isinstance(slug, str) or not isinstance(ts_raw, str):
                      continue
                  try:
                      parsed = datetime.fromisoformat(ts_raw)
                  except ValueError:
                      continue
                  if parsed.tzinfo is None:
                      parsed = parsed.replace(tzinfo=timezone.utc)
                  if parsed < cutoff:
                      continue
                  # Insertion order preserved by dict in Python 3.7+.
                  seen.setdefault(slug, None)
      except OSError:
          return []
      return list(seen.keys())[:_MAX_SLUGS_PER_RUN]
  
  
  def _invoke_recompute(slugs: list[str], session_id: str | None = None) -> int:
      if not slugs:
          return 0
      script = SRC / "skill_quality.py"
      if not script.is_file():
          print(f"[quality_on_session_end] missing {script}", file=sys.stderr)
          return 1
      # Propagate session_id via environment so the per-slug
      # skill.score_updated audit rows carry it. Without this the
      # dashboard's per-session timeline drops the middle event in the
      # load -> score_updated -> unload triad.
      env = dict(os.environ)
      if session_id:
          env["CTX_SESSION_ID"] = session_id
      try:
          result = subprocess.run(
              [sys.executable, str(script), "recompute", "--slugs", ",".join(slugs)],
              capture_output=True,
              text=True,
              timeout=120,
              check=False,
              env=env,
          )
          if result.stderr.strip():
              print(result.stderr.strip(), file=sys.stderr)
          return result.returncode
      except (OSError, subprocess.TimeoutExpired) as exc:
          print(f"[quality_on_session_end] recompute failed: {exc}", file=sys.stderr)
          return 1
  
  
  def

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.