Engine(feat[control-mode]): add protocol parser and tmux_cmd adapter

Author: tony

src/libtmux/_internal/engines/base.py

@@ -1,24 +1,164 @@
-"""Base engine for libtmux."""
+"""Engine abstractions and shared types for libtmux."""
 
 from __future__ import annotations
 
+import dataclasses
+import enum
 import typing as t
 from abc import ABC, abstractmethod
 
-if t.TYPE_CHECKING:
-    from libtmux.common import tmux_cmd
+from libtmux.common import tmux_cmd
+
+
+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_ADD = enum.auto()
+    WINDOW_CLOSE = enum.auto()
+    WINDOW_RENAMED = enum.auto()
+    WINDOW_PANE_CHANGED = enum.auto()
+    SESSION_CHANGED = enum.auto()
+    SESSIONS_CHANGED = enum.auto()
+    SESSION_WINDOW_CHANGED = 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."""
+    """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.
+    """
 
-    @abstractmethod
     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 the result."""
-        ...
+        """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
+
+    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