#!/usr/bin/env python3 """test_muse_choice_watcher.py — Focused tests for the Muse A/B/C watcher. Covers: prompt matching (A-default regex), once-per-prompt answering policy, socket-namespaced state files (the %0-on-two-sockets collision regression), and send-keys command construction. """ import sys import time 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_ABC = """Some agent output here. How should I proceed? A. Apply the fix now B. Show a diff first C. Skip this file Reply with A, B, or C: """ PROMPT_AB_WRAPPED = """Long thinking output... Which approach? A. Switch coverage tuples to a single cached registry call with ALL_NODES B. Keep per-node calls and retry Your choice (A/B)? """ PROMPT_STALE = ( "A. Old option one\nB. Old option two\nPick one (A/B)?\n" + "\n".join("filler line %d" % i for i in range(40)) ) class TestMatcher(unittest.TestCase): def test_matches_abc_with_cue(self): m = w.find_choice_prompt(PROMPT_ABC) self.assertIsNotNone(m) self.assertEqual(len(m["options"]), 3) self.assertTrue(m["options"][0].startswith("A.")) # Cue scan hits the question line above the options first; either cue # line proves the block was recognized as awaiting a reply. self.assertIn(m["cue"], ("How should I proceed?", "Reply with A, B, or C:")) self.assertEqual(len(m["sig"]), 16) def test_matches_ab_wrapped(self): m = w.find_choice_prompt(PROMPT_AB_WRAPPED) self.assertIsNotNone(m) self.assertEqual(len(m["options"]), 2) def test_rejects_single_option(self): self.assertIsNone(w.find_choice_prompt("A. Only one option\nSome text\n")) def test_rejects_lettered_list_without_cue(self): text = "A. Apples are red\nB. Bananas are yellow\nJust a grocery list.\n" self.assertIsNone(w.find_choice_prompt(text)) def test_rejects_stale_scrollback(self): self.assertIsNone(w.find_choice_prompt(PROMPT_STALE)) def test_matches_prompt_with_trailing_blanks(self): # Tall panes pad output with blank lines; a live prompt above the # padding must still match (scratch-pane regression). text = PROMPT_ABC + "\n" * 30 m = w.find_choice_prompt(text) self.assertIsNotNone(m) self.assertEqual(len(m["options"]), 3) def test_rejects_out_of_order(self): text = "B. Second thing\nA. First thing\nWhich (A/B)?\n" self.assertIsNone(w.find_choice_prompt(text)) def test_rejects_empty(self): self.assertIsNone(w.find_choice_prompt("")) self.assertIsNone(w.find_choice_prompt(None)) def test_sig_stable_and_sensitive(self): a = w.find_choice_prompt(PROMPT_ABC)["sig"] b = w.find_choice_prompt(PROMPT_ABC)["sig"] self.assertEqual(a, b) changed = PROMPT_ABC.replace("Apply the fix now", "Apply the fix later") c = w.find_choice_prompt(changed)["sig"] self.assertNotEqual(a, c) PROMPT_YN = """Migrating 12 threads... Proceed with the migration? (y/n) """ PROMPT_YN_BRACKET = """Target file exists. Overwrite existing file? [y/N] """ PROMPT_NUMBERED = """Requesting permission for: rm -rf /tmp/x (1) Allow once (2) Always allow Selection: """ class TestPromptKinds(unittest.TestCase): def test_letter_kind_and_key(self): m = w.find_choice_prompt(PROMPT_ABC) self.assertEqual(m["kind"], "letter") self.assertEqual(m["key"], "A") def test_yn_paren(self): m = w.find_choice_prompt(PROMPT_YN) self.assertIsNotNone(m) self.assertEqual((m["kind"], m["key"]), ("yn", "y")) def test_yn_bracket(self): m = w.find_choice_prompt(PROMPT_YN_BRACKET) self.assertIsNotNone(m) self.assertEqual((m["kind"], m["key"]), ("yn", "y")) def test_yn_rejects_mid_line_mention(self): self.assertIsNone(w.find_choice_prompt("use y/n for confirmation\nok\n")) def test_yn_rejects_stale(self): text = "Proceed? (y/n)\n" + "\n".join("filler %d" % i for i in range(10)) self.assertIsNone(w.find_choice_prompt(text)) def test_numbered_kind_and_key(self): m = w.find_choice_prompt(PROMPT_NUMBERED) self.assertIsNotNone(m) self.assertEqual((m["kind"], m["key"]), ("numbered", "1")) self.assertEqual(len(m["options"]), 2) def test_numbered_rejects_single(self): self.assertIsNone( w.find_choice_prompt("(1) Only option\nSelection:\n")) def test_numbered_rejects_without_cue(self): self.assertIsNone( w.find_choice_prompt("(1) one\n(2) two\nSome other text.\n")) def test_letter_beats_yn_priority(self): m = w.find_choice_prompt(PROMPT_ABC + "Proceed? (y/n)\n") self.assertIsNotNone(m) self.assertEqual(m["kind"], "letter") def test_sigs_namespaced_by_kind(self): a = w.find_choice_prompt(PROMPT_ABC)["sig"] b = w.find_choice_prompt(PROMPT_YN)["sig"] c = w.find_choice_prompt(PROMPT_NUMBERED)["sig"] self.assertEqual(len({a, b, c}), 3) class TestPollOnce(unittest.TestCase): def _run(self, captures, dry_run=False, prefill_cap=False): state = w.WatcherState() if prefill_cap: import time for i in range(w.MAX_ANSWERS_PER_HOUR): state.record_answer("old-%d" % i, time.time()) log = mock.Mock() with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", side_effect=captures) as cap, \ mock.patch.object(w, "send_answer", return_value=True) as send, \ mock.patch.object(w, "audit") as audit: outcomes = [w._poll_once("/tmp/s", "%1", state, log, dry_run=dry_run) for _ in range(len(captures) - 1)] return state, log, cap, send, audit, outcomes def _log_msgs(self, log): return [c.args[1] for c in log.log.call_args_list] def test_letter_answered_with_A(self): state, log, cap, send, audit, outcomes = self._run( [PROMPT_ABC, PROMPT_ABC, PROMPT_ABC]) self.assertEqual(outcomes, ["seen", "answered"]) send.assert_called_once_with("/tmp/s", "%1", "A", enter=True) audit.assert_called_once() self.assertIn("prompt seen", self._log_msgs(log)) self.assertIn("prompt stable, answering", self._log_msgs(log)) def test_yn_answered_with_y(self): state, log, cap, send, audit, outcomes = self._run( [PROMPT_YN, PROMPT_YN, PROMPT_YN]) self.assertEqual(outcomes[-1], "answered") send.assert_called_once_with("/tmp/s", "%1", "y", enter=True) self.assertEqual(audit.call_args[0][0], "muse-choice-answered") self.assertEqual(audit.call_args[1]["extra"]["key"], "y") def test_numbered_answered_with_1(self): state, log, cap, send, audit, outcomes = self._run( [PROMPT_NUMBERED, PROMPT_NUMBERED, PROMPT_NUMBERED]) self.assertEqual(outcomes[-1], "answered") send.assert_called_once_with("/tmp/s", "%1", "1", enter=True) def test_dry_run_records_without_sending(self): state, log, cap, send, audit, outcomes = self._run( [PROMPT_YN, PROMPT_YN, PROMPT_YN], dry_run=True) self.assertEqual(outcomes, ["seen", "dry-answered"]) send.assert_not_called() audit.assert_not_called() self.assertEqual(len(state.answered_sigs), 1) def test_vanished_prompt_skips_send(self): state, log, cap, send, audit, outcomes = self._run( [PROMPT_YN, PROMPT_YN, "something else entirely\n"]) self.assertEqual(outcomes, ["seen", "vanished"]) send.assert_not_called() audit.assert_not_called() def test_capped_logs_once(self): state, log, cap, send, audit, outcomes = self._run( [PROMPT_YN] * 5, prefill_cap=True) self.assertTrue(all(o == "capped" for o in outcomes[1:])) send.assert_not_called() warns = [c for c in log.log.call_args_list if c.args[0] == "warn"] self.assertEqual(len(warns), 1) def test_gone_and_capture_failed(self): state, log = w.WatcherState(), mock.Mock() with mock.patch.object(w, "pane_exists", return_value=False): self.assertEqual( w._poll_once("/tmp/s", "%1", state, log), "gone") with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", return_value=None): self.assertEqual( w._poll_once("/tmp/s", "%1", state, log), "capture-failed") class TestAnswerPolicy(unittest.TestCase): def test_needs_stability(self): st = w.WatcherState() m = w.find_choice_prompt(PROMPT_ABC) now = time.time() self.assertEqual(st.observe(m, now), "wait") self.assertEqual(st.observe(m, now), "answer") def test_once_per_prompt(self): st = w.WatcherState() m = w.find_choice_prompt(PROMPT_ABC) now = time.time() st.observe(m, now) st.observe(m, now) st.record_answer(m["sig"], now) self.assertEqual(st.observe(m, now), "none") # A new prompt answers again. m2 = w.find_choice_prompt(PROMPT_AB_WRAPPED) self.assertNotEqual(m2["sig"], m["sig"]) self.assertEqual(st.observe(m2, now), "wait") self.assertEqual(st.observe(m2, now), "answer") def test_none_resets_pending(self): st = w.WatcherState() m = w.find_choice_prompt(PROMPT_ABC) now = time.time() st.observe(m, now) self.assertEqual(st.observe(None, now), "none") self.assertEqual(st.observe(m, now), "wait") def test_hourly_cap(self): st = w.WatcherState() now = time.time() for i in range(w.MAX_ANSWERS_PER_HOUR): st.record_answer("sig-%d" % i, now) m = w.find_choice_prompt(PROMPT_ABC) st.observe(m, now) self.assertEqual(st.observe(m, now), "capped") def test_answered_ttl_allows_recovery(self): st = w.WatcherState() now = time.time() m = w.find_choice_prompt(PROMPT_ABC) st.observe(m, now) st.observe(m, now) st.record_answer(m["sig"], now) self.assertEqual(st.observe(m, now + 1), "none") # Same prompt still present past TTL => stuck dialog, re-answer. self.assertEqual(st.observe(m, now + w.ANSWERED_TTL_SECONDS + 1), "wait") self.assertEqual(st.observe(m, now + w.ANSWERED_TTL_SECONDS + 2), "answer") class TestNamespacing(unittest.TestCase): """Same pane id on different sockets must never share state files.""" def test_pidfile_differs_across_sockets(self): a = w.pidfile_for("/tmp/tmux-1000/default", "%0") b = w.pidfile_for("/tmp/tmux-1000/lte", "%0") self.assertNotEqual(a, b) def test_logfile_differs_across_sockets(self): a = w.logfile_for("/tmp/tmux-1000/default", "%0") b = w.logfile_for("/tmp/tmux-1000/lte", "%0") self.assertNotEqual(a, b) def test_slug_guards_same_basename(self): a = w.slug_socket("/tmp/a/default") b = w.slug_socket("/tmp/b/default") self.assertNotEqual(a, b) class TestSendAnswer(unittest.TestCase): def test_sends_A_then_enter_to_exact_pane(self): calls = [] def fake_tmux(sock, *args, timeout=5): calls.append((sock, args)) r = mock.Mock() r.returncode = 0 return r with mock.patch.object(w, "_tmux", side_effect=fake_tmux): self.assertTrue(w.send_answer("/tmp/tmux-1000/default", "%37")) self.assertEqual(len(calls), 2) self.assertEqual(calls[0][0], "/tmp/tmux-1000/default") self.assertEqual(calls[0][1][:3], ("send-keys", "-t", "%37")) self.assertEqual(calls[0][1][3], "A") self.assertEqual(calls[1][1][3], "Enter") def test_send_failure_returns_false(self): def failing(sock, *args, timeout=5): r = mock.Mock() r.returncode = 1 return r with mock.patch.object(w, "_tmux", side_effect=failing): self.assertFalse(w.send_answer("/tmp/tmux-1000/default", "%37")) class TestDesiredState(unittest.TestCase): def test_default_is_on(self): # Policy: undefined desired state means auto-approve on. with mock.patch.object(w, "DESIRED_STATE_FILE", "/nonexistent/x.json"): st = w.get_desired() self.assertTrue(st["enabled"]) self.assertFalse(st["dry_run"]) def test_missing_key_defaults_on(self): import json import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/muse-choices.json" with open(path, "w") as f: json.dump({"dry_run": False}, f) with mock.patch.object(w, "DESIRED_STATE_FILE", path): self.assertTrue(w.get_desired()["enabled"]) def test_explicit_off_is_respected(self): import json import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/muse-choices.json" with open(path, "w") as f: json.dump({"enabled": False, "dry_run": False}, f) with mock.patch.object(w, "DESIRED_STATE_FILE", path): self.assertFalse(w.get_desired()["enabled"]) def test_set_enabled_roundtrip(self): import json import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/muse-choices.json" with mock.patch.object(w, "DESIRED_STATE_FILE", path), \ mock.patch.object(w, "audit") as audit: st = w.set_enabled(True, dry_run=True, by="tester") self.assertTrue(st["enabled"]) self.assertTrue(st["dry_run"]) self.assertEqual(st["updated_by"], "tester") self.assertTrue(w.get_desired()["enabled"]) audit.assert_called_once() args, _ = audit.call_args self.assertEqual(args[0], "muse-choice-enabled") with open(path) as f: on_disk = json.load(f) self.assertTrue(on_disk["enabled"]) class TestAudit(unittest.TestCase): def test_local_fallback_shape(self): import json import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/box-ctl.jsonl" with mock.patch.object(w, "CTL_LOG", path): w._audit_local("muse-choice-answered", "default:%37", "watcher", {"sig": "abc"}) with open(path) as f: rec = json.loads(f.read()) self.assertEqual(rec["action"], "muse-choice-answered") self.assertEqual(rec["type"], "muse-choice") self.assertEqual(rec["name"], "default:%37") self.assertEqual(rec["sig"], "abc") self.assertIn("ts", rec) def test_audit_prefers_approvals_module(self): import sys fake = mock.Mock() with mock.patch.dict(sys.modules, {"approvals": fake}): w.audit("muse-choice-enabled", caller="box", extra={"dry_run": False}) fake.log_box_ctl.assert_called_once_with( "muse-choice-enabled", name=None, caller="box", extra={"dry_run": False}) def test_audit_never_raises(self): import sys with mock.patch.dict(sys.modules, {"approvals": None}), \ mock.patch.object(w, "_audit_local", side_effect=OSError("disk")): w.audit("muse-choice-answered") # must not raise class TestReconcile(unittest.TestCase): def _patch_common(self, enabled, dry_run=False): return (mock.patch.object(w, "get_desired", return_value={"enabled": enabled, "dry_run": dry_run}), mock.patch.object(w, "muse_panes", return_value=["%37"]), mock.patch.object(w, "_start_detached", return_value=True), mock.patch.object(w, "status_all", return_value=[]), mock.patch.object(w, "stop_all", return_value=[]), mock.patch.object(w, "audit"), mock.patch.object(w.os.path, "exists", return_value=True)) def test_enabled_starts_missing(self): patches = self._patch_common(True) with patches[0], patches[1], patches[2] as start, patches[3], \ patches[4], patches[5] as audit, patches[6], \ mock.patch.object(w, "is_running", return_value=None): res = w.reconcile(sockets=["/tmp/sock"]) start.assert_called_once_with("/tmp/sock", "%37", dry_run=False) self.assertEqual(res["started"], ["/tmp/sock:%37"]) audit.assert_called_once() # changed something -> audited def test_enabled_skips_running_and_stays_quiet(self): patches = self._patch_common(True) with patches[0], patches[1], patches[2] as start, patches[3], \ patches[4], patches[5] as audit, patches[6], \ mock.patch.object(w, "is_running", return_value=1234): res = w.reconcile(sockets=["/tmp/sock"]) start.assert_not_called() self.assertEqual(res["already"], ["/tmp/sock:%37"]) audit.assert_not_called() # no change -> no audit noise def test_disabled_stops_all(self): patches = self._patch_common(False) with patches[0], patches[1], patches[2] as start, patches[3], \ mock.patch.object(w, "stop_all", return_value=[{"pidfile": "x.pid", "pid": 1}]), \ patches[5] as audit, patches[6]: res = w.reconcile(sockets=["/tmp/sock"]) start.assert_not_called() self.assertFalse(res["enabled"]) self.assertEqual(len(res["stopped"]), 1) audit.assert_called_once() def test_prunes_dead_pidfiles(self): patches = self._patch_common(True) dead = [{"alive": False, "pidfile": "muse-choice-watcher-x.pid"}] with patches[0], patches[1], patches[2], \ mock.patch.object(w, "status_all", return_value=dead), \ patches[4], patches[5], patches[6], \ mock.patch.object(w, "is_running", return_value=999), \ mock.patch.object(w.os, "remove") as rm: res = w.reconcile(sockets=["/tmp/sock"]) rm.assert_called_once() self.assertEqual(res["pruned"], ["muse-choice-watcher-x.pid"]) class TestPidfileClaim(unittest.TestCase): def test_claims_missing_file(self): import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/x.pid" self.assertTrue(w._claim_pidfile(path)) with open(path) as f: self.assertEqual(f.read().strip(), str(w.os.getpid())) def test_refuses_live_other_watcher(self): import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/x.pid" with open(path, "w") as f: f.write("99999998") with mock.patch.object(w, "_pid_alive", return_value=True), \ mock.patch.object(w, "_pid_is_watcher", return_value=True): self.assertFalse(w._claim_pidfile(path)) def test_takes_over_dead_pid(self): import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/x.pid" with open(path, "w") as f: f.write("99999997") with mock.patch.object(w, "_pid_alive", return_value=False): self.assertTrue(w._claim_pidfile(path)) class TestAnsweredPersistence(unittest.TestCase): """Answered sigs survive restarts: a daemon that restarts while an answered prompt is still visible must not answer it again (live double-answer -> stray "1" in the input box).""" def test_restart_suppresses_answered_sig(self): import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/x.answered.json" s1 = w.WatcherState(persist_path=path) now = time.time() s1.record_answer("sig-abc", now) s2 = w.WatcherState(persist_path=path) self.assertEqual(s2.observe({"sig": "sig-abc"}, now + 5), "none") def test_expired_sig_not_loaded(self): import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/x.answered.json" s1 = w.WatcherState(persist_path=path) s1.record_answer("sig-old", time.time() - w.ANSWERED_TTL_SECONDS - 10) s2 = w.WatcherState(persist_path=path) self.assertNotIn("sig-old", s2.answered_sigs) def test_corrupt_store_loads_empty(self): import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/x.answered.json" with open(path, "w") as f: f.write("not json{{{") s = w.WatcherState(persist_path=path) self.assertEqual(s.answered_sigs, {}) def test_memory_only_without_path(self): s = w.WatcherState() s.record_answer("sig-x", time.time()) self.assertIn("sig-x", s.answered_sigs) class TestPidfileFlock(unittest.TestCase): """Single daemon per pane, kernel-enforced: concurrent starters must not pile up (double answers, '11' in the input box).""" def _child_claim(self, path): import subprocess code = ("import sys; sys.path.insert(0, %r); " "import muse_choice_watcher as w; " "sys.exit(0 if w._claim_pidfile(%r) else 3)") % ( str(BIN_DIR), path) return subprocess.run([sys.executable, "-c", code], capture_output=True, timeout=30) def test_second_process_refused_while_held(self): import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/x.pid" self.assertTrue(w._claim_pidfile(path)) try: r = self._child_claim(path) self.assertEqual(r.returncode, 3) finally: w._release_pidfile(path) def test_claim_succeeds_after_release(self): import os import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/x.pid" self.assertTrue(w._claim_pidfile(path)) w._release_pidfile(path) self.assertFalse(os.path.exists(path)) r = self._child_claim(path) self.assertEqual(r.returncode, 0) class TestReconcileFailed(unittest.TestCase): def test_failed_starts_recorded(self): with mock.patch.object(w, "get_desired", return_value={"enabled": True, "dry_run": False}), \ mock.patch.object(w, "muse_panes", return_value=["%37"]), \ mock.patch.object(w, "is_running", return_value=None), \ mock.patch.object(w, "_start_detached", return_value=False), \ mock.patch.object(w, "status_all", return_value=[]), \ mock.patch.object(w, "audit") as audit, \ mock.patch.object(w.os.path, "exists", return_value=True): res = w.reconcile(sockets=["/tmp/sock"]) self.assertEqual(res["failed"], ["/tmp/sock:%37"]) self.assertEqual(res["started"], []) audit.assert_called_once() class TestWatchCommand(unittest.TestCase): def test_watch_refused_when_claimed(self): with mock.patch.object(w, "_claim_pidfile", return_value=False): rc = w.main(["watch", "--socket", "/tmp/s", "--pane", "%1"]) self.assertEqual(rc, 3) def test_watch_runs_loop_and_releases_own_pidfile(self): import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/x.pid" with open(path, "w") as f: f.write(str(w.os.getpid())) with mock.patch.object(w, "pidfile_for", return_value=path), \ mock.patch.object(w, "_claim_pidfile", return_value=True), \ mock.patch.object(w, "watch_loop", return_value=0) as loop: rc = w.main(["watch", "--socket", "/tmp/s", "--pane", "%1"]) self.assertEqual(rc, 0) loop.assert_called_once_with("/tmp/s", "%1", dry_run=False) self.assertFalse(w.os.path.exists(path)) class TestWatchProcs(unittest.TestCase): def _mkproc(self, root, pid, argv): import os d = os.path.join(root, str(pid)) os.makedirs(d) with open(os.path.join(d, "cmdline"), "wb") as f: f.write(b"\0".join(x.encode() for x in argv) + b"\0") def test_exact_argv_scan(self): import os import tempfile with tempfile.TemporaryDirectory() as td: self._mkproc(td, 111, ["python3", "/x/muse_choice_watcher.py", "watch", "--socket", "/tmp/s", "--pane", "%1"]) self._mkproc(td, 222, ["python3", "bin/super-cli.py", "muse-choices", "reconcile"]) self._mkproc(td, 333, ["python3", "/x/muse_choice_watcher.py", "reconcile"]) os.makedirs(os.path.join(td, "self")) procs = w._watch_procs(proc_root=td) self.assertEqual(procs, [{"pid": 111, "socket": "/tmp/s", "pane": "%1"}]) def test_missing_root(self): self.assertEqual(w._watch_procs(proc_root="/nonexistent-proc"), []) class TestWatchDuplicate(unittest.TestCase): def test_watch_refuses_duplicate_argv(self): dup = [{"pid": 9991, "socket": "/tmp/s", "pane": "%1"}] with mock.patch.object(w, "_watch_procs", return_value=dup), \ mock.patch.object(w, "watch_loop") as loop: rc = w.main(["watch", "--socket", "/tmp/s", "--pane", "%1"]) self.assertEqual(rc, 3) loop.assert_not_called() def test_watch_allows_different_pane(self): other = [{"pid": 9991, "socket": "/tmp/s", "pane": "%2"}] with mock.patch.object(w, "_watch_procs", return_value=other), \ mock.patch.object(w, "_claim_pidfile", return_value=True), \ mock.patch.object(w, "watch_loop", return_value=0) as loop: rc = w.main(["watch", "--socket", "/tmp/s", "--pane", "%1"]) self.assertEqual(rc, 0) loop.assert_called_once() class TestStopAllOrphans(unittest.TestCase): def test_stop_all_kills_orphans(self): import signal orphan = {"pid": 8888, "socket": "/tmp/s", "pane": "%9"} with mock.patch("os.listdir", return_value=[]), \ mock.patch.object(w, "_watch_procs", return_value=[orphan]), \ mock.patch("os.kill") as kill: res = w.stop_all() kill.assert_called_once_with(8888, signal.SIGTERM) self.assertEqual(res[0]["status"], "stopped-orphan") self.assertEqual(res[0]["pane"], "%9") class TestStatusOrphans(unittest.TestCase): def test_status_lists_orphans(self): orphan = {"pid": 99999999, "socket": "/tmp/sock-x", "pane": "%9"} with mock.patch.object(w, "_watch_procs", return_value=[orphan]): rows = w.status_all() orphans = [r for r in rows if r.get("orphan")] self.assertEqual(len(orphans), 1) self.assertEqual(orphans[0]["pid"], 99999999) class TestRecentAnswers(unittest.TestCase): def test_filters_and_limits(self): import json import tempfile with tempfile.TemporaryDirectory() as td: path = td + "/box-ctl.jsonl" with open(path, "w") as f: f.write('{"action": "other"}\n') f.write('not json\n') for i in range(3): f.write(json.dumps({"action": "muse-choice-answered", "sig": "s%d" % i}) + "\n") with mock.patch.object(w, "CTL_LOG", path): recs = w.recent_answers(limit=2) self.assertEqual([r["sig"] for r in recs], ["s1", "s2"]) def test_missing_log_returns_empty(self): with mock.patch.object(w, "CTL_LOG", "/nonexistent/x.jsonl"): self.assertEqual(w.recent_answers(), []) class TestDaemonUnits(unittest.TestCase): def test_reconcile_unit_lets_daemons_survive(self): # Load-bearing line: without KillMode=process, systemd kills # timer-spawned watchers when the oneshot service exits. text = (REPO_ROOT / "systemd" / "muse-choices-reconcile.service" ).read_text() self.assertIn("KillMode=process", text) self.assertIn("muse_choice_watcher.py reconcile", text) def test_reconcile_timer_exists(self): text = (REPO_ROOT / "systemd" / "muse-choices-reconcile.timer" ).read_text() self.assertIn("OnUnitActiveSec=", text) class TestBoxWiring(unittest.TestCase): def test_box_status_json_shape(self): import json import subprocess cmd = [sys.executable, str(BIN_DIR / "super-cli.py"), "muse-choices", "status", "--json"] r = subprocess.run(cmd, capture_output=True, text=True, timeout=60) self.assertEqual(r.returncode, 0, r.stderr[:500]) data = json.loads(r.stdout) self.assertTrue(data["ok"]) self.assertIn("enabled", data["desired"]) self.assertIsInstance(data["watchers"], list) self.assertIsInstance(data["recent_answers"], list) CURSOR = "›" # Muse TUI menu cursor (U+203A) PROMPT_MUSE_APPROVAL = ( "Would you like to run the following\n" "\n" " $ tmux -S /tmp/tmux-1000/default capture-pane -p -t %39\n" "\n" + CURSOR + " 1. Yes, proceed (y)\n" " 2. No, and tell Muse Code what to do instead\n" ) PROMPT_MUSE_APPROVAL_WRAPPED = ( "Would you like to run the following\n" "\n" " $ ps -eo pid,etime,args | grep\n" " \"[m]use_choice_watcher.py\n" " watch\" | wc -l; box\n" " muse-choices status\n" "\n" + CURSOR + " 1. Yes, proceed (y)\n" " 2. No, and tell Muse Code what to\n" " do instead\n" ) PROMPT_MUSE_APPROVAL_DECIDED = ( PROMPT_MUSE_APPROVAL + "approval decision accepted\n" ) class TestMuseApproval(unittest.TestCase): """Native Muse TUI approval menu: Would-you-like + 1.Yes/2.No.""" def test_matches_native_approval(self): m = w.find_choice_prompt(PROMPT_MUSE_APPROVAL) self.assertIsNotNone(m) self.assertEqual((m["kind"], m["key"]), ("muse-approval", "1")) self.assertEqual(m["cue"], "Would you like to run the following") self.assertEqual(len(m["options"]), 2) def test_matches_wrapped_command_echo(self): m = w.find_choice_prompt(PROMPT_MUSE_APPROVAL_WRAPPED) self.assertIsNotNone(m) self.assertEqual((m["kind"], m["key"]), ("muse-approval", "1")) def test_matches_ascii_cursor(self): text = PROMPT_MUSE_APPROVAL.replace(CURSOR, ">") m = w.find_choice_prompt(text) self.assertIsNotNone(m) self.assertEqual(m["kind"], "muse-approval") def test_matches_cursor_on_no(self): text = PROMPT_MUSE_APPROVAL.replace(CURSOR + " 1.", " 1.") text = text.replace(" 2.", CURSOR + " 2.") m = w.find_choice_prompt(text) self.assertIsNotNone(m) self.assertEqual((m["kind"], m["key"]), ("muse-approval", "1")) def test_matches_observed_live_variant(self): # Verbatim shape answered on a live pane: cue with "command?", # option 2 ending "(esc)". text = ("Would you like to run the following command?\n" "\n" " $ tmux capture-pane -p\n" "\n" + CURSOR + " 1. Yes, proceed (y)\n" " 2. No, and tell Muse Code what to do differently (esc)\n") m = w.find_choice_prompt(text) self.assertIsNotNone(m) self.assertEqual((m["kind"], m["key"]), ("muse-approval", "1")) def test_rejects_decided_block(self): # An already-landed decision must never be double-answered. self.assertIsNone(w.find_choice_prompt(PROMPT_MUSE_APPROVAL_DECIDED)) def test_rejects_cue_without_pair(self): text = "Would you like to run the following\n\n $ foo\n" self.assertIsNone(w.find_choice_prompt(text)) def test_rejects_pair_without_cue(self): text = (CURSOR + " 1. Yes, proceed (y)\n" " 2. No, thanks\n") self.assertIsNone(w.find_choice_prompt(text)) def test_priority_over_yn(self): m = w.find_choice_prompt(PROMPT_MUSE_APPROVAL + "Proceed? (y/n)\n") self.assertIsNotNone(m) self.assertEqual(m["kind"], "muse-approval") def test_sig_namespaced(self): a = w.find_choice_prompt(PROMPT_MUSE_APPROVAL)["sig"] b = w.find_choice_prompt(PROMPT_ABC)["sig"] self.assertNotEqual(a, b) def test_sig_distinguishes_consecutive_approvals(self): # Live stuck-state regression: options+cue are byte-identical # across command approvals, so the sig must include the $ command # or every dialog after the first is swallowed by once-only. other = PROMPT_MUSE_APPROVAL.replace( "capture-pane -p -t %39", "capture-pane -p -t %29") a = w.find_choice_prompt(PROMPT_MUSE_APPROVAL)["sig"] b = w.find_choice_prompt(other)["sig"] self.assertNotEqual(a, b) def test_second_approval_answers_after_first(self): # End-to-end stuck-state regression through the poll loop. other = PROMPT_MUSE_APPROVAL.replace( "capture-pane -p -t %39", "capture-pane -p -t %29") state = w.WatcherState() log = mock.Mock() captures = [PROMPT_MUSE_APPROVAL] * 3 + [other] * 3 with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", side_effect=captures), \ mock.patch.object(w, "send_answer", return_value=True) as send, \ mock.patch.object(w, "audit"): outcomes = [w._poll_once("/tmp/s", "%1", state, log, dry_run=False) for _ in range(4)] self.assertEqual(outcomes, ["seen", "answered", "seen", "answered"]) self.assertEqual(send.call_count, 2) def test_poll_answers_with_1(self): state = w.WatcherState() log = mock.Mock() captures = [PROMPT_MUSE_APPROVAL] * 3 with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", side_effect=captures), \ mock.patch.object(w, "send_answer", return_value=True) as send, \ mock.patch.object(w, "audit") as audit: outcomes = [w._poll_once("/tmp/s", "%1", state, log, dry_run=False) for _ in range(len(captures) - 1)] self.assertEqual(outcomes, ["seen", "answered"]) send.assert_called_once_with("/tmp/s", "%1", "1", enter=True) audit.assert_called_once() self.assertEqual(audit.call_args[1]["extra"]["kind"], "muse-approval") PROMPT_INTERVIEW = ( "The daemon only approves today. Should this step add deny/escalate\n" "decisions informed by helpers, cover more prompt shapes, or both?\n" "\n" + CURSOR + " 1. Deny/escalate policy (Recommended) Keep approving by default.\n" " 2. More prompt shapes Teach the matcher more UIs.\n" " 3. Both Policy plus broader shapes.\n" " 4. None of the above Optionally add notes (tab).\n" ) class TestInterview(unittest.TestCase): """Agent interview UI: cursor + ordered 1./2. menu + question.""" def test_matches_interview(self): m = w.find_choice_prompt(PROMPT_INTERVIEW) self.assertIsNotNone(m) self.assertEqual((m["kind"], m["key"]), ("interview", "1")) self.assertEqual(len(m["options"]), 4) def test_matches_ascii_cursor(self): m = w.find_choice_prompt(PROMPT_INTERVIEW.replace(CURSOR, ">")) self.assertIsNotNone(m) self.assertEqual(m["kind"], "interview") def test_rejects_prose_list_without_cursor(self): # Same shape minus the selection cursor is prose, not a live menu. text = PROMPT_INTERVIEW.replace(CURSOR + " ", " ") self.assertIsNone(w.find_choice_prompt(text)) def test_rejects_single_option(self): text = ("Pick one?\n\n" + CURSOR + " 1. Only choice\n") self.assertIsNone(w.find_choice_prompt(text)) def test_rejects_options_without_question(self): text = ("Some statement here.\n\n" + CURSOR + " 1. First\n" " 2. Second\n") self.assertIsNone(w.find_choice_prompt(text)) def test_approval_wins_over_interview(self): # A native approval dialog also carries dotted options; the more # specific kind must win. m = w.find_choice_prompt(PROMPT_MUSE_APPROVAL) self.assertEqual(m["kind"], "muse-approval") def test_poll_answers_with_1(self): state = w.WatcherState() log = mock.Mock() captures = [PROMPT_INTERVIEW] * 3 with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", side_effect=captures), \ mock.patch.object(w, "send_answer", return_value=True) as send, \ mock.patch.object(w, "audit") as audit: outcomes = [w._poll_once("/tmp/s", "%1", state, log, dry_run=False) for _ in range(len(captures) - 1)] self.assertEqual(outcomes, ["seen", "answered"]) send.assert_called_once_with("/tmp/s", "%1", "1", enter=True) self.assertEqual(audit.call_args[1]["extra"]["kind"], "interview") class TestLaunchOptOut(unittest.TestCase): def test_bare_argv_answers(self): self.assertFalse(w.launch_opt_out(["/x/muse-bin-1.4"])) self.assertFalse(w.launch_opt_out([])) self.assertFalse(w.launch_opt_out(None)) def test_auto_flags_answer(self): self.assertFalse(w.launch_opt_out(["muse", "--yolo"])) self.assertFalse(w.launch_opt_out(["muse", "--disable-approval"])) self.assertFalse( w.launch_opt_out(["muse", "--approval-mode", "never"])) self.assertFalse( w.launch_opt_out(["muse", "--approval-mode=never"])) def test_explicit_mode_holds(self): self.assertTrue( w.launch_opt_out(["muse", "--approval-mode", "on-request"])) self.assertTrue( w.launch_opt_out(["muse", "--approval-mode", "untrusted"])) self.assertTrue( w.launch_opt_out(["muse", "--approval-mode=on-request"])) def test_poll_holds_opt_out_pane(self): state = w.WatcherState() log = mock.Mock() captures = [PROMPT_YN] * 4 with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", side_effect=captures), \ mock.patch.object(w, "pane_muse_argv", return_value=["muse", "--approval-mode", "on-request"]), \ mock.patch.object(w, "send_answer", return_value=True) as send, \ mock.patch.object(w, "audit") as audit: outcomes = [w._poll_once("/tmp/s", "%1", state, log, dry_run=False) for _ in range(len(captures) - 1)] self.assertEqual(outcomes, ["seen", "held", "none"]) send.assert_not_called() audit.assert_not_called() msgs = [c.args[1] for c in log.log.call_args_list] self.assertIn("held: pane opted out via launch flags", msgs) def test_poll_answers_bare_pane(self): state = w.WatcherState() log = mock.Mock() captures = [PROMPT_YN] * 3 with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", side_effect=captures), \ mock.patch.object(w, "pane_muse_argv", return_value=[]), \ mock.patch.object(w, "send_answer", return_value=True) as send, \ mock.patch.object(w, "audit"): outcomes = [w._poll_once("/tmp/s", "%1", state, log, dry_run=False) for _ in range(len(captures) - 1)] self.assertEqual(outcomes, ["seen", "answered"]) send.assert_called_once_with("/tmp/s", "%1", "y", enter=True) def test_pane_muse_argv(self): listing = mock.Mock(returncode=0, stdout="%1 1000\n%2 2000\n", stderr="") def fake_cmdline(pid): if pid == 1001: return ["/x/muse-bin", "--yolo"] return ["/bin/bash"] with mock.patch.object(w, "_tmux", return_value=listing) as t, \ mock.patch.object(w, "_child_pids", return_value=[1001]), \ mock.patch.object(w, "_cmdline", side_effect=fake_cmdline): argv = w.pane_muse_argv("/tmp/s", "%1") self.assertEqual(argv, ["/x/muse-bin", "--yolo"]) t.assert_called_once() def test_pane_muse_argv_missing(self): listing = mock.Mock(returncode=0, stdout="%2 2000\n", stderr="") with mock.patch.object(w, "_tmux", return_value=listing): self.assertEqual(w.pane_muse_argv("/tmp/s", "%1"), []) PROMPT_COLLAPSED_APPROVAL = ( "Would you like to run the following\n" "\n" " $ python3 -m unittest\n" " tests.test_muse_choice_watcher\n" " 2>&1 | tail -n 3 && cp\n" " ʼ 4 command rows omitted\n" " ctrl+o view full command\n" ) class TestCollapsedApproval(unittest.TestCase): """Collapsed approval: long command hides the 1/2 pair; ctrl+o expands.""" def test_matches_collapsed(self): m = w.find_choice_prompt(PROMPT_COLLAPSED_APPROVAL) self.assertIsNotNone(m) self.assertEqual(m["kind"], "muse-approval-collapsed") self.assertEqual(m["key"], "Enter") self.assertFalse(m["enter"]) def test_expanded_pair_beats_collapsed(self): m = w.find_choice_prompt(PROMPT_MUSE_APPROVAL) self.assertIsNotNone(m) self.assertEqual(m["kind"], "muse-approval") def test_rejects_cue_without_markers(self): text = "Would you like to run the following\n\n $ foo\n" self.assertIsNone(w.find_choice_prompt(text)) def test_rejects_markers_without_cue(self): text = (" $ foo bar baz\n" " 4 command rows omitted\n" " ctrl+o view full command\n") self.assertIsNone(w.find_choice_prompt(text)) def test_rejects_decided_collapsed(self): text = PROMPT_COLLAPSED_APPROVAL + "approval decision accepted\n" self.assertIsNone(w.find_choice_prompt(text)) def test_poll_sends_bare_enter(self): state = w.WatcherState() log = mock.Mock() captures = [PROMPT_COLLAPSED_APPROVAL] * 3 calls = [] def fake_tmux(sock, *args, timeout=5): calls.append((sock, args)) r = mock.Mock() r.returncode = 0 return r with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", side_effect=captures), \ mock.patch.object(w, "pane_muse_argv", return_value=[]), \ mock.patch.object(w, "_tmux", side_effect=fake_tmux), \ mock.patch.object(w, "audit") as audit: outcomes = [w._poll_once("/tmp/s", "%1", state, log, dry_run=False) for _ in range(len(captures) - 1)] self.assertEqual(outcomes, ["seen", "answered"]) self.assertEqual(len(calls), 1) self.assertEqual(calls[0][1][:3], ("send-keys", "-t", "%1")) self.assertEqual(calls[0][1][3], "Enter") audit.assert_called_once() self.assertEqual(audit.call_args[1]["extra"]["kind"], "muse-approval-collapsed") PROMPT_EXPLICIT = ( "Reply ACCEPT to approve this text as written (the 2 minutes\n" "included), or amend anything first. Note: accepting the scope\n" "finishes the interview and flips the record to Final.\n" ) class TestExplicitPhrase(unittest.TestCase): """Model asks the user to reply an explicit magic word.""" def test_matches_accept(self): m = w.find_choice_prompt(PROMPT_EXPLICIT) self.assertIsNotNone(m) self.assertEqual((m["kind"], m["key"]), ("explicit-phrase", "ACCEPT")) def test_matches_other_tokens(self): for token in ("YES", "GO", "OK", "CONTINUE", "PROCEED", "ABORT-1"): with self.subTest(token=token): m = w.find_choice_prompt("Reply %s to confirm.\n" % token) self.assertIsNotNone(m) self.assertEqual(m["key"], token) def test_rejects_lowercase_prose(self): self.assertIsNone( w.find_choice_prompt("Please reply soon to confirm.\n")) self.assertIsNone( w.find_choice_prompt("Reply yes please to continue.\n")) def test_rejects_stale(self): text = ("Reply ACCEPT to approve.\n" + "\n".join("filler %d" % i for i in range(12))) self.assertIsNone(w.find_choice_prompt(text)) def test_freshest_wins(self): text = ("Reply YES to confirm.\n" "Some agent chatter.\n" "Reply ACCEPT to approve this text as written.\n") m = w.find_choice_prompt(text) self.assertEqual(m["key"], "ACCEPT") def test_poll_answers_with_token(self): state = w.WatcherState() log = mock.Mock() captures = [PROMPT_EXPLICIT] * 3 with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", side_effect=captures), \ mock.patch.object(w, "send_answer", return_value=True) as send, \ mock.patch.object(w, "audit") as audit: outcomes = [w._poll_once("/tmp/s", "%1", state, log, dry_run=False) for _ in range(len(captures) - 1)] self.assertEqual(outcomes, ["seen", "answered"]) send.assert_called_once_with("/tmp/s", "%1", "ACCEPT", enter=True) self.assertEqual(audit.call_args[1]["extra"]["kind"], "explicit-phrase") class TestRulesEval(unittest.TestCase): def _use_rules(self, tmpdir, data): import json path = tmpdir + "/rules.json" if isinstance(data, str): with open(path, "w") as f: f.write(data) else: with open(path, "w") as f: json.dump(data, f) w._RULES_CACHE = {"key": None, "rules": []} return mock.patch.object(w, "RULES_FILE", path) def _rm_approval(self): return PROMPT_MUSE_APPROVAL.replace( "tmux -S /tmp/tmux-1000/default capture-pane -p -t %39", "rm -rf /tmp/scratch") def test_no_file_approves(self): import tempfile with tempfile.TemporaryDirectory() as td, \ mock.patch.object(w, "RULES_FILE", td + "/missing.json"): w._RULES_CACHE = {"key": None, "rules": []} d, r = w.evaluate_rules( w.find_choice_prompt(PROMPT_ABC)) self.assertEqual((d, r), ("approve", None)) def test_bad_json_fails_open(self): import tempfile with tempfile.TemporaryDirectory() as td: with self._use_rules(td, "{not json"): d, r = w.evaluate_rules( w.find_choice_prompt(PROMPT_ABC), mock.Mock()) self.assertEqual((d, r), ("approve", None)) def test_token_match_holds(self): import tempfile rules = {"rules": [{"id": "t", "kind": "explicit-phrase", "token": ["ABORT", "DELETE"], "decision": "hold"}]} with tempfile.TemporaryDirectory() as td: with self._use_rules(td, rules): m = w.find_choice_prompt("Reply ABORT to cancel.\n") d, r = w.evaluate_rules(m) self.assertEqual(d, "hold") self.assertEqual(r["id"], "t") def test_token_nonmatch_approves(self): import tempfile rules = {"rules": [{"id": "t", "kind": "explicit-phrase", "token": ["ABORT"], "decision": "hold"}]} with tempfile.TemporaryDirectory() as td: with self._use_rules(td, rules): m = w.find_choice_prompt("Reply ACCEPT to approve.\n") d, r = w.evaluate_rules(m) self.assertEqual((d, r), ("approve", None)) def test_command_match_holds(self): import tempfile rules = {"rules": [{"id": "c", "kind": "muse-approval", "command": r"\brm\s+-rf?\b", "decision": "hold"}]} with tempfile.TemporaryDirectory() as td: with self._use_rules(td, rules): m = w.find_choice_prompt(self._rm_approval()) self.assertIsNotNone(m) d, r = w.evaluate_rules(m) self.assertEqual(d, "hold") def test_kind_mismatch_skips(self): import tempfile rules = {"rules": [{"id": "c", "kind": "yn", "command": r"\brm\s+-rf?\b", "decision": "hold"}]} with tempfile.TemporaryDirectory() as td: with self._use_rules(td, rules): m = w.find_choice_prompt(self._rm_approval()) d, r = w.evaluate_rules(m) self.assertEqual((d, r), ("approve", None)) def test_deny_question_skipped(self): import tempfile rules = {"rules": [{"id": "d", "kind": "explicit-phrase", "token": ["ABORT"], "decision": "deny"}]} with tempfile.TemporaryDirectory() as td: with self._use_rules(td, rules): m = w.find_choice_prompt("Reply ABORT to cancel.\n") d, r = w.evaluate_rules(m) self.assertEqual((d, r), ("approve", None)) def test_deny_approval(self): import tempfile rules = {"rules": [{"id": "d", "kind": "muse-approval", "decision": "deny"}]} with tempfile.TemporaryDirectory() as td: with self._use_rules(td, rules): m = w.find_choice_prompt(PROMPT_MUSE_APPROVAL) d, r = w.evaluate_rules(m) self.assertEqual(d, "deny") def test_deny_collapsed_downgrades_to_hold(self): import tempfile rules = {"rules": [{"id": "d", "kind": "muse-approval-collapsed", "decision": "deny"}]} with tempfile.TemporaryDirectory() as td: with self._use_rules(td, rules): m = w.find_choice_prompt(PROMPT_COLLAPSED_APPROVAL) d, r = w.evaluate_rules(m) self.assertEqual(d, "hold") self.assertEqual(r["downgraded_from"], "deny") def test_unknown_decision_approves(self): import tempfile rules = {"rules": [{"id": "x", "decision": "explode"}]} with tempfile.TemporaryDirectory() as td: with self._use_rules(td, rules): d, r = w.evaluate_rules( w.find_choice_prompt(PROMPT_ABC)) self.assertEqual(d, "approve") def test_bad_regex_never_matches(self): import tempfile rules = {"rules": [{"id": "x", "command": "[invalid", "decision": "hold"}]} with tempfile.TemporaryDirectory() as td: with self._use_rules(td, rules): d, r = w.evaluate_rules( w.find_choice_prompt(PROMPT_MUSE_APPROVAL)) self.assertEqual((d, r), ("approve", None)) class TestHoldFlow(unittest.TestCase): def _polls(self, tmpdir, text, n, hold_rule=True): import time state = w.WatcherState() log = mock.Mock() rule = {"id": "t", "reason": "test hold"} eff = ("hold", rule) if hold_rule else ("approve", None) with mock.patch.object(w, "STATE_DIR", tmpdir), \ mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", side_effect=[text] * (2 * n)), \ mock.patch.object(w, "send_answer", return_value=True) as send, \ mock.patch.object(w, "audit") as audit, \ mock.patch.object(w, "evaluate_rules", return_value=eff): outcomes = [w._poll_once("/tmp/s", "%1", state, log, dry_run=False) for _ in range(n)] return state, log, send, audit, outcomes def test_hold_suppresses_and_writes_file(self): import tempfile with tempfile.TemporaryDirectory() as td: state, log, send, audit, outcomes = self._polls( td, PROMPT_YN, 2) self.assertEqual(outcomes, ["seen", "held"]) send.assert_not_called() held_calls = [c for c in audit.call_args_list if c.args[0] == "muse-choice-held"] self.assertEqual(len(held_calls), 1) def test_held_persists_without_reaudit(self): import tempfile with tempfile.TemporaryDirectory() as td: state, log, send, audit, outcomes = self._polls( td, PROMPT_YN, 4) self.assertEqual(outcomes, ["seen", "held", "held", "held"]) send.assert_not_called() held_calls = [c for c in audit.call_args_list if c.args[0] == "muse-choice-held"] self.assertEqual(len(held_calls), 1) def test_resolve_approve_releases(self): import tempfile with tempfile.TemporaryDirectory() as td: with mock.patch.object(w, "STATE_DIR", td): state, log, send, audit, outcomes = self._polls( td, PROMPT_YN, 2) hf = w.read_hold("/tmp/s", "%1") self.assertIsNotNone(hf) hf["directive"] = "approve" # what box resolve does w.write_hold("/tmp/s", "%1", hf) with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", return_value=PROMPT_YN), \ mock.patch.object(w, "send_answer", return_value=True) as send2, \ mock.patch.object(w, "audit") as audit2, \ mock.patch.object(w, "evaluate_rules") as ev: out = w._poll_once("/tmp/s", "%1", state, log, dry_run=False) self.assertEqual(out, "answered") send2.assert_called_once_with("/tmp/s", "%1", "y", enter=True) ev.assert_not_called() # release bypasses re-evaluation def test_resolve_deny_sends_negative(self): import tempfile with tempfile.TemporaryDirectory() as td: with mock.patch.object(w, "STATE_DIR", td): state, log, send, audit, outcomes = self._polls( td, PROMPT_YN, 2) hf = w.read_hold("/tmp/s", "%1") hf["directive"] = "deny" w.write_hold("/tmp/s", "%1", hf) with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", return_value=PROMPT_YN), \ mock.patch.object(w, "send_answer", return_value=True) as send2, \ mock.patch.object(w, "audit"): out = w._poll_once("/tmp/s", "%1", state, log, dry_run=False) self.assertEqual(out, "denied") send2.assert_called_once_with("/tmp/s", "%1", "n", enter=True) def test_resolve_deny_refused_on_questions(self): import tempfile text = "Reply ABORT to cancel.\n" with tempfile.TemporaryDirectory() as td: with mock.patch.object(w, "STATE_DIR", td): state, log, send, audit, outcomes = self._polls( td, text, 2) hf = w.read_hold("/tmp/s", "%1") hf["directive"] = "deny" w.write_hold("/tmp/s", "%1", hf) with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", return_value=text), \ mock.patch.object(w, "send_answer", return_value=True) as send2, \ mock.patch.object(w, "audit"): out = w._poll_once("/tmp/s", "%1", state, log, dry_run=False) self.assertIsNone(w.read_hold("/tmp/s", "%1") ["directive"]) self.assertEqual(out, "held") send2.assert_not_called() def test_expiry_releases_to_approve(self): import tempfile import time with tempfile.TemporaryDirectory() as td: with mock.patch.object(w, "STATE_DIR", td): state, log, send, audit, outcomes = self._polls( td, PROMPT_YN, 2) hf = w.read_hold("/tmp/s", "%1") hf["held_until"] = time.time() - 1 w.write_hold("/tmp/s", "%1", hf) with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", return_value=PROMPT_YN), \ mock.patch.object(w, "send_answer", return_value=True) as send2, \ mock.patch.object(w, "audit") as audit2, \ mock.patch.object(w, "evaluate_rules") as ev: out = w._poll_once("/tmp/s", "%1", state, log, dry_run=False) self.assertEqual(out, "answered") send2.assert_called_once() ev.assert_not_called() expired = [c for c in audit2.call_args_list if c.args[0] == "muse-choice-hold-expired"] self.assertEqual(len(expired), 1) def test_stale_hold_cleared(self): import tempfile with tempfile.TemporaryDirectory() as td: with mock.patch.object(w, "STATE_DIR", td): state, log, send, audit, outcomes = self._polls( td, PROMPT_YN, 2) self.assertIsNotNone(w.read_hold("/tmp/s", "%1")) with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", return_value=PROMPT_ABC), \ mock.patch.object(w, "send_answer", return_value=True), \ mock.patch.object(w, "audit"), \ mock.patch.object(w, "evaluate_rules", return_value=("approve", None)): out = w._poll_once("/tmp/s", "%1", state, log, dry_run=False) self.assertIsNone(w.read_hold("/tmp/s", "%1")) self.assertEqual(out, "seen") class TestBoxResolve(unittest.TestCase): @classmethod def setUpClass(cls): import importlib.util spec = importlib.util.spec_from_file_location( "supercli_test", str(BIN_DIR / "super-cli.py")) cls.cli = importlib.util.module_from_spec(spec) spec.loader.exec_module(cls.cli) def _ns(self, decision): import argparse return argparse.Namespace(mc_action="resolve", socket="/tmp/s", pane="%1", decision=decision, json=True) def _hold(self, tmpdir, kind="yn"): import time with mock.patch.object(w, "STATE_DIR", tmpdir): w.write_hold("/tmp/s", "%1", {"sig": "s1", "kind": kind, "key": "y", "text": "t", "rule": "r", "held_until": time.time() + 60, "directive": None}) def _run(self, tmpdir, decision): import io import json from contextlib import redirect_stdout buf = io.StringIO() with mock.patch.object(w, "STATE_DIR", tmpdir), \ mock.patch.object(w, "audit"), \ redirect_stdout(buf): self.cli.cmd_muse_choices(self._ns(decision)) return json.loads(buf.getvalue()) def test_resolve_approve_sets_directive(self): import tempfile with tempfile.TemporaryDirectory() as td: self._hold(td) data = self._run(td, "approve") self.assertTrue(data["ok"]) with mock.patch.object(w, "STATE_DIR", td): hf = w.read_hold("/tmp/s", "%1") self.assertEqual(hf["directive"], "approve") def test_resolve_deny_sets_directive(self): import tempfile with tempfile.TemporaryDirectory() as td: self._hold(td, kind="yn") data = self._run(td, "deny") self.assertTrue(data["ok"]) with mock.patch.object(w, "STATE_DIR", td): hf = w.read_hold("/tmp/s", "%1") self.assertEqual(hf["directive"], "deny") def test_resolve_deny_refused_on_questions(self): import tempfile with tempfile.TemporaryDirectory() as td: self._hold(td, kind="interview") data = self._run(td, "deny") self.assertFalse(data["ok"]) self.assertIn("D2", data["reason"]) def test_resolve_nothing_held(self): import tempfile with tempfile.TemporaryDirectory() as td: data = self._run(td, "approve") self.assertTrue(data["ok"]) self.assertIsNone(data["held"]) class TestPeerAnsweredReload(unittest.TestCase): """A peer watcher answering first must suppress this watcher. Concurrent duplicates share only the on-disk answered store: a second watcher that loaded before the peer's answer must re-check the store before typing, or both type '1' ('11' in the input box). """ def test_disk_answer_suppresses_send(self): import json import tempfile import time with tempfile.TemporaryDirectory() as td: path = td + "/x.answered.json" state = w.WatcherState(persist_path=path) log = mock.Mock() sig = w.find_choice_prompt(PROMPT_YN)["sig"] with mock.patch.object(w, "pane_exists", return_value=True), \ mock.patch.object(w, "capture_pane", side_effect=[PROMPT_YN] * 4), \ mock.patch.object(w, "send_answer", return_value=True) as send, \ mock.patch.object(w, "audit"): out1 = w._poll_once("/tmp/s", "%1", state, log, dry_run=False) self.assertEqual(out1, "seen") # Peer answers the same prompt and persists first. with open(path, "w") as f: json.dump({sig: time.time()}, f) out2 = w._poll_once("/tmp/s", "%1", state, log, dry_run=False) send.assert_not_called() self.assertIn(out2, ("duplicate-suppressed", "none")) self.assertIn(sig, state.answered_sigs) class TestCaptureJoinWrapped(unittest.TestCase): def test_capture_joins_wrapped_lines(self): with mock.patch.object(w, "_tmux") as t: t.return_value = mock.Mock(returncode=0, stdout="ok") w.capture_pane("/tmp/s", "%1") args = t.call_args[0] self.assertIn("-J", args) class TestLogPosture(unittest.TestCase): """Watcher start records the pane's permission posture. The watcher answers with per-choice logging in every mode; bypass postures (yolo / approval disabled) additionally audit, since the session then acts outside the choice trail. """ def test_bypass_audits(self): log = mock.Mock() with mock.patch.object(w, "pane_muse_argv", return_value=["muse", "--yolo"]), \ mock.patch.object(w, "audit") as audit: w._log_posture("/tmp/s", "%1", log) log.log.assert_called_once() self.assertEqual(log.log.call_args[0][1], "pane posture") self.assertEqual(log.log.call_args[1]["mode"], "yolo") audit.assert_called_once() self.assertEqual(audit.call_args[0][0], "muse-choice-posture") self.assertTrue(audit.call_args[1]["extra"]["bypass"]) def test_default_logs_without_audit(self): log = mock.Mock() with mock.patch.object(w, "pane_muse_argv", return_value=["muse"]), \ mock.patch.object(w, "audit") as audit: w._log_posture("/tmp/s", "%1", log) log.log.assert_called_once() audit.assert_not_called() def test_nondefault_profile_audits(self): log = mock.Mock() with mock.patch.object(w, "pane_muse_argv", return_value=["muse", "--permission-profile", ":unrestricted"]), \ mock.patch.object(w, "audit") as audit: w._log_posture("/tmp/s", "%1", log) audit.assert_called_once() extra = audit.call_args[1]["extra"] self.assertEqual(extra["mode"], ":unrestricted") self.assertFalse(extra["bypass"]) def test_argv_failure_never_raises(self): log = mock.Mock() with mock.patch.object(w, "pane_muse_argv", side_effect=RuntimeError("tmux")), \ mock.patch.object(w, "audit") as audit: w._log_posture("/tmp/s", "%1", log) # must not raise log.log.assert_not_called() audit.assert_not_called() class TestReconcileOrphan(unittest.TestCase): """Reconcile must not start a second watcher when an orphan with a lost pidfile is already alive for the pane.""" def test_orphan_counts_as_already(self): orphan = [{"pid": 7777, "socket": "/tmp/sock", "pane": "%37"}] with mock.patch.object(w, "get_desired", return_value={"enabled": True, "dry_run": False}), \ mock.patch.object(w, "muse_panes", return_value=["%37"]), \ mock.patch.object(w, "is_running", return_value=None), \ mock.patch.object(w, "_watch_procs", return_value=orphan), \ mock.patch.object(w, "_start_detached") as start, \ mock.patch.object(w, "status_all", return_value=[]), \ mock.patch.object(w, "audit"), \ mock.patch.object(w.os.path, "exists", return_value=True): res = w.reconcile(sockets=["/tmp/sock"]) start.assert_not_called() self.assertEqual(res["already"], ["/tmp/sock:%37"]) self.assertEqual(res["failed"], []) if __name__ == "__main__": unittest.main()