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    prompt: str114    callback_url: str  # http://<orchestrateur>/api/ingest/<mission_id>115116117SPOOL = HOME / "ka-guardian-spool"118119120def post_event(url: str, payload: dict[str, Any]) -> None:121    """Dépose l'événement dans le spool du courrier zsh (deploy/courier.sh).122123    macOS 26 « Local Network Privacy » refuse le trafic LAN dès que python124    (homebrew) est dans la chaîne de processus — même ses enfants ssh/curl.125    Le courrier (launchd 100 % zsh) expédie via ssh → curl localhost.126    """127    try:128        from urllib.parse import urlsplit129        u = urlsplit(url)130        for d in ("outbox", "done", "tmp"):131            (SPOOL / d).mkdir(parents=True, exist_ok=True)132        # id préfixé du temps ns: le courrier traite les jobs en ordre lexical,133        # les événements arrivent donc dans l'ordre d'émission.134        jid = f"{time.time_ns()}-{uuid.uuid4().hex[:8]}"135        tmp = SPOOL / "tmp" / f"{jid}.job"136        tmp.write_text(f"{u.hostname} {u.port or 80} {u.path} 15\n"137                       + json.dumps(payload, ensure_ascii=False))138        tmp.rename(SPOOL / "outbox" / f"{jid}.job")139    except Exception:140        pass  # l'orchestrateur relira le transcript au besoin141142143def condense_stream_line(obj: dict[str, Any]) -> list[dict[str, Any]]:144    """Transforme une ligne stream-json de claude en événements courts pour le feed."""145    events: list[dict[str, Any]] = []146    t = obj.get("type")147    if t == "system" and obj.get("subtype") == "init":148        events.append({"type": "init", "data": {"model": obj.get("model", "?")}})149    elif t == "assistant":150        for block in obj.get("message", {}).get("content", []):151            if block.get("type") == "text" and block.get("text", "").strip():152                events.append({"type": "text", "data": {"text": block["text"][:1500]}})153            elif block.get("type") == "tool_use":154                inp = json.dumps(block.get("input", {}), ensure_ascii=False)155                events.append({"type": "tool", "data": {"name": block.get("name", "?"), "input": inp[:400]}})156    elif t == "result":157        events.append({"type": "result", "data": {158            "subtype": obj.get("subtype"),159            "result": (obj.get("result") or "")[:4000],160            "cost_usd": obj.get("total_cost_usd"),161            "num_turns": obj.get("num_turns"),162            "duration_ms": obj.get("duration_ms"),163        }})164    return events165166167def extract_verdict(result_text: str) -> dict[str, Any]:168    """Extrait le dernier bloc JSON {verdict: ...} de la réponse finale."""169    import re170    for m in reversed(re.findall(r"\{[^{}]*\"verdict\"[\s\S]*?\}", result_text)):171        try:172            v = json.loads(m)173            if "verdict" in v:174                return v175        except Exception:176            continue177    return {"verdict": "inconnu", "diagnostic": result_text[-500:] if result_text else ""}178179180def run_mission(m: MissionIn) -> None:181    global _current182    workdir = expand(m.dir)183    transcript = TRANSCRIPTS / f"{m.mission_id}.jsonl"184    base_commit = ""185    try:186        # Snapshot pré-mission : tree sale → commit de sûreté pour un rollback net.187        if git(m.dir, "status", "--porcelain").stdout.strip():188            git(m.dir, "add", "-A")189            git(m.dir, "commit", "-m", f"[{m.agent}] snapshot pré-mission {m.source}")190        base_commit = git(m.dir, "rev-parse", "HEAD").stdout.strip()191        post_event(m.callback_url, {"type": "start", "data": {"base_commit": base_commit, "node": NODE}})192193        env = {194            **os.environ,195            **load_env_file(CLAUDE_ENV_FILE),196            "PATH": "/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin",197            "HOME": str(HOME),198        }199        cmd = [200            CLAUDE_BIN, "-p", m.prompt,201            "--output-format", "stream-json", "--verbose",202            "--model", m.model,203            "--max-turns", str(m.max_turns),204            "--dangerously-skip-permissions",205        ]206        proc = subprocess.Popen(207            cmd, cwd=workdir, env=env,208            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1,209        )210        _current["pid"] = proc.pid211        result_text, cost, turns = "", None, None212        deadline = time.time() + m.timeout_seconds213        with transcript.open("w") as tf:214            for line in proc.stdout:  # type: ignore[union-attr]215                tf.write(line)216                if time.time() > deadline:217                    proc.kill()218                    post_event(m.callback_url, {"type": "error", "data": {"error": "timeout mission"}})219                    break220                line = line.strip()221                if not line:222                    continue223                try:224                    obj = json.loads(line)225                except Exception:226                    continue227                for ev in condense_stream_line(obj):228                    if ev["type"] == "result":229                        result_text = ev["data"].get("result", "")230                        cost = ev["data"].get("cost_usd")231                        turns = ev["data"].get("num_turns")232                    post_event(m.callback_url, ev)233        proc.wait(timeout=60)234235        commits_raw = git(m.dir, "rev-list", "--oneline", f"{base_commit}..HEAD").stdout.strip()236        commits = commits_raw.splitlines() if commits_raw else []237        verdict = extract_verdict(result_text)238        health = healthcheck(m.web_port, m.pm2)239        post_event(m.callback_url, {"type": "final", "data": {240            "verdict": verdict, "commits": commits, "base_commit": base_commit,241            "cost_usd": cost, "num_turns": turns, "health": health,242            "exit_code": proc.returncode,243        }})244    except Exception as exc:245        post_event(m.callback_url, {"type": "error", "data": {"error": str(exc), "base_commit": base_commit}})246    finally:247        with _lock:248            _current.clear()249250251@app.get("/health")252def health() -> dict[str, Any]:253    return {"ok": True, "node": NODE, "busy": bool(_current), "current": _current.get("mission_id")}254255256@app.post("/missions")257def missions(m: MissionIn, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:258    check_token(x_ka_token)259    workdir = expand(m.dir)260    if not os.path.isdir(workdir):261        raise HTTPException(status_code=400, detail=f"dir introuvable: {workdir}")262    with _lock:263        if _current:264            raise HTTPException(status_code=409, detail=f"mission déjà en cours: {_current.get('mission_id')}")265        _current.update({"mission_id": m.mission_id, "service": m.service, "source": m.source, "started": time.time()})266    threading.Thread(target=run_mission, args=(m,), daemon=True).start()267    return {"accepted": True, "node": NODE}268269270class RollbackIn(BaseModel):271    dir: str272    base_commit: str273    pm2: list[str]274    web_port: int275276277@app.post("/rollback")278def rollback(r: RollbackIn, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:279    """Revient au commit d'avant mission (reset --hard si tous les commits sont de l'agent, sinon revert)."""280    check_token(x_ka_token)281    log = git(r.dir, "log", "--format=%s", f"{r.base_commit}..HEAD").stdout.strip()282    msgs = log.splitlines() if log else []283    if not msgs:284        mode = "aucun_commit"285    elif all(s.startswith("[ka") for s in msgs):286        git(r.dir, "reset", "--hard", r.base_commit)287        mode = "reset_hard"288    else:289        rc = git(r.dir, "revert", "--no-edit", f"{r.base_commit}..HEAD")290        if rc.returncode != 0:291            git(r.dir, "revert", "--abort")292            git(r.dir, "reset", "--hard", r.base_commit)293            mode = "reset_hard(fallback)"294        else:295            mode = "revert"296    # Pas de git clean : les process de sync de l'app écrivent en continu,297    # on ne supprime jamais de fichiers non trackés apparus pendant la mission.298    for name in r.pm2:299        sh(f"pm2 restart {shlex.quote(name)} --update-env", timeout=180)300    time.sleep(8)301    h = healthcheck(r.web_port, r.pm2)302    return {"ok": True, "mode": mode, "health": h, "head": git(r.dir, "rev-parse", "HEAD").stdout.strip()}303304305if __name__ == "__main__":306    import uvicorn307    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("KA_GUARDIAN_RUNNER_PORT", "7791")))308