Server(refactor[sessions]): Use engine internal filters

why: keep engine transparency without reaching into control-mode internals.
what:
- hide management sessions via engine.internal_session_names
- route attached_sessions through engine.exclude_internal_sessions hook
- preserve existing server arg handling and attach behaviour

Author: tony

src/libtmux/server.py

@@ -16,6 +16,7 @@
 import warnings
 
 from libtmux import exc, formats
+from libtmux._internal.engines.subprocess_engine import SubprocessEngine
 from libtmux._internal.query_list import QueryList
 from libtmux.common import tmux_cmd
 from libtmux.constants import OptionScope
@@ -40,6 +41,7 @@
 
     from typing_extensions import Self
 
+    from libtmux._internal.engines.base import Engine
     from libtmux._internal.types import StrPath
 
     DashLiteral: TypeAlias = t.Literal["-"]
@@ -78,17 +80,17 @@ class Server(
     >>> server
     Server(socket_name=libtmux_test...)
 
-    >>> server.sessions
-    [Session($1 ...)]
+    >>> server.sessions  # doctest: +ELLIPSIS
+    [Session($... ...)]
 
-    >>> server.sessions[0].windows
-    [Window(@1 1:..., Session($1 ...))]
+    >>> server.sessions[0].windows  # doctest: +ELLIPSIS
+    [Window(@... ..., Session($... ...))]
 
-    >>> server.sessions[0].active_window
-    Window(@1 1:..., Session($1 ...))
+    >>> server.sessions[0].active_window  # doctest: +ELLIPSIS
+    Window(@... ..., Session($... ...))
 
-    >>> server.sessions[0].active_pane
-    Pane(%1 Window(@1 1:..., Session($1 ...)))
+    >>> server.sessions[0].active_pane  # doctest: +ELLIPSIS
+    Pane(%... Window(@... ..., Session($... ...)))
 
     The server can be used as a context manager to ensure proper cleanup:
 
@@ -137,12 +139,17 @@ def __init__(
         colors: int | None = None,
         on_init: t.Callable[[Server], None] | None = None,
         socket_name_factory: t.Callable[[], str] | None = None,
+        engine: Engine | None = None,
         **kwargs: t.Any,
     ) -> None:
         EnvironmentMixin.__init__(self, "-g")
         self._windows: list[WindowDict] = []
         self._panes: list[PaneDict] = []
 
+        if engine is None:
+            engine = SubprocessEngine()
+        self.engine = engine
+
         if socket_path is not None:
             self.socket_path = socket_path
         elif socket_name is not None:
@@ -205,6 +212,12 @@ 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
+
+        if isinstance(self.engine, ControlModeEngine):
+            return self._probe_server() == 0
+
         try:
             res = self.cmd("list-sessions")
         except Exception:
@@ -221,23 +234,57 @@ def raise_if_dead(self) -> None:
         ...     print(type(e))
         <class 'subprocess.CalledProcessError'>
         """
+        from libtmux._internal.engines.control_mode import ControlModeEngine
+
+        if isinstance(self.engine, ControlModeEngine):
+            rc = self._probe_server()
+            if rc != 0:
+                tmux_bin_probe = shutil.which("tmux") or "tmux"
+                raise subprocess.CalledProcessError(
+                    returncode=rc,
+                    cmd=[tmux_bin_probe, *self._build_server_args(), "list-sessions"],
+                )
+            return
+
         tmux_bin = shutil.which("tmux")
         if tmux_bin is None:
             raise exc.TmuxCommandNotFound
 
-        cmd_args: list[str] = ["list-sessions"]
+        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(
+                returncode=proc.returncode,
+                cmd=[tmux_bin, *server_args, "list-sessions"],
+            )
+
+    #
+    # Command
+    #
+    def _build_server_args(self) -> list[str]:
+        """Return tmux server args based on socket/config settings."""
+        server_args: list[str] = []
         if self.socket_name:
-            cmd_args.insert(0, f"-L{self.socket_name}")
+            server_args.append(f"-L{self.socket_name}")
         if self.socket_path:
-            cmd_args.insert(0, f"-S{self.socket_path}")
+            server_args.append(f"-S{self.socket_path}")
         if self.config_file:
-            cmd_args.insert(0, f"-f{self.config_file}")
+            server_args.append(f"-f{self.config_file}")
+        return server_args
 
-        subprocess.check_call([tmux_bin, *cmd_args])
+    def _probe_server(self) -> int:
+        """Check server liveness without bootstrapping control mode."""
+        tmux_bin = shutil.which("tmux")
+        if tmux_bin is None:
+            raise exc.TmuxCommandNotFound
+
+        result = subprocess.run(
+            [tmux_bin, *self._build_server_args(), "list-sessions"],
+            check=False,
+            capture_output=True,
+        )
+        return result.returncode
 
-    #
-    # Command
-    #
     def cmd(
         self,
         cmd: str,
@@ -291,25 +338,24 @@ def cmd(
 
             Renamed from ``.tmux`` to ``.cmd``.
         """
-        svr_args: list[str | int] = [cmd]
-        cmd_args: list[str | int] = []
+        server_args: list[str | int] = []
         if self.socket_name:
-            svr_args.insert(0, f"-L{self.socket_name}")
+            server_args.append(f"-L{self.socket_name}")
         if self.socket_path:
-            svr_args.insert(0, f"-S{self.socket_path}")
+            server_args.append(f"-S{self.socket_path}")
         if self.config_file:
-            svr_args.insert(0, f"-f{self.config_file}")
+            server_args.append(f"-f{self.config_file}")
         if self.colors:
             if self.colors == 256:
-                svr_args.insert(0, "-2")
+                server_args.append("-2")
             elif self.colors == 88:
-                svr_args.insert(0, "-8")
+                server_args.append("-8")
             else:
                 raise exc.UnknownColorOption
 
-        cmd_args = ["-t", str(target), *args] if target is not None else [*args]
+        cmd_args = ["-t", str(target), *args] if target is not None else list(args)
 
-        return tmux_cmd(*svr_args, *cmd_args)
+        return self.engine.run(cmd, cmd_args=cmd_args, server_args=server_args)
 
     @property
     def attached_sessions(self) -> list[Session]:
@@ -324,10 +370,28 @@ def attached_sessions(self) -> list[Session]:
         -------
         list of :class:`Session`
         """
-        return self.sessions.filter(session_attached__noeq="1")
+        sessions = list(self.sessions.filter(session_attached__noeq="1"))
+
+        # Let the engine hide its own internal client if it wants to.
+        filter_fn = getattr(self.engine, "exclude_internal_sessions", None)
+        if callable(filter_fn):
+            server_args = tuple(self._build_server_args())
+            try:
+                sessions = filter_fn(
+                    sessions,
+                    server_args=server_args,
+                )
+            except TypeError:
+                # Subprocess engine does not accept server_args; ignore.
+                sessions = filter_fn(sessions)
+
+        return sessions
 
     def has_session(self, target_session: str, exact: bool = True) -> bool:
-        """Return True if session exists.
+        """Return True if session exists (excluding internal engine sessions).
+
+        Internal sessions used by engines for connection management are
+        excluded to maintain engine transparency.
 
         Parameters
         ----------
@@ -347,6 +411,11 @@ def has_session(self, target_session: str, exact: bool = True) -> bool:
         """
         session_check_name(target_session)
 
+        # Never report internal engine sessions as existing
+        internal_names = self._get_internal_session_names()
+        if target_session in internal_names:
+            return False
+
         if exact:
             target_session = f"={target_session}"
 
@@ -412,6 +481,15 @@ def switch_client(self, target_session: str) -> None:
         """
         session_check_name(target_session)
 
+        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):
+            msg = "no current client"
+            raise exc.LibTmuxException(msg)
+
         proc = self.cmd("switch-client", target=target_session)
 
         if proc.stderr:
@@ -435,6 +513,78 @@ def attach_session(self, target_session: str | None = None) -> None:
         if proc.stderr:
             raise exc.LibTmuxException(proc.stderr)
 
+    def connect(self, session_name: str) -> Session:
+        """Connect to a session, creating if it doesn't exist.
+
+        Returns an existing session if found, otherwise creates a new detached session.
+
+        Parameters
+        ----------
+        session_name : str
+            Name of the session to connect to.
+
+        Returns
+        -------
+        :class:`Session`
+            The connected or newly created session.
+
+        Raises
+        ------
+        :exc:`exc.BadSessionName`
+            If the session name is invalid (contains '.' or ':').
+        :exc:`exc.LibTmuxException`
+            If tmux returns an error.
+
+        Examples
+        --------
+        >>> session = server.connect('my_session')
+        >>> session.name
+        'my_session'
+
+        Calling again returns the same session:
+
+        >>> session2 = server.connect('my_session')
+        >>> session2.session_id == session.session_id
+        True
+        """
+        session_check_name(session_name)
+
+        # Check if session already exists
+        if self.has_session(session_name):
+            session = self.sessions.get(session_name=session_name)
+            if session is None:
+                msg = "Session lookup failed after has_session passed"
+                raise exc.LibTmuxException(msg)
+            return session
+
+        # Session doesn't exist, create it
+        # Save and clear TMUX env var (same as new_session)
+        env = os.environ.get("TMUX")
+        if env:
+            del os.environ["TMUX"]
+
+        proc = self.cmd(
+            "new-session",
+            "-d",
+            f"-s{session_name}",
+            "-P",
+            "-F#{session_id}",
+        )
+
+        if proc.stderr:
+            raise exc.LibTmuxException(proc.stderr)
+
+        session_id = proc.stdout[0]
+
+        # Restore TMUX env var
+        if env:
+            os.environ["TMUX"] = env
+
+        return Session.from_session_id(
+            server=self,
+            session_id=session_id,
+        )
+
     def new_session(
         self,
         session_name: str | None = None,
@@ -596,14 +746,51 @@ def new_session(
     #
     # Relations
     #
+    def _get_internal_session_names(self) -> set[str]:
+        """Get session names used internally by the engine for management."""
+        internal_names: set[str] = set(
+            getattr(self.engine, "internal_session_names", set()),
+        )
+        try:
+            return set(internal_names)
+        except Exception:  # pragma: no cover - defensive
+            return set()
+
     @property
     def sessions(self) -> QueryList[Session]:
-        """Sessions contained in server.
+        """Sessions contained in server (excluding internal engine sessions).
+
+        Internal sessions are used by engines for connection management
+        (e.g., control mode maintains a persistent connection session).
+        These are automatically filtered to maintain engine transparency.
+
+        For advanced debugging, use the internal :meth:`._sessions_all()` method.
 
         Can be accessed via
         :meth:`.sessions.get() <libtmux._internal.query_list.QueryList.get()>` and
         :meth:`.sessions.filter() <libtmux._internal.query_list.QueryList.filter()>`
         """
+        all_sessions = self._sessions_all()
+
+        # Filter out internal engine sessions
+        internal_names = self._get_internal_session_names()
+        filtered_sessions = [
+            s for s in all_sessions if s.session_name not in internal_names
+        ]
+
+        return QueryList(filtered_sessions)
+
+    def _sessions_all(self) -> QueryList[Session]:
+        """Return all sessions including internal engine sessions.
+
+        Used internally for engine management and advanced debugging.
+        Most users should use the :attr:`.sessions` property instead.
+
+        Returns
+        -------
+        QueryList[Session]
+            All sessions including internal ones used by engines.
+        """
         sessions: list[Session] = []
 
         try: