# ============================================
# Projet   : KA Guardian
# Fichier  : orchestrator/main.py
# Rôle     : Cerveau d'un agent gardien (ka2/ka4/ka6) — surveille la santé
#            des connecteurs via api-ka, ouvre des incidents, dépêche des
#            missions claude aux runners des nœuds, surveille la guérison,
#            rollback automatique si ça empire. Sert aussi le dashboard live.
# Author   : Simon-Pierre Boucher
# Date     : 2026-08-23
# ============================================
"""Instance unique par agent: AGENT=ka2|ka4|ka6 (env), port depuis topology.json."""
from __future__ import annotations

import asyncio
import json
import os
import pathlib
import sqlite3
import time
import uuid
from typing import Any

import httpx
from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.responses import FileResponse, HTMLResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles

import registry  # topologie vivante (registre mld) — même dossier que main.py

BASE = pathlib.Path(__file__).resolve().parent
ROOT = BASE.parent
HOME = pathlib.Path.home()

AGENT = os.environ.get("AGENT", "ka2")
TOPO = json.loads((ROOT / "topology.json").read_text())
ME = TOPO["agents"][AGENT]
POLICY = TOPO["policy"]
# SERVICES / NODES : dicts mutés EN PLACE par registry.refresh() — l'emplacement
# réel (nœud, IP, dir, port, PM2) vient du registre mld, topology.json n'est
# que le repli si le registre n'a jamais été reçu.
SERVICES: dict[str, Any] = TOPO["services"]
NODES: dict[str, Any] = TOPO["nodes"]
MY_SERVICES: set[str] = set(ME["services"])
RUNNER_PORT = TOPO["runner_port"]
try:  # première application au chargement (silencieuse : le hub SSE n'existe pas encore)
    for _m in registry.refresh(TOPO, SERVICES, NODES):
        print(f"[registre] {_m}", flush=True)
except Exception as _exc:
    print(f"[registre] indisponible au démarrage ({_exc}) — topology.json en repli", flush=True)


def node_alias(node_key: str) -> str:
    return (NODES.get(node_key) or {}).get("alias", node_key)


def node_ip(node_key: str) -> str | None:
    return (NODES.get(node_key) or {}).get("lan_ip")


def apply_registry(announce: bool = True) -> None:
    """Relit le registre mld s'il a changé et signale les déménagements."""
    try:
        changes = registry.refresh(TOPO, SERVICES, NODES)
    except Exception as exc:  # jamais fatal : on garde la dernière topologie connue
        print(f"[registre] échec: {exc}", flush=True)
        return
    for msg in changes:
        print(f"[registre] {msg}", flush=True)
        if announce:
            hub.publish_sync({"kind": "log", "level": "warn", "msg": f"registre mld — {msg}", "ts": now()})


_RUNNER_WARNED: dict[str, float] = {}


def runner_available(node_key: str, what: str) -> bool:
    """False si la sonde registry-sync dit qu'il n'y a PAS de runner sur ce nœud
    (app déménagée sur un nœud jamais équipé) — on n'envoie alors rien au
    courrier, et on avertit au plus une fois par heure par nœud."""
    ip = node_ip(node_key)
    rs = registry.runner_status(ip)
    if rs is None or rs.get("ok"):
        return True
    if now() - _RUNNER_WARNED.get(node_key, 0) > 3600:
        _RUNNER_WARNED[node_key] = now()
        msg = (f"aucun runner gardien sur {node_alias(node_key)} ({ip}) — {what} impossible ; "
               f"déployer avec deploy/deploy.sh runners (le nœud est listé par le registre)")
        print(f"[runner] {msg}", flush=True)
        hub.publish_sync({"kind": "log", "level": "error", "msg": msg, "ts": now()})
    return False

# ka6 (ou autre) ramasse les services non assignés qui apparaîtraient dans api-ka.
DEFAULT_AGENT = ME.get("default_for_unknown_services", False)
ASSIGNED = {s for a in TOPO["agents"].values() for s in a["services"]}


def load_env_file(path: pathlib.Path) -> dict[str, str]:
    out: dict[str, str] = {}
    if path.exists():
        for line in path.read_text().splitlines():
            line = line.strip()
            if line and not line.startswith("#") and "=" in line:
                k, v = line.split("=", 1)
                out[k.strip()] = v.strip().strip('"').strip("'")
    return out


TOKEN = load_env_file(HOME / ".ka-guardian.env").get("KA_GUARDIAN_TOKEN", "")


SPOOL = HOME / "ka-guardian-spool"


async def lan_post(ip: str, port: int, path: str, payload: dict[str, Any], timeout: int = 30) -> tuple[int, str]:
    """POST JSON vers un autre nœud via le courrier zsh (deploy/courier.sh).

    macOS 26 « Local Network Privacy » refuse le trafic LAN dès que python
    (homebrew) est dans la chaîne de processus — même ses enfants ssh/curl.
    On dépose donc un job dans un spool disque ; un service launchd 100 % zsh
    l'expédie (ssh → curl localhost du nœud cible) et dépose la réponse.
    """
    for d in ("outbox", "done", "tmp"):
        (SPOOL / d).mkdir(parents=True, exist_ok=True)
    jid = uuid.uuid4().hex
    tmp = SPOOL / "tmp" / f"{jid}.job"
    tmp.write_text(f"{ip} {port} {path} {timeout}\n" + jdump(payload))
    tmp.rename(SPOOL / "outbox" / f"{jid}.job")
    resp = SPOOL / "done" / f"{jid}.resp"
    deadline = time.time() + timeout + 25
    while time.time() < deadline:
        if resp.exists():
            txt = resp.read_text()
            resp.unlink(missing_ok=True)
            body, _, code = txt.strip().rpartition("\n")
            if code.startswith("courier_ssh_rc"):
                raise ConnectionError(f"{code} vers {ip}:{port}{path}")
            return int(code or "0"), body
        await asyncio.sleep(0.3)
    raise TimeoutError(f"courrier sans réponse pour {ip}:{port}{path}")
DATA = ROOT / "data"
DATA.mkdir(exist_ok=True)
DB_PATH = DATA / f"{AGENT}.db"

# ---------------------------------------------------------------- SQLite ---

def db() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    return conn


def init_db() -> None:
    with db() as c:
        c.executescript("""
        CREATE TABLE IF NOT EXISTS incidents(
          id TEXT PRIMARY KEY, service TEXT, source TEXT, status_detected TEXT,
          state TEXT, attempts INTEGER DEFAULT 0, created REAL, updated REAL,
          resolved REAL, detail TEXT);
        CREATE TABLE IF NOT EXISTS missions(
          id TEXT PRIMARY KEY, incident_id TEXT, service TEXT, source TEXT,
          node TEXT, state TEXT, base_commit TEXT, commits TEXT, verdict TEXT,
          cost_usd REAL, num_turns INTEGER, health TEXT, started REAL, ended REAL);
        CREATE TABLE IF NOT EXISTS events(
          id INTEGER PRIMARY KEY AUTOINCREMENT, mission_id TEXT, ts REAL,
          type TEXT, data TEXT);
        CREATE TABLE IF NOT EXISTS snapshots(
          ts REAL PRIMARY KEY, mine TEXT, ecosystem TEXT);
        CREATE INDEX IF NOT EXISTS ev_mission ON events(mission_id);
        """)


