Hooks

Where it lives

• hooks/_shell_scan.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """One Bash-shaped scanner for the PreToolUse guards that must read a command.
  
  Guards decide things like "is this argument the one bash will rewrite before
  `gh` sees it". Answering that from regex lookbehind or `shlex` fails in three
  ways a reviewer reproduced against bash itself: an escaped quote ends a
  `shlex(posix=False)` token early, an EVEN run of backslashes leaves `$(` active
  while a lookbehind reads the last one as an escape, and an escaped space looks
  like a word boundary. All three UNDER-deny, silently.
  
  So state is carried, not inferred from the previous character: quoting, escape,
  and whether we are at a word start. `bash` is the oracle the tests compare to.
  """
  from __future__ import annotations
  
  import os
  from dataclasses import dataclass, field
  from typing import List
  
  OPERATORS = (";;", "&&", "||", ";", "|", "&", "(", ")", "\n")
  # What bash SUBSTITUTES: command substitution only. `$VAR` is an ordinary
  # interpolation, not a code span, and denying it would cry wolf.
  _SUBST_OPENERS = ("`", "$(")
  
  
  @dataclass
  class Word:
      """One argv word, plus what the raw source said about it."""
  
      raw: str = ""
      text: str = ""
      quoted: str = ""          # the quote style that opened the VALUE: '' " or ''
      expands: bool = False     # an ACTIVE ` or $( — bash would run it
      is_operator: bool = False
  
      def basename_is(self, name: str, fold: bool = False) -> bool:
          """Is this word the program `name`, however it was spelled?
  
          Compares the EXPANDED text, so `/usr/bin/gh` and `"gh"` both match; with
          fold=True, `GH` does too — a case-insensitive filesystem runs it.
          """
          base = os.path.basename(self.text)
          return base.lower() == name.lower() if fold else base == name
  
  
  @dataclass
  class _State:
      quote: str = ""       # "", "'", or '"'
      escaped: bool = False
      in_word: bool = False
      words: List[Word] = field(default_factory=list)
      cur: Word = field(default_factory=Word)
  
  
  def _flush(st: _State) -> None:
      if st.in_word:
          st.words.append(st.cur)
      st.cur = Word()
      st.in_word = False
  
  
  _ANSI_C_SIMPLE = {"a": "\a", "b": "\b", "e": "\x1b", "E": "\x1b", "f": "\f",
                    "n": "\n", "r": "\r", "t": "\t", "v": "\v",
                    "\\": "\\", "'": "'", '"': '"', "?": "?"}
  
  
  def _ansi_c(body: str) -> str:
      """Bash's $'...' escapes, spelled out rather than borrowed from Python.
  
      `unicode_escape` is close but not bash — no \\e, no \\cX — and a guard that
      mis-decodes here reports a value gh never receives.
      """
      out, i = [], 0
      while i < len(body):
          ch = body[i]
          if ch != "\\" or i + 1 >= len(body):
              out.append(ch)
              i += 1
              continue
          nxt = body[i + 1]
          if nxt in _ANSI_C_SIMPLE:
              out.append(_ANSI_C_SIMPLE[nxt])
              i += 2
              continue
          if nxt == "x":
              j = i + 2
              while j < len(body) and j < i + 4 and body[j] in "0123456789abcdefABCDEF":
                  j += 1
              if j > i + 2:
                  out.append(chr(int(body[i + 2:j], 16)))
                  i = j
                  continue
          if nxt in "01234567":
              j = i + 1
              while j < len(body) and j < i + 4 and body[j] in "01234567":
                  j += 1
              out.append(chr(int(body[i + 1:j], 8) & 0xFF))
              i = j
              continue
          if nxt in ("u", "U"):
              width = 4 if nxt == "u" else 8
              j = i + 2
              while j < len(body) and j < i + 2 + width and body[j] in "0123456789abcdefABCDEF":
                  j += 1
              if j > i + 2:
                  out.append(chr(int(body[i + 2:j], 16)))
                  i = j
                  continue
          if nxt == "c" and i + 2 < len(body):
              out.append(chr(ord(body[i + 2].upper()) ^ 0x40))
              i += 3
              continue
          out.append("\\")
          out.append(nxt)
          i += 2
      # A NUL cannot survive into argv, so bash truncates THIS span at it — the
      # rest of the word still concatenates ($'a\\0b'c is "ac", not "a").
      return "".join(out).split("\x00", 1)[0]
  
  
  def _ansi_c_span(command: str, i: int):
      """(decoded, index past the closing quote) for a $'...' at i, or (None, -1)
      when it never closes."""
      j = i + 2
      body = []
      while j < len(command):
          c = command[j]
          if c == "\\" and j + 1 < len(command):
              body.append(c)
              body.append(command[j + 1])
              j += 2
              continue
          if c == "'":
              return _ansi_c("".join(body)), j + 1
          body.append(c)
          j += 1
      return None, -1
  
  
  def _double_quote_escapes(ch: str) -> bool:
      """Inside double quotes bash honours a backslash only before these; before
      anything else the backslash is a literal character."""
      return ch in '$`"\\\n'
  
  
  def words(command: str) -> List[Word]:
      """argv words plus operators, or [] when the line does not lex.
  
      An unterminated quote makes bash refuse the whole command, so returning []
      is the honest answer — a guard must not scan a fragment of something that
      will never run.
      """
      st = _State()
      i, n = 0, len(command)
      while i < n:
          ch = command[i]
  
          if st.escaped:
              # The backslash is consumed; the character it protected is literal
              # and — crucially — cannot start a word, end one, or open a comment.
              st.cur.raw += ch
              st.cur.text += ch
              st.escaped = False
              i += 1
              continue
  
          if ch == "\\" and st.quote != "'" and command[i + 1:i + 2] == "\n":
              # Line continuation: bash removes BOTH characters. Emitting the
              # newline would split the command at a point bash never splits.
              st.cur.raw += command[i:i + 2]
              i += 2
              continue
  
          if ch == "\\" and st.quote != "'":
              if st.quote == '"' and not _double_quote_

