Hooks

What navigator runs automatically, and when. A hook is a command Claude Code fires at a fixed moment, without you asking for it.

alekspetrov@alekspetrov

GitHub

From plugin

navigator

19631 skills2 agents16 hooks

Install

$ npx -y skills add alekspetrov/navigator --agent claude-code

Ships with navigator. Installing the plugin gets these hooks.

Where it lives

hooks/nav_brief.pyGitHub

Read the script

#!/usr/bin/env python3
"""
Navigator Intent-Brief Hook (TASK-56)

Claude Code hook that runs on UserPromptSubmit, scores the prompt for
ambiguity, and — when a task-shaped prompt scores at/above threshold —
instructs the model to render a one-screen INTENT BRIEF before implementing,
pre-filled from knowledge-graph memories relevant to the prompt.

Composes with hooks/workflow_enforcer.py (a sibling entry in plugin.json's
UserPromptSubmit array); this hook is deliberately read-only and non-blocking:

  - ALWAYS exits 0. Exit 2 on UserPromptSubmit blocks the prompt before the
    model runs (stderr reaches only the user), which would silently defeat
    a feature whose whole point is instructing the model. See mem-034.
  - Below threshold / question / confirmation / disabled -> prints nothing.
  - Memory recall degrades silently (missing graph, helper failure, timeout).

Input (from Claude Code):
    stdin JSON: {"prompt": "...", "cwd": "..."} for UserPromptSubmit

Output:
    stdout: NAV-BRIEF instruction block (surfaced into the model's context).
    exit 0 always.
"""

import json
import os
import re
import subprocess
import sys
from pathlib import Path

# workflow_enforcer's blocked-stderr sentinel: Claude Code echoes blocked
# stderr into the next prompt's context; excise it before scoring so echoed
# instruction text can't fake a task-shaped prompt (mem-034).
NAV_BLOCK_PATTERN = re.compile(
    re.escape("<nav-workflow-block>") + r".*?" + re.escape("</nav-workflow-block>"),
    re.DOTALL,
)

RECALL_TIMEOUT = 3
RECALL_LIMIT = 5
DEFAULT_THRESHOLD = 0.5
DEFAULT_MEMORY_BUDGET = 1200
MAX_CONCEPTS = 8

CONCEPT_STOPWORDS = {
    "this", "that", "these", "those", "with", "from", "into", "onto",
    "over", "under", "about", "after", "before", "please", "then",
    "them", "they", "their", "have", "been", "will", "would", "should",
    "could", "make", "sure", "need", "want", "like", "some", "more",
    "when", "what", "where", "which", "until", "done", "everything",
}

sys.path.insert(0, str(Path(__file__).parent.parent / "skills" / "nav-brief" / "functions"))

try:
    from ambiguity_scorer import score_ambiguity
except ImportError as exc:
    print(f"nav_brief: ambiguity_scorer import failed ({exc}); hook disabled this invocation.",
          file=sys.stderr)

    def score_ambiguity(prompt):
        return {"score": 0.0, "task_shaped": False,
                "undefined_dimensions": [], "matched_signals": []}


def read_stdin_payload() -> dict:
    """Read and parse the UserPromptSubmit stdin JSON exactly once.

    Mirrors workflow_enforcer.read_stdin_payload: {} when stdin absent/empty,
    {"prompt": raw} when the body is non-JSON text.
    """
    try:
        import select
        if select.select([sys.stdin], [], [], 0)[0]:
            raw = sys.stdin.read().strip()
            if raw:
                try:
                    return json.loads(raw)
                except json.JSONDecodeError:
                    return {"prompt": raw}
    except Exception:
        pass
    return {}


def get_user_message(stdin_data: dict) -> str:
    prompt = stdin_data.get("prompt") or stdin_data.get("user_message") or ""
    if not prompt:
        prompt = os.environ.get("CLAUDE_USER_MESSAGE", "")
    if "<nav-workflow-block>" in prompt:
        prompt = NAV_BLOCK_PATTERN.sub("", prompt)
    return prompt


def _project_root(stdin_data: dict) -> Path:
    """Resolve the project root the same way every other Navigator hook does."""
    cwd = stdin_data.get("cwd") or os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd()
    return Path(cwd)