def now() -> float:
    return time.time()


def jdump(x: Any) -> str:
    return json.dumps(x, ensure_ascii=False)


# ------------------------------------------------------------------- SSE ---

class Hub:
    def __init__(self) -> None:
        self.clients: set[asyncio.Queue] = set()

    async def publish(self, msg: dict[str, Any]) -> None:
        for q in list(self.clients):
            if q.qsize() < 500:
                q.put_nowait(msg)

    def publish_sync(self, msg: dict[str, Any]) -> None:
        if LOOP:
            asyncio.run_coroutine_threadsafe(self.publish(msg), LOOP)


hub = Hub()
LOOP: asyncio.AbstractEventLoop | None = None
LATEST: dict[str, Any] = {"mine": {}, "ecosystem": {}, "polled": 0, "paused": False}

# ------------------------------------------------------------- missions ----

def active_incident(c: sqlite3.Connection, service: str, source: str) -> sqlite3.Row | None:
    return c.execute(
        "SELECT * FROM incidents WHERE service=? AND source=? AND state IN "
        "('open','dispatched','fixing','watching','cooldown') ORDER BY created DESC LIMIT 1",
        (service, source)).fetchone()


def set_incident(c: sqlite3.Connection, iid: str, **kw: Any) -> None:
    kw["updated"] = now()
    keys = ",".join(f"{k}=?" for k in kw)
    c.execute(f"UPDATE incidents SET {keys} WHERE id=?", (*kw.values(), iid))


def incident_event(iid: str, service: str, source: str, state: str, note: str = "") -> None:
    hub.publish_sync({"kind": "incident", "incident_id": iid, "service": service,
                      "source": source, "state": state, "note": note, "ts": now()})


def service_situation(service: str) -> str:
    """Portrait santé de l'app entière — permet à l'agent de distinguer une
    panne isolée d'un problème systémique (app entière, réseau, quota)."""
    block = LATEST["mine"].get(service) or {}
    summary = block.get("summary") or {}
    app_row = block.get("app") or {}
    sick = [f"{c['source']} ({c['status']})" for c in block.get("connectors", [])
            if c.get("status") not in ("ok", None)][:15]
    lines = [f"- app entière: {app_row.get('status', '?')} — {summary.get('ok', '?')} ok, "
             f"{summary.get('degraded', 0)} dégradés, {summary.get('broken', 0)} cassés, {summary.get('stale', 0)} endormis"]
    if sick:
        lines.append(f"- autres connecteurs non-ok de cette app: {', '.join(sick)}")
        lines.append("- Si BEAUCOUP de connecteurs sont touchés en même temps, le problème est probablement "
                     "systémique (scheduler de l'app, réseau, quota API): diagnostique au niveau app, pas source par source.")
    else:
        lines.append("- Tous les autres connecteurs de l'app sont sains: la panne est isolée à cette source.")
    return "\n".join(lines)


def mission_history(service: str, source: str) -> str:
    """Résumé des missions précédentes sur ce connecteur (éviter de refaire ce qui a échoué)."""
    with db() as c:
        rows = c.execute(
            "SELECT m.started, m.verdict, m.commits FROM missions m WHERE m.service=? AND m.source=? "
            "AND m.state!='running' ORDER BY m.started DESC LIMIT 3", (service, source)).fetchall()
    if not rows:
        return "- aucune: c'est la première intervention sur ce connecteur."
    out = []
    for r in rows:
        v = json.loads(r["verdict"] or "{}")
        out.append(f"- {time.strftime('%Y-%m-%d %H:%M', time.localtime(r['started']))}: "
                   f"verdict={v.get('verdict', '?')} — {str(v.get('diagnostic', ''))[:200]} "
                   f"(actions: {str(v.get('actions', ''))[:150]})")
    out.append("NE RÉPÈTE PAS une approche qui a déjà échoué ci-dessus: change d'angle.")
    return "\n".join(out)


def build_prompt(service: str, source: str, health: dict[str, Any]) -> str:
    svc = SERVICES[service]
    sync_proc = next((p for p in svc["pm2"] if "sync" in p or "etl" in p), (svc["pm2"][-1] if svc["pm2"] else "?"))
    node_alias_ = node_alias(svc["node"])
    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', '')}).

== CONTEXTE SANTÉ DU CONNECTEUR (supervision api-ka, scan aux 2 h) ==
- statut détecté: {health.get('status')} | échecs consécutifs: {health.get('consecutive_failures')}
- dernier succès: {health.get('last_success')} | volume au dernier sync: {health.get('found_last')} (médiane historique: {health.get('median_found')})
- message: {health.get('message')}
Rappel 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.

== SITUATION DE L'APP ==
{service_situation(service)}

== INTERVENTIONS PRÉCÉDENTES SUR CE CONNECTEUR ==
{mission_history(service, source)}

== TON ENVIRONNEMENT ==
Tu 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).
L'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']}.
Le venv Python et/ou node_modules du repo sont déjà installés — utilise-les, jamais d'installation globale.

== DÉMARCHE IMPOSÉE (dans l'ordre) ==
1. 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.
2. 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).
3. 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.
4. 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é?
5. 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.
6. 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.
7. 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).
8. 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.
9. 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.

== INTERDITS ABSOLUS ==
Toucher 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.

== FIN DE MISSION ==
Termine ta TOUTE DERNIÈRE réponse par un bloc JSON exactement de cette forme:
{{"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"]}}
Sois 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."""


COMMON_RULES = """== RÈGLES COMMUNES (non négociables) ==
- 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.
- Respecte l'architecture et les patterns du repo (regarde comment les connecteurs voisins sont faits AVANT d'écrire).
- Utilise le venv/node_modules du repo; jamais d'installation globale.
- INTERDIT d'inventer, stubber ou câbler des données en dur. Chaque donnée vient d'un vrai fetch.
- 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.
- À 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."""


def effort_stack(service: str) -> str:
    svc = SERVICES[service]
    return f"""== BOÎTE À OUTILS (clés dans ~/.claude/.env) ==
- 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.
- 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).
- APIFY (APIFY_TOKEN): en dernier recours pour les plateformes dures, acteurs maison du compte (proxy résidentiel intégré).
- Regarde d'abord comment le repo de {svc['app']} utilise déjà ces services et fais pareil."""


def build_effort_prompt(kind: str, service: str, source: str, note: str) -> str:
    svc = SERVICES[service]
    sync_proc = next((p for p in svc["pm2"] if "sync" in p or "etl" in p), (svc["pm2"][-1] if svc["pm2"] else "?"))
    node_alias_ = node_alias(svc["node"])
    rules = COMMON_RULES.replace("%AGENT%", AGENT)
    env = f"""== TON ENVIRONNEMENT ==
Tu es sur le nœud {node_alias_}, répertoire courant = repo de l'app: {svc['dir']} (remote-first, source de vérité).
L'app tourne via pm2 ({', '.join(svc['pm2'])}); process de sync: {sync_proc}; site local port {svc['web_port']}.
Commence par lire le CLAUDE.md / README du repo et docs/connecteurs/ s'ils existent."""
    if kind == "effort_degrade":
        block = LATEST["mine"].get(service) or {}
        degraded = [c for c in block.get("connectors", []) if c.get("status") == "degraded"][:25]
        deg_lines = "\n".join(
            f"- {c['source']}: dernier volume {c.get('found_last')} vs médiane {round(c.get('median_found') or 0)} "
            f"(dernier succès: {c.get('last_success')}, message: {c.get('message')})"
            for c in degraded) or "- (aucun connecteur dégradé au dernier scan — re-vérifie l'état réel dans l'app)"
        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', '')}).
{"Consigne de l'opérateur: " + note if note else ""}

