herdr-session-switch/herdr_hub/cli.py

from __future__ import annotations

import argparse
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager
import curses
from dataclasses import dataclass, field, replace
import fcntl
from functools import wraps
import hashlib
import json
import os
from pathlib import Path
import shlex
import shutil
import stat
import subprocess
import sys
import tempfile
import threading
import time
from typing import Callable, Iterable, Mapping, Sequence
from urllib.parse import quote

from .aliases import AliasRegistry, AliasRegistryError, default_alias_registry_path
from .control import CONTROL_KEY, CONTROL_SHORTCUT, ControlMenu, HerdrControl
from .domain import validate_session_name
from .key_help import KeyHelpPopup
from .remotes import (
    RegistryError,
    RemoteEntry,
    RemoteRegistry,
    default_registry_path,
    delete_remote_session,
    ssh_destination_arguments,
    stop_remote_session,
    validate_target,
)
from .selector import (
    CatalogWindow,
    DeleteSessionOutcome,
    SelectorError,
    SelectorResult,
    SessionSelector,
    StopSessionOutcome,
)


DEFAULT_SOCKET_NAME = "herdr-hub"
DEFAULT_TMUX_SESSION = "hub"
# The Hub server is configured entirely by configure(); reading the user's
# ~/.tmux.conf would let theme status-formats and root-table bindings leak in.
DEFAULT_TMUX_CONFIG = "/dev/null"
DEFAULT_SESSION_NAME = "default"
BOOTSTRAP_WINDOW_NAME = "__herdr_hub_bootstrap__"
WINDOW_KEY_OPTION = "@herdr_hub_key"
WINDOW_COMMAND_OPTION = "@herdr_hub_command"
WINDOW_ENDPOINT_OPTION = "@herdr_hub_endpoint"
WINDOW_SHORT_LABEL_OPTION = "@herdr_hub_short_label"
SESSION_DISMISSED_OPTION = "@herdr_hub_dismissed"


class HubError(RuntimeError):
    """A user-facing failure."""


@dataclass(frozen=True)
class Endpoint:
    kind: str
    session: str = DEFAULT_SESSION_NAME
    target: str | None = None
    remote_keybindings: str = "local"
    remote_transport: str = "native"

    def __post_init__(self) -> None:
        if self.kind not in {"local", "remote"}:
            raise ValueError(f"unsupported endpoint kind: {self.kind}")
        validate_session_name(self.session)
        if self.kind == "local":
            if self.target is not None:
                raise ValueError("a local endpoint cannot have a remote target")
            if self.remote_transport != "native":
                raise ValueError("a local endpoint cannot use a remote transport")
            return
        validate_remote_target(self.target)
        if self.remote_keybindings not in {"local", "server"}:
            raise ValueError("remote keybindings must be 'local' or 'server'")
        if self.remote_transport not in {"native", "ssh"}:
            raise ValueError("remote transport must be 'native' or 'ssh'")

    @classmethod
    def local(cls, session: str = DEFAULT_SESSION_NAME) -> Endpoint:
        return cls(kind="local", session=session)

    @classmethod
    def remote(
        cls,
        target: str,
        session: str = DEFAULT_SESSION_NAME,
        remote_keybindings: str = "local",
        remote_transport: str = "native",
    ) -> Endpoint:
        return cls(
            kind="remote",
            target=target,
            session=session,
            remote_keybindings=remote_keybindings,
            remote_transport=remote_transport,
        )

    @property
    def key(self) -> str:
        if self.kind == "local":
            return f"local:{quote(self.session, safe='')}"
        assert self.target is not None
        return f"remote:{quote(self.target, safe='')}:{quote(self.session, safe='')}"

    @property
    def label(self) -> str:
        if self.kind == "local":
            return truncate_label(f"L {self.session}")
        assert self.target is not None
        return truncate_label(f"R {display_target(self.target)}/{self.session}")

    @property
    def short_label(self) -> str:
        if self.kind == "local":
            label = "local"
            if self.session != DEFAULT_SESSION_NAME:
                label += f"/{self.session}"
            return truncate_label(label, 32)
        assert self.target is not None
        label = display_target(self.target)
        if self.session != DEFAULT_SESSION_NAME:
            label += f"/{self.session}"
        return truncate_label(label, 32)

    def command(self, herdr_bin: str, ssh_bin: str = "ssh") -> list[str]:
        command = [herdr_bin]
        if self.kind == "remote":
            assert self.target is not None
            if self.remote_transport == "ssh":
                remote_command = ["herdr"]
                if self.session != DEFAULT_SESSION_NAME:
                    remote_command.extend(["--session", self.session])
                return [
                    ssh_bin,
                    "-tt",
                    "-o",
                    "ServerAliveInterval=15",
                    "-o",
                    "ServerAliveCountMax=3",
                    *ssh_destination_arguments(self.target),
                    shlex.join(remote_command),
                ]
            command.extend(["--remote", self.target])
            if self.remote_keybindings != "local":
                command.extend(["--remote-keybindings", self.remote_keybindings])
        if self.session != DEFAULT_SESSION_NAME:
            command.extend(["--session", self.session])
        return command


@dataclass
class DiscoveryReport:
    endpoints: list[Endpoint] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)

    def extend(self, other: DiscoveryReport) -> None:
        self.endpoints.extend(other.endpoints)
        self.warnings.extend(other.warnings)


@dataclass(frozen=True)
class ManagedWindow:
    window_id: str
    key: str
    name: str
    dead: bool
    active: bool
    endpoint: Endpoint | None = None
    short_label: str = ""


RunFunction = Callable[..., subprocess.CompletedProcess[str]]


def serialized_hub_mutation(method):  # type: ignore[no-untyped-def]
    @wraps(method)
    def wrapper(self, *args, **kwargs):  # type: ignore[no-untyped-def]
        with self.mutation_lock():
            return method(self, *args, **kwargs)

    return wrapper


def validate_remote_target(target: str | None) -> None:
    validate_target(target or "")


def truncate_label(label: str, limit: int = 48) -> str:
    clean = "".join(
        character if ord(character) >= 32 and ord(character) != 127 else "?"
        for character in label
    )
    if len(clean) <= limit:
        return clean
    return clean[: limit - 1] + "…"


def display_target(target: str) -> str:
    if target.startswith("ssh://"):
        return target.removeprefix("ssh://")
    return target


def endpoint_sort_key(endpoint: Endpoint) -> tuple[int, str, str]:
    if endpoint.kind == "local":
        session_rank = "" if endpoint.session == DEFAULT_SESSION_NAME else endpoint.session
        return (0, "", session_rank)
    return (1, endpoint.target or "", endpoint.session)


def dedupe_endpoints(endpoints: Iterable[Endpoint]) -> list[Endpoint]:
    by_key: dict[str, Endpoint] = {}
    for endpoint in endpoints:
        by_key.setdefault(endpoint.key, endpoint)
    return sorted(by_key.values(), key=endpoint_sort_key)


def discover_local_sessions(
    herdr_bin: str,
    *,
    run: RunFunction = subprocess.run,
) -> DiscoveryReport:
    try:
        result = run(
            [herdr_bin, "session", "list", "--json"],
            capture_output=True,
            text=True,
            timeout=5,
            check=False,
        )
    except (OSError, subprocess.SubprocessError) as error:
        return DiscoveryReport(warnings=[f"local Herdr session discovery failed: {error}"])

    if result.returncode != 0:
        detail = (result.stderr or result.stdout or "unknown error").strip()
        return DiscoveryReport(
            warnings=[f"local Herdr session discovery failed: {detail}"]
        )

    try:
        payload = json.loads(result.stdout)
        sessions = payload["sessions"]
    except (json.JSONDecodeError, KeyError, TypeError) as error:
        return DiscoveryReport(
            warnings=[f"local Herdr returned invalid session JSON: {error}"]
        )

    endpoints: list[Endpoint] = []
    for item in sessions:
        if not isinstance(item, dict) or item.get("running") is not True:
            continue
        name = item.get("name")
        if not isinstance(name, str):
            continue
        try:
            endpoints.append(Endpoint.local(name))
        except ValueError:
            continue
    return DiscoveryReport(endpoints=dedupe_endpoints(endpoints))


def parse_endpoint_argv(argv: Sequence[str]) -> Endpoint | None:
    if not argv or Path(argv[0]).name != "herdr":
        return None

    target: str | None = None
    session = DEFAULT_SESSION_NAME
    keybindings = "local"
    index = 1
    while index < len(argv):
        argument = argv[index]
        if argument == "--":
            break
        if argument == "--remote":
            if index + 1 >= len(argv):
                return None
            target = argv[index + 1]
            index += 2
            continue
        if argument.startswith("--remote="):
            target = argument.split("=", 1)[1]
            index += 1
            continue
        if argument == "--session":
            if index + 1 >= len(argv):
                return None
            session = argv[index + 1]
            index += 2
            continue
        if argument.startswith("--session="):
            session = argument.split("=", 1)[1]
            index += 1
            continue
        if argument == "--remote-keybindings":
            if index + 1 >= len(argv):
                return None
            keybindings = argv[index + 1]
            index += 2
            continue
        if argument.startswith("--remote-keybindings="):
            keybindings = argument.split("=", 1)[1]
            index += 1
            continue
        index += 1

    try:
        return (
            Endpoint.remote(target, session, keybindings)
            if target is not None
            else Endpoint.local(session)
        )
    except ValueError:
        return None


def parse_remote_argv(argv: Sequence[str]) -> Endpoint | None:
    endpoint = parse_endpoint_argv(argv)
    return endpoint if endpoint is not None and endpoint.kind == "remote" else None


def serialize_endpoint(endpoint: Endpoint) -> str:
    return json.dumps(
        {
            "version": 1,
            "kind": endpoint.kind,
            "session": endpoint.session,
            "target": endpoint.target,
            "remote_keybindings": endpoint.remote_keybindings,
            "remote_transport": endpoint.remote_transport,
        },
        ensure_ascii=False,
        separators=(",", ":"),
    )


def deserialize_endpoint(metadata: str, command: str = "") -> Endpoint | None:
    if metadata:
        try:
            payload = json.loads(metadata)
            if not isinstance(payload, dict) or payload.get("version") != 1:
                raise ValueError
            kind = payload.get("kind")
            session = payload.get("session")
            target = payload.get("target")
            keybindings = payload.get("remote_keybindings", "local")
            transport = payload.get("remote_transport", "native")
            if not isinstance(kind, str) or not isinstance(session, str):
                raise ValueError
            if not isinstance(keybindings, str):
                raise ValueError
            if not isinstance(transport, str):
                raise ValueError
            if kind == "local":
                return Endpoint.local(session)
            if kind == "remote" and isinstance(target, str):
                return Endpoint.remote(target, session, keybindings, transport)
            raise ValueError
        except (json.JSONDecodeError, TypeError, ValueError):
            pass
    if not command:
        return None
    try:
        arguments = shlex.split(command)
    except ValueError:
        return None
    if arguments and arguments[0] == "exec":
        arguments = arguments[1:]
    return parse_endpoint_argv(arguments)


def decode_cmdline(raw: bytes) -> list[str]:
    return [
        os.fsdecode(argument)
        for argument in raw.rstrip(b"\0").split(b"\0")
        if argument
    ]


def discover_remote_processes(
    proc_root: Path = Path("/proc"),
    *,
    user_id: int | None = None,
) -> DiscoveryReport:
    if user_id is None:
        user_id = os.getuid()
    if not proc_root.is_dir():
        return DiscoveryReport(
            warnings=[f"remote discovery unavailable: {proc_root} is not mounted"]
        )

    endpoints: list[Endpoint] = []
    unreadable = 0
    try:
        entries = sorted(
            (entry for entry in proc_root.iterdir() if entry.name.isdigit()),
            key=lambda entry: int(entry.name),
        )
    except OSError as error:
        return DiscoveryReport(warnings=[f"remote process discovery failed: {error}"])

    for entry in entries:
        cmdline_path = entry / "cmdline"
        try:
            if cmdline_path.stat().st_uid != user_id:
                continue
            raw = cmdline_path.read_bytes()
        except (FileNotFoundError, ProcessLookupError):
            continue
        except OSError:
            unreadable += 1
            continue
        endpoint = parse_remote_argv(decode_cmdline(raw))
        if endpoint is not None:
            endpoints.append(endpoint)

    warnings: list[str] = []
    if unreadable:
        warnings.append(f"could not inspect {unreadable} same-user process command line(s)")
    return DiscoveryReport(endpoints=dedupe_endpoints(endpoints), warnings=warnings)


def discover_endpoints(herdr_bin: str, proc_root: Path = Path("/proc")) -> DiscoveryReport:
    report = discover_local_sessions(herdr_bin)
    report.extend(discover_remote_processes(proc_root))
    report.endpoints = dedupe_endpoints(report.endpoints)
    return report


