Hooks

Where it lives

• hooks/conftest.pyGitHub ↗
  Read the script

  """Shared fixtures for hook tests."""
  
  from __future__ import annotations
  
  import textwrap
  
  import pytest
  
  VALID_STANDARD_BRIEF = textwrap.dedent("""\
      == TURNOVER BRIEF ==
      Ship: HMS Daring (destroyer)
      Role: Captain 1 — auth implementation
      Timestamp: 2026-01-01T12:00:00Z
      Reason for relief: Red hull
  
      Mission context:
      - Mission: Build auth system
      - Outcome: Working JWT auth
      - Success metric: All tests pass
  
      Task assignment:
      - Task ID: task-1
      - Task name: Build auth
      - Deliverable: JWT validation module
      - Action station: 1
      - File ownership: src/auth.py
      - Dependencies: none
  
      Progress log:
      - Implemented JWT validation in src/auth/validate.ts
      - Added tests in tests/auth.test.ts
  
      Running plot:
      - Working on refresh token rotation
      - Current state: halfway through implementation
  
      Files touched:
      - src/auth/validate.ts — JWT validation logic
      - tests/auth.test.ts — Test coverage
  
      Key decisions made:
      - Chose RS256 over HS256 — Rationale: asymmetric signing for microservices
  
      Hazards and blockers:
      - None discovered
  
      Recommended course of action:
      - Complete refresh token rotation
      - Add integration tests
  
      Relief chain:
      - This is the first ship on this task
  
      == END TURNOVER BRIEF ==
  """)
  
  VALID_FLAGSHIP_BRIEF = textwrap.dedent("""\
      == FLAGSHIP TURNOVER BRIEF ==
      Ship: Flagship HMS Victory
      Role: Admiral
      Timestamp: 2026-01-01T12:00:00Z
      Reason for relief: Amber hull at 65%
  
      Sailing orders:
      - Outcome: Build complete auth system
      - Success metric: All tests pass
      - Deadline: EOD
      - Constraints: None
      - Out of scope: OAuth
  
      Battle plan status:
      - Task task-1: Build auth | Owner: HMS Daring | Status: in_progress | Notes: on track
  
      Squadron state:
      - HMS Daring (destroyer) | Captain 1 | Task: task-1 | Hull: Green | Status: active
  
      Key decisions made:
      - Chose JWT over sessions — Rationale: stateless auth
  
      Active blockers and risks:
      - None
  
      Pending escalations:
      - None
  
      Quarterdeck rhythm:
      - Cadence: every 15 minutes
      - Last checkpoint: checkpoint 3
      - Next scheduled checkpoint: checkpoint 4
  
      Relief chain:
      - This is the first admiral on this mission
  
      Recommended course of action:
      - Continue monitoring HMS Daring progress
      - Check hull integrity at next checkpoint
  
      == END FLAGSHIP TURNOVER BRIEF ==
  """)
  
  
  @pytest.fixture()
  def standard_brief() -> str:
      return VALID_STANDARD_BRIEF
  
  
  @pytest.fixture()
  def flagship_brief() -> str:
      return VALID_FLAGSHIP_BRIEF
  

