box/bin/session-probe.py

#!/usr/bin/env python3
"""
session-probe.py — classify a fleet agent's muse.ai login state via CDP.

READ-ONLY: performs a single Runtime.evaluate reading document.title,
location.href, and body markers. Never clicks, navigates, types, or mutates
session state in any way. Uses the shared CDP queue at PRIORITY_LOW so it
never blocks operator sends or the harvester.

Login states (per operator AGENTS.md, 2026-10-03):
  LOGGED_IN         title contains "Chat —"/"Muse —" (SPA booted authenticated)
  LANDING           title == "muse.ai" (landing page, not logged in)
  LOGGED_OUT        body shows a "Log in" affordance
  OTP_PROMPT        body contains "To log in, enter the code"
  ACCOUNT_SELECTION body contains "Your email matches multiple accounts"
  UNKNOWN           none of the above matched
  CDP_UNREACHABLE   browser/CDP could not be reached at all

  parked_on_landing (bool, separate field): LOGGED_IN but the current URL is
  https://muse.ai/ — the watchdog relaunches browsers with the landing page
  as start URL, so a fresh browser is parked there until first nav. Session
  is fine; nav may proceed (but allow for SPA boot race).

Exit code: 0 when LOGGED_IN, 1 otherwise (suitable for pre-nav gating).
Stdout: one JSON line: {agent, state, title, url, has_input, checked_at}.

INVOCATION (important): CDP is only reachable from inside the node's netns
(Chromium binds DevTools to loopback). Always run via:
  netvm-exec.sh <agent> -- python3 /home/super/Projects/NetVM/bin/session-probe.py --agent <agent>
Running it on the bl host directly will report CDP_UNREACHABLE even when the
browser is healthy.
"""
import argparse
import contextlib
import importlib.util
import json
import os
import sys
import urllib.request
from datetime import datetime, timezone

BIN_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, BIN_DIR)

try:
    from cdp_queue import cdp_slot, PRIORITY_LOW
    HAS_CDP_QUEUE = True
except ImportError:
    HAS_CDP_QUEUE = False

import websocket  # noqa: E402  (after sys.path tweak, mirrors muse-chat-api.py)


def _load_accounts():
    path = os.path.join(BIN_DIR, "netvm-registry.py")
    spec = importlib.util.spec_from_file_location("netvm_registry", path)
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    accounts = {}
    for node, rec in mod.load().items():
        accounts[node] = (node, "http://127.0.0.1:%d/json/list" % rec["cdp_port"])
    return accounts


def _ev(ws, expr):
    """Runtime.evaluate with event draining (copied pattern from muse-chat-api.py)."""
    ws.send(json.dumps({
        "id": 1, "method": "Runtime.evaluate",
        "params": {"expression": expr, "returnByValue": True},
    }))
    for _ in range(50):
        resp = json.loads(ws.recv())
        if resp.get("id") == 1:
            break
    else:
        return None
    return resp.get("result", {}).get("result", {}).get("value")


_STATE_JS = """(() => {
  const body = document.body ? document.body.innerText.slice(0, 4000) : "";
  const has_input = !!document.querySelector(
    '[contenteditable="true"], textarea[placeholder*="Message"], div[role="textbox"]');
  return JSON.stringify({
    title: document.title || "",
    url: location.href || "",
    body: body,
    has_input: has_input,
  });
})()"""


def classify(title, url, body, has_input):
    t = (title or "").strip()
    b = (body or "")
    if "Your email matches multiple accounts" in b:
        return "ACCOUNT_SELECTION"
    if "To log in, enter the code" in b:
        return "OTP_PROMPT"
    if t == "muse.ai":
        return "LANDING"
    # A "Chat —"/"Muse —" title means the SPA booted with an authenticated
    # session (the landing page title is exactly "muse.ai"). The browser may
    # still be parked on "/" (fresh relaunch start URL) — reported separately
    # via parked_on_landing, not as a session failure.
    if ("Chat \u2014" in t) or ("Muse \u2014" in t) or ("Chat -" in t) or ("Muse -" in t):
        return "LOGGED_IN"
    if "Log in" in b:
        return "LOGGED_OUT"
    return "UNKNOWN"


def probe(agent, timeout=15):
    accounts = _load_accounts()
    if agent not in accounts:
        return {"agent": agent, "state": "UNKNOWN",
                "error": "no such agent in registry",
                "checked_at": _now()}
    node, cdp_url = accounts[agent]
    try:
        with urllib.request.urlopen(cdp_url, timeout=5) as r:
            targets = json.load(r)
    except Exception as e:
        return {"agent": agent, "state": "CDP_UNREACHABLE",
                "error": "cdp list failed: %s" % str(e)[:120],
                "checked_at": _now()}
    pages = [t for t in targets if t.get("type") == "page"]
    if not pages:
        return {"agent": agent, "state": "CDP_UNREACHABLE",
                "error": "no page target", "checked_at": _now()}

    slot = cdp_slot(node, priority=PRIORITY_LOW) if HAS_CDP_QUEUE \
        else contextlib.nullcontext()
    try:
        with slot:
            ws = websocket.create_connection(
                pages[0]["webSocketDebuggerUrl"], timeout=timeout)
            try:
                raw = _ev(ws, _STATE_JS)
            finally:
                ws.close()
    except Exception as e:
        return {"agent": agent, "state": "CDP_UNREACHABLE",
                "error": "cdp session failed: %s" % str(e)[:120],
                "checked_at": _now()}
    if not raw:
        return {"agent": agent, "state": "UNKNOWN",
                "error": "empty evaluate result", "checked_at": _now()}
    try:
        snap = json.loads(raw)
    except Exception:
        return {"agent": agent, "state": "UNKNOWN",
                "error": "unparseable evaluate result", "checked_at": _now()}

    state = classify(snap.get("title"), snap.get("url"),
                     snap.get("body"), snap.get("has_input"))
    url = snap.get("url") or ""
    parked = url.rstrip("/") in ("https://muse.ai", "https://muse.ai/")
    return {"agent": agent, "state": state, "title": snap.get("title"),
            "url": url[:120], "parked_on_landing": parked,
            "has_input": snap.get("has_input"),
            "checked_at": _now()}


def _now():
    return datetime.now(timezone.utc).isoformat()


def main():
    accounts = _load_accounts()
    p = argparse.ArgumentParser(
        description="Classify a fleet agent's muse.ai login state (read-only).")
    p.add_argument("--agent", required=True, choices=sorted(accounts.keys()))
    p.add_argument("--timeout", type=int, default=15)
    args = p.parse_args()
    result = probe(args.agent, timeout=args.timeout)
    print(json.dumps(result))
    sys.stdout.flush()
    sys.exit(0 if result.get("state") == "LOGGED_IN" else 1)


if __name__ == "__main__":
    main()