diff --git a/pyproject.toml b/pyproject.toml index e34e7bf6..b13363fd 100644 --- a/pyproject.toml +++ b/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 diff --git a/src/libtmux_mcp/server.py b/src/libtmux_mcp/server.py index f92a6404..4bf546fe 100644 --- a/src/libtmux_mcp/server.py +++ b/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() diff --git a/src/libtmux_mcp/tools/__init__.py b/src/libtmux_mcp/tools/__init__.py index 7a72f9ab..f6762f1d 100644 --- a/src/libtmux_mcp/tools/__init__.py +++ b/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) diff --git a/src/libtmux_mcp/tools/engine_plan.py b/src/libtmux_mcp/tools/engine_plan.py new file mode 100644 index 00000000..cf3eac10 --- /dev/null +++ b/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()) diff --git a/src/libtmux_mcp/tools/engine_watch.py b/src/libtmux_mcp/tools/engine_watch.py new file mode 100644 index 00000000..7f9f4dbb --- /dev/null +++ b/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() diff --git a/tests/test_engine_plan.py b/tests/test_engine_plan.py new file mode 100644 index 00000000..bab9d154 --- /dev/null +++ b/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" diff --git a/tests/test_engine_watch.py b/tests/test_engine_watch.py new file mode 100644 index 00000000..20df306a --- /dev/null +++ b/tests/test_engine_watch.py @@ -0,0 +1,139 @@ +"""Tests for the opt-in engine-ops watcher tools.""" + +from __future__ import annotations + +import asyncio +import typing as t + +import pytest +from fastmcp import Client, FastMCP + +from libtmux_mcp.tools import engine_watch + +if t.TYPE_CHECKING: + from collections.abc import Coroutine + + from libtmux.session import Session + +_T = t.TypeVar("_T") + + +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(mode: str | None, monkeypatch: pytest.MonkeyPatch) -> FastMCP: + """Return a fresh server with engine_watch registered under *mode* (off=None).""" + if mode is None: + monkeypatch.delenv(engine_watch.ENGINE_OPS_ENV, raising=False) + else: + monkeypatch.setenv(engine_watch.ENGINE_OPS_ENV, "1") + monkeypatch.setenv("LIBTMUX_MCP_EVENTS", mode) + mcp = FastMCP("test") + engine_watch.register(mcp) + return mcp + + +class _ModeCase(t.NamedTuple): + """An event mode and the watcher tools it must register.""" + + test_id: str + mode: str + tools: frozenset[str] + + +_MODE_CASES: tuple[_ModeCase, ...] = ( + _ModeCase("push", "push", frozenset({"wait_for_output", "watch_events"})), + _ModeCase("pull", "pull", frozenset({"wait_for_output", "poll_events"})), + _ModeCase( + "both", + "both", + frozenset({"wait_for_output", "watch_events", "poll_events"}), + ), +) + + +@pytest.mark.parametrize("case", _MODE_CASES, ids=[c.test_id for c in _MODE_CASES]) +def test_watcher_tools_registered_per_mode( + case: _ModeCase, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Each event mode registers its watcher tools; wait_for_output is always on.""" + tools = _tool_names(_registered(case.mode, monkeypatch)) + assert case.tools <= tools + _run(engine_watch.ashutdown()) # engine never started here -> harmless reset + + +def test_register_is_noop_when_disabled(monkeypatch: pytest.MonkeyPatch) -> None: + """With the flag off, no watcher tools are registered.""" + assert "wait_for_output" not in _tool_names(_registered(None, monkeypatch)) + + +def test_no_per_op_tools_leak(monkeypatch: pytest.MonkeyPatch) -> None: + """Only the watcher tools are added -- no op_* clutter.""" + assert not any( + n.startswith("op_") for n in _tool_names(_registered("both", monkeypatch)) + ) + + +def test_lifecycle_is_noop_when_off() -> None: + """astartup/ashutdown are safe no-ops when the tier was never registered.""" + engine_watch._engine = None + _run(engine_watch.astartup()) + _run(engine_watch.ashutdown()) + + +def test_wait_for_output_settles_live(session: Session) -> None: + """The registered monitor folds a real pane's output and settles when quiet.""" + from libtmux.experimental.engines import AsyncControlModeEngine + from libtmux.experimental.engines.base import CommandRequest + from libtmux.experimental.mcp.events import register_events + + server = session.server + pane = session.active_window.active_pane + assert pane is not None and pane.pane_id is not None + pane_id = pane.pane_id + + async def _go() -> t.Any: + async with AsyncControlModeEngine.for_server(server) as engine: + mcp = FastMCP("test") + register_events(mcp, engine, mode="push") + async with Client(mcp) as client: + + async def _produce() -> None: + await asyncio.sleep(0.3) # let the monitor subscribe first + await engine.run( + CommandRequest.from_args( + "send-keys", + "-t", + pane_id, + "echo WATCH_OK", + "Enter", + ), + ) + + producer = asyncio.ensure_future(_produce()) + try: + result = await client.call_tool( + "wait_for_output", + {"target": pane_id, "settle_ms": 400, "timeout": 10.0}, + ) + finally: + await producer + return result.data + + data = _run(_go()) + assert data.pane_id == pane_id + assert data.reason in ("settled", "byte_cap") + assert "WATCH_OK" in data.captured_text diff --git a/uv.lock b/uv.lock index a0730ab6..1e0bf09c 100644 --- a/uv.lock +++ b/uv.lock @@ -1177,11 +1177,7 @@ wheels = [ [[package]] name = "libtmux" version = "0.61.0" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/72/c2/6abbac4420c35b3f1140c72607117236b160173f74fd86941f99b28375d5/libtmux-0.61.0.tar.gz", hash = "sha256:2d6081081a629b9236a36a64f874667533811d2ce5a1b0caa9c821f4d9d2e618", size = 546122, upload-time = "2026-07-04T12:57:36.401Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/73/14/934d0d9076d1d46fcc71f3397daea7921363781674337084b15d1bbcbf84/libtmux-0.61.0-py3-none-any.whl", hash = "sha256:b9315db6600757f840a8f16438725103e579aaa4be3c32728c105f2c284330df", size = 118059, upload-time = "2026-07-04T12:57:34.598Z" }, -] +source = { git = "https://github.com/tmux-python/libtmux.git?branch=engine-ops#22557d9815bb332bfa0ef96fc1d7050f953aeba8" } [[package]] name = "libtmux-mcp" @@ -1244,7 +1240,7 @@ testing = [ [package.metadata] requires-dist = [ { name = "fastmcp", specifier = ">=3.4.2,<4.0.0" }, - { name = "libtmux", specifier = ">=0.61.0,<1.0" }, + { name = "libtmux", git = "https://github.com/tmux-python/libtmux.git?branch=engine-ops" }, ] [package.metadata.requires-dev]