Hooks

Where it lives

• hooks/nav_dispatch.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """nav_dispatch — single dispatcher entrypoint for every Navigator hook event (TASK-60).
  
  plugin.json routes the seven v6 surfaces here: python3 nav_dispatch.py <EventName>.
  Thin shim: read stdin once, delegate to nav_hook_lib.runtime.dispatch(), relay the
  single JSON doc + stderr (sentinels.emit_stderr, mem-034) + exit code. Fail-open:
  ANY failure — bad argv/stdin, missing runtime, escaping SystemExit, broken stdout
  pipe — exits 0 (BaseException catch-all: hooks are non-interactive). stdout flushes
  INSIDE the guard so a closed pipe cannot become exit 120 at interpreter shutdown.
  """
  import os
  import sys
  
  
  def main() -> int:
      sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
      from nav_hook_lib import hio, runtime, sentinels
  
      event = sys.argv[1]
      payload = hio.read_stdin_payload()
      result = runtime.dispatch(event, payload)
      if result.stdout is not None:
          print(result.stdout)
      sys.stdout.flush()  # broken pipe raises HERE, inside the catch-all
      if result.stderr is not None:
          sentinels.emit_stderr(result.stderr)
      return result.exit_code
  
  
  if __name__ == "__main__":
      try:
          exit_code = main()
      except BaseException:
          exit_code = 0  # fail-open: Navigator must never brick the harness
      try:
          sys.stdout.flush()
      except BaseException:
          os._exit(exit_code)  # dead stdout: bypass shutdown flush (exit-120 guard)
      sys.exit(exit_code)
  

