diff --git a/README.md b/README.md index 5b265f0..634cd8b 100644 --- a/README.md +++ b/README.md @@ -27,7 +27,7 @@ sidebar 插入 UI，也不能让 thin client 从一个 server 原地改连另一 - Linux - Python 3.11+ - tmux 3.4+ -- Herdr 0.8.2+ +- Herdr 0.9+（本地和需要 Fork 的远端目标均需支持 0.9 的控制 API） - OpenSSH client 检查当前环境： @@ -93,6 +93,7 @@ Hub 使用独立 tmux socket `herdr-hub`，不会接管普通 tmux server。外 | 操作 | 快捷键 / 命令 | | --- | --- | | 点击切换 | 鼠标点击顶部 session | +| 打开 Herdr 命令菜单 | `Ctrl+A h` | | 打开 Session 弹窗 | 点击左上角 `Sessions` 或 `Ctrl+A g` | | 上一个/下一个 window | `Ctrl+A p` / `Ctrl+A n` | | 按编号切换 | `Ctrl+A 0..9` | @@ -105,6 +106,31 @@ Hub 使用独立 tmux socket `herdr-hub`，不会接管普通 tmux server。外 | 重新进入 Hub | `herdr-hub` | | 停止 Hub clients | `herdr-hub stop` | +### Fork 当前 Agent 到新 Tab + +按 **Ctrl+A h** 打开独立的 Herdr 命令菜单，选择“Fork 当前会话到新 Tab”， +按 Enter 或单击执行，Esc 关闭。菜单只显示当前 Agent、Workspace、主操作和必要的状态信息。 +操作正在执行时需等待结果后再关闭，以免将关闭弹窗误认为取消已经发送的请求。 +Sessions 弹窗仍使用 `Ctrl+A g`；两个菜单都使用 Hub 外层的 `Ctrl+A` 前缀。 +快捷键集中定义在 `herdr_hub/control.py` 的 `CONTROL_KEY` 常量中。 + +Fork 复制的是 **Codex / Claude conversation**，并保留正在运行的原 Agent 和原 Tab。 +它在同一个 Herdr Workspace、相同 cwd 新建 Tab，使用当前 pane 已报告的 conversation ID： +Codex 执行 `codex fork UUID`；Claude 执行 `claude --resume SESSION_ID --fork-session`。 +新 Agent 被 Herdr 确认为就绪后才切换焦点。它不会新建外层 Herdr named session， +也不会创建 Git worktree；两个 Agent 会编辑同一工作目录，需自行协调并发修改。 + +支持本地 default / named session，以及通过 SSH 控制的远端 native / Windows SSH endpoint。 +控制目标取自触发菜单的 Hub window metadata，不使用 popup 继承的 Herdr socket 或 pane 环境变量。 +目标必须运行 Herdr 0.9+，并安装对应 Agent；当前 pane 必须已报告可用的 Agent session ID。 +远端控制使用非交互 SSH（需要已信任的主机密钥和可用认证）；Windows 需要 PowerShell。 +启动 blocked、失败、超时或响应不完整时，菜单保留错误信息，**不会自动关闭任何 Tab**。 +超时并不代表远端没有创建或启动成功，请先检查提示中的新 Tab，再重新打开菜单重试。 + +CLI 入口是 `herdr-hub control`。弹窗打开时会锁定 Hub 当前激活的 managed window， +再从其 Endpoint 对应的 Herdr server 读取当前页面的 Agent conversation。 +更新后运行 `herdr-hub refresh` 配置现有 Hub 的快捷键。`herdr-hub doctor` 会显示入口和版本要求。 + ## 远端 Session 弹窗 弹窗由 tmux `display-popup` 和 Hub 内置选择器组成，不依赖 fzf。它会立即显示已打开的 diff --git a/herdr_hub/cli.py b/herdr_hub/cli.py index c306acf..6b12843 100644 --- a/herdr_hub/cli.py +++ b/herdr_hub/cli.py @@ -22,6 +22,7 @@ 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 .remotes import ( RegistryError, @@ -581,7 +582,7 @@ class TmuxHub: ["set-option", "-g", "mouse", "on"], ["set-option", "-g", "focus-events", "on"], ["set-option", "-g", "status-left-length", "18"], - ["set-option", "-g", "status-right-length", "42"], + ["set-option", "-g", "status-right-length", "64"], [ "set-option", "-g", @@ -596,7 +597,7 @@ class TmuxHub: "status-right", "#{?#{<:#{client_width},80},," "#[dim]#{?pane_dead,连接已断开 · C-a R 重连," - "C-a g choose · q close · r refresh}#[default]}", + f"{CONTROL_SHORTCUT} commands · C-a g choose · q close · r refresh}}#[default]}}", ], ["set-option", "-g", "status-style", "bg=colour235,fg=colour250"], ["set-window-option", "-g", "window-status-separator", ""], @@ -638,12 +639,18 @@ class TmuxHub: close = shlex.join( [self.python_bin, self.script_path, "close", "--window", "#{window_id}"] ) + control = shlex.join([self.python_bin, self.script_path, "control"]) confirm_close = shlex.join(["run-shell", "-b", close]) # Remove tmux's default/legacy '&' binding when reconfiguring both new # and already-running Hub servers. self.run(["unbind-key", "-T", "prefix", "&"], check=False) self.run(["unbind-key", "-T", "prefix", "Q"], check=False) + # Remove the short-lived global binding used by the first control-menu + # prototype when reconfiguring an already-running Hub. + self.run(["unbind-key", "-T", "root", "C-g"], check=False) binding_commands = [ + ["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], [ @@ -1423,6 +1430,27 @@ def command_choose(arguments: argparse.Namespace) -> int: 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) + return curses.wrapper(ControlMenu(control).run) + + def command_remote(arguments: argparse.Namespace) -> int: registry = make_registry() action = arguments.remote_command @@ -1493,6 +1521,10 @@ def command_stop(arguments: argparse.Namespace) -> int: 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"]: @@ -1566,6 +1598,8 @@ def build_parser() -> argparse.ArgumentParser: 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("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") @@ -1609,6 +1643,7 @@ def main(argv: Sequence[str] | None = None) -> int: "refresh": command_refresh, "open": command_open, "choose": command_choose, + "control": command_control, "remote": command_remote, "reconnect": command_reconnect, "close": command_close, diff --git a/herdr_hub/control.py b/herdr_hub/control.py new file mode 100644 index 0000000..d0c617b --- /dev/null +++ b/herdr_hub/control.py @@ -0,0 +1,457 @@ +"""Herdr command menu and non-destructive conversation forks.""" +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +import curses +from dataclasses import dataclass +import json +import os +import subprocess +import threading +import time +from typing import TYPE_CHECKING, Callable, Sequence +import uuid +import unicodedata + +from .remotes import ( + RemoteEntry, decode_remote_output, parse_remote_session_result, + remote_control_command, remote_session_command, +) + +if TYPE_CHECKING: + from .cli import Endpoint + + +# A dedicated command inside the Hub's existing Ctrl+A prefix table. +CONTROL_KEY = "h" +CONTROL_SHORTCUT = "C-a h" +ROUTING_ENV = ( + "HERDR_SOCKET_PATH", "HERDR_CLIENT_SOCKET_PATH", "HERDR_SESSION", + "HERDR_PANE_ID", "HERDR_TAB_ID", "HERDR_WORKSPACE_ID", + "HERDR_ACTIVE_PANE_ID", "HERDR_ACTIVE_TAB_ID", "HERDR_ACTIVE_WORKSPACE_ID", +) + + +class ControlError(RuntimeError): + """A diagnostic safe to display inside the command menu.""" + + +class ControlCancelled(RuntimeError): + """A read-only control request cancelled while closing the popup.""" + + +def object_field(data: dict, name: str) -> dict: + value = data.get(name) + if not isinstance(value, dict): + raise ControlError(f"Herdr 返回缺少 {name} 对象（需要 Herdr 0.9+）") + return value + + +def string_field(data: dict, name: str) -> str: + value = data.get(name) + if not isinstance(value, str) or not value.strip() or "\0" in value: + raise ControlError(f"Herdr 返回缺少有效 {name}（需要 Herdr 0.9+）") + return value + + +def result_object(payload: object) -> dict: + if not isinstance(payload, dict): + raise ControlError("Herdr 返回的 JSON 不是对象") + if payload.get("error"): + raise ControlError(f"Herdr: {json.dumps(payload['error'], ensure_ascii=False)}") + return object_field(payload, "result") if "result" in payload else payload + + +@dataclass(frozen=True) +class ForkSource: + workspace_id: str + pane_id: str + tab_id: str + cwd: str + kind: str + session_id: str + workspace_label: str + existing_names: frozenset[str] + + @property + def arguments(self) -> list[str]: + if self.kind == "codex": + return ["fork", self.session_id] + return ["--resume", self.session_id, "--fork-session"] + + +def parse_snapshot(payload: object) -> ForkSource: + result = result_object(payload) + snapshot = object_field(result, "snapshot") if "snapshot" in result else result + workspace_id = string_field(snapshot, "focused_workspace_id") + pane_id = string_field(snapshot, "focused_pane_id") + panes = snapshot.get("panes") + if not isinstance(panes, list): + raise ControlError("Herdr snapshot 缺少 panes") + matches = [p for p in panes if isinstance(p, dict) and p.get("pane_id") == pane_id] + if len(matches) != 1: + raise ControlError("找不到唯一的当前 Agent pane") + pane = matches[0] + if pane.get("workspace_id") != workspace_id: + raise ControlError("当前 pane 与 Workspace 不一致，请重新打开菜单") + kind = pane.get("agent") + if not kind: + raise ControlError("当前 pane 没有运行 Agent") + if kind not in ("codex", "claude"): + raise ControlError(f"暂不支持 Fork {kind}；支持 Codex / Claude") + session = object_field(pane, "agent_session") + if session.get("kind") != "id" or session.get("agent") != kind: + raise ControlError("当前 Agent 尚未报告可 Fork 的 conversation ID") + session_id = string_field(session, "value") + if session_id.startswith("-") or any(ord(c) < 32 for c in session_id): + raise ControlError("Agent conversation ID 无效") + if kind == "codex": + try: + uuid.UUID(session_id) + except ValueError as error: + raise ControlError("Codex conversation ID 不是有效 UUID") from error + agents = snapshot.get("agents", []) + if not isinstance(agents, list): + raise ControlError("Herdr snapshot 的 agents 无效") + names = frozenset(a["name"] for a in agents if isinstance(a, dict) and isinstance(a.get("name"), str)) + workspaces = snapshot.get("workspaces", []) + label = workspace_id + if isinstance(workspaces, list): + for workspace in workspaces: + if isinstance(workspace, dict) and workspace.get("workspace_id") == workspace_id: + label = workspace.get("label") or workspace_id + return ForkSource(workspace_id, pane_id, string_field(pane, "tab_id"), string_field(pane, "cwd"), kind, session_id, str(label), names) + + +class HerdrControl: + def __init__(self, endpoint: Endpoint, herdr_bin: str, ssh_bin: str = "ssh") -> None: + self.endpoint = endpoint + self.herdr_bin = herdr_bin + self.ssh_bin = ssh_bin + self.platform: str | None = None + + def _execute( + self, + command: Sequence[str], + timeout: float, + cancel_event: threading.Event | None = None, + ) -> subprocess.CompletedProcess: + environment = {k: v for k, v in os.environ.items() if k not in ROUTING_ENV} + if cancel_event is not None and cancel_event.is_set(): + raise ControlCancelled() + if cancel_event is not None: + try: + process = subprocess.Popen( + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + stdin=subprocess.DEVNULL, + env=environment, + ) + except OSError as error: + raise ControlError(f"无法执行 Herdr 命令: {error}") from error + deadline = time.monotonic() + timeout + while True: + if cancel_event.wait(0.025): + process.terminate() + try: + process.communicate(timeout=0.25) + except subprocess.TimeoutExpired: + process.kill() + process.communicate() + raise ControlCancelled() + remaining = deadline - time.monotonic() + if remaining <= 0: + process.kill() + process.communicate() + raise ControlError( + f"命令超时（{timeout:g}s）；操作可能已完成，请检查 Herdr" + ) + try: + stdout, stderr = process.communicate( + timeout=min(0.05, remaining) + ) + except subprocess.TimeoutExpired: + continue + return subprocess.CompletedProcess( + command, + process.returncode, + stdout, + stderr, + ) + try: + return subprocess.run(command, capture_output=True, check=False, timeout=timeout, + env=environment, stdin=subprocess.DEVNULL) + except subprocess.TimeoutExpired as error: + raise ControlError(f"命令超时（{timeout:g}s）；操作可能已完成，请检查 Herdr") from error + except OSError as error: + raise ControlError(f"无法执行 Herdr 命令: {error}") from error + + def _remote_platform( + self, + entry: RemoteEntry, + cancel_event: threading.Event | None = None, + ) -> str: + if self.platform is None: + for fallback in (False, True): + result = self._execute( + remote_session_command( + self.ssh_bin, + entry, + posix_fallback=fallback, + ), + 8, + cancel_event, + ) + if result.returncode != 127 or fallback: + break + missing = decode_remote_output(result.stderr) + decode_remote_output(result.stdout) + if result.returncode != 0 and any(marker in missing.lower() for marker in ( + "not recognized", "不是内部或外部命令", "commandnotfoundexception", + )): + # A Windows shell can lack Herdr on PATH while the standard + # per-user executable is installed. This probe is read-only. + result = self._execute(remote_control_command(self.ssh_bin, entry, + ["session", "list", "--json"], platform="windows"), 8, + cancel_event) + parsed = parse_remote_session_result(entry, result.returncode, + decode_remote_output(result.stdout), decode_remote_output(result.stderr)) + if parsed.error: + raise ControlError(parsed.error) + if parsed.platform not in ("windows", "posix"): + raise ControlError("无法从远端 Session 路径识别平台，不能安全执行控制命令") + self.platform = parsed.platform + return self.platform + + def call( + self, + arguments: Sequence[str], + *, + timeout: float = 10, + cancel_event: threading.Event | None = None, + ) -> dict: + argv = ["--session", self.endpoint.session, *arguments] + if self.endpoint.kind == "remote": + entry = RemoteEntry(self.endpoint.target or "", self.endpoint.remote_keybindings, + self.endpoint.remote_transport) + command = remote_control_command(self.ssh_bin, entry, argv, + platform=self._remote_platform(entry, cancel_event), + unset_environment=ROUTING_ENV) + else: + command = [self.herdr_bin, *argv] + result = self._execute(command, timeout, cancel_event) + stdout, stderr = decode_remote_output(result.stdout), decode_remote_output(result.stderr) + if result.returncode != 0: + raise ControlError((stderr or stdout or f"Herdr exited {result.returncode}").strip()) + try: + return result_object(json.loads(stdout)) + except json.JSONDecodeError as error: + raise ControlError(f"Herdr 返回无效 JSON: {error}") from error + + def source( + self, + cancel_event: threading.Event | None = None, + ) -> ForkSource: + return parse_snapshot( + self.call(["api", "snapshot"], cancel_event=cancel_event) + ) + + def fork(self, source: ForkSource, progress: Callable[[str], None] = lambda _: None) -> str: + name = f"{source.kind}-fork-{uuid.uuid4().hex[:12]}" + while name in source.existing_names: + name = f"{source.kind}-fork-{uuid.uuid4().hex[:12]}" + progress("正在同一 Workspace 创建新 Tab…") + # Never delete either tab: a timed-out request may already have succeeded. + tab_id = "未知（请检查 Workspace）" + try: + created = self.call(["tab", "create", "--workspace", source.workspace_id, + "--cwd", source.cwd, "--no-focus"]) + tab_id = string_field(object_field(created, "tab"), "tab_id") + root_pane = object_field(created, "root_pane") + pane_id = string_field(root_pane, "pane_id") + if tab_id == source.tab_id or pane_id == source.pane_id: + raise ControlError("返回的 Tab / root pane 与原 Tab / pane 相同，已拒绝启动") + if (root_pane.get("workspace_id") != source.workspace_id + or root_pane.get("tab_id") != tab_id): + raise ControlError("新 root pane 的 Workspace / Tab 不一致，已拒绝启动") + progress(f"正在启动 {source.kind} Fork，等待就绪…") + started = self.call(["agent", "start", name, "--kind", source.kind, + "--pane", pane_id, "--", *source.arguments], timeout=40) + agent = object_field(started, "agent") + if (agent.get("pane_id") != pane_id or agent.get("agent") != source.kind + or agent.get("interactive_ready") is False + or agent.get("launch_pending") is True + or agent.get("agent_status") in ("blocked", "error")): + raise ControlError("新 Agent 未确认就绪: " + json.dumps(started, ensure_ascii=False)) + progress("Fork 已就绪，正在切换 Tab…") + self.call(["tab", "focus", tab_id]) + except ControlError as error: + raise ControlError(f"{error}\n原 Tab 保留；新增 Tab {tab_id} 未自动关闭。请检查后再重试。") from error + return tab_id + + +def display_chunks(text: str, width: int) -> list[str]: + chunks: list[str] = [] + line = "" + cells = 0 + for character in text: + if character == "\n": + chunks.append(line) + line, cells = "", 0 + continue + if ord(character) < 32 or ord(character) == 127: + character = " " + size = 0 if unicodedata.combining(character) else (2 if unicodedata.east_asian_width(character) in ("W", "F") else 1) + if cells + size > width and line: + chunks.append(line) + line, cells = "", 0 + line += character + cells += size + chunks.append(line) + return chunks + + +class ControlMenu: + """Small action menu; all subprocess work stays off the curses thread.""" + ACTIONS = ( + ( + "fork", + "Fork 当前会话到新 Tab", + "保留当前 Tab，在同一 Workspace 和工作目录继续", + ), + ) + + def __init__(self, control: HerdrControl) -> None: + self.control = control + + def run(self, screen) -> int: # type: ignore[no-untyped-def] + import queue + messages: queue.SimpleQueue[str] = queue.SimpleQueue() + screen.keypad(True) + screen.timeout(100) + try: + curses.set_escdelay(25) + curses.curs_set(0) + curses.mousemask(curses.BUTTON1_CLICKED | curses.BUTTON1_PRESSED) + except (AttributeError, curses.error): + pass + source: ForkSource | None = None + status = "" + selected = 0 + cancel_loading = threading.Event() + with ThreadPoolExecutor(max_workers=1) as executor: + future = executor.submit(self.control.source, cancel_loading) + phase = "load" + while True: + while not messages.empty(): + status = messages.get() + if future is not None and future.done(): + try: + value = future.result() + if phase == "load": + source = value + status = "" + else: + return 0 + except ControlCancelled: + return 0 + except Exception as error: + status = str(error) + if phase == "fork": + source = None # Avoid accidental duplicate forks after ambiguous failure. + future = None + screen.erase() + height, width = screen.getmaxyx() + if source is not None: + context = ( + f"当前：{source.kind.capitalize()} · " + f"{source.workspace_label}" + ) + elif future is not None and phase == "load": + context = "正在读取当前页面…" + else: + context = "无法读取当前页面" + self._add(screen, 1, 2, context, curses.A_BOLD) + action_top = 3 + for index, (_, label, detail) in enumerate(self.ACTIONS): + y = action_top + index * 2 + enabled = source is not None and future is None + attribute = ( + curses.A_REVERSE + if enabled and index == selected + else curses.A_NORMAL + if enabled + else curses.A_DIM + ) + self._add(screen, y, 2, f" {label} ", attribute) + self._add(screen, y + 1, 4, detail, curses.A_DIM) + + status_top = action_top + len(self.ACTIONS) * 2 + 1 + for offset, line in enumerate( + display_chunks(status, max(1, width - 5)) if status else [] + ): + self._add(screen, status_top + offset, 2, line, curses.A_DIM) + + footer = ( + "操作执行中，请等待结果" + if future is not None and phase == "fork" + else "Enter 执行 · Esc 关闭" + ) + self._add(screen, height - 2, 2, footer, curses.A_DIM) + screen.refresh() + key = screen.getch() + if key == 27: + if future is not None and phase == "load": + cancel_loading.set() + return 0 + if future is None: + return 0 + status = "操作执行中；完成后可关闭（避免误以为已取消 Fork）。" + elif key in (curses.KEY_UP, curses.KEY_DOWN): + selected = (selected + (1 if key == curses.KEY_DOWN else -1)) % len(self.ACTIONS) + else: + execute = key in (10, 13, curses.KEY_ENTER) + if key == curses.KEY_MOUSE: + try: + _, x, y, _, state = curses.getmouse() + action_index = (y - action_top) // 2 + execute = ( + 1 <= x < width - 1 + and 0 <= action_index < len(self.ACTIONS) + and y in { + action_top + action_index * 2, + action_top + action_index * 2 + 1, + } + and bool( + state + & ( + curses.BUTTON1_CLICKED + | curses.BUTTON1_PRESSED + ) + ) + ) + if execute: + selected = action_index + except curses.error: + execute = False + if execute and future is None and source is not None: + phase = "fork" + future = executor.submit(self.control.fork, source, messages.put) + status = "正在 Fork…" + + @staticmethod + def _add( + screen, + y: int, + x: int, + value: str, + attribute: int = curses.A_NORMAL, + ) -> None: # type: ignore[no-untyped-def] + height, width = screen.getmaxyx() + if y < 0 or y >= height or x >= width: + return + chunks = display_chunks(value, max(1, width - x - 1)) + try: + screen.addstr(y, x, chunks[0] if chunks else "", attribute) + except curses.error: + pass diff --git a/herdr_hub/remotes.py b/herdr_hub/remotes.py index 0bd5b81..a74904c 100644 --- a/herdr_hub/remotes.py +++ b/herdr_hub/remotes.py @@ -708,3 +708,58 @@ class RemoteQueryPool: except subprocess.TimeoutExpired: process.kill() process.communicate() + + +def remote_control_command( + ssh_bin: str, + entry: RemoteEntry, + arguments: Sequence[str], + *, + platform: str, + unset_environment: Sequence[str] = (), +) -> list[str]: + """Build one non-replayed mutation with platform-specific argv encoding. + + Windows SSH commonly uses cmd.exe; only a base64 PowerShell script crosses + that shell. ProcessStartInfo then receives Windows-quoted native arguments. + """ + import base64 + import re + import shlex + + if any(not re.fullmatch(r"[A-Z_][A-Z_0-9]*", key) for key in unset_environment): + raise ValueError("invalid environment variable name") + if any("\0" in argument for argument in arguments): + raise ValueError("NUL in remote argument") + if platform == "posix": + unset = "".join(f"unset {key}; " for key in unset_environment) + argv = shlex.join(arguments) + script = ( + unset + 'if [ -x "$HOME/.local/bin/herdr" ]; then ' + f'exec "$HOME/.local/bin/herdr" {argv}; ' + 'elif command -v herdr >/dev/null 2>&1; then ' + f"exec herdr {argv}; " + "else echo 'herdr is not installed on the remote host' >&2; exit 127; fi" + ) + elif platform == "windows": + def literal(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + unset = "".join(f"Remove-Item Env:{key} -ErrorAction SilentlyContinue; " for key in unset_environment) + script_text = ( + "$ErrorActionPreference='Stop'; " + unset + + "$exe=Join-Path $HOME '.local\\bin\\herdr.exe'; " + "if (!(Test-Path -LiteralPath $exe)) { $exe=(Get-Command herdr -ErrorAction Stop).Source }; " + "$p=New-Object System.Diagnostics.Process; " + "$p.StartInfo.FileName=$exe; " + "$p.StartInfo.UseShellExecute=$false; " + "$p.StartInfo.Arguments=" + literal(subprocess.list2cmdline(arguments)) + "; " + "$null=$p.Start(); $p.WaitForExit(); exit $p.ExitCode" + ) + encoded = base64.b64encode(script_text.encode("utf-16le")).decode("ascii") + script = "powershell.exe -NoLogo -NoProfile -NonInteractive -EncodedCommand " + encoded + else: + raise ValueError("unknown remote platform") + return [ssh_bin, "-T", "-o", "BatchMode=yes", "-o", "StrictHostKeyChecking=yes", + "-o", "ConnectionAttempts=1", "-o", "ConnectTimeout=4", + *ssh_destination_arguments(entry.target), script] diff --git a/herdr_hub/selector.py b/herdr_hub/selector.py index 6c4fe02..ec1d6de 100644 --- a/herdr_hub/selector.py +++ b/herdr_hub/selector.py @@ -588,8 +588,9 @@ class SessionSelector: screen.keypad(True) screen.timeout(100) try: + curses.set_escdelay(25) curses.curs_set(0) - except curses.error: + except (AttributeError, curses.error): pass try: curses.use_default_colors() diff --git a/tests/test_control.py b/tests/test_control.py new file mode 100644 index 0000000..60032de --- /dev/null +++ b/tests/test_control.py @@ -0,0 +1,304 @@ +from __future__ import annotations + +import argparse +import base64 +import curses +import json +import os +import subprocess +import threading +import time +import unittest +from unittest.mock import Mock, patch + +from herdr_hub.cli import Endpoint, HubError, ManagedWindow, build_parser, command_control +from herdr_hub.control import ( + ControlCancelled, + ControlError, + ControlMenu, + HerdrControl, + parse_snapshot, +) +from herdr_hub.remotes import RemoteEntry, remote_control_command + + +SESSION_ID = "0198f891-d531-745e-96f4-ef7d9092298b" + + +def snapshot(kind="codex"): + return {"id": "request", "result": {"snapshot": { + "focused_workspace_id": "w1", "focused_pane_id": "w1:p1", + "panes": [{"pane_id": "w1:p1", "workspace_id": "w1", "tab_id": "t1", "agent": kind, + "cwd": "/tmp/project with spaces", "agent_session": { + "agent": kind, "kind": "id", "value": SESSION_ID}}], + "agents": [{"name": "existing-agent"}], + "workspaces": [{"workspace_id": "w1", "label": "Project"}], + }}} + + +def completed(payload, code=0): + return subprocess.CompletedProcess([], code, json.dumps(payload).encode(), b"") + + +class SnapshotTests(unittest.TestCase): + def test_focused_pane_and_exact_native_arguments(self): + for kind, args in (("codex", ["fork", SESSION_ID]), + ("claude", ["--resume", SESSION_ID, "--fork-session"])): + with self.subTest(kind=kind): + source = parse_snapshot(snapshot(kind)) + self.assertEqual(source.arguments, args) + self.assertEqual(source.workspace_id, "w1") + self.assertEqual(source.workspace_label, "Project") + self.assertEqual(source.existing_names, {"existing-agent"}) + + def test_invalid_snapshots_fail_before_mutation(self): + for field, value in (("agent", None), ("agent", "gemini"), ("cwd", None), + ("workspace_id", "other"), ("agent_session", None), + ("agent_session", {"agent": "codex", "kind": "id", "value": "wrong"}), + ("agent_session", {"agent": "codex", "kind": "path", "value": "x"})): + payload = snapshot() + payload["result"]["snapshot"]["panes"][0][field] = value + with self.subTest(field=field, value=value), self.assertRaises(ControlError): + parse_snapshot(payload) + for field in ("focused_workspace_id", "focused_pane_id", "panes"): + payload = snapshot() + del payload["result"]["snapshot"][field] + with self.subTest(field=field), self.assertRaises(ControlError): + parse_snapshot(payload) + + +class ControlTests(unittest.TestCase): + def test_local_explicit_session_and_clean_environment(self): + for session in ("default", "named"): + with self.subTest(session=session), patch.dict(os.environ, { + "HERDR_SOCKET_PATH": "/wrong/socket", "HERDR_SESSION": "wrong", + "HERDR_PANE_ID": "wrong", "HERDR_CLIENT_SOCKET_PATH": "/wrong/client", + }), patch("herdr_hub.control.subprocess.run", return_value=completed(snapshot())) as run: + source = HerdrControl(Endpoint.local(session), "/fake/herdr").source() + self.assertEqual(source.kind, "codex") + self.assertEqual(run.call_args.args[0], ["/fake/herdr", "--session", session, "api", "snapshot"]) + self.assertNotIn("HERDR_SOCKET_PATH", run.call_args.kwargs["env"]) + self.assertNotIn("HERDR_SESSION", run.call_args.kwargs["env"]) + + def test_remote_native_and_ssh_route_to_target_named_server(self): + for transport, path in (("native", "/home/user/.herdr/default"), + ("ssh", "C:\\Users\\me\\.herdr\\default")): + listing = {"sessions": [{"name": "named", "running": True, "session_dir": path}]} + with self.subTest(transport=transport), patch("herdr_hub.control.subprocess.run", side_effect=[completed(listing), completed(snapshot())]) as run: + control = HerdrControl(Endpoint.remote("ssh://user@host:2222", "named", "server", transport), "/unused", "fake-ssh") + control.source() + command = run.call_args.args[0] + self.assertEqual(command[0], "fake-ssh") + self.assertIn("2222", command) + self.assertIn("host", command) + script = command[-1] + if transport == "ssh": + script = base64.b64decode(script.split()[-1]).decode("utf-16le") + self.assertIn("--session named api snapshot", script) + self.assertIn("HERDR_SOCKET_PATH", script) + + def test_windows_missing_path_uses_read_only_fallback_before_mutation(self): + missing = subprocess.CompletedProcess([], 1, b"", b"'herdr' is not recognized") + listing = {"sessions": [{"name": "named", "running": True, + "session_dir": "C:\\Users\\me\\sessions"}]} + with patch("herdr_hub.control.subprocess.run", side_effect=[missing, completed(listing), completed(snapshot())]) as run: + HerdrControl(Endpoint.remote("host", "named", remote_transport="ssh"), "unused").source() + script = base64.b64decode(run.call_args_list[1].args[0][-1].split()[-1]).decode("utf-16le") + self.assertIn("session list --json", script) + self.assertIn(".local", script) + self.assertEqual(len(run.call_args_list), 3) + + def test_success_creates_unfocused_tab_starts_then_focuses(self): + for kind, readiness in (("codex", {"interactive_ready": True}), + ("claude", {})): + control = HerdrControl(Endpoint.local(), "fake") + control.call = Mock(side_effect=[ + {"tab": {"tab_id": "t2"}, "root_pane": {"pane_id": "w1:p2", "workspace_id": "w1", "tab_id": "t2"}}, + {"agent": {"pane_id": "w1:p2", "agent": kind, **readiness}}, + {}, + ]) + source = parse_snapshot(snapshot(kind)) + self.assertEqual(control.fork(source), "t2") + calls = [call.args[0] for call in control.call.call_args_list] + self.assertEqual(calls[0], ["tab", "create", "--workspace", "w1", "--cwd", source.cwd, "--no-focus"]) + self.assertEqual(calls[1][3:], ["--kind", kind, "--pane", "w1:p2", "--", *source.arguments]) + self.assertRegex(calls[1][2], rf"^{kind}-fork-[a-f0-9]+$") + self.assertEqual(calls[2], ["tab", "focus", "t2"]) + + def test_failure_never_closes_original_or_created_tab(self): + for failure in (ControlError("agent_not_ready: blocked"), ControlError("timeout"), + {"agent": {"interactive_ready": False}}, {"bad": "response"}): + with self.subTest(failure=failure): + control = HerdrControl(Endpoint.local(), "fake") + control.call = Mock(side_effect=[ + {"tab": {"tab_id": "t2"}, "root_pane": {"pane_id": "w1:p2", "workspace_id": "w1", "tab_id": "t2"}}, failure]) + with self.assertRaisesRegex(ControlError, "新增 Tab t2 未自动关闭"): + control.fork(parse_snapshot(snapshot())) + self.assertEqual(len(control.call.call_args_list), 2) + self.assertFalse(any("close" in c.args[0] or "focus" in c.args[0] for c in control.call.call_args_list)) + + def test_wrong_created_pane_never_starts_in_original_tab(self): + for tab_id, pane_id, workspace in (("t1", "w1:p2", "w1"), + ("t2", "w1:p1", "w1"), + ("t2", "w2:p1", "w2")): + control = HerdrControl(Endpoint.local(), "fake") + control.call = Mock(return_value={"tab": {"tab_id": tab_id}, + "root_pane": {"pane_id": pane_id, "tab_id": tab_id, "workspace_id": workspace}}) + with self.assertRaises(ControlError): + control.fork(parse_snapshot(snapshot())) + control.call.assert_called_once() + + def test_remote_mutation_failure_is_never_replayed(self): + control = HerdrControl(Endpoint.remote("host", "named"), "unused") + control.platform = "posix" + with patch("herdr_hub.control.subprocess.run", return_value=completed({"error": "failed"}, 127)) as run: + with self.assertRaises(ControlError): + control.call(["tab", "create", "--no-focus"]) + run.assert_called_once() + + def test_ambiguous_create_failure_is_not_replayed_or_cleaned_up(self): + control = HerdrControl(Endpoint.local(), "fake") + control.call = Mock(side_effect=ControlError("timeout")) + with self.assertRaisesRegex(ControlError, "原 Tab 保留"): + control.fork(parse_snapshot(snapshot())) + control.call.assert_called_once() + + def test_timeout_and_invalid_json_are_diagnostics(self): + for result in (subprocess.TimeoutExpired(["fake"], 10), subprocess.CompletedProcess([], 0, b"not json", b"")): + with patch("herdr_hub.control.subprocess.run", side_effect=[result]): + with self.assertRaises(ControlError): + HerdrControl(Endpoint.local(), "fake").source() + + def test_read_only_request_can_be_cancelled_quickly(self): + control = HerdrControl(Endpoint.local(), "fake") + cancelled = threading.Event() + timer = threading.Timer(0.05, cancelled.set) + timer.start() + started = time.monotonic() + try: + with self.assertRaises(ControlCancelled): + control._execute(["sleep", "5"], 5, cancelled) # type: ignore[attr-defined] + finally: + timer.cancel() + self.assertLess(time.monotonic() - started, 1) + + +class RemoteCommandTests(unittest.TestCase): + def test_posix_quoting_executes_exact_argv_with_path_fallback(self): + # Shell injection is checked by running only a temporary fake executable. + import tempfile + from pathlib import Path + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + executable = root / ".local/bin/herdr" + executable.parent.mkdir(parents=True) + executable.write_text("#!/usr/bin/env python3\nimport json,sys\nprint(json.dumps(sys.argv[1:]))\n") + executable.chmod(0o755) + arguments = ["--session", "named", "tab", "create", "--cwd", "a b'\";$HOME`id`$(id)\nnext"] + command = remote_control_command("ssh", RemoteEntry("host"), arguments, platform="posix") + result = subprocess.run(["sh", "-c", command[-1]], env={**os.environ, "HOME": directory}, capture_output=True, text=True, check=True) + self.assertEqual(json.loads(result.stdout), arguments) + + def test_windows_encoded_script_keeps_untrusted_text_out_of_shell(self): + value = "C:\\a b\\quote'\"; $(whoami) & %PATH%" + command = remote_control_command("ssh", RemoteEntry("host", remote_transport="ssh"), ["tab", "create", "--cwd", value], platform="windows") + self.assertNotIn(value, command[-1]) + script = base64.b64decode(command[-1].split()[-1]).decode("utf-16le") + self.assertIn("UseShellExecute=$false", script) + self.assertIn(subprocess.list2cmdline(["tab", "create", "--cwd", value]).replace("'", "''"), script) + + +class ImmediateFuture: + def __init__(self, fn, *args): + try: + self.value, self.error = fn(*args), None + except Exception as error: + self.value, self.error = None, error + def done(self): + return True + def result(self): + if self.error: + raise self.error + return self.value + + +class ImmediateExecutor: + def __init__(self, **kwargs): + pass + def __enter__(self): + return self + def __exit__(self, *args): + pass + def submit(self, fn, *args): + return ImmediateFuture(fn, *args) + + +class MenuTests(unittest.TestCase): + def setUp(self): + self.control = Mock(endpoint=Endpoint.local()) + self.control.source.return_value = parse_snapshot(snapshot()) + self.control.fork.return_value = "t2" + self.screen = Mock() + self.screen.getmaxyx.return_value = (24, 90) + + def run_menu(self, keys): + self.screen.getch.side_effect = keys + with patch("herdr_hub.control.ThreadPoolExecutor", ImmediateExecutor), patch("curses.set_escdelay"), patch("curses.curs_set"), patch("curses.mousemask"): + return ControlMenu(self.control).run(self.screen) + + def test_enter_executes_and_esc_does_not(self): + self.assertEqual(self.run_menu([27]), 0) + self.control.fork.assert_not_called() + self.assertEqual(self.run_menu([10]), 0) + self.control.fork.assert_called_once() + + def test_click_executes_action(self): + with patch("curses.getmouse", return_value=(0, 3, 3, 0, curses.BUTTON1_PRESSED)): + self.run_menu([curses.KEY_MOUSE]) + self.control.fork.assert_called_once() + + def test_error_is_visible_and_enter_cannot_repeat_failed_fork(self): + self.control.fork.side_effect = ControlError("blocked: inspect t2") + self.run_menu([10, 10, 27]) + self.control.fork.assert_called_once() + self.assertTrue(any("blocked: inspect t2" in str(call) for call in self.screen.addstr.call_args_list)) + + def test_snapshot_error_does_not_execute(self): + self.control.source.side_effect = ControlError("没有 Agent") + self.run_menu([10, 27]) + self.control.fork.assert_not_called() + + +class CliControlTests(unittest.TestCase): + def test_active_window_metadata_selects_control_endpoint(self): + args = build_parser().parse_args(["control"]) + hub = Mock(ssh_bin="fake-ssh") + endpoint = Endpoint.remote("host", "named", "server", "ssh") + hub.managed_windows.return_value = [ + ManagedWindow("@17", endpoint.key, "Remote", False, True, endpoint), + ManagedWindow( + "@18", "local:default", "Other", False, False, Endpoint.local() + ), + ] + with patch("herdr_hub.cli.make_hub", return_value=hub), patch("herdr_hub.cli.HerdrControl") as controller, patch("herdr_hub.cli.curses.wrapper", return_value=0): + self.assertEqual(command_control(args), 0) + controller.assert_called_once_with(endpoint, "herdr", "fake-ssh") + + def test_explicit_window_remains_available_for_cli_use(self): + args = build_parser().parse_args(["control", "--window", "@17"]) + hub = Mock(ssh_bin="fake-ssh") + endpoint = Endpoint.local() + hub.managed_windows.return_value = [ + ManagedWindow("@17", endpoint.key, "Local", False, False, endpoint) + ] + with patch("herdr_hub.cli.make_hub", return_value=hub), patch( + "herdr_hub.cli.HerdrControl" + ) as controller, patch("herdr_hub.cli.curses.wrapper", return_value=0): + self.assertEqual(command_control(args), 0) + controller.assert_called_once() + + def test_unmanaged_window_rejected(self): + hub = Mock() + hub.managed_windows.return_value = [] + with patch("herdr_hub.cli.make_hub", return_value=hub), self.assertRaises(HubError): + command_control(argparse.Namespace(window="@404")) diff --git a/tests/test_tmux_integration.py b/tests/test_tmux_integration.py index 05d3d85..acbab10 100644 --- a/tests/test_tmux_integration.py +++ b/tests/test_tmux_integration.py @@ -59,6 +59,21 @@ class TmuxIntegrationTests(unittest.TestCase): pass self.socket_path.unlink(missing_ok=True) + def test_control_uses_existing_prefix_and_active_window(self) -> None: + self.hub.ensure_server() + binding = self.hub.run(["list-keys", "-T", "prefix", "h"]).stdout + self.assertIn("display-popup", binding) + self.assertIn("control", binding) + self.assertNotIn("--window", binding) + self.assertNotIn("#{window_id}", binding) + old_binding = self.hub.run( + ["list-keys", "-T", "root", "C-g"], check=False + ) + self.assertNotEqual(old_binding.returncode, 0) + self.assertEqual(self.hub.run(["show-options", "-gv", "prefix"]).stdout.strip(), "C-a") + self.assertEqual(self.hub.run(["show-options", "-gv", "prefix2"]).stdout.strip(), "None") + self.assertIn("choose", self.hub.run(["list-keys", "-T", "prefix", "g"]).stdout) + def test_configures_isolated_server_and_adds_without_duplicates(self) -> None: self.assertTrue(self.hub.ensure_server()) endpoints = [Endpoint.local(), Endpoint.remote("orangepi")]