Hooks

Where it lives

• hooks/a2a_bridge_daemon.pyGitHub ↗
  Read the script

  """
  ArmaraOS daemon discovery for the A2A subsystem.
  
  Replaces the old "launch a Python bridge" approach.
  ArmaraOS is the A2A bridge — we just discover it via ~/.armaraos/daemon.json.
  """
  
  import json
  import os
  import socket
  import time
  from pathlib import Path
  from typing import Dict, Any
  
  from shared.armaraos_daemon import (
      DAEMON_NOT_FOUND_REASON,
      LEGACY_DAEMON_URL_CACHE_NAME,
      daemon_url_cache_path,
      scan_daemon_listen_port,
  )
  
  
  def _pid_alive(pid: int) -> bool:
      try:
          os.kill(pid, 0)
          return True
      except (ProcessLookupError, PermissionError):
          return False
  
  
  def _port_open(host: str, port: int, timeout: float = 1.0) -> bool:
      try:
          s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
          s.settimeout(timeout)
          result = s.connect_ex((host, port))
          s.close()
          return result == 0
      except Exception:
          return False
  
  
  def _write_url_cache(plugin_root: Path, base_url: str, pid, version: str) -> None:
      """Write discovered daemon URL to plugin-local cache for fast reuse."""
      cache_file = daemon_url_cache_path(plugin_root)
      cache_file.parent.mkdir(parents=True, exist_ok=True)
      tmp = cache_file.with_suffix(".tmp")
      tmp.write_text(json.dumps({
          "base_url": base_url,
          "pid": pid,
          "version": version,
          "discovered_at": int(time.time()),
      }), encoding="utf-8")
      os.replace(tmp, cache_file)
  
  
  def ensure_bridge_running(plugin_root: Path, config: dict) -> Dict[str, Any]:
      """
      Discover the ArmaraOS daemon, cache its URL, and return its status.
  
      Discovery order:
        1. daemon.json — check if the recorded port is still alive
        2. lsof scan   — daemon may use dynamic ports on restart
      Result is written to a2a/armaraos_daemon_url.json so tools skip re-scanning.
      """
      a2a_cfg = config.get("a2a", {})
      if not a2a_cfg.get("enabled", True):
          return {"running": False, "reason": "disabled"}
  
      daemon_json_path = Path(
          a2a_cfg.get("daemon_json", "~/.armaraos/daemon.json")
      ).expanduser()
  
      pid = None
      version = "unknown"
  
      # ── Step 1: try daemon.json ───────────────────────────────────────────────
      if daemon_json_path.exists():
          try:
              daemon = json.loads(daemon_json_path.read_text(encoding="utf-8"))
              pid = daemon.get("pid")
              version = daemon.get("version", "unknown")
              listen_addr = daemon.get("listen_addr", "")
              if listen_addr:
                  host, _, port_str = listen_addr.rpartition(":")
                  port = int(port_str)
                  if _pid_alive(pid) and _port_open(host, port):
                      base_url = f"http://{listen_addr}"
                      _write_url_cache(plugin_root, base_url, pid, version)
                      return {"running": True, "pid": pid, "port": port, "host": host,
                              "base_url": base_url, "version": version, "source": "daemon.json"}
          except Exception:
              pass
  
      # ── Step 2: lsof scan for dynamic port ───────────────────────────────────
      host, port = scan_daemon_listen_port()
      if host and port:
          base_url = f"http://{host}:{port}"
          # Confirm it's actually the ArmaraOS API
          try:
              import urllib.request
              resp = urllib.request.urlopen(f"{base_url}/api/health", timeout=2)
              data = json.loads(resp.read())
              version = data.get("version", version)
              # Try to get PID from /api/health or keep what we have from daemon.json
          except Exception:
              pass
          _write_url_cache(plugin_root, base_url, pid, version)
          return {"running": True, "pid": pid, "port": port, "host": host,
                  "base_url": base_url, "version": version, "source": "lsof"}
  
      # ── Not found ─────────────────────────────────────────────────────────────
      # Clear stale cache so tools don't use a dead URL
      cache_file = daemon_url_cache_path(plugin_root)
      if cache_file.exists():
          cache_file.unlink(missing_ok=True)
      legacy_cache = plugin_root / "a2a" / LEGACY_DAEMON_URL_CACHE_NAME
      if legacy_cache.exists():
          legacy_cache.unlink(missing_ok=True)
  
      return {"running": False, "reason": DAEMON_NOT_FOUND_REASON}
  

