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, HTMLResponse, 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>"`. Si tu as touché plusieurs fichiers pour des raisons distinctes, fais des commits séparés et clairs.2399. POUSSE sur spbgit (le push fait partie du travail): `git push origin main`. NOTE: sur certains nœuds le push échoue en « no route to host » depuis une mission (macOS 26 Local Network Privacy) — c'est ATTENDU: le pousseur launchd (com.ka.pousseur) poussera tes commits dans les 5 minutes. Si le push échoue, signale-le simplement dans ton rapport final et passe à la suite; n'insiste pas et ne modifie JAMAIS ~/.ssh ni la config git pour « réparer » le push.240241== INTERDITS ABSOLUS ==242Toucher 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 --force` ou push d'une autre branche que main; 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.243244== FIN DE MISSION ==245Termine ta TOUTE DERNIÈRE réponse par un bloc JSON exactement de cette forme:246{{"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"]}}247Sois 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."""248249250COMMON_RULES = """== RÈGLES COMMUNES (non négociables) ==251- 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.252- Respecte l'architecture et les patterns du repo (regarde comment les connecteurs voisins sont faits AVANT d'écrire).253- Utilise le venv/node_modules du repo; jamais d'installation globale.254- INTERDIT d'inventer, stubber ou câbler des données en dur. Chaque donnée vient d'un vrai fetch.255- 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 --force` ou push d'une autre branche que main, ~/.ssh, clés API, suppression de données.256- À la fin: `pm2 restart <process de sync>`, vérifie que le site répond (curl localhost), committe en messages clairs préfixés [%AGENT%], puis POUSSE sur spbgit: `git push origin main`. Un échec « no route to host » est ATTENDU depuis une mission (macOS 26 LNP) — le pousseur launchd s'en chargera dans les 5 minutes: signale-le et passe à la suite, sans toucher à ~/.ssh ni à la config git."""257258259def effort_stack(service: str) -> str:260 svc = SERVICES[service]261 return f"""== BOÎTE À OUTILS (clés dans ~/.claude/.env) ==262- 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.263- 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).264- APIFY (APIFY_TOKEN): en dernier recours pour les plateformes dures, acteurs maison du compte (proxy résidentiel intégré).265- Regarde d'abord comment le repo de {svc['app']} utilise déjà ces services et fais pareil."""266267268def build_effort_prompt(kind: str, service: str, source: str, note: str) -> str:269 svc = SERVICES[service]270 sync_proc = next((p for p in svc["pm2"] if "sync" in p or "etl" in p), svc["pm2"][-1])271 node_alias = NODES[svc["node"]]["alias"]272 rules = COMMON_RULES.replace("%AGENT%", AGENT)273 env = f"""== TON ENVIRONNEMENT ==274Tu es sur le nœud {node_alias}, répertoire courant = repo de l'app: {svc['dir']} (remote-first, source de vérité).275L'app tourne via pm2 ({', '.join(svc['pm2'])}); process de sync: {sync_proc}; site local port {svc['web_port']}.276Commence par lire le CLAUDE.md / README du repo et docs/connecteurs/ s'ils existent."""277 if kind == "effort_degrade":278 block = LATEST["mine"].get(service) or {}279 degraded = [c for c in block.get("connectors", []) if c.get("status") == "degraded"][:25]280 deg_lines = "\n".join(281 f"- {c['source']}: dernier volume {c.get('found_last')} vs médiane {round(c.get('median_found') or 0)} "282 f"(dernier succès: {c.get('last_success')}, message: {c.get('message')})"283 for c in degraded) or "- (aucun connecteur dégradé au dernier scan — re-vérifie l'état réel dans l'app)"284 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', '')}).285{"Consigne de l'opérateur: " + note if note else ""}286287Un 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.288289== CONNECTEURS DÉGRADÉS AU DERNIER SCAN api-ka ==290{deg_lines}291292{env}293294== DÉMARCHE ==2951. 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.2962. 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>: …).2973. Pour les cas (b), ne touche à rien: consigne-les dans ton rapport final comme « baisse légitime ».2984. 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.2995. À la fin: `pm2 restart {sync_proc}`, vérifie le site (port {svc['web_port']}), puis pousse tous tes commits sur spbgit: `git push origin main`.300301{effort_stack(service)}302303{rules}304305== FIN DE MISSION ==306Termine ta TOUTE DERNIÈRE réponse par un bloc JSON:307{{"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"]}}"""308 if kind == "effort_new":309 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', '')}).310{"Consigne de l'opérateur: " + note if note else "Aucune consigne particulière: choisis la source la plus utile."}311312{env}313314== DÉMARCHE ==3151. 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).3162. 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.3173. 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.3184. 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).3195. DOCUMENTE: ajoute la fiche docs/connecteurs/<source>.md si le repo suit cette convention (URL, stratégie, pièges).3206. Redémarre {sync_proc}, vérifie le site, committe et pousse sur spbgit (`git push origin main`).321322{effort_stack(service)}323324{rules}325326== FIN DE MISSION ==327Termine ta TOUTE DERNIÈRE réponse par un bloc JSON:328{{"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"]}}"""329 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', '')}).330{"Consigne de l'opérateur: " + note if note else "Aucune consigne particulière: maximise la valeur (couverture, champs, robustesse)."}331332{env}333334== DÉMARCHE ==3351. É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.3362. 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.3373. IMPLÉMENTE proprement dans le pattern du repo, SANS casser le format de sortie existant (les consommateurs avals dépendent du schéma).3384. 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.3395. METS À JOUR la fiche docs/connecteurs/<source>.md si elle existe.3406. Redémarre {sync_proc}, vérifie le site, committe et pousse sur spbgit (`git push origin main`).341342{effort_stack(service)}343344{rules}345346== FIN DE MISSION ==347Termine ta TOUTE DERNIÈRE réponse par un bloc JSON:348{{"verdict": "livre|echec", "diagnostic": "état initial constaté", "actions": "enrichissements livrés", "test": "avant/après mesuré", "fichiers": ["fichiers modifiés"]}}"""349350351def build_site_down_prompt(service: str, detail: dict[str, Any]) -> str:352 svc = SERVICES[service]353 web_proc = svc["pm2"][0]354 node_alias = NODES[svc["node"]]["alias"]355 return f"""Tu es {AGENT}, agent gardien autonome du Groupe KA. MISSION URGENTE: le SITE {svc.get('site', '')} de l'app {svc['app']} NE RÉPOND PLUS, et le redémarrage pm2 automatique n'a pas suffi. Investigue, trouve la cause racine et remets le site en ligne.356Session 100 % AUTONOME: tu ne poses aucune question, tu travailles jusqu'au bout.357358== CE QUE LE VEILLEUR A DÉJÀ CONSTATÉ ==359- échecs consécutifs du check public: {detail.get('fails', '?')} (site muet depuis {detail.get('down_since', '?')})360- dernière erreur du check public: {detail.get('error', '?')}361- redémarrage pm2 automatique: {detail.get('restart', 'tenté')} — santé locale mesurée juste après: {jdump(detail.get('health')) if detail.get('health') else 'inconnue'}362363== TON ENVIRONNEMENT ==364Tu es sur le nœud {node_alias}, répertoire courant = repo de l'app: {svc['dir']} (remote-first, source de vérité).365Process pm2 de l'app: {', '.join(svc['pm2'])} (web: {web_proc}); le site local doit répondre sur http://localhost:{svc['web_port']}/ et le site public est {svc.get('site', '')} (tunnel ngrok sur ce nœud, souvent un process pm2 `<app>-ngrok` ou un service launchd).366367== DÉMARCHE IMPOSÉE (dans l'ordre) ==3681. CONSTATE: `curl -s -o /dev/null -w '%{{http_code}}' http://localhost:{svc['web_port']}/` puis la même chose sur {svc.get('site', 'le site public')}. `pm2 jlist` / `pm2 describe {web_proc}`: statut, nombre de restarts, uptime, mémoire.3692. LIS LES LOGS: `pm2 logs {web_proc} --nostream --lines 300` (stdout ET err) + dossier logs/ du repo. Cherche: crash en boucle, port déjà occupé (EADDRINUSE), exception au démarrage, OOM, module manquant, build corrompu, base de données verrouillée/corrompue, disque plein (`df -h`).3703. DISTINGUE les 3 familles de panne:371 a. Process web mort ou en boucle de crash → cause racine dans le code/l'env → corrige puis `pm2 restart {web_proc}`.372 b. Process online mais localhost:{svc['web_port']} muet ou en erreur → mauvais port/bind, build cassé → rebuild si le repo a une étape de build (npm run build…), puis redémarre.373 c. localhost OK mais site public muet → le tunnel: trouve le process (pm2 `*-ngrok` de CETTE app ou launchd) et relance-le (`pm2 restart <app>-ngrok` ou `launchctl kickstart -k gui/$(id -u)/<label>`). NE MODIFIE JAMAIS sa config ni son domaine.3744. CORRIGE la cause racine de façon minimale (pas de rustine qui masque le symptôme). Si des données/DB sont corrompues, répare prudemment SANS perte de données (copie de sauvegarde avant toute opération risquée).3755. VÉRIFIE pour de vrai: localhost ET le site public doivent répondre 200/3xx de façon STABLE (2 mesures espacées de 30 s).3766. Si tu as modifié du code ou de la config du repo: `git add -A && git commit -m "[{AGENT}] fix site down {service}: <cause>"` puis `git push origin main`. Un échec « no route to host » au push est ATTENDU (macOS 26 LNP) — le pousseur launchd s'en charge dans les 5 minutes: signale-le et passe à la suite, sans toucher à ~/.ssh ni à la config git.377378== INTERDITS ABSOLUS ==379Toucher aux autres apps du nœud; MODIFIER la config pm2/ngrok/launchd (relancer un process existant de CETTE app est permis, changer sa config non); supprimer des données; `git push --force` ou push d'une autre branche que main; ~/.ssh; clés API. Si la panne dépasse l'app (disque plein système, nœud malade), libère de l'espace UNIQUEMENT dans le repo de l'app (logs, caches, artefacts de build) et documente le reste dans ton diagnostic.380381== FIN DE MISSION ==382Termine ta TOUTE DERNIÈRE réponse par un bloc JSON exactement de cette forme:383{{"verdict": "repare|echec|rien_a_faire", "diagnostic": "cause racine en 1-2 phrases", "actions": "ce que tu as fait", "test": "codes HTTP local + public mesurés", "fichiers": ["fichiers modifiés"]}}384Sois honnête: « repare » exige que le SITE PUBLIC réponde réellement — le veilleur re-vérifie aux 2 minutes et un faux « repare » sera rollback."""385386387async def dispatch(incident: sqlite3.Row, health: dict[str, Any]) -> None:388 service, source, iid = incident["service"], incident["source"], incident["id"]389 svc = SERVICES[service]390 node = svc["node"]391 ip = NODES[node]["lan_ip"]392 mid = uuid.uuid4().hex[:12]393 my_ip = os.environ.get("KA_GUARDIAN_SELF_IP", "192.168.2.69")394 kind = incident["status_detected"]395 max_cost = None396 if kind in ("effort_new", "effort_enrich", "effort_degrade"):397 prompt = build_effort_prompt(kind, service, source, health.get("note", ""))398 max_turns = POLICY.get("effort_max_turns", 150)399 timeout = POLICY.get("effort_timeout_seconds", 7200)400 # Plafonds choisis par l'opérateur au moment de commander l'effort.401 if health.get("max_minutes"):402 timeout = min(timeout, int(health["max_minutes"]) * 60)403 if health.get("max_cost_usd"):404 max_cost = float(health["max_cost_usd"])405 elif kind == "site_down":406 prompt = build_site_down_prompt(service, health)407 max_turns, timeout = POLICY["mission_max_turns"], POLICY["mission_timeout_seconds"]408 else:409 prompt = build_prompt(service, source, health)410 max_turns, timeout = POLICY["mission_max_turns"], POLICY["mission_timeout_seconds"]411 payload = {412 "mission_id": mid, "agent": AGENT, "service": service, "source": source,413 "dir": svc["dir"], "pm2": svc["pm2"], "web_port": svc["web_port"],414 "model": ME.get("model", "sonnet"),415 "max_turns": max_turns,416 "timeout_seconds": timeout,417 "max_cost_usd": max_cost,418 "prompt": prompt,419 "callback_url": f"http://{my_ip}:{ME['port']}/api/ingest/{mid}",420 }421 try:422 code, body = await lan_post(ip, RUNNER_PORT, "/missions", payload, timeout=25)423 if code == 409:424 return # runner occupé, on retentera au prochain tick425 if code != 200:426 raise ConnectionError(f"runner {code}: {body[:200]}")427 except Exception as exc:428 print(f"[dispatch] {service}/{source} → {type(exc).__name__}: {exc}", flush=True)429 hub.publish_sync({"kind": "log", "level": "error", "msg": f"dispatch {source}: {exc}", "ts": now()})430 return431 with db() as c:432 c.execute("INSERT INTO missions(id,incident_id,service,source,node,state,started) VALUES(?,?,?,?,?,?,?)",433 (mid, iid, service, source, node, "running", now()))434 set_incident(c, iid, state="fixing", attempts=incident["attempts"] + 1)435 incident_event(iid, service, source, "fixing", f"mission {mid} dépêchée sur {NODES[node]['alias']}")436 hub.publish_sync({"kind": "mission", "mission_id": mid, "service": service,437 "source": source, "state": "running", "ts": now()})438439440async def rollback_mission(mission: sqlite3.Row, reason: str) -> dict[str, Any]:441 svc = SERVICES[mission["service"]]442 ip = NODES[mission["node"]]["lan_ip"]443 try:444 code, body = await lan_post(ip, RUNNER_PORT, "/rollback",445 {"dir": svc["dir"], "base_commit": mission["base_commit"],446 "pm2": svc["pm2"], "web_port": svc["web_port"]}, timeout=300)447 out = json.loads(body) if code == 200 else {"ok": False, "erreur": f"{code}: {body[:200]}"}448 except Exception as exc:449 out = {"ok": False, "erreur": str(exc)}450 hub.publish_sync({"kind": "rollback", "mission_id": mission["id"], "service": mission["service"],451 "source": mission["source"], "reason": reason, "result": out, "ts": now()})452 with db() as c:453 c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",454 (mission["id"], now(), "rollback", jdump({"reason": reason, "result": out})))455 return out456457458# --------------------------------------------------------------- moteur ----459460async def poll_once() -> None:461 async with httpx.AsyncClient() as cl:462 r = await cl.get(TOPO["apika_monitoring_url"], timeout=15)463 data = r.json()["data"]464 mine: dict[str, Any] = {}465 for service, block in data["services"].items():466 assigned_to_me = service in MY_SERVICES or (DEFAULT_AGENT and service not in ASSIGNED)467 if assigned_to_me:468 mine[service] = block469 LATEST.update({"mine": mine, "ecosystem": data["summary"], "polled": now()})470 with db() as c:471 summary_mine = {s: b["summary"] for s, b in mine.items()}472 c.execute("INSERT OR REPLACE INTO snapshots(ts,mine,ecosystem) VALUES(?,?,?)",473 (now(), jdump(summary_mine), jdump(data["summary"])))474 c.execute("DELETE FROM snapshots WHERE ts < ?", (now() - 30 * 86400,))475 hub.publish_sync({"kind": "snapshot", "mine": {s: b["summary"] for s, b in mine.items()},476 "ecosystem": data["summary"], "ts": now()})477478 trigger = set(POLICY["trigger_statuses"])479 with db() as c:480 for service, block in mine.items():481 if service not in SERVICES:482 continue # service inconnu de la topologie: visible au dashboard, pas d'action483 to_create: list[tuple[str, str, dict[str, Any]]] = []484 for conn in block.get("connectors", []):485 source, status = conn["source"], conn["status"]486 inc = active_incident(c, service, source)487 if status in trigger and inc is None:488 # Garde anti-boucle : une source déjà diagnostiquée irréparable489 # (site_source_mort, tentatives épuisées) ne redonne pas490 # d'incident tant que le délai de re-examen n'est pas écoulé.491 # Sans elle, un connecteur désactivé côté app mais rapporté492 # broken à perpétuité par api-ka relance une mission toutes493 # les ~10 min (2026-08-24 : 37 missions sur origin-north.ai).494 ab = c.execute(495 "SELECT updated FROM incidents WHERE service=? AND source=? "496 "AND state='abandoned' ORDER BY updated DESC LIMIT 1",497 (service, source)).fetchone()498 if ab and now() - ab["updated"] < POLICY.get("abandoned_retry_hours", 168) * 3600:499 continue500 to_create.append((source, status, conn))501 elif status in trigger and inc is not None and inc["state"] in ("open", "cooldown"):502 # Ne pas rafraîchir watching/fixing: `updated` sert de chrono503 # à la fenêtre de surveillance et au cooldown.504 c.execute("UPDATE incidents SET status_detected=?, detail=? WHERE id=?",505 (status, jdump(conn), inc["id"]))506 elif status == "ok" and inc is not None:507 if inc["state"] == "watching":508 set_incident(c, inc["id"], state="resolved", resolved=now())509 incident_event(inc["id"], service, source, "resolved", "connecteur de retour à ok — réparation confirmée")510 elif inc["state"] in ("open", "cooldown"):511 set_incident(c, inc["id"], state="self_healed", resolved=now())512 incident_event(inc["id"], service, source, "self_healed", "revenu à ok sans intervention")513 elif status == "retired" and inc is not None and inc["state"] in ("open", "cooldown", "watching"):514 # api-ka a retiré la source de la supervision (désactivée515 # côté app ou disparue du journal) : plus rien à réparer.516 set_incident(c, inc["id"], state="abandoned")517 incident_event(inc["id"], service, source, "abandoned", "source retirée de la supervision api-ka")518519 # Damper d'événement de masse : des centaines de sources qui520 # basculent d'un coup signalent une panne GLOBALE de l'app (sync521 # mort, /api/stats cassé), pas autant de pannes individuelles —522 # missionner source par source serait long et coûteux pour rien523 # (2026-08-24 : 794 incidents jobka créés après ~17 h d'arrêt de524 # job-ka-sync). On n'ouvre rien : la panne d'app se voit au525 # dashboard (bloc _app) et les sources encore cassées une fois la526 # vague retombée sous le seuil recevront leurs incidents.527 limit = POLICY.get("mass_incident_threshold", 20)528 if len(to_create) > limit:529 msg = (f"vague de {len(to_create)} connecteurs en panne sur {service} "530 f"(seuil {limit}) — panne globale probable de l'app, aucun incident créé")531 print(f"[poll] {msg}", flush=True)532 hub.publish_sync({"kind": "log", "level": "error", "msg": msg, "ts": now()})533 continue534 for source, status, conn in to_create:535 iid = uuid.uuid4().hex[:10]536 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "537 "VALUES(?,?,?,?,?,?,?,?)",538 (iid, service, source, status, "open", now(), now(), jdump(conn)))539 incident_event(iid, service, source, "open", f"détecté {status}")540541542async def tick() -> None:543 if LATEST["paused"]:544 return545 try:546 await poll_once()547 except Exception as exc:548 hub.publish_sync({"kind": "log", "level": "warn", "msg": f"poll api-ka échoué: {exc}", "ts": now()})549550 with db() as c:551 # 1. Réconciliation des missions en cours depuis > 10 min: on demande au552 # runner ce qu'il en est (événement final perdu? runner redémarré?).553 for m in c.execute("SELECT id FROM missions WHERE state='running' AND started < ?",554 (now() - 600,)).fetchall():555 if m["id"] not in RECONCILING:556 RECONCILING.add(m["id"])557 asyncio.create_task(reconcile_mission(m["id"]))558559 # 2. Watching → rollback si la fenêtre est passée et toujours cassé560 # (fenêtre courte pour les incidents de site: le veilleur mesure aux561 # 2 minutes, pas besoin d'attendre le rythme des scans api-ka)562 for inc in c.execute("SELECT * FROM incidents WHERE state='watching'").fetchall():563 window_h = POLICY.get("site_watch_window_hours", 1) if inc["source"] == "_site" else POLICY["watch_window_hours"]564 if now() - inc["updated"] < window_h * 3600:565 continue566 status = current_status(inc["service"], inc["source"])567 m = c.execute("SELECT * FROM missions WHERE incident_id=? AND base_commit IS NOT NULL "568 "ORDER BY started DESC LIMIT 1", (inc["id"],)).fetchone()569 if status in POLICY["trigger_statuses"] or status == "degraded":570 if m and (json.loads(m["commits"] or "[]")):571 asyncio.create_task(rollback_mission(m, "toujours cassé après la fenêtre de surveillance"))572 set_incident(c, inc["id"], state="cooldown")573 incident_event(inc["id"], inc["service"], inc["source"], "cooldown",574 f"non guéri après {POLICY['watch_window_hours']}h → rollback + cooldown")575 elif status == "ok":576 set_incident(c, inc["id"], state="resolved", resolved=now())577 incident_event(inc["id"], inc["service"], inc["source"], "resolved", "confirmé ok")578579 # 3. Cooldown expiré → réouverture ou abandon580 for inc in c.execute("SELECT * FROM incidents WHERE state='cooldown'").fetchall():581 cool_h = POLICY.get("site_cooldown_hours", 1) if inc["source"] == "_site" else POLICY["attempt_cooldown_hours"]582 if now() - inc["updated"] < cool_h * 3600:583 continue584 status = current_status(inc["service"], inc["source"])585 if status == "ok":586 set_incident(c, inc["id"], state="resolved", resolved=now())587 incident_event(inc["id"], inc["service"], inc["source"], "resolved", "guéri pendant le cooldown")588 elif inc["attempts"] >= POLICY["max_attempts_per_incident"]:589 set_incident(c, inc["id"], state="abandoned")590 incident_event(inc["id"], inc["service"], inc["source"], "abandoned",591 f"{inc['attempts']} tentatives épuisées — intervention humaine requise")592 else:593 set_incident(c, inc["id"], state="open")594 incident_event(inc["id"], inc["service"], inc["source"], "open", "cooldown terminé, nouvelle tentative")595596 # 4. Dispatch (1 mission à la fois par agent, broken avant stale, plus vieux d'abord)597 running = c.execute("SELECT COUNT(*) n FROM missions WHERE state='running'").fetchone()["n"]598 if running < POLICY["max_concurrent_missions"]:599 busy_nodes = {m["node"] for m in c.execute("SELECT node FROM missions WHERE state='running'").fetchall()}600 nxt = c.execute(601 "SELECT * FROM incidents WHERE state='open' "602 "ORDER BY CASE status_detected WHEN 'site_down' THEN 0 WHEN 'manual' THEN 0 "603 "WHEN 'effort_new' THEN 0 WHEN 'effort_enrich' THEN 0 WHEN 'broken' THEN 1 ELSE 2 END, created "604 ).fetchall()605 for inc in nxt:606 if inc["service"] not in SERVICES:607 continue608 if SERVICES[inc["service"]]["node"] in busy_nodes:609 continue610 set_incident(c, inc["id"], state="dispatched")611 asyncio.create_task(dispatch_row(inc["id"]))612 break613614615def current_status(service: str, source: str) -> str:616 if source == "_site":617 st = SITE.get(service) or {}618 if not st.get("checked"):619 return "inconnu"620 return "ok" if st.get("ok") else "broken"621 block = LATEST["mine"].get(service) or {}622 for conn in block.get("connectors", []):623 if conn["source"] == source:624 return conn["status"]625 return "inconnu"626627628async def dispatch_row(iid: str) -> None:629 with db() as c:630 inc = c.execute("SELECT * FROM incidents WHERE id=?", (iid,)).fetchone()631 if inc:632 await dispatch(inc, json.loads(inc["detail"] or "{}"))633 # dispatch() remet fixing; si l'appel a échoué/409, on relâche634 with db() as c:635 cur = c.execute("SELECT state FROM incidents WHERE id=?", (iid,)).fetchone()636 if cur and cur["state"] == "dispatched":637 set_incident(c, iid, state="open")638639640RECONCILING: set[str] = set()641642643async def reconcile_mission(mid: str) -> None:644 """Résout une mission « en cours » suspecte auprès de son runner."""645 try:646 with db() as c:647 m = c.execute("SELECT * FROM missions WHERE id=? AND state='running'", (mid,)).fetchone()648 if not m:649 return650 ip = NODES[m["node"]]["lan_ip"]651 try:652 code, body = await lan_post(ip, RUNNER_PORT, "/mission-result", {"mission_id": mid}, timeout=20)653 resp = json.loads(body) if code == 200 else {"status": "erreur"}654 except Exception:655 return # courrier/runner injoignable: on retentera au prochain tick656 if resp.get("status") == "running":657 return658 with db() as c:659 m = c.execute("SELECT * FROM missions WHERE id=? AND state='running'", (mid,)).fetchone()660 if not m:661 return662 if resp.get("status") == "done":663 data = (resp.get("final") or {}).get("data", {})664 c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",665 (mid, now(), "final", jdump(data)))666 await finalize(c, m, "final", data)667 hub.publish_sync({"kind": "mission_event", "mission_id": mid, "type": "final",668 "data": data, "ts": now()})669 elif now() - m["started"] > 900:670 # Le runner ne connaît pas cette mission: tuée en vol671 # (redémarrage). On la clôt et on relance l'incident.672 c.execute("UPDATE missions SET state='error', ended=? WHERE id=?", (now(), mid))673 inc = c.execute("SELECT * FROM incidents WHERE id=?", (m["incident_id"],)).fetchone()674 if inc and inc["state"] in ("fixing", "dispatched"):675 set_incident(c, inc["id"], state="open")676 incident_event(inc["id"], m["service"], m["source"], "open",677 "mission perdue (runner interrompu) — remise en file")678 finally:679 RECONCILING.discard(mid)680681682async def engine() -> None:683 global LOOP684 LOOP = asyncio.get_running_loop()685 await asyncio.sleep(3)686 while True:687 try:688 await tick()689 except Exception as exc:690 import traceback691 traceback.print_exc()692 hub.publish_sync({"kind": "log", "level": "error", "msg": f"tick: {exc}", "ts": now()})693 await asyncio.sleep(POLICY["poll_interval_seconds"])694695696# ----------------------------------------------------- veilleur de sites ----697# Surveillance du SITE WEB de chaque service assigné (indépendante des698# connecteurs api-ka). Down confirmé → pm2 restart automatique via le runner699# du nœud; toujours down → incident « site_down » et mission d'investigation.700701SITE: dict[str, dict[str, Any]] = {} # service → état du dernier check public702703704async def check_site(url: str) -> tuple[bool, str]:705 try:706 async with httpx.AsyncClient(follow_redirects=True, timeout=15) as cl:707 r = await cl.get(url, headers={"User-Agent": f"ka-guardian-{AGENT}/1.0"})708 return r.status_code < 500, f"HTTP {r.status_code}"709 except Exception as exc:710 return False, f"{type(exc).__name__}: {exc}"711712713async def runner_restart(service: str) -> dict[str, Any]:714 """pm2 restart de l'app via le runner de son nœud (endpoint /restart)."""715 svc = SERVICES[service]716 ip = NODES[svc["node"]]["lan_ip"]717 try:718 code, body = await lan_post(ip, RUNNER_PORT, "/restart",719 {"pm2": svc["pm2"], "web_port": svc["web_port"]}, timeout=240)720 return json.loads(body) if code == 200 else {"ok": False, "erreur": f"runner {code}: {body[:200]}"}721 except Exception as exc:722 return {"ok": False, "erreur": f"{type(exc).__name__}: {exc}"}723724725async def site_check_service(service: str) -> None:726 svc = SERVICES[service]727 url = svc.get("site")728 if not url:729 return730 up, info = await check_site(url)731 if not up: # contre-mesure avant de compter l'échec (blip réseau/ngrok)732 await asyncio.sleep(5)733 up, info = await check_site(url)734 st = SITE.setdefault(service, {"fails": 0, "ok": True, "down_since": None, "last_restart": 0.0})735 st.update({"checked": now(), "ok": up, "info": info})736 if up:737 if st["fails"]:738 hub.publish_sync({"kind": "log", "level": "info",739 "msg": f"site {url} de retour en ligne ({info})", "ts": now()})740 st["fails"], st["down_since"] = 0, None741 with db() as c:742 inc = active_incident(c, service, "_site")743 if inc and inc["state"] in ("open", "cooldown"):744 set_incident(c, inc["id"], state="self_healed", resolved=now())745 incident_event(inc["id"], service, "_site", "self_healed", f"site de retour en ligne ({info})")746 return747 st["fails"] += 1748 st["down_since"] = st["down_since"] or now()749 need = POLICY.get("site_check_fails", 2)750 hub.publish_sync({"kind": "log", "level": "warn",751 "msg": f"site {url} muet ({info}) — échec {st['fails']}/{need}", "ts": now()})752 if st["fails"] < need:753 return754 with db() as c:755 inc = active_incident(c, service, "_site")756 if inc and inc["state"] in ("dispatched", "fixing"):757 return # une mission d'investigation travaille déjà sur cette app758 # Étape 1: pm2 restart automatique (throttlé pour ne pas marteler l'app)759 restart_res: dict[str, Any] = {"ok": False, "erreur": "non tenté (throttle)"}760 if now() - st["last_restart"] >= POLICY.get("site_restart_throttle_seconds", 600):761 st["last_restart"] = now()762 hub.publish_sync({"kind": "log", "level": "warn",763 "msg": f"{svc['app']}: site down confirmé → pm2 restart automatique ({', '.join(svc['pm2'])})",764 "ts": now()})765 restart_res = await runner_restart(service)766 await asyncio.sleep(POLICY.get("site_restart_wait_seconds", 20))767 up, info = await check_site(url)768 if up:769 st.update({"ok": True, "fails": 0, "down_since": None, "info": info})770 with db() as c:771 if inc:772 set_incident(c, inc["id"], state="resolved", resolved=now())773 incident_event(inc["id"], service, "_site", "resolved", f"site rétabli par pm2 restart ({info})")774 else:775 # Trace au dashboard: le restart automatique a suffi.776 iid = uuid.uuid4().hex[:10]777 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,resolved,detail) "778 "VALUES(?,?,?,?,?,?,?,?,?)",779 (iid, service, "_site", "site_down", "resolved", now(), now(), now(),780 jdump({"note": "rétabli par pm2 restart automatique", "restart": restart_res})))781 incident_event(iid, service, "_site", "resolved", f"site rétabli par pm2 restart automatique ({info})")782 return783 # Étape 2: toujours down → incident + mission d'investigation (si aucun actif)784 if inc is not None:785 return # open/watching: la machinerie d'incidents suit déjà son cours786 with db() as c:787 # Garde anti-boucle (même logique que les connecteurs, délai plus court:788 # un site down est urgent, on re-tente quand même chaque jour).789 ab = c.execute("SELECT updated FROM incidents WHERE service=? AND source='_site' "790 "AND state='abandoned' ORDER BY updated DESC LIMIT 1", (service,)).fetchone()791 if ab and now() - ab["updated"] < POLICY.get("site_abandoned_retry_hours", 24) * 3600:792 return793 detail = {"fails": st["fails"],794 "down_since": time.strftime("%Y-%m-%d %H:%M", time.localtime(st["down_since"])),795 "error": info,796 "restart": "pm2 restart exécuté, site toujours muet" if restart_res.get("ok")797 else f"restart en échec: {restart_res.get('erreur') or restart_res}",798 "health": restart_res.get("health")}799 iid = uuid.uuid4().hex[:10]800 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "801 "VALUES(?,?,?,?,?,?,?,?)",802 (iid, service, "_site", "site_down", "open", now(), now(), jdump(detail)))803 incident_event(iid, service, "_site", "open",804 f"site hors ligne malgré le pm2 restart automatique ({info}) — mission d'investigation demandée")805806807async def site_engine() -> None:808 await asyncio.sleep(10)809 while True:810 if not LATEST["paused"]:811 for service in sorted(MY_SERVICES & set(SERVICES)):812 try:813 await site_check_service(service)814 except Exception as exc:815 hub.publish_sync({"kind": "log", "level": "warn",816 "msg": f"veilleur site {service}: {exc}", "ts": now()})817 await asyncio.sleep(POLICY.get("site_check_interval_seconds", 120))818819820# ------------------------------------------------------------------- app ---821822app = FastAPI(title=f"KA Guardian — {AGENT}", docs_url=None, redoc_url=None)823824825@app.on_event("startup")826async def startup() -> None:827 init_db()828 asyncio.create_task(engine())829 asyncio.create_task(site_engine())830831832def check_token(tok: str | None) -> None:833 if not TOKEN or tok != TOKEN:834 raise HTTPException(status_code=401)835836837@app.post("/api/ingest/{mission_id}")838async def ingest(mission_id: str, req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:839 check_token(x_ka_token)840 ev = await req.json()841 etype, data = ev.get("type"), ev.get("data", {})842 with db() as c:843 c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",844 (mission_id, now(), etype, jdump(data)))845 m = c.execute("SELECT * FROM missions WHERE id=?", (mission_id,)).fetchone()846 if m:847 if etype == "start":848 c.execute("UPDATE missions SET base_commit=? WHERE id=?", (data.get("base_commit"), mission_id))849 elif etype in ("final", "error"):850 await finalize(c, m, etype, data)851 hub.publish_sync({"kind": "mission_event", "mission_id": mission_id, "type": etype,852 "data": data, "ts": now()})853 return {"ok": True}854855856async def finalize(c: sqlite3.Connection, m: sqlite3.Row, etype: str, data: dict[str, Any]) -> None:857 inc = c.execute("SELECT * FROM incidents WHERE id=?", (m["incident_id"],)).fetchone()858 if etype == "error":859 c.execute("UPDATE missions SET state='error', ended=? WHERE id=?", (now(), m["id"]))860 if inc:861 set_incident(c, inc["id"], state="cooldown")862 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"erreur mission: {data.get('error')}")863 return864 verdict = data.get("verdict", {})865 commits = data.get("commits", [])866 health = data.get("health", {})867 c.execute("UPDATE missions SET state='done', ended=?, commits=?, verdict=?, cost_usd=?, num_turns=?, health=?, base_commit=COALESCE(base_commit,?) WHERE id=?",868 (now(), jdump(commits), jdump(verdict), data.get("cost_usd"),869 data.get("num_turns"), jdump(health), data.get("base_commit"), m["id"]))870 m2 = c.execute("SELECT * FROM missions WHERE id=?", (m["id"],)).fetchone()871 v = verdict.get("verdict", "inconnu")872 if not inc:873 return874 if inc["status_detected"] in ("effort_new", "effort_enrich", "effort_degrade"):875 # Les efforts commandés ne passent pas par la surveillance api-ka:876 # verdict + santé de l'app décident tout de suite.877 if not health.get("ok", True):878 asyncio.create_task(rollback_mission(m2, "healthcheck app en échec post-effort"))879 set_incident(c, inc["id"], state="abandoned")880 incident_event(inc["id"], m["service"], m["source"], "abandoned", "app en mauvaise santé → ROLLBACK immédiat")881 elif v == "livre":882 set_incident(c, inc["id"], state="resolved", resolved=now())883 incident_event(inc["id"], m["service"], m["source"], "resolved",884 f"effort livré ({len(commits)} commit{'s' if len(commits) > 1 else ''})")885 elif v == "echec" and commits:886 asyncio.create_task(rollback_mission(m2, "effort en échec déclaré → rollback préventif"))887 set_incident(c, inc["id"], state="abandoned")888 incident_event(inc["id"], m["service"], m["source"], "abandoned", "effort en échec → rollback")889 else:890 # Session interrompue (plafond durée/coût) ou verdict illisible:891 # l'app est saine, les commits incrémentaux testés sont conservés.892 set_incident(c, inc["id"], state="abandoned")893 incident_event(inc["id"], m["service"], m["source"], "abandoned",894 f"effort terminé sans verdict livré ({v}) — {len(commits)} commit(s) conservé(s), app saine")895 return896 if not health.get("ok", True):897 # L'app est tombée → rollback immédiat, quoi qu'ait dit l'agent.898 asyncio.create_task(rollback_mission(m2, "healthcheck app en échec post-mission"))899 set_incident(c, inc["id"], state="cooldown")900 incident_event(inc["id"], m["service"], m["source"], "cooldown", "app en mauvaise santé → ROLLBACK immédiat")901 elif v == "repare" and commits:902 set_incident(c, inc["id"], state="watching")903 incident_event(inc["id"], m["service"], m["source"], "watching",904 f"réparation déclarée ({len(commits)} commit) — surveillance {POLICY['watch_window_hours']}h")905 elif v in ("echec", "inconnu") and commits:906 asyncio.create_task(rollback_mission(m2, f"verdict {v} avec commits → annulation par prudence"))907 set_incident(c, inc["id"], state="cooldown")908 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"verdict {v} → rollback préventif")909 elif v == "site_source_mort":910 set_incident(c, inc["id"], state="abandoned")911 incident_event(inc["id"], m["service"], m["source"], "abandoned", "source définitivement morte (diagnostic agent)")912 elif v in ("rien_a_faire", "repare"):913 set_incident(c, inc["id"], state="watching")914 incident_event(inc["id"], m["service"], m["source"], "watching", f"verdict {v} — surveillance")915 else:916 set_incident(c, inc["id"], state="cooldown")917 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"verdict {v}")918919920@app.get("/api/state")921def state() -> dict[str, Any]:922 with db() as c:923 incidents = [dict(r) for r in c.execute(924 "SELECT * FROM incidents ORDER BY created DESC LIMIT 200").fetchall()]925 missions = [dict(r) for r in c.execute(926 "SELECT * FROM missions ORDER BY started DESC LIMIT 100").fetchall()]927 stats = c.execute("""SELECT928 (SELECT COUNT(*) FROM missions) n_missions,929 (SELECT COUNT(*) FROM incidents WHERE state='resolved') n_resolved,930 (SELECT COUNT(*) FROM incidents WHERE state IN ('open','dispatched','fixing','watching','cooldown')) n_active,931 (SELECT COUNT(*) FROM incidents WHERE state='abandoned') n_abandoned,932 (SELECT COUNT(*) FROM events WHERE type='rollback') n_rollbacks,933 (SELECT ROUND(SUM(cost_usd),2) FROM missions) cost_total,934 (SELECT ROUND(AVG(resolved-created)/3600.0,1) FROM incidents WHERE state='resolved' AND resolved IS NOT NULL) mttr_h935 """).fetchone()936 history = [dict(r) for r in c.execute(937 "SELECT ts, mine FROM snapshots WHERE ts > ? ORDER BY ts", (now() - 7 * 86400,)).fetchall()]938 return {939 "agent": AGENT, "identity": {k: ME.get(k) for k in ("domain", "accent", "accent2", "accent_soft", "tagline", "model")},940 "services": {s: {**SERVICES[s], "node_alias": NODES[SERVICES[s]["node"]]["alias"]} for s in ME["services"]},941 "siblings": {a: {"domain": v["domain"], "accent": v["accent"], "services": v["services"]}942 for a, v in TOPO["agents"].items() if a != AGENT},943 "latest": LATEST, "sites": SITE, "incidents": incidents, "missions": missions,944 "stats": dict(stats), "history": history, "policy": POLICY, "now": now(),945 }946947948@app.get("/api/missions/{mission_id}")949def mission_detail(mission_id: str) -> dict[str, Any]:950 with db() as c:951 m = c.execute("SELECT * FROM missions WHERE id=?", (mission_id,)).fetchone()952 if not m:953 raise HTTPException(status_code=404)954 evs = [dict(r) for r in c.execute(955 "SELECT ts,type,data FROM events WHERE mission_id=? ORDER BY id", (mission_id,)).fetchall()]956 return {"mission": dict(m), "events": evs}957958959@app.get("/events")960async def sse(request: Request) -> StreamingResponse:961 q: asyncio.Queue = asyncio.Queue()962 hub.clients.add(q)963964 async def gen():965 try:966 yield f"data: {jdump({'kind': 'hello', 'agent': AGENT, 'ts': now()})}\n\n"967 while True:968 if await request.is_disconnected():969 break970 try:971 msg = await asyncio.wait_for(q.get(), timeout=25)972 yield f"data: {jdump(msg)}\n\n"973 except asyncio.TimeoutError:974 yield ": keepalive\n\n"975 finally:976 hub.clients.discard(q)977978 return StreamingResponse(gen(), media_type="text/event-stream",979 headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})980981982@app.get("/api/connectors/{service}")983def connectors_list(service: str) -> dict[str, Any]:984 """Sources connues du service (pour le menu déroulant d'enrichissement)."""985 block = LATEST["mine"].get(service) or {}986 return {"service": service,987 "connectors": sorted(988 ({"source": c["source"], "status": c["status"]} for c in block.get("connectors", [])),989 key=lambda x: x["source"])}990991992@app.post("/api/admin/effort")993async def admin_effort(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:994 """Commande un effort autonome: nouveau connecteur ou enrichissement."""995 check_token(x_ka_token)996 body = await req.json()997 kind = body.get("kind")998 service = body.get("service")999 note = (body.get("note") or "").strip()[:2000]1000 if kind not in ("effort_new", "effort_enrich", "effort_degrade") or service not in SERVICES:1001 raise HTTPException(status_code=400, detail="kind ou service invalide")1002 if kind == "effort_enrich":1003 source = (body.get("source") or "").strip()1004 if not source:1005 raise HTTPException(status_code=400, detail="source requise pour un enrichissement")1006 elif kind == "effort_degrade":1007 source = f"dégradés·{uuid.uuid4().hex[:6]}"1008 else:1009 source = f"nouveau·{uuid.uuid4().hex[:6]}"1010 detail: dict[str, Any] = {"note": note, "kind": kind}1011 try:1012 if body.get("max_minutes"):1013 detail["max_minutes"] = max(10, min(240, int(body["max_minutes"])))1014 if body.get("max_cost_usd"):1015 detail["max_cost_usd"] = max(0.5, min(100.0, float(body["max_cost_usd"])))1016 except (TypeError, ValueError):1017 raise HTTPException(status_code=400, detail="max_minutes/max_cost_usd invalides")1018 iid = uuid.uuid4().hex[:10]1019 with db() as c:1020 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "1021 "VALUES(?,?,?,?,?,?,?,?)",1022 (iid, service, source, kind, "open", now(), now(), jdump(detail)))1023 labels = {"effort_new": "nouveau connecteur", "effort_enrich": f"enrichissement de {source}",1024 "effort_degrade": "inspection des connecteurs dégradés"}1025 incident_event(iid, service, source, "open",1026 f"effort commandé: {labels[kind]}"1027 + (f" · ≤{detail['max_minutes']} min" if detail.get("max_minutes") else "")1028 + (f" · ≤{detail['max_cost_usd']} $" if detail.get("max_cost_usd") else ""))1029 return {"ok": True, "incident_id": iid}103010311032@app.post("/api/admin/mission")1033async def admin_mission(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:1034 """Déclenche manuellement une mission sur un connecteur (même sain)."""1035 check_token(x_ka_token)1036 body = await req.json()1037 service, source = body["service"], body["source"]1038 if service not in SERVICES:1039 raise HTTPException(status_code=400, detail="service inconnu")1040 detail = {"source": source, "status": "manual", "message": body.get("note", "mission manuelle")}1041 for conn in (LATEST["mine"].get(service) or {}).get("connectors", []):1042 if conn["source"] == source:1043 detail = {**conn, "status": "manual"}1044 with db() as c:1045 inc = active_incident(c, service, source)1046 if inc is None:1047 iid = uuid.uuid4().hex[:10]1048 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "1049 "VALUES(?,?,?,?,?,?,?,?)",1050 (iid, service, source, "manual", "open", now(), now(), jdump(detail)))1051 else:1052 iid = inc["id"]1053 set_incident(c, iid, state="open", status_detected="manual")1054 incident_event(iid, service, source, "open", "mission manuelle demandée")1055 return {"ok": True, "incident_id": iid}105610571058@app.post("/api/admin/pause")1059async def admin_pause(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:1060 check_token(x_ka_token)1061 body = await req.json()1062 LATEST["paused"] = bool(body.get("paused", True))1063 return {"ok": True, "paused": LATEST["paused"]}106410651066@app.get("/health")1067def health() -> dict[str, Any]:1068 return {"ok": True, "agent": AGENT, "polled": LATEST["polled"], "paused": LATEST["paused"]}106910701071@app.get("/favicon.svg")1072def favicon() -> FileResponse:1073 return FileResponse(BASE / "web" / f"favicon-{AGENT}.svg", media_type="image/svg+xml")107410751076@app.get("/og.png")1077def og_image() -> FileResponse:1078 """Bannière de partage (Open Graph) propre à l'agent."""1079 return FileResponse(BASE / "web" / f"og-{AGENT}.png", media_type="image/png")108010811082def _render_page(name: str) -> str:1083 """Les pages sont statiques mais servies par agent : on remplace au1084 démarrage les jetons de partage (OG) par l'identité de l'agent —1085 les robots des réseaux sociaux n'exécutent pas le JS."""1086 html = (BASE / "web" / name).read_text(encoding="utf-8")1087 for token, value in {1088 "__NUM__": AGENT[-1],1089 "__DOMAIN__": ME["domain"],1090 "__TAGLINE__": ME.get("tagline", ""),1091 }.items():1092 html = html.replace(token, value)1093 return html109410951096PAGES = {name: _render_page(name) for name in ("index.html", "commander.html")}109710981099@app.get("/")1100def index() -> HTMLResponse:1101 return HTMLResponse(PAGES["index.html"])110211031104@app.get("/commander")1105def commander() -> HTMLResponse:1106 return HTMLResponse(PAGES["commander.html"])110711081109app.mount("/static", StaticFiles(directory=BASE / "web"), name="static")111011111112if __name__ == "__main__":1113 import uvicorn1114 uvicorn.run(app, host="0.0.0.0", port=ME["port"])1115