Un 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.

== CONNECTEURS DÉGRADÉS AU DERNIER SCAN api-ka ==
{deg_lines}

{env}

== DÉMARCHE ==
1. 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.
2. 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>: …).
3. Pour les cas (b), ne touche à rien: consigne-les dans ton rapport final comme « baisse légitime ».
4. 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.
5. À 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`.

{effort_stack(service)}

{rules}

== FIN DE MISSION ==
Termine ta TOUTE DERNIÈRE réponse par un bloc JSON:
{{"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"]}}"""
    if kind == "effort_new":
        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', '')}).
{"Consigne de l'opérateur: " + note if note else "Aucune consigne particulière: choisis la source la plus utile."}

{env}

== DÉMARCHE ==
1. 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).
2. 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.
3. 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.
4. 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).
5. DOCUMENTE: ajoute la fiche docs/connecteurs/<source>.md si le repo suit cette convention (URL, stratégie, pièges).
6. Redémarre {sync_proc}, vérifie le site, committe et pousse sur spbgit (`git push origin main`).

{effort_stack(service)}

{rules}

== FIN DE MISSION ==
Termine ta TOUTE DERNIÈRE réponse par un bloc JSON:
{{"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"]}}"""
    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', '')}).
{"Consigne de l'opérateur: " + note if note else "Aucune consigne particulière: maximise la valeur (couverture, champs, robustesse)."}

{env}

== DÉMARCHE ==
1. É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.
2. 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.
3. IMPLÉMENTE proprement dans le pattern du repo, SANS casser le format de sortie existant (les consommateurs avals dépendent du schéma).
4. 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.
5. METS À JOUR la fiche docs/connecteurs/<source>.md si elle existe.
6. Redémarre {sync_proc}, vérifie le site, committe et pousse sur spbgit (`git push origin main`).

{effort_stack(service)}

{rules}

== FIN DE MISSION ==
Termine ta TOUTE DERNIÈRE réponse par un bloc JSON:
{{"verdict": "livre|echec", "diagnostic": "état initial constaté", "actions": "enrichissements livrés", "test": "avant/après mesuré", "fichiers": ["fichiers modifiés"]}}"""


def build_site_down_prompt(service: str, detail: dict[str, Any]) -> str:
    svc = SERVICES[service]
    web_proc = svc["pm2"][0] if svc["pm2"] else "?"
    node_alias_ = node_alias(svc["node"])
    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.
Session 100 % AUTONOME: tu ne poses aucune question, tu travailles jusqu'au bout.

== CE QUE LE VEILLEUR A DÉJÀ CONSTATÉ ==
- échecs consécutifs du check public: {detail.get('fails', '?')} (site muet depuis {detail.get('down_since', '?')})
- dernière erreur du check public: {detail.get('error', '?')}
- 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'}

== TON ENVIRONNEMENT ==
Tu es sur le nœud {node_alias_}, répertoire courant = repo de l'app: {svc['dir']} (remote-first, source de vérité).
Process 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).

== DÉMARCHE IMPOSÉE (dans l'ordre) ==
1. 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.
2. 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`).
3. DISTINGUE les 3 familles de panne:
   a. Process web mort ou en boucle de crash → cause racine dans le code/l'env → corrige puis `pm2 restart {web_proc}`.
   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.
   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.
4. 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).
5. 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).
6. 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.

== INTERDITS ABSOLUS ==
Toucher 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.

