Hooks

Where it lives

• 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/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/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/human-action-bridge.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """human-action-bridge — PreToolUse hook that turns `AskUserQuestion` into a
  REMOTE question instead of a hard deny (human-action bridge v1, step 1).
  
  Today `hooks/skip-ask-user-question.py` denies AskUserQuestion outright because
  the core runs headless — a rendered prompt would hang the session forever. That
  protects the session but lobotomizes the interaction: every question the model
  judged worth asking is silently skipped.
  
  This hook upgrades deny → remote-ask:
  
      AskUserQuestion fires
        → durable pending-action file      (<workspace>/state/human-actions/)
        → card file for the owner         (results/proactive-ha-<id>.txt — the
                                            sanctioned proactive path; the channel
                                            bridge delivers it to the owner)
        → bounded wait, polling the action file for a decision
        → decision arrives  → allow + updatedInput carrying the owner's answers
                              (Claude continues as if answered locally)
        → timeout / no path → deny with the same decide-autonomously guidance as
                              skip-ask-user-question (EXACTLY today's behavior)
  
  Decisions are written into the action file by whoever holds the return path:
  the sparrow DecisionHandler (v1 step 3) when SPARROW_EVENTS is live, or the
  core itself when the owner's reply arrives as a normal task (works today).
  A decision is only honored while `status` is "pending"; terminal states are
  immutable and late answers are ignored by the hook (the writer may still record
  them for audit).
  
  Safety invariants (design: notes/tasks-events/human_action_bridge_design.md):
    - timeout NEVER approves — no response ⇒ deny-with-reason, never consent
    - fail-OPEN for the session (any hook error ⇒ exit 0 allow-passthrough;
      a crashing hook must never wedge the core) but fail-CLOSED for the
      decision (only an explicit resolved decision produces an allow)
    - every state transition is stamped in the action file for audit
  
  Registration: PreToolUse with matcher "AskUserQuestion", INSTEAD OF
  skip-ask-user-question.py (it subsumes it — the timeout branch IS that hook).
  See hooks/README.md. Test-only env overrides (documented here, used by
  tests/human-action-bridge.test.py): SUTANDO_HA_DIR (action-store dir),
  SUTANDO_HA_CARD_DIR (card dir), SUTANDO_HA_TIMEOUT (seconds, default 120),
  SUTANDO_HA_POLL (seconds, default 2).
  """
  import fcntl
  import hashlib
  import json
  import os
  import sys
  import time
  import uuid
  from pathlib import Path
  
  TOOL = "AskUserQuestion"
  
  # Timeout branch = the exact guidance skip-ask-user-question ships today, so the
  # fallback behavior is indistinguishable from the current hook.
  TIMEOUT_REASON = (
      "AskUserQuestion could not be answered remotely in time: Sutando's core agent "
      "runs headless and the owner did not respond to the question card within the "
      "window. Do NOT ask again — decide autonomously: pick the option you judge "
      "best, or state a clear assumption and proceed. If a choice is genuinely "
      "blocking AND irreversible, surface the question through a normal channel "
      "(per-host pending-questions.md or an owner notification) and keep working "
      "on other things. [human-action-bridge:{action_id}]"
  )
  
  
  def _workspace() -> Path:
      """Derive the workspace from CLAUDE_CONFIG_DIR — the Claude Code project
      tree lives at `<workspace>/.claude-sutando` per the workspace contract, so
      the nearest `.claude-sutando` ancestor's parent IS the workspace. Same
      pattern as context-source-guard.py: no subprocess (hooks are hot-path) and
      no __file__ walk (the bundled-symlink anti-pattern; also keeps this hook
      standalone-deployable). Falls back to the system's canonical last-ditch
      default, ~/sutando-workspace."""
      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:  # filesystem root — no `.claude-sutando` ancestor
              return Path(os.path.expanduser("~/sutando-workspace"))
          p = parent
  
  
  def _store_dir() -> Path:
      override = os.environ.get("SUTANDO_HA_DIR")
      return Path(override) if override else _workspace() / "state" / "human-actions"
  
  
  def _card_dir() -> Path:
      override = os.environ.get("SUTANDO_HA_CARD_DIR")
      return Path(override) if override else _workspace() / "results"
  
  
  def _atomic_write(path: Path, payload: dict) -> None:
      """Durable atomic write: unique per-writer temp (the resolver may write the
      same action concurrently — a fixed name could be clobbered mid-write),
      fsync the data before rename and the directory entry after, so a crash can
      never lose pending-action state (review blocker)."""
      tmp = path.parent / f"{path.name}.{os.getpid()}.{uuid.uuid4().hex[:8]}.tmp"
      with open(tmp, "w") as f:
          f.write(json.dumps(payload, ensure_ascii=False, indent=1))
          f.flush()
          os.fsync(f.fileno())
      os.replace(tmp, path)
      dfd = os.open(path.parent, os.O_RDONLY)
      try:
          os.fsync(dfd)
      finally:
          os.close(dfd)
  
  
  def _transition_lock(path: Path):
      """flock shared by EVERY writer of an action file — this hook's expiry path
      AND the sparrow resolver (ActionStore.transition_lock uses the same
      `<action_id>.lock` file). Serializes read→check→write so exactly one
      terminal transition wins (review blocker: the timeout could overwrite a
      decision that landed between its read and its write)."""
      lock_file = open(path.parent / (path.stem + ".lock"), "a+")
      fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
      return lock_file
  
  
  def _new_action(data: dict) -> dict:
      tool_input = data.get("tool_input") or {}
      questions = tool_input.get("questions") or []
      now = time.time()
      digest = hashlib.sha1(
          (json.dumps(qu

• hooks/result-file-marker-guard.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """result-file-marker-guard — PreToolUse hook that DENIES writing a result body
  whose ``[file:|send:|attach:]`` marker points outside the send allowlist **for
  the adapter that will actually deliver it**.
  
  Why (owner incident 2026-08-04, #susan): the agent finished a 6-minute video,
  wrote ``[file: …/skill-repos/video-production/…/talk.mp4]`` into a result, and
  reported the task delivered. ``skill-repos/`` is not on the allowlist
  (``src/send_allowlist.py``), so the bridge posted a literal
  ``(file not allowed: /Users/…/talk.mp4)`` into the owner's channel and delivered
  nothing. The owner found it, not the agent:
  
      "Can't see this file. And I don't want to babysit. Can you improve"
  
  **The failure reports success in every cheap way available to the author.** The
  file exists, the path is absolute and correct, the marker regex matches, the
  Write succeeds, the result file lands, and the bridge consumes it and archives
  the task — so every signal the agent normally checks says delivered. The
  allowlist is enforced at the *far* end, after the last point the agent looks.
  
  So the check belongs at the moment the marker is AUTHORED, and it must be a
  mechanism rather than a discipline
  (``feedback_guarantee_is_structural_not_disciplinary``).
  
  Scope — deliberately narrow:
    * Only Write/Edit/MultiEdit whose target resolves under ``<workspace>/results/``.
    * Only bodies containing an attachment marker.
    * Parsing and the verdict come from the SAME modules the delivery path uses
      (``result_markers`` + ``send_allowlist``), so the guard cannot drift from
      the policy it enforces. A re-implemented copy would eventually accept what
      the bridge rejects, and that false PASS is worse than no guard.
  
  ADAPTER CONTEXT (qingyun-wu, PR #2596 review). The allowlist is not global:
  Slack deliberately extends it with its adapter-local ``<workspace>/slack-inbox/``
  so an uploaded file can be echoed back (``src/slack-bridge.py:153-158``).
  Judging every result against the canonical Discord/Telegram policy would deny a
  currently-supported Slack reply. So the guard resolves the DESTINATION first —
  ``results/task-<id>.txt`` names the task, and the task file's ``source:`` field
  names the adapter — and applies that adapter's policy. When the destination
  cannot be established (a proactive body, or a task archived out from under us)
  it falls back to the CANONICAL roots only, never the union — see the comment at
  that branch for why the union was unsound and why the routing state cannot
  recover the answer.
  
  REPO ROOT (bassilkhilo-ag2 + qingyun-wu, PR #2596 review). This file needs
  ``src/`` on ``sys.path``. It must NOT discover that by walking up from
  ``__file__``: the deploy step copies the hook out of the checkout, so the walk
  resolves to the deploy dir, and the repo bans that pattern outright
  (``scripts/lint-workspace-resolution.sh`` — it breaks under symlinked/bundled
  layouts). The location is therefore CONFIGURED, not guessed: ``--repo <path>``
  (written by the registration snippet in hooks/README.md) or
  ``$SUTANDO_REPO_ROOT``. If neither resolves, the hook says so **loudly on
  stderr** and allows — the v1 of this hook exited silently, which made an
  unresolvable root indistinguishable from a clean pass, i.e. exactly the
  "reports success in every cheap way" defect it exists to prevent.
  
  Escape hatch: ``SUTANDO_SKIP_FILE_MARKER_GUARD=1``.
  
  Fail-OPEN on any internal error — a crashing hook must never wedge the core
  (same contract as ``skip-ask-user-question.py`` / ``gmail-write-guard.py``).
  Note the asymmetry with ``context-source-guard.py``, which fails CLOSED: that
  one prevents blacklisted content entering context, where the cost of being
  wrong is a leak. Here the cost of being wrong is a message the owner can see
  and re-request, so wedging the core would be the larger harm.
  
  Registration: manual per-node deploy — see hooks/README.md.
  """
  import json
  import os
  import re
  import sys
  
  WRITE_TOOLS = {"Write", "Edit", "MultiEdit", "NotebookEdit"}
  
  # Adapter -> extra roots beyond the canonical allowlist, relative to the
  # workspace. Mirrors what each bridge passes to is_path_sendable(extra_roots=…).
  # Keep in step with the bridges; a bridge that adds a root and not an entry here
  # gets a FALSE DENY, which the tests below are meant to make loud.
  ADAPTER_EXTRA_ROOTS = {
      "slack": ("slack-inbox",),          # src/slack-bridge.py:153-158
      "discord": (),
      "telegram": (),
  }
  
  
  def _warn(msg):
      print(f"[result-file-marker-guard] {msg}", file=sys.stderr)
  
  
  def _repo_root(argv):
      """CONFIGURED, never guessed — see the module docstring."""
      for i, a in enumerate(argv):
          if a == "--repo" and i + 1 < len(argv):
              return argv[i + 1]
          if a.startswith("--repo="):
              return a.split("=", 1)[1]
      return os.environ.get("SUTANDO_REPO_ROOT") or None
  
  
  def _body_from(tool_name, ti):
      """The text this call would put on disk, across the edit-tool shapes."""
      if tool_name == "Write":
          return ti.get("content") or ""
      if tool_name == "Edit":
          return ti.get("new_string") or ""
      if tool_name == "MultiEdit":
          return "\n".join(e.get("new_string") or "" for e in (ti.get("edits") or []))
      if tool_name == "NotebookEdit":
          return ti.get("new_source") or ""
      return ""
  
  
  def _adapter_for(result_path, workspace):
      """The bridge that will deliver this result, from the task it answers.
  
      `results/task-<id>.txt` -> `tasks/task-<id>.txt` (or its archive) -> `source:`.
      Returns None when it can't be established; the caller then uses the
      CANONICAL roots only (see that branch), never a union of provider-local ones.
      """
      m = re.match(r"^(?:[^.]+\.)?task-(.+)\.txt$", os.path.basename(result_path))
      if not m:
          return None
      tid = m.group(1)
      for rel in (f"tasks/task-{tid}.txt", f"tasks/archive/task-{tid}.txt"):
          p = os.path.join(str(workspace), rel)
          try:
              with open(p, encoding="utf-8", errors="rep

• hooks/skill-usage-telemetry.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Skill-usage telemetry — PostToolUse[Skill] hook.
  
  Emits ONE anonymous ``feature_used {feature: "skill:<name>"}`` product-telemetry
  event every time the core invokes a skill (the `Skill` tool). This is the
  chokepoint that broadens feature-usage coverage from the two hand-instrumented
  scripts (morning-briefing, daily-insight) to the *entire* skill surface without
  touching each skill — the loop already runs dozens of skills (proactive-loop,
  people-analysis, context-reconstruct, session-recap, task-orphan-check, …) and
  none of them reported until now.
  
  Why a hook, not per-skill calls: skills are markdown + scripts of every shape;
  wiring `feature_used()` into each is unmaintainable and misses future skills. A
  single PostToolUse matcher on the `Skill` tool captures all of them, for free,
  the moment they run.
  
  Privacy: sends ONLY the skill's short categorical name (prefixed ``skill:``).
  Never the skill arguments, task content, prompts, or any PII — same contract as
  `telemetry.feature_used`. Honors the same opt-out (DO_NOT_TRACK / telemetry-
  disabled) because it routes through `telemetry.capture()`, which checks opt-out
  on every call.
  
  Fail-OPEN, ALWAYS. Telemetry must never break a tool: any error (bad stdin,
  missing telemetry module, network) is swallowed and the hook exits 0 with no
  output. A PostToolUse observability hook has no decision to make — it observes.
  
  Registration: ``start-cli.sh`` passes this repo file's absolute path to
  ``build-core-settings.mjs``, which registers it under PostToolUse for the
  ``Skill`` matcher independently of the observability opt-in. Repo root is found
  from this file's location, or via ``$SUTANDO_REPO_ROOT`` for tests.
  """
  import sys
  import os
  import json
  
  
  def _repo_src() -> str:
      """Locate the repo's src/ dir (telemetry.py lives there)."""
      override = os.environ.get("SUTANDO_REPO_ROOT")
      if override:
          return os.path.join(override, "src")
      # hooks/ is a sibling of src/ in the repo; from the deployed copy in
      # ~/.claude/hooks we fall back to $SUTANDO_REPO_ROOT (set at install).
      here = os.path.dirname(os.path.abspath(__file__))
      return os.path.join(os.path.dirname(here), "src")
  
  
  def main() -> int:
      # Fail-open around EVERYTHING — never let telemetry break the tool.
      try:
          raw = sys.stdin.read()
          if not raw.strip():
              return 0
          payload = json.loads(raw)
  
          # Only the Skill tool is a "feature use" for this hook.
          if payload.get("tool_name") != "Skill":
              return 0
  
          tool_input = payload.get("tool_input") or {}
          skill = tool_input.get("skill")
          if not skill or not isinstance(skill, str):
              return 0
          # Guard against unbounded/garbage names getting into the property space.
          skill = skill.strip().lstrip("/")[:64]
          if not skill:
              return 0
  
          # Emit via a DETACHED subprocess, never in-process. This hook is
          # registered for EVERY Skill call, so it must return promptly: doing the
          # send here — even the bounded 1s flush path — would add a network RTT to
          # the tool run and compound across skill-heavy loops (CR #2254,
          # qingyun-wu). Instead we spawn `telemetry.py feature_used skill:<name>`
          # in its own session and DO NOT wait for it. The child (which the CLI
          # runs on the flush path, since it exits immediately) does the POST off
          # this hook's critical path; the hook forks-and-returns in ~ms.
          telemetry_py = os.path.join(_repo_src(), "telemetry.py")
          if not os.path.isfile(telemetry_py):
              return 0
          import subprocess
          subprocess.Popen(
              [sys.executable, telemetry_py, "feature_used", f"skill:{skill}"],
              stdin=subprocess.DEVNULL,
              stdout=subprocess.DEVNULL,
              stderr=subprocess.DEVNULL,
              start_new_session=True,  # detach: outlive this hook, don't get reaped with it
          )
      except Exception:
          # Swallow — observability must never surface an error to the tool run.
          pass
      return 0
  
  
  if __name__ == "__main__":
      sys.exit(main())
  

• hooks/skip-ask-user-question.pyGitHub ↗

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.