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"]}}"""349350351async def dispatch(incident: sqlite3.Row, health: dict[str, Any]) -> None:352 service, source, iid = incident["service"], incident["source"], incident["id"]353 svc = SERVICES[service]354 node = svc["node"]355 ip = NODES[node]["lan_ip"]356 mid = uuid.uuid4().hex[:12]357 my_ip = os.environ.get("KA_GUARDIAN_SELF_IP", "192.168.2.69")358 kind = incident["status_detected"]359 max_cost = None360 if kind in ("effort_new", "effort_enrich", "effort_degrade"):361 prompt = build_effort_prompt(kind, service, source, health.get("note", ""))362 max_turns = POLICY.get("effort_max_turns", 150)363 timeout = POLICY.get("effort_timeout_seconds", 7200)364 # Plafonds choisis par l'opérateur au moment de commander l'effort.365 if health.get("max_minutes"):366 timeout = min(timeout, int(health["max_minutes"]) * 60)367 if health.get("max_cost_usd"):368 max_cost = float(health["max_cost_usd"])369 else:370 prompt = build_prompt(service, source, health)371 max_turns, timeout = POLICY["mission_max_turns"], POLICY["mission_timeout_seconds"]372 payload = {373 "mission_id": mid, "agent": AGENT, "service": service, "source": source,374 "dir": svc["dir"], "pm2": svc["pm2"], "web_port": svc["web_port"],375 "model": ME.get("model", "sonnet"),376 "max_turns": max_turns,377 "timeout_seconds": timeout,378 "max_cost_usd": max_cost,379 "prompt": prompt,380 "callback_url": f"http://{my_ip}:{ME['port']}/api/ingest/{mid}",381 }382 try:383 code, body = await lan_post(ip, RUNNER_PORT, "/missions", payload, timeout=25)384 if code == 409:385 return # runner occupé, on retentera au prochain tick386 if code != 200:387 raise ConnectionError(f"runner {code}: {body[:200]}")388 except Exception as exc:389 print(f"[dispatch] {service}/{source} → {type(exc).__name__}: {exc}", flush=True)390 hub.publish_sync({"kind": "log", "level": "error", "msg": f"dispatch {source}: {exc}", "ts": now()})391 return392 with db() as c:393 c.execute("INSERT INTO missions(id,incident_id,service,source,node,state,started) VALUES(?,?,?,?,?,?,?)",394 (mid, iid, service, source, node, "running", now()))395 set_incident(c, iid, state="fixing", attempts=incident["attempts"] + 1)396 incident_event(iid, service, source, "fixing", f"mission {mid} dépêchée sur {NODES[node]['alias']}")397 hub.publish_sync({"kind": "mission", "mission_id": mid, "service": service,398 "source": source, "state": "running", "ts": now()})399400401async def rollback_mission(mission: sqlite3.Row, reason: str) -> dict[str, Any]:402 svc = SERVICES[mission["service"]]403 ip = NODES[mission["node"]]["lan_ip"]404 try:405 code, body = await lan_post(ip, RUNNER_PORT, "/rollback",406 {"dir": svc["dir"], "base_commit": mission["base_commit"],407 "pm2": svc["pm2"], "web_port": svc["web_port"]}, timeout=300)408 out = json.loads(body) if code == 200 else {"ok": False, "erreur": f"{code}: {body[:200]}"}409 except Exception as exc:410 out = {"ok": False, "erreur": str(exc)}411 hub.publish_sync({"kind": "rollback", "mission_id": mission["id"], "service": mission["service"],412 "source": mission["source"], "reason": reason, "result": out, "ts": now()})413 with db() as c:414 c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",415 (mission["id"], now(), "rollback", jdump({"reason": reason, "result": out})))416 return out417418419# --------------------------------------------------------------- moteur ----420421async def poll_once() -> None:422 async with httpx.AsyncClient() as cl:423 r = await cl.get(TOPO["apika_monitoring_url"], timeout=15)424 data = r.json()["data"]425 mine: dict[str, Any] = {}426 for service, block in data["services"].items():427 assigned_to_me = service in MY_SERVICES or (DEFAULT_AGENT and service not in ASSIGNED)428 if assigned_to_me:429 mine[service] = block430 LATEST.update({"mine": mine, "ecosystem": data["summary"], "polled": now()})431 with db() as c:432 summary_mine = {s: b["summary"] for s, b in mine.items()}433 c.execute("INSERT OR REPLACE INTO snapshots(ts,mine,ecosystem) VALUES(?,?,?)",434 (now(), jdump(summary_mine), jdump(data["summary"])))435 c.execute("DELETE FROM snapshots WHERE ts < ?", (now() - 30 * 86400,))436 hub.publish_sync({"kind": "snapshot", "mine": {s: b["summary"] for s, b in mine.items()},437 "ecosystem": data["summary"], "ts": now()})438439 trigger = set(POLICY["trigger_statuses"])440 with db() as c:441 for service, block in mine.items():442 if service not in SERVICES:443 continue # service inconnu de la topologie: visible au dashboard, pas d'action444 for conn in block.get("connectors", []):445 source, status = conn["source"], conn["status"]446 inc = active_incident(c, service, source)447 if status in trigger and inc is None:448 # Garde anti-boucle : une source déjà diagnostiquée irréparable449 # (site_source_mort, tentatives épuisées) ne redonne pas450 # d'incident tant que le délai de re-examen n'est pas écoulé.451 # Sans elle, un connecteur désactivé côté app mais rapporté452 # broken à perpétuité par api-ka relance une mission toutes453 # les ~10 min (2026-08-24 : 37 missions sur origin-north.ai).454 ab = c.execute(455 "SELECT updated FROM incidents WHERE service=? AND source=? "456 "AND state='abandoned' ORDER BY updated DESC LIMIT 1",457 (service, source)).fetchone()458 if ab and now() - ab["updated"] < POLICY.get("abandoned_retry_hours", 168) * 3600:459 continue460 iid = uuid.uuid4().hex[:10]461 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "462 "VALUES(?,?,?,?,?,?,?,?)",463 (iid, service, source, status, "open", now(), now(), jdump(conn)))464 incident_event(iid, service, source, "open", f"détecté {status}")465 elif status in trigger and inc is not None and inc["state"] in ("open", "cooldown"):466 # Ne pas rafraîchir watching/fixing: `updated` sert de chrono467 # à la fenêtre de surveillance et au cooldown.468 c.execute("UPDATE incidents SET status_detected=?, detail=? WHERE id=?",469 (status, jdump(conn), inc["id"]))470 elif status == "ok" and inc is not None:471 if inc["state"] == "watching":472 set_incident(c, inc["id"], state="resolved", resolved=now())473 incident_event(inc["id"], service, source, "resolved", "connecteur de retour à ok — réparation confirmée")474 elif inc["state"] in ("open", "cooldown"):475 set_incident(c, inc["id"], state="self_healed", resolved=now())476 incident_event(inc["id"], service, source, "self_healed", "revenu à ok sans intervention")477478479async def tick() -> None:480 if LATEST["paused"]:481 return482 try:483 await poll_once()484 except Exception as exc:485 hub.publish_sync({"kind": "log", "level": "warn", "msg": f"poll api-ka échoué: {exc}", "ts": now()})486487 with db() as c:488 # 1. Réconciliation des missions en cours depuis > 10 min: on demande au489 # runner ce qu'il en est (événement final perdu? runner redémarré?).490 for m in c.execute("SELECT id FROM missions WHERE state='running' AND started < ?",491 (now() - 600,)).fetchall():492 if m["id"] not in RECONCILING:493 RECONCILING.add(m["id"])494 asyncio.create_task(reconcile_mission(m["id"]))495496 # 2. Watching → rollback si la fenêtre est passée et toujours cassé497 for inc in c.execute("SELECT * FROM incidents WHERE state='watching'").fetchall():498 if now() - inc["updated"] < POLICY["watch_window_hours"] * 3600:499 continue500 status = current_status(inc["service"], inc["source"])501 m = c.execute("SELECT * FROM missions WHERE incident_id=? AND base_commit IS NOT NULL "502 "ORDER BY started DESC LIMIT 1", (inc["id"],)).fetchone()503 if status in POLICY["trigger_statuses"] or status == "degraded":504 if m and (json.loads(m["commits"] or "[]")):505 asyncio.create_task(rollback_mission(m, "toujours cassé après la fenêtre de surveillance"))506 set_incident(c, inc["id"], state="cooldown")507 incident_event(inc["id"], inc["service"], inc["source"], "cooldown",508 f"non guéri après {POLICY['watch_window_hours']}h → rollback + cooldown")509 elif status == "ok":510 set_incident(c, inc["id"], state="resolved", resolved=now())511 incident_event(inc["id"], inc["service"], inc["source"], "resolved", "confirmé ok")512513 # 3. Cooldown expiré → réouverture ou abandon514 for inc in c.execute("SELECT * FROM incidents WHERE state='cooldown'").fetchall():515 if now() - inc["updated"] < POLICY["attempt_cooldown_hours"] * 3600:516 continue517 status = current_status(inc["service"], inc["source"])518 if status == "ok":519 set_incident(c, inc["id"], state="resolved", resolved=now())520 incident_event(inc["id"], inc["service"], inc["source"], "resolved", "guéri pendant le cooldown")521 elif inc["attempts"] >= POLICY["max_attempts_per_incident"]:522 set_incident(c, inc["id"], state="abandoned")523 incident_event(inc["id"], inc["service"], inc["source"], "abandoned",524 f"{inc['attempts']} tentatives épuisées — intervention humaine requise")525 else:526 set_incident(c, inc["id"], state="open")527 incident_event(inc["id"], inc["service"], inc["source"], "open", "cooldown terminé, nouvelle tentative")528529 # 4. Dispatch (1 mission à la fois par agent, broken avant stale, plus vieux d'abord)530 running = c.execute("SELECT COUNT(*) n FROM missions WHERE state='running'").fetchone()["n"]531 if running < POLICY["max_concurrent_missions"]:532 busy_nodes = {m["node"] for m in c.execute("SELECT node FROM missions WHERE state='running'").fetchall()}533 nxt = c.execute(534 "SELECT * FROM incidents WHERE state='open' "535 "ORDER BY CASE status_detected WHEN 'manual' THEN 0 WHEN 'effort_new' THEN 0 "536 "WHEN 'effort_enrich' THEN 0 WHEN 'broken' THEN 1 ELSE 2 END, created "537 ).fetchall()538 for inc in nxt:539 if inc["service"] not in SERVICES:540 continue541 if SERVICES[inc["service"]]["node"] in busy_nodes:542 continue543 set_incident(c, inc["id"], state="dispatched")544 asyncio.create_task(dispatch_row(inc["id"]))545 break546547548def current_status(service: str, source: str) -> str:549 block = LATEST["mine"].get(service) or {}550 for conn in block.get("connectors", []):551 if conn["source"] == source:552 return conn["status"]553 return "inconnu"554555556async def dispatch_row(iid: str) -> None:557 with db() as c:558 inc = c.execute("SELECT * FROM incidents WHERE id=?", (iid,)).fetchone()559 if inc:560 await dispatch(inc, json.loads(inc["detail"] or "{}"))561 # dispatch() remet fixing; si l'appel a échoué/409, on relâche562 with db() as c:563 cur = c.execute("SELECT state FROM incidents WHERE id=?", (iid,)).fetchone()564 if cur and cur["state"] == "dispatched":565 set_incident(c, iid, state="open")566567568RECONCILING: set[str] = set()569570571async def reconcile_mission(mid: str) -> None:572 """Résout une mission « en cours » suspecte auprès de son runner."""573 try:574 with db() as c:575 m = c.execute("SELECT * FROM missions WHERE id=? AND state='running'", (mid,)).fetchone()576 if not m:577 return578 ip = NODES[m["node"]]["lan_ip"]579 try:580 code, body = await lan_post(ip, RUNNER_PORT, "/mission-result", {"mission_id": mid}, timeout=20)581 resp = json.loads(body) if code == 200 else {"status": "erreur"}582 except Exception:583 return # courrier/runner injoignable: on retentera au prochain tick584 if resp.get("status") == "running":585 return586 with db() as c:587 m = c.execute("SELECT * FROM missions WHERE id=? AND state='running'", (mid,)).fetchone()588 if not m:589 return590 if resp.get("status") == "done":591 data = (resp.get("final") or {}).get("data", {})592 c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",593 (mid, now(), "final", jdump(data)))594 await finalize(c, m, "final", data)595 hub.publish_sync({"kind": "mission_event", "mission_id": mid, "type": "final",596 "data": data, "ts": now()})597 elif now() - m["started"] > 900:598 # Le runner ne connaît pas cette mission: tuée en vol599 # (redémarrage). On la clôt et on relance l'incident.600 c.execute("UPDATE missions SET state='error', ended=? WHERE id=?", (now(), mid))601 inc = c.execute("SELECT * FROM incidents WHERE id=?", (m["incident_id"],)).fetchone()602 if inc and inc["state"] in ("fixing", "dispatched"):603 set_incident(c, inc["id"], state="open")604 incident_event(inc["id"], m["service"], m["source"], "open",605 "mission perdue (runner interrompu) — remise en file")606 finally:607 RECONCILING.discard(mid)608609610async def engine() -> None:611 global LOOP612 LOOP = asyncio.get_running_loop()613 await asyncio.sleep(3)614 while True:615 try:616 await tick()617 except Exception as exc:618 import traceback619 traceback.print_exc()620 hub.publish_sync({"kind": "log", "level": "error", "msg": f"tick: {exc}", "ts": now()})621 await asyncio.sleep(POLICY["poll_interval_seconds"])622623624# ------------------------------------------------------------------- app ---625626app = FastAPI(title=f"KA Guardian — {AGENT}", docs_url=None, redoc_url=None)627628629@app.on_event("startup")630async def startup() -> None:631 init_db()632 asyncio.create_task(engine())633634635def check_token(tok: str | None) -> None:636 if not TOKEN or tok != TOKEN:637 raise HTTPException(status_code=401)638639640@app.post("/api/ingest/{mission_id}")641async def ingest(mission_id: str, req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:642 check_token(x_ka_token)643 ev = await req.json()644 etype, data = ev.get("type"), ev.get("data", {})645 with db() as c:646 c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",647 (mission_id, now(), etype, jdump(data)))648 m = c.execute("SELECT * FROM missions WHERE id=?", (mission_id,)).fetchone()649 if m:650 if etype == "start":651 c.execute("UPDATE missions SET base_commit=? WHERE id=?", (data.get("base_commit"), mission_id))652 elif etype in ("final", "error"):653 await finalize(c, m, etype, data)654 hub.publish_sync({"kind": "mission_event", "mission_id": mission_id, "type": etype,655 "data": data, "ts": now()})656 return {"ok": True}657658659async def finalize(c: sqlite3.Connection, m: sqlite3.Row, etype: str, data: dict[str, Any]) -> None:660 inc = c.execute("SELECT * FROM incidents WHERE id=?", (m["incident_id"],)).fetchone()661 if etype == "error":662 c.execute("UPDATE missions SET state='error', ended=? WHERE id=?", (now(), m["id"]))663 if inc:664 set_incident(c, inc["id"], state="cooldown")665 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"erreur mission: {data.get('error')}")666 return667 verdict = data.get("verdict", {})668 commits = data.get("commits", [])669 health = data.get("health", {})670 c.execute("UPDATE missions SET state='done', ended=?, commits=?, verdict=?, cost_usd=?, num_turns=?, health=?, base_commit=COALESCE(base_commit,?) WHERE id=?",671 (now(), jdump(commits), jdump(verdict), data.get("cost_usd"),672 data.get("num_turns"), jdump(health), data.get("base_commit"), m["id"]))673 m2 = c.execute("SELECT * FROM missions WHERE id=?", (m["id"],)).fetchone()674 v = verdict.get("verdict", "inconnu")675 if not inc:676 return677 if inc["status_detected"] in ("effort_new", "effort_enrich", "effort_degrade"):678 # Les efforts commandés ne passent pas par la surveillance api-ka:679 # verdict + santé de l'app décident tout de suite.680 if not health.get("ok", True):681 asyncio.create_task(rollback_mission(m2, "healthcheck app en échec post-effort"))682 set_incident(c, inc["id"], state="abandoned")683 incident_event(inc["id"], m["service"], m["source"], "abandoned", "app en mauvaise santé → ROLLBACK immédiat")684 elif v == "livre":685 set_incident(c, inc["id"], state="resolved", resolved=now())686 incident_event(inc["id"], m["service"], m["source"], "resolved",687 f"effort livré ({len(commits)} commit{'s' if len(commits) > 1 else ''})")688 elif v == "echec" and commits:689 asyncio.create_task(rollback_mission(m2, "effort en échec déclaré → rollback préventif"))690 set_incident(c, inc["id"], state="abandoned")691 incident_event(inc["id"], m["service"], m["source"], "abandoned", "effort en échec → rollback")692 else:693 # Session interrompue (plafond durée/coût) ou verdict illisible:694 # l'app est saine, les commits incrémentaux testés sont conservés.695 set_incident(c, inc["id"], state="abandoned")696 incident_event(inc["id"], m["service"], m["source"], "abandoned",697 f"effort terminé sans verdict livré ({v}) — {len(commits)} commit(s) conservé(s), app saine")698 return699 if not health.get("ok", True):700 # L'app est tombée → rollback immédiat, quoi qu'ait dit l'agent.701 asyncio.create_task(rollback_mission(m2, "healthcheck app en échec post-mission"))702 set_incident(c, inc["id"], state="cooldown")703 incident_event(inc["id"], m["service"], m["source"], "cooldown", "app en mauvaise santé → ROLLBACK immédiat")704 elif v == "repare" and commits:705 set_incident(c, inc["id"], state="watching")706 incident_event(inc["id"], m["service"], m["source"], "watching",707 f"réparation déclarée ({len(commits)} commit) — surveillance {POLICY['watch_window_hours']}h")708 elif v in ("echec", "inconnu") and commits:709 asyncio.create_task(rollback_mission(m2, f"verdict {v} avec commits → annulation par prudence"))710 set_incident(c, inc["id"], state="cooldown")711 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"verdict {v} → rollback préventif")712 elif v == "site_source_mort":713 set_incident(c, inc["id"], state="abandoned")714 incident_event(inc["id"], m["service"], m["source"], "abandoned", "source définitivement morte (diagnostic agent)")715 elif v in ("rien_a_faire", "repare"):716 set_incident(c, inc["id"], state="watching")717 incident_event(inc["id"], m["service"], m["source"], "watching", f"verdict {v} — surveillance")718 else:719 set_incident(c, inc["id"], state="cooldown")720 incident_event(inc["id"], m["service"], m["source"], "cooldown", f"verdict {v}")721722723@app.get("/api/state")724def state() -> dict[str, Any]:725 with db() as c:726 incidents = [dict(r) for r in c.execute(727 "SELECT * FROM incidents ORDER BY created DESC LIMIT 200").fetchall()]728 missions = [dict(r) for r in c.execute(729 "SELECT * FROM missions ORDER BY started DESC LIMIT 100").fetchall()]730 stats = c.execute("""SELECT731 (SELECT COUNT(*) FROM missions) n_missions,732 (SELECT COUNT(*) FROM incidents WHERE state='resolved') n_resolved,733 (SELECT COUNT(*) FROM incidents WHERE state IN ('open','dispatched','fixing','watching','cooldown')) n_active,734 (SELECT COUNT(*) FROM incidents WHERE state='abandoned') n_abandoned,735 (SELECT COUNT(*) FROM events WHERE type='rollback') n_rollbacks,736 (SELECT ROUND(SUM(cost_usd),2) FROM missions) cost_total,737 (SELECT ROUND(AVG(resolved-created)/3600.0,1) FROM incidents WHERE state='resolved' AND resolved IS NOT NULL) mttr_h738 """).fetchone()739 history = [dict(r) for r in c.execute(740 "SELECT ts, mine FROM snapshots WHERE ts > ? ORDER BY ts", (now() - 7 * 86400,)).fetchall()]741 return {742 "agent": AGENT, "identity": {k: ME.get(k) for k in ("domain", "accent", "accent2", "accent_soft", "tagline", "model")},743 "services": {s: {**SERVICES[s], "node_alias": NODES[SERVICES[s]["node"]]["alias"]} for s in ME["services"]},744 "siblings": {a: {"domain": v["domain"], "accent": v["accent"], "services": v["services"]}745 for a, v in TOPO["agents"].items() if a != AGENT},746 "latest": LATEST, "incidents": incidents, "missions": missions,747 "stats": dict(stats), "history": history, "policy": POLICY, "now": now(),748 }749750751@app.get("/api/missions/{mission_id}")752def mission_detail(mission_id: str) -> dict[str, Any]:753 with db() as c:754 m = c.execute("SELECT * FROM missions WHERE id=?", (mission_id,)).fetchone()755 if not m:756 raise HTTPException(status_code=404)757 evs = [dict(r) for r in c.execute(758 "SELECT ts,type,data FROM events WHERE mission_id=? ORDER BY id", (mission_id,)).fetchall()]759 return {"mission": dict(m), "events": evs}760761762@app.get("/events")763async def sse(request: Request) -> StreamingResponse:764 q: asyncio.Queue = asyncio.Queue()765 hub.clients.add(q)766767 async def gen():768 try:769 yield f"data: {jdump({'kind': 'hello', 'agent': AGENT, 'ts': now()})}\n\n"770 while True:771 if await request.is_disconnected():772 break773 try:774 msg = await asyncio.wait_for(q.get(), timeout=25)775 yield f"data: {jdump(msg)}\n\n"776 except asyncio.TimeoutError:777 yield ": keepalive\n\n"778 finally:779 hub.clients.discard(q)780781 return StreamingResponse(gen(), media_type="text/event-stream",782 headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})783784785@app.get("/api/connectors/{service}")786def connectors_list(service: str) -> dict[str, Any]:787 """Sources connues du service (pour le menu déroulant d'enrichissement)."""788 block = LATEST["mine"].get(service) or {}789 return {"service": service,790 "connectors": sorted(791 ({"source": c["source"], "status": c["status"]} for c in block.get("connectors", [])),792 key=lambda x: x["source"])}793794795@app.post("/api/admin/effort")796async def admin_effort(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:797 """Commande un effort autonome: nouveau connecteur ou enrichissement."""798 check_token(x_ka_token)799 body = await req.json()800 kind = body.get("kind")801 service = body.get("service")802 note = (body.get("note") or "").strip()[:2000]803 if kind not in ("effort_new", "effort_enrich", "effort_degrade") or service not in SERVICES:804 raise HTTPException(status_code=400, detail="kind ou service invalide")805 if kind == "effort_enrich":806 source = (body.get("source") or "").strip()807 if not source:808 raise HTTPException(status_code=400, detail="source requise pour un enrichissement")809 elif kind == "effort_degrade":810 source = f"dégradés·{uuid.uuid4().hex[:6]}"811 else:812 source = f"nouveau·{uuid.uuid4().hex[:6]}"813 detail: dict[str, Any] = {"note": note, "kind": kind}814 try:815 if body.get("max_minutes"):816 detail["max_minutes"] = max(10, min(240, int(body["max_minutes"])))817 if body.get("max_cost_usd"):818 detail["max_cost_usd"] = max(0.5, min(100.0, float(body["max_cost_usd"])))819 except (TypeError, ValueError):820 raise HTTPException(status_code=400, detail="max_minutes/max_cost_usd invalides")821 iid = uuid.uuid4().hex[:10]822 with db() as c:823 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "824 "VALUES(?,?,?,?,?,?,?,?)",825 (iid, service, source, kind, "open", now(), now(), jdump(detail)))826 labels = {"effort_new": "nouveau connecteur", "effort_enrich": f"enrichissement de {source}",827 "effort_degrade": "inspection des connecteurs dégradés"}828 incident_event(iid, service, source, "open",829 f"effort commandé: {labels[kind]}"830 + (f" · ≤{detail['max_minutes']} min" if detail.get("max_minutes") else "")831 + (f" · ≤{detail['max_cost_usd']} $" if detail.get("max_cost_usd") else ""))832 return {"ok": True, "incident_id": iid}833834835@app.post("/api/admin/mission")836async def admin_mission(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:837 """Déclenche manuellement une mission sur un connecteur (même sain)."""838 check_token(x_ka_token)839 body = await req.json()840 service, source = body["service"], body["source"]841 if service not in SERVICES:842 raise HTTPException(status_code=400, detail="service inconnu")843 detail = {"source": source, "status": "manual", "message": body.get("note", "mission manuelle")}844 for conn in (LATEST["mine"].get(service) or {}).get("connectors", []):845 if conn["source"] == source:846 detail = {**conn, "status": "manual"}847 with db() as c:848 inc = active_incident(c, service, source)849 if inc is None:850 iid = uuid.uuid4().hex[:10]851 c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "852 "VALUES(?,?,?,?,?,?,?,?)",853 (iid, service, source, "manual", "open", now(), now(), jdump(detail)))854 else:855 iid = inc["id"]856 set_incident(c, iid, state="open", status_detected="manual")857 incident_event(iid, service, source, "open", "mission manuelle demandée")858 return {"ok": True, "incident_id": iid}859860861@app.post("/api/admin/pause")862async def admin_pause(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:863 check_token(x_ka_token)864 body = await req.json()865 LATEST["paused"] = bool(body.get("paused", True))866 return {"ok": True, "paused": LATEST["paused"]}867868869@app.get("/health")870def health() -> dict[str, Any]:871 return {"ok": True, "agent": AGENT, "polled": LATEST["polled"], "paused": LATEST["paused"]}872873874@app.get("/favicon.svg")875def favicon() -> FileResponse:876 return FileResponse(BASE / "web" / f"favicon-{AGENT}.svg", media_type="image/svg+xml")877878879@app.get("/og.png")880def og_image() -> FileResponse:881 """Bannière de partage (Open Graph) propre à l'agent."""882 return FileResponse(BASE / "web" / f"og-{AGENT}.png", media_type="image/png")883884885def _render_page(name: str) -> str:886 """Les pages sont statiques mais servies par agent : on remplace au887 démarrage les jetons de partage (OG) par l'identité de l'agent —888 les robots des réseaux sociaux n'exécutent pas le JS."""889 html = (BASE / "web" / name).read_text(encoding="utf-8")890 for token, value in {891 "__NUM__": AGENT[-1],892 "__DOMAIN__": ME["domain"],893 "__TAGLINE__": ME.get("tagline", ""),894 }.items():895 html = html.replace(token, value)896 return html897898899PAGES = {name: _render_page(name) for name in ("index.html", "commander.html")}900901902@app.get("/")903def index() -> HTMLResponse:904 return HTMLResponse(PAGES["index.html"])905906907@app.get("/commander")908def commander() -> HTMLResponse:909 return HTMLResponse(PAGES["commander.html"])910911912app.mount("/static", StaticFiles(directory=BASE / "web"), name="static")913914915if __name__ == "__main__":916 import uvicorn917 uvicorn.run(app, host="0.0.0.0", port=ME["port"])918