== FIN DE MISSION ==
Termine ta TOUTE DERNIÈRE réponse par un bloc JSON exactement de cette forme:
{{"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"]}}
Sois 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."""


async def dispatch(incident: sqlite3.Row, health: dict[str, Any]) -> None:
    service, source, iid = incident["service"], incident["source"], incident["id"]
    svc = SERVICES[service]
    node = svc["node"]
    ip = node_ip(node)
    if not ip:
        hub.publish_sync({"kind": "log", "level": "error", "ts": now(),
                          "msg": f"dispatch {source}: IP LAN inconnue pour le nœud {node_alias(node)} (registre incomplet)"})
        return
    if not runner_available(node, f"mission {source}"):
        return  # l'incident reste ouvert ; on retentera quand le runner existera
    mid = uuid.uuid4().hex[:12]
    # IP du gardien lui-même (callback du runner) : env launchd, sinon le registre mld.
    my_ip = os.environ.get("KA_GUARDIAN_SELF_IP") or (registry.app_entry(AGENT) or {}).get("ip") or "192.168.2.69"
    kind = incident["status_detected"]
    max_cost = None
    if kind in ("effort_new", "effort_enrich", "effort_degrade"):
        prompt = build_effort_prompt(kind, service, source, health.get("note", ""))
        max_turns = POLICY.get("effort_max_turns", 150)
        timeout = POLICY.get("effort_timeout_seconds", 7200)
        # Plafonds choisis par l'opérateur au moment de commander l'effort.
        if health.get("max_minutes"):
            timeout = min(timeout, int(health["max_minutes"]) * 60)
        if health.get("max_cost_usd"):
            max_cost = float(health["max_cost_usd"])
    elif kind == "site_down":
        prompt = build_site_down_prompt(service, health)
        max_turns, timeout = POLICY["mission_max_turns"], POLICY["mission_timeout_seconds"]
    else:
        prompt = build_prompt(service, source, health)
        max_turns, timeout = POLICY["mission_max_turns"], POLICY["mission_timeout_seconds"]
    payload = {
        "mission_id": mid, "agent": AGENT, "service": service, "source": source,
        "dir": svc["dir"], "pm2": svc["pm2"], "web_port": svc["web_port"],
        "model": ME.get("model", "sonnet"),
        "max_turns": max_turns,
        "timeout_seconds": timeout,
        "max_cost_usd": max_cost,
        "prompt": prompt,
        "callback_url": f"http://{my_ip}:{ME['port']}/api/ingest/{mid}",
    }
    try:
        code, body = await lan_post(ip, RUNNER_PORT, "/missions", payload, timeout=25)
        if code == 409:
            return  # runner occupé, on retentera au prochain tick
        if code != 200:
            raise ConnectionError(f"runner {code}: {body[:200]}")
    except Exception as exc:
        print(f"[dispatch] {service}/{source} → {type(exc).__name__}: {exc}", flush=True)
        hub.publish_sync({"kind": "log", "level": "error", "msg": f"dispatch {source}: {exc}", "ts": now()})
        return
    with db() as c:
        c.execute("INSERT INTO missions(id,incident_id,service,source,node,state,started) VALUES(?,?,?,?,?,?,?)",
                  (mid, iid, service, source, node, "running", now()))
        set_incident(c, iid, state="fixing", attempts=incident["attempts"] + 1)
    incident_event(iid, service, source, "fixing", f"mission {mid} dépêchée sur {node_alias(node)}")
    hub.publish_sync({"kind": "mission", "mission_id": mid, "service": service,
                      "source": source, "state": "running", "ts": now()})


async def rollback_mission(mission: sqlite3.Row, reason: str) -> dict[str, Any]:
    svc = SERVICES[mission["service"]]
    # Le repo (et son historique git) suit l'app : si elle a déménagé depuis la
    # mission, le rollback vise le nœud ACTUEL du service, pas celui de la mission.
    node = svc.get("node") or mission["node"]
    ip = node_ip(node) or node_ip(mission["node"])
    if node != mission["node"]:
        hub.publish_sync({"kind": "log", "level": "warn", "ts": now(),
                          "msg": f"rollback {mission['source']}: l'app a déménagé {node_alias(mission['node'])} → {node_alias(node)}, rollback sur le nœud actuel"})
    if not ip or not runner_available(node, f"rollback {mission['source']}"):
        out = {"ok": False, "erreur": f"runner indisponible sur {node_alias(node)}"}
        hub.publish_sync({"kind": "rollback", "mission_id": mission["id"], "service": mission["service"],
                          "source": mission["source"], "reason": reason, "result": out, "ts": now()})
        return out
    try:
        code, body = await lan_post(ip, RUNNER_PORT, "/rollback",
                                    {"dir": svc["dir"], "base_commit": mission["base_commit"],
                                     "pm2": svc["pm2"], "web_port": svc["web_port"]}, timeout=300)
        out = json.loads(body) if code == 200 else {"ok": False, "erreur": f"{code}: {body[:200]}"}
    except Exception as exc:
        out = {"ok": False, "erreur": str(exc)}
    hub.publish_sync({"kind": "rollback", "mission_id": mission["id"], "service": mission["service"],
                      "source": mission["source"], "reason": reason, "result": out, "ts": now()})
    with db() as c:
        c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",
                  (mission["id"], now(), "rollback", jdump({"reason": reason, "result": out})))
    return out


# --------------------------------------------------------------- moteur ----

async def poll_once() -> None:
    async with httpx.AsyncClient() as cl:
        r = await cl.get(TOPO["apika_monitoring_url"], timeout=15)
        data = r.json()["data"]
    mine: dict[str, Any] = {}
    for service, block in data["services"].items():
        assigned_to_me = service in MY_SERVICES or (DEFAULT_AGENT and service not in ASSIGNED)
        if assigned_to_me:
            mine[service] = block
    LATEST.update({"mine": mine, "ecosystem": data["summary"], "polled": now()})
    with db() as c:
        summary_mine = {s: b["summary"] for s, b in mine.items()}
        c.execute("INSERT OR REPLACE INTO snapshots(ts,mine,ecosystem) VALUES(?,?,?)",
                  (now(), jdump(summary_mine), jdump(data["summary"])))
        c.execute("DELETE FROM snapshots WHERE ts < ?", (now() - 30 * 86400,))
    hub.publish_sync({"kind": "snapshot", "mine": {s: b["summary"] for s, b in mine.items()},
                      "ecosystem": data["summary"], "ts": now()})

    trigger = set(POLICY["trigger_statuses"])
    with db() as c:
        for service, block in mine.items():
            if service not in SERVICES:
                continue  # service inconnu de la topologie: visible au dashboard, pas d'action
            to_create: list[tuple[str, str, dict[str, Any]]] = []
            for conn in block.get("connectors", []):
                source, status = conn["source"], conn["status"]
                inc = active_incident(c, service, source)
                if status in trigger and inc is None:
                    # Garde anti-boucle : une source déjà diagnostiquée irréparable
                    # (site_source_mort, tentatives épuisées) ne redonne pas
                    # d'incident tant que le délai de re-examen n'est pas écoulé.
                    # Sans elle, un connecteur désactivé côté app mais rapporté
                    # broken à perpétuité par api-ka relance une mission toutes
                    # les ~10 min (2026-08-24 : 37 missions sur origin-north.ai).
                    ab = c.execute(
                        "SELECT updated FROM incidents WHERE service=? AND source=? "
                        "AND state='abandoned' ORDER BY updated DESC LIMIT 1",
                        (service, source)).fetchone()
                    if ab and now() - ab["updated"] < POLICY.get("abandoned_retry_hours", 168) * 3600:
                        continue
                    to_create.append((source, status, conn))
                elif status in trigger and inc is not None and inc["state"] in ("open", "cooldown"):
                    # Ne pas rafraîchir watching/fixing: `updated` sert de chrono
                    # à la fenêtre de surveillance et au cooldown.
                    c.execute("UPDATE incidents SET status_detected=?, detail=? WHERE id=?",
                              (status, jdump(conn), inc["id"]))
                elif status == "ok" and inc is not None:
                    if inc["state"] == "watching":
                        set_incident(c, inc["id"], state="resolved", resolved=now())
                        incident_event(inc["id"], service, source, "resolved", "connecteur de retour à ok — réparation confirmée")
                    elif inc["state"] in ("open", "cooldown"):
                        set_incident(c, inc["id"], state="self_healed", resolved=now())
                        incident_event(inc["id"], service, source, "self_healed", "revenu à ok sans intervention")
                elif status == "retired" and inc is not None and inc["state"] in ("open", "cooldown", "watching"):
                    # api-ka a retiré la source de la supervision (désactivée
                    # côté app ou disparue du journal) : plus rien à réparer.
                    set_incident(c, inc["id"], state="abandoned")
                    incident_event(inc["id"], service, source, "abandoned", "source retirée de la supervision api-ka")

            # Damper d'événement de masse : des centaines de sources qui
            # basculent d'un coup signalent une panne GLOBALE de l'app (sync
            # mort, /api/stats cassé), pas autant de pannes individuelles —
            # missionner source par source serait long et coûteux pour rien
            # (2026-08-24 : 794 incidents jobka créés après ~17 h d'arrêt de
            # job-ka-sync). On n'ouvre rien : la panne d'app se voit au
            # dashboard (bloc _app) et les sources encore cassées une fois la
            # vague retombée sous le seuil recevront leurs incidents.
            limit = POLICY.get("mass_incident_threshold", 20)
            if len(to_create) > limit:
                msg = (f"vague de {len(to_create)} connecteurs en panne sur {service} "
                       f"(seuil {limit}) — panne globale probable de l'app, aucun incident créé")
                print(f"[poll] {msg}", flush=True)
                hub.publish_sync({"kind": "log", "level": "error", "msg": msg, "ts": now()})
                continue
            for source, status, conn in to_create:
                iid = uuid.uuid4().hex[:10]
                c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "
                          "VALUES(?,?,?,?,?,?,?,?)",
                          (iid, service, source, status, "open", now(), now(), jdump(conn)))
                incident_event(iid, service, source, "open", f"détecté {status}")


async def tick() -> None:
    apply_registry()  # même en pause : la topologie doit rester juste
    if LATEST["paused"]:
        return
    try:
        await poll_once()
    except Exception as exc:
        hub.publish_sync({"kind": "log", "level": "warn", "msg": f"poll api-ka échoué: {exc}", "ts": now()})

    with db() as c:
        # 1. Réconciliation des missions en cours depuis > 10 min: on demande au
        # runner ce qu'il en est (événement final perdu? runner redémarré?).
        for m in c.execute("SELECT id FROM missions WHERE state='running' AND started < ?",
                           (now() - 600,)).fetchall():
            if m["id"] not in RECONCILING:
                RECONCILING.add(m["id"])
                asyncio.create_task(reconcile_mission(m["id"]))

        # 2. Watching → rollback si la fenêtre est passée et toujours cassé
        # (fenêtre courte pour les incidents de site: le veilleur mesure aux
        # 2 minutes, pas besoin d'attendre le rythme des scans api-ka)
        for inc in c.execute("SELECT * FROM incidents WHERE state='watching'").fetchall():
            window_h = POLICY.get("site_watch_window_hours", 1) if inc["source"] == "_site" else POLICY["watch_window_hours"]
            if now() - inc["updated"] < window_h * 3600:
                continue
            status = current_status(inc["service"], inc["source"])
            m = c.execute("SELECT * FROM missions WHERE incident_id=? AND base_commit IS NOT NULL "
                          "ORDER BY started DESC LIMIT 1", (inc["id"],)).fetchone()
            if status in POLICY["trigger_statuses"] or status == "degraded":
                if m and (json.loads(m["commits"] or "[]")):
                    asyncio.create_task(rollback_mission(m, "toujours cassé après la fenêtre de surveillance"))
                set_incident(c, inc["id"], state="cooldown")
                incident_event(inc["id"], inc["service"], inc["source"], "cooldown",
                               f"non guéri après {POLICY['watch_window_hours']}h → rollback + cooldown")
            elif status == "ok":
                set_incident(c, inc["id"], state="resolved", resolved=now())
                incident_event(inc["id"], inc["service"], inc["source"], "resolved", "confirmé ok")

        # 3. Cooldown expiré → réouverture ou abandon
        for inc in c.execute("SELECT * FROM incidents WHERE state='cooldown'").fetchall():
            cool_h = POLICY.get("site_cooldown_hours", 1) if inc["source"] == "_site" else POLICY["attempt_cooldown_hours"]
            if now() - inc["updated"] < cool_h * 3600:
                continue
            status = current_status(inc["service"], inc["source"])
            if status == "ok":
                set_incident(c, inc["id"], state="resolved", resolved=now())
                incident_event(inc["id"], inc["service"], inc["source"], "resolved", "guéri pendant le cooldown")
            elif inc["attempts"] >= POLICY["max_attempts_per_incident"]:
                set_incident(c, inc["id"], state="abandoned")
                incident_event(inc["id"], inc["service"], inc["source"], "abandoned",
                               f"{inc['attempts']} tentatives épuisées — intervention humaine requise")
            else:
                set_incident(c, inc["id"], state="open")
                incident_event(inc["id"], inc["service"], inc["source"], "open", "cooldown terminé, nouvelle tentative")

        # 4. Dispatch (1 mission à la fois par agent, broken avant stale, plus vieux d'abord)
        running = c.execute("SELECT COUNT(*) n FROM missions WHERE state='running'").fetchone()["n"]
        if running < POLICY["max_concurrent_missions"]:
            busy_nodes = {m["node"] for m in c.execute("SELECT node FROM missions WHERE state='running'").fetchall()}
            nxt = c.execute(
                "SELECT * FROM incidents WHERE state='open' "
                "ORDER BY CASE status_detected WHEN 'site_down' THEN 0 WHEN 'manual' THEN 0 "
                "WHEN 'effort_new' THEN 0 WHEN 'effort_enrich' THEN 0 WHEN 'broken' THEN 1 ELSE 2 END, created "
            ).fetchall()
            for inc in nxt:
                if inc["service"] not in SERVICES:
                    continue
                if SERVICES[inc["service"]]["node"] in busy_nodes:
                    continue
                set_incident(c, inc["id"], state="dispatched")
                asyncio.create_task(dispatch_row(inc["id"]))
                break


def current_status(service: str, source: str) -> str:
    if source == "_site":
        st = SITE.get(service) or {}
        if not st.get("checked"):
            return "inconnu"
        return "ok" if st.get("ok") else "broken"
    block = LATEST["mine"].get(service) or {}
    for conn in block.get("connectors", []):
        if conn["source"] == source:
            return conn["status"]
    return "inconnu"


async def dispatch_row(iid: str) -> None:
    with db() as c:
        inc = c.execute("SELECT * FROM incidents WHERE id=?", (iid,)).fetchone()
    if inc:
        await dispatch(inc, json.loads(inc["detail"] or "{}"))
        # dispatch() remet fixing; si l'appel a échoué/409, on relâche
        with db() as c:
            cur = c.execute("SELECT state FROM incidents WHERE id=?", (iid,)).fetchone()
            if cur and cur["state"] == "dispatched":
                set_incident(c, iid, state="open")


RECONCILING: set[str] = set()


async def reconcile_mission(mid: str) -> None:
    """Résout une mission « en cours » suspecte auprès de son runner."""
    try:
        with db() as c:
            m = c.execute("SELECT * FROM missions WHERE id=? AND state='running'", (mid,)).fetchone()
        if not m:
            return
        ip = node_ip(m["node"])
        if not ip:
            return  # nœud disparu du registre et de topology.json : on retentera
        try:
            code, body = await lan_post(ip, RUNNER_PORT, "/mission-result", {"mission_id": mid}, timeout=20)
            resp = json.loads(body) if code == 200 else {"status": "erreur"}
        except Exception:
            return  # courrier/runner injoignable: on retentera au prochain tick
        if resp.get("status") == "running":
            return
        with db() as c:
            m = c.execute("SELECT * FROM missions WHERE id=? AND state='running'", (mid,)).fetchone()
            if not m:
                return
            if resp.get("status") == "done":
                data = (resp.get("final") or {}).get("data", {})
                c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",
                          (mid, now(), "final", jdump(data)))
                await finalize(c, m, "final", data)
                hub.publish_sync({"kind": "mission_event", "mission_id": mid, "type": "final",
                                  "data": data, "ts": now()})
            elif now() - m["started"] > 900:
                # Le runner ne connaît pas cette mission: tuée en vol
                # (redémarrage). On la clôt et on relance l'incident.
                c.execute("UPDATE missions SET state='error', ended=? WHERE id=?", (now(), mid))
                inc = c.execute("SELECT * FROM incidents WHERE id=?", (m["incident_id"],)).fetchone()
                if inc and inc["state"] in ("fixing", "dispatched"):
                    set_incident(c, inc["id"], state="open")
                    incident_event(inc["id"], m["service"], m["source"], "open",
                                   "mission perdue (runner interrompu) — remise en file")
    finally:
        RECONCILING.discard(mid)


async def engine() -> None:
    global LOOP
    LOOP = asyncio.get_running_loop()
    await asyncio.sleep(3)
    while True:
        try:
            await tick()
        except Exception as exc:
            import traceback
            traceback.print_exc()
            hub.publish_sync({"kind": "log", "level": "error", "msg": f"tick: {exc}", "ts": now()})
        await asyncio.sleep(POLICY["poll_interval_seconds"])


# ----------------------------------------------------- veilleur de sites ----
# Surveillance du SITE WEB de chaque service assigné (indépendante des
# connecteurs api-ka). Down confirmé → pm2 restart automatique via le runner
# du nœud; toujours down → incident « site_down » et mission d'investigation.

SITE: dict[str, dict[str, Any]] = {}  # service → état du dernier check public


async def check_site(url: str) -> tuple[bool, str]:
    try:
        async with httpx.AsyncClient(follow_redirects=True, timeout=15) as cl:
            r = await cl.get(url, headers={"User-Agent": f"ka-guardian-{AGENT}/1.0"})
        return r.status_code < 500, f"HTTP {r.status_code}"
    except Exception as exc:
        return False, f"{type(exc).__name__}: {exc}"


async def runner_restart(service: str) -> dict[str, Any]:
    """pm2 restart de l'app via le runner de son nœud (endpoint /restart)."""
    svc = SERVICES[service]
    ip = node_ip(svc["node"])
    if not ip or not runner_available(svc["node"], f"pm2 restart {svc['app']}"):
        return {"ok": False, "erreur": f"runner indisponible sur {node_alias(svc['node'])}"}
    try:
        code, body = await lan_post(ip, RUNNER_PORT, "/restart",
                                    {"pm2": svc["pm2"], "web_port": svc["web_port"]}, timeout=240)
        return json.loads(body) if code == 200 else {"ok": False, "erreur": f"runner {code}: {body[:200]}"}
    except Exception as exc:
        return {"ok": False, "erreur": f"{type(exc).__name__}: {exc}"}