• hooks/a2a_inbox_writer.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  A2A inbox writer — delegate script for HERMES_AINL_BRIDGE_CMD.
  
  The A2A bridge calls this script with the inbound message text on stdin.
  We write a structured JSON file to the plugin inbox and print "ok" to
  stdout so the bridge marks the task completed.
  
  Zero plugin imports — stdlib only so any Python 3.8+ installation works.
  """
  
  import json
  import os
  import sys
  import time
  import uuid
  from pathlib import Path
  
  
  PLUGIN_ROOT = Path(os.environ.get("AINL_PLUGIN_ROOT", Path(__file__).resolve().parent.parent))
  INBOX_DIR = PLUGIN_ROOT / "a2a" / "inbox"
  
  
  def extract_header(text: str, header: str, default: str = "") -> str:
      """Extract X-Header: value from message text (convention for structured sends)."""
      prefix = f"{header}: "
      for line in text.splitlines():
          if line.startswith(prefix):
              return line[len(prefix):].strip()
      return default
  
  
  def strip_headers(text: str) -> str:
      """Remove X-* header lines from message body."""
      lines = []
      for line in text.splitlines():
          if not (line.startswith("X-") and ": " in line):
              lines.append(line)
      return "\n".join(lines).strip()
  
  
  def main():
      raw = sys.stdin.read().strip()
      if not raw:
          print("ok")
          return
  
      msg_id = str(uuid.uuid4())
      from_agent = extract_header(raw, "X-From-Agent", "unknown")
      urgency = extract_header(raw, "X-Urgency", "normal")
      thread_id = extract_header(raw, "X-Thread-Id") or None
      task_id = extract_header(raw, "X-Task-Id") or None
      msg_type = "task_result" if task_id else "message"
      body = strip_headers(raw)
  
      msg = {
          "id": msg_id,
          "type": msg_type,
          "from_agent": from_agent,
          "to_agent": "claude-code",
          "thread_id": thread_id,
          "task_id": task_id,
          "message": body,
          "urgency": urgency if urgency in ("critical", "normal", "low") else "normal",
          "received_at": int(time.time()),
      }
  
      INBOX_DIR.mkdir(parents=True, exist_ok=True)
      tmp = INBOX_DIR / f"{msg_id}.tmp"
      dest = INBOX_DIR / f"{msg_id}.json"
      tmp.write_text(json.dumps(msg, indent=2), encoding="utf-8")
      os.replace(tmp, dest)
  
      print("ok")
  
  
  if __name__ == "__main__":
      main()
  

• hooks/ainl_detection.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """AINL opportunity detection hook for Claude Code.
  
  Detects when to suggest using .ainl files based on user prompts and context.
  Includes persona evolution tracking.
  """
  import json
  import re
  import sys
  from pathlib import Path
  from typing import Dict, List, Optional, Any
  
  # Add parent directory to path for imports
  sys.path.insert(0, str(Path(__file__).parent.parent))
  
  try:
      from mcp_server.persona_evolution import (
          PersonaEvolutionEngine,
          detect_action_from_context
      )
      PERSONA_AVAILABLE = True
  except ImportError:
      PERSONA_AVAILABLE = False
  
  
  # Trigger keywords that suggest AINL usage
  RECURRING_KEYWORDS = [
      "every", "hourly", "daily", "weekly", "monthly",
      "monitor", "check", "recurring", "scheduled", "schedule", "cron",
      "repeatedly", "periodic", "regular", "automation", "automat",
  ]
  
  WORKFLOW_KEYWORDS = [
      "workflow", "automation", "pipeline", "process",
      "multi-step", "sequence", "orchestrate", "coordinate"
  ]
  
  API_KEYWORDS = [
      "api", "endpoint", "fetch", "http", "webhook",
      "rest", "call", "request", "integration"
  ]
  
  BLOCKCHAIN_KEYWORDS = [
      "solana", "blockchain", "wallet", "crypto", "token",
      "nft", "defi", "web3", "balance", "transfer"
  ]
  
  COST_KEYWORDS = [
      "cost", "expensive", "budget", "save", "cheap",
      "token", "efficient", "optimize"
  ]
  
  CONDITIONAL_KEYWORDS = [
      "if", "when", "then", "else", "condition",
      "check if", "depending on", "based on"
  ]
  
  
  class AINLDetector:
      """Detects opportunities to suggest AINL."""
  
      def __init__(self, project_id: Optional[str] = None):
          self.confidence_threshold = 0.35
          self.project_id = project_id
  
          # Initialize persona engine if available
          self.persona_engine = None
          if PERSONA_AVAILABLE and project_id:
              try:
                  persona_db = Path.home() / ".claude" / "projects" / project_id / "persona.db"
                  persona_db.parent.mkdir(parents=True, exist_ok=True)
                  self.persona_engine = PersonaEvolutionEngine(persona_db)
              except Exception as e:
                  sys.stderr.write(f"Failed to initialize persona engine: {e}\n")
                  self.persona_engine = None
  
      def analyze_prompt(self, prompt: str, context: Dict[str, Any]) -> Dict[str, Any]:
          """
          Analyze user prompt to detect AINL opportunities.
  
          Args:
              prompt: User's message
              context: Additional context (working dir, files, etc.)
  
          Returns:
              {
                  "suggest_ainl": bool,
                  "confidence": float (0-1),
                  "reasons": List[str],
                  "use_case": str,
                  "suggestion_text": str (markdown)
              }
          """
          prompt_lower = prompt.lower()
          reasons = []
          confidence_score = 0.0
  
          # Check for .ainl files in workspace
          has_ainl_files = self._check_ainl_files(context)
          if has_ainl_files:
              confidence_score += 0.2
              reasons.append("Existing .ainl files in workspace")
  
          # Check for recurring/scheduled patterns
          recurring_matches = sum(1 for kw in RECURRING_KEYWORDS if kw in prompt_lower)
          if recurring_matches > 0:
              confidence_score += min(0.70, 0.35 + (recurring_matches - 1) * 0.15)
              reasons.append(f"Recurring pattern detected ({recurring_matches} keywords)")
  
          # Check for workflow patterns
          workflow_matches = sum(1 for kw in WORKFLOW_KEYWORDS if kw in prompt_lower)
          if workflow_matches > 0:
              confidence_score += min(0.3, workflow_matches * 0.15)
              reasons.append(f"Workflow pattern detected ({workflow_matches} keywords)")
  
          # Check for API integration
          api_matches = sum(1 for kw in API_KEYWORDS if kw in prompt_lower)
          if api_matches >= 1:
              confidence_score += min(0.25, api_matches * 0.1)
              reasons.append("API integration detected")
  
          # Check for blockchain
          blockchain_matches = sum(1 for kw in BLOCKCHAIN_KEYWORDS if kw in prompt_lower)
          if blockchain_matches > 0:
              confidence_score += 0.5  # Strong signal
              reasons.append("Blockchain interaction detected (AINL has Solana adapter)")
  
          # Check for cost concerns
          cost_matches = sum(1 for kw in COST_KEYWORDS if kw in prompt_lower)
          if cost_matches > 0:
              confidence_score += 0.3
              reasons.append("Cost/efficiency concern detected")
  
          # Check for conditional logic
          conditional_matches = sum(1 for kw in CONDITIONAL_KEYWORDS if kw in prompt_lower)
          if conditional_matches > 1:
              confidence_score += 0.2
              reasons.append("Conditional logic detected")
  
          # Determine use case
          use_case = self._determine_use_case(
              prompt_lower,
              recurring_matches,
              workflow_matches,
              blockchain_matches,
              api_matches
          )
  
          # Cap confidence at 1.0
          confidence_score = min(1.0, confidence_score)
  
          # Extract persona signals from prompt
          if self.persona_engine and PERSONA_AVAILABLE:
              try:
                  action = detect_action_from_context(prompt)
                  if action:
                      signals = self.persona_engine.extract_signals(action, context)
                      if signals:
                          self.persona_engine.ingest_signals(signals)
              except Exception as e:
                  sys.stderr.write(f"Persona signal extraction failed: {e}\n")
  
          # Generate suggestion text
          suggestion_text = ""
          persona_traits = ""
  
          if confidence_score >= self.confidence_threshold:
              suggestion_text = self._generate_suggestion(
                  use_case, confidence_score, reasons
              )
  
              # Add persona traits if available
              if self.persona_engine:
                  try:
                      persona_trait

• hooks/ainl_validator.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """AINL auto-validation hook for Claude Code.
  
  Automatically validates .ainl files after tool use (Read/Edit/Write).
  """
  import json
  import sys
  from pathlib import Path
  from typing import Dict, Any, Optional
  
  sys.path.insert(0, str(Path(__file__).parent))
  from shared.project_id import get_project_id
  
  # Try to import AINL tools
  try:
      sys.path.insert(0, str(Path(__file__).parent.parent))
      from mcp_server.ainl_tools import AINLTools, _HAS_AINL
  except ImportError:
      _HAS_AINL = False
  
  
  class AINLValidator:
      """Auto-validates .ainl files after Edit/Write tool use."""
  
      def __init__(self, project_id: Optional[str] = None):
          self.tools = AINLTools() if _HAS_AINL else None
          self.project_id = project_id
  
      def should_validate(self, event: Dict[str, Any]) -> Optional[str]:
          """
          Check if we should validate based on event.
  
          Returns:
              File path if should validate, None otherwise
          """
          # Claude Code PostToolUse payload uses snake_case field names
          tool_name = event.get("tool_name", "")
          if tool_name not in ["Edit", "Write"]:
              # Read doesn't change the file — no point re-validating on read
              return None
  
          tool_input = event.get("tool_input", {}) or {}
          file_path = tool_input.get("file_path")
          if file_path and file_path.endswith(".ainl"):
              return file_path
  
          return None
  
      def validate_file(self, file_path: str) -> Optional[Dict[str, Any]]:
          """
          Validate .ainl file and return diagnostics.
  
          Returns:
              Validation result or None if can't validate
          """
          if not self.tools:
              return None
  
          try:
              # Read file
              with open(file_path, 'r', encoding='utf-8') as f:
                  source = f.read()
  
              # Validate with strict mode
              result = self.tools.validate(source, strict=True)
  
              return result
  
          except FileNotFoundError:
              return None
          except Exception as e:
              return {
                  "valid": False,
                  "error": f"Validation error: {e}"
              }
  
      def format_validation_output(self, file_path: str, validation: Dict[str, Any]) -> str:
          """Format validation results as a compact markdown block."""
          name = Path(file_path).name
  
          if validation.get("valid"):
              next_tools = validation.get('recommended_next_tools', [])
              msg = validation.get('message', 'Valid')
              out = f"**AINL Validation:** ✅ {name}\n{msg}"
              if next_tools:
                  out += f"\n**Next steps:** {', '.join(next_tools)}"
              return out
  
          diagnostics = validation.get("diagnostics", [])
          primary = validation.get("primary_diagnostic")
  
          out = f"**AINL Validation:** ❌ {name}\n"
          if primary:
              out += f"**Error:** {primary.get('message', 'Unknown error')}\n"
              if "line" in primary:
                  out += f"**Line:** {primary['line']}\n"
  
          repair_steps = validation.get("agent_repair_steps", [])
          if repair_steps:
              out += "**How to fix:**\n" + "".join(f"- {s}\n" for s in repair_steps)
  
          if len(diagnostics) > 1:
              out += f"\n**{len(diagnostics) - 1} additional issue(s)**"
  
          resources = validation.get("recommended_resources", [])
          if resources:
              out += f"\n**Resources:** {', '.join(resources)}"
  
          return out
  
  
  def main():
      """Hook entry point for PostToolUse."""
      if not _HAS_AINL:
          # Silently skip if AINL not installed
          return
  
      try:
          from shared.stdin import read_stdin_json
          event = read_stdin_json(hook_name="ainl_validator")
  
          # projectId is not in PostToolUse payloads — compute from cwd
          cwd = Path(event.get("cwd", str(Path.cwd())))
          project_id = get_project_id(cwd)
          validator = AINLValidator(project_id=project_id)
  
          # Check if we should validate
          file_path = validator.should_validate(event)
          if not file_path:
              return
  
          # Validate
          validation = validator.validate_file(file_path)
          if not validation:
              return
  
          # Format output
          output_text = validator.format_validation_output(file_path, validation)
  
          # PostToolUse context injection: hookSpecificOutput.additionalContext
          output = {
              "hookSpecificOutput": {
                  "additionalContext": output_text
              }
          }
  
          print(json.dumps(output))
  
      except Exception as e:
          # Silent failure
          sys.stderr.write(f"AINL validator error: {e}\n")
          pass
  
  
  if __name__ == "__main__":
      main()
  

• hooks/notifications.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  Notification poller for ainl-cortex.
  
  Fetches https://www.ainativelang.com/notifications once per session, filters for
  this plugin, surfaces unseen entries in the SessionStart banner, and optionally
  applies auto-updates when the server marks a release safe.
  
  Client algorithm (matches server contract):
    1. Require schema_version == 1 (forward-compat: accept > 1 silently too).
    2. Filter: targets must include "claude-code-plugin", "ainativelang", "ainl", or "*".
    3. Drop: expires_at present and now > expires_at.
    4. Sort: priority desc (default 0), then published_at desc.
    5. Auto-update: enabled + artifact == "ainl-cortex" + version in range.
  """
  
  import json
  import os
  import ssl
  import subprocess
  import sys
  import urllib.error
  import urllib.request
  from datetime import datetime, timezone
  from pathlib import Path
  from typing import Any, Dict, List, Optional, Tuple
  
  
  def _ssl_context() -> ssl.SSLContext:
      """Return an SSL context that works on macOS without system cert config."""
      try:
          import certifi
          return ssl.create_default_context(cafile=certifi.where())
      except ImportError:
          pass
      return ssl.create_default_context()
  
  NOTIFICATIONS_URL = "https://www.ainativelang.com/notifications"
  PLUGIN_ARTIFACT = "ainl-cortex"
  OUR_TARGETS: frozenset = frozenset({"claude-code-plugin", "ainativelang", "ainl", "*"})
  SEEN_FILE_REL = Path("a2a") / "notifications_seen.json"
  PLUGIN_JSON_REL = Path(".claude-plugin") / "plugin.json"
  
  
  # ── version helpers ────────────────────────────────────────────────────────────
  
  def _read_plugin_version(plugin_root: Path) -> str:
      try:
          data = json.loads((plugin_root / PLUGIN_JSON_REL).read_text())
          return str(data.get("version", "0.0.0"))
      except Exception:
          return "0.0.0"
  
  
  def _ver_tuple(v: str) -> Tuple[int, ...]:
      try:
          return tuple(int(x) for x in v.split(".")[:3])
      except Exception:
          return (0, 0, 0)
  
  
  def _version_in_range(current: str, min_v: Optional[str], max_v: Optional[str]) -> bool:
      cur = _ver_tuple(current)
      if min_v is not None and cur < _ver_tuple(min_v):
          return False
      if max_v is not None and cur > _ver_tuple(max_v):
          return False
      return True
  
  
  # ── datetime helper ────────────────────────────────────────────────────────────
  
  def _parse_dt(s: Optional[str]) -> Optional[datetime]:
      if not s:
          return None
      try:
          return datetime.fromisoformat(s.replace("Z", "+00:00"))
      except Exception:
          return None
  
  
  # ── seen-ID persistence ────────────────────────────────────────────────────────
  
  def _load_seen(plugin_root: Path) -> set:
      path = plugin_root / SEEN_FILE_REL
      try:
          data = json.loads(path.read_text())
          return set(data.get("seen_ids", []))
      except Exception:
          return set()
  
  
  def _save_seen(plugin_root: Path, seen: set) -> None:
      path = plugin_root / SEEN_FILE_REL
      path.parent.mkdir(parents=True, exist_ok=True)
      tmp = path.with_suffix(".tmp")
      tmp.write_text(json.dumps({"seen_ids": sorted(seen)}, indent=2), encoding="utf-8")
      os.replace(tmp, path)
  
  
  # ── network ────────────────────────────────────────────────────────────────────
  
  def _fetch(url: str, version: str, timeout: float) -> Optional[Dict[str, Any]]:
      req = urllib.request.Request(url, method="GET")
      req.add_header("Accept", "application/json")
      req.add_header("User-Agent", f"ainl-cortex/{version}")
      try:
          ctx = _ssl_context()
          with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
              raw = resp.read().decode("utf-8")
              return json.loads(raw)
      except (urllib.error.URLError, OSError, json.JSONDecodeError):
          return None
  
  
  # ── auto-update ────────────────────────────────────────────────────────────────
  
  def _try_auto_update(plugin_root: Path, notif: Dict[str, Any], current_version: str) -> Optional[str]:
      """
      Runs `git pull --ff-only` inside plugin_root.
      Returns a human-readable result string, or None if skipped.
      """
      au = notif.get("auto_update")
      if not (isinstance(au, dict) and au.get("enabled")):
          return None
      if au.get("artifact") != PLUGIN_ARTIFACT:
          return None
      if not _version_in_range(current_version, au.get("min_version"), au.get("max_version")):
          return None
  
      try:
          r = subprocess.run(
              ["git", "pull", "--ff-only"],
              cwd=str(plugin_root),
              capture_output=True,
              text=True,
              timeout=30,
          )
          if r.returncode == 0:
              try:
                  sys.path.insert(0, str(plugin_root))
                  from mcp_server.build_stamp import write_install_stamp
                  from mcp_server.mcp_reload import request_mcp_reload
                  write_install_stamp(plugin_root)
                  request_mcp_reload(plugin_root, reason="git_pull_auto_update")
              except Exception:
                  pass
              return f"auto-updated ainl-cortex: {r.stdout.strip()[:200]}"
          else:
              return f"auto-update attempted but failed: {(r.stderr or r.stdout or '').strip()[:200]}"
      except Exception as e:
          return f"auto-update error: {e}"
  
  
  # ── public API ─────────────────────────────────────────────────────────────────
  
  def poll(plugin_root: Path, config: Dict[str, Any]) -> Tuple[List[Dict[str, Any]], List[str]]:
      """
      Check the notifications feed and return:
        new_notifs  — list of notification dicts not yet seen (to show in banner)
        update_msgs — list of strings from any auto-update attempts
  
      Reads config["notifications"] for:
        enabled               (bool, default True)
        url                   (str, overrides NOTIFICATIONS_URL)
        check_timeout_seconds (float, default 5.0)
        auto_update           (bool, default False) — gate for running git pull
      """
      notif_cfg: Dict[str, Any] = config.get("notifications", {})
      if not notif_cfg.get("

• hooks/post_compact.pyGitHub ↗
  Read the script

  #!/usr/bin/env python3
  """
  PostCompact Hook — Update anchored summary after context compaction.
  
  After compaction, update the anchored summary to reflect the compacted state.
  This ensures the next session start injection shows the correct "in progress"
  context rather than stale prior-session data.
  """
  
  import json
  import sys
  import time
  from pathlib import Path
  
  sys.path.insert(0, str(Path(__file__).parent))
  sys.path.insert(0, str(Path(__file__).parent.parent / "mcp_server"))
  sys.path.insert(0, str(Path(__file__).parent.parent))
  
  from shared.mcp_bootstrap import ensure_hook_mcp_imports
  ensure_hook_mcp_imports()
  
  from shared.project_id import get_project_id
  from shared.logger import log_event, get_logger
  
  
  def _get_compact_project_id(plugin_root: Path) -> str:
      """Get the session's project_id from the startup-written sidecar.
  
      PostCompact payload doesn't include cwd, so we fall back to the sidecar
      written by startup.py. Without this, get_project_id() uses Path.cwd()
      which is the plugin root, not the user's actual project."""
      try:
          cid_file = plugin_root / "inbox" / "current_project_id.txt"
          if cid_file.exists():
              pid = cid_file.read_text(encoding="utf-8").strip()
              if pid:
                  return pid
      except Exception:
          pass
      return get_project_id()
  
  logger = get_logger("post_compact")
  
  try:
      import ainl_native as _ainl_native
      _NATIVE_OK = True
  except ImportError:
      _ainl_native = None
      _NATIVE_OK = False
  
  
  def main():
      try:
          from shared.stdin import read_stdin_json
          input_data = read_stdin_json(hook_name="post_compact")
          plugin_root = Path(__file__).parent.parent
          project_id = _get_compact_project_id(plugin_root)
  
          messages_before = input_data.get('messagesBefore', 0)
          messages_after = input_data.get('messagesAfter', 0)
          messages_removed = messages_before - messages_after
          estimated_tokens_saved = messages_removed * 200
  
          log_event("post_compact", {
              "project_id": project_id,
              "messages_before": messages_before,
              "messages_after": messages_after,
              "messages_removed": messages_removed,
              "estimated_tokens_saved": estimated_tokens_saved,
          })
          logger.info(f"PostCompact: {messages_removed} messages removed, ~{estimated_tokens_saved} tokens saved")
  
          # Update anchored summary to reflect post-compaction state
          if _NATIVE_OK:
              try:
                  db_path = Path.home() / ".claude" / "projects" / project_id / "graph_memory"
                  db_path.mkdir(parents=True, exist_ok=True)
                  native_db = str(db_path / "ainl_native.db")
                  store = _ainl_native.AinlNativeStore.open(native_db)
  
                  # Fetch existing summary to preserve task context
                  prior_raw = store.fetch_anchored_summary("claude-code")
                  prior_summary = "session compacted"
                  prior_ts = int(time.time())
                  if prior_raw:
                      try:
                          p = json.loads(prior_raw)
                          prior_summary = p.get("task_summary", prior_summary)
                          prior_ts = p.get("session_ts", prior_ts)
                      except Exception:
                          pass
  
                  payload = json.dumps({
                      "schema_version": 1,
                      "task_summary": prior_summary,
                      "outcome": "in_progress",
                      "post_compaction": True,
                      "messages_after_compaction": messages_after,
                      "tokens_saved_by_compaction": estimated_tokens_saved,
                      "session_ts": prior_ts,
                      "compacted_at": int(time.time()),
                      "project_id": project_id,
                  }, separators=(",", ":"))
  
                  node_id = store.upsert_anchored_summary("claude-code", payload)
                  logger.info(f"PostCompact anchored summary updated: {node_id}")
              except Exception as e:
                  logger.debug(f"PostCompact summary update failed (non-fatal): {e}")
  
          print(json.dumps({}), file=sys.stdout)
  
      except Exception as e:
          logger.error(f"PostCompact error: {e}")
          print(json.dumps({}), file=sys.stdout)
      finally:
          sys.exit(0)
  
  
  if __name__ == "__main__":
      main()
  

• hooks/post_tool_use.pyGitHub ↗
• hooks/pre_compact.pyGitHub ↗
• hooks/session_banner.pyGitHub ↗
• hooks/startup.pyGitHub ↗
• hooks/stop.pyGitHub ↗
• hooks/telemetry.pyGitHub ↗
• hooks/user_prompt_expansion.pyGitHub ↗
• hooks/user_prompt_submit.pyGitHub ↗

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