svrnty-hermes-webui-plugin/scripts/cmo-natural-canvas-smoke.py

#!/usr/bin/env python3
"""Smoke a natural CMO Hermes turn into the WebUI Canvas bridge.

This proves the runtime handoff JP cares about:
CMO profile -> terminal helper -> Hermes WebUI plugin -> canva-editor -> replayable
Canvas events. The canva-editor is mocked here so the proof isolates routing; the
real Spark qwen3.6 generator is covered by canva-editor's real-model smoke.
"""

from __future__ import annotations

import argparse
import json
import os
import signal
import socket
import subprocess
import sys
import tempfile
import time
from datetime import datetime
from pathlib import Path
from urllib.error import URLError
from urllib.request import urlopen


PLUGIN_REPO = Path(__file__).resolve().parent.parent
HERMES_REPO = PLUGIN_REPO.parent
WEBUI_REPO = HERMES_REPO / "hermes-webui"
AGENT_REPO = HERMES_REPO / "hermes-agent"
CANVA_REPO = HERMES_REPO.parent / "cortex" / "L2-svrnty.tool-canva-editor"
CMO_REPO = HERMES_REPO / "cmo"
DEFAULT_OUT = HERMES_REPO / ".scratch" / "cmo-natural-canvas-proof.json"


def _free_port() -> int:
    sock = socket.socket()
    sock.bind(("127.0.0.1", 0))
    port = sock.getsockname()[1]
    sock.close()
    return port


def _wait_for(url: str, timeout: int = 45) -> bool:
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            with urlopen(url, timeout=1) as res:
                if res.status < 500:
                    return True
        except URLError:
            pass
        except Exception:
            pass
        time.sleep(0.25)
    return False


def _fetch_json(url: str, timeout: float = 8.0) -> dict:
    with urlopen(url, timeout=timeout) as res:
        return json.loads(res.read().decode("utf-8"))


def _start_canva(port: int, db_path: Path) -> subprocess.Popen:
    env = os.environ.copy()
    env["PATH"] = f"/home/svrnty/sdk/go1.26.0/bin{os.pathsep}{env.get('PATH', '')}"
    env["PORT"] = str(port)
    env["DB_PATH"] = str(db_path)
    return subprocess.Popen(
        ["go", "run", "./cmd/server", "--mock"],
        cwd=CANVA_REPO,
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        preexec_fn=os.setsid,
    )


def _start_webui(port: int, state_dir: Path, canva_base: str) -> subprocess.Popen:
    env = os.environ.copy()
    env.pop("HERMES_WEBUI_PASSWORD", None)
    env["CANVA_EDITOR_BASE_URL"] = canva_base
    env["HERMES_WEBUI_PYTHON_PLUGIN"] = "svrnty_hermes_webui_plugin"
    env["HERMES_WEBUI_PORT"] = str(port)
    env["HERMES_WEBUI_STATE_DIR"] = str(state_dir)
    env["HERMES_WEBUI_AGENT_DIR"] = str(AGENT_REPO)
    env["HERMES_REPO_ROOT"] = str(HERMES_REPO)
    env["PYTHONPATH"] = (
        f"{PLUGIN_REPO}{os.pathsep}{AGENT_REPO}"
        if not env.get("PYTHONPATH")
        else f"{PLUGIN_REPO}{os.pathsep}{AGENT_REPO}{os.pathsep}{env['PYTHONPATH']}"
    )
    py = WEBUI_REPO / "venv" / "bin" / "python"
    agent_py = AGENT_REPO / "venv" / "bin" / "python"
    python_exe = str(py) if py.exists() else str(agent_py) if agent_py.exists() else "python3"
    return subprocess.Popen(
        [python_exe, str(WEBUI_REPO / "server.py")],
        cwd=AGENT_REPO if AGENT_REPO.exists() else WEBUI_REPO,
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        preexec_fn=os.setsid,
    )


def _stop(proc: subprocess.Popen | None) -> str:
    if proc is None:
        return ""
    output = ""
    try:
        os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
        proc.wait(timeout=5)
    except Exception:
        try:
            os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
        except Exception:
            pass
    if proc.stdout:
        try:
            output = proc.stdout.read(4000)
        except Exception:
            output = ""
    return output


