Hooks

Where it lives

• hooks/activity-journal-guard.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """PreToolUse: ENFORCE the shared activity journal when mutating a tracked shared resource.
  
  General, config-driven version of the "force journaling" pattern (see
  rules/activity-journal-and-state-registry.md). When a Bash/PowerShell command reaches a
  configured shared resource (ssh/scp/rsync to one of its targets) and MUTATES state, and the
  command does NOT call that resource's journal and carries no `# claude-bypass: journal` marker,
  the command is BLOCKED until a journal call is added.
  
  Why: discipline decays under context pressure; an append-only journal + state registry only
  work if every mutation is logged. A mechanical guard keeps the journal complete so that
  "what is running, who started it, why" is always answerable.
  
  Config (so this is reusable, not tied to any one host) — first match wins:
      ~/.claude/activity-journal.config.json
      {
        "resources": [
          {
            "name": "mygpu",                       # shown in the block message
            "targets": ["myhost", "10.0.0.5"],     # ssh/scp destinations that mean "this resource"
            "journal_marker": "journal.py",        # substring proving the cmd logs (regex-free)
            "log_hint": "ssh myhost 'journal.py log <proj> \"what\" \"detail\"; <cmd>'"
          }
        ]
      }
  
  INERT BY DEFAULT: with no config file (or empty `resources`), this hook allows everything —
  shipping it in a shared config never blocks anyone who has not opted in. For a host-specific
  instance, fork this with your targets hardcoded (same pattern, fixed targets).
  
  Bypass: `# claude-bypass: journal` in the command, or CLAUDE_ALLOW_NO_JOURNAL=1.
  Fail-OPEN on any error (a hook bug must never block every command).
  """
  from __future__ import annotations
  
  import json
  import os
  import re
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent))
  from safety_common import allow, bash_command, block, bypass, read_event  # noqa: E402
  
  CONFIG_PATH = Path.home() / ".claude" / "activity-journal.config.json"
  
  SSH_TOKEN_RE = re.compile(r"(?<![a-z])(ssh|scp|rsync)(?![a-z])", re.IGNORECASE)
  
  # Tokens that indicate the remote/shared command CHANGES state (worth logging).
  MUTATING_RE = re.compile(
      r"\b(docker\s+(stop|start|restart|rm|run|kill|update|compose)|"
      r"rm\s|rmdir|pkill|killall|\bkill\s|mv\s|nohup|setsid|systemctl\s+(start|stop|restart|disable|enable)|"
      r"sed\s+-i|tee\b|chmod|chown|crontab|cryptsetup|mkfs|dd\s+if=|truncate|"
      r"createdb|dropdb|psql|mysql|>\s*/|>>\s*/)",
      re.IGNORECASE,
  )
  
  
  def _load_resources() -> list[dict]:
      try:
          if not CONFIG_PATH.exists():
              return []
          data = json.loads(CONFIG_PATH.read_text(encoding="utf-8"))
          res = data.get("resources", [])
          return res if isinstance(res, list) else []
      except (OSError, json.JSONDecodeError, ValueError):
          return []
  
  
  def main() -> None:
      if os.environ.get("CLAUDE_ALLOW_NO_JOURNAL") == "1":
          allow()
  
      resources = _load_resources()
      if not resources:
          allow()  # inert until configured
  
      event = read_event()
      if event.get("tool_name") not in ("Bash", "PowerShell"):
          allow()
  
      cmd = bash_command(event.get("tool_input", {}))
      if not cmd or not SSH_TOKEN_RE.search(cmd):
          allow()
  
      low = cmd.lower()
  
      # Find which configured resource this command targets (first match wins).
      hit = None
      for r in resources:
          targets = [str(t).lower() for t in r.get("targets", []) if t]
          if any(t in low for t in targets):
              hit = r
              break
      if hit is None:
          allow()  # not a tracked resource
  
      marker = str(hit.get("journal_marker", "journal.py")).lower()
      if marker and marker in low:
          allow()  # already journaling
      if bypass("journal", cmd):
          allow()
  
      if not MUTATING_RE.search(cmd):
          allow()  # read-only, no need to log
  
      name = hit.get("name", "shared resource")
      hint = hit.get("log_hint", "add a journal log call in the same command")
      block(
          f"📓 activity journal — BLOCKED: state-changing command on '{name}' with no journal entry.\n"
          f"The shared journal + state registry are how anyone sees what is running / who started it.\n"
          f"Add the journal call to THIS command, e.g.:\n  {hint}\n"
          f"Read-only commands pass. One-off / not worth logging → bypass: "
          f"`# claude-bypass: journal` + `# Reason: ...`. (rules/activity-journal-and-state-registry.md)"
      )
  
  
  if __name__ == "__main__":
      try:
          main()
      except Exception:
          # Fail-open: never let a guard bug block every command.
          allow()
  

