Hooks

Where it lives

• hooks/guard_code_quality.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """PostToolUse: catch gross code smells after write/edit."""
  from __future__ import annotations
  import re, sys
  from pathlib import Path
  sys.path.insert(0, str(Path(__file__).resolve().parent))
  from lib_payload import ( # type: ignore
      read_payload, detect_client, tool_name, file_path, is_edit_tool, emit_post_block,
  )
  
  def main() -> None:
      p = read_payload()
      client = detect_client(p)
      name = tool_name(p)
      if name and not is_edit_tool(name) and client != "":
          #  post may not send tool name the same way — still try path
          pass
      path = file_path(p)
      if not path or not Path(path).is_file():
          sys.exit(0)
      # skip agent contract paths
      if "/.agents/" in path or "/node_modules/" in path:
          sys.exit(0)
  
      text = Path(path).read_text(errors="replace")
      lines = text.splitlines()
      warnings: list[str] = []
      ext = Path(path).suffix.lower()
  
      def add(msg: str, hits: list[str]) -> None:
          if hits:
              warnings.append(msg + "\n" + "\n".join(hits[:12]))
  
      if ext in {".ts", ".tsx", ".mts", ".cts", ".js", ".jsx", ".mjs", ".cjs"}:
          any_hits = []
          for i, ln in enumerate(lines, 1):
              if "guardian: allow" in ln:
                  continue
              if re.search(r":\s*any(\s|;|,|\)|>|\[|=|$)|as\s+any\b|<any>", ln):
                  any_hits.append(f"{i}:{ln.strip()[:120]}")
          add(f"'any' type in {path}", any_hits)
  
          prisma = []
          for i, ln in enumerate(lines, 1):
              if "guardian: allow" in ln:
                  continue
              if re.search(r"\$queryRawUnsafe|\$executeRawUnsafe", ln):
                  prisma.append(f"{i}:{ln.strip()[:120]}")
          add(f"Prisma unsafe raw SQL in {path}", prisma)
  
          ts_ignore = []
          for i, ln in enumerate(lines, 1):
              if re.search(r"@ts-ignore|@ts-nocheck", ln) and "guardian: allow" not in ln:
                  ts_ignore.append(f"{i}:{ln.strip()[:120]}")
          add(f"@ts-ignore/@ts-nocheck in {path}", ts_ignore)
  
          evals = []
          for i, ln in enumerate(lines, 1):
              if re.search(r"\beval\s*\(|new\s+Function\s*\(", ln) and "guardian: allow" not in ln:
                  evals.append(f"{i}:{ln.strip()[:120]}")
          add(f"eval/new Function in {path}", evals)
  
      secrets = []
      for i, ln in enumerate(lines, 1):
          if "guardian: allow" in ln or "process.env" in ln or "os.environ" in ln:
              continue
          if re.search(r"(api[_-]?key|secret|password|token)\s*[:=]\s*['\"][A-Za-z0-9_\-]{16,}['\"]", ln, re.I):
              secrets.append(f"{i}:{ln.strip()[:100]}")
      add(f"Possible hardcoded secret in {path}", secrets)
  
      if not warnings:
          if client == "":
              print("{}")
          sys.exit(0)
  
      report = "=== agent-guard code quality ===\n\n" + "\n\n---\n\n".join(warnings)
      report += "\n\nFix before done. Bypass line: // guardian: allow <reason>"
      emit_post_block(client, report)
  
  if __name__ == "__main__":
      try:
          main()
      except Exception:
          sys.exit(0)
  