class TmuxHub:
    def __init__(
        self,
        *,
        tmux_bin: str,
        ssh_bin: str = "ssh",
        socket_name: str = DEFAULT_SOCKET_NAME,
        session_name: str = DEFAULT_TMUX_SESSION,
        config_file: str | None = None,
        script_path: str | None = None,
        python_bin: str | None = None,
        aliases: Mapping[str, str] | None = None,
        run: RunFunction = subprocess.run,
    ) -> None:
        self.tmux_bin = tmux_bin
        self.ssh_bin = ssh_bin
        self.socket_name = socket_name
        self.session_name = session_name
        self.config_file = config_file
        self.script_path = script_path or str(Path(sys.argv[0]).absolute())
        self.python_bin = python_bin or sys.executable
        self.aliases = dict(aliases or {})
        self._run = run
        self._local_mutation_lock = threading.RLock()
        self._mutation_depth = 0

    @property
    def mutation_lock_path(self) -> Path:
        identity = f"{os.getuid()}\0{self.socket_name}\0{self.session_name}"
        digest = hashlib.sha256(identity.encode("utf-8")).hexdigest()[:20]
        roots = []
        if configured_runtime := os.environ.get("XDG_RUNTIME_DIR"):
            roots.append(Path(configured_runtime).expanduser())
        temporary_root = Path(tempfile.gettempdir())
        if temporary_root not in roots:
            roots.append(temporary_root)

        failures: list[str] = []
        for runtime_root in roots:
            lock_directory = runtime_root / f"herdr-hub-{os.getuid()}"
            try:
                lock_directory.mkdir(mode=0o700, parents=False, exist_ok=True)
                metadata = lock_directory.lstat()
                if (
                    not stat.S_ISDIR(metadata.st_mode)
                    or metadata.st_uid != os.getuid()
                ):
                    failures.append(f"unsafe directory {lock_directory}")
                    continue
                os.chmod(lock_directory, 0o700)
                return lock_directory / f"{digest}.lock"
            except OSError as error:
                failures.append(f"{lock_directory}: {error}")
        raise HubError("could not prepare Hub lock directory: " + "; ".join(failures))

    @contextmanager
    def mutation_lock(self):  # type: ignore[no-untyped-def]
        """Serialize state-changing Hub commands across CLI processes."""
        with self._local_mutation_lock:
            if self._mutation_depth:
                self._mutation_depth += 1
                try:
                    yield
                finally:
                    self._mutation_depth -= 1
                return

            descriptor = os.open(
                self.mutation_lock_path,
                os.O_CREAT | os.O_RDWR,
                0o600,
            )
            try:
                os.fchmod(descriptor, 0o600)
                fcntl.flock(descriptor, fcntl.LOCK_EX)
                self._mutation_depth = 1
                try:
                    yield
                finally:
                    self._mutation_depth = 0
                    fcntl.flock(descriptor, fcntl.LOCK_UN)
            finally:
                os.close(descriptor)

    def base_command(self) -> list[str]:
        command = [self.tmux_bin, "-L", self.socket_name]
        if self.config_file is not None:
            command.extend(["-f", self.config_file])
        return command

    def run(
        self,
        arguments: Sequence[str],
        *,
        check: bool = True,
        capture_output: bool = True,
        text: bool = True,
    ) -> subprocess.CompletedProcess[str]:
        try:
            result = self._run(
                [*self.base_command(), *arguments],
                capture_output=capture_output,
                text=text,
                check=False,
            )
        except OSError as error:
            raise HubError(f"could not run tmux: {error}") from error
        if check and result.returncode != 0:
            detail = (result.stderr or result.stdout or "tmux command failed").strip()
            raise HubError(detail)
        return result

    def run_batch(self, commands: Sequence[Sequence[str]]) -> None:
        """Run several tmux commands in one client invocation.

        Every tmux client spawn is a fork/exec plus a socket round trip, which
        adds up on phones. tmux splits its command line on ';' arguments and
        on any argument that merely ends with ';', so a literal trailing
        semicolon is escaped. A failing command aborts the rest of the batch;
        the commands are then replayed one by one so the error is attributed
        to the exact command, as before batching.
        """
        pending = [list(command) for command in commands if command]
        if not pending:
            return
        if len(pending) == 1:
            self.run(pending[0])
            return
        arguments: list[str] = []
        for index, command in enumerate(pending):
            if index:
                arguments.append(";")
            arguments.extend(
                argument[:-1] + "\\;" if argument.endswith(";") else argument
                for argument in command
            )
        if self.run(arguments, check=False).returncode != 0:
            for command in pending:
                self.run(command)

    def exists(self) -> bool:
        result = self.run(
            ["has-session", "-t", self.session_name],
            check=False,
        )
        return result.returncode == 0

    @serialized_hub_mutation
    def ensure_server(self) -> bool:
        if self.exists():
            self.configure()
            return False
        bootstrap = shlex.join(["sleep", "infinity"])
        self.run(
            [
                "new-session",
                "-d",
                "-s",
                self.session_name,
                "-n",
                BOOTSTRAP_WINDOW_NAME,
                bootstrap,
            ]
        )
        self.configure()
        return True

    def configure(self) -> None:
        option_commands = [
            ["set-option", "-s", "set-clipboard", "on"],
            ["set-option", "-g", "prefix", "C-a"],
            ["set-option", "-g", "status", "on"],
            ["set-option", "-g", "status-position", "top"],
            ["set-option", "-g", "status-interval", "2"],
            ["set-option", "-g", "mouse", "on"],
            ["set-option", "-g", "focus-events", "on"],
            ["set-option", "-g", "status-left-length", "18"],
            ["set-option", "-g", "status-right-length", "14"],
            [
                "set-option",
                "-g",
                "status-left",
                "#{?#{<:#{client_width},50},"
                "#[bold#,fg=colour81] S #[default],"
                "#[bold#,fg=colour81] Sessions #[default]}",
            ],
            [
                "set-option",
                "-g",
                "status-right",
                "#{?#{<:#{client_width},50},"
                "#[dim]C-a ?#[default],"
                "#[dim]C-a ? 帮助#[default]}",
            ],
            ["set-option", "-g", "status-style", "bg=colour235,fg=colour250"],
            ["set-window-option", "-g", "window-status-separator", ""],
            [
                "set-window-option",
                "-g",
                "window-status-format",
                "#{?#{<:#{client_width},80},"
                " #{?pane_dead,×,○} #{=12:@herdr_hub_short_label} ,"
                " #{?pane_dead,× #W [断开],○ #W} }",
            ],
            [
                "set-window-option",
                "-g",
                "window-status-current-format",
                "#[bold,bg=colour31,fg=colour231]"
                "#{?#{<:#{client_width},80},"
                " #{?pane_dead,×,●} #{=12:@herdr_hub_short_label} ,"
                " #{?pane_dead,× #W [断开],● #W} }"
                "#[default]",
            ],
            ["set-window-option", "-g", "automatic-rename", "off"],
            ["set-window-option", "-g", "allow-rename", "off"],
            ["set-window-option", "-g", "remain-on-exit", "on"],
            [
                "set-window-option",
                "-g",
                "remain-on-exit-format",
                "#[bold,fg=colour196]连接已断开#[default]"
                "（退出状态 #{pane_dead_status}） · 按 Ctrl+A R 重连",
            ],
        ]

        refresh = shlex.join([self.python_bin, self.script_path, "refresh"])
        choose = shlex.join([self.python_bin, self.script_path, "choose"])
        reconnect = shlex.join([self.python_bin, self.script_path, "reconnect"])
        close = shlex.join(
            [self.python_bin, self.script_path, "close", "--window", "#{window_id}"]
        )
        control = shlex.join([self.python_bin, self.script_path, "control"])
        key_help = shlex.join([self.python_bin, self.script_path, "keys"])
        confirm_close = shlex.join(["run-shell", "-b", close])
        cleanup_commands = [
            # Remove tmux's default/legacy '&' binding when reconfiguring both
            # new and already-running Hub servers.
            ["unbind-key", "-q", "-T", "prefix", "&"],
            ["unbind-key", "-q", "-T", "prefix", "Q"],
            # Remove the short-lived global binding used by the first
            # control-menu prototype when reconfiguring an already-running Hub.
            ["unbind-key", "-q", "-T", "root", "C-g"],
        ]
        binding_commands = [
            ["bind-key", "-T", "prefix", "?", "display-popup", "-EE",
             "-w", "72%", "-h", "80%", "-T", "快捷键", key_help],
            ["bind-key", "-T", "prefix", CONTROL_KEY, "display-popup", "-EE",
             "-w", "80%", "-h", "50%", "-T", "Herdr 命令", control],
            ["bind-key", "-T", "prefix", "C-a", "send-prefix"],
            ["bind-key", "-T", "prefix", "r", "run-shell", "-b", refresh],
            [
                "bind-key",
                "-T",
                "prefix",
                "g",
                "display-popup",
                "-EE",
                "-w",
                "90%",
                "-h",
                "90%",
                "-T",
                "Sessions",
                choose,
            ],
            [
                "bind-key",
                "-T",
                "root",
                "MouseDown1StatusLeft",
                "display-popup",
                "-EE",
                "-w",
                "90%",
                "-h",
                "90%",
                "-T",
                "Sessions",
                choose,
            ],
            ["bind-key", "-T", "prefix", "R", "run-shell", "-b", reconnect],
            [
                "bind-key",
                "-T",
                "prefix",
                "q",
                "confirm-before",
                "-p",
                "关闭当前 Session Tab？ (y/n)",
                confirm_close,
            ],
        ]
        self.run_batch([*option_commands, *cleanup_commands, *binding_commands])
        self.refresh_labels()

    def managed_windows(self) -> list[ManagedWindow]:
        line_format = (
            f"#{{window_id}}\t#{{{WINDOW_KEY_OPTION}}}\t#{{window_name}}\t"
            f"#{{pane_dead}}\t#{{window_active}}\t#{{{WINDOW_ENDPOINT_OPTION}}}\t"
            f"#{{{WINDOW_COMMAND_OPTION}}}\t#{{{WINDOW_SHORT_LABEL_OPTION}}}"
        )
        result = self.run(
            ["list-windows", "-t", self.session_name, "-F", line_format],
            check=False,
        )
        if result.returncode != 0:
            # The common reason is that the Hub session does not exist yet;
            # only a failure against a live session is an error.
            if not self.exists():
                return []
            raise HubError((result.stderr or result.stdout or "tmux command failed").strip())
        windows: list[ManagedWindow] = []
        for line in result.stdout.splitlines():
            fields = line.split("\t", 7)
            if len(fields) != 8 or not fields[1]:
                continue
            windows.append(
                ManagedWindow(
                    window_id=fields[0],
                    key=fields[1],
                    name=fields[2],
                    dead=fields[3] == "1",
                    active=fields[4] == "1",
                    endpoint=deserialize_endpoint(fields[5], fields[6]),
                    short_label=fields[7],
                )
            )
        return windows

    def window_for_key(self, key: str) -> ManagedWindow | None:
        return next((window for window in self.managed_windows() if window.key == key), None)

    def refresh_labels(self) -> None:
        self.run_batch(
            [
                command
                for window in self.managed_windows()
                for command in self._label_commands(window)
            ]
        )

    def _labels_for(self, window: ManagedWindow) -> tuple[str, str]:
        alias = self.aliases.get(window.key)
        if alias:
            return truncate_label(alias), truncate_label(alias, 32)
        if window.endpoint is not None:
            return window.endpoint.label, window.endpoint.short_label
        return window.name, truncate_label(window.name, 32)

    def _label_commands(self, window: ManagedWindow) -> list[list[str]]:
        display_label, short_label = self._labels_for(window)
        commands: list[list[str]] = []
        if window.name != display_label:
            # A display name may start with '-', which tmux would otherwise
            # parse as a flag. set-option takes the value positionally and
            # does not accept '--', so only rename-window needs the guard.
            commands.append(["rename-window", "-t", window.window_id, "--", display_label])
        if window.short_label != short_label:
            commands.append(
                ["set-option", "-w", "-t", window.window_id, WINDOW_SHORT_LABEL_OPTION, short_label]
            )
        return commands

    def _apply_window_label(self, window: ManagedWindow) -> None:
        self.run_batch(self._label_commands(window))

    @serialized_hub_mutation
    def apply_alias(self, key: str, label: str) -> ManagedWindow | None:
        previous = self.aliases.get(key)
        if label:
            self.aliases[key] = label
        else:
            self.aliases.pop(key, None)
        window = self.window_for_key(key)
        if window is not None:
            try:
                self._apply_window_label(window)
            except HubError:
                # Keep the in-memory table consistent with what tmux shows.
                if previous is None:
                    self.aliases.pop(key, None)
                else:
                    self.aliases[key] = previous
                raise
        return self.window_for_key(key)

    def dismissed_keys(self) -> set[str]:
        result = self.run(
            [
                "show-options",
                "-v",
                "-t",
                self.session_name,
                SESSION_DISMISSED_OPTION,
            ],
            check=False,
        )
        if result.returncode != 0 or not result.stdout.strip():
            return set()
        try:
            payload = json.loads(result.stdout)
        except json.JSONDecodeError:
            return set()
        if not isinstance(payload, list):
            return set()
        return {key for key in payload if isinstance(key, str) and key}

    def set_dismissed_keys(self, keys: Iterable[str]) -> None:
        value = json.dumps(sorted(set(keys)), ensure_ascii=False, separators=(",", ":"))
        self.run(
            [
                "set-option",
                "-t",
                self.session_name,
                SESSION_DISMISSED_OPTION,
                value,
            ]
        )

    def restore_endpoint(self, key: str) -> None:
        dismissed = self.dismissed_keys()
        if key not in dismissed:
            return
        dismissed.remove(key)
        self.set_dismissed_keys(dismissed)

    @serialized_hub_mutation
    def create_window(self, endpoint: Endpoint, herdr_bin: str) -> ManagedWindow:
        existing = self.window_for_key(endpoint.key)
        if existing is not None:
            return existing
        dismissed = self.dismissed_keys()
        was_dismissed = endpoint.key in dismissed
        if was_dismissed:
            self.set_dismissed_keys(dismissed - {endpoint.key})
        command = "exec " + shlex.join(endpoint.command(herdr_bin, self.ssh_bin))
        alias = self.aliases.get(endpoint.key)
        display_label = truncate_label(alias) if alias else endpoint.label
        short_label = truncate_label(alias, 32) if alias else endpoint.short_label
        result = self.run(
            [
                "new-window",
                "-d",
                "-P",
                "-F",
                "#{window_id}",
                "-t",
                f"{self.session_name}:",
                "-n",
                display_label,
                command,
            ]
        )
        window_id = result.stdout.strip()
        if not window_id:
            if was_dismissed:
                self.set_dismissed_keys({*self.dismissed_keys(), endpoint.key})
            raise HubError("tmux did not return the new window id")
        try:
            self.run_batch(
                [
                    *(
                        ["set-option", "-w", "-t", window_id, option, value]
                        for option, value in [
                            (WINDOW_KEY_OPTION, endpoint.key),
                            (WINDOW_COMMAND_OPTION, command),
                            (WINDOW_ENDPOINT_OPTION, serialize_endpoint(endpoint)),
                            (WINDOW_SHORT_LABEL_OPTION, short_label),
                        ]
                    ),
                    ["set-window-option", "-t", window_id, "remain-on-exit", "on"],
                ]
            )
        except HubError:
            self.run(["kill-window", "-t", window_id], check=False)
            if was_dismissed:
                self.set_dismissed_keys({*self.dismissed_keys(), endpoint.key})
            raise
        return ManagedWindow(
            window_id=window_id,
            key=endpoint.key,
            name=display_label,
            dead=False,
            active=False,
            endpoint=endpoint,
            short_label=short_label,
        )

    @serialized_hub_mutation
    def add_missing(self, endpoints: Sequence[Endpoint], herdr_bin: str) -> list[ManagedWindow]:
        existing_keys = {window.key for window in self.managed_windows()}
        dismissed_keys = self.dismissed_keys()
        added: list[ManagedWindow] = []
        for endpoint in endpoints:
            if endpoint.key in existing_keys or endpoint.key in dismissed_keys:
                continue
            window = self.create_window(endpoint, herdr_bin)
            existing_keys.add(endpoint.key)
            added.append(window)
        if self.managed_windows():
            self.remove_bootstrap()
        return added

    @serialized_hub_mutation
    def remove_bootstrap(self) -> None:
        result = self.run(
            [
                "list-windows",
                "-t",
                self.session_name,
                "-F",
                "#{window_id}\t#{window_name}",
            ],
            check=False,
        )
        if result.returncode != 0:
            return
        for line in result.stdout.splitlines():
            fields = line.split("\t", 1)
            if len(fields) == 2 and fields[1] == BOOTSTRAP_WINDOW_NAME:
                self.run(["kill-window", "-t", fields[0]], check=False)

    def select_window(self, window_id: str) -> None:
        self.run(["select-window", "-t", window_id])

    @serialized_hub_mutation
    def dismiss_window(self, window_id: str | None = None) -> ManagedWindow:
        window_id = window_id or self.current_window_id()
        windows = self.managed_windows()
        window = next((item for item in windows if item.window_id == window_id), None)
        if window is None or window.endpoint is None:
            raise HubError("the selected window is not managed by herdr-hub")
        if len(windows) <= 1:
            raise HubError("至少保留一个 Session Tab；可用 herdr-hub stop 退出整个 Hub")

        previous = self.dismissed_keys()
        self.set_dismissed_keys({*previous, window.key})
        result = self.run(["kill-window", "-t", window.window_id], check=False)
        if result.returncode != 0:
            self.set_dismissed_keys(previous)
            detail = (result.stderr or result.stdout or "tmux command failed").strip()
            raise HubError(detail)
        return window

    def select_primary(self, endpoints: Sequence[Endpoint]) -> None:
        preferred = next(
            (endpoint for endpoint in endpoints if endpoint.key == Endpoint.local().key),
            endpoints[0] if endpoints else None,
        )
        if preferred is None:
            return
        window = self.window_for_key(preferred.key)
        if window is not None:
            self.select_window(window.window_id)

    def current_window_id(self) -> str:
        arguments = ["display-message", "-p"]
        if pane_id := os.environ.get("TMUX_PANE"):
            arguments.extend(["-t", pane_id])
        arguments.append("#{window_id}")
        result = self.run(arguments)
        window_id = result.stdout.strip()
        if not window_id:
            raise HubError("could not identify the current hub window")
        return window_id

    @serialized_hub_mutation
    def reconnect(self, window_id: str | None = None) -> None:
        window_id = window_id or self.current_window_id()
        command_result = self.run(
            ["show-options", "-w", "-v", "-t", window_id, WINDOW_COMMAND_OPTION],
            check=False,
        )
        command = command_result.stdout.strip()
        if command_result.returncode != 0 or not command:
            raise HubError("the selected window is not managed by herdr-hub")
        self.run(["respawn-window", "-k", "-t", window_id, command])

    @serialized_hub_mutation
    def replace_endpoint(
        self,
        window: ManagedWindow,
        endpoint: Endpoint,
        herdr_bin: str,
    ) -> ManagedWindow:
        command = "exec " + shlex.join(endpoint.command(herdr_bin, self.ssh_bin))
        alias = self.aliases.get(endpoint.key)
        short_label = truncate_label(alias, 32) if alias else endpoint.short_label
        replacements = [
            (WINDOW_COMMAND_OPTION, command),
            (WINDOW_ENDPOINT_OPTION, serialize_endpoint(endpoint)),
            (WINDOW_SHORT_LABEL_OPTION, short_label),
        ]
        snapshot = self.run(
            [
                "display-message",
                "-p",
                "-t",
                window.window_id,
                "\t".join(f"#{{{option}}}" for option, _ in replacements),
            ],
            check=False,
        )
        previous: dict[str, str] = {}
        if snapshot.returncode == 0:
            values = snapshot.stdout.rstrip("\n").split("\t", len(replacements) - 1)
            if len(values) == len(replacements):
                previous = {option: value for (option, _), value in zip(replacements, values)}
        try:
            self.run_batch(
                [
                    *(
                        ["set-option", "-w", "-t", window.window_id, option, value]
                        for option, value in replacements
                    ),
                    ["respawn-window", "-k", "-t", window.window_id, command],
                ]
            )
        except HubError:
            for option, value in previous.items():
                self.run(
                    ["set-option", "-w", "-t", window.window_id, option, value],
                    check=False,
                )
            raise
        replaced = self.window_for_key(endpoint.key)
        if replaced is None:
            raise HubError("the selected Session window disappeared during reconnect")
        self._apply_window_label(replaced)
        return self.window_for_key(endpoint.key) or replaced

    @serialized_hub_mutation
    def activate_endpoint(
        self,
        endpoint: Endpoint,
        herdr_bin: str,
        *,
        verify_startup: bool = False,
    ) -> ManagedWindow:
        window = self.window_for_key(endpoint.key)
        started = False
        if window is None:
            window = self.create_window(endpoint, herdr_bin)
            self.remove_bootstrap()
            started = True
        elif window.endpoint is not None and (
            window.endpoint.remote_transport != endpoint.remote_transport
            or window.endpoint.remote_keybindings != endpoint.remote_keybindings
        ):
            window = self.replace_endpoint(window, endpoint, herdr_bin)
            started = True
        elif window.dead:
            self.reconnect(window.window_id)
            started = True
        if verify_startup and started:
            window = self.wait_for_window_startup(window.window_id)
        self.select_window(window.window_id)
        return window

    def wait_for_window_startup(
        self,
        window_id: str,
        *,
        timeout: float = 0.4,
    ) -> ManagedWindow:
        """Catch commands that exit immediately before presenting their UI."""
        deadline = time.monotonic() + timeout
        while True:
            window = next(
                (
                    candidate
                    for candidate in self.managed_windows()
                    if candidate.window_id == window_id
                ),
                None,
            )
            if window is None:
                raise HubError("Session Tab 在启动过程中消失")
            if window.dead:
                capture = self.run(
                    ["capture-pane", "-p", "-t", window_id, "-S", "-20"],
                    check=False,
                )
                lines = [
                    line.strip()
                    for line in (capture.stdout or "").splitlines()
                    if line.strip()
                ]
                detail = " · ".join(lines[-3:])
                message = "Session 进程启动后立即退出"
                if detail:
                    message += f"：{detail}"
                raise HubError(message)
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                return window
            time.sleep(min(0.05, remaining))

    def attach(self) -> None:
        if inside_socket(self.socket_name):
            return
        ensure_not_nested(self.socket_name)
        result = self.run(
            ["attach-session", "-t", self.session_name],
            check=False,
            capture_output=False,
            text=True,
        )
        if result.returncode != 0:
            raise HubError("could not attach to the herdr-hub tmux session")

    def display_message(self, message: str) -> None:
        self.run(["display-message", message], check=False)

    @serialized_hub_mutation
    def stop(self) -> None:
        if not self.exists():
            return
        self.run(["kill-server"])


