# ============================================
# 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, StreamingResponse
from fastapi.staticfiles import StaticFiles

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: dict[str, Any] = TOPO["services"]
NODES: dict[str, Any] = TOPO["nodes"]
MY_SERVICES: set[str] = set(ME["services"])
RUNNER_PORT = TOPO["runner_port"]

# 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])
    node_alias = NODES[svc["node"]]["alias"]
    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>"`. NE PUSH JAMAIS. Si tu as touché plusieurs fichiers pour des raisons distinctes, fais des commits séparés et clairs.

== 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; 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, ~/.ssh, clés API, suppression de données.
- À la fin: `pm2 restart <process de sync>`, vérifie que le site répond (curl localhost), puis committe en messages clairs préfixés [%AGENT%]. NE PUSH JAMAIS."""


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])
    node_alias = NODES[svc["node"]]["alias"]
    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_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.

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

{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"]}}"""


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 = NODES[node]["lan_ip"]
    mid = uuid.uuid4().hex[:12]
    my_ip = os.environ.get("KA_GUARDIAN_SELF_IP", "192.168.2.69")
    kind = incident["status_detected"]
    if kind in ("effort_new", "effort_enrich"):
        prompt = build_effort_prompt(kind, service, source, health.get("note", ""))
        max_turns, timeout = POLICY.get("effort_max_turns", 150), POLICY.get("effort_timeout_seconds", 7200)
    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,
        "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 {NODES[node]['alias']}")
    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"]]
    ip = NODES[mission["node"]]["lan_ip"]
    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
            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:
                    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}")
                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")


async def tick() -> None:
    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. Missions zombies (runner mort / callback perdu)
        for m in c.execute("SELECT * FROM missions WHERE state='running' AND started < ?",
                           (now() - POLICY.get("effort_timeout_seconds", 7200) - 900,)).fetchall():
            c.execute("UPDATE missions SET state='error', ended=? WHERE id=?", (now(), m["id"]))
            inc = c.execute("SELECT * FROM incidents WHERE id=?", (m["incident_id"],)).fetchone()
            if inc:
                set_incident(c, inc["id"], state="cooldown")
                incident_event(inc["id"], m["service"], m["source"], "cooldown", "mission sans réponse (zombie)")

        # 2. Watching → rollback si la fenêtre est passée et toujours cassé
        for inc in c.execute("SELECT * FROM incidents WHERE state='watching'").fetchall():
            if now() - inc["updated"] < POLICY["watch_window_hours"] * 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():
            if now() - inc["updated"] < POLICY["attempt_cooldown_hours"] * 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 '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:
    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")


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"])


# ------------------------------------------------------------------- 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())


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"):
        # 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 ''})")
        else:
            if commits:
                asyncio.create_task(rollback_mission(m2, f"effort verdict {v} → rollback préventif"))
            set_incident(c, inc["id"], state="abandoned")
            incident_event(inc["id"], m["service"], m["source"], "abandoned", f"effort en échec ({v})")
        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": NODES[SERVICES[s]["node"]]["alias"]} for s in ME["services"]},
        "siblings": {a: {"domain": v["domain"], "accent": v["accent"], "services": v["services"]}
                     for a, v in TOPO["agents"].items() if a != AGENT},
        "latest": LATEST, "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") 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")
    else:
        source = f"nouveau·{uuid.uuid4().hex[:6]}"
    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({"note": note, "kind": kind})))
    incident_event(iid, service, source, "open",
                   "effort commandé: " + ("nouveau connecteur" if kind == "effort_new" else f"enrichissement de {source}"))
    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")


@app.get("/")
def index() -> FileResponse:
    return FileResponse(BASE / "web" / "index.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"])