Grok Bot

#!/usr/bin/env python3
"""grok-dashboard — Grok Bot's dock launch page.

Bridge launch page (one dashboard, one port, one URL). Status comes from
shelling `adom-bridge --ai-thread "Grok Bot"` so the grok bridge stays the
source of truth. Local probes cover Grok's computer (this box): KiCad on
PATH and `adom/core`.

  GET  /                     the app (dashboard/ui/index.html, read per request)
  GET  /favicon.svg
  GET  /health
  GET  /api/status[/force]   {state, leds:[{id,label,state,hint,short}]}
  GET  /api/targets
  GET  /console
  GET  /events               SSE
  GET  /demo/steps.json
  POST /api/say-send {phrase}
  POST /api/action {action,args}
  POST /ui/toast /ui/click
  GET/POST /updater/status, /ai-threads/run-prompt  (control-API proxy)

CLI:
  grok-dashboard serve [--port 8791] [--print-url]
  grok-dashboard ensure-desktop
  grok-dashboard status
  grok-dashboard health
"""
from __future__ import annotations

import json
import os
import queue
import re
import shutil
import signal
import subprocess
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse

VERSION = "1.0.3"
APP = "grok-dashboard"
AI_THREAD = "Grok Bot"
DEFAULT_PORT = 8791
STATE_PATH = Path.home() / ".adom" / "grok-dashboard.json"
RELAY_PUBLIC_URL_PATH = Path.home() / ".adom" / "relay-public-url"
CLOUDFLARED_LOG_PATH = Path("/tmp/cloudflared-ab.log")
RELAY_HEALTH_URL = "http://127.0.0.1:8766/health"
KNOWN_DESKTOPS = ("AdomLapper", "ConfRoomROG")
GROK_LAPTOP_ID = "a101b94c-a17b-4818-aa2c-503f3e4e2f8c"
WIN_AB_EXE = r"%LOCALAPPDATA%\Adom Bridge\adom-bridge.exe"
WIN_AB_CLI = r"%LOCALAPPDATA%\Adom Bridge\adom-bridge-cli.exe"
HERE = Path(__file__).resolve().parent
UI_DIR = HERE / "ui"
ROOT = HERE.parent
DEMO_DIR = HERE / "demo"
ICON_PATH = ROOT / "docs" / "icon.svg"
MIN_SHELL_VERSION = (1, 0, 142)

_SAY_FALLBACK = [
    "bind the grok webhook",
    "create a grok bot",
    "screenshot grok's computer",
    "open kicad on grok's computer",
    "install adom/core on grok",
    "fire a grok webhook ping",
]

MANIFEST_URL = (
    "https://wiki.adom.inc/api/v1/pages/grok-bridge/files/"
    "adom-bridge-grok-manifest.json"
)
ARDUINO_PCB = "$HOME/kicad-offload/Arduino_Uno/Arduino_Uno.kicad_pcb"


def say_phrases():
    for cand in (ROOT / "dockbar.json",
                 Path.home() / ".local" / "share" / "adom" / "grok-bridge" / "dockbar.json"):
        try:
            lst = json.loads(cand.read_text(encoding="utf-8")).get("say")
            if isinstance(lst, list) and lst:
                return [str(x) for x in lst][:6]
        except Exception:
            continue
    return list(_SAY_FALLBACK)


def _hydrogen_control_url():
    try:
        u = (Path.home() / ".adom" / "hydrogen-control-url").read_text(
            encoding="utf-8").strip()
        return u.rstrip("/") or None
    except Exception:
        return None


def _ver_tuple(v):
    out = []
    for part in str(v or "").strip().lstrip("v").split("."):
        digits = "".join(c for c in part if c.isdigit())
        out.append(int(digits) if digits else 0)
    return tuple(out[:3]) or (0,)


_EXACT_ROUTES = ("/health", "/console", "/events", "/api/targets",
                 "/api/action", "/ui/toast", "/ui/click", "/api/status",
                 "/api/say-send", "/api/claim-webview")
_SUFFIX_ROUTES = ("/newbie-panel.js", "/favicon.svg", "/favicon.png",
                  "/favicon.ico", "/icon.svg", "/demo.mp4",
                  "/updater/status", "/api/viewport",
                  "/ai-threads/run-prompt", "/ai-threads/run-prompt/status",
                  "/api/say-send",
                  "/demo/steps.json")
_REGEX_ROUTES = (r"/assets/(.+)$",
                 r"^/demo/audio/(.+)$",
                 r"^/(?:api/status|state)(/force)?$")
_OWNED_PREFIXES = ("/api/", "/ui/", "/updater/", "/assets/", "/ai-threads/",
                   "/demo/")
_ASSET_SUFFIXES = (".png", ".jpg", ".jpeg", ".webp", ".svg", ".ico", ".gif",
                   ".mp3", ".wav", ".ogg", ".mp4", ".webm",
                   ".js", ".mjs", ".css", ".map", ".json",
                   ".woff", ".woff2", ".ttf")


def _route_exists(path):
    if path in _EXACT_ROUTES or path.endswith(_SUFFIX_ROUTES):
        return True
    return any(re.search(rx, path) for rx in _REGEX_ROUTES)


def _wants_app_shell(handler, path):
    if path == "/":
        return True
    if path.endswith(_ASSET_SUFFIXES):
        return False
    if any(pre in path + "/" for pre in _OWNED_PREFIXES):
        return False
    return "text/html" in (handler.headers.get("Accept") or "")


def _proxy_control(handler, path, body=None, timeout=20):
    import urllib.error
    import urllib.request
    base = _hydrogen_control_url()
    if not base:
        return handler._json({"ok": False, "why": "no control api on this machine"}, 503)
    req = urllib.request.Request(
        base + path, method="POST" if body is not None else "GET",
        data=json.dumps(body).encode("utf-8") if body is not None else None,
        headers={"Content-Type": "application/json"} if body is not None else {})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return handler._send(r.status, r.headers.get("Content-Type", "application/json"),
                                 r.read())
    except urllib.error.HTTPError as e:
        return handler._send(e.code, e.headers.get("Content-Type", "application/json"),
                             e.read())
    except Exception as e:
        return handler._json({"ok": False, "why": f"{e.__class__.__name__}: {e}"}, 502)


