ControlModeEngine(abstraction): Engine hooks for probe/switch

Clean up abstraction leaks by replacing isinstance(ControlModeEngine)
checks with hook methods in the base Engine class:

- Add probe_server_alive() hook: allows engines to check server liveness
  without bootstrapping (ControlModeEngine does direct subprocess probe)
- Add can_switch_client() hook: allows engines to indicate if switch-client
  is meaningful (ControlModeEngine checks for non-control clients)
- Update Server.is_alive() to use probe hook instead of isinstance
- Update Server.raise_if_dead() to use probe hook instead of isinstance
- Update Server.switch_client() to use hook directly (was via getattr)
- Fix thread join in close() to skip current thread during GC

This removes all isinstance(ControlModeEngine) checks from Server,
making the Engine abstraction clean and extensible.

Author: tony

src/libtmux/_internal/engines/base.py

@@ -161,6 +161,39 @@ def iter_notifications(
         return
 
     # Optional hooks ---------------------------------------------------
+    def probe_server_alive(
+        self,
+        server_args: tuple[str | int, ...],
+    ) -> bool | None:
+        """Probe if tmux server is alive without starting the engine.
+
+        Returns
+        -------
+        bool | None
+            True if server is alive, False if dead, None to use default check.
+            Return None to fall back to running ``list-sessions`` via the engine.
+
+        Notes
+        -----
+        Override in engines that shouldn't start on probe (e.g., ControlModeEngine).
+        """
+        return None
+
+    def can_switch_client(
+        self,
+        *,
+        server_args: tuple[str | int, ...] | None = None,
+    ) -> bool:
+        """Check if switch-client is meaningful for this engine.
+
+        Returns
+        -------
+        bool
+            True if there is at least one client that can be switched.
+            Default implementation returns True (assumes switching is allowed).
+        """
+        return True
+
     @property
     def internal_session_names(self) -> set[str]:
         """Names of sessions reserved for engine internals."""

src/libtmux/_internal/engines/control_mode.py

@@ -169,10 +169,20 @@ def close(self) -> None:
             self._server_args = None
             self._protocol.mark_dead("engine closed")
 
-            # Join threads to ensure clean shutdown (non-daemon threads)
-            if self._reader_thread is not None and self._reader_thread.is_alive():
+            # Join threads to ensure clean shutdown (non-daemon threads).
+            # Skip join if called from within the thread itself (e.g., during GC).
+            current = threading.current_thread()
+            if (
+                self._reader_thread is not None
+                and self._reader_thread.is_alive()
+                and self._reader_thread is not current
+            ):
                 self._reader_thread.join(timeout=2)
-            if self._stderr_thread is not None and self._stderr_thread.is_alive():
+            if (
+                self._stderr_thread is not None
+                and self._stderr_thread.is_alive()
+                and self._stderr_thread is not current
+            ):
                 self._stderr_thread.join(timeout=2)
 
     def __del__(self) -> None:  # pragma: no cover - best effort cleanup
@@ -312,6 +322,31 @@ def internal_session_names(self) -> set[str]:
             return set()
         return {self._internal_session_name}
 
+    def probe_server_alive(
+        self,
+        server_args: tuple[str | int, ...],
+    ) -> bool | None:
+        """Check if tmux server is alive without starting control mode.
+
+        Performs a direct subprocess check to avoid bootstrapping the control
+        mode connection just to probe server liveness.
+
+        Returns
+        -------
+        bool
+            True if server is alive (list-sessions returns 0), False otherwise.
+        """
+        tmux_bin = shutil.which("tmux")
+        if tmux_bin is None:
+            return False
+
+        result = subprocess.run(
+            [tmux_bin, *[str(a) for a in server_args], "list-sessions"],
+            check=False,
+            capture_output=True,
+        )
+        return result.returncode == 0
+
     def exclude_internal_sessions(
         self,
         sessions: list[Session],

src/libtmux/server.py

@@ -212,12 +212,14 @@ def is_alive(self) -> bool:
         >>> tmux = Server(socket_name="no_exist")
         >>> assert not tmux.is_alive()
         """
-        # Avoid spinning up control-mode just to probe.
-        from libtmux._internal.engines.control_mode import ControlModeEngine
+        server_args = tuple(self._build_server_args())
 
-        if isinstance(self.engine, ControlModeEngine):
-            return self._probe_server() == 0
+        # Use engine hook to allow engines to probe without bootstrapping.
+        probe_result = self.engine.probe_server_alive(server_args)
+        if probe_result is not None:
+            return probe_result
 
+        # Default: run list-sessions through the engine.
         try:
             res = self.cmd("list-sessions")
         except Exception:
@@ -234,23 +236,24 @@ def raise_if_dead(self) -> None:
         ...     print(type(e))
         <class 'subprocess.CalledProcessError'>
         """
-        from libtmux._internal.engines.control_mode import ControlModeEngine
+        server_args = tuple(self._build_server_args())
 
-        if isinstance(self.engine, ControlModeEngine):
-            rc = self._probe_server()
-            if rc != 0:
+        # Use engine hook to allow engines to probe without bootstrapping.
+        probe_result = self.engine.probe_server_alive(server_args)
+        if probe_result is not None:
+            if not probe_result:
                 tmux_bin_probe = shutil.which("tmux") or "tmux"
                 raise subprocess.CalledProcessError(
-                    returncode=rc,
-                    cmd=[tmux_bin_probe, *self._build_server_args(), "list-sessions"],
+                    returncode=1,
+                    cmd=[tmux_bin_probe, *server_args, "list-sessions"],
                 )
             return
 
+        # Default: run list-sessions through the engine.
         tmux_bin = shutil.which("tmux")
         if tmux_bin is None:
             raise exc.TmuxCommandNotFound
 
-        server_args = self._build_server_args()
         proc = self.engine.run("list-sessions", server_args=server_args)
         if proc.returncode is not None and proc.returncode != 0:
             raise subprocess.CalledProcessError(
@@ -483,10 +486,9 @@ def switch_client(self, target_session: str) -> None:
 
         server_args = tuple(self._build_server_args())
 
-        # If the engine knows there are no "real" clients, mirror tmux's
-        # `no current client` error before dispatching.
-        can_switch = getattr(self.engine, "can_switch_client", None)
-        if callable(can_switch) and not can_switch(server_args=server_args):
+        # Use engine hook to check if switch-client is meaningful.
+        # For control mode, this ensures there is at least one non-control client.
+        if not self.engine.can_switch_client(server_args=server_args):
             msg = "no current client"
             raise exc.LibTmuxException(msg)