svrnty-hermes-webui-plugin/scripts/umbrella-visual-smoke.py

#!/usr/bin/env python3
"""Browser visual smoke for the Cortex-OS umbrella panel.

Starts hermes-webui with the Svrnty plugin against a temporary auth-free state
dir, opens the real /plugins/svrnty/umbrella.html page, waits for the graph to
render, captures screenshots, and records simple pixel statistics.
"""

from __future__ import annotations

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

from PIL import Image
from playwright.sync_api import sync_playwright


PLUGIN_REPO = Path(__file__).resolve().parent.parent
HERMES_REPO = PLUGIN_REPO.parent
WEBUI_REPO = HERMES_REPO / "hermes-webui"
AGENT_REPO = HERMES_REPO / "hermes-agent"
DEFAULT_OUT = HERMES_REPO / "sot" / "08-OUTPUTS" / "umbrella-visual-proof-2026-05-25.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 = 30) -> 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 _pixel_stats(path: Path) -> dict:
    img = Image.open(path).convert("RGB")
    width, height = img.size
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", DeprecationWarning)
        pixels = list(img.getdata())
    total = len(pixels)
    dark_bg = sum(1 for r, g, b in pixels if r < 28 and g < 32 and b < 42)
    bright = sum(1 for r, g, b in pixels if max(r, g, b) > 120)
    distinct = len(set(pixels[:: max(1, total // 20000)]))
    return {
        "path": str(path),
        "width": width,
        "height": height,
        "dark_bg_ratio": round(dark_bg / total, 4),
        "bright_ratio": round(bright / total, 4),
        "sampled_distinct_colors": distinct,
        "nonblank": bright > 500 and distinct > 20,
    }


def _start_webui(port: int, state_dir: Path) -> subprocess.Popen:
    env = os.environ.copy()
    env.pop("HERMES_WEBUI_PASSWORD", None)
    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"
    cmd = [python_exe, str(WEBUI_REPO / "server.py")]
    return subprocess.Popen(
        cmd,
        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 main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output-json", default=str(DEFAULT_OUT))
    parser.add_argument("--screenshot-dir", default=str(HERMES_REPO / "screenshots" / "umbrella"))
    args = parser.parse_args()

    out_json = Path(args.output_json)
    screenshot_dir = Path(args.screenshot_dir)
    screenshot_dir.mkdir(parents=True, exist_ok=True)
    port = _free_port()
    base = f"http://127.0.0.1:{port}"
    proc: subprocess.Popen | None = None
    server_log = ""

    with tempfile.TemporaryDirectory(prefix="hermes-webui-umbrella-") as tmp:
        proc = _start_webui(port, Path(tmp))
        try:
            if not _wait_for(f"{base}/health", timeout=30):
                if proc.stdout:
                    server_log = proc.stdout.read(4000)
                raise RuntimeError(f"webui did not become healthy at {base}/health")

            proof = {
                "generated_at": datetime.now().isoformat(timespec="seconds"),
                "base": base,
                "viewports": [],
            }
            with sync_playwright() as p:
                browser = p.chromium.launch(
                    headless=True,
                    executable_path="/usr/bin/google-chrome" if Path("/usr/bin/google-chrome").exists() else None,
                    args=["--no-sandbox"],
                )
                for label, size in (
                    ("desktop", {"width": 1440, "height": 1000}),
                    ("mobile", {"width": 390, "height": 844}),
                ):
                    page = browser.new_page(viewport=size)
                    page.goto(f"{base}/plugins/svrnty/umbrella.html", wait_until="networkidle", timeout=30000)
                    page.wait_for_function(
                        "() => window.__svrntyUmbrella && window.__svrntyUmbrella.ready === true",
                        timeout=30000,
                    )
                    state = page.evaluate("() => window.__svrntyUmbrella")
                    stats_text = page.locator("#stats").inner_text(timeout=5000)
                    assert "nodes" in stats_text and "edges" in stats_text
                    shot = screenshot_dir / f"umbrella-{label}-2026-05-25.png"
                    page.screenshot(path=str(shot), full_page=True)
                    pixels = _pixel_stats(shot)
                    proof["viewports"].append({
                        "label": label,
                        "viewport": size,
                        "state": state,
                        "stats_text": stats_text,
                        "pixel_stats": pixels,
                    })
                    page.close()
                browser.close()

            proof["pass"] = all(v["state"]["nodes"] > 0 and v["state"]["edges"] > 0 and v["pixel_stats"]["nonblank"] for v in proof["viewports"])
            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:
            if proc is not None:
                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 server_log:
                print(server_log, file=sys.stderr)


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