Add Herdr update and live handoff menu actions

The Ctrl+A h command menu gains two host-scoped actions: a plain
`herdr update` that streams updater output into the popup, and an
experimental update-and-hot-restart flow that verifies a live server
handoff and reconnects the originating Hub tab once the popup closes.

Remote updates use the remote host's own proxy settings. A standalone
helper reads literal proxy assignments from shell profiles and running
Clash/Mihomo listeners without executing startup files; Windows reads
process, user and machine proxy variables through PowerShell.

tests/test_control.py

@@ -6,6 +6,9 @@ import curses
 import json
 import os
 import subprocess
+import sys
+import tempfile
+from pathlib import Path
 import threading
 import time
 import unittest
@@ -18,6 +21,7 @@ from herdr_hub.control import (
     ControlMenu,
     HerdrControl,
     parse_snapshot,
+    update_failure,
 )
 from herdr_hub.remotes import RemoteEntry, remote_control_command
 
@@ -41,6 +45,13 @@ def completed(payload, code=0):
 
 
 class SnapshotTests(unittest.TestCase):
+    def test_missing_session_reports_identity_problem_not_version(self):
+        payload = snapshot()
+        del payload["result"]["snapshot"]["panes"][0]["agent_session"]
+        with self.assertRaisesRegex(ControlError, "尚未向 Herdr 上报会话 ID") as caught:
+            parse_snapshot(payload)
+        self.assertNotIn("需要 Herdr 0.9", str(caught.exception))
+
     def test_focused_pane_and_exact_native_arguments(self):
         for kind, args in (("codex", ["fork", SESSION_ID]),
                            ("claude", ["--resume", SESSION_ID, "--fork-session"])):
@@ -68,6 +79,99 @@ class SnapshotTests(unittest.TestCase):
 
 
 class ControlTests(unittest.TestCase):