def inside_any_tmux() -> bool:
    return bool(os.environ.get("TMUX"))


def inside_socket(socket_name: str) -> bool:
    value = os.environ.get("TMUX", "")
    socket_path = value.split(",", 1)[0]
    return bool(socket_path) and Path(socket_path).name == socket_name


def resolve_binary(name: str, override: str | None = None) -> str:
    if override:
        return override
    resolved = shutil.which(name)
    if resolved is None:
        raise HubError(f"required executable not found on PATH: {name}")
    return resolved


def invocation_script_path() -> str:
    invoked = Path(sys.argv[0]).absolute()
    if invoked.name == "__main__.py":
        repository_launcher = Path(__file__).resolve().parents[1] / "bin" / "herdr-hub"
        return str(repository_launcher)
    return str(invoked)


def configured_socket_name() -> str:
    return os.environ.get("HERDR_HUB_SOCKET", DEFAULT_SOCKET_NAME)


def configured_tmux_config() -> str:
    override = os.environ.get("HERDR_HUB_TMUX_CONFIG")
    return os.path.expanduser(override) if override else DEFAULT_TMUX_CONFIG


def load_alias_table() -> tuple[dict[str, str], str | None]:
    """Load display names, degrading to defaults when the file is unreadable.

    A corrupt optional file must not block start/refresh/open; `doctor`
    reports the problem and the warning is surfaced to the caller.
    """
    registry = make_alias_registry()
    try:
        return registry.load(), None
    except AliasRegistryError as error:
        return {}, f"display names unavailable, using defaults: {error}"


