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_new":277 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', '')}).278{"Consigne de l'opérateur: " + note if note else "Aucune consigne particulière: choisis la source la plus utile."}279280{env}281282== DÉMARCHE ==2831. 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).2842. 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.2853. 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.2864. 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).2875. DOCUMENTE: ajoute la fiche docs/connecteurs/<source>.md si le repo suit cette convention (URL, stratégie, pièges).2886. Redémarre {sync_proc}, vérifie le site, committe.289290{effort_stack(service)}291292{rules}293294== FIN DE MISSION ==295Termine ta TOUTE DERNIÈRE réponse par un bloc JSON:296{{"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"]}}"""297 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', '')}).298{"Consigne de l'opérateur: " + note if note else "Aucune consigne particulière: maximise la valeur (couverture, champs, robustesse)."}299300{env}301302== DÉMARCHE ==3031. É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.3042. 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.3053. IMPLÉMENTE proprement dans le pattern du repo, SANS casser le format de sortie existant (les consommateurs avals dépendent du schéma).3064. 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.3075. METS À JOUR la fiche docs/connecteurs/<source>.md si elle existe.3086. Redémarre {sync_proc}, vérifie le site, committe.309310{effort_stack(service)}311312{rules}313314== FIN DE MISSION ==315Termine ta TOUTE DERNIÈRE réponse par un bloc JSON:316{{"verdict": "livre|echec", "diagnostic": "état initial constaté", "actions": "enrichissements livrés", "test": "avant/après mesuré", "fichiers": ["fichiers modifiés"]}}"""317318319async def dispatch(incident: sqlite3.Row, health: dict[str, Any]) -> None:320 service, source, iid = incident["service"], incident["source"], incident["id"]321 svc = SERVICES[service]322 node = svc["node"]323 ip = NODES[node]["lan_ip"]324 mid = uuid.uuid4().hex[:12]325 my_ip = os.environ.get("KA_GUARDIAN_SELF_IP", "192.168.2.69")326 kind = incident["status_detected"]327 if kind in ("effort_new", "effort_enrich"):328 prompt = build_effort_prompt(kind, service, source, health.get("note", ""))329 max_turns, timeout = POLICY.get("effort_max_turns", 150), POLICY.get("effort_timeout_seconds", 7200)330 else:331 prompt = build_prompt(service, source, health)332 max_turns, timeout = POLICY["mission_max_turns"], POLICY["mission_timeout_seconds"]333 payload = {334 "mission_id": mid, "agent": AGENT, "service": service, "source": source,335 "dir": svc["dir"], "pm2": svc["pm2"], "web_port": svc["web_port"],336 "model": ME.get("model", "sonnet"),337 "max_turns": max_turns,338 "timeout_seconds": timeout,339 "prompt": prompt,340 "callback_url": f"http://{my_ip}:{ME['port']}/api/ingest/{mid}",341 }342 try:343 code, body = await lan_post(ip, RUNNER_PORT, "/missions", payload, timeout=25)344 if code == 409:345 return # runner occupé, on retentera au prochain tick346 if code != 200:347 raise ConnectionError(f"runner {code}: {body[:200]}")348 except Exception as exc:349 print(f"[dispatch] {service}/{source} → {type(exc).__name__}: {exc}", flush=True)350 hub.publish_sync({"kind": "log", "level": "error", "msg": f"dispatch {source}: {exc}", "ts": now()})351 return352 with db() as c:353 c.execute("INSERT INTO missions(id,incident_id,service,source,node,state,started) VALUES(?,?,?,?,?,?,?)",354 (mid, iid, service, source, node, "running", now()))355 set_incident(c, iid, state="fixing", attempts=incident["attempts"] + 1)356 incident_event(iid, service, source, "fixing", f"mission {mid} dépêchée sur {NODES[node]['alias']}")357 hub.publish_sync({"kind": "mission", "mission_id": mid, "service": service,358 "source": source, "state": "running", "ts": now()})359360361async def rollback_mission(mission: sqlite3.Row, reason: str) -> dict[str, Any]:362 svc = SERVICES[mission["service"]]363 ip = NODES[mission["node"]]["lan_ip"]364 try:365 code, body = await lan_post(ip, RUNNER_PORT, "/rollback",366 {"dir": svc["dir"], "base_commit": mission["base_commit"],367 "pm2": svc["pm2"], "web_port": svc["web_port"]}, timeout=300)368 out = json.loads(body) if code == 200 else {"ok": False, "erreur": f"{code}: {body[:200]}"}369 except Exception as exc:370 out = {"ok": False, "erreur": str(exc)}371 hub.publish_sync({"kind": "rollback", "mission_id": mission["id"], "service": mission["service"],372 "source": mission["source"], "reason": reason, "result": out, "ts": now()})373 with db() as c:374 c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",375 (mission["id"], now(), "rollback", jdump({"reason": reason, "result": out})))376 return out377378379# --------------------------------------------------------------- moteur ----380381async def poll_once() -> None:382 async with httpx.AsyncClient() as cl:383 r = await cl.get(TOPO["apika_monitoring_url"], timeout=15)384 data = r.json()["data"]385 mine: dict[str, Any] = {}386 for service, block in data["services"].items():387 assigned_to_me = service in MY_SERVICES or (DEFAULT_AGENT and service not in ASSIGNED)388 if assigned_to_me:389 mine[service] = block390 LATEST.update({"mine": mine, "ecosystem": data["summary"], "polled": now()})391 with db() as c:392 summary_mine = {s: b["summary"] for s, b in mine.items()}393 c.execute("INSERT OR REPLACE INTO snapshots(ts,mine,ecosystem) VALUES(?,?,?)",394 (now(), jdump(summary_mine), jdump(data["summary"])))395 c.execute("DELETE FROM snapshots WHERE ts < ?", (now() - 30 * 86400,))396 hub.publish_sync({"kind": "snapshot", "mine": {s: b["summary"] for s, b in mine.items()},397 "ecosystem": data["summary"], "ts": now()})398399 trigger = set(POLICY["trigger_statuses"])400 with db() as c:401 for service, block in mine.items():402 if service not in SERVICES:403 continue # service inconnu de la topologie: visible au dashboard, pas d'action404 for conn in block.get("connectors", []):405 source, status = conn["source"], conn["status"]406 inc = active_incident(c, service, source)407 if status in trigger and inc is None:408 iid = uuid.uuid4().hex[:10]409 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "410 "VALUES(?,?,?,?,?,?,?,?)",411 (iid, service, source, status, "open", now(), now(), jdump(conn)))412 incident_event(iid, service, source, "open", f"détecté {status}")413 elif status in trigger and inc is not None and inc["state"] in ("open", "cooldown"):414 # Ne pas rafraîchir watching/fixing: `updated` sert de chrono415 # à la fenêtre de surveillance et au cooldown.416 c.execute("UPDATE incidents SET status_detected=?, detail=? WHERE id=?",417 (status, jdump(conn), inc["id"]))418 elif status == "ok" and inc is not None:419 if inc["state"] == "watching":420 set_incident(c, inc["id"], state="resolved", resolved=now())421 incident_event(inc["id"], service, source, "resolved", "connecteur de retour à ok — réparation confirmée")422 elif inc["state"] in ("open", "cooldown"):423 set_incident(c, inc["id"], state="self_healed", resolved=now())424 incident_event(inc["id"], service, source, "self_healed", "revenu à ok sans intervention")425426427async def tick() -> None:428 if LATEST["paused"]:429 return430 try:431 await poll_once()432 except Exception as exc:433 hub.publish_sync({"kind": "log", "level": "warn", "msg": f"poll api-ka échoué: {exc}", "ts": now()})434435 with db() as c:436 # 1. Missions zombies (runner mort / callback perdu)437 for m in c.execute("SELECT * FROM missions WHERE state='running' AND started < ?",438 (now() - POLICY.get("effort_timeout_seconds", 7200) - 900,)).fetchall():439 c.execute("UPDATE missions SET state='error', ended=? WHERE id=?", (now(), m["id"]))440 inc = c.execute("SELECT * FROM incidents WHERE id=?", (m["incident_id"],)).fetchone()441 if inc:442 set_incident(c, inc["id"], state="cooldown")443 incident_event(inc["id"], m["service"], m["source"], "cooldown", "mission sans réponse (zombie)")444445 # 2. Watching → rollback si la fenêtre est passée et toujours cassé446 for inc in c.execute("SELECT * FROM incidents WHERE state='watching'").fetchall():447 if now() - inc["updated"] < POLICY["watch_window_hours"] * 3600:448 continue449 status = current_status(inc["service"], inc["source"])450 m = c.execute("SELECT * FROM missions WHERE incident_id=? AND base_commit IS NOT NULL "451 "ORDER BY started DESC LIMIT 1", (inc["id"],)).fetchone()452 if status in POLICY["trigger_statuses"] or status == "degraded":453 if m and (json.loads(m["commits"] or "[]")):454 asyncio.create_task(rollback_mission(m, "toujours cassé après la fenêtre de surveillance"))455 set_incident(c, inc["id"], state="cooldown")456 incident_event(inc["id"], inc["service"], inc["source"], "cooldown",457 f"non guéri après {POLICY['watch_window_hours']}h → rollback + cooldown")458 elif status == "ok":459 set_incident(c, inc["id"], state="resolved", resolved=now())460 incident_event(inc["id"], inc["service"], inc["source"], "resolved", "confirmé ok")461462 # 3. Cooldown expiré → réouverture ou abandon463 for inc in c.execute("SELECT * FROM incidents WHERE state='cooldown'").fetchall():464 if now() - inc["updated"] < POLICY["attempt_cooldown_hours"] * 3600:465 continue466 status = current_status(inc["service"], inc["source"])467 if status == "ok":468 set_incident(c, inc["id"], state="resolved", resolved=now())469 incident_event(inc["id"], inc["service"], inc["source"], "resolved", "guéri pendant le cooldown")470 elif inc["attempts"] >= POLICY["max_attempts_per_incident"]:471 set_incident(c, inc["id"], state="abandoned")472 incident_event(inc["id"], inc["service"], inc["source"], "abandoned",473 f"{inc['attempts']} tentatives épuisées — intervention humaine requise")474 else:475 set_incident(c, inc["id"], state="open")476 incident_event(inc["id"], inc["service"], inc["source"], "open", "cooldown terminé, nouvelle tentative")477478 # 4. Dispatch (1 mission à la fois par agent, broken avant stale, plus vieux d'abord)479 running = c.execute("SELECT COUNT(*) n FROM missions WHERE state='running'").fetchone()["n"]480 if running < POLICY["max_concurrent_missions"]:481 busy_nodes = {m["node"] for m in c.execute("SELECT node FROM missions WHERE state='running'").fetchall()}482 nxt = c.execute(483 "SELECT * FROM incidents WHERE state='open' "484 "ORDER BY CASE status_detected WHEN 'manual' THEN 0 WHEN 'effort_new' THEN 0 "485 "WHEN 'effort_enrich' THEN 0 WHEN 'broken' THEN 1 ELSE 2 END, created "486 ).fetchall()487 for inc in nxt:488 if inc["service"] not in SERVICES:489 continue490 if SERVICES[inc["service"]]["node"] in busy_nodes:491 continue492 set_incident(c, inc["id"], state="dispatched")493 asyncio.create_task(dispatch_row(inc["id"]))494 break495496497def current_status(service: str, source: str) -> str:498 block = LATEST["mine"].get(service) or {}499 for conn in block.get("connectors", []):500 if conn["source"] == source:501 return conn["status"]502 return "inconnu"503504505async def dispatch_row(iid: str) -> None:506 with db() as c:507 inc = c.execute("SELECT * FROM incidents WHERE id=?", (iid,)).fetchone()508 if inc:509 await dispatch(inc, json.loads(inc["detail"] or "{}"))510 # dispatch() remet fixing; si l'appel a échoué/409, on relâche511 with db() as c:512 cur = c.execute("SELECT state FROM incidents WHERE id=?", (iid,)).fetchone()513 if cur and cur["state"] == "dispatched":514 set_incident(c, iid, state="open")515516517async def engine() -> None:518 global LOOP519 LOOP = asyncio.get_running_loop()520 await asyncio.sleep(3)521 while True:522 try:523 await tick()524 except Exception as exc:525 import traceback526 traceback.print_exc()527 hub.publish_sync({"kind": "log", "level": "error", "msg": f"tick: {exc}", "ts": now()})528 await asyncio.sleep(POLICY["poll_interval_seconds"])529530531# ------------------------------------------------------------------- app ---532533app = FastAPI(title=f"KA Guardian — {AGENT}", docs_url=None, redoc_url=None)534535536@app.on_event("startup")537async def startup() -> None:538 init_db()539 asyncio.create_task(engine())540541542def check_token(tok: str | None) -> None:543 if not TOKEN or tok != TOKEN:544 raise HTTPException(status_code=401)545546547@app.post("/api/ingest/{mission_id}")548async def ingest(mission_id: str, req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:549 check_token(x_ka_token)550 ev = await req.json()551 etype, data = ev.get("type"), ev.get("data", {})552 with db() as c:553 c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",554 (mission_id, now(), etype, jdump(data)))555 m = c.execute("SELECT * FROM missions WHERE id=?", (mission_id,)).fetchone()556 if m:557 if etype == "start":558 c.execute("UPDATE missions SET base_commit=? WHERE id=?", (data.get("base_commit"), mission_id))559 elif etype in ("final", "error"):560 await finalize(c, m, etype, data)561 hub.publish_sync({"kind": "mission_event", "mission_id": mission_id, "type": etype,562 "data": data, "ts": now()})563 return {"ok": True}564565566async def finalize(c: sqlite3.Connection, m: sqlite3.Row, etype: str, data: dict[str, Any]) -> None:567 inc = c.execute("SELECT * FROM incidents WHERE id=?", (m["incident_id"],)).fetchone()568 if etype == "error":569 c.execute("UPDATE missions SET state='error', ended=? WHERE id=?", (now(), m["id"]))570 if inc:571 set_incident(c, inc["id"], state="cooldown")572 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"erreur mission: {data.get('error')}")573 return574 verdict = data.get("verdict", {})575 commits = data.get("commits", [])576 health = data.get("health", {})577 c.execute("UPDATE missions SET state='done', ended=?, commits=?, verdict=?, cost_usd=?, num_turns=?, health=?, base_commit=COALESCE(base_commit,?) WHERE id=?",578 (now(), jdump(commits), jdump(verdict), data.get("cost_usd"),579 data.get("num_turns"), jdump(health), data.get("base_commit"), m["id"]))580 m2 = c.execute("SELECT * FROM missions WHERE id=?", (m["id"],)).fetchone()581 v = verdict.get("verdict", "inconnu")582 if not inc:583 return584 if inc["status_detected"] in ("effort_new", "effort_enrich"):585 # Les efforts commandés ne passent pas par la surveillance api-ka:586 # verdict + santé de l'app décident tout de suite.587 if not health.get("ok", True):588 asyncio.create_task(rollback_mission(m2, "healthcheck app en échec post-effort"))589 set_incident(c, inc["id"], state="abandoned")590 incident_event(inc["id"], m["service"], m["source"], "abandoned", "app en mauvaise santé → ROLLBACK immédiat")591 elif v == "livre":592 set_incident(c, inc["id"], state="resolved", resolved=now())593 incident_event(inc["id"], m["service"], m["source"], "resolved",594 f"effort livré ({len(commits)} commit{'s' if len(commits) > 1 else ''})")595 else:596 if commits:597 asyncio.create_task(rollback_mission(m2, f"effort verdict {v} → rollback préventif"))598 set_incident(c, inc["id"], state="abandoned")599 incident_event(inc["id"], m["service"], m["source"], "abandoned", f"effort en échec ({v})")600 return601 if not health.get("ok", True):602 # L'app est tombée → rollback immédiat, quoi qu'ait dit l'agent.603 asyncio.create_task(rollback_mission(m2, "healthcheck app en échec post-mission"))604 set_incident(c, inc["id"], state="cooldown")605 incident_event(inc["id"], m["service"], m["source"], "cooldown", "app en mauvaise santé → ROLLBACK immédiat")606 elif v == "repare" and commits:607 set_incident(c, inc["id"], state="watching")608 incident_event(inc["id"], m["service"], m["source"], "watching",609 f"réparation déclarée ({len(commits)} commit) — surveillance {POLICY['watch_window_hours']}h")610 elif v in ("echec", "inconnu") and commits:611 asyncio.create_task(rollback_mission(m2, f"verdict {v} avec commits → annulation par prudence"))612 set_incident(c, inc["id"], state="cooldown")613 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"verdict {v} → rollback préventif")614 elif v == "site_source_mort":615 set_incident(c, inc["id"], state="abandoned")616 incident_event(inc["id"], m["service"], m["source"], "abandoned", "source définitivement morte (diagnostic agent)")617 elif v in ("rien_a_faire", "repare"):618 set_incident(c, inc["id"], state="watching")619 incident_event(inc["id"], m["service"], m["source"], "watching", f"verdict {v} — surveillance")620 else:621 set_incident(c, inc["id"], state="cooldown")622 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"verdict {v}")623624625@app.get("/api/state")626def state() -> dict[str, Any]:627 with db() as c:628 incidents = [dict(r) for r in c.execute(629 "SELECT * FROM incidents ORDER BY created DESC LIMIT 200").fetchall()]630 missions = [dict(r) for r in c.execute(631 "SELECT * FROM missions ORDER BY started DESC LIMIT 100").fetchall()]632 stats = c.execute("""SELECT633 (SELECT COUNT(*) FROM missions) n_missions,634 (SELECT COUNT(*) FROM incidents WHERE state='resolved') n_resolved,635 (SELECT COUNT(*) FROM incidents WHERE state IN ('open','dispatched','fixing','watching','cooldown')) n_active,636 (SELECT COUNT(*) FROM incidents WHERE state='abandoned') n_abandoned,637 (SELECT COUNT(*) FROM events WHERE type='rollback') n_rollbacks,638 (SELECT ROUND(SUM(cost_usd),2) FROM missions) cost_total,639 (SELECT ROUND(AVG(resolved-created)/3600.0,1) FROM incidents WHERE state='resolved' AND resolved IS NOT NULL) mttr_h640 """).fetchone()641 history = [dict(r) for r in c.execute(642 "SELECT ts, mine FROM snapshots WHERE ts > ? ORDER BY ts", (now() - 7 * 86400,)).fetchall()]643 return {644 "agent": AGENT, "identity": {k: ME.get(k) for k in ("domain", "accent", "accent2", "accent_soft", "tagline", "model")},645 "services": {s: {**SERVICES[s], "node_alias": NODES[SERVICES[s]["node"]]["alias"]} for s in ME["services"]},646 "siblings": {a: {"domain": v["domain"], "accent": v["accent"], "services": v["services"]}647 for a, v in TOPO["agents"].items() if a != AGENT},648 "latest": LATEST, "incidents": incidents, "missions": missions,649 "stats": dict(stats), "history": history, "policy": POLICY, "now": now(),650 }651652653@app.get("/api/missions/{mission_id}")654def mission_detail(mission_id: str) -> dict[str, Any]:655 with db() as c:656 m = c.execute("SELECT * FROM missions WHERE id=?", (mission_id,)).fetchone()657 if not m:658 raise HTTPException(status_code=404)659 evs = [dict(r) for r in c.execute(660 "SELECT ts,type,data FROM events WHERE mission_id=? ORDER BY id", (mission_id,)).fetchall()]661 return {"mission": dict(m), "events": evs}662663664@app.get("/events")665async def sse(request: Request) -> StreamingResponse:666 q: asyncio.Queue = asyncio.Queue()667 hub.clients.add(q)668669 async def gen():670 try:671 yield f"data: {jdump({'kind': 'hello', 'agent': AGENT, 'ts': now()})}\n\n"672 while True:673 if await request.is_disconnected():674 break675 try:676 msg = await asyncio.wait_for(q.get(), timeout=25)677 yield f"data: {jdump(msg)}\n\n"678 except asyncio.TimeoutError:679 yield ": keepalive\n\n"680 finally:681 hub.clients.discard(q)682683 return StreamingResponse(gen(), media_type="text/event-stream",684 headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})685686687@app.get("/api/connectors/{service}")688def connectors_list(service: str) -> dict[str, Any]:689 """Sources connues du service (pour le menu déroulant d'enrichissement)."""690 block = LATEST["mine"].get(service) or {}691 return {"service": service,692 "connectors": sorted(693 ({"source": c["source"], "status": c["status"]} for c in block.get("connectors", [])),694 key=lambda x: x["source"])}695696697@app.post("/api/admin/effort")698async def admin_effort(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:699 """Commande un effort autonome: nouveau connecteur ou enrichissement."""700 check_token(x_ka_token)701 body = await req.json()702 kind = body.get("kind")703 service = body.get("service")704 note = (body.get("note") or "").strip()[:2000]705 if kind not in ("effort_new", "effort_enrich") or service not in SERVICES:706 raise HTTPException(status_code=400, detail="kind ou service invalide")707 if kind == "effort_enrich":708 source = (body.get("source") or "").strip()709 if not source:710 raise HTTPException(status_code=400, detail="source requise pour un enrichissement")711 else:712 source = f"nouveau·{uuid.uuid4().hex[:6]}"713 iid = uuid.uuid4().hex[:10]714 with db() as c:715 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "716 "VALUES(?,?,?,?,?,?,?,?)",717 (iid, service, source, kind, "open", now(), now(), jdump({"note": note, "kind": kind})))718 incident_event(iid, service, source, "open",719 "effort commandé: " + ("nouveau connecteur" if kind == "effort_new" else f"enrichissement de {source}"))720 return {"ok": True, "incident_id": iid}721722723@app.post("/api/admin/mission")724async def admin_mission(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:725 """Déclenche manuellement une mission sur un connecteur (même sain)."""726 check_token(x_ka_token)727 body = await req.json()728 service, source = body["service"], body["source"]729 if service not in SERVICES:730 raise HTTPException(status_code=400, detail="service inconnu")731 detail = {"source": source, "status": "manual", "message": body.get("note", "mission manuelle")}732 for conn in (LATEST["mine"].get(service) or {}).get("connectors", []):733 if conn["source"] == source:734 detail = {**conn, "status": "manual"}735 with db() as c:736 inc = active_incident(c, service, source)737 if inc is None:738 iid = uuid.uuid4().hex[:10]739 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "740 "VALUES(?,?,?,?,?,?,?,?)",741 (iid, service, source, "manual", "open", now(), now(), jdump(detail)))742 else:743 iid = inc["id"]744 set_incident(c, iid, state="open", status_detected="manual")745 incident_event(iid, service, source, "open", "mission manuelle demandée")746 return {"ok": True, "incident_id": iid}747748749@app.post("/api/admin/pause")750async def admin_pause(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:751 check_token(x_ka_token)752 body = await req.json()753 LATEST["paused"] = bool(body.get("paused", True))754 return {"ok": True, "paused": LATEST["paused"]}755756757@app.get("/health")758def health() -> dict[str, Any]:759 return {"ok": True, "agent": AGENT, "polled": LATEST["polled"], "paused": LATEST["paused"]}760761762@app.get("/favicon.svg")763def favicon() -> FileResponse:764 return FileResponse(BASE / "web" / f"favicon-{AGENT}.svg", media_type="image/svg+xml")765766767@app.get("/")768def index() -> FileResponse:769 return FileResponse(BASE / "web" / "index.html")770771772app.mount("/static", StaticFiles(directory=BASE / "web"), name="static")773774775if __name__ == "__main__":776 import uvicorn777 uvicorn.run(app, host="0.0.0.0", port=ME["port"])778