+    def test_windows_hot_restart_rejected_before_update(self):
+        control = HerdrControl(Endpoint.remote("windows-host", remote_transport="ssh"), "unused")
+        control.platform = "windows"
+        with patch.object(control, "_execute") as execute:
+            with self.assertRaisesRegex(ControlError, "Windows 不支持热切换"):
+                control.update_and_restart(lambda _: None)
+        execute.assert_not_called()
+
+    def test_remote_hot_restart_runs_helper_on_exact_host(self):
+        control = HerdrControl(Endpoint.remote("ssh://me@host:2222", "named"), "unused", "fake-ssh")
+        control.platform = "posix"
+        with patch.object(control, "_execute", return_value=subprocess.CompletedProcess([], 0, b"ok", b"")) as execute:
+            control.update_and_restart(lambda _: None)
+        command = execute.call_args.args[0]
+        self.assertEqual(command[0], "fake-ssh")
+        self.assertIn("host", command)
+        self.assertIn("2222", command)
+        self.assertIn("exec python3 -u -c", command[-1])
+        self.assertIn(' "$hub_herdr" named', command[-1])
+        self.assertIn("unset HERDR_ENV;", command[-1])
+
+    def test_streaming_shows_unterminated_output_before_process_exits(self):
+        with tempfile.TemporaryDirectory() as directory:
+            marker = Path(directory) / "seen"
+            script = (
+                "import os, sys, time\n"
+                "from pathlib import Path\n"
+                "os.write(2, b'downloading 10%')\n"
+                "deadline = time.monotonic() + 3\n"
+                "while not Path(sys.argv[1]).exists():\n"
+                " if time.monotonic() > deadline: sys.exit(9)\n"
+                " time.sleep(.01)\n"
+                "print(' done')\n"
+            )
+            updates = []
+            def progress(value):
+                updates.append(value)
+                marker.touch()
+            result = HerdrControl(Endpoint.local(), "unused")._execute(
+                [sys.executable, "-c", script, str(marker)], 5, progress=progress
+            )
+        self.assertEqual(result.returncode, 0)
+        self.assertIn("downloading 10%", updates[0])
+        self.assertIn(b"done", result.stdout)
+
+    def test_streaming_timeout_preserves_diagnostic(self):
+        with self.assertRaisesRegex(ControlError, "waiting for network"):
+            HerdrControl(Endpoint.local(), "unused")._execute(
+                [sys.executable, "-c", "import os,time; os.write(2,b'waiting for network'); time.sleep(10)"],
+                .3, progress=lambda _: None,
+            )
+
+    def test_update_local_is_host_scoped_and_accepts_plain_text(self):
+        result = subprocess.CompletedProcess([], 0, b"Already up to date", b"notice")
+        with patch.dict(os.environ, {"HERDR_SESSION": "wrong"}), patch(
+            "herdr_hub.control.subprocess.run", return_value=result
+        ) as run:
+            output = HerdrControl(Endpoint.local("named"), "/fake/herdr").update()
+        self.assertIn("Already up to date", output)
+        self.assertIn("notice", output)
+        self.assertEqual(run.call_args.args[0], ["/fake/herdr", "update"])
+        self.assertEqual(run.call_args.kwargs["timeout"], 300)
+        self.assertEqual(run.call_args.kwargs["stdin"], subprocess.DEVNULL)
+        self.assertNotIn("HERDR_SESSION", run.call_args.kwargs["env"])
+
+    def test_update_remote_routes_once_on_both_platforms(self):
+        for transport, path in (("native", "/home/me/sessions"),
+                                ("ssh", "C:\\Users\\me\\sessions")):
+            listing = {"sessions": [{"name": "named", "running": True, "session_dir": path}]}
+            failure = subprocess.CompletedProcess([], 1, b"download started", b"network error")
+            with self.subTest(transport=transport), patch(
+                "herdr_hub.control.subprocess.run", side_effect=[completed(listing), failure]
+            ) as run:
+                control = HerdrControl(Endpoint.remote("host", "named", remote_transport=transport), "unused", "fake-ssh")
+                with self.assertRaisesRegex(ControlError, "network error"):
+                    control.update()
+                self.assertEqual(run.call_count, 2)
+                command = run.call_args.args[0]
+                self.assertEqual(command[0], "fake-ssh")
+                self.assertIn("host", command)
+                script = command[-1]
+                if transport == "ssh":
+                    script = base64.b64decode(script.split()[-1]).decode("utf-16le")
+                self.assertIn("update", script)
+                self.assertNotIn("--session", script)
+                self.assertNotIn("--handoff", script)
+
+    def test_update_timeout_is_not_replayed(self):
+        with patch("herdr_hub.control.subprocess.run", side_effect=subprocess.TimeoutExpired("herdr", 300)) as run:
+            with self.assertRaisesRegex(ControlError, "操作可能已完成"):
+                HerdrControl(Endpoint.local(), "herdr").update()
+        run.assert_called_once()
+
     def test_local_explicit_session_and_clean_environment(self):
         for session in ("default", "named"):
             with self.subTest(session=session), patch.dict(os.environ, {
@@ -234,6 +338,35 @@ class ImmediateExecutor:
 
 
 class MenuTests(unittest.TestCase):
+    def test_small_popup_shows_download_failure_before_proxy_notice(self):
+        self.screen.getmaxyx.return_value = (8, 71)
+        self.control.update_and_restart.side_effect = ControlError(update_failure(
+            "更新失败", "未检测到代理\n正在下载\ncurl: (28) Connection timed out\n没有自动执行普通重启；请检查当前 Session。"
+        ))
+        self.run_menu([curses.KEY_DOWN, curses.KEY_DOWN, 10, 27])
+        rendered = str(self.screen.addstr.call_args_list)
+        self.assertIn("curl: (28) Connection timed out", rendered)
+        self.assertNotIn("未检测到代理", rendered)
+
+    def test_hot_restart_success_requests_reconnect(self):
+        self.control.update_and_restart.return_value = "verified"
+        self.screen.getch.side_effect = [curses.KEY_DOWN, curses.KEY_DOWN, 10]
+        menu = ControlMenu(self.control)
+        with patch("herdr_hub.control.ThreadPoolExecutor", ImmediateExecutor), patch("curses.set_escdelay"), patch("curses.curs_set"), patch("curses.mousemask"):
+            self.assertEqual(menu.run(self.screen), 0)
+        self.assertTrue(menu.reconnect_requested)
+        self.control.update_and_restart.assert_called_once()
+
+    def test_hot_restart_failure_stays_visible_without_reconnect(self):
+        self.control.update_and_restart.side_effect = ControlError("handoff failed")
+        self.screen.getch.side_effect = [curses.KEY_DOWN, curses.KEY_DOWN, 10, 10, 27]
+        menu = ControlMenu(self.control)
+        with patch("herdr_hub.control.ThreadPoolExecutor", ImmediateExecutor), patch("curses.set_escdelay"), patch("curses.curs_set"), patch("curses.mousemask"):
+            self.assertEqual(menu.run(self.screen), 0)
+        self.assertFalse(menu.reconnect_requested)
+        self.control.update_and_restart.assert_called_once()
+        self.assertIn("handoff failed", str(self.screen.addstr.call_args_list))
+
     def setUp(self):
         self.control = Mock(endpoint=Endpoint.local())
         self.control.source.return_value = parse_snapshot(snapshot())
@@ -268,8 +401,102 @@ class MenuTests(unittest.TestCase):
         self.run_menu([10, 27])
         self.control.fork.assert_not_called()
 
+    def test_refresh_restores_fork_after_session_identity_becomes_available(self):
+        self.screen.getmaxyx.return_value = (8, 71)
+        self.control.source.side_effect = [ControlError("尚未上报会话 ID"), parse_snapshot(snapshot())]
+        self.run_menu([10, 18, 10])
+        self.assertEqual(self.control.source.call_count, 2)
+        self.control.fork.assert_called_once()
+        self.assertIn("不可用", str(self.screen.addstr.call_args_list))
+
+    def test_compact_mouse_fork_still_executes(self):
+        self.screen.getmaxyx.return_value = (8, 71)
+        with patch("curses.getmouse", return_value=(0, 3, 1, 0, curses.BUTTON1_PRESSED)):
+            self.run_menu([curses.KEY_MOUSE])
+        self.control.fork.assert_called_once()
+        self.control.update.assert_not_called()
+
+    def test_update_without_agent_keeps_result_and_prevents_repeat(self):
+        self.control.source.side_effect = ControlError("没有 Agent")
+        self.control.update.return_value = "更新成功"
+        self.run_menu([curses.KEY_DOWN, 10, 10, 27])
+        self.control.update.assert_called_once()
+        self.control.fork.assert_not_called()
+        rendered = str(self.screen.addstr.call_args_list)
+        self.assertIn("更新成功", rendered)
+        self.assertIn("更新目标：本机", rendered)
+
+    def test_click_update_reports_remote_target_and_failure(self):
+        self.control.endpoint = Endpoint.remote("my-host")
+        self.control.update.side_effect = ControlError("更新失败：连接断开")
+        with patch("curses.getmouse", return_value=(0, 3, 5, 0, curses.BUTTON1_PRESSED)):
+            self.run_menu([curses.KEY_MOUSE, 10, 27])
+        self.control.update.assert_called_once()
+        rendered = str(self.screen.addstr.call_args_list)
+        self.assertIn("更新目标：my-host", rendered)
+        self.assertIn("更新失败：连接断开", rendered)
+
+    def test_update_output_can_be_scrolled(self):
+        self.control.update.return_value = "\n".join(f"line-{i}" for i in range(60))
+        self.run_menu([curses.KEY_DOWN, 10, curses.KEY_NPAGE, curses.KEY_PPAGE, 27])
+        self.assertIn("line-20", str(self.screen.addstr.call_args_list))
+
+    def test_update_result_visible_in_actual_small_popup(self):
+        self.screen.getmaxyx.return_value = (8, 71)
+        self.control.update.return_value = "更新命令已完成\nAlready up to date"
+        self.run_menu([curses.KEY_DOWN, 10, 27])
+        calls = self.screen.addstr.call_args_list
+        results = [call.args for call in calls if "Already up to date" in call.args[2]]
+        self.assertTrue(results, "结果必须在真实的 8 行弹窗内绘制")
+        self.assertTrue(all(3 <= args[0] < 7 for args in results))
+        self.assertTrue(all(0 <= call.args[0] < 8 for call in calls))
+
+    def test_compact_mouse_update_row(self):
+        self.screen.getmaxyx.return_value = (8, 71)
+        self.control.update.return_value = "完成"
+        with patch("curses.getmouse", return_value=(0, 3, 2, 0, curses.BUTTON1_PRESSED)):
+            self.run_menu([curses.KEY_MOUSE, 27])
+        self.control.update.assert_called_once()
+        self.control.fork.assert_not_called()
+
+    def test_update_during_snapshot_load_is_not_ignored(self):
+        cancel = None
+        class LoadingFuture:
+            def done(self):
+                return cancel.is_set()
+            def result(self):
+                raise ControlCancelled()
+        class LoadingExecutor(ImmediateExecutor):
+            def submit(inner, fn, *args):
+                nonlocal cancel
+                if fn == self.control.source:
+                    cancel = args[0]
+                    return LoadingFuture()
+                return ImmediateFuture(fn, *args)
+        self.control.update.return_value = "完成"
+        self.screen.getch.side_effect = [curses.KEY_DOWN, 10, -1, 27]
+        with patch("herdr_hub.control.ThreadPoolExecutor", LoadingExecutor), patch("curses.set_escdelay"), patch("curses.curs_set"), patch("curses.mousemask"):
+            ControlMenu(self.control).run(self.screen)
+        self.control.update.assert_called_once()
+
 
 class CliControlTests(unittest.TestCase):
+    def test_hot_restart_reconnects_original_window_after_popup_closes(self):
+        args = build_parser().parse_args(["control"])
+        endpoint = Endpoint.local()
+        original = ManagedWindow("@17", endpoint.key, "Local", False, True, endpoint)
+        other = ManagedWindow("@18", "local:other", "Other", False, True, Endpoint.local("other"))
+        hub = Mock(ssh_bin="fake-ssh")
+        hub.managed_windows.side_effect = [[original], [original, other]]
+        def close_popup(run):
+            hub.reconnect.assert_not_called()
+            run.__self__.reconnect_requested = True
+            return 0
+        with patch("herdr_hub.cli.make_hub", return_value=hub), patch("herdr_hub.cli.curses.wrapper", side_effect=close_popup):
+            self.assertEqual(command_control(args), 0)
+        hub.reconnect.assert_called_once_with("@17")
+        hub.wait_for_window_startup.assert_called_once_with("@17")
+
     def test_active_window_metadata_selects_control_endpoint(self):
         args = build_parser().parse_args(["control"])
         hub = Mock(ssh_bin="fake-ssh")

tests/test_proxy_env.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+import base64
+import os
+from pathlib import Path
+import shlex
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from herdr_hub.cli import Endpoint
+from herdr_hub.control import HerdrControl
+from herdr_hub.proxy_env import configure_application_proxy, configure_proxy, proxy_assignments, running_proxy_paths
+
+
+class RemoteProxyTests(unittest.TestCase):
+    def test_running_mihomo_custom_config_is_discovered(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            core = root / "1277"
+            core.mkdir()
+            (core / "cmdline").write_bytes(
+                b"/opt/mihomo-linux-arm64\0-d\0/home/orangepi/repo/clash\0-f\0/home/orangepi/repo/clash/config-systemd.yaml\0"
+            )
+            unrelated = root / "1278"
+            unrelated.mkdir()
+            (unrelated / "cmdline").write_bytes(b"python3\0-c\0mihomo\0-f\0/irrelevant.yaml\0")
+            self.assertEqual(running_proxy_paths(root), [Path("/home/orangepi/repo/clash/config-systemd.yaml")])
+
+    def test_running_clash_listener_used_when_shell_has_no_proxy(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "clash-verge.yaml"
+            path.write_text("mixed-port: 7897\nbind-address: '*'\nallow-lan: false\n")
+            environment = {"no_proxy": "localhost,internal"}
+            with patch("herdr_hub.proxy_env.socket.create_connection") as connect:
+                source = configure_application_proxy(environment, [path])
+            self.assertEqual(source, str(path))
+            self.assertEqual(environment["https_proxy"], "http://127.0.0.1:7897")
+            self.assertEqual(environment["no_proxy"], "localhost,internal")
+            connect.assert_called_once_with(("127.0.0.1", 7897), timeout=1)
+
+    def test_stopped_application_and_authenticated_proxy_not_adopted(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "config.yaml"
+            for contents in ("mixed-port: 7897\n", "mixed-port: 7897\nauthentication:\n  - user:secret\n"):
+                path.write_text(contents)
+                environment = {}
+                with patch("herdr_hub.proxy_env.socket.create_connection", side_effect=ConnectionRefusedError):
+                    self.assertIsNone(configure_application_proxy(environment, [path]))
+                self.assertEqual(environment, {})
+
+    def test_application_detection_respects_existing_proxy_and_disable(self):
+        for value in ("", "http://configured:8080"):
+            environment = {"HTTPS_PROXY": value}
+            with patch("herdr_hub.proxy_env.socket.create_connection") as connect:
+                self.assertIsNone(configure_application_proxy(environment, [Path("/unused")]))
+                connect.assert_not_called()
+            self.assertEqual(environment, {"HTTPS_PROXY": value})
+
+    def test_literal_exports_and_uppercase_are_read_without_shell_evaluation(self):
+        values = proxy_assignments('''
+export https_proxy="http://remote-user:secret@127.0.0.1:7890" # local to remote
+HTTP_PROXY=http://remote-host:8080
+export no_proxy='localhost,127.0.0.1'
+export all_proxy=$(dangerous_command)
+export all_proxy=http://other:8080; dangerous_command
+    export all_proxy=http://inside-function:8080
+''')
+        self.assertEqual(values, {
+            "https_proxy": "http://remote-user:secret@127.0.0.1:7890",
+            "http_proxy": "http://remote-host:8080",
+            "no_proxy": "localhost,127.0.0.1",
+        })
+
+    def test_remote_environment_wins_including_explicit_disable(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / ".bashrc"
+            path.write_text("export https_proxy=http://file:7890\nexport all_proxy=http://fallback:7890\n")
+            environment = {"HTTPS_PROXY": "http://remote-env:7890", "all_proxy": ""}
+            configure_proxy(environment, [path])
+        self.assertEqual(environment["https_proxy"], "http://remote-env:7890")
+        self.assertEqual(environment["all_proxy"], "")
+
+    def test_wrapper_loads_remote_bashrc_and_never_embeds_local_proxy(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            marker = root / "should-not-exist"
+            remote_proxy = "http://remote-user:remote-secret@127.0.0.1:7891"
+            (root / ".bashrc").write_text(
+                "case $- in *i*) ;; *) return;; esac\n"
+                + f"export https_proxy={shlex.quote(remote_proxy)}\n"
+                + f"touch {shlex.quote(str(marker))}\n"
+            )
+            controller = HerdrControl(Endpoint.remote("host"), "unused")
+            controller.platform = "posix"
+            child = shlex.join([sys.executable, "-c", "import os; assert os.environ['https_proxy'] == " + repr(remote_proxy)])
+            with patch.dict(os.environ, {"https_proxy": "http://local-secret@127.0.0.1:9999"}):
+                command = controller._update_proxy(["ssh", "host", child], None)
+            self.assertEqual(command[:-1], ["ssh", "host"])
+            self.assertNotIn("local-secret", command[-1])
+            environment = {k: v for k, v in os.environ.items() if k.lower() not in {
+                "http_proxy", "https_proxy", "all_proxy", "no_proxy",
+            }}
+            environment.update(HOME=directory, SHELL="/bin/bash")
+            result = subprocess.run(["sh", "-c", command[-1]], env=environment, capture_output=True, text=True, timeout=10)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            self.assertIn("使用远端自身代理", result.stdout)
+            self.assertNotIn("remote-secret", result.stdout + result.stderr)
+            self.assertFalse(marker.exists())
+
+    def test_windows_loads_only_remote_environment_scopes(self):
+        controller = HerdrControl(Endpoint.remote("windows-host"), "unused")
+        controller.platform = "windows"
+        with patch.dict(os.environ, {"https_proxy": "http://local-secret@127.0.0.1:9999"}):
+            command = controller._update_proxy(["ssh", "windows-host", "unused"], None)
+        script = base64.b64decode(command[-1].split()[-1]).decode("utf-16le")
+        for scope in ("Process", "User", "Machine"):
+            self.assertIn(f"GetEnvironmentVariable($key,'{scope}')", script)
+        self.assertNotIn("local-secret", script)
+        self.assertNotIn("-R", command)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_restart.py

@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import os
+from pathlib import Path
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+from unittest.mock import patch
+
+from herdr_hub.cli import Endpoint
+from herdr_hub.control import ControlError, HerdrControl
+from herdr_hub.restart import update_and_handoff
+
+
+class HandoffTests(unittest.TestCase):
+    def setUp(self):
+        identity = patch("herdr_hub.restart.binary_identity", return_value=(1, 1, 1, 1))
+        identity.start()
+        self.addCleanup(identity.stop)
+
+    def test_failed_update_never_sends_handoff(self):
+        with patch("herdr_hub.restart.subprocess.check_output", return_value="status: running\nsocket: /tmp/fake.sock\n"), patch(
+            "herdr_hub.restart.request", return_value={"capabilities": {"live_handoff": True}}
+        ) as call, patch("herdr_hub.restart.subprocess.run", side_effect=subprocess.CalledProcessError(1, "update")), contextlib.redirect_stdout(io.StringIO()):
+            with self.assertRaises(subprocess.CalledProcessError):
+                update_and_handoff("/fake/herdr", "named")
+        self.assertEqual([c.args[1] for c in call.call_args_list], ["ping"])
+
+    def test_native_082_update_uses_handoff_and_does_not_repeat_it(self):
+        old = {"version": "0.8.2", "capabilities": {"live_handoff": True}}
+        with patch("herdr_hub.restart.subprocess.check_output", return_value="status: running\nsocket: /tmp/fake.sock\n"), patch(
+            "herdr_hub.restart.request", side_effect=[old, {"version": "0.9.1"}]
+        ) as call, patch("herdr_hub.restart.subprocess.run") as run, patch(
+            "herdr_hub.restart.installed_version", return_value="0.9.1"
+        ), contextlib.redirect_stdout(io.StringIO()):
+            update_and_handoff("/fake/herdr", "named")
+        self.assertEqual(run.call_args.args[0], ["/fake/herdr", "--session", "named", "update", "--handoff"])
+        self.assertEqual([c.args[1] for c in call.call_args_list], ["ping", "ping"])
+
+    def test_native_handoff_failure_after_install_is_not_replayed(self):
+        old = {"version": "0.8.2", "capabilities": {"live_handoff": True}}
+        with patch("herdr_hub.restart.subprocess.check_output", return_value="status: running\nsocket: /tmp/fake.sock\n"), patch(
+            "herdr_hub.restart.request", side_effect=[old, old]
+        ) as call, patch("herdr_hub.restart.subprocess.run"), patch(
+            "herdr_hub.restart.installed_version", return_value="0.9.1"
+        ), patch("herdr_hub.restart.binary_identity", side_effect=[(1, 1, 1, 1), (1, 2, 1, 1)]), contextlib.redirect_stdout(io.StringIO()):
+            with self.assertRaisesRegex(RuntimeError, "不会重复发送热切换"):
+                update_and_handoff("/fake/herdr", "named")
+        self.assertEqual([c.args[1] for c in call.call_args_list], ["ping", "ping"])
+
+    def test_unverified_handoff_is_failure_not_success(self):
+        old = {"version": "0.9.0", "capabilities": {"live_handoff": True}}
+        with patch("herdr_hub.restart.subprocess.check_output", return_value="status: running\nsocket: /tmp/fake.sock\n"), patch(
+            "herdr_hub.restart.request", side_effect=[old, old, {}, old]
+        ) as call, patch("herdr_hub.restart.subprocess.run"), patch(
+            "herdr_hub.restart.installed_version", return_value="0.9.1"
+        ), patch("herdr_hub.restart.time.monotonic", side_effect=[0, 21]), contextlib.redirect_stdout(io.StringIO()):
+            with self.assertRaisesRegex(RuntimeError, "未确认新服务器就绪"):
+                update_and_handoff("/fake/herdr", "named")
+        self.assertEqual([c.args[1] for c in call.call_args_list].count("server.live_handoff"), 1)
+
+    def test_real_helper_updates_old_server_even_when_binary_already_latest(self):
+        self._run_real_helper()
+
+    def test_real_helper_failure_does_not_stop_or_retry_server(self):
+        self._run_real_helper(fail_handoff=True)
+
+    def _run_real_helper(self, fail_handoff=False):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            socket_path = root / "api.sock"
+            record = root / "commands.jsonl"
+            binary = root / "herdr"
+            binary.write_text(
+                f"#!{sys.executable}\n"
+                "import json, os, sys\n"
+                f"with open({str(record)!r}, 'a') as f: f.write(json.dumps(sys.argv[1:])+'\\n')\n"
+                "if os.environ.get('HERDR_ENV') == '1': sys.exit(8)\n"
+                "if sys.argv[1:] == ['--version']: print('herdr 0.9.1')\n"
+                "elif sys.argv[1:] == ['--session', 'named', 'status', 'server']:\n"
+                f" print('status: running\\nsocket: {socket_path}')\n"
+                "elif sys.argv[1:] == ['--session', 'named', 'update', '--handoff']:\n"
+                " print('already up to date (0.9.1)', flush=True)\n"
+                "else: sys.exit(9)\n"
+            )
+            binary.chmod(0o755)
+            methods = []
+            done = threading.Event()
+            errors = []
+            server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            server.bind(str(socket_path))
+            server.listen()
+            server.settimeout(.1)
+            def serve():
+                version = "0.9.0"
+                try:
+                    while not done.is_set():
+                        try:
+                            connection, _ = server.accept()
+                        except socket.timeout:
+                            continue
+                        with connection, connection.makefile("rb") as stream:
+                            message = json.loads(stream.readline())
+                            methods.append(message)
+                            response = {"id": message["id"]}
+                            if message["method"] == "server.live_handoff":
+                                if fail_handoff:
+                                    response["error"] = {"message": "handoff refused"}
+                                else:
+                                    version = "0.9.1"
+                                    response["result"] = {}
+                            elif message["method"] == "ping":
+                                response["result"] = {"version": version, "capabilities": {"live_handoff": True}}
+                            else:
+                                raise AssertionError(message)
+                            connection.sendall((json.dumps(response) + "\n").encode())
+                except Exception as error:
+                    errors.append(error)
+            thread = threading.Thread(target=serve)
+            thread.start()
+            output = []
+            try:
+                with patch.dict(os.environ, {"HERDR_ENV": "1", "HERDR_SESSION": "wrong"}):
+                    control = HerdrControl(Endpoint.local("named"), str(binary))
+                    if fail_handoff:
+                        with self.assertRaisesRegex(ControlError, "handoff refused"):
+                            control.update_and_restart(output.append)
+                    else:
+                        result = control.update_and_restart(output.append)
+                        self.assertIn("服务器已确认使用 0.9.1", result)
+            finally:
+                done.set()
+                thread.join(3)
+                server.close()
+            self.assertFalse(errors)
+            handoffs = [m for m in methods if m["method"] == "server.live_handoff"]
+            self.assertEqual(len(handoffs), 1)
+            self.assertEqual(handoffs[0]["params"]["import_exe"], str(binary.resolve()))
+            self.assertEqual(handoffs[0]["params"]["expected_version"], "0.9.1")
+            commands = [json.loads(line) for line in record.read_text().splitlines()]
+            self.assertEqual(commands.count(["--session", "named", "update", "--handoff"]), 1)
+            self.assertFalse(any("stop" in command for command in commands))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_tmux_integration.py

@@ -8,6 +8,7 @@ import tempfile
 import time
 import unittest
 import uuid
+from unittest.mock import patch
 
 from herdr_hub.cli import (
     BOOTSTRAP_WINDOW_NAME,
@@ -15,11 +16,32 @@ from herdr_hub.cli import (
     HubError,
     TmuxHub,
     WINDOW_SHORT_LABEL_OPTION,
+    build_parser,
+    command_control,
 )
 
 
 @unittest.skipUnless(shutil.which("tmux"), "tmux is required")
 class TmuxIntegrationTests(unittest.TestCase):
+    def test_verified_handoff_reconnects_only_originating_client(self) -> None:
+        self.hub.ensure_server()
+        original = self.hub.create_window(Endpoint.local("first"), str(self.fake_herdr))
+        other = self.hub.create_window(Endpoint.local("second"), str(self.fake_herdr))
+        self.hub.select_window(original.window_id)
+        def pid(window):
+            return self.hub.run(["display-message", "-p", "-t", window.window_id, "#{pane_pid}"]).stdout.strip()
+        original_pid, other_pid = pid(original), pid(other)
+        def close_popup(run):
+            self.hub.select_window(other.window_id)
+            run.__self__.reconnect_requested = True
+            return 0
+        with patch("herdr_hub.cli.make_hub", return_value=self.hub), patch(
+            "herdr_hub.cli.resolve_herdr", return_value=str(self.fake_herdr)
+        ), patch("herdr_hub.cli.curses.wrapper", side_effect=close_popup):
+            self.assertEqual(command_control(build_parser().parse_args(["control"])), 0)
+        self.assertNotEqual(pid(original), original_pid)
+        self.assertEqual(pid(other), other_pid)
+
     def setUp(self) -> None:
         self.temporary_directory = tempfile.TemporaryDirectory()
         self.addCleanup(self.temporary_directory.cleanup)