feat(rust-backend) add server_kind/control hooks

Author: tony

.github/workflows/tests.yml

@@ -12,6 +12,11 @@ jobs:
       matrix:
         python-version: ['3.14']
         tmux-version: ['3.2a', '3.3a', '3.4', '3.5', '3.6', 'master']
+        backend: ['default']
+        include:
+          - python-version: '3.14'
+            tmux-version: '3.6'
+            backend: 'rust-control'
     steps:
       - uses: actions/checkout@v6
 
@@ -78,6 +83,24 @@ jobs:
           export PATH=$HOME/tmux-builds/tmux-${{ matrix.tmux-version }}/bin:$PATH
           ls $HOME/tmux-builds/tmux-${{ matrix.tmux-version }}/bin
           tmux -V
+          backend="${{ matrix.backend }}"
+          if [ "$backend" != "default" ]; then
+            if ! uv run python - <<'PY'
+import importlib.util
+import sys
+sys.exit(0 if importlib.util.find_spec("vibe_tmux") else 1)
+PY
+            then
+              echo "vibe_tmux not installed; skipping rust backend tests"
+              exit 0
+            fi
+            export LIBTMUX_BACKEND=rust
+            export LIBTMUX_RUST_CONNECTION_KIND=protocol
+            if [ "$backend" = "rust-control" ]; then
+              export LIBTMUX_RUST_CONTROL_MODE=1
+              export LIBTMUX_RUST_CONTROL_AUTOSTART=1
+            fi
+          fi
           uv run py.test --cov=./ --cov-append --cov-report=xml -n auto --verbose
         env:
           COV_CORE_SOURCE: .

src/libtmux/common.py

@@ -41,10 +41,12 @@
 
 _RUST_BACKEND = os.getenv("LIBTMUX_BACKEND") == "rust"
 _RUST_SERVER_CACHE: dict[
-    tuple[str | None, str | None, int | None, str | None, str | None], t.Any
+    tuple[str | None, str | None, int | None, str | None, str | None, bool | None],
+    t.Any,
 ] = {}
 _RUST_SERVER_CONFIG: dict[
-    tuple[str | None, str | None, int | None, str | None, str | None], set[str]
+    tuple[str | None, str | None, int | None, str | None, str | None, bool | None],
+    set[str],
 ] = {}
 
 
@@ -67,6 +69,18 @@ def _resolve_rust_socket_path(socket_path: str | None, socket_name: str | None)
     return str(socket_dir / name)
 
 
+def _env_bool(name: str) -> bool | None:
+    value = os.getenv(name)
+    if value is None:
+        return None
+    value = value.strip().lower()
+    if value in {"1", "true", "yes", "on"}:
+        return True
+    if value in {"0", "false", "no", "off"}:
+        return False
+    return None
+
+
 def _rust_run_with_config(
     socket_path: str | None,
     socket_name: str | None,
@@ -167,7 +181,15 @@ def _rust_server(
 ) -> t.Any:
     connection_kind = os.getenv("LIBTMUX_RUST_CONNECTION_KIND")
     server_kind = os.getenv("LIBTMUX_RUST_SERVER_KIND")
-    key = (socket_name, socket_path, colors, connection_kind, server_kind)
+    control_autostart = _env_bool("LIBTMUX_RUST_CONTROL_AUTOSTART")
+    key = (
+        socket_name,
+        socket_path,
+        colors,
+        connection_kind,
+        server_kind,
+        control_autostart,
+    )
     server = _RUST_SERVER_CACHE.get(key)
     if server is None:
         from libtmux import _rust as rust_backend
@@ -177,13 +199,16 @@ def _rust_server(
             kwargs["connection_kind"] = connection_kind
         if server_kind:
             kwargs["server_kind"] = server_kind
+        if control_autostart is not None:
+            kwargs["control_autostart"] = control_autostart
         with libtmux_trace.span(
             "rust_server_init",
             layer="python",
             socket_name=socket_name,
             socket_path=socket_path,
             connection_kind=connection_kind,
             server_kind=server_kind,
+            control_autostart=control_autostart,
         ):
             server = rust_backend.Server(
                 socket_path=socket_path,
@@ -223,6 +248,7 @@ def _rust_cmd_result(
 
     connection_kind = os.getenv("LIBTMUX_RUST_CONNECTION_KIND")
     server_kind = os.getenv("LIBTMUX_RUST_SERVER_KIND")
+    control_autostart = _env_bool("LIBTMUX_RUST_CONTROL_AUTOSTART")
     with libtmux_trace.span(
         "rust_cmd_result",
         layer="python",
@@ -232,13 +258,21 @@ def _rust_cmd_result(
         config_file=config_file,
         connection_kind=connection_kind,
         server_kind=server_kind,
+        control_autostart=control_autostart,
     ):
         if connection_kind in {"bin", "tmux-bin"} and config_file:
             cmd_parts = ["-f", config_file, *cmd_parts]
             config_file = None
 
         server = _rust_server(socket_name, socket_path, colors)
-        key = (socket_name, socket_path, colors, connection_kind, server_kind)
+        key = (
+            socket_name,
+            socket_path,
+            colors,
+            connection_kind,
+            server_kind,
+            control_autostart,
+        )
         if config_file:
             loaded = _RUST_SERVER_CONFIG.setdefault(key, set())
             if config_file not in loaded:

src/libtmux/pytest_plugin.py

@@ -13,6 +13,7 @@
 import pytest
 
 from libtmux import exc
+from libtmux.common import _rust_server
 from libtmux.server import Server
 from libtmux.test.constants import TEST_SESSION_PREFIX
 from libtmux.test.random import get_test_session_name, namer
@@ -24,6 +25,13 @@
 USING_ZSH = "zsh" in os.getenv("SHELL", "")
 
 
+def _env_truthy(name: str) -> bool:
+    value = os.getenv(name)
+    if value is None:
+        return False
+    return value.strip().lower() in {"1", "true", "yes", "on"}
+
+
 @pytest.fixture(scope="session")
 def home_path(tmp_path_factory: pytest.TempPathFactory) -> pathlib.Path:
     """Temporary `/home/` path."""
@@ -141,8 +149,25 @@ def server(
         >>> result.assert_outcomes(passed=1)
     """
     server = Server(socket_name=f"libtmux_test{next(namer)}")
+    rust_refresh = None
+    rust_server = None
+
+    if os.getenv("LIBTMUX_BACKEND") == "rust" and _env_truthy(
+        "LIBTMUX_RUST_CONTROL_MODE"
+    ):
+        socket_path = (
+            str(server.socket_path)
+            if isinstance(server.socket_path, pathlib.Path)
+            else server.socket_path
+        )
+        rust_server = _rust_server(server.socket_name, socket_path, server.colors)
+        rust_refresh = rust_server.subscribe(10)
+        setattr(server, "_rust_refresh", rust_refresh)
 
     def fin() -> None:
+        if rust_server is not None:
+            with contextlib.suppress(Exception):
+                rust_server.close()
         server.kill()
 
     request.addfinalizer(fin)