#!/usr/bin/env python3 """Aria Meta-Verify (KAR-589) — wasserdichter Self-Check ueber ALLE Regeln/Skripte/ Hooks/Timer/Standing-Orders. Findet kaputte Pfade, tote Timer, fehlende Hooks und Regeln ohne Enforcement — damit nichts still stirbt und Kais nicht manuell erinnern muss. Stufen: 1. scripts — alle scripts/*.py|*.sh auf kaputte Pfade ($HOME-literal, /projects/-root/, tote Refs) 2. timers — alle aria-*.timer enabled? service failed? 3. hooks — alle in settings.json referenzierten Hook-Files existieren + executable 4. standing — Standing-Order-Memories: Enforcement-Coverage (Hook vs nur-Text) 5. services — failed aria-services Exit 0 = clean, Exit 2 = kritische Findings (kaputte Pfade / tote Timer / fehlende Hooks). Aufruf: python3 aria-verify.py [--json] [--telegram] """ from __future__ import annotations import argparse import json import os import re import subprocess import sys import urllib.parse import urllib.request from pathlib import Path SCRIPTS = Path("/root/aria/scripts") MEMORY = Path("/root/.claude/projects/-/memory") SETTINGS = Path("/root/.claude/settings.json") HOOKS_DIR = Path("/root/.claude/hooks") CHAT_ID = "1164395546" CRIT, WARN, INFO = "CRITICAL", "WARN", "INFO" def sh(cmd: list[str], timeout=20) -> str: try: return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout).stdout except Exception as e: return f"[err {e}]" # ---------- Stufe 1: scripts ---------- def check_scripts(findings): if not SCRIPTS.exists(): return for p in sorted(SCRIPTS.glob("*.py")) + sorted(SCRIPTS.glob("*.sh")): if p.name == "aria-verify.py": continue # the linter contains the patterns by definition — never self-scan try: text = p.read_text(errors="replace") except Exception: continue is_py = p.suffix == ".py" in_doc = False # crude triple-quote docstring tracker (skip prose) for n, line in enumerate(text.splitlines(), 1): stripped = line.strip() if stripped.startswith("#"): continue if is_py and (stripped.count('"""') == 1 or stripped.count("'''") == 1): in_doc = not in_doc continue if in_doc: continue # literal $HOME inside a python string = never expands — # UNLESS wrapped in expandvars/expanduser/shell-call/getenv (those DO expand) if is_py and re.search(r'["\'][^"\']*\$HOME', line): safe = re.search(r"expandvars|expanduser|\.system\(|shell\s*=\s*True|getenv|environ|popen", line) if not safe: findings.append((CRIT, "scripts", f"{p.name}:{n} literal $HOME in Python-String ohne expandvars (expandiert nie)")) # /projects/-root/ : memory MUST be /projects/-/ ; jsonl/session dirs are cwd-dependent (both exist) → context-check if "/projects/-root/" in line: if "memory" in line.lower(): findings.append((CRIT, "scripts", f"{p.name}:{n} Memory-Pfad /projects/-root/memory (soll /projects/-/memory)")) else: findings.append((WARN, "scripts", f"{p.name}:{n} /projects/-root/ cwd-abhaengig (jsonl/session) — pruefen ob cwd=/root stimmt")) # hardcoded absolute file refs -> verify existence for m in re.finditer(r'["\'](/root/(?:aria|\.claude)/[^"\']+\.(?:sh|py))["\']', line): ref = m.group(1) if not Path(ref).exists(): findings.append((WARN, "scripts", f"{p.name}:{n} referenziert nicht-existente Datei {ref}")) # ---------- Stufe 2: timers ---------- def check_timers(findings, info): out = sh(["systemctl", "list-timers", "--all", "--no-legend"]) timers = sorted(set(re.findall(r"(aria-[\w-]+\.timer)", out))) # also units on disk that may not be listed disk = sorted(set(t.name for t in Path("/etc/systemd/system").glob("aria-*.timer"))) info["timers_total"] = len(disk) for t in disk: en = sh(["systemctl", "is-enabled", t]).strip() svc = t.replace(".timer", ".service") failed = sh(["systemctl", "is-failed", svc]).strip() if en not in ("enabled", "static", "enabled-runtime"): # disabled may be intentional (e.g. akp-to-linear) — WARN not CRIT findings.append((WARN, "timers", f"{t} ist {en} (beabsichtigt? sonst tot)")) if failed == "failed": findings.append((CRIT, "timers", f"{svc} im failed-Zustand")) # ---------- Stufe 3: hooks ---------- def check_hooks(findings, info): if not SETTINGS.exists(): findings.append((CRIT, "hooks", "settings.json fehlt")) return d = json.loads(SETTINGS.read_text()) cmds = [] for evt, groups in d.get("hooks", {}).items(): for g in groups: for h in g.get("hooks", []): cmds.append((evt, h.get("command", ""))) info["hooks_total"] = len(cmds) wired = set() for evt, c in cmds: m = re.search(r'(/root/\.claude/hooks/[\w.-]+\.sh)', c) if m: wired.add(Path(m.group(1)).name) sh_path = Path(m.group(1)) if not sh_path.exists(): findings.append((CRIT, "hooks", f"{evt}: Hook-Skript fehlt {m.group(1)}")) else: # Wasserdicht: ein Hook der einen Validator/Helper per absolutem # Pfad aufruft (z.B. brain-write-validator.sh → validators/*.py) # darf nicht stillschweigend auf eine fehlende Datei zeigen. try: sh_text = sh_path.read_text(errors="replace") except Exception: sh_text = "" for dep in re.finditer(r'(/root/\.claude/hooks/[\w./-]+\.py)', sh_text): dp = Path(dep.group(1)) if not dp.exists(): findings.append((CRIT, "hooks", f"{sh_path.name} ruft fehlende Datei {dp} auf")) # detect orphan hooks: hooks/*.sh that exist but are NOT wired in settings.json # (these are silently dead — e.g. a safety hook that never runs) KNOWN_HELPERS = {"telegram-log-in.sh", "telegram-log-out.sh"} # called indirectly / logging for hp in HOOKS_DIR.glob("*.sh"): if hp.name not in wired and hp.name not in KNOWN_HELPERS: sev = CRIT if "safety" in hp.name or "guard" in hp.name or "critic" in hp.name else WARN findings.append((sev, "hooks", f"{hp.name} existiert aber ist NICHT in settings.json gewired (toter Hook?)")) # ---------- Stufe 4: standing-order enforcement coverage ---------- def check_standing(findings, info): if not MEMORY.exists(): return feedbacks = [p for p in MEMORY.glob("feedback_*.md")] info["standing_total"] = len(feedbacks) # collect text of all hook scripts to detect enforcement hook_text = "" for hp in HOOKS_DIR.glob("*.sh"): try: hook_text += hp.read_text(errors="replace").lower() except Exception: pass # validators dir too vdir = HOOKS_DIR / "validators" if vdir.exists(): for hp in vdir.glob("*"): try: hook_text += hp.read_text(errors="replace").lower() except Exception: pass enforced = 0 unenforced = [] for fb in feedbacks: slug = fb.stem.replace("feedback_", "").replace("_", "-") slug_us = fb.stem.replace("feedback_", "") # heuristic: slug or key tokens appear in a hook script toks = [t for t in re.split(r"[-_]", slug_us) if len(t) > 4][:3] hit = slug in hook_text or slug_us in hook_text or (toks and all(t in hook_text for t in toks)) if hit: enforced += 1 else: unenforced.append(fb.stem) info["standing_enforced"] = enforced info["standing_unenforced"] = len(unenforced) # coverage is informational (most rules are judgment-based, not hook-able) — report ratio findings.append((INFO, "standing", f"Enforcement-Coverage: {enforced}/{len(feedbacks)} Standing-Orders haben einen Hook " f"({len(unenforced)} sind nur-Text/Judgment)")) # ---------- Stufe 5: services ---------- def check_services(findings): out = sh(["systemctl", "--failed", "--no-legend"]) for line in out.splitlines(): if "aria-" in line: unit = line.split()[0] if line.split() else line findings.append((CRIT, "services", f"failed service: {unit}")) def md2(s): return re.sub(r"([_*\[\]()~`>#+\-=|{}.!\\])", r"\\\1", str(s)) def get_tg_token(): for path in ["/root/.claude/channels/telegram/.env", "/root/aria/.env"]: p = Path(path) if p.exists(): for line in p.read_text().splitlines(): if "=" in line and not line.strip().startswith("#"): k, _, v = line.partition("=") if k.strip() in ("TG_TOKEN", "TELEGRAM_BOT_TOKEN", "TELEGRAM_TOKEN") and v.strip(): return v.strip().strip('"').strip("'") return os.environ.get("TG_TOKEN") def send_tg(text): tok = get_tg_token() if not tok: return False data = urllib.parse.urlencode({"chat_id": CHAT_ID, "text": text, "parse_mode": "MarkdownV2", "disable_web_page_preview": "true"}).encode() try: with urllib.request.urlopen(urllib.request.Request( f"https://api.telegram.org/bot{tok}/sendMessage", data=data, method="POST"), timeout=15) as r: return r.status == 200 except Exception as e: print(f"[verify] tg error: {e}", file=sys.stderr); return False def main(): ap = argparse.ArgumentParser() ap.add_argument("--json", action="store_true") ap.add_argument("--telegram", action="store_true") args = ap.parse_args() findings = [] info = {} check_scripts(findings) check_timers(findings, info) check_hooks(findings, info) check_standing(findings, info) check_services(findings) crit = [f for f in findings if f[0] == CRIT] warn = [f for f in findings if f[0] == WARN] inf = [f for f in findings if f[0] == INFO] if args.json: print(json.dumps({"info": info, "critical": len(crit), "warn": len(warn), "findings": [{"sev": s, "area": a, "msg": m} for s, a, m in findings]}, indent=2)) else: print(f"=== aria-verify === {info}") for s, a, m in crit + warn + inf: print(f"[{s}] {a}: {m}") print(f"\nSUMMARY: {len(crit)} CRITICAL, {len(warn)} WARN, {len(inf)} INFO") if args.telegram: L = ["*Aria Meta\\-Verify*", ""] L.append(f"Timer {info.get('timers_total','?')} · Hooks {info.get('hooks_total','?')} · " f"Standing\\-Orders {info.get('standing_total','?')} \\(enforced {info.get('standing_enforced','?')}\\)") L.append("") if crit: L.append(f"🔴 *{len(crit)} CRITICAL:*") for s, a, m in crit[:8]: L.append(f"• {md2(m)}") else: L.append("✅ *0 kritische Findings* — alle Pfade/Timer/Hooks ok") if warn: L.append("") L.append(f"🟠 {len(warn)} WARN \\(z\\.B\\. bewusst disabled\\)") send_tg("\n".join(L)) return 2 if crit else 0 if __name__ == "__main__": sys.exit(main())