• hooks/test_nav_dispatch.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Subprocess contract harness for hooks/nav_dispatch.py (TASK-60 Phase 5).
  
  Template: hooks/test_workflow_enforcer.py (TASK-45 precedent) — each test
  builds a throwaway project dir and drives the dispatcher as a subprocess
  (`python3 hooks/nav_dispatch.py <event>`) with a JSON payload on stdin,
  asserting on stdout/exit code only. stdlib unittest (pytest not installed).
  
  Mandatory cases from the TASK-60 acceptance list covered here:
  
    - Pristine v6.18.1 config fixture -> exit 0 on every v6 event surface,
      with at most ONE JSON document on stdout.
    - Project with NO .agent/ -> exit 0 AND empty stdout on every event, and
      the dispatcher never CREATES .agent/ (silent degradation).
    - PILOT_EXECUTOR=1 -> no blocking output on any event. NOTE: TASK-61 ops
      do not exist yet, so this is the dispatch-level synthetic check (clean
      exit, no block/deny/continue:false emissions); op-level bypass cases
      (e.g. read_guard deny suppressed under Pilot) land with the TASK-61
      ports that plug into this harness.
    - Per-behavior off-switch: a config with every v6 toggle block disabled
      (read_guard_hook.enabled=false etc.) -> exit 0 everywhere, and the
      config gate skips ops BEFORE import (no missing-module op_errors notes
      for gated-off ops).
    - Malformed stdin fails open (exit 0); missing/unknown event arg -> exit
      0, no output.
    - Crash-injection: NOT feasible at the subprocess level right now — the
      real op modules land in TASK-61, so there is nothing on the live
      registry to crash. Instead a tmp-dir driver script imports
      nav_hook_lib.runtime directly and dispatches a SYNTHETIC registry
      (contract: the `registry` param overrides EVENT_OPS for testability)
      with fake op modules: crash isolation, sentinel stderr hygiene
      (mem-034: no payload echo), meta.op_errors, health-file write +
      SessionStart surfacing, and gate short-circuit of rightward phases.
    - Timing: UserPromptSubmit dispatched 20x against
      nav_hook_lib/fixtures/timing_prompt.json; p95 <= 200ms * NAV_TIMING_MULT
      (env, default 1 — set >1 on slow CI runners).
  
  Out of scope here (other TASK-60 groups): mem-036 three-env-variant manifest
  command tests (need .claude-plugin/plugin.json — integrator), state
  read-once/write-once unit verification (runtime builder's colocated tests).
  """
  
  import json
  import math
  import os
  import subprocess
  import sys
  import tempfile
  import time
  import unittest
  from pathlib import Path
  
  HOOKS_DIR = Path(__file__).resolve().parent
  HOOK = str(HOOKS_DIR / "nav_dispatch.py")
  FIXTURES = HOOKS_DIR / "nav_hook_lib" / "fixtures"
  PRISTINE_CONFIG = (FIXTURES / "nav-config-v6.18.1.json").read_text()
  TIMING_FIXTURE = FIXTURES / "timing_prompt.json"
  
  # The seven v6 manifest event surfaces (registry.EVENT_OPS keys — TASK-60
  # registers ONLY these; new routing-matrix events belong to TASK-62).
  EVENTS = (
      "SessionStart",
      "UserPromptSubmit",
      "PreToolUse",
      "PostToolUse",
      "Stop",
      "PreCompact",
      "PostCompact",
  )
  
  # Realistic per-event payload shapes (tool_name chosen to HIT the coarse
  # registry matchers: 'Read' and 'Edit|Write|MultiEdit|NotebookEdit').
  EVENT_PAYLOAD_EXTRAS = {
      "SessionStart": {"source": "startup"},
      "UserPromptSubmit": {"prompt": "hello there"},
      "PreToolUse": {"tool_name": "Read", "tool_input": {"file_path": "/tmp/x.md"}},
      "PostToolUse": {
          "tool_name": "Edit",
          "tool_input": {"file_path": "/tmp/x.py"},
          "tool_response": {},
      },
      "Stop": {"stop_hook_active": False},
      "PreCompact": {"trigger": "manual"},
      "PostCompact": {},
  }
  
  # v6 toggle block names (from fixtures/nav-config-v6.18.1.json) -> the op
  # names the registry gates on them. Used by the off-switch tests.
  TOGGLE_BLOCKS = (
      "session_start_hook",
      "workflow_enforcer_hook",
      "brief_hook",
      "read_guard_hook",
      "task_graph_sync_hook",
      "profile_sync_hook",
      "workflow_state_hook",
      "compact_hook",
  )
  GATED_OPS = (
      "session_start",
      "prompt_gate",
      "prompt_brief",
      "read_guard",
      "graph_sync",
      "profile_sync",
      "stop_state",
      "compact_marker",
  )
  ALL_TOGGLES_OFF = json.dumps({block: {"enabled": False} for block in TOGGLE_BLOCKS})
  
  
  def clean_env(extra=None):
      """Subprocess env with the escape hatches / project redirects removed.
  
      CLAUDE_PROJECT_DIR must go: with malformed stdin the payload is {}, and
      an inherited value would redirect hio.resolve_cwd() at the REAL repo.
      """
      env = os.environ.copy()
      for var in ("PILOT_EXECUTOR", "CLAUDE_PROJECT_DIR", "CLAUDE_USER_MESSAGE"):
          env.pop(var, None)
      if extra:
          env.update(extra)
      return env
  
  
  def run_dispatch(project_dir, event=None, payload=None, raw_stdin=None, env_extra=None):
      """Invoke `python3 nav_dispatch.py [<event>]` rooted at project_dir.
  
      payload: sent as JSON on stdin ({} when None).
      raw_stdin: sent verbatim (overrides payload) — malformed-input cases.
      event: omitted from argv entirely when None (missing-arg case).
      """
      argv = [sys.executable, HOOK]
      if event is not None:
          argv.append(event)
      if raw_stdin is None:
          raw_stdin = json.dumps(payload if payload is not None else {})
      return subprocess.run(
          argv,
          input=raw_stdin,
          capture_output=True,
          text=True,
          cwd=project_dir,
          env=clean_env(env_extra),
      )
  
  
  def event_payload(project, event, session_id="s1"):
      payload = {"cwd": project, "session_id": session_id}
      payload.update(EVENT_PAYLOAD_EXTRAS.get(event, {}))
      return payload
  
  
  def parse_single_doc(testcase, stdout):
      """Contract: EXACTLY ONE JSON doc on stdout, or nothing. Returns doc/None.
  
      json.loads rejects concatenated documents, so a second doc fails here.
      """
      if not stdout.strip():
          return None
      try:
          doc = json.loads(stdout)
      except json.JSONDecodeError:
          testcase.fail(f"stdout is not a single JSON document: {stdout!r}")
  

• hooks/test_nav_dispatch_manifest.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Manifest + shim contract tests for nav_dispatch (TASK-60, shim-manifest group).
  
  stdlib unittest only, subprocess-driven per the test_workflow_enforcer.py
  template. Three concerns:
  
  1. Manifest shape — .claude-plugin/plugin.json references ONLY
     hooks/nav_dispatch.py, registers exactly the seven v6 event surfaces
     plus the six TASK-62 events that survived the validate-or-drop gate
     (SubagentStart, PostToolUseFailure, TaskCreated, TaskCompleted,
     ConfigChange, Setup — all accepted by `claude plugin validate` on CC
     2.1.205), with the contract matchers (PreToolUse: Read; PostToolUse:
     Edit|Write|MultiEdit|NotebookEdit; no matcher on any TASK-62 event) and
     timeouts (SessionStart 10, PostToolUse 10, PreCompact 30, PostCompact 10
     — the v6 allowances — all others 5, including every TASK-62 event).
     Each command is the fail-OPEN shell guard: the dispatcher
     file is [ -f ]-checked before exec, so a resolution miss exits 0
     silently instead of python3's loud exit 2 (which the harness treats as
     a block on every event).
  
  2. mem-036 env variants — every event's LITERAL manifest command runs as a
     subprocess (sh -c, so ${VAR:-fallback} expands exactly as the harness
     would) under three CLAUDE_PLUGIN_ROOT variants:
       * set-to-repo: resolves to this checkout; the dispatcher must exit 0 on
         an empty payload (fail-open), and a seeded health file proves the
         guard actually DISPATCHES (not silently skips) when the file exists.
       * unset: sh substitutes the $HOME fallback. HOME is pointed at a temp
         dir carrying a STALE marketplace clone (pre-v7: no nav_dispatch.py) —
         the realistic fallback state. The guard must exit 0 silently: the old
         loud-exit-2 behavior fails CLOSED and blocks every event.
       * empty string: the :- operator treats empty like unset, so behavior
         must match the unset variant (same fallback resolution).
  
  3. Shim contract — hooks/nav_dispatch.py stays under 40 lines and exits 0
     with no output on: empty JSON stdin, garbage stdin, missing event arg,
     unknown event (fail-open; must hold with or without nav_hook_lib.runtime).
  """
  
  import json
  import os
  import re
  import subprocess
  import tempfile
  import unittest
  from pathlib import Path
  
  HOOKS_DIR = Path(__file__).resolve().parent
  REPO_ROOT = HOOKS_DIR.parent
  SHIM = HOOKS_DIR / "nav_dispatch.py"
  PLUGIN_JSON = REPO_ROOT / ".claude-plugin" / "plugin.json"
  
  EVENTS = (
      "SessionStart",
      "UserPromptSubmit",
      "PreToolUse",
      "PostToolUse",
      "Stop",
      "PreCompact",
      "PostCompact",
      # TASK-62 event surfaces (validate-or-drop survivors, CC 2.1.205):
      "SubagentStart",
      "PostToolUseFailure",
      "TaskCreated",
      "TaskCompleted",
      "ConfigChange",
      "Setup",
  )
  # v6 allowances: PostToolUse hooks had 10s EACH, PostCompact 10s — the shared
  # dispatcher must not regress them to the 5s default.
  EXPECTED_TIMEOUTS = {
      "SessionStart": 10, "PostToolUse": 10, "PreCompact": 30, "PostCompact": 10,
  }
  DEFAULT_TIMEOUT = 5
  EXPECTED_MATCHERS = {
      "PreToolUse": "Read",
      "PostToolUse": "Edit|Write|MultiEdit|NotebookEdit",
  }
  # Fail-OPEN shell guard: missing dispatcher file -> exit 0 silently (stdin
  # passes through the exec). %s is the event name.
  COMMAND_GUARD = (
      "sh -c 'f=\"${CLAUDE_PLUGIN_ROOT:-"
      "$HOME/.claude/plugins/marketplaces/navigator-marketplace}"
      "/hooks/nav_dispatch.py\"; "
      "if [ -f \"$f\" ]; then exec python3 \"$f\" %s; fi'"
  )
  FALLBACK_RELPATH = ".claude/plugins/marketplaces/navigator-marketplace"
  # Wedge guard: generous vs the 5-30s manifest timeouts; TimeoutExpired = fail.
  SUBPROCESS_TIMEOUT = 20
  
  
  def load_manifest():
      return json.loads(PLUGIN_JSON.read_text(encoding="utf-8"))
  
  
  def single_hook_entry(testcase, manifest, event):
      """Assert the event has exactly one matcher group with one hook; return both."""
      groups = manifest["hooks"][event]
      testcase.assertEqual(len(groups), 1, f"{event}: expected exactly one matcher group")
      hooks = groups[0]["hooks"]
      testcase.assertEqual(len(hooks), 1, f"{event}: expected exactly one hook entry")
      return groups[0], hooks[0]
  
  
  def clean_env(extra=None, drop=()):
      env = os.environ.copy()
      for key in ("PILOT_EXECUTOR", "CLAUDE_PROJECT_DIR", "CLAUDE_USER_MESSAGE"):
          env.pop(key, None)
      for key in drop:
          env.pop(key, None)
      if extra:
          env.update(extra)
      return env
  
  
  def run_shim(args, stdin="", env=None, cwd=None):
      return subprocess.run(
          ["python3", str(SHIM), *args],
          input=stdin,
          capture_output=True,
          text=True,
          cwd=cwd or REPO_ROOT,
          env=env or clean_env(),
          timeout=SUBPROCESS_TIMEOUT,
      )
  
  
  def run_manifest_command(command, env, cwd, stdin="{}"):
      """Run the literal manifest command through sh -c, as the harness does."""
      return subprocess.run(
          ["/bin/sh", "-c", command],
          input=stdin,
          capture_output=True,
          text=True,
          cwd=cwd,
          env=env,
          timeout=SUBPROCESS_TIMEOUT,
      )
  
  
  class ManifestShapeTest(unittest.TestCase):
      """plugin.json routes every event through nav_dispatch.py, nothing else."""
  
      def setUp(self):
          self.manifest = load_manifest()
  
      def test_only_nav_dispatch_referenced(self):
          raw = PLUGIN_JSON.read_text(encoding="utf-8")
          scripts = set(re.findall(r"hooks/[A-Za-z0-9_]+\.py", raw))
          self.assertEqual(scripts, {"hooks/nav_dispatch.py"})
  
      def test_exactly_the_registered_event_surfaces(self):
          # Seven v6 surfaces + six TASK-62 validate-or-drop survivors; every
          # registration maps to a committed dispatcher (v5.1.0 lesson) and to
          # a registry EVENT_OPS row (asserted in nav_hook_lib/test_registry).
          self.assertEqual(set(self.manifest["hooks"].keys()), set(EVENTS))
  
      def test_command_shape_per_event(self):
          for event in EVENTS:
              _, hook = single_hook_entry(self, self.manifest, event)
           

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.