def say_send(phrase):
    import urllib.request
    base = _hydrogen_control_url()
    if not base:
        return {"ok": False, "why": "no AI panel on this machine (hydrogen-control-url missing)"}

    def _get(path, timeout):
        with urllib.request.urlopen(base + path, timeout=timeout) as r:
            return json.loads(r.read().decode("utf-8") or "{}")

    try:
        st = _get("/updater/status", 8)
        d = st.get("data") if isinstance(st.get("data"), dict) else st
        cur = (d.get("updater") or d).get("current_version") or d.get("current_version")
        if cur and _ver_tuple(cur) < MIN_SHELL_VERSION:
            return {"ok": False, "shellTooOld": True, "currentVersion": str(cur),
                    "why": f"your Hydrogen ({cur}) predates the Send to AI fix; update to "
                           f"1.0.142 or newer to enable it. The phrase was copied instead."}
    except Exception:
        pass

    # Wiki #4 (Hydrogen 1.0.213 AI picker): NO `type`, so the human picks the AI in
    # the shell's picker; the app: tag keys its "remember for this app" checkbox.
    # background:true (1.0.219): the answer is an immediate {job, state:"accepted"}
    # and the picker's 5 to 45 s happen detached; the status route narrates them.
    body = {"new": True, "prompt": phrase, "submit": True, "background": True,
            "name": "Grok Bot - say it", "tags": ["dock:grok-bridge", "app:adom/grok-bridge"]}
    req = urllib.request.Request(base + "/ai-threads/run-prompt",
                                 data=json.dumps(body).encode("utf-8"), method="POST",
                                 headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=120) as r:
            res = json.loads(r.read().decode("utf-8") or "{}")
    except Exception as e:
        return {"ok": False, "why": f"the AI panel did not answer ({e.__class__.__name__}). "
                                    f"Not retried, so no duplicate tab. The phrase was copied instead."}

    d = res.get("data") if isinstance(res.get("data"), dict) else res
    if isinstance(d, dict) and d.get("cancelled"):
        return {"ok": False, "cancelled": True,
                "why": "you dismissed the AI picker; nothing was opened or sent. The phrase is still on your clipboard."}
    job = d.get("job") if isinstance(d, dict) else None
    if job:
        return _poll_run_prompt(_get, job, d)
    inject = d.get("inject") if isinstance(d.get("inject"), dict) else {}
    delivered = (d.get("delivered_to")
                 or inject.get("delivered_in_conversation")
                 or d.get("delivered_in_conversation"))
    if not delivered:
        hint = d.get("_hints") or d.get("_hint") or res.get("_hints") or ""
        if isinstance(hint, dict):
            hint = " ".join(str(v) for v in hint.values())
        if d.get("ok") or res.get("ok"):
            return {"ok": False, "unconfirmed": True,
                    "why": "the thread opened and the prompt was submitted, but delivery is "
                           "not confirmed yet. Check the new AI tab; the phrase was also "
                           "copied so you can paste it if it did not land."}
        return {"ok": False, "why": ("the AI panel opened a thread but did not confirm "
                                     "delivery. " + (str(hint)[:200] if hint else "")).strip()}
    return {"ok": True, "deliveredTo": str(delivered)[:200],
            "tid": d.get("tid"), "sid": d.get("sid")}



# Hydrogen 1.0.219 run-prompt job states: accepted, picking, picked, opening,
# prefilled | tab_found, submitted, working, needs_you | done; Codex and Kimi end
# at delivered. `terminal:true` ends the job; `human` is the ready toast line.
_SENT_STATES = ("working", "needs_you", "done", "delivered")
_FAIL_STATES = ("cancelled", "rejected", "open_failed", "open_unconfirmed", "tab_not_found",
                "inject_failed", "submit_failed", "failed", "timeout",
                "submitted_unidentified", "submit_unconfirmed", "still_working")


def _poll_run_prompt(_get, job, d):
    import urllib.parse
    last = str(d.get("state") or "accepted"); human = str(d.get("human") or "")
    picked = None
    deadline = time.monotonic() + 75.0          # the picker alone may hold 45 s
    while time.monotonic() < deadline:
        terminal = False
        try:
            st = _get("/ai-threads/run-prompt/status?job=" + urllib.parse.quote(str(job)), 4)
            sd = st.get("data") if isinstance(st.get("data"), dict) else st
            last = str(sd.get("state") or last); human = str(sd.get("human") or human)
            jb = sd.get("job") if isinstance(sd.get("job"), dict) else {}
            picked = picked or jb.get("type") or sd.get("type")
            terminal = bool(sd.get("terminal"))
        except Exception:
            pass
        if last == "cancelled":
            # the user pressed Escape in the picker: never retry, never name an AI instead
            return {"ok": False, "cancelled": True, "job": job, "state": last,
                    "why": human or "you dismissed the AI picker; nothing was opened or sent. The phrase is still on your clipboard."}
        if last in _SENT_STATES:
            return {"ok": True, "job": job, "state": last, "ai": picked, "deliveredTo": picked,
                    "why": human or f"sent to {picked or 'your AI'}"}
        if last in _FAIL_STATES or terminal:
            why = ("the prompt is sitting in the new tab and Send was never clicked; press Enter there"
                   if last == "submit_failed" else (human or f"the AI panel reported {last}"))
            return {"ok": False, "job": job, "state": last, "why": why + ". The phrase was also copied."}
        time.sleep(1.0)
    return {"ok": False, "unconfirmed": True, "job": job, "state": last,
            "why": (human or f"the new thread is still {last}") + " after 75 s; check the new AI tab. The phrase was also copied."}

def ab(verb, args=None, timeout=25, target=None):
    cmd = ["adom-bridge", "--ai-thread", AI_THREAD]
    if target and target != "-":
        cmd += ["--target", target]
    cmd += [verb, json.dumps(args or {})]
    try:
        p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
        d = json.loads(p.stdout or "{}")
        out = d.get("output")
        if isinstance(out, str) and out.strip().startswith("{"):
            try:
                inner = json.loads(out)
                merged = dict(d)
                merged.update(inner.get("data") if isinstance(inner.get("data"), dict) else inner)
                return merged
            except Exception:
                pass
        return d
    except subprocess.TimeoutExpired:
        return {"error": f"no answer within {timeout}s", "timedOut": True}
    except FileNotFoundError:
        return {"error": "adom-bridge not on PATH", "errorCode": "no_bridge"}
    except Exception as e:
        return {"error": str(e).split("\n")[0][:120]}


def get_targets():
    try:
        p = subprocess.run(["adom-bridge", "targets"], capture_output=True, text=True, timeout=20)
        d = json.loads(p.stdout or "{}")
        return [t.get("name") for t in (d.get("targets") or []) if t.get("name")]
    except Exception:
        return []