def check_config(root: Path) -> dict:
    config_path = root / ".agent" / ".nav-config.json"
    if config_path.exists():
        try:
            with open(config_path) as f:
                return json.load(f)
        except Exception:
            pass
    return {}


def _resolve_plugin_dir() -> Path | None:
    """Env var first (tests inject fakes this way), then file-relative."""
    env = os.environ.get("CLAUDE_PLUGIN_ROOT") or os.environ.get("CLAUDE_PLUGIN_DIR")
    if env and Path(env).is_dir():
        return Path(env)
    here = Path(__file__).resolve().parent.parent
    if (here / "skills" / "nav-graph").is_dir():
        return here
    return None


def _extract_concepts(message: str) -> list:
    tokens = re.findall(r"[a-z][a-z\-_]{3,}", message.lower())
    concepts = []
    for token in tokens:
        if token in CONCEPT_STOPWORDS or token in concepts:
            continue
        concepts.append(token)
        if len(concepts) >= MAX_CONCEPTS:
            break
    return concepts


def _recall_memories(root: Path, message: str, budget: int) -> str:
    """Fetch relevant memories via memory_recall.py; empty string on any failure."""
    graph_path = root / ".agent" / "knowledge" / "graph.json"
    if not graph_path.is_file():
        return ""
    plugin_dir = _resolve_plugin_dir()
    if plugin_dir is None:
        return ""
    recall = plugin_dir / "skills" / "nav-graph" / "functions" / "memory_recall.py"
    if not recall.is_file():
        return ""
    concepts = _extract_concepts(message)
    if not concepts:
        return ""
    try:
        out = subprocess.run(
            [
                sys.executable, str(recall),
                "--concepts", ",".join(concepts),
                "--agent-dir", str(root / ".agent"),
                "--graph-path", str(graph_path),
                "--limit", str(RECALL_LIMIT),
                "--format", "compact",
            ],
            capture_output=True,
            text=True,
            timeout=RECALL_TIMEOUT,
        )
        if out.returncode != 0:
            return ""
        return (out.stdout or "").strip()[:budget]
    except Exception:
        return ""


