Hooks

Where it lives

• hooks/compound_nudge.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env -S uv run --script
  # /// script
  # requires-python = ">=3.12"
  # dependencies = ["pydantic>=2"]
  # ///
  """Stop hook: one-shot nudge to run the ERPAVal Compound phase.
  
  Stop hooks cannot inject additionalContext per the authoritative schema
  (code.claude.com/docs/en/hooks.md) — the only Claude-facing channel is
  `decision: "block"` + `reason`. This hook uses that channel exactly like
  codeprobe's review-before-commit pattern: block the first turn-end that
  matches the gates, feed Claude the instruction, then step out of the way.
  
  Firing gates (ALL must hold):
  
    1. This Claude session actually wrote to .erpaval/ — marker file
       dropped by validate_packet.py. Prevents nudges in sessions that
       never touched ERPAVal.
    2. A .erpaval/sessions/session-<hex>/ dir in cwd has validation.yaml
       but no lessons.yaml (the original pending-Compound signal).
    3. validation.yaml mtime < 2h ago. Abandoned sessions stop nagging.
    4. Session dir name matches `session-<hex>` — filters hand-named dirs.
    5. Not nudged this Claude session (HookState).
    6. Session ID not in .erpaval/sessions/.nudged ledger — dismiss-once
       across Claude sessions.
  
  On fire: returns `decision: "block"` with instructions to run Compound.
  Claude either runs it (lessons.yaml lands, gate 2 fails next turn) or
  skips it (HookState + ledger prevent re-fire). Either way: bounded.
  
  Harness caveat: stop_hook_active caps consecutive Stop blocks at 2, so
  even a buggy nudge cannot trap a session.
  """
  
  import os
  import re
  import sys
  import time
  from pathlib import Path
  
  sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
  
  from framework import (
      HookInput,
      HookState,
      StopInput,
      SyncHookOutput,
      block,
      run_hook,
  )
  
  
  SESSION_RE = re.compile(r"^session-[a-f0-9]{6,}$")
  RECENCY_WINDOW_SECONDS = 2 * 60 * 60  # 2 hours
  
  ERPAVAL_DOT_DIR = ".erpaval"
  ERPAVAL_MARKER_PREFIX = "claude-erpaval-active-"
  
  
  def _erpaval_active(session_id: str) -> bool:
      """Did this Claude session write to .erpaval/? Set by validate_packet.py."""
      return (Path("/tmp") / f"{ERPAVAL_MARKER_PREFIX}{session_id}").exists()
  
  
  def _find_pending_compound(cwd: Path) -> Path | None:
      """Return the first session dir with pending Compound, or None.
  
      Applies: name-shape filter, recency gate, per-project dismiss ledger.
      """
      sessions = cwd / ERPAVAL_DOT_DIR / "sessions"
      if not sessions.is_dir():
          return None
  
      dismissed = _load_ledger(sessions)
      now = time.time()
  
      for session in sorted(sessions.iterdir()):
          if not session.is_dir():
              continue
          if not SESSION_RE.match(session.name):
              continue
          if session.name in dismissed:
              continue
          validation = session / "validation.yaml"
          lessons = session / "lessons.yaml"
          if not validation.exists() or lessons.exists():
              continue
          try:
              age = now - validation.stat().st_mtime
          except OSError:
              continue
          if age > RECENCY_WINDOW_SECONDS:
              continue
          return session
  
      return None
  
  
  def _load_ledger(sessions_dir: Path) -> set[str]:
      ledger = sessions_dir / ".nudged"
      if not ledger.exists():
          return set()
      try:
          return {line.strip() for line in ledger.read_text().splitlines() if line.strip()}
      except OSError:
          return set()
  
  
  def _append_ledger(sessions_dir: Path, session_name: str) -> None:
      ledger = sessions_dir / ".nudged"
      try:
          with ledger.open("a") as fh:
              fh.write(session_name + "\n")
      except OSError:
          pass
  
  
  def handle(input: HookInput, state: HookState) -> SyncHookOutput | None:
      if not isinstance(input, StopInput):
          return None
      if state.has("compound_nudged"):
          return None
      if not _erpaval_active(input.session_id):
          return None
  
      session = _find_pending_compound(Path(input.cwd))
      if session is None:
          return None
  
      state.set("compound_nudged", True)
      _append_ledger(session.parent, session.name)
  
      reason = (
          f"ERPAVal Compound phase is pending for {session}. "
          "validation.yaml exists but lessons.yaml does not — the session "
          "has not written its dual-track lessons yet.\n\n"
          "Before ending: run CL-LESSONS against the session trace and, for "
          "each novel + reusable candidate, write .erpaval/solutions/<category>/"
          "<slug>.md, then update .erpaval/INDEX.md and write "
          f"{session}/lessons.yaml. See "
          "${CLAUDE_PLUGIN_ROOT}/skills/erpaval/references/compound.md "
          "for the full operating sequence.\n\n"
          "If the session was abandoned and no lessons should be persisted, "
          "write an empty {lessons_written: []} lessons.yaml so the gate "
          "clears. This nudge fires at most once per Claude session and is "
          "suppressed permanently for this session-id via "
          ".erpaval/sessions/.nudged."
      )
      return block(reason)
  
  
  if __name__ == "__main__":
      run_hook(handle, name="erpaval_compound_nudge")
  

