ControlMode(core): Add engine stack and protocol bridge

why: introduce control-mode execution path with protocol parsing while keeping
public cmd API compatible.

what:
- add Engine hooks (internal_session_names, exclude_internal_sessions) plus control/subprocess engines
- parse control-mode stream via ControlProtocol and surface CommandResult metadata
- retain tmux_cmd compatibility and control-aware capture_pane trimming/retry
- extend exception types for control-mode timeouts/connection/protocol errors

Author: tony

src/libtmux/_internal/engines/base.py

@@ -0,0 +1,191 @@
+"""Engine abstractions and shared types for libtmux."""
+
+from __future__ import annotations
+
+import dataclasses
+import enum
+import typing as t
+from abc import ABC, abstractmethod
+
+from libtmux.common import tmux_cmd
+
+if t.TYPE_CHECKING:
+    from libtmux.session import Session
+
+
+class ExitStatus(enum.Enum):
+    """Exit status returned by tmux control mode commands."""
+
+    OK = 0
+    ERROR = 1
+
+
+@dataclasses.dataclass
+class CommandResult:
+    """Canonical result shape produced by engines.
+
+    This is the internal representation used by engines. Public-facing APIs
+    still return :class:`libtmux.common.tmux_cmd` for compatibility; see
+    :func:`command_result_to_tmux_cmd`.
+    """
+
+    argv: list[str]
+    stdout: list[str]
+    stderr: list[str]
+    exit_status: ExitStatus
+    cmd_id: int | None = None
+    start_time: float | None = None
+    end_time: float | None = None
+    tmux_time: int | None = None
+    flags: int | None = None
+
+    @property
+    def returncode(self) -> int:
+        """Return a POSIX-style return code matching tmux expectations."""
+        return 0 if self.exit_status is ExitStatus.OK else 1
+
+
+class NotificationKind(enum.Enum):
+    """High-level categories for tmux control-mode notifications."""
+
+    PANE_OUTPUT = enum.auto()
+    PANE_EXTENDED_OUTPUT = enum.auto()
+    PANE_MODE_CHANGED = enum.auto()
+    WINDOW_LAYOUT_CHANGED = enum.auto()
+    WINDOW_ADD = enum.auto()
+    WINDOW_CLOSE = enum.auto()
+    UNLINKED_WINDOW_ADD = enum.auto()
+    UNLINKED_WINDOW_CLOSE = enum.auto()
+    UNLINKED_WINDOW_RENAMED = enum.auto()
+    WINDOW_RENAMED = enum.auto()
+    WINDOW_PANE_CHANGED = enum.auto()
+    SESSION_CHANGED = enum.auto()
+    CLIENT_SESSION_CHANGED = enum.auto()
+    CLIENT_DETACHED = enum.auto()
+    SESSION_RENAMED = enum.auto()
+    SESSIONS_CHANGED = enum.auto()
+    SESSION_WINDOW_CHANGED = enum.auto()
+    PASTE_BUFFER_CHANGED = enum.auto()
+    PASTE_BUFFER_DELETED = enum.auto()
+    PAUSE = enum.auto()
+    CONTINUE = enum.auto()
+    SUBSCRIPTION_CHANGED = enum.auto()
+    EXIT = enum.auto()
+    RAW = enum.auto()
+
+
+@dataclasses.dataclass
+class Notification:
+    """Parsed notification emitted by tmux control mode."""
+
+    kind: NotificationKind
+    when: float
+    raw: str
+    data: dict[str, t.Any]
+
+
+@dataclasses.dataclass
+class EngineStats:
+    """Light-weight diagnostics about engine state."""
+
+    in_flight: int
+    notif_queue_depth: int
+    dropped_notifications: int
+    restarts: int
+    last_error: str | None
+    last_activity: float | None
+
+
+def command_result_to_tmux_cmd(result: CommandResult) -> tmux_cmd:
+    """Adapt :class:`CommandResult` into the legacy ``tmux_cmd`` wrapper."""
+    proc = tmux_cmd(
+        cmd=result.argv,
+        stdout=result.stdout,
+        stderr=result.stderr,
+        returncode=result.returncode,
+    )
+    # Preserve extra metadata for consumers that know about it.
+    proc.exit_status = result.exit_status  # type: ignore[attr-defined]
+    proc.cmd_id = result.cmd_id  # type: ignore[attr-defined]
+    proc.tmux_time = result.tmux_time  # type: ignore[attr-defined]
+    proc.flags = result.flags  # type: ignore[attr-defined]
+    proc.start_time = result.start_time  # type: ignore[attr-defined]
+    proc.end_time = result.end_time  # type: ignore[attr-defined]
+    return proc
+
+
+class Engine(ABC):
+    """Abstract base class for tmux execution engines.
+
+    Engines produce :class:`CommandResult` internally but surface ``tmux_cmd``
+    to the existing libtmux public surface. Subclasses should implement
+    :meth:`run_result` and rely on the base :meth:`run` adapter unless they have
+    a strong reason to override both.
+    """
+
+    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 and return a ``tmux_cmd`` wrapper."""
+        return command_result_to_tmux_cmd(
+            self.run_result(
+                cmd=cmd,
+                cmd_args=cmd_args,
+                server_args=server_args,
+                timeout=timeout,
+            ),
+        )
+
+    @abstractmethod
+    def run_result(
+        self,
+        cmd: str,
+        cmd_args: t.Sequence[str | int] | None = None,
+        server_args: t.Sequence[str | int] | None = None,
+        timeout: float | None = None,
+    ) -> CommandResult:
+        """Run a tmux command and return a :class:`CommandResult`."""
+
+    def iter_notifications(
+        self,
+        *,
+        timeout: float | None = None,
+    ) -> t.Iterator[Notification]:  # pragma: no cover - default noop
+        """Yield control-mode notifications if supported by the engine."""
+        if False:  # keeps the function a generator for typing
+            yield timeout
+        return
+
+    # Optional hooks ---------------------------------------------------
+    @property
+    def internal_session_names(self) -> set[str]:
+        """Names of sessions reserved for engine internals."""
+        return set()
+
+    def exclude_internal_sessions(
+        self,
+        sessions: list[Session],
+        *,
+        server_args: tuple[str | int, ...] | None = None,
+    ) -> list[Session]:  # pragma: no cover - overridden by control mode
+        """Allow engines to hide internal/management sessions from user lists."""
+        return sessions
+
+    def get_stats(self) -> EngineStats:  # pragma: no cover - default noop
+        """Return engine diagnostic stats."""
+        return EngineStats(
+            in_flight=0,
+            notif_queue_depth=0,
+            dropped_notifications=0,
+            restarts=0,
+            last_error=None,
+            last_activity=None,
+        )
+
+    def close(self) -> None:  # pragma: no cover - default noop
+        """Clean up any engine resources."""
+        return None