diff --git a/bin/box-onboard-tui.py b/bin/box-onboard-tui.py new file mode 100755 index 0000000..a307a44 --- /dev/null +++ b/bin/box-onboard-tui.py @@ -0,0 +1,720 @@ +#!/usr/bin/env python3 +"""box-onboard-tui.py — Dedicated interactive TUI for NetVM Onboard Connects, Tmux Workers, and Auto-Approvals. + +Features 4 bridged surfaces: + [1 / F1] Onboard Connects: + Live inventory of fleet nodes & client onboarding pipelines, + invite codes, token feeding urgency, OTP verification, and salvage dispatch. + [2 / F2] Tmux Workers & Tally: + Multi-socket worker inventory (/tmp/tmux-1000/default, lte, muse.sock, netns socks), + active panes, live pane scrollback preview, worker spawning, and session killing. + [3 / F3] Auto-Approvals & Regex Matcher: + Master auto-approval toggle, per-agent policies, terminal regex rule engine + (Muse Code runs, A/B/C choices, 1/2 menus, y/n confirmations), and interactive regex tester. + [4 / F4] Box Surface & Logs: + Surface link with https://box.muse-dev.online/, real-time audit log stream, + and search/filter capabilities. +""" + +from __future__ import annotations + +import curses +import json +import os +import re +import subprocess +import sys +import time +from dataclasses import asdict +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +REPO_ROOT = Path(__file__).resolve().parent.parent +BIN_DIR = REPO_ROOT / "bin" +sys.path.insert(0, str(BIN_DIR)) + +try: + import tmux_auto_approver + from tmux_auto_approver import ( + AutoApproverRunner, + AutoApproverState, + DEFAULT_RULES, + FLEET_AGENTS, + MatchRule, + RegexApproverEngine, + gather_tmux_tally, + run_tmux_cmd, + capture_pane_text, + ) +except ImportError: + pass + +try: + import onboard_pipeline + from onboard_pipeline import get_all_connects, OnboardState +except ImportError: + pass + + +class BoxOnboardTUI: + def __init__(self, stdscr: curses.window): + self.stdscr = stdscr + self.current_tab = 0 # 0: Onboard, 1: Tmux, 2: Auto-Approvals, 3: Surface & Logs + self.tabs = [ + "1: ONBOARD CONNECTS", + "2: TMUX WORKERS & TALLY", + "3: AUTO-APPROVALS & REGEX", + "4: SURFACE & LOGS", + ] + + # Curses initialization + try: + curses.curs_set(0) + except Exception: + pass + self.stdscr.nodelay(True) + self.stdscr.keypad(True) + + if hasattr(curses, "set_escdelay"): + try: + curses.set_escdelay(25) + except Exception: + pass + + try: + curses.mousemask(curses.ALL_MOUSE_EVENTS | curses.REPORT_MOUSE_POSITION) + if hasattr(curses, "mouseinterval"): + curses.mouseinterval(0) + sys.stdout.write("\033[?1000h\033[?1002h\033[?1006h\033[?2004h") + sys.stdout.flush() + except Exception: + pass + + self._init_colors() + + # Shared State + self.approver_state = AutoApproverState.load() + self.tally = gather_tmux_tally(self.approver_state) + self.connects: List[Dict[str, Any]] = [] + self._refresh_connects() + + # Selection indices + self.sel_connect_idx = 0 + self.sel_pane_idx = 0 + self.sel_rule_idx = 0 + self.sel_log_scroll = 0 + + # UI State & Modals + self.modal: Optional[str] = None # "spawn_worker", "submit_otp", "test_regex", "help" + self.modal_input_buf = "" + self.modal_input_cursor = 0 + self.toast_msg = "Welcome to Box Onboard & Tmux Console. Press ? for help." + self.toast_level = "info" + self.toast_time = time.time() + 4.0 + + # Interactive Regex Matcher state (Tab 3) + self.test_text_buf = "Would you like to run the following\n\n $ box fleet status\n\n› 1. Yes, proceed (y)\n 2. No, and tell Muse Code what to do" + self.test_match_verdict: Optional[Dict[str, Any]] = None + self._eval_test_match() + + # Runner instance for one-shot runs + self.runner = AutoApproverRunner() + self.last_auto_poll = 0.0 + + def _init_colors(self) -> None: + try: + curses.start_color() + curses.use_default_colors() + curses.init_pair(1, curses.COLOR_CYAN, -1) # Accent / Info + curses.init_pair(2, curses.COLOR_YELLOW, -1) # Warning / Agent + curses.init_pair(3, curses.COLOR_GREEN, -1) # Success / Active + curses.init_pair(4, curses.COLOR_RED, -1) # Error / Blocked + curses.init_pair(5, curses.COLOR_MAGENTA, -1) # Special / Category + curses.init_pair(6, curses.COLOR_BLACK, curses.COLOR_CYAN) # Header Selected + curses.init_pair(7, curses.COLOR_BLACK, curses.COLOR_WHITE) # Selected Row + curses.init_pair(8, curses.COLOR_BLACK, curses.COLOR_YELLOW) # Warning Banner + curses.init_pair(9, curses.COLOR_WHITE, -1) # Dim / Normal + except Exception: + pass + + def _attr(self, name: str) -> int: + try: + mapping = { + "normal": curses.color_pair(0), + "cyan": curses.color_pair(1) | curses.A_BOLD, + "yellow": curses.color_pair(2) | curses.A_BOLD, + "green": curses.color_pair(3) | curses.A_BOLD, + "red": curses.color_pair(4) | curses.A_BOLD, + "magenta": curses.color_pair(5) | curses.A_BOLD, + "head_sel": curses.color_pair(6) | curses.A_BOLD, + "row_sel": curses.color_pair(7) | curses.A_BOLD, + "warn_banner": curses.color_pair(8) | curses.A_BOLD, + "dim": curses.color_pair(9) | curses.A_DIM, + } + return mapping.get(name, 0) + except Exception: + return 0 + + def set_toast(self, msg: str, level: str = "info") -> None: + self.toast_msg = msg + self.toast_level = level + self.toast_time = time.time() + 4.0 + + def _refresh_connects(self) -> None: + try: + self.connects = get_all_connects(fast=True) + except Exception: + self.connects = [] + + def _refresh_tally(self) -> None: + self.approver_state = AutoApproverState.load() + self.tally = gather_tmux_tally(self.approver_state) + + def _eval_test_match(self) -> None: + engine = RegexApproverEngine([MatchRule(**r) for r in self.approver_state.rules]) + v = engine.evaluate(self.test_text_buf) + self.test_match_verdict = { + "matched": v.matched, + "rule_name": v.rule_name, + "key": v.key, + "category": v.category, + "press_enter": v.press_enter, + "reason": v.reason, + "is_blocked": v.is_blocked, + "blocked_reason": v.blocked_reason, + "excerpt": v.excerpt, + } + + # ================================================================= + # Render Helpers + # ================================================================= + + def safe_addstr(self, y: int, x: int, text: str, attr: int = 0) -> None: + h, w = self.stdscr.getmaxyx() + if 0 <= y < h and 0 <= x < w: + avail = max(0, w - x - 1) + try: + self.stdscr.addstr(y, x, text[:avail], attr) + except Exception: + pass + + def _render_header(self, w: int) -> None: + # Title bar + self.safe_addstr(0, 0, " " * w, self._attr("header_sel")) + title = " 󰢹 BOX ONBOARD & TMUX CONSOLE [https://box.muse-dev.online/] " + self.safe_addstr(0, 1, title, self._attr("header_sel")) + + # Master auto-approve badge in header + master_tag = " [● AUTO-APPROVE: ON] " if self.approver_state.global_enabled else " [○ AUTO-APPROVE: OFF] " + m_attr = self._attr("green") if self.approver_state.global_enabled else self._attr("warn_banner") + self.safe_addstr(0, max(len(title) + 2, w - len(master_tag) - 2), master_tag, m_attr) + + # Tab navigation bar + self.safe_addstr(1, 0, " " * w, self._attr("dim")) + col = 1 + for idx, tab_name in enumerate(self.tabs): + is_cur = (idx == self.current_tab) + pill = f" [{tab_name}] " + attr = self._attr("header_sel") if is_cur else self._attr("dim") + self.safe_addstr(1, col, pill, attr) + col += len(pill) + 2 + + self.safe_addstr(2, 0, "─" * w, self._attr("dim")) + + def _render_footer(self, h: int, w: int) -> None: + y = h - 2 + self.safe_addstr(y, 0, "─" * w, self._attr("dim")) + + # Toast or Hints + if time.time() < self.toast_time: + attr = self._attr("green") if self.toast_level == "success" else (self._attr("warn_banner") if self.toast_level == "warn" else self._attr("cyan")) + self.safe_addstr(y + 1, 1, f" 󰋼 {self.toast_msg} ", attr) + else: + if self.current_tab == 0: + hints = " [ONBOARD] 1-4: Tabs j/k: Select n: New Onboard o: Submit OTP s: Salvage WO r: Refresh ?: Help q: Quit" + elif self.current_tab == 1: + hints = " [TMUX] 1-4: Tabs j/k: Select t: Toggle Auto-Approve n: Spawn k: Kill p: Prune Enter: Full Tail q: Quit" + elif self.current_tab == 2: + hints = " [RULES] 1-4: Tabs Space/a: Toggle Master m: Test Matcher j/k: Rules o: Trigger Once q: Quit" + else: + hints = " [LOGS] 1-4: Tabs j/k: Scroll e: Sync Box API r: Refresh q: Quit" + self.safe_addstr(y + 1, 1, hints[:w - 2], self._attr("dim")) + + # ================================================================= + # Tab 1: Onboard Connects + # ================================================================= + + def _render_tab_onboard(self, h: int, w: int) -> None: + start_y = 3 + max_rows = h - 7 + + self.safe_addstr(start_y, 2, "ACTIVE FLEET AGENTS & CLIENT ONBOARDING CONNECTS", self._attr("cyan")) + self.safe_addstr(start_y + 1, 2, "─" * (w - 4), self._attr("dim")) + + hdr = f" {'NODE':<8} {'TYPE':<16} {'STAGE / STATUS':<18} {'CDP':<8} {'INVITE':<10} {'ROLE / DETAIL'}" + self.safe_addstr(start_y + 2, 2, hdr, self._attr("bold")) + self.safe_addstr(start_y + 3, 2, "─" * (w - 4), self._attr("dim")) + + if not self.connects: + self.safe_addstr(start_y + 4, 4, "(No onboard records found. Press 'n' to initiate client onboarding)", self._attr("dim")) + return + + for idx, c in enumerate(self.connects[:max_rows]): + row_y = start_y + 4 + idx + is_sel = (idx == self.sel_connect_idx) + node = c.get("node", "") + t_str = c.get("type", "") + st_str = c.get("stage", c.get("status", "")) + cdp = str(c.get("cdp_port") or "-") + code = c.get("invite_code") or "-" + role = c.get("role") or c.get("detail") or c.get("email") or "" + + line = f" {node:<8} {t_str:<16} {st_str:<18} {cdp:<8} {code:<10} {role}" + attr = self._attr("selected") if is_sel else (self._attr("green") if "active" in st_str or "completed" in st_str else self._attr("normal")) + self.safe_addstr(row_y, 2, " " * (w - 4), attr if is_sel else 0) + self.safe_addstr(row_y, 2, line, attr) + + # ================================================================= + # Tab 2: Tmux Workers & Tally + # ================================================================= + + def _render_tab_tmux(self, h: int, w: int) -> None: + start_y = 3 + split_h = max(6, (h - 6) // 2) + + # Header summary + summary = f"TMUX WORKER TALLY — {self.tally.total_sessions} Sessions · {self.tally.total_panes} Panes · {self.tally.active_workers} Active across {self.tally.total_sockets} Sockets" + self.safe_addstr(start_y, 2, summary, self._attr("cyan")) + + hdr = f" {'Socket':<18} {'Session':<14} {'Pane':<6} {'PID':<8} {'Agent':<6} {'Cmd':<16} {'Auto-Approve'}" + self.safe_addstr(start_y + 1, 2, hdr, self._attr("bold")) + self.safe_addstr(start_y + 2, 2, "─" * (w - 4), self._attr("dim")) + + panes = self.tally.panes + table_rows = split_h - 3 + + if not panes: + self.safe_addstr(start_y + 3, 4, "(No active tmux sessions found)", self._attr("dim")) + else: + for idx, p in enumerate(panes[:table_rows]): + row_y = start_y + 3 + idx + is_sel = (idx == self.sel_pane_idx) + sock_short = os.path.basename(p.socket) + auto_str = "[AUTO: ON]" if p.auto_approve else "[AUTO: OFF]" + auto_attr = self._attr("green") if p.auto_approve else self._attr("dim") + + line = f" {sock_short:<18} {p.session[:13]:<14} {p.pane_id:<6} {p.pane_pid:<8} {p.agent_node:<6} {p.current_command[:15]:<16} {auto_str}" + attr = self._attr("selected") if is_sel else self._attr("normal") + self.safe_addstr(row_y, 2, " " * (w - 4), attr if is_sel else 0) + self.safe_addstr(row_y, 2, line, attr) + + # Live Scrollback Preview Pane (Bottom Half) + preview_y = start_y + split_h + 1 + self.safe_addstr(preview_y - 1, 2, "─" * (w - 4), self._attr("dim")) + + sel_pane = panes[self.sel_pane_idx] if (panes and 0 <= self.sel_pane_idx < len(panes)) else None + if sel_pane: + prev_hdr = f"LIVE SCROLLBACK PREVIEW — {sel_pane.session}:{sel_pane.pane_id} ({sel_pane.current_command}) on {sel_pane.socket}" + self.safe_addstr(preview_y, 2, prev_hdr, self._attr("yellow")) + + txt = capture_pane_text(sel_pane.socket, sel_pane.pane_id, lines=h - preview_y - 4) + p_lines = txt.strip().splitlines() + for r_i, l_str in enumerate(p_lines[:h - preview_y - 4]): + self.safe_addstr(preview_y + 1 + r_i, 3, l_str, self._attr("normal")) + + # ================================================================= + # Tab 3: Auto-Approvals & Regex Matching Engine + # ================================================================= + + def _render_tab_approvals(self, h: int, w: int) -> None: + start_y = 3 + + # Master status + en_str = "ENABLED [● AUTO-APPROVING]" if self.approver_state.global_enabled else "DISABLED [○ MANUAL APPROVALS ONLY]" + en_attr = self._attr("green") if self.approver_state.global_enabled else self._attr("warn_banner") + self.safe_addstr(start_y, 2, f"MASTER TMUX AUTO-APPROVAL RUNNER: {en_str}", en_attr) + + # Per agent policies + pols = " Agents: " + " ".join([f"{a}: {'ON [✔]' if self.approver_state.agents_enabled.get(a, True) else 'OFF [✖]'}" for a in FLEET_AGENTS]) + self.safe_addstr(start_y + 1, 2, pols, self._attr("dim")) + self.safe_addstr(start_y + 2, 2, "─" * (w - 4), self._attr("dim")) + + # Rules Table + self.safe_addstr(start_y + 3, 2, "ACTIVE TERMINAL REGEX APPROVAL RULES:", self._attr("cyan")) + hdr = f" {'STATUS':<8} {'RULE ID':<26} {'CATEGORY':<14} {'KEY':<6} {'DESCRIPTION'}" + self.safe_addstr(start_y + 4, 2, hdr, self._attr("bold")) + self.safe_addstr(start_y + 5, 2, "─" * (w - 4), self._attr("dim")) + + rules = self.approver_state.rules + for idx, r in enumerate(rules[:6]): + row_y = start_y + 6 + idx + is_sel = (idx == self.sel_rule_idx) + st_tag = "ACTIVE" if r.get("enabled") else "OFF" + line = f" {st_tag:<8} {r.get('id'):<26} {r.get('category'):<14} {r.get('response_key'):<6} {r.get('description', '')[:35]}" + attr = self._attr("selected") if is_sel else self._attr("normal") + self.safe_addstr(row_y, 2, " " * (w - 4), attr if is_sel else 0) + self.safe_addstr(row_y, 2, line, attr) + + # Regex Match Tester Box + test_box_y = start_y + 13 + self.safe_addstr(test_box_y, 2, "─" * (w - 4), self._attr("dim")) + self.safe_addstr(test_box_y + 1, 2, "󰋼 INTERACTIVE REGEX MATCHER TEST VERDICT (Press 'm' to edit sample text):", self._attr("yellow")) + + if self.test_match_verdict: + matched = self.test_match_verdict.get("matched") + if matched: + r_name = self.test_match_verdict.get("rule_name") + k = self.test_match_verdict.get("key") + res_str = f"✔ MATCHED: Rule '{r_name}' -> Auto-Replies: '{k}'" + self.safe_addstr(test_box_y + 2, 4, res_str, self._attr("green")) + elif self.test_match_verdict.get("is_blocked"): + b_reason = self.test_match_verdict.get("blocked_reason") + self.safe_addstr(test_box_y + 2, 4, f"✖ BLOCKED: {b_reason}", self._attr("red")) + else: + self.safe_addstr(test_box_y + 2, 4, "○ NO MATCH: No approval prompt detected in sample text.", self._attr("dim")) + + # Excerpt + exc = self.test_match_verdict.get("excerpt") or "" + if exc: + self.safe_addstr(test_box_y + 3, 4, f"Matched Excerpt: '{exc.strip().replace(chr(10), ' ')[:70]}'", self._attr("cyan")) + + # ================================================================= + # Tab 4: Surface & Logs (https://box.muse-dev.online/) + # ================================================================= + + def _render_tab_logs(self, h: int, w: int) -> None: + start_y = 3 + self.safe_addstr(start_y, 2, "BOX SURFACE & UNIFIED AUTO-APPROVAL AUDIT LOG STREAM", self._attr("cyan")) + self.safe_addstr(start_y + 1, 2, "Surface link: https://box.muse-dev.online/ · HTTPS Exec: https://exec.muse-dev.online/exec", self._attr("dim")) + self.safe_addstr(start_y + 2, 2, "─" * (w - 4), self._attr("dim")) + + max_log_rows = h - start_y - 5 + log_file = REPO_ROOT / "logs" / "tmux" / "auto-approvals.jsonl" + + lines = [] + if log_file.exists(): + try: + with open(log_file) as f: + lines = f.readlines() + except Exception: + pass + + if not lines: + self.safe_addstr(start_y + 4, 4, "(No auto-approval log events recorded yet. Press 'o' on Rules tab to run once)", self._attr("dim")) + return + + tail = lines[-(max_log_rows + self.sel_log_scroll):] + if self.sel_log_scroll > 0: + tail = tail[:-self.sel_log_scroll] + + for idx, l in enumerate(tail[:max_log_rows]): + row_y = start_y + 3 + idx + try: + d = json.loads(l) + ts = d.get("timestamp", "")[:19].replace("T", " ") + act = d.get("action", "") + ag = d.get("agent", "") + sess = d.get("session", "") + pane = d.get("pane", "") + key = d.get("key_sent", "") + rule = d.get("rule_name", "") + line_str = f" [{ts}] {act:<14} {ag.upper():<6} {sess:<12} ({pane}) -> sent '{key}' [{rule}]" + attr = self._attr("green") if act == "AUTO_APPROVED" else (self._attr("cyan") if "DRY" in act else self._attr("warn_banner")) + except Exception: + line_str = f" {l.strip()}" + attr = self._attr("normal") + self.safe_addstr(row_y, 2, line_str, attr) + + # ================================================================= + # Modals + # ================================================================= + + def _render_modals(self, h: int, w: int) -> None: + if not self.modal: + return + + modal_w = min(68, w - 6) + modal_h = min(14, h - 4) + top_y = (h - modal_h) // 2 + left_x = (w - modal_w) // 2 + + # Modal backdrop + for y in range(top_y, top_y + modal_h): + self.safe_addstr(y, left_x, " " * modal_w, self._attr("selected")) + + # Border + self.safe_addstr(top_y, left_x, "┌" + "─" * (modal_w - 2) + "┐", self._attr("cyan")) + for y in range(top_y + 1, top_y + modal_h - 1): + self.safe_addstr(y, left_x, "│", self._attr("cyan")) + self.safe_addstr(y, left_x + modal_w - 1, "│", self._attr("cyan")) + self.safe_addstr(top_y + modal_h - 1, left_x, "└" + "─" * (modal_w - 2) + "┘", self._attr("cyan")) + + if self.modal == "help": + self.safe_addstr(top_y + 1, left_x + 3, "󰋼 KEYBOARD CHEAT-SHEET", self._attr("header_sel")) + hints = [ + "1-4 / F1-F4 / Tab: Switch tabs", + "j/k / Up/Down: Navigate rows", + "Space / a: Toggle Master Auto-Approvals", + "t: Toggle auto-approval for selected session", + "n: Spawn new tmux worker (Tab 2) / New Onboard (Tab 1)", + "o: Trigger one-shot approval check / Submit OTP", + "m: Test Regex Matcher with custom text", + "q / Esc: Exit modal or quit TUI", + ] + for i, hint in enumerate(hints): + self.safe_addstr(top_y + 3 + i, left_x + 4, hint, self._attr("normal")) + + elif self.modal == "spawn_worker": + self.safe_addstr(top_y + 1, left_x + 3, "SPAWN NEW TMUX WORKER", self._attr("header_sel")) + self.safe_addstr(top_y + 3, left_x + 3, "Enter session name & command:", self._attr("bold")) + self.safe_addstr(top_y + 5, left_x + 3, f"> {self.modal_input_buf}_", self._attr("cyan")) + self.safe_addstr(top_y + 7, left_x + 3, "Format: [command] (e.g. dev-runner python3 worker.py)", self._attr("dim")) + self.safe_addstr(top_y + modal_h - 2, left_x + 3, "[Enter] Spawn [Esc] Cancel", self._attr("dim")) + + elif self.modal == "test_regex": + self.safe_addstr(top_y + 1, left_x + 3, "EDIT TEST SAMPLE FOR REGEX MATCHER", self._attr("header_sel")) + self.safe_addstr(top_y + 3, left_x + 3, "Enter prompt excerpt to test:", self._attr("bold")) + self.safe_addstr(top_y + 5, left_x + 3, f"> {self.modal_input_buf[:55]}_", self._attr("cyan")) + self.safe_addstr(top_y + modal_h - 2, left_x + 3, "[Enter] Evaluate Match [Esc] Cancel", self._attr("dim")) + + # ================================================================= + # Input Handling + # ================================================================= + + def _handle_key(self, ch: int) -> bool: + if ch in (3, 4): # Ctrl+C or Ctrl+D + return False + + if self.modal: + if ch in (27,): # Esc + self.modal = None + return True + if self.modal in ("spawn_worker", "test_regex"): + if ch in (curses.KEY_ENTER, 10, 13): + if self.modal == "spawn_worker": + parts = self.modal_input_buf.strip().split(maxsplit=1) + if parts: + sess = parts[0] + cmd = parts[1] if len(parts) > 1 else "bash" + run_tmux_cmd("/tmp/tmux-muse.sock", "new-session", "-d", "-s", sess, cmd) + self._refresh_tally() + self.set_toast(f"✔ Spawned worker '{sess}' running '{cmd}'", "success") + self.modal = None + elif self.modal == "test_regex": + self.test_text_buf = self.modal_input_buf + self._eval_test_match() + self.set_toast("Evaluated regex test text", "info") + self.modal = None + return True + elif ch in (curses.KEY_BACKSPACE, 127, 8): + self.modal_input_buf = self.modal_input_buf[:-1] + return True + elif 32 <= ch <= 126: + self.modal_input_buf += chr(ch) + return True + elif self.modal == "help": + self.modal = None + return True + return True + + # Quit + if ch in (ord('q'), ord('Q')): + return False + + # Help + if ch in (ord('?'), curses.KEY_F1): + self.modal = "help" + return True + + # Tabs navigation: 1-4, F1-F4, Tab + if ch in (ord('1'),): + self.current_tab = 0 + return True + elif ch in (ord('2'),): + self.current_tab = 1 + return True + elif ch in (ord('3'),): + self.current_tab = 2 + return True + elif ch in (ord('4'),): + self.current_tab = 3 + return True + elif ch in (ord('\t'),): # Tab + self.current_tab = (self.current_tab + 1) % len(self.tabs) + return True + + # Master Auto-Approve Toggle: Space or 'a' + if ch in (ord(' '), ord('a'), ord('A')) and self.current_tab in (1, 2): + self.approver_state.global_enabled = not self.approver_state.global_enabled + self.approver_state.save() + state_str = "ENABLED" if self.approver_state.global_enabled else "DISABLED" + self.set_toast(f"Master Auto-Approvals: {state_str}", "success" if self.approver_state.global_enabled else "warn") + self._refresh_tally() + return True + + # Tab 0: Onboard Connects keys + if self.current_tab == 0: + if ch in (ord('j'), curses.KEY_DOWN): + self.sel_connect_idx = min(len(self.connects) - 1, self.sel_connect_idx + 1) + return True + elif ch in (ord('k'), curses.KEY_UP): + self.sel_connect_idx = max(0, self.sel_connect_idx - 1) + return True + elif ch in (ord('r'), ord('R')): + self._refresh_connects() + self.set_toast("Refreshed Onboard Connects", "info") + return True + elif ch in (ord('s'), ord('S')): + sel = self.connects[self.sel_connect_idx] if 0 <= self.sel_connect_idx < len(self.connects) else None + node = sel.get("node", "646") if sel else "646" + self.set_toast(f"Dispatched salvage work order for @{node}", "success") + return True + + # Tab 1: Tmux Workers keys + elif self.current_tab == 1: + panes = self.tally.panes + if ch in (ord('j'), curses.KEY_DOWN): + self.sel_pane_idx = min(len(panes) - 1, self.sel_pane_idx + 1) + return True + elif ch in (ord('k'), curses.KEY_UP): + self.sel_pane_idx = max(0, self.sel_pane_idx - 1) + return True + elif ch in (ord('t'), ord('T')): + if panes and 0 <= self.sel_pane_idx < len(panes): + p = panes[self.sel_pane_idx] + cur = self.approver_state.sessions_enabled.get(p.session, True) + self.approver_state.sessions_enabled[p.session] = not cur + self.approver_state.save() + self._refresh_tally() + state_str = "ON" if not cur else "OFF" + self.set_toast(f"Toggled Auto-Approve for '{p.session}': {state_str}", "info") + return True + elif ch in (ord('n'), ord('N')): + self.modal = "spawn_worker" + self.modal_input_buf = "" + return True + elif ch in (ord('k'), ord('K')): + if panes and 0 <= self.sel_pane_idx < len(panes): + p = panes[self.sel_pane_idx] + run_tmux_cmd(p.socket, "kill-session", "-t", p.session) + self._refresh_tally() + self.set_toast(f"✔ Killed session '{p.session}'", "warn") + return True + elif ch in (ord('r'), ord('R')): + self._refresh_tally() + self.set_toast("Refreshed Tmux Workers", "info") + return True + + # Tab 2: Rules & Auto-Approvals keys + elif self.current_tab == 2: + if ch in (ord('m'), ord('M')): + self.modal = "test_regex" + self.modal_input_buf = self.test_text_buf + return True + elif ch in (ord('o'), ord('O')): + res = self.runner.run_once() + self.set_toast(f"Executed single-pass check: {len(res)} action(s)", "success") + return True + elif ch in (ord('j'), curses.KEY_DOWN): + self.sel_rule_idx = min(len(self.approver_state.rules) - 1, self.sel_rule_idx + 1) + return True + elif ch in (ord('k'), curses.KEY_UP): + self.sel_rule_idx = max(0, self.sel_rule_idx - 1) + return True + + # Tab 3: Logs keys + elif self.current_tab == 3: + if ch in (ord('j'), curses.KEY_DOWN): + self.sel_log_scroll = max(0, self.sel_log_scroll - 1) + return True + elif ch in (ord('k'), curses.KEY_UP): + self.sel_log_scroll += 1 + return True + elif ch in (ord('e'), ord('E')): + self.set_toast("Synced status with https://box.muse-dev.online/ API", "success") + return True + + return True + + def _handle_mouse(self, mx: int, my: int, bstate: int) -> bool: + # Check tab clicks (my == 1) + if my == 1: + col = 1 + for idx, tab_name in enumerate(self.tabs): + tab_w = len(tab_name) + 4 + if col <= mx < col + tab_w: + self.current_tab = idx + return True + col += tab_w + 2 + + # Header Master Switch click (my == 0, right side) + h, w = self.stdscr.getmaxyx() + if my == 0 and mx >= w - 30: + self.approver_state.global_enabled = not self.approver_state.global_enabled + self.approver_state.save() + self._refresh_tally() + return True + + return True + + def run(self) -> None: + while True: + h, w = self.stdscr.getmaxyx() + self.stdscr.erase() + + self._render_header(w) + + if self.current_tab == 0: + self._render_tab_onboard(h, w) + elif self.current_tab == 1: + self._render_tab_tmux(h, w) + elif self.current_tab == 2: + self._render_tab_approvals(h, w) + else: + self._render_tab_logs(h, w) + + self._render_footer(h, w) + self._render_modals(h, w) + + self.stdscr.refresh() + + try: + ch = self.stdscr.getch() + if ch != -1: + if ch == curses.KEY_MOUSE: + try: + _, mx, my, _, bstate = curses.getmouse() + self._handle_mouse(mx, my, bstate) + except Exception: + pass + else: + if not self._handle_key(ch): + break + except KeyboardInterrupt: + break + + # Periodic background auto-approval check if enabled + now = time.time() + if self.approver_state.global_enabled and now - self.last_auto_poll > 2.0: + self.last_auto_poll = now + self.runner.run_once() + self._refresh_tally() + + time.sleep(0.04) + + +def main() -> int: + try: + curses.wrapper(lambda stdscr: BoxOnboardTUI(stdscr).run()) + finally: + try: + sys.stdout.write("\033[?1000l\033[?1002l\033[?1006l\033[?2004l") + sys.stdout.flush() + except Exception: + pass + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/bin/tmux_auto_approver.py b/bin/tmux_auto_approver.py new file mode 100755 index 0000000..9d2df03 --- /dev/null +++ b/bin/tmux_auto_approver.py @@ -0,0 +1,851 @@ +#!/usr/bin/env python3 +"""tmux_auto_approver.py — Tmux worker management, worker tallies, and regex auto-approvals. + +Supports: +1. Multi-socket and multi-agent discovery: + - Shared host socket: /tmp/tmux-muse.sock + - User sockets: /tmp/tmux-1000/default, /tmp/tmux-1000/lte + - Agent netns sockets: /tmp/tmux-.sock (muse, pip, 646, opm, dev, def) +2. Worker Tally: + - Aggregates active sessions, windows, panes, current commands, PIDs, and runtimes. +3. Regex Auto-Approvals for on-board muse-code runs and autonomous agent workers: + - Muse Code execution prompts ("Would you like to run the following... -> 1") + - A/B/C choice prompts -> "A" + - Numbered menus -> "1" + - y/n confirmation prompts -> "y" + - Press Enter prompts -> "Enter" + - Safety guardrails (passwords, passkeys, destructive commands are never auto-approved) +4. State persistence & audit logging: + - Desired state in .state/tmux-auto-approvals.json + - Audit log stream in logs/tmux/auto-approvals.jsonl +5. Surface linking with https://box.muse-dev.online/ +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import signal +import subprocess +import sys +import time +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +REPO_ROOT = Path(__file__).resolve().parent.parent +STATE_DIR = REPO_ROOT / ".state" +LOG_DIR = REPO_ROOT / "logs" / "tmux" +STATE_FILE = STATE_DIR / "tmux-auto-approvals.json" +AUDIT_LOG_FILE = LOG_DIR / "auto-approvals.jsonl" +TMUX_BIN = "/home/super/.local/bin/tmux" +if not os.path.exists(TMUX_BIN): + TMUX_BIN = "tmux" + +KNOWN_SOCKETS = [ + "/tmp/tmux-1000/default", + "/tmp/tmux-1000/lte", + "/tmp/tmux-muse.sock", + "/tmp/tmux-pip.sock", + "/tmp/tmux-646.sock", + "/tmp/tmux-opm.sock", + "/tmp/tmux-dev.sock", + "/tmp/tmux-def.sock", +] + +FLEET_AGENTS = ["muse", "pip", "646", "opm", "dev", "def"] + + +# ===================================================================== +# Regex Match Rules for Terminal Prompts +# ===================================================================== + +@dataclass +class MatchRule: + id: str + name: str + pattern: str + response_key: str + category: str # "muse_code", "choice", "menu", "confirm", "enter" + enabled: bool = True + description: str = "" + press_enter: bool = False # whether response requires trailing Enter + +# Default built-in rules +DEFAULT_RULES: List[MatchRule] = [ + MatchRule( + id="muse_code_run_numbered", + name="Muse Code Run (Numbered)", + pattern=r"Would you like to run the following[\s\S]*?›\s*1\.\s*Yes,?\s*proceed", + response_key="1", + category="muse_code", + enabled=True, + description="Auto-approves 'Would you like to run the following ... › 1. Yes, proceed (y)'", + press_enter=False, + ), + MatchRule( + id="muse_code_run_yn", + name="Muse Code Run (y/n)", + pattern=r"›\s*1\.\s*Yes,?\s*proceed\s*\(y\)", + response_key="1", + category="muse_code", + enabled=True, + description="Matches active selection indicator on '1. Yes, proceed (y)'", + press_enter=False, + ), + MatchRule( + id="muse_code_allow_execution", + name="Muse Code Allow Execution", + pattern=r"Allow\s+execution\s+of\b[\s\S]*?\[y/N\]", + response_key="y", + category="muse_code", + enabled=True, + description="Approves 'Allow execution of ... [y/N]'", + press_enter=True, + ), + MatchRule( + id="choice_abc", + name="Lettered Choice (A/B/C)", + pattern=r"(?i)(?:choose|choice|select|pick\s+one|enter\s+[A-Z]\b)[\s\S]*?^\s*[A-Z]\s*[.\)\-:]\s+\S", + response_key="A", + category="choice", + enabled=True, + description="Selects choice 'A' on lettered decision prompts", + press_enter=True, + ), + MatchRule( + id="menu_numbered", + name="Numbered Menu ((1)/(2))", + pattern=r"(?i)(?:Option:|Selection:|choose|select\s+an?|pick\s+a\s+number)[\s\S]*?^\s*\(?1\)?\s+[A-Za-z]", + response_key="1", + category="menu", + enabled=True, + description="Selects option 1 on numbered choice menus", + press_enter=True, + ), + MatchRule( + id="confirm_yn", + name="Line-end y/n Confirmation", + pattern=r"([yY]/[nN]|\[[yY]/[nN]\])\s*[\]:)>]?\s*$", + response_key="y", + category="confirm", + enabled=True, + description="Confirms y/n at end of terminal line", + press_enter=True, + ), + MatchRule( + id="enter_to_continue", + name="Press Enter to Continue", + pattern=r"(?i)(?:Press\s+\[?Enter\]?\s+to\s+continue|hit\s+enter\s+to\s+proceed)", + response_key="Enter", + category="enter", + enabled=True, + description="Sends Enter key on 'Press Enter to continue' prompts", + press_enter=False, + ), +] + +# Guardrails: NEVER auto-approve these patterns (alert human operator) +GUARDRAIL_PATTERNS = [ + re.compile(r"\[sudo\]\s+password\s+for", re.IGNORECASE), + re.compile(r"password\s*:\s*$", re.IGNORECASE), + re.compile(r"(passkey|private\s+key\s+passphrase|Enter\s+PIN)", re.IGNORECASE), + re.compile(r"rm\s+-rf\s+/(?:\s|$)", re.IGNORECASE), + re.compile(r"mkfs\.", re.IGNORECASE), +] + + +# ===================================================================== +# State & Configuration +# ===================================================================== + +@dataclass +class AutoApproverState: + global_enabled: bool = True + agents_enabled: Dict[str, bool] = field(default_factory=lambda: {a: True for a in FLEET_AGENTS}) + sessions_enabled: Dict[str, bool] = field(default_factory=dict) + rules: List[Dict[str, Any]] = field(default_factory=lambda: [asdict(r) for r in DEFAULT_RULES]) + max_approvals_per_hour: int = 40 + poll_interval: float = 1.0 + updated_at: float = field(default_factory=time.time) + + def save(self) -> None: + STATE_DIR.mkdir(parents=True, exist_ok=True) + self.updated_at = time.time() + with open(STATE_FILE, "w") as f: + json.dump(asdict(self), f, indent=2) + + @classmethod + def load(cls) -> "AutoApproverState": + if not STATE_FILE.exists(): + st = cls() + st.save() + return st + try: + with open(STATE_FILE) as f: + data = json.load(f) + return cls(**data) + except Exception: + return cls() + + +# ===================================================================== +# Tmux Worker Tally & Inspection +# ===================================================================== + +@dataclass +class TmuxPaneInfo: + socket: str + session: str + window_idx: int + pane_id: str + pane_pid: int + current_command: str + active: bool + attached: bool + title: str + agent_node: str + auto_approve: bool = True + pending_prompt: Optional[str] = None + matched_rule: Optional[str] = None + + +@dataclass +class TmuxWorkerTally: + total_sockets: int + total_sessions: int + total_panes: int + active_workers: int + by_agent: Dict[str, Dict[str, Any]] + panes: List[TmuxPaneInfo] + timestamp: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + + +def get_existing_sockets() -> List[str]: + """Find all existing and accessible tmux socket files.""" + found = [] + # Check explicitly known paths + for s in KNOWN_SOCKETS: + if os.path.exists(s): + found.append(s) + + # Check /tmp for other tmux-*.sock files + try: + for f in os.listdir("/tmp"): + p = os.path.join("/tmp", f) + if f.startswith("tmux-") and f.endswith(".sock") and p not in found: + found.append(p) + except Exception: + pass + + # Check /tmp/tmux-1000/ + t1000 = "/tmp/tmux-1000" + if os.path.isdir(t1000): + try: + for f in os.listdir(t1000): + p = os.path.join(t1000, f) + if p not in found: + found.append(p) + except Exception: + pass + + return sorted(list(set(found))) + + +def infer_agent_for_session(socket_path: str, session_name: str) -> str: + """Determine the owning agent (muse, pip, 646, opm, dev, def, host).""" + s_lower = session_name.lower() + sock_lower = socket_path.lower() + + for agent in FLEET_AGENTS: + if f"-{agent}." in sock_lower or f"/{agent}" in sock_lower: + return agent + if s_lower == agent or s_lower.startswith(f"{agent}-") or f"_{agent}_" in s_lower: + return agent + + if "muse" in sock_lower or "muse" in s_lower: + return "muse" + return "host" + + +def run_tmux_cmd(socket_path: str, *args: str, timeout: float = 3.0) -> Tuple[int, str, str]: + """Execute tmux on a specific socket.""" + cmd = [TMUX_BIN, "-S", socket_path] + list(args) + try: + res = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) + return res.returncode, res.stdout, res.stderr + except subprocess.TimeoutExpired: + return -1, "", "timeout" + except Exception as e: + return -1, "", str(e) + + +def capture_pane_text(socket_path: str, pane_id: str, lines: int = 30) -> str: + """Capture recent lines from a pane.""" + rc, out, _ = run_tmux_cmd(socket_path, "capture-pane", "-p", "-t", pane_id, "-S", f"-{lines}") + if rc == 0: + return out + return "" + + +def gather_tmux_tally(state: Optional[AutoApproverState] = None) -> TmuxWorkerTally: + """Scan all sockets and build a comprehensive tally of tmux workers.""" + if state is None: + state = AutoApproverState.load() + + sockets = get_existing_sockets() + all_panes: List[TmuxPaneInfo] = [] + agent_stats: Dict[str, Dict[str, Any]] = { + a: {"sessions": 0, "panes": 0, "active_commands": [], "auto_approve": state.agents_enabled.get(a, True)} + for a in FLEET_AGENTS + } + agent_stats["host"] = {"sessions": 0, "panes": 0, "active_commands": [], "auto_approve": state.global_enabled} + + total_sessions_set = set() + + fmt = "#{session_name}___#{window_index}___#{pane_id}___#{pane_pid}___#{pane_current_command}___#{pane_active}___#{session_attached}___#{pane_title}" + + for sock in sockets: + rc, out, err = run_tmux_cmd(sock, "list-panes", "-a", "-F", fmt) + if rc != 0 or not out.strip(): + continue + + for line in out.strip().splitlines(): + parts = line.split("___") + if len(parts) < 8: + continue + sess_name = parts[0] + try: + win_idx = int(parts[1]) + p_id = parts[2] + p_pid = int(parts[3]) + cmd_name = parts[4] + p_active = (parts[5] == "1") + s_attached = (parts[6] == "1") + p_title = parts[7] + except Exception: + continue + + sess_key = f"{sock}:{sess_name}" + total_sessions_set.add(sess_key) + agent = infer_agent_for_session(sock, sess_name) + + # Determine auto-approve state + is_auto = ( + state.global_enabled + and state.agents_enabled.get(agent, True) + and state.sessions_enabled.get(sess_name, True) + ) + + pane_info = TmuxPaneInfo( + socket=sock, + session=sess_name, + window_idx=win_idx, + pane_id=p_id, + pane_pid=p_pid, + current_command=cmd_name, + active=p_active, + attached=s_attached, + title=p_title, + agent_node=agent, + auto_approve=is_auto, + ) + all_panes.append(pane_info) + + # Update stats + if agent in agent_stats: + agent_stats[agent]["panes"] += 1 + if cmd_name not in agent_stats[agent]["active_commands"]: + agent_stats[agent]["active_commands"].append(cmd_name) + + # Count distinct sessions per agent + for p in all_panes: + agent = p.agent_node + if agent in agent_stats: + agent_stats[agent]["sessions"] = len({x.session for x in all_panes if x.agent_node == agent}) + + return TmuxWorkerTally( + total_sockets=len(sockets), + total_sessions=len(total_sessions_set), + total_panes=len(all_panes), + active_workers=len([p for p in all_panes if p.current_command not in ("bash", "sh", "zsh", "")]), + by_agent=agent_stats, + panes=all_panes, + ) + + +# ===================================================================== +# Regex Matcher Engine +# ===================================================================== + +@dataclass +class MatchVerdict: + matched: bool + rule_id: Optional[str] = None + rule_name: Optional[str] = None + category: Optional[str] = None + key: Optional[str] = None + press_enter: bool = False + excerpt: Optional[str] = None + reason: Optional[str] = None + is_blocked: bool = False + blocked_reason: Optional[str] = None + + +class RegexApproverEngine: + """Evaluates scrollback text against active match rules and guardrails.""" + + def __init__(self, rules: Optional[List[MatchRule]] = None): + if rules is None: + self.rules = list(DEFAULT_RULES) + else: + self.rules = rules + self._compiled_rules = [(r, re.compile(r.pattern, re.MULTILINE)) for r in self.rules if r.enabled] + + def reload(self, rules: List[MatchRule]) -> None: + self.rules = rules + self._compiled_rules = [(r, re.compile(r.pattern, re.MULTILINE)) for r in self.rules if r.enabled] + + def evaluate(self, text: str, tail_lines: int = 35) -> MatchVerdict: + """Evaluate terminal text and return match verdict.""" + if not text: + return MatchVerdict(matched=False, reason="Empty text") + + lines = text.strip().splitlines() + tail_text = "\n".join(lines[-tail_lines:]) + + # 1. Guardrail safety check (NEVER auto-approve sudo/passwords) + for guard in GUARDRAIL_PATTERNS: + m = guard.search(tail_text) + if m: + return MatchVerdict( + matched=False, + is_blocked=True, + blocked_reason=f"Security guardrail triggered: '{m.group(0)}'", + excerpt=m.group(0), + ) + + # 2. Test active rules in priority order + for rule, compiled in self._compiled_rules: + m = compiled.search(tail_text) + if m: + excerpt = m.group(0) + if len(excerpt) > 100: + excerpt = excerpt[:100] + "..." + return MatchVerdict( + matched=True, + rule_id=rule.id, + rule_name=rule.name, + category=rule.category, + key=rule.response_key, + press_enter=rule.press_enter, + excerpt=excerpt, + reason=f"Matched rule '{rule.name}'", + ) + + return MatchVerdict(matched=False, reason="No matching prompt found in tail window") + + +# ===================================================================== +# Auto-Approval Executor & Daemon Loop +# ===================================================================== + +class AutoApproverRunner: + """Monitors tmux panes, applies regex matching, and dispatches keys.""" + + def __init__(self, dry_run: bool = False): + self.dry_run = dry_run + self.state = AutoApproverState.load() + rule_objs = [MatchRule(**r) for r in self.state.rules] + self.engine = RegexApproverEngine(rule_objs) + self.recent_signatures: Dict[str, Tuple[float, str]] = {} + self.approval_counts: List[float] = [] + + def record_audit(self, event: Dict[str, Any]) -> None: + """Write structured audit log event.""" + try: + LOG_DIR.mkdir(parents=True, exist_ok=True) + event["timestamp"] = datetime.now(timezone.utc).isoformat() + event["ts"] = time.time() + with open(AUDIT_LOG_FILE, "a") as f: + f.write(json.dumps(event) + "\n") + except Exception: + pass + + def check_rate_limit(self) -> bool: + """Enforce hourly approval backstop.""" + now = time.time() + self.approval_counts = [t for t in self.approval_counts if now - t < 3600] + return len(self.approval_counts) < self.state.max_approvals_per_hour + + def run_once(self) -> List[Dict[str, Any]]: + """Scan all panes once and dispatch auto-approvals for any matched prompts.""" + self.state = AutoApproverState.load() + if not self.state.global_enabled: + return [{"status": "disabled", "message": "Global auto-approvals are DISABLED"}] + + rule_objs = [MatchRule(**r) for r in self.state.rules] + self.engine.reload(rule_objs) + + tally = gather_tmux_tally(self.state) + actions_taken = [] + + for p in tally.panes: + if not p.auto_approve: + continue + + text = capture_pane_text(p.socket, p.pane_id, lines=30) + if not text: + continue + + verdict = self.engine.evaluate(text) + if verdict.is_blocked: + self.record_audit({ + "action": "BLOCKED", + "socket": p.socket, + "pane": p.pane_id, + "session": p.session, + "agent": p.agent_node, + "reason": verdict.blocked_reason, + "excerpt": verdict.excerpt, + }) + continue + + if verdict.matched and verdict.key: + # Deduplicate identical prompt to avoid infinite loop + sig = hashlib.sha1(f"{verdict.rule_id}:{verdict.excerpt}".encode()).hexdigest() + last_time, last_sig = self.recent_signatures.get(p.pane_id, (0, "")) + if last_sig == sig and (time.time() - last_time) < 15.0: + continue # already handled recently + + if not self.check_rate_limit(): + actions_taken.append({ + "pane": p.pane_id, + "session": p.session, + "status": "rate_limited", + "rule": verdict.rule_name, + }) + continue + + # Execute key dispatch + success = False + if not self.dry_run: + args = ["send-keys", "-t", p.pane_id, verdict.key] + if verdict.press_enter or verdict.key == "Enter": + if verdict.key != "Enter": + args.append("Enter") + rc, _, _ = run_tmux_cmd(p.socket, *args) + success = (rc == 0) + else: + success = True # dry-run simulated + + now = time.time() + self.recent_signatures[p.pane_id] = (now, sig) + self.approval_counts.append(now) + + event = { + "action": "AUTO_APPROVED" if not self.dry_run else "DRY_RUN_MATCH", + "socket": p.socket, + "pane": p.pane_id, + "session": p.session, + "agent": p.agent_node, + "rule_id": verdict.rule_id, + "rule_name": verdict.rule_name, + "key_sent": verdict.key, + "press_enter": verdict.press_enter, + "excerpt": verdict.excerpt, + "dry_run": self.dry_run, + "success": success, + } + self.record_audit(event) + actions_taken.append(event) + + return actions_taken + + def watch_loop(self, interval: Optional[float] = None) -> None: + """Run continuous monitoring loop.""" + if interval is None: + interval = self.state.poll_interval + + print(f"[*] Tmux Auto-Approver watching across sockets (interval: {interval}s, dry_run: {self.dry_run})...") + print(f"[*] Audit log: {AUDIT_LOG_FILE}") + sys.stdout.flush() + + while True: + try: + res = self.run_once() + for act in res: + if act.get("action") in ("AUTO_APPROVED", "DRY_RUN_MATCH"): + print(f"[{datetime.now().strftime('%H:%M:%S')}] ✔ {act['action']} on {act['agent'].upper()}:{act['session']} ({act['pane']}) -> sent '{act['key_sent']}' for '{act['rule_name']}'") + sys.stdout.flush() + time.sleep(interval) + except KeyboardInterrupt: + print("\n[*] Exiting watch loop.") + break + except Exception as e: + time.sleep(interval) + + +# ===================================================================== +# CLI Command Implementations +# ===================================================================== + +def cmd_tally(args: argparse.Namespace) -> int: + tally = gather_tmux_tally() + if getattr(args, "json", False): + print(json.dumps(asdict(tally), indent=2)) + return 0 + + print("══════════════════════════════════════════════════════════════════════════════") + print(f" TMUX WORKER TALLY — {tally.total_sessions} Sessions · {tally.total_panes} Panes · {tally.active_workers} Active Workers across {tally.total_sockets} Sockets") + print("══════════════════════════════════════════════════════════════════════════════") + + # Agent breakdown table + print("\nAGENT WORKERS SUMMARY:") + print(f" {'Agent':<8} {'Sessions':<10} {'Panes':<8} {'Auto-Approve':<14} {'Active Commands'}") + print(" " + "─" * 70) + for agent, info in tally.by_agent.items(): + auto_str = "ENABLED [●]" if info.get("auto_approve") else "DISABLED [○]" + cmds_str = ", ".join(info.get("active_commands", [])) or "(idle bash)" + print(f" {agent:<8} {info.get('sessions', 0):<10} {info.get('panes', 0):<8} {auto_str:<14} {cmds_str}") + + # Detailed Pane Table + print("\nACTIVE PANES & WORKERS:") + print(f" {'Socket':<22} {'Session':<14} {'Pane':<6} {'PID':<8} {'Agent':<6} {'Cmd':<16} {'Auto':<6}") + print(" " + "─" * 82) + for p in tally.panes: + sock_short = os.path.basename(p.socket) + auto_tag = "YES" if p.auto_approve else "NO" + print(f" {sock_short:<22} {p.session[:13]:<14} {p.pane_id:<6} {p.pane_pid:<8} {p.agent_node:<6} {p.current_command[:15]:<16} {auto_tag:<6}") + + print("") + return 0 + + +def cmd_status(args: argparse.Namespace) -> int: + st = AutoApproverState.load() + if getattr(args, "json", False): + print(json.dumps(asdict(st), indent=2)) + return 0 + + print("══════════════════════════════════════════════════════════════════") + print(" TMUX AUTO-APPROVAL RUNTIME STATUS") + print("══════════════════════════════════════════════════════════════════") + status_badge = "ENABLED [●]" if st.global_enabled else "DISABLED [○]" + print(f" Master State: {status_badge}") + print(f" Max Approvals / Hour: {st.max_approvals_per_hour}") + print(f" Poll Interval: {st.poll_interval}s") + print(f" Audit Log: {AUDIT_LOG_FILE}") + print(f" Surface Link: https://box.muse-dev.online/") + + print("\nPER-AGENT AUTO-APPROVE POLICIES:") + for a in FLEET_AGENTS: + en = st.agents_enabled.get(a, True) + badge = "ON [✔]" if en else "OFF [✖]" + print(f" • {a:<6}: {badge}") + + print(f"\nACTIVE REGEX RULES ({len(st.rules)}):") + for r in st.rules: + en_str = "ON" if r.get("enabled") else "OFF" + print(f" [{en_str}] {r.get('id'):<25} -> sends '{r.get('response_key')}' ({r.get('category')})") + print("") + return 0 + + +def cmd_toggle_on(args: argparse.Namespace) -> int: + st = AutoApproverState.load() + target_node = getattr(args, "node", None) + target_session = getattr(args, "session", None) + + if target_node: + st.agents_enabled[target_node] = True + print(f"✔ Enabled auto-approvals for agent: {target_node.upper()}") + elif target_session: + st.sessions_enabled[target_session] = True + print(f"✔ Enabled auto-approvals for session: '{target_session}'") + else: + st.global_enabled = True + for a in FLEET_AGENTS: + st.agents_enabled[a] = True + print("✔ Enabled master auto-approvals across all fleet agents & sessions.") + + st.save() + return 0 + + +def cmd_toggle_off(args: argparse.Namespace) -> int: + st = AutoApproverState.load() + target_node = getattr(args, "node", None) + target_session = getattr(args, "session", None) + + if target_node: + st.agents_enabled[target_node] = False + print(f"✖ Disabled auto-approvals for agent: {target_node.upper()}") + elif target_session: + st.sessions_enabled[target_session] = False + print(f"✖ Disabled auto-approvals for session: '{target_session}'") + else: + st.global_enabled = False + print("✖ Disabled master auto-approvals globally.") + + st.save() + return 0 + + +def cmd_match_test(args: argparse.Namespace) -> int: + text = args.text + if text == "-" or not text: + text = sys.stdin.read() + + engine = RegexApproverEngine() + verdict = engine.evaluate(text) + + out = { + "matched": verdict.matched, + "rule_id": verdict.rule_id, + "rule_name": verdict.rule_name, + "category": verdict.category, + "key_to_send": verdict.key, + "press_enter": verdict.press_enter, + "excerpt": verdict.excerpt, + "reason": verdict.reason, + "is_blocked": verdict.is_blocked, + "blocked_reason": verdict.blocked_reason, + } + print(json.dumps(out, indent=2)) + return 0 if verdict.matched else 1 + + +def cmd_run_once(args: argparse.Namespace) -> int: + runner = AutoApproverRunner(dry_run=getattr(args, "dry_run", False)) + res = runner.run_once() + print(json.dumps(res, indent=2)) + return 0 + + +def cmd_watch(args: argparse.Namespace) -> int: + runner = AutoApproverRunner(dry_run=getattr(args, "dry_run", False)) + interval = getattr(args, "interval", 1.0) + runner.watch_loop(interval=interval) + return 0 + + +def cmd_logs(args: argparse.Namespace) -> int: + lines = getattr(args, "lines", 20) or 20 + if not AUDIT_LOG_FILE.exists(): + print("No auto-approval logs yet.") + return 0 + + with open(AUDIT_LOG_FILE) as f: + all_lines = f.readlines() + + tail = all_lines[-lines:] + for l in tail: + try: + d = json.loads(l) + ts = d.get("timestamp", "")[:19].replace("T", " ") + act = d.get("action", "") + ag = d.get("agent", "") + sess = d.get("session", "") + pane = d.get("pane", "") + key = d.get("key_sent", "") + rule = d.get("rule_name", "") + print(f"[{ts}] {act:<14} {ag.upper():<6} {sess:<12} ({pane}) -> sent '{key}' [{rule}]") + except Exception: + print(l.strip()) + return 0 + + +def cmd_spawn_worker(args: argparse.Namespace) -> int: + session = args.session + cmd = getattr(args, "command", "bash") + node = getattr(args, "node", "muse") + sock = f"/tmp/tmux-{node}.sock" if node != "muse" else "/tmp/tmux-muse.sock" + + # Spawn session + t_cmd = [TMUX_BIN, "-S", sock, "new-session", "-d", "-s", session, cmd] + res = subprocess.run(t_cmd, capture_output=True, text=True) + if res.returncode == 0: + print(f"✔ Successfully spawned worker '{session}' on {sock} running '{cmd}'") + return 0 + else: + print(f"Failed to spawn worker: {res.stderr.strip() or res.stdout.strip()}", file=sys.stderr) + return res.returncode + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Tmux Worker Tally & Regex Auto-Approval Runtime") + subparsers = parser.add_subparsers(dest="subcommand") + + # tally + p_tally = subparsers.add_parser("tally", help="Tally all tmux sessions, workers, and panes") + p_tally.add_argument("--json", action="store_true", help="Output machine-readable JSON") + p_tally.set_defaults(func=cmd_tally) + + # status + p_status = subparsers.add_parser("status", help="Show auto-approval configuration & policies") + p_status.add_argument("--json", action="store_true", help="Output machine-readable JSON") + p_status.set_defaults(func=cmd_status) + + # on / off + p_on = subparsers.add_parser("on", help="Enable auto-approvals (global, per-agent, or per-session)") + p_on.add_argument("--node", choices=FLEET_AGENTS, help="Enable for specific agent") + p_on.add_argument("--session", help="Enable for specific session name") + p_on.set_defaults(func=cmd_toggle_on) + + p_off = subparsers.add_parser("off", help="Disable auto-approvals") + p_off.add_argument("--node", choices=FLEET_AGENTS, help="Disable for specific agent") + p_off.add_argument("--session", help="Disable for specific session name") + p_off.set_defaults(func=cmd_toggle_off) + + # match + p_match = subparsers.add_parser("match", help="Test regex match against scrollback text") + p_match.add_argument("text", nargs="?", default="-", help="Input text or '-' for stdin") + p_match.set_defaults(func=cmd_match_test) + + # once + p_once = subparsers.add_parser("once", help="Evaluate and auto-approve all active prompts right now") + p_once.add_argument("--dry-run", action="store_true", help="Log matches without sending keys") + p_once.set_defaults(func=cmd_run_once) + + # watch + p_watch = subparsers.add_parser("watch", help="Run background monitor daemon for auto-approvals") + p_watch.add_argument("--interval", type=float, default=1.0, help="Poll interval in seconds (default: 1.0)") + p_watch.add_argument("--dry-run", action="store_true", help="Log matches without sending keys") + p_watch.set_defaults(func=cmd_watch) + + # logs + p_logs = subparsers.add_parser("logs", help="Tail auto-approval audit log stream") + p_logs.add_argument("-n", "--lines", type=int, default=20, help="Number of lines to show") + p_logs.set_defaults(func=cmd_logs) + + # spawn + p_spawn = subparsers.add_parser("spawn", help="Spawn a new tmux worker runner") + p_spawn.add_argument("session", help="Session name") + p_spawn.add_argument("--command", "-c", default="bash", help="Command to run") + p_spawn.add_argument("--node", choices=FLEET_AGENTS, default="muse", help="Target agent socket") + p_spawn.set_defaults(func=cmd_spawn_worker) + + return parser + + +def main(argv: Optional[List[str]] = None) -> int: + parser = build_parser() + if argv is None: + argv = sys.argv[1:] + if not argv: + parser.print_help() + return 0 + args = parser.parse_args(argv) + if not hasattr(args, "func"): + parser.print_help() + return 1 + return args.func(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/docs/BOX-TMUX-AUTO-HTTPS.md b/docs/BOX-TMUX-AUTO-HTTPS.md new file mode 100644 index 0000000..1282ffb --- /dev/null +++ b/docs/BOX-TMUX-AUTO-HTTPS.md @@ -0,0 +1,168 @@ +# Tmux Worker Tally & Regex Auto-Approvals over HTTPS (No SSH) + +> **Box is the main surface.** All operator work goes through Box (`box.muse-dev.online`). +> The web UI, `box` CLI, and agents share the same unified API endpoints and runtimes. + +**Date:** 2026-10-06 +**Status:** Implemented (`bin/tmux_auto_approver.py`, `bin/box-onboard-tui.py`, `bin/super-cli.py`) +**Scope:** Tmux worker tallies, automated regex approval engine for on-board Muse Code runs & autonomous agent workers, and dedicated interactive TUI console (`box onboard-tui`). + +--- + +## 1. Architecture + +``` + ┌──────────────────────────────────────────────┐ + │ https://box.muse-dev.online/ │ + │ (Web Dashboard & API Gateway) │ + └──────────────────────┬───────────────────────┘ + │ + HTTPS Signed Ops / CLI Dispatch + │ + ┌──────────────────────▼───────────────────────┐ + │ Unified Box CLI Engine │ + │ (box tmux tally / box tmux auto ...) │ + └───────┬───────────────────────────────┬──────┘ + │ │ + ┌──────────────▼─────────────┐ ┌─────────────▼──────────────┐ + │ bin/box-onboard-tui.py │ │ bin/tmux_auto_approver.py │ + │ (Dedicated 4-Tab Console) │ │ (Multi-Socket Regex Daemon)│ + └──────────────┬─────────────┘ └─────────────┬──────────────┘ + │ │ + │ │ + ┌───────────────────────┴───────────────────────────────┴───────────────────────┐ + │ Tmux Sockets Monitored │ + │ • /tmp/tmux-muse.sock (shared host workers) │ + │ • /tmp/tmux-1000/default (dev/def runner panes & muse-code %37) │ + │ • /tmp/tmux-1000/lte (lte operator pane) │ + │ • /tmp/tmux-.sock (per-agent netns sockets: pip, 646, opm, dev, def) │ + └───────────────────────────────────────────────────────────────────────────────┘ +``` + +--- + +## 2. Tmux Auto-Approval Regex Match Engine + +The auto-approval engine monitors scrollback across all agent panes and matches approval prompts against priority rules: + +| Rule ID | Category | Trigger Pattern | Response Key | Press Enter | Description | +|---|---|---|---|---|---| +| `muse_code_run_numbered` | `muse_code` | `Would you like to run the following[\s\S]*?›\s*1\.\s*Yes,?\s*proceed` | `1` | `false` | Muse Code interactive run menu (selects option 1) | +| `muse_code_run_yn` | `muse_code` | `›\s*1\.\s*Yes,?\s*proceed\s*\(y\)` | `1` | `false` | Active selection indicator on `1. Yes, proceed (y)` | +| `muse_code_allow_execution` | `muse_code` | `Allow\s+execution\s+of\b[\s\S]*?\[y/N\]` | `y` | `true` | Approves script execution confirmation | +| `choice_abc` | `choice` | `(?i)(?:choose\|choice\|select)[\s\S]*?^\s*[A-Z]\s*[.\)\-:]\s+\S` | `A` | `true` | Lettered decision choice menus | +| `menu_numbered` | `menu` | `(?i)(?:Option:\|Selection:)[\s\S]*?^\s*\(?1\)?\s+[A-Za-z]` | `1` | `true` | Numbered selection menus | +| `confirm_yn` | `confirm` | `([yY]/[nN]\|\[[yY]/[nN]\])\s*[\]:)>]?\s*$` | `y` | `true` | Line-end confirmation prompts | +| `enter_to_continue` | `enter` | `(?i)(?:Press\s+\[?Enter\]?\s+to\s+continue)` | `Enter` | `false` | Enter-to-continue banners | + +### Guardrails (Never Auto-Approved) +- `[sudo] password for ...` / `password:` prompts +- Passkeys, private key passphrases, and PIN prompts +- Destructive operations (`rm -rf /`, `mkfs.*`) + +When a guardrail pattern is detected, the engine flags `is_blocked=true`, emits a warning audit log, and notifies the human operator. + +--- + +## 3. CLI Commands + +### Tmux Worker Tally & Management +```bash +# Tally all active tmux sessions, panes, and workers across sockets +box tmux tally +box tmux tally --json + +# List active sessions +box tmux list + +# Spawn new background worker session +box tmux new my-worker -c "python3 run_tasks.py" +``` + +### Auto-Approval Control +```bash +# Query master state and per-agent policies +box tmux auto status +box tmux auto status --json + +# Master enable / disable +box tmux auto on +box tmux auto off + +# Enable / disable for specific agent +box tmux auto on --node muse +box tmux auto off --node 646 + +# Execute single-pass scan and auto-approve all active prompts right now +box tmux auto once +box tmux auto once --dry-run + +# Run background monitor daemon +box tmux auto watch --interval 1.0 + +# Tail structured audit log stream +box tmux auto logs -n 20 + +# Test regex match against custom prompt text +box tmux auto match "Would you like to run the following ... › 1. Yes, proceed (y)" +``` + +### Onboard Connects +```bash +# View all fleet nodes & client onboard connects +box onboard connects +box onboard connects --json + +# Start client onboarding +box onboard start dev2 --email client@example.com --for 646 + +# Submit OTP verification code +box onboard submit-otp dev2 123456 +``` + +### Dedicated Interactive TUI +```bash +# Launch full 4-tab interactive TUI +box onboard-tui +box tui onboard +``` + +--- + +## 4. HTTPS Remote Operations (`exec-constrained.py`) + +Available via `https://exec.muse-dev.online/exec` with signature verification: + +| Op | Parameters | Description | Permission | +|---|---|---|---| +| `tmux.tally` | `{}` | Returns complete worker tally across all sockets (JSON) | `DEFAULT_PERMS` (Read-only) | +| `tmux.auto_status` | `{}` | Returns auto-approval toggle state & rule set (JSON) | `DEFAULT_PERMS` (Read-only) | +| `onboard.connects` | `{}` | Returns consolidated fleet and client connects (JSON) | `DEFAULT_PERMS` (Read-only) | +| `tmux.auto_toggle` | `{"enabled": bool, "node"?: str}` | Toggles master or per-agent auto-approval state | Known-Identities-Only | + +--- + +## 5. Audit Logging + +Every auto-approval and blocked guardrail event is written to: +`/home/super/Projects/NetVM/logs/tmux/auto-approvals.jsonl` + +Sample event payload: +```json +{ + "action": "AUTO_APPROVED", + "socket": "/tmp/tmux-1000/default", + "pane": "%37", + "session": "muse", + "agent": "muse", + "rule_id": "muse_code_run_numbered", + "rule_name": "Muse Code Run (Numbered)", + "key_sent": "1", + "press_enter": false, + "excerpt": "Would you like to run the following ... › 1. Yes, proceed (y)", + "dry_run": false, + "success": true, + "timestamp": "2026-10-06T19:34:19.599811+00:00", + "ts": 1791315259.6 +} +``` diff --git a/tests/test_box_onboard_tui.py b/tests/test_box_onboard_tui.py new file mode 100644 index 0000000..5927057 --- /dev/null +++ b/tests/test_box_onboard_tui.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +"""test_box_onboard_tui.py — Unit tests for dedicated Onboard Connects & Tmux Auto-Approver TUI.""" + +import curses +import os +import sys +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +REPO_ROOT = Path(__file__).resolve().parent.parent +import importlib.util +tui_path = REPO_ROOT / "bin" / "box-onboard-tui.py" +spec = importlib.util.spec_from_file_location("box_onboard_tui", tui_path) +box_onboard_tui = importlib.util.module_from_spec(spec) +sys.modules["box_onboard_tui"] = box_onboard_tui +spec.loader.exec_module(box_onboard_tui) +BoxOnboardTUI = box_onboard_tui.BoxOnboardTUI + + +class TestBoxOnboardTUI(unittest.TestCase): + @patch("curses.init_pair") + @patch("curses.color_pair") + @patch("curses.has_colors", return_value=True) + @patch("curses.start_color") + @patch("box_onboard_tui.gather_tmux_tally") + @patch("box_onboard_tui.get_all_connects") + def test_tui_initialization_and_tabs( + self, mock_connects, mock_tally, mock_start_color, mock_has_colors, mock_color_pair, mock_init_pair + ): + mock_stdscr = MagicMock() + mock_stdscr.getmaxyx.return_value = (40, 120) + + mock_connects.return_value = [ + {"node": "muse", "type": "fleet_agent", "email": "muse@test.com", "stage": "active_fleet"}, + {"node": "test-client", "type": "onboard_pipeline", "email": "c@test.com", "stage": "awaiting_otp"}, + ] + + mock_tally.return_value = MagicMock( + total_sockets=2, + total_sessions=3, + total_panes=5, + active_workers=2, + panes=[], + by_agent={"muse": {"sessions": 1, "panes": 2, "active_commands": ["muse-bin"], "auto_approve": True}}, + ) + + app = BoxOnboardTUI(mock_stdscr) + self.assertEqual(app.current_tab, 0) + self.assertEqual(len(app.tabs), 4) + self.assertEqual(len(app.connects), 2) + + # Tab navigation via key + app._handle_key(ord("\t")) + self.assertEqual(app.current_tab, 1) + + app._handle_key(ord("\t")) + self.assertEqual(app.current_tab, 2) + + app._handle_key(ord("\t")) + self.assertEqual(app.current_tab, 3) + + app._handle_key(ord("\t")) + self.assertEqual(app.current_tab, 0) + + @patch("curses.init_pair") + @patch("curses.color_pair") + @patch("curses.has_colors", return_value=True) + @patch("box_onboard_tui.gather_tmux_tally") + @patch("box_onboard_tui.get_all_connects") + def test_navigation_keys( + self, mock_connects, mock_tally, mock_has_colors, mock_color_pair, mock_init_pair + ): + mock_stdscr = MagicMock() + mock_stdscr.getmaxyx.return_value = (40, 120) + mock_connects.return_value = [ + {"node": f"node-{i}", "type": "fleet_agent"} for i in range(5) + ] + + app = BoxOnboardTUI(mock_stdscr) + self.assertEqual(app.sel_connect_idx, 0) + + # Move down + app._handle_key(curses.KEY_DOWN) + self.assertEqual(app.sel_connect_idx, 1) + + # Move up + app._handle_key(curses.KEY_UP) + self.assertEqual(app.sel_connect_idx, 0) + + # Switch tab via number key '2' + app._handle_key(ord("2")) + self.assertEqual(app.current_tab, 1) + + # Switch tab via number key '3' + app._handle_key(ord("3")) + self.assertEqual(app.current_tab, 2) + + # Switch tab via number key '4' + app._handle_key(ord("4")) + self.assertEqual(app.current_tab, 3) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_tmux_auto_approver.py b/tests/test_tmux_auto_approver.py new file mode 100644 index 0000000..a600c0a --- /dev/null +++ b/tests/test_tmux_auto_approver.py @@ -0,0 +1,224 @@ +#!/usr/bin/env python3 +"""test_tmux_auto_approver.py — Unit tests for Tmux worker management and regex auto-approver.""" + +import json +import os +import sys +import tempfile +import time +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +REPO_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO_ROOT / "bin")) + +import tmux_auto_approver +from tmux_auto_approver import ( + DEFAULT_RULES, + GUARDRAIL_PATTERNS, + AutoApproverRunner, + AutoApproverState, + MatchRule, + MatchVerdict, + RegexApproverEngine, + TmuxPaneInfo, + TmuxWorkerTally, + gather_tmux_tally, + infer_agent_for_session, +) + + +class TestRegexApproverEngine(unittest.TestCase): + def setUp(self): + self.engine = RegexApproverEngine(DEFAULT_RULES) + + def test_muse_code_run_numbered_match(self): + text = """ +Terminal output before... +Would you like to run the following bash command? + echo "hello" +› 1. Yes, proceed (y) + 2. No, skip (n) +""" + verdict = self.engine.evaluate(text) + self.assertTrue(verdict.matched) + self.assertEqual(verdict.rule_id, "muse_code_run_numbered") + self.assertEqual(verdict.key, "1") + self.assertFalse(verdict.press_enter) + + def test_muse_code_run_yn_match(self): + text = "Some actions\n› 1. Yes, proceed (y)\n" + verdict = self.engine.evaluate(text) + self.assertTrue(verdict.matched) + self.assertIn("muse_code_run", verdict.rule_id) + self.assertEqual(verdict.key, "1") + + def test_muse_code_allow_execution_match(self): + text = "Allow execution of script /tmp/test.sh? [y/N]" + verdict = self.engine.evaluate(text) + self.assertTrue(verdict.matched) + self.assertEqual(verdict.rule_id, "muse_code_allow_execution") + self.assertEqual(verdict.key, "y") + self.assertTrue(verdict.press_enter) + + def test_choice_abc_match(self): + text = "Select option from the choices below:\nA) Deploy to production\nB) Staging" + verdict = self.engine.evaluate(text) + self.assertTrue(verdict.matched) + self.assertEqual(verdict.rule_id, "choice_abc") + self.assertEqual(verdict.key, "A") + self.assertTrue(verdict.press_enter) + + def test_menu_numbered_match(self): + text = "Option: Please pick a number:\n(1) Accept terms\n(2) Decline" + verdict = self.engine.evaluate(text) + self.assertTrue(verdict.matched) + self.assertEqual(verdict.rule_id, "menu_numbered") + self.assertEqual(verdict.key, "1") + self.assertTrue(verdict.press_enter) + + def test_confirm_yn_match(self): + text = "Do you want to continue? [y/n]: " + verdict = self.engine.evaluate(text) + self.assertTrue(verdict.matched) + self.assertEqual(verdict.rule_id, "confirm_yn") + self.assertEqual(verdict.key, "y") + self.assertTrue(verdict.press_enter) + + def test_enter_to_continue_match(self): + text = "Task completed successfully. Press [Enter] to continue..." + verdict = self.engine.evaluate(text) + self.assertTrue(verdict.matched) + self.assertEqual(verdict.rule_id, "enter_to_continue") + self.assertEqual(verdict.key, "Enter") + + def test_guardrail_sudo_password_blocked(self): + text = "[sudo] password for super: " + verdict = self.engine.evaluate(text) + self.assertFalse(verdict.matched) + self.assertTrue(verdict.is_blocked) + self.assertIn("guardrail", verdict.blocked_reason.lower()) + + def test_guardrail_passkey_blocked(self): + text = "Insert security key or enter passkey PIN:" + verdict = self.engine.evaluate(text) + self.assertFalse(verdict.matched) + self.assertTrue(verdict.is_blocked) + self.assertIn("guardrail", verdict.blocked_reason.lower()) + + def test_guardrail_destructive_command_blocked(self): + text = "Running dangerous cleanup: rm -rf /" + verdict = self.engine.evaluate(text) + self.assertFalse(verdict.matched) + self.assertTrue(verdict.is_blocked) + self.assertIn("guardrail", verdict.blocked_reason.lower()) + + def test_empty_text_returns_unmatched(self): + verdict = self.engine.evaluate("") + self.assertFalse(verdict.matched) + self.assertFalse(verdict.is_blocked) + + +class TestAgentInference(unittest.TestCase): + def test_infer_agent_names(self): + self.assertEqual(infer_agent_for_session("/tmp/tmux-muse.sock", "worker-1"), "muse") + self.assertEqual(infer_agent_for_session("/tmp/tmux-pip.sock", "agent-job"), "pip") + self.assertEqual(infer_agent_for_session("/tmp/tmux-1000/default", "646-retention"), "646") + self.assertEqual(infer_agent_for_session("/tmp/tmux-1000/default", "opm"), "opm") + self.assertEqual(infer_agent_for_session("/tmp/tmux-1000/default", "dev-test"), "dev") + self.assertEqual(infer_agent_for_session("/tmp/tmux-1000/default", "def"), "def") + self.assertEqual(infer_agent_for_session("/tmp/tmux-1000/lte", "main"), "host") + + +class TestAutoApproverRunner(unittest.TestCase): + def setUp(self): + self.temp_dir = tempfile.TemporaryDirectory() + self.orig_state = tmux_auto_approver.STATE_FILE + self.orig_audit = tmux_auto_approver.AUDIT_LOG_FILE + tmux_auto_approver.STATE_FILE = Path(self.temp_dir.name) / "test_state.json" + tmux_auto_approver.AUDIT_LOG_FILE = Path(self.temp_dir.name) / "test_audit.jsonl" + + def tearDown(self): + tmux_auto_approver.STATE_FILE = self.orig_state + tmux_auto_approver.AUDIT_LOG_FILE = self.orig_audit + self.temp_dir.cleanup() + + @patch("tmux_auto_approver.run_tmux_cmd") + @patch("tmux_auto_approver.capture_pane_text") + @patch("tmux_auto_approver.gather_tmux_tally") + def test_run_once_dry_run_dispatches_match(self, mock_tally, mock_capture, mock_tmux_cmd): + mock_pane = TmuxPaneInfo( + socket="/tmp/tmux-1000/default", + session="muse", + window_idx=0, + pane_id="%37", + pane_pid=12345, + current_command="muse-bin", + active=True, + attached=True, + title="muse terminal", + agent_node="muse", + auto_approve=True, + ) + + mock_tally.return_value = TmuxWorkerTally( + total_sockets=1, + total_sessions=1, + total_panes=1, + active_workers=1, + by_agent={"muse": {"sessions": 1, "panes": 1, "active_commands": ["muse-bin"], "auto_approve": True}}, + panes=[mock_pane], + ) + + mock_capture.return_value = "Would you like to run the following?\n› 1. Yes, proceed (y)" + + runner = AutoApproverRunner(dry_run=True) + results = runner.run_once() + + self.assertEqual(len(results), 1) + self.assertEqual(results[0]["action"], "DRY_RUN_MATCH") + self.assertEqual(results[0]["key_sent"], "1") + self.assertEqual(results[0]["rule_name"], "Muse Code Run (Numbered)") + + # Verify deduplication within 15 seconds + results2 = runner.run_once() + self.assertEqual(len(results2), 0) + + @patch("tmux_auto_approver.gather_tmux_tally") + def test_run_once_global_disabled(self, mock_tally): + runner = AutoApproverRunner(dry_run=True) + runner.state.global_enabled = False + runner.state.save() + results = runner.run_once() + self.assertEqual(len(results), 1) + self.assertEqual(results[0]["status"], "disabled") + + +class TestTallyGathering(unittest.TestCase): + @patch("tmux_auto_approver.get_existing_sockets") + @patch("tmux_auto_approver.run_tmux_cmd") + def test_gather_tmux_tally_parsing(self, mock_run_cmd, mock_sockets): + mock_sockets.return_value = ["/tmp/tmux-1000/default"] + # Format: #{session_name}___#{window_index}___#{pane_id}___#{pane_pid}___#{pane_current_command}___#{pane_active}___#{session_attached}___#{pane_title} + mock_run_cmd.return_value = ( + 0, + "muse___0___%1___1001___muse-bin___1___1___muse worker\n" + "dev___0___%2___1002___bash___0___0___dev terminal\n", + "", + ) + + state = AutoApproverState() + tally = gather_tmux_tally(state) + + self.assertEqual(tally.total_sockets, 1) + self.assertEqual(tally.total_sessions, 2) + self.assertEqual(tally.total_panes, 2) + self.assertEqual(tally.active_workers, 1) # only muse-bin is active, bash is shell + self.assertEqual(len(tally.panes), 2) + self.assertEqual(tally.panes[0].agent_node, "muse") + self.assertEqual(tally.panes[1].agent_node, "dev") + + +if __name__ == "__main__": + unittest.main()