def _run_cmo_turn(webui_base: str, timeout_s: int) -> subprocess.CompletedProcess:
    prompt = f"""
You are CMO for Plan B. This is a routing smoke for the Canvas design command center.

Use the terminal tool to run exactly this command:
python3 "{CMO_REPO / "canvas_command.py"}" command "Create one CMO-routed Plan B promo card for a family meal campaign" --variants 1 --brand planb --project "Hermes CMO Natural Smoke" --wait

After the terminal command returns, reply with a compact summary containing command_id, project_id, number of screens, design_context.status, and secondbrain_status. Do not call canva-editor directly.
""".strip()
    env = os.environ.copy()
    env["HERMES_WEBUI_URL"] = webui_base
    env["CMO_LIB"] = str(CMO_REPO)
    env["HERMES_INFERENCE_PROVIDER"] = "vllm"
    env["HERMES_INFERENCE_MODEL"] = "qwen3.6-35b-a3b"
    return subprocess.run(
        [
            "hermes",
            "-p",
            "cmo-planb",
            "--provider",
            "vllm",
            "--model",
            "qwen3.6-35b-a3b",
            "-z",
            prompt,
            "--toolsets",
            "terminal",
            "--accept-hooks",
            "--yolo",
        ],
        cwd=HERMES_REPO,
        env=env,
        capture_output=True,
        text=True,
        timeout=timeout_s,
    )


def _canvas_event_summary(webui_base: str) -> dict:
    payload = _fetch_json(f"{webui_base}/api/canvas/events?format=json&since=0")
    events = payload.get("events") or []
    completed = [event for event in events if event.get("type") == "canvas.command.completed"]
    persisted = [event for event in events if event.get("type") == "canvas.screen.persisted"]
    failed = [event for event in events if event.get("type") == "canvas.command.failed"]
    command_ids = sorted({event.get("command_id") for event in completed if event.get("command_id")})
    return {
        "cursor": payload.get("cursor"),
        "event_count": len(events),
        "completed_count": len(completed),
        "persisted_count": len(persisted),
        "failed_count": len(failed),
        "command_ids": command_ids,
        "last_completed": completed[-1] if completed else None,
        "last_persisted": persisted[-1] if persisted else None,
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output-json", default=str(DEFAULT_OUT))
    parser.add_argument("--timeout", type=int, default=180)
    args = parser.parse_args()

    out_json = Path(args.output_json)
    canva_port = _free_port()
    webui_port = _free_port()
    canva_base = f"http://127.0.0.1:{canva_port}"
    webui_base = f"http://127.0.0.1:{webui_port}"
    canva_proc: subprocess.Popen | None = None
    webui_proc: subprocess.Popen | None = None

    with tempfile.TemporaryDirectory(prefix="hermes-cmo-canvas-") as tmp:
        tmp_path = Path(tmp)
        proof = {
            "generated_at": datetime.now().isoformat(timespec="seconds"),
            "webui_base": webui_base,
            "canva_base": canva_base,
            "profile": "cmo-planb",
            "provider": "vllm",
            "model": "qwen3.6-35b-a3b",
        }
        try:
            canva_proc = _start_canva(canva_port, tmp_path / "canva-editor.db")
            if not _wait_for(f"{canva_base}/health"):
                raise RuntimeError(f"canva-editor did not become healthy at {canva_base}/health")

            webui_proc = _start_webui(webui_port, tmp_path / "webui-state", canva_base)
            if not _wait_for(f"{webui_base}/health"):
                raise RuntimeError(f"webui did not become healthy at {webui_base}/health")

            result = _run_cmo_turn(webui_base, args.timeout)
            events = _canvas_event_summary(webui_base)
            proof["cmo_turn"] = {
                "returncode": result.returncode,
                "stdout": result.stdout[-4000:],
                "stderr": result.stderr[-2000:],
            }
            proof["events"] = events
            proof["pass"] = bool(
                result.returncode == 0
                and events["completed_count"] >= 1
                and events["persisted_count"] >= 1
                and events["failed_count"] == 0
                and events["command_ids"]
            )
            out_json.parent.mkdir(parents=True, exist_ok=True)
            out_json.write_text(json.dumps(proof, indent=2), encoding="utf-8")
            print(json.dumps(proof, indent=2))
            return 0 if proof["pass"] else 1
        finally:
            webui_log = _stop(webui_proc)
            canva_log = _stop(canva_proc)
            if webui_log:
                print(webui_log, file=sys.stderr)
            if canva_log:
                print(canva_log, file=sys.stderr)


if __name__ == "__main__":
    raise SystemExit(main())