Hooks

Where it lives

• hooks/compact-recovery.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  SessionStart("compact") hook: Re-inject active chain state after compaction.
  
  Triggers after: Context compaction (manual /compact or auto-compaction).
  
  Reads chain state from server's SQLite state.db (SSOT) and outputs a
  continuation directive to stdout. Claude Code adds stdout to the
  post-compaction context, bridging the gap between compaction and the
  next UserPromptSubmit (where prompt-suggest.py would also catch it).
  
  This replaces pre-compact.py which used PreCompact — a side-effects-only
  event that cannot inject context into the conversation.
  """
  
  import json
  import sys
  from pathlib import Path
  
  # Add hooks lib to path
  sys.path.insert(0, str(Path(__file__).parent / "lib"))
  
  from db_reader import load_active_chain_state
  from session_state import ChainState, format_chain_reminder, load_session_state
  
  
  def parse_hook_input() -> dict:
      """Parse JSON input from Claude Code hook system."""
      try:
          return json.load(sys.stdin)
      except (json.JSONDecodeError, EOFError):
          return {}
  
  
  def main():
      hook_input = parse_hook_input()
      session_id = hook_input.get("session_id", "")
  
      # SSOT: read from server's state.db (works without PostToolUse hook)
      state: ChainState | None = load_active_chain_state()  # type: ignore[assignment]
  
      # Fallback: hooks-state.db (if PostToolUse populated it)
      if not state and session_id:
          state = load_session_state(session_id)
  
      if not state:
          sys.exit(0)
  
      # Only inject if there's active chain/gate/verify state
      chain_id = state.get("chain_id", "")
      step = state.get("current_step", 0)
      total = state.get("total_steps", 0)
      pending_gate = state.get("pending_gate")
      pending_verify = state.get("pending_shell_verify")
  
      has_chain = step > 0
      has_gate = pending_gate is not None
      has_verify = pending_verify is not None
  
      if not has_chain and not has_gate and not has_verify:
          sys.exit(0)
  
      # Build continuation directive matching prompt-suggest.py patterns
      reminder = format_chain_reminder(state, mode="full")
  
      if pending_gate:
          directive = f'<GATE-REVIEW>chain_id="{chain_id}" gates="{pending_gate}" → Submit gate_verdict</GATE-REVIEW>'
      elif pending_verify:
          directive = (
              f"<CALL-TOOL>\n"
              f'prompt_engine | chain_id:"{chain_id}"\n'
              f"REQUIRED: Shell verification pending. Run implementation, "
              f"then prompt_engine validates.\n"
              f"</CALL-TOOL>"
          )
      elif step > 0 and step <= total:
          directive = (
              f"<CALL-TOOL>\n"
              f'prompt_engine | chain_id:"{chain_id}"\n'
              f"REQUIRED: Continue active chain (step {step}/{total}). "
              f"Do not respond without advancing.\n"
              f"</CALL-TOOL>"
          )
      else:
          directive = ""
  
      # Output to stdout — Claude Code injects this into post-compaction context
      output_lines = [
          "## Active Chain State (recovered after compaction)",
          "",
          reminder,
      ]
      if directive:
          output_lines.append("")
          output_lines.append(directive)
  
      print("\n".join(output_lines))
      sys.exit(0)
  
  
  if __name__ == "__main__":
      main()
  

• hooks/delegation-enforce.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  PreToolUse hook: Enforce delegation when ==> operator requires sub-agent execution.
  
  Fires on Edit|Write|Bash|Task tool calls.
  
  Behavior:
  - Task while delegation pending → clear state and allow (agent delegating correctly)
  - Read-only tools while delegation pending → allow (research before delegation is fine)
  - Action tools (Edit/Write/Bash) while delegation pending → DENY (hard block)
  - No delegation pending → no-op
  """
  
  import json
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent / "lib"))
  
  from session_state import clear_delegation_state, load_session_state
  
  # Tools allowed during pending delegation (read-only + delegation itself)
  ALLOW_LIST = {"Task", "Read", "Glob", "Grep", "WebSearch", "WebFetch", "ListMcpResourcesTool"}
  
  
  def log(msg: str) -> None:
      """Print to stderr for --debug visibility."""
      print(f"[delegation-enforce] {msg}", file=sys.stderr)
  
  
  def parse_hook_input() -> dict:
      """Parse JSON input from Claude Code hook system."""
      try:
          return json.load(sys.stdin)
      except json.JSONDecodeError:
          return {}
  
  
  def main():
      hook_input = parse_hook_input()
  
      session_id = hook_input.get("session_id", "")
      if not session_id:
          sys.exit(0)
  
      tool_name = hook_input.get("tool_name", "")
  
      state = load_session_state(session_id)
      if not state or not state.get("pending_delegation"):
          sys.exit(0)
  
      agent_type = state.get("delegation_agent_type", "chain-executor")
      model_hint = state.get("delegation_model_hint")
  
      # Task tool call = agent is delegating correctly — clear state and allow
      if tool_name == "Task":
          log(f"Task tool invoked, clearing delegation state (agent_type={agent_type})")
          clear_delegation_state(session_id)
          sys.exit(0)
  
      # Read-only tools: allow silently (research before delegation is fine)
      if tool_name in ALLOW_LIST:
          sys.exit(0)
  
      # Action tools (Edit/Write/Bash) while delegation pending — hard block
      model_part = f' model="{model_hint}"' if model_hint else ""
      log(f"delegation pending, BLOCKING {tool_name} (agent_type={agent_type})")
  
      response = {
          "hookSpecificOutput": {
              "hookEventName": "PreToolUse",
              "permissionDecision": "deny",
              "permissionDecisionReason": (
                  f"Delegation pending: use Task tool "
                  f'(subagent_type="{agent_type}"{model_part}) '
                  f"before making direct edits. "
                  f"The ==> operator requires sub-agent execution."
              ),
          }
      }
      print(json.dumps(response))
      sys.exit(0)
  
  
  if __name__ == "__main__":
      main()
  