• hooks/agent-skill-contract.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Pre/PostToolUse delegation: bind a skill/evidence contract to each child.
  
  The semantic loader can select a skill for a parent request, but a child task
  starts with a separate prompt.  A coordinator therefore renders this compact,
  task-bound contract before delegation.  The hook validates the exact schema
  and binding at Claude Code's ``Task`` boundary.
  
  It intentionally does *not* apply keyword matching to arbitrary child prose:
  quoted data and translations can contain a trigger phrase.  The CLI performs
  the curated routing before dispatch; the hook proves that the resulting,
  task-bound contract was carried across the boundary.  Every child task gets a
  contract, including an explicit ``no-high-confidence-match`` result.
  
  Codex exposes ``spawn_agent`` through the local-function hook path (matcher
  alias ``Agent``).  The pre-hook adds or validates the exact contract, then the
  post-hook binds that route to the returned ``agent_id`` for SubagentStop.
  """
  from __future__ import annotations
  
  import argparse
  import hashlib
  import importlib.util
  import json
  import re
  from dataclasses import dataclass
  from pathlib import Path
  
  from safety_common import allow, block, log, read_event
  from skill_contract_state import bind_agent, save_pending
  
  
  def _load_router():
      router_path = Path(__file__).resolve().with_name("keyword-skill-router.py")
      spec = importlib.util.spec_from_file_location("keyword_skill_router", router_path)
      if spec is None or spec.loader is None:
          raise RuntimeError(f"cannot load skill router from {router_path}")
      module = importlib.util.module_from_spec(spec)
      spec.loader.exec_module(module)
      return module
  
  
  detect_keywords = _load_router().detect_keywords
  
  
  CONTRACT_VERSION = "4"
  CONTRACT_OPEN = f'<agent-skill-contract version="{CONTRACT_VERSION}">'
  CONTRACT_CLOSE = "</agent-skill-contract>"
  CONTRACT_RE = re.compile(
      r'<agent-skill-contract version="(?P<version>[0-9]+)">'
      r"(?P<body>.*?)" + re.escape(CONTRACT_CLOSE),
      re.DOTALL,
  )
  SKILL_LINE_RE = re.compile(r"^- ([a-z0-9][a-z0-9:_-]*)$")
  SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
  ROUTING_SOURCE = "keyword-skill-router-v1"
  SKILL_GAP_CHECKLIST = "skills/agent-harness-design/references/agent-skill-install-checklist.md"
  SKILL_OPT_OUT_RE = re.compile(
      r"(?i)^(?:"
      r"(?:do\s+not|don't)\s+(?:use|load|invoke)\s+(?:any\s+)?skills?"
      r"|(?:не\s+используй(?:те)?|не\s+использовать)\s+"
      r"(?:никакие\s+)?(?:скилл\w*|навык\w*)"
      r"|без\s+(?:никаких\s+)?(?:скилл\w*|навык\w*)"
      r")\b"
  )
  
  # Machine-readable parity contract.  The runtime branches below intentionally
  # differ because Claude calls the boundary ``Task`` while Codex exposes several
  # aliases for ``spawn_agent``.  Keep the complete accepted set literal so the
  # wiring audit does not have to guess control flow from the first comparison.
  HARNESS_ACCEPTED_TOOLS = {
      "Task",
      "spawn_agent",
      "Agent",
      "collaboration.spawn_agent",
      "collaboration__spawn_agent",
  }
  CODEX_AGENT_TOOLS = HARNESS_ACCEPTED_TOOLS - {"Task"}
  
  
  @dataclass(frozen=True)
  class TaskContract:
      profile: str
      skills: list[str]
      missing_skills: list[str]
      route: str
  
  
  @dataclass(frozen=True)
  class TaskSelection:
      profile: str
      skills: list[str]
      missing_skills: list[str]
      route: str
  
  
  def skill_routing_opted_out(task_text: str) -> bool:
      """Honor only a leading top-level directive, never quoted/literal payload."""
      for raw_line in task_text.splitlines():
          if not raw_line.strip():
              continue
          # The plain Task payload has no provenance metadata. Restricting this
          # authority-changing switch to its leading, unindented line makes the
          # boundary deterministic: blockquotes, fenced code, indented literals,
          # and later quoted examples cannot disable routing.
          if raw_line[:1].isspace() or raw_line.lstrip().startswith((">", "`", "~", "'", '"')):
              return False
          return bool(SKILL_OPT_OUT_RE.match(raw_line.strip()))
      return False
  
  
  def route_for_task(task_text: str, skills: list[str], missing_skills: list[str] | None = None) -> str:
      if skill_routing_opted_out(task_text):
          return "user-opt-out"
      if missing_skills:
          return "skill-gap"
      return "curated" if skills else "no-high-confidence-match"
  
  
  def select_task(task_text: str, profile: str = "claude") -> TaskSelection:
      """Select only routes the receiving client can load and expose real gaps."""
      profile = str(profile).strip().lower()
      if profile not in {"claude", "codex"}:
          raise ValueError(f"unsupported client profile: {profile}")
      if skill_routing_opted_out(task_text):
          return TaskSelection(profile, [], [], "user-opt-out")
      matches = detect_keywords(task_text, profile=profile)
      usable = [item for item in matches if "skill" in item]
      unavailable = [item for item in matches if "unavailable_skill" in item]
      required_usable = [str(item["skill"]) for item in usable if item.get("required")]
      required_missing = [
          str(item["unavailable_skill"]) for item in unavailable if item.get("required")
      ]
      if required_missing:
          # A required unavailable route is not usable context. Resolve that gap
          # first; do not dilute it by also claiming a different skill was used.
          skills, missing = [], list(dict.fromkeys(required_missing))
      elif required_usable:
          skills, missing = list(dict.fromkeys(required_usable)), []
      elif usable:
          skills, missing = [str(usable[0]["skill"])], []
      elif unavailable:
          skills, missing = [], [str(unavailable[0]["unavailable_skill"])]
      else:
          skills, missing = [], []
      return TaskSelection(profile, skills, missing, route_for_task(task_text, skills, missing))
  
  
  def selected_skills(task_text: str, profile: str = "claude") -> list[str]:
      """Compatibility helper returning usable routes for one concrete client."""
      return select_task(task_text, profile)

• hooks/agent_policy_gates.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Agent policy gates: the Claude-side half of a repo's `.github/agent-policy.json`.
  
  Adopted 2026-09-25 from «блекпилл недели» (a CTO's rules for teams where agents
  write most of the code): rules that live in a prompt are not followed, so every
  rule that can be a check becomes one. The repo's local CI checks the finished
  branch (`scripts/quality_gates.py`); this hook stops the three things an agent
  does *during* the work that CI sees too late or not at all:
  
    * editing a human-owned document (AGENTS.md, the incident log, the policy);
    * opening a PR that is too big to review or has no template description;
    * running the widget test suites bare, so their Tk windows flash on the
      owner's screen (25.09: «окна чата открываются и закрываются по кругу»).
  
  Only repos that declare a policy are affected; elsewhere the hook is silent.
  Wired the same way on Codex (~/.codex/hooks.json), where file edits arrive as
  `apply_patch` and name their files only in the patch headers.
  
  Human-owned documents. The hook cannot verify a human's approval (same limit as
  human-confirmation-guard), but UserPromptSubmit carries the human's own words: a
  prompt that names a protected file (path or file name) unlocks exactly that file
  for that turn. Text inside <pasted_content> does not count — pasted material is
  data, not the owner's instruction — and the next prompt resets the unlock.
  
  Fail-open on the hook's own errors (a broken hook must not wedge every session);
  fail-closed on a match.
  """
  
  from __future__ import annotations
  
  import fnmatch
  import json
  import re
  import shlex
  import subprocess
  import sys
  import time
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent))
  from safety_common import allow, bash_command, block, file_path, log, read_event  # noqa: E402
  
  HOOK = "agent_policy_gates"
  POLICY = Path(".github") / "agent-policy.json"
  UNLOCKS = Path.home() / ".claude" / "state" / "human-owned-unlocks"
  PASTED = re.compile(r"<pasted_content\b[^>]*>.*?</pasted_content\b[^>]*>", re.S | re.I)
  CD_PREFIX = re.compile(r"""^\s*(?:cd|Set-Location)\s+("[^"]+"|'[^']+'|[^\s;&|]+)\s*(?:&&|;)""", re.I)
  WRITES = (r"(?:\btee\b|\bsed\s+-i|Set-Content|Add-Content|Out-File|Copy-Item|Move-Item|"
            r"Rename-Item|Remove-Item|\bcp\b|\bmv\b|\brm\b|\bunlink\b|\bdel\b|os\.remove|"
            r"write_text|write_bytes|WriteAllText|git\s+(?:checkout|restore)\b)")
  # A redirect INTO the file; `2>/dev/null` next to a read is not a write (review 25.09).
  REDIRECT = r"(?<![0-9&])>>?\s*[\"']?[^\s\"'|;&]*"
  TEST_RUN = re.compile(r"-m\s*(?:unittest|pytest)\b|\bpytest\b|python\S*\s+(?:-\S+\s+)*\S*test_\w+\.py")
  FILE_TOOLS = {"Write", "Edit", "MultiEdit", "NotebookEdit", "apply_patch"}
  # Codex edits files through apply_patch: no file_path, the targets are the patch's
  # section headers. `*** Move to:` counts too: renaming a file onto a protected path
  # overwrites it (measured 2026-09-28: without it that rename passed the guard).
  PATCH_FILE = re.compile(r"(?m)^[ \t]*\*\*\* (?:(?:Add|Update|Delete) File|Move to): (?P<path>.+?)\s*$")
  # Read by scripts/check_harness_parity.py: the tools this hook acts on, both harnesses.
  HARNESS_ACCEPTED_TOOLS = FILE_TOOLS | {"Bash", "PowerShell"}
  # Commands only the owner may start: her prompt must name the flag. The other
  # spellings reach the same reset without it (review 25.09: python -c ... reset_rounds).
  OWNER_COMMAND = "--reset-review-rounds"
  OWNER_SPELLINGS = (OWNER_COMMAND, "reset_rounds(", "quality-review-reset")
  
  
  # ---------- repo and policy ----------
  
  def windows_path(text: str) -> Path:
      """`/d/x` (Git Bash) -> `D:/x`; quotes stripped."""
      text = text.strip("\"'")
      match = re.match(r"^/([a-zA-Z])/(.*)$", text)
      return Path(f"{match.group(1).upper()}:/{match.group(2)}") if match else Path(text)
  
  
  def repo_root(start: Path) -> Path | None:
      """The nearest folder with a `.git` (directory, or file in a worktree).
  
      A walk, not `git rev-parse`: under load git outran its 5 s timeout, the hook
      failed open, and a human-owned file went unprotected (25.09, hook suite)."""
      probe = start if start.is_dir() else start.parent
      while not probe.exists() and probe != probe.parent:
          probe = probe.parent
      for folder in (probe, *probe.parents):
          if (folder / ".git").exists():
              return folder.resolve()
      return None
  
  
  def load_policy(root: Path | None) -> dict | None:
      if root is None or not (root / POLICY).is_file():
          return None
      return json.loads((root / POLICY).read_text(encoding="utf-8"))
  
  
  def relative(root: Path, path: Path) -> str | None:
      try:
          return path.resolve().relative_to(root).as_posix().lower()
      except ValueError:
          return None
  
  
  def protected(policy: dict) -> list[str]:
      return [p.replace("\\", "/").lower() for p in policy.get("human_owned", [])]
  
  
  def named_in(text: str, paths: list[str]) -> list[str]:
      """Protected paths the text names, by full path or by file name."""
      low = text.lower().replace("\\", "/")
      return [p for p in paths
              if p in low or re.search(rf"(?<![\w.-]){re.escape(p.rsplit('/', 1)[-1])}(?![\w-])", low)]
  
  
  # ---------- human-owned documents ----------
  
  def unlock_file(session: str) -> Path:
      return UNLOCKS / (re.sub(r"[^\w.-]", "_", session or "unknown") + ".json")
  
  
  def on_prompt(event: dict) -> None:
      """Every prompt rewrites the session's unlock: what this prompt names, nothing more."""
      root = repo_root(Path(event.get("cwd") or "."))
      policy = load_policy(root)
      if policy is None:
          return
      words = PASTED.sub(" ", str(event.get("prompt", "")))
      record = {"root": str(root), "paths": named_in(words, protected(policy)),
                "commands": [OWNER_COMMAND] if OWNER_COMMAND in words else [], "at": time.time()}
      UNLOCKS.mkdir(parents=True, exist_ok=True)
      unlock_file(event.get("session_id", "")).write_text(json.dumps(record), encoding="utf-8")
  
  
  def unlocked(event: dict, root: Path, rel: str, fi

• hooks/api-key-leak-detector.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """PostToolUse: detect API keys in tool output.
  
  Runs after Bash / Read / any tool that produces output. Scans the output
  for well-known API key patterns. Cannot retroactively block (tool already
  ran), but emits a loud warning to stderr so the user sees the leak.
  
  Key patterns recognized:
   - Anthropic: sk-ant-*
   - OpenAI: sk-*
   - GitHub PAT: ghp_*, gho_*, ghu_*, ghs_*, ghr_*
   - AWS access key: AKIA[0-9A-Z]{16}
   - AWS secret key: 40-char base64-like after aws_secret_access_key
   - Stripe: sk_live_*, sk_test_*, pk_live_*, pk_test_*
   - Slack: xoxb-*, xoxp-*, xoxa-*, xoxr-*
   - Google: AIza[0-9A-Za-z_-]{35}
   - Private keys: -----BEGIN * PRIVATE KEY-----
   - JWT: eyJ*.eyJ*.* (three base64url segments)
  
  This is a detective control, not preventive. Use alongside block_secrets.py
  which is preventive. If this fires, secret was exposed in tool output and
  is likely in context. Advise rotation.
  """
  from __future__ import annotations
  
  import json
  import re
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent))
  from safety_common import log, read_event  # noqa: E402
  
  # (label, regex) pairs. Ordered by specificity (most specific first).
  PATTERNS = [
      ("Anthropic API key", re.compile(r"sk-ant-[A-Za-z0-9\-_]{32,}")),
      ("OpenAI API key", re.compile(r"sk-(?:proj-)?[A-Za-z0-9]{32,}")),
      ("GitHub PAT", re.compile(r"gh[pousr]_[A-Za-z0-9]{36,}")),
      ("GitHub fine-grained", re.compile(r"github_pat_[A-Za-z0-9_]{80,}")),
      ("AWS access key", re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b")),
      ("AWS secret key",
       re.compile(r"aws[_\-\s]*secret[_\-\s]*access[_\-\s]*key[\"'\s=:]+([A-Za-z0-9/+=]{40})",
                  re.IGNORECASE)),
      ("Stripe live key", re.compile(r"\b(?:sk|rk|pk)_live_[0-9a-zA-Z]{24,}")),
      ("Stripe test key", re.compile(r"\b(?:sk|rk|pk)_test_[0-9a-zA-Z]{24,}")),
      ("Slack token", re.compile(r"\bxox[baprs]-[A-Za-z0-9\-]{10,}")),
      ("Google API key", re.compile(r"\bAIza[0-9A-Za-z_\-]{35}\b")),
      ("Private key block",
       re.compile(r"-----BEGIN (?:RSA |EC |DSA |OPENSSH |PGP |ENCRYPTED )?PRIVATE KEY-----")),
      ("JWT token",
       re.compile(r"\beyJ[A-Za-z0-9_\-]{10,}\.eyJ[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}")),
      ("Generic bearer token",
       re.compile(r"\b[Bb]earer\s+[A-Za-z0-9_\-\.=]{40,}")),
  ]
  
  
  def extract_output(event: dict) -> str:
      """Pull the tool's output from the PostToolUse event.
  
      Event shape varies by tool; we try common keys.
      """
      response = event.get("tool_response", {})
      if isinstance(response, str):
          return response
      if not isinstance(response, dict):
          return ""
      # Most tools put text here
      for key in ("stdout", "output", "content", "text", "result"):
          v = response.get(key)
          if isinstance(v, str):
              return v
          if isinstance(v, list):
              return "\n".join(str(x) for x in v)
      # Fallback - serialize entire response
      try:
          return json.dumps(response, ensure_ascii=False)
      except (TypeError, ValueError):
          return str(response)
  
  
  def redact_match(label: str, snippet: str) -> str:
      """Show first 8 chars + asterisks + last 4 so user sees what to rotate."""
      if len(snippet) <= 12:
          return f"[{label}] {'*' * len(snippet)}"
      return f"[{label}] {snippet[:8]}...{snippet[-4:]}"
  
  
  def main() -> None:
      event = read_event()
      tool_name = event.get("tool_name", "")
      output = extract_output(event)
      if not output:
          sys.exit(0)
  
      hits: list[tuple[str, str]] = []
      for label, pat in PATTERNS:
          for m in pat.finditer(output):
              hits.append((label, m.group(0)))
              if len(hits) >= 5:  # don't spam, cap at 5 findings
                  break
          if len(hits) >= 5:
              break
  
      if not hits:
          sys.exit(0)
  
      # Log each finding
      for label, snippet in hits:
          log("ALERT", "detect_api_key_leak", "found", label,
              f"{tool_name}: {snippet[:40]}")
  
      # Emit loud warning to stderr - user will see this
      lines = [
          "",
          "=" * 70,
          "[api-key-leak] WARNING: tool output contains what looks like API keys",
          "=" * 70,
          f"Tool: {tool_name}",
          f"Findings: {len(hits)}",
      ]
      for label, snippet in hits[:5]:
          lines.append(f"  {redact_match(label, snippet)}")
      lines.extend([
          "",
          "Action items:",
          "  1) The secret is now in Claude Code context. Consider the session compromised",
          "  2) Rotate the exposed key(s) in the respective service",
          "  3) Check git history / log files for the same pattern",
          "  4) If committed recently: force-push after BFG Repo-Cleaner rewrite",
          "=" * 70,
          "",
      ])
      sys.stderr.write("\n".join(lines))
      sys.exit(0)
  
  
  if __name__ == "__main__":
      main()
  

• hooks/article-voice-gate.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """PreToolUse: an article reaches the knowledge base only with a check behind it.
  
  The news pipeline writes knowledge-base pages by machine. Two passes stand between
  the assembled facts and the public repository: an editor that rewrites the sentences
  in the house voice under mechanical guards, and an independent checker that reads the
  result against the same research and names anything invented. Both leave a receipt.
  
  This gate refuses the publishing step - pushing a `kb/` branch, opening or merging its
  pull request - when that receipt is missing or says the page was refused. It is the
  mechanical half of the owner's rule (2026-09-04): "из головы писать не должен, надо
  человеческий стиль написания и проверять все данные".
  
  Three states, and the middle one is not green (see rules/absence-of-signal.md):
    receipt says checked, nothing invented  -> allow
    receipt says refused / invented         -> block, naming the finding
    no receipt for this branch              -> block as UNVERIFIED, not as fine
  
  Scope: one pipeline, not every repository with a `kb/` in it (2026-09-27). A `kb/`
  token alone is not evidence of an article - `kb/docs/platform/...` is an engineering
  reference filed by hand in another repository, with no renderer, no ledger and no
  `ops/codex/article_voice.py` anywhere in that repository. For such a file no receipt
  can ever exist, so the gate had exactly one exit there: its own bypass. A gate that is
  always bypassed has stopped being a gate, and the habit travels to the pages where the
  check is real. See `in_scope` for the three things that put a command inside the
  pipeline - and for why the pipeline's own marker is not one of them on its own.
  
  Bypass: `# claude-bypass: article-voice` in the command, or CLAUDE_ALLOW_ARTICLE_VOICE=1.
  Self-test: python article-voice-gate.py --self-test
  """
  from __future__ import annotations
  
  import json
  import os
  import re
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).resolve().parent))
  try:
      from safety_common import block, bypass, read_event  # type: ignore
  except Exception:  # a hook that cannot load its helpers must not wedge the session
      block = bypass = read_event = None  # type: ignore
  
  REPO = "AnastasiyaW/knowledge-space"
  BRANCH = re.compile(r"(?:^|[\s\"'/:])(kb/[A-Za-z0-9._-]+)")
  # Git takes its own options before the verb, and `git -C <clone> push` is exactly how
  # one reaches another repository - the publish verb has to survive them, or the gate
  # reads that command as no publish at all (found by the scope controls, 2026-09-27).
  GIT_GLOBAL = r"(?:-C\s+\S+|-c\s+\S+|--git-dir=\S+|--work-tree=\S+|--no-pager)"
  PUBLISHES = re.compile(rf"(?i)\bgh\s+pr\s+(create|merge)\b|\bgit\b(?:\s+{GIT_GLOBAL})*\s+push\b")
  # What identifies the pipeline's own repository from inside it. The renderer, the
  # editor and the checker live in ops/codex; every receipt lands in var/kb-articles.
  PIPELINE_MARKERS = ("ops/codex/article_voice.py", "var/kb-articles")
  # What identifies the knowledge base itself. The writer commits and pushes with its
  # working directory set to a *clone of the knowledge base*, and that clone carries no
  # pipeline marker at all - measured 2026-09-27. A marker-only scope would therefore have waved the real publishing
  # path straight through, which is the opposite of the point.
  KB_SLUG = "knowledge-space"
  # `git -C <path>` moves the repository out from under the working directory.
  GIT_C = re.compile(r"-C\s+(\"[^\"]+\"|'[^']+'|\S+)")
  # Where the writer files its receipts. The first that exists wins; a project may name
  # its own through CLAUDE_NEWS_LEDGERS (semicolon-separated).
  # Built from the home directory, never written out: this file lives in a public
  # repository and a literal user path is exactly what its scan refuses.
  _WORK = Path.home() / "Desktop" / "Codex+Code"
  DEFAULT_LEDGERS = [
      _WORK / "worktrees" / "happyin-news-v1-4" / "diffusion-love" / "var" / "kb-articles",
      _WORK / "diffusion-love" / "var" / "kb-articles",
      Path.cwd() / "var" / "kb-articles",
  ]
  LEDGER_FILES = ("ledger.jsonl", "revoice.jsonl")
  
  
  def ledger_dirs() -> list[Path]:
      configured = os.environ.get("CLAUDE_NEWS_LEDGERS", "").strip()
      if configured:
          return [Path(p) for p in configured.split(";") if p.strip()]
      return DEFAULT_LEDGERS
  
  
  def receipts_for(branch: str) -> list[dict]:
      """Every receipt that names this branch, oldest first."""
      found: list[dict] = []
      for directory in ledger_dirs():
          for name in LEDGER_FILES:
              path = directory / name
              if not path.is_file():
                  continue
              try:
                  for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
                      if branch not in line:
                          continue
                      try:
                          record = json.loads(line)
                      except json.JSONDecodeError:
                          continue
                      if record.get("branch") == branch:
                          found.append(record)
              except OSError:
                  continue
      return found
  
  
  def verdict(branch: str) -> tuple[str, str]:
      """(state, why) where state is 'checked', 'refused' or 'unverified'."""
      records = receipts_for(branch)
      if not records:
          return "unverified", f"no receipt names {branch} in {', '.join(str(d) for d in ledger_dirs())}"
      latest = records[-1]
      check = latest.get("check") or {}
      voice = latest.get("voice") or {}
      invented = check.get("invented") or []
      if invented:
          return "refused", f"the independent check found {len(invented)} invented specific(s): {str(invented[0])[:160]}"
      if not check:
          return "unverified", "the receipt carries no check: the page was written before the gate, or with --no-voice"
      if voice.get("status") == "refused":
          why = "; ".join(str(c) for c in (voice.get("complaints") or [])[:2])
          return "checked", f"the rewrite was refuse

