Experiment: Control mode

.github/workflows/tests.yml

@@ -83,6 +83,13 @@ jobs:
           COV_CORE_SOURCE: .
           COV_CORE_CONFIG: .coveragerc
           COV_CORE_DATAFILE: .coverage.eager
+      - name: Test with pytest (control engine)
+        continue-on-error: ${{ matrix.tmux-version == 'master' }}
+        run: |
+          sudo apt install libevent-2.1-7
+          export PATH=$HOME/tmux-builds/tmux-${{ matrix.tmux-version }}/bin:$PATH
+          tmux -V
+          uv run py.test --engine=control -n auto --verbose
       - uses: codecov/codecov-action@v5
         with:
           token: ${{ secrets.CODECOV_TOKEN }}

CHANGES

@@ -34,6 +34,61 @@ $ uvx --from 'libtmux' --prerelease allow python
 
 _Upcoming changes will be written here._
 
+### Features
+
+#### Control mode engine (experimental) (#605)
+- Control-mode–first engine stack with structured `CommandResult`, notification parsing
+  (%layout-change, unlinked window add/close/rename, client session change/detach, paste buffer),
+  and stats, while keeping the public `cmd` API returning `tmux_cmd`.
+- Internal control session is automatically filtered from `Server.sessions` /
+  `Server.has_session()`; advanced users can inspect all sessions via
+  `Server._sessions_all()`.
+- `ControlModeEngine` accepts `internal_session_name` (default `libtmux_control_mode`)
+  and `control_session` for advanced connection strategies.
+
+Example:
+
+```python
+from libtmux._internal.engines.control_mode import ControlModeEngine
+from libtmux.server import Server
+
+engine = ControlModeEngine(command_timeout=5)
+server = Server(engine=engine)
+session = server.new_session(session_name="ctrl")
+for notif in engine.iter_notifications(timeout=0.1):
+    print(notif.kind, notif.data)
+```
+
+#### Convenience APIs (#605)
+- `Server.connect()` returns an existing session or creates a new detached one,
+  working with both subprocess and control-mode engines.
+
+#### Diagnostics and resilience (#605)
+- Bounded notification queue with drop counting; exposed via engine stats.
+- Expanded exceptions: `ControlModeTimeout`, `ControlModeProtocolError`,
+  `ControlModeConnectionError`, `SubprocessTimeout`; documented retry/timeout behaviour.
+- Control-mode capture-pane trims trailing whitespace-only lines to mirror subprocess
+  semantics; first capture after send-keys briefly retries to avoid prompt races.
+
+#### Testing utilities (#605)
+- Control sandbox fixture provides a hermetic control-mode tmux server (isolated HOME,
+  TMUX_TMPDIR, unique socket); handy for integration-style tests.
+  Example:
+
+  ```python
+  @pytest.mark.engines(["control"])
+  def test_control_sandbox(control_sandbox):
+      with control_sandbox as server:
+          out = server.cmd("display-message", "-p", "hi")
+          assert out.stdout == ["hi"]
+  ```
+
+### Compatibility
+
+- Control mode internal session filtering is engine-driven; callers no longer need
+  to manually exclude `libtmux_control_mode`. APIs stay unchanged; additional metadata
+  is attached for advanced users. (#605)
+
 ## libtmux 0.50.1 (2025-12-06)
 
 ### Documentation (#612)

conftest.py

@@ -10,12 +10,17 @@
 
 from __future__ import annotations
 
+import contextlib
+import pathlib
 import shutil
+import subprocess
 import typing as t
+import uuid
 
 import pytest
 from _pytest.doctest import DoctestItem
 
+from libtmux._internal.engines.control_protocol import CommandContext, ControlProtocol
 from libtmux.pane import Pane
 from libtmux.pytest_plugin import USING_ZSH
 from libtmux.server import Server
@@ -73,3 +78,141 @@ def setup_session(
     """Session-level test configuration for pytest."""
     if USING_ZSH:
         request.getfixturevalue("zshrc")
+
+
+# ---------------------------------------------------------------------------
+# Control-mode sandbox helper
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+@contextlib.contextmanager
+def control_sandbox(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path_factory: pytest.TempPathFactory,
+) -> t.Iterator[Server]:
+    """Provide an isolated control-mode server for a test.
+
+    - Creates a unique tmux socket name per invocation
+    - Isolates HOME and TMUX_TMPDIR under a per-test temp directory
+    - Clears TMUX env var to avoid inheriting user sessions
+    - Uses ControlModeEngine; on exit, kills the server best-effort
+    """
+    socket_name = f"libtmux_test_{uuid.uuid4().hex[:8]}"
+    base = tmp_path_factory.mktemp("ctrl_sandbox")
+    home = base / "home"
+    tmux_tmpdir = base / "tmux"
+    home.mkdir()
+    tmux_tmpdir.mkdir()
+
+    monkeypatch.setenv("HOME", str(home))
+    monkeypatch.setenv("TMUX_TMPDIR", str(tmux_tmpdir))
+    monkeypatch.delenv("TMUX", raising=False)
+
+    from libtmux._internal.engines.control_mode import ControlModeEngine
+
+    server = Server(socket_name=socket_name, engine=ControlModeEngine())
+
+    try:
+        yield server
+    finally:
+        with contextlib.suppress(Exception):
+            server.kill()
+
+
+@pytest.fixture
+def control_client_logs(
+    control_sandbox: t.ContextManager[Server],
+    tmp_path_factory: pytest.TempPathFactory,
+) -> t.Iterator[tuple[subprocess.Popen[str], ControlProtocol]]:
+    """Spawn a raw tmux -C client against the sandbox and log stdout/stderr."""
+    base = tmp_path_factory.mktemp("ctrl_logs")
+    stdout_path = base / "control_stdout.log"
+    stderr_path = base / "control_stderr.log"
+
+    with control_sandbox as server:
+        cmd = [
+            "tmux",
+            "-L",
+            server.socket_name or "",
+            "-C",
+            "attach-session",
+            "-t",
+            "ctrltest",
+        ]
+        # Ensure ctrltest exists
+        server.cmd("new-session", "-d", "-s", "ctrltest")
+        stdout_path.open("w+", buffering=1)
+        stderr_f = stderr_path.open("w+", buffering=1)
+        proc = subprocess.Popen(
+            cmd,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=stderr_f,
+            text=True,
+            bufsize=1,
+        )
+        proto = ControlProtocol()
+        # tmux -C will emit a %begin/%end pair for this initial attach-session;
+        # queue a matching context so the parser has a pending command.
+        proto.register_command(CommandContext(argv=list(cmd)))
+        try:
+            yield proc, proto
+        finally:
+            with contextlib.suppress(Exception):
+                if proc.stdin:
+                    proc.stdin.write("kill-session -t ctrltest\n")
+                    proc.stdin.flush()
+            proc.terminate()
+            proc.wait(timeout=2)
+
+
+def pytest_addoption(parser: pytest.Parser) -> None:
+    """Add CLI options for selecting tmux engine."""
+    parser.addoption(
+        "--engine",
+        action="store",
+        default="subprocess",
+        choices=["subprocess", "control"],
+        help="Select tmux engine for fixtures (default: subprocess).",
+    )
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    """Register custom markers."""
+    config.addinivalue_line(
+        "markers",
+        (
+            "engines(names): run the test once for each engine in 'names' "
+            "(e.g. ['control', 'subprocess'])."
+        ),
+    )
+
+
+def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
+    """Parametrize engine_name when requested by tests."""
+    if "engine_name" in metafunc.fixturenames:
+        marker = metafunc.definition.get_closest_marker("engines")
+        if marker:
+            params = list(marker.args[0])
+        else:
+            params = [metafunc.config.getoption("--engine")]
+        metafunc.parametrize("engine_name", params, indirect=True)
+
+
+def pytest_collection_modifyitems(
+    config: pytest.Config,
+    items: list[pytest.Item],
+) -> None:
+    """Filter out doctests when running with control engine.
+
+    Remove doctests from collection to avoid pytest's _use_item_location
+    bug: DoctestItem.reportinfo() returns None lineno for fixture doctests,
+    which triggers assertion failure in _pytest/reports.py:420 when skipped.
+    """
+    engine_opt = config.getoption("--engine", default="subprocess")
+    if engine_opt != "control":
+        return
+
+    # Filter out DoctestItems - can't use skip markers due to pytest bug
+    items[:] = [item for item in items if not isinstance(item, DoctestItem)]

docs/api/index.md

@@ -4,6 +4,12 @@
 
 # API Reference
 
+:::{note}
+Looking for the new control-mode engine? See {ref}`control-mode` for an
+experimental, protocol-focused entry point that still preserves the public
+``tmux_cmd`` return type.
+:::
+
 ```{toctree}
 
 properties

docs/pytest-plugin/index.md

@@ -137,6 +137,53 @@ def set_home(
     monkeypatch.setenv("HOME", str(user_path))
 ```
 
+## Selecting tmux engines (experimental)
+
+Fixtures can run against different execution engines. By default the
+`subprocess` engine is used. You can choose control mode globally:
+
+```console
+$ pytest --engine=control
+```
+
+Or per-test via the `engines` marker (uses parametrization) and the `engine_name`
+fixture:
+
+```python
+import pytest
+
+@pytest.mark.engines(["subprocess", "control"])
+def test_my_flow(server, engine_name):
+    # server uses the selected engine, engine_name reflects the current one
+    assert engine_name in {"subprocess", "control"}
+    assert server.is_alive()
+```
+
+`TestServer` also respects the selected engine. Control mode is experimental and
+its APIs may change between releases.
+
+### Control sandbox fixture (experimental)
+
+Use ``control_sandbox`` when you need a hermetic control-mode server for a test:
+
+```python
+import typing as t
+import pytest
+from libtmux.server import Server
+
+@pytest.mark.engines(["control"])
+def test_control_sandbox(control_sandbox: t.ContextManager[Server]):
+    with control_sandbox as server:
+        session = server.new_session(session_name="sandbox", attach=False)
+        out = server.cmd("display-message", "-p", "hi")
+        assert out.stdout == ["hi"]
+```
+
+The fixture:
+- Spins up a unique socket name and isolates ``HOME`` / ``TMUX_TMPDIR``
+- Clears inherited ``TMUX`` so it never attaches to the user's server
+- Uses ``ControlModeEngine`` and cleans up the server on exit
+
 ## Fixtures
 
 ```{eval-rst}

docs/quickstart.md

@@ -157,6 +157,23 @@ in your server object. `libtmux.Server(socket_name='mysocket')` is
 equivalent to `$ tmux -L mysocket`.
 :::
 
+### Optional: Control mode (experimental)
+
+Control mode keeps a persistent tmux client open and streams
+notifications. Enable it by injecting a control-mode engine:
+
+```python
+from libtmux._internal.engines.control_mode import ControlModeEngine
+from libtmux.server import Server
+
+engine = ControlModeEngine()
+server = Server(engine=engine)
+session = server.new_session(session_name="ctrl")
+print(session.name)
+```
+
+See {ref}`control-mode` for details, caveats, and notification handling.
+
 `server` is now a living object bound to the tmux server's Sessions,
 Windows and Panes.
 
@@ -263,6 +280,41 @@ Session($1 foo)
 
 to give us a `session` object to play with.
 
+## Connect to or create a session
+
+A simpler approach is to use {meth}`Server.connect()`, which returns an existing
+session or creates it if it doesn't exist:
+
+```python
+>>> session = server.connect('my_project')
+>>> session.name
+'my_project'
+
+>>> # Calling again returns the same session
+>>> session2 = server.connect('my_project')
+>>> session2.session_id == session.session_id
+True
+```
+
+This is particularly useful for:
+
+- Development workflows where you want to reuse existing sessions
+- Scripts that should create a session on first run, then reattach
+- Working with both subprocess and control-mode engines transparently
+
+Compare with the traditional approach:
+
+```python
+>>> # Traditional: check then create
+>>> if server.has_session('my_project'):
+...     session = server.sessions.get(session_name='my_project')
+... else:
+...     session = server.new_session('my_project')
+
+>>> # Simpler with connect()
+>>> session = server.connect('my_project')
+```
+
 ## Playing with our tmux session
 
 We now have access to `session` from above with all of the methods