diff --git a/herdr_hub/cli.py b/herdr_hub/cli.py index e3b3e53..de2f5d0 100644 --- a/herdr_hub/cli.py +++ b/herdr_hub/cli.py @@ -178,6 +178,7 @@ class ManagedWindow: dead: bool active: bool endpoint: Endpoint | None = None + short_label: str = "" RunFunction = Callable[..., subprocess.CompletedProcess[str]] @@ -549,6 +550,34 @@ class TmuxHub: 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], @@ -634,8 +663,6 @@ class TmuxHub: "（退出状态 #{pane_dead_status}） · 按 Ctrl+A R 重连", ], ] - for command in option_commands: - self.run(command) refresh = shlex.join([self.python_bin, self.script_path, "refresh"]) choose = shlex.join([self.python_bin, self.script_path, "choose"]) @@ -646,13 +673,15 @@ class TmuxHub: 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]) - # 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) + 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], @@ -702,26 +731,29 @@ class TmuxHub: confirm_close, ], ] - for command in binding_commands: - self.run(command) - + self.run_batch([*option_commands, *cleanup_commands, *binding_commands]) self.refresh_labels() def managed_windows(self) -> list[ManagedWindow]: - if not self.exists(): - return [] 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}}}" + f"#{{{WINDOW_COMMAND_OPTION}}}\t#{{{WINDOW_SHORT_LABEL_OPTION}}}" ) result = self.run( - ["list-windows", "-t", self.session_name, "-F", line_format] + ["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", 6) - if len(fields) != 7 or not fields[1]: + fields = line.split("\t", 7) + if len(fields) != 8 or not fields[1]: continue windows.append( ManagedWindow( @@ -731,6 +763,7 @@ class TmuxHub: dead=fields[3] == "1", active=fields[4] == "1", endpoint=deserialize_endpoint(fields[5], fields[6]), + short_label=fields[7], ) ) return windows @@ -739,33 +772,38 @@ class TmuxHub: return next((window for window in self.managed_windows() if window.key == key), None) def refresh_labels(self) -> None: - for window in self.managed_windows(): - self._apply_window_label(window) + self.run_batch( + [ + command + for window in self.managed_windows() + for command in self._label_commands(window) + ] + ) - def _apply_window_label(self, window: ManagedWindow) -> None: + def _labels_for(self, window: ManagedWindow) -> tuple[str, str]: alias = self.aliases.get(window.key) - display_label = truncate_label(alias) if alias else window.name - short_label = truncate_label(alias, 32) if alias else truncate_label(window.name, 32) + if alias: + return truncate_label(alias), truncate_label(alias, 32) if window.endpoint is not None: - display_label = truncate_label(alias) if alias else window.endpoint.label - short_label = ( - truncate_label(alias, 32) if alias else window.endpoint.short_label - ) + 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. - self.run(["rename-window", "-t", window.window_id, "--", display_label]) - self.run( - [ - "set-option", - "-w", - "-t", - window.window_id, - WINDOW_SHORT_LABEL_OPTION, - short_label, - ] - ) + 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: @@ -832,8 +870,10 @@ class TmuxHub: existing = self.window_for_key(endpoint.key) if existing is not None: return existing - was_dismissed = endpoint.key in self.dismissed_keys() - self.restore_endpoint(endpoint.key) + 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 @@ -858,14 +898,20 @@ class TmuxHub: self.set_dismissed_keys({*self.dismissed_keys(), endpoint.key}) raise HubError("tmux did not return the new window id") try: - 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), - ]: - self.run(["set-option", "-w", "-t", window_id, option, value]) - self.run(["set-window-option", "-t", window_id, "remain-on-exit", "on"]) + 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: @@ -878,6 +924,7 @@ class TmuxHub: dead=False, active=False, endpoint=endpoint, + short_label=short_label, ) @serialized_hub_mutation @@ -985,20 +1032,31 @@ class TmuxHub: (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] = {} - for option, _ in replacements: - result = self.run( - ["show-options", "-w", "-v", "-t", window.window_id, option], - check=False, - ) - if result.returncode == 0: - previous[option] = result.stdout.rstrip("\n") + 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: - for option, value in replacements: - self.run( - ["set-option", "-w", "-t", window.window_id, option, value] - ) - self.run(["respawn-window", "-k", "-t", window.window_id, command]) + 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( diff --git a/tests/test_tmux_integration.py b/tests/test_tmux_integration.py index 5017b5e..10d090a 100644 --- a/tests/test_tmux_integration.py +++ b/tests/test_tmux_integration.py @@ -351,6 +351,59 @@ class TmuxIntegrationTests(unittest.TestCase): self.assertIn(str(self.fake_ssh), command) self.assertIn("-tt", command) + def test_reconfiguration_batches_tmux_calls_and_escapes_semicolons(self) -> None: + calls: list[list[str]] = [] + real_run = self.hub._run # type: ignore[attr-defined] + + def counting_run(command, **keywords): # type: ignore[no-untyped-def] + calls.append(list(command)) + return real_run(command, **keywords) + + self.hub._run = counting_run # type: ignore[attr-defined] + self.hub.ensure_server() + for name in ("a", "b", "c"): + self.hub.create_window(Endpoint.local(name), str(self.fake_herdr)) + self.hub.remove_bootstrap() + + # A no-op refresh (nothing new discovered, labels unchanged) used to + # spawn one tmux client per option and per window label. + calls.clear() + self.hub.ensure_server() + self.hub.add_missing([], str(self.fake_herdr)) + self.hub.managed_windows() + self.assertLessEqual(len(calls), 10, calls) + self.assertEqual(self.hub.run(["show-options", "-gv", "prefix"]).stdout.strip(), "C-a") + self.assertIn("display-popup", self.hub.run(["list-keys", "-T", "prefix", "g"]).stdout) + + # tmux treats any argument ending in ';' as a command separator. + self.hub.run_batch( + [ + ["set-option", "-g", "@herdr_hub_test_one", "value;"], + ["set-option", "-g", "@herdr_hub_test_two", "other"], + ] + ) + self.assertEqual( + self.hub.run(["show-options", "-gv", "@herdr_hub_test_one"]).stdout.strip(), + "value;", + ) + self.assertEqual( + self.hub.run(["show-options", "-gv", "@herdr_hub_test_two"]).stdout.strip(), + "other", + ) + # A failing command is attributed precisely instead of silently + # skipping the rest of the batch. + with self.assertRaisesRegex(HubError, "no-such-option"): + self.hub.run_batch( + [ + ["set-option", "-g", "@herdr_hub_test_three", "set"], + ["set-option", "-g", "no-such-option", "1"], + ] + ) + self.assertEqual( + self.hub.run(["show-options", "-gv", "@herdr_hub_test_three"]).stdout.strip(), + "set", + ) + def test_concurrent_open_of_the_same_endpoint_creates_one_window(self) -> None: self.hub.ensure_server() second_hub = TmuxHub(