• hooks/ask-question-guard.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """PreToolUse(AskUserQuestion): block deferral/menu questions on reversible work.
  
  Root cause this closes (2026-06-16): the behavioral canon "on reversible work
  decide yourself and proceed; ask only for irreversible / genuine-fork decisions"
  (rules/autonomy-risk-tiers.md + finish-the-task.md) was enforced only by
  stop-phrase-guard, which scans the final assistant TEXT at the Stop event. But
  AskUserQuestion is a TOOL CALL — its menu text lives in tool_input, never in the
  scanned assistant text — so "ask via the tool" was a total enforcement blind spot
  (a "куда продолжаем?" / "what next?" menu sailed straight through).
  
  This hook scans the AskUserQuestion payload itself (question + header + option
  labels) for deferral / menu / scope-degree phrasing and blocks it. A SPECIFIC
  genuine fork ("Postgres or MySQL?", "delete prod table X — confirm?") does not
  match these patterns and passes. Bypass for a real irreversible/fork question:
  CLAUDE_ALLOW_ASK=1 in the session.
  
  Exit conventions per safety_common: block = {"decision":"block"} on stdout.
  """
  from __future__ import annotations
  
  import sys
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent))
  from safety_common import allow, any_match, block, bypass, log, read_event  # noqa: E402
  
  # Deferral / menu / scope-degree phrasing. Regex, case-insensitive.
  # These signal "what should I do next / which of these / how far" — decisions the
  # agent should make itself on reversible work, not punt to the user.
  DEFERRAL_PATTERNS = [
      # RU — "what next / which of these / what do we do"
      r"что (дальше|делать дальше|делаем(?: дальше)?|теперь делаем|по плану дальше)",
      r"что (выбираешь|предпочитаешь|приоритетн|важнее)",
      r"что из (этого|них|трёх|двух|перечисл|предложен|вариант)",
      r"куда (дальше|продолжа|двига|ид[её]м|приложить)",
      r"\bпродолжаем\b",
      r"какой из (вариантов|подходов|трёх|двух|них)",
      r"какой (вариант|подход) (выбрать|предпочт|брать)",
      # RU — scope-degree ("how aggressive / how deep / how far") = agent's call
      r"насколько (агрессивн|глубок|далеко|сильно|подробн|широк)",
      r"в каком (объёме|масштабе)",
      # EN — what next / which of these / how far
      r"what (would you like|do you want|next|should i (do|tackle|work on|focus|start))",
      r"which (option|one|of these|of the|direction|do you (want|prefer))",
      r"how (aggressive|far|deep|much) (should|do you|to go)",
      r"want me to (keep going|continue|proceed|do)",
      r"should i (continue|proceed|keep going|also)",
  ]
  
  # Generic menu headers (AskUserQuestion `header` field) that signal a deferral menu
  # rather than a specific decision. Compared lowercased, exact-ish.
  DEFERRAL_HEADERS = {
      "next", "next step", "scope", "direction", "priority", "objem", "объём",
      "дальше", "выбор", "приоритет", "направление", "что дальше",
  }
  
  
  def main() -> None:
      event = read_event()
      if event.get("tool_name") != "AskUserQuestion":
          allow()
  
      ti = event.get("tool_input", {})
      questions = ti.get("questions", []) or []
  
      texts: list[str] = []
      headers: list[str] = []
      for q in questions:
          if not isinstance(q, dict):
              continue
          texts.append(str(q.get("question", "")))
          h = str(q.get("header", "")).strip().lower()
          headers.append(h)
          texts.append(h)
          for o in q.get("options", []) or []:
              if isinstance(o, dict):
                  texts.append(str(o.get("label", "")))
      blob = "\n".join(t for t in texts if t)
      if not blob.strip():
          allow()
  
      hit = any_match(blob, DEFERRAL_PATTERNS)
      header_hit = next((h for h in headers if h in DEFERRAL_HEADERS), None)
      if not hit and not header_hit:
          allow()  # specific genuine-fork question — let it through
  
      signal = hit or f"header:{header_hit}"
  
      if bypass("ask", blob):
          log("WARN", "ask_question_guard", "bypass", signal, blob[:300])
          allow()
  
      log("BLOCK", "ask_question_guard", "deny", signal, blob[:300])
      block(
          "Deferral/menu question detected (/" + signal + "/). "
          "Канон autonomy-risk-tiers + finish-the-task: на ОБРАТИМОЙ работе НЕ "
          "спрашиваем — выбираем лучший вариант сами и делаем, отчёт ПОСЛЕ. "
          "Вопрос-меню «что дальше / куда продолжаем / насколько / which of these» "
          "= откладывание.\n"
          "Если решение реально НЕОБРАТИМОЕ или это genuine fork, который должен "
          "решить пользователь (DROP/DELETE, прод-миграция, выбор архитектуры с "
          "разными последствиями) — переформулируй как КОНКРЕТНЫЙ вопрос (не «что "
          "дальше», а «удалить таблицу X? / Postgres или MySQL?») ИЛИ выставь "
          "CLAUDE_ALLOW_ASK=1 в сессии и переспроси."
      )
  
  
  if __name__ == "__main__":
      main()
  

