feat(watchers): add box stability watcher daemon, recovery guardrails, and CLI integration

- Add dedicated watchers/ project folder with box-stability-watcher.py supervisor
- Monitor host load, memory, swap saturation, and crash-looping services
- Implement tiered mitigations: yellow renicing, orange SIGSTOP pause with 60s grace, red shedding
- Distinguish user-launched agents (allowed on desktop default socket) from automated box workloads
- Wire first-class box stability CLI subcommand and top-line host status in fleet status
- Harden tmux.service with cgroup memory limits to prevent OS freeze and OOM avalanches
- Add 10-test unit test suite covering thresholds, safety whitelist, pause/resume, and isolation
This commit is contained in:
operator
2026-10-07 17:49:14 +00:00
parent c11d1d83ae
commit 0a45133d28
8 changed files with 1616 additions and 22 deletions
+1
View File
@@ -0,0 +1 @@
../watchers/box-stability-watcher.py
+575 -20
View File
@@ -348,6 +348,25 @@ def cmd_fleet_status(args):
return return
print("\n" + c_bold("=== NETVM FLEET STATUS ===") + c_dim(f" ({datetime.now().strftime('%H:%M:%S')} local)\n")) print("\n" + c_bold("=== NETVM FLEET STATUS ===") + c_dim(f" ({datetime.now().strftime('%H:%M:%S')} local)\n"))
# Top-line host stability badge
try:
sys.path.insert(0, str(NETVM_ROOT / "watchers"))
import importlib.util
spec = importlib.util.spec_from_file_location("box_stability_watcher", str(NETVM_ROOT / "watchers" / "box-stability-watcher.py"))
bsw = importlib.util.module_from_spec(spec)
spec.loader.exec_module(bsw)
m = bsw.get_system_metrics()
load_str = f"Load: {m['load_1m']} (1m) | {m['load_5m']} (5m) [Cores: {m['cpu_count']}]"
ram_str = f"RAM: {m['ram_used_pct']}%"
if m['load_1m'] < 20 and m['ram_used_pct'] < 80:
stab_badge = badge_ok("STABLE")
elif m['load_1m'] >= 60 or m['ram_used_pct'] >= 95:
stab_badge = badge_err("EMERGENCY")
else:
stab_badge = badge_warn("ELEVATED")
print(f" HOST {c_bold('bl')}: {stab_badge} {c_dim(load_str + ' ' + ram_str)}\n")
except Exception:
pass
headers = ["NODE", "STATUS", "PEER IP:PORT", "LATENCY", "QUEUE", "ACTIVE PAGE / THREAD"] headers = ["NODE", "STATUS", "PEER IP:PORT", "LATENCY", "QUEUE", "ACTIVE PAGE / THREAD"]
rows = [] rows = []
@@ -931,6 +950,49 @@ def cmd_approvals(args):
print(f" Reason: {c_cyan(reason)}") print(f" Reason: {c_cyan(reason)}")
print(c_dim(f" Operator can approve with: box approvals allow {node}")) print(c_dim(f" Operator can approve with: box approvals allow {node}"))
elif action in ("gates", "gate"):
scope = getattr(args, "scope", None)
if scope:
res = approvals.verify_coordinator_signoff(scope)
if getattr(args, "json", False):
print(json.dumps(res, indent=2))
return
if res.get("ok"):
print(c_green(f"\n✔ Coordinator gate for scope '{scope}' is SIGNED-OFF."))
print(f" Coordinator: {c_cyan(res.get('coordinator'))}")
print(f" Accepted At: {c_dim(res.get('accepted_at'))}")
print(f" Record: {res.get('doc_name')}\n")
else:
print(c_red(f"\n✖ Coordinator gate verification FAILED for scope '{scope}':"))
print(f" Error: {res.get('error')}\n")
sys.exit(1)
return
gates = approvals.scan_coordinator_gates()
if getattr(args, "json", False):
print(json.dumps({"ok": True, "gates": gates}, indent=2))
return
print("\n" + c_bold("=== COORDINATOR GATES & DECISION RECORDS ===\n"))
if not gates:
print(c_dim(" (no coordinator decision records found in docs/)"))
print()
return
headers = ["RECORD", "SCOPE", "STATUS", "COORDINATOR", "ACCEPTED AT", "TARGETS"]
rows = []
for g in gates:
st = c_green("SIGNED-OFF") if g.get("is_signed_off") else c_yellow(str(g.get("status", "DRAFT")).upper())
targets = ", ".join(g.get("signoff_targets") or []) if isinstance(g.get("signoff_targets"), list) else str(g.get("signoff_targets") or "-")
rows.append([
g.get("doc_name", "-"),
c_cyan(g.get("scope", "-")),
st,
g.get("coordinator", "-"),
str(g.get("accepted_at", "-"))[:19],
targets or "-",
])
print_table(headers, rows)
print()
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Domain: RUNTIME (agentic management of Muse CLI tmux runtimes) # Domain: RUNTIME (agentic management of Muse CLI tmux runtimes)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
@@ -939,18 +1001,24 @@ def cmd_runtime(args):
import muse_choice_watcher as mcw import muse_choice_watcher as mcw
action = getattr(args, "rt_action", None) or "list" action = getattr(args, "rt_action", None) or "list"
as_json = getattr(args, "json", False) as_json = getattr(args, "json", False)
if getattr(args, "socket", None):
args.socket = mcw.resolve_socket(args.socket)
if action == "list": if action == "list":
sock = getattr(args, "socket", None) sock = getattr(args, "socket", None)
rows = mcw.all_runtime_rows([sock] if sock else None) errors = {}
rows = mcw.all_runtime_rows([sock] if sock else None, errors=errors)
if getattr(args, "muse_only", False): if getattr(args, "muse_only", False):
rows = [r for r in rows if r["is_muse"]] rows = [r for r in rows if r["is_muse"]]
if as_json: if as_json:
print(json.dumps({"ok": True, "runtimes": rows}, indent=2)) print(json.dumps({"ok": True, "runtimes": rows,
"errors": errors}, indent=2))
return return
print(c_bold("\n=== MUSE RUNTIMES ===\n")) print(c_bold("\n=== MUSE RUNTIMES ===\n"))
if not rows: if not rows:
print(c_dim(" No panes found.")) print(c_dim(" No panes found."))
for s, e in errors.items():
print(c_dim(" %s: %s" % (s, e)))
print() print()
return return
headers = ["SOCKET", "SESSION", "NODE", "PANE", "CMD", headers = ["SOCKET", "SESSION", "NODE", "PANE", "CMD",
@@ -962,10 +1030,14 @@ def cmd_runtime(args):
state = "%s(%s/%s)" % (state, r["prompt_kind"], state = "%s(%s/%s)" % (state, r["prompt_kind"],
r["prompt_key"] or "…") r["prompt_key"] or "…")
if r["is_muse"]: if r["is_muse"]:
approve = (badge_ok("YES") if r["auto_approve"] eff_bypass = r.get("effective_bypass")
if eff_bypass is None:
eff_bypass = bool(r["auto_approve"])
approve = (badge_ok("YES") if eff_bypass
else badge_err("NO")) else badge_err("NO"))
mode = r.get("permission_mode") or "default" mode = (r.get("effective_mode")
if r.get("permission_bypass") and mode != "yolo": or r.get("permission_mode") or "default")
if eff_bypass and mode != "yolo":
mode += "!" mode += "!"
mode = mode[:14] mode = mode[:14]
else: else:
@@ -979,6 +1051,8 @@ def cmd_runtime(args):
r["pane"], (r["cmd"] or "")[:26], state, r["pane"], (r["cmd"] or "")[:26], state,
approve, mode, watcher]) approve, mode, watcher])
print_table(headers, table) print_table(headers, table)
for s, e in errors.items():
print(c_dim(" %s: %s" % (s, e)))
print() print()
elif action == "send": elif action == "send":
@@ -990,8 +1064,14 @@ def cmd_runtime(args):
if pre.get("error"): if pre.get("error"):
if as_json: if as_json:
print(json.dumps({"ok": False, "error": pre["error"], print(json.dumps({"ok": False, "error": pre["error"],
"socket": sock, "pane": pane})) "socket": sock, "pane": pane,
"detail": pre.get("detail")}))
return return
if pre["error"] == "socket_unreachable":
print(c_red("Error: cannot reach socket %s: %s"
% (sock, pre.get("detail") or "tmux error")),
file=sys.stderr)
else:
print(c_red("Error: no such pane %s on %s" % (pane, sock)), print(c_red("Error: no such pane %s on %s" % (pane, sock)),
file=sys.stderr) file=sys.stderr)
sys.exit(1) sys.exit(1)
@@ -1022,19 +1102,89 @@ def cmd_runtime(args):
sys.exit(1) sys.exit(1)
print() print()
elif action == "launch": elif action == "open":
sock = getattr(args, "socket", None) or mcw.KNOWN_SOCKETS[0] sock = getattr(args, "socket", None) or mcw.KNOWN_SOCKETS[0]
session = getattr(args, "session", None)
dry_run = getattr(args, "dry_run", False)
def _fail_open(msg, candidates=None):
if as_json:
print(json.dumps({"ok": False, "error": msg,
"socket": sock, "session": session,
"sessions": candidates or []}))
return
print(c_red("Error: %s" % msg), file=sys.stderr)
if candidates:
print(" Sessions on %s: %s" % (sock, ", ".join(candidates)),
file=sys.stderr)
sys.exit(1)
def _sessions():
r = mcw._tmux(sock, "list-sessions", "-F", "#{session_name}",
timeout=10)
if r.returncode != 0:
return []
return [ln.strip() for ln in (r.stdout or "").split("\n")
if ln.strip()]
if not os.path.exists(sock):
_fail_open("no such socket %s" % sock)
return
if not session:
names = _sessions()
if len(names) == 1:
session = names[0]
else:
_fail_open("no session given and %s on %s"
% ("no sessions found" if not names
else "multiple sessions", sock), names)
return
else:
probe = mcw._tmux(sock, "has-session", "-t", session,
timeout=10)
if probe.returncode != 0:
_fail_open("no such session %s on %s" % (session, sock),
_sessions())
return
argv = ["tmux", "-S", sock, "attach-session", "-t", session]
if dry_run or as_json:
if as_json:
print(json.dumps({"ok": True, "dry_run": bool(dry_run),
"socket": sock, "session": session,
"argv": argv}, indent=2))
return
print(c_bold("\n=== RUNTIME OPEN (dry-run) ===\n"))
print(" Socket: %s" % sock)
print(" Session: %s" % c_cyan(session))
print(" Command: %s" % shlex.join(argv))
print()
return
try:
os.execvp("tmux", argv)
except OSError as e:
print(c_red("Error: cannot exec tmux: %s" % e),
file=sys.stderr)
sys.exit(1)
elif action == "launch":
sock = getattr(args, "socket", None) or mcw.FLEET_SOCKETS[0]
# Track box-launched session
try:
box_state_file = NETVM_ROOT / ".state" / "box-launched-sessions.json"
box_state_file.parent.mkdir(parents=True, exist_ok=True)
box_data = {}
if box_state_file.exists():
import json
box_data = json.loads(box_state_file.read_text())
box_data[args.session] = {"socket": sock, "launched_at": datetime.now(timezone.utc).isoformat(), "origin": "box-cli"}
box_state_file.write_text(json.dumps(box_data, indent=2))
except Exception:
pass
session = args.session session = args.session
window = getattr(args, "window", None) window = getattr(args, "window", None)
dry_run = getattr(args, "dry_run", False) dry_run = getattr(args, "dry_run", False)
muse_args = list(getattr(args, "muse_args", None) or []) muse_args = list(getattr(args, "muse_args", None) or [])
if muse_args[:1] == ["--"]: cmdline, injected = mcw.muse_launch_cmdline(muse_args)
muse_args = muse_args[1:]
posture = mcw.muse_approval_flags(muse_args)
injected = [] if posture["flags"] else ["--disable-approval"]
launcher = (shutil.which("muse-code")
or "/home/super/.local/bin/muse-code")
cmdline = shlex.join([launcher] + injected + muse_args)
if dry_run: if dry_run:
if as_json: if as_json:
print(json.dumps({ print(json.dumps({
@@ -1047,7 +1197,7 @@ def cmd_runtime(args):
print(" Session: %s" % c_cyan(session)) print(" Session: %s" % c_cyan(session))
print(" Command: %s" % cmdline) print(" Command: %s" % cmdline)
if injected: if injected:
print(" %s auto-approve injected: %s" % ( print(" %s approval trail injected: %s" % (
c_green("✔"), " ".join(injected))) c_green("✔"), " ".join(injected)))
else: else:
print(" %s caller already sets approval flags; " print(" %s caller already sets approval flags; "
@@ -1077,6 +1227,7 @@ def cmd_runtime(args):
print(c_red("Error: launch failed: %s" % err), print(c_red("Error: launch failed: %s" % err),
file=sys.stderr) file=sys.stderr)
sys.exit(1) sys.exit(1)
mcw.wait_for_session_pane(sock, session, timeout=10)
rec = mcw.reconcile(sockets=[sock]) rec = mcw.reconcile(sockets=[sock])
if as_json: if as_json:
print(json.dumps({"ok": True, "socket": sock, print(json.dumps({"ok": True, "socket": sock,
@@ -1088,6 +1239,9 @@ def cmd_runtime(args):
print(" Command: %s" % c_dim(cmdline)) print(" Command: %s" % c_dim(cmdline))
for s in rec.get("started") or []: for s in rec.get("started") or []:
print(" %s watcher %s" % (badge_ok("STARTED"), c_cyan(s))) print(" %s watcher %s" % (badge_ok("STARTED"), c_cyan(s)))
for f in rec.get("failed") or []:
print(" %s watcher %s (retry: box muse-choices reconcile)"
% (badge_err("FAILED"), c_cyan(f)))
print() print()
elif action == "layout": elif action == "layout":
@@ -1184,6 +1338,50 @@ def cmd_runtime(args):
badge_err("FAILED"), f["pane"], f["error"])) badge_err("FAILED"), f["pane"], f["error"]))
print() print()
elif action == "reconcile":
import runtime_reconcile as rec
manifest = (getattr(args, "manifest", None)
or str(NETVM_ROOT / "fleet" / "agents.json"))
tasks = getattr(args, "tasks", None)
dry_run = getattr(args, "dry_run", False)
adopt = getattr(args, "adopt", False)
report = rec.run_reconcile(manifest, tasks_dir=tasks,
dry_run=dry_run, adopt=adopt)
if as_json:
print(json.dumps(report, indent=2))
return
print(c_bold("\n=== RUNTIME RECONCILE%s ===\n" % (
" (dry-run)" if dry_run else "")))
if not report["agents"] and not report["errors"]:
print(c_dim(" Manifest declares no agents."))
for a in report["agents"]:
if a["action"] in ("launched", "briefed", "adopted"):
mark = badge_ok(a["action"].upper())
elif a["action"] in ("failed",):
mark = badge_err("FAILED")
elif a["action"] in ("launch", "brief"):
mark = c_cyan("WOULD " + a["action"].upper())
else:
mark = c_dim("• " + a["action"])
print(" %s %s [%s]: %s" % (
mark, c_cyan(a["session"]), a["hat"], a["detail"]))
for c in report["claims"]["requeued"]:
print(" %s task %s (%s)" % (
badge_ok("REQUEUED"), c_cyan(c["task"]), c["reason"]))
for n in report.get("nudges") or []:
print(" %s %s nudge to %s: %s" % (
badge_ok("NUDGED"), n["kind"],
c_cyan(n["session"]), n["detail"]))
for e in report["claims"]["errors"] + report["errors"]:
print(" %s %s" % (badge_err("ERROR"), e))
w = report.get("watchers") or {}
if w.get("started"):
print(" %s watchers: %s" % (
badge_ok("STARTED"), ", ".join(w["started"])))
print()
if not report["ok"]:
sys.exit(1)
else: else:
if as_json: if as_json:
print(json.dumps({"ok": False, print(json.dumps({"ok": False,
@@ -1202,6 +1400,8 @@ def cmd_muse_choices(args):
import muse_choice_watcher as mcw import muse_choice_watcher as mcw
action = getattr(args, "mc_action", None) or "status" action = getattr(args, "mc_action", None) or "status"
as_json = getattr(args, "json", False) as_json = getattr(args, "json", False)
if getattr(args, "socket", None):
args.socket = mcw.resolve_socket(args.socket)
caller = os.environ.get("BOX_CALLER") or getattr(args, "from_agent", None) or "super" caller = os.environ.get("BOX_CALLER") or getattr(args, "from_agent", None) or "super"
if action == "on": if action == "on":
@@ -1358,9 +1558,15 @@ def cmd_muse_choices(args):
left = max(0, int(float(h.get("held_until", 0)) - time.time())) left = max(0, int(float(h.get("held_until", 0)) - time.time()))
except (TypeError, ValueError): except (TypeError, ValueError):
left = -1 left = -1
if mcw.hold_renews_forever(h, h.get("kind")):
when = "gate (renews, never auto-approves)"
elif left >= 0:
when = f"expires in {left}s"
else:
when = "expiry unknown"
print(f" • {badge_warn('HELD')} {c_bold(h.get('pane', '?'))} " print(f" • {badge_warn('HELD')} {c_bold(h.get('pane', '?'))} "
f"{h.get('kind')}/{h.get('key')} rule={h.get('rule')} " f"{h.get('kind')}/{h.get('key')} rule={h.get('rule')} "
f"expires in {left}s") f"{when}")
print(f" {c_dim((h.get('reason') or '')[:100])}") print(f" {c_dim((h.get('reason') or '')[:100])}")
print(f" {c_dim((h.get('text') or '')[:100])}") print(f" {c_dim((h.get('text') or '')[:100])}")
if answers: if answers:
@@ -2218,8 +2424,12 @@ def cmd_dm_send(args):
"--to", recipient, "--to", recipient,
"--target", target, "--target", target,
"--raw", "--raw",
signed_wire
] ]
if getattr(args, "expect_reply", False):
cmd.append("--expect-reply")
if getattr(args, "reply_timeout", None):
cmd.extend(["--reply-timeout", str(args.reply_timeout)])
cmd.append(signed_wire)
print(f"Dispatching Cryptographically Signed DM [{c_green('verified from:' + sender)}] -> [{c_bold(recipient)}/{target}]...") print(f"Dispatching Cryptographically Signed DM [{c_green('verified from:' + sender)}] -> [{c_bold(recipient)}/{target}]...")
res = subprocess.run(cmd) res = subprocess.run(cmd)
if res.returncode != 0: if res.returncode != 0:
@@ -5623,6 +5833,15 @@ def cmd_tmux_dispatch(args):
sys.exit(res.returncode) sys.exit(res.returncode)
def cmd_flow_dispatch(args):
"""Bridge 'box flow' commands directly to bin/flow_engine.py."""
flow_bin = str(BIN_DIR / "flow_engine.py")
f_args = getattr(args, "flow_args", []) or []
cmd = [sys.executable, flow_bin] + f_args
res = subprocess.run(cmd)
sys.exit(res.returncode)
def cmd_muse_dispatch(args): def cmd_muse_dispatch(args):
"""Bridge 'box muse' commands to muse-cli-node or interactive REPL / multi-node lookups.""" """Bridge 'box muse' commands to muse-cli-node or interactive REPL / multi-node lookups."""
m_args = getattr(args, "muse_args", []) or [] m_args = getattr(args, "muse_args", []) or []
@@ -5681,6 +5900,289 @@ def cmd_muse_dispatch(args):
sys.exit(res.returncode) sys.exit(res.returncode)
# ---------------------------------------------------------------------------
# Domain: SYSOP (one-shot fleet installer)
# ---------------------------------------------------------------------------
SYSOP_SYSTEMD_DIR = NETVM_ROOT / "systemd"
# `systemctl show -p NextElapseUSecRealtime --value` reports this when a
# timer has no computed next elapse (UINT64_MAX = USEC_INFINITY).
SYSOP_NO_NEXT = {"", "0", "n/a", "18446744073709551615"}
def sysop_user_units_dir():
return Path.home() / ".config" / "systemd" / "user"
def sysop_unit_files(systemd_dir=None):
"""Sorted unit filenames (*.timer + *.service) shipped in systemd/."""
d = Path(systemd_dir) if systemd_dir else SYSOP_SYSTEMD_DIR
if not d.is_dir():
return []
return sorted(p.name for p in d.iterdir()
if p.is_file() and p.suffix in (".timer", ".service"))
def sysop_timer_needs_anchor(timer_path):
"""True when a timer uses relative triggers (OnBootSec/OnActiveSec/
OnUnitActiveSec) whose NEXT stays empty until the service runs once."""
try:
text = Path(timer_path).read_text()
except OSError:
return False
return any(k in text for k in ("OnUnitActiveSec=", "OnBootSec=",
"OnActiveSec="))
def sysop_is_daemon(service_path):
"""True for long-lived daemons that must not be one-shot started."""
try:
text = Path(service_path).read_text()
except OSError:
return False
return "Restart=always" in text
def _sysop_next(run, timer):
"""NEXT elapse for a timer, or None when absent/unparseable."""
rc, out = run(["systemctl", "--user", "show", timer,
"-p", "NextElapseUSecRealtime", "--value"])
nxt = (out or "").strip()
if rc == 0 and nxt not in SYSOP_NO_NEXT:
return nxt
return None
def _sysop_service_state(run, service):
"""ActiveState for a service ('' when unknown)."""
rc, out = run(["systemctl", "--user", "show", service,
"-p", "ActiveState", "--value"])
return (out or "").strip() if rc == 0 else ""
def sysop_install_units(systemd_dir=None, user_dir=None, dry_run=False,
run=None, progress=None, anchor_timeout=90,
poll_interval=3):
"""Link fleet units, reload, enable timers, anchor + verify them.
Returns a result dict with per-unit reports and a failures list.
Idempotent: correct symlinks are kept, systemctl calls are re-runnable.
With dry_run=True nothing is changed and no command is executed.
progress, when given, is called with short status lines as work
happens (the run is otherwise silent for minutes behind slow
service starts). Anchors never block indefinitely: services are
started --no-block and NEXT is polled up to anchor_timeout; a
still-activating service is reported in-progress (it self-anchors
on completion) rather than failed.
"""
run = run or _sh
say = progress or (lambda line: None)
src_dir = Path(systemd_dir) if systemd_dir else SYSOP_SYSTEMD_DIR
dst_dir = Path(user_dir) if user_dir else sysop_user_units_dir()
units = sysop_unit_files(src_dir)
timers = [u for u in units if u.endswith(".timer")]
services = {u for u in units if u.endswith(".service")}
result = {"ok": True, "dry_run": dry_run, "units": [],
"daemon_reload": {"ok": True, "detail": ""},
"timers": [], "failures": []}
def fail(msg):
result["failures"].append(msg)
result["ok"] = False
if not timers:
fail("no *.timer units discovered in %s" % src_dir)
return result
# (1) Symlink every shipped unit into the user systemd dir.
if not dry_run:
try:
dst_dir.mkdir(parents=True, exist_ok=True)
except OSError as e:
fail("cannot create %s: %s" % (dst_dir, e))
return result
for name in units:
src = src_dir / name
dst = dst_dir / name
if dry_run:
result["units"].append(
{"unit": name, "status": "would-link",
"src": str(src), "dst": str(dst)})
continue
try:
if dst.is_symlink() and dst.resolve() == src.resolve():
result["units"].append(
{"unit": name, "status": "already-linked",
"dst": str(dst)})
continue
if dst.is_symlink() or dst.exists():
dst.unlink()
dst.symlink_to(src)
result["units"].append(
{"unit": name, "status": "linked", "dst": str(dst)})
except OSError as e:
result["units"].append(
{"unit": name, "status": "failed", "error": str(e)})
fail("link %s: %s" % (name, e))
if dry_run:
for timer in timers:
result["timers"].append(
{"timer": timer, "enabled": None, "anchored": None,
"anchor_skipped": None, "next": None,
"note": "would enable --now, anchor, and verify"})
return result
# (2) Reload the user manager.
say("daemon-reload ...")
rc, out = run(["systemctl", "--user", "daemon-reload"])
result["daemon_reload"] = {"ok": rc == 0, "detail": out}
if rc != 0:
fail("daemon-reload: %s" % (out or ("exit %d" % rc)))
# (3)-(5) Enable, anchor, and verify each timer.
for timer in timers:
entry = {"timer": timer, "enabled": False, "anchored": False,
"anchor_skipped": None, "next": None, "note": None}
say("enable --now %s ..." % timer)
rc, out = run(["systemctl", "--user", "enable", "--now", timer])
entry["enabled"] = rc == 0
if rc != 0:
fail("enable --now %s: %s" % (timer, out or ("exit %d" % rc)))
# (4) Anchor relative timers: start the matching oneshot
# service once so OnUnitActiveSec gains a reference timestamp
# (fresh-install-mid-boot otherwise leaves NEXT empty). The
# start is --no-block: slow first runs (backlog scrapes)
# would otherwise stall the whole install with no output.
service = timer[:-len(".timer")] + ".service"
if service not in services:
entry["anchor_skipped"] = "no matching service shipped"
elif not sysop_timer_needs_anchor(src_dir / timer):
entry["anchor_skipped"] = "calendar timer needs no anchor"
elif sysop_is_daemon(src_dir / service):
entry["anchor_skipped"] = "long-lived daemon, not started"
else:
state = _sysop_service_state(run, service)
if state == "activating":
say("anchor %s: already running, waiting for NEXT ..."
% service)
else:
say("anchor %s: starting ..." % service)
rc, out = run(["systemctl", "--user", "start",
"--no-block", service])
if rc != 0:
fail("start %s: %s"
% (service, out or ("exit %d" % rc)))
state = "start-failed"
# (5) Poll for a real NEXT elapse (bounded).
if state != "start-failed":
polled = _sysop_poll_next(
run, say, timer, service, anchor_timeout,
poll_interval)
msg = polled.pop("_fail", None)
if msg:
fail(msg)
entry.update(polled)
# Timers whose anchor was skipped still get one NEXT check.
if entry["anchor_skipped"] and entry["next"] is None:
nxt = _sysop_next(run, timer)
if nxt is not None:
entry["next"] = nxt
else:
fail("verify %s: no NEXT elapse" % timer)
result["timers"].append(entry)
return result
def _sysop_poll_next(run, say, timer, service, timeout, interval):
"""Poll until the timer shows NEXT, the service fails, or timeout.
Returns a partial timer entry (anchored/next/note). A service
still activating at timeout is reported in-progress rather than
failed: its running start job anchors the timer on completion.
"""
deadline = time.monotonic() + max(0, timeout)
while True:
nxt = _sysop_next(run, timer)
if nxt is not None:
return {"anchored": True, "next": nxt, "note": None}
state = _sysop_service_state(run, service)
if state == "failed":
return {"anchored": False, "next": None,
"note": None, "_fail":
"anchor %s: service failed "
"(journalctl --user -u %s)" % (service, service)}
if time.monotonic() >= deadline:
if state == "activating":
say("anchor %s: still running, timer self-anchors "
"on completion" % service)
return {"anchored": "in-progress", "next": None,
"note": "anchor running; verify NEXT shortly"}
return {"anchored": False, "next": None,
"note": None, "_fail":
"verify %s: no NEXT elapse "
"(service state: %s)" % (timer, state or "?")}
say("anchor %s: waiting for NEXT ..." % service)
time.sleep(max(0, interval))
def cmd_sysop_install(args):
dry_run = getattr(args, "dry_run", False)
as_json = getattr(args, "json", False)
# Live progress: stdout in text mode, stderr in --json mode so
# stdout stays pure machine-readable JSON.
progress = (lambda line: print(" ... %s" % line, flush=True,
file=sys.stderr if as_json else sys.stdout))
result = sysop_install_units(dry_run=dry_run, progress=progress)
if as_json:
print(json.dumps(result, indent=2))
sys.exit(0 if result["ok"] else 1)
print(c_bold("\n=== SYSOP INSTALL%s ===\n" % (
" (dry-run)" if dry_run else "")))
for u in result["units"]:
st = u["status"]
mark = (badge_ok("LINKED") if st == "linked"
else badge_dim("KEPT") if st == "already-linked"
else c_cyan("○ WOULD-LINK") if st == "would-link"
else badge_err("FAILED"))
print(" %s %s" % (mark, u["unit"]))
if not dry_run:
dr = result["daemon_reload"]
print(" %s daemon-reload" % (
badge_ok("OK") if dr["ok"] else badge_err("FAILED")))
print()
for t in result["timers"]:
en = t["enabled"]
emark = (badge_dim("?") if en is None
else badge_ok("ON") if en else badge_err("OFF"))
if t.get("anchor_skipped"):
amark = c_dim("- %s" % t["anchor_skipped"])
elif t.get("anchored") == "in-progress":
amark = c_cyan("○ ANCHORING")
elif t["anchored"]:
amark = badge_ok("ANCHORED")
elif t["anchored"] is None:
amark = badge_dim("?")
else:
amark = badge_err("ANCHOR-FAILED")
nxt = (t["next"] or c_dim(t["note"]) if t.get("note")
else t["next"] or (c_dim("pending (dry-run)")
if dry_run else badge_err("NO NEXT")))
print(" %s %-32s %s NEXT=%s" % (emark, t["timer"], amark,
nxt))
print()
if result["failures"]:
for f in result["failures"]:
print(" %s %s" % (c_red("✘"), f))
print()
sys.exit(1)
print(" %s fleet units installed and verified.\n" % c_green("✔"))
sys.exit(0)
def build_parser(): def build_parser():
common = argparse.ArgumentParser(add_help=False) common = argparse.ArgumentParser(add_help=False)
common.add_argument("--json", action="store_true", help="Output machine-readable JSON") common.add_argument("--json", action="store_true", help="Output machine-readable JSON")
@@ -5771,6 +6273,9 @@ def build_parser():
p_app_req_key.add_argument("node", choices=VALID_NODES, help="Target node requesting key") p_app_req_key.add_argument("node", choices=VALID_NODES, help="Target node requesting key")
p_app_req_key.add_argument("--reason", default="Passkey authentication required", help="Reason for key request") p_app_req_key.add_argument("--reason", default="Passkey authentication required", help="Reason for key request")
p_app_gates = app_sub.add_parser("gates", aliases=["gate"], parents=[common], help="Inspect coordinator decision record gates")
p_app_gates.add_argument("--scope", default=None, help="Check specific gate scope (e.g. role-layer, merges-to-main)")
# Domain: MUSE-CHOICES # Domain: MUSE-CHOICES
p_mc = subparsers.add_parser("muse-choices", parents=[common], help="Muse TUI A/B/C choice auto-answer daemon (on/off/status/logs)") p_mc = subparsers.add_parser("muse-choices", parents=[common], help="Muse TUI A/B/C choice auto-answer daemon (on/off/status/logs)")
mc_sub = p_mc.add_subparsers(dest="mc_action") mc_sub = p_mc.add_subparsers(dest="mc_action")
@@ -5795,7 +6300,7 @@ def build_parser():
p_mc_res.add_argument("decision", choices=["approve", "deny"], help="Release the hold to approve, or deny it (permission kinds only)") p_mc_res.add_argument("decision", choices=["approve", "deny"], help="Release the hold to approve, or deny it (permission kinds only)")
# Domain: RUNTIME # Domain: RUNTIME
p_rt = subparsers.add_parser("runtime", parents=[common], help="Muse CLI tmux runtimes: list states, send input, launch auto-approved") p_rt = subparsers.add_parser("runtime", parents=[common], help="Muse CLI tmux runtimes: list states, send input, launch with approval trail")
rt_sub = p_rt.add_subparsers(dest="rt_action") rt_sub = p_rt.add_subparsers(dest="rt_action")
p_rt_list = rt_sub.add_parser("list", parents=[common], help="List panes with runtime state + approval posture (default)") p_rt_list = rt_sub.add_parser("list", parents=[common], help="List panes with runtime state + approval posture (default)")
@@ -5808,8 +6313,13 @@ def build_parser():
p_rt_send.add_argument("keys", help="Keys / text to send") p_rt_send.add_argument("keys", help="Keys / text to send")
p_rt_send.add_argument("--no-enter", action="store_true", help="Do not send Enter after keys") p_rt_send.add_argument("--no-enter", action="store_true", help="Do not send Enter after keys")
p_rt_launch = rt_sub.add_parser("launch", parents=[common], help="Launch a Muse session with auto-approve injected") p_rt_open = rt_sub.add_parser("open", parents=[common], help="Attach to a session on a socket (execs tmux attach)")
p_rt_launch.add_argument("--socket", default=None, help="Tmux socket (default: /tmp/tmux-1000/default)") p_rt_open.add_argument("--socket", default=None, help="Tmux socket (default: /tmp/tmux-1000/default)")
p_rt_open.add_argument("--session", default=None, help="Session name (default: the only session)")
p_rt_open.add_argument("--dry-run", action="store_true", help="Print the attach plan without attaching")
p_rt_launch = rt_sub.add_parser("launch", parents=[common], help="Launch a Muse session with approval trail injected (on-request)")
p_rt_launch.add_argument("--socket", default=None, help="Tmux socket (default: /tmp/tmux-muse.sock for Box fleet)")
p_rt_launch.add_argument("--session", required=True, help="New tmux session name") p_rt_launch.add_argument("--session", required=True, help="New tmux session name")
p_rt_launch.add_argument("--window", "-w", default=None, help="Initial window name") p_rt_launch.add_argument("--window", "-w", default=None, help="Initial window name")
p_rt_launch.add_argument("--dry-run", action="store_true", help="Print the launch plan without creating") p_rt_launch.add_argument("--dry-run", action="store_true", help="Print the launch plan without creating")
@@ -5823,6 +6333,12 @@ def build_parser():
p_rt_spread.add_argument("--session", default=None, help="Only this tmux session") p_rt_spread.add_argument("--session", default=None, help="Only this tmux session")
p_rt_spread.add_argument("--dry-run", action="store_true", help="Print the spread plan without moving panes") p_rt_spread.add_argument("--dry-run", action="store_true", help="Print the spread plan without moving panes")
p_rt_rec = rt_sub.add_parser("reconcile", parents=[common], help="Enforce fleet/agents.json: relaunch missing, brief fresh panes, requeue stale claims")
p_rt_rec.add_argument("--manifest", default=None, help="Manifest path (default: fleet/agents.json)")
p_rt_rec.add_argument("--tasks", default=None, help="Task queue dir (default: alongside manifest)")
p_rt_rec.add_argument("--dry-run", action="store_true", help="Print the plan without changing anything")
p_rt_rec.add_argument("--adopt", action="store_true", help="Record live sessions as briefed without sending")
# Domain: INVITE # Domain: INVITE
p_invite = subparsers.add_parser("invite", parents=[common], help="Muse.ai invite codes: find per-agent codes and redeem") p_invite = subparsers.add_parser("invite", parents=[common], help="Muse.ai invite codes: find per-agent codes and redeem")
p_invite.add_argument("--node", choices=VALID_NODES, default=None, help="Filter by node (status)") p_invite.add_argument("--node", choices=VALID_NODES, default=None, help="Filter by node (status)")
@@ -6398,6 +6914,10 @@ def build_parser():
p_sub_spawn.add_argument("--wait", type=int, default=30, help="Seconds to wait for subagent response") p_sub_spawn.add_argument("--wait", type=int, default=30, help="Seconds to wait for subagent response")
# Domain: FLOW (Agentic workflows in persistent tmux panes with delta read-backs)
p_flow = subparsers.add_parser("flow", parents=[common], help="Manage agentic flows in persistent tmux panes (start, read, send, list, stop)")
p_flow.add_argument("flow_args", nargs=argparse.REMAINDER, help="Arguments passed directly to flow_engine.py")
# Domain: TMUX (Headless background tmux sessions with automatic logging) # Domain: TMUX (Headless background tmux sessions with automatic logging)
p_tmux = subparsers.add_parser("tmux", parents=[common], help="Manage headless background tmux sessions on /tmp/tmux-muse.sock") p_tmux = subparsers.add_parser("tmux", parents=[common], help="Manage headless background tmux sessions on /tmp/tmux-muse.sock")
p_tmux.add_argument("tmux_args", nargs=argparse.REMAINDER, help="Arguments passed directly to muse-tmux.py") p_tmux.add_argument("tmux_args", nargs=argparse.REMAINDER, help="Arguments passed directly to muse-tmux.py")
@@ -6423,6 +6943,12 @@ def build_parser():
p_tui = subparsers.add_parser("tui", parents=[common], help="Interactive full-screen Muse TUI & Box fleet console") p_tui = subparsers.add_parser("tui", parents=[common], help="Interactive full-screen Muse TUI & Box fleet console")
p_tui.add_argument("tui_args", nargs=argparse.REMAINDER, help="Arguments passed directly to muse-tui.py") p_tui.add_argument("tui_args", nargs=argparse.REMAINDER, help="Arguments passed directly to muse-tui.py")
# Domain: SYSOP (one-shot fleet installer: link units, enable timers, anchor + verify)
p_sysop = subparsers.add_parser("sysop", parents=[common], help="Fleet operations: one-shot systemd unit installer")
sysop_sub = p_sysop.add_subparsers(dest="sysop_action")
p_sysop_install = sysop_sub.add_parser("install", parents=[common], help="Link systemd units, enable timers, anchor and verify them")
p_sysop_install.add_argument("--dry-run", action="store_true", help="Print actions without changing anything")
return parser return parser
def main(): def main():
@@ -6431,6 +6957,8 @@ def main():
tui_args = sys.argv[2:] tui_args = sys.argv[2:]
if tui_args and tui_args[0] in ("onboard", "tmux", "connects", "approvals"): if tui_args and tui_args[0] in ("onboard", "tmux", "connects", "approvals"):
cmd = [sys.executable, str(BIN_DIR / "box-onboard-tui.py")] + tui_args[1:] cmd = [sys.executable, str(BIN_DIR / "box-onboard-tui.py")] + tui_args[1:]
elif tui_args and tui_args[0] in ("fleet", "oversight"):
cmd = [sys.executable, str(BIN_DIR / "box-fleet-tui.py")] + tui_args[1:]
else: else:
cmd = [sys.executable, str(BIN_DIR / "muse-tui.py"), "--mode", "box"] + tui_args cmd = [sys.executable, str(BIN_DIR / "muse-tui.py"), "--mode", "box"] + tui_args
res = subprocess.run(cmd) res = subprocess.run(cmd)
@@ -6439,6 +6967,10 @@ def main():
cmd = [sys.executable, str(BIN_DIR / "box-onboard-tui.py")] + sys.argv[2:] cmd = [sys.executable, str(BIN_DIR / "box-onboard-tui.py")] + sys.argv[2:]
res = subprocess.run(cmd) res = subprocess.run(cmd)
sys.exit(res.returncode) sys.exit(res.returncode)
elif sys.argv[1] in ("fleet-tui",):
cmd = [sys.executable, str(BIN_DIR / "box-fleet-tui.py")] + sys.argv[2:]
res = subprocess.run(cmd)
sys.exit(res.returncode)
elif sys.argv[1] == "tmux": elif sys.argv[1] == "tmux":
if len(sys.argv) > 2 and sys.argv[2] in ("tally", "auto", "watch", "once", "match", "rules", "status"): if len(sys.argv) > 2 and sys.argv[2] in ("tally", "auto", "watch", "once", "match", "rules", "status"):
if sys.argv[2] == "auto": if sys.argv[2] == "auto":
@@ -6459,6 +6991,21 @@ def main():
cmd = [sys.executable, str(BIN_DIR / "docs-lookup.py")] + sys.argv[2:] cmd = [sys.executable, str(BIN_DIR / "docs-lookup.py")] + sys.argv[2:]
res = subprocess.run(cmd) res = subprocess.run(cmd)
sys.exit(res.returncode) sys.exit(res.returncode)
elif sys.argv[1] in ("stability", "stable"):
cmd = [sys.executable, str(NETVM_ROOT / "watchers" / "box-stability-watcher.py")]
sub = sys.argv[2:]
if not sub or sub[0] == "status":
cmd.append("--status")
elif sub[0] == "check":
cmd.append("--check")
elif sub[0] == "json":
cmd.extend(["--status", "--json"])
elif sub[0] == "resume" and len(sub) > 1:
cmd.extend(["--resume", sub[1]])
else:
cmd.extend(sub)
res = subprocess.run(cmd)
sys.exit(res.returncode)
elif sys.argv[1] == "muse": elif sys.argv[1] == "muse":
m_args = sys.argv[2:] m_args = sys.argv[2:]
if not m_args: if not m_args:
@@ -6759,12 +7306,20 @@ def main():
cmd_lookup(args) cmd_lookup(args)
elif args.domain in ("passkey", "key"): elif args.domain in ("passkey", "key"):
cmd_passkey_info(args) cmd_passkey_info(args)
elif args.domain == "flow":
cmd_flow_dispatch(args)
elif args.domain == "tmux": elif args.domain == "tmux":
cmd_tmux_dispatch(args) cmd_tmux_dispatch(args)
elif args.domain == "muse": elif args.domain == "muse":
cmd_muse_dispatch(args) cmd_muse_dispatch(args)
elif args.domain in ("docs", "doc"): elif args.domain in ("docs", "doc"):
cmd_docs_dispatch(args) cmd_docs_dispatch(args)
elif args.domain == "sysop":
act = getattr(args, "sysop_action", None)
if act == "install":
cmd_sysop_install(args)
else:
parser.print_help()
else: else:
parser.print_help() parser.print_help()
+56
View File
@@ -0,0 +1,56 @@
# Box Stability Watcher & Host Load Mitigation
**Date:** 2026-10-07
**Scope:** Host `bl` (100.123.153.75), NetVM execution stability, and preventative runaway containment.
---
## 1. Incident Post-Mortem (2026-10-07)
### Symptoms
- Host `bl` stopped responding over SSH and Tailscale ("went dark") at ~16:44 UTC.
- Connections timed out during SSH banner exchange (`Connection timed out during banner exchange`).
- Tailscale direct connections dropped, failing back to DERP relay `nyc` before dropping entirely.
### Forensics & Root Cause
1. **Tmux Memory Leak & OOM Killer:**
- At 16:44:52 UTC, `systemd` triggered an OOM-kill on `tmux.service`:
```
tmux.service: Consumed 5h 52min 34s CPU time ... 22.3G memory peak, 1.3G memory swap peak.
tmux.service: Failed with result "oom-kill".
```
- Multiple `muse-bin` worker processes running inside background tmux windows had accumulated 22.3 GB of memory against the host 28 GB RAM and 4 GB swap.
2. **Avalanche Load Spike (Load Avg: 515.17):**
- The unconstrained crash triggered core dump collection (`systemd-coredump`) and simultaneous resurrection of multiple background sessions.
- Host 1-minute load average spiked to **515.17** (on a 16-core CPU), starving kernel network processing and dropping incoming SSH and Tailscale packets.
3. **Crash Loop Contributor:**
- Concurrently, `audio-patchbay.service` was stuck in a tight infinite failure loop (restarting >1,075,000 times) due to a headless GTK initialization panic, generating relentless fork/exit churn.
---
## 2. Hardening Measures Implemented
### A. Dedicated Watcher Directory (`Projects/NetVM/watchers/`)
Created a dedicated project folder inside `Projects/NetVM/` containing:
- `watchers/box-stability-watcher.py`: Core stability supervisor.
- `watchers/box-stability.json`: Operational configuration & thresholds.
- `watchers/README.md`: Architecture and usage guide.
- `bin/box-stability-watcher.py`: Symlink for operator CLI access.
### B. Proactive Mitigation Tiers
- **Tier GREEN (<20 load, <80% RAM):** Passive observation.
- **Tier YELLOW (20-35 load, 80-90% RAM):** Renices rogue CPU hogs to `nice +15` to protect interactive SSH and Tailscale responsiveness.
- **Tier ORANGE (35-60 load, >90% RAM, or process RSS >3000MB):** Proactively sends `SIGTERM` to individual leaky worker processes (`muse-bin`, headless renderers) before system memory is exhausted and kernel OOM kills `tmux.service`.
- **Tier RED (>60 load, >95% RAM, or >92% Swap):** Emergency shedder that terminates non-protected heavy consumers (>1500MB) to avert complete system lockup.
### C. Systemd Hardening & Service Cleanup
1. **Disabled Runaway Service:** Stopped and disabled `audio-patchbay.service`, instantly halting 1M+ iterations of process restart overhead.
2. **Cgroup Memory Limits on `tmux.service`:** Configured `MemoryHigh=18G` and `MemoryMax=22G` in `~/.config/systemd/user/tmux.service` so that a rogue subagent cannot consume 100% of the host RAM.
3. **Daemonized Watcher:** Enabled `box-stability-watcher.service` as a persistent user systemd service with a 256MB memory cap and automatic restart.
---
## 3. Verification
- Watcher unit test suite: `tests/test_box_stability_watcher.py` (10/10 tests passed).
- Live evaluation: `box-stability-watcher.py --status` reports `GREEN` with host load normalized to ~2.5.
+16
View File
@@ -0,0 +1,16 @@
[Unit]
Description=NetVM Box & Host Stability Watcher Daemon
After=network.target
[Service]
Type=simple
ExecStart=/usr/bin/python3 /home/super/Projects/NetVM/watchers/box-stability-watcher.py --daemon
Restart=always
RestartSec=5
MemoryMax=256M
CPUQuota=50%
StandardOutput=journal
StandardError=journal
[Install]
WantedBy=default.target
+169
View File
@@ -0,0 +1,169 @@
#!/usr/bin/env python3
"""test_box_stability_watcher.py — Comprehensive unit tests for Box Stability Watcher."""
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest import mock
REPO_ROOT = Path("/home/super/Projects/NetVM")
WATCHERS_DIR = REPO_ROOT / "watchers"
sys.path.insert(0, str(WATCHERS_DIR))
import importlib.util
spec = importlib.util.spec_from_file_location("box_stability_watcher", str(WATCHERS_DIR / "box-stability-watcher.py"))
w = importlib.util.module_from_spec(spec)
spec.loader.exec_module(w)
class TestConfigAndSafety(unittest.TestCase):
def test_load_config_defaults(self):
with tempfile.TemporaryDirectory() as td:
non_existent = Path(td) / "missing.json"
cfg = w.load_config(non_existent)
self.assertIn("thresholds", cfg)
self.assertIn("protected_commands", cfg)
self.assertEqual(cfg["thresholds"]["load_warning"], 20.0)
def test_is_protected(self):
cfg = {"protected_commands": ["sshd", "tailscaled", "tmux", "systemd", "ghostty"]}
self.assertTrue(w.is_protected(1, "systemd", "/sbin/init", cfg))
self.assertTrue(w.is_protected(os.getpid(), "python3", "some_script", cfg))
self.assertTrue(w.is_protected(999, "sshd", "/usr/sbin/sshd -D", cfg))
self.assertTrue(w.is_protected(888, "tmux", "tmux new-session -s main", cfg))
self.assertTrue(w.is_protected(777, "tailscaled", "/usr/sbin/tailscaled", cfg))
self.assertFalse(w.is_protected(1234, "muse-bin", "/home/super/.local/bin/muse-bin-1.4.3 resume abc", cfg))
self.assertFalse(w.is_protected(5678, "chromium", "/usr/lib/chromium/chromium --type=renderer", cfg))
class TestStabilityEvaluation(unittest.TestCase):
def setUp(self):
self.cfg = {
"thresholds": {
"load_warning": 20.0,
"load_critical": 35.0,
"load_emergency": 60.0,
"ram_warning_pct": 80.0,
"ram_critical_pct": 90.0,
"swap_warning_pct": 75.0,
"swap_critical_pct": 85.0,
"process_rss_warning_mb": 2000,
"process_rss_critical_mb": 3000,
},
"protected_commands": ["sshd", "tmux"]
}
def test_green_tier(self):
metrics = {"load_1m": 2.5, "ram_used_pct": 30.0, "swap_used_pct": 10.0}
procs = [
{"pid": 101, "cmdline": "muse-bin", "rss_mb": 400, "cpu_pct": 5.0, "is_protected": False, "nice": 0}
]
tier, reasons, actions = w.evaluate_stability(metrics, procs, self.cfg)
self.assertEqual(tier, "GREEN")
self.assertEqual(reasons, [])
self.assertEqual(actions, [])
def test_yellow_tier_elevated_load(self):
metrics = {"load_1m": 22.5, "ram_used_pct": 50.0, "swap_used_pct": 20.0}
procs = [
{"pid": 102, "cmdline": "python worker", "rss_mb": 500, "cpu_pct": 90.0, "is_protected": False, "nice": 0}
]
tier, reasons, actions = w.evaluate_stability(metrics, procs, self.cfg)
self.assertEqual(tier, "YELLOW")
self.assertTrue(any("Elevated load/memory" in r for r in reasons))
self.assertEqual(len(actions), 1)
self.assertEqual(actions[0]["action"], "renice")
self.assertEqual(actions[0]["pid"], 102)
def test_orange_tier_pause_leaky_process(self):
metrics = {"load_1m": 2.0, "ram_used_pct": 40.0, "swap_used_pct": 10.0}
procs = [
{"pid": 202, "cmdline": "muse-bin leaky", "rss_mb": 3400, "cpu_pct": 10.0, "is_protected": False, "nice": 0}
]
tier, reasons, actions = w.evaluate_stability(metrics, procs, self.cfg)
self.assertEqual(tier, "ORANGE")
self.assertTrue(any("exceeded critical RSS" in r for r in reasons))
self.assertEqual(len(actions), 1)
self.assertEqual(actions[0]["action"], "pause")
self.assertEqual(actions[0]["pid"], 202)
def test_red_emergency_tier(self):
metrics = {"load_1m": 75.0, "ram_used_pct": 96.0, "swap_used_pct": 94.0}
procs = [
{"pid": 301, "cmdline": "muse-bin heavy", "rss_mb": 1800, "cpu_pct": 50.0, "is_protected": False, "nice": 0},
{"pid": 302, "cmdline": "sshd daemon", "rss_mb": 2000, "cpu_pct": 2.0, "is_protected": True, "nice": 0}
]
tier, reasons, actions = w.evaluate_stability(metrics, procs, self.cfg)
self.assertEqual(tier, "RED")
self.assertTrue(any("Emergency host pressure" in r for r in reasons))
pause_pids = [a["pid"] for a in actions if a["action"] == "pause"]
self.assertIn(301, pause_pids)
self.assertNotIn(302, pause_pids)
class TestSocketIsolation(unittest.TestCase):
def test_distinguishes_user_from_box_launched_agents(self):
# PID 555 is an automated job on default socket
# PID 666 is an agent on the correct fleet socket
# PID 777 is a user-launched interactive agent on default socket
procs = [
{"pid": 555, "cmdline": "muse-bin auto-work sweep", "tmux_sock": "/tmp/tmux-1000/default", "is_protected": False},
{"pid": 666, "cmdline": "muse-bin auto-work sweep", "tmux_sock": "/tmp/tmux-muse.sock", "is_protected": False},
{"pid": 777, "cmdline": "muse-bin interactive chat", "tmux_sock": "/tmp/tmux-1000/default", "is_protected": False},
]
box_sessions = {"auto-work": {}}
violations, allowed = w.check_socket_isolation_violations(procs, box_sessions=box_sessions)
self.assertEqual(len(violations), 1)
self.assertEqual(violations[0]["pid"], 555)
self.assertEqual(len(allowed), 1)
self.assertEqual(allowed[0]["pid"], 777)
class TestPauseResumeAndExpiry(unittest.TestCase):
@mock.patch("os.kill")
def test_pause_and_state_persistence(self, mock_kill):
with tempfile.TemporaryDirectory() as td:
state_file = Path(td) / "paused.json"
actions = [{"action": "pause", "pid": 4321, "cmd": "muse-bin", "reason": "RSS high"}]
executed = w.execute_actions(actions, state_file=state_file, dry_run=False)
self.assertEqual(len(executed), 1)
mock_kill.assert_called_once_with(4321, 19) # SIGSTOP = 19
self.assertTrue(state_file.exists())
data = json.loads(state_file.read_text())
self.assertIn("4321", data)
@mock.patch("os.kill")
def test_reconcile_expired_pause(self, mock_kill):
with tempfile.TemporaryDirectory() as td:
state_file = Path(td) / "paused.json"
now = w.now_epoch()
state_data = {
"4321": {"pid": 4321, "cmd": "muse-bin", "paused_at_epoch": now - 70}
}
state_file.write_text(json.dumps(state_data))
expired = w.reconcile_paused_processes(state_file=state_file, dry_run=False)
self.assertEqual(len(expired), 1)
self.assertEqual(expired[0]["action"], "cull_expired")
self.assertEqual(expired[0]["pid"], 4321)
mock_kill.assert_called_once_with(4321, 15) # SIGTERM = 15
remaining = json.loads(state_file.read_text())
self.assertNotIn("4321", remaining)
@mock.patch("os.kill")
def test_resume_process(self, mock_kill):
with tempfile.TemporaryDirectory() as td:
state_file = Path(td) / "paused.json"
state_file.write_text(json.dumps({"4321": {"pid": 4321}}))
res = w.resume_process(4321, state_file=state_file)
self.assertTrue(res["success"])
mock_kill.assert_called_once_with(4321, 18) # SIGCONT = 18
remaining = json.loads(state_file.read_text())
self.assertNotIn("4321", remaining)
if __name__ == "__main__":
unittest.main()
+60
View File
@@ -0,0 +1,60 @@
# NetVM Watchers
Dedicated directory for background autonomous health, resource, and stability watchers on NetVM host `bl`.
## Components
- **`box-stability-watcher.py`**: Host resource supervisor and load shedder. Proactively monitors:
- System 1m/5m/15m load averages against core counts.
- Host RAM and Swap pressure percentages.
- Per-process memory leaks (critical RSS thresholds for `muse-bin`, headless Chromium renderers, Python workers).
- Rogue/leaked CPU hogs starving SSH/Tailscale.
- Failing systemd user services trapped in tight restart loops.
- Socket isolation violations (automated workers running on `/tmp/tmux-1000/default` instead of `/tmp/tmux-muse.sock`).
- **`box-stability.json`**: Tunable operational thresholds, notifications, and protected process whitelist.
- **`systemd/box-stability-watcher.service`**: Systemd user daemon running the watcher continuously with 10s evaluation ticks.
## Operational Tiers & Mitigations
| Tier | Status | Trigger Condition | Automated Action |
| :--- | :--- | :--- | :--- |
| **GREEN** | Normal | Load < 20, RAM < 80%, Swap < 75% | Silent monitoring. |
| **YELLOW** | Warning | Load >= 20, RAM >= 80%, or process RSS >= 2000MB | Renice CPU hogs (+15) to preserve interactive SSH responsiveness; log warning. |
| **ORANGE** | Critical | Load >= 35, RAM >= 90%, or process RSS >= 3000MB | **Pause (SIGSTOP)** runaway worker, record in `.state/stability-paused.json`, post alert to `646 tasks` sidechat, and allow 60s operator inspection before SIGTERM. |
| **RED** | Emergency | Load >= 60, RAM >= 95%, or Swap >= 92% | Emergency load shedding of non-protected heavy consumers (>1500MB). |
## Paused Process Lifecycle (60s Grace Window)
When a process is paused:
1. Sent `SIGSTOP` immediately.
2. Recorded in `.state/stability-paused.json` with timestamp and command info.
3. Alert posted to `646 tasks` sidechat.
4. An operator can inspect the runtime or resume it via:
```bash
box stability resume <PID>
```
5. If unresumed after 60 seconds, the watcher automatically culls the process via `SIGTERM`.
## Protected Whitelist
The watcher will **never** terminate or renice:
`sshd`, `tailscaled`, `tailscale`, `systemd`, `dbus-broker`, `pipewire`, `wireplumber`, `tmux` (main server), `bash`, `zsh`, `ghostty`, `alacritty`.
## Unified Box CLI Integration
```bash
# Host stability status & active socket warnings
box stability status
# Machine-readable JSON output
box stability json
# Single evaluation check
box stability check [--dry-run]
# Resume a paused process
box stability resume <PID>
# Top-line host health indicator
box fleet status
```
+693
View File
@@ -0,0 +1,693 @@
#!/usr/bin/env python3
"""box-stability-watcher.py — Proactive host and fleet stability guardrail for NetVM & Box.
Features:
1. System Load & Memory Monitoring (1m/5m/15m load, RAM%, Swap%)
2. Multi-tier Mitigation:
- GREEN: Normal.
- YELLOW: Renice runaway CPU hogs to +15.
- ORANGE: Pause runaway workers via SIGSTOP, record in .state/stability-paused.json,
post alert to sidechat (646 tasks), and give 60s grace period before SIGTERM.
- RED: Emergency shedder for processes >1500MB.
3. Tmux Socket Origin & Isolation:
- Allows user-launched agents on interactive desktop socket (/tmp/tmux-1000/default).
- Detects and logs automated workloads launched through Box on the desktop socket,
recommending migration to /tmp/tmux-muse.sock.
4. Systemd Loop Detection: Identifies crashing services trapped in tight restart loops.
5. State Management: Supports explicit "freeze" and "resume" commands.
"""
import argparse
import json
import os
import signal
import subprocess
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
try:
import psutil
except ImportError:
psutil = None
WATCHERS_DIR = Path(__file__).resolve().parent
REPO_ROOT = WATCHERS_DIR.parent
DEFAULT_CONFIG = WATCHERS_DIR / "box-stability.json"
DEFAULT_LOG = REPO_ROOT / "logs" / "box-stability.jsonl"
PAUSED_STATE_FILE = REPO_ROOT / ".state" / "stability-paused.json"
BOX_LAUNCHED_SESSIONS_FILE = REPO_ROOT / ".state" / "box-launched-sessions.json"
PAUSE_GRACE_SECONDS = 60
def now_iso() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def now_epoch() -> float:
return time.time()
def load_config(config_path: Optional[Path] = None) -> Dict[str, Any]:
path = config_path or DEFAULT_CONFIG
if path.exists():
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return {
"thresholds": {
"load_warning": 20.0,
"load_critical": 35.0,
"load_emergency": 60.0,
"ram_warning_pct": 80.0,
"ram_critical_pct": 90.0,
"swap_warning_pct": 75.0,
"swap_critical_pct": 85.0,
"process_rss_warning_mb": 2000,
"process_rss_critical_mb": 3000,
},
"protected_commands": [
"sshd", "tailscaled", "tailscale", "systemd", "dbus-broker",
"pipewire", "wireplumber", "tmux", "bash", "zsh", "ghostty", "alacritty"
],
"monitored_targets": [
"muse-bin", "chromium", "python3", "parec"
],
"actions": {
"enable_renice": True,
"enable_cull_runaway": True,
"enable_service_freeze": True,
"notify_sidechat": True,
},
"notification": {
"agent": "646",
"target": "646 tasks",
},
"interval_sec": 10,
"log_file": "logs/box-stability.jsonl",
}
def rotate_log_if_needed(log_path: Path, max_bytes: int = 10485760):
try:
if log_path.exists() and log_path.stat().st_size > max_bytes:
rotated = log_path.with_name(f"{log_path.name}.1")
os.replace(log_path, rotated)
except Exception:
pass
def append_stability_event(event: Dict[str, Any], log_path: Optional[Path] = None):
target = log_path or DEFAULT_LOG
try:
target.parent.mkdir(parents=True, exist_ok=True)
rotate_log_if_needed(target)
with open(target, "a", encoding="utf-8") as f:
f.write(json.dumps(event) + "\n")
except Exception:
pass
def read_paused_state(state_file: Optional[Path] = None) -> Dict[str, Any]:
target = state_file or PAUSED_STATE_FILE
if target.exists():
try:
with open(target, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return {}
def write_paused_state(state: Dict[str, Any], state_file: Optional[Path] = None):
target = state_file or PAUSED_STATE_FILE
try:
target.parent.mkdir(parents=True, exist_ok=True)
tmp = target.with_suffix(".tmp")
with open(tmp, "w", encoding="utf-8") as f:
json.dump(state, f, indent=2)
os.replace(tmp, target)
except Exception:
pass
def get_box_launched_sessions(sessions_file: Optional[Path] = None) -> Dict[str, Any]:
target = sessions_file or BOX_LAUNCHED_SESSIONS_FILE
if target.exists():
try:
with open(target, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return {}
def send_sidechat_alert(message: str, config: Dict[str, Any], dry_run: bool = False):
if dry_run or not config.get("actions", {}).get("notify_sidechat", True):
return
notif = config.get("notification", {})
agent = notif.get("agent", "646")
target = notif.get("target", "646 tasks")
box_cli = REPO_ROOT / "bin" / "super-cli.py"
if box_cli.exists():
cmd = [
sys.executable, str(box_cli), "dm", "send",
"--agent", agent,
"--to", agent,
"--target", target,
f"[STABILITY ALERT] {message}"
]
try:
subprocess.run(cmd, capture_output=True, timeout=10)
except Exception:
pass
def get_system_metrics() -> Dict[str, Any]:
load1, load5, load15 = os.getloadavg()
cpu_count = os.cpu_count() or 1
ram_total_mb = 0
ram_avail_mb = 0
ram_used_pct = 0.0
swap_total_mb = 0
swap_used_mb = 0
swap_used_pct = 0.0
if psutil:
vm = psutil.virtual_memory()
ram_total_mb = int(vm.total / (1024 * 1024))
ram_avail_mb = int(vm.available / (1024 * 1024))
ram_used_pct = round(vm.percent, 1)
sm = psutil.swap_memory()
swap_total_mb = int(sm.total / (1024 * 1024))
swap_used_mb = int(sm.used / (1024 * 1024))
swap_used_pct = round(sm.percent, 1)
else:
try:
meminfo = {}
with open("/proc/meminfo", "r") as f:
for line in f:
parts = line.split(":")
if len(parts) == 2:
key = parts[0].strip()
val = parts[1].strip().split()[0]
meminfo[key] = int(val)
total_kb = meminfo.get("MemTotal", 1)
avail_kb = meminfo.get("MemAvailable", meminfo.get("MemFree", 0))
ram_total_mb = total_kb // 1024
ram_avail_mb = avail_kb // 1024
ram_used_pct = round((1.0 - (avail_kb / total_kb)) * 100, 1)
swap_tot_kb = meminfo.get("SwapTotal", 0)
swap_free_kb = meminfo.get("SwapFree", 0)
if swap_tot_kb > 0:
swap_total_mb = swap_tot_kb // 1024
swap_used_mb = (swap_tot_kb - swap_free_kb) // 1024
swap_used_pct = round(((swap_tot_kb - swap_free_kb) / swap_tot_kb) * 100, 1)
except Exception:
pass
return {
"load_1m": round(load1, 2),
"load_5m": round(load5, 2),
"load_15m": round(load15, 2),
"cpu_count": cpu_count,
"ram_total_mb": ram_total_mb,
"ram_avail_mb": ram_avail_mb,
"ram_used_pct": ram_used_pct,
"swap_total_mb": swap_total_mb,
"swap_used_mb": swap_used_mb,
"swap_used_pct": swap_used_pct,
}
def is_protected(pid: int, name: str, cmdline: str, config: Dict[str, Any]) -> bool:
if pid in (os.getpid(), os.getppid(), 1):
return True
low_name = name.lower()
low_cmd = cmdline.lower()
for prot in config.get("protected_commands", []):
p_low = prot.lower()
if p_low == low_name or p_low in low_cmd.split():
return True
if "tmux new-session" in low_cmd or (low_name == "tmux" and "main" in low_cmd):
return True
return False
def get_tmux_socket_for_pid(pid: int) -> str:
try:
with open(f"/proc/{pid}/environ", "rb") as f:
env_data = f.read().split(b"\0")
for item in env_data:
if item.startswith(b"TMUX="):
val = item.decode("utf-8", errors="ignore")
parts = val.split(",")
if parts:
return parts[0].replace("TMUX=", "")
except Exception:
pass
return ""
def inspect_processes(config: Dict[str, Any]) -> List[Dict[str, Any]]:
results = []
if not psutil:
return results
targets = [t.lower() for t in config.get("monitored_targets", [])]
for p in psutil.process_iter(["pid", "name", "cmdline", "cpu_percent", "memory_info", "nice"]):
try:
info = p.info
pid = info["pid"]
name = info["name"] or ""
cmdline_list = info["cmdline"] or []
cmdline = " ".join(cmdline_list)
mem_info = info["memory_info"]
rss_mb = int(mem_info.rss / (1024 * 1024)) if mem_info else 0
cpu_pct = info["cpu_percent"] or 0.0
nice = info["nice"] or 0
low_cmd = cmdline.lower()
low_name = name.lower()
matches_target = any(t in low_name or t in low_cmd for t in targets)
tmux_sock = get_tmux_socket_for_pid(pid) if matches_target else ""
results.append({
"pid": pid,
"name": name,
"cmdline": cmdline[:200],
"rss_mb": rss_mb,
"cpu_pct": cpu_pct,
"nice": nice,
"matches_target": matches_target,
"tmux_sock": tmux_sock,
"is_protected": is_protected(pid, name, cmdline, config),
})
except (psutil.NoSuchProcess, psutil.AccessDenied):
continue
return results
def check_socket_isolation_violations(processes: List[Dict[str, Any]], box_sessions: Optional[Dict[str, Any]] = None) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
"""Distinguish user-launched agents (allowed) from Box/agent-launched workloads on the desktop socket.
Returns (violations, user_allowed).
"""
violations = []
user_allowed = []
tracked_box = box_sessions if box_sessions is not None else get_box_launched_sessions()
# Read subagent sessions as well
subagent_file = REPO_ROOT / "subagent-sessions.json"
subagent_ids = set()
if subagent_file.exists():
try:
with open(subagent_file, "r") as f:
data = json.load(f)
if isinstance(data, dict):
subagent_ids = set(data.keys())
except Exception:
pass
for p in processes:
if p.get("is_protected", False):
continue
sock = p.get("tmux_sock", "")
cmd = p.get("cmdline", "").lower()
if "default" in sock and ("muse-bin" in cmd or "auto-work" in cmd):
# Check if this workload originated from Box / automated scheduling
is_box_spawned = False
origin_reason = ""
# 1. Matches an automated session explicitly launched by Box CLI
for s_name in tracked_box:
if s_name.lower() in cmd:
is_box_spawned = True
origin_reason = f"tracked Box session '{s_name}'"
break
# 2. Matches subagent tracker UUID
if not is_box_spawned:
for sub_id in subagent_ids:
if sub_id.lower() in cmd:
is_box_spawned = True
origin_reason = f"tracked subagent '{sub_id[:8]}'"
break
# 3. Matches automated batch / flow naming conventions
if not is_box_spawned:
for pattern in ["auto-work", "flow-", "muse--runtime--"]:
if pattern in cmd:
is_box_spawned = True
origin_reason = f"automated job pattern '{pattern}'"
break
if is_box_spawned:
violations.append({
"pid": p["pid"],
"cmd": p["cmdline"][:60],
"socket": sock,
"origin": origin_reason,
"issue": f"Automated worker ({origin_reason}) running on desktop socket (/tmp/tmux-1000/default) instead of /tmp/tmux-muse.sock"
})
else:
# User-launched agent in interactive session
user_allowed.append({
"pid": p["pid"],
"cmd": p["cmdline"][:60],
"socket": sock,
"status": "user-launched (allowed in desktop socket)"
})
return violations, user_allowed
def evaluate_stability(metrics: Dict[str, Any], processes: List[Dict[str, Any]], config: Dict[str, Any]) -> Tuple[str, List[str], List[Dict[str, Any]]]:
th = config.get("thresholds", {})
tier = "GREEN"
reasons = []
actions_planned = []
load1 = metrics.get("load_1m", 0.0)
ram_pct = metrics.get("ram_used_pct", 0.0)
swap_pct = metrics.get("swap_used_pct", 0.0)
if load1 >= th.get("load_emergency", 60.0) or ram_pct >= 95.0 or swap_pct >= 92.0:
tier = "RED"
reasons.append(f"Emergency host pressure: load={load1}, RAM={ram_pct}%, Swap={swap_pct}%")
elif load1 >= th.get("load_critical", 35.0) or ram_pct >= th.get("ram_critical_pct", 90.0) or swap_pct >= th.get("swap_critical_pct", 85.0):
tier = "ORANGE"
reasons.append(f"Critical load/memory: load={load1}, RAM={ram_pct}%, Swap={swap_pct}%")
elif load1 >= th.get("load_warning", 20.0) or ram_pct >= th.get("ram_warning_pct", 80.0) or swap_pct >= th.get("swap_warning_pct", 75.0):
tier = "YELLOW"
reasons.append(f"Elevated load/memory: load={load1}, RAM={ram_pct}%, Swap={swap_pct}%")
rss_crit_mb = th.get("process_rss_critical_mb", 3000)
rss_warn_mb = th.get("process_rss_warning_mb", 2000)
for p in processes:
if p.get("is_protected", False):
continue
pid = p["pid"]
rss_mb = p.get("rss_mb", 0)
cpu_pct = p.get("cpu_pct", 0.0)
cmd_short = p.get("cmdline", "")[:60]
if rss_mb >= rss_crit_mb:
if tier in ("GREEN", "YELLOW"):
tier = "ORANGE"
reasons.append(f"Process PID {pid} exceeded critical RSS {rss_mb}MB >= {rss_crit_mb}MB: {cmd_short}")
actions_planned.append({
"action": "pause",
"pid": pid,
"reason": f"RSS {rss_mb}MB >= {rss_crit_mb}MB",
"cmd": cmd_short,
})
elif rss_mb >= rss_warn_mb:
if tier == "GREEN":
tier = "YELLOW"
reasons.append(f"Process PID {pid} elevated RSS {rss_mb}MB >= {rss_warn_mb}MB: {cmd_short}")
if tier in ("YELLOW", "ORANGE", "RED") and cpu_pct > 80.0 and p.get("nice", 0) < 10:
actions_planned.append({
"action": "renice",
"pid": pid,
"reason": f"CPU hog {cpu_pct}% under elevated load",
"nice_value": 15,
"cmd": cmd_short,
})
if tier == "RED":
for p in processes:
if not p.get("is_protected", False) and p.get("rss_mb", 0) > 1500:
actions_planned.append({
"action": "pause",
"pid": p["pid"],
"reason": f"Emergency RED shedder: RSS {p['rss_mb']}MB",
"cmd": p.get("cmdline", "")[:60],
})
return tier, reasons, actions_planned
def reconcile_paused_processes(state_file: Optional[Path] = None, dry_run: bool = False) -> List[Dict[str, Any]]:
actions = []
state = read_paused_state(state_file)
if not state:
return actions
now = now_epoch()
new_state = {}
for s_pid, info in state.items():
try:
pid = int(s_pid)
except ValueError:
continue
paused_at = info.get("paused_at_epoch", now)
elapsed = now - paused_at
cmd = info.get("cmd", "")
if elapsed >= PAUSE_GRACE_SECONDS:
entry = {
"action": "cull_expired",
"pid": pid,
"cmd": cmd,
"reason": f"Grace period of {PAUSE_GRACE_SECONDS}s expired without resume",
"dry_run": dry_run,
"success": False,
}
if not dry_run:
try:
os.kill(pid, signal.SIGTERM)
entry["status"] = "SIGTERM sent"
entry["success"] = True
except ProcessLookupError:
entry["status"] = "process already dead"
entry["success"] = True
except Exception as e:
entry["status"] = f"error: {e}"
else:
entry["status"] = "skipped (dry-run)"
actions.append(entry)
else:
new_state[s_pid] = info
if not dry_run:
write_paused_state(new_state, state_file)
return actions
def execute_actions(actions: List[Dict[str, Any]], state_file: Optional[Path] = None, dry_run: bool = False) -> List[Dict[str, Any]]:
executed = []
paused_state = read_paused_state(state_file) if not dry_run else {}
for act in actions:
kind = act.get("action")
pid = act.get("pid")
entry = dict(act)
entry["dry_run"] = dry_run
entry["success"] = False
if dry_run:
entry["status"] = "skipped (dry-run)"
executed.append(entry)
continue
try:
if kind == "renice":
nice_val = act.get("nice_value", 15)
os.setpriority(os.PRIO_PROCESS, pid, nice_val)
entry["status"] = f"reniced to {nice_val}"
entry["success"] = True
elif kind == "pause":
os.kill(pid, signal.SIGSTOP)
entry["status"] = "SIGSTOP sent (paused for 60s inspection)"
entry["success"] = True
paused_state[str(pid)] = {
"pid": pid,
"cmd": act.get("cmd", ""),
"reason": act.get("reason", ""),
"paused_at": now_iso(),
"paused_at_epoch": now_epoch(),
}
elif kind == "cull":
os.kill(pid, signal.SIGTERM)
entry["status"] = "SIGTERM sent"
entry["success"] = True
except ProcessLookupError:
entry["status"] = "process already gone"
entry["success"] = True
except PermissionError:
entry["status"] = "permission denied"
except Exception as e:
entry["status"] = f"error: {e}"
executed.append(entry)
if not dry_run and paused_state:
write_paused_state(paused_state, state_file)
return executed
def resume_process(pid: int, state_file: Optional[Path] = None) -> Dict[str, Any]:
state = read_paused_state(state_file)
res = {"pid": pid, "action": "resume", "success": False}
try:
os.kill(pid, signal.SIGCONT)
res["success"] = True
res["status"] = "SIGCONT sent (resumed)"
except ProcessLookupError:
res["status"] = "process does not exist"
except Exception as e:
res["status"] = f"error: {e}"
if str(pid) in state:
del state[str(pid)]
write_paused_state(state, state_file)
return res
def run_cycle(config: Dict[str, Any], dry_run: bool = False) -> Dict[str, Any]:
metrics = get_system_metrics()
processes = inspect_processes(config)
socket_violations, user_allowed = check_socket_isolation_violations(processes)
tier, reasons, actions_planned = evaluate_stability(metrics, processes, config)
actions_taken = execute_actions(actions_planned, dry_run=dry_run)
expired_actions = reconcile_paused_processes(dry_run=dry_run)
actions_taken.extend(expired_actions)
if socket_violations:
for sv in socket_violations:
reasons.append(f"Automated workload on desktop socket: PID {sv['pid']} ({sv.get('origin', 'box')})")
event = {
"timestamp": now_iso(),
"tier": tier,
"metrics": metrics,
"reasons": reasons,
"socket_violations": socket_violations,
"user_allowed": user_allowed,
"actions_taken": actions_taken,
"dry_run": dry_run,
}
if tier in ("ORANGE", "RED") or any(a.get("action") == "pause" for a in actions_taken):
alert_msg = f"Tier: {tier}. Reasons: {reasons}. Actions: {actions_taken}"
send_sidechat_alert(alert_msg, config, dry_run=dry_run)
if tier != "GREEN" or actions_taken or socket_violations:
log_path = Path(config.get("log_file", DEFAULT_LOG))
if not log_path.is_absolute():
log_path = REPO_ROOT / log_path
append_stability_event(event, log_path)
return event
def print_status(event: Dict[str, Any]):
m = event["metrics"]
tier = event["tier"]
color_code = {
"GREEN": "\033[92m● GREEN\033[0m",
"YELLOW": "\033[93m▲ YELLOW\033[0m",
"ORANGE": "\033[91m■ ORANGE\033[0m",
"RED": "\033[1;41m✖ RED (EMERGENCY)\033[0m",
}.get(tier, tier)
print(f"\n=== BOX STABILITY STATUS: {color_code} ===")
print(f" Load Average: {m['load_1m']} (1m) | {m['load_5m']} (5m) | {m['load_15m']} (15m) [Cores: {m['cpu_count']}]")
print(f" RAM Usage: {m['ram_used_pct']}% ({m['ram_total_mb'] - m['ram_avail_mb']}MB used / {m['ram_total_mb']}MB total)")
print(f" Swap Usage: {m['swap_used_pct']}% ({m['swap_used_mb']}MB used / {m['swap_total_mb']}MB total)")
paused = read_paused_state()
if paused:
print("\n Paused Processes (60s Grace Window):")
for s_pid, info in paused.items():
print(f" - PID {s_pid}: {info.get('cmd')} (paused at {info.get('paused_at')})")
if event.get("socket_violations"):
print("\n Automated Workload Warnings (Detected on Desktop Socket):")
for v in event["socket_violations"]:
print(f" - PID {v['pid']} ({v.get('origin', 'box')}): {v['cmd']}")
if event.get("user_allowed"):
print(f"\n User-Launched Agents on Desktop Socket: {len(event['user_allowed'])} active (allowed)")
if event.get("reasons"):
print("\n Active Issues:")
for r in event["reasons"]:
print(f" - {r}")
if event.get("actions_taken"):
print("\n Mitigations:")
for a in event["actions_taken"]:
print(f" - [{a['action']}] PID {a['pid']} ({a['cmd']}): {a.get('status')}")
print("")
def main():
parser = argparse.ArgumentParser(description="Box & Host Stability Watcher")
parser.add_argument("--config", type=Path, default=None, help="Path to box-stability.json")
parser.add_argument("--dry-run", action="store_true", help="Evaluate without executing kills or renices")
parser.add_argument("--check", "--once", dest="once", action="store_true", help="Run a single evaluation cycle and exit")
parser.add_argument("--status", action="store_true", help="Print human-readable status overview")
parser.add_argument("--json", action="store_true", help="Output JSON result")
parser.add_argument("--resume", type=int, help="Resume a paused PID with SIGCONT and remove from pause state")
parser.add_argument("--daemon", action="store_true", help="Run continuously in background daemon loop")
args = parser.parse_args()
config = load_config(args.config)
if args.resume:
res = resume_process(args.resume)
print(json.dumps(res, indent=2))
return 0 if res["success"] else 1
if args.status or args.once:
event = run_cycle(config, dry_run=args.dry_run or args.status)
if args.json:
print(json.dumps(event, indent=2))
else:
print_status(event)
return 0
if args.daemon:
interval = config.get("interval_sec", 10)
print(f"[{now_iso()}] Starting Box Stability Watcher daemon (interval: {interval}s)...")
while True:
try:
run_cycle(config, dry_run=args.dry_run)
time.sleep(interval)
except KeyboardInterrupt:
print(f"[{now_iso()}] Watcher stopped by user.")
break
except Exception as e:
print(f"[{now_iso()}] Watcher cycle error: {e}", file=sys.stderr)
time.sleep(interval)
return 0
event = run_cycle(config, dry_run=args.dry_run)
if args.json:
print(json.dumps(event, indent=2))
else:
print_status(event)
return 0
if __name__ == "__main__":
sys.exit(main())
+44
View File
@@ -0,0 +1,44 @@
{
"thresholds": {
"load_warning": 20.0,
"load_critical": 35.0,
"load_emergency": 60.0,
"ram_warning_pct": 80.0,
"ram_critical_pct": 90.0,
"swap_warning_pct": 75.0,
"swap_critical_pct": 85.0,
"process_rss_warning_mb": 2000,
"process_rss_critical_mb": 3000
},
"protected_commands": [
"sshd",
"tailscaled",
"tailscale",
"systemd",
"dbus-broker",
"pipewire",
"wireplumber",
"tmux",
"bash",
"zsh",
"ghostty",
"alacritty"
],
"monitored_targets": [
"muse-bin",
"chromium",
"python3",
"parec"
],
"actions": {
"enable_renice": true,
"enable_cull_runaway": true,
"enable_service_freeze": true
},
"service_guardrails": {
"max_restarts_per_window": 10,
"window_seconds": 60
},
"interval_sec": 10,
"log_file": "logs/box-stability.jsonl"
}