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. Réconciliation des missions en cours depuis > 10 min: on demande au476 # runner ce qu'il en est (événement final perdu? runner redémarré?).477 for m in c.execute("SELECT id FROM missions WHERE state='running' AND started < ?",478 (now() - 600,)).fetchall():479 if m["id"] not in RECONCILING:480 RECONCILING.add(m["id"])481 asyncio.create_task(reconcile_mission(m["id"]))482483 # 2. Watching → rollback si la fenêtre est passée et toujours cassé484 for inc in c.execute("SELECT * FROM incidents WHERE state='watching'").fetchall():485 if now() - inc["updated"] < POLICY["watch_window_hours"] * 3600:486 continue487 status = current_status(inc["service"], inc["source"])488 m = c.execute("SELECT * FROM missions WHERE incident_id=? AND base_commit IS NOT NULL "489 "ORDER BY started DESC LIMIT 1", (inc["id"],)).fetchone()490 if status in POLICY["trigger_statuses"] or status == "degraded":491 if m and (json.loads(m["commits"] or "[]")):492 asyncio.create_task(rollback_mission(m, "toujours cassé après la fenêtre de surveillance"))493 set_incident(c, inc["id"], state="cooldown")494 incident_event(inc["id"], inc["service"], inc["source"], "cooldown",495 f"non guéri après {POLICY['watch_window_hours']}h → rollback + cooldown")496 elif status == "ok":497 set_incident(c, inc["id"], state="resolved", resolved=now())498 incident_event(inc["id"], inc["service"], inc["source"], "resolved", "confirmé ok")499500 # 3. Cooldown expiré → réouverture ou abandon501 for inc in c.execute("SELECT * FROM incidents WHERE state='cooldown'").fetchall():502 if now() - inc["updated"] < POLICY["attempt_cooldown_hours"] * 3600:503 continue504 status = current_status(inc["service"], inc["source"])505 if status == "ok":506 set_incident(c, inc["id"], state="resolved", resolved=now())507 incident_event(inc["id"], inc["service"], inc["source"], "resolved", "guéri pendant le cooldown")508 elif inc["attempts"] >= POLICY["max_attempts_per_incident"]:509 set_incident(c, inc["id"], state="abandoned")510 incident_event(inc["id"], inc["service"], inc["source"], "abandoned",511 f"{inc['attempts']} tentatives épuisées — intervention humaine requise")512 else:513 set_incident(c, inc["id"], state="open")514 incident_event(inc["id"], inc["service"], inc["source"], "open", "cooldown terminé, nouvelle tentative")515516 # 4. Dispatch (1 mission à la fois par agent, broken avant stale, plus vieux d'abord)517 running = c.execute("SELECT COUNT(*) n FROM missions WHERE state='running'").fetchone()["n"]518 if running < POLICY["max_concurrent_missions"]:519 busy_nodes = {m["node"] for m in c.execute("SELECT node FROM missions WHERE state='running'").fetchall()}520 nxt = c.execute(521 "SELECT * FROM incidents WHERE state='open' "522 "ORDER BY CASE status_detected WHEN 'manual' THEN 0 WHEN 'effort_new' THEN 0 "523 "WHEN 'effort_enrich' THEN 0 WHEN 'broken' THEN 1 ELSE 2 END, created "524 ).fetchall()525 for inc in nxt:526 if inc["service"] not in SERVICES:527 continue528 if SERVICES[inc["service"]]["node"] in busy_nodes:529 continue530 set_incident(c, inc["id"], state="dispatched")531 asyncio.create_task(dispatch_row(inc["id"]))532 break533534535def current_status(service: str, source: str) -> str:536 block = LATEST["mine"].get(service) or {}537 for conn in block.get("connectors", []):538 if conn["source"] == source:539 return conn["status"]540 return "inconnu"541542543async def dispatch_row(iid: str) -> None:544 with db() as c:545 inc = c.execute("SELECT * FROM incidents WHERE id=?", (iid,)).fetchone()546 if inc:547 await dispatch(inc, json.loads(inc["detail"] or "{}"))548 # dispatch() remet fixing; si l'appel a échoué/409, on relâche549 with db() as c:550 cur = c.execute("SELECT state FROM incidents WHERE id=?", (iid,)).fetchone()551 if cur and cur["state"] == "dispatched":552 set_incident(c, iid, state="open")553554555RECONCILING: set[str] = set()556557558async def reconcile_mission(mid: str) -> None:559 """Résout une mission « en cours » suspecte auprès de son runner."""560 try:561 with db() as c:562 m = c.execute("SELECT * FROM missions WHERE id=? AND state='running'", (mid,)).fetchone()563 if not m:564 return565 ip = NODES[m["node"]]["lan_ip"]566 try:567 code, body = await lan_post(ip, RUNNER_PORT, "/mission-result", {"mission_id": mid}, timeout=20)568 resp = json.loads(body) if code == 200 else {"status": "erreur"}569 except Exception:570 return # courrier/runner injoignable: on retentera au prochain tick571 if resp.get("status") == "running":572 return573 with db() as c:574 m = c.execute("SELECT * FROM missions WHERE id=? AND state='running'", (mid,)).fetchone()575 if not m:576 return577 if resp.get("status") == "done":578 data = (resp.get("final") or {}).get("data", {})579 c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",580 (mid, now(), "final", jdump(data)))581 await finalize(c, m, "final", data)582 hub.publish_sync({"kind": "mission_event", "mission_id": mid, "type": "final",583 "data": data, "ts": now()})584 elif now() - m["started"] > 900:585 # Le runner ne connaît pas cette mission: tuée en vol586 # (redémarrage). On la clôt et on relance l'incident.587 c.execute("UPDATE missions SET state='error', ended=? WHERE id=?", (now(), mid))588 inc = c.execute("SELECT * FROM incidents WHERE id=?", (m["incident_id"],)).fetchone()589 if inc and inc["state"] in ("fixing", "dispatched"):590 set_incident(c, inc["id"], state="open")591 incident_event(inc["id"], m["service"], m["source"], "open",592 "mission perdue (runner interrompu) — remise en file")593 finally:594 RECONCILING.discard(mid)595596597async def engine() -> None:598 global LOOP599 LOOP = asyncio.get_running_loop()600 await asyncio.sleep(3)601 while True:602 try:603 await tick()604 except Exception as exc:605 import traceback606 traceback.print_exc()607 hub.publish_sync({"kind": "log", "level": "error", "msg": f"tick: {exc}", "ts": now()})608 await asyncio.sleep(POLICY["poll_interval_seconds"])609610611# ------------------------------------------------------------------- app ---612613app = FastAPI(title=f"KA Guardian — {AGENT}", docs_url=None, redoc_url=None)614615616@app.on_event("startup")617async def startup() -> None:618 init_db()619 asyncio.create_task(engine())620621622def check_token(tok: str | None) -> None:623 if not TOKEN or tok != TOKEN:624 raise HTTPException(status_code=401)625626627@app.post("/api/ingest/{mission_id}")628async def ingest(mission_id: str, req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:629 check_token(x_ka_token)630 ev = await req.json()631 etype, data = ev.get("type"), ev.get("data", {})632 with db() as c:633 c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",634 (mission_id, now(), etype, jdump(data)))635 m = c.execute("SELECT * FROM missions WHERE id=?", (mission_id,)).fetchone()636 if m:637 if etype == "start":638 c.execute("UPDATE missions SET base_commit=? WHERE id=?", (data.get("base_commit"), mission_id))639 elif etype in ("final", "error"):640 await finalize(c, m, etype, data)641 hub.publish_sync({"kind": "mission_event", "mission_id": mission_id, "type": etype,642 "data": data, "ts": now()})643 return {"ok": True}644645646async def finalize(c: sqlite3.Connection, m: sqlite3.Row, etype: str, data: dict[str, Any]) -> None:647 inc = c.execute("SELECT * FROM incidents WHERE id=?", (m["incident_id"],)).fetchone()648 if etype == "error":649 c.execute("UPDATE missions SET state='error', ended=? WHERE id=?", (now(), m["id"]))650 if inc:651 set_incident(c, inc["id"], state="cooldown")652 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"erreur mission: {data.get('error')}")653 return654 verdict = data.get("verdict", {})655 commits = data.get("commits", [])656 health = data.get("health", {})657 c.execute("UPDATE missions SET state='done', ended=?, commits=?, verdict=?, cost_usd=?, num_turns=?, health=?, base_commit=COALESCE(base_commit,?) WHERE id=?",658 (now(), jdump(commits), jdump(verdict), data.get("cost_usd"),659 data.get("num_turns"), jdump(health), data.get("base_commit"), m["id"]))660 m2 = c.execute("SELECT * FROM missions WHERE id=?", (m["id"],)).fetchone()661 v = verdict.get("verdict", "inconnu")662 if not inc:663 return664 if inc["status_detected"] in ("effort_new", "effort_enrich", "effort_degrade"):665 # Les efforts commandés ne passent pas par la surveillance api-ka:666 # verdict + santé de l'app décident tout de suite.667 if not health.get("ok", True):668 asyncio.create_task(rollback_mission(m2, "healthcheck app en échec post-effort"))669 set_incident(c, inc["id"], state="abandoned")670 incident_event(inc["id"], m["service"], m["source"], "abandoned", "app en mauvaise santé → ROLLBACK immédiat")671 elif v == "livre":672 set_incident(c, inc["id"], state="resolved", resolved=now())673 incident_event(inc["id"], m["service"], m["source"], "resolved",674 f"effort livré ({len(commits)} commit{'s' if len(commits) > 1 else ''})")675 elif v == "echec" and commits:676 asyncio.create_task(rollback_mission(m2, "effort en échec déclaré → rollback préventif"))677 set_incident(c, inc["id"], state="abandoned")678 incident_event(inc["id"], m["service"], m["source"], "abandoned", "effort en échec → rollback")679 else:680 # Session interrompue (plafond durée/coût) ou verdict illisible:681 # l'app est saine, les commits incrémentaux testés sont conservés.682 set_incident(c, inc["id"], state="abandoned")683 incident_event(inc["id"], m["service"], m["source"], "abandoned",684 f"effort terminé sans verdict livré ({v}) — {len(commits)} commit(s) conservé(s), app saine")685 return686 if not health.get("ok", True):687 # L'app est tombée → rollback immédiat, quoi qu'ait dit l'agent.688 asyncio.create_task(rollback_mission(m2, "healthcheck app en échec post-mission"))689 set_incident(c, inc["id"], state="cooldown")690 incident_event(inc["id"], m["service"], m["source"], "cooldown", "app en mauvaise santé → ROLLBACK immédiat")691 elif v == "repare" and commits:692 set_incident(c, inc["id"], state="watching")693 incident_event(inc["id"], m["service"], m["source"], "watching",694 f"réparation déclarée ({len(commits)} commit) — surveillance {POLICY['watch_window_hours']}h")695 elif v in ("echec", "inconnu") and commits:696 asyncio.create_task(rollback_mission(m2, f"verdict {v} avec commits → annulation par prudence"))697 set_incident(c, inc["id"], state="cooldown")698 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"verdict {v} → rollback préventif")699 elif v == "site_source_mort":700 set_incident(c, inc["id"], state="abandoned")701 incident_event(inc["id"], m["service"], m["source"], "abandoned", "source définitivement morte (diagnostic agent)")702 elif v in ("rien_a_faire", "repare"):703 set_incident(c, inc["id"], state="watching")704 incident_event(inc["id"], m["service"], m["source"], "watching", f"verdict {v} — surveillance")705 else:706 set_incident(c, inc["id"], state="cooldown")707 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"verdict {v}")708709710@app.get("/api/state")711def state() -> dict[str, Any]:712 with db() as c:713 incidents = [dict(r) for r in c.execute(714 "SELECT * FROM incidents ORDER BY created DESC LIMIT 200").fetchall()]715 missions = [dict(r) for r in c.execute(716 "SELECT * FROM missions ORDER BY started DESC LIMIT 100").fetchall()]717 stats = c.execute("""SELECT718 (SELECT COUNT(*) FROM missions) n_missions,719 (SELECT COUNT(*) FROM incidents WHERE state='resolved') n_resolved,720 (SELECT COUNT(*) FROM incidents WHERE state IN ('open','dispatched','fixing','watching','cooldown')) n_active,721 (SELECT COUNT(*) FROM incidents WHERE state='abandoned') n_abandoned,722 (SELECT COUNT(*) FROM events WHERE type='rollback') n_rollbacks,723 (SELECT ROUND(SUM(cost_usd),2) FROM missions) cost_total,724 (SELECT ROUND(AVG(resolved-created)/3600.0,1) FROM incidents WHERE state='resolved' AND resolved IS NOT NULL) mttr_h725 """).fetchone()726 history = [dict(r) for r in c.execute(727 "SELECT ts, mine FROM snapshots WHERE ts > ? ORDER BY ts", (now() - 7 * 86400,)).fetchall()]728 return {729 "agent": AGENT, "identity": {k: ME.get(k) for k in ("domain", "accent", "accent2", "accent_soft", "tagline", "model")},730 "services": {s: {**SERVICES[s], "node_alias": NODES[SERVICES[s]["node"]]["alias"]} for s in ME["services"]},731 "siblings": {a: {"domain": v["domain"], "accent": v["accent"], "services": v["services"]}732 for a, v in TOPO["agents"].items() if a != AGENT},733 "latest": LATEST, "incidents": incidents, "missions": missions,734 "stats": dict(stats), "history": history, "policy": POLICY, "now": now(),735 }736737738@app.get("/api/missions/{mission_id}")739def mission_detail(mission_id: str) -> dict[str, Any]:740 with db() as c:741 m = c.execute("SELECT * FROM missions WHERE id=?", (mission_id,)).fetchone()742 if not m:743 raise HTTPException(status_code=404)744 evs = [dict(r) for r in c.execute(745 "SELECT ts,type,data FROM events WHERE mission_id=? ORDER BY id", (mission_id,)).fetchall()]746 return {"mission": dict(m), "events": evs}747748749@app.get("/events")750async def sse(request: Request) -> StreamingResponse:751 q: asyncio.Queue = asyncio.Queue()752 hub.clients.add(q)753754 async def gen():755 try:756 yield f"data: {jdump({'kind': 'hello', 'agent': AGENT, 'ts': now()})}\n\n"757 while True:758 if await request.is_disconnected():759 break760 try:761 msg = await asyncio.wait_for(q.get(), timeout=25)762 yield f"data: {jdump(msg)}\n\n"763 except asyncio.TimeoutError:764 yield ": keepalive\n\n"765 finally:766 hub.clients.discard(q)767768 return StreamingResponse(gen(), media_type="text/event-stream",769 headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})770771772@app.get("/api/connectors/{service}")773def connectors_list(service: str) -> dict[str, Any]:774 """Sources connues du service (pour le menu déroulant d'enrichissement)."""775 block = LATEST["mine"].get(service) or {}776 return {"service": service,777 "connectors": sorted(778 ({"source": c["source"], "status": c["status"]} for c in block.get("connectors", [])),779 key=lambda x: x["source"])}780781782@app.post("/api/admin/effort")783async def admin_effort(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:784 """Commande un effort autonome: nouveau connecteur ou enrichissement."""785 check_token(x_ka_token)786 body = await req.json()787 kind = body.get("kind")788 service = body.get("service")789 note = (body.get("note") or "").strip()[:2000]790 if kind not in ("effort_new", "effort_enrich", "effort_degrade") or service not in SERVICES:791 raise HTTPException(status_code=400, detail="kind ou service invalide")792 if kind == "effort_enrich":793 source = (body.get("source") or "").strip()794 if not source:795 raise HTTPException(status_code=400, detail="source requise pour un enrichissement")796 elif kind == "effort_degrade":797 source = f"dégradés·{uuid.uuid4().hex[:6]}"798 else:799 source = f"nouveau·{uuid.uuid4().hex[:6]}"800 detail: dict[str, Any] = {"note": note, "kind": kind}801 try:802 if body.get("max_minutes"):803 detail["max_minutes"] = max(10, min(240, int(body["max_minutes"])))804 if body.get("max_cost_usd"):805 detail["max_cost_usd"] = max(0.5, min(100.0, float(body["max_cost_usd"])))806 except (TypeError, ValueError):807 raise HTTPException(status_code=400, detail="max_minutes/max_cost_usd invalides")808 iid = uuid.uuid4().hex[:10]809 with db() as c:810 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "811 "VALUES(?,?,?,?,?,?,?,?)",812 (iid, service, source, kind, "open", now(), now(), jdump(detail)))813 labels = {"effort_new": "nouveau connecteur", "effort_enrich": f"enrichissement de {source}",814 "effort_degrade": "inspection des connecteurs dégradés"}815 incident_event(iid, service, source, "open",816 f"effort commandé: {labels[kind]}"817 + (f" · ≤{detail['max_minutes']} min" if detail.get("max_minutes") else "")818 + (f" · ≤{detail['max_cost_usd']} $" if detail.get("max_cost_usd") else ""))819 return {"ok": True, "incident_id": iid}820821822@app.post("/api/admin/mission")823async def admin_mission(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:824 """Déclenche manuellement une mission sur un connecteur (même sain)."""825 check_token(x_ka_token)826 body = await req.json()827 service, source = body["service"], body["source"]828 if service not in SERVICES:829 raise HTTPException(status_code=400, detail="service inconnu")830 detail = {"source": source, "status": "manual", "message": body.get("note", "mission manuelle")}831 for conn in (LATEST["mine"].get(service) or {}).get("connectors", []):832 if conn["source"] == source:833 detail = {**conn, "status": "manual"}834 with db() as c:835 inc = active_incident(c, service, source)836 if inc is None:837 iid = uuid.uuid4().hex[:10]838 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "839 "VALUES(?,?,?,?,?,?,?,?)",840 (iid, service, source, "manual", "open", now(), now(), jdump(detail)))841 else:842 iid = inc["id"]843 set_incident(c, iid, state="open", status_detected="manual")844 incident_event(iid, service, source, "open", "mission manuelle demandée")845 return {"ok": True, "incident_id": iid}846847848@app.post("/api/admin/pause")849async def admin_pause(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:850 check_token(x_ka_token)851 body = await req.json()852 LATEST["paused"] = bool(body.get("paused", True))853 return {"ok": True, "paused": LATEST["paused"]}854855856@app.get("/health")857def health() -> dict[str, Any]:858 return {"ok": True, "agent": AGENT, "polled": LATEST["polled"], "paused": LATEST["paused"]}859860861@app.get("/favicon.svg")862def favicon() -> FileResponse:863 return FileResponse(BASE / "web" / f"favicon-{AGENT}.svg", media_type="image/svg+xml")864865866@app.get("/")867def index() -> FileResponse:868 return FileResponse(BASE / "web" / "index.html")869870871@app.get("/commander")872def commander() -> FileResponse:873 return FileResponse(BASE / "web" / "commander.html")874875876app.mount("/static", StaticFiles(directory=BASE / "web"), name="static")877878879if __name__ == "__main__":880 import uvicorn881 uvicorn.run(app, host="0.0.0.0", port=ME["port"])882