def make_hub(
    arguments: argparse.Namespace,
    *,
    load_aliases: bool = True,
    warnings: list[str] | None = None,
) -> TmuxHub:
    tmux_bin = resolve_binary("tmux", os.environ.get("HERDR_HUB_TMUX_BIN"))
    ssh_bin = os.environ.get("HERDR_HUB_SSH_BIN") or shutil.which("ssh") or "ssh"
    aliases: dict[str, str] = {}
    if load_aliases:
        aliases, alias_warning = load_alias_table()
        if alias_warning is not None:
            if warnings is not None:
                warnings.append(alias_warning)
            else:
                print(f"herdr-hub: warning: {alias_warning}", file=sys.stderr)
    return TmuxHub(
        tmux_bin=tmux_bin,
        ssh_bin=ssh_bin,
        socket_name=configured_socket_name(),
        session_name=os.environ.get("HERDR_HUB_SESSION", DEFAULT_TMUX_SESSION),
        config_file=configured_tmux_config(),
        script_path=invocation_script_path(),
        aliases=aliases,
    )


def resolve_herdr() -> str:
    return resolve_binary("herdr", os.environ.get("HERDR_HUB_HERDR_BIN"))


def resolve_ssh() -> str:
    return resolve_binary("ssh", os.environ.get("HERDR_HUB_SSH_BIN"))