def emit_brief_instruction(result: dict, threshold: float, memories: str):
    print(f"🧭 NAV-BRIEF: ambiguous task-shaped prompt "
          f"(score={result['score']}, threshold={threshold})")
    if result["undefined_dimensio

hooks/nav_post_compact.pyGitHub

Read the script

#!/usr/bin/env python3
"""
Navigator PostCompact hook — append Claude Code's compact summary to the
marker that PreCompact just wrote.

Fires after manual `/compact` or auto-compact finishes. Reads `.active`
to find the marker created by PreCompact, then appends a `## Compact
Summary (Claude Code)` section containing the official summary that
replaced the conversation.

Spec: https://docs.claude.com/en/docs/claude-code/hooks#postcompact
- stdin JSON includes: cwd, transcript_path, session_id, compact_summary
- Always exit 0 — never block.
- If `.active` doesn't exist (PreCompact didn't fire), do nothing.
"""
from __future__ import annotations

import json
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Any


def _safe_read(path: Path, max_bytes: int = 1000) -> str | None:
    try:
        if not path.is_file():
            return None
        return path.read_text(encoding="utf-8", errors="replace")[:max_bytes]
    except Exception as e:
        print(f"nav_post_compact: skip {path}: {e}", file=sys.stderr)
        return None


def _safe_json(path: Path) -> dict | None:
    raw = _safe_read(path, max_bytes=200_000)
    if not raw:
        return None
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        return None


def _project_root(stdin_data: dict) -> Path:
    cwd = stdin_data.get("cwd") or os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd()
    return Path(cwd)


def _hook_enabled(root: Path) -> bool:
    cfg = _safe_json(root / ".agent" / ".nav-config.json") or {}
    hook_cfg = cfg.get("compact_hook") or {}
    if not hook_cfg.get("enabled", True):
        return False
    return bool(hook_cfg.get("append_post_compact_summary", True))


def main() -> int:
    raw_stdin = sys.stdin.read() if not sys.stdin.isatty() else ""
    stdin_data: dict[str, Any] = {}
    if raw_stdin.strip():
        try:
            stdin_data = json.loads(raw_stdin)
        except json.JSONDecodeError:
            stdin_data = {}

    root = _project_root(stdin_data)
    if not (root / ".agent").is_dir():
        print(json.dumps({}))
        return 0

    if not _hook_enabled(root):
        print(json.dumps({}))
        return 0

    markers_dir = root / ".agent" / ".context-markers"
    active_path = markers_dir / ".active"
    active_name = _safe_read(active_path, max_bytes=200)
    if not active_name:
        # PreCompact didn't fire — nothing to append to.
        print(json.dumps({}))
        return 0

    active_name = active_name.strip()
    marker_path = markers_dir / active_name
    if not marker_path.is_file():
        print(f"nav_post_compact: marker {marker_path} missing", file=sys.stderr)
        print(json.dumps({}))
        return 0

    summary = stdin_data.get("compact_summary") or "_[no summary provided by Claude Code]_"
    if isinstance(summary, (dict, list)):
        summary = json.dumps(summary, indent=2, default=str)
    summary = str(summary).strip()

    try:
        with marker_path.open("a", encoding="utf-8") as fh:
            fh.write("\n\n---\n\n")
            fh.write("## Compact Summary (Claude Code)\n\n")
            fh.write(f"_Appended by PostCompact hook at {datetime.now().isoformat(timespec='seconds')}._\n\n")
            fh.write(summary + "\n")
        print(f"nav_post_compact: appended summary to {marker_path}", file=sys.stderr)
    except Exception as e:
        print(f"nav_post_compact: append failed: {e}", file=sys.stderr)

    print(json.dumps({}))
    return 0


if __name__ == "__main__":
    sys.exit(main())

hooks/nav_pre_compact.pyGitHub

Read the script

#!/usr/bin/env python3
"""
Navigator PreCompact hook — compact-resilient marker writer.

Fires before manual `/compact` or auto-compact. Reads the conversation
transcript, extracts a heuristic summary, captures git state, and writes
a marker to `.agent/.context-markers/` along with the `.active` pointer.

On the next session start, `hooks/nav_session_start.py` picks up `.active`
and surfaces the marker for restore — so context survives both manual and
auto-compacts.

Spec: https://docs.claude.com/en/docs/claude-code/hooks#precompact
- stdin JSON includes: cwd, transcript_path, session_id, trigger ("manual"|"auto")
- Always exit 0 — never block compact (exit 2 on auto-compact surfaces the
  underlying API error and breaks the session).
- stdout is empty JSON `{}` — we don't inject context, we write a file.
"""
from __future__ import annotations

import json
import os
import re
import subprocess
import sys
from datetime import datetime
from pathlib import Path
from typing import Any


# ──────────────────────────────────────────────────────────────────────────────
# Utilities (mirror nav_session_start.py for cross-hook consistency)
# ──────────────────────────────────────────────────────────────────────────────

def _safe_read(path: Path, max_bytes: int = 200_000) -> str | None:
    try:
        if not path.is_file():
            return None
        return path.read_text(encoding="utf-8", errors="replace")[:max_bytes]
    except Exception as e:
        print(f"nav_pre_compact: skip {path}: {e}", file=sys.stderr)
        return None


def _safe_json(path: Path) -> dict | None:
    raw = _safe_read(path)
    if not raw:
        return None
    try:
        return json.loads(raw)
    except json.JSONDecodeError as e:
        print(f"nav_pre_compact: invalid JSON in {path}: {e}", file=sys.stderr)
        return None


def _project_root(stdin_data: dict) -> Path:
    cwd = stdin_data.get("cwd") or os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd()
    return Path(cwd)


def _read_hook_config(root: Path) -> dict:
    cfg = _safe_json(root / ".agent" / ".nav-config.json") or {}
    hook_cfg = cfg.get("compact_hook") or {}
    return {
        "enabled": hook_cfg.get("enabled", True),
        "include_transcript_summary": hook_cfg.get("include_transcript_summary", True),
        "include_git_state": hook_cfg.get("include_git_state", True),
        "char_budget": int(hook_cfg.get("char_budget", 8000)),
    }


# ──────────────────────────────────────────────────────────────────────────────
# Heuristic transcript summarization (mirrors marker_compressor.py)
# ──────────────────────────────────────────────────────────────────────────────

def _flatten_transcript(transcript_path: Path) -> str:
    """Read the JSONL transcript and flatten to plain text for heuristic scan."""
    if not transcript_path.is_file():
        return ""
    lines_out: list[str] = []
    try:
        with transcript_path.open("r", encoding="utf-8", errors="replace") as fh:
            for raw in fh:
                raw = raw.strip()
                if not raw:
                    continue
                try:
                    obj = json.loads(raw)
                except json.JSONDecodeError:
                    lines_out.append(raw)
                    continue
                # Claude Code transcripts: nested message.content with various shapes
                msg = obj.get("message") or obj
                content = msg.get("content") if isinstance(msg, dict) else None
                if isinstance(content, str):
                    lines_out.append(content)
                elif isinstance(content, list):
                    for block in content:
                        if isinstance(block, dict):
                            text = block.get("text") or block.get("input") or block.get("output")
                            if isinstance(text, str):
                                lines_out.append(text)
                            elif text is not None:
                                lines_out.append(json.dumps(text, default=str)[:2000])
    except Exception as e:
        print(f"nav_pre_compact: transcript flatten failed: {e}", file=sys.stderr)
        return ""
    return "\n".join(lines_out)


# Path-shaped token: a slash/word run ending in a known source extension at a
# word boundary. Real path characters must precede the dot, so prose like
# "see the .py docs" is NOT captured while "hooks/token_monitor.py" is.
_PATH_RE = re.compile(r"[\w./-]+\.(?:tsx|json|md|ts|py|sh|js)\b")


def _compress_context(text: str, max_length: int = 5000) -> str:
    """Heuristic compressor — files/code/errors/recent context. Mirrors
    skills/nav-marker/functions/marker_compressor.py."""
    if not text:
        return "_[transcript unavailable]_"

    lines = text.split("\n")
    # Sample head + tail so paths/markers from both the conversation's start
    # (task setup) and end (recent work) survive; only the mid-section is
    # dropped when the transcript is long.
    scan_lines = lines[:100] + lines[-100:] if len(lines) > 200 else lines

    code_blocks: list[str] = []
    file_paths: list[str] = []
    errors: list[str] = []

    in_code_block = False
    code_buffer: list[str] = []

    for line in scan_lines:
        stripped = line.strip()
        if stripped.startswith("```"):
            if in_code_block:
                code_blocks.append("\n".join(code_buffer))
                code_buffer = []
            in_code_block = not in_code_block
        elif in_code_block:
            code_buffer.append(line)

        file_paths.extend(_PATH_RE.findall(line))

        low = line.lower()
        if "error" in low or "failed" in low or "traceback" in low:
            errors.append(stripped)

    # "Recent" = the true tail of the transcript, independent of sampling.
    recent_context = lines[-20:]

    parts: list[str] = []
    if file_paths:
        unique = list(dict.fromkeys(file_paths))[:10]
        parts.append("**Files/paths men

hooks/nav_profile_sync.pyGitHub

Read the script

#!/usr/bin/env python3
"""
Navigator user-profile → corrections-memory sync hook (Opp 3 / v6.11.0).

Fires on `PostToolUse` after `Write` or `Edit` calls. When the touched
file is `.agent/.user-profile.json`, this hook diffs the corrections
array against a tracked last_synced_count and runs
`correction_to_memory.py --action sync --last-synced N` only when the
array grew. Replaces the soft "monitor ALL conversations for corrections"
rule that depended on the model remembering to invoke nav-profile.

Idempotency is provided by `.agent/.nav-profile-sync-state.json`, which
records the corrections count after the last successful sync.

Spec: https://docs.claude.com/en/docs/claude-code/hooks#posttooluse
- stdin JSON: tool_name, tool_input (with file_path), cwd
- Exit 0 always — never block tool execution.
- stdout is empty JSON `{}` — pure side effect.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any


def _safe_read(path: Path, max_bytes: int = 500_000) -> str | None:
    try:
        if not path.is_file():
            return None
        return path.read_text(encoding="utf-8", errors="replace")[:max_bytes]
    except Exception as e:
        print(f"nav_profile_sync: skip {path}: {e}", file=sys.stderr)
        return None


def _safe_json(path: Path) -> dict | None:
    raw = _safe_read(path)
    if not raw:
        return None
    try:
        return json.loads(raw)
    except json.JSONDecodeError as e:
        print(f"nav_profile_sync: invalid JSON in {path}: {e}", file=sys.stderr)
        return None


def _project_root(stdin_data: dict) -> Path:
    cwd = stdin_data.get("cwd") or os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd()
    return Path(cwd)


def _hook_enabled(root: Path) -> bool:
    cfg = _safe_json(root / ".agent" / ".nav-config.json") or {}
    hook_cfg = cfg.get("profile_sync_hook") or {}
    return hook_cfg.get("enabled", True)


def _resolve_plugin_dir() -> Path | None:
    env = os.environ.get("CLAUDE_PLUGIN_ROOT") or os.environ.get("CLAUDE_PLUGIN_DIR")
    if env and Path(env).is_dir():
        return Path(env)
    candidates = [
        Path.home() / ".claude" / "plugins" / "cache" / "navigator-marketplace" / "navigator",
        Path.home() / ".claude" / "plugins" / "marketplaces" / "navigator-marketplace",
    ]
    for c in candidates:
        if (c / "skills" / "nav-graph").is_dir():
            return c
    here = Path(__file__).resolve().parent.parent
    if (here / "skills" / "nav-graph").is_dir():
        return here
    return None


def _is_profile_write(stdin_data: dict, root: Path) -> Path | None:
    """Return the profile path if this tool call touched it."""
    tool_input = stdin_data.get("tool_input") or {}
    candidate = tool_input.get("file_path")
    if not isinstance(candidate, str) or not candidate:
        return None
    p = Path(candidate)
    if not p.is_absolute():
        p = root / candidate
    # Match either the conventional .agent/.user-profile.json or any path
    # whose basename is .user-profile.json (defensive)
    if p.name == ".user-profile.json" and p.is_file():
        return p
    return None


def _load_state(root: Path) -> dict:
    state_path = root / ".agent" / ".nav-profile-sync-state.json"
    return _safe_json(state_path) or {}


def _save_state(root: Path, state: dict) -> None:
    state_path = root / ".agent" / ".nav-profile-sync-state.json"
    try:
        state_path.parent.mkdir(parents=True, exist_ok=True)
        state_path.write_text(
            json.dumps(state, indent=2) + "\n", encoding="utf-8"
        )
    except Exception as e:
        print(f"nav_profile_sync: state save failed: {e}", file=sys.stderr)


def main() -> int:
    raw_stdin = sys.stdin.read() if not sys.stdin.isatty() else ""
    stdin_data: dict[str, Any] = {}
    if raw_stdin.strip():
        try:
            stdin_data = json.loads(raw_stdin)
        except json.JSONDecodeError:
            stdin_data = {}

    root = _project_root(stdin_data)
    if not (root / ".agent").is_dir():
        print(json.dumps({}))
        return 0

    if not _hook_enabled(root):
        print(json.dumps({}))
        return 0

    profile_path = _is_profile_write(stdin_data, root)
    if profile_path is None:
        print(json.dumps({}))
        return 0

    profile = _safe_json(profile_path)
    if profile is None:
        print(json.dumps({}))
        return 0

    corrections = profile.get("corrections") or []
    current_count = len(corrections) if isinstance(corrections, list) else 0

    state = _load_state(root)
    last_synced = int(state.get("last_synced_count") or 0)

    if current_count <= last_synced:
        # No new corrections — nothing to sync. Pure no-op for non-correction
        # profile edits (preferences, goals, etc.).
        print(json.dumps({}))
        return 0

    graph_path = root / ".agent" / "knowledge" / "graph.json"
    if not graph_path.is_file():
        # No graph initialized — skip silently
        print(json.dumps({}))
        return 0

    plugin_dir = _resolve_plugin_dir()
    if plugin_dir is None:
        print("nav_profile_sync: plugin dir not found", file=sys.stderr)
        print(json.dumps({}))
        return 0

    syncer = plugin_dir / "skills" / "nav-graph" / "functions" / "correction_to_memory.py"
    if not syncer.is_file():
        print(f"nav_profile_sync: {syncer} missing", file=sys.stderr)
        print(json.dumps({}))
        return 0

    try:
        result = subprocess.run(
            [
                sys.executable,
                str(syncer),
                "--action",
                "sync",
                "--profile-path",
                str(profile_path),
                "--graph-path",
                str(graph_path),
                "--last-synced",
                str(last_synced),
            ],
            capture_output=True,
            text=True,
            timeout=8,
        )
        if re

hooks/nav_read_guard.pyGitHub

Read the script

#!/usr/bin/env python3
"""
Navigator `.agent/` bulk-read guard (Opp 6 / v6.12.0).

Fires on `PreToolUse` Read. Counts non-allowlisted reads of `.agent/` files
per turn and emits an escalating warning when the count crosses thresholds.

The tail-risk this prevents: sequential bulk-loads of `.agent/` documentation
(tasks, SOPs, system docs) that can consume 50k+ tokens in a single turn and
crash mid-session — the exact anti-pattern Navigator's lazy-loading strategy
exists to avoid.

Behavior (v6.12.1+):
  - Counter file: `.agent/.nav-read-counter.json` (per-session, reset on Stop)
  - Allowlist (do not count): DEVELOPMENT-README.md, .nav-config.json,
    .user-profile.json, knowledge/graph.json
  - Warn at count >= warn_threshold (default 3): stderr advisory, exit 0
  - Escalate at count >= escalate_threshold (default 5):
      * strict_block=true (default) → exit 2, sentinel-wrapped stderr block
      * strict_block=false → stderr advisory only, exit 0
  - Counter satisfies mem-027's three-condition gate: this hook writes the
    state file on prior invocations; reading count >= threshold IS state-file
    confirmation that violations already occurred this turn.

Spec: https://docs.claude.com/en/docs/claude-code/hooks#pretooluse
  - stdin JSON: tool_name, tool_input.file_path, cwd, session_id
  - file_path may be absolute OR relative (handled per mem-027 discipline).
  - Output channel: stderr only. mem-035 confirmed PreToolUse stdout AND
    hookSpecificOutput.additionalContext are silent to the model. Block
    (exit 2) is the only behavior-affecting channel; warn (exit 0) text on
    stderr surfaces in the CC UI but does not influence model behavior.
"""
from __future__ import annotations

import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


COUNTER_FILE = ".agent/.nav-read-counter.json"

# Files in `.agent/` that Navigator itself reads during legitimate session
# start or on-demand patterns. These are exempt from the counter.
DEFAULT_ALLOWLIST = frozenset({
    "DEVELOPMENT-README.md",
    ".nav-config.json",
    ".user-profile.json",
    "knowledge/graph.json",
})

DEFAULT_WARN_THRESHOLD = 3
DEFAULT_ESCALATE_THRESHOLD = 5
# Primary counter reset is the Stop hook; if Stop did not fire (mem-036), a
# stale counter from a prior turn must not block this turn's legitimate reads.
# Treat the counter as fresh-from-zero once its last update predates this window.
DEFAULT_STALE_AFTER_SECONDS = 300


def _safe_read(path: Path, max_bytes: int = 200_000) -> str | None:
    try:
        if not path.is_file():
            return None
        return path.read_text(encoding="utf-8", errors="replace")[:max_bytes]
    except Exception:
        return None


def _safe_json(path: Path) -> dict | None:
    raw = _safe_read(path)
    if not raw:
        return None
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        return None


def _project_root(stdin_data: dict) -> Path:
    cwd = stdin_data.get("cwd") or os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd()
    return Path(cwd)


def _hook_cfg(root: Path) -> dict:
    cfg = _safe_json(root / ".agent" / ".nav-config.json") or {}
    return cfg.get("read_guard_hook") or {}


def _resolve_agent_relative(file_path: str, root: Path) -> str | None:
    """Return the path relative to .agent/ if the file is under it, else None.

    Handles both absolute and relative file_path inputs (OQ-1 defensive).
    """
    if not file_path:
        return None
    p = Path(file_path)
    if not p.is_absolute():
        p = root / file_path
    try:
        resolved = p.resolve()
        rel = resolved.relative_to((root / ".agent").resolve())
    except (ValueError, OSError):
        return None
    return rel.as_posix()


def _is_counted(agent_rel: str, allowlist: frozenset[str]) -> bool:
    """Return True if this `.agent/`-relative path should increment the counter.

    Allowlist matches exact path-from-.agent (e.g., "DEVELOPMENT-README.md" or
    "knowledge/graph.json") — not basename. Knowledge graph subtree files
    other than graph.json (e.g., memory `.md` files) ARE counted.
    """
    return agent_rel not in allowlist


def _load_counter(root: Path) -> dict:
    data = _safe_json(root / COUNTER_FILE) or {}
    if not isinstance(data, dict):
        data = {}
    return data


def _save_counter(root: Path, data: dict) -> None:
    target = root / COUNTER_FILE
    try:
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
    except Exception as e:
        print(f"nav_read_guard: counter write failed: {e}", file=sys.stderr)


def _is_stale(updated_at: str | None, stale_after_s: int) -> bool:
    """True when the counter's last update predates the staleness window.

    Guards against a missed Stop reset (mem-036): a stale counter would
    otherwise persist into the next turn and falsely block. Missing or
    unparseable timestamps are treated as NOT stale (preserve prior behavior).
    """
    if not updated_at or stale_after_s <= 0:
        return False
    try:
        ts = datetime.fromisoformat(updated_at)
    except (ValueError, TypeError):
        return False
    if ts.tzinfo is None:
        ts = ts.replace(tzinfo=timezone.utc)
    age = (datetime.now(timezone.utc) - ts).total_seconds()
    return age > stale_after_s


def _increment_counter(
    root: Path, session_id: str | None, stale_after_s: int = DEFAULT_STALE_AFTER_SECONDS
) -> int:
    state = _load_counter(root)
    prior_session = state.get("session_id")
    # Reset when the session changed (secondary path; primary is the Stop hook)
    # or when the prior count is stale (Stop may not have fired — mem-036).
    if session_id and prior_session and prior_session != session_id:
        state = {}
    elif _is_stale(state.get("updated_at"), stale_after_s):
        state = {}
    new_count = int(state.get("turn_count", 0)

hooks/nav_session_start.pyGitHub

Read the script

#!/usr/bin/env python3
"""
Navigator SessionStart hook — zero-Read context injection.

Invoked by Claude Code on session start (and on --resume). Emits a JSON
payload whose `additionalContext` is injected as a system reminder into
the model's context window, eliminating the ~6 Read tool calls that
nav-start would otherwise make.

Parity goal: produce the SAME content nav-start currently renders
(navigator + active marker + config + graph stats + profile + auto-update).
Only the *delivery mechanism* changes.

Spec: https://docs.claude.com/en/docs/claude-code/hooks#sessionstart
- stdout MUST be valid JSON
- additionalContext capped at 10_000 chars by Claude Code
- stderr is logged but does not block; non-zero exit is tolerated

Sentinel: every successful injection emits `<!-- nav-session-start-injected:v1 -->`
so nav-start can detect prior injection and skip its own reads.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any

SENTINEL = "<!-- nav-session-start-injected:v1 -->"
CHAR_BUDGET = 9500  # leave headroom under Claude Code's 10k limit
TRUNCATION_FOOTER = (
    "\n\n[truncated: ask nav-start for full detail]"
)


def _safe_read(path: Path, max_bytes: int = 50_000) -> str | None:
    try:
        if not path.is_file():
            return None
        return path.read_text(encoding="utf-8", errors="replace")[:max_bytes]
    except Exception as e:
        print(f"nav_session_start: skip {path}: {e}", file=sys.stderr)
        return None


def _safe_json(path: Path) -> dict | None:
    raw = _safe_read(path)
    if not raw:
        return None
    try:
        return json.loads(raw)
    except json.JSONDecodeError as e:
        print(f"nav_session_start: invalid JSON in {path}: {e}", file=sys.stderr)
        return None


def _project_root(stdin_data: dict) -> Path:
    cwd = stdin_data.get("cwd") or os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd()
    return Path(cwd)


def _section_navigator(root: Path) -> str | None:
    nav = _safe_read(root / ".agent" / "DEVELOPMENT-README.md", max_bytes=8_000)
    if not nav:
        return None
    return f"## Navigator Index (.agent/DEVELOPMENT-README.md)\n\n{nav}"


def _section_active_marker(root: Path) -> str | None:
    active = root / ".agent" / ".context-markers" / ".active"
    name = _safe_read(active, max_bytes=200)
    if not name:
        return None
    name = name.strip()
    if not name:
        return None
    marker_file = root / ".agent" / ".context-markers" / name
    body = _safe_read(marker_file, max_bytes=6_000)
    if not body:
        return f"## Active Marker\n\n`{name}` referenced but file missing."
    return (
        f"## Active Marker: `{name}`\n\n"
        f"User was working on this before compact. Offer to resume.\n\n"
        f"{body}"
    )


def _section_config(root: Path) -> str | None:
    cfg = _safe_json(root / ".agent" / ".nav-config.json")
    if not cfg:
        return None
    summary = {
        "version": cfg.get("version"),
        "project_management": cfg.get("project_management"),
        "task_prefix": cfg.get("task_prefix"),
        "team_chat": cfg.get("team_chat"),
        "loop_mode": (cfg.get("loop_mode") or {}).get("enabled"),
        "task_mode": (cfg.get("task_mode") or {}).get("enabled"),
        "knowledge_graph": (cfg.get("knowledge_graph") or {}).get("enabled"),
        "tom_features": cfg.get("tom_features"),
        "auto_update": (cfg.get("auto_update") or {}).get("enabled"),
    }
    return (
        "## Navigator Config (.agent/.nav-config.json)\n\n"
        f"```json\n{json.dumps(summary, indent=2)}\n```"
    )


def _section_user_profile(root: Path) -> str | None:
    profile = _safe_json(root / ".agent" / ".user-profile.json")
    if not profile:
        return None
    prefs = profile.get("preferences", {})
    corrections = profile.get("corrections", [])
    goals = profile.get("goals", [])
    body = {
        "preferences": prefs,
        "recent_corrections": corrections[-5:] if corrections else [],
        "goals": goals[-5:] if goals else [],
    }
    return (
        "## User Profile (Theory of Mind)\n\n"
        "Apply these preferences for this session.\n\n"
        f"```json\n{json.dumps(body, indent=2, default=str)}\n```"
    )


def _section_graph_stats(root: Path, plugin_dir: Path | None) -> str | None:
    graph_path = root / ".agent" / "knowledge" / "graph.json"
    if not graph_path.is_file():
        return None
    if plugin_dir is None:
        return f"## Knowledge Graph\n\nGraph present at {graph_path} (stats unavailable: plugin dir unknown)."
    manager = plugin_dir / "skills" / "nav-graph" / "functions" / "graph_manager.py"
    if not manager.is_file():
        return f"## Knowledge Graph\n\nGraph present (stats helper not found)."
    try:
        out = subprocess.run(
            [
                sys.executable,
                str(manager),
                "--action",
                "stats",
                "--graph-path",
                str(graph_path),
            ],
            capture_output=True,
            text=True,
            timeout=4,
        )
        text = (out.stdout or "").strip()
        if not text:
            return None
        return f"## Knowledge Graph Stats\n\n```\n{text[:1500]}\n```"
    except Exception as e:
        print(f"nav_session_start: graph stats failed: {e}", file=sys.stderr)
        return None


def _section_auto_update(root: Path, plugin_dir: Path | None) -> str | None:
    """Read-only version-drift notice for session start.

    Historically this invoked auto_updater with no args, which triggered a
    *mutating* `claude plugin update` (a 30s marketplace refresh + a 60s
    update) from inside the SessionStart hook's 10s budget — the update could
    never finish and the blocking hook risked timing out. We now run the
    updater in read-only `--check-drift` mode: it only compares the installed
    plugin version ag