ControlModeEngine(fix[timeout]): Add timeouts and cleanup

Author: tony

src/libtmux/_internal/engines/base.py

@@ -18,6 +18,7 @@ def run(
         cmd: str,
         cmd_args: t.Sequence[str | int] | None = None,
         server_args: t.Sequence[str | int] | None = None,
+        timeout: float | None = None,
     ) -> tmux_cmd:
         """Run a tmux command and return the result."""
         ...

src/libtmux/_internal/engines/control_mode.py

@@ -19,17 +19,23 @@
 class ControlModeEngine(Engine):
     """Engine that runs tmux commands via a persistent Control Mode process."""
 
-    def __init__(self) -> None:
+    def __init__(self, command_timeout: float | None = 10.0) -> None:
         self.process: subprocess.Popen[str] | None = None
         self._lock = threading.Lock()
-        self._server_args: t.Sequence[str | int] | None = None
+        self._server_args: tuple[str | int, ...] | None = None
+        self.command_timeout = command_timeout
 
     def close(self) -> None:
         """Terminate the tmux control mode process."""
         if self.process:
             self.process.terminate()
-            self.process.wait()
+            try:
+                self.process.wait(timeout=1)
+            except subprocess.TimeoutExpired:
+                self.process.kill()
+                self.process.wait()
             self.process = None
+            self._server_args = None
 
     def __del__(self) -> None:
         """Cleanup the process on destruction."""
@@ -41,13 +47,15 @@ def _start_process(self, server_args: t.Sequence[str | int] | None) -> None:
         if not tmux_bin:
             raise exc.TmuxCommandNotFound
 
+        normalized_args: tuple[str | int, ...] = tuple(server_args or ())
+
         cmd = [tmux_bin]
-        if server_args:
-            cmd.extend(str(a) for a in server_args)
+        if normalized_args:
+            cmd.extend(str(a) for a in normalized_args)
         cmd.append("-C")
         cmd.extend(["new-session", "-A", "-s", "libtmux_control_mode"])
 
-        logger.debug(f"Starting Control Mode process: {cmd}")
+        logger.debug("Starting Control Mode process: %s", cmd)
         self.process = subprocess.Popen(
             cmd,
             stdin=subprocess.PIPE,
@@ -57,38 +65,58 @@ def _start_process(self, server_args: t.Sequence[str | int] | None) -> None:
             bufsize=0,  # Unbuffered
             errors="backslashreplace",
         )
-        self._server_args = server_args
+        self._server_args = normalized_args
 
-        # Consume startup command output
+        # Consume startup command output with the same timeout used for commands.
         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 bootstrap_reader() -> None:
+            assert self.process is not None
+            assert self.process.stdout is not None
+            while True:
+                line = self.process.stdout.readline()
+                if not line:
+                    logger.warning("Control Mode process exited immediately")
+                    break
+                if line.startswith("%end") or line.startswith("%error"):
+                    break
+
+        if self.command_timeout is None:
+            bootstrap_reader()
+        else:
+            bootstrap_thread = threading.Thread(target=bootstrap_reader, daemon=True)
+            bootstrap_thread.start()
+            bootstrap_thread.join(timeout=self.command_timeout)
+            if bootstrap_thread.is_alive():
+                logger.error("Control Mode bootstrap command timed out")
+                self.close()
+                msg = "tmux control mode command timed out"
+                raise exc.ControlModeTimeout(msg)
 
     def run(
         self,
         cmd: str,
         cmd_args: t.Sequence[str | int] | None = None,
         server_args: t.Sequence[str | int] | None = None,
+        timeout: float | None = None,
     ) -> tmux_cmd:
         """Run a tmux command via Control Mode."""
         with self._lock:
+            incoming_server_args = tuple(server_args or ())
+
             if self.process is None:
-                self._start_process(server_args)
-            elif server_args != self._server_args:
+                self._start_process(incoming_server_args)
+            elif incoming_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}"
+                    "Server args changed; restarting Control Mode process. "
+                    "Old: %s, New: %s",
+                    self._server_args,
+                    incoming_server_args,
                 )
                 self.close()
-                self._start_process(server_args)
+                self._start_process(incoming_server_args)
 
             assert self.process is not None
             assert self.process.stdin is not None