• hooks/gate-enforce.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  PreToolUse hook: Enforce gate verdicts on prompt_engine calls.
  
  Blocks:
  1. GATE_REVIEW: FAIL without retry attempt
  2. Missing gate_verdict when resuming a chain that requires it
  
  Allows Claude to self-correct before the tool executes.
  """
  
  import json
  import re
  import sys
  from pathlib import Path
  
  # Add hooks lib to path
  sys.path.insert(0, str(Path(__file__).parent / "lib"))
  
  from session_state import load_session_state
  
  
  def parse_hook_input() -> dict:
      """Parse JSON input from Claude Code hook system."""
      try:
          return json.load(sys.stdin)
      except json.JSONDecodeError:
          return {}
  
  
  def main():
      hook_input = parse_hook_input()
  
      tool_name = hook_input.get("tool_name", "")
  
      # Only process prompt_engine calls
      if "prompt_engine" not in tool_name:
          sys.exit(0)
  
      tool_input = hook_input.get("tool_input", {})
  
      # Extract parameters
      chain_id = tool_input.get("chain_id", "")
      gate_verdict = tool_input.get("gate_verdict", "")
  
      # Check 1: FAIL verdict should trigger retry guidance.
      # gate_verdict has two schema shapes: the structured object
      # {overall, rationale, per_gate[]} (preferred) and the legacy
      # "GATE_REVIEW: FAIL - reason" string.
      if isinstance(gate_verdict, dict):
          if gate_verdict.get("overall", "").upper() == "FAIL":
              reason = str(gate_verdict.get("rationale", "unspecified"))[:50]
  
              hook_response = {
                  "hookSpecificOutput": {
                      "hookEventName": "PreToolUse",
                      "permissionDecision": "deny",
                      "permissionDecisionReason": (
                          f"Gate FAIL: {reason}. Review the failing criteria, "
                          "address the gaps in your output, then resubmit your verdict."
                      ),
                  }
              }
              print(json.dumps(hook_response))
              sys.exit(0)
      elif gate_verdict:
          # Parse verdict: "GATE_REVIEW: FAIL - reason" or "GATE_REVIEW: PASS - reason"
          fail_match = re.search(r"GATE_REVIEW:\s*FAIL", gate_verdict, re.IGNORECASE)
          if fail_match:
              # Extract the reason
              reason_match = re.search(r"FAIL\s*[-:]\s*(.+)", gate_verdict, re.IGNORECASE)
              reason = reason_match.group(1).strip()[:50] if reason_match else "unspecified"
  
              hook_response = {
                  "hookSpecificOutput": {
                      "hookEventName": "PreToolUse",
                      "permissionDecision": "deny",
                      "permissionDecisionReason": (
                          f"Gate FAIL: {reason}. Review the failing criteria, "
                          "address the gaps in your output, then resubmit your verdict."
                      ),
                  }
              }
              print(json.dumps(hook_response))
              sys.exit(0)
  
      # Check 2: Resuming chain without required gate_verdict
      if chain_id and not gate_verdict:
          # Load session state to check if gate was pending
          session_id = hook_input.get("session_id", "")
          state = load_session_state(session_id) if session_id else None
  
          if state and state.get("pending_gate"):
              gate = state["pending_gate"]
              hook_response = {
                  "hookSpecificOutput": {
                      "hookEventName": "PreToolUse",
                      "permissionDecision": "deny",
                      "permissionDecisionReason": (
                          f"Gate review required: {gate}. Review your output against the gate criteria before continuing."
                      ),
                  }
              }
              print(json.dumps(hook_response))
              sys.exit(0)
  
      # All checks passed - allow tool execution
      sys.exit(0)
  
  
  if __name__ == "__main__":
      main()
  

• hooks/post-prompt-engine.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  PostToolUse hook: Track chain/gate state from prompt_engine responses.
  
  Triggers after: mcp__claude-prompts__prompt_engine
  
  Parses the response to:
  1. Track current chain step
  2. Detect pending gates
  3. Inject reminders for gate reviews
  """
  
  import json
  import sys
  from pathlib import Path
  
  # Add hooks lib to path
  sys.path.insert(0, str(Path(__file__).parent / "lib"))
  
  from session_state import (
      parse_prompt_engine_response,
      save_session_state,
  )
  
  
  def parse_hook_input() -> dict:
      """Parse JSON input from Claude Code hook system."""
      try:
          return json.load(sys.stdin)
      except json.JSONDecodeError:
          return {}
  
  
  def main():
      hook_input = parse_hook_input()
  
      tool_name = hook_input.get("tool_name", "")
      session_id = hook_input.get("session_id", "")
  
      # Only process prompt_engine calls
      if "prompt_engine" not in tool_name:
          sys.exit(0)
  
      tool_response = hook_input.get("tool_response", {})
      tool_input = hook_input.get("tool_input", {})
  
      # Parse response for chain/gate state
      if isinstance(tool_response, dict):
          content = tool_response.get("content", "")
          # Handle array of content blocks
          if isinstance(content, list):
              content = " ".join(block.get("text", "") if isinstance(block, dict) else str(block) for block in content)
      else:
          content = str(tool_response)
  
      state = parse_prompt_engine_response(content)
  
      if not state:
          sys.exit(0)
  
      # Extract chain_id from tool_input (higher priority than regex parsing)
      if isinstance(tool_input, dict):
          input_chain_id = tool_input.get("chain_id", "")
          if input_chain_id:
              state["chain_id"] = input_chain_id
  
      # Save state for this session
      save_session_state(session_id, state)
  
      # Detect delegation: command contains ==> and chain has remaining steps
      command = tool_input.get("command", "") if isinstance(tool_input, dict) else ""
      chain_id = state.get("chain_id", "")
      pending_gate = state.get("pending_gate")
      step = state.get("current_step", 0)
      total = state.get("total_steps", 0)
  
      if "==>" in command and step > 0 and step < total and not pending_gate:
          state["pending_delegation"] = True
          save_session_state(session_id, state)
  
      if pending_gate:
          # CLAUDE DIRECTIVE ONLY: Guide Claude to submit verdict (token-efficient)
          # User sees server's "Gate Review Required" message in tool response
          directive = f'<GATE-REVIEW>chain_id="{chain_id}" gates="{pending_gate}" → Submit gate_verdict</GATE-REVIEW>'
  
          hook_response = {"hookSpecificOutput": {"hookEventName": "PostToolUse", "additionalContext": directive}}
          print(json.dumps(hook_response))
          sys.exit(0)
  
      # Imperative directive: force Claude to continue chain
      if step > 0 and total > 0 and step < total:
          directive = (
              f"<CALL-TOOL>\n"
              f'prompt_engine | chain_id:"{chain_id}"\n'
              f"REQUIRED: Continue active chain (step {step}/{total}). "
              f"Do not respond without advancing.\n"
              f"</CALL-TOOL>"
          )
          hook_response = {"hookSpecificOutput": {"hookEventName": "PostToolUse", "additionalContext": directive}}
          print(json.dumps(hook_response))
          sys.exit(0)
  
      sys.exit(0)  # No output needed
  
  
  if __name__ == "__main__":
      main()
  

