721 lines
31 KiB
Python
721 lines
31 KiB
Python
|
#!/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: <session_name> [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())
|