@@ -103,15 +131,15 @@ def run(
 
             command_line = shlex.join(full_args)
 
-            logger.debug(f"Sending to Control Mode: {command_line}")
+            logger.debug("Sending to Control Mode: %s", 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)
+                self._start_process(incoming_server_args)
                 assert self.process is not None
                 assert self.process.stdin is not None
                 assert self.process.stdout is not None
@@ -123,47 +151,72 @@ def run(
             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
-                    # %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
-                    # %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
-                    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)
+            def reader() -> None:
+                nonlocal returncode, stdout_lines, stderr_lines
+                assert self.process is not None
+                assert self.process.stdout is not None
+                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"):
+                        # %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 time id flags
+                        parts = line.split()
+                        if len(parts) > 3:
+                            flags = int(parts[3])
+                            if flags & 1:
+                                returncode = 1
+                        break
+                    elif line.startswith("%error"):
+                        returncode = 1
+                        stderr_lines = stdout_lines
+                        stdout_lines = []
+                        break
+                    elif line.startswith("%"):
+                        logger.debug("Control Mode Notification: %s", line)
+                        continue
+                    else:
+                        stdout_lines.append(line)
+
+            reader_exc: list[BaseException] = []
+
+            def wrapped_reader() -> None:
+                try:
+                    reader()
+                except BaseException as read_exc:  # pragma: no cover - defensive
+                    reader_exc.append(read_exc)
+
+            effective_timeout = timeout if timeout is not None else self.command_timeout
+            if effective_timeout is None:
+                wrapped_reader()
+            else:
+                reader_thread = threading.Thread(target=wrapped_reader, daemon=True)
+                reader_thread.start()
+                reader_thread.join(timeout=effective_timeout)
+                if reader_thread.is_alive():
+                    logger.error(
+                        "Control Mode command timed out waiting for response: %s",
+                        command_line,
+                    )
+                    self.close()
+                    msg = "tmux control mode command timed out"
+                    raise exc.ControlModeTimeout(msg)
+                if reader_exc:
+                    raise reader_exc[0]
 
             # Tmux usually puts error message in stdout (captured above) for %error
             # But we moved it to stderr_lines if %error occurred.

src/libtmux/_internal/engines/subprocess_engine.py

@@ -22,8 +22,11 @@ def run(
         cmd: str,
         cmd_args: t.Sequence[str | int] | None = None,
         server_args: t.Sequence[str | int] | None = None,
+        timeout: float | None = None,
     ) -> tmux_cmd:
         """Run a tmux command using subprocess.Popen."""
+        # Kept for API parity with ControlModeEngine; subprocess variant is synchronous.
+        _ = timeout
         tmux_bin = shutil.which("tmux")
         if not tmux_bin:
             raise exc.TmuxCommandNotFound

src/libtmux/exc.py

@@ -92,6 +92,10 @@ class WaitTimeout(LibTmuxException):
     """Function timed out without meeting condition."""
 
 
+class ControlModeTimeout(LibTmuxException):
+    """tmux control mode command did not return within the timeout."""
+
+
 class VariableUnpackingError(LibTmuxException):
     """Error unpacking variable."""
 

tests/test_control_mode_engine.py

@@ -2,8 +2,15 @@
 
 from __future__ import annotations
 
+import io
 import pathlib
+import subprocess
+import time
+import typing as t
 
+import pytest
+
+from libtmux import exc
 from libtmux._internal.engines.control_mode import ControlModeEngine
 from libtmux.server import Server
 
@@ -50,3 +57,44 @@ def test_control_mode_engine_basic(tmp_path: pathlib.Path) -> None:
 
     engine.process.wait(timeout=2)
     assert engine.process.poll() is not None
+
+
+def test_control_mode_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
+    """ControlModeEngine should surface timeouts and clean up the process."""
+
+    class BlockingStdout:
+        def readline(self) -> str:
+            time.sleep(0.05)
+            return ""
+
+    class FakeProcess:
+        def __init__(self) -> None:
+            self.stdin = io.StringIO()
+            self.stdout = BlockingStdout()
+            self._terminated = False
+
+        def poll(self) -> None:
+            return None
+
+        def terminate(self) -> None:
+            self._terminated = True
+
+        def wait(self, timeout: float | None = None) -> None:
+            return None
+
+    engine = ControlModeEngine(command_timeout=0.01)
+
+    fake_process = FakeProcess()
+
+    def fake_start(
+        server_args: t.Sequence[str | int] | None,
+    ) -> None:  # pragma: no cover - simple stub
+        engine.process = t.cast(subprocess.Popen[str], fake_process)
+        engine._server_args = tuple(server_args or ())
+
+    monkeypatch.setattr(engine, "_start_process", fake_start)
+
+    with pytest.raises(exc.ControlModeTimeout):
+        engine.run("list-sessions")
+
+    assert engine.process is None