feat: Implement ControlModeEngine and refactor Server.cmd

Author: tony

src/libtmux/_internal/engines/base.py

@@ -13,6 +13,11 @@ class Engine(ABC):
     """Abstract base class for tmux execution engines."""
 
     @abstractmethod
-    def run(self, *args: t.Any) -> tmux_cmd:
+    def run(
+        self,
+        cmd: str,
+        cmd_args: t.Sequence[str | int] | None = None,
+        server_args: t.Sequence[str | int] | None = None,
+    ) -> tmux_cmd:
         """Run a tmux command and return the result."""
         ...

src/libtmux/_internal/engines/control_mode.py

@@ -0,0 +1,153 @@
+"""Control Mode engine for libtmux."""
+
+from __future__ import annotations
+
+import logging
+import shutil
+import subprocess
+import threading
+import typing as t
+
+from libtmux import exc
+from libtmux._internal.engines.base import Engine
+from libtmux.common import tmux_cmd
+
+logger = logging.getLogger(__name__)
+
+
+class ControlModeEngine(Engine):
+    """Engine that runs tmux commands via a persistent Control Mode process."""
+
+    def __init__(self) -> None:
+        self.process: subprocess.Popen[str] | None = None
+        self._lock = threading.Lock()
+        self._server_args: t.Sequence[str | int] | None = None
+
+    def close(self) -> None:
+        """Terminate the tmux control mode process."""
+        if self.process:
+            self.process.terminate()
+            self.process.wait()
+            self.process = None
+
+    def __del__(self) -> None:
+        """Cleanup the process on destruction."""
+        self.close()
+
+    def _start_process(self, server_args: t.Sequence[str | int] | None) -> None:
+        """Start the tmux control mode process."""
+        tmux_bin = shutil.which("tmux")
+        if not tmux_bin:
+            raise exc.TmuxCommandNotFound
+
+        cmd = [tmux_bin]
+        if server_args:
+            cmd.extend(str(a) for a in server_args)
+        cmd.append("-C")
+
+        logger.debug(f"Starting Control Mode process: {cmd}")
+        self.process = subprocess.Popen(
+            cmd,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            bufsize=0,  # Unbuffered
+            errors="backslashreplace",
+        )
+        self._server_args = server_args
+
+    def run(
+        self,
+        cmd: str,
+        cmd_args: t.Sequence[str | int] | None = None,
+        server_args: t.Sequence[str | int] | None = None,
+    ) -> tmux_cmd:
+        """Run a tmux command via Control Mode."""
+        with self._lock:
+            if self.process is None:
+                self._start_process(server_args)
+            elif server_args != self._server_args:
+                # If server_args changed, we might need a new process.
+                # For now, just warn or restart. Restarting is safer.
+                logger.warning(
+                    "Server args changed, restarting Control Mode process. "
+                    f"Old: {self._server_args}, New: {server_args}"
+                )
+                self.close()
+                self._start_process(server_args)
+
+            assert self.process is not None
+            assert self.process.stdin is not None
+            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.
+            full_args = [cmd]
+            if cmd_args:
+                full_args.extend(str(a) for a in cmd_args)
+
+            command_line = subprocess.list2cmdline(full_args)
+
+            logger.debug(f"Sending to Control Mode: {command_line}")
+            try:
+                self.process.stdin.write(command_line + "\n")
+                self.process.stdin.flush()
+            except BrokenPipeError:
+                # Process died?
+                logger.exception("Control Mode process died, restarting...")
+                self.close()
+                self._start_process(server_args)
+                assert self.process is not None
+                assert self.process.stdin is not None
+                assert self.process.stdout is not None
+                self.process.stdin.write(command_line + "\n")
+                self.process.stdin.flush()
+
+            # Read response
+            stdout_lines: list[str] = []
+            stderr_lines: list[str] = []
+            returncode = 0
+
+            while True:
+                line = self.process.stdout.readline()
+                if not line:
+                    # EOF
+                    logger.error("Unexpected EOF from Control Mode process")
+                    returncode = 1
+                    break
+
+                line = line.rstrip("\n")
+
+                if line.startswith("%begin"):
+                    # Start of response
+                    continue
+                elif line.startswith("%end"):
+                    # End of success response
+                    returncode = 0
+                    break
+                elif line.startswith("%error"):
+                    # End of error response
+                    returncode = 1
+                    # Captured lines are the error message
+                    stderr_lines = stdout_lines
+                    stdout_lines = []
+                    break
+                elif line.startswith("%"):
+                    # Notification (ignore for now)
+                    logger.debug(f"Control Mode Notification: {line}")
+                    continue
+                else:
+                    stdout_lines.append(line)
+
+            # Tmux usually puts error message in stdout (captured above) for %error
+            # But we moved it to stderr_lines if %error occurred.
+
+            # Mimic subprocess.communicate output structure
+            return tmux_cmd(
+                cmd=[cmd] + (list(map(str, cmd_args)) if cmd_args else []),
+                stdout=stdout_lines,
+                stderr=stderr_lines,
+                returncode=returncode,
+            )

