feat(ControlMode): filter internal session + add configurability

tony · tony · commit 8659d8779de0 · 2025-11-23T09:03:00.000-06:00
Phase 1: Automatic Internal Session Filtering - Filter control mode's internal session from Server.sessions property - Update has_session() to exclude internal sessions - Add Server._sessions_all() for advanced debugging - Add Server._get_internal_session_names() to query engine for internal sessions Phase 2: Engine Configurability - Add session_name parameter to ControlModeEngine for custom internal session names - Add attach_to parameter to attach to existing sessions (advanced opt-in) - Update _start_process() to handle both new-session and attach-session modes - Dynamic filtering based on engine configuration Tests: - test_sessions_excludes_internal_control_mode: verify filtering works - test_has_session_excludes_control_mode: verify has_session consistency - test_session_count_engine_agnostic: verify engine transparency - test_control_mode_custom_session_name: verify custom naming - test_control_mode_attach_to_existing: verify attach mode Documentation: - Update control_mode.md with filtering behavior and advanced options - Add warnings about attach_to notification spam - Update CHANGES with features and compatibility notes This makes control mode engine truly transparent - users see the same session behavior regardless of which engine they choose. Internal sessions are used for connection management but hidden from user-facing APIs. Related: #605
diff --git a/CHANGES b/CHANGES
@@ -40,16 +40,22 @@ _Future release notes will be placed here_
   notification parsing (layout changes, unlinked windows, client detach/session change,
   session rename, paste-buffer events), and stats while keeping existing
   `Server/Session/Window/Pane.cmd` return type (`tmux_cmd`) stable. (#605)
+- Control mode engine's internal connection session is now automatically filtered from
+  `Server.sessions` and `Server.has_session()`, making engine choice transparent to
+  users. Advanced users can access all sessions via `Server._sessions_all()`. (#605)
+- `ControlModeEngine` accepts optional `session_name` and `attach_to` parameters for
+  advanced session management scenarios. (#605)
 - `Server.connect()`: New convenience method for session management. Returns an
   existing session if found, otherwise creates a new detached session. Simplifies
   common session reuse patterns and works transparently with both subprocess and
   control-mode engines.
 
 ### Compatibility
 
-- Control mode creates a bootstrap session named ``libtmux_control_mode``; callers that
-  enumerate sessions should ignore or filter it. APIs remain unchanged for tmux command
-  return objects; new metadata is attached for advanced users. (#605)
+- Control mode's internal session is now automatically filtered from user-facing APIs.
+  Code that previously filtered `libtmux_control_mode` manually can be simplified.
+  APIs remain unchanged for tmux command return objects; new metadata is attached for
+  advanced users. (#605)
 
 ### Testing
 
diff --git a/docs/topics/control_mode.md b/docs/topics/control_mode.md
@@ -42,8 +42,9 @@ for notif in engine.iter_notifications(timeout=0.1):
 ```
 
 :::{note}
-Control mode creates a bootstrap tmux session named ``libtmux_control_mode``.
-If your code enumerates sessions, filter it out.
+Control mode creates an internal session for connection management (default name:
+`libtmux_control_mode`). This session is automatically filtered from
+`Server.sessions` and `Server.has_session()` to maintain engine transparency.
 :::
 
 ## Session management with Control Mode
@@ -69,6 +70,50 @@ assert session2.session_id == session.session_id
 This works transparently with both control mode and subprocess engines, making it
 easy to switch between them without changing your code.
 
+## Advanced Configuration
+
+### Custom Internal Session Name
+
+For testing or advanced scenarios, you can customize the internal session name:
+
+```python
+from libtmux._internal.engines.control_mode import ControlModeEngine
+from libtmux.server import Server
+
+engine = ControlModeEngine(session_name="my_control_session")
+server = Server(engine=engine)
+
+# Internal session is still filtered
+user_session = server.new_session("my_app")
+len(server.sessions)  # 1 (only my_app visible)
+
+# But exists internally
+len(server._sessions_all())  # 2 (my_app + my_control_session)
+```
+
+### Attach to Existing Session
+
+For expert use cases, control mode can attach to an existing session instead of
+creating an internal one:
+
+```python
+# Create a session first
+server.new_session("shared")
+
+# Control mode attaches to it for its connection
+engine = ControlModeEngine(attach_to="shared")
+server = Server(engine=engine)
+
+# The shared session is visible (not filtered)
+len(server.sessions)  # 1 (shared session)
+```
+
+:::{warning}
+Attaching to active user sessions will generate notification traffic from pane
+output and layout changes. This increases protocol parsing overhead and may impact
+performance. Use only when you need control mode notifications for a specific session.
+:::
+
 ## Parsing notifications directly
 
 The protocol parser can be used without tmux to understand the wire format.
diff --git a/src/libtmux/_internal/engines/control_mode.py b/src/libtmux/_internal/engines/control_mode.py
@@ -26,13 +26,41 @@
 
 
 class ControlModeEngine(Engine):
-    """Engine that runs tmux commands via a persistent Control Mode process."""
+    """Engine that runs tmux commands via a persistent Control Mode process.
+
+    By default, creates an internal session for connection management.
+    This session is hidden from user-facing APIs like Server.sessions.
+    """
 
     def __init__(
         self,
         command_timeout: float | None = 10.0,
         notification_queue_size: int = 4096,
+        session_name: str | None = None,
+        attach_to: str | None = None,
     ) -> None:
+        """Initialize control mode engine.
+
+        Parameters
+        ----------
+        command_timeout : float, optional
+            Timeout for tmux commands in seconds. Default: 10.0
+        notification_queue_size : int
+            Size of notification queue. Default: 4096
+        session_name : str, optional
+            Custom name for internal control session.
+            Default: "libtmux_control_mode"
+
+            The internal session is used for connection management and is
+            automatically filtered from user-facing APIs.
+        attach_to : str, optional
+            Attach to existing session instead of creating internal one.
+            When set, control mode attaches to this session for its connection.
+
+            .. warning::
+               Attaching to user sessions can cause notification spam from
+               pane output. Use for advanced scenarios only.
+        """
         self.process: subprocess.Popen[str] | None = None
         self._lock = threading.Lock()
         self._server_args: tuple[str | int, ...] | None = None
@@ -45,6 +73,8 @@ def __init__(
             notification_queue_size=notification_queue_size,
         )
         self._restarts = 0
+        self._session_name = session_name or "libtmux_control_mode"
+        self._attach_to = attach_to
 
     # Lifecycle ---------------------------------------------------------
     def close(self) -> None:
@@ -167,15 +197,43 @@ def _start_process(self, server_args: tuple[str | int, ...]) -> None:
             notification_queue_size=self._notification_queue_size,
         )
 
-        cmd = [
-            tmux_bin,
-            *[str(a) for a in server_args],
-            "-C",
-            "new-session",
-            "-A",
-            "-s",
-            "libtmux_control_mode",
-        ]
+        # Build command based on configuration
+        if self._attach_to:
+            # Attach to existing session (advanced mode)
+            cmd = [
+                tmux_bin,
+                *[str(a) for a in server_args],
+                "-C",
+                "attach-session",
+                "-t",
+                self._attach_to,
+            ]
+            bootstrap_argv = [
+                tmux_bin,
+                *[str(a) for a in server_args],
+                "attach-session",
+                "-t",
+                self._attach_to,
+            ]
+        else:
+            # Create or attach to internal session (default)
+            cmd = [
+                tmux_bin,
+                *[str(a) for a in server_args],
+                "-C",
+                "new-session",
+                "-A",
+                "-s",
+                self._session_name,
+            ]
+            bootstrap_argv = [
+                tmux_bin,
+                *[str(a) for a in server_args],
+                "new-session",
+                "-A",
+                "-s",
+                self._session_name,
+            ]
 
         logger.debug("Starting Control Mode process: %s", cmd)
         self.process = subprocess.Popen(
@@ -188,18 +246,9 @@ def _start_process(self, server_args: tuple[str | int, ...]) -> None:
             errors="backslashreplace",
         )
 
-        # The initial command (new-session) emits an output block; register
-        # a context so the protocol can consume it.
-        bootstrap_ctx = CommandContext(
-            argv=[
-                tmux_bin,
-                *[str(a) for a in server_args],
-                "new-session",
-                "-A",
-                "-s",
-                "libtmux_control_mode",
-            ],
-        )
+        # The initial command (new-session or attach-session) emits an output
+        # block; register a context so the protocol can consume it.
+        bootstrap_ctx = CommandContext(argv=bootstrap_argv)
         self._protocol.register_command(bootstrap_ctx)
 
         # Start IO threads after registration to avoid early protocol errors.
diff --git a/src/libtmux/server.py b/src/libtmux/server.py
@@ -327,7 +327,10 @@ def attached_sessions(self) -> list[Session]:
         return self.sessions.filter(session_attached__noeq="1")
 
     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
         ----------
@@ -348,6 +351,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 and has_gte_version("2.1"):
             target_session = f"={target_session}"
 
@@ -675,14 +683,60 @@ def new_session(
     #
     # Relations
     #
+    def _get_internal_session_names(self) -> set[str]:
+        """Get session names used internally by the engine.
+
+        These sessions are used for engine management (e.g., control mode
+        connection) and should be hidden from user-facing APIs.
+
+        Returns
+        -------
+        set[str]
+            Set of internal session names to filter.
+        """
+        from libtmux._internal.engines.control_mode import ControlModeEngine
+
+        if isinstance(self.engine, ControlModeEngine) and not self.engine._attach_to:
+            # Only filter if not attaching to user session
+            return {self.engine._session_name}
+
+        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:
diff --git a/tests/test_control_mode_engine.py b/tests/test_control_mode_engine.py
@@ -37,13 +37,17 @@ def test_control_mode_engine_basic(tmp_path: pathlib.Path) -> None:
     assert engine.process.poll() is None
 
     # list sessions
-    # ControlModeEngine creates a bootstrap session "libtmux_control_mode", so we
-    # expect 2 sessions
+    # Control mode bootstrap session is now filtered from server.sessions
     sessions = server.sessions
-    assert len(sessions) >= 1
+    assert len(sessions) == 1
     session_names = [s.name for s in sessions]
     assert "test_sess" in session_names
-    assert "libtmux_control_mode" in session_names
+
+    # Verify bootstrap session exists but is filtered (use internal method)
+    all_sessions = server._sessions_all()
+    all_session_names = [s.name for s in all_sessions]
+    assert "libtmux_control_mode" in all_session_names
+    assert len(all_sessions) == 2  # test_sess + libtmux_control_mode
 
     # run a command that returns output
     output_cmd = server.cmd("display-message", "-p", "hello")
@@ -106,3 +110,66 @@ def fake_start(server_args: t.Sequence[str | int] | None) -> None:
         engine.run("list-sessions")
 
     assert engine.process is None
+
+
+def test_control_mode_custom_session_name(tmp_path: pathlib.Path) -> None:
+    """Control mode engine can use custom internal session name."""
+    socket_path = tmp_path / "tmux-custom-session-test"
+    engine = ControlModeEngine(session_name="my_control_session")
+    server = Server(socket_path=socket_path, engine=engine)
+
+    # Cleanup if exists
+    if server.is_alive():
+        server.kill()
+
+    # Create user session
+    server.new_session(session_name="user_app")
+
+    # Only user session visible via public API
+    assert len(server.sessions) == 1
+    assert server.sessions[0].name == "user_app"
+
+    # Custom internal session exists but is filtered
+    all_sessions = server._sessions_all()
+    all_names = [s.name for s in all_sessions]
+    assert "my_control_session" in all_names
+    assert "user_app" in all_names
+    assert len(all_sessions) == 2
+
+    # Cleanup
+    server.kill()
+    assert engine.process is not None
+    engine.process.wait(timeout=2)
+
+
+def test_control_mode_attach_to_existing(tmp_path: pathlib.Path) -> None:
+    """Control mode can attach to existing session (advanced opt-in)."""
+    socket_path = tmp_path / "tmux-attach-test"
+
+    # Create session first with subprocess engine
+    from libtmux._internal.engines.subprocess_engine import SubprocessEngine
+
+    subprocess_engine = SubprocessEngine()
+    server1 = Server(socket_path=socket_path, engine=subprocess_engine)
+
+    if server1.is_alive():
+        server1.kill()
+
+    server1.new_session(session_name="shared_session")
+
+    # Control mode attaches to existing session (no internal session created)
+    control_engine = ControlModeEngine(attach_to="shared_session")
+    server2 = Server(socket_path=socket_path, engine=control_engine)
+
+    # Should see the shared session
+    assert len(server2.sessions) == 1
+    assert server2.sessions[0].name == "shared_session"
+
+    # No internal session was created
+    all_sessions = server2._sessions_all()
+    assert len(all_sessions) == 1  # Only shared_session
+
+    # Cleanup
+    server2.kill()
+    assert control_engine.process is not None
+    control_engine.process.wait(timeout=2)
diff --git a/tests/test_server.py b/tests/test_server.py