• hooks/prompt-suggest.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  UserPromptSubmit hook: Directive injection for claude-prompts.
  
  Detects >>syntax and operators, outputs:
  - systemMessage: Compact user confirmation
  - additionalContext: Structured directive for Claude
  
  Architecture:
  - Layer 1: MCP-native prompts (via registerPrompt) - standard protocol
  - Layer 2: This hook - efficiency layer translating >>syntax to prompt_engine calls
  
  Operators detected (from SSOT registry):
  - `-->` chain, `==>` delegation, `::` gate, `@` framework, `#` style, `* N` repetition
  """
  
  import json
  import re
  import sys
  from pathlib import Path
  
  # Add hooks lib to path
  sys.path.insert(0, str(Path(__file__).parent / "lib"))
  
  from cache_manager import (
      ArgumentInfo,
      PromptInfo,
      fuzzy_match_prompt_id,
      get_chains_only,
      get_prompt_by_id,
      get_valid_frameworks,
      get_valid_styles,
      load_prompts_cache,
      match_prompts_to_intent,
  )
  from config_loader import is_expanded_output
  from db_reader import load_active_chain_state
  from session_state import ChainState, format_chain_reminder, load_session_state
  
  # Import generated operator patterns (SSOT: server/tooling/contracts/operators.json)
  try:
      from operators import OPERATORS, detect_operator, get_delimiter_symbols
  
      HAS_GENERATED_OPERATORS = True
  except ImportError:
      HAS_GENERATED_OPERATORS = False
      OPERATORS = {}
  
      def get_delimiter_symbols() -> list[str]:
          return ["-->", "==>"]
  
  
  def format_arguments(prompt_id: str) -> dict[str, str]:
      """
      Extract argument info from prompt metadata.
  
      Returns dict of arg_name -> "type (required|optional)"
      If options are available, shows: "opt1 | opt2 | opt3 (required)"
      """
      prompt = get_prompt_by_id(prompt_id)
      if not prompt:
          return {}
      args = prompt.get("arguments", [])
      result: dict[str, str] = {}
      for arg in args:
          if not isinstance(arg, dict) or not arg.get("name"):
              continue
          name = arg.get("name", "")
          options = arg.get("options")
          if options and isinstance(options, list) and len(options) > 0:
              # Show options inline: "tutorial | howto | reference"
              type_str = " | ".join(options)
          else:
              type_str = arg.get("type", "string")
          req = "required" if arg.get("required") else "optional"
          result[name] = f"{type_str} ({req})"
      return result
  
  
  def parse_hook_input() -> dict:
      """Parse JSON input from Claude Code hook system."""
      try:
          return json.load(sys.stdin)
      except json.JSONDecodeError:
          return {}
  
  
  def detect_prompt_invocation(message: str) -> str | None:
      """
      Detect >> prompt invocation syntax.
      Returns the prompt ID/name if found.
  
      Examples:
          >>deep_analysis -> "deep_analysis"
          >> code_review -> "code_review"
          >>research-comprehensive -> "research-comprehensive"
          @CAGEERF >>analyze -> "analyze"
          #analytical >>report -> "report"
      """
      # Try exact start first
      match = re.match(r"^>>\s*([a-zA-Z0-9_-]+)", message.strip())
      if match:
          # Normalize to lowercase for case-insensitive matching (aligns with MCP server)
          return match.group(1).lower()
  
      # Also check for >> after operators (@framework, #style)
      match = re.search(r">>\s*([a-zA-Z0-9_-]+)", message)
      if match:
          # Normalize to lowercase for case-insensitive matching (aligns with MCP server)
          return match.group(1).lower()
  
      return None
  
  
  def detect_explicit_request(message: str) -> bool:
      """Detect explicit prompt suggestion requests."""
      triggers = [
          r"\bsuggest\s+prompts?\b",
          r"\blist\s+prompts?\b",
          r"\bavailable\s+prompts?\b",
          r"\bshow\s+prompts?\b",
          r"\bwhat\s+prompts?\b",
          r"\bprompt\s+suggestions?\b",
          r"\brecommend\s+prompts?\b",
      ]
      message_lower = message.lower()
      return any(re.search(trigger, message_lower) for trigger in triggers)
  
  
  def detect_chain_syntax(message: str) -> list[str]:
      """
      Detect chain syntax (-->, ==>) in message.
      Returns list of prompt IDs in chain order (normalized to lowercase).
  
      Splits on delimiter operators from SSOT registry (plus → unicode alias),
      then extracts >>prompt_id from each segment. Handles arguments between
      prompt ID and delimiter correctly.
  
      Example: >>analyze --> >>implement --> >>test
      Example with args: >>analyze scope:"backend" --> >>implement
      Example with delegation: >>step1 ==> >>step2
      """
      # Build split pattern from SSOT delimiter symbols
      delimiters = get_delimiter_symbols()
      escaped = [re.escape(d) for d in delimiters] + ["→"]
      split_pattern = r"\s*(?:" + "|".join(escaped) + r")\s*"
  
      parts = re.split(split_pattern, message)
      if len(parts) <= 1:
          return []
  
      # Extract >>prompt_id from each segment (ignores arguments safely)
      prompts = []
      for part in parts:
          match = re.search(r">>\s*([a-zA-Z0-9_-]+)", part)
          if match:
              prompts.append(match.group(1).lower())
  
      return prompts
  
  
  def detect_inline_gates(message: str) -> list[str]:
      """
      Detect :: gate syntax in message.
      Returns list of gate criteria/IDs.
  
      Examples:
          :: 'must check security' -> ["must check security"]
          :: security-check -> ["security-check"]
  
      Note: Always uses semantic extraction (not generated patterns) because
      we need gate content, not the :: symbol itself.
      """
      # Always use semantic patterns - generated pattern returns operator symbol too
      quoted_pattern = r'::\s*[\'"]([^\'"]+)[\'"]'
      id_pattern = r"::\s*([a-zA-Z][a-zA-Z0-9_-]*)\b"
  
      quoted = re.findall(quoted_pattern, message)
      ids = re.findall(id_pattern, message)
  
      return quoted + ids
  
  
  def detect_framework(message: str) -> str | None:
      """
      Detect @FRAMEWORK syntax in message.
      Returns the framework ID if found (normalized to lowercase).
  
      Examples:
          @CAGEERF >>analyze -> "cageerf"
          @ReACT >>debu