def pick_target(targets):
    names = [str(t) for t in (targets or []) if t]
    lower = {n.lower(): n for n in names}
    for want in KNOWN_DESKTOPS:
        if want.lower() in lower:
            return lower[want.lower()]
    return names[0] if names else ""


def _load_state():
    try:
        d = json.loads(STATE_PATH.read_text(encoding="utf-8"))
        return d if isinstance(d, dict) else {}
    except Exception:
        return {}


def _write_state_merge(updates):
    """Merge keys into grok-dashboard.json. Never store tokens/keys."""
    st = _load_state()
    for k, v in (updates or {}).items():
        lk = str(k).lower()
        if any(s in lk for s in ("token", "secret", "password", "apikey", "api-key", "auth_token", "authtoken")):
            continue
        st[k] = v
    STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
    STATE_PATH.write_text(json.dumps(st, indent=2) + "\n", encoding="utf-8")
    return st



def _host_resolves(url):
    from urllib.parse import urlparse
    import socket
    host = urlparse(url).hostname
    if not host:
        return False
    try:
        socket.getaddrinfo(host, 443)
        return True
    except Exception:
        return False


def _is_loopback_url(url):
    u = (url or "").strip().lower()
    if not u:
        return True
    return any(x in u for x in ("127.0.0.1", "localhost", "[::1]", "0.0.0.0"))


def _is_public_relay_url(url):
    u = (url or "").strip()
    if not u or _is_loopback_url(u):
        return False
    return u.startswith(("wss://", "ws://", "https://", "http://"))


def public_relay_url(setup=None):
    setup = setup or {}
    try:
        file_url = RELAY_PUBLIC_URL_PATH.read_text(encoding="utf-8").strip()
    except Exception:
        file_url = ""
    if _is_public_relay_url(file_url):
        return file_url
    su = str(setup.get("server_url") or "").strip()
    if _is_public_relay_url(su):
        return su
    return file_url or su or ""


_TRYCLOUDFLARE_HTTPS = re.compile(
    r"https://([A-Za-z0-9-]+\.trycloudflare\.com)", re.I
)


def _to_wss(url):
    u = (url or "").strip().rstrip("/")
    if u.startswith("https://"):
        return "wss://" + u[len("https://"):]
    if u.startswith("http://"):
        return "ws://" + u[len("http://"):]
    return u


def live_trycloudflare_from_log(log_path=None):
    """Return wss://host from a live cloudflared log, or "".

    Last printed trycloudflare hostname that currently resolves wins.
    Does not copy tokens.
    """
    path = Path(log_path) if log_path else CLOUDFLARED_LOG_PATH
    try:
        blob = path.read_text(encoding="utf-8", errors="replace")
    except Exception:
        return ""
    hosts = [h.lower() for h in _TRYCLOUDFLARE_HTTPS.findall(blob)]
    for host in reversed(hosts):
        wss = "wss://" + host
        if _host_resolves(wss):
            return wss
    return ""


def heal_public_relay_url(setup=None):
    """If relay-public-url does not resolve, refresh from cloudflared log.

    Returns (url, how). url is "" when the public tunnel is down — callers
    must not push a dead hostname to the laptop. how is resolves |
    refreshed_from_log | down. Writes the healed line back to
    RELAY_PUBLIC_URL_PATH. Does not copy tokens.
    """
    pub = _to_wss(public_relay_url(setup))
    if pub and _host_resolves(pub):
        return pub, "resolves"
    live = live_trycloudflare_from_log()
    if live:
        try:
            RELAY_PUBLIC_URL_PATH.parent.mkdir(parents=True, exist_ok=True)
            RELAY_PUBLIC_URL_PATH.write_text(live + "\n", encoding="utf-8")
        except Exception:
            pass
        return live, "refreshed_from_log"
    return "", "down"


def _relay_health():
    import urllib.request
    try:
        with urllib.request.urlopen(RELAY_HEALTH_URL, timeout=1.5) as r:
            return json.loads(r.read().decode("utf-8") or "{}")
    except Exception:
        return None