• hooks/nelson_hooks.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  Nelson Hook Enforcement Script.
  
  Enforces Nelson's structural guarantees at the tool level via Claude Code
  hooks. Each subcommand maps to a hook event type:
  
    preflight      — PreToolUse on Agent: station tier gate, file ownership
                     conflicts, mode-tool consistency
    brief-validate — PostToolUse on Write/Edit: turnover brief quality gate
    task-complete  — TaskCompleted: validation evidence and station controls
    idle-ship      — TeammateIdle: paid-off standing order advisory
    session-init   — SessionStart: record admiral transcript_path for the
                     TaskCreate captain-misuse gate
    session-check  — PreToolUse on TaskCreate: reject captain TaskCreate calls
                     in subagents/single-session mode (admiral exception via
                     the marker written by session-init)
  
  Exit codes:
    0 — allow (action proceeds)
    2 — reject (action blocked, stderr feedback sent to agent)
  
  All hooks degrade gracefully: if no active Nelson mission is found, they
  exit 0 and do not interfere with non-Nelson workflows.
  
  Requires only Python stdlib (no pip dependencies).
  """
  
  from __future__ import annotations
  
  import argparse
  import json
  import re
  import sys
  from datetime import UTC, datetime
  from pathlib import Path
  from typing import Any, NoReturn
  
  # ---------------------------------------------------------------------------
  # Shared utilities
  # ---------------------------------------------------------------------------
  
  
  ADMIRAL_SESSION_MARKER = "admiral.session"
  # NOTE: must stay in sync with
  # skills/nelson/scripts/nelson_data_utils.py:ADMIRAL_SESSION_MARKER.
  
  
  def _read_stdin() -> dict[str, Any]:
      """Parse JSON from stdin (Claude Code hook payload)."""
      try:
          raw = sys.stdin.read()
          if not raw.strip():
              return {}
          return json.loads(raw)
      except (json.JSONDecodeError, OSError):
          return {}
  
  
  def _find_mission_dir(cwd: Path) -> Path | None:
      """Find the active Nelson mission directory.
  
      Scans for ``.nelson/.active-*`` marker files in the working directory.
      Returns the mission directory path if exactly one active session exists,
      or None if no active mission is found.
      """
      nelson_dir = cwd / ".nelson"
      if not nelson_dir.is_dir():
          return None
  
      active_files = sorted(nelson_dir.glob(".active-*"))
      if not active_files:
          return None
  
      try:
          mission_path = active_files[0].read_text(encoding="utf-8").strip()
          mission_dir = Path(mission_path)
          if mission_dir.is_dir():
              return mission_dir
      except OSError:
          pass
  
      return None
  
  
  def _read_json(path: Path) -> dict[str, Any]:
      """Read and parse a JSON file. Returns empty dict on failure."""
      try:
          return json.loads(path.read_text(encoding="utf-8"))
      except (json.JSONDecodeError, OSError):
          return {}
  
  
  def _reject(message: str) -> NoReturn:
      """Print rejection feedback to stderr and exit with code 2."""
      print(message, file=sys.stderr)
      sys.exit(2)
  
  
  def _allow() -> NoReturn:
      """Exit successfully — action proceeds."""
      sys.exit(0)
  
  
  def _get_mode(battle_plan: dict[str, Any]) -> str:
      """Extract execution mode from battle plan."""
      return battle_plan.get("squadron", {}).get("mode", "subagents")
  
  
  def _get_tasks(battle_plan: dict[str, Any]) -> list[dict[str, Any]]:
      """Extract tasks list from battle plan."""
      return battle_plan.get("tasks", [])
  
  
  def _load_mission_context(
      payload: dict[str, Any],
  ) -> tuple[Path, dict[str, Any]] | None:
      """Load battle plan from the active mission. Returns None if unavailable."""
      cwd = Path(payload.get("cwd", "."))
      mission_dir = _find_mission_dir(cwd)
      if mission_dir is None:
          return None
      bp = _read_json(mission_dir / "battle-plan.json")
      if not bp:
          return None
      return mission_dir, bp
  
  
  def _write_admiral_marker(nelson_dir: Path, transcript_path: str) -> bool:
      """Write the admiral session marker. Returns True on success, False on failure.
  
      The marker stores the admiral's transcript_path so cmd_session_check can
      distinguish admiral calls (match) from captain subagent calls (mismatch).
      """
      if not nelson_dir.is_dir():
          return False
      if not transcript_path.strip():
          return False
      try:
          (nelson_dir / ADMIRAL_SESSION_MARKER).write_text(
              transcript_path.strip() + "\n",
              encoding="utf-8",
          )
          return True
      except OSError:
          return False
  
  
  # ---------------------------------------------------------------------------
  # Preflight helpers
  # ---------------------------------------------------------------------------
  
  
  def _check_station_tiers(tasks: list[dict[str, Any]]) -> str | None:
      """Return rejection message if any task lacks a station tier."""
      unclassified = [t for t in tasks if t.get("station_tier") is None]
      if not unclassified:
          return None
      names = ", ".join(f"'{t.get('name', t.get('id', 'unknown'))}'" for t in unclassified)
      return (
          f"Standing order violation (unclassified-engagement): "
          f"Tasks without station tier: {names}. "
          f"Classify all tasks via references/action-stations.md before "
          f"spawning agents."
      )
  
  
  def _check_file_ownership(tasks: list[dict[str, Any]]) -> str | None:
      """Return rejection message if any file is owned by multiple captains."""
      file_owners: dict[str, list[str]] = {}
      for task in tasks:
          owner = task.get("owner", "unknown")
          for filepath in task.get("file_ownership", []):
              file_owners.setdefault(filepath, []).append(owner)
  
      conflicts = {fp: owners for fp, owners in file_owners.items() if len(set(owners)) > 1}
      if not conflicts:
          return None
      details = "; ".join(f"'{fp}' owned by {', '.join(sorted(set(owners)))}" for fp, owners in conflicts.items())
      return (
          f"Standing order violation (split-keel

• hooks/test_nelson_hooks.pyGitHub ↗
  Read the script

  """Tests for nelson_hooks.py — hook enforcement script.
  
  Tests the preflight, brief-validate, task-complete, and idle-ship
  subcommands using temporary mission directories and monkeypatched stdin.
  """
  
  from __future__ import annotations
  
  import argparse
  import json
  import os
  import sys
  from io import StringIO
  from pathlib import Path
  from typing import Any
  from unittest.mock import patch
  
  import pytest
  
  # Add hooks directory to path for imports
  hooks_dir = os.path.dirname(os.path.abspath(__file__))
  if hooks_dir not in sys.path:
      sys.path.insert(0, hooks_dir)
  
  from conftest import VALID_FLAGSHIP_BRIEF, VALID_STANDARD_BRIEF
  from nelson_hooks import (
      ADMIRAL_SESSION_MARKER,
      ROLLBACK_PATTERNS,
      VALIDATION_EVIDENCE_PATTERNS,
      _check_running_plot_nonempty,
      _check_section_present,
      _find_mission_dir,
      _get_mode,
      _get_tasks,
      _has_evidence,
      cmd_brief_validate,
      cmd_idle_ship,
      cmd_preflight,
      cmd_session_check,
      cmd_session_init,
      cmd_task_complete,
  )
  
  # ---------------------------------------------------------------------------
  # Helpers
  # ---------------------------------------------------------------------------
  
  
  def _make_mission(
      tmp_path: Path,
      *,
      mode: str = "subagents",
      tasks: list[dict[str, Any]] | None = None,
      fleet_status: dict[str, Any] | None = None,
  ) -> Path:
      """Create a minimal active Nelson mission directory."""
      nelson_dir = tmp_path / ".nelson"
      nelson_dir.mkdir(exist_ok=True)
      mission_dir = nelson_dir / "missions" / "2026-01-01_120000_test"
      mission_dir.mkdir(parents=True, exist_ok=True)
      (nelson_dir / ".active-test-session").write_text(
          str(mission_dir),
          encoding="utf-8",
      )
      battle_plan: dict[str, Any] = {
          "version": 1,
          "squadron": {
              "mode": mode,
              "admiral": {"ship_name": "HMS Victory", "model": "opus"},
              "captains": [
                  {
                      "ship_name": "HMS Daring",
                      "ship_class": "destroyer",
                      "model": "sonnet",
                      "task_id": "task-1",
                  },
              ],
          },
          "tasks": tasks or [],
      }
      (mission_dir / "battle-plan.json").write_text(
          json.dumps(battle_plan),
          encoding="utf-8",
      )
      if fleet_status is not None:
          (mission_dir / "fleet-status.json").write_text(
              json.dumps(fleet_status),
              encoding="utf-8",
          )
      return mission_dir
  
  
  def _stdin(payload: dict[str, Any], cwd: str = ".") -> StringIO:
      return StringIO(json.dumps({"cwd": cwd, **payload}))
  
  
  def _run(cmd_fn: Any, payload: dict[str, Any], cwd: str = ".") -> int:
      """Run a hook command and return its exit code."""
      with patch("sys.stdin", _stdin(payload, cwd)), pytest.raises(SystemExit) as exc:
          cmd_fn(argparse.Namespace())
      return exc.value.code
  
  
  # ---------------------------------------------------------------------------
  # Shared utilities
  # ---------------------------------------------------------------------------
  
  
  class TestFindMissionDir:
      def test_no_nelson_dir(self, tmp_path: Path) -> None:
          assert _find_mission_dir(tmp_path) is None
  
      def test_no_active_files(self, tmp_path: Path) -> None:
          (tmp_path / ".nelson").mkdir()
          assert _find_mission_dir(tmp_path) is None
  
      def test_active_mission_found(self, tmp_path: Path) -> None:
          mission_dir = _make_mission(tmp_path)
          assert _find_mission_dir(tmp_path) == mission_dir
  
      def test_active_file_points_to_missing_dir(self, tmp_path: Path) -> None:
          nelson_dir = tmp_path / ".nelson"
          nelson_dir.mkdir()
          (nelson_dir / ".active-test").write_text("/nonexistent/path", encoding="utf-8")
          assert _find_mission_dir(tmp_path) is None
  
  
  class TestGetMode:
      def test_default_mode(self) -> None:
          assert _get_mode({}) == "subagents"
  
      def test_explicit_mode(self) -> None:
          assert _get_mode({"squadron": {"mode": "agent-team"}}) == "agent-team"
  
  
  class TestGetTasks:
      def test_no_tasks(self) -> None:
          assert _get_tasks({}) == []
  
      def test_with_tasks(self) -> None:
          tasks = [{"id": "task-1", "name": "Test"}]
          assert _get_tasks({"tasks": tasks}) == tasks
  
  
  class TestHasEvidence:
      def test_finds_test_keyword(self) -> None:
          assert _has_evidence("All tests passed", VALIDATION_EVIDENCE_PATTERNS)
  
      def test_finds_rollback(self) -> None:
          assert _has_evidence("Rollback: git revert abc", ROLLBACK_PATTERNS)
  
      def test_no_match(self) -> None:
          assert not _has_evidence("Hello world", ROLLBACK_PATTERNS)
  
      def test_case_insensitive(self) -> None:
          assert _has_evidence("VERIFIED output", VALIDATION_EVIDENCE_PATTERNS)
  
  
  class TestCheckSectionPresent:
      def test_present(self) -> None:
          assert _check_section_present("Ship: HMS Daring\n", "Ship:")
  
      def test_absent(self) -> None:
          assert not _check_section_present("Role: Captain\n", "Ship:")
  
      def test_case_insensitive(self) -> None:
          assert _check_section_present("ship: HMS Daring\n", "Ship:")
  
  
  class TestCheckRunningPlotNonempty:
      def test_nonempty(self) -> None:
          assert _check_running_plot_nonempty(
              "Running plot:\n- Working on auth module\n\nFiles touched:",
          )
  
      def test_empty_section(self) -> None:
          assert not _check_running_plot_nonempty("Running plot:\n\nFiles touched:")
  
      def test_missing_section(self) -> None:
          assert not _check_running_plot_nonempty("Progress log:\n- Did stuff\n")
  
      def test_section_with_only_whitespace(self) -> None:
          assert not _check_running_plot_nonempty("Running plot:\n   \n\nFiles touched:")
  
  
  # ---------------------------------------------------------------------------
  # Preflight
  # ---------------------------------------------------------------------------
  
  
  def _task(
      tid: str = "task-1",
      name: str = "Build auth",
      ti

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.