#!/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:<basename>" 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:<basename>" 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())