#!/usr/bin/env python3 """ Aria Hook-Manifest (KAR-686, AgentScope-Pattern #1) =================================================== Macht Arias Hook-Setup legibel, geordnet und validiert — das adoptierbare Äquivalent zu AgentScopes Onion-Middleware-Chain (der Dispatch selbst gehört Claude Codes Runtime, NICHT diesem Script). Liest /root/.claude/settings.json und erzeugt ein geordnetes Manifest aller Hooks: pro Event in Ausführungs-Reihenfolge (order = die "Chain"), mit Matcher, Script, Timeout, Zweck. Modi: --check Validierung: jedes referenzierte bash-Script existiert? Orphan-Scripts im hooks-Dir die in keinem Event registriert sind? Exit 1 bei fehlendem Script (Drift/kaputt). --render Markdown-Tabelle (für die HOOKS.md Auto-Sektion zwischen den Markern) — killt Drift. --json Manifest als JSON. Default ohne Flag: --check. """ from __future__ import annotations import argparse import json import re import sys import time from pathlib import Path SETTINGS_PATH = Path("/root/.claude/settings.json") HOOKS_DIR = Path("/root/.claude/hooks") EVENT_ORDER = [ "SessionStart", "UserPromptSubmit", "PreToolUse", "PostToolUse", "PreCompact", "Stop", "SessionEnd", ] # Kurz-Zweck pro Script (gepflegt; unbekannte -> "" und vom --check als Doku-Lücke flagbar) PURPOSE = { "session-start.sh": "Brain-Snapshot + Linear + Mode laden", "telegram-log-in.sh": "Inbound-Telegram → Chat-Log", "auto-memory-inject.sh": "Memory/Brain-Auto-Context injizieren (KAR-80)", "telegram-reply-discipline.sh": "Reminder: Telegram-Inbound braucht reply-Tool", "aria-effort-detect.sh": "Effort-Auto-Eskalation erkennen", "pre-tool-safety.sh": "Bash-Safety-Gate (destructive Commands)", "tdd-guard.sh": "TDD-Gate vor Write/Edit", "gateguard.sh": "Fact-Forcing-Gate vor Write/Edit (KAR-636)", "pre-output-critic.sh": "Format-Lint vor Telegram-Reply (KAR-65)", "telegram-log-out.sh": "Outbound-Telegram → Chat-Log", "brain-write-validator.sh": "Brain-Frontmatter/Wikilink-Validierung", "post-tool-failure.sh": "Tool-Failure-Diagnose-Hinweis", "pre-compact.sh": "ACTIVE_TASK-Backup + HANDOFF-Stub vor Compaction", "session-stop.sh": "Session-Wrap-Tasks", } def _script_path_from_command(command: str) -> str | None: """Extrahiert den bash-Script-Pfad aus 'bash /pfad/x.sh ...'. None bei Nicht-Script (z.B. 'rtk hook claude').""" m = re.match(r"^\s*bash\s+(\S+\.sh)", command) return m.group(1) if m else None def build_manifest(settings: dict) -> list[dict]: """Pure: settings-dict -> geordnete Hook-Liste. order = Ausführungsindex pro Event.""" hooks_cfg = settings.get("hooks", {}) rows: list[dict] = [] # bekannte Events zuerst in kanonischer Reihenfolge, dann unbekannte events = [e for e in EVENT_ORDER if e in hooks_cfg] events += [e for e in hooks_cfg if e not in EVENT_ORDER] for event in events: order = 0 for group in hooks_cfg.get(event, []): matcher = group.get("matcher", "*") for h in group.get("hooks", []): cmd = h.get("command", "") spath = _script_path_from_command(cmd) rows.append({ "event": event, "order": order, "matcher": matcher, "command": cmd, "type": h.get("type", "command"), "timeout": h.get("timeout"), "script_path": spath, "script_name": Path(spath).name if spath else None, "is_external": spath is None, # z.B. rtk-Binary "purpose": PURPOSE.get(Path(spath).name, "") if spath else "", }) order += 1 return rows def validate(manifest: list[dict], hooks_dir: Path) -> dict: """Findet fehlende Scripts (kaputt) + Orphan-Scripts (im Dir, nirgends registriert) + Doku-Lücken.""" missing, no_purpose = [], [] registered = set() for r in manifest: if r["is_external"]: continue registered.add(r["script_name"]) if not Path(r["script_path"]).is_file(): missing.append(r["script_path"]) elif not r["purpose"]: no_purpose.append(r["script_name"]) # Orphans: *.sh im hooks-Dir die in keinem Event hängen (Sub-Helper ausgenommen via Heuristik nicht möglich → alle listen) on_disk = {p.name for p in hooks_dir.glob("*.sh") if not p.name.endswith(".bak") and ".bak." not in p.name} orphans = sorted(on_disk - registered) return {"missing": missing, "orphans": orphans, "no_purpose": sorted(set(no_purpose))} def render_markdown(manifest: list[dict]) -> str: """Markdown-Tabellen pro Event, in Ausführungs-Reihenfolge.""" out = ["", ""] cur = None for r in manifest: if r["event"] != cur: cur = r["event"] out += [f"### Event: `{cur}`", "", "| # | Matcher | Hook | Timeout | Zweck |", "|---|---|---|---|---|"] name = r["script_name"] or f"`{r['command']}` (extern)" to = f"{r['timeout']}s" if r["timeout"] else "—" out.append(f"| {r['order']} | `{r['matcher']}` | {name} | {to} | {r['purpose'] or '—'} |") out.append("") return "\n".join(out) MARKER_START = "" MARKER_END = "" HOOKS_MD_HEADING = "## 3. Claude Code Hooks" def splice_section(text: str, rendered: str) -> str: """Setzt den Auto-Manifest-Block in HOOKS.md ein/aktualisiert ihn — ohne die manuelle Detail-Tabelle zu zerstören (additiv beim ersten Mal, marker-replace danach).""" block = f"{MARKER_START}\n{rendered}\n{MARKER_END}" if MARKER_START in text and MARKER_END in text: pre = text[:text.index(MARKER_START)] post = text[text.index(MARKER_END) + len(MARKER_END):] return pre + block + post # Erst-Insertion: direkt nach der Sektions-Überschrift idx = text.find(HOOKS_MD_HEADING) if idx == -1: # Heading fehlt -> ans Ende anhängen return text.rstrip() + "\n\n" + HOOKS_MD_HEADING + "\n\n" + block + "\n" line_end = text.find("\n", idx) insert_at = line_end + 1 return text[:insert_at] + "\n" + block + "\n" + text[insert_at:] def main(argv: list[str]) -> int: ap = argparse.ArgumentParser() ap.add_argument("--check", action="store_true") ap.add_argument("--render", action="store_true") ap.add_argument("--json", action="store_true") ap.add_argument("--update-hooks-md", metavar="PATH", nargs="?", const="/root/aria/brain/HOOKS.md", default=None) ap.add_argument("--settings", default=str(SETTINGS_PATH)) args = ap.parse_args(argv) try: settings = json.loads(Path(args.settings).read_text()) except (OSError, json.JSONDecodeError) as e: print(f"[hook-manifest] FEHLER beim Lesen {args.settings}: {e}", file=sys.stderr) return 2 manifest = build_manifest(settings) if args.json: print(json.dumps(manifest, indent=2, ensure_ascii=False)) return 0 if args.render: print(render_markdown(manifest)) return 0 if args.update_hooks_md: target = Path(args.update_hooks_md) try: old = target.read_text(encoding="utf-8") except OSError as e: print(f"[hook-manifest] FEHLER beim Lesen {target}: {e}", file=sys.stderr) return 2 new = splice_section(old, render_markdown(manifest)) if new == old: print(f"[hook-manifest] {target} bereits aktuell, kein Write.") return 0 bak = target.with_suffix(target.suffix + f".bak.{int(time.time())}") bak.write_text(old, encoding="utf-8") target.write_text(new, encoding="utf-8") print(f"[hook-manifest] {target} aktualisiert (Backup: {bak.name}).") return 0 # Default: --check v = validate(manifest, HOOKS_DIR) n_ext = sum(1 for r in manifest if r["is_external"]) print(f"[hook-manifest] {len(manifest)} Hooks über {len({r['event'] for r in manifest})} Events " f"({n_ext} extern/Binary).") if v["orphans"]: print(f" ⚠ Orphan-Scripts (im hooks-Dir, nirgends registriert): {v['orphans']}") if v["no_purpose"]: print(f" ⚠ Ohne Zweck-Doku (PURPOSE ergänzen): {v['no_purpose']}") if v["missing"]: print(f" ✗ FEHLENDE Scripts (kaputter Hook!): {v['missing']}") return 1 print(" ✓ Alle registrierten Hook-Scripts existieren.") return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))