def ensure_relay():
    h = _relay_health()
    if isinstance(h, dict) and h.get("status") == "ok":
        return True, "relay already answering"
    try:
        subprocess.Popen(
            ["adom-bridge", "serve", "--supervise"],
            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
    except FileNotFoundError:
        return False, "adom-bridge not on PATH"
    except Exception as e:
        return False, ("could not start relay: " + str(e))[:120]
    for _ in range(20):
        time.sleep(0.25)
        h = _relay_health()
        if isinstance(h, dict) and h.get("status") == "ok":
            return True, "started adom-bridge serve --supervise"
    return bool(_relay_health()), "started serve --supervise; health not yet ok"


def _parse_ab_json(stdout):
    try:
        d = json.loads(stdout or "{}")
    except Exception:
        return {"error": "unreadable json", "raw": (stdout or "")[:200]}
    out = d.get("output")
    if isinstance(out, str) and out.strip().startswith("{"):
        try:
            inner = json.loads(out)
            merged = dict(d)
            merged.update(inner.get("data") if isinstance(inner.get("data"), dict) else inner)
            return merged
        except Exception:
            pass
    return d


def _ab_local(verb, args=None, timeout=25):
    cmd = ["adom-bridge", "--ai-thread", AI_THREAD, verb,
           json.dumps(args if args is not None else {})]
    try:
        p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
        return _parse_ab_json(p.stdout)
    except subprocess.TimeoutExpired:
        return {"error": f"no answer within {timeout}s", "timedOut": True}
    except FileNotFoundError:
        return {"error": "adom-bridge not on PATH", "errorCode": "no_bridge"}
    except Exception as e:
        return {"error": str(e).split("\n")[0][:120]}


def _run_register_container(setup, url):
    """Run register_container with the public URL. Do not dump paste-JSON on ok:false."""
    cfg = setup.get("server_config") if isinstance(setup.get("server_config"), dict) else {}
    name = cfg.get("name") or "grok-bot"
    if not url:
        return {"ok": False, "error": "no public relay URL"}
    d = _ab_local("register_container", {"name": name, "url": url}, timeout=20)
    ok = bool(d.get("ok") or d.get("success")) and not d.get("error")
    err = d.get("error") or d.get("message") or d.get("why") or ""
    if not ok:
        return {
            "ok": False,
            "error": (str(err)[:200] or
                      "register_container returned ok:false (not an Adom cloud container)"),
        }
    return {"ok": True, "name": d.get("name") or name}


def persist_ensure(info):
    safe = {
        "desktop_connected": bool(info.get("desktop_connected")),
        "last_ensure": info.get("last_ensure") or time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "error": info.get("error") or "",
        "hint": info.get("hint") or "",
        "targets": list(info.get("targets") or []),
        "public_url": info.get("public_url") or "",
    }
    _write_state_merge(safe)
    return safe


_ENSURE_BG_LOCK = threading.Lock()
_ENSURE_BG_STARTED = False


def ensure_desktop(wait_s=28):
    """First-run auto-setup: laptop Adom Bridge on this relay. Never paste JSON."""
    steps = []
    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    result = {
        "ok": False,
        "desktop_connected": False,
        "last_ensure": now,
        "error": "",
        "hint": "",
        "targets": [],
        "public_url": "",
        "steps": steps,
    }

    up, msg = ensure_relay()
    steps.append("relay: " + msg)
    result["relay_up"] = up
    if not up:
        result["error"] = msg
        result["hint"] = ("adom-bridge serve --supervise did not come up. "
                          "Run grok-dashboard ensure-desktop again.")
        persist_ensure(result)
        return result

    def _snapshot():
        names = get_targets()
        result["targets"] = names
        result["desktop_connected"] = bool(names)
        result["ok"] = bool(names)
        return names

    names = _snapshot()
    if names:
        result["hint"] = "Adom Bridge already connected: " + ", ".join(names[:3])
        persist_ensure(result)
        return result

    setup = _ab_local("setup_desktop", {}, timeout=20)
    pub, heal_how = heal_public_relay_url(setup)
    result["public_url"] = pub
    result["tunnel_heal"] = heal_how
    steps.append("setup_desktop: server_url=" + str(setup.get("server_url") or "")[:80])
    if heal_how == "refreshed_from_log":
        steps.append("healed public_url from " + str(CLOUDFLARED_LOG_PATH) + ": " + pub)
        result["tunnel_resolves"] = True
    elif pub:
        steps.append("public_url: " + pub)
        result["tunnel_resolves"] = True
    else:
        steps.append(
            "public tunnel is down; not pushing dead name from "
            + str(RELAY_PUBLIC_URL_PATH)
            + " (no live URL in " + str(CLOUDFLARED_LOG_PATH) + ")"
        )
        result["tunnel_resolves"] = False
        result["hint"] = (
            "public tunnel is down. "
            + str(RELAY_PUBLIC_URL_PATH)
            + " host does not resolve and "
            + str(CLOUDFLARED_LOG_PATH)
            + " has no live trycloudflare URL."
        )
    if setup.get("error") and not setup.get("register_command") and not setup.get("server_url"):
        steps.append("setup_desktop error: " + str(setup.get("error"))[:120])

    if pub:
        reg = _run_register_container(setup, pub)
        if reg.get("ok"):
            steps.append("register_container: ok")
            result["register_ok"] = True
        else:
            steps.append("register_container: ok:false (not an Adom cloud box; not dumping JSON)")
            result["register_ok"] = False
            result["register_error"] = str(reg.get("error") or "")[:200]
    else:
        steps.append("register_container: skipped (no resolvable public URL)")
        result["register_ok"] = False

    names = _snapshot()
    if names:
        result["hint"] = "Adom Bridge connected after register_container: " + ", ".join(names[:3])
        persist_ensure(result)
        return result

    if pub:
        result["laptop_push"] = {
            "label": "AdomLapper",
            "machineId": GROK_LAPTOP_ID,
            "also": "ConfRoomROG",
            "exe": WIN_AB_EXE,
            "cli": WIN_AB_CLI,
            "url": pub,
            "how": (
                "Grok ListMachines, then Shell on AdomLapper. If Adom Bridge is installed, "
                "launch adom-bridge.exe without stealing focus if it is not running, then "
                'adom-bridge-cli.exe --ai-thread "Grok Bot" server_add with the public URL. '
                "Do not paste JSON into the GUI. Do not write a guessed config path."
            ),
        }
        steps.append("laptop_push: recipe for AdomLapper (this Linux box cannot Shell the laptop)")
    else:
        result["laptop_push"] = {
            "label": "AdomLapper",
            "machineId": GROK_LAPTOP_ID,
            "also": "ConfRoomROG",
            "exe": WIN_AB_EXE,
            "cli": WIN_AB_CLI,
            "url": "",
            "how": (
                "public tunnel is down; do not push a dead hostname. See "
                + str(RELAY_PUBLIC_URL_PATH)
                + " and " + str(CLOUDFLARED_LOG_PATH)
                + ". Restart cloudflared, then grok-dashboard ensure-desktop."
            ),
        }
        steps.append(
            "laptop_push: withheld dead URL from "
            + str(RELAY_PUBLIC_URL_PATH)
        )

    deadline = time.time() + max(5, int(wait_s))
    while time.time() < deadline:
        names = _snapshot()
        if names:
            break
        time.sleep(2)

    if names:
        result["error"] = ""
        result["hint"] = "Adom Bridge connected: " + ", ".join(names[:3])
    else:
        if result.get("tunnel_resolves") is False:
            result["error"] = "public relay URL does not resolve; laptop Adom Bridge cannot dial this box"
            result["hint"] = (
                "public tunnel is down. "
                + str(RELAY_PUBLIC_URL_PATH)
                + " host does not resolve and "
                + str(CLOUDFLARED_LOG_PATH)
                + " has no live trycloudflare URL. Restart the quick tunnel, then "
                "grok-dashboard ensure-desktop. Do not paste JSON."
            )
        else:
            result["error"] = result.get("register_error") or "no desktop on this relay"
            result["hint"] = (
                "Adom Bridge on the laptop is not connected. Run grok-dashboard ensure-desktop "
                "or adom-bridge setup_desktop. From Grok Bot, ListMachines and Shell AdomLapper "
                "to launch Adom Bridge with the public relay URL. Do not paste JSON."
            )
    persist_ensure(result)
    return result


def start_ensure_background():
    global _ENSURE_BG_STARTED
    with _ENSURE_BG_LOCK:
        if _ENSURE_BG_STARTED:
            return
        _ENSURE_BG_STARTED = True

    def _run():
        try:
            log("ensure-desktop: starting (background, once)")
            r = ensure_desktop(wait_s=28)
            names = ",".join(r.get("targets") or [])[:80]
            log("ensure-desktop: connected=" + str(bool(r.get("desktop_connected")))
                + " targets=" + names)
        except Exception as e:
            log("ensure-desktop failed: " + str(e).split("\n")[0][:120])

    threading.Thread(target=_run, name="ensure-desktop", daemon=True).start()


class Hub:
    def __init__(self):
        self._clients = []
        self._lock = threading.Lock()

    def add(self):
        q = queue.Queue(maxsize=64)
        with self._lock:
            self._clients.append(q)
        return q

    def drop(self, q):
        with self._lock:
            if q in self._clients:
                self._clients.remove(q)

    def count(self):
        with self._lock:
            return len(self._clients)

    def broadcast(self, event, data):
        msg = f"event: {event}\ndata: {json.dumps(data)}\n\n"
        with self._lock:
            for q in list(self._clients):
                try:
                    q.put_nowait(msg)
                except queue.Full:
                    pass


HUB = Hub()
CONSOLE = []
_console_lock = threading.Lock()


def log(line):
    with _console_lock:
        CONSOLE.append(f"{time.strftime('%H:%M:%S')} {line}")
        del CONSOLE[:-300]
    print(line, file=sys.stderr, flush=True)


def _led(lid, label, state, short, hint, detail=""):
    return {"id": lid, "label": label, "state": state, "short": str(short)[:24],
            "hint": hint, "detail": detail or hint}


def _core_installed():
    try:
        p = subprocess.run(["adom-wiki", "pkg", "list"], capture_output=True, text=True, timeout=20)
        d = json.loads(p.stdout or "{}")
        pkgs = d.get("packages") or d.get("installed") or []
        for row in pkgs:
            name = (row.get("name") if isinstance(row, dict) else str(row)) or ""
            if name in ("adom/core", "core") or name.endswith("/core"):
                return True, row.get("version") if isinstance(row, dict) else ""
    except Exception:
        pass
    # A skills-dir marker is not proof of the bootstrap pkg; only the registry list is.
    return False, ""


def _looks_like_agent_box():
    home = Path.home()
    return any((
        Path("/home/box").is_dir(),
        (home / "kicad-offload").is_dir(),
        Path("/home/box/agent-data").is_dir(),
        Path("/home/box/kicad-offload").is_dir(),
    ))


_CRITICAL = ("bridge", "host", "webhook", "computer")
_ADVISORY = ("kicad", "core")


def rollup(leds):
    by = {l["id"]: l for l in leds}
    def st(k):
        return (by.get(k) or {}).get("state")
    crit = [st(k) for k in _CRITICAL if st(k)]
    if "err" in crit:
        worst, why = "err", [k for k in _CRITICAL if st(k) == "err"]
    elif "warn" in crit:
        worst, why = "warn", [k for k in _CRITICAL if st(k) == "warn"]
    else:
        worst, why = "ok", []
    degraded = [k for k in _ADVISORY if st(k) in ("warn", "err")]
    reason = (", ".join(why) if why else
              ("healthy" + (f" (advisory: {', '.join(degraded)})" if degraded else "")))
    return {"state": worst, "reason": reason,
            "criticalLeds": list(_CRITICAL), "advisoryLeds": list(_ADVISORY)}


def build_status():
    targets = get_targets()
    target = pick_target(targets)
    st = _load_state()
    leds = []

    if targets:
        leds.append(_led(
            "bridge", "Adom Bridge", "ok", (target or targets[0])[:24],
            "Adom Bridge on the laptop is connected to this Grok Bot box.",
            "targets: " + ", ".join(targets[:3])))
    else:
        err = st.get("error") or "no desktop"
        leds.append(_led(
            "bridge", "Adom Bridge", "warn", "no desktop",
            "Adom Bridge on the laptop is not connected. Run grok-dashboard ensure-desktop "
            "(it calls adom-bridge setup_desktop and reconnects the laptop). "
            "Do not paste JSON. Do not ask the user to copy connection JSON.",
            str(err)[:200]))

    ready = {}
    no_bridge = False
    timed = False
    if targets:
        ready = ab("grok_readiness", {}, timeout=20, target=target or None)
        no_bridge = ready.get("errorCode") == "no_bridge" or "not on PATH" in str(ready.get("error") or "")
        timed = bool(ready.get("timedOut") or ready.get("error"))
    if not targets or no_bridge or (timed and not ready.get("hostAppInstalled") and not ready.get("success")):
        # Not probed — warn, not red. A missing desktop is not a failed check.
        if not targets:
            leds.append(_led(
                "host", "Grok Bot.exe", "warn", "not probed",
                "No laptop Adom Bridge yet, so grok_readiness was not called. "
                "Run grok-dashboard ensure-desktop / adom-bridge setup_desktop first. "
                "Never paste JSON."))
            leds.append(_led(
                "webhook", "Webhook", "warn", "not probed",
                "No laptop Adom Bridge yet. After grok-dashboard ensure-desktop connects, "
                "call grok_webhook_ensure {}. Never paste the key."))
        else:
            leds.append(_led(
                "host", "Grok Bot.exe", "warn", "no desktop",
                "Could not reach grok_readiness. Run grok-dashboard ensure-desktop / "
                "adom-bridge setup_desktop. Do not paste JSON."))
            leds.append(_led(
                "webhook", "Webhook", "warn", "not probed",
                "Could not read grok_webhook_status. Call grok_webhook_ensure {} "
                "once Grok Bot is reachable. Never paste the key."))
    else:
        installed = bool(ready.get("hostAppInstalled"))
        running = bool(ready.get("hostAppRunning"))
        fallback = ready.get("hostAppFallback")
        if installed and not fallback:
            leds.append(_led(
                "host", "Grok Bot.exe", "ok" if running else "warn",
                "running" if running else "installed",
                "Grok Bot.exe is running." if running else
                "Grok Bot.exe is installed but not running. grok_open {} launches it in the background.",
                ready.get("_hint") or ""))
        elif installed and fallback:
            leds.append(_led(
                "host", "Grok Bot.exe", "warn", "cursor fallback",
                "Grok Bot.exe is missing; Cursor is last-resort host. Prefer Grok Bot. "
                "Never kill Cursor. grok_open {} launches the fallback in the background."))
        else:
            leds.append(_led(
                "host", "Grok Bot.exe", "err", "not installed",
                "Grok Bot.exe is not installed. Offer grok_open {} — do not send the user "
                "to a download page. Never kill Cursor."))

        wh = ab("grok_webhook_status", {}, timeout=20, target=target or None)
        bound = bool(wh.get("bound"))
        host = wh.get("urlHost") or ""
        if bound:
            leds.append(_led(
                "webhook", "Webhook", "ok", "bound",
                f"Webhook bound to host {host} (key not shown). Fire with grok_webhook_fire.",
                f"bound to {host}" if host else "bound"))
        else:
            leds.append(_led(
                "webhook", "Webhook", "warn", "not bound",
                "No webhook bound. Call grok_webhook_ensure {} — it scrapes the Grok Bridge "
                "webhook routine pane. Never paste the key."))

    # Grok's computer is the agent box (this machine when the dashboard is
    # serving from Grok Bot's container). A connected box is extra evidence.
    host_name = os.uname().nodename if hasattr(os, "uname") else "this machine"
    if ready.get("computerReachable") is False:
        # Wiki #5: Grok Bot shows "Couldn't Reach Grok Bot's Computer"; a connected
        # laptop is not evidence the box is up.
        leds.append(_led(
            "computer", "Grok's computer", "err", "unreachable",
            "Grok Bot says it cannot reach its computer. Webhook events are accepted but "
            "nothing replies. Press Recover Grok Bot's Computer / Retry in Grok Bot.",
            str(ready.get("computerBanner") or "")[:200]))
    elif _looks_like_agent_box() or targets:
        leds.append(_led(
            "computer", "Grok's computer", "ok", host_name[:24] or "this box",
            "This box is Grok Bot's computer (the agent box), not the Grok Bot.exe window. "
            "Ask grok to screenshot its computer or open KiCad here.",
            f"box {host_name}" + (f" · targets: {', '.join(targets[:3])}" if targets else "")))
    else:
        leds.append(_led(
            "computer", "Grok's computer", "warn", "unknown",
            "No agent-box markers and no adom-bridge targets. Connect a box, then "
            "grok_readiness {} will name it."))

    kicad_bin = shutil.which("kicad") or shutil.which("kicad-cli")
    if kicad_bin:
        leds.append(_led(
            "kicad", "KiCad on grok", "ok", "on PATH",
            "KiCad is on PATH on Grok's computer. Open a board with kicad_open_board on "
            f"the agent box (Arduino Uno at {ARDUINO_PCB}).",
            kicad_bin))
    else:
        leds.append(_led(
            "kicad", "KiCad on grok", "warn", "not on PATH",
            "KiCad is not on PATH on Grok's computer. Ask the AI to install KiCad on grok's "
            "computer so the EE does not have to use their laptop."))

    core_ok, core_ver = _core_installed()
    if core_ok:
        leds.append(_led(
            "core", "adom/core", "ok", core_ver or "installed",
            "adom/core is installed on Grok's computer. The Adom ecosystem (skills hub, wiki, "
            "KiCad/Fusion skills) is on this box."))
    else:
        leds.append(_led(
            "core", "adom/core", "warn", "not installed",
            "adom/core is not installed on Grok's computer. Run: adom-wiki pkg install adom/core"))

    rolled = rollup(leds)
    return {
        "state": rolled["state"],
        "reason": rolled["reason"],
        "leds": leds,
        "target": target or "this machine",
        "targets": targets,
        "dashboardVersion": VERSION,
        "generatedAt": time.strftime("%H:%M:%S"),
        "criticalLeds": rolled["criticalLeds"],
        "advisoryLeds": rolled["advisoryLeds"],
        "_hint": "Each LED's hint names the verb that fixes it (grok-dashboard ensure-desktop, "
                 "adom-bridge setup_desktop, grok_readiness, grok_webhook_ensure, "
                 "adom-wiki pkg install adom/core). Never paste webhook keys. Never paste JSON.",
    }


_cache = {"t": 0.0, "data": None}
_cache_lock = threading.Lock()


def cached_status(force=False):
    with _cache_lock:
        if not force and _cache["data"] and time.time() - _cache["t"] < 8:
            return _cache["data"]
    data = build_status()
    with _cache_lock:
        _cache.update(t=time.time(), data=data)
    HUB.broadcast("state", data)
    return data


LAST_VIEWPORT = {}
VIEWPORT_LOG = []


def do_action(name, args):
    args = args or {}
    if name == "say":
        return {"success": True, "phrases": say_phrases()}
    if name == "refresh":
        return cached_status(force=True)
    if name == "steps":
        p = DEMO_DIR / "steps.json"
        try:
            return {"success": True, "steps": json.loads(p.read_text(encoding="utf-8"))}
        except Exception as e:
            return {"success": False, "error": str(e)}
    return {"error": f"unknown action {name}",
            "_hint": "Actions: say, refresh, steps. Drive grok via an AI thread, not clicks."}


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"
    server_version = APP + "/" + VERSION

    def log_message(self, *a):
        pass

    def _send(self, code, ctype, body, extra=None):
        if isinstance(body, str):
            body = body.encode("utf-8")
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        for k, v in (extra or {}).items():
            self.send_header(k, v)
        self.end_headers()
        self.wfile.write(body)

    def _json(self, obj, code=200):
        self._send(code, "application/json", json.dumps(obj))

    def _read_body(self):
        try:
            n = int(self.headers.get("Content-Length") or 0)
            return json.loads(self.rfile.read(n) or b"{}") if n else {}
        except Exception:
            return {}

    def _file(self, path: Path, mime, extra=None):
        if not path.is_file():
            return self._json({"error": "not found", "path": str(path.name)}, 404)
        extra = dict(extra or {})
        extra.setdefault("Cache-Control", "no-cache")
        return self._send(200, mime, path.read_bytes(), extra)

    def do_GET(self):
        path = urlparse(self.path).path.rstrip("/") or "/"
        if path.endswith("/newbie-panel.js"):
            return self._file(UI_DIR / "newbie-panel.js",
                              "application/javascript; charset=utf-8",
                              {"Cache-Control": "no-store"})
        if path.endswith(("/favicon.svg", "/favicon.png", "/favicon.ico", "/icon.svg")):
            p = ICON_PATH if ICON_PATH.is_file() else UI_DIR / "icon.svg"
            return self._file(p, "image/svg+xml")
        if path.endswith("/demo.mp4"):
            return self._file(UI_DIR / "demo.mp4", "video/mp4")
        if path.endswith("/demo/steps.json"):
            return self._file(DEMO_DIR / "steps.json", "application/json")
        am = re.search(r"/demo/audio/(.+)$", path)
        if am:
            f = (DEMO_DIR / "audio" / am.group(1)).resolve()
            if not str(f).startswith(str((DEMO_DIR / "audio").resolve())) or not f.is_file():
                return self._json({"error": "no such audio"}, 404)
            return self._file(f, "audio/mpeg")
        if path == "/health":
            return self._json({"ok": True, "app": APP, "version": VERSION,
                               "pid": os.getpid(), "sseClients": HUB.count(),
                               "port": DEFAULT_PORT})
        am = re.search(r"/assets/(.+)$", path)
        if am:
            f = (UI_DIR / "assets" / am.group(1)).resolve()
            if not str(f).startswith(str((UI_DIR / "assets").resolve())) or not f.is_file():
                return self._json({"error": "no such asset"}, 404)
            mime = {"svg": "image/svg+xml", "png": "image/png", "jpg": "image/jpeg",
                    "webp": "image/webp", "mp4": "video/mp4"}.get(
                        f.suffix.lstrip(".").lower(), "application/octet-stream")
            return self._file(f, mime)
        if path.endswith("/ai-threads/run-prompt/status"):
            # the module polls this once a second; the ?job= query must travel intact
            q = urlparse(self.path).query or ""
            return _proxy_control(self, "/ai-threads/run-prompt/status" + (("?" + q) if q else ""),
                                  timeout=8)
        if path.endswith("/updater/status"):
            return _proxy_control(self, "/updater/status", timeout=8)
        if path.endswith("/api/viewport"):
            if "matrix" in (urlparse(self.path).query or ""):
                return self._json({"reports": sorted(VIEWPORT_LOG,
                                                     key=lambda r: r.get("innerWidth") or 0)})
            return self._json(LAST_VIEWPORT or {"note": "no page has reported a viewport yet"})
        if path == "/console":
            with _console_lock:
                return self._json({"lines": list(CONSOLE)})
        if path == "/events":
            return self._sse()
        if path == "/api/targets":
            return self._json({"targets": [{"name": n} for n in get_targets()]})
        m = re.match(r"^/(?:api/status|state)(/force)?$", path)
        if m:
            force = bool(m.group(1))
            log(f"status{' force' if force else ''}")
            return self._json(cached_status(force))
        if not _wants_app_shell(self, path):
            return self._json({"error": "not found", "path": path}, 404)
        try:
            html = (UI_DIR / "index.html").read_text(encoding="utf-8")
            html = html.replace("</head>",
                                "<script>window.__SAY_PHRASES__=" +
                                json.dumps(say_phrases()) + ";window.__DASH_VERSION__=" +
                                json.dumps(VERSION) + ";</script></head>", 1)
            return self._send(200, "text/html; charset=utf-8", html,
                              {"Cache-Control": "no-store, must-revalidate"})
        except Exception as e:
            return self._json({"error": f"ui missing: {e}"}, 500)

    def do_OPTIONS(self):
        path = urlparse(self.path).path.rstrip("/") or "/"
        if not _route_exists(path):
            return self._json({"error": "not found", "path": path}, 404)
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.send_header("Access-Control-Max-Age", "86400")
        self.end_headers()

    def do_POST(self):
        path = urlparse(self.path).path.rstrip("/") or "/"
        body = self._read_body()
        if path == "/api/action":
            name = body.get("action") or ""
            log(f"action {name}")
            return self._json(do_action(name, body.get("args")))
        if path == "/ui/toast":
            msg, kind = body.get("message") or "", body.get("kind") or "info"
            log(f"toast [{kind}] {msg}")
            HUB.broadcast("toast", {"message": msg, "kind": kind})
            return self._json({"ok": True, "delivered": HUB.count()})
        if path.endswith("/api/viewport"):
            global LAST_VIEWPORT
            LAST_VIEWPORT = dict(body or {})
            LAST_VIEWPORT["at"] = time.strftime("%H:%M:%S")
            VIEWPORT_LOG.append(dict(LAST_VIEWPORT))
            del VIEWPORT_LOG[:-16]
            return self._json({"ok": True})
        if path.endswith("/ai-threads/run-prompt"):
            # Wiki #4: strip any `type` (absent = Hydrogen's AI picker asks the user),
            # carry the app: tag so a remembered pick applies here, and go background
            # so the picker never sits inside this request. Reply goes back verbatim.
            b = dict(body or {}); b.pop("type", None)
            b.setdefault("background", True)
            tags = list(b.get("tags") or [])
            if "app:adom/grok-bridge" not in tags:
                tags.append("app:adom/grok-bridge")
            b["tags"] = tags
            return _proxy_control(self, "/ai-threads/run-prompt", body=b, timeout=125)
        if path.endswith("/api/say-send"):
            phrase = str((body or {}).get("phrase") or "").strip()
            if not phrase:
                return self._json({"ok": False, "why": "phrase required"}, 400)
            log(f"say-send: {phrase[:60]}")
            return self._json(say_send(phrase))
        if path.endswith("/api/claim-webview"):
            name = str((body or {}).get("name") or os.environ.get("ADOM_DOCK_WEBVIEW") or "").strip()
            if not name:
                return self._json({"ok": False, "why": "ADOM_DOCK_WEBVIEW unset (hand-run)"})
            port = DEFAULT_PORT
            threading.Thread(target=claim_dock_webview, args=(port, name), daemon=True).start()
            log(f"claim-webview {name}")
            return self._json({"ok": True, "name": name, "url": direct_url(port)})
        if path == "/ui/click":
            t = body.get("target") or ""
            log(f"ui click {t}")
            HUB.broadcast("click", {"target": t})
            return self._json({"ok": True, "delivered": HUB.count()})
        return self._json({"error": "not found"}, 404)

    def _sse(self):
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Cache-Control", "no-cache, no-transform")
        self.send_header("X-Accel-Buffering", "no")
        self.send_header("Access-Control-Allow-Origin", "*")
        self.send_header("Connection", "keep-alive")
        self.end_headers()
        q = HUB.add()
        log(f"sse client connected ({HUB.count()} live)")
        try:
            self.wfile.write((": " + "p" * 2048 + "\n\n").encode())
            self.wfile.write(f"event: hello\ndata: {json.dumps({'version': VERSION})}\n\n".encode())
            self.wfile.flush()
            while True:
                try:
                    msg = q.get(timeout=15)
                except queue.Empty:
                    msg = ": ping\n\n"
                self.wfile.write(msg.encode())
                self.wfile.flush()
        except Exception:
            pass
        finally:
            HUB.drop(q)
            log(f"sse client disconnected ({HUB.count()} live)")


def direct_url(port):
    return f"http://127.0.0.1:{port}/"


def write_state(port):
    _write_state_merge({
        "app": APP, "pid": os.getpid(), "port": int(port),
        "version": VERSION, "url": direct_url(port),
    })


def claim_dock_webview(port, name=None):
    """open-or-refresh ONLY when ADOM_DOCK_WEBVIEW is set (user machine).
    Never call adom-cli hydrogen from a cloud container — that drives hc."""
    name = name or os.environ.get("ADOM_DOCK_WEBVIEW") or ""
    if not name:
        return
    try:
        subprocess.run(["adom-cli", "hydrogen", "webview", "open-or-refresh",
                        "--name", name, "--url", direct_url(port)],
                       capture_output=True, timeout=20)
        log(f"claimed dock webview tab '{name}' -> {direct_url(port)}")
    except Exception:
        pass


def local_call(port, path, payload=None):
    import urllib.request
    url = f"http://127.0.0.1:{port}{path}"
    data = json.dumps(payload).encode() if payload is not None else None
    req = urllib.request.Request(url, data=data, method="POST" if data else "GET",
                                 headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=20) as r:
            return json.loads(r.read().decode())
    except Exception as e:
        return {"error": str(e),
                "_hint": f"Is the dashboard serving? `grok-dashboard serve --port {port}`."}


def _running_version(p_):
    import urllib.request
    try:
        with urllib.request.urlopen(f"http://127.0.0.1:{p_}/health", timeout=1.5) as r:
            d_ = json.loads(r.read().decode("utf-8") or "{}")
        if bool(d_.get("ok")) and d_.get("app") == APP:
            return str(d_.get("version") or "?")
    except Exception:
        pass
    return None


def _pid_owning_port(p_):
    """Port ownership via /proc, never argv-as-selector (kicad killed a bystander)."""
    want = []
    for path in ("/proc/net/tcp", "/proc/net/tcp6"):
        try:
            for line in open(path).read().splitlines()[1:]:
                f = line.split()
                if len(f) < 4 or f[3] != "0A":
                    continue
                if int(f[1].split(":")[1], 16) == int(p_):
                    want.append(f[9])
        except Exception:
            pass
    if not want:
        return None
    me = os.getpid()
    for pid in os.listdir("/proc"):
        if not pid.isdigit() or int(pid) == me:
            continue
        try:
            for fd in os.listdir(f"/proc/{pid}/fd"):
                try:
                    link = os.readlink(f"/proc/{pid}/fd/{fd}")
                except Exception:
                    continue
                if link.startswith("socket:[") and link[8:-1] in want:
                    return int(pid)
        except Exception:
            continue
    return None


def _retire_stale(p_, their):
    log(f"grok-dashboard {their} is serving on {p_} but this is {VERSION}; "
        f"retiring the stale instance")
    victim = _pid_owning_port(p_)
    if victim is None:
        log(f"could not identify the process owning port {p_}; not killing anything")
        return False
    try:
        argv_ = open(f"/proc/{victim}/cmdline", "rb").read().split(b"\0")
    except Exception:
        argv_ = []
    # SAFETY FILTER on that one pid. Never select by argv — Cursor and other
    # dashboards must never be collateral.
    ours = False
    joined = b" ".join(argv_).lower()
    if b"grok_dashboard.py" in joined and b"serve" in joined:
        ours = True
    if not ours:
        log(f"port {p_} is held by pid {victim} which is not grok-dashboard; refusing to kill")
        return False
    try:
        os.kill(victim, signal.SIGTERM)
    except Exception:
        pass
    for _ in range(40):
        if _running_version(p_) is None:
            return True
        time.sleep(0.1)
    try:
        os.kill(victim, signal.SIGKILL)
    except Exception:
        pass
    time.sleep(0.5)
    return _running_version(p_) is None


def main():
    argv = sys.argv[1:] or ["serve"]
    cmd = argv[0]

    def flag(name, default=None):
        return argv[argv.index(name) + 1] if name in argv and argv.index(name) + 1 < len(argv) else default

    port = int(flag("--port", DEFAULT_PORT))

    if cmd == "serve":
        their = _running_version(port)
        if their is not None and their != VERSION:
            if not _retire_stale(port, their):
                print(f"grok-dashboard: a stale {their} instance owns port {port} and "
                      f"would not exit; kill it and relaunch.", file=sys.stderr, flush=True)
                raise SystemExit(2)
            their = None
        if their is not None:
            # Refresh the pid file from the LISTENER's /health, never this process
            # (we are about to exit; writing os.getpid() here would orphan the file).
            try:
                import urllib.request
                with urllib.request.urlopen(f"http://127.0.0.1:{port}/health", timeout=1.5) as r:
                    h = json.loads(r.read().decode("utf-8") or "{}")
                _write_state_merge({
                    "app": APP, "pid": h.get("pid"), "port": int(port),
                    "version": VERSION, "url": direct_url(port),
                })
            except Exception:
                pass
            # Reuse launch: re-claim the interstitial tab the dock already opened.
            name = os.environ.get("ADOM_DOCK_WEBVIEW") or ""
            if name:
                local_call(port, "/api/claim-webview", {"name": name})
            if "--print-url" in argv:
                print(direct_url(port), flush=True)
            else:
                log(f"grok-dashboard {VERSION} already serving on {port}; nothing to do")
            return
        try:
            srv = ThreadingHTTPServer(("0.0.0.0", port), Handler)
        except OSError as e:
            if _running_version(port) is not None:
                if "--print-url" in argv:
                    print(direct_url(port), flush=True)
                return
            print(f"grok-dashboard: cannot serve on port {port}: {e}. "
                  f"Something else owns it - pick another with --port.",
                  file=sys.stderr, flush=True)
            raise SystemExit(2)
        srv.daemon_threads = True
        write_state(port)

        if "--print-url" in argv:
            url = direct_url(port)
            if hasattr(os, "fork"):
                if os.fork() > 0:
                    print(url, flush=True)
                    os._exit(0)
                os.setsid()
                _null = os.open(os.devnull, os.O_RDWR)
                for _fd in (0, 1, 2):
                    try:
                        os.dup2(_null, _fd)
                    except Exception:
                        pass
                write_state(port)
                claim_dock_webview(port)
            else:
                print(url, flush=True)
                claim_dock_webview(port)
        log(f"grok-dashboard {VERSION} serving on {port}")
        start_ensure_background()
        try:
            srv.serve_forever()
        except KeyboardInterrupt:
            pass
        return

    if cmd == "ensure-desktop":
        print(json.dumps(ensure_desktop(wait_s=28), indent=2))
        return

    if cmd == "status":
        print(json.dumps(build_status(), indent=2))
        return

    if cmd == "health":
        print(json.dumps(local_call(port, "/health")))
        return

    print(json.dumps({"error": f"unknown command {cmd}",
                      "_hint": "serve [--print-url] [--port 8791] | ensure-desktop | status | health"}))


if __name__ == "__main__":
    main()