Hooks

Where it lives

• hooks/detect_session_changes.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Detect file changes made by Claude during a session.
  
  Reads a JSON object from stdin with {"transcript_path": "..."}.
  Streams the JSONL transcript line-by-line, extracts file paths from
  Write/Edit/NotebookEdit tool_use blocks, and reports which files exist.
  
  Prints a single JSON line to stdout.
  """
  
  import json
  import os
  import sys
  
  TRACKED_TOOLS = {"Write", "Edit", "NotebookEdit"}
  
  
  def scan_transcript(path: str) -> dict:
      files_seen: dict[str, set[str]] = {}  # file_path -> set of tool names
      tools_used: dict[str, int] = {}
  
      try:
          with open(path, "r", encoding="utf-8", errors="replace") as f:
              for line in f:
                  line = line.strip()
                  if not line:
                      continue
                  try:
                      obj = json.loads(line)
                  except (json.JSONDecodeError, ValueError):
                      continue
  
                  message = obj.get("message") if isinstance(obj, dict) else None
                  if not isinstance(message, dict):
                      continue
  
                  content = message.get("content")
                  if not isinstance(content, list):
                      continue
  
                  for block in content:
                      if not isinstance(block, dict):
                          continue
                      if block.get("type") != "tool_use":
                          continue
  
                      name = block.get("name", "")
                      if name not in TRACKED_TOOLS:
                          continue
  
                      inp = block.get("input")
                      if not isinstance(inp, dict):
                          continue
  
                      file_path = inp.get("file_path") or inp.get("notebook_path")
                      if not file_path or not isinstance(file_path, str):
                          continue
  
                      tools_used[name] = tools_used.get(name, 0) + 1
  
                      if file_path not in files_seen:
                          files_seen[file_path] = set()
                      files_seen[file_path].add(name)
  
      except OSError as e:
          return {"files_changed": -1, "files": [], "tools_used": {}, "error": str(e)}
  
      # Only report files that still exist on disk
      existing = [p for p in files_seen if os.path.exists(p)]
      existing.sort()
  
      return {
          "files_changed": len(existing),
          "files": existing,
          "tools_used": tools_used,
      }
  
  
  def main() -> None:
      try:
          raw = sys.stdin.read()
          request = json.loads(raw)
      except (json.JSONDecodeError, ValueError) as e:
          json.dump(
              {"files_changed": -1, "files": [], "tools_used": {}, "error": f"invalid input JSON: {e}"},
              sys.stdout,
          )
          print()
          return
  
      if not isinstance(request, dict):
          json.dump(
              {"files_changed": -1, "files": [], "tools_used": {}, "error": "input must be a JSON object"},
              sys.stdout,
          )
          print()
          return
  
      transcript_path = request.get("transcript_path")
      if not transcript_path or not isinstance(transcript_path, str):
          json.dump(
              {"files_changed": -1, "files": [], "tools_used": {}, "error": "missing transcript_path"},
              sys.stdout,
          )
          print()
          return
  
      if not os.path.isfile(transcript_path):
          json.dump(
              {"files_changed": -1, "files": [], "tools_used": {}, "error": f"file not found: {transcript_path}"},
              sys.stdout,
          )
          print()
          return
  
      result = scan_transcript(transcript_path)
      json.dump(result, sys.stdout)
      print()
  
  
  if __name__ == "__main__":
      main()
  

• hooks/eval_checkpoint.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """PostToolUse(Write|Edit) checkpoint — continuous QA during work.
  
  Runs the project's test command at configurable intervals during a session.
  Injects feedback via systemMessage when tests fail, without blocking.
  
  Hook protocol (PostToolUse hook, matcher: Write|Edit):
    - exit 0 with no stdout → no-op (silent pass or not yet at interval)
    - exit 0 with {"systemMessage": "..."} → inject test failure feedback
  """
  
  from __future__ import annotations
  
  import json
  import os
  import re
  import shlex
  import subprocess
  import sys
  import time
  from pathlib import Path
  from typing import Optional
  
  
  # -- Configuration --
  
  CHECKPOINT_ENABLED = os.getenv("LACP_EVAL_CHECKPOINT_ENABLED", "0") == "1"
  CHECKPOINT_INTERVAL = int(os.getenv("LACP_EVAL_CHECKPOINT_INTERVAL", "10"))
  TEST_TIMEOUT = int(os.getenv("LACP_EVAL_CHECKPOINT_TIMEOUT", "30"))
  
  _LACP_STATE_DIR = Path.home() / ".lacp" / "hooks" / "state"
  
  # Allowed test runner prefixes (same as stop_quality_gate.py)
  _ALLOWED_TEST_RUNNERS = (
      "bun test", "pnpm test", "yarn test", "npm test",
      "make test", "cargo test", "python3 -m pytest",
      "bin/lacp-test",
  )
  
  HOOKS_DIR = Path(__file__).parent
  sys.path.insert(0, str(HOOKS_DIR))
  
  
  def _safe_session_id(session_id: str) -> str:
      return re.sub(r"[^a-zA-Z0-9_-]", "_", session_id) if session_id else "default"
  
  
  def _session_state_dir(session_id: str) -> Path:
      safe_id = _safe_session_id(session_id)
      d = _LACP_STATE_DIR / safe_id
      d.mkdir(parents=True, exist_ok=True, mode=0o700)
      return d
  
  
  def _get_write_count(session_id: str) -> int:
      counter_file = _session_state_dir(session_id) / "write-count"
      try:
          return int(counter_file.read_text().strip()) if counter_file.exists() else 0
      except (ValueError, OSError):
          return 0
  
  
  def _set_write_count(session_id: str, count: int) -> None:
      counter_file = _session_state_dir(session_id) / "write-count"
      try:
          counter_file.write_text(str(count))
      except OSError:
          pass
  
  
  def _get_test_command(session_id: str) -> Optional[str]:
      """Read cached test command from session start contract."""
      try:
          from hook_contracts import read_contract
          contract = read_contract("session_start", session_id)
          if contract and contract.get("test_cmd"):
              cmd = contract["test_cmd"]
              if any(cmd.startswith(prefix) for prefix in _ALLOWED_TEST_RUNNERS):
                  return cmd
      except Exception:
          pass
      return None
  
  
  def _update_checkpoint_contract(session_id: str, result: str, fail_count: int) -> None:
      """Update the eval checkpoint contract for telemetry/stop hook consumption."""
      try:
          from hook_contracts import EvalCheckpoint, write_contract
          checkpoint = EvalCheckpoint(
              write_count=_get_write_count(session_id),
              last_check_at=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
              last_result=result,
              fail_count=fail_count,
          )
          write_contract("eval_checkpoint", checkpoint, session_id)
      except Exception:
          pass
  
  
  def _get_fail_count(session_id: str) -> int:
      """Read current fail count from checkpoint contract."""
      try:
          from hook_contracts import read_contract
          contract = read_contract("eval_checkpoint", session_id)
          if contract:
              return int(contract.get("fail_count", 0))
      except Exception:
          pass
      return 0
  
  
  def main() -> None:
      if not CHECKPOINT_ENABLED:
          return
  
      raw = sys.stdin.read()
      try:
          hook_input = json.loads(raw) if raw.strip() else {}
      except json.JSONDecodeError:
          return
  
      session_id = hook_input.get("session_id") or ""
      cwd = hook_input.get("cwd") or ""
  
      # Increment write counter
      count = _get_write_count(session_id) + 1
      _set_write_count(session_id, count)
  
      # Check if we're at an interval
      if count % CHECKPOINT_INTERVAL != 0:
          return
  
      # Find test command
      test_cmd = _get_test_command(session_id)
      if not test_cmd:
          return
  
      # Run tests
      try:
          result = subprocess.run(
              shlex.split(test_cmd),
              shell=False,
              capture_output=True,
              text=True,
              timeout=TEST_TIMEOUT,
              cwd=cwd or None,
          )
      except (subprocess.TimeoutExpired, OSError):
          return  # Fail-open
  
      fail_count = _get_fail_count(session_id)
  
      if result.returncode == 0:
          _update_checkpoint_contract(session_id, "pass", fail_count)
          return  # Silent success
  
      # Tests failed — inject feedback (don't block)
      fail_count += 1
      _update_checkpoint_contract(session_id, "fail", fail_count)
  
      output = (result.stdout or "") + (result.stderr or "")
      last_lines = "\n".join(output.strip().splitlines()[-8:])
  
      # Think-mode enhancement: inject thinking prompt alongside failure report
      think_prompt = ""
      if os.getenv("LACP_CONTEXT_MODE", "") in ("think", "tdd", "debugging"):
          think_prompt = (
              "\n\nBefore making changes, think about: "
              "What changed since tests last passed? "
              "Which specific file most likely caused this failure? "
              "What is the minimal fix (not a rewrite)?"
          )
  
      msg = (
          f"Checkpoint ({count} writes): tests failing (exit {result.returncode}). "
          f"Fix before continuing:\n{last_lines}{think_prompt}"
      )
      print(json.dumps({"systemMessage": msg}))
  
  
  if __name__ == "__main__":
      main()
  

• hooks/extract_memories.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Per-turn memory extraction — captures durable signals from conversation.
  
  Lightweight Stop hook that scans the last assistant message for memory-worthy
  signals and writes them to a staging file. Brain-expand later promotes from
  staging into the knowledge graph.
  
  No LLM calls — heuristic pattern matching only. Keeps hook latency <100ms.
  
  Hook protocol (Stop hook):
    - Reads JSON from stdin: {transcript_path, session_id, cwd, ...}
    - Writes signals to staging file: ~/.lacp/memory-staging/pending.jsonl
    - Exits 0 (never blocks stop)
  """
  from __future__ import annotations
  
  import json
  import os
  import re
  import sys
  from datetime import UTC, datetime
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent.parent / "automation" / "scripts"))
  
  STAGING_DIR = Path.home() / ".lacp" / "memory-staging"
  STAGING_FILE = STAGING_DIR / "pending.jsonl"
  MAX_STAGING_LINES = 500  # rotate after this many
  
  # Patterns that indicate durable, memory-worthy content
  MEMORY_INDICATORS = [
      # Decisions and preferences
      re.compile(r"(?:we decided|the decision is|going with|chose to|prefer)\s+(.{20,})", re.I),
      # Lessons learned
      re.compile(r"(?:lesson learned|takeaway|key insight|important note|remember that)\s*:?\s*(.{20,})", re.I),
      # Architecture/design choices
      re.compile(r"(?:architecture|design choice|trade-?off|we.ll use|stack is)\s*:?\s*(.{20,})", re.I),
      # Process/workflow
      re.compile(r"(?:workflow|process|convention|rule|policy)\s*:?\s*(.{20,})", re.I),
      # User corrections (feedback type)
      re.compile(r"(?:don.t|never|always|stop|instead of)\s+(.{15,})", re.I),
  ]
  
  # Anti-patterns — skip these even if they match indicators
  SKIP_PATTERNS = [
      re.compile(r"```[\s\S]{100,}```"),  # large code blocks
      re.compile(r"^\s*[-*]\s+`[^`]+`\s*$", re.M),  # bullet list of code refs
      re.compile(r"(?:error|traceback|stack trace)", re.I),  # error output
  ]
  
  
  def extract_signals(text: str, session_id: str, cwd: str) -> list[dict]:
      """Extract memory-worthy signals from assistant message text."""
      if not text or len(text) < 50:
          return []
  
      # Skip if text is mostly code
      code_ratio = text.count("```") / max(1, len(text) / 100)
      if code_ratio > 0.3:
          return []
  
      # Check anti-patterns
      for pattern in SKIP_PATTERNS:
          if pattern.search(text):
              return []
  
      signals = []
      for pattern in MEMORY_INDICATORS:
          for match in pattern.finditer(text):
              signal_text = match.group(1).strip() if match.lastindex else match.group(0).strip()
              # Clean up
              signal_text = re.sub(r"\s+", " ", signal_text)
              if len(signal_text) < 20 or len(signal_text) > 300:
                  continue
              signals.append({
                  "text": signal_text[:280],
                  "source": "per-turn-extraction",
                  "session_id": session_id,
                  "cwd": cwd,
                  "timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
                  "day": datetime.now(UTC).strftime("%Y-%m-%d"),
              })
  
      # Deduplicate within this extraction
      seen = set()
      unique = []
      for s in signals:
          key = s["text"][:50].lower()
          if key not in seen:
              seen.add(key)
              unique.append(s)
  
      return unique[:5]  # cap at 5 per turn
  
  
  def append_to_staging(signals: list[dict]) -> int:
      """Append signals to staging JSONL file. Returns count written."""
      if not signals:
          return 0
  
      STAGING_DIR.mkdir(parents=True, exist_ok=True)
  
      # Rotate if too large
      if STAGING_FILE.exists():
          line_count = sum(1 for _ in STAGING_FILE.open())
          if line_count > MAX_STAGING_LINES:
              archive = STAGING_DIR / f"staging-{datetime.now(UTC).strftime('%Y%m%d')}.jsonl"
              STAGING_FILE.rename(archive)
  
      with STAGING_FILE.open("a", encoding="utf-8") as f:
          for signal in signals:
              f.write(json.dumps(signal) + "\n")
  
      return len(signals)
  
  
  def _extract_last_assistant(transcript_path: str) -> str:
      """Extract last assistant message from JSONL transcript."""
      if not transcript_path or not os.path.isfile(transcript_path):
          return ""
      last = ""
      try:
          with open(transcript_path, encoding="utf-8", errors="ignore") as f:
              for line in f:
                  line = line.strip()
                  if not line:
                      continue
                  try:
                      msg = json.loads(line)
                      if msg.get("type") == "assistant":
                          content = msg.get("message", {}).get("content", "")
                          if isinstance(content, list):
                              # Extract text parts
                              text_parts = [p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text"]
                              content = "\n".join(text_parts)
                          if isinstance(content, str) and len(content) > 50:
                              last = content
                  except json.JSONDecodeError:
                      continue
      except OSError:
          pass
      return last
  
  
  def main() -> int:
      try:
          hook_input = json.loads(sys.stdin.read())
      except (json.JSONDecodeError, EOFError):
          return 0  # silent exit on bad input
  
      session_id = hook_input.get("session_id", "")
      cwd = hook_input.get("cwd", "")
      transcript_path = hook_input.get("transcript_path", "")
  
      # Get the last assistant message
      last_message = hook_input.get("last_assistant_message", "")
      if not last_message:
          last_message = _extract_last_assistant(transcript_path)
  
      if not last_message:
          return 0
  
      signals = extract_signals(last_message, session_id, cwd)
      count = append_to_staging(signals)
  
      if count > 0 and os.environ.get("LACP_EXTRACT_DEBUG") == "1":
          print(json.dumps({"extracted": count, "signals": signals}), file=sys.stderr)
  
      retur

• hooks/hook_contracts.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Typed hook state contracts — shared state between hooks within a session.
  
  Provides a simple contract system where hooks can write structured data
  that other hooks can read with schema validation. Contracts are stored
  per-session and auto-cleaned.
  
  Inspired by herm's tool interface (Definition → ToolDefinition, Execute → (string, error))
  and langdag's typed ContentBlock/Node structures.
  """
  
  from __future__ import annotations
  
  import json
  import os
  import re
  from dataclasses import asdict, dataclass
  from pathlib import Path
  from typing import Optional
  
  
  def _safe_session_id(session_id: str) -> str:
      """Sanitize session_id for use in file paths (L1: CWE-22)."""
      return re.sub(r"[^a-zA-Z0-9_-]", "_", session_id) if session_id else "default"
  
  
  # Contract storage location
  def _contracts_dir(session_id: str) -> Path:
      return Path.home() / ".lacp" / "hooks" / "contracts" / _safe_session_id(session_id)
  
  
  @dataclass
  class SessionStartOutput:
      test_cmd: Optional[str] = None
      git_branch: Optional[str] = None
      context_mode: Optional[str] = None
      started_at: Optional[str] = None
      context_budget_hint: Optional[int] = None
  
  
  @dataclass
  class StopGateInput:
      test_cmd: Optional[str] = None
      session_changes: Optional[list[str]] = None
      transcript_path: Optional[str] = None
      session_started_at: Optional[str] = None
      context_budget_hint: Optional[int] = None
      tool_use_count: Optional[int] = None
  
  
  def write_contract(name: str, data: object, session_id: str | None = None) -> Path | None:
      """Write a contract to disk. Returns the path written, or None on failure."""
      sid = session_id or os.getenv("CLAUDE_SESSION_ID", "default")
      contracts = _contracts_dir(sid)
      try:
          contracts.mkdir(parents=True, exist_ok=True)
          path = contracts / f"{name}.json"
          payload = asdict(data) if hasattr(data, "__dataclass_fields__") else data
          path.write_text(json.dumps(payload, default=str))
          return path
      except (OSError, TypeError):
          return None
  
  
  def read_contract(name: str, session_id: str | None = None) -> dict | None:
      """Read a contract from disk. Returns dict or None if missing/invalid."""
      sid = session_id or os.getenv("CLAUDE_SESSION_ID", "default")
      path = _contracts_dir(sid) / f"{name}.json"
      if not path.is_file():
          return None
      try:
          return json.loads(path.read_text())
      except (json.JSONDecodeError, OSError):
          return None
  
  
  @dataclass
  class SprintContract:
      acceptance_criteria: list[str]
      expected_files: list[str]
      expected_tests: list[str]
      agreed_at: str = ""
  
  
  @dataclass
  class EvalCheckpoint:
      write_count: int = 0
      last_check_at: str = ""
      last_result: str = ""  # "pass", "fail", or ""
      fail_count: int = 0
  
  
  @dataclass
  class HandoffArtifact:
      task_summary: str = ""
      files_modified: list[str] = None
      open_issues: list[str] = None
      next_steps: list[str] = None
      test_status: str = "unknown"  # "pass", "fail", "unknown"
      git_branch: str = ""
      git_diff_summary: str = ""
      created_at: str = ""
  
      def __post_init__(self):
          if self.files_modified is None:
              self.files_modified = []
          if self.open_issues is None:
              self.open_issues = []
          if self.next_steps is None:
              self.next_steps = []
  
  
  @dataclass
  class TaskPlan:
      subtasks: list[dict] = None  # [{name, files, depends_on, status, scope}]
      created_at: str = ""
  
      def __post_init__(self):
          if self.subtasks is None:
              self.subtasks = []
  
  
  def cleanup_contracts(session_id: str | None = None) -> int:
      """Remove all contracts for a session. Returns count of files removed."""
      sid = session_id or os.getenv("CLAUDE_SESSION_ID", "default")
      contracts = _contracts_dir(sid)
      if not contracts.is_dir():
          return 0
      count = 0
      try:
          for f in contracts.iterdir():
              f.unlink()
              count += 1
          contracts.rmdir()
      except OSError:
          pass
      return count
  
  
  def cleanup_stale_contracts(max_age_hours: int = 48) -> int:
      """Remove contract directories older than max_age_hours. Returns count removed."""
      import time as _time
      contracts_root = Path.home() / ".lacp" / "hooks" / "contracts"
      if not contracts_root.is_dir():
          return 0
      cutoff = _time.time() - (max_age_hours * 3600)
      removed = 0
      try:
          for d in contracts_root.iterdir():
              if not d.is_dir():
                  continue
              try:
                  mtime = max(f.stat().st_mtime for f in d.iterdir()) if any(d.iterdir()) else d.stat().st_mtime
                  if mtime < cutoff:
                      for f in d.iterdir():
                          f.unlink()
                      d.rmdir()
                      removed += 1
              except OSError:
                  continue
      except OSError:
          pass
      return removed
  
  
  def cleanup_stale_state(max_age_hours: int = 48) -> int:
      """Remove session state directories older than max_age_hours. Returns count removed."""
      import time as _time
      state_root = Path.home() / ".lacp" / "hooks" / "state"
      if not state_root.is_dir():
          return 0
      cutoff = _time.time() - (max_age_hours * 3600)
      removed = 0
      try:
          for d in state_root.iterdir():
              if not d.is_dir():
                  continue
              if d.name == "quality-gate.log":
                  continue  # skip the debug log file
              try:
                  mtime = max(f.stat().st_mtime for f in d.iterdir()) if any(d.iterdir()) else d.stat().st_mtime
                  if mtime < cutoff:
                      for f in d.iterdir():
                          f.unlink()
                      d.rmdir()
                      removed += 1
              except OSError:
                  continue
      except OSError:
          pass
      return removed
  

• hooks/hook_telemetry.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Lightweight telemetry logger for Claude Code hooks.
  
  Append-only JSONL log with auto-rotation. Usable as CLI or importable module.
  Python 3.11 stdlib only.
  """
  
  import argparse
  import fcntl
  import json
  import os
  import sys
  from datetime import datetime, timezone
  from pathlib import Path
  
  TELEMETRY_DIR = Path.home() / ".local" / "share" / "claude-hooks"
  TELEMETRY_FILE = TELEMETRY_DIR / "telemetry.jsonl"
  MAX_SIZE = 10 * 1024 * 1024  # 10 MB
  MAX_ROTATIONS = 3
  
  
  def _rotate_if_needed() -> None:
      """Rotate telemetry file if it exceeds MAX_SIZE."""
      try:
          if not TELEMETRY_FILE.exists() or TELEMETRY_FILE.stat().st_size < MAX_SIZE:
              return
      except OSError:
          return
  
      # Shift .3 -> delete, .2 -> .3, .1 -> .2, current -> .1
      for i in range(MAX_ROTATIONS, 0, -1):
          src = TELEMETRY_FILE.with_suffix(f".{i}")
          if i == MAX_ROTATIONS:
              src.unlink(missing_ok=True)
          else:
              dst = TELEMETRY_FILE.with_suffix(f".{i + 1}")
              if src.exists():
                  src.rename(dst)
  
      TELEMETRY_FILE.rename(TELEMETRY_FILE.with_suffix(".1"))
  
  
  def log_decision(
      hook: str,
      decision: str,
      reason: str,
      session_id: str = "unknown",
      elapsed_ms: int = 0,
      details: dict | None = None,
  ) -> None:
      """Append a single telemetry entry (thread-safe)."""
      TELEMETRY_DIR.mkdir(parents=True, exist_ok=True)
      _rotate_if_needed()
  
      entry = {
          "ts": datetime.now(timezone.utc).astimezone().isoformat(),
          "hook": hook,
          "session_id": session_id,
          "decision": decision,
          "reason": reason,
          "elapsed_ms": elapsed_ms,
          "details": details or {},
      }
  
      line = json.dumps(entry, separators=(",", ":")) + "\n"
  
      fd = os.open(str(TELEMETRY_FILE), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644)
      try:
          fcntl.flock(fd, fcntl.LOCK_EX)
          os.write(fd, line.encode())
      finally:
          fcntl.flock(fd, fcntl.LOCK_UN)
          os.close(fd)
  
  
  def main() -> None:
      parser = argparse.ArgumentParser(description="Log a hook telemetry decision")
      parser.add_argument("--hook", required=True, help="Hook name")
      parser.add_argument("--session-id", default="unknown", help="Session ID")
      parser.add_argument(
          "--decision",
          required=True,
          choices=["allow", "block", "skip"],
          help="Decision outcome",
      )
      parser.add_argument("--reason", required=True, help="Human-readable reason")
      parser.add_argument("--elapsed", type=int, default=0, help="Elapsed time in ms")
      parser.add_argument(
          "--detail",
          action="append",
          default=[],
          metavar="KEY=VALUE",
          help="Extra key=value detail (repeatable)",
      )
  
      args = parser.parse_args()
  
      details = {}
      for kv in args.detail:
          if "=" in kv:
              k, v = kv.split("=", 1)
              details[k] = v
          else:
              details[kv] = True
  
      log_decision(
          hook=args.hook,
          session_id=args.session_id,
          decision=args.decision,
          reason=args.reason,
          elapsed_ms=args.elapsed,
          details=details,
      )
  
  
  if __name__ == "__main__":
      main()
  

• hooks/isolation_guard.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """Log Claude Code subagent lifecycle events. Does not replace WorktreeCreate."""
  from __future__ import annotations
  
  import json
  import os
  import sys
  from datetime import datetime, timezone
  from pathlib import Path
  
  
  def knowledge_root() -> Path:
      raw = os.environ.get("LACP_KNOWLEDGE_ROOT", "").strip()
      if raw:
          return Path(raw).expanduser()
      return Path.home() / ".lacp" / "knowledge"
  
  
  def main() -> int:
      raw = sys.stdin.read()
      if not raw.strip():
          return 0
      try:
          payload = json.loads(raw)
      except json.JSONDecodeError:
          return 0
      event = str(payload.get("hook_event_name") or payload.get("event") or "unknown")
      record = {
          "captured_at_utc": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
          "event": event,
          "agent_id": payload.get("agent_id"),
          "agent_type": payload.get("agent_type"),
          "session_id": payload.get("session_id"),
          "cwd": payload.get("cwd"),
      }
      out_dir = knowledge_root() / "data"
      out_dir.mkdir(parents=True, exist_ok=True)
      with (out_dir / "isolation-events.jsonl").open("a", encoding="utf-8") as handle:
          handle.write(json.dumps(record, separators=(",", ":")) + "\n")
      return 0
  
  
  if __name__ == "__main__":
      raise SystemExit(main())
  

• hooks/pretool_guard.pyGitHub ↗
• hooks/self_memory_system.pyGitHub ↗
• hooks/session_orient.shGitHub ↗
• hooks/session_start.pyGitHub ↗
• hooks/stop_quality_gate.pyGitHub ↗
• hooks/stop_quality_gate.shGitHub ↗
• hooks/test_stop_hook.pyGitHub ↗
• hooks/thinking_nudge.pyGitHub ↗
• hooks/write_validate.pyGitHub ↗

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