#!/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 -- python3 /home/super/Projects/NetVM/bin/session-probe.py --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()