spb/ka-guardian
Public
1# ============================================2# Projet : KA Guardian3# Fichier : orchestrator/main.py4# Rôle : Cerveau d'un agent gardien (ka2/ka4/ka6) — surveille la santé5# des connecteurs via api-ka, ouvre des incidents, dépêche des6# missions claude aux runners des nœuds, surveille la guérison,7# rollback automatique si ça empire. Sert aussi le dashboard live.8# Author : Simon-Pierre Boucher9# Date : 2026-08-2310# ============================================11"""Instance unique par agent: AGENT=ka2|ka4|ka6 (env), port depuis topology.json."""12from __future__ import annotations1314import asyncio15import json16import os17import pathlib18import sqlite319import time20import uuid21from typing import Any2223import httpx24from fastapi import FastAPI, Header, HTTPException, Request25from fastapi.responses import FileResponse, StreamingResponse26from fastapi.staticfiles import StaticFiles2728BASE = pathlib.Path(__file__).resolve().parent29ROOT = BASE.parent30HOME = pathlib.Path.home()3132AGENT = os.environ.get("AGENT", "ka2")33TOPO = json.loads((ROOT / "topology.json").read_text())34ME = TOPO["agents"][AGENT]35POLICY = TOPO["policy"]36SERVICES: dict[str, Any] = TOPO["services"]37NODES: dict[str, Any] = TOPO["nodes"]38MY_SERVICES: set[str] = set(ME["services"])39RUNNER_PORT = TOPO["runner_port"]4041# ka6 (ou autre) ramasse les services non assignés qui apparaîtraient dans api-ka.42DEFAULT_AGENT = ME.get("default_for_unknown_services", False)43ASSIGNED = {s for a in TOPO["agents"].values() for s in a["services"]}444546def load_env_file(path: pathlib.Path) -> dict[str, str]:47 out: dict[str, str] = {}48 if path.exists():49 for line in path.read_text().splitlines():50 line = line.strip()51 if line and not line.startswith("#") and "=" in line:52 k, v = line.split("=", 1)53 out[k.strip()] = v.strip().strip('"').strip("'")54 return out555657TOKEN = load_env_file(HOME / ".ka-guardian.env").get("KA_GUARDIAN_TOKEN", "")585960SPOOL = HOME / "ka-guardian-spool"616263async def lan_post(ip: str, port: int, path: str, payload: dict[str, Any], timeout: int = 30) -> tuple[int, str]:64 """POST JSON vers un autre nœud via le courrier zsh (deploy/courier.sh).6566 macOS 26 « Local Network Privacy » refuse le trafic LAN dès que python67 (homebrew) est dans la chaîne de processus — même ses enfants ssh/curl.68 On dépose donc un job dans un spool disque ; un service launchd 100 % zsh69 l'expédie (ssh → curl localhost du nœud cible) et dépose la réponse.70 """71 for d in ("outbox", "done", "tmp"):72 (SPOOL / d).mkdir(parents=True, exist_ok=True)73 jid = uuid.uuid4().hex74 tmp = SPOOL / "tmp" / f"{jid}.job"75 tmp.write_text(f"{ip} {port} {path} {timeout}\n" + jdump(payload))76 tmp.rename(SPOOL / "outbox" / f"{jid}.job")77 resp = SPOOL / "done" / f"{jid}.resp"78 deadline = time.time() + timeout + 2579 while time.time() < deadline:80 if resp.exists():81 txt = resp.read_text()82 resp.unlink(missing_ok=True)83 body, _, code = txt.strip().rpartition("\n")84 if code.startswith("courier_ssh_rc"):85 raise ConnectionError(f"{code} vers {ip}:{port}{path}")86 return int(code or "0"), body87 await asyncio.sleep(0.3)88 raise TimeoutError(f"courrier sans réponse pour {ip}:{port}{path}")89DATA = ROOT / "data"90DATA.mkdir(exist_ok=True)91DB_PATH = DATA / f"{AGENT}.db"9293# ---------------------------------------------------------------- SQLite ---9495def db() -> sqlite3.Connection:96 conn = sqlite3.connect(DB_PATH)97 conn.row_factory = sqlite3.Row98 conn.execute("PRAGMA journal_mode=WAL")99 return conn100101102def init_db() -> None:103 with db() as c:104 c.executescript("""105 CREATE TABLE IF NOT EXISTS incidents(106 id TEXT PRIMARY KEY, service TEXT, source TEXT, status_detected TEXT,107 state TEXT, attempts INTEGER DEFAULT 0, created REAL, updated REAL,108 resolved REAL, detail TEXT);109 CREATE TABLE IF NOT EXISTS missions(110 id TEXT PRIMARY KEY, incident_id TEXT, service TEXT, source TEXT,111 node TEXT, state TEXT, base_commit TEXT, commits TEXT, verdict TEXT,112 cost_usd REAL, num_turns INTEGER, health TEXT, started REAL, ended REAL);113 CREATE TABLE IF NOT EXISTS events(114 id INTEGER PRIMARY KEY AUTOINCREMENT, mission_id TEXT, ts REAL,115 type TEXT, data TEXT);116 CREATE TABLE IF NOT EXISTS snapshots(117 ts REAL PRIMARY KEY, mine TEXT, ecosystem TEXT);118 CREATE INDEX IF NOT EXISTS ev_mission ON events(mission_id);119 """)120121122def now() -> float:123 return time.time()124125126def jdump(x: Any) -> str:127 return json.dumps(x, ensure_ascii=False)128129130# ------------------------------------------------------------------- SSE ---131132class Hub:133 def __init__(self) -> None:134 self.clients: set[asyncio.Queue] = set()135136 async def publish(self, msg: dict[str, Any]) -> None:137 for q in list(self.clients):138 if q.qsize() < 500:139 q.put_nowait(msg)140141 def publish_sync(self, msg: dict[str, Any]) -> None:142 if LOOP:143 asyncio.run_coroutine_threadsafe(self.publish(msg), LOOP)144145146hub = Hub()147LOOP: asyncio.AbstractEventLoop | None = None148LATEST: dict[str, Any] = {"mine": {}, "ecosystem": {}, "polled": 0, "paused": False}149150# ------------------------------------------------------------- missions ----151152def active_incident(c: sqlite3.Connection, service: str, source: str) -> sqlite3.Row | None:153 return c.execute(154 "SELECT * FROM incidents WHERE service=? AND source=? AND state IN "155 "('open','dispatched','fixing','watching','cooldown') ORDER BY created DESC LIMIT 1",156 (service, source)).fetchone()157158159def set_incident(c: sqlite3.Connection, iid: str, **kw: Any) -> None:160 kw["updated"] = now()161 keys = ",".join(f"{k}=?" for k in kw)162 c.execute(f"UPDATE incidents SET {keys} WHERE id=?", (*kw.values(), iid))163164165def incident_event(iid: str, service: str, source: str, state: str, note: str = "") -> None:166 hub.publish_sync({"kind": "incident", "incident_id": iid, "service": service,167 "source": source, "state": state, "note": note, "ts": now()})168169170def service_situation(service: str) -> str:171 """Portrait santé de l'app entière — permet à l'agent de distinguer une172 panne isolée d'un problème systémique (app entière, réseau, quota)."""173 block = LATEST["mine"].get(service) or {}174 summary = block.get("summary") or {}175 app_row = block.get("app") or {}176 sick = [f"{c['source']} ({c['status']})" for c in block.get("connectors", [])177 if c.get("status") not in ("ok", None)][:15]178 lines = [f"- app entière: {app_row.get('status', '?')} — {summary.get('ok', '?')} ok, "179 f"{summary.get('degraded', 0)} dégradés, {summary.get('broken', 0)} cassés, {summary.get('stale', 0)} endormis"]180 if sick:181 lines.append(f"- autres connecteurs non-ok de cette app: {', '.join(sick)}")182 lines.append("- Si BEAUCOUP de connecteurs sont touchés en même temps, le problème est probablement "183 "systémique (scheduler de l'app, réseau, quota API): diagnostique au niveau app, pas source par source.")184 else:185 lines.append("- Tous les autres connecteurs de l'app sont sains: la panne est isolée à cette source.")186 return "\n".join(lines)187188189def mission_history(service: str, source: str) -> str:190 """Résumé des missions précédentes sur ce connecteur (éviter de refaire ce qui a échoué)."""191 with db() as c:192 rows = c.execute(193 "SELECT m.started, m.verdict, m.commits FROM missions m WHERE m.service=? AND m.source=? "194 "AND m.state!='running' ORDER BY m.started DESC LIMIT 3", (service, source)).fetchall()195 if not rows:196 return "- aucune: c'est la première intervention sur ce connecteur."197 out = []198 for r in rows:199 v = json.loads(r["verdict"] or "{}")200 out.append(f"- {time.strftime('%Y-%m-%d %H:%M', time.localtime(r['started']))}: "201 f"verdict={v.get('verdict', '?')} — {str(v.get('diagnostic', ''))[:200]} "202 f"(actions: {str(v.get('actions', ''))[:150]})")203 out.append("NE RÉPÈTE PAS une approche qui a déjà échoué ci-dessus: change d'angle.")204 return "\n".join(out)205206207def build_prompt(service: str, source: str, health: dict[str, Any]) -> str:208 svc = SERVICES[service]209 sync_proc = next((p for p in svc["pm2"] if "sync" in p or "etl" in p), svc["pm2"][-1])210 node_alias = NODES[svc["node"]]["alias"]211 return f"""Tu es {AGENT}, agent gardien autonome des connecteurs du Groupe KA. Mission: réparer le connecteur « {source} » de l'app {svc['app']} (service api-ka: {service}, site {svc.get('site', '')}).212213== CONTEXTE SANTÉ DU CONNECTEUR (supervision api-ka, scan aux 2 h) ==214- statut détecté: {health.get('status')} | échecs consécutifs: {health.get('consecutive_failures')}215- dernier succès: {health.get('last_success')} | volume au dernier sync: {health.get('found_last')} (médiane historique: {health.get('median_found')})216- message: {health.get('message')}217Rappel des statuts: broken = ≥3 syncs consécutifs en échec ou à 0 résultat; stale = aucun succès depuis > 2× la cadence attendue; degraded = volume < 50 % de la médiane.218219== SITUATION DE L'APP ==220{service_situation(service)}221222== INTERVENTIONS PRÉCÉDENTES SUR CE CONNECTEUR ==223{mission_history(service, source)}224225== TON ENVIRONNEMENT ==226Tu es sur le nœud {node_alias}, ton répertoire courant est le repo de l'app: {svc['dir']} (source de vérité, développement remote-first).227L'app tourne via pm2 ({', '.join(svc['pm2'])}); le process de synchronisation des connecteurs est {sync_proc}; site web local sur le port {svc['web_port']}.228Le venv Python et/ou node_modules du repo sont déjà installés — utilise-les, jamais d'installation globale.229230== DÉMARCHE IMPOSÉE (dans l'ordre) ==2311. IMPRÈGNE-TOI DU REPO: lis le CLAUDE.md et/ou README du repo s'ils existent, et surtout la doc du connecteur si elle existe: cherche `docs/connecteurs/` (fiches générées par app, souvent une par source: URL de la source, stratégie de fetch, format, pièges connus). `grep -ri "{source}"` pour localiser le code du connecteur, sa config et son éventuelle entrée de registre.2322. LIS LES LOGS: dossier logs/ du repo + `pm2 logs {sync_proc} --nostream --lines 300` — cherche les traces d'erreur de « {source} » (traceback, code HTTP, timeout).2333. REPRODUIS: exécute le connecteur ou son fetch directement (la plupart des apps ont un runner par source — la doc/le code du scheduler te montrera comment; sinon appelle la fonction de fetch dans un petit script via le venv du repo). Constate l'erreur réelle.2344. DIAGNOSTIQUE la cause racine: HTML/sélecteurs changés? endpoint JSON déplacé? 403/429 anti-bot? pagination cassée? redirection? certificat/TLS (gotcha connu: certains sites cassent avec requests → utiliser curl via subprocess)? sitemap figé?2355. CORRIGE de façon MINIMALE et ciblée, en respectant les conventions et l'architecture du repo (mêmes patterns que les connecteurs voisins). Si le site bloque, la pile d'escalade du Groupe KA est disponible, clés dans ~/.claude/.env: requêtes directes → curl → Scrapfly (SCRAPFLY_API_KEY, render_js/asp) → Bright Data Web Unlocker → proxies résidentiels Oxylabs (pr.oxylabs.io:7777, -cc-CA) → Serper/Tavily pour retrouver une source déplacée. Regarde comment le repo utilise déjà ces services et fais pareil.2366. RE-TESTE réellement: le connecteur doit rapporter un volume plausible (ordre de grandeur de la médiane {health.get('median_found')}). INTERDIT d'inventer, stubber ou câbler des données en dur. Un test qui retourne 0 ou 3 items quand la médiane est {health.get('median_found')} N'EST PAS une réparation.2377. REDÉMARRE uniquement le process concerné: `pm2 restart {sync_proc}`. Puis vérifie que le site répond: `curl -s -o /dev/null -w '%{{http_code}}' http://localhost:{svc['web_port']}/` (attendu: 200/3xx).2388. COMMITTE: `git add -A && git commit -m "[{AGENT}] fix connecteur {source}: <résumé court>"`. NE PUSH JAMAIS. Si tu as touché plusieurs fichiers pour des raisons distinctes, fais des commits séparés et clairs.239240== INTERDITS ABSOLUS ==241Toucher aux autres connecteurs ou aux autres apps du nœud; modifier la config pm2/ngrok/launchd ou le schéma de la base; installer des paquets globaux; supprimer des données; git push; toucher à ~/.ssh, aux clés API ou aux fichiers hors du repo. Ne désactive JAMAIS un connecteur pour « réparer » sa santé. Si la source est définitivement morte (site fermé, domaine à vendre, 404 permanent confirmé), ne force rien: documente et conclus en verdict site_source_mort.242243== FIN DE MISSION ==244Termine ta TOUTE DERNIÈRE réponse par un bloc JSON exactement de cette forme:245{{"verdict": "repare|echec|site_source_mort|rien_a_faire", "diagnostic": "cause racine en 1-2 phrases", "actions": "ce que tu as fait", "test": "résultat mesuré du test final (volume obtenu)", "fichiers": ["fichiers modifiés"]}}246Sois honnête: si ta réparation n'est pas prouvée par un test réel, le verdict est echec — un faux « repare » sera détecté par la supervision et rollback automatiquement."""247248249COMMON_RULES = """== RÈGLES COMMUNES (non négociables) ==250- Session 100 % AUTONOME: tu ne poses AUCUNE question, tu ne demandes AUCUNE validation, tu travailles jusqu'au bout. S'il faut trancher, tranche selon les conventions du repo et le bon sens, et documente ton choix dans le commit.251- Respecte l'architecture et les patterns du repo (regarde comment les connecteurs voisins sont faits AVANT d'écrire).252- Utilise le venv/node_modules du repo; jamais d'installation globale.253- INTERDIT d'inventer, stubber ou câbler des données en dur. Chaque donnée vient d'un vrai fetch.254- INTERDITS: toucher aux autres apps du nœud, config pm2/ngrok/launchd, schéma de la base (sauf migration prévue par le repo), git push, ~/.ssh, clés API, suppression de données.255- À la fin: `pm2 restart <process de sync>`, vérifie que le site répond (curl localhost), puis committe en messages clairs préfixés [%AGENT%]. NE PUSH JAMAIS."""256257258def effort_stack(service: str) -> str:259 svc = SERVICES[service]260 return f"""== BOÎTE À OUTILS (clés dans ~/.claude/.env) ==261- DÉCOUVERTE: Serper (SERPER_API_KEY, POST https://google.serper.dev/search, gl=ca hl=fr) pour trouver sources/sitemaps/endpoints; Tavily (TAVILY_API_KEY) en complément.262- FETCH — escalade dans cet ordre: requêtes directes (requests/fetch du repo) → curl via subprocess (contourne les gotchas TLS de requests) → Scrapfly (SCRAPFLY_API_KEY, render_js/asp, backend=auto) → Bright Data Web Unlocker → proxies résidentiels Oxylabs (pr.oxylabs.io:7777, user avec -cc-CA, sessions collantes).263- APIFY (APIFY_TOKEN): en dernier recours pour les plateformes dures, acteurs maison du compte (proxy résidentiel intégré).264- Regarde d'abord comment le repo de {svc['app']} utilise déjà ces services et fais pareil."""265266267def build_effort_prompt(kind: str, service: str, source: str, note: str) -> str:268 svc = SERVICES[service]269 sync_proc = next((p for p in svc["pm2"] if "sync" in p or "etl" in p), svc["pm2"][-1])270 node_alias = NODES[svc["node"]]["alias"]271 rules = COMMON_RULES.replace("%AGENT%", AGENT)272 env = f"""== TON ENVIRONNEMENT ==273Tu es sur le nœud {node_alias}, répertoire courant = repo de l'app: {svc['dir']} (remote-first, source de vérité).274L'app tourne via pm2 ({', '.join(svc['pm2'])}); process de sync: {sync_proc}; site local port {svc['web_port']}.275Commence par lire le CLAUDE.md / README du repo et docs/connecteurs/ s'ils existent."""276 if kind == "effort_degrade":277 block = LATEST["mine"].get(service) or {}278 degraded = [c for c in block.get("connectors", []) if c.get("status") == "degraded"][:25]279 deg_lines = "\n".join(280 f"- {c['source']}: dernier volume {c.get('found_last')} vs médiane {round(c.get('median_found') or 0)} "281 f"(dernier succès: {c.get('last_success')}, message: {c.get('message')})"282 for c in degraded) or "- (aucun connecteur dégradé au dernier scan — re-vérifie l'état réel dans l'app)"283 return f"""Tu es {AGENT}, agent gardien autonome du Groupe KA. EFFORT COMMANDÉ: INSPECTER LES CONNECTEURS DÉGRADÉS de l'app {svc['app']} (service {service}, site {svc.get('site', '')}).284{"Consigne de l'opérateur: " + note if note else ""}285286Un connecteur « dégradé » livre encore des données, mais moins de 50 % de sa médiane historique — souvent le signe d'une pagination cassée, d'un filtre qui se resserre, d'une section du site source disparue ou d'un blocage partiel.287288== CONNECTEURS DÉGRADÉS AU DERNIER SCAN api-ka ==289{deg_lines}290291{env}292293== DÉMARCHE ==2941. TRIE: pour chaque connecteur dégradé, regarde vite (logs + un fetch de contrôle) si la baisse est (a) réelle et réparable, (b) légitime (le site source a vraiment moins d'items — saison, inventaire réduit), ou (c) un blocage.2952. PRIORISE les cas (a) au plus fort potentiel de volume récupéré, et répare-les UN PAR UN: cause racine, correctif minimal, test réel avec volume mesuré avant/après. Commit séparé par connecteur réparé ([{AGENT}] fix connecteur <source>: …).2963. Pour les cas (b), ne touche à rien: consigne-les dans ton rapport final comme « baisse légitime ».2974. Traite autant de connecteurs que ton budget de temps le permet, en gardant la qualité: mieux vaut 3 vraies réparations prouvées que 10 rustines.2985. À la fin: `pm2 restart {sync_proc}`, vérifie le site (port {svc['web_port']}).299300{effort_stack(service)}301302{rules}303304== FIN DE MISSION ==305Termine ta TOUTE DERNIÈRE réponse par un bloc JSON:306{{"verdict": "livre|echec", "diagnostic": "portrait global des dégradés", "actions": "connecteurs réparés (avec volumes avant/après) / baisses légitimes / blocages", "test": "mesures", "fichiers": ["fichiers modifiés"]}}"""307 if kind == "effort_new":308 return f"""Tu es {AGENT}, agent gardien autonome du Groupe KA. EFFORT COMMANDÉ: ajouter UN NOUVEAU CONNECTEUR de qualité production à l'app {svc['app']} (service {service}, site {svc.get('site', '')}).309{"Consigne de l'opérateur: " + note if note else "Aucune consigne particulière: choisis la source la plus utile."}310311{env}312313== DÉMARCHE ==3141. CARTOGRAPHIE L'EXISTANT: liste les connecteurs actuels de l'app (registre/scheduler + docs/connecteurs) pour comprendre ce qui est déjà couvert et le pattern d'implémentation exact d'un connecteur (structure, signature, enregistrement, cache, dédup).3152. DÉCOUVRE avec Serper: cherche des sources québécoises pertinentes pour {svc['app']} NON couvertes (sitemaps, pages listant des items, endpoints JSON internes découverts via les pages). Évalue 3-5 candidats: volume estimé, faisabilité du fetch, qualité des données, stabilité. CHOISIS le meilleur.3163. IMPLÉMENTE le connecteur en suivant le pattern exact des connecteurs voisins: fetch (avec escalade si bloqué), parsing complet (tous les champs que l'app affiche, fiche détail incluse si le pattern le fait), dédup, enregistrement au registre/scheduler.3174. TESTE réellement: exécute-le au complet, il doit rapporter un volume substantiel et des items complets et corrects (vérifie 3-4 items à la main contre le site source).3185. DOCUMENTE: ajoute la fiche docs/connecteurs/<source>.md si le repo suit cette convention (URL, stratégie, pièges).3196. Redémarre {sync_proc}, vérifie le site, committe.320321{effort_stack(service)}322323{rules}324325== FIN DE MISSION ==326Termine ta TOUTE DERNIÈRE réponse par un bloc JSON:327{{"verdict": "livre|echec", "diagnostic": "source choisie et pourquoi", "actions": "ce qui a été construit", "test": "volume obtenu + vérifications", "fichiers": ["fichiers créés/modifiés"]}}"""328 return f"""Tu es {AGENT}, agent gardien autonome du Groupe KA. EFFORT COMMANDÉ: ENRICHIR le connecteur existant « {source} » de l'app {svc['app']} (service {service}, site {svc.get('site', '')}).329{"Consigne de l'opérateur: " + note if note else "Aucune consigne particulière: maximise la valeur (couverture, champs, robustesse)."}330331{env}332333== DÉMARCHE ==3341. ÉTUDIE le connecteur « {source} »: code, fiche docs/connecteurs, logs récents, volume actuel vs médiane, champs remplis vs champs que l'app sait afficher.3352. IDENTIFIE les enrichissements à plus forte valeur, par exemple: pagination complète (couvre TOUT le site source, pas la première page), fiche détail (champs manquants: descriptions, photos, prix, coordonnées, dates), robustesse (retries, escalade anti-bot, tolérance aux changements de DOM), fraîcheur (détection de retraits), dédup plus fine.3363. IMPLÉMENTE proprement dans le pattern du repo, SANS casser le format de sortie existant (les consommateurs avals dépendent du schéma).3374. TESTE réellement: exécute le connecteur au complet, compare avant/après (volume, complétude des champs), vérifie 3-4 items à la main.3385. METS À JOUR la fiche docs/connecteurs/<source>.md si elle existe.3396. Redémarre {sync_proc}, vérifie le site, committe.340341{effort_stack(service)}342343{rules}344345== FIN DE MISSION ==346Termine ta TOUTE DERNIÈRE réponse par un bloc JSON:347{{"verdict": "livre|echec", "diagnostic": "état initial constaté", "actions": "enrichissements livrés", "test": "avant/après mesuré", "fichiers": ["fichiers modifiés"]}}"""348349350async def dispatch(incident: sqlite3.Row, health: dict[str, Any]) -> None:351 service, source, iid = incident["service"], incident["source"], incident["id"]352 svc = SERVICES[service]353 node = svc["node"]354 ip = NODES[node]["lan_ip"]355 mid = uuid.uuid4().hex[:12]356 my_ip = os.environ.get("KA_GUARDIAN_SELF_IP", "192.168.2.69")357 kind = incident["status_detected"]358 max_cost = None359 if kind in ("effort_new", "effort_enrich", "effort_degrade"):360 prompt = build_effort_prompt(kind, service, source, health.get("note", ""))361 max_turns = POLICY.get("effort_max_turns", 150)362 timeout = POLICY.get("effort_timeout_seconds", 7200)363 # Plafonds choisis par l'opérateur au moment de commander l'effort.364 if health.get("max_minutes"):365 timeout = min(timeout, int(health["max_minutes"]) * 60)366 if health.get("max_cost_usd"):367 max_cost = float(health["max_cost_usd"])368 else:369 prompt = build_prompt(service, source, health)370 max_turns, timeout = POLICY["mission_max_turns"], POLICY["mission_timeout_seconds"]371 payload = {372 "mission_id": mid, "agent": AGENT, "service": service, "source": source,373 "dir": svc["dir"], "pm2": svc["pm2"], "web_port": svc["web_port"],374 "model": ME.get("model", "sonnet"),375 "max_turns": max_turns,376 "timeout_seconds": timeout,377 "max_cost_usd": max_cost,378 "prompt": prompt,379 "callback_url": f"http://{my_ip}:{ME['port']}/api/ingest/{mid}",380 }381 try:382 code, body = await lan_post(ip, RUNNER_PORT, "/missions", payload, timeout=25)383 if code == 409:384 return # runner occupé, on retentera au prochain tick385 if code != 200:386 raise ConnectionError(f"runner {code}: {body[:200]}")387 except Exception as exc:388 print(f"[dispatch] {service}/{source} → {type(exc).__name__}: {exc}", flush=True)389 hub.publish_sync({"kind": "log", "level": "error", "msg": f"dispatch {source}: {exc}", "ts": now()})390 return391 with db() as c:392 c.execute("INSERT INTO missions(id,incident_id,service,source,node,state,started) VALUES(?,?,?,?,?,?,?)",393 (mid, iid, service, source, node, "running", now()))394 set_incident(c, iid, state="fixing", attempts=incident["attempts"] + 1)395 incident_event(iid, service, source, "fixing", f"mission {mid} dépêchée sur {NODES[node]['alias']}")396 hub.publish_sync({"kind": "mission", "mission_id": mid, "service": service,397 "source": source, "state": "running", "ts": now()})398399400async def rollback_mission(mission: sqlite3.Row, reason: str) -> dict[str, Any]:401 svc = SERVICES[mission["service"]]402 ip = NODES[mission["node"]]["lan_ip"]403 try:404 code, body = await lan_post(ip, RUNNER_PORT, "/rollback",405 {"dir": svc["dir"], "base_commit": mission["base_commit"],406 "pm2": svc["pm2"], "web_port": svc["web_port"]}, timeout=300)407 out = json.loads(body) if code == 200 else {"ok": False, "erreur": f"{code}: {body[:200]}"}408 except Exception as exc:409 out = {"ok": False, "erreur": str(exc)}410 hub.publish_sync({"kind": "rollback", "mission_id": mission["id"], "service": mission["service"],411 "source": mission["source"], "reason": reason, "result": out, "ts": now()})412 with db() as c:413 c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",414 (mission["id"], now(), "rollback", jdump({"reason": reason, "result": out})))415 return out416417418# --------------------------------------------------------------- moteur ----419420async def poll_once() -> None:421 async with httpx.AsyncClient() as cl:422 r = await cl.get(TOPO["apika_monitoring_url"], timeout=15)423 data = r.json()["data"]424 mine: dict[str, Any] = {}425 for service, block in data["services"].items():426 assigned_to_me = service in MY_SERVICES or (DEFAULT_AGENT and service not in ASSIGNED)427 if assigned_to_me:428 mine[service] = block429 LATEST.update({"mine": mine, "ecosystem": data["summary"], "polled": now()})430 with db() as c:431 summary_mine = {s: b["summary"] for s, b in mine.items()}432 c.execute("INSERT OR REPLACE INTO snapshots(ts,mine,ecosystem) VALUES(?,?,?)",433 (now(), jdump(summary_mine), jdump(data["summary"])))434 c.execute("DELETE FROM snapshots WHERE ts < ?", (now() - 30 * 86400,))435 hub.publish_sync({"kind": "snapshot", "mine": {s: b["summary"] for s, b in mine.items()},436 "ecosystem": data["summary"], "ts": now()})437438 trigger = set(POLICY["trigger_statuses"])439 with db() as c:440 for service, block in mine.items():441 if service not in SERVICES:442 continue # service inconnu de la topologie: visible au dashboard, pas d'action443 for conn in block.get("connectors", []):444 source, status = conn["source"], conn["status"]445 inc = active_incident(c, service, source)446 if status in trigger and inc is None:447 iid = uuid.uuid4().hex[:10]448 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "449 "VALUES(?,?,?,?,?,?,?,?)",450 (iid, service, source, status, "open", now(), now(), jdump(conn)))451 incident_event(iid, service, source, "open", f"détecté {status}")452 elif status in trigger and inc is not None and inc["state"] in ("open", "cooldown"):453 # Ne pas rafraîchir watching/fixing: `updated` sert de chrono454 # à la fenêtre de surveillance et au cooldown.455 c.execute("UPDATE incidents SET status_detected=?, detail=? WHERE id=?",456 (status, jdump(conn), inc["id"]))457 elif status == "ok" and inc is not None:458 if inc["state"] == "watching":459 set_incident(c, inc["id"], state="resolved", resolved=now())460 incident_event(inc["id"], service, source, "resolved", "connecteur de retour à ok — réparation confirmée")461 elif inc["state"] in ("open", "cooldown"):462 set_incident(c, inc["id"], state="self_healed", resolved=now())463 incident_event(inc["id"], service, source, "self_healed", "revenu à ok sans intervention")464465466async def tick() -> None:467 if LATEST["paused"]:468 return469 try:470 await poll_once()471 except Exception as exc:472 hub.publish_sync({"kind": "log", "level": "warn", "msg": f"poll api-ka échoué: {exc}", "ts": now()})473474 with db() as c:475 # 1. Missions zombies (runner mort / callback perdu)476 for m in c.execute("SELECT * FROM missions WHERE state='running' AND started < ?",477 (now() - POLICY.get("effort_timeout_seconds", 7200) - 900,)).fetchall():478 c.execute("UPDATE missions SET state='error', ended=? WHERE id=?", (now(), m["id"]))479 inc = c.execute("SELECT * FROM incidents WHERE id=?", (m["incident_id"],)).fetchone()480 if inc:481 set_incident(c, inc["id"], state="cooldown")482 incident_event(inc["id"], m["service"], m["source"], "cooldown", "mission sans réponse (zombie)")483484 # 2. Watching → rollback si la fenêtre est passée et toujours cassé485 for inc in c.execute("SELECT * FROM incidents WHERE state='watching'").fetchall():486 if now() - inc["updated"] < POLICY["watch_window_hours"] * 3600:487 continue488 status = current_status(inc["service"], inc["source"])489 m = c.execute("SELECT * FROM missions WHERE incident_id=? AND base_commit IS NOT NULL "490 "ORDER BY started DESC LIMIT 1", (inc["id"],)).fetchone()491 if status in POLICY["trigger_statuses"] or status == "degraded":492 if m and (json.loads(m["commits"] or "[]")):493 asyncio.create_task(rollback_mission(m, "toujours cassé après la fenêtre de surveillance"))494 set_incident(c, inc["id"], state="cooldown")495 incident_event(inc["id"], inc["service"], inc["source"], "cooldown",496 f"non guéri après {POLICY['watch_window_hours']}h → rollback + cooldown")497 elif status == "ok":498 set_incident(c, inc["id"], state="resolved", resolved=now())499 incident_event(inc["id"], inc["service"], inc["source"], "resolved", "confirmé ok")500501 # 3. Cooldown expiré → réouverture ou abandon502 for inc in c.execute("SELECT * FROM incidents WHERE state='cooldown'").fetchall():503 if now() - inc["updated"] < POLICY["attempt_cooldown_hours"] * 3600:504 continue505 status = current_status(inc["service"], inc["source"])506 if status == "ok":507 set_incident(c, inc["id"], state="resolved", resolved=now())508 incident_event(inc["id"], inc["service"], inc["source"], "resolved", "guéri pendant le cooldown")509 elif inc["attempts"] >= POLICY["max_attempts_per_incident"]:510 set_incident(c, inc["id"], state="abandoned")511 incident_event(inc["id"], inc["service"], inc["source"], "abandoned",512 f"{inc['attempts']} tentatives épuisées — intervention humaine requise")513 else:514 set_incident(c, inc["id"], state="open")515 incident_event(inc["id"], inc["service"], inc["source"], "open", "cooldown terminé, nouvelle tentative")516517 # 4. Dispatch (1 mission à la fois par agent, broken avant stale, plus vieux d'abord)518 running = c.execute("SELECT COUNT(*) n FROM missions WHERE state='running'").fetchone()["n"]519 if running < POLICY["max_concurrent_missions"]:520 busy_nodes = {m["node"] for m in c.execute("SELECT node FROM missions WHERE state='running'").fetchall()}521 nxt = c.execute(522 "SELECT * FROM incidents WHERE state='open' "523 "ORDER BY CASE status_detected WHEN 'manual' THEN 0 WHEN 'effort_new' THEN 0 "524 "WHEN 'effort_enrich' THEN 0 WHEN 'broken' THEN 1 ELSE 2 END, created "525 ).fetchall()526 for inc in nxt:527 if inc["service"] not in SERVICES:528 continue529 if SERVICES[inc["service"]]["node"] in busy_nodes:530 continue531 set_incident(c, inc["id"], state="dispatched")532 asyncio.create_task(dispatch_row(inc["id"]))533 break534535536def current_status(service: str, source: str) -> str:537 block = LATEST["mine"].get(service) or {}538 for conn in block.get("connectors", []):539 if conn["source"] == source:540 return conn["status"]541 return "inconnu"542543544async def dispatch_row(iid: str) -> None:545 with db() as c:546 inc = c.execute("SELECT * FROM incidents WHERE id=?", (iid,)).fetchone()547 if inc:548 await dispatch(inc, json.loads(inc["detail"] or "{}"))549 # dispatch() remet fixing; si l'appel a échoué/409, on relâche550 with db() as c:551 cur = c.execute("SELECT state FROM incidents WHERE id=?", (iid,)).fetchone()552 if cur and cur["state"] == "dispatched":553 set_incident(c, iid, state="open")554555556async def engine() -> None:557 global LOOP558 LOOP = asyncio.get_running_loop()559 await asyncio.sleep(3)560 while True:561 try:562 await tick()563 except Exception as exc:564 import traceback565 traceback.print_exc()566 hub.publish_sync({"kind": "log", "level": "error", "msg": f"tick: {exc}", "ts": now()})567 await asyncio.sleep(POLICY["poll_interval_seconds"])568569570# ------------------------------------------------------------------- app ---571572app = FastAPI(title=f"KA Guardian — {AGENT}", docs_url=None, redoc_url=None)573574575@app.on_event("startup")576async def startup() -> None:577 init_db()578 asyncio.create_task(engine())579580581def check_token(tok: str | None) -> None:582 if not TOKEN or tok != TOKEN:583 raise HTTPException(status_code=401)584585586@app.post("/api/ingest/{mission_id}")587async def ingest(mission_id: str, req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:588 check_token(x_ka_token)589 ev = await req.json()590 etype, data = ev.get("type"), ev.get("data", {})591 with db() as c:592 c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",593 (mission_id, now(), etype, jdump(data)))594 m = c.execute("SELECT * FROM missions WHERE id=?", (mission_id,)).fetchone()595 if m:596 if etype == "start":597 c.execute("UPDATE missions SET base_commit=? WHERE id=?", (data.get("base_commit"), mission_id))598 elif etype in ("final", "error"):599 await finalize(c, m, etype, data)600 hub.publish_sync({"kind": "mission_event", "mission_id": mission_id, "type": etype,601 "data": data, "ts": now()})602 return {"ok": True}603604605async def finalize(c: sqlite3.Connection, m: sqlite3.Row, etype: str, data: dict[str, Any]) -> None:606 inc = c.execute("SELECT * FROM incidents WHERE id=?", (m["incident_id"],)).fetchone()607 if etype == "error":608 c.execute("UPDATE missions SET state='error', ended=? WHERE id=?", (now(), m["id"]))609 if inc:610 set_incident(c, inc["id"], state="cooldown")611 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"erreur mission: {data.get('error')}")612 return613 verdict = data.get("verdict", {})614 commits = data.get("commits", [])615 health = data.get("health", {})616 c.execute("UPDATE missions SET state='done', ended=?, commits=?, verdict=?, cost_usd=?, num_turns=?, health=?, base_commit=COALESCE(base_commit,?) WHERE id=?",617 (now(), jdump(commits), jdump(verdict), data.get("cost_usd"),618 data.get("num_turns"), jdump(health), data.get("base_commit"), m["id"]))619 m2 = c.execute("SELECT * FROM missions WHERE id=?", (m["id"],)).fetchone()620 v = verdict.get("verdict", "inconnu")621 if not inc:622 return623 if inc["status_detected"] in ("effort_new", "effort_enrich", "effort_degrade"):624 # Les efforts commandés ne passent pas par la surveillance api-ka:625 # verdict + santé de l'app décident tout de suite.626 if not health.get("ok", True):627 asyncio.create_task(rollback_mission(m2, "healthcheck app en échec post-effort"))628 set_incident(c, inc["id"], state="abandoned")629 incident_event(inc["id"], m["service"], m["source"], "abandoned", "app en mauvaise santé → ROLLBACK immédiat")630 elif v == "livre":631 set_incident(c, inc["id"], state="resolved", resolved=now())632 incident_event(inc["id"], m["service"], m["source"], "resolved",633 f"effort livré ({len(commits)} commit{'s' if len(commits) > 1 else ''})")634 elif v == "echec" and commits:635 asyncio.create_task(rollback_mission(m2, "effort en échec déclaré → rollback préventif"))636 set_incident(c, inc["id"], state="abandoned")637 incident_event(inc["id"], m["service"], m["source"], "abandoned", "effort en échec → rollback")638 else:639 # Session interrompue (plafond durée/coût) ou verdict illisible:640 # l'app est saine, les commits incrémentaux testés sont conservés.641 set_incident(c, inc["id"], state="abandoned")642 incident_event(inc["id"], m["service"], m["source"], "abandoned",643 f"effort terminé sans verdict livré ({v}) — {len(commits)} commit(s) conservé(s), app saine")644 return645 if not health.get("ok", True):646 # L'app est tombée → rollback immédiat, quoi qu'ait dit l'agent.647 asyncio.create_task(rollback_mission(m2, "healthcheck app en échec post-mission"))648 set_incident(c, inc["id"], state="cooldown")649 incident_event(inc["id"], m["service"], m["source"], "cooldown", "app en mauvaise santé → ROLLBACK immédiat")650 elif v == "repare" and commits:651 set_incident(c, inc["id"], state="watching")652 incident_event(inc["id"], m["service"], m["source"], "watching",653 f"réparation déclarée ({len(commits)} commit) — surveillance {POLICY['watch_window_hours']}h")654 elif v in ("echec", "inconnu") and commits:655 asyncio.create_task(rollback_mission(m2, f"verdict {v} avec commits → annulation par prudence"))656 set_incident(c, inc["id"], state="cooldown")657 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"verdict {v} → rollback préventif")658 elif v == "site_source_mort":659 set_incident(c, inc["id"], state="abandoned")660 incident_event(inc["id"], m["service"], m["source"], "abandoned", "source définitivement morte (diagnostic agent)")661 elif v in ("rien_a_faire", "repare"):662 set_incident(c, inc["id"], state="watching")663 incident_event(inc["id"], m["service"], m["source"], "watching", f"verdict {v} — surveillance")664 else:665 set_incident(c, inc["id"], state="cooldown")666 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"verdict {v}")667668669@app.get("/api/state")670def state() -> dict[str, Any]:671 with db() as c:672 incidents = [dict(r) for r in c.execute(673 "SELECT * FROM incidents ORDER BY created DESC LIMIT 200").fetchall()]674 missions = [dict(r) for r in c.execute(675 "SELECT * FROM missions ORDER BY started DESC LIMIT 100").fetchall()]676 stats = c.execute("""SELECT677 (SELECT COUNT(*) FROM missions) n_missions,678 (SELECT COUNT(*) FROM incidents WHERE state='resolved') n_resolved,679 (SELECT COUNT(*) FROM incidents WHERE state IN ('open','dispatched','fixing','watching','cooldown')) n_active,680 (SELECT COUNT(*) FROM incidents WHERE state='abandoned') n_abandoned,681 (SELECT COUNT(*) FROM events WHERE type='rollback') n_rollbacks,682 (SELECT ROUND(SUM(cost_usd),2) FROM missions) cost_total,683 (SELECT ROUND(AVG(resolved-created)/3600.0,1) FROM incidents WHERE state='resolved' AND resolved IS NOT NULL) mttr_h684 """).fetchone()685 history = [dict(r) for r in c.execute(686 "SELECT ts, mine FROM snapshots WHERE ts > ? ORDER BY ts", (now() - 7 * 86400,)).fetchall()]687 return {688 "agent": AGENT, "identity": {k: ME.get(k) for k in ("domain", "accent", "accent2", "accent_soft", "tagline", "model")},689 "services": {s: {**SERVICES[s], "node_alias": NODES[SERVICES[s]["node"]]["alias"]} for s in ME["services"]},690 "siblings": {a: {"domain": v["domain"], "accent": v["accent"], "services": v["services"]}691 for a, v in TOPO["agents"].items() if a != AGENT},692 "latest": LATEST, "incidents": incidents, "missions": missions,693 "stats": dict(stats), "history": history, "policy": POLICY, "now": now(),694 }695696697@app.get("/api/missions/{mission_id}")698def mission_detail(mission_id: str) -> dict[str, Any]:699 with db() as c:700 m = c.execute("SELECT * FROM missions WHERE id=?", (mission_id,)).fetchone()701 if not m:702 raise HTTPException(status_code=404)703 evs = [dict(r) for r in c.execute(704 "SELECT ts,type,data FROM events WHERE mission_id=? ORDER BY id", (mission_id,)).fetchall()]705 return {"mission": dict(m), "events": evs}706707708@app.get("/events")709async def sse(request: Request) -> StreamingResponse:710 q: asyncio.Queue = asyncio.Queue()711 hub.clients.add(q)712713 async def gen():714 try:715 yield f"data: {jdump({'kind': 'hello', 'agent': AGENT, 'ts': now()})}\n\n"716 while True:717 if await request.is_disconnected():718 break719 try:720 msg = await asyncio.wait_for(q.get(), timeout=25)721 yield f"data: {jdump(msg)}\n\n"722 except asyncio.TimeoutError:723 yield ": keepalive\n\n"724 finally:725 hub.clients.discard(q)726727 return StreamingResponse(gen(), media_type="text/event-stream",728 headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})729730731@app.get("/api/connectors/{service}")732def connectors_list(service: str) -> dict[str, Any]:733 """Sources connues du service (pour le menu déroulant d'enrichissement)."""734 block = LATEST["mine"].get(service) or {}735 return {"service": service,736 "connectors": sorted(737 ({"source": c["source"], "status": c["status"]} for c in block.get("connectors", [])),738 key=lambda x: x["source"])}739740741@app.post("/api/admin/effort")742async def admin_effort(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:743 """Commande un effort autonome: nouveau connecteur ou enrichissement."""744 check_token(x_ka_token)745 body = await req.json()746 kind = body.get("kind")747 service = body.get("service")748 note = (body.get("note") or "").strip()[:2000]749 if kind not in ("effort_new", "effort_enrich", "effort_degrade") or service not in SERVICES:750 raise HTTPException(status_code=400, detail="kind ou service invalide")751 if kind == "effort_enrich":752 source = (body.get("source") or "").strip()753 if not source:754 raise HTTPException(status_code=400, detail="source requise pour un enrichissement")755 elif kind == "effort_degrade":756 source = f"dégradés·{uuid.uuid4().hex[:6]}"757 else:758 source = f"nouveau·{uuid.uuid4().hex[:6]}"759 detail: dict[str, Any] = {"note": note, "kind": kind}760 try:761 if body.get("max_minutes"):762 detail["max_minutes"] = max(10, min(240, int(body["max_minutes"])))763 if body.get("max_cost_usd"):764 detail["max_cost_usd"] = max(0.5, min(100.0, float(body["max_cost_usd"])))765 except (TypeError, ValueError):766 raise HTTPException(status_code=400, detail="max_minutes/max_cost_usd invalides")767 iid = uuid.uuid4().hex[:10]768 with db() as c:769 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "770 "VALUES(?,?,?,?,?,?,?,?)",771 (iid, service, source, kind, "open", now(), now(), jdump(detail)))772 labels = {"effort_new": "nouveau connecteur", "effort_enrich": f"enrichissement de {source}",773 "effort_degrade": "inspection des connecteurs dégradés"}774 incident_event(iid, service, source, "open",775 f"effort commandé: {labels[kind]}"776 + (f" · ≤{detail['max_minutes']} min" if detail.get("max_minutes") else "")777 + (f" · ≤{detail['max_cost_usd']} $" if detail.get("max_cost_usd") else ""))778 return {"ok": True, "incident_id": iid}779780781@app.post("/api/admin/mission")782async def admin_mission(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:783 """Déclenche manuellement une mission sur un connecteur (même sain)."""784 check_token(x_ka_token)785 body = await req.json()786 service, source = body["service"], body["source"]787 if service not in SERVICES:788 raise HTTPException(status_code=400, detail="service inconnu")789 detail = {"source": source, "status": "manual", "message": body.get("note", "mission manuelle")}790 for conn in (LATEST["mine"].get(service) or {}).get("connectors", []):791 if conn["source"] == source:792 detail = {**conn, "status": "manual"}793 with db() as c:794 inc = active_incident(c, service, source)795 if inc is None:796 iid = uuid.uuid4().hex[:10]797 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "798 "VALUES(?,?,?,?,?,?,?,?)",799 (iid, service, source, "manual", "open", now(), now(), jdump(detail)))800 else:801 iid = inc["id"]802 set_incident(c, iid, state="open", status_detected="manual")803 incident_event(iid, service, source, "open", "mission manuelle demandée")804 return {"ok": True, "incident_id": iid}805806807@app.post("/api/admin/pause")808async def admin_pause(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:809 check_token(x_ka_token)810 body = await req.json()811 LATEST["paused"] = bool(body.get("paused", True))812 return {"ok": True, "paused": LATEST["paused"]}813814815@app.get("/health")816def health() -> dict[str, Any]:817 return {"ok": True, "agent": AGENT, "polled": LATEST["polled"], "paused": LATEST["paused"]}818819820@app.get("/favicon.svg")821def favicon() -> FileResponse:822 return FileResponse(BASE / "web" / f"favicon-{AGENT}.svg", media_type="image/svg+xml")823824825@app.get("/")826def index() -> FileResponse:827 return FileResponse(BASE / "web" / "index.html")828829830@app.get("/commander")831def commander() -> FileResponse:832 return FileResponse(BASE / "web" / "commander.html")833834835app.mount("/static", StaticFiles(directory=BASE / "web"), name="static")836837838if __name__ == "__main__":839 import uvicorn840 uvicorn.run(app, host="0.0.0.0", port=ME["port"])841