• hooks/activity-emitter.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """activity-emitter — async Claude Code hook that journals the core's activity
  as AWP activity objects (Activity outbox Phase 2, step 1 — owner's pick
  2026-07-24, delivered via the human-action bridge's first live card).
  
  The AWP roadmap's Phase 2 is a durable Agent Activity outbox: the owner (and
  later the Workspace) should be able to see WHAT the agent is doing — session
  lifecycle, tool activity, turn completions — without attaching to tmux. Claude
  Code hooks are the structured source (NOT tmux scraping): each hook fires with
  JSON on stdin; this emitter normalizes it to an AWP activity object and appends
  it to a durable local journal. Upstream HTTP delivery is a later step (needs a
  broker /v1/activities endpoint); the journal means nothing is lost meanwhile,
  and the dashboard gets a local activity feed for free.
  
  Design (workspace notes/tasks-events/human_action_bridge_design.md §2.6):
    hook JSON → normalize → append JSONL line to
    <workspace>/state/activity-journal/YYYY-MM-DD.jsonl
  
  Hook → activity mapping (first ring, per the usecase doc):
    SessionStart        → agent.session.started
    UserPromptSubmit    → task.execution.started
    PreToolUse          → agent.tool.started
    PostToolUse         → agent.tool.completed
    PostToolUseFailure  → agent.tool.failed
    Notification        → agent.attention.required
    Stop                → task.turn.completed
    SessionEnd          → agent.session.ended
    (unknown hook)      → agent.activity (generic — forward-compatible)
  
  Attribution: if the Execution Binding Registry file exists
  (<workspace>/state/bindings/active-execution.json, written by the Core at task
  pickup), its task_id/room_id ride on every activity so downstream consumers can
  group by task. Absent registry → activities still journal, unattributed.
  
  Invariants:
    - FAIL-OPEN and FAST: any error → exit 0 silently. An emitter crash or slow
      path must never wedge or lag the core (register with "async": true).
    - Append-only JSONL, one line per activity, day-rotated by filename. No
      fsync — activities are telemetry, not decisions (the human-action store is
      the durable-by-contract one); a crash losing the tail of a telemetry
      journal is acceptable, wedging the core is not.
    - No secrets: tool_input is reduced to a short display summary, never the
      full payload (commands can carry tokens).
  
  Registration (NOT auto-registered yet — see hooks/README.md): add async
  command-hook entries for the events above pointing at this file, argv[1] =
  the hook name (SessionStart etc.) as a fallback when stdin lacks
  hook_event_name. Test: tests/activity-emitter.test.py.
  Test-only env override: SUTANDO_ACTIVITY_DIR (journal dir).
  """
  import json
  import os
  import sys
  import time
  import urllib.parse
  import uuid
  from pathlib import Path
  
  HOOK_TO_TYPE = {
      "SessionStart": "agent.session.started",
      "UserPromptSubmit": "task.execution.started",
      "PreToolUse": "agent.tool.started",
      "PostToolUse": "agent.tool.completed",
      "PostToolUseFailure": "agent.tool.failed",
      "Notification": "agent.attention.required",
      "Stop": "task.turn.completed",
      "SessionEnd": "agent.session.ended",
  }
  
  _SUMMARY_LIMIT = 160
  
  
  def _workspace() -> Path:
      """CLAUDE_CONFIG_DIR walk — same derivation as context-source-guard /
      human-action-bridge (no subprocess, no __file__ walk, deploy-safe)."""
      p = os.path.normpath(os.environ.get("CLAUDE_CONFIG_DIR")
                           or os.path.expanduser("~/.claude"))
      while True:
          if os.path.basename(p) == ".claude-sutando":
              return Path(os.path.dirname(p))
          parent = os.path.dirname(p)
          if parent == p:
              return Path(os.path.expanduser("~/sutando-workspace"))
          p = parent
  
  
  def _journal_dir(ws: Path) -> Path:
      override = os.environ.get("SUTANDO_ACTIVITY_DIR")
      return Path(override) if override else ws / "state" / "activity-journal"
  
  
  def _binding(ws: Path) -> dict:
      try:
          with open(ws / "state" / "bindings" / "active-execution.json") as f:
              b = json.load(f)
          return {k: b[k] for k in ("task_id", "room_id", "generation") if k in b}
      except (OSError, ValueError):
          return {}
  
  
  def _tool_summary(data: dict) -> dict:
      """Reduce tool info to display-safe fields — never the raw input payload."""
      name = data.get("tool_name")
      if not name:
          return {}
      ti = data.get("tool_input") or {}
      # NOTE: deliberately NOT ti["command"] — raw command lines are the
      # likeliest secret carriers; Bash calls surface via their description.
      hint = (ti.get("description") or ti.get("file_path") or ti.get("path")
              or ti.get("pattern") or "")
      if not hint and ti.get("url"):
          # URLs carry secrets in query strings/fragments (presigned sigs, OAuth
          # codes) AND in userinfo (https://user:token@host — netloc includes it,
          # so reusing netloc leaked credentials; second review). Journal
          # scheme + hostname[:port] + path ONLY, authority REBUILT from parts.
          try:
              u = urllib.parse.urlsplit(str(ti["url"]))
              host = u.hostname or ""
              authority = host + (f":{u.port}" if u.port else "")
              hint = urllib.parse.urlunsplit(
                  (u.scheme, authority, u.path, "", "")) if host else ""
          except ValueError:
              hint = ""
      hint = str(hint).splitlines()[0][:_SUMMARY_LIMIT] if hint else ""
      return {"tool": {"kind": name, **({"display": hint} if hint else {})}}
  
  
  def build_activity(data: dict, argv_hook: "str | None" = None,
                     ws: "Path | None" = None) -> dict:
      ws = ws or _workspace()
      hook = data.get("hook_event_name") or argv_hook or "unknown"
      activity = {
          "activity_id": f"act_{uuid.uuid4().hex[:12]}",
          "type": HOOK_TO_TYPE.get(hook, "agent.activity"),
          "hook": hook,
          "occurred_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
          "execution": {"runtime": "claude-code",

• hooks/comment-signature-guard.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """PreToolUse: a published comment or PR body must carry this agent's MXID.
  
  Attribution across the shared `qingyun-wu` login rests on the body signature —
  the login cannot tell two agents apart, and neither can the commit email. That
  rule was followed by discipline alone, and the discipline drifted: measured
  2026-09-03 across three PRs, 39 comments carry the older `Signed: @<mxid>` form
  and 2 the newer `— sutando-qingyun-air (@<mxid>)`. Both are fine; an unsigned
  body is not, because it is indistinguishable from the peer's.
  
  So the check is on the MXID, never the surrounding prose — matching one literal
  is what let the format drift go unnoticed for weeks.
  """
  import json
  import os
  import re
  import shlex
  import sys
  
  # No default: a node that deploys this without SUTANDO_AGENT_MXID would
  # otherwise enforce another agent's identity and deny every comment.
  MXID = os.environ.get("SUTANDO_AGENT_MXID", "")
  BODY_FLAGS = {"--body", "-b"}
  FILE_FLAGS = {"--body-file", "-F"}
  # Only subcommands that PUBLISH prose under this login. `gh pr view`, `gh api`
  # and friends carry no authored body a reader would attribute.
  PUBLISHING = (("pr", "comment"), ("issue", "comment"), ("pr", "create"), ("issue", "create"))
  EQUALS_FORM = re.compile(r"(--body|--body-file)=")
  
  
  def _is_gh(word):
      return word.rsplit("/", 1)[-1] == "gh"
  
  
  def _publishes(words):
      """`gh <a> <b>` where (a, b) is a publishing pair, allowing global flags
      between them — `gh -R o/r pr comment` publishes exactly as `gh pr comment`."""
      for i, w in enumerate(words):
          if not _is_gh(w):
              continue
          rest = words[i + 1:]
          # Adjacency, not position: a global flag TAKES A VALUE (`-R owner/repo`),
          # so dropping flags alone still leaves the value ahead of the subcommand.
          for j in range(len(rest) - 1):
              for a, b in PUBLISHING:
                  if rest[j] == a and rest[j + 1] == b:
                      return f"{a} {b}"
      return None
  
  
  def unsigned_body(command):
      """The body this command would publish, when it carries no MXID."""
      if not isinstance(command, str) or "gh" not in command:
          return None
      command = EQUALS_FORM.sub(r"\1 ", command)
      try:
          lex = shlex.shlex(command, posix=True, punctuation_chars=True)
          lex.whitespace_split = True
          words = list(lex)
      except ValueError:
          return None
      sub = _publishes(words)
      if sub is None:
          return None
      for i, w in enumerate(words):
          if w in BODY_FLAGS and i + 1 < len(words):
              if MXID not in words[i + 1]:
                  return (sub, "--body")
          if w in FILE_FLAGS and i + 1 < len(words):
              path = words[i + 1]
              try:
                  with open(path, encoding="utf-8", errors="replace") as handle:
                      text = handle.read()
              except OSError:
                  return None  # unreadable: the gate cannot answer, so it does not
              if MXID not in text:
                  return (sub, path)
      return None
  
  
  def main(argv):
      if os.environ.get("SUTANDO_ALLOW_UNSIGNED_COMMENT") == "1":
          return 0
      if not MXID:
          print("comment-signature-guard: SUTANDO_AGENT_MXID unset — not enforcing",
                file=sys.stderr)
          return 0
      try:
          payload = json.load(sys.stdin)
      except Exception:
          return 0
      if payload.get("tool_name") != "Bash":
          return 0
      found = unsigned_body((payload.get("tool_input") or {}).get("command"))
      if not found:
          return 0
      sub, where = found
      print(json.dumps({"hookSpecificOutput": {
          "hookEventName": "PreToolUse",
          "permissionDecision": "deny",
          "permissionDecisionReason": (
              f"BLOCKED: this `gh {sub}` body carries no `{MXID}`, so a reader cannot tell it "
              f"from the peer agent's — both push under the same login and the commit email is "
              f"many-to-one too. Body checked: {where}. Add a signature line containing the MXID "
              f"(either `— sutando-qingyun-air ({MXID})` or the older `Signed: @{MXID}` form; the "
              f"check is on the MXID, not the wording). Override once with "
              f"SUTANDO_ALLOW_UNSIGNED_COMMENT=1. [comment-signature-guard]"),
      }}))
      return 0
  
  
  if __name__ == "__main__":
      sys.exit(main(sys.argv[1:]))
  

• hooks/context-source-guard.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Context-source guard — PreToolUse hook enforcing the contextNotFrom rule on the
  agent's OWN channel reads (the path the gated reader + CLAUDE.md instruction can't
  force, because a raw `curl` bypasses an instruction).
  
  Rule: "if a channel is in contextNotFrom, skip its content when building context."
  A private channel's content can't be un-seen once it's in the model's context, so
  this blocks the READ before the content ever lands.
  
  SERVING-RELATIVE: the blacklist is the `contextNotFrom` of the channel currently
  being SERVED — not a global ban on the target. We learn the serving channel from
  the task the agent is processing:
    * PreToolUse[Read] of a tasks/task-*.txt        -> record its channel_id as "serving"
    * PreToolUse[Bash] that reads a task-file path  -> same (agents often `cat` the task)
    * PreToolUse[Bash] curling `…/channels/<id>/messages` -> if <id> (or its guild) is
      in the serving channel's contextNotFrom -> DENY (nothing fetched).
  So serving a private channel can still read that private channel; serving a public
  channel whose contextNotFrom lists the private guild cannot.
  
  Fail-OPEN for pre-check errors (parsing, tool detection, serving/blacklist lookup) and
  for unknown-serving-channel (interactive/diagnostic reads outside task processing).
  Once a non-empty blacklist is confirmed, any subsequent error is fail-CLOSED — never
  let an exception bypass an active guard.
  
  Deploy: copy to ~/.claude/hooks/ and register under PreToolUse for BOTH the "Bash"
  and "Read" matchers in ~/.claude/settings.json. See hooks/README.md. Config paths are
  env-overridable (SUTANDO_DISCORD_ACCESS_FILE / SUTANDO_DISCORD_ENV_FILE) for testing.
  """
  import sys
  import json
  import os
  import re
  import time
  import urllib.request
  
  # Resolve the Claude config dir via $CLAUDE_CONFIG_DIR (set by Claude Code),
  # matching read_discord_channel.py/discord-bridge's claude_home_path — NOT a
  # hardcoded ~/.claude. On a relocated install the hardcode read a DIFFERENT,
  # stale access.json than the bridge writes, so a configured contextNotFrom was
  # invisible to the hook → it silently failed OPEN (the one component you least
  # want failing open). Flagged by Sutando-Pro on PR #1698, 2026-06-18.
  # Mirrors util_paths.claude_home_path ($CLAUDE_CONFIG_DIR -> $CLAUDE_HOME -> ~/.claude); standalone hook.
  _CFG = os.environ.get("CLAUDE_CONFIG_DIR") or os.environ.get("CLAUDE_HOME") or os.path.expanduser("~/.claude")
  ACCESS_FILE = os.environ.get("SUTANDO_DISCORD_ACCESS_FILE",
                               os.path.join(_CFG, "channels", "discord", "access.json"))
  ENV_FILE = os.environ.get("SUTANDO_DISCORD_ENV_FILE",
                            os.path.join(_CFG, "channels", "discord", ".env"))
  # Workspace resolution (M0/#1440 residual, #1698): SUTANDO_WORKSPACE was dropped
  # for resolution in v0.8 and its old default was the pre-M0 legacy home-dir
  # location — so this hook stranded active-serving-channel.json in the LEGACY
  # workspace, a different dir than discord-bridge.py (M0 resolve_workspace) reads,
  # and the guard then couldn't find the serving channel → silently failed open.
  # Derive the workspace from CLAUDE_CONFIG_DIR instead: the Claude Code project
  # tree lives at `<workspace>/.claude-sutando` per the workspace contract, so the
  # parent of _CFG IS the workspace (and it moves with a config-relocated
  # workspace). No subprocess — this hook fires on every Bash/Read; no __file__
  # walk — that's the bundled-symlink anti-pattern in src/workspace_default.py.
  # Walk up to the nearest `.claude-sutando` ANCESTOR (not just an exact-leaf
  # basename match): src/startup.sh floats narrowing CLAUDE_CONFIG_DIR to a per-host
  # subdir (`<workspace>/.claude-sutando/hosts/<host>`); there the leaf is the
  # hostname and an exact match would wrongly fall through to the fallback.
  _cfg_norm = os.path.normpath(_CFG)
  WS = None
  _p = _cfg_norm
  while True:
      if os.path.basename(_p) == ".claude-sutando":
          WS = os.path.dirname(_p)
          break
      _parent = os.path.dirname(_p)
      if _parent == _p:  # reached filesystem root, no `.claude-sutando` ancestor
          break
      _p = _parent
  if WS is None:
      # CLAUDE_CONFIG_DIR unset (→ _CFG=~/.claude, no `.claude-sutando` ancestor) or
      # otherwise unresolvable. Fail open to the SAME canonical last-ditch the rest of
      # the system uses — workspace_default.default_workspace_dir() returns
      # ~/sutando-workspace (_DEFAULT_SUBPATH=("sutando-workspace",)). NOT ~/.sutando
      # (pre-v0.8). Not imported here: this hook is standalone-deployed to ~/.claude/hooks/.
      WS = os.path.expanduser("~/sutando-workspace")
  STATE = os.path.join(WS, "state", "active-serving-channel.json")
  API = "https://discord.com/api/v10"
  UA = "DiscordBot (https://github.com/sonichi/sutando, 1.0)"
  _TASK_RE = re.compile(r"task-\d+\.txt$")
  _TASKPATH_RE = re.compile(r"([^\s'\"]*task-\d+\.txt)")  # a task-file path inside a Bash command
  _CH_READ_RE = re.compile(r"channels/(\d+)/messages")
  _GUILD_CACHE = os.path.join(WS, "state", ".channel-guild-cache.json")
  
  
  def _read_channel_id_from_task(path):
      try:
          with open(path, encoding="utf-8", errors="replace") as f:
              for line in f:
                  if line.startswith("channel_id:"):
                      return line.split(":", 1)[1].strip()
      except Exception:
          pass
      return None
  
  
  def _record_serving(cid):
      try:
          os.makedirs(os.path.dirname(STATE), exist_ok=True)
          with open(STATE, "w") as f:
              json.dump({"channel_id": str(cid), "ts": int(time.time())}, f)
      except Exception:
          pass
  
  
  def _active_serving():
      try:
          with open(STATE) as f:
              return str(json.load(f).get("channel_id") or "") or None
      except Exception:
          return None
  
  
  def _blacklist(serving_cid):
      try:
          data = json.load(open(ACCESS_FILE))
          grp = data.get("groups", {}).get(str(serving_cid))
          if isinstance(grp, dict):
              return {str(c) for c in (grp.get("contextNotFrom") or []

• hooks/dedup-staging-guard.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """PreToolUse: gate a Bash `mv` that lands a dedup-staged result into `results/`
  on `skills/proactive-loop/scripts/check-dedup-targets.py`, for ANY caller — not
  just proactive-loop step 1's own checklist.
  
  WHY THIS EXISTS. Step 1 stages a grouped `[deduped: X]` reply under
  `state/dedup-staging/<file>` and only promotes it with:
  
    S="$WORKSPACE/state/dedup-staging/<file>"
    python3 skills/proactive-loop/scripts/check-dedup-targets.py "$S" && mv -f "$S" "$WORKSPACE/results/<file>"
  
  `check-dedup-targets.py` refuses (exit 1) a staged file whose dedup target
  resolves to nothing — `[no-send]` or absent — because the bridge would then
  tell the room "see task X" for an X that delivers nothing. That refusal only
  holds if the `&&` chain is actually typed. A bare `mv` of a dedup-staged file
  into `results/`, run from any skill or any live session with the chain
  dropped, bypasses it completely — same architectural gap as gh-policy-gate.py
  and memory-index-guard.py, and the same fix: move enforcement to the action.
  
  WHAT COUNTS AS THE ADDITION. A PreToolUse hook sees `tool_input.command` as
  the RAW, unexpanded shell text: `$S` is a literal variable reference, not the
  path it will resolve to, so this hook cannot resolve it — building real shell
  variable tracking is out of scope (fragile, and unlike a `gh`/`Edit`/`Write`
  call, a Bash command carries no resolved value the hook can read directly).
  Instead it matches literal substrings the way the skill's own prose documents
  the pattern (`state/dedup-staging/` -> `results/`): a command (a) invokes `mv`
  in some segment whose arguments mention `results` (the destination `mv -f
  "$S" ".../results/<file>"` always writes literally, even when the SOURCE is
  hidden behind `$S`), (b) the command as a whole — any segment, not just the
  `mv`'s own — mentions `dedup-staging` (usually the earlier `S=...` assignment,
  `;`-separated from the `mv`), and (c) NO segment anywhere in the same command
  invokes `check-dedup-targets.py`. `&&` splits into a SEPARATE segment from
  what precedes it (verified against `_shell_scan.segments()` directly, not
  assumed — the checker call and the `mv` it gates are almost always in
  different segments, unlike gh-policy-gate's same-segment `gh` matches), so (b)
  and (c) are scanned across the WHOLE command rather than per segment; only (a)
  stays segment-scoped, since a `results`-mentioning `mv` that has nothing to do
  with dedup staging (no `dedup-staging` mention anywhere in the command) is not
  this pattern.
  
  FAILS OPEN ON UNCERTAINTY, DENIES ONLY ON A POSITIVE FINDING. A command
  mentioning only one of `dedup-staging` / `results`, or no `mv` at all, is
  allowed — same contract as gh-policy-gate.py and memory-index-guard.py. This
  hook only sees ONE Bash command string at a time: a multi-tool-call sequence
  (`S=...` in one call, the check+mv in the next) is outside what any single
  PreToolUse invocation can observe, the same scope limit the other two accept.
  
  Tokenizes via the shared `_shell_scan` scanner, same reasoning as
  gh-policy-gate.py's own docstring for why `shlex`/lookbehind under-denies.
  """
  import json
  import os
  import sys
  from pathlib import Path
  
  sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
  import _shell_scan  # noqa: E402  (sibling module; path set above)
  
  _SOURCE_MARKER = "dedup-staging"
  _DEST_MARKER = "results"
  _CHECKER_MARKER = "check-dedup-targets.py"
  
  
  def _mv_segments(command):
      """Each `mv`-invoking segment of `command`, as (words-after-mv) texts."""
      if not isinstance(command, str) or "mv" not in command:
          return []
      out = []
      for seg in _shell_scan.segments(command):
          for i, w in enumerate(seg):
              if w.basename_is("mv"):
                  out.append([x.text for x in seg[i + 1:]])
                  break
      return out
  
  
  def _flat_word_texts(command):
      """Every non-operator word's text across the WHOLE command, all segments —
      used for the two checks that must see past the `&&`/`;` that splits the
      staging assignment, the checker call, and the `mv` into separate
      segments (see module docstring)."""
      if not isinstance(command, str):
          return []
      return [w.text for w in _shell_scan.words(command) if not w.is_operator]
  
  
  def check_dedup_staging_bypass(command):
      """Returns a deny reason string, or None to allow."""
      mv_segs = _mv_segments(command)
      if not mv_segs:
          return None
      dest_segs = [seg for seg in mv_segs if any(_DEST_MARKER in w for w in seg)]
      if not dest_segs:
          return None
      all_words = _flat_word_texts(command)
      if not any(_SOURCE_MARKER in w for w in all_words):
          return None
      if any(_CHECKER_MARKER in w for w in all_words):
          return None
      seg_text = " ".join(dest_segs[0])
      return (f"`mv {seg_text}` moves a dedup-staged file into results/ with no "
              f"check-dedup-targets.py call anywhere in the command")
  
  
  def main(argv):
      if os.environ.get("SUTANDO_ALLOW_UNGATED_DEDUP_STAGING") == "1":
          return 0
      try:
          payload = json.load(sys.stdin)
      except Exception:
          return 0
      if payload.get("tool_name") != "Bash":
          return 0
      reason = check_dedup_staging_bypass((payload.get("tool_input") or {}).get("command"))
      if not reason:
          return 0
      print(json.dumps({"hookSpecificOutput": {
          "hookEventName": "PreToolUse",
          "permissionDecision": "deny",
          "permissionDecisionReason": (
              f"BLOCKED: {reason}. Chain "
              f"skills/proactive-loop/scripts/check-dedup-targets.py \"$S\" && mv ... first. "
              f"Override once with SUTANDO_ALLOW_UNGATED_DEDUP_STAGING=1. "
              f"[dedup-staging-guard]"),
      }}))
      return 0
  
  
  if __name__ == "__main__":
      sys.exit(main(sys.argv[1:]))
  

• hooks/gmail-write-guard.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """gmail-write-guard — PreToolUse hook that denies the claude.ai Gmail MCP
  connector's WRITE-scoped tools and routes writes to the IMAP/SMTP path.
  
  Why (field report 05cb849a, michael@actoneventures.com, 2026-07-13): every
  Gmail WRITE operation through the claude.ai connector is unreliable or broken
  while reads work fine —
  
    * ``create_draft`` caused 7 documented incidents over ~5 weeks on one
      install (drafts not reflecting what actually gets sent — one
      wrong-recipient send), and the connector exposes no delete-draft tool, so
      each cleanup needed raw IMAP ``UID SEARCH`` + ``STORE \\Deleted`` +
      ``EXPUNGE``.
    * ``label_thread`` (and the other label/unlabel tools) fail outright with
      "Request had insufficient authentication scopes" — the connector's OAuth
      flow doesn't actually grant the Gmail write scopes it needs, even when
      read tools (search_threads / get_thread / list_labels) work.
  
  Nothing in the tools' own descriptions warns about this; an install can only
  discover it by getting burned. This hook is the generalized version of the
  per-install block Michael built: deny the connector's Gmail write tools BEFORE
  they run, with a reason that points the model at the app-password IMAP/SMTP
  path (see docs/built-in-tools.md → Email) that actually works.
  
  Scope — deliberately narrow:
    * Only MCP tools (``mcp__…``) whose server/tool name mentions gmail.
    * Only WRITE verbs (create/send/label/unlabel/delete/trash/archive/modify/
      update/move/apply/remove/mark). Read tools — search_threads, get_thread,
      list_labels, get_message, … — pass through untouched (they work fine).
    * Non-Gmail tools: no-op (exit 0), safe to register under a broad matcher.
  
  Escape hatch: set ``SUTANDO_ALLOW_GMAIL_CONNECTOR_WRITES=1`` to disable the
  guard (e.g. if/when the connector's OAuth scopes are fixed upstream).
  
  Fail-OPEN on any error — a crashing hook must never wedge the core (same
  contract as skip-ask-user-question.py).
  
  Registration: manual per-node deploy like context-source-guard.py — see
  hooks/README.md.
  """
  import json
  import os
  import sys
  
  # Write-verb tokens. Matched against the '_'-split tokens of the MCP tool's
  # trailing tool-name segment, so `list_labels` (token "labels") stays allowed
  # while `label_thread` / `create_label` / `unlabel_thread` are denied.
  WRITE_TOKENS = {
      "create", "send", "delete", "trash", "archive", "modify", "update",
      "move", "apply", "remove", "mark", "label", "unlabel", "insert",
      "batchmodify", "batchdelete", "untrash",
  }
  
  REASON = (
      "Gmail writes through the claude.ai MCP connector are blocked on this install: "
      "the connector's write scopes are broken/unreliable (label/archive fail with "
      "'insufficient authentication scopes'; create_draft has a documented history of "
      "drafts not matching what gets sent, incl. a wrong-recipient send — field report "
      "05cb849a). Gmail READS through the connector are fine and remain allowed. "
      "For this write, use the app-password IMAP/SMTP path instead (docs/built-in-tools.md "
      "-> Email: scripts using imaplib/smtplib with the vaulted app password). "
      "If the connector's OAuth scopes get fixed upstream, set "
      "SUTANDO_ALLOW_GMAIL_CONNECTOR_WRITES=1 to lift this guard. [gmail-write-guard]"
  )
  
  
  def is_gmail_connector_write(tool_name: str) -> bool:
      """True only for MCP Gmail tools whose name carries a write verb."""
      if not tool_name.startswith("mcp__"):
          return False
      lowered = tool_name.lower()
      if "gmail" not in lowered:
          return False
      # Trailing segment = the tool itself (server names also split on __).
      tool_part = lowered.rsplit("__", 1)[-1]
      tokens = set(tool_part.split("_"))
      return bool(tokens & WRITE_TOKENS)
  
  
  def main() -> None:
      if os.environ.get("SUTANDO_ALLOW_GMAIL_CONNECTOR_WRITES", "").strip() == "1":
          sys.exit(0)
      data = json.loads(sys.stdin.read())
      tool_name = str(data.get("tool_name") or "")
      if is_gmail_connector_write(tool_name):
          print(json.dumps({"hookSpecificOutput": {
              "hookEventName": "PreToolUse",
              "permissionDecision": "deny",
              "permissionDecisionReason": REASON,
          }}))
      # Everything else (and the deny above) exits 0; PreToolUse only blocks
      # when a deny decision is present in the JSON payload.
      sys.exit(0)
  
  
  if __name__ == "__main__":
      try:
          main()
      except SystemExit:
          raise
      except Exception as e:  # fail-open: never wedge the core on a hook error
          print(f"[gmail-write-guard] non-fatal error, allowing: {e}", file=sys.stderr)
          sys.exit(0)
  

• hooks/hitl-hook-driver.pyGitHub ↗
• hooks/human-action-bridge.pyGitHub ↗
• hooks/inline-body-substitution-guard.pyGitHub ↗
• hooks/memory-index-guard.pyGitHub ↗
• hooks/release-target-guard.pyGitHub ↗
• hooks/result-file-marker-guard.pyGitHub ↗
• hooks/review-authority-guard.pyGitHub ↗
• hooks/skill-usage-telemetry.pyGitHub ↗
• hooks/skip-ask-user-question.pyGitHub ↗

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