Keep session manager open on launch failure

README.md

@@ -151,6 +151,19 @@ Session 会在远端运行 `herdr --session NAME`。这种模式仍支持 Tab
 提供的本地图片剪贴板桥接。空 Windows 主机或平台检测失败时，可在“添加远端”中选择
 “Windows / SSH”，也可手动执行 `herdr-hub open --remote TARGET --transport ssh`。
 
+rootless 容器或 UID 映射环境可能把 `/etc/ssh/ssh_config` 显示为 `nobody` 所有，导致原生
+`herdr --remote` 报 `Bad owner or permissions`，即使系统 `ssh` 本身仍可使用。不要放宽该
+文件的写权限；可在本机 `~/.config/herdr/config.toml` 中让 Herdr 直接使用系统 SSH：
+
+```toml
+[remote]
+manage_ssh_config = false
+```
+
+这会停用 Herdr 自动生成的 keepalive 和连接复用配置，但不会禁用用户自己的 SSH 配置。
+执行 `herdr config check` 验证后重新打开 Session 即可。若 Session 进程启动后立即退出，
+Sessions 弹窗会保留在该 Session 的操作页，显示 pane 返回的错误并允许直接重试。
+
 也可通过 CLI 管理远端：
 
 ```bash

herdr_hub/cli.py

@@ -17,6 +17,7 @@ import subprocess
 import sys
 import tempfile
 import threading
+import time
 from typing import Callable, Iterable, Mapping, Sequence
 from urllib.parse import quote
 
@@ -986,21 +987,72 @@ class TmuxHub:
         return self.window_for_key(endpoint.key) or replaced
 
     @serialized_hub_mutation
-    def activate_endpoint(self, endpoint: Endpoint, herdr_bin: str) -> ManagedWindow:
+    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_any_tmux():
             if inside_socket(self.socket_name):
@@ -1292,6 +1344,46 @@ def command_choose(arguments: argparse.Namespace) -> int:
             ]
         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),
@@ -1299,6 +1391,7 @@ def command_choose(arguments: argparse.Namespace) -> int:
             ssh_bin,
             local_sessions=(endpoint.session for endpoint in local_report.endpoints),
             aliases=hub.aliases,
+            activate_handler=activate_selection,
             rename_handler=rename_session,
             close_window_handler=close_window,
             delete_session_handler=delete_session,
@@ -1326,34 +1419,7 @@ def command_choose(arguments: argparse.Namespace) -> int:
         assert result.window_id is not None
         hub.dismiss_window(result.window_id)
         return 0
-    if result.kind == "window":
-        assert result.window_id is not None
-        if result.dead:
-            hub.reconnect(result.window_id)
-        hub.select_window(result.window_id)
-        return 0
-    if result.kind == "local":
-        assert result.session is not None
-        hub.activate_endpoint(Endpoint.local(result.session), resolve_herdr())
-        return 0
-    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())
+    activate_selection(result)
     return 0
 
 

herdr_hub/selector.py

@@ -114,6 +114,7 @@ class StopSessionOutcome:
 DeleteSessionHandler = Callable[[str, str, str | None], DeleteSessionOutcome | None]
 StopSessionHandler = Callable[[str, str, str], StopSessionOutcome | list[CatalogWindow]]
 CloseWindowHandler = Callable[[str], Iterable[CatalogWindow]]
+ActivateHandler = Callable[["SelectorResult"], None]
 
 
 @dataclass(frozen=True)
@@ -523,10 +524,12 @@ class SessionSelector:
         close_window_handler: CloseWindowHandler | None = None,
         delete_session_handler: DeleteSessionHandler | None = None,
         stop_session_handler: StopSessionHandler | None = None,
+        activate_handler: ActivateHandler | None = None,
     ) -> None:
         self.windows = list(windows)
         self.local_sessions = list(local_sessions)
         self.aliases = dict(aliases or {})
+        self.activate_handler = activate_handler
         self.rename_handler = rename_handler
         self.close_window_handler = close_window_handler
         self.delete_session_handler = delete_session_handler
@@ -976,7 +979,10 @@ class SessionSelector:
             return None
         if row.kind in {"window", "action-window"}:
             assert row.window_id is not None
-            return SelectorResult.window(row.window_id, dead=row.dead)
+            return self._activate_result(
+                SelectorResult.window(row.window_id, dead=row.dead),
+                row,
+            )
         if row.kind == "action-close":
             assert row.window_id is not None
             if self.close_window_handler is None:
@@ -985,14 +991,17 @@ class SessionSelector:
             return None
         if row.kind in {"local", "action-local"}:
             assert row.session is not None
-            return SelectorResult.local(row.session)
+            return self._activate_result(SelectorResult.local(row.session), row)
         if row.kind in {"remote", "action-remote"}:
             assert row.target is not None and row.session is not None
-            return SelectorResult.remote(
-                row.target,
-                row.session,
-                row.remote_keybindings,
-                row.remote_transport,
+            return self._activate_result(
+                SelectorResult.remote(
+                    row.target,
+                    row.session,
+                    row.remote_keybindings,
+                    row.remote_transport,
+                ),
+                row,
             )
         if row.kind in {"new-session", "add-remote", "remote-empty"}:
             self._show_actions(row)
@@ -1094,13 +1103,36 @@ class SessionSelector:
         target = self.input_target
         assert target is not None
         entry = self.entries[target]
-        return SelectorResult.remote(
-            target,
-            session,
-            entry.remote_keybindings,
-            self.input_remote_transport,
+        return self._activate_result(
+            SelectorResult.remote(
+                target,
+                session,
+                entry.remote_keybindings,
+                self.input_remote_transport,
+            )
         )
 
+    def _activate_result(
+        self,
+        result: SelectorResult,
+        row: PickerRow | None = None,
+    ):
+        if self.activate_handler is None:
+            return result
+        try:
+            self.activate_handler(result)
+        except (OSError, RuntimeError, ValueError) as error:
+            detail = str(error).strip() or "未知错误"
+            self.message = f"打开失败：{detail}"
+            if row is not None:
+                source = self.context_row or row
+                self.context_row = replace(source, detail="打开失败", error=detail)
+                self.view = "actions"
+                self.selected_key = f"action:open:{source.key}"
+                self.scroll = 0
+            return None
+        return False
+
     def _finish_rename(self, alias: str):
         try:
             normalized = normalize_alias(alias)

tests/test_selector.py

@@ -316,6 +316,55 @@ class SelectorCatalogTests(unittest.TestCase):
 
 
 class SelectorActionTests(unittest.TestCase):
+    def test_open_failure_stays_in_context_and_retry_can_succeed(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            attempts: list[SelectorResult] = []
+
+            def activate(result: SelectorResult) -> None:
+                attempts.append(result)
+                if len(attempts) == 1:
+                    raise RuntimeError(
+                        "Bad owner or permissions on /etc/ssh/ssh_config"
+                    )
+
+            selector = SessionSelector(
+                [
+                    CatalogWindow(
+                        "@1",
+                        "remote:pi:default",
+                        "pi",
+                        True,
+                        False,
+                        kind="remote",
+                        target="pi",
+                        session="default",
+                    )
+                ],
+                RemoteRegistry(Path(directory) / "remotes.json"),
+                "/missing/ssh",
+                activate_handler=activate,
+            )
+            self.addCleanup(selector.pool.close)
+            row = next(
+                item
+                for item in selector._rows()  # type: ignore[attr-defined]
+                if item.key == "window:@1"
+            )
+
+            self.assertIsNone(selector._activate(row))  # type: ignore[attr-defined]
+            self.assertEqual(selector.view, "actions")
+            self.assertIn("打开失败", selector.message)
+            self.assertIn("ssh_config", selector.context_row.error)
+            retry = next(
+                item
+                for item in selector._rows()  # type: ignore[attr-defined]
+                if item.kind == "action-window"
+            )
+            self.assertEqual(selector.selected_key, retry.key)
+
+            self.assertFalse(selector._activate(retry))  # type: ignore[attr-defined]
+            self.assertEqual(len(attempts), 2)
+
     def test_search_is_explicit_and_escape_preserves_selection(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             window = CatalogWindow(

tests/test_tmux_integration.py

@@ -28,7 +28,8 @@ class TmuxIntegrationTests(unittest.TestCase):
         self.fake_herdr.write_text(
             "#!/bin/sh\n"
             "if [ \"${1:-}\" = \"--session\" ] && [ \"${2:-}\" = \"dead\" ]; then\n"
-            "  exit 0\n"
+            "  printf '%s\\n' 'fake startup failure' >&2\n"
+            "  exit 7\n"
             "fi\n"
             "exec sleep 30\n",
             encoding="utf-8",
@@ -219,6 +220,20 @@ class TmuxIntegrationTests(unittest.TestCase):
         self.hub.reconnect(window.window_id)
         self.assertIsNotNone(self.hub.window_for_key("local:dead"))
 
+    def test_startup_verification_reports_immediate_process_failure(self) -> None:
+        self.hub.ensure_server()
+        with self.assertRaisesRegex(HubError, "fake startup failure"):
+            self.hub.activate_endpoint(
+                Endpoint.local("dead"),
+                str(self.fake_herdr),
+                verify_startup=True,
+            )
+
+        failed = self.hub.window_for_key("local:dead")
+        self.assertIsNotNone(failed)
+        assert failed is not None
+        self.assertTrue(failed.dead)
+
     def test_alias_is_applied_immediately_and_can_be_reset(self) -> None:
         self.hub.aliases["remote:orangepi:work"] = "客厅开发"
         self.hub.ensure_server()