async def site_check_service(service: str) -> None:
    svc = SERVICES[service]
    url = svc.get("site")
    if not url:
        return
    up, info = await check_site(url)
    if not up:  # contre-mesure avant de compter l'échec (blip réseau/ngrok)
        await asyncio.sleep(5)
        up, info = await check_site(url)
    st = SITE.setdefault(service, {"fails": 0, "ok": True, "down_since": None, "last_restart": 0.0})
    st.update({"checked": now(), "ok": up, "info": info})
    if up:
        if st["fails"]:
            hub.publish_sync({"kind": "log", "level": "info",
                              "msg": f"site {url} de retour en ligne ({info})", "ts": now()})
        st["fails"], st["down_since"] = 0, None
        with db() as c:
            inc = active_incident(c, service, "_site")
            if inc and inc["state"] in ("open", "cooldown"):
                set_incident(c, inc["id"], state="self_healed", resolved=now())
                incident_event(inc["id"], service, "_site", "self_healed", f"site de retour en ligne ({info})")
        return
    st["fails"] += 1
    st["down_since"] = st["down_since"] or now()
    need = POLICY.get("site_check_fails", 2)
    hub.publish_sync({"kind": "log", "level": "warn",
                      "msg": f"site {url} muet ({info}) — échec {st['fails']}/{need}", "ts": now()})
    if st["fails"] < need:
        return
    with db() as c:
        inc = active_incident(c, service, "_site")
    if inc and inc["state"] in ("dispatched", "fixing"):
        return  # une mission d'investigation travaille déjà sur cette app
    # Étape 1: pm2 restart automatique (throttlé pour ne pas marteler l'app)
    restart_res: dict[str, Any] = {"ok": False, "erreur": "non tenté (throttle)"}
    if now() - st["last_restart"] >= POLICY.get("site_restart_throttle_seconds", 600):
        st["last_restart"] = now()
        hub.publish_sync({"kind": "log", "level": "warn",
                          "msg": f"{svc['app']}: site down confirmé → pm2 restart automatique ({', '.join(svc['pm2'])})",
                          "ts": now()})
        restart_res = await runner_restart(service)
        await asyncio.sleep(POLICY.get("site_restart_wait_seconds", 20))
        up, info = await check_site(url)
        if up:
            st.update({"ok": True, "fails": 0, "down_since": None, "info": info})
            with db() as c:
                if inc:
                    set_incident(c, inc["id"], state="resolved", resolved=now())
                    incident_event(inc["id"], service, "_site", "resolved", f"site rétabli par pm2 restart ({info})")
                else:
                    # Trace au dashboard: le restart automatique a suffi.
                    iid = uuid.uuid4().hex[:10]
                    c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,resolved,detail) "
                              "VALUES(?,?,?,?,?,?,?,?,?)",
                              (iid, service, "_site", "site_down", "resolved", now(), now(), now(),
                               jdump({"note": "rétabli par pm2 restart automatique", "restart": restart_res})))
                    incident_event(iid, service, "_site", "resolved", f"site rétabli par pm2 restart automatique ({info})")
            return
    # Étape 2: toujours down → incident + mission d'investigation (si aucun actif)
    if inc is not None:
        return  # open/watching: la machinerie d'incidents suit déjà son cours
    with db() as c:
        # Garde anti-boucle (même logique que les connecteurs, délai plus court:
        # un site down est urgent, on re-tente quand même chaque jour).
        ab = c.execute("SELECT updated FROM incidents WHERE service=? AND source='_site' "
                       "AND state='abandoned' ORDER BY updated DESC LIMIT 1", (service,)).fetchone()
        if ab and now() - ab["updated"] < POLICY.get("site_abandoned_retry_hours", 24) * 3600:
            return
        detail = {"fails": st["fails"],
                  "down_since": time.strftime("%Y-%m-%d %H:%M", time.localtime(st["down_since"])),
                  "error": info,
                  "restart": "pm2 restart exécuté, site toujours muet" if restart_res.get("ok")
                             else f"restart en échec: {restart_res.get('erreur') or restart_res}",
                  "health": restart_res.get("health")}
        iid = uuid.uuid4().hex[:10]
        c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "
                  "VALUES(?,?,?,?,?,?,?,?)",
                  (iid, service, "_site", "site_down", "open", now(), now(), jdump(detail)))
    incident_event(iid, service, "_site", "open",
                   f"site hors ligne malgré le pm2 restart automatique ({info}) — mission d'investigation demandée")


