spb/ka-guardian
Public
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 tf.flush() # le transcript doit survivre à une mort brutale du runner241 if time.time() > deadline:242 proc.kill()243 post_event(m.callback_url, {"type": "error", "data": {"error": f"durée max atteinte ({m.timeout_seconds // 60} min) — session interrompue"}})244 break245 line = line.strip()246 if not line:247 continue248 try:249 obj = json.loads(line)250 except Exception:251 continue252 if obj.get("type") == "assistant":253 spent_estimate += usage_cost_usd(obj.get("message", {}).get("usage") or {}, m.model)254 if m.max_cost_usd and spent_estimate > m.max_cost_usd:255 proc.kill()256 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"}})257 break258 for ev in condense_stream_line(obj):259 if ev["type"] == "result":260 result_text = ev["data"].get("result", "")261 cost = ev["data"].get("cost_usd")262 turns = ev["data"].get("num_turns")263 post_event(m.callback_url, ev)264 proc.wait(timeout=60)265266 commits_raw = git(m.dir, "rev-list", "--oneline", f"{base_commit}..HEAD").stdout.strip()267 commits = commits_raw.splitlines() if commits_raw else []268 verdict = extract_verdict(result_text)269 health = healthcheck(m.web_port, m.pm2)270 final = {"type": "final", "data": {271 "verdict": verdict, "commits": commits, "base_commit": base_commit,272 "cost_usd": cost, "num_turns": turns, "health": health,273 "exit_code": proc.returncode,274 }}275 # Résultat final persisté: l'orchestrateur peut le réclamer même si276 # l'événement s'est perdu (redémarrage, courrier en panne).277 (TRANSCRIPTS / f"{m.mission_id}.final.json").write_text(278 json.dumps(final, ensure_ascii=False))279 post_event(m.callback_url, final)280 except Exception as exc:281 post_event(m.callback_url, {"type": "error", "data": {"error": str(exc), "base_commit": base_commit}})282 finally:283 with _lock:284 _current.clear()285286287@app.get("/health")288def health() -> dict[str, Any]:289 return {"ok": True, "node": NODE, "busy": bool(_current), "current": _current.get("mission_id")}290291292@app.post("/missions")293def missions(m: MissionIn, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:294 check_token(x_ka_token)295 workdir = expand(m.dir)296 if not os.path.isdir(workdir):297 raise HTTPException(status_code=400, detail=f"dir introuvable: {workdir}")298 with _lock:299 if _current:300 raise HTTPException(status_code=409, detail=f"mission déjà en cours: {_current.get('mission_id')}")301 _current.update({"mission_id": m.mission_id, "service": m.service, "source": m.source, "started": time.time()})302 threading.Thread(target=run_mission, args=(m,), daemon=True).start()303 return {"accepted": True, "node": NODE}304305306class MissionResultIn(BaseModel):307 mission_id: str308309310@app.post("/mission-result")311def mission_result(q: MissionResultIn, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:312 """Réconciliation: l'orchestrateur réclame le sort d'une mission.313314 running = encore en cours ici; done = terminée (payload final joint);315 unknown = jamais terminée ici (probablement tuée) — à l'orchestrateur de trancher.316 """317 check_token(x_ka_token)318 if _current.get("mission_id") == q.mission_id:319 return {"status": "running"}320 fpath = TRANSCRIPTS / f"{q.mission_id}.final.json"321 if fpath.exists():322 return {"status": "done", "final": json.loads(fpath.read_text())}323 return {"status": "unknown"}324325326class RollbackIn(BaseModel):327 dir: str328 base_commit: str329 pm2: list[str]330 web_port: int331332333@app.post("/rollback")334def rollback(r: RollbackIn, x_ka_token: str | None = Header(default=None)) -> dict[str, Any]:335 """Revient au commit d'avant mission (reset --hard si tous les commits sont de l'agent, sinon revert)."""336 check_token(x_ka_token)337 log = git(r.dir, "log", "--format=%H %s", f"{r.base_commit}..HEAD").stdout.strip()338 entries = [ln.split(" ", 1) for ln in log.splitlines() if ln.strip()]339 agent_shas = [h for h, s in entries if s.startswith("[ka")]340 if not entries:341 mode = "aucun_commit"342 else:343 # Revert CIBLÉ, du plus récent au plus ancien, UNIQUEMENT les commits de344 # l'agent — jamais reset --hard ni revert de plage : les commits étrangers345 # (docs, humains, autres sessions) doivent survivre au rollback (leçon346 # fabri-ka 2026-08-24 : un revert de plage avait emporté la page /doc et347 # un connecteur), et les commits déjà poussés sur spbgit ne doivent348 # jamais exiger de force-push.349 reverted: list[str] = []350 skipped: list[str] = []351 for sha in agent_shas: # git log liste déjà du plus récent au plus ancien352 rc = git(r.dir, "revert", "--no-edit", sha)353 if rc.returncode != 0:354 git(r.dir, "revert", "--abort")355 skipped.append(sha[:9])356 else:357 reverted.append(sha[:9])358 mode = f"revert_cible({len(reverted)}/{len(agent_shas)})"359 if skipped:360 mode += f" sautés:{','.join(skipped)}"361 # Pas de git clean : les process de sync de l'app écrivent en continu,362 # on ne supprime jamais de fichiers non trackés apparus pendant la mission.363 # Push simple (les reverts s'empilent, fast-forward). Échec attendu sous364 # macOS 26 Local Network Privacy (runner = python) : le pousseur launchd365 # (com.ka.pousseur, chaîne binaires Apple) poussera dans les 5 minutes.366 pushed = None367 if entries:368 p = git(r.dir, "push", "origin", "HEAD")369 pushed = p.returncode == 0370 for name in r.pm2:371 sh(f"pm2 restart {shlex.quote(name)} --update-env", timeout=180)372 time.sleep(8)373 h = healthcheck(r.web_port, r.pm2)374 return {"ok": True, "mode": mode, "pushed": pushed, "health": h, "head": git(r.dir, "rev-parse", "HEAD").stdout.strip()}375376377if __name__ == "__main__":378 import uvicorn379 uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("KA_GUARDIAN_RUNNER_PORT", "7791")))380