Hooks

Where it lives

• hooks/finish-the-work.shGitHub ↗
  Read the script

  #!/bin/bash
  # finish-the-work.sh — Stop hook
  # Detects early termination where the agent only PROMISES work without doing it, and re-engages it.
  # Deterministic (regex). Infinite loops are guarded by stop_hook_active.
  # stdin: JSON { transcript_path, stop_hook_active, ... }
  # stdout: {"decision":"block","reason":"..."} to continue, otherwise empty (exit 0)
  
  set -e
  input=$(cat)
  
  # Loop guard: if this hook already forced one continuation, do not block again.
  active=$(printf '%s' "$input" | python3 -c "import sys,json; print(json.load(sys.stdin).get('stop_hook_active', False))" 2>/dev/null || echo "False")
  if [ "$active" = "True" ]; then exit 0; fi
  
  tpath=$(printf '%s' "$input" | python3 -c "import sys,json; print(json.load(sys.stdin).get('transcript_path',''))" 2>/dev/null || echo "")
  if [ -z "$tpath" ] || [ ! -f "$tpath" ]; then exit 0; fi
  # Reject paths outside TMPDIR or HOME to block path-traversal via crafted hook input.
  case "$tpath" in
      "${TMPDIR:-/tmp}"/* | "$HOME"/*) ;;
      *) exit 0 ;;
  esac
  
  # Extract the last assistant message's text and whether it ended with a tool call.
  python3 - "$tpath" <<'PY'
  import sys, json, re
  
  path = sys.argv[1]
  last_text = ""
  last_had_tool = False
  try:
      with open(path) as f:
          for line in f:
              line = line.strip()
              if not line:
                  continue
              try:
                  obj = json.loads(line)
              except json.JSONDecodeError:
                  continue
              msg = obj.get("message", obj)
              if obj.get("type") == "assistant" or msg.get("role") == "assistant":
                  content = msg.get("content", [])
                  if isinstance(content, list):
                      texts = [b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text"]
                      tools = [b for b in content if isinstance(b, dict) and b.get("type") == "tool_use"]
                      if texts or tools:
                          last_text = "\n".join(texts).strip()
                          last_had_tool = bool(tools)
  except Exception:
      sys.exit(0)
  
  # Ended with a tool call (still working) or with no text -> not an early stop.
  if last_had_tool or not last_text:
      sys.exit(0)
  
  # Inspect only the closing paragraph, not the whole report.
  tail = last_text[-400:]
  
  # Unfulfilled-promise patterns (English + Korean). Future/intent only; past tense is excluded.
  promise = re.search(
      r"\b(I'?ll|I will|let me|next,? I|now I'?ll)\b[^.]{0,60}\b(now|next|then|implement|create|write|add|run|fix|save|build|start|proceed)\b",
      tail, re.IGNORECASE)
  
  # A legitimate stop that ends by asking the user passes through.
  asks_user = re.search(
      r"(\?|shall i|would you like|do you want|let me know|which option)",
      tail, re.IGNORECASE)
  
  if promise and not asks_user:
      out = {
          "decision": "block",
          "reason": "Your previous response ended by stating an intent to do work without actually doing it. "
                    "Do that work now with tool calls. End the turn only when the task is complete or you are "
                    "blocked on input that only the user can provide."
      }
      print(json.dumps(out, ensure_ascii=False))
  sys.exit(0)
  PY
  

• hooks/gate_post_tool.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """fablize observation gate — PostToolUse.
  
  Records observed evidence after each Bash/Edit/Write tool call: whether files
  changed (and their kind), and whether a verification command ran and observably
  succeeded or failed. Fails open.
  """
  
  from __future__ import annotations
  
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts" / "gate"))
  
  from ledger import add_unique, emit_json, read_stdin_json, update_ledger
  from parse_tool_result import (
      changed_kinds,
      command_from_input,
      detect_failure,
      repeated_failure,
      verification_record,
  )
  
  
  def main() -> int:
      input_data = read_stdin_json()
      kinds = changed_kinds(input_data)
      failure = detect_failure(input_data)
      verification = verification_record(input_data)
      command = command_from_input(input_data)
  
      def apply(ledger):
          if kinds:
              ledger["changed_files_seen"] = True
              add_unique(ledger, "change_kinds", kinds)
          if verification:
              ledger["verification_results"].append(verification)
              if command:
                  ledger["verification_commands"].append(verification["command"])
          if failure:
              ledger["failures"].append(failure)
  
      ledger = update_ledger(input_data, apply)
  
      # silent-recovery guard: recover quietly from one-offs, but if the SAME class
      # of failure repeats (>=2), disclose it instead of retrying silently.
      repeat = repeated_failure(ledger.get("failures", [])) if failure else None
      if repeat:
          _sig, count = repeat
          emit_json(
              {
                  "hookSpecificOutput": {
                      "hookEventName": "PostToolUse",
                      "additionalContext": (
                          f"fablize: the same class of failure has repeated {count} times. "
                          "Stop retrying silently — report it briefly (what failed / recovery "
                          "already tried / next path)."
                      ),
                  }
              }
          )
      elif failure:
          emit_json(
              {
                  "hookSpecificOutput": {
                      "hookEventName": "PostToolUse",
                      "additionalContext": "fablize gate observed a tool failure. Do not report completion until it is fixed, isolated as a known baseline, or explicitly documented.",
                  }
              }
          )
      else:
          emit_json({})
      return 0
  
  
  if __name__ == "__main__":
      try:
          raise SystemExit(main())
      except Exception as exc:  # noqa: BLE001 — fail open
          emit_json({"systemMessage": f"fablize gate post-tool hook failed open: {exc}"})
          raise SystemExit(0)
  

• hooks/gate_prompt.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """fablize observation gate — UserPromptSubmit.
  
  Classifies the new prompt's task mode and resets the per-prompt ledger so the
  Stop gate judges only this turn's evidence. Fails open (emits {} on any error).
  """
  
  from __future__ import annotations
  
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts" / "gate"))
  
  from ledger import add_unique, emit_json, read_stdin_json, update_ledger
  from classify_task import classify_prompt, context_for_mode
  
  
  def main() -> int:
      input_data = read_stdin_json()
      prompt = str(input_data.get("prompt") or input_data.get("user_prompt") or "")
      mode, risks = classify_prompt(prompt)
  
      def apply(ledger):
          ledger["task_mode"] = mode
          ledger["changed_files_seen"] = False
          ledger["change_kinds"] = []
          ledger["risk_flags"] = []
          ledger["verification_commands"] = []
          ledger["verification_results"] = []
          ledger["failures"] = []
          ledger["stop_blocks"] = 0
          add_unique(ledger, "risk_flags", risks)
  
      update_ledger(input_data, apply)
  
      emit_json(
          {
              "hookSpecificOutput": {
                  "hookEventName": "UserPromptSubmit",
                  "additionalContext": context_for_mode(mode, risks),
              }
          }
      )
      return 0
  
  
  if __name__ == "__main__":
      try:
          raise SystemExit(main())
      except Exception as exc:  # noqa: BLE001 — fail open, never block on our own bug
          emit_json({"systemMessage": f"fablize gate prompt hook failed open: {exc}"})
          raise SystemExit(0)
  

• hooks/gate_stop.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """fablize observation gate — Stop.
  
  Blocks completion when this turn's observed ledger shows a non-quick, non-docs
  task that changed files but has no observed successful verification. Capped at
  MAX_STOP_BLOCKS reminders, then allows with a warning. Fails open.
  
  Runs alongside finish-the-work.sh (promise-no-act guard): the two cover
  different failure modes — claim-without-observation vs intent-without-action.
  """
  
  from __future__ import annotations
  
  import os
  import sys
  from pathlib import Path
  
  _HERE = Path(__file__).resolve().parent
  sys.path.insert(0, str(_HERE.parent / "scripts" / "gate"))
  sys.path.insert(0, str(_HERE.parent / "scripts" / "shadow"))
  
  from ledger import emit_json, load_ledger, read_stdin_json, save_ledger
  from verify_state import should_block_stop, warning_after_max_blocks
  
  
  def _holdout_suppresses(input_data: dict) -> bool:
      """M3 holdout: env-gated, opt-in. When FABLIZE_HOLDOUT=1, sessions in the
      'off' arm skip the gate (pure baseline for measuring whether the gate
      helps). Default OFF -> gate behaviour is identical, so the 16/16 regression
      (which tests should_block_stop directly) is unaffected. The arm is computed
      out-of-band and is never shown to the model."""
      if os.environ.get("FABLIZE_HOLDOUT") != "1":
          return False
      try:
          from shadow_logger import holdout_arm
          return holdout_arm(input_data.get("session_id") or "") == "off"
      except Exception:
          return False
  
  
  def _log_holdout(input_data: dict, reason: str) -> None:
      """Record the holdout suppression out-of-band (events.jsonl). Never emits to
      the model and never raises into the gate path."""
      try:
          from shadow_logger import append_event, make_event
          append_event(make_event(input_data.get("session_id") or "", "holdout_suppress",
                                  {"would_block_reason": reason}))
      except Exception:
          pass
  
  
  def main() -> int:
      input_data = read_stdin_json()
      # Respect the loop guard so we never block twice in a row on the same stop.
      if input_data.get("stop_hook_active") is True:
          emit_json({})
          return 0
  
      ledger = load_ledger(input_data)
      block, reason = should_block_stop(ledger)
      if block:
          # M3 holdout (env-gated, default off): 'off' arm sessions skip the gate
          # so we can measure the gate's effect against a pure baseline.
          if _holdout_suppresses(input_data):
              _log_holdout(input_data, reason)
              emit_json({})
              return 0
          ledger["stop_blocks"] = int(ledger.get("stop_blocks") or 0) + 1
          save_ledger(input_data, ledger)
          emit_json({"decision": "block", "reason": reason})
          return 0
  
      warning = warning_after_max_blocks(ledger)
      if warning:
          emit_json(
              {
                  "systemMessage": warning,
                  "hookSpecificOutput": {"hookEventName": "Stop", "additionalContext": warning},
              }
          )
      else:
          emit_json({})
      return 0
  
  
  if __name__ == "__main__":
      try:
          raise SystemExit(main())
      except Exception as exc:  # noqa: BLE001 — fail open
          emit_json({"systemMessage": f"fablize gate stop hook failed open: {exc}"})
          raise SystemExit(0)
  

• hooks/router.shRunsGitHub ↗
  Read the script

  #!/usr/bin/env bash
  # fablize UserPromptSubmit router — when a task signal is detected, inject the relevant pack discipline as context.
  # Routing: smallest matching pack only / overlap only when genuinely multi-category / mimic observable behavior only.
  # Only verified packs are auto-routed.
  # stdin: JSON {"prompt": "..."}. stdout: extra context (only when a signal matches). Always exits 0.
  set -uo pipefail
  
  # Plugin root: prefer the runtime-injected var, else fall back to this script's location.
  ROOT="${CLAUDE_PLUGIN_ROOT:-$(cd "$(dirname "$0")/.." && pwd)}"
  PACKS="$ROOT/packs"
  
  prompt="$(python3 -c 'import sys,json
  try: print(json.load(sys.stdin).get("prompt",""))
  except Exception: pass' 2>/dev/null || true)"
  [ -z "${prompt:-}" ] && exit 0
  low="$(printf '%s' "$prompt" | tr '[:upper:]' '[:lower:]')"
  
  emit=""
  add() { emit="${emit:+$emit
  }$1"; }
  
  # Debugging / root-cause → investigation-protocol
  case "$low" in
    *debug*|*bug*|*error*|*traceback*|*"stack trace"*|*crash*|*failing*|*"not working"*)
      add "[fablize:investigation] Debugging/root-cause signal — follow $PACKS/investigation-protocol.txt: reproduce first, form 3+ competing hypotheses, gather evidence per hypothesis, trace the full causal chain, verify before/after, and report the hypotheses you rejected." ;;
  esac
  # Render/executable artifacts → verification-grounding
  case "$low" in
    *html*|*svg*|*game*|*canvas*|*chart*|*render*|*website*|*webpage*)
      add "[fablize:grounding] Render/executable artifact signal — follow $PACKS/verification-grounding-pack.txt grounding loop: run it in the real renderer, observe the actual output, fix what the observation reveals, then re-run. A static check is not observation." ;;
  esac
  
  [ -n "$emit" ] && printf '%s\n' "$emit"
  exit 0
  

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.