def configured_proc_root() -> Path:
    return Path(os.environ.get("HERDR_HUB_PROC_ROOT", "/proc"))


def make_registry() -> RemoteRegistry:
    return RemoteRegistry(default_registry_path())


def make_alias_registry() -> AliasRegistry:
    return AliasRegistry(default_alias_registry_path())


def remember_remote_endpoints(endpoints: Iterable[Endpoint]) -> str | None:
    entries = [
        RemoteEntry(
            endpoint.target,
            endpoint.remote_keybindings,
            endpoint.remote_transport,
        )
        for endpoint in endpoints
        if endpoint.kind == "remote" and endpoint.target is not None
    ]
    if not entries:
        return None
    try:
        make_registry().remember_many(entries)
    except RegistryError as error:
        return f"remote registry: {error}"
    return None


def catalog_windows(windows: Iterable[ManagedWindow]) -> list[CatalogWindow]:
    return [
        CatalogWindow(
            window_id=window.window_id,
            key=window.key,
            label=window.name,
            dead=window.dead,
            active=window.active,
            kind=window.endpoint.kind if window.endpoint is not None else None,
            target=window.endpoint.target if window.endpoint is not None else None,
            session=window.endpoint.session if window.endpoint is not None else None,
            default_label=(
                window.endpoint.label if window.endpoint is not None else window.name
            ),
            remote_transport=(
                window.endpoint.remote_transport
                if window.endpoint is not None
                else "native"
            ),
        )
        for window in windows
    ]


def ensure_not_nested(socket_name: str) -> None:
    if inside_any_tmux() and not inside_socket(socket_name):
        raise HubError(
            "already inside another tmux server; detach from it before attaching to herdr-hub"
        )


def command_start(arguments: argparse.Namespace) -> int:
    # Fail before discovery and server creation: a Hub nobody can attach to
    # would otherwise be left running in the background.
    ensure_not_nested(configured_socket_name())
    herdr_bin = resolve_herdr()
    report = discover_endpoints(herdr_bin, configured_proc_root())
    if warning := remember_remote_endpoints(report.endpoints):
        report.warnings.append(warning)
    hub = make_hub(arguments)
    created = hub.ensure_server()
    added = hub.add_missing(report.endpoints, herdr_bin)
    windows = hub.managed_windows()
    if not windows:
        hub.stop()
        detail = "; ".join(report.warnings)
        suffix = f" ({detail})" if detail else ""
        raise HubError(
            "no running Herdr sessions or live --remote clients were found; "
            f"use 'herdr-hub open' to create one{suffix}"
        )
    if created:
        hub.select_primary(report.endpoints)
    if added and not created:
        hub.display_message(f"herdr-hub: added {len(added)} session(s)")
    hub.attach()
    return 0


def command_refresh(arguments: argparse.Namespace) -> int:
    herdr_bin = resolve_herdr()
    report = discover_endpoints(herdr_bin, configured_proc_root())
    if warning := remember_remote_endpoints(report.endpoints):
        report.warnings.append(warning)
    hub = make_hub(arguments)
    hub.ensure_server()
    added = hub.add_missing(report.endpoints, herdr_bin)
    message = f"herdr-hub: added {len(added)} session(s)"
    if report.warnings:
        message += f"; {report.warnings[0]}"
    has_windows = bool(hub.managed_windows())
    if not has_windows:
        hub.stop()
    if inside_socket(hub.socket_name) and has_windows:
        hub.display_message(message)
    else:
        print(message)
    return 0


def command_open(arguments: argparse.Namespace) -> int:
    herdr_bin = resolve_herdr()
    try:
        endpoint = (
            Endpoint.remote(
                arguments.remote,
                arguments.session,
                arguments.remote_keybindings,
                arguments.transport,
            )
            if arguments.remote is not None
            else Endpoint.local(arguments.session)
        )
    except ValueError as error:
        raise HubError(str(error)) from error
    if endpoint.kind == "remote":
        assert endpoint.target is not None
        try:
            make_registry().upsert(
                RemoteEntry(
                    endpoint.target,
                    endpoint.remote_keybindings,
                    endpoint.remote_transport,
                )
            )
        except RegistryError as error:
            print(f"herdr-hub: warning: remote registry: {error}", file=sys.stderr)
    hub = make_hub(arguments)
    hub.ensure_server()
    hub.activate_endpoint(endpoint, herdr_bin)
    hub.attach()
    return 0


