From c9143a558bd64a06965d7dda3dcdde8c1d532bd4 Mon Sep 17 00:00:00 2001 From: Muse Sidechat Date: Tue, 6 Oct 2026 18:16:14 +0000 Subject: [PATCH] fix: truthful fleet status in blind shells + agent-health circuit breaker box fleet status / approvals check misreported every node as STOPPED / CDP-unreachable from sandboxed shells (own PID+net namespaces: pgrep blind, no route to 10.201.x.x, no sudo). Fleet was healthy throughout. - bin/host_evidence.py (new): host watchdog evidence fallback. Recent timer runs (journal -o json, exact UNIT match) with no newer failure line in cdp-relay-watchdog.log / chromebox-watchdog.log (both silent-when-healthy) prove a node is up. def/dev have no watchdog coverage: browser verdict via chromebox-.log freshness (alive-only), CDP verdict unknown. - super-cli.py: effective status/source/evidence per node. Host evidence decides ONLY the fully-blind pattern (both local probes negative); live local signals always win. New UNKNOWN badge, [*] footnote; approvals UNREACHABLE splits into BLIND / OFFLINE(host agrees) / unreachable-evidence-inconclusive, with honest footer. proc_alive/cdp_ok keep local-probe meaning; status/source/evidence are new JSON fields. - approvals.py: host_cdp_ok flag on the unreachable path. - agent-health.sh: restart circuit breaker. 3 consecutive futile restarts (restart leaves agent still failing) opens the circuit: no more kills for 1800s, ALERT to log+journal, half-open probe after cooldown, reset on any success. Stops the def murder loop (57 restarts / 155 API FAILs for an account-layer failure). - tests/test_fleet_status.py (25), tests/test_agent_health.py (6). - CHROMEBOX-RUNBOOK.md: blind-shell status + futile-restart sections. Tests: 98/98 focused green (agent_health + fleet_status + completion + tool_calls). Live-verified: 4 ACTIVE [*] + 2 UNKNOWN. --- bin/agent-health.sh | 64 ++++++- bin/approvals.py | 14 ++ bin/host_evidence.py | 348 +++++++++++++++++++++++++++++++++++++ bin/super-cli.py | 86 ++++++++- docs/CHROMEBOX-RUNBOOK.md | 31 ++++ tests/test_agent_health.py | 87 ++++++++++ tests/test_fleet_status.py | 280 +++++++++++++++++++++++++++++ 7 files changed, 901 insertions(+), 9 deletions(-) create mode 100644 bin/host_evidence.py create mode 100644 tests/test_agent_health.py create mode 100644 tests/test_fleet_status.py diff --git a/bin/agent-health.sh b/bin/agent-health.sh index 5bb7d0c..9cc56c3 100755 --- a/bin/agent-health.sh +++ b/bin/agent-health.sh @@ -115,6 +115,57 @@ restart_browser() { echo "$(date -Iseconds) $agent: browser restarted" >> "$LOG" } +# 2026-10-06: restart circuit breaker. A restart that leaves the agent +# still failing is FUTILE (observed 2026-10-06: def's API check failed +# 155x while its CDP port was up; 57 kill+restart cycles murdered a +# healthy browser for an account-layer failure restarts cannot fix). +# After FUTILE_THRESHOLD consecutive futile restarts, open the circuit: +# stop killing/restarting and alert, until CIRCUIT_COOLDOWN seconds pass +# (one half-open probe restart) or any check succeeds. Manual reset: +# rm /tmp/agent-health-state/circuit- /tmp/agent-health-state/futile- +FUTILE_THRESHOLD=3 +CIRCUIT_COOLDOWN=1800 + +# circuit_allows : return 0 if a restart may proceed now. +circuit_allows() { + local agent=$1 now opened retry_in + local cf="$STATE_DIR/circuit-$agent" + [ -f "$cf" ] || return 0 + opened=$(cat "$cf" 2>/dev/null || echo 0) + now=$(date +%s) + if [ $(( now - opened )) -ge $CIRCUIT_COOLDOWN ]; then + echo "$(date -Iseconds) $agent: circuit half-open after ${CIRCUIT_COOLDOWN}s cooldown, one probe restart" >> "$LOG" + return 0 + fi + retry_in=$(( (opened + CIRCUIT_COOLDOWN - now + 59) / 60 )) + echo "$(date -Iseconds) $agent: CIRCUIT OPEN - skipping kill/restart (restarts futile, probe retry in ~${retry_in}m; manual reset: rm $cf)" >> "$LOG" + return 1 +} + +# circuit_note_restart : record a restart outcome. +circuit_note_restart() { + local agent=$1 outcome=$2 count=0 + local ff="$STATE_DIR/futile-$agent" cf="$STATE_DIR/circuit-$agent" + if [ "$outcome" = "ok" ]; then + rm -f "$ff" "$cf" 2>/dev/null + return 0 + fi + [ -f "$ff" ] && count=$(cat "$ff" 2>/dev/null || echo 0) + count=$(( count + 1 )) + echo "$count" > "$ff" + if [ "$count" -ge "$FUTILE_THRESHOLD" ]; then + date +%s > "$cf" + local msg="$agent: ALERT - $count consecutive futile restarts, circuit OPEN for ${CIRCUIT_COOLDOWN}s (no more kills until probe; manual reset: rm $cf $ff)" + echo "$(date -Iseconds) $msg" >> "$LOG" + echo "agent-health ALERT: $msg" + fi +} + +# Allow sourcing for tests without running checks. +if [ "${AGENT_HEALTH_LIB_ONLY:-}" = "1" ]; then + return 0 2>/dev/null || exit 0 +fi + # Main echo "=== Health check $(date -Iseconds) ===" >> "$LOG" @@ -126,8 +177,9 @@ check_one() { local rc=$? if [ $rc -eq 0 ]; then - # Healthy: reset the consecutive-API-failure counter. - rm -f "$STATE_DIR/failcount-$agent" 2>/dev/null + # Healthy: reset the consecutive-API-failure counter and close + # any open restart circuit. + rm -f "$STATE_DIR/failcount-$agent" "$STATE_DIR/futile-$agent" "$STATE_DIR/circuit-$agent" 2>/dev/null return 0 fi @@ -158,6 +210,11 @@ check_one() { return 0 fi + # Circuit breaker: repeated futile restarts stop here until cooldown. + if ! circuit_allows "$agent"; then + return 0 + fi + restart_browser "$agent" "$cdp_port" # 2026-10-04: post-restart re-check grace extended to ~60s total # (restart_browser sleeps 15s internally + 45s here), matching @@ -166,9 +223,10 @@ check_one() { sleep 45 if ! check_agent "$agent" "$agent" "$cdp_port"; then echo "$(date -Iseconds) $agent: CRITICAL - still down after restart" >> "$LOG" - # TODO: Alert operator (e.g., via board post or email) + circuit_note_restart "$agent" fail else echo "$(date -Iseconds) $agent: RECOVERED after restart" >> "$LOG" + circuit_note_restart "$agent" ok fi } diff --git a/bin/approvals.py b/bin/approvals.py index 855131b..2556070 100755 --- a/bin/approvals.py +++ b/bin/approvals.py @@ -616,11 +616,25 @@ def inspect_node_approvals(node: str) -> dict: "ws_url": "", "key_request": key_req, } + # Local CDP probe failed. Ask host evidence whether the node is + # really down or this shell is just blind (sandboxed netns). + host_ok = None + try: + import host_evidence + ev = host_evidence.collect([node]).get(node) or {} + bv, cv = ev.get("browser"), ev.get("cdp") + if cv == "down" or bv == "down": + host_ok = False + elif cv == "healthy" or bv == "healthy": + host_ok = True + except Exception: + host_ok = None return { "node": node, "status": "UNREACHABLE", "error": str(e), "has_pending": False, + "host_cdp_ok": host_ok, } all_input_waits = [] diff --git a/bin/host_evidence.py b/bin/host_evidence.py new file mode 100644 index 0000000..acdf3f7 --- /dev/null +++ b/bin/host_evidence.py @@ -0,0 +1,348 @@ +#!/usr/bin/env python3 +"""Host-side fleet evidence for network/PID-blind shells. + +`box fleet status` probes each node live (pgrep for the chromium process, +HTTP to the CDP relay on the peer IP). Both probes assume the caller's +network + PID namespace is the bl host's. From a sandboxed shell (own PID +and net namespaces, no sudo, no route to 10.201.x.x) both probes always +fail, so every node misreports as STOPPED even with a healthy fleet. + +This module provides the fallback signal: evidence written by the +host-side watchdogs that run on bl unsandboxed via systemd timers: + + - cdp-relay-watchdog (every 5 min, all active registry nodes): + log ``cdp-relay-watchdog.log`` + journal unit + ``cdp-relay-watchdog.service``. Proves the CDP relay path end to end. + - chromebox-watchdog (every 2 min, same nodes, one timer per node): + log ``chromebox-watchdog.log`` + journal units + ``chromebox-watchdog@.service``. Curls CDP inside the node netns, + so it proves browser + in-netns CDP. + - ``chromebox-.log`` mtime: chromium's own stdout. Fresh output + proves the browser process is alive. Used only for nodes outside + watchdog coverage — and only to conclude "alive", never "dead". + +Both watchdogs are silent-when-healthy: a failure is ALWAYS logged, so a +recent timer run (journal "Starting" line) with no newer failure line for +the node means that run found the node healthy. + +Verdicts: "healthy" | "degraded" | "down" | "unknown". +""" + +import json +import os +import re +import subprocess +import time +from datetime import datetime, timezone +from pathlib import Path + +NETVM_ROOT = Path("/home/super/Projects/NetVM") +RELAY_LOG = NETVM_ROOT / "cdp-relay-watchdog.log" +CHROMEBOX_LOG = NETVM_ROOT / "chromebox-watchdog.log" + +ALL_NODES = ("muse", "pip", "646", "opm", "def", "dev") + + +def _covered_nodes(): + """Nodes with watchdog coverage, from the fleet registry. + + Both watchdogs supervise every active registry node. Falls back to + ALL_NODES when the registry is unreadable, so a broken registry can + never silently narrow fleet status to a subset of the fleet. + """ + try: + import importlib.util + spec = importlib.util.spec_from_file_location( + "netvm_registry", NETVM_ROOT / "bin" / "netvm-registry.py") + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return tuple(sorted(mod.active_nodes())) + except Exception: + return ALL_NODES + + +RELAY_NODES = _covered_nodes() +CHROMEBOX_NODES = _covered_nodes() + +RELAY_UNIT = "cdp-relay-watchdog.service" +CHROMEBOX_UNIT_TMPL = "chromebox-watchdog@{node}.service" + +# A watchdog run older than this proves nothing (timer may be dead). +RELAY_STALE_MIN = 15 +CHROMEBOX_STALE_MIN = 8 +# Chromium stdout older than this proves nothing (idle browsers go quiet). +CHROME_LOG_FRESH_MIN = 20 + +_LOG_TS_RE = re.compile(r"^\[(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})Z\]") + + +def _utcnow(): + return datetime.now(timezone.utc) + + +def parse_log_ts(line): + """Parse a ``[YYYY-MM-DDTHH:MM:SSZ]`` log prefix. None if absent.""" + m = _LOG_TS_RE.match(line) + if not m: + return None + try: + return datetime.strptime(m.group(1), "%Y-%m-%dT%H:%M:%S").replace( + tzinfo=timezone.utc) + except ValueError: + return None + + +def classify_chromebox_line(line): + """Classify one chromebox-watchdog log line. + + Returns "healthy" | "degraded" | "down", or None when the line + carries no verdict (rotation markers, relay stdout passthrough). + """ + if "log rotated" in line: + return None + if "relaunch FAILED" in line: + return "down" + if "relaunch OK" in line: + return "healthy" + if "recovered" in line and "relaunch not needed" in line: + return "healthy" + if "not healthy yet" in line: + return "degraded" + if "proceeding with chrome relaunch" in line: + return "degraded" + if "relaunching chromebox" in line: + return "degraded" + if "warp partition detected" in line: + return "degraded" + if "skipping relaunch (probably still starting)" in line: + return "degraded" + return None + + +def classify_relay_line(line): + """Classify one cdp-relay-watchdog log line (None = no verdict).""" + if "relay restart FAILED" in line: + return "down" + if "FAIL_LOUD" in line: + return "down" + if "relay restarted OK" in line: + return "healthy" + if "relay unhealthy" in line and "restarting" in line: + # Always followed by an OK/FAILED line; a trailing one means the + # restart crashed mid-flight. + return "down" + return None + + +def last_verdict(lines, node, classify): + """Newest (verdict, ts, line) for ``[node]``. None if no verdict line.""" + tag = "[%s]" % node + best = None + for line in lines: + if tag not in line: + continue + verdict = classify(line) + if verdict is None: + continue + ts = parse_log_ts(line) + if ts is None: + continue + if best is None or ts >= best[1]: + best = (verdict, ts, line.strip()[:160]) + return best + + +def _tail_lines(path, max_bytes=65536): + try: + size = os.path.getsize(path) + with open(path, "rb") as f: + if size > max_bytes: + f.seek(size - max_bytes) + f.readline() # drop partial first line + return f.read().decode("utf-8", errors="replace").splitlines() + except OSError: + return [] + + +def query_journal_starts(units, since_min=25, timeout=20): + """Map each unit -> newest run-start (aware UTC). Missing on failure. + + Uses ``-o json``: the short-format "Starting" line carries the unit + description, not the unit name, so exact per-unit matching needs the + structured UNIT field. + """ + cmd = ["journalctl", "--no-pager", "-o", "json", + "--since", "%d min ago" % since_min] + for u in units: + cmd.extend(["-u", u]) + try: + r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + except (OSError, subprocess.TimeoutExpired): + return {} + if r.returncode != 0: + return {} + want = set(units) + starts = {} + for line in (r.stdout or "").splitlines(): + try: + e = json.loads(line) + except ValueError: + continue + if e.get("UNIT") not in want: + continue + if not (e.get("MESSAGE") or "").startswith("Starting"): + continue + try: + ts = datetime.fromtimestamp( + int(e["__REALTIME_TIMESTAMP"]) / 1e6, tz=timezone.utc) + except (KeyError, ValueError, TypeError, OverflowError): + continue + u = e["UNIT"] + if u not in starts or ts > starts[u]: + starts[u] = ts + return starts + + +def _verdict_since_run(verdict_row, run_ts): + """True when the verdict line is newer than (or from) the last run.""" + if verdict_row is None or run_ts is None: + return False + return verdict_row[1] >= run_ts + + +def browser_verdict(node, chromebox_lines, run_ts, chrome_log_mtime=None, + now=None): + """(verdict, detail) for the node's browser process.""" + now = now or _utcnow() + row = last_verdict(chromebox_lines, node, classify_chromebox_line) + if node in CHROMEBOX_NODES: + if run_ts is None: + return ("unknown", "no chromebox-watchdog run in journal window") + if (now - run_ts).total_seconds() > CHROMEBOX_STALE_MIN * 60: + return ("unknown", "chromebox-watchdog run is stale") + if _verdict_since_run(row, run_ts): + return (row[0], "watchdog: %s" % row[2]) + return ("healthy", "watchdog run silent (silent-when-healthy)") + # Nodes outside watchdog coverage: chromium stdout proves alive only. + if chrome_log_mtime is not None and ( + now - chrome_log_mtime).total_seconds() < CHROME_LOG_FRESH_MIN * 60: + return ("healthy", "chromebox-%s.log fresh" % node) + return ("unknown", "no watchdog coverage for %s" % node) + + +def cdp_verdict(node, relay_lines, run_ts, now=None): + """(verdict, detail) for the node's host-reachable CDP relay path.""" + now = now or _utcnow() + if node not in RELAY_NODES: + return ("unknown", "no relay-monitor coverage for %s" % node) + if run_ts is None: + return ("unknown", "no cdp-relay-watchdog run in journal window") + if (now - run_ts).total_seconds() > RELAY_STALE_MIN * 60: + return ("unknown", "cdp-relay-watchdog run is stale") + row = last_verdict(relay_lines, node, classify_relay_line) + if _verdict_since_run(row, run_ts): + return (row[0], "relay watchdog: %s" % row[2]) + return ("healthy", "relay watchdog run silent (silent-when-healthy)") + + +def chrome_log_mtime(node): + """Mtime of chromium's stdout log as aware UTC. None if missing.""" + try: + return datetime.fromtimestamp( + os.path.getmtime(NETVM_ROOT / ("chromebox-%s.log" % node)), + tz=timezone.utc) + except OSError: + return None + + +_CACHE = {"at": 0.0, "nodes": frozenset(), "data": {}} +_CACHE_TTL_S = 60 + + +def collect(nodes=None, _journal_starts=None, _relay_lines=None, + _chromebox_lines=None, _chrome_mtimes=None, _now=None): + """Per-node host evidence. Underscore args are seams for tests.""" + nodes = list(nodes or ALL_NODES) + live = (_journal_starts is None and _relay_lines is None + and _chromebox_lines is None and _chrome_mtimes is None + and _now is None) + if live: + key = frozenset(nodes) + if (key <= _CACHE["nodes"] + and time.monotonic() - _CACHE["at"] < _CACHE_TTL_S): + return {n: _CACHE["data"][n] for n in nodes if n in _CACHE["data"]} + now = _now or _utcnow() + if _journal_starts is None: + units = [RELAY_UNIT] + [CHROMEBOX_UNIT_TMPL.format(node=n) + for n in nodes if n in CHROMEBOX_NODES] + _journal_starts = query_journal_starts(units) + if _relay_lines is None: + _relay_lines = _tail_lines(RELAY_LOG) + if _chromebox_lines is None: + _chromebox_lines = _tail_lines(CHROMEBOX_LOG) + out = {} + for node in nodes: + if _chrome_mtimes is not None and node in _chrome_mtimes: + mtime = _chrome_mtimes[node] + else: + mtime = chrome_log_mtime(node) if node not in CHROMEBOX_NODES else None + b_verd, b_det = browser_verdict( + node, _chromebox_lines, + _journal_starts.get(CHROMEBOX_UNIT_TMPL.format(node=node)), + chrome_log_mtime=mtime, now=now) + c_verd, c_det = cdp_verdict( + node, _relay_lines, _journal_starts.get(RELAY_UNIT), now=now) + out[node] = { + "browser": b_verd, + "browser_detail": b_det, + "cdp": c_verd, + "cdp_detail": c_det, + } + if live: + _CACHE["at"] = time.monotonic() + _CACHE["nodes"] = frozenset(nodes) + _CACHE["data"] = out + return out + + +def effective_status(local_proc, local_cdp, browser_v, cdp_v): + """Map (local probes, host verdicts) -> (status, source). + + Host evidence only ever overrides the fully-blind pattern (both + local probes negative — the sandbox signature). It never overrides + a live local signal, so a fresh outage on the host always wins. + """ + if local_cdp: + # CDP answers: the browser is definitionally alive. + return ("ACTIVE", "local") + if local_proc: + return ("CDP_DOWN", "local") + # Both local probes negative: consult host evidence. + if browser_v == "down": + return ("STOPPED", "host-evidence") + if browser_v == "unknown": + return ("UNKNOWN", "host-evidence") + if cdp_v == "healthy": + return ("ACTIVE", "host-evidence") + if cdp_v == "down": + return ("CDP_DOWN", "host-evidence") + return ("UNKNOWN", "host-evidence") + + +def main(argv=None): + import argparse + ap = argparse.ArgumentParser(description="Show host-side fleet evidence") + ap.add_argument("--json", action="store_true") + args = ap.parse_args(argv) + data = collect() + if args.json: + print(json.dumps({"ok": True, "evidence": data}, indent=2)) + return + for node, ev in data.items(): + print("%-6s browser=%-8s cdp=%-8s" % (node, ev["browser"], ev["cdp"])) + print(" browser: %s" % ev["browser_detail"]) + print(" cdp: %s" % ev["cdp_detail"]) + + +if __name__ == "__main__": + main() diff --git a/bin/super-cli.py b/bin/super-cli.py index 9492ae3..708fef9 100755 --- a/bin/super-cli.py +++ b/bin/super-cli.py @@ -252,6 +252,53 @@ def get_queue_depth(node: str) -> int: # --------------------------------------------------------------------------- # Domain: FLEET # --------------------------------------------------------------------------- +def _host_evidence_for(results): + """Host watchdog evidence, or None when unneeded/unavailable. + + Queried only when at least one node is fully blind (both local + probes negative). Never raises: evidence must not break status. + """ + if not any(not r["proc_alive"] and not r["cdp_ok"] for r in results): + return None + try: + import host_evidence + return host_evidence.collect([r["node"] for r in results]) + except Exception: + return None + + +def _resolve_statuses(results) -> None: + """Attach effective status/source/evidence to each collected node. + + proc_alive/cdp_ok keep their local-probe meaning; status is the + display verdict. Host evidence only decides the fully-blind pattern + (both local probes negative); live local signals always win. + """ + try: + import host_evidence + have_mod = True + except ImportError: + have_mod = False + ev_map = _host_evidence_for(results) if have_mod else None + for r in results: + ev = (ev_map or {}).get(r["node"]) if ev_map else None + if not r["proc_alive"] and r["cdp_ok"]: + # CDP answers: the browser is definitionally alive even when + # pgrep cannot see it (PID-blind shell, renamed profile path). + r["status"], r["source"] = "ACTIVE", "local" + elif r["proc_alive"] and not r["cdp_ok"]: + r["status"], r["source"] = "CDP_DOWN", "local" + elif r["proc_alive"] and r["cdp_ok"]: + r["status"], r["source"] = "ACTIVE", "local" + elif ev is None: + r["status"], r["source"] = "STOPPED", "local" + else: + r["status"], r["source"] = host_evidence.effective_status( + False, False, ev.get("browser", "unknown"), + ev.get("cdp", "unknown")) + r["evidence"] = ev + + def collect_fleet_data() -> list: results = [] for node in VALID_NODES: @@ -287,6 +334,7 @@ def collect_fleet_data() -> list: "approval_pending": approval_pending, "approval_detail": approval_detail, }) + _resolve_statuses(results) return results def cmd_fleet_status(args): @@ -301,14 +349,16 @@ def cmd_fleet_status(args): rows = [] has_any_approval = False for item in data: - # Status calculation + # Status calculation (effective status; see _resolve_statuses) if item.get("approval_pending"): status = badge_warn("APPROVAL_REQ") has_any_approval = True - elif item["proc_alive"] and item["cdp_ok"]: + elif item.get("status") == "ACTIVE": status = badge_ok("ACTIVE") - elif item["proc_alive"] and not item["cdp_ok"]: + elif item.get("status") == "CDP_DOWN": status = badge_warn("CDP_DOWN") + elif item.get("status") == "UNKNOWN": + status = badge_dim("UNKNOWN") else: status = badge_err("STOPPED") @@ -338,6 +388,8 @@ def cmd_fleet_status(args): ]) print_table(headers, rows) + if any(r.get("source") == "host-evidence" for r in data): + print(c_dim(" [*] status via host watchdog evidence (local probes blind in this shell)")) if has_any_approval: print("\n" + c_yellow(" ⚠ Agent(s) held up on browser approval. Run 'box approvals' to inspect/allow.")) print("\n" + c_dim(" Commands: super fleet watch | super fleet restart | box approvals [check|allow|auto]") + "\n") @@ -446,9 +498,16 @@ def cmd_approvals(args): btns = c_dim("answer in task") input_wait_count += 1 elif st == "UNREACHABLE": - badge = badge_dim("OFFLINE") + if it.get("host_cdp_ok") is True: + badge = badge_dim("BLIND") + purp = c_dim("CDP ok on host; blind here") + elif it.get("host_cdp_ok") is False: + badge = badge_err("OFFLINE") + purp = c_dim("CDP down (host agrees)") + else: + badge = badge_dim("OFFLINE") + purp = c_dim("CDP unreachable") target = "-" - purp = c_dim("CDP unreachable") trust = "-" btns = "-" elif st == "ERROR": @@ -478,7 +537,22 @@ def cmd_approvals(args): print(c_cyan(" All pending approvals are for trusted infrastructure. Run 'box approvals auto' to resolve.")) print() elif input_wait_count == 0: - print("\n" + c_green(" ✔ All agent approval queues clear. No agents blocked.") + "\n") + unreach = [it for it in fleet if it.get("status") == "UNREACHABLE"] + blind_ok = [it["node"] for it in unreach + if it.get("host_cdp_ok") is True] + blind_unknown = [it["node"] for it in unreach + if it.get("host_cdp_ok") is None] + if blind_ok: + print("\n" + c_dim(" ? %d node(s) blind from this shell " + "(CDP ok on host); queues unverified: %s." + % (len(blind_ok), ", ".join(blind_ok))) + "\n") + if blind_unknown: + print("\n" + c_dim(" ? %d node(s) unreachable, host evidence " + "inconclusive: %s." + % (len(blind_unknown), + ", ".join(blind_unknown))) + "\n") + if not blind_ok and not blind_unknown: + print("\n" + c_green(" ✔ All agent approval queues clear. No agents blocked.") + "\n") # Verbose or inspect breakdown is_verbose = getattr(args, "verbose", False) or action == "inspect" diff --git a/docs/CHROMEBOX-RUNBOOK.md b/docs/CHROMEBOX-RUNBOOK.md index 7f057e8..aea3650 100644 --- a/docs/CHROMEBOX-RUNBOOK.md +++ b/docs/CHROMEBOX-RUNBOOK.md @@ -190,6 +190,19 @@ recent-launch guard. **Fix:** fixed 2026-10-04 — 4 retries over 60s + skip kill if launched <2 min ago. If you see this pattern again, the guard may need tuning (longer window). +### Futile-restart loop (agent-health killing a healthy browser) +**Symptoms:** `FAIL (api timeout)` + `restarting browser...` + `CRITICAL - +still down after restart` repeating every ~10 min for one node while its +CDP port stays up (observed 2026-10-06: def, 57 restarts, 155 API FAILs). +The API/account layer is broken; restarts cannot fix it, they just murder +a working browser. +**Fix:** `agent-health.sh` circuit breaker — after 3 consecutive futile +restarts the circuit OPENS (alert in log + journal, no more kills) until +a 30-min half-open probe or any successful check. Manual reset: +`rm /tmp/agent-health-state/circuit- /tmp/agent-health-state/futile-`. +Then fix the actual API-layer failure (account session/auth/chat-state), +not the browser. + ### Relay on wrong IP **Symptoms:** relay process exists but on the wrong veth IP (e.g., muse's relay on pip's `10.201.87.2` instead of muse's `10.201.35.2`). Port responds on the @@ -212,6 +225,24 @@ in the NetVM repo — check `git status` if it's gone. **Cause:** old node-ups or queue tests. Harmless but confusing. **Fix:** kill them. Only 9410/9420/9430/9440 should be running. +## Fleet Status From Blind Shells + +`box fleet status` probes live (pgrep + peer-IP CDP). Sandboxed shells +(own PID/net namespaces, no sudo, no route to 10.201.x.x) fail both +probes for every node. Instead of misreporting STOPPED, fleet status +falls back to host watchdog evidence (`bin/host_evidence.py`): + +- Recent watchdog timer runs (journal) with no newer failure line in + `cdp-relay-watchdog.log` / `chromebox-watchdog.log` (both are + silent-when-healthy) prove the node is up → `ACTIVE [*]`. +- `UNKNOWN` means neither live probes nor host evidence could decide + (e.g. def/dev have no relay-monitor coverage). +- Host evidence never overrides a live local signal, so a fresh outage + observed on the host always wins over a minutes-old watchdog run. + +Same rule drives `box approvals check`: `BLIND` = CDP ok on host, this +shell cannot reach it; approval queues are unverified, not clear. + ## Key Reference **Node bring-up:** `sudo /home/super/Projects/NetVM/bin/netvm-node-up.sh ` diff --git a/tests/test_agent_health.py b/tests/test_agent_health.py new file mode 100644 index 0000000..119ead5 --- /dev/null +++ b/tests/test_agent_health.py @@ -0,0 +1,87 @@ +"""Tests for the agent-health.sh restart circuit breaker. + +Drives the real shell functions (sourced with AGENT_HEALTH_LIB_ONLY=1) +against a scratch STATE_DIR/LOG: futile-restart counting, circuit open, +half-open probe after cooldown, and reset on success. +""" +import subprocess +import tempfile +import unittest +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent +SCRIPT = REPO_ROOT / "bin" / "agent-health.sh" + + +def _run(state_dir, log, snippet): + prog = ( + "source '%s'\n" + "STATE_DIR='%s'; LOG='%s'\n" + "%s\n" % (SCRIPT, state_dir, log, snippet) + ) + env = {"AGENT_HEALTH_LIB_ONLY": "1", "PATH": "/usr/bin:/bin"} + return subprocess.run(["bash", "-c", prog], capture_output=True, + text=True, env=env, timeout=30) + + +class CircuitBreaker(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.state = str(Path(self.tmp.name) / "state") + Path(self.state).mkdir() + self.log = str(Path(self.tmp.name) / "log") + + def tearDown(self): + self.tmp.cleanup() + + def bash(self, snippet): + return _run(self.state, self.log, snippet) + + def test_allows_when_closed(self): + r = self.bash("circuit_allows def") + self.assertEqual(r.returncode, 0, r.stderr) + + def test_allows_below_threshold(self): + r = self.bash("circuit_note_restart def fail\n" + "circuit_note_restart def fail\n" + "circuit_allows def") + self.assertEqual(r.returncode, 0, r.stderr) + self.assertEqual((Path(self.state) / "futile-def").read_text().strip(), "2") + + def test_opens_after_threshold_and_alerts(self): + r = self.bash("circuit_note_restart def fail\n" + "circuit_note_restart def fail\n" + "circuit_note_restart def fail") + self.assertEqual(r.returncode, 0, r.stderr) + self.assertTrue((Path(self.state) / "circuit-def").exists()) + self.assertIn("ALERT", r.stdout) + self.assertIn("ALERT", Path(self.log).read_text()) + r2 = self.bash("circuit_allows def") + self.assertNotEqual(r2.returncode, 0) + self.assertIn("CIRCUIT OPEN", Path(self.log).read_text()) + + def test_half_open_after_cooldown(self): + old = "echo $(( $(date +%%s) - 1900 )) > '%s/circuit-def'" % self.state + r = self.bash(old + "\ncircuit_allows def") + self.assertEqual(r.returncode, 0, r.stderr) + self.assertIn("half-open", Path(self.log).read_text()) + + def test_ok_resets(self): + r = self.bash("circuit_note_restart def fail\n" + "circuit_note_restart def fail\n" + "circuit_note_restart def fail\n" + "circuit_note_restart def ok") + self.assertEqual(r.returncode, 0, r.stderr) + self.assertFalse((Path(self.state) / "futile-def").exists()) + self.assertFalse((Path(self.state) / "circuit-def").exists()) + + def test_per_agent_isolation(self): + r = self.bash("circuit_note_restart def fail\n" + "circuit_note_restart def fail\n" + "circuit_note_restart def fail\n" + "circuit_allows pip") + self.assertEqual(r.returncode, 0, r.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_fleet_status.py b/tests/test_fleet_status.py new file mode 100644 index 0000000..298bbf9 --- /dev/null +++ b/tests/test_fleet_status.py @@ -0,0 +1,280 @@ +"""Tests for host-side fleet evidence (bin/host_evidence.py). + +Covers log-line classification, verdict assembly (silent-when-healthy +runs vs. fresh failures vs. stale/missing runs), and the effective +status mapping used by `box fleet status` in PID/net-blind shells. +""" +import importlib.util +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent + + +def _load(mod_name, rel_path): + spec = importlib.util.spec_from_file_location(mod_name, REPO_ROOT / rel_path) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +he = _load("host_evidence_test", "bin/host_evidence.py") + +NOW = datetime(2026, 10, 6, 8, 40, 0, tzinfo=timezone.utc) + + +def _ts(dt): + return dt.strftime("[%Y-%m-%dT%H:%M:%SZ]") + + +def _run(minutes_ago): + return NOW - timedelta(minutes=minutes_ago) + + +class ParseLogTs(unittest.TestCase): + def test_valid(self): + self.assertEqual( + he.parse_log_ts("[2026-10-06T05:55:25Z] [opm] relay restart FAILED"), + datetime(2026, 10, 6, 5, 55, 25, tzinfo=timezone.utc)) + + def test_garbage(self): + self.assertIsNone(he.parse_log_ts("cdp relay already running")) + self.assertIsNone(he.parse_log_ts("")) + self.assertIsNone(he.parse_log_ts("[2026-13-99T99:99:99Z] [muse] x")) + + +class ClassifyChromebox(unittest.TestCase): + def test_down(self): + self.assertEqual( + he.classify_chromebox_line("[muse] relaunch FAILED (x) — needs operator attention"), + "down") + + def test_healthy(self): + self.assertEqual( + he.classify_chromebox_line("[muse] relaunch OK (attempt 2)"), "healthy") + self.assertEqual( + he.classify_chromebox_line( + "[muse] tunnel restart recovered warp egress, chrome relaunch not needed"), + "healthy") + + def test_degraded(self): + for line in ( + "[muse] relaunch attempt 3 not healthy yet (CDP up but no page)", + "[muse] tunnel restart did not recover egress (x), proceeding with chrome relaunch", + "[muse] unhealthy (x), relaunching chromebox", + "[muse] warp partition detected (x), restarting tunnel via netvm-node-up.sh", + "[muse] browser launched recently (pid 1), skipping relaunch (probably still starting)", + ): + self.assertEqual(he.classify_chromebox_line(line), "degraded", line) + + def test_no_verdict(self): + for line in ( + "[2026-10-06T08:00:00Z] [muse] log rotated", + "cdp relay already running", + "tunnel already up (egress=1.2.3.4), skipping handshake wait", + "node=muse netns=warp-muse egress=1.2.3.4", + "Traceback (most recent call last):", + ): + self.assertIsNone(he.classify_chromebox_line(line), line) + + +class ClassifyRelay(unittest.TestCase): + def test_down(self): + self.assertEqual( + he.classify_relay_line("[opm] relay restart FAILED on 10.201.157.2:9440 — needs operator"), + "down") + self.assertEqual( + he.classify_relay_line("[pip] FAIL_LOUD: host veth ve-x missing IP 10.201.87.1"), + "down") + # Trailing "restarting" with no OK/FAILED after it: crashed mid-restart. + self.assertEqual( + he.classify_relay_line("[muse] relay unhealthy on 10.201.35.2:9410, restarting"), + "down") + + def test_healthy(self): + self.assertEqual( + he.classify_relay_line("[646] relay restarted OK on 10.201.202.2:9430"), + "healthy") + + def test_no_verdict(self): + self.assertIsNone(he.classify_relay_line(" File \"/x/netvm-cdp-relay.py\", line 37, in main")) + self.assertIsNone(he.classify_relay_line("")) + + +class LastVerdict(unittest.TestCase): + def test_newest_wins_per_node(self): + lines = [ + "%s [muse] relaunch OK (attempt 1)" % _ts(_run(30)), + "%s [pip] relaunch FAILED (x)" % _ts(_run(20)), + "%s [muse] unhealthy (x), relaunching chromebox" % _ts(_run(5)), + "%s [muse] log rotated" % _ts(_run(1)), + ] + v, ts, _ = he.last_verdict(lines, "muse", he.classify_chromebox_line) + self.assertEqual(v, "degraded") + self.assertEqual(ts, _run(5)) + v, _, _ = he.last_verdict(lines, "pip", he.classify_chromebox_line) + self.assertEqual(v, "down") + + def test_none_when_no_verdict(self): + self.assertIsNone(he.last_verdict( + ["[2026-10-06T08:00:00Z] [muse] log rotated", "noise"], + "muse", he.classify_chromebox_line)) + self.assertIsNone(he.last_verdict([], "muse", he.classify_chromebox_line)) + + +class BrowserVerdict(unittest.TestCase): + def test_silent_run_is_healthy(self): + v, d = he.browser_verdict("muse", [], _run(2), now=NOW) + self.assertEqual(v, "healthy") + v, _ = he.browser_verdict( + "muse", ["%s [muse] relaunch FAILED (x)" % _ts(_run(60))], + _run(2), now=NOW) + self.assertEqual(v, "healthy") # old failure, silent since + + def test_fresh_failure_wins(self): + v, _ = he.browser_verdict( + "muse", ["%s [muse] relaunch FAILED (x)" % _ts(_run(1))], + _run(2), now=NOW) + self.assertEqual(v, "down") + v, _ = he.browser_verdict( + "muse", ["%s [muse] unhealthy (x), relaunching chromebox" % _ts(_run(1))], + _run(2), now=NOW) + self.assertEqual(v, "degraded") + + def test_fresh_recovery_is_healthy(self): + v, _ = he.browser_verdict( + "muse", ["%s [muse] tunnel restart recovered warp egress, chrome relaunch not needed" + % _ts(_run(1))], + _run(2), now=NOW) + self.assertEqual(v, "healthy") + + def test_stale_or_missing_run_is_unknown(self): + v, _ = he.browser_verdict("muse", [], _run(30), now=NOW) + self.assertEqual(v, "unknown") + v, _ = he.browser_verdict("muse", [], None, now=NOW) + self.assertEqual(v, "unknown") + + def test_def_uses_watchdog_like_core_nodes(self): + v, _ = he.browser_verdict("def", [], _run(2), now=NOW) + self.assertEqual(v, "healthy") # silent run + v, _ = he.browser_verdict("def", [], None, now=NOW) + self.assertEqual(v, "unknown") # no timer run (yet) + v, _ = he.browser_verdict( + "def", ["%s [def] relaunch FAILED (x)" % _ts(_run(1))], + _run(2), now=NOW) + self.assertEqual(v, "down") + + def test_uncovered_node_uses_chrome_log_freshness(self): + v, _ = he.browser_verdict("ghost", [], None, + chrome_log_mtime=_run(5), now=NOW) + self.assertEqual(v, "healthy") + v, d = he.browser_verdict("ghost", [], None, + chrome_log_mtime=_run(60), now=NOW) + self.assertEqual(v, "unknown") + v, _ = he.browser_verdict("ghost", [], None, + chrome_log_mtime=None, now=NOW) + self.assertEqual(v, "unknown") + + +class CdpVerdict(unittest.TestCase): + def test_silent_run_is_healthy(self): + v, _ = he.cdp_verdict("opm", [], _run(4), now=NOW) + self.assertEqual(v, "healthy") + + def test_fresh_failure_is_down(self): + v, _ = he.cdp_verdict( + "opm", ["%s [opm] relay restart FAILED on 10.201.157.2:9440" % _ts(_run(1))], + _run(4), now=NOW) + self.assertEqual(v, "down") + + def test_stale_or_missing_run_is_unknown(self): + v, _ = he.cdp_verdict("opm", [], _run(60), now=NOW) + self.assertEqual(v, "unknown") + v, _ = he.cdp_verdict("opm", [], None, now=NOW) + self.assertEqual(v, "unknown") + + def test_dev_def_monitored_like_core_nodes(self): + for node in ("def", "dev"): + v, _ = he.cdp_verdict(node, [], _run(4), now=NOW) + self.assertEqual(v, "healthy", node) # silent run + v, _ = he.cdp_verdict( + node, + ["%s [%s] relay restart FAILED on 10.0.0.1:1" % (_ts(_run(1)), node)], + _run(4), now=NOW) + self.assertEqual(v, "down", node) + + def test_unmonitored_node_is_unknown(self): + v, d = he.cdp_verdict("ghost", [], _run(1), now=NOW) + self.assertEqual(v, "unknown") + self.assertIn("no relay-monitor coverage", d) + + +class EffectiveStatus(unittest.TestCase): + def test_live_local_signals_win(self): + self.assertEqual(he.effective_status(True, True, "down", "down"), + ("ACTIVE", "local")) + self.assertEqual(he.effective_status(True, False, "healthy", "healthy"), + ("CDP_DOWN", "local")) + + def test_cdp_without_proc_is_active(self): + self.assertEqual(he.effective_status(False, True, "unknown", "unknown"), + ("ACTIVE", "local")) + + def test_blind_with_evidence(self): + self.assertEqual( + he.effective_status(False, False, "healthy", "healthy"), + ("ACTIVE", "host-evidence")) + self.assertEqual( + he.effective_status(False, False, "degraded", "healthy"), + ("ACTIVE", "host-evidence")) + self.assertEqual( + he.effective_status(False, False, "down", "healthy"), + ("STOPPED", "host-evidence")) + self.assertEqual( + he.effective_status(False, False, "healthy", "down"), + ("CDP_DOWN", "host-evidence")) + + def test_blind_without_evidence_is_unknown(self): + self.assertEqual( + he.effective_status(False, False, "unknown", "unknown"), + ("UNKNOWN", "host-evidence")) + self.assertEqual( + he.effective_status(False, False, "healthy", "unknown"), + ("UNKNOWN", "host-evidence")) + self.assertEqual( + he.effective_status(False, False, "unknown", "healthy"), + ("UNKNOWN", "host-evidence")) + + +class CollectSeams(unittest.TestCase): + def test_assembles_per_node_verdicts(self): + starts = {"cdp-relay-watchdog.service": _run(4), + "chromebox-watchdog@muse.service": _run(1), + "chromebox-watchdog@def.service": _run(1)} + relay_lines = ["%s [muse] relay restarted OK on 10.201.35.2:9410" % _ts(_run(3))] + chrome_lines = ["%s [muse] relaunch OK (attempt 1)" % _ts(_run(1))] + out = he.collect(["muse", "def"], + _journal_starts=starts, + _relay_lines=relay_lines, + _chromebox_lines=chrome_lines, + _chrome_mtimes={"def": _run(5)}, + _now=NOW) + self.assertEqual(out["muse"]["browser"], "healthy") + self.assertEqual(out["muse"]["cdp"], "healthy") + # def is watchdog-covered: silent runs read healthy (chrome-log + # freshness is no longer consulted for covered nodes). + self.assertEqual(out["def"]["browser"], "healthy") + self.assertEqual(out["def"]["cdp"], "healthy") + + +class CoverageTuples(unittest.TestCase): + def test_watchdog_tuples_cover_registry(self): + reg = _load("netvm_registry_cov", "bin/netvm-registry.py") + for node in reg.active_nodes(): + self.assertIn(node, he.RELAY_NODES) + self.assertIn(node, he.CHROMEBOX_NODES) + + +if __name__ == "__main__": + unittest.main()