feat(tmux): add server death watchdog daemon and multi-socket approver enhancements

tests/test_box_runtime.py

@@ -131,6 +131,44 @@ class TestApprovalFlags(unittest.TestCase):
         p = w.muse_approval_flags([])
         self.assertFalse(p["auto_approve"])
 
+    def test_permission_profile(self):
+        p = w.muse_approval_flags(
+            ["muse", "--permission-profile", ":unrestricted"])
+        self.assertEqual(p["profile"], ":unrestricted")
+        self.assertEqual(p["mode"], ":unrestricted")
+        self.assertIn("permission-profile=:unrestricted", p["flags"])
+        self.assertFalse(p["bypass"])
+
+    def test_permission_profile_equals(self):
+        p = w.muse_approval_flags(
+            ["muse", "--permission-profile=:read-only"])
+        self.assertEqual(p["profile"], ":read-only")
+        self.assertEqual(p["mode"], ":read-only")
+
+    def test_yolo_mode_and_bypass(self):
+        p = w.muse_approval_flags(["muse", "--yolo"])
+        self.assertEqual(p["mode"], "yolo")
+        self.assertTrue(p["bypass"])
+        self.assertIsNone(p["profile"])
+
+    def test_bypass_without_profile(self):
+        p = w.muse_approval_flags(["muse", "--disable-approval"])
+        self.assertTrue(p["bypass"])
+        self.assertEqual(p["mode"], "default")
+
+    def test_default_mode(self):
+        p = w.muse_approval_flags(["muse"])
+        self.assertEqual(p["mode"], "default")
+        self.assertFalse(p["bypass"])
+        self.assertIsNone(p["profile"])
+
+    def test_sandbox_and_trust_flags(self):
+        p = w.muse_approval_flags(
+            ["muse", "--disable-sandbox", "--trust-workspace"])
+        self.assertIn("disable-sandbox", p["flags"])
+        self.assertIn("trust-workspace", p["flags"])
+        self.assertFalse(p["bypass"])
+
 
 def _tmux_result(returncode=0, stdout="", stderr=""):
     r = mock.Mock()
@@ -238,6 +276,33 @@ class TestRuntimeRows(unittest.TestCase):
         self.assertEqual(rows[0]["height"], 7)
         self.assertTrue(rows[1]["squeezed"])
 
+    def test_rows_carry_permission_posture(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",
+                                "--permission-profile", ":unrestricted",
+                                "--disable-sandbox"]})
+        with patches[0], patches[1], patches[2], patches[3], patches[4]:
+            rows = w.runtime_rows("/tmp/sock")
+        self.assertEqual(rows[0]["permission_mode"], ":unrestricted")
+        self.assertEqual(rows[0]["permission_profile"], ":unrestricted")
+        self.assertFalse(rows[0]["permission_bypass"])
+        self.assertIn("permission-profile=:unrestricted",
+                      rows[0]["approval_flags"])
+        self.assertIsNone(rows[1]["permission_mode"])
+
+    def test_narrow_but_tall_not_squeezed(self):
+        # Empirical sizing: 35-wide panes answer cleanly when tall
+        # enough; only short height drops dialog text. Width minimum
+        # must not flag working panes as squeezed.
+        listing = "muse\t1\t%37\tmuse-bin-1.4\t2880158\t35\t35\n"
+        patches = self._patched(
+            tmux_stdout=listing, captures={"%37": STATE_OPEN})
+        with patches[0], patches[1], patches[2], patches[3], patches[4]:
+            rows = w.runtime_rows("/tmp/sock")
+        self.assertFalse(rows[0]["squeezed"])
+
 
 class TestNodeFromSession(unittest.TestCase):
     def test_conforming_sessions(self):

tests/test_tmux_auto_approver.py

@@ -220,5 +220,128 @@ class TestTallyGathering(unittest.TestCase):
         self.assertEqual(tally.panes[1].agent_node, "dev")
 
 
