box/bin/muse-tmux.py

#!/usr/bin/env python3
"""
muse-tmux.py: Shared tmux socket manager for Muse agents and operators.
Socket location: /tmp/tmux-muse.sock (shared across fleet agents & super).
"""
import sys
import os
import re
import subprocess
import argparse

SOCKET_PATH = "/tmp/tmux-muse.sock"
TMUX_BIN = "/home/super/.local/bin/tmux"
if not os.path.exists(TMUX_BIN):
    TMUX_BIN = "tmux"

def _run_tmux(*args, capture=True):
    cmd = [TMUX_BIN, "-S", SOCKET_PATH] + list(args)
    res = subprocess.run(cmd, capture_output=capture, text=True)
    return res.returncode, res.stdout, res.stderr

def cmd_new(args):
    session = args.session
    window = getattr(args, "window", None)
    command = getattr(args, "command", None)
    
    # Check if session already exists
    rc, out, _ = _run_tmux("has-session", "-t", session)
    if rc == 0:
        print(f"Session '{session}' already exists on shared socket {SOCKET_PATH}.")
        return 0
        
    t_args = ["new-session", "-d", "-s", session]
    if window:
        t_args.extend(["-n", window])
    if command:
        t_args.append(command)
        
    rc, out, err = _run_tmux(*t_args)
    if rc == 0:
        print(f"✔ Created session '{session}' on {SOCKET_PATH}")
        return 0
    else:
        print(f"Failed to create session: {err.strip() or out.strip()}", file=sys.stderr)
        return rc

def cmd_send(args):
    session = args.session
    keys = args.keys
    enter = not getattr(args, "no_enter", False)
    
    # Check session
    rc, _, _ = _run_tmux("has-session", "-t", session)
    if rc != 0:
        # Auto-create if not exists
        print(f"Notice: Session '{session}' does not exist; creating now...", file=sys.stderr)
        _run_tmux("new-session", "-d", "-s", session)
        
    t_args = ["send-keys", "-t", session, keys]
    if enter:
        t_args.append("Enter")
        
    rc, out, err = _run_tmux(*t_args)
    if rc == 0:
        print(f"✔ Sent keys to '{session}'")
        return 0
    else:
        print(f"Failed to send keys: {err.strip() or out.strip()}", file=sys.stderr)
        return rc

def cmd_capture(args):
    session = args.session
    lines = getattr(args, "lines", 30) or 30
    
    t_args = ["capture-pane", "-p", "-t", session, "-S", f"-{lines}"]
    rc, out, err = _run_tmux(*t_args)
    if rc == 0:
        # Strip trailing blank lines
        clean = out.rstrip()
        print(clean)
        return 0
    else:
        print(f"Failed to capture pane: {err.strip() or out.strip()}", file=sys.stderr)
        return rc

def cmd_list(args):
    rc, out, err = _run_tmux("list-sessions")
    if rc == 0:
        print(out.strip() or "No active sessions.")
        return 0
    elif "no server running" in (err or "").lower() or rc == 1:
        print(f"No active tmux server on {SOCKET_PATH}.")
        return 0
    else:
        print(f"Error: {err.strip() or out.strip()}", file=sys.stderr)
        return rc

def cmd_kill(args):
    session = args.session
    rc, out, err = _run_tmux("kill-session", "-t", session)
    if rc == 0:
        print(f"✔ Killed session '{session}'")
        return 0
    else:
        print(f"Error: {err.strip() or out.strip()}", file=sys.stderr)
        return rc

def cmd_attach(args):
    session = args.session
    cmd = [TMUX_BIN, "-S", SOCKET_PATH, "attach", "-t", session]
    # Replace current process for interactive attach
    os.execv(TMUX_BIN, cmd)

def main():
    parser = argparse.ArgumentParser(description="Shared Muse tmux socket manager")
    subparsers = parser.add_subparsers(dest="action")
    
    # list
    p_ls = subparsers.add_parser("list", aliases=["ls"], help="List sessions on shared socket")
    p_ls.set_defaults(func=cmd_list)
    
    # new
    p_new = subparsers.add_parser("new", help="Create new background session")
    p_new.add_argument("session", help="Session name")
    p_new.add_argument("--window", "-w", help="Initial window name")
    p_new.add_argument("--command", "-c", help="Command to run in session")
    p_new.set_defaults(func=cmd_new)
    
    # send / send-keys
    p_send = subparsers.add_parser("send", aliases=["send-keys"], help="Send keys to a session")
    p_send.add_argument("session", help="Target session name")
    p_send.add_argument("keys", help="Keys / command string to send")
    p_send.add_argument("--no-enter", action="store_true", help="Do not send Enter key after keys")
    p_send.set_defaults(func=cmd_send)
    
    # capture
    p_cap = subparsers.add_parser("capture", aliases=["cap", "tail"], help="Capture pane output")
    p_cap.add_argument("session", help="Target session name")
    p_cap.add_argument("--lines", "-n", type=int, default=30, help="Number of scrollback lines to capture (default: 30)")
    p_cap.set_defaults(func=cmd_capture)
    
    # kill
    p_kill = subparsers.add_parser("kill", help="Kill a session")
    p_kill.add_argument("session", help="Session name")
    p_kill.set_defaults(func=cmd_kill)
    
    # attach
    p_att = subparsers.add_parser("attach", help="Attach to a session interactively")
    p_att.add_argument("session", help="Session name")
    p_att.set_defaults(func=cmd_attach)
    
    if len(sys.argv) == 1:
        parser.print_help()
        sys.exit(0)
        
    args = parser.parse_args()
    if not hasattr(args, "func"):
        parser.print_help()
        sys.exit(1)
        
    sys.exit(args.func(args))

if __name__ == "__main__":
    main()