• hooks/guard_shell.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """PreToolUse: block destructive shell across CLIs."""
  from __future__ import annotations
  import hashlib, ipaddress, json, os, re, shlex, socket, subprocess, sys, threading, time
  from pathlib import Path
  sys.dont_write_bytecode = True
  sys.path.insert(0, str(Path(__file__).resolve().parent))
  from lib_payload import (  # type: ignore
      read_payload, detect_client, tool_name, shell_command, file_path,
      is_shell_tool, is_edit_tool, emit_allow, emit_deny, tool_input,
  )
  
  PM_AGENTS = {"dev-orchestrator", "frontend-orchestrator", "marketing-orchestrator"}
  LANE_PILOT_PM_AGENT_TYPES = {"lane-pilot-pm"}
  LANE_PILOT_HELPER_ROLES = {
      "errand", "specialist", "browser-qa", "pm-read", "pm-reader",
      "critic", "plan-critic", "code-critic", "specialist-reviewer",
      "night-reviewer", "docs", "docs-maintainer", "onboarder",
      "memory-maintainer", "project-life", "gate-triage", "council-seat",
      "rules-analyzer",
      "design-lead", "copy-lead", "seo-specialist", "tavily",
  }
  
  
  def agent_key(agent: object) -> str | None:
      if not isinstance(agent, str) or not agent.strip():
          return None
      return agent.strip().rsplit(":", 1)[-1]
  
  
  LANE_PILOT_READ_COMMANDS = {
      "cat", "cd", "cmp", "cut", "echo", "file", "find", "grep", "head", "jq",
      "ls", "printf", "pwd", "readlink", "realpath", "rg", "sed", "sha256sum",
      "sort", "stat", "tail", "test", "true", "false", "type", "wc", "which",
      "yq",
  }
  LANE_PILOT_GIT_READONLY = {"status", "diff", "log", "show"}
  LANE_PILOT_RUN_CONTROLLER_SUBCOMMANDS = {"run", "start", "status", "watch"}
  LANE_PILOT_LANE_CTL_SUBCOMMANDS = {
      "start", "status", "tail", "events", "cancel", "retry", "fallback",
      "verify", "accept",
  }
  LANE_PILOT_ADOC_COMMANDS = {"adoc", "agents-doctor"}
  LANE_PILOT_FORBIDDEN_SCRIPT_RUNNERS = {
      "bash", "sh", "python", "python3", "node", "nodejs",
  }
  PM_READ_COMMANDS = {
      "cat", "cd", "cmp", "cut", "date", "df", "du", "echo", "file", "find",
      "gitnexus", "grep", "head", "journalctl", "jq", "ls", "lsof", "pgrep",
      "printf", "ps", "pstree", "pwd", "readlink", "realpath", "rg", "sed",
      "sha256sum", "sort", "stat", "tail", "test", "true", "false", "type",
      "uniq", "wc", "which", "yq",
  }
  # Read-only bb CLI calls the PM may use to look around; anything that changes state stays denied.
  PM_BB_READ_COMMANDS = {
      ("status",), ("guide",), ("thread", "show"), ("thread", "get"), ("thread", "log"), ("thread", "messages"),
      ("thread", "list"), ("thread", "output"), ("thread", "search"), ("thread", "wait"), ("thread", "history"),
      ("thread", "context"), ("thread", "count"),
      ("memory", "search"), ("memory", "get"), ("memory", "catalog"),
      ("project", "list"), ("project", "show"),
      # Env Catalog and Lane Pilot's own CLI (accounts, settings, schedules, anamnesis) are used by the PM directly: no caller fences (owner decision 2026-10-08).
      ("env-catalog", "list"), ("env-catalog", "request"), ("env-catalog", "get"), ("env-catalog", "set"), ("env-catalog", "delete"), ("env-catalog", "export"),
      ("lane-pilot",),
  }
  # `bb plugin rpc call <plugin> ...` and `bb plugin run lane-pilot ...` for the plugins an agent works with every day.
  _PM_OPEN_PLUGINS = {"lane-pilot", "bb-plugin-lane-pilot", "env-catalog"}
  
  
  def _pm_plugin_call_open(words: list[str]) -> bool:
      return (words[:3] == ["plugin", "rpc", "call"] and len(words) > 3 and words[3] in _PM_OPEN_PLUGINS) or (
          words[:3] == ["plugin", "run", "lane-pilot"])
  # Agents coordinate by messaging each other's threads; starting or changing threads stays denied.
  PM_BB_MESSAGE_COMMANDS = {
      ("thread", "tell"), ("thread", "message"),
      ("thread", "queue", "create"), ("thread", "queue", "send"), ("thread", "queue", "list"),
  }
  # A Lane Pilot PM stops a stuck thread, archives it and cancels its plan (owner, 2026-10-09). Starting one (bb thread new) stays
  # with Lane Pilot's tools (lane_pilot_dispatch_writer, lane_pilot_specialist, lane_pilot_errand), so a writer never bypasses critique.
  PM_BB_THREAD_CONTROL = {("thread", "stop"), ("thread", "archive"), ("thread", "unarchive"), ("thread", "cancel-plan")}
  # Settings are read-only for the PM; the setters (bb settings general …) stay denied.
  PM_BB_SETTINGS_READ = {("settings", "show"), ("settings", "usage"), ("settings", "version")}
  # A Lane Pilot PM is the owner's own chat and ships too (owner, 2026-10-03): it may read BB state and reload or
  # install plugins.
  LANE_PILOT_BB_COMMANDS = PM_BB_READ_COMMANDS | PM_BB_MESSAGE_COMMANDS | PM_BB_THREAD_CONTROL | PM_BB_SETTINGS_READ | {
      ("version",), ("memory", "catalog"), ("project", "get"),
      ("plugin", "list"), ("plugin", "logs"), ("plugin", "info"), ("plugin", "show"), ("plugin", "status"),
      ("plugin", "source"), ("plugin", "outdated"), ("plugin", "install-jobs"), ("plugin", "update-jobs"),
      ("environment", "list"), ("environment", "show"), ("environment", "get"), ("environment", "providers"),
      ("host", "list"), ("host", "show"), ("provider", "list"), ("provider", "models"), ("skill", "list"),
  }
  # Shipping a plugin is the job of that plugin's own PM: reload/install/update only from a bb-plugin-* checkout, so a
  # product PM cannot reload Lane Pilot under running writers.
  LANE_PILOT_BB_PLUGIN_SHIP = {("plugin", "build"), ("plugin", "reload"), ("plugin", "install"), ("plugin", "update")}
  
  
  def _is_plugin_checkout(cwd: object) -> bool:
      return isinstance(cwd, str) and any(part.startswith("bb-plugin-") for part in Path(cwd).parts)
  
  
  # The bb CLI by name, by absolute path, or through the BB_CLI variable BB sets for agents.
  PM_BB_EXECUTABLES = {"bb", "$BB_CLI", "${BB_CLI}"}
  # Typed control-plane CLIs the PM may run directly (not writer lifecycle).
  # lane-ctl / run-controller start|watch|status stay delegated to supervisors.
  PM_CONTROL_COMMANDS = {
      "agents-doctor",
      "check-owns-paths",
      "plan-critique",
      "pm_read",
      "lane-stall-check",
      "resume-project",
      "run-board",
      "run-finalize",
      "run-init",
      "run-val

• hooks/lane_statusline_session.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """SessionStart/SessionEnd helper: remember agent_type for statusLine routing.
  
  Claude Code statusLine JSON sometimes omits agent_type. SessionStart always has
  it for agent sessions — we stash it under ~/.agents/statusline/sessions/<id>.
  
  statusLine (lane-statusline) then maps session_id → lane HUD vs claude-pulse.
  """
  from __future__ import annotations
  
  import json
  import os
  import sys
  import time
  from pathlib import Path
  
  # allow import of lane_statusline_lib from install bin
  _CANDIDATE_BINS = (
      Path(__file__).resolve().parent.parent / "bin",
      Path.home() / ".agents" / "bin",
  )
  for _b in _CANDIDATE_BINS:
      if _b.is_dir() and str(_b) not in sys.path:
          sys.path.insert(0, str(_b))
  
  try:
      from lane_statusline_lib import (  # noqa: E402
          DEFAULT_LANE_AGENTS,
          extract_agent_type,
          lane_agent_set,
          session_mark_path,
          write_session_agent,
      )
  except ImportError:
      # minimal fallback if lib not installed yet
      def extract_agent_type(raw):  # type: ignore
          try:
              d = json.loads(raw) if isinstance(raw, str) else raw
          except Exception:
              return ""
          if not isinstance(d, dict):
              return ""
          return str(d.get("agent_type") or "").strip()
  
      def write_session_agent(session_id, agent):  # type: ignore
          return None
  
      def session_mark_path(session_id):  # type: ignore
          return Path.home() / ".agents" / "statusline" / "sessions" / f"{session_id}.json"
  
      DEFAULT_LANE_AGENTS = ("dev-orchestrator",)
      def lane_agent_set():  # type: ignore
          return set(DEFAULT_LANE_AGENTS)
  
  
  def _session_id(data: dict) -> str:
      for key in ("session_id", "sessionId", "transcript_path", "transcriptPath"):
          val = data.get(key)
          if isinstance(val, str) and val.strip():
              # transcript path → basename stem as id
              if "transcript" in key.lower() or val.endswith(".jsonl"):
                  return Path(val).stem
              return val.strip()
      nested = data.get("data")
      if isinstance(nested, dict):
          return _session_id(nested)
      return ""
  
  
  def main() -> int:
      raw = sys.stdin.read()
      try:
          data = json.loads(raw) if raw.strip() else {}
      except json.JSONDecodeError:
          return 0
      if not isinstance(data, dict):
          return 0
  
      # SessionEnd: clear mark
      event = str(data.get("hook_event_name") or data.get("event") or "").lower()
      sid = _session_id(data)
      agent = extract_agent_type(data)
  
      if "end" in event and sid:
          path = session_mark_path(sid)
          try:
              if path.is_file():
                  path.unlink()
          except OSError:
              pass
          return 0
  
      if not agent or agent not in lane_agent_set():
          # non-lane session: ensure no stale mark if we know session id
          if sid:
              path = session_mark_path(sid)
              try:
                  if path.is_file():
                      path.unlink()
              except OSError:
                  pass
          return 0
  
      if sid:
          write_session_agent(sid, agent)
      else:
          # last-resort: write "active" stamp (short TTL used by statusline)
          try:
              root = Path.home() / ".agents" / "statusline"
              root.mkdir(parents=True, exist_ok=True)
              (root / "last_lane_agent.json").write_text(
                  json.dumps({"agent_type": agent, "ts": time.time()}),
                  encoding="utf-8",
              )
          except OSError:
              pass
      return 0
  
  
  if __name__ == "__main__":
      raise SystemExit(main())
  

• hooks/lib_payload.pyGitHub ↗
  Read the script

  """Normalize hook stdin payloads across Claude / Codex / Grok."""
  from __future__ import annotations
  import json, os, sys
  from typing import Any
  
  def read_payload() -> dict[str, Any] | None:
      raw = sys.stdin.read()
      if not raw.strip():
          return None
      try:
          payload = json.loads(raw)
      except json.JSONDecodeError:
          return None
      return payload if isinstance(payload, dict) else None
  
  def detect_client(p: dict) -> str:
      env = os.environ.get("AGENT_HOOK_CLIENT", "").lower()
      if env in ("claude", "codex", "grok", ""):
          return env
      # Heuristics
      if "toolCall" in p and isinstance(p.get("toolCall"), dict):
          return ""
      if p.get("hookEventName") in ("pre_tool_use", "post_tool_use") or "toolName" in p:
          return "grok"
      if "agent_type" in p or "tool_name" in p:
          # Claude + Codex both use similar; codex often has session_id + tool_name
          return "claude" if os.environ.get("CLAUDE_PROJECT_DIR") or "agent_type" in p else "codex"
      return "claude"
  
  def tool_name(p: dict) -> str:
      if isinstance(p.get("toolCall"), dict):
          return str(p["toolCall"].get("name") or "")
      for k in ("tool_name", "toolName", "tool"):
          if p.get(k):
              return str(p[k])
      ti = p.get("tool_input") or p.get("toolInput") or {}
      if isinstance(ti, dict) and ti.get("name"):
          return str(ti["name"])
      return ""
  
  def tool_input(p: dict) -> dict:
      if isinstance(p.get("toolCall"), dict):
          args = p["toolCall"].get("args") or p["toolCall"].get("arguments") or {}
          if isinstance(args, str):
              try:
                  args = json.loads(args)
              except json.JSONDecodeError:
                  return {}
          return args if isinstance(args, dict) else {}
      ti = p.get("tool_input") or p.get("toolInput") or {}
      return ti if isinstance(ti, dict) else {}
  
  def shell_command(p: dict) -> str:
      ti = tool_input(p)
      for k in ("command", "cmd", "CommandLine", "commandLine"):
          v = ti.get(k)
          if isinstance(v, str) and v.strip():
              return v
      #  sometimes nests
      for k in ("command", "cmd"):
          v = p.get(k)
          if isinstance(v, str):
              return v
      return ""
  
  def file_path(p: dict) -> str:
      ti = tool_input(p)
      for k in (
          "file_path", "filePath", "notebook_path", "notebookPath", "path",
          "target_file", "TargetFile", "file",
      ):
          v = ti.get(k)
          if isinstance(v, str) and v:
              return v
      # apply_patch may not have path
      return ""
  
  def is_shell_tool(name: str) -> bool:
      n = name.lower()
      return n in {
          "bash", "run_terminal_command", "run_command", "shell",
          "execute", "terminal", "local_shell",
      }
  
  def is_edit_tool(name: str) -> bool:
      n = name.lower()
      return n in {
          "edit", "write", "multiedit", "notebookedit", "search_replace", "str_replace",
          "apply_patch", "replace_file_content", "multi_replace_file_content",
          "write_to_file", "create_file", "edit_file",
      }
  
  def emit_allow(client: str) -> None:
      if client == "":
          print(json.dumps({"decision": "allow"}))
      elif client == "grok":
          print(json.dumps({"decision": "allow"}))
      # Claude/Codex: silent allow via exit 0 is fine
      sys.exit(0)
  
  def emit_deny(client: str, reason: str, *, block_exit: int = 2) -> None:
      if client == "":
          print(json.dumps({"decision": "deny", "reason": reason}, ensure_ascii=False))
          sys.exit(0)
      if client == "grok":
          print(json.dumps({"decision": "deny", "reason": reason}, ensure_ascii=False))
          sys.exit(2)
      # Exit 2 makes Claude prefix "hook error". pm_read passes block_exit=0:
      # the JSON deny still blocks the tool, without looking like a crash.
      print(json.dumps({
          "decision": "block",
          "reason": reason,
          "hookSpecificOutput": {
              "hookEventName": "PreToolUse",
              "permissionDecision": "deny",
              "permissionDecisionReason": reason,
          },
      }, ensure_ascii=False))
      sys.exit(block_exit)
  
  def emit_post_block(client: str, reason: str) -> None:
      """PostToolUse quality feedback (Claude guardian style)."""
      if client in ("claude", "codex"):
          print(json.dumps({"decision": "block", "reason": reason}, ensure_ascii=False))
      elif client == "grok":
          # PostToolUse can't block on Grok; still print for UI/log
          print(json.dumps({"decision": "allow", "reason": reason, "systemMessage": reason}, ensure_ascii=False))
      else:
          #  PostToolUse expects {}
          print("{}")
          # Also write to stderr so agent sees it
          print(reason, file=sys.stderr)
      sys.exit(0)
  

• hooks/living_memory.pyGitHub ↗
  Read the script

  """Canonical paths for living project memory (PROGRESS / LESSONS).
  
  Prefer `.agents/<name>` so root cleanups and accidental `rm *.md` do not
  wipe session state. Legacy repo-root files are still read; writers migrate
  them into `.agents/` on first touch.
  """
  from __future__ import annotations
  
  import shutil
  from pathlib import Path
  
  PROGRESS = "PROGRESS.md"
  LESSONS = "LESSONS.md"
  MEMORY_NAMES = (PROGRESS, LESSONS)
  
  
  def preferred(repo: Path, name: str) -> Path:
      return Path(repo) / ".agents" / name
  
  
  def legacy(repo: Path, name: str) -> Path:
      return Path(repo) / name
  
  
  def resolve(repo: Path, name: str) -> Path:
      """Path to read/write: preferred if present, else legacy, else preferred."""
      pref = preferred(repo, name)
      if pref.is_file():
          return pref
      leg = legacy(repo, name)
      if leg.is_file():
          return leg
      return pref
  
  
  def progress_path(repo: Path) -> Path:
      return resolve(repo, PROGRESS)
  
  
  def lessons_path(repo: Path) -> Path:
      return resolve(repo, LESSONS)
  
  
  def migrate_legacy(repo: Path, name: str) -> Path:
      """Ensure living memory lives under `.agents/`. Move legacy root file if needed."""
      pref = preferred(repo, name)
      leg = legacy(repo, name)
      if pref.is_file():
          return pref
      pref.parent.mkdir(parents=True, exist_ok=True)
      if leg.is_file():
          shutil.move(str(leg), str(pref))
          return pref
      return pref
  
  
  def migrate_all(repo: Path) -> None:
      for name in MEMORY_NAMES:
          migrate_legacy(repo, name)
  

• hooks/merge_claude_settings.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Merge the lane-stack guard + statusLine into Claude settings without wiping user config."""
  
  from __future__ import annotations
  
  import argparse
  import sys
  import json
  import os
  import re
  import shlex
  import shutil
  import subprocess
  import tempfile
  from pathlib import Path
  from typing import Any
  
  
  MATCHER = "Bash|Edit|Write|MultiEdit|NotebookEdit|WebSearch|WebFetch|mcp__bb-bridge__lane_pilot_helpers"  # web tools: the Lane Pilot PM calls lane_pilot_route first (2026-10-09)
  GUARD_COMMAND = re.compile(
      r"(?:^|[\s/])(?:guard_shell\.py|guard-orchestrator-no-direct-edits\.sh)"
      r"(?:['\"])?(?:\s|$)"
  )
  
  
  def load_settings(path: Path) -> dict[str, Any]:
      if not path.exists():
          return {}
      settings = json.loads(path.read_text(encoding="utf-8"))
      if not isinstance(settings, dict):
          raise ValueError(f"{path}: settings root must be an object")
      hooks = settings.get("hooks", {})
      if hooks is not None and not isinstance(hooks, dict):
          raise ValueError(f"{path}: hooks must be an object")
      if isinstance(hooks, dict):
          pre_tool_use = hooks.get("PreToolUse", [])
          if pre_tool_use is not None and not isinstance(pre_tool_use, list):
              raise ValueError(f"{path}: hooks.PreToolUse must be an array")
      return settings
  
  
  def is_guard(hook: object) -> bool:
      return (
          isinstance(hook, dict)
          and isinstance(hook.get("command"), str)
          and GUARD_COMMAND.search(hook["command"]) is not None
      )
  
  
  def merge_guard(settings: dict[str, Any], guard_path: Path) -> dict[str, Any]:
      hooks = settings.setdefault("hooks", {})
      entries = hooks.setdefault("PreToolUse", [])
      # Always absolute: Claude runs hooks with cwd=project, so relative paths like
      # "hooks/guard_shell.py" fail as /bin/sh: hooks/guard_shell.py: not found.
      guard_abs = str(Path(guard_path).expanduser().resolve())
      guard = {
          "type": "command",
          "command": f"AGENT_HOOK_CLIENT=claude {shlex.quote(guard_abs)}",
          "timeout": 2,
      }
      merged: list[object] = []
      found = False
      for entry in entries:
          if not isinstance(entry, dict) or not isinstance(entry.get("hooks"), list):
              merged.append(entry)
              continue
          remaining = [hook for hook in entry["hooks"] if not is_guard(hook)]
          if len(remaining) == len(entry["hooks"]):
              merged.append(entry)
          elif not found:
              if remaining:
                  merged.append({**entry, "hooks": remaining})
              merged.append({"matcher": MATCHER, "hooks": [guard]})
              found = True
          elif remaining:
              merged.append({**entry, "hooks": remaining})
      if not found:
          merged.append({"matcher": MATCHER, "hooks": [guard]})
      hooks["PreToolUse"] = merged
      return settings
  
  
  PREV_STATUSLINE = Path.home() / ".agents" / "state" / "statusline.prev.json"
  
  
  def _ours_statusline(cmd: str) -> bool:
      return "lane-statusline" in cmd or "claude_status" in cmd or "claude-pulse" in cmd
  
  
  def merge_statusline(settings: dict[str, Any], statusline_path: Path) -> dict[str, Any]:
      """Wire lane-statusline router (pulse for normal, HUD for orchestrators).
  
      Overwrites previous lane-statusline / claude-pulse statusLine; leaves other
      custom statusLine commands alone only if LANE_STATUSLINE_FORCE=0.
      Default: always set (install owns the status line for stack users).
      """
      force = os.environ.get("LANE_STATUSLINE_FORCE", "1").strip() not in {
          "0",
          "false",
          "no",
      }
      existing = settings.get("statusLine")
      if not force and isinstance(existing, dict):
          cmd = str(existing.get("command") or "")
          ours = "lane-statusline" in cmd or "claude_status" in cmd or "claude-pulse" in cmd
          if cmd and not ours:
              return settings
  
      if isinstance(existing, dict) and existing.get("command") and not _ours_statusline(str(existing.get("command"))):
          # Restore the own status line: the one a person had before is kept for lane-stack-remove.
          PREV_STATUSLINE.parent.mkdir(parents=True, exist_ok=True)
          PREV_STATUSLINE.write_text(json.dumps(existing, ensure_ascii=False), encoding="utf-8")
      path = statusline_path.expanduser().resolve()
      settings["statusLine"] = {
          "type": "command",
          "command": f"python3 {shlex.quote(str(path))}",
          "padding": 0,
      }
      return settings
  
  
  SESSION_MARK_RE = re.compile(r"lane_statusline_session\.py")
  TEAMMATE_IDLE_RE = re.compile(r"teammate_idle_sentinel\.py")
  SESSION_LEDGER_RE = re.compile(r"session_ledger\.py")
  PM_STOP_RE = re.compile(r"pm_stop_sentinel\.py")
  PM_BULK_READ_RE = re.compile(r"pm_bulk_read\.py")
  STALL_CHECK_RE = re.compile(r"stall_check\.py")
  OWNER_PRESENCE_RE = re.compile(r"owner_presence\.py")
  SMA_HOOKS_RE = re.compile(r"sma_hooks\.py")
  GITNEXUS_HOOK_RE = re.compile(r"gitnexus-hook")
  GITNEXUS_HOOK_FILES = (
      "gitnexus-hook.cjs",
      "hook-lock.cjs",
      "hook-db-lock-probe.cjs",
      "resolve-analyze-cmd.cjs",
      "registry-query.cjs",
      "win-rm-list-json.ps1",
  )
  GITNEXUS_CLI_PATH_LITERAL = (
      "let cliPath = path.resolve(__dirname, '..', '..', 'dist', 'cli', 'index.js');"
  )
  
  
  def _replace_event_hooks(
      settings: dict[str, Any],
      event: str,
      command: str,
      drop_re: re.Pattern[str],
      *,
      timeout: int = 2,
  ) -> None:
      """Idempotently set a single-command hook entry for an event (drop prior ours)."""
      hooks = settings.setdefault("hooks", {})
      entry = {
          "hooks": [
              {
                  "type": "command",
                  "command": command,
                  "timeout": timeout,
              }
          ]
      }
      entries = hooks.setdefault(event, [])
      if not isinstance(entries, list):
          hooks[event] = [entry]
          return
      cleaned: list[object] = []
      for item in entries:
          if not isinstance(item, dict):
              cleaned.append(item)
              continue
          hlist = item.get("h

• hooks/owner_presence.pyGitHub ↗
• hooks/pm_bulk_read.pyGitHub ↗
• hooks/pm_stop_sentinel.pyGitHub ↗
• hooks/session_ledger.pyGitHub ↗
• hooks/stall_check.pyGitHub ↗
• hooks/teammate_idle_sentinel.pyGitHub ↗

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