Hooks

Where it lives

• hooks/_captain_hook_lite.pyGitHub ↗
  Read the script

  # Vendored from captain-hook/ — re-sync when captain-hook schema changes.
  # Source: captain-hook/src/captain_hook/
  #
  # Self-contained subset of captain-hook Pydantic models, covering only the
  # hooks consumed by live_session_tracker.py:
  #   - SessionStart, SessionEnd
  #   - UserPromptSubmit
  #   - PostToolUse
  #   - Notification, PermissionRequest
  #   - Stop
  #   - SubagentStart, SubagentStop
  #
  # Field set intentionally minimal — only what the tracker reads today, plus
  # fields recently surfaced as useful (SessionStart.model, SessionStart.agent_type,
  # Notification.message, PermissionRequest.message).
  #
  # Pydantic is required at hook runtime. If unavailable, parse_hook_event()
  # returns a lightweight dict-shim with attribute access so callers keep working.
  
  from __future__ import annotations
  
  from typing import Any, Dict, Optional, Union
  
  try:
      from pydantic import BaseModel, ConfigDict, Field
  
      HAS_PYDANTIC = True
  except ImportError:  # pragma: no cover - fallback only when pydantic missing
      HAS_PYDANTIC = False
  
  
  if HAS_PYDANTIC:
  
      class _BaseHook(BaseModel):
          """Base for all hook events. extra='allow' for forward compatibility."""
  
          model_config = ConfigDict(extra="allow")
  
          session_id: str = Field(..., description="Claude Code session UUID")
          transcript_path: str = Field("", description="Path to session JSONL transcript")
          cwd: str = Field("", description="Working directory when hook fired")
          permission_mode: str = Field("default", description="Current permission mode")
          hook_event_name: str = Field(..., description="Hook event name")
  
      class SessionStartHook(_BaseHook):
          hook_event_name: str = Field("SessionStart")
          source: Optional[str] = Field(
              None, description="startup, resume, clear, or compact"
          )
          model: Optional[str] = Field(None, description="Claude model identifier")
          agent_type: Optional[str] = Field(
              None, description="Agent type if started with --agent flag"
          )
  
      class SessionEndHook(_BaseHook):
          hook_event_name: str = Field("SessionEnd")
          reason: Optional[str] = Field(
              None, description="prompt_input_exit, clear, logout, other"
          )
  
      class UserPromptSubmitHook(_BaseHook):
          hook_event_name: str = Field("UserPromptSubmit")
          prompt: Optional[str] = Field(None, description="The user's submitted text")
  
      class PostToolUseHook(_BaseHook):
          hook_event_name: str = Field("PostToolUse")
          tool_name: Optional[str] = Field(None)
          tool_use_id: Optional[str] = Field(None)
  
      class NotificationHook(_BaseHook):
          hook_event_name: str = Field("Notification")
          notification_type: str = Field("", description="permission_prompt, idle_prompt, ...")
          message: Optional[str] = Field(None, description="Notification text")
  
      class PermissionRequestHook(_BaseHook):
          hook_event_name: str = Field("PermissionRequest")
          notification_type: str = Field("")
          message: Optional[str] = Field(None, description="Permission prompt text")
  
      class StopHook(_BaseHook):
          hook_event_name: str = Field("Stop")
          # A missing field means no stop hook is running — a natural stop.
          stop_hook_active: bool = Field(
              False,
              description="True if already continuing from a previous Stop hook",
          )
  
      class SubagentStartHook(_BaseHook):
          hook_event_name: str = Field("SubagentStart")
          agent_id: Optional[str] = Field(None)
          agent_type: str = Field("unknown")
  
      class SubagentStopHook(_BaseHook):
          hook_event_name: str = Field("SubagentStop")
          agent_id: Optional[str] = Field(None)
          agent_transcript_path: Optional[str] = Field(None)
          stop_hook_active: bool = Field(False)
  
      HookEvent = Union[
          SessionStartHook,
          SessionEndHook,
          UserPromptSubmitHook,
          PostToolUseHook,
          NotificationHook,
          PermissionRequestHook,
          StopHook,
          SubagentStartHook,
          SubagentStopHook,
          _BaseHook,
      ]
  
      _HOOK_TYPE_MAP: Dict[str, type] = {
          "SessionStart": SessionStartHook,
          "SessionEnd": SessionEndHook,
          "UserPromptSubmit": UserPromptSubmitHook,
          "PostToolUse": PostToolUseHook,
          "Notification": NotificationHook,
          "PermissionRequest": PermissionRequestHook,
          "Stop": StopHook,
          "SubagentStart": SubagentStartHook,
          "SubagentStop": SubagentStopHook,
      }
  
      def parse_hook_event(data: Dict[str, Any]) -> HookEvent:
          """Parse hook dict into a typed model. Falls back to _BaseHook for unknowns."""
          hook_name = data.get("hook_event_name")
          if not hook_name:
              raise ValueError("Missing 'hook_event_name' in hook data")
          hook_class = _HOOK_TYPE_MAP.get(hook_name, _BaseHook)
          return hook_class.model_validate(data)
  
  else:  # pragma: no cover - fallback path
      # Dict-shim with attribute access so the rest of the tracker can keep
      # using `hook.field` even without pydantic installed.
  
      class _DictShim:
          """Lightweight attribute-access wrapper around a hook dict."""
  
          __slots__ = ("_data",)
  
          def __init__(self, data: Dict[str, Any]) -> None:
              self._data = data
  
          def __getattr__(self, name: str) -> Any:
              if name.startswith("_"):
                  raise AttributeError(name)
              return self._data.get(name)
  
          def model_dump(self) -> Dict[str, Any]:
              return dict(self._data)
  
      HookEvent = _DictShim  # type: ignore[assignment,misc]
  
      def parse_hook_event(data: Dict[str, Any]) -> _DictShim:
          """Fallback parser when pydantic is unavailable — returns dict-shim."""
          if not data.get("hook_event_name"):
              raise ValueError("Missing 'hook_event_name' in hook data")
          return _DictShim(data)
  
  
  __all__ = ["parse_hook_event", "HookEvent", "HAS_PYDAN

• hooks/cron_state_capture.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  Cron live-state capture hook for Claude Code Karma.
  
  Optional PostToolUse hook. Watches CronCreate / CronDelete / CronList tool
  calls and writes per-session event logs + the latest CronList snapshot to:
  
      ~/.claude_karma/cron-state/{session_uuid}/
          events.jsonl     append-only — one record per CronCreate/Delete/List
          snapshot.json    overwrite   — the latest CronList response
  
  The karma indexer ingests events.jsonl into the cron_state_snapshots SQLite
  table (idempotent via UNIQUE(session_uuid, trigger_event, captured_at)).
  
  Why this is opt-in:
      Cron state in Claude Code lives in-memory only; karma can normally
      reconstruct CronCreate/CronDelete events from JSONL after the fact.
      The hook adds ground-truth CronList snapshots so karma can show "what
      is scheduled RIGHT NOW" instead of only "what was scheduled in this
      session's history." Off by default to keep karma minimal.
  
  Install:
      Add to ~/.claude/settings.json:
          "hooks": {
              "PostToolUse": [{
                  "matcher": "CronCreate|CronDelete|CronList",
                  "hooks": [{
                      "type": "command",
                      "command": "/path/to/claude-karma/hooks/cron_state_capture.py"
                  }]
              }]
          }
  
  The hook silently no-ops on any error so it never blocks a Claude Code
  session. Errors are logged to ~/.claude_karma/cron-state/.errors.log so
  they remain debuggable.
  """
  
  from __future__ import annotations
  
  import json
  import sys
  import traceback
  from datetime import datetime, timezone
  from pathlib import Path
  from typing import Any, Dict
  
  KARMA_BASE = Path.home() / ".claude_karma"
  STATE_ROOT = KARMA_BASE / "cron-state"
  ERROR_LOG = STATE_ROOT / ".errors.log"
  
  WATCH_TOOLS = {"CronCreate", "CronDelete", "CronList"}
  
  
  def _now_z() -> str:
      """ISO-8601 with Z suffix; matches karma's stored timestamp format."""
      return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
  
  
  def _log_error(detail: str) -> None:
      """Append to a per-user error log. Best-effort: ignores write failures."""
      try:
          STATE_ROOT.mkdir(parents=True, exist_ok=True)
          with ERROR_LOG.open("a", encoding="utf-8") as fp:
              fp.write(f"{_now_z()}\t{detail}\n")
      except OSError:
          pass
  
  
  def main() -> int:
      """Read PostToolUse event JSON from stdin and persist if relevant."""
      try:
          raw = sys.stdin.read()
          if not raw:
              return 0
          event: Dict[str, Any] = json.loads(raw)
      except (json.JSONDecodeError, OSError) as e:
          _log_error(f"stdin parse: {e}")
          return 0
  
      tool_name = event.get("tool_name")
      if tool_name not in WATCH_TOOLS:
          return 0
  
      session_uuid = event.get("session_id") or event.get("sessionId")
      if not session_uuid:
          _log_error("missing session_id")
          return 0
  
      try:
          out_dir = STATE_ROOT / session_uuid
          out_dir.mkdir(parents=True, exist_ok=True)
  
          record = {
              "captured_at": _now_z(),
              "trigger_event": tool_name,
              "tool_input": event.get("tool_input"),
              "tool_response": event.get("tool_response"),
              "cwd": event.get("cwd"),
          }
  
          # Append every event to events.jsonl. Atomic line-write — no rotation,
          # files stay small (one line per cron tool call).
          with (out_dir / "events.jsonl").open("a", encoding="utf-8") as fp:
              fp.write(json.dumps(record, default=str) + "\n")
  
          # On CronList: overwrite snapshot.json with the live state.
          if tool_name == "CronList":
              snapshot = {
                  "captured_at": record["captured_at"],
                  "session_uuid": session_uuid,
                  "jobs": event.get("tool_response"),
              }
              snapshot_path = out_dir / "snapshot.json"
              tmp_path = snapshot_path.with_suffix(".json.tmp")
              with tmp_path.open("w", encoding="utf-8") as fp:
                  json.dump(snapshot, fp, indent=2, default=str)
              tmp_path.replace(snapshot_path)  # atomic on POSIX
      except OSError as e:
          _log_error(f"write: {e}")
          return 0
      except Exception:  # noqa: BLE001 — hook must never propagate
          _log_error(f"unexpected:\n{traceback.format_exc()}")
          return 0
  
      return 0
  
  
  if __name__ == "__main__":
      sys.exit(main())
  

• hooks/live_session_tracker.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  Live session state tracker for Claude Code Karma.
  
  Writes session state to ``~/.claude_karma/live-sessions/{session_id}.json``
  based on Claude Code hook events.
  
  Session states:
  - STARTING: Session started, waiting for first message
  - LIVE: Session actively running (tool execution)
  - WAITING: Claude needs user input (AskUserQuestion, permission dialog)
  - STOPPED: Agent finished but session still open
  - STALE: User has been idle for 60+ seconds
  - ENDED: Session terminated
  
  Hook → state mapping:
  - SessionStart → STARTING
  - UserPromptSubmit → LIVE
  - PostToolUse → LIVE
  - Notification(permission_prompt) → WAITING
  - Notification(idle_prompt) → STALE (unless already WAITING)
  - Stop (stop_hook_active=false) → STOPPED
  - SessionEnd → ENDED (with end_reason)
  
  Files are always keyed by ``session_id``. The legacy slug-based filename
  scheme was removed — Claude Code never emits a top-level ``slug`` field
  outside SummaryMessage, so every state file on disk had ``slug=null``.
  The ``slug`` field on the schema is preserved but never written by this
  hook (older files with a populated slug remain readable).
  """
  
  from __future__ import annotations
  
  import json
  import os
  import subprocess
  import sys
  import traceback
  from datetime import datetime, timezone
  from pathlib import Path
  from typing import Any, Callable, Dict, Optional
  
  # Vendored captain-hook subset — self-contained, no /api or /captain-hook deps.
  from _captain_hook_lite import parse_hook_event
  
  # Platform-specific file locking
  try:
      import fcntl
  
      HAS_FCNTL = True
  except ImportError:
      HAS_FCNTL = False
  
  try:
      import msvcrt
      import time
  
      HAS_MSVCRT = True
  except ImportError:
      HAS_MSVCRT = False
  
  LIVE_SESSIONS_DIR = Path.home() / ".claude_karma" / "live-sessions"
  ERROR_LOG = LIVE_SESSIONS_DIR / ".errors.log"
  
  
  def _log_error(detail: str) -> None:
      """Append to a per-user error log. Best-effort: ignores write failures.
  
      Mirrors hooks/cron_state_capture.py so a hook failure stays debuggable
      without ever propagating to (and breaking) the Claude Code session.
      """
      try:
          LIVE_SESSIONS_DIR.mkdir(parents=True, exist_ok=True)
          ts = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
          with ERROR_LOG.open("a", encoding="utf-8") as fp:
              fp.write(f"{ts}\t{detail}\n")
      except OSError:
          pass
  
  
  def resolve_git_root(cwd: str) -> Optional[str]:
      """Resolve the real git root from cwd.
  
      For worktrees, ``--show-toplevel`` returns the worktree root (not the
      main repo). We use ``--git-common-dir`` to find the shared .git
      directory, whose parent is the actual repository root.
      """
      try:
          result = subprocess.run(
              ["git", "rev-parse", "--show-toplevel", "--git-common-dir"],
              cwd=cwd,
              capture_output=True,
              text=True,
              timeout=5,
          )
          if result.returncode != 0:
              return None
  
          lines = result.stdout.strip().splitlines()
          toplevel = lines[0]
          common_dir = lines[1] if len(lines) > 1 else None
  
          if common_dir:
              common_path = Path(common_dir)
              if not common_path.is_absolute():
                  common_path = (Path(cwd) / common_path).resolve()
              real_root = str(common_path.parent)
              if real_root != toplevel:
                  return real_root
  
          return toplevel
      except (subprocess.TimeoutExpired, OSError):
          pass
      return None
  
  
  def _tty_of_pid(pid: Optional[int]) -> Optional[str]:
      """Resolve a process's controlling tty via ``ps``, e.g. '/dev/ttys006'."""
      if not pid:
          return None
      try:
          result = subprocess.run(
              ["ps", "-o", "tty=", "-p", str(pid)],
              capture_output=True,
              text=True,
              timeout=5,
          )
          tty = result.stdout.strip()
          if result.returncode == 0 and tty and tty not in ("??", "-"):
              return tty if tty.startswith("/dev/") else f"/dev/{tty}"
      except (subprocess.SubprocessError, OSError):
          pass
      return None
  
  
  def resolve_terminal() -> Dict[str, Any]:
      """Capture the terminal/pane a session is running in from the environment.
  
      Claude Code spawns hooks with the terminal's environment inherited, so
      the same env vars the user's shell sees are visible here. We record the
      identifiers the API later uses to raise the right window/pane:
  
      - ``TMUX`` / ``TMUX_PANE``   → running inside tmux (works on any OS)
      - ``TERM_PROGRAM`` / ``TERM_SESSION_ID`` → macOS terminal apps (osascript)
      - ``ITERM_SESSION_ID``       → iTerm2 tab/session UUID (survives pid death)
      - ``__CFBundleIdentifier``   → exact macOS app that spawned the shell
      - ``WINDOWID``               → Linux X11 window (xdotool/wmctrl)
  
      ``pid`` is the hook's parent — the live ``claude`` process. ``tty`` is
      that process's controlling terminal, resolved NOW while the process is
      alive; the API prefers it over a click-time pid→tty lookup so tab
      matching survives pid death and recycling. Inside tmux the tty is the
      tmux server's pty (not a GUI tab) — the API detects that via ``tmux``.
      All fields are best-effort; missing ones stay ``None``.
      """
      env = os.environ
      tmux_pane = env.get("TMUX_PANE") or None
      try:
          pid: Optional[int] = os.getppid()
      except OSError:
          pid = None
      return {
          "tmux": bool(env.get("TMUX")),
          "tmux_pane": tmux_pane,
          "term_program": env.get("TERM_PROGRAM") or None,
          "term_session_id": env.get("TERM_SESSION_ID") or None,
          "iterm_session_id": env.get("ITERM_SESSION_ID") or None,
          "bundle_id": env.get("__CFBundleIdentifier") or None,
          "window_id": env.get("WINDOWID") or None,
          "pid": pid,
          "tty": _tty_of_pid(pid),
      }
  
  
  def _terminal_for(existing_terminal: Optional[Dict[str, Any]]) -> Dict[str, Any]:
      """Keep a current-format terminal dict; (re-)resolve 

• hooks/plan_approval.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  Plan approval hook for Claude Code's PermissionRequest hook.
  
  Intercepts ExitPlanMode calls and checks plan approval status
  in Claude Code Karma before allowing Claude to proceed.
  
  Decision Logic:
  - If tool is not ExitPlanMode: continue (allow Claude to proceed)
  - If plan status is "approved": allow
  - If plan status is "changes_requested" or has annotations: deny with feedback
  - If plan is pending or API error: deny, prompt user to review in Claude Code Karma UI
  
  Usage:
      This script is called by Claude Code's PermissionRequest hook when
      ExitPlanMode is invoked. Configure in hooks.json:
  
      {
        "hooks": {
          "PermissionRequest": [
            {
              "matcher": "ExitPlanMode",
              "hooks": [
                {
                  "type": "command",
                  "command": "python /path/to/plan_approval.py",
                  "timeout": 30
                }
              ]
            }
          ]
        }
      }
  """
  
  import json
  import sys
  import urllib.request
  import urllib.error
  from pathlib import Path
  from typing import Optional
  
  # Claude Code Karma API base URL
  API_BASE_URL = "http://localhost:8020"
  
  
  def output_continue() -> None:
      """Output continue response (don't intercept this permission request)."""
      print(json.dumps({"continue": True}))
  
  
  def output_allow() -> None:
      """Output allow response (let Claude proceed with ExitPlanMode)."""
      print(json.dumps({
          "hookSpecificOutput": {
              "decision": {
                  "behavior": "allow"
              }
          }
      }))
  
  
  def output_deny(message: str) -> None:
      """Output deny response with feedback message."""
      print(json.dumps({
          "hookSpecificOutput": {
              "decision": {
                  "behavior": "deny",
                  "message": message
              }
          }
      }))
  
  
  def extract_slug_from_tool_input(tool_input: dict) -> Optional[str]:
      """
      Extract the plan slug from tool_input.
  
      The plan content in tool_input may contain a reference to the plan file,
      or we can derive the slug from the plan file path if available.
  
      Args:
          tool_input: The tool_input dictionary from the hook event
  
      Returns:
          Plan slug if found, None otherwise
      """
      # Check if there's a plan_path field
      plan_path = tool_input.get("plan_path") or tool_input.get("planPath")
      if plan_path:
          # Extract slug from path like ~/.claude/plans/{slug}.md
          path = Path(plan_path)
          if path.suffix == ".md":
              return path.stem
  
      # Check if there's a slug field directly
      slug = tool_input.get("slug") or tool_input.get("plan_slug")
      if slug:
          return slug
  
      # Try to extract from plan content (look for a slug pattern in first line)
      plan_content = tool_input.get("plan", "")
      if plan_content:
          # Plans often have a slug in their metadata or filename reference
          # Look for common patterns like "Plan: {slug}" or "# {slug}"
          first_lines = plan_content.split('\n')[:5]
          for line in first_lines:
              # Skip empty lines and markdown headers
              line = line.strip()
              if line.startswith("Plan:"):
                  potential_slug = line.replace("Plan:", "").strip()
                  if "-" in potential_slug and len(potential_slug) < 50:
                      return potential_slug
  
      return None
  
  
  def get_active_plan_slug() -> Optional[str]:
      """
      Get the currently active plan slug from the plans directory.
  
      Falls back to finding the most recently modified plan.
  
      Returns:
          Plan slug if found, None otherwise
      """
      plans_dir = Path.home() / ".claude" / "plans"
      if not plans_dir.exists():
          return None
  
      # Find most recently modified plan
      plan_files = list(plans_dir.glob("*.md"))
      if not plan_files:
          return None
  
      # Sort by modification time, most recent first
      plan_files.sort(key=lambda p: p.stat().st_mtime, reverse=True)
      return plan_files[0].stem
  
  
  def api_get(endpoint: str) -> tuple[Optional[dict], Optional[str]]:
      """
      Make a GET request to the Claude Code Karma API.
  
      Args:
          endpoint: API endpoint (e.g., "/plans/{slug}/status")
  
      Returns:
          Tuple of (response_data, error_message)
      """
      url = f"{API_BASE_URL}{endpoint}"
      try:
          req = urllib.request.Request(url, headers={"Accept": "application/json"})
          with urllib.request.urlopen(req, timeout=10) as response:
              data = json.loads(response.read().decode("utf-8"))
              return data, None
      except urllib.error.HTTPError as e:
          if e.code == 404:
              return None, f"Not found: {endpoint}"
          return None, f"HTTP {e.code}: {e.reason}"
      except urllib.error.URLError as e:
          return None, f"Connection error: {e.reason}"
      except json.JSONDecodeError:
          return None, "Invalid JSON response from API"
      except Exception as e:
          return None, f"Error: {str(e)}"
  
  
  def format_annotation(annotation: dict) -> str:
      """
      Format a single annotation for display in the deny message.
  
      Args:
          annotation: Annotation dictionary from the API
  
      Returns:
          Formatted annotation string
      """
      ann_type = annotation.get("type", "UNKNOWN")
      original_text = annotation.get("original_text", "")
      new_text = annotation.get("new_text")
      comment = annotation.get("comment")
  
      # Truncate long text
      if len(original_text) > 100:
          original_text = original_text[:100] + "..."
  
      lines = [f"- [{ann_type}]"]
  
      if original_text:
          lines.append(f"  Original: \"{original_text}\"")
  
      if ann_type == "REPLACEMENT" and new_text:
          if len(new_text) > 100:
              new_text = new_text[:100] + "..."
          lines.append(f"  Replace with: \"{new_text}\"")
      elif ann_type == "INSERTION" and new_text:
          if len(new_text) > 100:
              new_text = new_text[:100] + "..."
          lines.append(f"  Insert: \"{new_text}\"")
     

• hooks/project_identity.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  Project identity sidecar writer for Claude Code Karma.
  
  Fires on SessionStart. Reads the session's cwd, asks git for the
  `remote.origin.url` and the repo toplevel, and writes
  `karma-project.json` next to the session transcript
  (~/.claude/projects/<encoded>/). After the user syncs ~/.claude to
  another machine, Karma there can match the project by git identity
  instead of the machine-specific encoded path.
  
  Sidecar shape:
  
      {
        "git_identity": "owner/repo",
        "subdir": "api",
        "project_path": "/Users/me/repo/api",
        "written_by": "hook"
      }
  
  Silent on every failure — this hook NEVER blocks SessionStart.
  Stdlib only; Python 3.9+.
  """
  
  from __future__ import annotations
  
  import json
  import os
  import re
  import subprocess
  import sys
  from pathlib import Path
  from typing import Optional
  
  SIDECAR_NAME = "karma-project.json"
  TMP_NAME = ".karma-project.json.tmp"
  
  # Claude Code's project dir naming (see encode_project_dir)
  _ALNUM_RE = re.compile(r"[A-Za-z0-9]")
  _DRIVE_RE = re.compile(r"^[A-Za-z]:")
  _MAX_ENCODED_LEN = 200
  _BASE36 = "0123456789abcdefghijklmnopqrstuvwxyz"
  
  # Inlined from api/services/git_identity.py (hooks can't import api).
  _SCP_RE = re.compile(r"^[^@]+@[^:]+:(?P<path>.+)$")
  _URL_RE = re.compile(r"^[a-z]+://(?:[^@/]+@)?[^/]+/(?P<path>.+)$", re.IGNORECASE)
  
  
  def normalize_git_url(url: Optional[str]) -> Optional[str]:
      """Parse a git remote URL into canonical `owner/repo` (lowercase)."""
      if not url:
          return None
      url = url.strip()
      if not url:
          return None
  
      # A scheme URL with a port (ssh://git@host:2222/...) would also match the scp regex
      m = _URL_RE.match(url) if "://" in url else _SCP_RE.match(url)
      if m is None:
          return None
  
      path = m.group("path").strip("/")
      if path.lower().endswith(".git"):
          path = path[:-4]
  
      parts = [p for p in path.split("/") if p]
      # Azure DevOps: https://dev.azure.com/org/project/_git/repo
      if len(parts) >= 3 and parts[-2] == "_git":
          del parts[-2]
      if len(parts) < 2:
          return None
  
      owner, repo = parts[-2], parts[-1]
      return f"{owner}/{repo}".lower()
  
  
  def _git(cwd: str, *args: str) -> Optional[str]:
      """Run `git -C cwd <args>`; return stripped stdout or None."""
      try:
          result = subprocess.run(
              ["git", "-C", cwd, *args],
              capture_output=True,
              # git prints UTF-8 paths; the locale codec would mangle them on Windows
              encoding="utf-8",
              errors="replace",
              timeout=2,
              check=False,
          )
      except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
          return None
      if result.returncode != 0:
          return None
      out = result.stdout.strip()
      return out or None
  
  
  def _norm_path(p: str) -> str:
      return p.replace("\\", "/").rstrip("/")
  
  
  def _relative(cwd: str, root: str) -> Optional[str]:
      c, r = _norm_path(cwd), _norm_path(root)
      if c.lower() == r.lower():
          return ""
      prefix = r + "/"
      if c.lower().startswith(prefix.lower()):
          return c[len(prefix) :]
      return None
  
  
  def compute_subdir(cwd: str, git_root: str) -> Optional[str]:
      """cwd relative to git_root, '/'-separated; '' at root, None if unrelated."""
      rel = _relative(cwd, git_root)
      if rel is None:
          try:
              rel = _relative(os.path.realpath(cwd), os.path.realpath(git_root))
          except (OSError, ValueError):
              rel = None
      return rel
  
  
  def build_identity(cwd: str) -> Optional[dict]:
      """Return the sidecar dict for cwd, or None if no git identity."""
      if not cwd:
          return None
      identity = normalize_git_url(_git(cwd, "config", "--get", "remote.origin.url"))
      if not identity:
          return None
      root = _git(cwd, "rev-parse", "--show-toplevel")
      if root is None:
          # rev-parse failed or timed out: the subdir is unknown, so write nothing
          return None
      subdir = compute_subdir(cwd, root)
      if subdir is None:
          # A wrong subdir would merge this project into its repo root elsewhere
          return None
      return {
          "git_identity": identity,
          "subdir": subdir,
          "project_path": cwd,
          "written_by": "hook",
      }
  
  
  def _utf16_units(s: str) -> list:
      data = s.encode("utf-16-le", "surrogatepass")
      return [int.from_bytes(data[i : i + 2], "little") for i in range(0, len(data), 2)]
  
  
  def _base36(n: int) -> str:
      out = ""
      while True:
          n, r = divmod(n, 36)
          out = _BASE36[r] + out
          if not n:
              return out
  
  
  def encode_project_dir(path: str) -> str:
      """Claude Code's project dir name for a cwd (per UTF-16 unit; long names hashed)."""
      units = _utf16_units(path)
      encoded = "".join(
          chr(u) if u < 128 and _ALNUM_RE.match(chr(u)) else "-" for u in units
      )
      if len(encoded) <= _MAX_ENCODED_LEN:
          return encoded
      h = 0
      for u in units:  # (h << 5) - h + u, as a signed 32-bit int
          h = (h * 31 + u) & 0xFFFFFFFF
      if h >= 0x80000000:
          h -= 0x100000000
      return encoded[:_MAX_ENCODED_LEN] + "-" + _base36(abs(h))
  
  
  def dir_matches_cwd(dir_name: str, cwd: str) -> bool:
      """True if dir_name is cwd's own project dir, not a start dir left after a cd."""
      candidates = {cwd}
      try:
          candidates.add(os.path.realpath(cwd))
      except (OSError, ValueError):
          pass
      # Windows paths (drive letters show up in either case) compare case-insensitively
      if _DRIVE_RE.match(cwd):
          return any(encode_project_dir(c).lower() == dir_name.lower() for c in candidates)
      return any(encode_project_dir(c) == dir_name for c in candidates)
  
  
  def drop_misplaced_sidecar(project_dir: Path) -> None:
      """Delete a sidecar an older version wrote after a cd (project_path of another dir)."""
      target = project_dir / SIDECAR_NAME
      try:
          existing = json.loads(target.read_text(encoding="utf-8"))
      except (OSError, ValueError):
          return
      pat

• hooks/session_title_generator.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  Session title generator hook for Claude Code Karma.
  
  Fires on SessionEnd, reads the session JSONL, extracts context,
  and generates a concise title. Prefers git commit messages when
  available (free, no LLM). Falls back to Claude Haiku via
  `claude -p --no-session-persistence` to avoid creating bloat sessions.
  
  """
  
  import json
  import os
  import re
  import subprocess
  import sys
  from pathlib import Path
  from typing import Optional, Tuple
  
  API_BASE = os.environ.get("CLAUDE_KARMA_API", "http://localhost:8020")
  MAX_PROMPT_LENGTH = 500
  MAX_RESPONSE_LENGTH = 300
  TITLE_MAX_WORDS = 10
  FALLBACK_TITLE_MAX_CHARS = 60
  
  
  def _strip_system_tags(text: str) -> str:
      """Remove XML system/skill tags, keeping any real user text around them."""
      cleaned = re.sub(
          r"<(?:command-message|system-reminder|command-name|user-prompt-submit-hook)[^>]*>.*?</(?:command-message|system-reminder|command-name|user-prompt-submit-hook)>",
          "",
          text,
          flags=re.DOTALL,
      )
      # Also strip self-closing variants
      cleaned = re.sub(r"<(?:command-message|system-reminder|command-name|user-prompt-submit-hook)[^/]*/?>", "", cleaned)
      return cleaned.strip()
  
  
  def truncate_title(text: str, limit: int = FALLBACK_TITLE_MAX_CHARS) -> str:
      """Collapse whitespace and cut at a word boundary to at most `limit` chars, ending in one "…"."""
      text = " ".join(text.split())
      if len(text) <= limit:
          return text
      cut = text[: limit - 1]
      space = cut.rfind(" ")
      # Back up to the last space unless a word ends right at the cut; a lone long word (path, URL) is cut mid-word
      if text[limit - 1] != " " and space >= limit // 2:
          cut = cut[:space]
      return cut.rstrip(" ,.;:-") + "…"
  
  
  def _extract_text(msg: dict) -> str:
      """Extract plain text from a message's content (handles string and list forms)."""
      content = msg.get("content", "")
      if isinstance(content, list):
          texts = [c.get("text", "") for c in content if c.get("type") == "text"]
          content = " ".join(texts)
      return content.strip() if isinstance(content, str) else ""
  
  
  def _get_session_start_iso(transcript_path: str) -> Optional[str]:
      """Extract the ISO timestamp of the first entry in a JSONL transcript."""
      try:
          with open(transcript_path, "r", encoding="utf-8", errors="replace") as f:
              for line in f:
                  line = line.strip()
                  if not line:
                      continue
                  try:
                      entry = json.loads(line)
                      ts = entry.get("timestamp")
                      if ts:
                          return ts
                  except json.JSONDecodeError:
                      continue
      except (OSError, IOError):
          pass
      return None
  
  
  def main():
      try:
          # Claude Code sends UTF-8; the locale codec (cp1252 on Windows) would mangle paths
          data = json.loads(sys.stdin.buffer.read().decode("utf-8", errors="replace"))
      except (json.JSONDecodeError, EOFError):
          return
  
      session_id = data.get("session_id", "")
      transcript_path = data.get("transcript_path", "")
      cwd = data.get("cwd", "")
      reason = data.get("reason", "")
  
      # Skip if no transcript or if cleared (not meaningful sessions)
      if not transcript_path or not Path(transcript_path).exists():
          return
      if reason in ("clear",):
          return
  
      # Extract context from JSONL
      initial_prompt, first_response = extract_session_context(transcript_path)
      if not initial_prompt:
          return
  
      # Get git commits during session
      git_context = get_git_context(cwd, transcript_path)
  
      # Generate title
      title, source = generate_title(initial_prompt, first_response, git_context)
  
      if title:
          post_title(session_id, title)
  
  
  def extract_session_context(transcript_path: str) -> Tuple[Optional[str], Optional[str]]:
      """Extract initial prompt and first assistant response from JSONL."""
      initial_prompt = None
      first_response = None
  
      try:
          with open(transcript_path, "r", encoding="utf-8", errors="replace") as f:
              for line in f:
                  line = line.strip()
                  if not line:
                      continue
                  try:
                      entry = json.loads(line)
                  except json.JSONDecodeError:
                      continue
  
                  msg = entry.get("message", {})
                  role = msg.get("role", "")
  
                  # Skip sidechain messages (subagents)
                  if entry.get("isSidechain"):
                      continue
  
                  if role == "user" and initial_prompt is None:
                      text = _strip_system_tags(_extract_text(msg))
                      if text:
                          initial_prompt = text[:MAX_PROMPT_LENGTH]
  
                  elif role == "assistant" and initial_prompt and first_response is None:
                      text = _extract_text(msg)
                      if text:
                          first_response = text[:MAX_RESPONSE_LENGTH]
                          break
      except (OSError, IOError):
          pass
  
      return initial_prompt, first_response
  
  
  def get_git_context(cwd: str, transcript_path: str) -> Optional[str]:
      """Get git commits made during the session timeframe.
  
      Uses the JSONL transcript's first timestamp to scope the git log to
      the actual session duration rather than an arbitrary fixed window.
      Falls back to 1 hour if the timestamp can't be extracted.
      """
      if not cwd or not Path(cwd).exists():
          return None
  
      since_arg = "1 hour ago"
      start_ts = _get_session_start_iso(transcript_path)
      if start_ts:
          since_arg = start_ts
  
      try:
          result = subprocess.run(
              ["git", "log", "--oneline", f"--since={since_arg}", "--no-merges", "-10"],
              cwd=cwd,
              capture_output=True,
              text=True,
              encoding="utf-8",
              errors="replace",
              timeout=5,
   

• hooks/ticket_branch_detector.pyGitHub ↗

All 7 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.