Typed operations & engines: spine, 6 engines, plans, models, facades (#689)

CHANGES

@@ -45,6 +45,34 @@ $ uvx --from 'libtmux' --prerelease allow python
 _Notes on the upcoming release will go here._
 <!-- END PLACEHOLDER - ADD NEW CHANGELOG ENTRIES BELOW THIS LINE -->
 
+### What's new
+
+#### Experimental operations and engines (#690)
+
+Operations describe tmux commands as data. Each renders its argv against a tmux
+version (dropping flags an older tmux cannot accept), adapts raw output into a
+typed result, and serializes to and from plain dicts -- all without a running
+tmux server. The set spans the read seam (``list-*``, ``has-session``,
+``display-message``, ``show-options``, ``show-buffer``) and the
+mutating/creating surface for panes, windows, the server, options, environment,
+hooks, and paste buffers. A registry-generated catalog on the {ref}`experimental`
+page always matches the code.
+
+Engines run those operations behind one protocol, so the same operation returns
+the same typed result whether it goes through a subprocess (the classic path
+that reproduces today's libtmux behavior), an in-memory simulator for tests and
+dry runs, a persistent ``tmux -C`` control connection, an async transport, or
+tmux's native binary peer protocol. Results never raise on construction;
+raising is opt-in via ``raise_for_status()``, and how a failed result is handled
+is each engine's policy.
+
+A {class}`~libtmux.experimental.ops.plan.LazyPlan` records operations and yields
+forward references so a later operation can target an object that does not exist
+yet, resolved against captured ids at execution time. How a plan becomes tmux
+dispatches is a pluggable {class}`~libtmux.experimental.ops.planner.Planner`
+(sequential, ``;``-folding, or ``{marked}``-folding), so dispatch strategies can
+be A/B tested against the same plan with identical results.
+
 ## libtmux 0.58.1 (2026-06-16)
 
 libtmux 0.58.1 restores compatibility with pytest 9.1. The bundled

docs/_ext/tmuxop.py

@@ -0,0 +1,120 @@
+"""Sphinx directive that renders the experimental operation catalog.
+
+``.. tmuxop-catalog::`` (or the MyST fenced form) walks
+:func:`libtmux.experimental.ops.catalog` and emits a table of operations with
+their scope, safety tier, result type, minimum tmux version, and summary. The
+operation registry is the single source of truth, so the rendered reference
+cannot drift from the code.
+
+Options
+-------
+``:scope:`` / ``:safety:``
+    Filter to one scope (``pane``/``window``/``session``/``server``/``client``)
+    or safety tier (``readonly``/``mutating``/``destructive``).
+``:primitive-only:``
+    Show only operations that wrap a single tmux command.
+
+This is the in-repo renderer; a full gp-sphinx ``tmuxop`` domain (cross-reference
+roles + an operations index) can later replace it under the same directive name.
+"""
+
+from __future__ import annotations
+
+import typing as t
+
+from docutils import nodes
+from docutils.parsers.rst import directives
+from sphinx.util import logging
+from sphinx.util.docutils import SphinxDirective
+
+if t.TYPE_CHECKING:
+    from collections.abc import Sequence
+
+    from sphinx.application import Sphinx
+
+logger = logging.getLogger(__name__)
+
+_HEADERS = ("Operation", "Command", "Scope", "Safety", "Result", "Min tmux", "Summary")
+
+
+def _row(cells: Sequence[str]) -> nodes.row:
+    """Build a docutils table row from string cells."""
+    row = nodes.row()
+    for cell in cells:
+        entry = nodes.entry()
+        entry += nodes.paragraph(text=cell)
+        row += entry
+    return row
+
+
+def _table(headers: Sequence[str], rows: Sequence[Sequence[str]]) -> nodes.table:
+    """Build a simple docutils table."""
+    table = nodes.table()
+    tgroup = nodes.tgroup(cols=len(headers))
+    table += tgroup
+    for _ in headers:
+        tgroup += nodes.colspec(colwidth=1)
+    thead = nodes.thead()
+    thead += _row(headers)
+    tgroup += thead
+    tbody = nodes.tbody()
+    for row in rows:
+        tbody += _row(row)
+    tgroup += tbody
+    return table
+
+
+class TmuxopCatalogDirective(SphinxDirective):
+    """Render the operation catalog as a table."""
+
+    has_content = False
+    option_spec: t.ClassVar[dict[str, t.Any]] = {
+        "scope": directives.unchanged,
+        "safety": directives.unchanged,
+        "primitive-only": directives.flag,
+    }
+
+    def run(self) -> list[nodes.Node]:
+        """Build the catalog table from the operation registry."""
+        from libtmux.experimental.ops import catalog
+
+        entries = catalog()
+        scope = self.options.get("scope")
+        safety = self.options.get("safety")
+        if scope:
+            entries = [entry for entry in entries if entry.scope == scope]
+        if safety:
+            entries = [entry for entry in entries if entry.safety == safety]
+        if "primitive-only" in self.options:
+            entries = [entry for entry in entries if entry.primitive]
+
+        if not entries:
+            logger.warning(
+                "tmuxop-catalog: no operations matched the given filters",
+                location=self.get_location(),
+            )
+            return []
+
+        rows = [
+            (
+                entry.kind,
+                entry.command,
+                entry.scope,
+                entry.safety,
+                entry.result_type,
+                entry.min_version or "-",
+                entry.summary,
+            )
+            for entry in entries
+        ]
+        return [_table(_HEADERS, rows)]
+
+
+def setup(app: Sphinx) -> dict[str, t.Any]:
+    """Register the directive."""
+    app.add_directive("tmuxop-catalog", TmuxopCatalogDirective)
+    return {
+        "version": "0.1",
+        "parallel_read_safe": True,
+        "parallel_write_safe": True,
+    }

docs/conf.py

@@ -15,6 +15,7 @@
 project_src = project_root / "src"
 
 sys.path.insert(0, str(project_src))
+sys.path.insert(0, str(cwd / "_ext"))
 
 # package data
 about: dict[str, str] = {}
@@ -34,6 +35,7 @@
         "sphinx_autodoc_api_style",
         "sphinx_autodoc_pytest_fixtures",
         "sphinx.ext.todo",
+        "tmuxop",
     ],
     intersphinx_mapping={
         "python": ("https://docs.python.org/", None),

docs/experimental.md

@@ -0,0 +1,136 @@
+(experimental)=
+
+# Experimental: operations & engines
+
+```{warning}
+Everything under {mod}`libtmux.experimental` is **not** covered by the
+versioning policy and may change or be removed between any two releases.
+```
+
+`libtmux.experimental` hosts an inert, typed *operation* substrate and the
+*engines* that execute it. An operation describes a tmux command (it renders
+argv, carries its result type and metadata, and serializes) without dispatching;
+an engine runs operations and returns typed results. The same operation returns
+the same typed result whether executed by a subprocess, an in-memory simulator,
+a persistent `tmux -C` control connection, or an async transport.
+
+See ``tmux-python/libtmux`` issue 689 for the operationalization plan.
+
+## Running an operation
+
+An operation is a value; ``run`` (or ``arun`` for async) hands it to an engine
+and returns the engine's typed result. Results never raise on construction --
+inspect ``ok``/``status``, or opt into raising with ``raise_for_status()``:
+
+```python
+>>> from libtmux.experimental.ops import HasSession, run
+>>> from libtmux.experimental.ops._types import SessionId
+>>> from libtmux.experimental.engines import ConcreteEngine
+>>> result = run(HasSession(target=SessionId("$0")), ConcreteEngine())
+>>> result.ok
+True
+>>> result.raise_for_status() is result
+True
+```
+
+How a *failed* result is treated is the engine's policy: the classic subprocess
+path raises in its facade to match today's libtmux behavior, while the newer
+engines hand the result back and let the caller decide.
+
+## Choosing an engine
+
+Every engine satisfies the same ``TmuxEngine`` (or ``AsyncTmuxEngine``)
+protocol, so swapping engines never changes an operation or its result type --
+only *how* and *where* the command runs.
+
+| Engine | Transport | Use it for |
+| --- | --- | --- |
+| ``SubprocessEngine`` | one ``tmux`` process per command | the classic path; reproduces today's libtmux behavior |
+| ``ConcreteEngine`` | in-memory, no tmux | tests and dry runs (deterministic, fabricated output) |
+| ``ControlModeEngine`` | a persistent ``tmux -C`` connection | many commands over one long-lived session |
+| ``ImsgEngine`` | tmux's native binary peer protocol | an opt-in easter egg |
+
+Each has an ``Async*`` counterpart (``AsyncSubprocessEngine``,
+``AsyncConcreteEngine``, ``AsyncControlModeEngine``) behind ``AsyncTmuxEngine``.
+Construct one directly, bind it to a live server with
+``SubprocessEngine.for_server(server)``, or select one by name from the engine
+registry:
+
+```python
+>>> from libtmux.experimental.engines import available_engines, create_engine
+>>> from libtmux.experimental.ops import HasSession, run
+>>> from libtmux.experimental.ops._types import SessionId
+>>> available_engines()
+('concrete', 'control_mode', 'imsg', 'subprocess')
+>>> engine = create_engine("concrete")
+>>> run(HasSession(target=SessionId("$0")), engine).status
+'complete'
+```
+
+## Lazy plans and planners
+
+A {class}`~libtmux.experimental.ops.plan.LazyPlan` records operations without
+running them, returning a forward *slot reference* for each created object so a
+later operation can target something that does not exist yet. ``execute``
+(or ``aexecute``) resolves those references against captured ids as it goes:
+
+```python
+>>> from libtmux.experimental.ops import LazyPlan, SplitWindow, SendKeys
+>>> from libtmux.experimental.ops._types import WindowId
+>>> from libtmux.experimental.engines import ConcreteEngine
+>>> plan = LazyPlan()
+>>> pane = plan.add(SplitWindow(target=WindowId("@1")))
+>>> _ = plan.add(SendKeys(target=pane, keys="echo hi", enter=True))
+>>> outcome = plan.execute(ConcreteEngine())
+>>> outcome.ok
+True
+>>> [r.status for r in outcome.results]
+['complete', 'complete']
+```
+
+Operations also compose with ``>>`` into a chain, which a plan can run as one
+dispatch when the members are chainable.
+
+*How* a plan turns into dispatches is a pluggable
+{class}`~libtmux.experimental.ops.planner.Planner`, so strategies can be A/B
+tested against the same plan:
+
+- ``SequentialPlanner`` -- one dispatch per operation (the default).
+- ``FoldingPlanner`` -- folds adjacent chainable operations into a single
+  ``;``-separated dispatch.
+- ``MarkedPlanner`` -- folds a "create then decorate the new pane" run into one
+  dispatch using tmux's ``{marked}`` register.
+
+Every planner produces the same per-operation result; they differ only in how
+many times tmux is invoked:
+
+```python
+>>> from libtmux.experimental.ops import LazyPlan, SplitWindow, SendKeys, FoldingPlanner
+>>> from libtmux.experimental.ops._types import WindowId
+>>> from libtmux.experimental.engines import ConcreteEngine
+>>> plan = LazyPlan()
+>>> pane = plan.add(SplitWindow(target=WindowId("@1")))
+>>> _ = plan.add(SendKeys(target=pane, keys="echo hi", enter=True))
+>>> plan.execute(ConcreteEngine(), planner=FoldingPlanner()).ok
+True
+```
+
+## Operation catalog
+
+The catalog below is generated from the operation registry, so it always matches
+the code.
+
+```{tmuxop-catalog}
+```
+
+### Read-only operations
+
+```{tmuxop-catalog}
+:safety: readonly
+```
+
+### Destructive operations
+
+```{tmuxop-catalog}
+:safety: destructive
+```

docs/index.md

@@ -102,6 +102,7 @@ api/index
 api/testing/index
 internals/index
 project/index
+experimental
 history
 migration
 glossary

docs/topics/automation_patterns.md

@@ -76,13 +76,18 @@ True
 
 ### Waiting for specific output
 
+> **Note:** This polls with `capture_pane` + `sleep` — correct for the
+> synchronous library. If you drive tmux through the libtmux MCP server, prefer
+> the event-backed `wait_for_output` tool instead: it folds live `%output` and
+> returns when the pane settles, with no polling.
+
 ```python
 >>> import time
 
 >>> monitor_window = session.new_window(window_name='monitor', attach=False)
 >>> monitor_pane = monitor_window.active_pane
 
->>> def wait_for_output(pane, text, timeout=5.0, poll_interval=0.1):
+>>> def wait_for_text(pane, text, timeout=5.0, poll_interval=0.1):
 ...     """Wait for specific text to appear in pane output."""
 ...     start = time.time()
 ...     while time.time() - start < timeout:
@@ -93,7 +98,7 @@ True
 ...     return False
 
 >>> monitor_pane.send_keys('sleep 0.2; echo "READY"')
->>> wait_for_output(monitor_pane, 'READY', timeout=2.0)
+>>> wait_for_text(monitor_pane, 'READY', timeout=2.0)
 True
 
 >>> # Clean up

fastmcp.json

@@ -0,0 +1,11 @@
+{
+  "$schema": "https://gofastmcp.com/public/schemas/fastmcp.json/v1.json",
+  "source": {
+    "type": "filesystem",
+    "path": "src/libtmux/experimental/mcp/__init__.py",
+    "entrypoint": "default_async_server"
+  },
+  "deployment": {
+    "transport": "stdio"
+  }
+}