• hooks/framework.pyGitHub ↗
  Read the script

  """Strongly-typed, stateful hook framework for Claude Code.
  
  Provides Pydantic models for all 26 hook input types and their output types,
  a session-scoped HookState class, and a run_hook() entry point that handles
  stdin/stdout plumbing.
  
  Usage from a UV shebang hook script:
  
      #!/usr/bin/env -S uv run --script
      # /// script
      # requires-python = ">=3.12"
      # dependencies = ["pydantic>=2"]
      # ///
      import os, sys
      sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
      from framework import run_hook, deny, PreToolUseInput, HookInput, HookState, SyncHookOutput
  
      def handle(input: HookInput, state: HookState) -> SyncHookOutput | None:
          ...
  
      if __name__ == "__main__":
          run_hook(handle, name="my_hook")
  """
  
  from __future__ import annotations
  
  import json
  import sys
  from datetime import datetime, timezone
  from pathlib import Path
  from typing import Annotated, Any, Callable, Literal, Union
  
  from pydantic import BaseModel, ConfigDict, Field, TypeAdapter
  
  
  # ---------------------------------------------------------------------------
  # Input Models -- all 26 hook events per code.claude.com/docs/en/hooks.md
  # ---------------------------------------------------------------------------
  
  class BaseHookInput(BaseModel):
      """Fields common to all hook events."""
  
      model_config = ConfigDict(extra="allow")
  
      session_id: str
      transcript_path: str
      cwd: str
      hook_event_name: str
      permission_mode: str | None = None
  
  
  # -- Session lifecycle events -----------------------------------------------
  
  class SessionStartInput(BaseHookInput):
      hook_event_name: Literal["SessionStart"]
      source: Literal["startup", "resume", "clear", "compact"] | None = None
      model: str | None = None
  
  
  class InstructionsLoadedInput(BaseHookInput):
      hook_event_name: Literal["InstructionsLoaded"]
      file_path: str
      memory_type: str  # User | Project | Local | Managed
      load_reason: str  # session_start | nested_traversal | path_glob_match | include | compact
      globs: list[str] | None = None
      trigger_file_path: str | None = None
      parent_file_path: str | None = None
  
  
  class SessionEndInput(BaseHookInput):
      hook_event_name: Literal["SessionEnd"]
  
  
  # -- User input events ------------------------------------------------------
  
  class UserPromptSubmitInput(BaseHookInput):
      hook_event_name: Literal["UserPromptSubmit"]
      prompt: str
  
  
  # -- Tool lifecycle events ---------------------------------------------------
  
  class PreToolUseInput(BaseHookInput):
      hook_event_name: Literal["PreToolUse"]
      tool_name: str
      tool_input: dict[str, Any]
      tool_use_id: str
      agent_id: str | None = None
      agent_type: str | None = None
  
  
  class PostToolUseInput(BaseHookInput):
      hook_event_name: Literal["PostToolUse"]
      tool_name: str
      tool_input: dict[str, Any]
      tool_response: Any = None
      tool_use_id: str
      agent_id: str | None = None
      agent_type: str | None = None
  
  
  class PostToolUseFailureInput(BaseHookInput):
      hook_event_name: Literal["PostToolUseFailure"]
      tool_name: str
      tool_input: dict[str, Any]
      tool_use_id: str
      error: str
      is_interrupt: bool | None = None
      agent_id: str | None = None
      agent_type: str | None = None
  
  
  class PermissionRequestInput(BaseHookInput):
      hook_event_name: Literal["PermissionRequest"]
      tool_name: str
      tool_input: dict[str, Any]
      permission_suggestions: list[Any] | None = None
      agent_id: str | None = None
      agent_type: str | None = None
  
  
  class PermissionDeniedInput(BaseHookInput):
      hook_event_name: Literal["PermissionDenied"]
      tool_name: str
      tool_input: dict[str, Any]
      tool_use_id: str
      reason: str
      agent_id: str | None = None
      agent_type: str | None = None
  
  
  # -- Agent/subagent events ---------------------------------------------------
  
  class SubagentStartInput(BaseHookInput):
      hook_event_name: Literal["SubagentStart"]
      agent_id: str
      agent_type: str
  
  
  class SubagentStopInput(BaseHookInput):
      hook_event_name: Literal["SubagentStop"]
      stop_hook_active: bool
      agent_id: str
      agent_transcript_path: str
      agent_type: str
      last_assistant_message: str | None = None
  
  
  # -- Task/team events --------------------------------------------------------
  
  class TaskCreatedInput(BaseHookInput):
      hook_event_name: Literal["TaskCreated"]
      task_id: str
      task_subject: str
      task_description: str | None = None
      teammate_name: str | None = None
      team_name: str | None = None
  
  
  class TaskCompletedInput(BaseHookInput):
      hook_event_name: Literal["TaskCompleted"]
      task_id: str
      task_subject: str
      task_description: str | None = None
      teammate_name: str | None = None
      team_name: str | None = None
  
  
  class TeammateIdleInput(BaseHookInput):
      hook_event_name: Literal["TeammateIdle"]
  
  
  # -- Stop events --------------------------------------------------------------
  
  class StopInput(BaseHookInput):
      hook_event_name: Literal["Stop"]
      assistant_response: str | None = None
  
  
  class StopFailureInput(BaseHookInput):
      hook_event_name: Literal["StopFailure"]
      error_type: str  # rate_limit | authentication_failed | billing_error | etc.
      error_message: str | None = None
  
  
  # -- Compaction events --------------------------------------------------------
  
  class PreCompactInput(BaseHookInput):
      hook_event_name: Literal["PreCompact"]
      trigger: Literal["manual", "auto"]
      custom_instructions: str | None = None
  
  
  class PostCompactInput(BaseHookInput):
      hook_event_name: Literal["PostCompact"]
  
  
  # -- Notification events ------------------------------------------------------
  
  class NotificationInput(BaseHookInput):
      hook_event_name: Literal["Notification"]
      message: str
      title: str | None = None
      notification_type: str
  
  
  # -- Config/filesystem events -------------------------------------------------
  
  class ConfigChangeInput(BaseHookInput):
      hook_event_name: Literal["ConfigChange"]
      source: str  # 

• hooks/session_start_bootstrap.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env -S uv run --script
  # /// script
  # requires-python = ">=3.12"
  # dependencies = ["pydantic>=2"]
  # ///
  """SessionStart hook: emit category counts + INDEX pointer when this repo
  has recent ERPAVal activity.
  
  SessionStart genuinely supports additionalContext per the authoritative
  schema (code.claude.com/docs/en/hooks.md), so injection is legal here.
  
  Firing gates (ALL must hold):
  
    1. .erpaval/solutions/ exists with at least one lesson.
    2. A session-<hex>/ dir under .erpaval/sessions/ was modified in the
       last 24h — proxy for active-build rhythm or recent resume.
       Without this, cold repos with old lessons emit on every unrelated
       session just because the directory exists on disk.
    3. Fires at most once per Claude session (HookState `bootstrapped`).
  
  Skip conditions:
    - No .erpaval/ at all → return None (usual no-op).
    - Lessons exist but no recent session activity → silent. User hasn't
      opted into an ERPAVal run this week.
  """
  
  import os
  import re
  import sys
  import time
  from pathlib import Path
  
  sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
  
  from framework import (
      HookInput,
      HookState,
      SessionStartInput,
      SyncHookOutput,
      add_context,
      run_hook,
  )
  
  
  SESSION_RE = re.compile(r"^session-[a-f0-9]{6,}$")
  ACTIVITY_WINDOW_SECONDS = 24 * 60 * 60  # 24 hours
  
  ERPAVAL_DOT_DIR = ".erpaval"
  
  
  def _recent_session_activity(erpaval: Path) -> bool:
      """True iff any session-<hex>/ dir has been touched in the last 24h."""
      sessions = erpaval / "sessions"
      if not sessions.is_dir():
          return False
      cutoff = time.time() - ACTIVITY_WINDOW_SECONDS
      for session in sessions.iterdir():
          if not session.is_dir() or not SESSION_RE.match(session.name):
              continue
          try:
              if session.stat().st_mtime >= cutoff:
                  return True
          except OSError:
              continue
      return False
  
  
  def _bootstrap_summary(solutions: Path) -> str | None:
      counts: dict[str, int] = {}
      for md in solutions.rglob("*.md"):
          counts[md.parent.name] = counts.get(md.parent.name, 0) + 1
      total = sum(counts.values())
      if total == 0:
          return None
      lines = [f"prior ERPAVal lessons: {total} across {len(counts)} categories"]
      for cat in sorted(counts):
          lines.append(f"  {cat}: {counts[cat]}")
      index = solutions.parent / "INDEX.md"
      if index.exists():
          lines.append(f"index: {index}")
      return "\n".join(lines)
  
  
  def handle(input: HookInput, state: HookState) -> SyncHookOutput | None:
      if not isinstance(input, SessionStartInput):
          return None
      if state.has("bootstrapped"):
          return None
      erpaval = Path(input.cwd) / ERPAVAL_DOT_DIR
      solutions = erpaval / "solutions"
      if not solutions.is_dir():
          return None
      if not _recent_session_activity(erpaval):
          return None
      summary = _bootstrap_summary(solutions)
      if summary is None:
          return None
      state.set("bootstrapped", True)
      return add_context(summary, event_name="SessionStart")
  
  
  if __name__ == "__main__":
      run_hook(handle, name="erpaval_session_start_bootstrap")
  

• hooks/validate_packet.pyRunsGitHub ↗
  Read the script

  #!/usr/bin/env -S uv run --script
  # /// script
  # requires-python = ">=3.12"
  # dependencies = ["pydantic>=2", "pyyaml"]
  # ///
  """PostToolUse hook: validate `.erpaval/sessions/<id>/*.yaml` writes.
  
  Early-exits for non-`.erpaval` paths and non-YAML files. On schema violation,
  injects a systemMessage via additionalContext so Claude sees the error but is
  not blocked (fail-open: bad validation must never wedge a session).
  
  Per-task packets (`tasks/T-*.md`) are validated by parsing YAML frontmatter
  (first `---` block). Body is intentionally unchecked.
  """
  
  import os
  import sys
  from pathlib import Path
  from typing import Literal
  
  import yaml
  from pydantic import BaseModel, Field, ValidationError
  
  sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
  
  from framework import (
      HookInput,
      HookState,
      PostToolUseInput,
      SyncHookOutput,
      add_context,
      run_hook,
  )
  
  
  # ---------- schemas mirroring tools/erpaval-validate.py ----------
  
  
  class IntakeInferred(BaseModel):
      scope: Literal["coding", "non-coding"] | None = None
      complexity: Literal["1-file-fix", "multi-module", "rebuild"] | None = None
      dir_state: Literal["empty", "existing", "rebuild-in-place"] | None = None
      variant: Literal["greenfield", "brownfield", "rip-and-replace"] | None = None
      rigor_needed: list[Literal["hmw", "ears"]] | None = None
  
  
  class IntakeGit(BaseModel):
      is_repo: bool
      branch: str | None = None
      dirty: bool | None = None
  
  
  class IntakeSchema(BaseModel):
      session_id: str = Field(pattern=r"^session-[a-f0-9]{6,}$")
      working_dir: str
      raw_request: str
      inferred: IntakeInferred
      git: IntakeGit | None = None
      upstream_artifacts: dict | None = None
      env_snapshot: dict | None = None
  
  
  class RecallSchema(BaseModel):
      applicable_lessons: list[dict] = Field(default_factory=list)
      injection_strategy: str | None = None
  
  
  class ExploreSchema(BaseModel):
      agent_id: str
      scope: str
      findings: dict
  
  
  class ResearchSchema(BaseModel):
      agent_id: str
      domain: str
      libraries: list[dict] = Field(default_factory=list)
  
  
  class TaskFrontmatter(BaseModel):
      task_id: str
      ac_source: str | None = None
      agent_name: str | None = None
      model: Literal["haiku", "sonnet", "opus"] | None = None
      isolation: Literal["worktree"] | None = None
      status: Literal["IN_PROGRESS", "COMPLETE", "BLOCKED"] | None = None
  
  
  class ValidationSchema(BaseModel):
      validation_id: str
      layers: dict
      auto_merge_eligible: bool | None = None
      disposition: str | None = None
  
  
  class LessonsSchema(BaseModel):
      lessons_written: list[dict] = Field(default_factory=list)
      claude_md_updated: bool | None = None
      index_md_updated: bool | None = None
  
  
  class SessionSchema(BaseModel):
      session_id: str = Field(pattern=r"^session-[a-f0-9]{6,}$")
      status: Literal["active", "completed", "abandoned"]
      variant: str | None = None
      classifier_trace: list = Field(default_factory=list)
      cycles_executed: dict = Field(default_factory=dict)
      packets: dict = Field(default_factory=dict)
      merge: dict | None = None
  
  
  ERPAVAL_DOT_DIR = ".erpaval"
  ERPAVAL_MARKER_PREFIX = "claude-erpaval-active-"
  
  
  SCHEMAS: dict[str, type[BaseModel]] = {
      "intake": IntakeSchema,
      "recall": RecallSchema,
      "explore": ExploreSchema,
      "validation": ValidationSchema,
      "lessons": LessonsSchema,
      "session": SessionSchema,
  }
  
  
  def _schema_for_yaml(path: Path) -> type[BaseModel] | None:
      name = path.stem
      if name in SCHEMAS:
          return SCHEMAS[name]
      if name.startswith("research-"):
          return ResearchSchema
      return None
  
  
  def _parse_frontmatter(text: str) -> dict | None:
      if not text.startswith("---\n"):
          return None
      try:
          end = text.index("\n---", 4)
      except ValueError:
          return None
      return yaml.safe_load(text[4:end]) or {}
  
  
  def _validate(path: Path) -> list[str]:
      if not path.exists():
          return []
  
      if path.suffix in (".yaml", ".yml"):
          schema = _schema_for_yaml(path)
          if schema is None:
              return []
          try:
              data = yaml.safe_load(path.read_text())
          except yaml.YAMLError as e:
              return [f"{path}: YAML parse error: {e}"]
          if data is None:
              return [f"{path}: file is empty"]
      elif path.suffix == ".md" and path.parent.name == "tasks":
          schema = TaskFrontmatter
          data = _parse_frontmatter(path.read_text())
          if data is None:
              return []  # no frontmatter yet — skeleton being written
      else:
          return []
  
      try:
          schema.model_validate(data)
      except ValidationError as e:
          return [
              f"{path}: {'.'.join(str(x) for x in err['loc'])}: {err['msg']}"
              for err in e.errors()
          ]
      return []
  
  
  def _mark_erpaval_active(session_id: str) -> None:
      """Drop a cross-hook marker so compound_nudge knows this Claude session
      actually wrote to .erpaval/. Prevents the nudge from firing when the user
      is working in some other repo that happens to have a stale .erpaval/.
      """
      marker = Path("/tmp") / f"{ERPAVAL_MARKER_PREFIX}{session_id}"
      try:
          marker.touch(exist_ok=True)
      except OSError:
          pass
  
  
  def handle(input: HookInput, state: HookState) -> SyncHookOutput | None:
      if not isinstance(input, PostToolUseInput):
          return None
      file_path = input.tool_input.get("file_path", "")
      if f"/{ERPAVAL_DOT_DIR}/" not in file_path:
          return None
      path = Path(file_path)
      if path.suffix not in (".yaml", ".yml", ".md"):
          return None
  
      _mark_erpaval_active(input.session_id)
  
      errors = _validate(path)
      if not errors:
          return None
  
      message = "ERPAVal packet validation failed:\n" + "\n".join(errors)
      return add_context(message, event_name="PostToolUse")
  
  
  if __name__ == "__main__":
      run_hook(handle, name="erpaval_validate_packet")
  

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.