+class TestMuseDeferral(unittest.TestCase):
+    """tmux approver must defer muse panes owned by muse_choice_watcher.
+
+    Live double-answer regression: both daemons answered the same
+    Would-you-like prompt within the same second (box-ctl + tmux audit
+    overlap on %40/%0/%2), producing '11' + stray keys in the input box.
+    When a per-pane muse watcher is alive, tmux must skip the pane.
+    """
+
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.orig_state = tmux_auto_approver.STATE_FILE
+        self.orig_audit = tmux_auto_approver.AUDIT_LOG_FILE
+        tmux_auto_approver.STATE_FILE = Path(self.temp_dir.name) / "test_state.json"
+        tmux_auto_approver.AUDIT_LOG_FILE = Path(self.temp_dir.name) / "test_audit.jsonl"
+
+    def tearDown(self):
+        tmux_auto_approver.STATE_FILE = self.orig_state
+        tmux_auto_approver.AUDIT_LOG_FILE = self.orig_audit
+        self.temp_dir.cleanup()
+
+    def _muse_pane(self, pane_id="%37", socket="/tmp/tmux-1000/default"):
+        return TmuxPaneInfo(
+            socket=socket, session="muse", window_idx=0, pane_id=pane_id,
+            pane_pid=12345, current_command="muse-bin", active=True,
+            attached=True, title="muse terminal", agent_node="muse",
+            auto_approve=True,
+        )
+
+    def _tally(self, panes):
+        return TmuxWorkerTally(
+            total_sockets=1, total_sessions=1, total_panes=len(panes),
+            active_workers=len(panes), by_agent={}, panes=panes,
+        )
+
+    @patch("tmux_auto_approver.run_tmux_cmd")
+    @patch("tmux_auto_approver.capture_pane_text")
+    @patch("tmux_auto_approver.gather_tmux_tally")
+    def test_muse_pane_skipped_when_watcher_alive(
+            self, mock_tally, mock_capture, mock_tmux_cmd):
+        import muse_choice_watcher as mcw
+        mock_tally.return_value = self._tally([self._muse_pane()])
+        mock_capture.return_value = (
+            "Would you like to run the following?\n› 1. Yes, proceed (y)")
+        with patch.object(mcw, "is_running", return_value=99999):
+            runner = AutoApproverRunner(dry_run=True)
+            results = runner.run_once()
+        self.assertEqual(results, [])
+        mock_tmux_cmd.assert_not_called()
+
+    @patch("tmux_auto_approver.run_tmux_cmd")
+    @patch("tmux_auto_approver.capture_pane_text")
+    @patch("tmux_auto_approver.gather_tmux_tally")
+    def test_non_muse_pane_still_approved(
+            self, mock_tally, mock_capture, mock_tmux_cmd):
+        pane = TmuxPaneInfo(
+            socket="/tmp/tmux-pip.sock", session="worker", window_idx=0,
+            pane_id="%1", pane_pid=999, current_command="agent-worker",
+            active=True, attached=True, title="w", agent_node="pip",
+            auto_approve=True,
+        )
+        mock_tally.return_value = self._tally([pane])
+        mock_capture.return_value = (
+            "Would you like to run the following?\n› 1. Yes, proceed (y)")
+        runner = AutoApproverRunner(dry_run=True)
+        results = runner.run_once()
+        self.assertEqual(len(results), 1)
+        self.assertEqual(results[0]["action"], "DRY_RUN_MATCH")
+
+
+class TestDedupSocketScoped(unittest.TestCase):
+    """Dedup must be keyed by socket:pane, not bare pane id.
+
+    Same %N exists on every tmux socket; bare-pane dedup suppresses a
+    real prompt on socket B because socket A saw one (the %0-on-two-
+    sockets collision, tmux-side).
+    """
+
+    def setUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.orig_state = tmux_auto_approver.STATE_FILE
+        self.orig_audit = tmux_auto_approver.AUDIT_LOG_FILE
+        tmux_auto_approver.STATE_FILE = Path(self.temp_dir.name) / "test_state.json"
+        tmux_auto_approver.AUDIT_LOG_FILE = Path(self.temp_dir.name) / "test_audit.jsonl"
+
+    def tearDown(self):
+        tmux_auto_approver.STATE_FILE = self.orig_state
+        tmux_auto_approver.AUDIT_LOG_FILE = self.orig_audit
+        self.temp_dir.cleanup()
+
+    @patch("tmux_auto_approver.run_tmux_cmd")
+    @patch("tmux_auto_approver.capture_pane_text")
+    @patch("tmux_auto_approver.gather_tmux_tally")
+    def test_same_pane_id_on_two_sockets_both_approved(
+            self, mock_tally, mock_capture, mock_tmux_cmd):
+        def mk(sock):
+            return TmuxPaneInfo(
+                socket=sock, session="w", window_idx=0, pane_id="%1",
+                pane_pid=999, current_command="agent-worker", active=True,
+                attached=True, title="w", agent_node="pip",
+                auto_approve=True,
+            )
+        mock_tally.return_value = TmuxWorkerTally(
+            total_sockets=2, total_sessions=2, total_panes=2,
+            active_workers=2, by_agent={},
+            panes=[mk("/tmp/tmux-pip.sock"), mk("/tmp/tmux-opm.sock")],
+        )
+        mock_capture.return_value = (
+            "Would you like to run the following?\n› 1. Yes, proceed (y)")
+        runner = AutoApproverRunner(dry_run=True)
+        results = runner.run_once()
+        self.assertEqual(len(results), 2)
+
+
+class TestCaptureJoinWrapped(unittest.TestCase):
+    def test_capture_joins_wrapped_lines(self):
+        with patch("tmux_auto_approver.run_tmux_cmd",
+                   return_value=(0, "ok", "")) as m:
+            tmux_auto_approver.capture_pane_text("/tmp/s", "%1", lines=30)
+        args = m.call_args[0]
+        self.assertIn("-J", args)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_tmux_server_watchdog.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""test_tmux_server_watchdog.py — Death-capture transition logic.
+
+Covers: steady state is quiet, pid change yields restart, alive->dead
+yields a death bundle, dead->alive yields started, corrupt state file
+is tolerated.
+"""
+
+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 tmux_server_watchdog as w
+
+
+class TestEvaluate(unittest.TestCase):
+    def test_steady_alive_is_quiet(self):
+        prev = {"/s": {"pid": 100, "since": "t"}}
+        new, events = w.evaluate(prev, {"/s": 100})
+        self.assertEqual(events, [])
+        self.assertEqual(new["/s"]["pid"], 100)
+
+    def test_pid_change_is_restart(self):
+        prev = {"/s": {"pid": 100, "since": "t"}}
+        new, events = w.evaluate(prev, {"/s": 200})
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0]["type"], "restart")
+        self.assertEqual(events[0]["old_pid"], 100)
+
+    def test_alive_to_dead_is_death(self):
+        prev = {"/s": {"pid": 100, "since": "t"}}
+        new, events = w.evaluate(prev, {"/s": None})
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0]["type"], "death")
+        self.assertIsNone(new["/s"]["pid"])
+
+    def test_dead_stays_dead_is_quiet(self):
+        prev = {"/s": {"pid": None, "died": "t", "last_pid": 100}}
+        _, events = w.evaluate(prev, {"/s": None})
+        self.assertEqual(events, [])
+
+    def test_dead_to_alive_is_started(self):
+        prev = {"/s": {"pid": None, "died": "t", "last_pid": 100}}
+        _, events = w.evaluate(prev, {"/s": 300})
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0]["type"], "started")
+
+    def test_unknown_socket_first_seen_is_started(self):
+        _, events = w.evaluate({}, {"/s": 300})
+        self.assertEqual(events[0]["type"], "started")
+
+
+class TestCheck(unittest.TestCase):
+    def _iso(self, td):
+        state = str(td / "servers.json")
+        log = str(td / "deaths.jsonl")
+        p1 = mock.patch.object(w, "STATE_FILE", state)
+        p2 = mock.patch.object(w, "DEATH_LOG", log)
+        return p1, p2, state, log
+
+    def test_death_writes_bundle(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            td = Path(td)
+            p1, p2, state, log = self._iso(td)
+            with open(state, "w") as f:
+                json.dump({"/s": {"pid": 100, "since": "t"}}, f)
+            with p1, p2, \
+                 mock.patch.object(w, "probe", return_value=None), \
+                 mock.patch.object(w, "collect_forensics",
+                                   return_value={"ts": "t", "socket": "/s",
+                                                 "last_pid": 100}):
+                res = w.check(sockets=["/s"])
+            self.assertEqual(res["events"][0]["type"], "death")
+            bundle = json.loads(open(log).read().strip())
+            self.assertEqual(bundle["event"], "death")
+            self.assertEqual(bundle["last_pid"], 100)
+            self.assertIsNone(json.load(open(state))["/s"]["pid"])
+
+    def test_dry_run_writes_nothing(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            td = Path(td)
+            p1, p2, state, log = self._iso(td)
+            with p1, p2, \
+                 mock.patch.object(w, "probe", return_value=100):
+                res = w.check(sockets=["/s"], dry_run=True)
+            self.assertEqual(res["events"][0]["type"], "started")
+            self.assertFalse(Path(state).exists())
+            self.assertFalse(Path(log).exists())
+
+    def test_corrupt_state_tolerated(self):
+        import tempfile
+        with tempfile.TemporaryDirectory() as td:
+            td = Path(td)
+            p1, p2, state, _ = self._iso(td)
+            with open(state, "w") as f:
+                f.write("{{{nope")
+            with p1, p2, \
+                 mock.patch.object(w, "probe", return_value=100):
+                res = w.check(sockets=["/s"])
+            self.assertEqual(res["events"][0]["type"], "started")
+
+
+if __name__ == "__main__":
+    unittest.main()