agent-runtime/assay_cli.py

"""CLI helper for reading assay API — trace, history, state, send."""

import json
import sys
import httpx

API = "https://assay.loop42.de"
TOKEN = "7Oorb9S3OpwFyWgm4zi_Tq7GeamefbjjTgooPVPWAwPDOf6B4TvgvQlLbhmT4DjsqBS_D1g"
HEADERS = {"Authorization": f"Bearer {TOKEN}", "Content-Type": "application/json"}


def _request(method, path, **kwargs):
    """Make an HTTP request with error handling. Fail fast on any error."""
    timeout = kwargs.pop("timeout", 15)
    try:
        r = getattr(httpx, method)(f"{API}{path}", headers=HEADERS, timeout=timeout, **kwargs)
    except httpx.TimeoutException:
        print(f"TIMEOUT: {method.upper()} {path} (>{timeout}s)", file=sys.stderr)
        sys.exit(1)
    except httpx.ConnectError:
        print(f"CONNECTION REFUSED: {API}{path} — is the pod running?", file=sys.stderr)
        sys.exit(1)
    except httpx.HTTPError as e:
        print(f"HTTP ERROR: {e}", file=sys.stderr)
        sys.exit(1)

    if r.status_code >= 400:
        print(f"HTTP {r.status_code}: {r.text[:200]}", file=sys.stderr)
        sys.exit(1)

    try:
        return r.json()
    except json.JSONDecodeError:
        print(f"INVALID JSON: {r.text[:200]}", file=sys.stderr)
        sys.exit(1)


def trace(last=20, filter_events=None):
    data = _request("get", f"/api/trace?last={last}")
    lines = data.get("lines", [])
    if not lines:
        print("(no trace events)")
        return
    for t in lines:
        event = t.get("event", "")
        if filter_events and event not in filter_events:
            continue
        node = t.get("node", "")
        if event == "tool_call":
            print(f"  CALL: {t.get('tool')} -> {str(t.get('input', ''))[:120]}")
        elif event == "tool_result":
            print(f"  RESULT: {t.get('tool')} ({t.get('rows', '?')} rows) -> {str(t.get('output', ''))[:120]}")
        elif event == "controls":
            ctrls = t.get("controls", [])
            types = {}
            for c in ctrls:
                types[c.get("type", "?")] = types.get(c.get("type", "?"), 0) + 1
            print(f"  CONTROLS: {types}")
        elif event == "s3_audit":
            print(f"  S3*: {t.get('check', '')} — {t.get('detail', '')}")
        elif event == "director_plan":
            print(f"  PLAN: {t.get('goal', '')} [{len(t.get('steps', []))} steps]")
        elif event in ("perceived", "decided", "director_updated", "machine_created",
                       "machine_transition", "machine_destroyed"):
            detail = t.get("instruction", t.get("detail", t.get("id", "")))
            print(f"  {node:12} {event:20} {str(detail)[:100]}")
        elif event == "tick":
            deltas = t.get("deltas", {})
            if deltas:
                print(f"  {node:12} tick #{t.get('tick', 0):3} {' '.join(f'{k}={v}' for k,v in deltas.items())}")


def history(last=20):
    data = _request("get", f"/api/history?last={last}")
    msgs = data.get("messages", [])
    if not msgs:
        print("(no messages)")
        return
    for m in msgs:
        print(f"\n--- {m['role']} ---")
        print(m["content"][:300])


def state():
    data = _request("get", "/api/state")
    print(json.dumps(data, indent=2, ensure_ascii=False))


def send(text):
    data = _request("post", "/api/send", json={"text": text}, timeout=90)
    resp = data.get("response", "")
    if not resp:
        print("WARNING: empty response", file=sys.stderr)
    print(resp[:500])


def clear():
    data = _request("post", "/api/clear", json={})
    print(data)


def graph():
    data = _request("get", "/api/graph/active")
    print(f"{data.get('name')} — {len(data.get('nodes', {}))} nodes, {len(data.get('edges', []))} edges")
    print(f"  {data.get('description', '')}")


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: assay_cli.py <command> [args]")
        print("  trace [last] [event_filter]  — show trace events")
        print("  history [last]               — show chat history")
        print("  state                        — show memorizer state")
        print("  send <text>                  — send a message")
        print("  clear                        — clear session")
        print("  graph                        — show active graph")
        sys.exit(0)

    cmd = sys.argv[1]
    if cmd == "trace":
        last = int(sys.argv[2]) if len(sys.argv) > 2 else 20
        filt = sys.argv[3].split(",") if len(sys.argv) > 3 else None
        trace(last, filt)
    elif cmd == "history":
        last = int(sys.argv[2]) if len(sys.argv) > 2 else 20
        history(last)
    elif cmd == "state":
        state()
    elif cmd == "send":
        if len(sys.argv) < 3:
            print("ERROR: send requires text argument", file=sys.stderr)
            sys.exit(1)
        send(" ".join(sys.argv[2:]))
    elif cmd == "clear":
        clear()
    elif cmd == "graph":
        graph()
    else:
        print(f"Unknown command: {cmd}", file=sys.stderr)
        sys.exit(1)