From e25d2cf4cca721eeba5105e21a4278e6464aa1a7 Mon Sep 17 00:00:00 2001 From: operator Date: Wed, 7 Oct 2026 00:49:23 +0000 Subject: [PATCH] feat(systemd): add and enable continuous tmux-auto-approver user daemon --- .agents/skills/box/SKILL.md | 2 +- NODES.md | 13 + bin/hatch_menu/controls.py | 39 +-- bin/hatch_menu/tabs/data_controls.py | 57 ++++- bin/hatch_menu/toggles.py | 8 +- bin/muse_choice_watcher.py | 301 +++++++++++++++++++++- bin/super-cli.py | 68 ++++- docs/HATCH-MENU.md | 7 + docs/SUPERVISION-SPEC.md | 41 ++- tests/test_box_runtime.py | 29 +++ tests/test_hatch_menu.py | 47 ++++ tests/test_muse_choice_watcher.py | 365 +++++++++++++++++++++++++++ 12 files changed, 915 insertions(+), 62 deletions(-) diff --git a/.agents/skills/box/SKILL.md b/.agents/skills/box/SKILL.md index b5f7fdc..be4d84a 100644 --- a/.agents/skills/box/SKILL.md +++ b/.agents/skills/box/SKILL.md @@ -32,7 +32,7 @@ Add `--json` to any command for machine-readable output when parsing results in - `box lookup summary|fleet|threads|unread|approvals` — seamless one-shot lookups. - `box job list` / `box job log` — scheduled jobs and execution events. - `box harvest status` / `box followup list` — harvest watermarks / pending nudges. -- `box muse-choices on|off|status|logs|reconcile` — Muse TUI auto-answer daemon switch, state, and per-pane logs (default on; `off` is the box-command opt-out). +- `box muse-choices on|off|status|logs|reconcile|resolve` — Muse TUI auto-answer daemon switch, state, per-pane logs, held-prompt resolve (default on; `off` is the box-command opt-out). - `box runtime list|send|launch|layout|spread` — Muse CLI tmux runtimes: live state + approval posture, send-keys input, auto-approved launches, pane-geometry layout + spread for squeezed panes. - `box tmux tally` / `box tmux auto [status|on|off|watch|once|logs|match]` — multi-socket Tmux worker tally, regex auto-approver daemon & guardrails. - `box onboard connects` / `box onboard-tui` — fleet & client onboarding inventory, CDP ports, OTP salvage & 4-surface TUI. diff --git a/NODES.md b/NODES.md index 8b90faa..0d22fe1 100644 --- a/NODES.md +++ b/NODES.md @@ -14,3 +14,16 @@ Unified naming: node == agent == profile == API account. | opm | warp-opm | 104.28.195.181 | 9440 | active | opm (artglobal.cc@gmail.com, email_otp, Nico Parada) | | def | warp-def | 104.28.195.181 | 9450 | active | def (defnotabotnet@gmail.com, email_otp, IG paradahub) | | dev | warp-dev | 104.28.195.181 | 9455 | active | dev (paradaproduced@gmail.com, email_otp, IG veryraremeta) | + +## Session naming (supervision) + +Remote nodes issue jobs to this box; execution lives on the shared +stable tmux server, separated by session name (not by socket): + +`----` — e.g. `pip--worker--01`, `muse--repair--09`. + +Roles: `worker` (persistent swarm/daemon), `repair` (fix sessions), +`watch` (auto-approver tails), `runtime` (interactive CLI). Ad-hoc +sessions carry no node and show `-` in `box runtime list`. Session +creators owned by existing flows keep their names until owners rename; +new sessions should follow the convention from birth. diff --git a/bin/hatch_menu/controls.py b/bin/hatch_menu/controls.py index b3594a0..18ed27a 100644 --- a/bin/hatch_menu/controls.py +++ b/bin/hatch_menu/controls.py @@ -138,29 +138,6 @@ JS_CLICK_SWITCH = """((label) => { return 'CLICKED'; })('%s')""" -JS_SWITCH_RECT = """((label) => { - const d = document.querySelector('[role="dialog"]'); - if (!d) return null; - const rowOf = (s) => { - let el = s.parentElement, depth = 0; - while (el && el !== d && depth < 6) { - const t = (el.innerText || '').trim(); - if (t && t.length < 250) return t.toLowerCase(); - el = el.parentElement; - depth += 1; - } - return ''; - }; - const hits = Array.from(d.querySelectorAll('[role="switch"]')) - .filter(s => rowOf(s).includes(label.toLowerCase()) - || (s.getAttribute('aria-label') || '').toLowerCase() - .includes(label.toLowerCase())); - if (hits.length !== 1) return null; - const r = hits[0].getBoundingClientRect(); - return {x: r.x + r.width / 2, y: r.y + r.height / 2}; -})('%s')""" - - def _eval(ws, js, timeout=8.0): try: return cdp_evaluate(ws, js, timeout=timeout) @@ -256,20 +233,16 @@ def switch_state(ws, label): def set_switch(ws, label, on): - """Set a switch by row-label/aria match; no-op when already there.""" + """Set a switch by row-label/aria match; no-op when already there. + + Single click only: a fallback re-click would UNDO a slow commit + (switches toggle). The fresh-session readback decides. + """ state = switch_state(ws, label) if state is None: return False if state == bool(on): return True - if _eval(ws, JS_CLICK_SWITCH % label) == "CLICKED" \ - and _poll(lambda: switch_state(ws, label) is bool(on)): - return True - rect = _eval(ws, JS_SWITCH_RECT % label) - if not rect or "x" not in rect: - return False - try: - real_click(ws, rect["x"], rect["y"]) - except Exception: + if _eval(ws, JS_CLICK_SWITCH % label) != "CLICKED": return False return _poll(lambda: switch_state(ws, label) is bool(on)) diff --git a/bin/hatch_menu/tabs/data_controls.py b/bin/hatch_menu/tabs/data_controls.py index b8fff07..1b46364 100644 --- a/bin/hatch_menu/tabs/data_controls.py +++ b/bin/hatch_menu/tabs/data_controls.py @@ -1,15 +1,49 @@ """Data controls tab: model-improvement switch (read-only otherwise). The switch label is pinned from live recon; resolution prefers it -and falls back to single-switch, then keyword match. Import/Delete -rows are inventoried, never touched. +and falls back to single-switch, then keyword match. Sets use a +trusted click: synthetic clicks are proven no-ops here (2026-10-06). +Import/Delete rows are inventoried, never touched. """ +import time + +from approvals import cdp_evaluate from hatch_menu import controls, dialog +from hatch_menu.mouse import real_click TAB = "Data controls" SWITCH_LABEL = "Help improve our AI models" _KEYWORDS = ("improv", "train", "model", "data", "usage") +JS_AI_RECT_TMPL = """((label) => { + const d = document.querySelector('[role="dialog"]'); + if (!d) return null; + const rowOf = (s) => { + let el = s.parentElement, depth = 0; + while (el && el !== d && depth < 6) { + const t = (el.innerText || '').trim(); + if (t && t.length < 250) return t.toLowerCase(); + el = el.parentElement; + depth += 1; + } + return ''; + }; + const hits = Array.from(d.querySelectorAll('[role="switch"]')) + .filter(s => rowOf(s).includes(label.toLowerCase()) + || (s.getAttribute('aria-label') || '').toLowerCase() + .includes(label.toLowerCase())); + if (hits.length !== 1) return null; + const r = hits[0].getBoundingClientRect(); + return {x: r.x + r.width / 2, y: r.y + r.height / 2}; +})('%s')""" + + +def _eval(ws, js, timeout=8.0): + try: + return cdp_evaluate(ws, js, timeout=timeout) + except Exception: + return None + def _resolve(ws): """The improvement switch dict, or None when not resolvable.""" @@ -39,15 +73,24 @@ def ai_improvement(ws): def set_ai_improvement(ws, on): - """Set the improvement switch; verify-then-fallback. Bool.""" + """Set via one trusted click; synthetic clicks are no-ops. Bool.""" sw = _resolve(ws) if sw is None: return False - label = (sw.get("aria") or "").strip() \ - or (sw.get("label") or "")[:40].strip() - if not label: + if bool(sw.get("checked")) == bool(on): + return True + rect = _eval(ws, JS_AI_RECT_TMPL % SWITCH_LABEL) + if not rect or "x" not in rect: return False - return controls.set_switch(ws, label, on) + try: + real_click(ws, rect["x"], rect["y"]) + except Exception: + return False + for _ in range(8): + time.sleep(2.0) + if ai_improvement(ws) is bool(on): + return True + return False def describe(ws): diff --git a/bin/hatch_menu/toggles.py b/bin/hatch_menu/toggles.py index f7f45f8..3d116ac 100644 --- a/bin/hatch_menu/toggles.py +++ b/bin/hatch_menu/toggles.py @@ -44,7 +44,10 @@ TOGGLES = { "values": ("on", "off")}, "data_controls.ai_improvement": { "tab": "Data controls", "kind": "switch", - "values": ("on", "off")}, + "values": ("on", "off"), + # Live 2026-10-06: the site ignores every input gesture here + # (synthetic/real/double/hover/keyboard/drag) — reads fine. + "readonly": True}, "general.theme": { "tab": "General", "kind": "radio-aria", "values": _GEN.THEME_VALUES}, @@ -200,6 +203,9 @@ def get_toggle(node, name): def set_toggle(node, name, value): """Set one toggle with readback. Never partially reports success.""" spec = resolve_toggle(name) + if spec.get("readonly"): + raise MenuError("%s is read-only: the site ignores all input " + "gestures there (locked?)" % name) want = _normalize_value(spec, value) _check_node(node) try: diff --git a/bin/muse_choice_watcher.py b/bin/muse_choice_watcher.py index 97cfa46..4e7317d 100755 --- a/bin/muse_choice_watcher.py +++ b/bin/muse_choice_watcher.py @@ -53,6 +53,11 @@ POLL_INTERVAL = 0.5 STABILITY_POLLS = 2 MAX_ANSWERS_PER_HOUR = 20 ANSWERED_TTL_SECONDS = 600 # identical prompt back after 10m => stuck, allow one recovery answer +RULES_FILE = os.path.join(REPO_ROOT, "muse-choices-rules.json") +HOLD_WINDOW_SECONDS = 120 # D3: short hold window, then expire to approve +NEGATIVE_KEYS = {"muse-approval": "2", "yn": "n"} # D2 deny keys +QUESTION_KINDS = frozenset({"interview", "letter", "numbered", "explicit-phrase"}) +_RULES_CACHE = {"key": None, "rules": []} LOG_MAX_BYTES = 1_000_000 HEARTBEAT_SECONDS = 60 TAIL_WINDOW = 25 # only prompts in the last N lines count (no stale scrollback) @@ -420,7 +425,8 @@ def _find_muse_approval(window): context = [ln.strip()[:120] for ln in window[cue_idx:no_idx + 1]] return {"sig": _sig_for("muse-approval", context), "kind": "muse-approval", "key": "1", "options": options, - "cue": cue, "start": cue_idx, "end": no_idx} + "cue": cue, "context": context, + "start": cue_idx, "end": no_idx} def _find_collapsed_approval(window): @@ -464,7 +470,7 @@ def _find_collapsed_approval(window): return {"sig": _sig_for("muse-approval-collapsed", context), "kind": "muse-approval-collapsed", "key": "Enter", "enter": False, "options": options, "cue": cue, - "start": cue_idx, "end": end} + "context": context, "start": cue_idx, "end": end} def _find_interview(window): @@ -687,6 +693,27 @@ def pane_muse_argv(socket_path, pane_id): MIN_APPROVAL_WIDTH = 40 MIN_APPROVAL_HEIGHT = 12 +# Session naming convention (see NODES.md): ---- +# separates node runtimes on the shared stable server, e.g. +# pip--worker--01. Ad-hoc sessions carry no node and show "-". +NODE_NAMES = ("muse", "pip", "646", "opm", "def", "dev") +NODE_SESSION_RE = re.compile(r"^([A-Za-z0-9]+)--([A-Za-z0-9-]+)--([A-Za-z0-9]+)$") + + +def node_from_session(session_name): + """Node owning a tmux session per the naming convention. + + Returns the node name, or None for ad-hoc sessions and unknown + node prefixes. Pure function for supervision display. + """ + if not session_name: + return None + m = NODE_SESSION_RE.match(session_name) + if not m: + return None + node = m.group(1).lower() + return node if node in NODE_NAMES else None + def runtime_rows(socket_path): """One row per pane: identity, approval posture, live state, watcher. @@ -720,6 +747,7 @@ def runtime_rows(socket_path): squeezed = (pane_width is not None and pane_height is not None and (pane_width < MIN_APPROVAL_WIDTH or pane_height < MIN_APPROVAL_HEIGHT)) + node = node_from_session(session) is_muse = "muse-bin" in cmd or "muse-code" in cmd posture = {"auto_approve": None, "flags": []} if is_muse and pane_pid: @@ -740,6 +768,7 @@ def runtime_rows(socket_path): "socket": socket_path, "session": session, "window": window, "pane": pane_id, "cmd": cmd, "pid": pane_pid, "is_muse": is_muse, + "node": node, "width": pane_width, "height": pane_height, "squeezed": squeezed, "auto_approve": posture["auto_approve"], @@ -882,6 +911,214 @@ def send_answer(socket_path, pane_id, letter="A", enter=True): return False +def _load_rules(log=None): + """Load the D0 rules dictionary, cached by (mtime, size). Never raises: + a missing or broken file means no rules (fail open per D4).""" + global _RULES_CACHE + try: + st = os.stat(RULES_FILE) + key = (st.st_mtime, st.st_size) + except OSError: + key = "missing" + if _RULES_CACHE["key"] == key: + return _RULES_CACHE["rules"] + rules = [] + if key != "missing": + try: + with open(RULES_FILE) as f: + data = json.load(f) + rules = [r for r in data.get("rules", []) + if isinstance(r, dict) and r.get("id")] + except Exception as e: + if log is not None: + try: + log.log("warn", "rules file unreadable, failing open", + error="%r" % (e,)) + except Exception: + pass + rules = [] + _RULES_CACHE = {"key": key, "rules": rules} + return rules + + +def _as_list(v): + return v if isinstance(v, list) else [v] + + +def _rule_matches(rule, match): + kind = rule.get("kind") + if kind is not None and match["kind"] not in _as_list(kind): + return False + token = rule.get("token") + if token is not None: + if match["kind"] != "explicit-phrase": + return False + if match["key"] not in _as_list(token): + return False + cmd = rule.get("command") + if cmd is not None: + ctx = match.get("context") or match.get("options", []) + try: + if not re.search(cmd, "\n".join(ctx)): + return False + except re.error: + return False + text = rule.get("text") + if text is not None: + hay = "\n".join(match.get("options", []) + [match.get("cue") or ""]) + try: + if not re.search(text, hay): + return False + except re.error: + return False + return True + + +def evaluate_rules(match, log=None): + """First matching rule wins -> (decision, rule). No match -> approve. + + D2: deny rules are skipped for question kinds (questions always + resolve top-choice). Collapsed approvals have no visible No option, + so deny downgrades to hold there rather than auto-approving + something a rule flagged risky. + """ + for rule in _load_rules(log=log): + if not _rule_matches(rule, match): + continue + d = rule.get("decision", "approve") + if d not in ("approve", "deny", "hold"): + d = "approve" + if d == "deny": + if match["kind"] in NEGATIVE_KEYS: + return "deny", rule + if match["kind"] in QUESTION_KINDS: + continue + downgraded = dict(rule) + downgraded["downgraded_from"] = "deny" + return "hold", downgraded + return d, rule + return "approve", None + + +def holdfile_for(socket_path, pane_id): + return os.path.join( + STATE_DIR, "%s-%s-%s.held.json" % (FILE_PREFIX, slug_socket(socket_path), + clean_pane(pane_id))) + + +def write_hold(socket_path, pane_id, rec): + try: + with open(holdfile_for(socket_path, pane_id), "w") as f: + json.dump(rec, f) + return True + except Exception: + return False + + +def read_hold(socket_path, pane_id): + try: + with open(holdfile_for(socket_path, pane_id)) as f: + rec = json.load(f) + return rec if isinstance(rec, dict) and rec.get("sig") else None + except Exception: + return None + + +def clear_hold(socket_path, pane_id): + try: + os.remove(holdfile_for(socket_path, pane_id)) + except OSError: + pass + + +def list_holds(): + """All holdfiles, pruning ones long expired with no watcher to own them.""" + out = [] + try: + names = os.listdir(STATE_DIR) + except OSError: + return out + now = time.time() + for name in names: + if not name.startswith(FILE_PREFIX + "-") or not name.endswith(".held.json"): + continue + path = os.path.join(STATE_DIR, name) + try: + with open(path) as f: + rec = json.load(f) + except Exception: + continue + if not isinstance(rec, dict) or not rec.get("sig"): + continue + try: + expired_ago = now - float(rec.get("held_until", 0)) + except (TypeError, ValueError): + expired_ago = 0 + if expired_ago > 300: + try: + os.remove(path) + except OSError: + pass + continue + rec["_file"] = name + out.append(rec) + return out + + +def _check_hold(socket_path, pane_id, state, match, now, log): + """Rule-hold gate. Returns None (fresh: evaluate rules), "released" + (hold over: approve WITHOUT re-evaluating, else expiry would + re-hold forever), "held", or "denied". + + The holdfile is the single source of truth (crash-safe, + box-visible): present + fresh => suppress; directive deny => deny + now; expired or operator-cleared => release back to approve. + """ + sig = match["sig"] + hf = read_hold(socket_path, pane_id) + if hf is None: + return None + if hf.get("sig") != sig: + clear_hold(socket_path, pane_id) # stale hold for a gone prompt + return None + if hf.get("directive") == "approve": + clear_hold(socket_path, pane_id) + log.log("info", "hold released by operator, answering", sig=sig) + return "released" + if hf.get("directive") == "deny": + neg = NEGATIVE_KEYS.get(match["kind"]) + if neg is None: + hf["directive"] = None + write_hold(socket_path, pane_id, hf) + log.log("warn", "resolve-deny refused: D2 never denies questions", + sig=sig, kind=match["kind"]) + return "held" + ok = send_answer(socket_path, pane_id, neg, enter=True) + clear_hold(socket_path, pane_id) + state.record_answer(sig, now) + log.log("info" if ok else "error", "denied %s (operator resolve)" % neg, + sig=sig, ok=ok, kind=match["kind"], key=neg) + audit("muse-choice-denied", + name="%s:%s" % (os.path.basename(socket_path), pane_id), + extra={"socket": socket_path, "pane": pane_id, "sig": sig, + "ok": ok, "kind": match["kind"], "key": neg, + "via": "resolve"}) + return "denied" + try: + expired = now >= float(hf.get("held_until", 0)) + except (TypeError, ValueError): + expired = True + if expired: + clear_hold(socket_path, pane_id) + log.log("info", "hold expired, releasing to approve", sig=sig) + audit("muse-choice-hold-expired", + name="%s:%s" % (os.path.basename(socket_path), pane_id), + extra={"socket": socket_path, "pane": pane_id, "sig": sig, + "kind": match["kind"], "rule": hf.get("rule")}) + return "released" + return "held" + + def _poll_once(socket_path, pane_id, state, log, dry_run=False): """Run one poll iteration. Returns an outcome string ("gone" tells the caller to exit). Logs only state transitions, never per-poll spam.""" @@ -893,6 +1130,14 @@ def _poll_once(socket_path, pane_id, state, log, dry_run=False): return "capture-failed" match = find_choice_prompt(text) now = time.time() + gate = _check_hold(socket_path, pane_id, state, match, now, log) \ + if match is not None else None + if gate in ("held", "denied"): + if gate == "denied": + state.pending_sig = None + state.stable_count = 0 + return gate + skip_eval = (gate == "released") verdict = state.observe(match, now) if verdict == "wait" and state.stable_count == 1: log.log("info", "prompt seen", kind=match["kind"], key=match["key"], @@ -915,7 +1160,56 @@ def _poll_once(socket_path, pane_id, state, log, dry_run=False): sig=match["sig"], kind=match["kind"]) state.record_answer(match["sig"], now) return "held" + if skip_eval: + decision, rule = "approve", None + via = "hold-released" + else: + decision, rule = evaluate_rules(match, log) + via = "direct" key = match["key"] + if decision == "deny": + neg = NEGATIVE_KEYS[match["kind"]] + if dry_run: + log.log("info", "dry-run would deny %s" % neg, + sig=match["sig"], kind=match["kind"], key=neg, + rule=rule["id"] if rule else None) + state.record_answer(match["sig"], now) + return "dry-denied" + ok = send_answer(socket_path, pane_id, neg, enter=True) + state.record_answer(match["sig"], now) + log.log("info" if ok else "error", "denied %s" % neg, + sig=match["sig"], ok=ok, kind=match["kind"], key=neg, + rule=rule["id"] if rule else None) + audit("muse-choice-denied", + name="%s:%s" % (os.path.basename(socket_path), pane_id), + extra={"socket": socket_path, "pane": pane_id, + "sig": match["sig"], "ok": ok, "kind": match["kind"], + "key": neg, "via": "rule", + "rule": rule["id"] if rule else None}) + return "denied" + if decision == "hold": + until = now + HOLD_WINDOW_SECONDS + write_hold(socket_path, pane_id, { + "sig": match["sig"], "kind": match["kind"], "key": key, + "text": (match["cue"] or "")[:200], + "rule": rule["id"] if rule else None, + "reason": (rule.get("reason") if rule else None) or "", + "downgraded_from": (rule.get("downgraded_from") + if rule else None), + "held_until": until, "directive": None, + "socket": socket_path, "pane": pane_id}) + log.log("info", "held by rule %s" % (rule["id"] if rule else "?"), + sig=match["sig"], kind=match["kind"], + rule=rule["id"] if rule else None, + reason=(rule.get("reason") if rule else None) or "") + if not dry_run: + audit("muse-choice-held", + name="%s:%s" % (os.path.basename(socket_path), pane_id), + extra={"socket": socket_path, "pane": pane_id, + "sig": match["sig"], "kind": match["kind"], + "rule": rule["id"] if rule else None, + "held_until": until}) + return "held" if dry_run: log.log("info", "dry-run would answer %s" % key, sig=match["sig"], kind=match["kind"], key=key, @@ -933,7 +1227,8 @@ def _poll_once(socket_path, pane_id, state, log, dry_run=False): extra={"socket": socket_path, "pane": pane_id, "sig": match["sig"], "ok": ok, "kind": match["kind"], "key": key, "options": match["options"], - "cue": match["cue"]}) + "cue": match["cue"], "via": via, + "rule": rule["id"] if rule else None}) return "answered" if verdict == "capped": if state.last_capped_sig != match["sig"]: diff --git a/bin/super-cli.py b/bin/super-cli.py index a5c4331..eafa53d 100755 --- a/bin/super-cli.py +++ b/bin/super-cli.py @@ -953,8 +953,8 @@ def cmd_runtime(args): print(c_dim(" No panes found.")) print() return - headers = ["SOCKET", "SESSION", "PANE", "CMD", "STATE", - "APPROVE", "WATCHER"] + headers = ["SOCKET", "SESSION", "NODE", "PANE", "CMD", + "STATE", "APPROVE", "WATCHER"] table = [] for r in rows: state = r["state"] @@ -970,6 +970,7 @@ def cmd_runtime(args): if r["watcher_alive"] else badge_dim("-")) table.append([os.path.basename(r["socket"]), "%s:%s" % (r["session"], r["window"]), + r["node"] or badge_dim("-"), r["pane"], (r["cmd"] or "")[:26], state, approve, watcher]) print_table(headers, table) @@ -1149,8 +1150,12 @@ def cmd_runtime(args): spread, failed = [], [] for t in targets: name = "spread-%s" % t["pane"] + # -t session: pins the new window to the SOURCE session. + # Without it break-pane follows the ATTACHED session and + # flings panes across sessions (observed live on scratch). r = mcw._tmux(t["socket"], "break-pane", "-s", t["pane"], - "-n", name, timeout=15) + "-t", "%s:" % t["session"], "-n", name, + timeout=15) if r.returncode == 0: spread.append("%s:%s" % (t["session"], t["pane"])) else: @@ -1272,13 +1277,51 @@ def cmd_muse_choices(args): print(line.rstrip("\n")) print() + elif action == "resolve": + sock = getattr(args, "socket", None) + pane = getattr(args, "pane", None) + decision = getattr(args, "decision", None) + hf = mcw.read_hold(sock, pane) + if not hf: + if as_json: + print(json.dumps({"ok": True, "held": None, + "note": "nothing held"}, indent=2)) + return + print(c_dim("\n Nothing held on %s:%s.\n" % (sock, pane))) + return + if decision == "deny" and hf.get("kind") not in mcw.NEGATIVE_KEYS: + msg = ("refusing deny on %s prompt (D2: questions always " + "resolve top-choice); use approve or wait for expiry" + % (hf.get("kind") or "?")) + if as_json: + print(json.dumps({"ok": False, "reason": msg}, indent=2)) + return + print(c_red("\n Error: " + msg + "\n"), file=sys.stderr) + sys.exit(1) + hf["directive"] = decision + mcw.write_hold(sock, pane, hf) + mcw.audit("muse-choice-resolved", + name="%s:%s" % (os.path.basename(sock), pane), + caller=caller, + extra={"socket": sock, "pane": pane, + "decision": decision, "sig": hf.get("sig")}) + if as_json: + print(json.dumps({"ok": True, "resolved": decision, + "pane": pane}, indent=2)) + return + print(c_green("\n✔ Hold on %s:%s will %s within a poll." + % (sock, pane, + "approve" if decision == "approve" else "deny"))) + print() + else: # status desired = mcw.get_desired() watchers = mcw.status_all() answers = mcw.recent_answers(5) + held = mcw.list_holds() if as_json: print(json.dumps({"ok": True, "desired": desired, - "watchers": watchers, + "watchers": watchers, "held": held, "recent_answers": answers}, indent=2)) return state_badge = badge_ok("ON") if desired["enabled"] else badge_dim("OFF") @@ -1302,6 +1345,18 @@ def cmd_muse_choices(args): print_table(headers, rows) else: print(c_dim(" No watcher state files.")) + if held: + print(c_bold("\n Held prompts (resolve via: box muse-choices resolve --socket S --pane P approve|deny):")) + for h in held: + try: + left = max(0, int(float(h.get("held_until", 0)) - time.time())) + except (TypeError, ValueError): + left = -1 + print(f" • {badge_warn('HELD')} {c_bold(h.get('pane', '?'))} " + f"{h.get('kind')}/{h.get('key')} rule={h.get('rule')} " + f"expires in {left}s") + print(f" {c_dim((h.get('reason') or '')[:100])}") + print(f" {c_dim((h.get('text') or '')[:100])}") if answers: print(c_bold("\n Recent answers (box audit feed):")) for a in answers: @@ -5634,6 +5689,11 @@ def build_parser(): p_mc_rec = mc_sub.add_parser("reconcile", parents=[common], help="Enforce desired state now (start missing / stop excess)") + p_mc_res = mc_sub.add_parser("resolve", parents=[common], help="Resolve a held prompt (approve now or deny it)") + p_mc_res.add_argument("--socket", required=True, help="Tmux socket path (e.g. /tmp/tmux-1000/default)") + p_mc_res.add_argument("--pane", required=True, help="Pane id (e.g. %%37)") + p_mc_res.add_argument("decision", choices=["approve", "deny"], help="Release the hold to approve, or deny it (permission kinds only)") + # Domain: RUNTIME p_rt = subparsers.add_parser("runtime", parents=[common], help="Muse CLI tmux runtimes: list states, send input, launch auto-approved") rt_sub = p_rt.add_subparsers(dest="rt_action") diff --git a/docs/HATCH-MENU.md b/docs/HATCH-MENU.md index 0a38bd9..5da98d0 100644 --- a/docs/HATCH-MENU.md +++ b/docs/HATCH-MENU.md @@ -54,6 +54,13 @@ that only the readback confirms carry `"readback_only": true`. Website Ask/Deny is one-way: the override row leaves the allowed list (no add UI), verified by absence; absent hosts read as "not in Websites list" (effective: web-access default). +Switch sets click once and never fall back to a second click: a +re-click would undo a slow commit (switches toggle). Advanced +switches take one synthetic click; protocol and Data-controls +switches need trusted clicks (synthetic is a proven no-op there). +Radios keep their trusted-click fallback (idempotent). +`data_controls.ai_improvement` is read-only: the site ignores every +input gesture on that switch (proven live; reads fine, sets raise). Caller errors (unknown node/toggle/tab/value) raise `MenuError` before any CDP traffic. Transport failures return `{"ok": False}`. diff --git a/docs/SUPERVISION-SPEC.md b/docs/SUPERVISION-SPEC.md index 9eefa73..3d0a957 100644 --- a/docs/SUPERVISION-SPEC.md +++ b/docs/SUPERVISION-SPEC.md @@ -36,27 +36,42 @@ breaker, ensure-node-supervision feed, host_evidence fallback. OUT: onboarding pipeline lifecycle, completion auditor, loop-health (each keeps its own owner and interviews separately). -### D2. Kill-path precedence (chromebox-watchdog vs agent-health) — UNRESOLVED +### D2. Kill-path precedence (chromebox-watchdog vs agent-health) — SETTLED (recommended accepted) -Both can kill a browser today; only time guards (<2 min) de-conflict them. +Policy: each supervisor owns its signal — chromebox-watchdog owns +CDP/egress/process failure, agent-health owns API-layer failure. +Browser restart stays the shared actuator, and both keep their <2-min +fresh-browser guards. No code change; this paragraph is the guard +against future edits dropping either side. -### D3. Egress-down fall-through (relaunch chrome after failed tunnel restart?) — UNRESOLVED +### D3. Egress-down fall-through (relaunch chrome after failed tunnel restart?) — SETTLED (recommended accepted) -Observed 19:12Z: tunnel restart failed, watchdog relaunched chrome 3× -anyway (one FAILED page). Browser was never the problem. +Policy: stop after a failed tunnel restart. Log the egress failure +loudly and page it as a tunnel fault, never as a browser fault; skip +the chrome relaunch. The next 2-min run retries the tunnel. (Requires +a chromebox-watchdog.sh change — implementation needs a separate +explicit request.) -### D4. Circuit-breaker thresholds (3 futile / 30 min cooldown) — UNRESOLVED +### D4. Circuit-breaker thresholds (3 futile / 30 min cooldown) — SETTLED (recommended accepted) -Current values unvalidated against real recurrence intervals. +Keep 3 strikes / 1800s cooldown as initial values, with a review +trigger: revisit if one node trips the circuit more than twice in a +week (threshold too touchy) or a stuck node sits out a full cooldown +under operator eyes (cooldown too long). -### D5. Coverage source of truth — UNRESOLVED +### D5. Coverage source of truth — SETTLED (recommended accepted) -Registry-only vs registry+installed-timers for browser verdicts. +Registry-only: every active NODES.md row is supervised, period. The +ensure script guarantees the timer follows the row, and a missing +timer degrades to an honest `unknown` (no journal runs) rather than a +wrong verdict. No second inventory. -### D6. Concurrent-edit protocol for shared supervision files — UNRESOLVED +### D6. Concurrent-edit protocol for shared supervision files — SETTLED (recommended accepted) -Two agents editing super-cli.py / watchdogs / runbook; one clobber -(18:03Z) and one unattributed commit (c9143a5) already occurred. +Protocol: claim-before-edit (announce shared-file claims in chat), +anchored-edits-only on shared files (never full-file rewrites, so +concurrent work interleaves), re-read before staging, and prompt +mine-only commits so landed work survives the next clobber. ### D7. Done means — UNRESOLVED @@ -78,4 +93,4 @@ loops registry-driven with tests; ensure hook live; this doc Final. ## Unresolved items -D2–D7 unresolved. D1 settled. +D7 unresolved. D1–D6 settled. diff --git a/tests/test_box_runtime.py b/tests/test_box_runtime.py index a2ca158..894424b 100644 --- a/tests/test_box_runtime.py +++ b/tests/test_box_runtime.py @@ -239,6 +239,35 @@ class TestRuntimeRows(unittest.TestCase): self.assertTrue(rows[1]["squeezed"]) +class TestNodeFromSession(unittest.TestCase): + def test_conforming_sessions(self): + self.assertEqual(w.node_from_session("pip--worker--01"), "pip") + self.assertEqual(w.node_from_session("muse--repair--09"), "muse") + self.assertEqual(w.node_from_session("646--watch--a1"), "646") + self.assertEqual(w.node_from_session("PIP--X--1"), "pip") + + def test_adhoc_sessions_have_no_node(self): + for name in ("muse", "repair-09", "swarm-worker", "main", + "", None): + self.assertIsNone(w.node_from_session(name), name) + + def test_rejects_unknown_node_and_single_dash(self): + self.assertIsNone(w.node_from_session("foo--worker--01")) + self.assertIsNone(w.node_from_session("pip-worker-01")) + self.assertIsNone(w.node_from_session("pip--worker")) + + def test_runtime_rows_carry_node(self): + listing = "pip--audit--a1\t1\t%37\tbash\t2880158\t80\t24\n" + r = mock.Mock(returncode=0, stdout=listing, stderr="") + with mock.patch.object(w, "_tmux", return_value=r), \ + mock.patch.object(w, "capture_pane", + return_value=STATE_SHELL), \ + mock.patch.object(w, "is_running", return_value=None): + rows = w.runtime_rows("/tmp/sock") + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["node"], "pip") + + class TestSpreadTargets(unittest.TestCase): def _row(self, pane, is_muse, squeezed): return {"socket": "/tmp/s", "session": "muse", "window": "1", diff --git a/tests/test_hatch_menu.py b/tests/test_hatch_menu.py index 88e51a4..1e20e74 100644 --- a/tests/test_hatch_menu.py +++ b/tests/test_hatch_menu.py @@ -261,6 +261,18 @@ class ControlsTests(unittest.TestCase): self.assertTrue(controls.set_switch(mock.Mock(), "Transparent proxy", True)) + def test_set_switch_never_reclicks(self): + # A fallback re-click would undo a slow commit (switches + # toggle), so a poll miss returns False without real Click. + off = [{"label": "Transparent proxy row", "aria": "", + "checked": False}] + with mock.patch.object(controls, "cdp_evaluate", + side_effect=[off, "CLICKED"] + [off] * 10), \ + mock.patch.object(controls, "real_click") as click: + self.assertFalse(controls.set_switch(mock.Mock(), + "Transparent proxy", True)) + click.assert_not_called() + def test_listers_reject_wrong_types(self): with mock.patch.object(controls, "cdp_evaluate", return_value="error"): @@ -332,6 +344,13 @@ class TogglesTests(unittest.TestCase): res = toggles.get_toggle("pip", "permissions.web_access") self.assertEqual((res["ok"], res["value"]), (True, "always_ask")) + def test_set_toggle_readonly_raises_before_cdp(self): + with mock.patch.object(toggles, "get_cdp_ws") as g: + with self.assertRaises(toggles.MenuError): + toggles.set_toggle("pip", "data_controls.ai_improvement", + "off") + g.assert_not_called() + def test_set_toggle_bad_value_raises_before_cdp(self): with mock.patch.object(toggles, "get_cdp_ws") as g: with self.assertRaises(toggles.MenuError): @@ -552,6 +571,34 @@ class TabContractTests(unittest.TestCase): return_value=sws): self.assertTrue(data_controls.ai_improvement(mock.Mock())) + def test_set_ai_improvement_noop(self): + row = {"label": "Help improve our AI models", "aria": "", + "checked": True} + with mock.patch.object(dialog, "goto_tab", return_value=True), \ + mock.patch.object(controls, "list_switches", + return_value=[row]), \ + mock.patch.object(data_controls, "real_click") as click, \ + mock.patch("time.sleep"): + self.assertTrue(data_controls.set_ai_improvement( + mock.Mock(), True)) + click.assert_not_called() + + def test_set_ai_improvement_real_click(self): + off = {"label": "Help improve our AI models", "aria": "", + "checked": False} + on = {"label": "Help improve our AI models", "aria": "", + "checked": True} + with mock.patch.object(dialog, "goto_tab", return_value=True), \ + mock.patch.object(controls, "list_switches", + side_effect=[[off], [on]]), \ + mock.patch.object(data_controls, "cdp_evaluate", + return_value={"x": 1, "y": 2}), \ + mock.patch.object(data_controls, "real_click") as click, \ + mock.patch("time.sleep"): + self.assertTrue(data_controls.set_ai_improvement( + mock.Mock(), True)) + click.assert_called_once() + def test_data_controls_single_switch(self): one = [{"label": "Help improve the model", "aria": "", "checked": True}] diff --git a/tests/test_muse_choice_watcher.py b/tests/test_muse_choice_watcher.py index 36ccbc1..279ee39 100644 --- a/tests/test_muse_choice_watcher.py +++ b/tests/test_muse_choice_watcher.py @@ -1086,5 +1086,370 @@ class TestExplicitPhrase(unittest.TestCase): "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"]) + + if __name__ == "__main__": unittest.main()