#!/usr/bin/env python3 """aria-brain-remediate — Brain-Health-Score + Dependency-ordered Remediation Loop. KAR-624: Aria-Analog zu `gbrain doctor --remediate --target-score 90 --max-usd 5`. Scoring-Formel (0–100): Basis: 100 Punkte — Abzüge für jede Kategorie (max pro Kategorie dokumentiert): 1. Identity-Files fehlen : -10 pro fehlendem File (max -50, 5 sind critical) 2. Dangling Wikilinks : -0.05 pro dirtym Link (max -20) 3. Orphan-Notizen : -0.03 pro Orphan (max -15), YouTube-/akp-Hashes ignoriert 4. Fehlende Folder-INDEX.md : -3 pro Folder >5 Notes ohne INDEX (max -21, 7 Folders) 5. Uncommitted Changes : -5 wenn >0, -10 wenn >20 Changes 6. Stale Notes : -0.5 pro Note >90 Tage nicht bearbeitet (max -10) 7. Git-Sync-Lag : -5 wenn letzte Commit >7 Tage Score = max(0, 100 - sum(Abzüge)) Dependency-Reihenfolge (wichtig!): 1. Identity-Fix (ohne Identity können andere Checks falsch sein) 2. Dangling-Wikilink-Fix (bevor Orphan-Graph berechnet: dangling verzerrt Orphan-Zählung) 3. Missing-INDEX generieren (strukturell → danach Graph stabiler) 4. Git-Commit (nach Fixes, bevor Sync-Check) 5. Stale-Notes prüfen (manuell — nie auto-archivieren) 6. Orphans prüfen (letzter: sauberster Graph nach allen Fixes) Cost-Cap-Design: Alle lokalen Schritte = $0.00 USD. Zukünftige LLM-gestützte Schritte (z.B. Auto-Beschreibung für Orphan-Notes) würden `est_usd` > 0 melden und werden gestoppt wenn `spent_usd + est_usd > max_usd`. Heute: alle Schritte sind $0 — das Interface ist trotzdem vollständig implementiert. Usage: python3 aria-brain-remediate.py # dry-run (Standard) python3 aria-brain-remediate.py --dry-run # explizit dry-run python3 aria-brain-remediate.py --apply --target-score 90 --max-usd 5.0 python3 aria-brain-remediate.py --json # JSON-Output (immer dry-run sofern kein --apply) python3 aria-brain-remediate.py --apply --json # JSON + apply Safety-Garantien: - --apply führt NUR reversible, gut verstandene Fixes aus. - Notizen werden NIEMALS gelöscht. - Alles als "needs-human" markierte wird übersprungen und nur gelistet. - Git-Commits werden NICHT automatisch gepusht. """ from __future__ import annotations import argparse import json import os import re import subprocess import sys from collections import defaultdict from datetime import datetime, timedelta from difflib import SequenceMatcher from pathlib import Path from typing import Any # --------------------------------------------------------------------------- # Konfiguration # --------------------------------------------------------------------------- BRAIN_ROOT = Path("/root/aria/brain") CITATION_FIXER = Path("/root/aria/scripts/aria-citation-fixer.py") SKIP_DIRS = {".git", "akp", "youtube", ".obsidian", "HANDOFF.archive", "CORRECTIONS.archive"} IDENTITY_FILES = [ "SOUL.md", "IDENTITY.md", "USER.md", "TOOLS.md", "CORRECTIONS.md", "SELF-IMPROVEMENT.md", "HEARTBEAT.md", "HOOKS.md", "CLAUDE.md", "HANDOFF.md", ] # Folders die einen INDEX.md verdienen wenn sie >5 Notes haben INDEXED_FOLDERS = [ "00-Inbox", "01-Projekte", "02-Wissen", "03-Recherchen", "04-Feedback", "05-Referenzen", "06-Daily", ] WIKILINK_RE = re.compile(r"\[\[([^\]\|\n]+?)(?:\|([^\]\n]+?))?\]\]") # Hash-artiger Dateiname (YouTube-/AKP-Hashes) → kein echter Orphan HASH_STEM_RE = re.compile(r"^[0-9a-f]{6}[-—]") STALE_DAYS = 90 GIT_STALE_DAYS = 7 # Maximale Abzüge pro Kategorie MAX_IDENTITY_PENALTY = 50 # 10 × 5 kritische Files MAX_DANGLING_PENALTY = 20 MAX_ORPHAN_PENALTY = 15 MAX_INDEX_PENALTY = 21 # 3 × 7 Folders MAX_UNCOMMITTED_PENALTY = 10 MAX_STALE_PENALTY = 10 MAX_GIT_LAG_PENALTY = 5 # --------------------------------------------------------------------------- # Daten-Sammlung # --------------------------------------------------------------------------- def _build_slug_index(root: Path) -> dict[str, list[Path]]: idx: dict[str, list[Path]] = defaultdict(list) for p in root.rglob("*.md"): if any(skip in p.parts for skip in SKIP_DIRS): continue idx[p.stem].append(p) return idx def _collect_all_notes(root: Path) -> list[Path]: notes = [] for p in root.rglob("*.md"): if any(skip in p.parts for skip in SKIP_DIRS): continue notes.append(p) return notes def check_identity_files(root: Path) -> dict[str, Any]: missing = [] present = [] for f in IDENTITY_FILES: p = root / f if p.exists(): present.append(f) else: missing.append(f) return {"missing": missing, "present": present, "total": len(IDENTITY_FILES)} def check_dangling_wikilinks(root: Path) -> dict[str, Any]: """Nutzt die Logik aus aria-citation-fixer.""" slug_index = _build_slug_index(root) all_slugs = set(slug_index.keys()) dirty: dict[str, list[tuple[int, str, str | None]]] = defaultdict(list) total_links = 0 dirty_count = 0 for p in root.rglob("*.md"): if any(skip in p.parts for skip in SKIP_DIRS): continue try: text = p.read_text(encoding="utf-8", errors="replace") except OSError: continue for m in WIKILINK_RE.finditer(text): total_links += 1 slug = m.group(1).strip() clean = slug.split("#")[0].strip() if not clean: continue base = Path(clean).name if base in all_slugs or clean in all_slugs: continue # Fuzzy-Suggestion target_l = base.lower() best, best_score = None, 0.0 for c in all_slugs: score = SequenceMatcher(None, target_l, c.lower()).ratio() if score > best_score: best, best_score = c, score suggestion = best if best_score >= 0.85 else None rel = p.relative_to(root).as_posix() dirty[rel].append((m.start(), slug, suggestion)) dirty_count += 1 fixable = sum( 1 for drifts in dirty.values() for _, _, sug in drifts if sug is not None ) return { "total_links": total_links, "dirty_count": dirty_count, "fixable_count": fixable, "files_with_drift": dict(dirty), } def check_orphan_notes(root: Path) -> dict[str, Any]: """Notizen die von keinem [[Wikilink]] referenziert werden.""" slug_index = _build_slug_index(root) referenced: set[str] = set() for p in root.rglob("*.md"): if any(skip in p.parts for skip in SKIP_DIRS): continue try: text = p.read_text(encoding="utf-8", errors="replace") except OSError: continue for m in WIKILINK_RE.finditer(text): slug = m.group(1).strip().split("#")[0].strip() base = Path(slug).name referenced.add(base) referenced.add(slug) orphans = [] hash_orphans = [] for stem, paths in slug_index.items(): if stem not in referenced: if HASH_STEM_RE.match(stem): hash_orphans.append(stem) else: orphans.append(stem) return { "orphan_count": len(orphans), "hash_orphan_count": len(hash_orphans), "orphan_samples": sorted(orphans)[:10], "total_notes": sum(len(v) for v in slug_index.values()), } def check_missing_indexes(root: Path) -> dict[str, Any]: """Folders mit >5 Notes ohne INDEX.md oder README.md.""" missing = [] present = [] for folder_name in INDEXED_FOLDERS: folder = root / folder_name if not folder.exists(): continue has_index = (folder / "INDEX.md").exists() or (folder / "README.md").exists() note_count = len(list(folder.glob("*.md"))) if has_index: present.append({"folder": folder_name, "notes": note_count}) else: if note_count > 5: missing.append({"folder": folder_name, "notes": note_count}) elif note_count > 0: # Kleiner Folder → kein Index nötig present.append({"folder": folder_name, "notes": note_count, "small": True}) return {"missing": missing, "present": present} def check_git_status(root: Path) -> dict[str, Any]: try: r = subprocess.run( ["git", "status", "--porcelain"], capture_output=True, text=True, cwd=root, timeout=15 ) uncommitted = [l.strip() for l in r.stdout.splitlines() if l.strip()] r2 = subprocess.run( ["git", "log", "-1", "--format=%ct"], capture_output=True, text=True, cwd=root, timeout=15 ) last_commit_ts = int(r2.stdout.strip()) if r2.stdout.strip() else 0 last_commit_dt = datetime.fromtimestamp(last_commit_ts) if last_commit_ts else None age_days = (datetime.now() - last_commit_dt).days if last_commit_dt else 999 r3 = subprocess.run( ["git", "log", "-1", "--format=%h %s"], capture_output=True, text=True, cwd=root, timeout=15 ) last_commit_msg = r3.stdout.strip()[:80] return { "uncommitted_count": len(uncommitted), "uncommitted_files": uncommitted[:10], "age_days": age_days, "last_commit": last_commit_msg, "git_ok": True, } except Exception as e: return {"uncommitted_count": 0, "age_days": 0, "git_ok": False, "error": str(e)} def check_stale_notes(root: Path) -> dict[str, Any]: """Notizen die >STALE_DAYS Tage nicht bearbeitet wurden.""" threshold = datetime.now() - timedelta(days=STALE_DAYS) stale = [] for p in root.rglob("*.md"): if any(skip in p.parts for skip in SKIP_DIRS): continue # Daily-Logs sind per Design alt → kein Stale-Marker parts = p.parts if "06-Daily" in parts or "CORRECTIONS.archive" in parts or "HANDOFF.archive" in parts: continue mtime = datetime.fromtimestamp(p.stat().st_mtime) if mtime < threshold: stale.append({ "path": p.relative_to(root).as_posix(), "mtime": mtime.strftime("%Y-%m-%d"), }) return { "stale_count": len(stale), "stale_samples": stale[:5], } # --------------------------------------------------------------------------- # Score-Berechnung # --------------------------------------------------------------------------- def compute_score( identity: dict, dangling: dict, orphans: dict, indexes: dict, git: dict, stale: dict, ) -> tuple[float, dict[str, float]]: """Gibt (score, breakdown) zurück. score ∈ [0, 100].""" penalties: dict[str, float] = {} # 1. Identity identity_penalty = min(len(identity["missing"]) * 10.0, MAX_IDENTITY_PENALTY) if identity_penalty > 0: penalties["identity_missing"] = identity_penalty # 2. Dangling Wikilinks dangling_penalty = min(dangling["dirty_count"] * 0.05, MAX_DANGLING_PENALTY) if dangling_penalty > 0: penalties["dangling_wikilinks"] = round(dangling_penalty, 2) # 3. Orphan-Notizen (Hash-Orphans ignorieren) orphan_penalty = min(orphans["orphan_count"] * 0.03, MAX_ORPHAN_PENALTY) if orphan_penalty > 0: penalties["orphan_notes"] = round(orphan_penalty, 2) # 4. Fehlende Folder-INDEXes index_penalty = min(len(indexes["missing"]) * 3.0, MAX_INDEX_PENALTY) if index_penalty > 0: penalties["missing_indexes"] = index_penalty # 5. Uncommitted Changes uc = git.get("uncommitted_count", 0) if uc > 20: penalties["uncommitted_heavy"] = 10.0 elif uc > 0: penalties["uncommitted_light"] = 5.0 # 6. Stale Notes stale_penalty = min(stale["stale_count"] * 0.5, MAX_STALE_PENALTY) if stale_penalty > 0: penalties["stale_notes"] = round(stale_penalty, 2) # 7. Git-Sync-Lag if git.get("age_days", 0) > GIT_STALE_DAYS: penalties["git_sync_lag"] = 5.0 total_penalty = sum(penalties.values()) score = max(0.0, 100.0 - total_penalty) return round(score, 1), penalties def compute_max_reachable(score: float, penalties: dict, identity: dict, dangling: dict) -> float: """Maximaler Score nach allen auto-fix-fähigen Schritten (ohne human-steps).""" # Auto-fix-fähig: identity (wenn fixable), dangling (fixable subset), missing indexes, git commit removable = 0.0 removable += penalties.get("identity_missing", 0.0) # generell fixable wenn Files kreiert # Dangling: nur fixable subset fixable_pct = dangling["fixable_count"] / max(1, dangling["dirty_count"]) removable += penalties.get("dangling_wikilinks", 0.0) * fixable_pct removable += penalties.get("missing_indexes", 0.0) removable += penalties.get("uncommitted_light", 0.0) + penalties.get("uncommitted_heavy", 0.0) removable += penalties.get("git_sync_lag", 0.0) # Stale + Orphans → needs-human return round(min(100.0, score + removable), 1) # --------------------------------------------------------------------------- # Remediation-Plan # --------------------------------------------------------------------------- def build_plan( score: float, identity: dict, dangling: dict, orphans: dict, indexes: dict, git: dict, stale: dict, penalties: dict, ) -> list[dict]: """Dependency-ordered Plan. Jeder Schritt enthält: step, description, would_fix (Anzahl Issues), est_usd, auto_applicable, needs_human. """ plan = [] # Schritt 1: Identity-Files erzeugen (falls fehlend) if identity["missing"]: plan.append({ "step": "fix_identity_files", "order": 1, "description": f"Fehlende Identity-Files anlegen: {', '.join(identity['missing'])}", "would_fix": len(identity["missing"]), "penalty_removed": penalties.get("identity_missing", 0.0), "est_usd": 0.0, "auto_applicable": True, "needs_human": False, "method": "create_stub_identity_file", }) # Schritt 2: Dangling Wikilinks fixen (citation-fixer --apply) if dangling["dirty_count"] > 0: fixable = dangling["fixable_count"] unfixable = dangling["dirty_count"] - fixable plan.append({ "step": "fix_dangling_wikilinks", "order": 2, "description": ( f"Dangling Wikilinks reparieren via aria-citation-fixer --apply " f"({fixable} auto-fixable, {unfixable} ohne Fuzzy-Match → übersprungen)" ), "would_fix": fixable, "total_dirty": dangling["dirty_count"], "unfixable": unfixable, "penalty_removed": round( penalties.get("dangling_wikilinks", 0.0) * (fixable / max(1, dangling["dirty_count"])), 2, ), "est_usd": 0.0, "auto_applicable": fixable > 0, "needs_human": unfixable > 0, "method": "invoke_citation_fixer_apply", "human_note": f"{unfixable} Links ohne Fuzzy-Match müssen manuell geprüft werden.", }) # Schritt 3: Fehlende Folder-INDEX.md generieren if indexes["missing"]: plan.append({ "step": "generate_folder_indexes", "order": 3, "description": ( f"INDEX.md für {len(indexes['missing'])} Folder generieren: " + ", ".join(f["folder"] for f in indexes["missing"]) ), "would_fix": len(indexes["missing"]), "folders": [f["folder"] for f in indexes["missing"]], "penalty_removed": penalties.get("missing_indexes", 0.0), "est_usd": 0.0, "auto_applicable": True, "needs_human": False, "method": "generate_index_files", }) # Schritt 4: Git Commit (nach lokalen Fixes) uc = git.get("uncommitted_count", 0) if uc > 0: plan.append({ "step": "git_commit_fixes", "order": 4, "description": f"git add -A && git commit '{uc} uncommitted changes'", "would_fix": uc, "penalty_removed": penalties.get("uncommitted_light", 0.0) + penalties.get("uncommitted_heavy", 0.0), "est_usd": 0.0, "auto_applicable": True, "needs_human": False, "method": "git_commit", "safety_note": "Kein git push — nur lokaler Commit.", }) # Schritt 5: Stale Notes → immer needs-human if stale["stale_count"] > 0: plan.append({ "step": "review_stale_notes", "order": 5, "description": ( f"{stale['stale_count']} Notizen sind >{STALE_DAYS} Tage nicht bearbeitet. " "Manuell prüfen: superseden, archivieren oder als aktuell markieren." ), "would_fix": stale["stale_count"], "penalty_removed": penalties.get("stale_notes", 0.0), "est_usd": 0.0, "auto_applicable": False, "needs_human": True, "samples": stale["stale_samples"], "human_note": "Auto-Archivierung ist nicht sicher — Kais entscheidet.", }) # Schritt 6: Orphan-Notes → immer needs-human (keine Löschung!) real_orphans = orphans["orphan_count"] if real_orphans > 0: plan.append({ "step": "review_orphan_notes", "order": 6, "description": ( f"{real_orphans} Notizen werden von keinem [[Wikilink]] referenziert " f"(+ {orphans['hash_orphan_count']} Hash-Stubs ignoriert). " "Manuell verlinken oder archivieren." ), "would_fix": real_orphans, "penalty_removed": penalties.get("orphan_notes", 0.0), "est_usd": 0.0, "auto_applicable": False, "needs_human": True, "samples": orphans["orphan_samples"], "human_note": "Niemals auto-löschen. Verlinkung oder Archivierung ist Kais-Entscheidung.", }) return plan # --------------------------------------------------------------------------- # Apply-Logik # --------------------------------------------------------------------------- def apply_fix_identity_files(brain: Path, step: dict) -> tuple[int, str]: """Stub-Identity-Files anlegen (nur wenn wirklich fehlt).""" fixed = 0 notes = [] for fname in IDENTITY_FILES: p = brain / fname if p.exists(): continue stub = f"""--- title: {fname.replace('.md', '')} type: identity tags: [identity, auto-generated] date: {datetime.now().strftime('%Y-%m-%d')} status: draft --- # {fname.replace('.md', '')} > Auto-generierter Stub von aria-brain-remediate.py (KAR-624). > Bitte mit echtem Inhalt befüllen. """ p.write_text(stub, encoding="utf-8") fixed += 1 notes.append(fname) msg = f"Angelegt: {', '.join(notes)}" if notes else "Keine Änderungen nötig." return fixed, msg def apply_fix_dangling_wikilinks(brain: Path, step: dict) -> tuple[int, str]: """aria-citation-fixer --apply aufrufen.""" if not CITATION_FIXER.exists(): return 0, f"FEHLER: {CITATION_FIXER} nicht gefunden." try: r = subprocess.run( [sys.executable, str(CITATION_FIXER), "--apply", "--threshold", "0.85"], capture_output=True, text=True, timeout=120, ) output = r.stdout + r.stderr # Anzahl Fixes aus Output extrahieren m = re.search(r"applied (\d+) fix", output) applied = int(m.group(1)) if m else 0 return applied, output.strip()[-300:] except subprocess.TimeoutExpired: return 0, "TIMEOUT beim Ausführen von citation-fixer." except Exception as e: return 0, f"FEHLER: {e}" def apply_generate_folder_indexes(brain: Path, step: dict) -> tuple[int, str]: """Einfache INDEX.md für jeden fehlenden Folder erzeugen.""" fixed = 0 notes = [] for folder_name in step.get("folders", []): folder = brain / folder_name if not folder.exists(): continue index_path = folder / "INDEX.md" if index_path.exists(): continue # Alle Notes im Folder auflisten md_files = sorted(folder.glob("*.md")) entries = "\n".join( f"- [[{p.stem}]]" for p in md_files if p.name != "INDEX.md" ) content = f"""--- title: INDEX — {folder_name} type: inbox tags: [index, moc, auto-generated] date: {datetime.now().strftime('%Y-%m-%d')} status: aktiv --- # {folder_name} — Inhaltsverzeichnis > Auto-generiert von aria-brain-remediate.py (KAR-624). Bitte pflegen. ## Notizen {entries if entries else "_(leer)_"} """ index_path.write_text(content, encoding="utf-8") fixed += 1 notes.append(folder_name) msg = f"INDEX.md erstellt für: {', '.join(notes)}" if notes else "Keine Änderungen nötig." return fixed, msg def apply_git_commit(brain: Path, step: dict) -> tuple[int, str]: """Lokaler git commit (kein push).""" try: r1 = subprocess.run( ["git", "add", "-A"], capture_output=True, text=True, cwd=brain, timeout=30, ) r2 = subprocess.run( ["git", "commit", "-m", "brain-remediate: auto-fix (KAR-624)"], capture_output=True, text=True, cwd=brain, timeout=30, ) if r2.returncode == 0: return 1, r2.stdout.strip()[:200] else: return 0, r2.stderr.strip()[:200] or r2.stdout.strip()[:200] except Exception as e: return 0, f"FEHLER: {e}" APPLY_METHODS: dict[str, Any] = { "fix_identity_files": apply_fix_identity_files, "fix_dangling_wikilinks": apply_fix_dangling_wikilinks, "generate_folder_indexes": apply_generate_folder_indexes, "git_commit_fixes": apply_git_commit, } # --------------------------------------------------------------------------- # Haupt-Orchestrierung # --------------------------------------------------------------------------- def run( dry_run: bool = True, apply: bool = False, target_score: float = 90.0, max_usd: float = 5.0, json_output: bool = False, ) -> dict: """Führt den Remediation-Loop durch. Gibt das vollständige Ergebnis-Dict zurück.""" # --- Daten sammeln --- identity = check_identity_files(BRAIN_ROOT) dangling = check_dangling_wikilinks(BRAIN_ROOT) orphans = check_orphan_notes(BRAIN_ROOT) indexes = check_missing_indexes(BRAIN_ROOT) git = check_git_status(BRAIN_ROOT) stale = check_stale_notes(BRAIN_ROOT) # --- Score --- score, penalties = compute_score(identity, dangling, orphans, indexes, git, stale) max_reachable = compute_max_reachable(score, penalties, identity, dangling) # --- Plan --- plan = build_plan(score, identity, dangling, orphans, indexes, git, stale, penalties) # --- Ergebnis-Struktur --- result: dict[str, Any] = { "score": score, "max_reachable_score": max_reachable, "target_score": target_score, "penalties": penalties, "plan": [], "spent_usd": 0.0, "apply_mode": apply and not dry_run, "stopped_reason": None, "raw": { "identity": identity, "dangling_summary": { "total_links": dangling["total_links"], "dirty_count": dangling["dirty_count"], "fixable_count": dangling["fixable_count"], }, "orphans": { "orphan_count": orphans["orphan_count"], "hash_orphan_count": orphans["hash_orphan_count"], }, "missing_indexes": indexes["missing"], "git": {k: v for k, v in git.items() if k != "uncommitted_files"}, "stale": {"stale_count": stale["stale_count"]}, }, } spent_usd = 0.0 current_score = score for step in plan: step_result = dict(step) step_result["applied"] = False step_result["apply_output"] = None est = step.get("est_usd", 0.0) if apply and not dry_run: # Cost-Cap-Check if spent_usd + est > max_usd: step_result["skipped_reason"] = f"cost_cap: {spent_usd:.2f}+{est:.2f} > {max_usd:.2f}" result["plan"].append(step_result) result["stopped_reason"] = "cost_cap" continue if current_score >= target_score: step_result["skipped_reason"] = f"target_reached: score={current_score:.1f} >= {target_score}" result["plan"].append(step_result) result["stopped_reason"] = "target_reached" continue if step.get("needs_human") and not step.get("auto_applicable"): step_result["skipped_reason"] = "needs_human" result["plan"].append(step_result) continue if not step.get("auto_applicable", False): step_result["skipped_reason"] = "not_auto_applicable" result["plan"].append(step_result) continue # Fix anwenden method_fn = APPLY_METHODS.get(step["step"]) if method_fn: fixed_count, output_msg = method_fn(BRAIN_ROOT, step) step_result["applied"] = True step_result["fixed_count"] = fixed_count step_result["apply_output"] = output_msg spent_usd += est # Re-check Score nach Fix i2 = check_identity_files(BRAIN_ROOT) d2 = check_dangling_wikilinks(BRAIN_ROOT) o2 = check_orphan_notes(BRAIN_ROOT) idx2 = check_missing_indexes(BRAIN_ROOT) g2 = check_git_status(BRAIN_ROOT) s2 = check_stale_notes(BRAIN_ROOT) new_score, new_penalties = compute_score(i2, d2, o2, idx2, g2, s2) step_result["score_after"] = new_score step_result["score_delta"] = round(new_score - current_score, 1) current_score = new_score penalties = new_penalties else: step_result["skipped_reason"] = f"kein apply-handler für {step['step']}" else: step_result["skipped_reason"] = "dry_run" result["plan"].append(step_result) result["final_score"] = current_score result["spent_usd"] = spent_usd return result # --------------------------------------------------------------------------- # Output-Formatierung # --------------------------------------------------------------------------- def print_human_report(result: dict) -> None: score = result["score"] max_r = result["max_reachable_score"] target = result["target_score"] penalties = result["penalties"] mode = "APPLY" if result["apply_mode"] else "DRY-RUN" print("=" * 60) print(f" ARIA BRAIN REMEDIATION REPORT [{mode}]") print("=" * 60) print() print(f" Score: {score:>6.1f} / 100") print(f" Max erreichbar: {max_r:>6.1f} / 100 (nach allen Auto-Fixes)") print(f" Ziel-Score: {target:>6.1f}") print() if penalties: print(" Abzüge:") for cat, val in sorted(penalties.items(), key=lambda x: -x[1]): print(f" -{val:>5.2f} {cat}") print() raw = result.get("raw", {}) d = raw.get("dangling_summary", {}) o = raw.get("orphans", {}) g = raw.get("git", {}) idx = raw.get("missing_indexes", []) print(" Rohdaten:") print(f" Dangling Wikilinks: {d.get('dirty_count', 0):>4} (davon {d.get('fixable_count', 0)} auto-fixbar)") print(f" Orphan-Notizen: {o.get('orphan_count', 0):>4} ({o.get('hash_orphan_count', 0)} Hash-Stubs ignoriert)") print(f" Fehlende INDEXes: {len(idx):>4} Folders") print(f" Uncommitted: {g.get('uncommitted_count', 0):>4} Files") print(f" Git-Alter: {g.get('age_days', 0):>4} Tage") print() print(" Remediation-Plan (dependency-ordered):") print() for i, step in enumerate(result["plan"], 1): status = "" if result["apply_mode"] and step.get("applied"): delta = step.get("score_delta", 0) status = f" ✓ APPLIED (+{delta:.1f} Score)" elif step.get("skipped_reason") == "dry_run": status = " [dry-run]" elif step.get("skipped_reason") == "needs_human": status = " ⚠ NEEDS HUMAN" elif step.get("skipped_reason") == "target_reached": status = " ✓ TARGET REACHED — skip" elif step.get("skipped_reason", "").startswith("cost_cap"): status = f" ✗ COST CAP — {step['skipped_reason']}" elif step.get("skipped_reason"): status = f" – skipped ({step['skipped_reason']})" marker = "✓" if step.get("auto_applicable") and not step.get("needs_human") else "⚠" print(f" {i}. [{marker}] {step['step']}{status}") print(f" {step['description']}") print(f" Würde fixen: {step['would_fix']} | Score+: {step.get('penalty_removed', 0):.2f} | " f"Kosten: ${step.get('est_usd', 0):.2f}") if step.get("needs_human"): print(f" ⚠ NEEDS HUMAN: {step.get('human_note', '')}") if step.get("apply_output") and result["apply_mode"]: short = step["apply_output"][:200].replace("\n", " | ") print(f" Output: {short}") print() print("=" * 60) if result["apply_mode"]: print(f" Endscore: {result['final_score']:.1f} | Ausgegeben: ${result['spent_usd']:.4f}") else: print(" Keine Änderungen vorgenommen (dry-run).") if result.get("stopped_reason"): print(f" Gestoppt wegen: {result['stopped_reason']}") print("=" * 60) # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser( description="Aria Brain Health Score + Remediation Loop (KAR-624)", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) ap.add_argument( "--dry-run", dest="dry_run", action="store_true", default=True, help="Nur Analyse, keine Änderungen (Standard)", ) ap.add_argument( "--apply", dest="apply", action="store_true", default=False, help="Remediations anwenden (überschreibt --dry-run)", ) ap.add_argument( "--target-score", dest="target_score", type=float, default=90.0, help="Ziel-Score (Standard: 90). Stop wenn erreicht.", ) ap.add_argument( "--max-usd", dest="max_usd", type=float, default=5.0, help="Maximale Ausgaben in USD (Standard: 5.0). Lokale Schritte = $0.", ) ap.add_argument( "--json", dest="json_output", action="store_true", default=False, help="JSON-Output anstatt lesbarem Report.", ) args = ap.parse_args(argv) dry_run = not args.apply result = run( dry_run=dry_run, apply=args.apply, target_score=args.target_score, max_usd=args.max_usd, json_output=args.json_output, ) if args.json_output: # Plan-Einträge bereinigen (Path-Objekte etc.) import copy out = copy.deepcopy(result) for step in out.get("plan", []): step.pop("files_with_drift", None) print(json.dumps(out, ensure_ascii=False, indent=2)) else: print_human_report(result) return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))