feat: implement robust ControlModeEngine with startup session

Author: tony

src/libtmux/_internal/engines/control_mode.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import shlex
 import shutil
 import subprocess
 import threading
@@ -44,6 +45,7 @@ def _start_process(self, server_args: t.Sequence[str | int] | None) -> None:
         if server_args:
             cmd.extend(str(a) for a in server_args)
         cmd.append("-C")
+        cmd.extend(["new-session", "-A", "-s", "libtmux_control_mode"])
 
         logger.debug(f"Starting Control Mode process: {cmd}")
         self.process = subprocess.Popen(
@@ -57,6 +59,17 @@ def _start_process(self, server_args: t.Sequence[str | int] | None) -> None:
         )
         self._server_args = server_args
 
+        # Consume startup command output
+        assert self.process.stdout is not None
+        while True:
+            line = self.process.stdout.readline()
+            if not line:
+                # EOF immediately?
+                logger.warning("Control Mode process exited immediately")
+                break
+            if line.startswith("%end") or line.startswith("%error"):
+                break
+
     def run(
         self,
         cmd: str,
@@ -82,13 +95,13 @@ def run(
             assert self.process.stdout is not None
 
             # Construct the command line
-            # We use subprocess.list2cmdline for basic quoting, but we need to be
-            # careful. tmux control mode accepts a single line.
+            # We use shlex.join for correct shell-like quoting, required by tmux control
+            # mode.
             full_args = [cmd]
             if cmd_args:
                 full_args.extend(str(a) for a in cmd_args)
 
-            command_line = subprocess.list2cmdline(full_args)
+            command_line = shlex.join(full_args)
 
             logger.debug(f"Sending to Control Mode: {command_line}")
             try:
@@ -122,10 +135,21 @@ def run(
 
                 if line.startswith("%begin"):
                     # Start of response
+                    # %begin time id flags
+                    parts = line.split()
+                    if len(parts) > 3:
+                        flags = int(parts[3])
+                        if flags & 1:
+                            returncode = 1
                     continue
                 elif line.startswith("%end"):
                     # End of success response
-                    returncode = 0
+                    # %end time id flags
+                    parts = line.split()
+                    if len(parts) > 3:
+                        flags = int(parts[3])
+                        if flags & 1:
+                            returncode = 1
                     break
                 elif line.startswith("%error"):
                     # End of error response
@@ -144,6 +168,10 @@ def run(
             # Tmux usually puts error message in stdout (captured above) for %error
             # But we moved it to stderr_lines if %error occurred.
 
+            # If we detected failure via flags but got %end, treat stdout as potentially
+            # containing info?
+            # For now, keep stdout as is.
+
             # Mimic subprocess.communicate output structure
             return tmux_cmd(
                 cmd=[cmd] + (list(map(str, cmd_args)) if cmd_args else []),

tests/test_control_mode_engine.py

@@ -0,0 +1,52 @@
+"""Tests for ControlModeEngine."""
+
+from __future__ import annotations
+
+import pathlib
+
+from libtmux._internal.engines.control_mode import ControlModeEngine
+from libtmux.server import Server
+
+
+def test_control_mode_engine_basic(tmp_path: pathlib.Path) -> None:
+    """Test basic functionality of ControlModeEngine."""
+    socket_path = tmp_path / "tmux-control-mode-test"
+    engine = ControlModeEngine()
+
+    # Server should auto-start engine on first cmd
+    server = Server(socket_path=socket_path, engine=engine)
+
+    # kill server if exists (cleanup from previous runs if any)
+    if server.is_alive():
+        server.kill()
+
+    # new session
+    session = server.new_session(session_name="test_sess", kill_session=True)
+    assert session.name == "test_sess"
+
+    # check engine process is running
+    assert engine.process is not None
+    assert engine.process.poll() is None
+
+    # list sessions
+    # ControlModeEngine creates a bootstrap session "libtmux_control_mode", so we
+    # expect 2 sessions
+    sessions = server.sessions
+    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
+
+    # run a command that returns output
+    output = server.cmd("display-message", "-p", "hello").stdout
+    assert output == ["hello"]
+
+    # cleanup
+    server.kill()
+    # Engine process should terminate eventually (ControlModeEngine.close is called
+    # manually or via weakref/del)
+    # Server.kill() kills the tmux SERVER. The control mode client process should
+    # exit as a result.
+
+    engine.process.wait(timeout=2)
+    assert engine.process.poll() is not None