#!/usr/bin/env python3 """test_box_runtime.py — Runtime state sensing + `box runtime` management. Covers: runtime_state classification (approval-pending/working/open-prompt), muse argv approval-posture parsing, runtime_rows assembly (mocked tmux), and the `box runtime` CLI surface. """ import json import sys import unittest from pathlib import Path from unittest import mock REPO_ROOT = Path("/home/super/Projects/NetVM") BIN_DIR = REPO_ROOT / "bin" sys.path.insert(0, str(BIN_DIR)) import muse_choice_watcher as w PROMPT = "❯" # Muse TUI input glyph (U+276F) STATE_OPEN = ( "Some completed agent output here.\n" "\n" "───────────────────────────────────\n" + PROMPT + "\n" "───────────────────────────────────\n" " muse-spark-1.3-con… · Auto-review\n" ) STATE_WORKING = ( "Partial agent output...\n" "\n" "… — running (10s · esc to interrupt)\n" "\n" "───────────────────────────────────\n" + PROMPT + "\n" "───────────────────────────────────\n" " muse-spark-1.3-con… · Auto-review\n" ) STATE_WORKING_CUT = ( "Partial agent output...\n" "◆ Calling tools (5m 53s · esc to in\n" "\n" "───────────────────────────────────\n" + PROMPT + "\n" ) STATE_APPROVAL = ( "───────────────────────────────────\n" "Would you like to run the following\n" "\n" " $ tmux capture-pane -p\n" "\n" "› 1. Yes, proceed (y)\n" " 2. No, and tell Muse Code what to do instead\n" ) STATE_SHELL = "[super@bl NetVM]$ printf 'hi'\nhi\n[super@bl NetVM]$ " class TestRuntimeState(unittest.TestCase): def test_open_prompt(self): st = w.runtime_state(STATE_OPEN) self.assertEqual(st["state"], "open-prompt") self.assertIsNone(st["match"]) def test_working(self): st = w.runtime_state(STATE_WORKING) self.assertEqual(st["state"], "working") def test_working_edge_cut_indicator(self): st = w.runtime_state(STATE_WORKING_CUT) self.assertEqual(st["state"], "working") def test_approval_pending(self): st = w.runtime_state(STATE_APPROVAL) self.assertEqual(st["state"], "approval-pending") self.assertEqual(st["match"]["kind"], "muse-approval") self.assertEqual(st["match"]["key"], "1") def test_approval_beats_working(self): st = w.runtime_state(STATE_WORKING + STATE_APPROVAL) self.assertEqual(st["state"], "approval-pending") def test_working_beats_open_prompt(self): # A working pane still renders its prompt footer. st = w.runtime_state(STATE_WORKING) self.assertEqual(st["state"], "working") def test_shell_is_unknown(self): self.assertEqual(w.runtime_state(STATE_SHELL)["state"], "unknown") def test_empty_is_unknown(self): self.assertEqual(w.runtime_state("")["state"], "unknown") self.assertEqual(w.runtime_state(None)["state"], "unknown") class TestApprovalFlags(unittest.TestCase): def test_bare_is_not_auto(self): p = w.muse_approval_flags(["/home/super/.local/bin/muse-bin-1.4.3"]) self.assertFalse(p["auto_approve"]) self.assertEqual(p["flags"], []) def test_yolo(self): p = w.muse_approval_flags(["muse", "--yolo"]) self.assertTrue(p["auto_approve"]) self.assertIn("yolo", p["flags"]) def test_disable_approval(self): p = w.muse_approval_flags(["muse", "--disable-approval"]) self.assertTrue(p["auto_approve"]) def test_approval_mode_never(self): p = w.muse_approval_flags(["muse", "--approval-mode", "never"]) self.assertTrue(p["auto_approve"]) self.assertIn("approval-mode=never", p["flags"]) def test_approval_mode_equals(self): p = w.muse_approval_flags(["muse", "--approval-mode=never"]) self.assertTrue(p["auto_approve"]) def test_approval_mode_on_request_is_not_auto(self): p = w.muse_approval_flags(["muse", "--approval-mode", "on-request"]) self.assertFalse(p["auto_approve"]) self.assertIn("approval-mode=on-request", p["flags"]) def test_empty_argv(self): p = w.muse_approval_flags([]) self.assertFalse(p["auto_approve"]) def _tmux_result(returncode=0, stdout="", stderr=""): r = mock.Mock() r.returncode = returncode r.stdout = stdout r.stderr = stderr return r class TestRuntimeRows(unittest.TestCase): LISTING = ("muse\t1\t%37\tmuse-bin-1.4.3-R5018.1\t2880158\t71\t27\n" "muse\t1\t%38\tbash\t2880200\t100\t30\n") def _patched(self, tmux_stdout=LISTING, tmux_rc=0, captures=None, children=None, cmdlines=None, watcher=None): captures = captures or {} cmdlines = cmdlines or {} children = children or {} return (mock.patch.object(w, "_tmux", return_value=_tmux_result( tmux_rc, tmux_stdout)), mock.patch.object(w, "capture_pane", side_effect=lambda s, p: captures.get(p)), mock.patch.object(w, "_child_pids", side_effect=lambda p: children.get(p, [])), mock.patch.object(w, "_cmdline", side_effect=lambda p: cmdlines.get(p, [])), mock.patch.object(w, "is_running", return_value=watcher)) def test_rows_shape(self): patches = self._patched( captures={"%37": STATE_OPEN, "%38": STATE_SHELL}, children={2880158: [2881158]}, cmdlines={2881158: ["/home/super/.local/bin/muse-bin-1.4.3", "--disable-approval"]}, watcher=1234) with patches[0], patches[1], patches[2], patches[3], patches[4]: rows = w.runtime_rows("/tmp/sock") self.assertEqual(len(rows), 2) muse = rows[0] self.assertEqual(muse["pane"], "%37") self.assertTrue(muse["is_muse"]) self.assertTrue(muse["auto_approve"]) self.assertEqual(muse["approval_flags"], ["disable-approval"]) self.assertEqual(muse["state"], "open-prompt") self.assertTrue(muse["watcher_alive"]) self.assertEqual(muse["watcher_pid"], 1234) self.assertEqual(muse["width"], 71) self.assertEqual(muse["height"], 27) self.assertFalse(muse["squeezed"]) shell = rows[1] self.assertFalse(shell["is_muse"]) self.assertIsNone(shell["auto_approve"]) self.assertEqual(shell["state"], "unknown") def test_bare_muse_reports_not_auto(self): patches = self._patched( captures={"%37": STATE_WORKING, "%38": STATE_SHELL}, children={2880158: [2881158]}, cmdlines={2881158: ["/home/super/.local/bin/muse-bin-1.4.3"]}) with patches[0], patches[1], patches[2], patches[3], patches[4]: rows = w.runtime_rows("/tmp/sock") self.assertFalse(rows[0]["auto_approve"]) self.assertEqual(rows[0]["state"], "working") self.assertFalse(rows[0]["watcher_alive"]) def test_approval_pending_row_carries_kind(self): patches = self._patched(captures={"%37": STATE_APPROVAL, "%38": STATE_SHELL}) with patches[0], patches[1], patches[2], patches[3], patches[4]: rows = w.runtime_rows("/tmp/sock") self.assertEqual(rows[0]["state"], "approval-pending") self.assertEqual(rows[0]["prompt_kind"], "muse-approval") self.assertEqual(rows[0]["prompt_key"], "1") def test_tmux_failure_returns_empty(self): patches = self._patched(tmux_rc=1, tmux_stdout="") with patches[0], patches[1], patches[2], patches[3], patches[4]: self.assertEqual(w.runtime_rows("/tmp/sock"), []) def test_vanished_pane_skipped(self): patches = self._patched(captures={"%37": None, "%38": STATE_SHELL}) with patches[0], patches[1], patches[2], patches[3], patches[4]: rows = w.runtime_rows("/tmp/sock") self.assertEqual([r["pane"] for r in rows], ["%38"]) def test_pane_state_found_and_missing(self): patches = self._patched(captures={"%37": STATE_OPEN, "%38": STATE_SHELL}) with patches[0], patches[1], patches[2], patches[3], patches[4]: hit = w.pane_state("/tmp/sock", "%37") miss = w.pane_state("/tmp/sock", "%99") self.assertEqual(hit["pane"], "%37") self.assertEqual(miss["error"], "no_such_pane") def test_rows_flag_squeezed(self): listing = ("muse\t1\t%37\tmuse-bin-1.4\t2880158\t35\t7\n" "muse\t1\t%38\tbash\t2880200\t35\t7\n") patches = self._patched( tmux_stdout=listing, captures={"%37": STATE_OPEN, "%38": STATE_SHELL}) with patches[0], patches[1], patches[2], patches[3], patches[4]: rows = w.runtime_rows("/tmp/sock") self.assertTrue(rows[0]["squeezed"]) self.assertEqual(rows[0]["width"], 35) self.assertEqual(rows[0]["height"], 7) self.assertTrue(rows[1]["squeezed"]) class TestNodeFromSession(unittest.TestCase): def test_conforming_sessions(self): self.assertEqual(w.node_from_session("pip--worker--01"), "pip") self.assertEqual(w.node_from_session("muse--repair--09"), "muse") self.assertEqual(w.node_from_session("646--watch--a1"), "646") self.assertEqual(w.node_from_session("PIP--X--1"), "pip") def test_adhoc_sessions_have_no_node(self): for name in ("muse", "repair-09", "swarm-worker", "main", "", None): self.assertIsNone(w.node_from_session(name), name) def test_rejects_unknown_node_and_single_dash(self): self.assertIsNone(w.node_from_session("foo--worker--01")) self.assertIsNone(w.node_from_session("pip-worker-01")) self.assertIsNone(w.node_from_session("pip--worker")) def test_runtime_rows_carry_node(self): listing = "pip--audit--a1\t1\t%37\tbash\t2880158\t80\t24\n" r = mock.Mock(returncode=0, stdout=listing, stderr="") with mock.patch.object(w, "_tmux", return_value=r), \ mock.patch.object(w, "capture_pane", return_value=STATE_SHELL), \ mock.patch.object(w, "is_running", return_value=None): rows = w.runtime_rows("/tmp/sock") self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["node"], "pip") class TestSpreadTargets(unittest.TestCase): def _row(self, pane, is_muse, squeezed): return {"socket": "/tmp/s", "session": "muse", "window": "1", "pane": pane, "is_muse": is_muse, "squeezed": squeezed} def test_selects_squeezed_muse_only(self): rows = [self._row("%22", True, False), self._row("%23", True, True), self._row("%38", False, True)] targets = w.spread_targets(rows) self.assertEqual([t["pane"] for t in targets], ["%23"]) def test_empty_when_nothing_squeezed(self): rows = [self._row("%22", True, False)] self.assertEqual(w.spread_targets(rows), []) def test_tolerates_missing_keys(self): self.assertEqual(w.spread_targets([{"pane": "%1"}]), []) class TestBoxRuntimeCLI(unittest.TestCase): def _box(self, *argv, timeout=60): import subprocess cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "runtime"] + list(argv) return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) def test_list_empty_socket_json(self): r = self._box("list", "--socket", "/nonexistent.sock", "--json") self.assertEqual(r.returncode, 0, r.stderr[:500]) data = json.loads(r.stdout) self.assertTrue(data["ok"]) self.assertEqual(data["runtimes"], []) def test_list_muse_only_flag_accepted(self): r = self._box("list", "--socket", "/nonexistent.sock", "--json", "--muse-only") self.assertEqual(r.returncode, 0, r.stderr[:500]) self.assertTrue(json.loads(r.stdout)["ok"]) def test_subcommand_help(self): for sub in ("list", "send", "launch", "layout", "spread"): r = self._box(sub, "--help") self.assertEqual(r.returncode, 0, sub) def test_launch_dry_run_injects_approve(self): r = self._box("launch", "--session", "probe-x", "--dry-run", "--json") self.assertEqual(r.returncode, 0, r.stderr[:500]) data = json.loads(r.stdout) self.assertTrue(data["ok"]) self.assertTrue(data["dry_run"]) self.assertEqual(data["injected"], ["--disable-approval"]) self.assertIn("--disable-approval", data["cmdline"]) self.assertIn("muse-code", data["cmdline"]) def test_launch_dry_run_respects_caller_flags(self): r = self._box("launch", "--session", "probe-x", "--dry-run", "--json", "--", "--yolo") self.assertEqual(r.returncode, 0, r.stderr[:500]) data = json.loads(r.stdout) self.assertEqual(data["injected"], []) self.assertIn("--yolo", data["cmdline"]) self.assertNotIn("--disable-approval", data["cmdline"]) def test_send_missing_pane_json(self): r = self._box("send", "--socket", "/nonexistent.sock", "%99", "hi", "--json") self.assertEqual(r.returncode, 0, r.stderr[:500]) data = json.loads(r.stdout) self.assertFalse(data["ok"]) self.assertEqual(data["error"], "no_such_pane") def test_layout_empty_socket_json(self): r = self._box("layout", "--socket", "/nonexistent.sock", "--json") self.assertEqual(r.returncode, 0, r.stderr[:500]) data = json.loads(r.stdout) self.assertTrue(data["ok"]) self.assertEqual(data["runtimes"], []) self.assertIn("width", data["minimum"]) self.assertIn("height", data["minimum"]) def test_spread_empty_socket_json(self): r = self._box("spread", "--socket", "/nonexistent.sock", "--json") self.assertEqual(r.returncode, 0, r.stderr[:500]) data = json.loads(r.stdout) self.assertTrue(data["ok"]) self.assertEqual(data["spread"], []) def test_spread_dry_run_empty_socket_json(self): r = self._box("spread", "--socket", "/nonexistent.sock", "--dry-run", "--json") self.assertEqual(r.returncode, 0, r.stderr[:500]) data = json.loads(r.stdout) self.assertTrue(data["dry_run"]) self.assertEqual(data["targets"], []) if __name__ == "__main__": unittest.main()