def local_session_listing(herdr_bin: str) -> tuple[list[str], list[str]]:
    """Background half of the Sessions popup: names and warnings only."""
    report = discover_local_sessions(herdr_bin)
    return [endpoint.session for endpoint in report.endpoints], report.warnings


def command_choose(arguments: argparse.Namespace) -> int:
    startup_warnings: list[str] = []
    hub = make_hub(arguments, warnings=startup_warnings)
    if not hub.exists():
        raise HubError("herdr-hub is not running")
    windows = hub.managed_windows()
    herdr_bin = resolve_herdr()
    registry_warning = remember_remote_endpoints(
        window.endpoint for window in windows if window.endpoint is not None
    )

    alias_registry = make_alias_registry()
    remote_registry = make_registry()
    ssh_bin = resolve_ssh()
    # `herdr session list` can take seconds; the popup must not wait for it.
    local_executor = ThreadPoolExecutor(
        max_workers=1, thread_name_prefix="herdr-hub-local"
    )
    local_future = local_executor.submit(local_session_listing, herdr_bin)

    def close_window(window_id: str) -> list[CatalogWindow]:
        hub.dismiss_window(window_id)
        return catalog_windows(hub.managed_windows())

    def rename_session(endpoint_key: str, alias: str) -> None:
        # Apply to tmux first: a name tmux rejects must never be persisted,
        # or every later start/refresh would fail while re-applying it.
        previous = hub.aliases.get(endpoint_key, "")
        if hub.apply_alias(endpoint_key, alias) is None:
            hub.apply_alias(endpoint_key, previous)
            raise HubError("the selected Session is no longer open")
        try:
            alias_registry.set(endpoint_key, alias)
        except AliasRegistryError:
            hub.apply_alias(endpoint_key, previous)
            raise

    def delete_session(
        target: str,
        session: str,
        window_id: str | None,
    ) -> DeleteSessionOutcome:
        delete_remote_session(ssh_bin, RemoteEntry(target), session)
        endpoint_key = Endpoint.remote(target, session).key
        warnings: list[str] = []
        tab_closed = False
        if window_id is not None:
            try:
                hub.dismiss_window(window_id)
                tab_closed = True
            except HubError as error:
                warnings.append(f"Tab 未关闭：{error}")
        try:
            alias_registry.set(endpoint_key, "")
        except AliasRegistryError as error:
            warnings.append(f"显示名称清理失败：{error}")
        hub.aliases.pop(endpoint_key, None)
        try:
            hub.restore_endpoint(endpoint_key)
        except HubError as error:
            warnings.append(f"隐藏记录清理失败：{error}")
        return DeleteSessionOutcome(
            warning="；".join(warnings) or None,
            windows=tuple(catalog_windows(hub.managed_windows())),
            tab_closed=tab_closed,
        )

    def stop_session(
        window_id: str,
        target: str,
        session: str,
    ) -> StopSessionOutcome:
        stop_remote_session(ssh_bin, RemoteEntry(target), session)
        warning: str | None = None
        tab_closed = False
        try:
            hub.dismiss_window(window_id)
            tab_closed = True
        except HubError as error:
            warning = f"Session 已停止，但 Tab 未关闭：{error}"
        updated = catalog_windows(hub.managed_windows())
        if not tab_closed:
            updated = [
                replace(window, dead=True)
                if window.window_id == window_id
                else window
                for window in updated
            ]
        return StopSessionOutcome(tuple(updated), tab_closed, warning)

    def activate_selection(result: SelectorResult) -> None:
        if result.kind == "window":
            assert result.window_id is not None
            if result.dead:
                hub.reconnect(result.window_id)
                hub.wait_for_window_startup(result.window_id)
            hub.select_window(result.window_id)
            return
        if result.kind == "local":
            assert result.session is not None
            hub.activate_endpoint(
                Endpoint.local(result.session),
                resolve_herdr(),
                verify_startup=True,
            )
            return

        assert result.target is not None and result.session is not None
        endpoint = Endpoint.remote(
            result.target,
            result.session,
            result.remote_keybindings,
            result.remote_transport,
        )
        try:
            remote_registry.upsert(
                RemoteEntry(
                    endpoint.target,
                    endpoint.remote_keybindings,
                    endpoint.remote_transport,
                )
            )
        except RegistryError as error:
            hub.display_message(f"herdr-hub: remote registry: {error}")
        hub.activate_endpoint(
            endpoint,
            resolve_herdr(),
            verify_startup=True,
        )

    try:
        selector = SessionSelector(
            catalog_windows(windows),
            remote_registry,
            ssh_bin,
            aliases=hub.aliases,
            activate_handler=activate_selection,
            rename_handler=rename_session,
            close_window_handler=close_window,
            delete_session_handler=delete_session,
            stop_session_handler=stop_session,
            pending_local_sessions=local_future,
        )
        if registry_warning:
            startup_warnings.append(registry_warning)
        if startup_warnings:
            selector.message = "; ".join(startup_warnings)
        result = selector.run()
    except (RegistryError, SelectorError) as error:
        raise HubError(str(error)) from error
    finally:
        local_executor.shutdown(wait=False, cancel_futures=True)
    if result is None:
        return 0
    if result.kind == "rename":
        assert result.endpoint_key is not None and result.alias is not None
        rename_session(result.endpoint_key, result.alias)
        window = hub.window_for_key(result.endpoint_key)
        assert window is not None
        action = "renamed" if result.alias else "restored the default name for"
        hub.display_message(f"herdr-hub: {action} {window.name}")
        return 0
    if result.kind == "close":
        assert result.window_id is not None
        hub.dismiss_window(result.window_id)
        return 0
    activate_selection(result)
    return 0


def command_control(arguments: argparse.Namespace) -> int:
    hub = make_hub(arguments, load_aliases=False)
    windows = hub.managed_windows()
    if arguments.window:
        window = next(
            (item for item in windows if item.window_id == arguments.window),
            None,
        )
    else:
        active = [item for item in windows if item.active]
        window = active[0] if len(active) == 1 else None
    if window is None or window.endpoint is None:
        raise HubError("无法确定当前 Herdr Hub Tab，请回到 Session 页面后重试")
    if window.dead:
        raise HubError("当前 Herdr 连接已断开，请先按 Ctrl+A R 重连")
    # Endpoint metadata is authoritative; popup environment may belong to another server.
    binary = resolve_herdr() if window.endpoint.kind == "local" else "herdr"
    control = HerdrControl(window.endpoint, binary, hub.ssh_bin)
    menu = ControlMenu(control)
    result = curses.wrapper(menu.run)
    if menu.reconnect_requested:
        # Close curses first; respawning the client must not destroy its popup.
        current = next((item for item in hub.managed_windows() if item.window_id == window.window_id), None)
        if current is None or current.endpoint != window.endpoint:
            raise HubError("服务器已更新，但原 Hub Tab 已改变；请手动连接目标 Session")
        hub.reconnect(window.window_id)
        hub.wait_for_window_startup(window.window_id)
        hub.display_message("Herdr 服务器已更新，当前 Tab 已重新连接")
    return result


def command_keys(arguments: argparse.Namespace) -> int:
    try:
        return curses.wrapper(KeyHelpPopup().run)
    except curses.error as error:
        raise HubError(f"无法打开快捷键帮助: {error}") from error


