#!/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 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") if __name__ == "__main__": unittest.main()