Hooks

Where it lives

• hooks/compact-recovery.pyGitHub ↗
  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_recoverable_chain_state
  from session_state import ChainState, format_chain_reminder
  
  
  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", "")
  
      # Session-scoped recovery: only the chain THIS conversation recorded via
      # PostToolUse tracking is eligible; state.db refreshes that chain and can
      # never substitute another client's newer chain (cross-client leakage fix).
      state: ChainState | None = load_recoverable_chain_state(session_id)  # type: ignore[assignment]
  
      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.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  PreToolUse hook: Enforce delegation when ==> operator requires sub-agent execution.
  
  Fires on Edit|Write|Bash|Task|Agent tool calls.
  
  Behavior:
  - Task/Agent while delegation pending → clear state and allow (agent delegating correctly)
  - Read-only + task-tracking tools while delegation pending → allow (research and
    Task* tracking calls before delegation are 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).
  # "Agent" is this Claude Code build's reported subagent-invocation tool name
  # (extension-alignment drift vs "Task"); TaskCreate/TaskUpdate/TaskGet/TaskList/
  # TaskOutput/TaskStop are task-tracking calls, not action tools, and would
  # otherwise be caught by the unanchored hooks.json matcher.
  ALLOW_LIST = {
      "Task",
      "Agent",
      "Read",
      "Glob",
      "Grep",
      "WebSearch",
      "WebFetch",
      "ListMcpResourcesTool",
      "TaskCreate",
      "TaskUpdate",
      "TaskGet",
      "TaskList",
      "TaskOutput",
      "TaskStop",
  }
  
  
  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)
  
      # Default mirrors CLAUDE_CODE_DEFAULT_AGENT_TYPE in the server's delegation strategy.
      agent_type = state.get("delegation_agent_type", "general-purpose")
      model_hint = state.get("delegation_model_hint")
  
      # Task/Agent tool call = agent is delegating correctly — clear state and allow.
      # "Agent" is this client's reported name for subagent invocation; "Task"
      # covers other clients/older builds.
      if tool_name in {"Task", "Agent"}:
          log(f"{tool_name} 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.pyGitHub ↗
  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. Chain resumes carrying NO pending-run resolution parameter while the run is HOLDING.
     Two holds reach this check: an ordinary gate review, and the reserved
     `__unknown_interrupt__` synthetic review a blocking unknown raises under
     `budget.pauseOnBlocking`. They are denied with different prose because they accept
     different exits — no gate_verdict clears an interrupt — but by ONE rule.
     The accepted resolution parameters (gate_verdict, gate_action, cancel, ...) are generated
     from the server contract into lib/_generated/resolution_verbs.py — never hardcoded
     here, so the hook cannot deny an exit the server supports.
  
  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 (
      UNKNOWN_INTERRUPT_LABEL,
      interrupt_exits,
      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 load_resolution_params() -> frozenset[str] | None:
      """Load the pending-run resolution verbs generated from the server contract.
  
      The set is emitted by server/scripts/generate-contracts.ts from parameters flagged
      `resolvesPendingRun` in tooling/contracts/prompt-engine.json, so this hook accepts
      exactly the moves the server accepts. A hardcoded model here rotted twice — it denied
      `gate_action: "abort"` and `cancel: true`, trapping sessions behind their own pending
      gate (2026-08-20).
  
      Returns None when the artifact is missing or unreadable: the caller fails open,
      because the server enforces gates authoritatively and a broken hook must never
      re-create the trap.
      """
      try:
          from _generated.resolution_verbs import PENDING_RUN_RESOLUTION_PARAMS
  
          return PENDING_RUN_RESOLUTION_PARAMS
      except Exception:
          return None
  
  
  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 any resolution parameter while a gate is pending.
      # Any contract-flagged resolution verb (gate_verdict, gate_action, cancel, ...) passes:
      # they are all server-supported responses to a pending gate, and denying them is how a
      # gate blocks its own abort.
      if chain_id:
          resolution_params = load_resolution_params()
          if resolution_params is None:
              # Fail open: without the generated verb set this hook cannot tell a valid
              # resolution from a bare resume — the server still enforces the gate.
              sys.exit(0)
  
          if not any(tool_input.get(name) for name in resolution_params):
              # 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"]
                  if gate == UNKNOWN_INTERRUPT_LABEL:
                      # A blocking-unknown interrupt is NOT a gate review: the run issued no step
                      # and no gate_verdict clears it, so a message telling the caller to submit
                      # one would name an exit the server refuses. The verbs come from the run's
                      # own interrupt section (session_state captures what the server printed).
                      reason = (
                          f"Chain paused: {gate}. The run issued no step and no gate_verdict "
                          "clears this hold. Resolve it with one of: "
   

• hooks/post-prompt-engine.pyGitHub ↗
  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 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 (
      clear_session_state,
      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:
          # A verdict/cancel submission whose response carries no further chain
          # markers means the tracked state RESOLVED (e.g. gate PASS → "Execution
          # complete."). Clear the row: sessions previously kept their pending
          # state for the 24h retention window, and recovery hooks re-injected
          # long-resolved gates after compaction.
          if isinstance(tool_input, dict) and (tool_input.get("gate_verdict") or tool_input.get("cancel")):
              clear_session_state(session_id)
          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
  
      # Terminal boundary: an explicit completion marker with nothing pending
      # means the run is over — clear the row instead of saving a snapshot that
      # only the 24h sweep would ever remove. The step numbers alone cannot
      # decide this: "Step 2 of 2" (final step delivered, still in flight) and
      # "Chain complete (2/2)" both parse to 2/2, so the marker text is the
      # discriminator.
      run_is_complete = bool(re.search(r"[Cc]hain complete|Execution complete", content))
      if run_is_complete and not state.get("pending_gate") and not state.get("pending_shell_verify"):
          clear_session_state(session_id)
          sys.exit(0)
  
      # Save state for this session
      save_session_state(session_id, state)
  
      # Detect delegation: RESPONSE (not command prose) contains a delegation CTA
      # and the chain has remaining steps. The server only renders these markers
      # (strategy.ts formatToolCall/getHandoffFooterInstruction) when the next step
      # is actually delegated — a command that merely mentions "==>" in prose must
      # not arm enforcement.
      chain_id = state.get("chain_id", "")
      pending_gate = state.get("pending_gate")
      step = state.get("current_step", 0)
      total = state.get("total_steps", 0)
  
      subagent_match = re.search(r'subagent_type:\s*"([^"]+)"', content)
      has_delegation_cta = bool(subagent_match) or "Handoff via Task tool" in content
  
      if has_delegation_cta and step > 0 and step < total and not pending_gate:
          state["pending_delegation"] = True
          # Fallback mirrors CLAUDE_CODE_DEFAULT_AGENT_TYPE (server/src/engine/execution/delegation/
          # strategy.ts). The Claude strategy always emits subagent_type, so this only fires for a
          # CTA shape the regex above does not know.
          state["delegation_agent_type"] = subagent_match.group(1) if subagent_match else "general-purpose"
          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.pyGitHub ↗
  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 os
  import re
  import sys
  from pathlib import Path
  from typing import cast
  
  # 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_recoverable_chain_state
  from session_state import ChainState, format_chain_reminder
  
  # 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"
    

• hooks/python-hook-runner.cjsRunsGitHub ↗
  Read the script

  #!/usr/bin/env node
  
  const assert = require("node:assert/strict");
  const { statSync } = require("node:fs");
  const { spawnSync } = require("node:child_process");
  const path = require("node:path");
  
  function interpreterCandidates(platform) {
    return platform === "win32"
      ? [
          { command: "py", prefixArgs: ["-3"] },
          { command: "python3", prefixArgs: [] },
          { command: "python", prefixArgs: [] },
        ]
      : [
          { command: "python3", prefixArgs: [] },
          { command: "python", prefixArgs: [] },
        ];
  }
  
  function isPythonFile(filePath) {
    try {
      return path.extname(filePath) === ".py" && statSync(filePath).isFile();
    } catch {
      return false;
    }
  }
  
  function runPythonHook({
    script,
    args = [],
    platform = process.platform,
    spawn = spawnSync,
    fileExists = isPythonFile,
  }) {
    if (!script || !fileExists(script)) {
      return {
        exitCode: 2,
        message: `Python hook script is missing or invalid: ${script || "<missing>"}`,
      };
    }
  
    for (const candidate of interpreterCandidates(platform)) {
      const result = spawn(
        candidate.command,
        [...candidate.prefixArgs, script, ...args],
        {
          shell: false,
          stdio: "inherit",
          windowsHide: true,
        },
      );
  
      if (result.error?.code === "ENOENT") continue;
      if (result.error) {
        return {
          exitCode: 1,
          message: `Failed to launch ${candidate.command}: ${result.error.message}`,
        };
      }
      if (typeof result.status === "number") return { exitCode: result.status };
      return {
        exitCode: 1,
        message: result.signal
          ? `Python hook terminated by signal ${result.signal}`
          : "Python hook exited without a status",
      };
    }
  
    return {
      exitCode: 127,
      message:
        "No Python 3 interpreter found. Install Python 3 with the Windows py launcher or provide python3/python on PATH.",
    };
  }
  
  function selfTest() {
    assert.deepEqual(interpreterCandidates("win32"), [
      { command: "py", prefixArgs: ["-3"] },
      { command: "python3", prefixArgs: [] },
      { command: "python", prefixArgs: [] },
    ]);
    assert.deepEqual(interpreterCandidates("linux"), [
      { command: "python3", prefixArgs: [] },
      { command: "python", prefixArgs: [] },
    ]);
  
    const attempted = [];
    const fallback = runPythonHook({
      script: "/fixture/hook.py",
      platform: "win32",
      fileExists: () => true,
      spawn: (command, args) => {
        attempted.push([command, args]);
        return command === "py"
          ? { error: Object.assign(new Error("missing"), { code: "ENOENT" }) }
          : { status: 0 };
      },
    });
    assert.equal(fallback.exitCode, 0);
    assert.deepEqual(
      attempted.map(([command]) => command),
      ["py", "python3"],
    );
    assert.deepEqual(attempted[0][1], ["-3", "/fixture/hook.py"]);
  
    let failureAttempts = 0;
    const scriptFailure = runPythonHook({
      script: "/fixture/hook.py",
      platform: "win32",
      fileExists: () => true,
      spawn: () => {
        failureAttempts += 1;
        return { status: 7 };
      },
    });
    assert.equal(scriptFailure.exitCode, 7);
    assert.equal(failureAttempts, 1);
  
    assert.equal(runPythonHook({ script: "missing.py" }).exitCode, 2);
    assert.equal(
      runPythonHook({
        script: "/fixture/hook.py",
        fileExists: () => true,
        spawn: () => ({
          error: Object.assign(new Error("missing"), { code: "ENOENT" }),
        }),
      }).exitCode,
      127,
    );
  
    process.stdout.write("python-hook-runner self-test — 7/7 checks passed\n");
  }
  
  function main() {
    if (process.argv[2] === "--self-test") {
      selfTest();
      return;
    }
  
    const result = runPythonHook({
      script: process.argv[2],
      args: process.argv.slice(3),
    });
    if (result.message) process.stderr.write(`${result.message}\n`);
    process.exitCode = result.exitCode;
  }
  
  if (require.main === module) main();
  

• hooks/ralph-context-tracker.pyGitHub ↗
• hooks/ralph-stop.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.