def command_remote(arguments: argparse.Namespace) -> int:
    registry = make_registry()
    action = arguments.remote_command
    if action == "list":
        entries = registry.load()
        if not entries:
            print("no registered remote targets")
            return 0
        for entry in entries:
            print(
                f"{entry.target}\t{entry.remote_keybindings}\t{entry.remote_transport}"
            )
        return 0
    if action == "add":
        entry = RemoteEntry(
            arguments.target,
            arguments.remote_keybindings,
            arguments.transport,
        )
        changed = registry.upsert(entry)
        print(f"{'registered' if changed else 'already registered'} {entry.target}")
        return 0
    if action == "remove":
        removed = registry.remove(arguments.target)
        print(f"{'removed' if removed else 'not registered'} {arguments.target}")
        return 0
    raise HubError("use 'herdr-hub remote list|add|remove'")


def command_reconnect(arguments: argparse.Namespace) -> int:
    hub = make_hub(arguments, load_aliases=False)
    if not hub.exists():
        raise HubError("herdr-hub is not running")
    hub.reconnect(arguments.window)
    if inside_socket(hub.socket_name):
        hub.display_message("herdr-hub: reconnecting current session")
    return 0


def command_close(arguments: argparse.Namespace) -> int:
    hub = make_hub(arguments, load_aliases=False)
    if not hub.exists():
        raise HubError("herdr-hub is not running")
    try:
        window = hub.dismiss_window(arguments.window)
    except HubError as error:
        if inside_socket(hub.socket_name):
            hub.display_message(f"herdr-hub: {error}")
            return 1
        raise
    message = f"herdr-hub: closed {window.name}; Herdr Session is still running"
    if inside_socket(hub.socket_name):
        hub.display_message(message)
    else:
        print(message)
    return 0


def command_stop(arguments: argparse.Namespace) -> int:
    hub = make_hub(arguments, load_aliases=False)
    if hub.exists():
        hub.stop()
        print("stopped herdr-hub; Herdr servers were left running")
    else:
        print("herdr-hub is not running")
    return 0


def command_doctor(arguments: argparse.Namespace) -> int:
    failures = 0
    print(
        f"control: {CONTROL_SHORTCUT} (Hub prefix); "
        "Fork requires Herdr 0.9+ on target"
    )
    print(f"python: {sys.version.split()[0]} ({sys.executable})")
    print(f"curses: {getattr(curses, 'version', 'available')}")
    for binary in ["tmux", "herdr", "ssh"]:
        override = os.environ.get(f"HERDR_HUB_{binary.upper()}_BIN")
        path = shutil.which(binary) if override is None else override
        if path is None:
            print(f"{binary}: missing")
            failures += 1
            continue
        version_args = [path, "-V"] if binary in {"tmux", "ssh"} else [path, "--version"]
        try:
            result = subprocess.run(
                version_args,
                capture_output=True,
                text=True,
                check=False,
                timeout=10,
            )
        except (OSError, subprocess.SubprocessError) as error:
            print(f"{binary}: cannot run {path} ({error})")
            failures += 1
            continue
        version = (result.stdout or result.stderr).strip()
        print(f"{binary}: {version or path}")
    registry = make_registry()
    try:
        registered = registry.load()
        print(f"remotes: {len(registered)} registered ({registry.path})")
    except RegistryError as error:
        print(f"remotes: invalid ({error})")
        failures += 1
    alias_registry = make_alias_registry()
    try:
        aliases = alias_registry.load()
        print(f"aliases: {len(aliases)} configured ({alias_registry.path})")
    except AliasRegistryError as error:
        print(f"aliases: invalid ({error})")
        failures += 1
    proc_root = configured_proc_root()
    print(f"proc: {proc_root} ({'available' if proc_root.is_dir() else 'missing'})")

    try:
        herdr_bin = resolve_herdr()
    except HubError as error:
        print(f"discovery: unavailable ({error})")
        return 1
    report = discover_endpoints(herdr_bin, proc_root)
    print(f"discovered: {len(report.endpoints)} endpoint(s)")
    for endpoint in report.endpoints:
        print(f"  {endpoint.label} [{endpoint.key}]")
    for warning in report.warnings:
        print(f"warning: {warning}")
    return failures


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="herdr-hub",
        description="Switch local and live --remote Herdr sessions through a tmux top bar.",
    )
    subparsers = parser.add_subparsers(dest="command")
    subparsers.add_parser("start", help="scan, create and attach to the hub")
    subparsers.add_parser("refresh", help="add newly discovered sessions")

    open_parser = subparsers.add_parser("open", help="open a session directly in the hub")
    open_parser.add_argument("--remote", metavar="TARGET")
    open_parser.add_argument("--session", default=DEFAULT_SESSION_NAME)
    open_parser.add_argument(
        "--remote-keybindings",
        choices=["local", "server"],
        default="local",
    )
    open_parser.add_argument(
        "--transport",
        choices=["native", "ssh"],
        default="native",
        help="remote attach transport; Windows targets require ssh",
    )

    control_parser = subparsers.add_parser("control", help="open Herdr commands for the current Hub window")
    control_parser.add_argument("--window", help="originating tmux window ID")
    subparsers.add_parser("keys", help="show Hub shortcut mappings")
    subparsers.add_parser("choose", help="browse and open local or remote sessions")
    remote_parser = subparsers.add_parser("remote", help="manage remembered remote targets")
    remote_subparsers = remote_parser.add_subparsers(dest="remote_command")
    remote_subparsers.add_parser("list", help="list remembered remote targets")
    remote_add_parser = remote_subparsers.add_parser("add", help="remember a remote target")
    remote_add_parser.add_argument("target", metavar="TARGET")
    remote_add_parser.add_argument(
        "--remote-keybindings",
        choices=["local", "server"],
        default="local",
    )
    remote_add_parser.add_argument(
        "--transport",
        choices=["native", "ssh"],
        default="native",
        help="preferred attach transport; use ssh for Windows targets",
    )
    remote_remove_parser = remote_subparsers.add_parser(
        "remove", help="forget a remote target"
    )
    remote_remove_parser.add_argument("target", metavar="TARGET")
    reconnect_parser = subparsers.add_parser(
        "reconnect", help="respawn a dead managed window"
    )
    reconnect_parser.add_argument("--window")
    close_parser = subparsers.add_parser(
        "close", help="hide a managed window without stopping its Herdr session"
    )
    close_parser.add_argument("--window")
    subparsers.add_parser("doctor", help="check dependencies and discovery")
    subparsers.add_parser("stop", help="stop the hub clients, leaving Herdr servers running")
    return parser


def main(argv: Sequence[str] | None = None) -> int:
    parser = build_parser()
    arguments = parser.parse_args(argv)
    command = arguments.command or "start"
    handlers = {
        "start": command_start,
        "refresh": command_refresh,
        "open": command_open,
        "choose": command_choose,
        "control": command_control,
        "keys": command_keys,
        "remote": command_remote,
        "reconnect": command_reconnect,
        "close": command_close,
        "doctor": command_doctor,
        "stop": command_stop,
    }
    try:
        return handlers[command](arguments)
    except (AliasRegistryError, HubError, RegistryError, ValueError) as error:
        print(f"herdr-hub: {error}", file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        return 130


if __name__ == "__main__":
    raise SystemExit(main())