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/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
  
  
  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 _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") 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:
          data = json.loads(sys.stdin.read())
      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") 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,
              timeout=5,
          )
          if result.returncode == 0 and result.stdout.strip():
              return result.stdout.strip()[:300]
      except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
          pass
  
      return None
  
  
  def generate_title(
      initial_prompt: str,
      first_response: Optional[str],
      git_context: Optional[str],
  ) -> Tuple[Optional[str], str]:
      """Generate a concise session title.
  
      Priority:
      1. Git commit messages (if available) — free, no LLM needed
      2. Claude Haiku via `claude -p` with --no-session-persistence to avoid bloat
      3. Fallback: truncated initial prompt
  
      Returns (title, source) where source is 'git', 'haiku', or 'fallback'.
      """
      # 1. Use most recent git commit message as title if available
      if git_context:
          # git_context is "hash msg\nhash msg\n..." — take the first (m

• hooks/ticket_branch_detector.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  Branch-name → ticket-link detector for Claude Code Karma.
  
  Fires on SessionStart. Reads the session's cwd, asks git for the current
  branch, matches against user-configured patterns (default: Linear/Jira
  style `ABC-123`), and POSTs a link to the karma API. Silent on every
  failure — this hook NEVER blocks SessionStart.
  
  Configuration (~/.claude_karma/config.json):
  
      {
        "branch_detect_enabled": false,
        "ticket_branch_patterns": [
          {"regex": "(?P<key>[A-Z][A-Z0-9_]+-\\d+)", "provider": "linear"}
        ]
      }
  
  Opt-in: branch_detect_enabled defaults to False so users must explicitly
  flip it on to avoid surprise links on personal-projects directories.
  
  See: docs/superpowers/specs/2026-05-13-session-ticket-linking-design.md
  """
  
  from __future__ import annotations
  
  import json
  import os
  import re
  import subprocess
  import sys
  import urllib.error
  import urllib.request
  from datetime import datetime, timezone
  from pathlib import Path
  from typing import Optional
  
  API_BASE = os.environ.get("CLAUDE_KARMA_API", "http://localhost:8020")
  CONFIG_PATH = Path.home() / ".claude_karma" / "config.json"
  LIVE_SESSIONS_DIR = Path.home() / ".claude_karma" / "live-sessions"
  LOG_PATH = Path.home() / ".claude_karma" / "logs" / "ticket_branch_detector.log"
  HTTP_TIMEOUT_SEC = 3
  
  DEFAULT_CONFIG = {
      "branch_detect_enabled": False,
      "ticket_branch_patterns": [],
  }
  
  
  def _log(msg: str) -> None:
      """Append a timestamped line to the hook log. Best-effort."""
      try:
          LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
          ts = datetime.now(timezone.utc).isoformat(timespec="seconds")
          with LOG_PATH.open("a") as f:
              f.write(f"{ts} {msg}\n")
      except Exception:
          pass
  
  
  def load_config() -> dict:
      """Load ~/.claude_karma/config.json with sensible defaults on any error."""
      if not CONFIG_PATH.exists():
          return DEFAULT_CONFIG
      try:
          loaded = json.loads(CONFIG_PATH.read_text())
          if not isinstance(loaded, dict):
              return DEFAULT_CONFIG
          return {**DEFAULT_CONFIG, **loaded}
      except (OSError, json.JSONDecodeError) as e:
          _log(f"config load failed: {e!r}")
          return DEFAULT_CONFIG
  
  
  def git_current_branch(cwd: str) -> Optional[str]:
      """Return the current git branch, or None if not in a git repo / detached HEAD."""
      if not cwd:
          return None
      try:
          result = subprocess.run(
              ["git", "symbolic-ref", "--short", "HEAD"],
              cwd=cwd,
              capture_output=True,
              text=True,
              timeout=2,
              check=False,
          )
      except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
          return None
  
      if result.returncode != 0:
          return None
      branch = result.stdout.strip()
      return branch or None
  
  
  def lookup_slug_from_live_sessions(cwd: str) -> Optional[str]:
      """Best-effort slug lookup via the live-sessions tracker's state files.
  
      Picks the live-sessions entry whose `cwd` matches and whose
      `last_updated` is most recent. Returns None if nothing matches —
      the link will then be deduped by session_uuid only, which is fine
      for first-of-its-kind sessions.
      """
      if not cwd or not LIVE_SESSIONS_DIR.exists():
          return None
  
      best_slug: Optional[str] = None
      best_ts = ""
  
      try:
          for path in LIVE_SESSIONS_DIR.glob("*.json"):
              try:
                  data = json.loads(path.read_text())
              except (OSError, json.JSONDecodeError):
                  continue
              if data.get("cwd") != cwd:
                  continue
              ts = data.get("last_updated") or data.get("started_at") or ""
              if ts >= best_ts:
                  best_ts = ts
                  best_slug = data.get("slug")
      except OSError:
          return None
  
      return best_slug
  
  
  def match_pattern(branch: str, patterns: list) -> Optional[tuple[str, str]]:
      """Return (provider, ref) for the first matching pattern, else None."""
      for entry in patterns:
          if not isinstance(entry, dict):
              continue
          regex = entry.get("regex")
          provider = entry.get("provider")
          if not regex or provider not in ("linear", "jira", "github"):
              continue
          try:
              m = re.search(regex, branch)
          except re.error as e:
              _log(f"bad regex {regex!r}: {e!r}")
              continue
          if not m:
              continue
          if "key" in m.groupdict() and m.group("key"):
              ref = m.group("key")
          else:
              ref = m.group(0)
          return provider, ref
      return None
  
  
  def post_link(
      session_uuid: str,
      ref: str,
      provider: str,
      session_slug: Optional[str],
  ) -> bool:
      """POST the link to karma. Returns True on success; never raises."""
      url = f"{API_BASE}/sessions/{session_uuid}/tickets"
      body: dict = {
          "ref": ref,
          "provider": provider,
          "source": "branch",
      }
      if session_slug:
          body["session_slug"] = session_slug
  
      data = json.dumps(body).encode("utf-8")
      req = urllib.request.Request(
          url,
          data=data,
          headers={"Content-Type": "application/json"},
          method="POST",
      )
      try:
          urllib.request.urlopen(req, timeout=HTTP_TIMEOUT_SEC)
          return True
      except (urllib.error.URLError, OSError) as e:
          _log(f"POST {url} failed: {e!r}")
          return False
  
  
  def main() -> None:
      """Hook entry point. Never raises, never blocks SessionStart."""
      try:
          raw = sys.stdin.read()
          if not raw:
              return
          payload = json.loads(raw)
      except (json.JSONDecodeError, OSError):
          return
  
      session_uuid = payload.get("session_id")
      cwd = payload.get("cwd") or ""
      if not session_uuid:
          return
  
      config = load_config()
      if not config.get("branch_detect_enabled"):
          return
  
      patterns = config.get("ticket_branch_patterns") or []
      if not patte

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.