• hooks/backup-retention-cleanup.pyGitHub ↗
• hooks/benjamin-plus-inject.pyGitHub ↗
• hooks/claude-attribution-guard.pyGitHub ↗
• hooks/codex_hook_dispatch.pyGitHub ↗
• hooks/command-injection-guard.pyGitHub ↗
• hooks/continuity-contract-guard.pyGitHub ↗
• hooks/continuity-session-check.pyGitHub ↗
• hooks/conversation-history-capture.pyGitHub ↗
• hooks/coord-claim-guard.pyGitHub ↗
• hooks/cyrillic-bash-guard.pyGitHub ↗
• hooks/db-snapshot-guard.pyGitHub ↗
• hooks/dependency-currency-guard.pyGitHub ↗
• hooks/dependency-provenance-guard.pyGitHub ↗
• hooks/destructive-command-guard.pyGitHub ↗
• hooks/directory-creation-guard.pyGitHub ↗
• hooks/disk-pressure-advisor.pyGitHub ↗
• hooks/docs-staleness-guard.pyGitHub ↗
• hooks/feature-list-validator.pyGitHub ↗
• hooks/feedback-pending-show.pyGitHub ↗
• hooks/file-cohesion-guard.pyGitHub ↗
• hooks/git-auto-backup.pyGitHub ↗
• hooks/git-destructive-guard.pyGitHub ↗
• hooks/git-lock-homework-guard.pyGitHub ↗
• hooks/git-source-gate.pyGitHub ↗
• hooks/github-workflow-security.pyGitHub ↗
• hooks/handoff-closure-audit-guard.pyGitHub ↗
• hooks/handoff-resume-gate.pyGitHub ↗
• hooks/handoff-to-todo.pyGitHub ↗
• hooks/harness-load-advisor.pyGitHub ↗
• hooks/hook-tree-drift-check.pyGitHub ↗
• hooks/human-confirmation-guard.pyGitHub ↗
• hooks/kb-validate-gate.pyGitHub ↗
• hooks/keyword-skill-router.pyGitHub ↗
• hooks/launch-watch-guard.pyGitHub ↗
• hooks/live-tree-guard.pyGitHub ↗
• hooks/long-run-detector.pyGitHub ↗
• hooks/module-shape-advisor.pyGitHub ↗
• hooks/open-items-are-work-orders.pyGitHub ↗
• hooks/outward-claim-evidence-guard.pyGitHub ↗
• hooks/over-engineering-advisor.pyGitHub ↗
• hooks/plan-gate.pyGitHub ↗
• hooks/powershell-dynamic-execution-guard.pyGitHub ↗
• hooks/pre-push-claude-attribution.pyGitHub ↗
• hooks/pre-push-personal-email-guard.pyGitHub ↗
• hooks/pre-push-public-repo-scan.pyGitHub ↗
• hooks/precompact-handoff-guard.pyGitHub ↗
• hooks/problems-md-validator.pyGitHub ↗
• hooks/repeated-attempt-guard.pyGitHub ↗
• hooks/root-cause-delivery-guard.pyGitHub ↗
• hooks/safety_common.pyGitHub ↗
• hooks/secret-leak-guard.pyGitHub ↗
• hooks/self-harm-guard.pyGitHub ↗
• hooks/session-drift-validator.pyGitHub ↗
• hooks/session-feedback-capture.pyGitHub ↗
• hooks/session-handoff-check.pyGitHub ↗
• hooks/session-handoff-reminder.pyGitHub ↗
• hooks/shape_common.pyGitHub ↗
• hooks/shared-branch-guard.pyGitHub ↗
• hooks/skill_contract_state.pyGitHub ↗
• hooks/stop-phrase-guard.pyGitHub ↗
• hooks/subagent-evidence-receipt.pyGitHub ↗
• hooks/subagent-skill-context.pyGitHub ↗
• hooks/task-cycle-controller.pyGitHub ↗
• hooks/task-inbox-show.pyGitHub ↗
• hooks/telegram-mass-send-guard.pyGitHub ↗
• hooks/test-gate-stop-hook.pyGitHub ↗
• hooks/test-muting-guard.pyGitHub ↗
• hooks/todo-inbox.pyGitHub ↗
• hooks/transfer-contract-guard.pyGitHub ↗
• hooks/unbuffered-progress-advisor.pyGitHub ↗
• hooks/user-task-completion-guard.pyGitHub ↗
• hooks/verify-deleted-guard.pyGitHub ↗
• hooks/widget-test-guard.pyGitHub ↗

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