diff --git a/bin/agent-manager.py b/bin/agent-manager.py new file mode 100755 index 0000000..64ecdc1 --- /dev/null +++ b/bin/agent-manager.py @@ -0,0 +1,905 @@ +#!/usr/bin/env python3 +"""agent-manager.py — Agent runs across tailnet devices (stdlib curses). + +Read-only dashboard that aggregates agent harness runs (any CLI / bin +runtime: muse, agy, and whatever else matches the open taxonomy) over +SSH to every reachable tailnet device, then sorts by agent type. + +Surfaces 3 read-only views (no actions in v1): + [1] RUNS — every run, sorted by agent type, then device. + [2] TYPES — counts per agent type with per-device breakdown. + [3] DEVICES — per-device reachability, run counts, and notes. + +Structure mirrors bin/box-fleet-tui.py: all data-gathering lives in +pure, testable functions taking injected runners (see gather_*); the +curses UI is a thin renderer over those functions. The backend differs: +instead of local repo files, each refresh fans out over tailnet SSH +(one call per device: tmux panes + process table, joined locally). +Unreachable devices yield "n/a" rows — never a crash. + +Usage: + python3 bin/agent-manager.py [--once [--json]] # non-interactive dump +""" + +from __future__ import annotations + +import concurrent.futures +import curses +import json +import os +import re +import subprocess +import sys +import time +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple + +REPO_ROOT = Path(__file__).resolve().parent.parent + +NA = "n/a" + +RunFn = Callable[..., Tuple[int, str]] + +# Per-device SSH budget; whole-fleet refresh runs devices in parallel. +DEVICE_TIMEOUT_S = 15 +SSH_OPTS = ["-o", "BatchMode=yes", "-o", "ConnectTimeout=5"] + +# Only these OS classes get an SSH probe (from `tailscale status`). +SSH_OS = ("linux", "macos") + +# ===================================================================== +# Agent-type taxonomy (open: unknown bin runtimes still show up) +# ===================================================================== +# +# Classification keys off the argv[0] basename so a wrapper path never +# matters (/home/super/.local/bin/agy.bin -> agy). Scripts (*.py, +# *.sh) are never harness runtimes. Anything shaped like a bin runtime +# (*.bin, *-bin-*) that is not otherwise known shows under its own +# "other:" type instead of being dropped. + +HARNESS_EXACT = { + "agy": "agy", + "agy.bin": "agy", + "muse-code": "muse", + "Muse": "muse", + "claude": "claude", + "codex": "codex", + "gemini": "gemini", + "aider": "aider", + "opencode": "opencode", + "crush": "crush", + "amp": "amp", +} + +HARNESS_PREFIX = ( + ("muse-bin", "muse"), +) + +# Bin-shaped names that are infrastructure, not agent harnesses. +HARNESS_DENY = frozenset({"tmux.bin", "tmux", "ssh.bin"}) + +SCRIPT_SUFFIXES = (".py", ".pyc", ".sh", ".pl", ".rb", ".js") + + +def classify_harness(argv0: str) -> Optional[Tuple[str, str]]: + """Map an argv[0] to (agent_type, bin_name); None when not a harness. + + Known harnesses collapse to a canonical type ("agy"); unknown bin + runtimes keep their own "other:" type so new harnesses + appear without a code change. + """ + base = os.path.basename((argv0 or "").strip().strip("'\"")) + if not base: + return None + low = base.lower() + if low in HARNESS_DENY: + return None + if low.endswith(SCRIPT_SUFFIXES): + return None + if low in HARNESS_EXACT: + return HARNESS_EXACT[low], base + for prefix, typ in HARNESS_PREFIX: + if low.startswith(prefix): + return typ, base + if low.endswith(".bin") or "-bin-" in low or low.startswith("bin-"): + return "other:%s" % base, base + return None + + +def type_sort_key(agent_type: str) -> Tuple[int, str]: + """Known types first (alpha), then other:* (alpha).""" + if agent_type.startswith("other:"): + return 1, agent_type + return 0, agent_type + + +# ===================================================================== +# Default IO primitives (injectable seams for tests) +# ===================================================================== + +def _run(cmd: List[str], timeout: int = 15) -> Tuple[int, str]: + """Run cmd, capture output. Returns (returncode, combined_output).""" + try: + r = subprocess.run(cmd, capture_output=True, text=True, + timeout=timeout) + return r.returncode, ((r.stdout or "") + (r.stderr or "")).strip() + except subprocess.TimeoutExpired: + return 124, "timed out after %ds: %s" % (timeout, " ".join(cmd)) + except OSError as e: + return 127, str(e) + + +def _run_ssh(device: str, remote_cmd: str, + timeout: int = DEVICE_TIMEOUT_S, + run: Optional[RunFn] = None) -> Tuple[int, str]: + """Run one remote command over tailnet SSH. Fails closed, never raises.""" + run = run or _run + try: + return run(["ssh"] + SSH_OPTS + [device, remote_cmd], + timeout=timeout) + except Exception as e: + return 127, str(e) + + +# ===================================================================== +# Pure parsers +# ===================================================================== + +def parse_tailscale_status(out: str) -> List[Dict[str, Any]]: + """Parse `tailscale status` into [{name, ip, os, online, detail}]. + + Unparseable lines are skipped; a warning preamble is ignored. + """ + devices: List[Dict[str, Any]] = [] + for line in (out or "").splitlines(): + line = line.rstrip() + if not line or line.startswith("Warning:"): + continue + parts = line.split() + if len(parts) < 4: + continue + ip, name, _user, osname = parts[0], parts[1], parts[2], parts[3] + if not re.match(r"^\d+\.\d+\.\d+\.\d+$", ip): + continue + detail = " ".join(parts[4:]) if len(parts) > 4 else "" + online = not detail.startswith("offline") + devices.append({"name": name, "ip": ip, "os": osname.lower(), + "online": online, "detail": detail or NA}) + return devices + + +PANE_PREFIX = "PANE|" +PS_MARKER = "__PS__" + + +def parse_panes(out: str) -> List[Dict[str, Any]]: + """Parse tmux pane lines (PANE|sock|id|pid|session|cmd|title). + + Pane ids repeat across servers, so (socket, id) is the identity. + Session groups repeat a pane under several sessions; dedupe by + (socket, id), joining session names. The legacy 4-field shape + (no socket) still parses with sock="". + """ + seen: Dict[Tuple[str, str], Dict[str, Any]] = {} + order: List[Tuple[str, str]] = [] + for line in (out or "").splitlines(): + if not line.startswith(PANE_PREFIX): + continue + fields = line[len(PANE_PREFIX):].split("|") + if len(fields) >= 6: + sock, pane_id, pid_s, session, cmd = ( + fields[0].strip(), fields[1].strip(), fields[2].strip(), + fields[3].strip(), fields[4].strip()) + title = "|".join(fields[5:]).strip() + elif len(fields) >= 4: + sock, pane_id, pid_s, session, cmd = ( + "", fields[0].strip(), fields[1].strip(), + fields[2].strip(), fields[3].strip()) + title = "|".join(fields[4:]).strip() + else: + continue + try: + pid = int(pid_s) + except ValueError: + continue + session = session or NA + key = (sock, pane_id) + if key in seen: + prev = seen[key]["session"] + if session not in prev.split(","): + seen[key]["session"] = prev + "," + session + continue + seen[key] = {"sock": sock, "pane": pane_id, "pid": pid, + "session": session, "cmd": cmd, "title": title} + order.append(key) + return [seen[k] for k in order] + + +def display_session(run: Dict[str, Any]) -> str: + """Session label; socket-qualified unless it is the default server.""" + session = str(run.get("session", NA)) + sock = run.get("sock") or "" + if session == NA or sock in ("", "default"): + return session + return "%s:%s" % (sock, session) + + +def parse_ps(out: str) -> Dict[int, Dict[str, Any]]: + """Parse `ps -eo pid,ppid,etime,command` into {pid: rec}.""" + procs: Dict[int, Dict[str, Any]] = {} + for line in (out or "").splitlines(): + parts = line.split(None, 3) + if len(parts) < 4: + continue + try: + pid, ppid = int(parts[0]), int(parts[1]) + except ValueError: + continue # header row + procs[pid] = {"pid": pid, "ppid": ppid, "etime": parts[2], + "args": parts[3]} + return procs + + +def split_scan(out: str) -> Tuple[str, str]: + """Split a device scan into (pane_text, ps_text) at the marker.""" + if PS_MARKER in out: + pane_text, _, ps_text = out.partition(PS_MARKER) + return pane_text, ps_text + return out, "" + + +def _descendants(procs: Dict[int, Dict[str, Any]], + root: int) -> List[int]: + """Pids under root (breadth-first via ppid links).""" + kids: Dict[int, List[int]] = {} + for pid, rec in procs.items(): + kids.setdefault(rec["ppid"], []).append(pid) + out: List[int] = [] + queue = list(kids.get(root, [])) + seen = {root} + while queue: + pid = queue.pop(0) + if pid in seen: + continue + seen.add(pid) + out.append(pid) + queue.extend(kids.get(pid, [])) + return out + + +def join_runs(panes: List[Dict[str, Any]], + procs: Dict[int, Dict[str, Any]]) -> List[Dict[str, Any]]: + """Join tmux panes with harness processes into run records. + + A pane is a run when its current command classifies as a harness + or a harness binary runs among its descendants. Harness processes + under no pane surface as bare runs (session/pane n/a). Returns + records sorted by (agent_type, session, pane). + """ + runs: List[Dict[str, Any]] = [] + claimed: set = set() + pane_roots = {p["pid"] for p in panes} + + for pane in panes: + argv = (pane.get("cmd") or "").strip() + pane_hit = classify_harness(argv.split()[0] if argv else "") + # Always resolve the live harness descendant: the pane's root + # is usually a shell, so its pid/etime/bin would mislead. + # The pane-command match is only a fallback (stale command). + hit = None + hpid: Optional[int] = None + for pid in _descendants(procs, pane["pid"]): + rec = procs.get(pid) + if not rec: + continue + first = rec["args"].split()[0] if rec["args"] else "" + hit = classify_harness(first) + if hit is not None: + hpid = pid + break + if hit is None: + if pane_hit is None: + continue + hit, hpid = pane_hit, pane["pid"] + agent_type, _bin = hit + claimed.add(hpid) + prec = procs.get(hpid, {}) + if prec.get("args"): + binn = os.path.basename(prec["args"].split()[0]) + elif argv: + binn = argv.split()[0] + else: + binn = NA + runs.append({ + "type": agent_type, + "bin": binn, + "sock": pane.get("sock", ""), + "session": pane.get("session", NA), + "pane": pane.get("pane", NA), + "pid": hpid, + "etime": prec.get("etime", NA), + "title": (pane.get("title") or "")[:48], + }) + + # Bare harness processes (no tmux pane above them). + for pid, rec in procs.items(): + if pid in claimed: + continue + first = rec["args"].split()[0] if rec["args"] else "" + hit = classify_harness(first) + if hit is None: + continue + # Skip when some pane root is an ancestor (already covered). + anc, under_pane = rec["ppid"], False + hops = 0 + while anc in procs and hops < 64: + if anc in pane_roots: + under_pane = True + break + anc = procs[anc]["ppid"] + hops += 1 + if under_pane: + continue + claimed.add(pid) + runs.append({ + "type": hit[0], + "bin": os.path.basename(first), + "sock": "", + "session": NA, + "pane": NA, + "pid": pid, + "etime": rec.get("etime", NA), + "title": "", + }) + + runs.sort(key=lambda r: (type_sort_key(r["type"]), + str(r["session"]), str(r["pane"]))) + return runs + + +# ===================================================================== +# Surfaces: devices + runs +# ===================================================================== + +PANE_FORMAT = ("PANE|%s|#{pane_id}|#{pane_pid}|#{session_name}|" + "#{pane_current_command}|#{pane_title}") + +REMOTE_SCAN_CMD = ( + "for d in \"${TMUX_TMPDIR:-/tmp}/tmux-$(id -u)\" " + "\"${TMPDIR:-/tmp}/tmux-$(id -u)\" /tmp/tmux-$(id -u); do " + "for s in \"$d\"/*; do [ -S \"$s\" ] || continue; " + "n=$(basename \"$s\"); " + "tmux -S \"$s\" list-panes -a -F \"PANE|$n|#{pane_id}|#{pane_pid}|" + "#{session_name}|#{pane_current_command}|#{pane_title}\" " + "2>/dev/null; done; done; " + "echo '%s'; ps -eo pid,ppid,etime,command 2>/dev/null" % PS_MARKER +) + + +def local_tmux_sockets() -> List[str]: + """Absolute tmux socket paths on this machine (may be empty).""" + uid = os.getuid() if hasattr(os, "getuid") else 0 + cands = [] + for base in (os.environ.get("TMUX_TMPDIR") or "/tmp", + os.environ.get("TMPDIR") or "/tmp", "/tmp"): + cands.append(os.path.join(base, "tmux-%d" % uid)) + found: List[str] = [] + seen_dirs = set() + for d in cands: + if d in seen_dirs: + continue + seen_dirs.add(d) + try: + names = sorted(os.listdir(d)) + except Exception: + continue + for n in names: + p = os.path.join(d, n) + try: + import stat + if stat.S_ISSOCK(os.stat(p).st_mode): + found.append(p) + except Exception: + continue + # Same server via two spellings: keep first per basename. + dedup: List[str] = [] + seen_base = set() + for p in found: + b = os.path.basename(p) + if b not in seen_base: + seen_base.add(b) + dedup.append(p) + return dedup + + +def local_device_name() -> str: + """Short hostname of this machine (never raises).""" + try: + import socket + return socket.gethostname().split(".")[0] + except Exception: + return "localhost" + + +def gather_devices(run: Optional[RunFn] = None, + local_name: Optional[str] = None) -> Dict[str, Any]: + """Tailnet devices from `tailscale status` + the local machine. + + Returns {"devices": [{name, ip, os, online, local, ssh, detail}], + "note": str}. Devices are stable-sorted: local first, then by name. + "ssh" marks whether v1 probes the device (online + ssh-capable OS). + """ + run = run or _run + local_name = local_name or local_device_name() + rc, out = run(["tailscale", "status"], timeout=10) + if rc != 0: + return {"devices": [{"name": local_name, "ip": NA, "os": NA, + "online": True, "local": True, "ssh": False, + "detail": "local only"}], + "note": "tailscale status failed (%s); local only." + % (out.strip().splitlines()[-1][:80] if out.strip() + else "rc=%d" % rc)} + devices = [] + for d in parse_tailscale_status(out): + local = (d["name"] == local_name) + ssh = bool(d["online"] and d["os"] in SSH_OS and not local) + devices.append({"name": d["name"], "ip": d["ip"], "os": d["os"], + "online": d["online"], "local": local, "ssh": ssh, + "detail": d["detail"]}) + if not any(d["local"] for d in devices): + devices.append({"name": local_name, "ip": NA, "os": NA, + "online": True, "local": True, "ssh": False, + "detail": "local"}) + devices.sort(key=lambda d: (not d["local"], d["name"])) + return {"devices": devices, "note": ""} + + +def gather_device_runs(device: Dict[str, Any], + run: Optional[RunFn] = None) -> Dict[str, Any]: + """One device scan -> {device, ok, runs, note}. Never raises.""" + run = run or _run + name = device.get("name", "?") + try: + if device.get("local"): + pane_chunks = [] + for sock in local_tmux_sockets(): + rc1, chunk = run( + ["tmux", "-S", sock, "list-panes", "-a", "-F", + PANE_FORMAT % os.path.basename(sock)], + timeout=DEVICE_TIMEOUT_S) + if rc1 == 0 and chunk: + pane_chunks.append(chunk) + _rc2, ps_out = run(["ps", "-eo", "pid,ppid,etime,command"], + timeout=DEVICE_TIMEOUT_S) + out = ("\n".join(pane_chunks) + "\n" + PS_MARKER + "\n" + + ps_out) + else: + rc, out = _run_ssh(name, REMOTE_SCAN_CMD, run=run) + ssh_err = "" if rc == 0 else out.strip().splitlines() + ssh_err = ssh_err[-1][:100] if ssh_err else "rc=%d" % rc + if rc != 0: + return {"device": name, "ok": False, "runs": [], + "note": "ssh failed: %s" % ssh_err} + pane_text, ps_text = split_scan(out) + runs = join_runs(parse_panes(pane_text), parse_ps(ps_text)) + for r in runs: + r["device"] = name + return {"device": name, "ok": True, "runs": runs, "note": ""} + except Exception as e: + return {"device": name, "ok": False, "runs": [], + "note": "scan failed: %s" % e} + + +def gather_all(run: Optional[RunFn] = None, + devices: Optional[List[Dict[str, Any]]] = None, + max_workers: int = 8) -> Dict[str, Any]: + """One-shot snapshot: devices + runs sorted by agent type. + + Devices scan in parallel (threads); each device is isolated — one + failure never blocks the rest. Returns {"devices": [...], + "runs": [...] (sorted by type/device), "by_type": {type: {total, + devices: {name: n}}}, "unreachable": [names], "note": str}. + """ + run = run or _run + dev_info = gather_devices(run=run) + if devices is None: + devices = [d for d in dev_info["devices"] + if d.get("local") or d.get("ssh")] + else: + devices = [d for d in devices + if d.get("local") or d.get("ssh")] + probed = {d["name"] for d in devices} + + results: List[Dict[str, Any]] = [] + if devices: + with concurrent.futures.ThreadPoolExecutor( + max_workers=min(max_workers, len(devices))) as pool: + futs = {pool.submit(gather_device_runs, d, run): d["name"] + for d in devices} + for fut in concurrent.futures.as_completed(futs): + try: + results.append(fut.result()) + except Exception as e: + results.append({"device": futs[fut], "ok": False, + "runs": [], "note": "scan error: %s" % e}) + + runs: List[Dict[str, Any]] = [] + unreachable: List[str] = [] + for res in results: + if not res.get("ok"): + unreachable.append(res["device"]) + continue + runs.extend(res.get("runs", [])) + runs.sort(key=lambda r: (type_sort_key(r["type"]), r.get("device", ""), + str(r.get("session", "")))) + + by_type: Dict[str, Dict[str, Any]] = {} + for r in runs: + bucket = by_type.setdefault(r["type"], {"total": 0, "devices": {}}) + bucket["total"] += 1 + dev = r.get("device", "?") + bucket["devices"][dev] = bucket["devices"].get(dev, 0) + 1 + + skipped = sorted(d["name"] for d in dev_info["devices"] + if d["name"] not in probed) + notes = [dev_info["note"]] if dev_info["note"] else [] + if skipped: + notes.append("skipped (offline/mobile/key): %s" % ", ".join(skipped)) + return {"devices": dev_info["devices"], "runs": runs, + "by_type": by_type, "unreachable": sorted(unreachable), + "note": "; ".join(notes)} + + +# ===================================================================== +# Curses UI (thin read-only renderer over gather_*) +# ===================================================================== + +AUTO_REFRESH_S = 30.0 + + +class AgentManagerTUI: + """Read-only agent-run console. q quits, r refreshes, ? helps.""" + + def __init__(self, stdscr: "curses.window"): + self.stdscr = stdscr + self.current_tab = 0 + self.tabs = [ + "1: RUNS", + "2: TYPES", + "3: DEVICES", + ] + 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 + self._init_colors() + + self.scroll = 0 + self.show_help = False + self.status_msg = "Scanning tailnet devices..." + self.last_refresh = 0.0 + self.snapshot: Dict[str, Any] = {} + self.refresh() + + # -- setup ------------------------------------------------------ + + def _init_colors(self) -> None: + try: + curses.start_color() + curses.use_default_colors() + curses.init_pair(1, curses.COLOR_CYAN, -1) + curses.init_pair(2, curses.COLOR_YELLOW, -1) + curses.init_pair(3, curses.COLOR_GREEN, -1) + curses.init_pair(4, curses.COLOR_RED, -1) + curses.init_pair(5, curses.COLOR_MAGENTA, -1) + curses.init_pair(6, curses.COLOR_BLACK, curses.COLOR_CYAN) + curses.init_pair(7, curses.COLOR_BLACK, curses.COLOR_WHITE) + curses.init_pair(8, curses.COLOR_BLACK, curses.COLOR_YELLOW) + 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": curses.color_pair(8) | curses.A_BOLD, + "dim": curses.A_DIM, + } + return mapping.get(name, 0) + except Exception: + return 0 + + # -- data ------------------------------------------------------- + + def refresh(self) -> None: + try: + self.snapshot = gather_all() + runs = len(self.snapshot.get("runs", [])) + devs = len([d for d in self.snapshot.get("devices", []) + if d.get("local") or d.get("ssh")]) + self.status_msg = ( + "Snapshot %s: %d runs on %d devices " + "(auto-refresh %ds; r=refresh)" % ( + datetime.now().strftime("%H:%M:%S"), runs, devs, + int(AUTO_REFRESH_S))) + except Exception as e: + self.snapshot = {} + self.status_msg = "Refresh failed (showing n/a): %s" % e + self.last_refresh = time.time() + self.scroll = 0 + + # -- 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: + try: + self.stdscr.addstr(y, x, text[:max(0, w - x - 1)], attr) + except Exception: + pass + + def _render_header(self, w: int) -> None: + self.safe_addstr(0, 0, " " * w, self._attr("head_sel")) + title = " AGENT MANAGER (read-only) [agent-manager.py] " + self.safe_addstr(0, 1, title, self._attr("head_sel")) + self.safe_addstr(1, 0, " " * w, self._attr("dim")) + col = 1 + for idx, tab_name in enumerate(self.tabs): + pill = " [%s] " % tab_name + attr = self._attr("head_sel") if idx == self.current_tab \ + else self._attr("dim") + self.safe_addstr(1, col, pill, attr) + col += len(pill) + 1 + self.safe_addstr(2, 0, "-" * w, self._attr("dim")) + + def _render_footer(self, h: int, w: int) -> None: + self.safe_addstr(h - 2, 0, "-" * w, self._attr("dim")) + hints = " 1-3/Tab: Tabs j/k: Scroll r: Refresh ?: Help q: Quit" + self.safe_addstr(h - 1, 1, self.status_msg[: w - 2], + self._attr("dim")) + if len(self.status_msg) + len(hints) + 2 < w: + self.safe_addstr(h - 1, w - len(hints) - 1, hints, + self._attr("dim")) + + def _body(self, h: int, w: int, title: str, + lines: List[Tuple[str, str]]) -> None: + self.safe_addstr(3, 2, title, self._attr("cyan")) + self.safe_addstr(4, 2, "-" * (w - 4), self._attr("dim")) + max_rows = h - 8 + visible = lines[self.scroll:self.scroll + max_rows] + for i, (text, attr_name) in enumerate(visible): + self.safe_addstr(5 + i, 2, text, self._attr(attr_name)) + if self.scroll > 0: + self.safe_addstr(5, w - 6, "^more", self._attr("dim")) + if self.scroll + max_rows < len(lines): + self.safe_addstr(h - 3, w - 6, "vmore", self._attr("dim")) + + def _note_lines(self, w: int) -> List[Tuple[str, str]]: + note = self.snapshot.get("note", "") + unreach = self.snapshot.get("unreachable", []) + lines: List[Tuple[str, str]] = [] + if unreach: + lines.append(("", "normal")) + lines.append(("unreachable: %s" % ", ".join(unreach), + "red")) + if note: + lines.append(("", "normal")) + lines.append(("note: %s" % note[: w - 10], "yellow")) + return lines + + # -- per-tab renderers ------------------------------------------ + + def _render_runs(self, h: int, w: int) -> None: + runs = self.snapshot.get("runs", []) + lines: List[Tuple[str, str]] = [ + ("%-14s %-12s %-16s %-14s %-6s %-11s %s" + % ("TYPE", "DEVICE", "BIN", "SESSION", "PANE", "ELAPSED", + "TITLE"), "dim"), + ] + last_type = None + for r in runs: + typ = r.get("type", "?") + if typ != last_type: + lines.append(("", "normal")) + last_type = typ + attr = "green" if not typ.startswith("other:") else "yellow" + lines.append(( + "%-14s %-12s %-16s %-14s %-6s %-11s %s" % ( + typ[:14], r.get("device", "?")[:12], + r.get("bin", NA)[:16], display_session(r)[:14], + str(r.get("pane", NA))[:6], + str(r.get("etime", NA))[:11], + r.get("title", "")[: w - 80]), attr)) + if not runs: + lines.append(("(No agent runs found on probed devices.)", + "dim")) + lines.extend(self._note_lines(w)) + self._body(h, w, "AGENT RUNS SORTED BY TYPE (%d)" % len(runs), + lines) + + def _render_types(self, h: int, w: int) -> None: + by_type = self.snapshot.get("by_type", {}) + lines: List[Tuple[str, str]] = [] + total = sum(b.get("total", 0) for b in by_type.values()) + lines.append(("Agent types: %d | total runs: %d" + % (len(by_type), total), "cyan")) + lines.append(("", "normal")) + for typ in sorted(by_type, key=type_sort_key): + bucket = by_type[typ] + attr = "green" if not typ.startswith("other:") else "yellow" + lines.append(("%-16s %d" % (typ, bucket.get("total", 0)), + attr)) + for dev, n in sorted(bucket.get("devices", {}).items()): + lines.append((" %-14s %d" % (dev, n), "normal")) + lines.append(("", "normal")) + if not by_type: + lines.append(("(No agent types observed.)", "dim")) + lines.extend(self._note_lines(w)) + self._body(h, w, "COUNTS BY AGENT TYPE", lines) + + def _render_devices(self, h: int, w: int) -> None: + devices = self.snapshot.get("devices", []) + runs = self.snapshot.get("runs", []) + counts: Dict[str, int] = {} + for r in runs: + dev = r.get("device", "?") + counts[dev] = counts.get(dev, 0) + 1 + unreach = set(self.snapshot.get("unreachable", [])) + lines: List[Tuple[str, str]] = [ + ("%-24s %-15s %-7s %-7s %-5s %s" + % ("DEVICE", "IP", "OS", "PROBED", "RUNS", "DETAIL"), "dim"), + ] + for d in devices: + name = d.get("name", "?") + probed = bool(d.get("local") or d.get("ssh")) + if name in unreach: + attr = "red" + elif not probed: + attr = "dim" + elif counts.get(name): + attr = "green" + else: + attr = "normal" + lines.append(( + "%-24s %-15s %-7s %-7s %-5s %s" % ( + name[:24] + (" *" if d.get("local") else ""), + d.get("ip", NA)[:15], str(d.get("os", NA))[:7], + "yes" if probed else "no", + counts.get(name, 0) if probed else NA, + str(d.get("detail", ""))[: w - 66]), attr)) + lines.append(("", "normal")) + lines.append(("* = local (no SSH); unreachable shows n/a, never " + "blocks the rest.", "dim")) + lines.extend(self._note_lines(w)) + self._body(h, w, "TAILNET DEVICES", lines) + + def _render_help(self, h: int, w: int) -> None: + modal_w = min(64, w - 6) + modal_h = 13 + top = (h - modal_h) // 2 + left = (w - modal_w) // 2 + for y in range(top, top + modal_h): + self.safe_addstr(y, left, " " * modal_w, self._attr("row_sel")) + self.safe_addstr(top, left, "+" + "-" * (modal_w - 2) + "+", + self._attr("cyan")) + for y in range(top + 1, top + modal_h - 1): + self.safe_addstr(y, left, "|", self._attr("cyan")) + self.safe_addstr(y, left + modal_w - 1, "|", + self._attr("cyan")) + self.safe_addstr(top + modal_h - 1, left, + "+" + "-" * (modal_w - 2) + "+", + self._attr("cyan")) + self.safe_addstr(top + 1, left + 3, "AGENT MANAGER HELP (read-only)", + self._attr("cyan")) + for i, hint in enumerate([ + "1-3 / Tab: switch surfaces", + "j/k / Up/Down: scroll", + "r: refresh snapshot now", + "q / Esc: quit (Esc closes help first)", + "", + "One SSH call per device per refresh.", + "Missing data shows as 'n/a' — never a crash.", + ]): + self.safe_addstr(top + 3 + i, left + 4, hint, + self._attr("normal")) + + # -- input + main loop ------------------------------------------ + + def _handle_key(self, ch: int) -> bool: + if ch in (3, 4): # Ctrl+C / Ctrl+D + return False + if self.show_help: + if ch in (27, ord("q"), ord("Q"), ord("?")): + self.show_help = False + return True + if ch in (ord("q"), ord("Q")): + return False + if ch in (27, ord("?")): + self.show_help = True + return True + if ch in (ord("1"), ord("2"), ord("3")): + self.current_tab = ch - ord("1") + self.scroll = 0 + return True + if ch == ord("\t"): + self.current_tab = (self.current_tab + 1) % len(self.tabs) + self.scroll = 0 + return True + if ch in (ord("j"), curses.KEY_DOWN): + self.scroll += 1 + return True + if ch in (ord("k"), curses.KEY_UP): + self.scroll = max(0, self.scroll - 1) + return True + if ch in (ord("r"), ord("R")): + self.refresh() + return True + return True + + def run(self) -> None: + renderers = [self._render_runs, self._render_types, + self._render_devices] + while True: + h, w = self.stdscr.getmaxyx() + self.stdscr.erase() + self._render_header(w) + try: + renderers[self.current_tab](h, w) + except Exception as e: + self.safe_addstr(5, 4, "Render error (n/a): %s" % e, + self._attr("red")) + self._render_footer(h, w) + if self.show_help: + self._render_help(h, w) + self.stdscr.refresh() + try: + ch = self.stdscr.getch() + if ch != -1 and not self._handle_key(ch): + break + except KeyboardInterrupt: + break + if time.time() - self.last_refresh > AUTO_REFRESH_S: + self.refresh() + time.sleep(0.05) + + +def main(argv: Optional[List[str]] = None) -> int: + argv = list(sys.argv[1:] if argv is None else argv) + if "--once" in argv: + snap = gather_all() + if "--json" in argv: + print(json.dumps(snap, indent=2, default=str)) + else: + print("== RUNS (%d) ==" % len(snap.get("runs", []))) + for r in snap.get("runs", []): + print("%-14s %-12s %-16s %s/%s %s" % ( + r.get("type"), r.get("device"), r.get("bin"), + display_session(r), r.get("pane"), r.get("etime"))) + print("== BY TYPE == ") + for typ, b in sorted(snap.get("by_type", {}).items()): + print("%s: %d %s" % (typ, b["total"], b["devices"])) + print("unreachable: %s" % snap.get("unreachable")) + if snap.get("note"): + print("note: %s" % snap["note"]) + return 0 + curses.wrapper(lambda stdscr: AgentManagerTUI(stdscr).run()) + return 0 + + +if __name__ == "__main__": + sys.exit(main())