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, format_pause_directive
  
  
  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)
  
      # A paused run is restored as paused: the concise directive, never a continue instruction.
      paused = format_pause_directive(state)
      if paused:
          print(f"## Paused Workflow (recovered after compaction)\n\n{paused}")
          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, pinned to the foreground → clear state
    and allow (agent delegating correctly)
  - Task/Agent while a DETACHED step (`await: run`) is pending, pinned to the
    background (`run_in_background: true`, what the server rendered) → clear
    and allow; not backgrounded → DENY naming the detached step
  - Task/Agent while delegation pending, NOT pinned to the foreground → DENY
    (a backgrounded spawn cannot report the `HANDOFF RESULT` trailer before the
    delegating agent continues); state is left pending, unchanged
  - 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
  - Any call from inside a subagent (payload carries `agent_id`) → no-op: it
    arrives under the parent's session_id but is not the delegating session
  - An owned exact pause of the delegated run → no demand and state untouched, so
    `pending_delegation` and the report identity survive and the demand returns on
    resume; a retained, unconfirmed or foreign pause suspends nothing.
  
  The server owns the floor: it renders `run_in_background: false` in the
  handoff instructions and refuses to resume a delegated node whose reply
  lacks the worker's `HANDOFF RESULT` trailer. This hook can only TIGHTEN on
  top of that floor for Claude Code — it denies a spawn call that was not
  actually pinned to the foreground before the server ever sees a resume.
  """
  
  import json
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent / "lib"))
  
  from session_state import (
      ChainState,
      clear_delegation_state,
      exact_run_identity,
      holds_pause_exemption,
      is_subagent_payload,
      load_session_state,
  )
  from spawn_pin import spawn_call_is_pinned
  
  # 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 paused_for_conversation(state: ChainState) -> bool:
      """Whether this conversation owns an exact pause of the run its pending delegation belongs to.
  
      The host binding is the tracked run itself: a recorded identity naming another chain id is a
      foreign pause, and `holds_pause_exemption` refuses a retained or unconfirmed one.
      """
      identity = exact_run_identity(state.get("run_identity"))
      if identity is None or identity["chain_id"] != state.get("chain_id"):
          return False
      return holds_pause_exemption(state, identity)
  
  
  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(client_id: str = "claude-code") -> None:
      """Enforce delegation using the caller's existing spawn-pin contract."""
      hook_input = parse_hook_input()
  
      session_id = hook_input.get("session_id", "")
      if not session_id:
          sys.exit(0)
  
      # A subagent's call carries its parent's session_id but is not the session that
      # rendered the brief: allow it and leave the parent's state alone — its own
      # Agent call must not clear the parent's pending delegation either.
      if is_subagent_payload(hook_input):
          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)
  
      # Paused for discussion: no demand, and no clear either — a spawn now is not the delegated
      # step, so the worker's report is still owed once the run resumes.
      if paused_for_conversation(state):
          log(f"workflow paused for conversation, delegation demand suspended for {tool_name}")
          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 — but Claude Code spawns
      # subagents in the BACKGROUND by default, and a backgrounded worker
      # cannot supply the server's required `HANDOFF RESULT` trailer before
      # this agent continues. Clear state and allow only when the call is
      # pinned to the foreground; otherwise deny and leave state pending.
      # "Agent" is this client's reported name for subagent invocation; "Task"
      # covers other clients/older builds.
      if tool_name in {"Task", "Agent"}:
          tool_input = hook_input.get("tool_input", {}) or {}
          mode = state.get("delegation_mode", "blocking")
          pinned, pin_reason = spawn_call_is_pinned(client_id, tool_input, mode)
          if not pinned and mode == "detached":
              log(f"{tool_name} tool invoked for a detached step but not backgrounded, BLOCKING ({pin_reason})")
              response = {
                  "hookSpecificOutput": {
                      "hookEventName": "PreToolUse",
                      "permissionDecision": "deny",
                      "permissionDecisionReason": (
                          f"This step is detached (await: run): {pin_reason} — the chain does not "
                          "wait fo

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

  #!/usr/bin/env python3
  """
  PreToolUse hook: Validate verdict parameter shape and pending-run resolution.
  
  Blocks:
  1. Non-object gate_verdict or malformed overall/rationale fields.
  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) 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.
  3. Workflow mutations on a run this conversation holds paused (owned, exact identity): denied
     with resume-or-abort. Controls (gate_action pause/resume/abort by name, handoff, claim_token)
     always reach the server, which decides them; retry/skip/accept_alternative are mutations.
  
  Allows Claude to self-correct before the tool executes.
  """
  
  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 (
      UNKNOWN_INTERRUPT_LABEL,
      ChainState,
      exact_run_identity,
      format_pause_directive,
      holds_pause_exemption,
      interrupt_exits,
      is_subagent_payload,
      load_session_state,
  )
  
  # The gate_action values that control a run rather than work it (contract enum, prompt-engine.json).
  RUN_CONTROLS = frozenset({"pause", "resume", "abort"})
  
  
  def is_run_control(tool_input: dict) -> bool:
      """A pause/resume/abort or custody call: the server decides it, so no hold may block it."""
      claim = tool_input.get("claim_token")
      return tool_input.get("gate_action") in RUN_CONTROLS or tool_input.get("handoff") is True or isinstance(claim, str)
  
  
  def paused_instruction(state: ChainState | None, chain_id: str) -> str | None:
      """The resume-or-abort instruction when this conversation owns an exact pause of `chain_id`."""
      identity = exact_run_identity(state.get("run_identity")) if state else None
      if not state or identity is None or identity["chain_id"] != chain_id or not holds_pause_exemption(state, identity):
          return None
      return format_pause_directive(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"`, trapping sessions behind their own pending gate.
  
      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")
  
      # Shape admission only: an honest FAIL must reach the server's authoritative judgment.
      if "gate_verdict" in tool_input and (
          not isinstance(gate_verdict, dict)
          or gate_verdict.get("overall") not in ("PASS", "FAIL")
          or not isinstance(gate_verdict.get("rationale"), str)
          or not gate_verdict["rationale"].strip()
      ):
          hook_response = {
              "hookSpecificOutput": {
                  "hookEventName": "PreToolUse",
                  "permissionDecision": "deny",
                  "permissionDecisionReason": (
                      'gate_verdict requires an object with overall "PASS" or "FAIL" and a '
                      "non-empty rationale. Follow the returned server template for per_gate, "
                      "reminders and semantic evaluation."
                  ),
              }
          }
          print(json.dumps(hook_response))
          sys.exit(0)
  
      # Check 2 reads the pending gate by session_id, and a subagent's call arrives under its
      # PARENT's: a worker resuming its own run is not held by the parent's review. The server
      # still gates the worker's run. Shape admission reads only the caller's own verdict, so it
      # applies to a worker as it does to its parent.
      if is_subagent_payload(hook_input):
          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) passes:
      # they are all server-supported responses to a pending gate, and denying them is how a
      # gate blocks its own abort.
      session_id = hook_input.get("session_id", "")
      state = load_session_state(session_id) if chain_id and session_id else None
      paused = None if is_run_control(tool_input) else paused_instruction(state, chain_id)
      if paused:
          # Check 3 precedes check 2: while paused, every hold waits behind the explicit resume.
          output = {"hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": paused}
          print(json.dumps({"hookSpecificOutput": output}))
          sys.exit(0)
  
      # Check 2 admits pause/resume/abort through the generated

• 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
  4. Track a conversation pause from an applied `conversation_control` packet only, read from the
     structured content in whichever shape the host sends (`read_tool_response`); reply text (and the
     structured `message`, which carries the recovery token) is never stored.
  5. Record the exact run behind every tracked chain step (`run_identity`), so Stop can find that
     run's verification loop: kept across the same chain's steps, otherwise resolved from the one live
     `chain_sessions` row for the chain id, and absent when that is ambiguous.
  6. Clear the tracked run on an abort whose `run_control` outcome is `cancelled`, and on a completion
     marker in the reply text. Every read of `tool_response` goes through `read_tool_response`: under
     Claude Code no MCP payload carries `isError`, so the absence of an error is the success signal.
  """
  
  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 db_reader import resolve_run_identity
  from session_state import (
      UNKNOWN_INTERRUPT_LABEL,
      ChainState,
      RunIdentity,
      clear_session_state,
      exact_run_identity,
      format_gate_directive,
      is_subagent_payload,
      load_session_state,
      parse_prompt_engine_response,
      pause_fields,
      read_tool_response,
      save_session_state,
  )
  
  COMPLETION_MARKER = re.compile(r"(?m)^(?:[✓✅][ \t]*)?(?:Chain complete\b|Execution complete\.)")
  # The outcomes a packet carries only when the control applied (conversation-control-response.ts);
  # `refused` -- busy, paused-run refusals, wrong scope -- never applies, since no MCP payload under
  # Claude Code carries `isError` to say so.
  PAUSE_OUTCOMES = frozenset({"paused", "already-paused", "claimed"})
  APPLIED_OUTCOMES = PAUSE_OUTCOMES | {"resumed"}
  # What a resumed run's owed obligation restores when the tracker lost the hold (e.g. a claim).
  OBLIGATION_HOLDS = {"blocking-unknown": UNKNOWN_INTERRUPT_LABEL, "review": "review", "exhausted-retry": "review"}
  
  
  def _control_packet(structured: dict | None) -> dict | None:
      """The structured content's `conversation_control`, or None when it carries none."""
      packet = structured.get("conversation_control") if structured is not None else None
      return packet if isinstance(packet, dict) and packet.get("kind") == "conversation_control" else None
  
  
  def _packet_identity(run: object) -> RunIdentity | None:
      """The packet run's exact identity in hook-state keys, or None when any field is missing."""
      run = run if isinstance(run, dict) else {}
      scope = run.get("scope") if isinstance(run.get("scope"), dict) else {}
      raw = {"run_id": run.get("runId"), "chain_id": run.get("chainId"), "organization_id": scope.get("organizationId")}
      raw |= {"workspace_id": scope.get("workspaceId"), "continuity_scope_id": scope.get("continuityScopeId")}
      return exact_run_identity(raw)
  
  
  def _apply_control_packet(session_id: str, request_id: str, packet: dict) -> None:
      """Record a pause, claim or resume of this conversation's exact run; anything else is a no-op."""
      run = packet.get("run")
      identity = _packet_identity(run)
      outcome = packet.get("outcome")
      if not isinstance(run, dict) or identity is None or outcome not in APPLIED_OUTCOMES:
          return
      if request_id and request_id != identity["chain_id"]:
          return
      tracked = load_session_state(session_id) or {}
      same_run = tracked.get("chain_id") == identity["chain_id"]
      if same_run and exact_run_identity(tracked.get("run_identity")) not in (None, identity):
          return  # same chain id, another run or scope
      # The same run keeps every hold detail (unknown verbs, delegation, shell verification).
      state: ChainState = {**tracked} if same_run else {"pending_gate": None}
      if outcome == "resumed":
          for key in ("conversation_pause", "owed", "pause_custody"):
              state.pop(key, None)
          owed = [item for item in run.get("obligations") or [] if isinstance(item, dict)]
          holds = [OBLIGATION_HOLDS[item["kind"]] for item in owed if item.get("kind") in OBLIGATION_HOLDS]
          if holds and not state.get("pending_gate"):
              state["pending_gate"] = holds[0]
      else:
          fields = pause_fields(run.get("pause"), run.get("obligations"))
          if not fields:
              return
          state.update(fields)
          state["pause_custody"] = "owned"
      step = run.get("step") if isinstance(run.get("step"), dict) else {}
      state.update({"chain_id": identity["chain_id"], "run_identity": identity})
      state["current_step"], state["total_steps"] = int(step.get("current") or 0), int(step.get("total") or 0)
      save_session_state(session_id, state)
  
  
  def _step_identity(chain_id: str, tracked: ChainState | None) -> RunIdentity | None:
      """The exact run behind an ordinary step of `chain_id`.
  
      The identity already tracked for this chain (a control packet or an earlier step recorded it)
      is kept; a step of another chain drops it, and resolves its own from the projection.
      """
      kept = exact_run_identity((tracked or {}).get("run_identity"))
      if kept is not None and kept["chain_id"] == chain_id and (tracked or {}).get("chain_id") == chain_id:
          return kept
      return resolve_run_identity(chain_id) if chain_id else None
  
  
  def _abort_cancelled_tracked_run(session_id: str, request_id: str, structured: dict | None) -> bool:
      """Whether an abort reply's `run_control` says this conversation's tracked run is cancelled.
  
      Only `cancelled` releases the run; `release-incomplete` still holds verify-loop state and
      `not-applied` changed nothing. The reply text cannot decide: a failed release leads with the
      same "Chain Cancelled" words,

• 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 (`@` is a deprecated alias), `#` 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 builtin_commands import is_builtin_command
  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 operators import annotate_prompt_references, detect_operator, has_chain_delimiter
  from session_state import (
      UNKNOWN_INTERRUPT_LABEL,
      ChainState,
      format_chain_reminder,
      format_gate_directive,
      format_pause_directive,
  )
  
  
  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"
      """
      annotations = annotate_prompt_references(message)
      return annotations[0] if annotations else 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 complete prompt references in chain order with their typed spelling.
  
      Detects unquoted delimiter operators from the SSOT registry,
      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
      """
      annotations = annotate_prompt_references(message)
      return (
          [prompt for prompt in annotations if prompt is not None]
          if has_chain_delimiter(message) and None not in annotations
          else []
      )
  
  
  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 (or the deprecated @FRAMEWORK alias) in message.
      Returns the framework ID if found (normalized to lowercase).
  
      Examples:
          ^CAGEERF >>analyze -> "cageerf"
          @ReACT >>debug -> "react"  (deprecated alias)
  
      Note: Framework IDs are normalized to lowercase to align with MCP server storage.
      """
      matches = detect_operator(message, "framework")
      # Normalize to lowercase (MCP server stores framework keys as lowercase)
      return matches[0].lower() if matches else None
  
  
  def detect_repetition(message: str) -> int | None:
      """
      Detect * N repetition syntax in message.
      Returns the repetition count if found.
  
      Examples:
          >>prompt * 3 -> 3
          >>analyze * 5 --> >>summarize -> 5
      """
      matches = detect_operator(message, "repetition")
      return int(matches[0]) if matches else None
  
  
  def parse_inline_args(message: str) -> dict[str, str]:
      """
      Parse key:"value" or key:'value' argument patterns from message

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