feat(engines): add control-mode engine support

Author: tony

CHANGES

@@ -13,7 +13,8 @@ $ pip install --user --upgrade --pre libtmux
 
 <!-- To maintainers and contributors: Please add notes for the forthcoming version below -->
 
-- _Future release notes will be placed here_
+- Server now accepts pluggable command engines, defaulting to the existing subprocess runner, so future backends (e.g., control mode) can be injected without breaking compatibility.
+- Added a control mode engine (`ControlModeEngine`) that executes commands via `tmux -C`, paving the way for control-mode powered workflows while keeping the existing API and defaults intact.
 
 ## libtmux 0.46.2 (2025-05-26)
 

src/libtmux/common.py

@@ -9,11 +9,11 @@
 
 import logging
 import re
-import shutil
-import subprocess
 import sys
 import typing as t
 
+from libtmux.engines import CommandResult, SubprocessEngine, TmuxEngine
+
 from . import exc
 from ._compat import LooseVersion
 
@@ -190,8 +190,23 @@ def getenv(self, name: str) -> str | bool | None:
         return opts_dict.get(name)
 
 
+_default_engine: TmuxEngine = SubprocessEngine()
+
+
+def _apply_result(target: tmux_cmd | None, result: CommandResult) -> tmux_cmd:
+    cmd_obj: tmux_cmd = object.__new__(tmux_cmd) if target is None else target
+
+    cmd_obj.cmd = list(result.cmd)
+    cmd_obj.stdout = result.stdout
+    cmd_obj.stderr = result.stderr
+    cmd_obj.returncode = result.returncode
+    cmd_obj.process = result.process
+
+    return cmd_obj
+
+
 class tmux_cmd:
-    """Run any :term:`tmux(1)` command through :py:mod:`subprocess`.
+    """Run any :term:`tmux(1)` command through the configured engine.
 
     Examples
     --------
@@ -219,52 +234,50 @@ class tmux_cmd:
         Renamed from ``tmux`` to ``tmux_cmd``.
     """
 
+    cmd: list[str]
+    stdout: list[str]
+    stderr: list[str]
+    returncode: int
+    process: object | None
+
     def __init__(self, *args: t.Any) -> None:
-        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]
-
-        self.cmd = cmd
-
-        try:
-            self.process = subprocess.Popen(
-                cmd,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
-                text=True,
-                errors="backslashreplace",
-            )
-            stdout, stderr = self.process.communicate()
-            returncode = self.process.returncode
-        except Exception:
-            logger.exception(f"Exception for {subprocess.list2cmdline(cmd)}")
-            raise
-
-        self.returncode = returncode
-
-        stdout_split = stdout.split("\n")
-        # remove trailing newlines from stdout
-        while stdout_split and stdout_split[-1] == "":
-            stdout_split.pop()
-
-        stderr_split = stderr.split("\n")
-        self.stderr = list(filter(None, stderr_split))  # filter empty values
-
-        if "has-session" in cmd and len(self.stderr) and not stdout_split:
-            self.stdout = [self.stderr[0]]
-        else:
-            self.stdout = stdout_split
-
-        logger.debug(
-            "self.stdout for {cmd}: {stdout}".format(
-                cmd=" ".join(cmd),
-                stdout=self.stdout,
-            ),
-        )
+        result = _default_engine.run(*args)
+        _apply_result(self, result)
+
+    @classmethod
+    def from_result(cls, result: CommandResult) -> tmux_cmd:
+        """Create a :class:`tmux_cmd` instance from a raw result."""
+        return _apply_result(None, result)
+
+
+def get_default_engine() -> TmuxEngine:
+    """Return the process-based default engine."""
+    return _default_engine
+
+
+def set_default_engine(engine: TmuxEngine) -> None:
+    """Override the process-based default engine.
+
+    Parameters
+    ----------
+    engine : TmuxEngine
+        Engine implementation that should back :class:`tmux_cmd` and helper functions.
+    """
+    global _default_engine
+    _default_engine = engine
+
+
+def run_tmux_command(engine: TmuxEngine, *args: t.Any) -> tmux_cmd:
+    """Execute a tmux command using the provided engine.
+
+    Falls back to the global default engine when ``engine`` matches the default,
+    preserving legacy monkeypatching that targets :class:`tmux_cmd` directly.
+    """
+    if engine is _default_engine:
+        return tmux_cmd(*args)
+
+    result = engine.run(*args)
+    return tmux_cmd.from_result(result)
 
 
 def get_version() -> LooseVersion:

src/libtmux/engines/__init__.py

@@ -0,0 +1,14 @@
+"""Engine implementations for libtmux."""
+
+from __future__ import annotations
+
+from .base import CommandResult, TmuxEngine
+from .control_mode import ControlModeEngine
+from .subprocess import SubprocessEngine
+
+__all__ = [
+    "CommandResult",
+    "ControlModeEngine",
+    "SubprocessEngine",
+    "TmuxEngine",
+]

src/libtmux/engines/base.py

@@ -0,0 +1,28 @@
+"""Core abstractions for libtmux command engines."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import Protocol
+
+
+@dataclass
+class CommandResult:
+    """Result of executing a tmux command."""
+
+    cmd: Sequence[str]
+    stdout: list[str]
+    stderr: list[str]
+    returncode: int
+    process: object | None = None
+
+
+class TmuxEngine(Protocol):
+    """Protocol for components that can execute tmux commands."""
+
+    def run(self, *args: str | int) -> CommandResult:  # pragma: no cover
+        """Execute a tmux command and return a :class:`CommandResult`.
+
+        Implementations may rely on structural typing rather than inheritance.
+        """

