spb/ka-guardian

1# ============================================2# Projet   : KA Guardian3# Fichier  : runner/runner.py4# Rôle     : Bras d'exécution sur chaque nœud d'app — lance les missions5#            claude -p headless dans le repo de l'app, streame les événements6#            vers l'orchestrateur (ka2/ka4/ka6), et sait rollback + restart.7# Author   : Simon-Pierre Boucher8# Date     : 2026-08-239# ============================================10"""KA Guardian Runner.1112Un seul runner par nœud (port 7791), une seule mission à la fois par nœud.13Auth: header X-KA-Token == $KA_GUARDIAN_TOKEN (fichier ~/.ka-guardian.env).14"""15from __future__ import annotations1617import json18import os19import pathlib20import shlex21import socket22import subprocess23import threading24import time25import uuid26from typing import Any2728import httpx29from fastapi import FastAPI, Header, HTTPException, Request30from pydantic import BaseModel3132HOME = pathlib.Path.home()33ENV_FILE = HOME / ".ka-guardian.env"34CLAUDE_ENV_FILE = HOME / ".claude" / ".env"35TRANSCRIPTS = HOME / "ka-guardian-runner" / "transcripts"36TRANSCRIPTS.mkdir(parents=True, exist_ok=True)37CLAUDE_BIN = "/opt/homebrew/bin/claude"383940def load_env_file(path: pathlib.Path) -> dict[str, str]:41    out: dict[str, str] = {}42    if path.exists():43        for line in path.read_text().splitlines():44            line = line.strip()45            if line and not line.startswith("#") and "=" in line:46                k, v = line.split("=", 1)47                out[k.strip()] = v.strip().strip('"').strip("'")48    return out495051LOCAL_ENV = load_env_file(ENV_FILE)52TOKEN = LOCAL_ENV.get("KA_GUARDIAN_TOKEN", "")53NODE = LOCAL_ENV.get("KA_GUARDIAN_NODE", socket.gethostname())5455app = FastAPI(title="KA Guardian Runner", docs_url=None, redoc_url=None)5657_lock = threading.Lock()58_current: dict[str, Any] = {}  # mission en cours (métadonnées)596061def check_token(x_ka_token: str | None) -> None:62    if not TOKEN or x_ka_token != TOKEN:63        raise HTTPException(status_code=401, detail="token invalide")646566def expand(dir_: str) -> str:67    return os.path.expanduser(dir_)686970def git(dir_: str, *args: str) -> subprocess.CompletedProcess:71    return subprocess.run(72        ["git", "-C", expand(dir_), *args],73        capture_output=True, text=True, timeout=120,74    )757677def sh(cmd: str, timeout: int = 120) -> subprocess.CompletedProcess:78    env = {**os.environ, "PATH": "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin"}79    return subprocess.run(["/bin/zsh", "-c", cmd], capture_output=True, text=True, timeout=timeout, env=env)808182def healthcheck(web_port: int, pm2_names: list[str]) -> dict[str, Any]:83    """Vérifie que l'app répond en HTTP et que ses process pm2 sont online."""84    http_ok = False85    try:86        r = httpx.get(f"http://127.0.0.1:{web_port}/", timeout=10, follow_redirects=True)87        http_ok = r.status_code < 50088    except Exception:89        http_ok = False90    pm2_status: dict[str, str] = {}91    try:92        out = sh("pm2 jlist").stdout93        procs = {p["name"]: p.get("pm2_env", {}).get("status", "?") for p in json.loads(out)}94        for name in pm2_names:95            pm2_status[name] = procs.get(name, "absent")96    except Exception as exc:  # pm2 absent ou jlist illisible97        pm2_status = {n: f"inconnu ({exc})" for n in pm2_names}98    ok = http_ok and all(s in ("online", "launching") for s in pm2_status.values())99    return {"ok": ok, "http_ok": http_ok, "pm2": pm2_status}100101102class MissionIn(BaseModel):103    mission_id: str104    agent: str105    service: str106    source: str107    dir: str108    pm2: list[str]109    web_port: int110    model: str = "sonnet"111    max_turns: int = 70112    timeout_seconds: int = 3600113    max_cost_usd: float | None = None  # plafond opérateur (estimation par usage)114    prompt: str115    callback_url: str  # http://<orchestrateur>/api/ingest/<mission_id>116117118# Tarifs $/MTok (entrée, sortie) — pour ESTIMER le coût en cours de mission et119# appliquer le plafond opérateur. Cache: lecture ≈ 0,1× entrée, écriture ≈ 1,25×.120MODEL_RATES = {"fable": (10.0, 50.0), "opus": (5.0, 25.0), "sonnet": (3.0, 15.0), "haiku": (1.0, 5.0)}121122123def rates_for(model: str) -> tuple[float, float]:124    for key, r in MODEL_RATES.items():125        if key in model:126            return r127    return MODEL_RATES["fable"]128129130def usage_cost_usd(usage: dict[str, Any], model: str) -> float:131    rin, rout = rates_for(model)132    return (133        (usage.get("input_tokens") or 0) * rin134        + (usage.get("output_tokens") or 0) * rout135        + (usage.get("cache_read_input_tokens") or 0) * rin * 0.1136        + (usage.get("cache_creation_input_tokens") or 0) * rin * 1.25137    ) / 1e6138139140SPOOL = HOME / "ka-guardian-spool"141142143def post_event(url: str, payload: dict[str, Any]) -> None:144    """Dépose l'événement dans le spool du courrier zsh (deploy/courier.sh).145146    macOS 26 « Local Network Privacy » refuse le trafic LAN dès que python147    (homebrew) est dans la chaîne de processus — même ses enfants ssh/curl.148    Le courrier (launchd 100 % zsh) expédie via ssh → curl localhost.149    """150    try:151        from urllib.parse import urlsplit152        u = urlsplit(url)153        for d in ("outbox", "done", "tmp"):154            (SPOOL / d).mkdir(parents=True, exist_ok=True)155        # id préfixé du temps ns: le courrier traite les jobs en ordre lexical,156        # les événements arrivent donc dans l'ordre d'émission.157        jid = f"{time.time_ns()}-{uuid.uuid4().hex[:8]}"158        tmp = SPOOL / "tmp" / f"{jid}.job"159        tmp.write_text(f"{u.hostname} {u.port or 80} {u.path} 15\n"160                       + json.dumps(payload, ensure_ascii=False))161        tmp.rename(SPOOL / "outbox" / f"{jid}.job")162    except Exception:163        pass  # l'orchestrateur relira le transcript au besoin164165166def condense_stream_line(obj: dict[str, Any]) -> list[dict[str, Any]]:167    """Transforme une ligne stream-json de claude en événements courts pour le feed."""168    events: list[dict[str, Any]] = []169    t = obj.get("type")170    if t == "system" and obj.get("subtype") == "init":171        events.append({"type": "init", "data": {"model": obj.get("model", "?")}})172    elif t == "assistant":173        for block in obj.get("message", {}).get("content", []):174            if block.get("type") == "text" and block.get("text", "").strip():175                events.append({"type": "text", "data": {"text": block["text"][:1500]}})176            elif block.get("type") == "tool_use":177                inp = json.dumps(block.get("input", {}), ensure_ascii=False)178                events.append({"type": "tool", "data": {"name": block.get("name", "?"), "input": inp[:400]}})179    elif t == "result":180        events.append({"type": "result", "data": {181            "subtype": obj.get("subtype"),182            "result": (obj.get("result") or "")[:4000],183            "cost_usd": obj.get("total_cost_usd"),184            "num_turns": obj.get("num_turns"),185            "duration_ms": obj.get("duration_ms"),186        }})187    return events188189190def extract_verdict(result_text: str) -> dict[str, Any]:191    """Extrait le dernier bloc JSON {verdict: ...} de la réponse finale."""192    import re193    for m in reversed(re.findall(r"\{[^{}]*\"verdict\"[\s\S]*?\}", result_text)):194        try:195            v = json.loads(m)196            if "verdict" in v:197                return v198        except Exception:199            continue200    return {"verdict": "inconnu", "diagnostic": result_text[-500:] if result_text else ""}201202203def run_mission(m: MissionIn) -> None:204    global _current205    workdir = expand(m.dir)206    transcript = TRANSCRIPTS / f"{m.mission_id}.jsonl"207    base_commit = ""208    try:209        # Snapshot pré-mission : tree sale → commit de sûreté pour un rollback net.210        if git(m.dir, "status", "--porcelain").stdout.strip():211            git(m.dir, "add", "-A")212            git(m.dir, "commit", "-m", f"[{m.agent}] snapshot pré-mission {m.source}")213        base_commit = git(m.dir, "rev-parse", "HEAD").stdout.strip()214        post_event(m.callback_url, {"type": "start", "data": {"base_commit": base_commit, "node": NODE}})215216        env = {217            **os.environ,218            **load_env_file(CLAUDE_ENV_FILE),219            "PATH": "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin",220            "HOME": str(HOME),221        }222        cmd = [223            CLAUDE_BIN, "-p", m.prompt,224            "--output-format", "stream-json", "--verbose",225            "--model", m.model,226            "--max-turns", str(m.max_turns),227            "--dangerously-skip-permissions",228        ]229        proc = subprocess.Popen(230            cmd, cwd=workdir, env=env,231            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1,232        )233        _current["pid"] = proc.pid234        result_text, cost, turns = "", None, None235        spent_estimate = 0.0236        deadline = time.time() + m.timeout_seconds237        with transcript.open("w") as tf:238            for line in proc.stdout:  # type: ignore[union-attr]239                tf.write(line)240                if time.time() > deadline:241                    proc.kill()242                    post_event(m.callback_url, {"type": "error", "data": {"error": f"durée max atteinte ({m.timeout_seconds // 60} min) — session interrompue"}})243                    break244                line = line.strip()245                if not line:246                    continue247                try:248                    obj = json.loads(line)249                except Exception:250                    continue251                if obj.get("type") == "assistant":252                    spent_estimate += usage_cost_usd(obj.get("message", {}).get("usage") or {}, m.model)253                    if m.max_cost_usd and spent_estimate > m.max_cost_usd:254                        proc.kill()255                        post_event(m.callback_url, {"type": "error", "data": {"error": f"coût max atteint (~{spent_estimate:.2f} $ estimé, plafond {m.max_cost_usd:.2f} $) — session interrompue"}})256                        break257                for ev in condense_stream_line(obj):258                    if ev["type"] == "result":259                        result_text = ev["data"].get("result", "")260                        cost = ev["data"].get("cost_usd")261                        turns = ev["data"].get("num_turns")262                    post_event(m.callback_url, ev)263        proc.wait(timeout=60)264265        commits_raw = git(m.dir, "rev-list", "--oneline", f"{base_commit}..HEAD").stdout.strip()266        commits = commits_raw.splitlines() if commits_raw else []267        verdict = extract_verdict(result_text)268        health = healthcheck(m.web_port, m.pm2)269        post_event(m.callback_url, {"type": "final", "data": {270            "verdict": verdict, "commits": commits, "base_commit": base_commit,271            "cost_usd": cost, "num_turns": turns, "health": health,272            "exit_code": proc.returncode,273        }})274    except Exception as exc:275        post_event(m.callback_url, {"type": "error", "data": {"error": str(exc), "base_commit": base_commit}})276    finally:277        with _lock:278            _current.clear()279280281@app.get("/health")282def health() -> dict[str, Any]:283    return {"ok": True, "node": NODE, "busy": bool(_current), "current": _current.get("mission_id")}284285286@app.post("/missions")287def missions(m: MissionIn, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:288    check_token(x_ka_token)289    workdir = expand(m.dir)290    if not os.path.isdir(workdir):291        raise HTTPException(status_code=400, detail=f"dir introuvable: {workdir}")292    with _lock:293        if _current:294            raise HTTPException(status_code=409, detail=f"mission déjà en cours: {_current.get('mission_id')}")295        _current.update({"mission_id": m.mission_id, "service": m.service, "source": m.source, "started": time.time()})296    threading.Thread(target=run_mission, args=(m,), daemon=True).start()297    return {"accepted": True, "node": NODE}298299300class RollbackIn(BaseModel):301    dir: str302    base_commit: str303    pm2: list[str]304    web_port: int305306307@app.post("/rollback")308def rollback(r: RollbackIn, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:309    """Revient au commit d'avant mission (reset --hard si tous les commits sont de l'agent, sinon revert)."""310    check_token(x_ka_token)311    log = git(r.dir, "log", "--format=%s", f"{r.base_commit}..HEAD").stdout.strip()312    msgs = log.splitlines() if log else []313    if not msgs:314        mode = "aucun_commit"315    elif all(s.startswith("[ka") for s in msgs):316        git(r.dir, "reset", "--hard", r.base_commit)317        mode = "reset_hard"318    else:319        rc = git(r.dir, "revert", "--no-edit", f"{r.base_commit}..HEAD")320        if rc.returncode != 0:321            git(r.dir, "revert", "--abort")322            git(r.dir, "reset", "--hard", r.base_commit)323            mode = "reset_hard(fallback)"324        else:325            mode = "revert"326    # Pas de git clean : les process de sync de l'app écrivent en continu,327    # on ne supprime jamais de fichiers non trackés apparus pendant la mission.328    for name in r.pm2:329        sh(f"pm2 restart {shlex.quote(name)} --update-env", timeout=180)330    time.sleep(8)331    h = healthcheck(r.web_port, r.pm2)332    return {"ok": True, "mode": mode, "health": h, "head": git(r.dir, "rev-parse", "HEAD").stdout.strip()}333334335if __name__ == "__main__":336    import uvicorn337    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("KA_GUARDIAN_RUNNER_PORT", "7791")))338