#!/usr/bin/env python3 """Plan and apply guarded self-healing actions from Jeannie status findings.""" from __future__ import annotations import argparse import json import os import subprocess import sys from dataclasses import dataclass from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] @dataclass(frozen=True) class HealRule: check: str command: str risk: str reason: str auto: bool RULES = [ HealRule("Pimox workers running", "./jeannie start-cluster", "low", "Worker VMs are expected cluster capacity and start-cluster is idempotent.", True), HealRule("Kubernetes API", "./jeannie start-cluster", "low", "API is down; start-cluster starts kubelet/containerd and worker VMs without destroying state.", True), HealRule("Kubernetes nodes Ready", "./jeannie start-cluster", "low", "NotReady worker nodes commonly recover by starting the saved cluster runtime.", True), HealRule("Gitea container", "./jeannie deploy-gitea", "medium", "Reapplies the Debian-hosted Gitea Compose service.", False), HealRule("Gitea local HTTP", "./jeannie deploy-gitea", "medium", "Local Gitea HTTP is down; redeploying Compose is usually safe but still mutates Git service runtime.", False), HealRule("Pi-hole DNS", "./jeannie rpi-services", "medium", "Reapplies RPi DNS services; safe for the lab but can briefly disrupt DNS.", False), HealRule("Uptime Kuma HTTP", "./jeannie rpi-services", "medium", "Reapplies RPi service Compose stack.", False), HealRule("RPi Docker root state", "./jeannie rpi-services", "medium", "May repair service runtime but should be reviewed if storage is failing.", False), HealRule("Traefik deployment", "./jeannie promote validate", "diagnostic", "Validate gates first; deployment rollout failures are usually downstream of cluster/node health.", False), HealRule("Website deployment", "./jeannie promote validate", "diagnostic", "Validate gates first; deployment rollout failures are usually downstream of cluster/node health.", False), HealRule("Traefik LoadBalancer HTTP", "./jeannie doctor-edge", "diagnostic", "Edge/LB failures need diagnosis after cluster nodes are healthy.", False), HealRule("Website public URL", "./jeannie doctor-edge", "diagnostic", "Public URL failures need edge and cluster diagnosis after local services recover.", False), HealRule("Gitea public route", "./jeannie doctor-gitea", "diagnostic", "Public Gitea route needs Gitea and edge diagnosis.", False), HealRule("No problem pods", "./jeannie explain status", "diagnostic", "Pod states need root-cause context; avoid blind deletes.", False), HealRule("Recent deployments healthy", "./jeannie explain status", "diagnostic", "Deployment drift can be a symptom of node health, image pulls, or scheduling.", False), HealRule("Pod restart pressure", "./jeannie explain status", "diagnostic", "Restart pressure needs workload-specific diagnosis before mutation.", False), HealRule("Traefik 5xx/404 signals", "./jeannie doctor-edge", "diagnostic", "Use edge logs and route checks before changing config.", False), ] def load_status(path: Path | None) -> dict[str, object]: if path: return json.loads(path.read_text(encoding="utf-8")) process = subprocess.run( [str(REPO_ROOT / "jeannie"), "status", "--json"], cwd=REPO_ROOT, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, ) if not process.stdout.strip(): print(process.stderr.strip() or "status produced no JSON", file=sys.stderr) raise SystemExit(1) return json.loads(process.stdout) def status_rows(status: dict[str, object]) -> list[dict[str, str]]: rows = status.get("rows", []) if not isinstance(rows, list): return [] return [row for row in rows if isinstance(row, dict) and row.get("status") in {"fail", "warn"}] def rule_for(row: dict[str, str]) -> HealRule | None: check = str(row.get("check", "")) for rule in RULES: if rule.check == check: return rule return None def build_plan(status: dict[str, object]) -> list[dict[str, object]]: planned: list[dict[str, object]] = [] seen_commands: set[str] = set() for row in status_rows(status): rule = rule_for(row) if rule is None: planned.append( { "check": row.get("check", ""), "area": row.get("area", ""), "status": row.get("status", ""), "command": "./jeannie explain status", "risk": "diagnostic", "auto": False, "reason": "No specific heal rule exists yet.", "summary": row.get("summary", ""), } ) continue if rule.command in seen_commands and rule.auto: continue seen_commands.add(rule.command) planned.append( { "check": row.get("check", ""), "area": row.get("area", ""), "status": row.get("status", ""), "command": rule.command, "risk": rule.risk, "auto": rule.auto, "reason": rule.reason, "summary": row.get("summary", ""), } ) return planned def print_plan(plan: list[dict[str, object]], ai: bool) -> None: auto = [item for item in plan if item["auto"]] review = [item for item in plan if not item["auto"]] print("Jeannie Heal Plan") print("=================") print("mode: guarded") print(f"auto_actions={len(auto)} review_actions={len(review)}") print() if auto: print("Auto-eligible:") for item in auto: print(f" - {item['command']}") print(f" fixes: {item['area']} / {item['check']}") print(f" why: {item['reason']}") else: print("Auto-eligible: none") print() if review: print("Needs review or diagnostic only:") for item in review: print(f" - {item['command']}") print(f" finding: {item['area']} / {item['check']} ({item['status']})") print(f" why: {item['reason']}") if ai: print_ai_context(plan) def print_ai_context(plan: list[dict[str, object]]) -> None: query = " ".join(str(item["check"]) for item in plan[:5]) if not query: return process = subprocess.run( [str(REPO_ROOT / "scripts" / "query-homelab-ai-index"), "--citations", "--limit", "3", query], cwd=REPO_ROOT, text=True, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, check=False, ) if process.returncode == 0 and process.stdout.strip(): print() print("AI/RAG Context") print("--------------") print(process.stdout.strip()) def apply_plan(plan: list[dict[str, object]], yes: bool) -> int: commands: list[str] = [] for item in plan: if item["auto"]: command = str(item["command"]) if command not in commands: commands.append(command) if not commands: print("No auto-eligible heal actions.") return 0 print("Jeannie Heal Apply") print("==================") for command in commands: sandbox = subprocess.run( [str(REPO_ROOT / "scripts" / "agent-sandbox"), "check", command, "--json"], cwd=REPO_ROOT, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=False, ) try: decision = json.loads(sandbox.stdout).get("decision") except json.JSONDecodeError: decision = "approval" if decision != "allow": print(f"skip: {command} blocked by agent sandbox decision={decision}") continue if not yes: print(f"would run: {command}") continue print(f"run: {command}") process = subprocess.run(command.split(), cwd=REPO_ROOT, text=True, check=False) if process.returncode != 0: return process.returncode if not yes: print() print("Dry run only. Re-run with: ./jeannie heal apply --yes") return 0 def main() -> int: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) plan_parser = subparsers.add_parser("plan") plan_parser.add_argument("--status-json", type=Path) plan_parser.add_argument("--ai", action="store_true") apply_parser = subparsers.add_parser("apply") apply_parser.add_argument("--status-json", type=Path) apply_parser.add_argument("--yes", action="store_true") apply_parser.add_argument("--ai", action="store_true") args = parser.parse_args() status = load_status(getattr(args, "status_json", None)) plan = build_plan(status) if args.command == "plan": print_plan(plan, args.ai) return 0 if args.command == "apply": print_plan(plan, args.ai) print() return apply_plan(plan, args.yes) return 2 if __name__ == "__main__": raise SystemExit(main())