src/libtmux/engines/control_mode.py

@@ -0,0 +1,87 @@
+"""Control-mode backed tmux engine."""
+
+from __future__ import annotations
+
+import logging
+import shutil
+import subprocess
+
+from libtmux import exc
+from libtmux.engines.base import CommandResult, TmuxEngine
+
+logger = logging.getLogger(__name__)
+
+
+def _decode_control_payload(payload: str) -> str:
+    """Decode tmux control mode payload sequences."""
+    if not payload:
+        return ""
+    # Control mode escapes mirror C-style escapes. ``unicode_escape`` handles both
+    # simple escapes and octal sequences (e.g. ``\012``).
+    try:
+        return bytes(payload, "utf-8").decode("unicode_escape")
+    except Exception:  # pragma: no cover - defensive, should not happen
+        logger.debug("Failed to decode control payload %r", payload, exc_info=True)
+        return payload
+
+
+class ControlModeEngine(TmuxEngine):
+    """Execute tmux commands via control mode (``tmux -C``)."""
+
+    def run(self, *args: str | int) -> CommandResult:
+        """Execute a tmux command using control mode and return structured output."""
+        tmux_bin = shutil.which("tmux")
+        if tmux_bin is None:
+            raise exc.TmuxCommandNotFound
+
+        cmd: list[str] = [tmux_bin, "-C"]
+        cmd.extend(str(value) for value in args)
+
+        try:
+            process = subprocess.Popen(
+                cmd,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                errors="backslashreplace",
+            )
+            stdout, stderr = process.communicate()
+            returncode = process.returncode
+        except Exception:
+            logger.exception("Exception for %s", subprocess.list2cmdline(cmd))
+            raise
+
+        stdout_lines: list[str] = []
+        stderr_lines: list[str] = []
+
+        for line in stdout.splitlines():
+            if line.startswith("%output"):
+                parts = line.split(" ", 4)
+                payload = parts[4] if len(parts) >= 5 else ""
+                stdout_lines.append(_decode_control_payload(payload))
+            elif line.startswith("%error"):
+                # ``%error time client flags code message``
+                parts = line.split(" ", 5)
+                payload = parts[5] if len(parts) >= 6 else ""
+                stderr_lines.append(_decode_control_payload(payload))
+            elif line.startswith(("%begin", "%end", "%exit")):
+                continue
+            else:
+                # Unexpected line: keep behaviour predictable by surfacing it
+                stdout_lines.append(line)
+
+        # Combine regular stderr with parsed control errors
+        stderr_lines.extend(line for line in stderr.splitlines() if line)
+
+        logger.debug("control-mode stdout for %s: %s", " ".join(cmd), stdout_lines)
+
+        return CommandResult(
+            cmd=cmd,
+            stdout=stdout_lines,
+            stderr=stderr_lines,
+            returncode=returncode,
+            process=None,
+        )
+
+
+__all__ = ["ControlModeEngine"]

src/libtmux/engines/subprocess.py

@@ -0,0 +1,65 @@
+"""Subprocess-backed tmux engine."""
+
+from __future__ import annotations
+
+import logging
+import shutil
+import subprocess
+
+from libtmux import exc
+from libtmux.engines.base import CommandResult, TmuxEngine
+
+logger = logging.getLogger(__name__)
+
+
+class SubprocessEngine(TmuxEngine):
+    """Execute tmux commands via the tmux CLI binary."""
+
+    def run(self, *args: str | int) -> CommandResult:
+        """Execute a tmux command via subprocess and return structured output."""
+        tmux_bin = shutil.which("tmux")
+        if tmux_bin is None:
+            raise exc.TmuxCommandNotFound
+
+        cmd: list[str] = [tmux_bin]
+        cmd.extend(str(value) for value in args)
+
+        try:
+            process = subprocess.Popen(
+                cmd,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                text=True,
+                errors="backslashreplace",
+            )
+            stdout, stderr = process.communicate()
+            returncode = process.returncode
+        except Exception:
+            logger.exception("Exception for %s", subprocess.list2cmdline(cmd))
+            raise
+
+        stdout_lines = stdout.split("\n")
+        while stdout_lines and stdout_lines[-1] == "":
+            stdout_lines.pop()
+
+        stderr_lines = [line for line in stderr.split("\n") if line]
+
+        if "has-session" in cmd and stderr_lines and not stdout_lines:
+            stdout_lines = [stderr_lines[0]]
+
+        logger.debug(
+            "self.stdout for %s: %s",
+            " ".join(cmd),
+            stdout_lines,
+        )
+
+        return CommandResult(
+            cmd=cmd,
+            stdout=stdout_lines,
+            stderr=stderr_lines,
+            returncode=returncode,
+            process=process,
+        )
+
+
+__all__ = ["SubprocessEngine"]

src/libtmux/neo.py

@@ -8,7 +8,7 @@
 from collections.abc import Iterable
 
 from libtmux import exc
-from libtmux.common import tmux_cmd
+from libtmux.common import run_tmux_command
 from libtmux.formats import FORMAT_SEPARATOR
 
 if t.TYPE_CHECKING:
@@ -216,7 +216,7 @@ def fetch_objs(
 
     tmux_cmds.append("-F{}".format("".join(tmux_formats)))
 
-    proc = tmux_cmd(*tmux_cmds)  # output
+    proc = run_tmux_command(server.engine, *tmux_cmds)
 
     if proc.stderr:
         raise exc.LibTmuxException(proc.stderr)