feat: opt-in engine-ops chained plan tools and watchers

pyproject.toml

@@ -111,6 +111,13 @@ lint = [
 requires = ["hatchling"]
 build-backend = "hatchling.build"
 
+# Development pin only (this branch): the chained plan tools + watchers
+# consume libtmux's unreleased experimental MCP adapter
+# (libtmux.experimental.mcp / .engines). Points libtmux at the engine-ops
+# branch so those APIs are importable. Not for release.
+[tool.uv.sources]
+libtmux = { git = "https://github.com/tmux-python/libtmux.git", branch = "engine-ops" }
+
 [tool.uv.exclude-newer-package]
 # git-pull packages release in lockstep with their workspaces, so a
 # fresh release blocking on the 3-day cooldown blocks every

src/libtmux_mcp/server.py

@@ -279,9 +279,14 @@ async def _lifespan(_app: FastMCP) -> t.AsyncIterator[None]:
     if shutil.which("tmux") is None:
         msg = "tmux binary not found on PATH"
         raise RuntimeError(msg)
+    # Start/stop the opt-in engine-ops control-mode engine (no-op when off).
+    from libtmux_mcp.tools import engine_watch
+
+    await engine_watch.astartup()
     try:
         yield
     finally:
+        await engine_watch.ashutdown()
         _gc_mcp_buffers(_server_cache)
         _server_cache.clear()
 

src/libtmux_mcp/tools/__init__.py

@@ -13,6 +13,8 @@ def register_tools(mcp: FastMCP) -> None:
     from libtmux_mcp.tools import (
         batch_tools,
         buffer_tools,
+        engine_plan,
+        engine_watch,
         env_tools,
         hook_tools,
         option_tools,
@@ -33,3 +35,7 @@ def register_tools(mcp: FastMCP) -> None:
     wait_for_tools.register(mcp)
     buffer_tools.register(mcp)
     hook_tools.register(mcp)
+    # Opt-in (LIBTMUX_MCP_ENGINE_OPS=1): chained plan tools + watchers over
+    # engine-ops. engine_watch's engine is started/closed by the server lifespan.
+    engine_plan.register(mcp)
+    engine_watch.register(mcp)

src/libtmux_mcp/tools/engine_plan.py

@@ -0,0 +1,65 @@
+"""Chained plan tools from libtmux's experimental engine-ops adapter.
+
+Opt-in (set ``LIBTMUX_MCP_ENGINE_OPS=1``). When enabled, this registers the
+plan tier of libtmux's experimental MCP adapter onto the existing server:
+``preview_plan`` / ``explain_plan`` / ``result_schema`` (readonly) and
+``execute_plan`` (mutating). An agent chains multiple tmux operations that
+*fold* into a few ``tmux a ; b`` dispatches, learning a step's result schema and
+resolving forward-reference ids through the returned ``bindings`` map before
+composing the next step.
+
+The plan tools drive a subprocess engine bound to the target tmux server
+(``LIBTMUX_SOCKET`` or the default socket), so no persistent connection is held.
+The op-tier tools they reference are registered hidden behind the ``per-op`` tag.
+The engine-ops adapter is unreleased (``libtmux.experimental``); this module is
+usable only with the branch pin in ``pyproject.toml``.
+"""
+
+from __future__ import annotations
+
+import os
+import typing as t
+
+if t.TYPE_CHECKING:
+    from fastmcp import FastMCP
+
+#: Environment flag that opts into the engine-ops tiers (default off).
+ENGINE_OPS_ENV = "LIBTMUX_MCP_ENGINE_OPS"
+
+_TRUE = frozenset({"1", "true", "yes", "on"})
+
+
+def enabled() -> bool:
+    """Return whether the engine-ops tiers are opted in via the environment.
+
+    Examples
+    --------
+    >>> import os
+    >>> _prev = os.environ.pop(ENGINE_OPS_ENV, None)
+    >>> enabled()
+    False
+    >>> os.environ[ENGINE_OPS_ENV] = "1"
+    >>> enabled()
+    True
+    >>> _ = os.environ.pop(ENGINE_OPS_ENV, None)
+    >>> os.environ.update({ENGINE_OPS_ENV: _prev} if _prev is not None else {})
+    """
+    return os.environ.get(ENGINE_OPS_ENV, "").strip().lower() in _TRUE
+
+
+def register(mcp: FastMCP) -> None:
+    """Register the chained plan tools onto *mcp* when opted in (else a no-op)."""
+    if not enabled():
+        return
+    from libtmux.experimental.engines import AsyncSubprocessEngine
+    from libtmux.experimental.mcp.fastmcp_adapter import register_plan_tools
+    from libtmux.experimental.mcp.registry import OperationToolRegistry
+
+    from libtmux_mcp._utils import _get_server
+
+    engine = AsyncSubprocessEngine.for_server(_get_server())
+    # The plan tools read the registry for result_schema and deserialize ops via
+    # LazyPlan.from_list for execute_plan -- they need the registry, not the
+    # per-op tool surface, so the op_* tools are intentionally not registered
+    # (they would be un-hidden by libtmux-mcp's safety-tag gate and clutter it).
+    register_plan_tools(mcp, engine, is_async=True, registry=OperationToolRegistry())

src/libtmux_mcp/tools/engine_watch.py

@@ -0,0 +1,92 @@
+"""Watcher tools from libtmux's experimental engine-ops event stream.
+
+Opt-in via the same ``LIBTMUX_MCP_ENGINE_OPS=1`` flag as
+:mod:`libtmux_mcp.tools.engine_plan`. When enabled this registers libtmux's
+control-mode event/monitor tools onto the existing server:
+
+- ``wait_for_output`` -- a needle-free settle monitor: it folds a pane's live
+  ``%output`` and returns when the pane goes quiet after a command, reporting
+  whether the pane's process exited. It replaces sleep-then-``capture_pane``
+  polling.
+- ``watch_events`` (push mode) and/or the ``tmux://events`` resource +
+  ``poll_events`` cursor (pull mode) over the control-mode notification stream.
+
+These require a persistent ``tmux -C`` control connection, so this module holds
+one process-global :class:`~libtmux.experimental.engines.AsyncControlModeEngine`
+bound to the target server; :func:`astartup` / :func:`ashutdown` open and close
+it from the server lifespan. Streaming is non-blocking: the engine's single
+reader fans out to bounded per-subscriber queues (drop-oldest), and the settle
+monitor pulls frames with a bounded wait -- consumers never block the producer.
+
+The engine-ops adapter is unreleased (``libtmux.experimental``); this module is
+usable only with the branch pin in ``pyproject.toml``.
+"""
+
+from __future__ import annotations
+
+import os
+import typing as t
+
+from libtmux_mcp.tools.engine_plan import ENGINE_OPS_ENV, enabled
+
+if t.TYPE_CHECKING:
+    from fastmcp import FastMCP
+    from libtmux.experimental.engines import AsyncControlModeEngine
+
+__all__ = ["ENGINE_OPS_ENV", "ashutdown", "astartup", "enabled", "register"]
+
+#: The single control-mode engine backing the watcher tools, or ``None`` when
+#: the engine-ops tier is not opted in. Opened by :func:`astartup`, closed by
+#: :func:`ashutdown`.
+_engine: AsyncControlModeEngine | None = None
+
+
+def register(mcp: FastMCP) -> None:
+    """Register the watcher tools onto *mcp* when opted in (else a no-op).
+
+    Builds one control-mode engine bound to the target server and registers the
+    event/monitor tools; the engine is started and closed by :func:`astartup` /
+    :func:`ashutdown` from the server lifespan.
+    """
+    global _engine
+    if not enabled():
+        return
+    from libtmux.experimental.engines import AsyncControlModeEngine
+    from libtmux.experimental.mcp.events import register_events
+
+    from libtmux_mcp._utils import _get_server
+
+    engine = AsyncControlModeEngine.for_server(_get_server())
+    # os.environ.get returns str; register_events wants the EventMode/EventSource
+    # literals, so narrow (an out-of-range value registers only the monitor).
+    mode = t.cast("t.Any", os.environ.get("LIBTMUX_MCP_EVENTS", "push"))
+    source = t.cast("t.Any", os.environ.get("LIBTMUX_MCP_EVENT_SOURCE", "subscription"))
+    register_events(mcp, engine, mode=mode, source=source)
+    _engine = engine
+
+
+async def astartup() -> None:
+    """Start the control-mode engine on the server loop (a no-op when off).
+
+    A ``list-sessions`` probe that doubles as the engine's first use: it spawns
+    ``tmux -C`` and launches the reader task, so ``watch_events`` / ``poll_events``
+    subscribe to a live stream rather than a dead one.
+    """
+    if _engine is None:
+        return
+    from libtmux.experimental.engines.base import CommandRequest
+
+    try:
+        await _engine.run(CommandRequest.from_args("list-sessions"))
+    except Exception as error:
+        msg = f"tmux control-mode engine preflight failed: {error}"
+        raise RuntimeError(msg) from error
+
+
+async def ashutdown() -> None:
+    """Close the control-mode engine on shutdown (a no-op when off)."""
+    global _engine
+    if _engine is None:
+        return
+    engine, _engine = _engine, None
+    await engine.aclose()

tests/test_engine_plan.py

@@ -0,0 +1,124 @@
+"""Tests for the opt-in engine-ops chained plan tools."""
+
+from __future__ import annotations
+
+import asyncio
+import typing as t
+
+import pytest
+from fastmcp import Client, FastMCP
+from libtmux.experimental.ops import LazyPlan, SendKeys, SplitWindow
+from libtmux.experimental.ops._types import WindowId
+
+from libtmux_mcp.tools import engine_plan
+
+if t.TYPE_CHECKING:
+    from collections.abc import Coroutine
+
+_T = t.TypeVar("_T")
+
+
+def _forward_ref_plan_ops() -> list[dict[str, t.Any]]:
+    """Serialize a 2-op plan whose send-keys targets the split's new pane."""
+    plan = LazyPlan()
+    pane = plan.add(SplitWindow(target=WindowId("@1")))
+    plan.add(SendKeys(target=pane, keys="vim"))
+    return plan.to_list()
+
+
+def _run(coro: Coroutine[t.Any, t.Any, _T]) -> _T:
+    """Run *coro* synchronously (the repo's async-test convention)."""
+    return asyncio.run(coro)
+
+
+def _tool_names(mcp: FastMCP) -> set[str]:
+    """List the visible tool names on *mcp* via an in-process client."""
+
+    async def _go() -> set[str]:
+        async with Client(mcp) as client:
+            return {tool.name for tool in await client.list_tools()}
+
+    return _run(_go())
+
+
+def _registered(*, enabled: bool, monkeypatch: pytest.MonkeyPatch) -> FastMCP:
+    """Return a fresh server with engine_plan registered (or not) per *enabled*."""
+    if enabled:
+        monkeypatch.setenv(engine_plan.ENGINE_OPS_ENV, "1")
+    else:
+        monkeypatch.delenv(engine_plan.ENGINE_OPS_ENV, raising=False)
+    mcp = FastMCP("test")
+    engine_plan.register(mcp)
+    return mcp
+
+
+def test_enabled_reads_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    """enabled() reflects the opt-in environment flag."""
+    monkeypatch.delenv(engine_plan.ENGINE_OPS_ENV, raising=False)
+    assert engine_plan.enabled() is False
+    monkeypatch.setenv(engine_plan.ENGINE_OPS_ENV, "1")
+    assert engine_plan.enabled() is True
+
+
+def test_register_is_noop_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None:
+    """With the flag off, no engine-ops tools are registered."""
+    mcp = _registered(enabled=False, monkeypatch=monkeypatch)
+    assert "preview_plan" not in _tool_names(mcp)
+
+
+class _ChainToolCase(t.NamedTuple):
+    """A chain-tier tool that must appear when the flag is on."""
+
+    test_id: str
+    tool: str
+
+
+_CHAIN_TOOL_CASES: tuple[_ChainToolCase, ...] = (
+    _ChainToolCase("preview_plan", "preview_plan"),
+    _ChainToolCase("explain_plan", "explain_plan"),
+    _ChainToolCase("result_schema", "result_schema"),
+    _ChainToolCase("execute_plan", "execute_plan"),
+)
+
+
+@pytest.mark.parametrize(
+    "case",
+    _CHAIN_TOOL_CASES,
+    ids=[c.test_id for c in _CHAIN_TOOL_CASES],
+)
+def test_chain_tool_registered_when_enabled(
+    case: _ChainToolCase,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Each chain-tier tool is registered when opted in."""
+    mcp = _registered(enabled=True, monkeypatch=monkeypatch)
+    assert case.tool in _tool_names(mcp)
+
+
+def test_no_per_op_tools_leak(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Only the plan tier is added -- the hidden op_* tools are not registered."""
+    mcp = _registered(enabled=True, monkeypatch=monkeypatch)
+    assert not any(name.startswith("op_") for name in _tool_names(mcp))
+
+
+def test_preview_and_explain_a_folded_forward_ref_plan(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The chain previews a forward-ref plan and explains how it folds."""
+    mcp = _registered(enabled=True, monkeypatch=monkeypatch)
+    ops = _forward_ref_plan_ops()
+
+    async def _go() -> tuple[t.Any, t.Any]:
+        async with Client(mcp) as client:
+            preview = await client.call_tool("preview_plan", {"operations": ops})
+            explain = await client.call_tool(
+                "explain_plan",
+                {"operations": ops, "planner": "marked"},
+            )
+        return preview.data, explain.data
+
+    preview, explain = _run(_go())
+    # the send-keys targets the not-yet-created pane, so preview cannot resolve it
+    assert preview["ok"] is False
+    # under the marked planner the create + its decorate fold into one dispatch
+    assert explain["steps"][0]["reason"] == "marked-fold"