box/bin/tmux_auto_approver.py

#!/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-<node>.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"
   - Interview navigate+select menus (cursor on 1 -> Enter)
   - 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="interview_select",
        name="Interview Menu (cursor on 1)",
        pattern=(r"\?\s*\n"
                 r"(?:[^\n]*\n){0,8}"
                 r"[ \t]*(?:›|>)[ \t]*1\.[ \t]+\S[^\n]*\n"
                 r"(?:[^\n]*\n){0,10}"
                 r"[ \t]*2\.[ \t]+\S"),
        response_key="Enter",
        category="enter",
        enabled=True,
        description=("Selects highlighted option 1 on navigate+select "
                     "menus (cursor on 1. + 2. + ?-question above)"),
        press_enter=False,
    ),
    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)
            st = cls(**data)
        except Exception:
            return cls()
        # Migrate: append built-in rules missing from stored state (a new
        # default must reach the daemon without wiping operator toggles).
        try:
            have = {r.get("id") for r in st.rules
                    if isinstance(r, dict)}
            missing = [asdict(r) for r in DEFAULT_RULES
                       if r.id not in have]
            if missing:
                st.rules.extend(missing)
                st.save()
        except Exception:
            pass
        return st


# =====================================================================
# 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, joining wrapped rows.

    -J joins physical wrapped lines into logical lines so matching is
    width-independent: narrow panes wrap the same dialog onto more
    rows, which otherwise breaks cue/option regexes.
    """
    rc, out, _ = run_tmux_cmd(socket_path, "capture-pane", "-p", "-J",
                              "-t", pane_id, "-S", f"-{lines}")
    if rc == 0:
        return out
    return ""


MUSE_COMMAND_HINTS = ("muse-bin", "muse-code")

# (socket, pane) ever observed running a muse runtime. pane_current_command
# flickers to the child tool while the agent works, so a muse pane stays
# muse-owned when its foreground reads "python3" (observed live: the hint
# gate missed tool-running panes and both daemons stacked 'y' answers).
_MUSE_PANES_SEEN = set()

# tmux rule category -> muse watcher kind for verified sends. Text-input
# categories verify render + submit with one retry; single-key widgets
# (and unknown categories) stay blind.
_CATEGORY_KIND_MAP = {
    "choice": "letter",
    "menu": "numbered",
    "confirm": "yn",
    "muse_code": "muse-approval",
    "enter": None,
}


def should_defer_to_muse_watcher(socket_path: str, pane_id: str,
                                 current_command: str) -> bool:
    """True when a per-pane muse watcher owns this pane.

    Single-owner rule: muse_choice_watcher is authoritative for muse
    panes (stability + re-verify + once-per-prompt + decided-block
    guard). When its daemon is alive for this socket:pane, tmux must
    skip the pane entirely, or both daemons answer the same prompt
    within the same second ('11' + stray keys, observed live; later the
    same hole stacked 'y' answers when the foreground flickered to a
    child tool mid-poll). Never raises: import or liveness failures
    mean no owner, handle here.
    """
    try:
        import muse_choice_watcher as mcw
        cmd = current_command or ""
        key = (socket_path, pane_id)
        if any(h in cmd for h in MUSE_COMMAND_HINTS):
            _MUSE_PANES_SEEN.add(key)
            alive = getattr(mcw, "watcher_alive", mcw.is_running)
            return alive(socket_path, pane_id) is not None
        if key in _MUSE_PANES_SEEN:
            alive = getattr(mcw, "watcher_alive", mcw.is_running)
            return alive(socket_path, pane_id) is not None
        # Never observed as muse: cheap pidfile check only (covers a
        # watcher racing ahead of our first observation of the pane).
        return mcw.is_running(socket_path, pane_id) is not None
    except Exception:
        return False


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
            if should_defer_to_muse_watcher(p.socket, p.pane_id,
                                            p.current_command):
                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.
                # Keyed by socket:pane: bare pane ids repeat on every
                # tmux socket, so %1 on pip must not suppress %1 on opm.
                sig = hashlib.sha1(f"{verdict.rule_id}:{verdict.excerpt}".encode()).hexdigest()
                dedup_key = "%s:%s" % (p.socket, p.pane_id)
                last_time, last_sig = self.recent_signatures.get(dedup_key, (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 through the verified send path:
                # literal text paced apart from Enter (a single-call
                # burst arrives as paste and lands a newline in
                # composers instead of submitting, then re-fires past
                # dedup and stacks). Text-input categories also verify
                # render + submit with one retry; single-key widgets
                # stay blind.
                success = False
                detail = {"verified": None, "retried": False}
                if not self.dry_run:
                    import muse_choice_watcher as mcw
                    want_enter = (verdict.key != "Enter"
                                  and bool(verdict.press_enter))
                    ok, detail = mcw.send_answer(
                        p.socket, p.pane_id, verdict.key,
                        enter=want_enter,
                        kind=_CATEGORY_KIND_MAP.get(verdict.category),
                        sig=sig)
                    success = bool(ok)
                else:
                    success = True  # dry-run simulated

                now = time.time()
                self.recent_signatures[dedup_key] = (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,
                    "verified": detail["verified"],
                    "retried": detail["retried"],
                }
                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())