ui/app.py

"""Test UI + API for the union command agent.  python ui/app.py  ->  http://localhost:8000
Model (numpy runtime) -> executor plan (dry-run by default) -> optional execution with risk gating."""
import json, os, sys, platform as _pf
HERE = os.path.dirname(os.path.abspath(__file__)); ROOT = os.path.dirname(HERE)
sys.path.insert(0, os.path.join(ROOT, "model")); sys.path.insert(0, os.path.join(ROOT, "executor"))
from fastapi import FastAPI  # noqa
from fastapi.responses import HTMLResponse, JSONResponse  # noqa
from pydantic import BaseModel  # noqa
from runtime import CommandModel  # noqa
import executor as EX  # noqa

MODEL = CommandModel(os.path.join(ROOT, "model"))
CAT = EX.CATALOG
app = FastAPI(title="union command agent")

class Req(BaseModel):
    text: str; platform: str = "auto"; execute: bool = False; allow_caution: bool = False; confirm: str = ""
class Batch(BaseModel):
    texts: list; platform: str = "auto"

def _plat(p): return EX.platform_name() if p in ("auto", "", None) else ("win" if p.startswith("win") else "pi")

def handle(text, platform="auto", execute=False, allow_caution=False, confirm=""):
    plat = _plat(platform); out = MODEL.predict(text); cmd = {k: v for k, v in out.items() if k != "_meta"}
    p = EX.plan(cmd, plat, text); ok, why = EX.gate(p, allow_caution, confirm or None)
    res = {"text": text, "platform": plat, "model": cmd, "meta": out["_meta"], "plan": p.to_dict(), "gate": {"ok": ok, "reason": why}, "spec": CAT.get(cmd["action"], {}), "run": None}
    if execute:
        res["run"] = EX.run(p, cmd, dry_run=False) if ok else {"status": "blocked", "output": why, "command": p.command}
    return res

@app.get("/", response_class=HTMLResponse)
def index(): return open(os.path.join(HERE, "static", "index.html"), encoding="utf-8").read()
@app.get("/api/info")
def info(): return {"actions": len(CAT), "host": _pf.node(), "os": _pf.system(), "platform": EX.platform_name(), "model_params": int(sum(v.size for v in MODEL.W.values()))}
@app.get("/api/prompts")
def prompts(): return json.load(open(os.path.join(HERE, "prompts.json"), encoding="utf-8"))
@app.get("/api/catalog")
def catalog(): return CAT
@app.post("/api/predict")
def predict(r: Req): return handle(r.text, r.platform, r.execute, r.allow_caution, r.confirm)
@app.post("/api/batch")
def batch(b: Batch): return [handle(t, b.platform) for t in b.texts if t.strip()]

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=int(os.environ.get("PORT", 8000)))