async def site_engine() -> None:
    await asyncio.sleep(10)
    while True:
        if not LATEST["paused"]:
            for service in sorted(MY_SERVICES & set(SERVICES)):
                try:
                    await site_check_service(service)
                except Exception as exc:
                    hub.publish_sync({"kind": "log", "level": "warn",
                                      "msg": f"veilleur site {service}: {exc}", "ts": now()})
        await asyncio.sleep(POLICY.get("site_check_interval_seconds", 120))


# ------------------------------------------------------------------- app ---

app = FastAPI(title=f"KA Guardian — {AGENT}", docs_url=None, redoc_url=None)


@app.on_event("startup")
async def startup() -> None:
    init_db()
    asyncio.create_task(engine())
    asyncio.create_task(site_engine())


def check_token(tok: str | None) -> None:
    if not TOKEN or tok != TOKEN:
        raise HTTPException(status_code=401)


@app.post("/api/ingest/{mission_id}")
async def ingest(mission_id: str, req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:
    check_token(x_ka_token)
    ev = await req.json()
    etype, data = ev.get("type"), ev.get("data", {})
    with db() as c:
        c.execute("INSERT INTO events(mission_id,ts,type,data) VALUES(?,?,?,?)",
                  (mission_id, now(), etype, jdump(data)))
        m = c.execute("SELECT * FROM missions WHERE id=?", (mission_id,)).fetchone()
        if m:
            if etype == "start":
                c.execute("UPDATE missions SET base_commit=? WHERE id=?", (data.get("base_commit"), mission_id))
            elif etype in ("final", "error"):
                await finalize(c, m, etype, data)
    hub.publish_sync({"kind": "mission_event", "mission_id": mission_id, "type": etype,
                      "data": data, "ts": now()})
    return {"ok": True}


async def finalize(c: sqlite3.Connection, m: sqlite3.Row, etype: str, data: dict[str, Any]) -> None:
    inc = c.execute("SELECT * FROM incidents WHERE id=?", (m["incident_id"],)).fetchone()
    if etype == "error":
        c.execute("UPDATE missions SET state='error', ended=? WHERE id=?", (now(), m["id"]))
        if inc:
            set_incident(c, inc["id"], state="cooldown")
            incident_event(inc["id"], m["service"], m["source"], "cooldown", f"erreur mission: {data.get('error')}")
        return
    verdict = data.get("verdict", {})
    commits = data.get("commits", [])
    health = data.get("health", {})
    c.execute("UPDATE missions SET state='done', ended=?, commits=?, verdict=?, cost_usd=?, num_turns=?, health=?, base_commit=COALESCE(base_commit,?) WHERE id=?",
              (now(), jdump(commits), jdump(verdict), data.get("cost_usd"),
               data.get("num_turns"), jdump(health), data.get("base_commit"), m["id"]))
    m2 = c.execute("SELECT * FROM missions WHERE id=?", (m["id"],)).fetchone()
    v = verdict.get("verdict", "inconnu")
    if not inc:
        return
    if inc["status_detected"] in ("effort_new", "effort_enrich", "effort_degrade"):
        # Les efforts commandés ne passent pas par la surveillance api-ka:
        # verdict + santé de l'app décident tout de suite.
        if not health.get("ok", True):
            asyncio.create_task(rollback_mission(m2, "healthcheck app en échec post-effort"))
            set_incident(c, inc["id"], state="abandoned")
            incident_event(inc["id"], m["service"], m["source"], "abandoned", "app en mauvaise santé → ROLLBACK immédiat")
        elif v == "livre":
            set_incident(c, inc["id"], state="resolved", resolved=now())
            incident_event(inc["id"], m["service"], m["source"], "resolved",
                           f"effort livré ({len(commits)} commit{'s' if len(commits) > 1 else ''})")
        elif v == "echec" and commits:
            asyncio.create_task(rollback_mission(m2, "effort en échec déclaré → rollback préventif"))
            set_incident(c, inc["id"], state="abandoned")
            incident_event(inc["id"], m["service"], m["source"], "abandoned", "effort en échec → rollback")
        else:
            # Session interrompue (plafond durée/coût) ou verdict illisible:
            # l'app est saine, les commits incrémentaux testés sont conservés.
            set_incident(c, inc["id"], state="abandoned")
            incident_event(inc["id"], m["service"], m["source"], "abandoned",
                           f"effort terminé sans verdict livré ({v}) — {len(commits)} commit(s) conservé(s), app saine")
        return
    if not health.get("ok", True):
        # L'app est tombée → rollback immédiat, quoi qu'ait dit l'agent.
        asyncio.create_task(rollback_mission(m2, "healthcheck app en échec post-mission"))
        set_incident(c, inc["id"], state="cooldown")
        incident_event(inc["id"], m["service"], m["source"], "cooldown", "app en mauvaise santé → ROLLBACK immédiat")
    elif v == "repare" and commits:
        set_incident(c, inc["id"], state="watching")
        incident_event(inc["id"], m["service"], m["source"], "watching",
                       f"réparation déclarée ({len(commits)} commit) — surveillance {POLICY['watch_window_hours']}h")
    elif v in ("echec", "inconnu") and commits:
        asyncio.create_task(rollback_mission(m2, f"verdict {v} avec commits → annulation par prudence"))
        set_incident(c, inc["id"], state="cooldown")
        incident_event(inc["id"], m["service"], m["source"], "cooldown", f"verdict {v} → rollback préventif")
    elif v == "site_source_mort":
        set_incident(c, inc["id"], state="abandoned")
        incident_event(inc["id"], m["service"], m["source"], "abandoned", "source définitivement morte (diagnostic agent)")
    elif v in ("rien_a_faire", "repare"):
        set_incident(c, inc["id"], state="watching")
        incident_event(inc["id"], m["service"], m["source"], "watching", f"verdict {v} — surveillance")
    else:
        set_incident(c, inc["id"], state="cooldown")
        incident_event(inc["id"], m["service"], m["source"], "cooldown", f"verdict {v}")


@app.get("/api/state")
def state() -> dict[str, Any]:
    with db() as c:
        incidents = [dict(r) for r in c.execute(
            "SELECT * FROM incidents ORDER BY created DESC LIMIT 200").fetchall()]
        missions = [dict(r) for r in c.execute(
            "SELECT * FROM missions ORDER BY started DESC LIMIT 100").fetchall()]
        stats = c.execute("""SELECT
          (SELECT COUNT(*) FROM missions) n_missions,
          (SELECT COUNT(*) FROM incidents WHERE state='resolved') n_resolved,
          (SELECT COUNT(*) FROM incidents WHERE state IN ('open','dispatched','fixing','watching','cooldown')) n_active,
          (SELECT COUNT(*) FROM incidents WHERE state='abandoned') n_abandoned,
          (SELECT COUNT(*) FROM events WHERE type='rollback') n_rollbacks,
          (SELECT ROUND(SUM(cost_usd),2) FROM missions) cost_total,
          (SELECT ROUND(AVG(resolved-created)/3600.0,1) FROM incidents WHERE state='resolved' AND resolved IS NOT NULL) mttr_h
        """).fetchone()
        history = [dict(r) for r in c.execute(
            "SELECT ts, mine FROM snapshots WHERE ts > ? ORDER BY ts", (now() - 7 * 86400,)).fetchall()]
    return {
        "agent": AGENT, "identity": {k: ME.get(k) for k in ("domain", "accent", "accent2", "accent_soft", "tagline", "model")},
        "services": {s: {**SERVICES[s], "node_alias": node_alias(SERVICES[s]["node"]),
                         "lan_ip": node_ip(SERVICES[s]["node"]),
                         "runner": registry.runner_status(node_ip(SERVICES[s]["node"]))}
                     for s in ME["services"] if s in SERVICES},
        "registry": registry.info(),
        "siblings": {a: {"domain": v["domain"], "accent": v["accent"], "services": v["services"]}
                     for a, v in TOPO["agents"].items() if a != AGENT},
        "latest": LATEST, "sites": SITE, "incidents": incidents, "missions": missions,
        "stats": dict(stats), "history": history, "policy": POLICY, "now": now(),
    }


@app.get("/api/missions/{mission_id}")
def mission_detail(mission_id: str) -> dict[str, Any]:
    with db() as c:
        m = c.execute("SELECT * FROM missions WHERE id=?", (mission_id,)).fetchone()
        if not m:
            raise HTTPException(status_code=404)
        evs = [dict(r) for r in c.execute(
            "SELECT ts,type,data FROM events WHERE mission_id=? ORDER BY id", (mission_id,)).fetchall()]
    return {"mission": dict(m), "events": evs}


@app.get("/events")
async def sse(request: Request) -> StreamingResponse:
    q: asyncio.Queue = asyncio.Queue()
    hub.clients.add(q)

    async def gen():
        try:
            yield f"data: {jdump({'kind': 'hello', 'agent': AGENT, 'ts': now()})}\n\n"
            while True:
                if await request.is_disconnected():
                    break
                try:
                    msg = await asyncio.wait_for(q.get(), timeout=25)
                    yield f"data: {jdump(msg)}\n\n"
                except asyncio.TimeoutError:
                    yield ": keepalive\n\n"
        finally:
            hub.clients.discard(q)

    return StreamingResponse(gen(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})


@app.get("/api/connectors/{service}")
def connectors_list(service: str) -> dict[str, Any]:
    """Sources connues du service (pour le menu déroulant d'enrichissement)."""
    block = LATEST["mine"].get(service) or {}
    return {"service": service,
            "connectors": sorted(
                ({"source": c["source"], "status": c["status"]} for c in block.get("connectors", [])),
                key=lambda x: x["source"])}


@app.post("/api/admin/effort")
async def admin_effort(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:
    """Commande un effort autonome: nouveau connecteur ou enrichissement."""
    check_token(x_ka_token)
    body = await req.json()
    kind = body.get("kind")
    service = body.get("service")
    note = (body.get("note") or "").strip()[:2000]
    if kind not in ("effort_new", "effort_enrich", "effort_degrade") or service not in SERVICES:
        raise HTTPException(status_code=400, detail="kind ou service invalide")
    if kind == "effort_enrich":
        source = (body.get("source") or "").strip()
        if not source:
            raise HTTPException(status_code=400, detail="source requise pour un enrichissement")
    elif kind == "effort_degrade":
        source = f"dégradés·{uuid.uuid4().hex[:6]}"
    else:
        source = f"nouveau·{uuid.uuid4().hex[:6]}"
    detail: dict[str, Any] = {"note": note, "kind": kind}
    try:
        if body.get("max_minutes"):
            detail["max_minutes"] = max(10, min(240, int(body["max_minutes"])))
        if body.get("max_cost_usd"):
            detail["max_cost_usd"] = max(0.5, min(100.0, float(body["max_cost_usd"])))
    except (TypeError, ValueError):
        raise HTTPException(status_code=400, detail="max_minutes/max_cost_usd invalides")
    iid = uuid.uuid4().hex[:10]
    with db() as c:
        c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "
                  "VALUES(?,?,?,?,?,?,?,?)",
                  (iid, service, source, kind, "open", now(), now(), jdump(detail)))
    labels = {"effort_new": "nouveau connecteur", "effort_enrich": f"enrichissement de {source}",
              "effort_degrade": "inspection des connecteurs dégradés"}
    incident_event(iid, service, source, "open",
                   f"effort commandé: {labels[kind]}"
                   + (f" · ≤{detail['max_minutes']} min" if detail.get("max_minutes") else "")
                   + (f" · ≤{detail['max_cost_usd']} $" if detail.get("max_cost_usd") else ""))
    return {"ok": True, "incident_id": iid}


@app.post("/api/admin/mission")
async def admin_mission(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:
    """Déclenche manuellement une mission sur un connecteur (même sain)."""
    check_token(x_ka_token)
    body = await req.json()
    service, source = body["service"], body["source"]
    if service not in SERVICES:
        raise HTTPException(status_code=400, detail="service inconnu")
    detail = {"source": source, "status": "manual", "message": body.get("note", "mission manuelle")}
    for conn in (LATEST["mine"].get(service) or {}).get("connectors", []):
        if conn["source"] == source:
            detail = {**conn, "status": "manual"}
    with db() as c:
        inc = active_incident(c, service, source)
        if inc is None:
            iid = uuid.uuid4().hex[:10]
            c.execute("INSERT INTO incidents(id,service,source,status_detected,state,created,updated,detail) "
                      "VALUES(?,?,?,?,?,?,?,?)",
                      (iid, service, source, "manual", "open", now(), now(), jdump(detail)))
        else:
            iid = inc["id"]
            set_incident(c, iid, state="open", status_detected="manual")
    incident_event(iid, service, source, "open", "mission manuelle demandée")
    return {"ok": True, "incident_id": iid}


@app.post("/api/admin/pause")
async def admin_pause(req: Request, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:
    check_token(x_ka_token)
    body = await req.json()
    LATEST["paused"] = bool(body.get("paused", True))
    return {"ok": True, "paused": LATEST["paused"]}


@app.get("/health")
def health() -> dict[str, Any]:
    return {"ok": True, "agent": AGENT, "polled": LATEST["polled"], "paused": LATEST["paused"]}


@app.get("/favicon.svg")
def favicon() -> FileResponse:
    return FileResponse(BASE / "web" / f"favicon-{AGENT}.svg", media_type="image/svg+xml")


# Safari (macOS/iOS) ne charge pas les favicons SVG → fallback PNG/ICO obligatoire,
# sinon l'onglet affiche un « ? ». ICO multi-tailles + PNG 32 + apple-touch 180.
@app.get("/favicon.ico")
def favicon_ico() -> FileResponse:
    return FileResponse(BASE / "web" / f"favicon-{AGENT}.ico", media_type="image/x-icon")


@app.get("/favicon.png")
def favicon_png() -> FileResponse:
    return FileResponse(BASE / "web" / f"favicon-{AGENT}-32.png", media_type="image/png")


@app.get("/apple-touch-icon.png")
def apple_touch() -> FileResponse:
    return FileResponse(BASE / "web" / f"apple-touch-{AGENT}.png", media_type="image/png")


@app.get("/og.png")
def og_image() -> FileResponse:
    """Bannière de partage (Open Graph) propre à l'agent."""
    return FileResponse(BASE / "web" / f"og-{AGENT}.png", media_type="image/png")


def _render_page(name: str) -> str:
    """Les pages sont statiques mais servies par agent : on remplace au
    démarrage les jetons de partage (OG) par l'identité de l'agent —
    les robots des réseaux sociaux n'exécutent pas le JS."""
    html = (BASE / "web" / name).read_text(encoding="utf-8")
    for token, value in {
        "__NUM__": AGENT[-1],
        "__DOMAIN__": ME["domain"],
        "__TAGLINE__": ME.get("tagline", ""),
    }.items():
        html = html.replace(token, value)
    return html


PAGES = {name: _render_page(name) for name in ("index.html", "commander.html")}


@app.get("/")
def index() -> HTMLResponse:
    return HTMLResponse(PAGES["index.html"])


@app.get("/commander")
def commander() -> HTMLResponse:
    return HTMLResponse(PAGES["commander.html"])


app.mount("/static", StaticFiles(directory=BASE / "web"), name="static")


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=ME["port"])