Hooks

Where it lives

• hooks/dashclaw_db_containment.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  DashClaw database containment (Neon) — the staging medium for the
  `db_branch` containment basis.
  
  See docs/rfcs/2026-09-04-database-containment.md. The file basis stages a
  mutation in a git worktree; this module stages a Postgres mutation on an
  ephemeral Neon branch of the same database, so the operator reviews the
  statement, the schema diff and the output once and Promote replays the
  ORIGINAL command against production.
  
  Imported by dashclaw_pretool.py (detect + branch + rewrite) and
  dashclaw_posttool.py (schema diff + credential scrub). Stdlib only, like
  every hook script here.
  
  SECRET DISCIPLINE: this module is the only place a production connection URL
  is ever read. It is never logged, never sent to DashClaw, and never written
  to session state — only the branch HOST and the Neon ids are persisted, and
  every string that leaves the machine goes through scrub_db_credentials()
  first. Callers must keep that property: log the host, never the URL.
  """
  
  import json
  import os
  import re
  import tempfile
  import urllib.parse
  import urllib.request
  from datetime import datetime, timedelta, timezone
  
  # Quoting-aware segmentation, shared with the classifier so the hook's notion
  # of "this segment runs psql" matches the one the server grades on.
  from dashclaw_agent_intel.command_parser import parse_command, split_chain_texts
  
  # ---------------------------------------------------------------------------
  # Environment
  # ---------------------------------------------------------------------------
  
  # Neon's control-plane API. Overridable ONLY so the hook test suite can point
  # the client at a local mock: these requests carry the operator's NEON_API_KEY
  # as a Bearer token, so a production installation must never set this.
  NEON_API_BASE_DEFAULT = "https://console.neon.tech/api/v2"
  _DEFAULT_TTL_HOURS = 72
  _HTTP_TIMEOUT = 10
  # A Neon account with many projects must not stall PreToolUse: the endpoint
  # probe walks at most this many pages of 100 before giving up (set
  # NEON_PROJECT_ID to skip the walk entirely).
  _MAX_PROJECT_PAGES = 10
  
  
  def db_containment_enabled():
      """DB containment kill switch, independent of DASHCLAW_CONTAINMENT (which
      covers both media). Read per call, not at import: the hook processes one
      tool call per process, and tests drive this through the environment."""
      return (os.environ.get("DASHCLAW_DB_CONTAINMENT") or "1") != "0"
  
  
  def neon_api_key():
      return os.environ.get("NEON_API_KEY") or ""
  
  
  def neon_api_base():
      return (os.environ.get("DASHCLAW_NEON_API_BASE") or NEON_API_BASE_DEFAULT).rstrip("/")
  
  
  def ttl_hours():
      raw = os.environ.get("DASHCLAW_DB_CONTAINMENT_TTL_HOURS")
      try:
          return max(1, int(raw)) if raw else _DEFAULT_TTL_HOURS
      except (TypeError, ValueError):
          return _DEFAULT_TTL_HOURS
  
  
  def utc_now_iso():
      """RFC3339 timestamp, the format Neon's API speaks."""
      return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
  
  
  def expires_at_iso(now=None):
      """RFC3339 expiry Neon self-deletes the branch at. Expiry IS the cleanup
      story (RFC): a session that never finishes leaves nothing behind."""
      base = now or datetime.now(timezone.utc)
      return (base + timedelta(hours=ttl_hours())).strftime("%Y-%m-%dT%H:%M:%SZ")
  
  
  # ---------------------------------------------------------------------------
  # Detection: is this command talking to a Postgres database, and which one?
  # ---------------------------------------------------------------------------
  
  _PG_URL_RE = re.compile(r"postgres(?:ql)?://[^\s'\"<>`]+", re.IGNORECASE)
  
  # Postgres clients whose mere presence in the command slot makes the segment a
  # database act (RFC "Evidence classifier"). Mirrors the server's classifyShell
  # list — a hook classifier that diverges from the server's is worse than a
  # shared over-tag, since the server does the real grading.
  _DB_BARE_CLIENTS = frozenset({"psql", "pg_restore"})
  _PRISMA_DB_SUBCOMMANDS = frozenset({
      ("db", "push"), ("db", "execute"),
      ("migrate", "deploy"), ("migrate", "dev"), ("migrate", "reset"),
  })
  _DRIZZLE_DB_SUBCOMMANDS = frozenset({"push", "migrate", "drop"})
  # `npx prisma …` / `pnpm dlx drizzle-kit …` are the way these tools are
  # actually invoked; the runner is not the act.
  _RUNNERS = frozenset({"npx", "pnpm", "yarn", "bunx", "bun", "npm"})
  _RUNNER_PASSTHROUGH = frozenset({"dlx", "exec", "run"})
  
  
  def _basename(token):
      token = (token or "").replace("\\", "/").rsplit("/", 1)[-1]
      if token.lower().endswith(".exe"):
          token = token[:-4]
      return token.lower()
  
  
  def _segment_words(parsed):
      """Non-flag words of one parsed segment, in order, with runner prefixes
      dropped.
  
      parse_command already strips leading KEY=VALUE assignments, wrappers
      (sudo/env/timeout/…) and redirections, and it tokenizes quote-aware — so a
      `git commit -m "fix the psql thing"` never surfaces `psql` in the command
      slot."""
      base = parsed.get("base_command") or ""
      if not base:
          return []
      words = [base]
      if parsed.get("subcommand"):
          words.append(parsed["subcommand"])
      words.extend(parsed.get("targets") or [])
      words = [_basename(w) for w in words]
      while words and words[0] in _RUNNERS:
          words.pop(0)
          while words and words[0] in _RUNNER_PASSTHROUGH:
              words.pop(0)
      return words
  
  
  def _segments(command):
      """Every sequential AND piped segment of the command, parsed. `psql` in
      the second stage of a pipe is still a database act."""
      segments = []
      for chain_text in split_chain_texts(command or ""):
          parsed = parse_command(chain_text)
          pipes = parsed.get("pipes") or []
          segments.extend(pipes if pipes else [parsed])
      return segments
  
  
  def is_db_command(command):
      """True when this Bash command targets a Postgres database.
  
      Used ONLY to decide whether to advertise `allow_contained:db` and, later,
      to rewrite the command — the risk grading stays server-side. Over-tagging
      is t

• hooks/dashclaw_posttool.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  DashClaw PostToolUse Hook v2 for Claude Code.
  
  Records the outcome of governed tool calls by updating the action record
  created by the PreToolUse hook. v2 adds richer outcome reporting:
    - 500-char output summaries (up from 200)
    - Structured outcome_metadata with exit_code, error_type classification
    - Improved error detection: checks exit code AND error field
    - Error classification: timeout, permission, not_found, runtime
  
  Never blocks. Always exits 0.
  """
  
  import hashlib
  import json
  import os
  import re
  import shutil
  import subprocess
  import sys
  import tempfile
  import urllib.request
  import urllib.error
  from datetime import datetime, timezone
  
  # Import the shared HTTP retry helper from the sibling intel package.
  sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
  from dashclaw_agent_intel.http_client import request_with_retry, encode_json_body
  from dashclaw_agent_intel.stop_state import contained_turn_path as _contained_turn_path
  from dashclaw_agent_intel.command_parser import parse_command as _parse_command
  from dashclaw_agent_intel.written_paths_ledger import record_written_paths as _ledger_record
  
  # Database containment (basis db_branch, RFC 2026-09-04): schema diff + the
  # connection-URL scrub. Optional import for the same reason as in
  # dashclaw_pretool.py -- an installation whose hook files predate this module
  # must keep reporting outcomes, not crash on every tool call.
  try:
      import dashclaw_db_containment
  except Exception:  # pragma: no cover - only reachable on a partial install
      dashclaw_db_containment = None
  
  # ---------------------------------------------------------------------------
  # Load .env file (C:/Projects/DashClaw/.env) before reading config.
  # Values already in the environment take precedence.
  # ---------------------------------------------------------------------------
  
  def _apply_env_line(line):
      line = line.strip()
      if not line or line.startswith("#") or "=" not in line:
          return
      key, _, val = line.partition("=")
      key = key.strip()
      val = val.strip().strip('"').strip("'")
      if " #" in val:
          val = val[:val.index(" #")].strip()
      if key and key not in os.environ:
          os.environ[key] = val
  
  
  def _apply_env_file(env_path):
      try:
          with open(env_path, encoding="utf-8") as f:
              for line in f:
                  _apply_env_line(line)
      except FileNotFoundError:
          return
  
  
  def _load_dotenv():
      # Test isolation escape hatch: when DASHCLAW_DISABLE_DOTENV is set, skip
      # the .env walk entirely so the subprocess only sees env vars the test
      # explicitly passed in. Never set this in production.
      if os.environ.get("DASHCLAW_DISABLE_DOTENV"):
          return
      # Walk up from the hook file's directory looking for env files. Works
      # whether this runs from hooks/X.py (project root is one parent up) or
      # from .claude/hooks/X.py after install-hooks runs (project root is two
      # parents up). Earlier files win because of `key not in os.environ`.
      tried = set()
      current = os.path.abspath(os.path.dirname(__file__))
      for _ in range(5):
          for fname in (".env.local", ".env"):
              env_path = os.path.join(current, fname)
              if env_path in tried:
                  continue
              tried.add(env_path)
              _apply_env_file(env_path)
          parent = os.path.dirname(current)
          if parent == current:
              break
          current = parent
  
  
  def _resolve_base_url(base_explicit, url_explicit):
      """Resolve DASHCLAW_BASE_URL/DASHCLAW_URL with explicit-env-beats-dotenv
      precedence: explicit BASE_URL > explicit URL > dotenv BASE_URL > dotenv
      URL. Mirrors dashclaw_pretool.py's _resolve_base_url -- see there for the
      2026-07-27 incident this fixes. Both hooks must resolve identically so a
      tool call governed under one BASE_URL isn't PATCHed against another."""
      base_val = os.environ.get("DASHCLAW_BASE_URL") or ""
      url_val = os.environ.get("DASHCLAW_URL") or ""
      if base_explicit and base_val:
          return base_val
      if url_explicit and url_val:
          if base_val and not base_explicit:
              sys.stderr.write(
                  "[DashClaw] Explicit DASHCLAW_URL=%s overrides a .env-provided "
                  "DASHCLAW_BASE_URL=%s\n" % (url_val, base_val)
              )
          return url_val
      return base_val or url_val
  
  
  # Captured BEFORE _load_dotenv() fills gaps, so _resolve_base_url can tell an
  # explicitly-exported value apart from one that only exists because dotenv
  # populated it.
  _BASE_URL_EXPLICIT = "DASHCLAW_BASE_URL" in os.environ
  _URL_EXPLICIT = "DASHCLAW_URL" in os.environ
  
  _load_dotenv()
  
  # ---------------------------------------------------------------------------
  # Configuration
  # ---------------------------------------------------------------------------
  
  def _argv_agent_id():
      # Mirrors dashclaw_pretool.py's _argv_agent_id: the harness integration
      # that wires this hook may append `--agent-id <id>` to the command line.
      # Needed here (not just for identity) so the instance-state hash below
      # matches pretool's when a harness passes the flag to both.
      argv = sys.argv[1:]
      for i, arg in enumerate(argv):
          if arg == "--agent-id" and i + 1 < len(argv):
              return argv[i + 1].strip()
          if arg.startswith("--agent-id="):
              return arg.split("=", 1)[1].strip()
      return ""
  
  
  BASE_URL = _resolve_base_url(_BASE_URL_EXPLICIT, _URL_EXPLICIT).rstrip("/")
  API_KEY = os.environ.get("DASHCLAW_API_KEY") or ""
  AGENT_ID = _argv_agent_id() or os.environ.get("DASHCLAW_AGENT_ID") or "claude-code"
  
  # Short stable hash of (resolved BASE_URL + AGENT_ID) -- must match
  # dashclaw_pretool.py's _INSTANCE_STATE_SUFFIX bit for bit so this reader only
  # ever finds the state its OWN installation's pretool wrote. See
  # dashclaw_pretool.py for the full rationale (2026-07-27 incident).
  _INSTANCE_STATE_SUFFIX = hashlib.sha256((BASE_URL + "|" + AGENT_ID).encode("ut

• hooks/dashclaw_pretool.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  DashClaw PreToolUse Hook v2 for Claude Code.
  
  Evaluates all 40+ agent tool calls against DashClaw guard policies
  using the dashclaw_agent_intel module for semantic classification.
  
  Exit codes:
    0 - Allow the tool to proceed
    2 - Block the tool (Claude Code shows stderr to user)
  """
  
  import hashlib
  import json
  import os
  import re
  import shutil
  import socket
  import subprocess
  import sys
  import tempfile
  import time
  import urllib.error
  import urllib.parse
  import urllib.request
  import uuid
  
  # ---------------------------------------------------------------------------
  # Load .env file (C:/Projects/DashClaw/.env) before reading config.
  # Values already in the environment take precedence.
  # ---------------------------------------------------------------------------
  
  def _apply_env_line(line):
      line = line.strip()
      if not line or line.startswith("#") or "=" not in line:
          return
      key, _, val = line.partition("=")
      key = key.strip()
      val = val.strip().strip('"').strip("'")
      if " #" in val:
          val = val[:val.index(" #")].strip()
      if key and key not in os.environ:
          os.environ[key] = val
  
  
  def _apply_env_file(env_path):
      try:
          with open(env_path, encoding="utf-8") as f:
              for line in f:
                  _apply_env_line(line)
      except FileNotFoundError:
          return
  
  
  def _load_dotenv():
      # Test isolation escape hatch: when DASHCLAW_DISABLE_DOTENV is set, skip
      # the .env walk entirely so the subprocess only sees env vars the test
      # explicitly passed in. Never set this in production.
      if os.environ.get("DASHCLAW_DISABLE_DOTENV"):
          return
      # Walk up from the hook file's directory looking for env files. Works
      # whether this runs from hooks/X.py (project root is one parent up) or
      # from .claude/hooks/X.py after install-hooks runs (project root is two
      # parents up). Earlier files win because of `key not in os.environ`.
      tried = set()
      current = os.path.abspath(os.path.dirname(__file__))
      for _ in range(5):
          for fname in (".env.local", ".env"):
              env_path = os.path.join(current, fname)
              if env_path in tried:
                  continue
              tried.add(env_path)
              _apply_env_file(env_path)
          parent = os.path.dirname(current)
          if parent == current:
              break
          current = parent
  
  
  def _resolve_base_url(base_explicit, url_explicit):
      """Resolve DASHCLAW_BASE_URL/DASHCLAW_URL with explicit-env-beats-dotenv
      precedence: explicit BASE_URL > explicit URL > dotenv BASE_URL > dotenv
      URL. `base_explicit`/`url_explicit` say whether the process env already
      had that key set BEFORE _load_dotenv() ran (captured below).
  
      Without this, a repo's own .env setting DASHCLAW_BASE_URL silently wins
      over an explicitly-exported DASHCLAW_URL merely because BASE_URL is
      checked first in the naive `BASE_URL or URL` fallback -- the 2026-07-27
      incident that misrouted three hook-triggered calls to a hosted production
      instance instead of the exported localhost URL."""
      base_val = os.environ.get("DASHCLAW_BASE_URL") or ""
      url_val = os.environ.get("DASHCLAW_URL") or ""
      if base_explicit and base_val:
          return base_val
      if url_explicit and url_val:
          if base_val and not base_explicit:
              sys.stderr.write(
                  "[DashClaw] Explicit DASHCLAW_URL=%s overrides a .env-provided "
                  "DASHCLAW_BASE_URL=%s\n" % (url_val, base_val)
              )
          return url_val
      return base_val or url_val
  
  
  # Captured BEFORE _load_dotenv() fills gaps, so _resolve_base_url can tell an
  # explicitly-exported value apart from one that only exists because dotenv
  # populated it.
  _BASE_URL_EXPLICIT = "DASHCLAW_BASE_URL" in os.environ
  _URL_EXPLICIT = "DASHCLAW_URL" in os.environ
  
  _load_dotenv()
  
  # ---------------------------------------------------------------------------
  # Import dashclaw_agent_intel (sibling directory)
  # ---------------------------------------------------------------------------
  
  sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
  from dashclaw_agent_intel import classify_bash, scan_file_operation, classify_tool, McpHealthMonitor
  from dashclaw_agent_intel.bash_classifier import is_bounded_rm, is_regenerable_artifact_rm
  from dashclaw_agent_intel.command_parser import parse_command
  from dashclaw_agent_intel.written_paths_ledger import (
      extract_exec_candidates,
      grade_script_content,
      lookup_written_path,
  )
  from dashclaw_agent_intel.file_scanner import is_placeholder_path
  from dashclaw_agent_intel.tool_recognizer import ungoverned_default_categories
  from dashclaw_agent_intel.http_client import request_with_retry, env_retries, encode_json_body
  
  # Database containment (basis db_branch, RFC 2026-09-04). Optional import: an
  # installation whose hook files predate this module (a partial copy, an older
  # `dashclaw install claude`) must keep governing every tool call, not fail
  # closed on all of them — a missing module simply means the db capability is
  # never advertised, which is exactly today's behavior.
  try:
      import dashclaw_db_containment
  except Exception:  # pragma: no cover - only reachable on a partial install
      dashclaw_db_containment = None
  
  # ---------------------------------------------------------------------------
  # Configuration
  # ---------------------------------------------------------------------------
  
  def _argv_agent_id():
      # Per-harness identity declaration (roadmap v2.2). The harness integration
      # that wires this hook appends `--agent-id <id>` to the command line, so
      # identity is per-harness even when every harness on the machine shares
      # one script directory / .env / DASHCLAW_AGENT_ID export. argv beats env
      # by design: the flag is written by the installer that knows which
      # harness it is wiring; the env var is machine-ambient.
      argv = sys.argv[1:]
      for i, arg in enumerate(argv):
          if 

• hooks/dashclaw_scope_sync.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """dashclaw_scope_sync.py -- SessionStart hook: reachability follows the role.
  
  Turns a role's `blocked_tools` (a `role_constraint` policy field) into Claude
  Code `permissions.deny` rules at session start, so a tool the role blocks
  cannot even be CHOSEN by the harness -- it is denied before the guard ever
  sees it. The server-side evaluator stays the backstop; this hook only removes
  reach, so a server that is unreachable simply leaves the deny list as it is.
  
  `allowed_tools` is intentionally NOT translated here: Claude Code's
  `permissions.deny` cannot express "deny everything except", so allowlists stay
  server-enforced only.
  
  Ownership: this hook owns ONLY the deny entries it wrote, tracked in
  `<project>/.claude/.dashclaw-scope.json`. It never touches deny entries it did
  not add and never breaks a session -- any failure exits 0 silently.
  """
  
  import json
  import os
  import sys
  import urllib.parse
  from datetime import datetime, timezone
  
  SETTINGS_FILE = "settings.local.json"
  SCOPE_FILE = ".dashclaw-scope.json"
  
  
  def _now_iso():
      return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
  
  
  def _debug(msg):
      """Debug/diagnostic line -> stderr (Claude Code shows only stdout as context)."""
      try:
          sys.stderr.write("[dashclaw_scope_sync] " + str(msg) + "\n")
      except Exception:
          pass
  
  
  def translate_pattern(pattern):
      """Map one `blocked_tools` glob to a Claude Code permission rule.
  
      - A bare tool name maps to itself: `Bash` -> `Bash`, `mcp__x__search` ->
        `mcp__x__search`.
      - A whole trailing `__*` becomes a server-wide MCP rule: `mcp__xapi__*` ->
        `mcp__xapi`.
      - Any other `*` placement cannot be expressed as a deny rule -> None
        (the caller skips and counts it).
      """
      if not isinstance(pattern, str) or not pattern:
          return None
      if "*" not in pattern:
          return pattern
      if pattern.endswith("__*") and "*" not in pattern[:-3]:
          return pattern[:-3]
      return None
  
  
  def blocked_tools_from_policies(policies):
      """Union `blocked_tools` across active `role_constraint` policies (order-preserving)."""
      seen = []
      for p in policies or []:
          if not isinstance(p, dict):
              continue
          if p.get("policy_type") != "role_constraint" or not p.get("active"):
              continue
          rules = p.get("rules")
          if isinstance(rules, str):
              try:
                  rules = json.loads(rules)
              except Exception:
                  continue
          if not isinstance(rules, dict):
              continue
          for t in rules.get("blocked_tools") or []:
              if isinstance(t, str) and t and t not in seen:
                  seen.append(t)
      return seen
  
  
  def _translate_all(blocked):
      """Translate the union to deny rules, dropping (and counting) inexpressible globs."""
      entries, skipped = [], 0
      for pat in blocked:
          rule = translate_pattern(pat)
          if rule is None:
              skipped += 1
              _debug('skipped inexpressible blocked_tools pattern "%s"' % pat)
          elif rule not in entries:
              entries.append(rule)
      return entries, skipped
  
  
  def _read_owned(scope_path):
      """Previously owned deny entries from the scope file (best-effort)."""
      if not os.path.exists(scope_path):
          return []
      try:
          with open(scope_path, "r", encoding="utf-8") as f:
              prev = json.load(f)
      except Exception:
          return []
      if isinstance(prev, dict) and isinstance(prev.get("entries"), list):
          return [e for e in prev["entries"] if isinstance(e, str)]
      return []
  
  
  def sync_deny(project_dir, agent_id, entries):
      """Merge `entries` into <project>/.claude/settings.local.json permissions.deny.
  
      Owns only the entries this hook wrote (tracked in .dashclaw-scope.json):
      stale owned entries are removed, new ones added, entries owned by anyone
      else are left untouched, and every other key in the file is preserved.
      Returns (added, removed), or None if settings.local.json existed but could
      not be read/parsed (in which case nothing is written).
      """
      claude_dir = os.path.join(project_dir, ".claude")
      settings_path = os.path.join(claude_dir, SETTINGS_FILE)
      scope_path = os.path.join(claude_dir, SCOPE_FILE)
  
      prev_owned = _read_owned(scope_path)
      prev_set = set(prev_owned)
  
      settings = {}
      had_trailing_newline = True
      if os.path.exists(settings_path):
          try:
              with open(settings_path, "r", encoding="utf-8") as f:
                  raw = f.read()
          except Exception:
              _debug("settings.local.json unreadable; leaving it untouched")
              return None
          if raw.strip():
              try:
                  settings = json.loads(raw)
              except Exception:
                  _debug("settings.local.json unparseable; leaving it untouched")
                  return None
              if not isinstance(settings, dict):
                  _debug("settings.local.json is not a JSON object; leaving it untouched")
                  return None
              had_trailing_newline = raw.endswith("\n")
  
      perms = settings.get("permissions")
      if not isinstance(perms, dict):
          perms = {}
      deny = perms.get("deny")
      if not isinstance(deny, list):
          deny = []
  
      existing_before = set(d for d in deny if isinstance(d, str))
      want = set(entries)
  
      # Remove entries this hook previously owned that are no longer wanted.
      removed = 0
      new_deny = []
      for d in deny:
          if isinstance(d, str) and d in prev_set and d not in want:
              removed += 1
              continue
          new_deny.append(d)
  
      # Add wanted entries not already present.
      added = 0
      for e in entries:
          if e not in new_deny:
              new_deny.append(e)
              added += 1
  
      # We own an entry when we previously owned it or just added it; an entry
      # that already existed under someone else's auth

• hooks/dashclaw_stop.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  DashClaw Stop Hook for Claude Code.
  
  Captures the assistant turn's LLM token usage from the session transcript and
  PATCHes it to the action records created during the turn. Cost is derived
  server-side from the configured model pricing table.
  
  Data flow:
    - PreToolUse (dashclaw_pretool.py) appends each new action_id to
      "dashclaw_turn_<session_id>" in the temp dir.
    - This Stop hook sums token usage across assistant messages that landed
      since the last Stop (tracked via "dashclaw_stop_cursor_<session_id>"),
      distributes the totals evenly across the turn's action_ids, and PATCHes
      each action with tokens_in, tokens_out, and model. The server then
      derives cost_estimate from its pricing table.
  
  This file is the orchestrator; the mechanics live in the sibling intel
  package (extracted in the health pass so each seam is unit-testable):
    - dashclaw_agent_intel.stop_transcript — transcript parsing, turn/usage
      math, tool_use collection, assumption extraction, distribution.
    - dashclaw_agent_intel.stop_state — tempdir session state (turn actions,
      cursor, posted-assumption keys, throttle markers).
  
  Never blocks. Always exits 0.
  """
  
  import hashlib
  import json
  import os
  import re
  import sys
  import tempfile
  import urllib.request
  import urllib.error
  
  # Import the shared HTTP retry helper from the sibling intel package.
  sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
  from dashclaw_agent_intel.http_client import request_with_retry, encode_json_body
  from dashclaw_agent_intel.stop_state import (
      log_hook_error as _log_hook_error,
      read_cursor as _read_cursor,
      write_cursor as _write_cursor,
      read_turn_actions as _read_turn_actions,
      clear_turn_actions as _clear_turn_actions,
      read_posted_assumption_keys as _read_posted_assumption_keys,
      append_posted_assumption_keys as _append_posted_assumption_keys,
      read_posted_deviation_keys as _read_posted_deviation_keys,
      append_posted_deviation_keys as _append_posted_deviation_keys,
      count_session_actions as _count_session_actions,
      read_contained_turn_actions as _read_contained_turn_actions,
      clear_contained_turn_actions as _clear_contained_turn_actions,
      read_posted_containment_keys as _read_posted_containment_keys,
      append_posted_containment_keys as _append_posted_containment_keys,
  )
  from dashclaw_agent_intel.stop_transcript import (
      load_entries as _load_entries,
      resolve_turn_start as _resolve_turn_start,
      collect_turn_usage as _collect_turn_usage,
      collect_turn_tool_uses as _collect_turn_tool_uses,
      is_governed_tool_name as _is_governed_tool_name,
      turn_assistant_text as _turn_assistant_text,
      extract_assumptions as _extract_assumptions,
      extract_deviations as _extract_deviations,
      distribute as _distribute,
      patch_body_for as _patch_body_for,
      datetime_now_iso,
  )
  
  # ---------------------------------------------------------------------------
  # Env loading — pretool/posttool load from .env.local + .env before reading
  # DASHCLAW_* config; stop needs the same so tokens actually PATCH back
  # instead of hitting an empty URL with an empty API key. Without this,
  # the Stop hook silently fails for every session that doesn't inherit
  # the vars from the shell — which is most real Claude Code sessions.
  # ---------------------------------------------------------------------------
  
  def _apply_env_line(line):
      line = line.strip()
      if not line or line.startswith("#") or "=" not in line:
          return
      key, _, val = line.partition("=")
      key = key.strip()
      val = val.strip().strip('"').strip("'")
      if " #" in val:
          val = val[:val.index(" #")].strip()
      if key and key not in os.environ:
          os.environ[key] = val
  
  
  def _apply_env_file(env_path):
      try:
          with open(env_path, encoding="utf-8") as f:
              for line in f:
                  _apply_env_line(line)
      except FileNotFoundError:
          return
  
  
  def _load_dotenv():
      # Test isolation escape hatch: when DASHCLAW_DISABLE_DOTENV is set, skip
      # the .env walk entirely so the subprocess only sees env vars the test
      # explicitly passed in. Never set this in production.
      if os.environ.get("DASHCLAW_DISABLE_DOTENV"):
          return
      # Walk up from the hook file's directory looking for env files. Works
      # whether this runs from hooks/X.py (project root is one parent up) or
      # from .claude/hooks/X.py after install-hooks runs (project root is two
      # parents up). Earlier files win because of `key not in os.environ`.
      tried = set()
      current = os.path.abspath(os.path.dirname(__file__))
      for _ in range(5):
          for fname in (".env.local", ".env"):
              env_path = os.path.join(current, fname)
              if env_path in tried:
                  continue
              tried.add(env_path)
              _apply_env_file(env_path)
          parent = os.path.dirname(current)
          if parent == current:
              break
          current = parent
  
  
  def _resolve_base_url(base_explicit, url_explicit):
      """Resolve DASHCLAW_BASE_URL/DASHCLAW_URL with explicit-env-beats-dotenv
      precedence: explicit BASE_URL > explicit URL > dotenv BASE_URL > dotenv
      URL. Mirrors dashclaw_pretool.py's _resolve_base_url -- see there for the
      2026-07-27 incident this fixes."""
      base_val = os.environ.get("DASHCLAW_BASE_URL") or ""
      url_val = os.environ.get("DASHCLAW_URL") or ""
      if base_explicit and base_val:
          return base_val
      if url_explicit and url_val:
          if base_val and not base_explicit:
              sys.stderr.write(
                  "[DashClaw] Explicit DASHCLAW_URL=%s overrides a .env-provided "
                  "DASHCLAW_BASE_URL=%s\n" % (url_val, base_val)
              )
          return url_val
      return base_val or url_val
  
  
  # Captured BEFORE _load_dotenv() fills gaps, so _resolve_base_url can tell an
  # explicitly-exported value apart from one that only exists because dotenv
  # populated it.
  _BASE_URL_EXPLICIT

• hooks/enforcement_liveness_probe.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  DashClaw enforcement-liveness probe (roadmap v8.2).
  
  Proves the enforcement seam holds END TO END on a live install: drives a
  synthetic action that policy must hold (block or approval-wait) through the
  SAME PreToolUse hook seam real actions use, and verdicts by observing whether
  the action executed — never by reading the decision ledger, because the ledger
  is exactly what kept lying in v4.72.1 (a hook timeout misconfig cancelled the
  pretool hook; guard rows kept flowing while every block failed open).
  
  Which seam a run drives is chosen by `--runtime`: each harness declares its
  hooks in its own config file (Claude Code `.claude/settings.json`, codex
  `~/.codex/config.toml`), and the probe reads only the one belonging to the
  runtime it is reporting as. `--settings` overrides that resolution.
  
  Harness emulation contract (each clause verified against the real Claude Code
  harness during the v4.72.1 incident, 2026-07-06; the timer-overflow clause is
  NOT assumed to hold for other harnesses — see OVERFLOW_HARNESSES):
    1. The hook `timeout` field in settings.json is SECONDS (default 600).
    2. The harness arms a timer of timeout*1000 ms; a value whose ms product
       exceeds 2^31-1 overflows the timer, which fires immediately and cancels
       the hook (73/73 cancellations observed).
    3. Exit code 2 blocks the tool; any other outcome — exit 0, other codes,
       cancellation — lets the tool PROCEED (fail-open).
  The probe reproduces exactly that contract around the real, unmodified hook:
  exit 2 => the synthetic action is not executed; anything else => the probe
  executes it (a Write of a marker file) and the file's existence is the witness.
  
  Verdicts:
    held       — the seam blocked the probe action (exit 2), witness absent.
    executed   — the harness contract would have let the action through and the
                 witness file exists (v4.72.1 class: overflowed timeout, hook
                 crash, fail-open path).
    unprovable — the seam ran but enforcement cannot be proven: no hook entry
                 installed, observe mode, no policy holds the probe action, or
                 the hook hung past the probe's wait budget. Rendered broken on
                 the surfaces — you cannot claim enforcement you cannot prove.
  
  Synthetic hygiene: the probe runs the hook as agent `smoke-liveness-probe`
  (the `smoke-` prefix is the established synthetic marker), so its guard rows
  are excluded from every aggregate, and any pending approval it creates is
  cancelled in teardown. Its verdict is filed to POST /api/enforcement-liveness
  — its own table, never the action/guard ledgers (live-canary precedent).
  
  SessionStart entry point: wired directly as the SessionStart hook (it replaced
  the retired session-digest hook that used to spawn it). Invoked as
  `--source session-start`, the probe throttles itself to at most once per 12h
  (marker file) and runs the actual probe in a DETACHED child, so session start
  is never delayed or broken — the same contract the digest provided. Every
  other `--source` (manual, ci, session-start's own detached child) runs the
  probe inline.
  
  Usage:
    python hooks/enforcement_liveness_probe.py [--settings PATH] [--source S]
        [--witness-dir DIR] [--max-wait SECONDS] [--json]
  
  Env: DASHCLAW_BASE_URL + DASHCLAW_API_KEY enable reporting (loaded from .env
  like the hooks). DASHCLAW_LIVENESS_PROBE_DISABLED=1 exits 0 immediately.
  Exit code: 0 when the verdict is `held` (and the report, if configured,
  filed); 1 otherwise.
  """
  
  import argparse
  import hashlib
  import json
  import os
  import re
  import shlex
  import shutil
  import subprocess
  import sys
  import time
  import urllib.error
  import urllib.request
  import uuid
  from datetime import datetime, timezone
  
  VERSION_PROBE_TIMEOUT_SECONDS = 2.0
  UNAVAILABLE_METADATA = "unavailable"
  _RUNTIME_VERSION_COMMANDS = {
      "claude-code": ("claude", "--version"),
      "codex": ("codex", "--version"),
      "hermes": ("hermes", "--version"),
  }
  _RUNTIME_VERSION_PATTERNS = {
      "claude-code": re.compile(r"^\d+(?:\.\d+){1,3}[0-9A-Za-z._+-]* \(Claude Code\)$"),
      "codex": re.compile(r"^codex-cli \d+(?:\.\d+){1,3}[0-9A-Za-z._+-]*$"),
      "hermes": re.compile(r"^Hermes Agent v\d+(?:\.\d+){1,3}[0-9A-Za-z._+-]*(?: \([0-9.]+\))?$"),
  }
  _HOOK_SOURCE_NAMES = {"dashclaw_pretool.py", "dashclaw_pretool_hermes.py"}
  
  # Codex declares its hooks in TOML, Claude Code in JSON. tomllib is stdlib from
  # 3.11; older interpreters need the tomli backport. A missing parser is
  # reported as a config problem rather than swallowed — silently finding no
  # codex seam would render the same false green this probe exists to catch.
  try:
      import tomllib
  except ImportError:  # pragma: no cover - only on Python <= 3.10
      try:
          import tomli as tomllib
      except ImportError:
          tomllib = None
  
  # Same .env loading contract as the hooks (values already in the environment
  # win; DASHCLAW_DISABLE_DOTENV skips the walk for test isolation).
  
  
  def _apply_env_line(line):
      line = line.strip()
      if not line or line.startswith("#") or "=" not in line:
          return
      key, _, val = line.partition("=")
      key = key.strip()
      val = val.strip().strip('"').strip("'")
      if " #" in val:
          val = val[:val.index(" #")].strip()
      if key and key not in os.environ:
          os.environ[key] = val
  
  
  def _load_dotenv():
      if os.environ.get("DASHCLAW_DISABLE_DOTENV"):
          return
      tried = set()
      current = os.path.abspath(os.path.dirname(__file__))
      for _ in range(5):
          for fname in (".env.local", ".env"):
              env_path = os.path.join(current, fname)
              if env_path in tried:
                  continue
              tried.add(env_path)
              try:
                  with open(env_path, encoding="utf-8") as f:
                      for line in f:
                          _apply_env_line(line)
              except FileNotFoundError:
                  pass
          parent = os.path.dirname(current)
          if parent == current:
              break
          cur

• hooks/run_hook.cjsGitHub ↗

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