src/libtmux/_internal/engines/subprocess_engine.py

@@ -17,19 +17,29 @@
 class SubprocessEngine(Engine):
     """Engine that runs tmux commands via subprocess."""
 
-    def run(self, *args: t.Any) -> tmux_cmd:
+    def run(
+        self,
+        cmd: str,
+        cmd_args: t.Sequence[str | int] | None = None,
+        server_args: t.Sequence[str | int] | None = None,
+    ) -> tmux_cmd:
         """Run a tmux command using subprocess.Popen."""
         tmux_bin = shutil.which("tmux")
         if not tmux_bin:
             raise exc.TmuxCommandNotFound
 
-        cmd = [tmux_bin]
-        cmd += args  # add the command arguments to cmd
-        cmd = [str(c) for c in cmd]
+        full_cmd: list[str | int] = [tmux_bin]
+        if server_args:
+            full_cmd += list(server_args)
+        full_cmd.append(cmd)
+        if cmd_args:
+            full_cmd += list(cmd_args)
+
+        full_cmd_str = [str(c) for c in full_cmd]
 
         try:
             process = subprocess.Popen(
-                cmd,
+                full_cmd_str,
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE,
                 text=True,
@@ -38,7 +48,7 @@ def run(self, *args: t.Any) -> tmux_cmd:
             stdout_str, stderr_str = process.communicate()
             returncode = process.returncode
         except Exception:
-            logger.exception(f"Exception for {subprocess.list2cmdline(cmd)}")
+            logger.exception(f"Exception for {subprocess.list2cmdline(full_cmd_str)}")
             raise
 
         stdout_split = stdout_str.split("\n")
@@ -49,20 +59,20 @@ def run(self, *args: t.Any) -> tmux_cmd:
         stderr_split = stderr_str.split("\n")
         stderr = list(filter(None, stderr_split))  # filter empty values
 
-        if "has-session" in cmd and len(stderr) and not stdout_split:
+        if "has-session" in full_cmd_str and len(stderr) and not stdout_split:
             stdout = [stderr[0]]
         else:
             stdout = stdout_split
 
         logger.debug(
             "self.stdout for {cmd}: {stdout}".format(
-                cmd=" ".join(cmd),
+                cmd=" ".join(full_cmd_str),
                 stdout=stdout,
             ),
         )
 
         return tmux_cmd(
-            cmd=cmd,
+            cmd=full_cmd_str,
             stdout=stdout,
             stderr=stderr,
             returncode=returncode,

src/libtmux/server.py

@@ -221,19 +221,19 @@ def raise_if_dead(self) -> None:
         if tmux_bin is None:
             raise exc.TmuxCommandNotFound
 
-        cmd_args: list[str] = ["list-sessions"]
+        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}")
 
-        proc = self.engine.run(*cmd_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, *cmd_args],
+                cmd=[tmux_bin, *server_args, "list-sessions"],
             )
 
     #
@@ -292,25 +292,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 self.engine.run(*svr_args, *cmd_args)
+        return self.engine.run(cmd, cmd_args=cmd_args, server_args=server_args)
 
     @property
     def attached_sessions(self) -> list[Session]: