super/box
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f2527f183c3033905a9ef3c04621ea099bdbb4e6
box/tests/test_muse_choice_watcher.py
T

1456 lines
60 KiB
Python
Raw Normal View History

feat(supervision): add choice watcher daemon, HTTPS spec docs, and test suites
2026-10-07 00:25:46 +00:00
#!/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")
feat(systemd): add and enable continuous tmux-auto-approver user daemon
2026-10-07 00:49:23 +00:00
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"])
feat(supervision): add choice watcher daemon, HTTPS spec docs, and test suites
2026-10-07 00:25:46 +00:00
if __name__ == "__main__":
unittest.main()
Reference in New Issue Copy Permalink