• hooks/ralph-context-tracker.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  PostToolUse hook: Track file changes and tool usage for Ralph loops.
  
  Triggers after: Edit, Write, Bash (during active Ralph sessions)
  
  Records:
  1. File modifications (Edit/Write tools)
  2. Command executions (Bash tool)
  3. Extracts lessons from Claude's reasoning
  
  This data feeds the session story for context-isolated Ralph instances.
  """
  
  import json
  import sys
  from pathlib import Path
  
  # Add hooks lib to path
  sys.path.insert(0, str(Path(__file__).parent / "lib"))
  
  from lesson_extractor import summarize_error
  from session_tracker import get_session_tracker
  from verify_active_store import load_verify_active_state
  
  
  def parse_hook_input() -> dict:
      """Parse JSON input from Claude Code hook system."""
      try:
          return json.load(sys.stdin)
      except json.JSONDecodeError:
          return {}
  
  
  def get_active_ralph_session(hook_session_id: str | None = None) -> str | None:
      """
      Get the currently active Ralph session ID.
  
      Checks verify-state.db (source of truth for Ralph sessions).
      When hook_session_id is provided, scopes lookup to that client's state.
      """
      state = load_verify_active_state(hook_session_id)
      if not state:
          return None
      return state.get("sessionId")
  
  
  def extract_file_change_details(tool_input: dict, tool_name: str) -> dict | None:
      """Extract file change details from Edit/Write tool input."""
      if "Edit" in tool_name:
          return {
              "file": tool_input.get("file_path", "unknown"),
              "type": "modify",
              "details": f"Edit: {tool_input.get('old_string', '')[:50]}... → {tool_input.get('new_string', '')[:50]}...",
          }
      elif "Write" in tool_name:
          return {
              "file": tool_input.get("file_path", "unknown"),
              "type": "add",
              "details": f"Write: {len(tool_input.get('content', ''))} chars",
          }
      return None
  
  
  def extract_bash_details(tool_input: dict, tool_response: str) -> dict | None:
      """Extract command execution details from Bash tool."""
      command = tool_input.get("command", "")
      if not command:
          return None
  
      # Truncate long commands
      cmd_summary = command[:100] + "..." if len(command) > 100 else command
  
      # Check if it's a verification command (test, lint, build, etc.)
      verification_indicators = ["test", "npm run", "yarn", "pytest", "cargo test", "go test", "make"]
      is_verification = any(ind in command.lower() for ind in verification_indicators)
  
      return {
          "command": cmd_summary,
          "is_verification": is_verification,
          "output_summary": summarize_error(tool_response) if tool_response else None,
      }
  
  
  def main():
      hook_input = parse_hook_input()
  
      tool_name = hook_input.get("tool_name", "")
      session_id = hook_input.get("session_id", "")
  
      # Only track Edit, Write, and Bash tools
      tracked_tools = ["Edit", "Write", "Bash"]
      if not any(t in tool_name for t in tracked_tools):
          sys.exit(0)
  
      # Only track during active Ralph sessions (scoped to this client)
      ralph_session = get_active_ralph_session(session_id or None)
      if not ralph_session:
          # No active Ralph session, no tracking needed
          sys.exit(0)
  
      tool_input = hook_input.get("tool_input", {})
      tool_response = hook_input.get("tool_response", "")
  
      # Convert response to string if needed
      if isinstance(tool_response, dict):
          content = tool_response.get("content", "")
          if isinstance(content, list):
              tool_response = " ".join(
                  block.get("text", "") if isinstance(block, dict) else str(block) for block in content
              )
          else:
              tool_response = str(content)
      else:
          tool_response = str(tool_response)
  
      # Get session tracker
      tracker = get_session_tracker(ralph_session)
  
      # Track file changes
      if "Edit" in tool_name or "Write" in tool_name:
          change = extract_file_change_details(tool_input, tool_name)
          if change:
              tracker.record_file_change(file_path=change["file"], change_type=change["type"], details=change["details"])
  
      # Track Bash commands (for context about what was run)
      if "Bash" in tool_name:
          bash_details = extract_bash_details(tool_input, tool_response)
          if bash_details and bash_details["is_verification"]:
              # This is a verification command - its output will be captured
              # by ralph-stop.py when the verification completes
              pass
  
      # No output needed - silent tracking
      sys.exit(0)
  
  
  if __name__ == "__main__":
      main()
  

• hooks/ralph-stop.pyRunsGitHub ↗
• hooks/subagent-gate-enforce.pyGitHub ↗

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