From 59227d3a375a6f8f786b15d51ab36874a3112461 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 4 Jul 2026 12:41:54 -0500 Subject: [PATCH 01/13] Query(feat): Split-type forward-ref pane handles why: a fluent build needs to target a pane an earlier op will create. Splitting the concrete PaneRef into two handle types makes reading an id off a not-yet-created pane a static type error under mypy and ty, not a runtime surprise. what: - Add _PaneRefBase (shared verbs: cmd, split, do), ForwardPaneRef (no snapshot reads), and PaneRef (concrete: pane_id/active/pane) - split() records the create and returns a forward handle; do() threads a recorder into a fluent chain - Widen BoundPaneCommands.target to Target so a forward SlotRef flows - Cover the split/do verbs and the concrete-vs-forward read surface --- src/libtmux/experimental/query.py | 100 +++++++++++++++++++++++++----- tests/experimental/test_query.py | 59 +++++++++++++++++- 2 files changed, 144 insertions(+), 15 deletions(-) diff --git a/src/libtmux/experimental/query.py b/src/libtmux/experimental/query.py index 05d752fa9..0ff2ec57d 100644 --- a/src/libtmux/experimental/query.py +++ b/src/libtmux/experimental/query.py @@ -43,15 +43,18 @@ RespawnPane, SelectPane, SendKeys, + SplitWindow, run, ) if t.TYPE_CHECKING: from collections.abc import Callable, Mapping, Sequence + from typing_extensions import Self + from libtmux.experimental.models.snapshots import PaneSnapshot from libtmux.experimental.ops import Planner, PlanResult - from libtmux.experimental.ops._types import SlotRef + from libtmux.experimental.ops._types import SlotRef, Target #: A source of pane snapshots: an engine to read from, or pre-taken snapshots. PaneSource = t.Union["TmuxEngine", "Sequence[PaneSnapshot]"] @@ -149,11 +152,13 @@ class BoundPaneCommands: Each method appends a typed operation targeting the bound pane to the plan and returns its :class:`~..ops._types.SlotRef`, so commands compose and the - plan folds to a single tmux dispatch. + plan folds to a single tmux dispatch. ``target`` is a :data:`~..ops._types.Target` + so a forward :class:`~..ops._types.SlotRef` (a pane an earlier op creates) + flows through as well as a concrete :class:`~..ops._types.PaneId`. """ plan: LazyPlan - target: PaneId + target: Target def send_keys( self, @@ -198,26 +203,93 @@ def kill(self) -> SlotRef: @dataclass(frozen=True) -class PaneRef: - """A matched pane plus a ``cmd`` namespace that records into a plan.""" +class _PaneRefBase: + """The verbs shared by concrete and forward pane handles. + + An immutable pointer into a *mutable* :class:`LazyPlan`. Structural verbs + (:meth:`split`) record a create op and return a *forward* handle to the + not-yet-created pane; leaf commands live under :attr:`cmd`; :meth:`do` + threads a side-effecting recorder into a fluent chain. + """ - pane: PaneSnapshot plan: LazyPlan + target: Target + + @property + def cmd(self) -> BoundPaneCommands: + """Pane commands bound to this handle's target (recorded into the plan).""" + return BoundPaneCommands(self.plan, self.target) + + def split(self, *, horizontal: bool = False) -> ForwardPaneRef: + """Split this pane; return a forward handle to the new pane. + + Examples + -------- + >>> plan = LazyPlan() + >>> new = _PaneRefBase(plan, PaneId("%1")).split() + >>> isinstance(new, ForwardPaneRef) + True + >>> [op.kind for op in plan.operations] + ['split_window'] + """ + slot = self.plan.add(SplitWindow(target=self.target, horizontal=horizontal)) + return ForwardPaneRef(self.plan, slot) + + def do(self, fn: Callable[[BoundPaneCommands], object]) -> Self: + """Apply *fn* to this handle's :attr:`cmd`, returning the handle. + + Examples + -------- + >>> plan = LazyPlan() + >>> h = _PaneRefBase(plan, PaneId("%1")) + >>> h.do(lambda c: c.send_keys("vim")) is h + True + >>> [op.kind for op in plan.operations] + ['send_keys'] + """ + fn(self.cmd) + return self + + +@dataclass(frozen=True) +class ForwardPaneRef(_PaneRefBase): + """A pane an earlier operation will create. + + Carries a forward :class:`~..ops._types.SlotRef`; it has no snapshot, so + reading a pane id/attribute off it is a *static* type error (the id is + unknown until the plan runs). Keep building instead -- ``split().do(...)``. + """ + + +@dataclass(frozen=True) +class PaneRef(_PaneRefBase): + """A concrete matched pane: the shared verbs plus snapshot reads. + + Examples + -------- + >>> from libtmux.experimental.models.snapshots import PaneSnapshot + >>> snap = PaneSnapshot.from_format({"pane_id": "%1", "pane_active": "1"}) + >>> ref = PaneRef(LazyPlan(), PaneId("%1"), snapshot=snap) + >>> ref.pane_id, ref.active + ('%1', True) + """ + + snapshot: PaneSnapshot + + @property + def pane(self) -> PaneSnapshot: + """The underlying pane snapshot.""" + return self.snapshot @property def pane_id(self) -> str: """The pane's id.""" - return self.pane.pane_id + return self.snapshot.pane_id @property def active(self) -> bool: """Whether the pane is active in its window.""" - return self.pane.active - - @property - def cmd(self) -> BoundPaneCommands: - """Pane commands bound to this pane (recorded into the plan).""" - return BoundPaneCommands(self.plan, PaneId(self.pane.pane_id)) + return self.snapshot.active @dataclass(frozen=True) @@ -250,7 +322,7 @@ def to_plan(self, source: PaneSource) -> LazyPlan: """ plan = LazyPlan() for pane in self.query.all(source): - self.mapper(PaneRef(pane, plan)) + self.mapper(PaneRef(plan, PaneId(pane.pane_id), snapshot=pane)) return plan def run( diff --git a/tests/experimental/test_query.py b/tests/experimental/test_query.py index 01e366f24..a3d9c3a41 100644 --- a/tests/experimental/test_query.py +++ b/tests/experimental/test_query.py @@ -4,8 +4,11 @@ import typing as t +import pytest + from libtmux.experimental.models.snapshots import PaneSnapshot -from libtmux.experimental.query import PaneQuery, panes +from libtmux.experimental.ops import LazyPlan, PaneId +from libtmux.experimental.query import ForwardPaneRef, PaneQuery, PaneRef, panes if t.TYPE_CHECKING: from libtmux.session import Session @@ -29,6 +32,60 @@ def _pane(pane_id: str, index: int, *, active: bool, command: str) -> PaneSnapsh ) +def _concrete(plan: LazyPlan) -> PaneRef: + return PaneRef( + plan, + PaneId("%1"), + snapshot=_pane("%1", 0, active=True, command="vim"), + ) + + +def test_split_returns_forward_handle() -> None: + """A structural verb records a create and returns a forward handle.""" + plan = LazyPlan() + new = _concrete(plan).split() + assert isinstance(new, ForwardPaneRef) + assert [op.kind for op in plan.operations] == ["split_window"] + + +def test_do_chains_on_the_handle() -> None: + """do() records into the plan and returns the same handle.""" + plan = LazyPlan() + ref = _concrete(plan) + assert ref.do(lambda c: c.send_keys("vim")) is ref + assert [op.kind for op in plan.operations] == ["send_keys"] + + +class _ReadCase(t.NamedTuple): + """Whether a handle exposes concrete pane reads (concrete) or not (forward).""" + + test_id: str + forward: bool + + +_READ_CASES: tuple[_ReadCase, ...] = ( + _ReadCase("concrete_reads", forward=False), + _ReadCase("forward_no_reads", forward=True), +) + + +@pytest.mark.parametrize("case", _READ_CASES, ids=[c.test_id for c in _READ_CASES]) +def test_handle_read_surface(case: _ReadCase) -> None: + """Concrete handles expose pane reads; forward handles have none. + + The forward handle's absence of ``pane_id`` is what makes a premature read a + *static* type error (mypy + ty), with the structural absence as its runtime + shadow. + """ + plan = LazyPlan() + concrete = _concrete(plan) + handle: object = concrete.split() if case.forward else concrete + assert hasattr(handle, "pane_id") is (not case.forward) + if not case.forward: + assert concrete.pane_id == "%1" + assert concrete.active is True + + def test_panes_returns_query() -> None: """panes() starts an empty, immutable query.""" assert panes() == PaneQuery() From 6a95d63e3265b0ceaa7c23686c29ec1ce5be7363 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 4 Jul 2026 12:44:43 -0500 Subject: [PATCH 02/13] Fluent(feat): Add plan() forward-ref build tier why: the split-type pane handles give forward-ref panes, but a user still needs a fluent entry to declare a whole session tree. plan() records a session/window/pane build into one LazyPlan that folds to a few tmux dispatches instead of one call per operation. what: - Add experimental/fluent.py: plan() -> PlanBuilder, SessionRef, WindowRef - Name-address sessions/windows (folds stay intact); reach a pane via the creator's captured first-pane SlotRef (a forward handle) - run()/arun() default to MarkedPlanner; preview() is a pure argv dry-run - Cover the build shape, an offline fold over ConcreteEngine, and a live build --- src/libtmux/experimental/fluent.py | 196 +++++++++++++++++++++++++++++ tests/experimental/test_fluent.py | 89 +++++++++++++ 2 files changed, 285 insertions(+) create mode 100644 src/libtmux/experimental/fluent.py create mode 100644 tests/experimental/test_fluent.py diff --git a/src/libtmux/experimental/fluent.py b/src/libtmux/experimental/fluent.py new file mode 100644 index 000000000..5641287aa --- /dev/null +++ b/src/libtmux/experimental/fluent.py @@ -0,0 +1,196 @@ +"""A fluent, forward-ref builder that folds a session build to a few dispatches. + +``plan()`` opens a :class:`PlanBuilder` -- a thin recorder over a Core +:class:`~libtmux.experimental.ops.plan.LazyPlan`. Navigating it +(:meth:`PlanBuilder.new_session` -> :class:`SessionRef` -> +:class:`WindowRef` -> :meth:`WindowRef.pane`) records create operations and +hands back forward handles (:class:`~libtmux.experimental.query.ForwardPaneRef`) +that address objects the plan will create. Nothing runs until +:meth:`PlanBuilder.run` (or its async twin :meth:`PlanBuilder.arun`), which folds +the recorded operations into a handful of ``tmux a ; b`` dispatches by default +(a :class:`~libtmux.experimental.ops.planner.MarkedPlanner`). + +Named objects (sessions, windows) are addressed by name so their sub-operations +fold; a pane -- which has no name -- is addressed by a forward +:class:`~libtmux.experimental.ops._types.SlotRef`, resolved from the creating +operation's captured id at execution. + +Examples +-------- +>>> from libtmux.experimental.engines.concrete import ConcreteEngine +>>> p = plan() +>>> pane = p.new_session("dev").window().pane() +>>> bottom = pane.do(lambda c: c.send_keys("vim")).split() +>>> bottom.do(lambda c: c.send_keys("htop")) is bottom +True +>>> [op.kind for op in p.plan.operations] +['new_session', 'send_keys', 'split_window', 'send_keys'] +>>> p.run(ConcreteEngine()).ok +True +""" + +from __future__ import annotations + +import typing as t +from dataclasses import dataclass, field + +from libtmux.experimental.ops import ( + LazyPlan, + MarkedPlanner, + NameRef, + NewSession, + NewWindow, +) +from libtmux.experimental.query import ForwardPaneRef + +if t.TYPE_CHECKING: + from libtmux.experimental.engines.base import AsyncTmuxEngine, TmuxEngine + from libtmux.experimental.ops import Planner, PlanResult + from libtmux.experimental.ops._types import SlotRef + + +@dataclass(frozen=True) +class WindowRef: + """A window in a plan; navigate to its first pane. + + ``first_pane`` is a forward :class:`~..ops._types.SlotRef` to the window's + first pane (captured by the creating ``new-session`` / ``new-window``). + """ + + plan: LazyPlan + first_pane: SlotRef + + def pane(self) -> ForwardPaneRef: + """Return a forward handle to the window's first pane. + + Examples + -------- + >>> p = plan() + >>> ref = p.new_session("dev").window().pane() + >>> isinstance(ref, ForwardPaneRef) + True + """ + return ForwardPaneRef(self.plan, self.first_pane) + + +@dataclass(frozen=True) +class SessionRef: + """A session in a plan; reach its first window or add another. + + The session is name-addressed (so its window operations fold); ``create`` is + the ``new-session`` slot, whose captured first pane backs the first window. + """ + + plan: LazyPlan + name: str + create: SlotRef + + def window(self) -> WindowRef: + """Return the session's first window. + + Examples + -------- + >>> isinstance(plan().new_session("dev").window(), WindowRef) + True + """ + return WindowRef(self.plan, self.create.pane) + + def new_window(self, name: str) -> WindowRef: + """Create another window in this session (name-addressed). + + Examples + -------- + >>> p = plan() + >>> _ = p.new_session("dev").new_window("logs") + >>> [op.kind for op in p.plan.operations] + ['new_session', 'new_window'] + """ + slot = self.plan.add( + NewWindow(target=NameRef(self.name), name=name, capture_pane=True), + ) + return WindowRef(self.plan, slot.pane) + + +@dataclass(frozen=True) +class PlanBuilder: + """A fluent recorder over a :class:`LazyPlan`; :meth:`run` folds by default.""" + + plan: LazyPlan = field(default_factory=LazyPlan) + + def new_session(self, name: str) -> SessionRef: + """Create a session, capturing its first pane for forward refs. + + Examples + -------- + >>> p = plan() + >>> ref = p.new_session("dev") + >>> isinstance(ref, SessionRef) + True + >>> [op.kind for op in p.plan.operations] + ['new_session'] + """ + slot = self.plan.add(NewSession(session_name=name, capture_panes=True)) + return SessionRef(self.plan, name, slot) + + def run( + self, + engine: TmuxEngine, + *, + version: str | None = None, + planner: Planner | None = None, + ) -> PlanResult: + """Build over *engine*, folding to a few dispatches (``MarkedPlanner``). + + Examples + -------- + >>> from libtmux.experimental.engines.concrete import ConcreteEngine + >>> p = plan() + >>> _ = p.new_session("dev").window().pane().do(lambda c: c.send_keys("vim")) + >>> p.run(ConcreteEngine()).ok + True + """ + return self.plan.execute( + engine, + version=version, + planner=planner or MarkedPlanner(), + ) + + async def arun( + self, + engine: AsyncTmuxEngine, + *, + version: str | None = None, + planner: Planner | None = None, + ) -> PlanResult: + """Async twin of :meth:`run` (same fold, ``await``ed).""" + return await self.plan.aexecute( + engine, + version=version, + planner=planner or MarkedPlanner(), + ) + + def preview(self, *, version: str | None = None) -> list[tuple[str, ...] | None]: + """Render a pure argv dry-run of the recorded plan (no engine). + + Examples + -------- + >>> p = plan() + >>> _ = p.new_session("dev") + >>> argv = p.preview()[0] + >>> argv[:4] + ('new-session', '-d', '-s', 'dev') + >>> argv[-1] + '#{session_id} #{window_id} #{pane_id}' + """ + return self.plan.preview(version=version) + + +def plan() -> PlanBuilder: + """Start a fluent, forward-ref plan build. + + Examples + -------- + >>> plan().plan.operations + () + """ + return PlanBuilder() diff --git a/tests/experimental/test_fluent.py b/tests/experimental/test_fluent.py new file mode 100644 index 000000000..8e596b7a0 --- /dev/null +++ b/tests/experimental/test_fluent.py @@ -0,0 +1,89 @@ +"""Tests for the fluent forward-ref plan builder (``plan()``).""" + +from __future__ import annotations + +import typing as t + +import pytest + +from libtmux.experimental.engines.concrete import ConcreteEngine +from libtmux.experimental.fluent import PlanBuilder, plan +from libtmux.experimental.query import ForwardPaneRef + +if t.TYPE_CHECKING: + from libtmux.session import Session + + +def _one_window_two_panes(p: PlanBuilder) -> None: + pane = p.new_session("dev").window().pane() + pane.do(lambda c: c.send_keys("vim")).split().do( + lambda c: c.send_keys("pytest -q"), + ) + + +def _two_windows(p: PlanBuilder) -> None: + sess = p.new_session("dev") + sess.window().pane().do(lambda c: c.send_keys("vim")) + sess.new_window("logs").pane().do(lambda c: c.send_keys("tail -f log")) + + +class _BuildCase(t.NamedTuple): + """A fluent build and the operation sequence it should record.""" + + test_id: str + build: t.Callable[[PlanBuilder], None] + kinds: list[str] + + +_BUILD_CASES: tuple[_BuildCase, ...] = ( + _BuildCase( + "one_window_two_panes", + _one_window_two_panes, + ["new_session", "send_keys", "split_window", "send_keys"], + ), + _BuildCase( + "two_windows", + _two_windows, + ["new_session", "send_keys", "new_window", "send_keys"], + ), +) + + +@pytest.mark.parametrize("case", _BUILD_CASES, ids=[c.test_id for c in _BUILD_CASES]) +def test_builder_records_ops(case: _BuildCase) -> None: + """The fluent build records the expected operation sequence.""" + p = plan() + case.build(p) + assert [op.kind for op in p.plan.operations] == case.kinds + + +@pytest.mark.parametrize("case", _BUILD_CASES, ids=[c.test_id for c in _BUILD_CASES]) +def test_builder_runs_offline(case: _BuildCase) -> None: + """The build resolves forward refs and folds over the in-memory engine.""" + p = plan() + case.build(p) + assert p.run(ConcreteEngine()).ok + + +def test_window_pane_is_forward_handle() -> None: + """A window's first pane is a forward handle with no snapshot reads.""" + ref = plan().new_session("dev").window().pane() + assert isinstance(ref, ForwardPaneRef) + assert not hasattr(ref, "pane_id") + + +def test_build_session_live(session: Session) -> None: + """A fluent build creates a real session with the declared panes.""" + from libtmux.experimental.engines.subprocess import SubprocessEngine + + engine = SubprocessEngine.for_server(session.server) + p = plan() + pane = p.new_session("fluentdev").window().pane() + pane.do(lambda c: c.send_keys("echo top", enter=False)).split().do( + lambda c: c.send_keys("echo bottom", enter=False), + ) + p.run(engine).raise_for_status() + + built = [s for s in session.server.sessions if s.session_name == "fluentdev"] + assert len(built) == 1 + assert len(built[0].windows[0].panes) == 2 From 614e3f68b8a2104eab1c7e5db83ee9901fc427b7 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 4 Jul 2026 12:53:38 -0500 Subject: [PATCH 03/13] Workspace(feat): Add freeze (live server to IR) why: analyze() and to_dict() cover the config->IR->dict round-trip, but there was no way back from a LIVE server to declarative IR. freeze() closes the loop -- reverse-analyze a running session into a Workspace you can rebuild elsewhere (tmuxp `freeze`). what: - Add experimental/workspace/freeze.py: freeze(ServerSnapshot) -> Workspace (pure), plus freeze_server/afreeze_server rebuilding the whole tree from one list-panes -a read - Export freeze/freeze_server/afreeze_server from the workspace package --- .../experimental/workspace/__init__.py | 8 + src/libtmux/experimental/workspace/freeze.py | 249 ++++++++++++++++++ tests/experimental/test_freeze.py | 213 +++++++++++++++ 3 files changed, 470 insertions(+) create mode 100644 src/libtmux/experimental/workspace/freeze.py create mode 100644 tests/experimental/test_freeze.py diff --git a/src/libtmux/experimental/workspace/__init__.py b/src/libtmux/experimental/workspace/__init__.py index 99b572142..7eb3e9049 100644 --- a/src/libtmux/experimental/workspace/__init__.py +++ b/src/libtmux/experimental/workspace/__init__.py @@ -38,6 +38,11 @@ WindowCreated, WorkspaceBuilt, ) +from libtmux.experimental.workspace.freeze import ( + afreeze_server, + freeze, + freeze_server, +) from libtmux.experimental.workspace.ir import ( Command, Float, @@ -65,9 +70,12 @@ "WorkspaceBuilt", "WorkspaceCompileError", "abuild_workspace", + "afreeze_server", "analyze", "build_workspace", "compile_full", "compile_workspace", "confirm", + "freeze", + "freeze_server", ) diff --git a/src/libtmux/experimental/workspace/freeze.py b/src/libtmux/experimental/workspace/freeze.py new file mode 100644 index 000000000..84c424fb0 --- /dev/null +++ b/src/libtmux/experimental/workspace/freeze.py @@ -0,0 +1,249 @@ +"""Reverse-analyze a live server snapshot into the declarative IR -- the round-trip. + +:func:`~libtmux.experimental.workspace.analyzer.analyze` lowers a tmuxp-style +config *into* a :class:`~libtmux.experimental.workspace.ir.Workspace`; +:func:`freeze` is its inverse over **live** state. It walks an immutable +:class:`~libtmux.experimental.models.snapshots.ServerSnapshot` back into a +``Workspace`` that :meth:`~..ir.Workspace.build` / :meth:`~..ir.Workspace.compile` +can replay, so a running session can be captured as reusable, version-controllable +IR (tmuxp's ``freeze``). It is **lossy by design**: scrollback, live process +state, and a pane sitting at a bare shell are not reconstructed. + +The north star -- *fewest backend calls* -- holds: :func:`freeze_server` / +:func:`afreeze_server` rebuild the **entire** session/window/pane tree from a +**single** ``list-panes -a -F`` read; the mapping itself is pure. +""" + +from __future__ import annotations + +import typing as t + +from libtmux.experimental.models.snapshots import ServerSnapshot +from libtmux.experimental.workspace.ir import Pane, Window, Workspace + +if t.TYPE_CHECKING: + from collections.abc import Collection, Iterable + + from libtmux.experimental.engines.base import AsyncTmuxEngine, TmuxEngine + from libtmux.experimental.models.snapshots import ( + PaneSnapshot, + SessionSnapshot, + WindowSnapshot, + ) + +#: Bare shells whose presence as a pane's *current command* means "no command": +#: freezing such a pane yields an empty pane, not a nested shell (tmuxp parity). +#: Override via the ``shells`` argument to keep or widen the set. +SHELLS: frozenset[str] = frozenset( + { + "sh", + "bash", + "zsh", + "fish", + "dash", + "ksh", + "tcsh", + "csh", + "ash", + "nu", + "xonsh", + "elvish", + "pwsh", + }, +) + +#: The tmux fields one ``list-panes -a -F`` read needs to rebuild the whole tree. +FREEZE_FIELDS: tuple[str, ...] = ( + "session_id", + "session_name", + "window_id", + "window_index", + "window_name", + "window_layout", + "window_active", + "pane_id", + "pane_index", + "pane_active", + "pane_current_command", + "pane_current_path", +) +_SEP = "\t" +#: The ``-F`` format string covering :data:`FREEZE_FIELDS` (one read, whole tree). +FREEZE_FORMAT: str = _SEP.join(f"#{{{field}}}" for field in FREEZE_FIELDS) + + +def _pick_session( + server: ServerSnapshot, + selector: str | None, +) -> SessionSnapshot: + """Choose the one session to freeze (by name/id, or the sole one).""" + sessions = server.sessions + if not sessions: + msg = "cannot freeze an empty server (no sessions)" + raise ValueError(msg) + if selector is None: + if len(sessions) == 1: + return sessions[0] + names = ", ".join(s.name or s.session_id for s in sessions) + msg = ( + f"ambiguous freeze: {len(sessions)} sessions ({names}); " + f"pass session= to choose one" + ) + raise ValueError(msg) + for session in sessions: + if selector in (session.name, session.session_id): + return session + names = ", ".join(s.name or s.session_id for s in sessions) + msg = f"no session matching {selector!r} (have: {names})" + raise ValueError(msg) + + +def _freeze_pane(pane: PaneSnapshot, shells: Collection[str]) -> Pane: + """Map one pane snapshot to a declarative :class:`~..ir.Pane`. + + A pane sitting at a bare shell (its ``current_command`` is in *shells*) + freezes to an empty pane -- replaying it as a command would nest a shell. + """ + command = pane.current_command + run = None if command is None or command in shells else command + return Pane(run=run, focus=pane.active, start_directory=pane.current_path) + + +def _freeze_window(window: WindowSnapshot, shells: Collection[str]) -> Window: + """Map one window snapshot and its panes to a declarative :class:`~..ir.Window`.""" + return Window( + name=window.name, + layout=window.layout, + focus=window.active, + panes=[_freeze_pane(pane, shells) for pane in window.panes], + ) + + +def freeze( + snapshot: ServerSnapshot, + *, + session: str | None = None, + shells: Collection[str] = SHELLS, +) -> Workspace: + """Reverse-analyze a live :class:`ServerSnapshot` into a declarative Workspace. + + The inverse of :func:`~..analyzer.analyze`: capture what is *running* as + reusable IR. Pure -- no tmux. Lossy by design (no scrollback / process state; + a bare-shell pane becomes an empty pane). + + Parameters + ---------- + snapshot : ServerSnapshot + The live server tree (e.g. from :meth:`ServerSnapshot.from_pane_rows`). + session : str or None + Which session to freeze, by ``session_name`` or ``session_id``. ``None`` + freezes the sole session and raises when the server holds several. + shells : Collection[str] + Commands treated as "a bare shell" -> an empty pane (default + :data:`SHELLS`). + + Returns + ------- + Workspace + A declarative spec that ``build``/``compile`` replays. + + Raises + ------ + ValueError + When the server is empty, *session* is ambiguous, or the named session + is absent. + + Examples + -------- + >>> from libtmux.experimental.models.snapshots import ServerSnapshot + >>> server = ServerSnapshot.from_pane_rows([ + ... {"session_id": "$0", "session_name": "dev", "window_id": "@1", + ... "window_index": "0", "window_name": "editor", "pane_id": "%1", + ... "pane_index": "0", "pane_active": "1", "pane_current_command": "vim"}, + ... {"session_id": "$0", "session_name": "dev", "window_id": "@1", + ... "window_index": "0", "window_name": "editor", "pane_id": "%2", + ... "pane_index": "1", "pane_current_command": "zsh"}, + ... ]) + >>> ws = freeze(server) + >>> ws.name + 'dev' + >>> [c.cmd for c in ws.windows[0].panes[0].commands] + ['vim'] + >>> ws.windows[0].panes[1].run is None # a bare shell -> empty pane + True + """ + chosen = _pick_session(snapshot, session) + return Workspace( + name=chosen.name or chosen.session_id, + windows=[_freeze_window(window, shells) for window in chosen.windows], + ) + + +def _rows(stdout: Iterable[str]) -> list[dict[str, str]]: + """Parse ``list-panes -F`` tab-separated lines into per-pane field dicts.""" + rows: list[dict[str, str]] = [] + for line in stdout: + if not line: + continue + parts = line.split(_SEP) + # zip(strict=False) tolerates a short row (a trailing empty field tmux drops) + rows.append(dict(zip(FREEZE_FIELDS, parts, strict=False))) + return rows + + +def freeze_server( + engine: TmuxEngine, + *, + session: str | None = None, + shells: Collection[str] = SHELLS, +) -> Workspace: + r"""Freeze a live server into IR with a **single** ``list-panes`` read. + + Reads the whole session/window/pane tree in one ``list-panes -a -F`` dispatch, + builds a :class:`ServerSnapshot`, and reverse-analyzes it via :func:`freeze`. + + Examples + -------- + >>> from libtmux.experimental.engines.base import CommandResult + >>> class _Engine: # one read returns the whole tree + ... def run(self, request): + ... row = "$0\tdev\t@1\t0\teditor\t\t1\t%1\t0\t1\tvim\t/work" + ... return CommandResult(cmd=("tmux",), stdout=(row,)) + >>> freeze_server(_Engine()).name + 'dev' + """ + from libtmux.experimental.engines.base import CommandRequest + + result = engine.run( + CommandRequest.from_args("list-panes", "-a", "-F", FREEZE_FORMAT), + ) + server = ServerSnapshot.from_pane_rows(_rows(result.stdout)) + return freeze(server, session=session, shells=shells) + + +async def afreeze_server( + engine: AsyncTmuxEngine, + *, + session: str | None = None, + shells: Collection[str] = SHELLS, +) -> Workspace: + r"""Async twin of :func:`freeze_server` (one awaited ``list-panes`` read). + + Examples + -------- + >>> import asyncio + >>> from libtmux.experimental.engines.base import CommandResult + >>> class _AEngine: + ... async def run(self, request): + ... row = "$0\tdev\t@1\t0\tmain\t\t1\t%1\t0\t1\tvim\t/w" + ... return CommandResult(cmd=("tmux",), stdout=(row,)) + >>> asyncio.run(afreeze_server(_AEngine())).name + 'dev' + """ + from libtmux.experimental.engines.base import CommandRequest + + result = await engine.run( + CommandRequest.from_args("list-panes", "-a", "-F", FREEZE_FORMAT), + ) + server = ServerSnapshot.from_pane_rows(_rows(result.stdout)) + return freeze(server, session=session, shells=shells) diff --git a/tests/experimental/test_freeze.py b/tests/experimental/test_freeze.py new file mode 100644 index 000000000..54651c71a --- /dev/null +++ b/tests/experimental/test_freeze.py @@ -0,0 +1,213 @@ +"""Tests for ``freeze`` -- a live server snapshot reverse-analyzed into IR. + +The pure core (:func:`freeze`) maps an immutable +:class:`~libtmux.experimental.models.snapshots.ServerSnapshot` into a declarative +:class:`~libtmux.experimental.workspace.ir.Workspace`, closing the round-trip +``analyze`` opens. These units feed synthetic snapshots -- no tmux. +""" + +from __future__ import annotations + +import asyncio +import typing as t + +import pytest + +from libtmux.experimental.engines import ConcreteEngine +from libtmux.experimental.engines.async_control_mode import AsyncControlModeEngine +from libtmux.experimental.models.snapshots import ServerSnapshot +from libtmux.experimental.workspace.freeze import SHELLS, afreeze_server, freeze + +if t.TYPE_CHECKING: + from libtmux.experimental.workspace.ir import Workspace + from libtmux.session import Session + + +def _server(*rows: dict[str, str]) -> ServerSnapshot: + """Build a ServerSnapshot from flat per-pane rows (one list-panes read).""" + return ServerSnapshot.from_pane_rows(rows) + + +def test_freeze_maps_session_window_pane() -> None: + """A single session's tree becomes a Workspace of Windows of Panes.""" + server = _server( + { + "session_id": "$0", + "session_name": "dev", + "window_id": "@1", + "window_index": "0", + "window_name": "editor", + "window_layout": "main-vertical", + "window_active": "1", + "pane_id": "%1", + "pane_index": "0", + "pane_active": "1", + "pane_current_command": "vim", + "pane_current_path": "/home/d/work", + }, + ) + ws = freeze(server) + assert ws.name == "dev" + assert [w.name for w in ws.windows] == ["editor"] + win = ws.windows[0] + assert win.layout == "main-vertical" + assert win.focus is True # the active window + pane = win.panes[0] + assert [c.cmd for c in pane.commands] == ["vim"] + assert pane.start_directory == "/home/d/work" + assert pane.focus is True # the active pane + + +def test_freeze_drops_shell_command() -> None: + """A pane sitting at a bare shell freezes to an empty pane (no nested shell).""" + server = _server( + { + "session_id": "$0", + "session_name": "dev", + "window_id": "@1", + "window_index": "0", + "window_name": "main", + "pane_id": "%1", + "pane_index": "0", + "pane_current_command": "zsh", + }, + ) + pane = freeze(server).windows[0].panes[0] + assert pane.run is None + assert "zsh" in SHELLS # documents the default filter + + +def test_freeze_keeps_non_shell_command() -> None: + """A pane running a real program freezes that program as the pane command.""" + server = _server( + { + "session_id": "$0", + "session_name": "dev", + "window_id": "@1", + "window_index": "0", + "window_name": "logs", + "pane_id": "%1", + "pane_index": "0", + "pane_current_command": "tail", + }, + ) + assert [c.cmd for c in freeze(server).windows[0].panes[0].commands] == ["tail"] + + +def test_freeze_selects_session_by_name() -> None: + """With many sessions, ``session=`` picks one to freeze.""" + server = _server( + { + "session_id": "$0", + "session_name": "a", + "window_id": "@1", + "window_index": "0", + "window_name": "w", + "pane_id": "%1", + "pane_index": "0", + }, + { + "session_id": "$1", + "session_name": "b", + "window_id": "@2", + "window_index": "0", + "window_name": "w", + "pane_id": "%2", + "pane_index": "0", + }, + ) + assert freeze(server, session="b").name == "b" + assert freeze(server, session="$0").name == "a" + + +def test_freeze_ambiguous_session_raises() -> None: + """With many sessions and no selector, freeze refuses to guess.""" + server = _server( + { + "session_id": "$0", + "session_name": "a", + "window_id": "@1", + "window_index": "0", + "pane_id": "%1", + "pane_index": "0", + }, + { + "session_id": "$1", + "session_name": "b", + "window_id": "@2", + "window_index": "0", + "pane_id": "%2", + "pane_index": "0", + }, + ) + with pytest.raises(ValueError, match=r"ambiguous|multiple|session="): + freeze(server) + + +def test_freeze_unknown_session_raises() -> None: + """A named session that is not present is an error, not an empty workspace.""" + server = _server( + { + "session_id": "$0", + "session_name": "a", + "window_id": "@1", + "window_index": "0", + "pane_id": "%1", + "pane_index": "0", + }, + ) + with pytest.raises(ValueError, match="nope"): + freeze(server, session="nope") + + +def test_freeze_round_trips_into_a_buildable_workspace() -> None: + """``freeze`` output compiles and builds -- the declarative round-trip closes.""" + server = _server( + { + "session_id": "$0", + "session_name": "dev", + "window_id": "@1", + "window_index": "0", + "window_name": "editor", + "pane_id": "%1", + "pane_index": "0", + "pane_active": "1", + "pane_current_command": "vim", + }, + { + "session_id": "$0", + "session_name": "dev", + "window_id": "@1", + "window_index": "0", + "window_name": "editor", + "pane_id": "%2", + "pane_index": "1", + "pane_current_command": "tail", + }, + ) + ws = freeze(server) + assert ws.compile().operations[0].kind == "new_session" + assert ws.build(ConcreteEngine(), preflight=False).ok + + +def test_afreeze_server_captures_live_tree(session: Session) -> None: + """A real server freezes in ONE list-panes read, reproducing its windows. + + Validates ``FREEZE_FORMAT`` against live tmux: the frozen Workspace must carry + the live session's name and every window name, and remain buildable. + """ + session.new_window(window_name="logs") + live_names = {w.window_name for w in session.windows} + + async def main() -> Workspace: + engine = AsyncControlModeEngine.for_server(session.server) + try: + return await afreeze_server(engine, session=session.name) + finally: + await engine.aclose() + + ws = asyncio.run(main()) + assert ws.name == session.name + assert {w.name for w in ws.windows} == live_names + # The captured tree is a valid, buildable spec. + assert ws.compile().operations[0].kind == "new_session" From 7092768e7723e02a7544f21bbc41a9ce0c442312 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 4 Jul 2026 12:59:42 -0500 Subject: [PATCH 04/13] Workspace(feat): Add variant expand + workspace sets why: building several near-identical sessions (per-app, per-env) meant declaring each by hand and dispatching each separately. expand() fans one workspace into named variants; WorkspaceSet folds N workspaces into one rebased plan so the whole batch builds in a single folded run. what: - Add workspace/expand.py: expand(workspace, variants) renders $name / ${name} tokens, leaving unknown tokens intact - Add workspace/sets.py: WorkspaceSet + compile_workspaces/build_workspaces /abuild_workspaces, rebasing each workspace's SlotRefs and host steps by a per-workspace offset into one LazyPlan - Export expand, WorkspaceSet(+Result/Compiled), build/abuild/compile_workspaces --- .../experimental/workspace/__init__.py | 16 + src/libtmux/experimental/workspace/expand.py | 87 ++++ src/libtmux/experimental/workspace/sets.py | 441 ++++++++++++++++++ .../contract/test_workspace_expand.py | 66 +++ .../contract/test_workspace_sets.py | 167 +++++++ 5 files changed, 777 insertions(+) create mode 100644 src/libtmux/experimental/workspace/expand.py create mode 100644 src/libtmux/experimental/workspace/sets.py create mode 100644 tests/experimental/contract/test_workspace_expand.py create mode 100644 tests/experimental/contract/test_workspace_sets.py diff --git a/src/libtmux/experimental/workspace/__init__.py b/src/libtmux/experimental/workspace/__init__.py index 7eb3e9049..1e7126c39 100644 --- a/src/libtmux/experimental/workspace/__init__.py +++ b/src/libtmux/experimental/workspace/__init__.py @@ -38,6 +38,7 @@ WindowCreated, WorkspaceBuilt, ) +from libtmux.experimental.workspace.expand import expand from libtmux.experimental.workspace.freeze import ( afreeze_server, freeze, @@ -52,11 +53,20 @@ Workspace, ) from libtmux.experimental.workspace.runner import abuild_workspace, build_workspace +from libtmux.experimental.workspace.sets import ( + CompiledWorkspaceSet, + WorkspaceSet, + WorkspaceSetResult, + abuild_workspaces, + build_workspaces, + compile_workspaces, +) __all__ = ( "BuildEvent", "Command", "Compiled", + "CompiledWorkspaceSet", "ConfirmReport", "Float", "FloatingPane", @@ -69,13 +79,19 @@ "Workspace", "WorkspaceBuilt", "WorkspaceCompileError", + "WorkspaceSet", + "WorkspaceSetResult", "abuild_workspace", + "abuild_workspaces", "afreeze_server", "analyze", "build_workspace", + "build_workspaces", "compile_full", "compile_workspace", + "compile_workspaces", "confirm", + "expand", "freeze", "freeze_server", ) diff --git a/src/libtmux/experimental/workspace/expand.py b/src/libtmux/experimental/workspace/expand.py new file mode 100644 index 000000000..fe9af1e48 --- /dev/null +++ b/src/libtmux/experimental/workspace/expand.py @@ -0,0 +1,87 @@ +"""Pure variant expansion for declarative workspace specs.""" + +from __future__ import annotations + +import collections.abc +import dataclasses +import re +import typing as t +from collections.abc import Callable, Iterable, Mapping + +from libtmux.experimental.workspace.ir import Workspace + +Variant: t.TypeAlias = Mapping[str, object] +NameFactory: t.TypeAlias = Callable[[str, Mapping[str, object]], str] + +_TOKEN_RE = re.compile( + r"\$(?P\$)|\$\{(?P[A-Za-z_][A-Za-z0-9_]*)\}" + r"|\$(?P[A-Za-z_][A-Za-z0-9_]*)", +) + + +def expand( + workspace: Workspace, + variants: Iterable[Mapping[str, object]], + *, + variables: Mapping[str, object] | None = None, + name: NameFactory | None = None, +) -> tuple[Workspace, ...]: + """Return one rendered workspace per variant, without mutating *workspace*. + + String fields use shell-style ``$name`` / ``${name}`` placeholders. Unknown + variables stay intact, so shell variables and tmux formats survive expansion. + + Examples + -------- + >>> from libtmux.experimental.workspace import Pane, Window, Workspace, expand + >>> base = Workspace("svc-$app", windows=[Window("$app", panes=[Pane("$cmd")])]) + >>> [ws.name for ws in expand(base, [{"app": "api", "cmd": "uvicorn"}])] + ['svc-api'] + """ + expanded: list[Workspace] = [] + for variant in variants: + context: dict[str, object] = dict(variables or {}) + context.update(variant) + rendered = t.cast("Workspace", _render(workspace, context)) + if name is not None: + rendered = dataclasses.replace( + rendered, + name=name(workspace.name, context), + ) + expanded.append(rendered) + return tuple(expanded) + + +def _render(value: t.Any, context: collections.abc.Mapping[str, object]) -> t.Any: + """Recursively render strings inside dataclasses, mappings, and sequences.""" + if isinstance(value, str): + return _render_string(value, context) + if dataclasses.is_dataclass(value) and not isinstance(value, type): + changes = { + field.name: _render(getattr(value, field.name), context) + for field in dataclasses.fields(value) + } + return dataclasses.replace(value, **changes) + if isinstance(value, collections.abc.Mapping): + return { + _render(key, context): _render(item, context) for key, item in value.items() + } + if isinstance(value, tuple): + return tuple(_render(item, context) for item in value) + if isinstance(value, list): + return [_render(item, context) for item in value] + return value + + +def _render_string(value: str, context: collections.abc.Mapping[str, object]) -> str: + """Render known ``$name`` tokens and leave unknown shell text intact.""" + + def repl(match: re.Match[str]) -> str: + if match.group("escaped") is not None: + return "$" + key = match.group("braced") or match.group("named") + if key is None or key not in context: + return match.group(0) + return str(context[key]) + + return _TOKEN_RE.sub(repl, value) diff --git a/src/libtmux/experimental/workspace/sets.py b/src/libtmux/experimental/workspace/sets.py new file mode 100644 index 000000000..157ab2971 --- /dev/null +++ b/src/libtmux/experimental/workspace/sets.py @@ -0,0 +1,441 @@ +"""Batch declarative workspaces into one folded Core plan. + +``WorkspaceSet`` is the Declarative tier's collection primitive: a group of +workspace specs that compile into one :class:`~libtmux.experimental.ops.LazyPlan` +and therefore run through the same chainable, async-capable engine path as a +single workspace. It is deliberately still a library value -- no database, server +process, or product workflow -- so callers can layer worktrees, dashboards, or +agent launch policy outside libtmux. +""" + +from __future__ import annotations + +import dataclasses +import typing as t +from dataclasses import dataclass, field + +from libtmux.experimental.ops import HasSession, KillSession, LazyPlan, arun, run +from libtmux.experimental.ops._types import NameRef, SlotRef, Target +from libtmux.experimental.ops.plan import PlanResult, StepReport +from libtmux.experimental.ops.planner import BoundedPlanner, MarkedPlanner +from libtmux.experimental.workspace.compiler import Compiled, HostStep, compile_full +from libtmux.experimental.workspace.events import WorkspaceBuilt, events_for +from libtmux.experimental.workspace.expand import ( + NameFactory, + Variant, + expand, +) +from libtmux.experimental.workspace.runner import _run_host_async, _run_host_sync + +if t.TYPE_CHECKING: + from collections.abc import Awaitable, Callable, Iterable, Mapping + + from typing_extensions import Self + + from libtmux.experimental.engines.base import AsyncTmuxEngine, TmuxEngine + from libtmux.experimental.ops.operation import Operation + from libtmux.experimental.ops.planner import Planner + from libtmux.experimental.workspace.events import BuildEvent + from libtmux.experimental.workspace.ir import Workspace + + +@dataclass(frozen=True) +class CompiledWorkspaceSet: + """A merged workspace-set plan plus batch metadata. + + Parameters + ---------- + plan : LazyPlan + The combined Core operation spine. + host_after : Mapping[int, tuple[HostStep, ...]] + Host steps scheduled after rebased operation indices. + pre : tuple[HostStep, ...] + Host steps to run before the first operation. + sessions : tuple[str, ...] + Session names in the input order. + session_slots : Mapping[str, int] + The plan index of each workspace's ``new-session`` operation. + end_indices : Mapping[str, int] + The final operation index for each workspace. + """ + + plan: LazyPlan + host_after: Mapping[int, tuple[HostStep, ...]] = field(default_factory=dict) + pre: tuple[HostStep, ...] = () + sessions: tuple[str, ...] = () + session_slots: Mapping[str, int] = field(default_factory=dict) + end_indices: Mapping[str, int] = field(default_factory=dict) + + +@dataclass(frozen=True) +class WorkspaceSetResult: + """Result of building a workspace set.""" + + result: PlanResult + sessions: tuple[str, ...] + reused: tuple[str, ...] = () + + @property + def ok(self) -> bool: + """Whether every dispatched operation completed successfully.""" + return self.result.ok + + @property + def bindings(self) -> dict[int | tuple[int, str], str]: + """Forward-ref bindings from the underlying plan result.""" + return self.result.bindings + + def raise_for_status(self) -> Self: + """Raise on the first failed operation; return ``self`` when OK.""" + self.result.raise_for_status() + return self + + +@dataclass(frozen=True) +class WorkspaceSet: + """A collection of declared workspaces compiled and built as one unit. + + Examples + -------- + >>> from libtmux.experimental.engines import ConcreteEngine + >>> from libtmux.experimental.workspace import Pane, Window, Workspace + >>> ws = Workspace("dev", windows=[Window("w", panes=[Pane("echo hi")])]) + >>> WorkspaceSet((ws,)).build(ConcreteEngine(), preflight=False).ok + True + """ + + workspaces: tuple[Workspace, ...] + + def __init__(self, workspaces: Iterable[Workspace]) -> None: + object.__setattr__(self, "workspaces", _workspace_tuple(workspaces)) + + @classmethod + def from_variants( + cls, + workspace: Workspace, + variants: Iterable[Variant], + *, + variables: Mapping[str, object] | None = None, + name: NameFactory | None = None, + ) -> WorkspaceSet: + """Expand *workspace* over *variants* and wrap the rendered specs.""" + return cls(expand(workspace, variants, variables=variables, name=name)) + + def compile(self, *, version: str | None = None) -> CompiledWorkspaceSet: + """Compile this set into one rebased Core plan.""" + return compile_workspaces(self.workspaces, version=version) + + def build( + self, + engine: TmuxEngine, + *, + version: str | None = None, + preflight: bool = True, + on_event: Callable[[BuildEvent], None] | None = None, + planner: Planner | None = None, + ) -> WorkspaceSetResult: + """Build this set synchronously over *engine*.""" + return build_workspaces( + self.workspaces, + engine, + version=version, + preflight=preflight, + on_event=on_event, + planner=planner, + ) + + async def abuild( + self, + engine: AsyncTmuxEngine, + *, + version: str | None = None, + preflight: bool = True, + on_event: Callable[[BuildEvent], Awaitable[None]] | None = None, + planner: Planner | None = None, + ) -> WorkspaceSetResult: + """Build this set asynchronously over *engine*.""" + return await abuild_workspaces( + self.workspaces, + engine, + version=version, + preflight=preflight, + on_event=on_event, + planner=planner, + ) + + +def _workspace_tuple(workspaces: Iterable[Workspace]) -> tuple[Workspace, ...]: + """Return workspaces as a tuple, rejecting duplicate session names.""" + rows = tuple(workspaces) + seen: set[str] = set() + duplicates: list[str] = [] + for workspace in rows: + if workspace.name in seen: + duplicates.append(workspace.name) + seen.add(workspace.name) + if duplicates: + msg = f"workspace set declares duplicate sessions: {', '.join(duplicates)}" + raise ValueError(msg) + return rows + + +def _rebase_slot(ref: SlotRef, offset: int) -> SlotRef: + """Return *ref* shifted by *offset* operation slots.""" + return dataclasses.replace(ref, slot=ref.slot + offset) + + +def _rebase_target(target: Target | None, offset: int) -> Target | None: + """Shift deferred targets by *offset* while leaving concrete ids unchanged.""" + if isinstance(target, SlotRef): + return _rebase_slot(target, offset) + return target + + +def _rebase_operation(operation: Operation[t.Any], offset: int) -> Operation[t.Any]: + """Shift operation targets from a per-workspace plan into the merged plan.""" + return dataclasses.replace( + operation, + target=_rebase_target(operation.target, offset), + src_target=_rebase_target(operation.src_target, offset), + ) + + +def _rebase_host_step(step: HostStep, offset: int) -> HostStep: + """Shift the pane ref carried by a host step, if any.""" + if step.pane is None: + return step + return dataclasses.replace(step, pane=_rebase_slot(step.pane, offset)) + + +def _extend_plan(plan: LazyPlan, compiled: Compiled, offset: int) -> None: + """Append one compiled workspace to *plan* with rebased refs.""" + for operation in compiled.plan.operations: + plan.add(_rebase_operation(operation, offset)) + + +def compile_workspaces( + workspaces: Iterable[Workspace], + *, + version: str | None = None, +) -> CompiledWorkspaceSet: + """Compile multiple workspaces into one rebased Core plan. + + Examples + -------- + >>> from libtmux.experimental.workspace import Pane, Window, Workspace + >>> compiled = compile_workspaces([ + ... Workspace("a", windows=[Window("w", panes=[Pane("one")])]), + ... Workspace("b", windows=[Window("w", panes=[Pane("two")])]), + ... ]) + >>> [op.kind for op in compiled.plan.operations].count("new_session") + 2 + """ + rows = _workspace_tuple(workspaces) + plan = LazyPlan() + pre: list[HostStep] = [] + host_after: dict[int, list[HostStep]] = {} + session_slots: dict[str, int] = {} + end_indices: dict[str, int] = {} + + for workspace in rows: + offset = len(plan) + compiled = compile_full(workspace, version=version) + if offset == 0: + pre.extend(_rebase_host_step(step, offset) for step in compiled.pre) + elif compiled.pre: + host_after.setdefault(offset - 1, []).extend( + _rebase_host_step(step, offset) for step in compiled.pre + ) + + for index, steps in compiled.host_after.items(): + host_after.setdefault(index + offset, []).extend( + _rebase_host_step(step, offset) for step in steps + ) + + _extend_plan(plan, compiled, offset) + if len(compiled.plan) > 0: + session_slots[workspace.name] = offset + end_indices[workspace.name] = offset + len(compiled.plan) - 1 + + return CompiledWorkspaceSet( + plan, + {key: tuple(value) for key, value in host_after.items()}, + tuple(pre), + tuple(workspace.name for workspace in rows), + session_slots, + end_indices, + ) + + +def _preflight_sync( + workspace: Workspace, + engine: TmuxEngine, + version: str | None, +) -> bool: + """Apply one workspace's ``on_exists`` policy before a batch build.""" + exists = run(HasSession(target=NameRef(workspace.name)), engine, version=version) + if not exists.exists: + return False + if workspace.on_exists == "replace": + run(KillSession(target=NameRef(workspace.name)), engine, version=version) + return False + if workspace.on_exists == "reuse": + return True + msg = f"session {workspace.name!r} already exists (on_exists='error')" + raise FileExistsError(msg) + + +async def _preflight_async( + workspace: Workspace, + engine: AsyncTmuxEngine, + version: str | None, +) -> bool: + """Async sibling of :func:`_preflight_sync`.""" + result = await arun( + HasSession(target=NameRef(workspace.name)), + engine, + version=version, + ) + if not result.exists: + return False + if workspace.on_exists == "replace": + await arun(KillSession(target=NameRef(workspace.name)), engine, version=version) + return False + if workspace.on_exists == "reuse": + return True + msg = f"session {workspace.name!r} already exists (on_exists='error')" + raise FileExistsError(msg) + + +def _split_reused_sync( + workspaces: tuple[Workspace, ...], + engine: TmuxEngine, + version: str | None, + preflight: bool, +) -> tuple[tuple[Workspace, ...], tuple[str, ...]]: + """Return workspaces to build plus names skipped by ``on_exists='reuse'``.""" + if not preflight: + return workspaces, () + active: list[Workspace] = [] + reused: list[str] = [] + for workspace in workspaces: + if _preflight_sync(workspace, engine, version): + reused.append(workspace.name) + else: + active.append(workspace) + return tuple(active), tuple(reused) + + +async def _split_reused_async( + workspaces: tuple[Workspace, ...], + engine: AsyncTmuxEngine, + version: str | None, + preflight: bool, +) -> tuple[tuple[Workspace, ...], tuple[str, ...]]: + """Async sibling of :func:`_split_reused_sync`.""" + if not preflight: + return workspaces, () + active: list[Workspace] = [] + reused: list[str] = [] + for workspace in workspaces: + if await _preflight_async(workspace, engine, version): + reused.append(workspace.name) + else: + active.append(workspace) + return tuple(active), tuple(reused) + + +def build_workspaces( + workspaces: Iterable[Workspace], + engine: TmuxEngine, + *, + version: str | None = None, + preflight: bool = True, + on_event: Callable[[BuildEvent], None] | None = None, + planner: Planner | None = None, +) -> WorkspaceSetResult: + """Compile and execute multiple workspaces synchronously over *engine*.""" + rows = _workspace_tuple(workspaces) + active, reused = _split_reused_sync(rows, engine, version, preflight) + if not active: + return WorkspaceSetResult( + PlanResult((), {}), + tuple(ws.name for ws in rows), + reused, + ) + + compiled = compile_workspaces(active, version=version) + ops = compiled.plan.operations + end_to_session = {index: name for name, index in compiled.end_indices.items()} + for step in compiled.pre: + _run_host_sync(step, engine, {}, version) + + def on_step(report: StepReport) -> None: + for index, result in zip(report.step.indices, report.results, strict=True): + if on_event is not None: + for event in events_for(ops[index], result): + on_event(event) + for host_step in compiled.host_after.get(index, ()): + _run_host_sync(host_step, engine, report.bindings, version) + if on_event is not None and index in end_to_session: + slot = compiled.session_slots[end_to_session[index]] + on_event(WorkspaceBuilt(report.bindings.get(slot, ""))) + + result = compiled.plan.execute( + engine, + version=version, + planner=BoundedPlanner( + planner or MarkedPlanner(), + frozenset(compiled.host_after), + ), + on_step=on_step, + ) + return WorkspaceSetResult(result, tuple(ws.name for ws in rows), reused) + + +async def abuild_workspaces( + workspaces: Iterable[Workspace], + engine: AsyncTmuxEngine, + *, + version: str | None = None, + preflight: bool = True, + on_event: Callable[[BuildEvent], Awaitable[None]] | None = None, + planner: Planner | None = None, +) -> WorkspaceSetResult: + """Compile and execute multiple workspaces asynchronously over *engine*.""" + rows = _workspace_tuple(workspaces) + active, reused = await _split_reused_async(rows, engine, version, preflight) + if not active: + return WorkspaceSetResult( + PlanResult((), {}), + tuple(ws.name for ws in rows), + reused, + ) + + compiled = compile_workspaces(active, version=version) + ops = compiled.plan.operations + end_to_session = {index: name for name, index in compiled.end_indices.items()} + for step in compiled.pre: + await _run_host_async(step, engine, {}, version) + + async def on_step(report: StepReport) -> None: + for index, result in zip(report.step.indices, report.results, strict=True): + if on_event is not None: + for event in events_for(ops[index], result): + await on_event(event) + for host_step in compiled.host_after.get(index, ()): + await _run_host_async(host_step, engine, report.bindings, version) + if on_event is not None and index in end_to_session: + slot = compiled.session_slots[end_to_session[index]] + await on_event(WorkspaceBuilt(report.bindings.get(slot, ""))) + + result = await compiled.plan.aexecute( + engine, + version=version, + planner=BoundedPlanner( + planner or MarkedPlanner(), + frozenset(compiled.host_after), + ), + on_step=on_step, + ) + return WorkspaceSetResult(result, tuple(ws.name for ws in rows), reused) diff --git a/tests/experimental/contract/test_workspace_expand.py b/tests/experimental/contract/test_workspace_expand.py new file mode 100644 index 000000000..fe7a80629 --- /dev/null +++ b/tests/experimental/contract/test_workspace_expand.py @@ -0,0 +1,66 @@ +"""Tests for pure workspace variant expansion.""" + +from __future__ import annotations + +from libtmux.experimental.workspace import Pane, Window, Workspace, expand + + +def test_expand_renders_variants_without_mutating_base() -> None: + """Expand returns one rendered workspace per variant and leaves the base pure.""" + base = Workspace( + name="svc-$app", + start_directory="${root}/$app", + environment={"APP": "$app", "UNCHANGED": "$HOME"}, + windows=[ + Window( + name="$app", + panes=[ + Pane( + run=["cd ${root}/$app", "$cmd", "echo $(pwd) #{pane_id}"], + environment={"APP": "$app"}, + ), + ], + ), + ], + ) + + expanded = expand( + base, + [ + {"app": "api", "cmd": "uvicorn app:app"}, + {"app": "worker", "cmd": "python worker.py"}, + ], + variables={"root": "/srv"}, + ) + + assert [ws.name for ws in expanded] == ["svc-api", "svc-worker"] + assert expanded[0].start_directory == "/srv/api" + assert expanded[1].windows[0].name == "worker" + assert [cmd.cmd for cmd in expanded[0].windows[0].panes[0].commands] == [ + "cd /srv/api", + "uvicorn app:app", + "echo $(pwd) #{pane_id}", + ] + assert expanded[0].environment == {"APP": "api", "UNCHANGED": "$HOME"} + assert expanded[0].windows[0].panes[0].environment == {"APP": "api"} + assert base.name == "svc-$app" + assert base.windows[0].panes[0].commands[0].cmd == "cd ${root}/$app" + + +def test_expand_name_callable_controls_workspace_name() -> None: + """A name callable can build names outside the template strings.""" + base = Workspace(name="dev", windows=[Window("py-$python", panes=[Pane("tox")])]) + + expanded = expand( + base, + [{"python": "3.12"}, {"python": "3.13"}], + name=lambda base_name, variant: f"{base_name}-py{variant['python']}", + ) + + assert [ws.name for ws in expanded] == ["dev-py3.12", "dev-py3.13"] + assert [ws.windows[0].name for ws in expanded] == ["py-3.12", "py-3.13"] + + +def test_expand_empty_variants_returns_empty_tuple() -> None: + """No variants means no expanded workspaces.""" + assert expand(Workspace(name="dev"), []) == () diff --git a/tests/experimental/contract/test_workspace_sets.py b/tests/experimental/contract/test_workspace_sets.py new file mode 100644 index 000000000..4a2f4bd37 --- /dev/null +++ b/tests/experimental/contract/test_workspace_sets.py @@ -0,0 +1,167 @@ +"""Workspace sets batch declarative builds without losing plan semantics.""" + +from __future__ import annotations + +import asyncio +import dataclasses +import typing as t + +from libtmux.experimental.engines import AsyncConcreteEngine, ConcreteEngine +from libtmux.experimental.engines.base import CommandResult +from libtmux.experimental.ops import SequentialPlanner +from libtmux.experimental.ops._types import SlotRef +from libtmux.experimental.workspace import ( + BuildEvent, + Pane, + Window, + Workspace, + WorkspaceBuilt, + WorkspaceSet, + build_workspaces, + compile_workspaces, +) + +if t.TYPE_CHECKING: + from collections.abc import Sequence + + from libtmux.experimental.engines.base import CommandRequest, TmuxEngine + + +@dataclasses.dataclass +class _RecordingEngine: + """Record dispatches while forwarding to an inner engine.""" + + inner: TmuxEngine = dataclasses.field(default_factory=ConcreteEngine) + calls: list[tuple[str, ...]] = dataclasses.field(default_factory=list) + + def run(self, request: CommandRequest) -> CommandResult: + """Record the argv and forward, faking a ready cursor for waits.""" + self.calls.append(request.args) + if "display-message" in request.args: + return CommandResult(cmd=("tmux", *request.args), stdout=("1,1",)) + return self.inner.run(request) + + def run_batch(self, requests: Sequence[CommandRequest]) -> list[CommandResult]: + """Execute each request in order.""" + return [self.run(req) for req in requests] + + +def _workspace(name: str, *, wait_pane: bool = False) -> Workspace: + """Return a two-pane workspace with a command after a split.""" + return Workspace( + name=name, + windows=[ + Window( + "editor", + panes=[ + Pane(run="echo first"), + Pane(run=["echo second", "echo third"]), + ], + ), + ], + wait_pane=wait_pane, + ) + + +def test_workspace_set_from_variants_expands_base() -> None: + """WorkspaceSet.from_variants delegates to expand and preserves ordering.""" + base = Workspace(name="dev-${app}", windows=[Window("w", panes=[Pane("${cmd}")])]) + workspace_set = WorkspaceSet.from_variants( + base, + [{"app": "api", "cmd": "pytest"}, {"app": "docs", "cmd": "sphinx-build"}], + ) + + assert [ws.name for ws in workspace_set.workspaces] == ["dev-api", "dev-docs"] + assert [ + ws.windows[0].panes[0].commands[0].cmd for ws in workspace_set.workspaces + ] == ["pytest", "sphinx-build"] + + +def test_compile_workspaces_rebases_slot_refs_and_host_steps() -> None: + """Merged plans offset later workspaces' SlotRefs and host-step targets.""" + compiled = compile_workspaces( + [ + _workspace("one"), + _workspace("two", wait_pane=True), + ], + ) + first_len = len(_workspace("one").compile().operations) + second_ops = compiled.plan.operations[first_len:] + send_ops = [op for op in second_ops if op.kind == "send_keys"] + assert send_ops + deferred_targets = [op.target for op in send_ops if isinstance(op.target, SlotRef)] + assert min(target.slot for target in deferred_targets) >= first_len + + wait_steps = [ + step + for steps in compiled.host_after.values() + for step in steps + if step.kind == "wait_pane" + ] + assert wait_steps + assert all( + step.pane is not None and step.pane.slot >= first_len for step in wait_steps + ) + + +def test_build_workspaces_folds_across_workspace_boundaries() -> None: + """Batch builds still use the folding planner over the merged operation stream.""" + default = _RecordingEngine() + build_workspaces([_workspace("one"), _workspace("two")], default, preflight=False) + sequential = _RecordingEngine() + build_workspaces( + [_workspace("one"), _workspace("two")], + sequential, + preflight=False, + planner=SequentialPlanner(), + ) + + assert len(default.calls) < len(sequential.calls) + assert any(";" in argv for argv in default.calls) + + +def test_workspace_set_all_reused_returns_noop_result() -> None: + """Preflight reuse skips every existing workspace without executing the plan.""" + reused = Workspace( + name="already", + windows=[Window("w", panes=[Pane("echo nope")])], + on_exists="reuse", + ) + engine = ConcreteEngine() + + first = build_workspaces([reused], engine, preflight=False) + second = build_workspaces([reused], engine) + + assert first.ok + assert second.ok + assert second.reused == ("already",) + assert second.result.results == () + + +def test_workspace_set_emits_built_event_per_workspace() -> None: + """Each workspace emits its own WorkspaceBuilt event.""" + events: list[BuildEvent] = [] + outcome = build_workspaces( + [_workspace("one"), _workspace("two")], + ConcreteEngine(), + preflight=False, + on_event=events.append, + ) + + built = [event for event in events if isinstance(event, WorkspaceBuilt)] + assert outcome.ok + assert len(built) == 2 + + +def test_workspace_set_async_build_matches_sync_shape() -> None: + """The async runner exposes the same result shape as the sync runner.""" + workspace_set = WorkspaceSet((_workspace("one"), _workspace("two"))) + outcome = asyncio.run( + workspace_set.abuild(AsyncConcreteEngine(), preflight=False), + ) + + assert outcome.ok + assert outcome.sessions == ("one", "two") + assert len(outcome.result.results) == len( + compile_workspaces(workspace_set.workspaces).plan.operations, + ) From 5ae8cf9e002d37c97e24044e1202863d70288cd8 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 4 Jul 2026 13:07:52 -0500 Subject: [PATCH 05/13] Ops(feat): Add plan explain() and astream() why: folding collapses N ops into a few dispatches, which hides per-op structure -- a caller can't see why the chain broke where it did, and had no way to observe a plan step by step as it ran. what: - LazyPlan.explain(planner): annotate each dispatch step with why it is a boundary (marked-fold/folded/created-id/capture/single) -- pure, no I/O - LazyPlan.astream(engine): async-generator twin of aexecute over the same drive core, yielding a StepDone per bound step and a terminal PlanDone; pull-based, so backpressure needs no buffer and the loop never blocks - Export StepExplanation, StepDone, PlanDone, PlanEvent --- src/libtmux/experimental/ops/__init__.py | 14 ++- src/libtmux/experimental/ops/plan.py | 120 ++++++++++++++++++++++- tests/experimental/ops/test_plan.py | 72 ++++++++++++++ 3 files changed, 204 insertions(+), 2 deletions(-) diff --git a/src/libtmux/experimental/ops/__init__.py b/src/libtmux/experimental/ops/__init__.py index 75d58453e..131020366 100644 --- a/src/libtmux/experimental/ops/__init__.py +++ b/src/libtmux/experimental/ops/__init__.py @@ -107,7 +107,15 @@ ) from libtmux.experimental.ops.execute import arun, run from libtmux.experimental.ops.operation import Operation -from libtmux.experimental.ops.plan import LazyPlan, PlanResult, StepReport +from libtmux.experimental.ops.plan import ( + LazyPlan, + PlanDone, + PlanEvent, + PlanResult, + StepDone, + StepExplanation, + StepReport, +) from libtmux.experimental.ops.planner import ( BoundedPlanner, FoldingPlanner, @@ -202,6 +210,8 @@ "PaneId", "PasteBuffer", "PipePane", + "PlanDone", + "PlanEvent", "PlanResult", "PlanStep", "Planner", @@ -241,6 +251,8 @@ "SplitWindowResult", "StartServer", "Status", + "StepDone", + "StepExplanation", "StepReport", "SuspendClient", "SwapPane", diff --git a/src/libtmux/experimental/ops/plan.py b/src/libtmux/experimental/ops/plan.py index 8b3bb52f6..83f25ba11 100644 --- a/src/libtmux/experimental/ops/plan.py +++ b/src/libtmux/experimental/ops/plan.py @@ -38,7 +38,7 @@ from libtmux.experimental.ops.serialize import operation_from_dict, operation_to_dict if t.TYPE_CHECKING: - from collections.abc import Generator, Iterator + from collections.abc import AsyncGenerator, Generator, Iterator from typing_extensions import Self @@ -99,6 +99,40 @@ class _Host: report: StepReport +@dataclass(frozen=True) +class StepExplanation: + """Why one dispatch step is its own tmux call (from :meth:`LazyPlan.explain`). + + ``reason`` is one of ``"marked-fold"`` (a pane create plus its ``{marked}`` + decorates), ``"folded"`` (a ``;``-chain of chainable ops), ``"created-id"`` + (a create whose captured id a later op must target -- a true blocker), + ``"capture"`` (a non-chainable op whose stdout can't merge into a chain), or + ``"single"`` (a lone chainable op with nothing to fold with). + """ + + step: PlanStep + kinds: tuple[str, ...] + reason: str + + +@dataclass(frozen=True) +class StepDone: + """Stream event: a plan step finished and its results have bound.""" + + report: StepReport + + +@dataclass(frozen=True) +class PlanDone: + """Stream event: the plan finished; carries the full :class:`PlanResult`.""" + + result: PlanResult + + +#: An event yielded by :meth:`LazyPlan.astream`. +PlanEvent = StepDone | PlanDone + + def _target_from_id(value: str) -> Target: """Map a captured concrete id back to its typed target.""" if value.startswith("%"): @@ -275,6 +309,45 @@ def _render(op: Operation[t.Any]) -> tuple[str, ...] | None: return [_render(op) for op in self._operations] + def explain(self, planner: Planner | None = None) -> list[StepExplanation]: + """Explain why *planner* breaks the plan into the dispatches it does. + + A pure companion to :meth:`preview`: folding hides per-op structure, so + this annotates each dispatch step with the reason it can't fold further + (see :class:`StepExplanation`). Defaults to + :class:`~.planner.SequentialPlanner`. + + Examples + -------- + >>> from libtmux.experimental.ops import SplitWindow, SendKeys, MarkedPlanner + >>> from libtmux.experimental.ops._types import WindowId + >>> plan = LazyPlan() + >>> pane = plan.add(SplitWindow(target=WindowId("@1"))) + >>> _ = plan.add(SendKeys(target=pane, keys="vim")) + >>> [(e.kinds, e.reason) for e in plan.explain(MarkedPlanner())] + [(('split_window', 'send_keys'), 'marked-fold')] + >>> [(e.kinds, e.reason) for e in plan.explain()] + [(('split_window',), 'created-id'), (('send_keys',), 'single')] + """ + steps = (planner or SequentialPlanner()).plan(self._operations) + out: list[StepExplanation] = [] + for step in steps: + kinds = tuple(self._operations[i].kind for i in step.indices) + if step.marked: + reason = "marked-fold" + elif len(step.indices) > 1: + reason = "folded" + else: + op = self._operations[step.indices[0]] + if op.effects.creates is not None: + reason = "created-id" + elif not op.chainable: + reason = "capture" + else: + reason = "single" + out.append(StepExplanation(step, kinds, reason)) + return out + def _drive( self, version: str | None, @@ -413,3 +486,48 @@ async def _adispatch( if isinstance(request, _Chain): return await engine.run(CommandRequest.from_args(*request.argv)) return await arun(request.op, engine, version=version) + + async def astream( + self, + engine: AsyncTmuxEngine, + *, + version: str | None = None, + planner: Planner | None = None, + ) -> AsyncGenerator[PlanEvent, None]: + """Execute the plan, streaming a :data:`PlanEvent` per step as it binds. + + The observe-as-you-go twin of :meth:`aexecute` over the same sans-I/O + resolution core: it yields a :class:`StepDone` after each dispatch binds + and a terminal :class:`PlanDone` carrying the full :class:`PlanResult`, so + ``[e async for e in plan.astream(engine)][-1].result`` equals ``await + plan.aexecute(engine)``. The stream is pull-based -- a slow ``async for`` + naturally paces the plan, so backpressure needs no buffer and the event + loop is never blocked between dispatches. Run one ``astream`` per engine + at a time (the engine's write order is shared). + + Examples + -------- + >>> import asyncio + >>> from libtmux.experimental.engines.concrete import AsyncConcreteEngine + >>> from libtmux.experimental.ops import SendKeys + >>> from libtmux.experimental.ops._types import PaneId + >>> plan = LazyPlan() + >>> _ = plan.add(SendKeys(target=PaneId("%1"), keys="vim")) + >>> async def drain() -> list[str]: + ... engine = AsyncConcreteEngine() + ... return [type(e).__name__ async for e in plan.astream(engine)] + >>> asyncio.run(drain()) + ['StepDone', 'PlanDone'] + """ + version = resolve_engine_version(engine, version) + gen = self._drive(version, planner or SequentialPlanner()) + try: + request = next(gen) + while True: + if isinstance(request, _Host): + yield StepDone(request.report) # pull point: consumer paces here + request = gen.send(None) + else: + request = gen.send(await self._adispatch(request, engine, version)) + except StopIteration as stop: + yield PlanDone(t.cast("PlanResult", stop.value)) diff --git a/tests/experimental/ops/test_plan.py b/tests/experimental/ops/test_plan.py index cb5a8f24b..43ce28ebc 100644 --- a/tests/experimental/ops/test_plan.py +++ b/tests/experimental/ops/test_plan.py @@ -211,3 +211,75 @@ def test_plan_unresolvable_ref_fails_closed() -> None: assert exc_info.value.slot == 0 # points at the non-capturing creator # ForwardCaptureError stays an OperationError, so broad handlers keep working assert isinstance(exc_info.value, OperationError) + + +class _ExplainCase(t.NamedTuple): + """A planner and the (kinds, reason) each of its dispatch steps should carry.""" + + test_id: str + planner: t.Any + expected: list[tuple[tuple[str, ...], str]] + + +def _split_then_send() -> LazyPlan: + plan = LazyPlan() + pane = plan.add(SplitWindow(target=WindowId("@1"))) + plan.add(SendKeys(target=pane, keys="vim")) + return plan + + +_EXPLAIN_CASES: tuple[_ExplainCase, ...] = ( + _ExplainCase( + "sequential_created_then_single", + SequentialPlanner(), + [(("split_window",), "created-id"), (("send_keys",), "single")], + ), + _ExplainCase( + "marked_fold", + MarkedPlanner(), + [(("split_window", "send_keys"), "marked-fold")], + ), +) + + +@pytest.mark.parametrize( + "case", + _EXPLAIN_CASES, + ids=[c.test_id for c in _EXPLAIN_CASES], +) +def test_explain_annotates_dispatch_boundaries(case: _ExplainCase) -> None: + """explain() reports why each step is its own dispatch under a planner.""" + steps = _split_then_send().explain(case.planner) + assert [(e.kinds, e.reason) for e in steps] == case.expected + + +def test_astream_yields_step_then_plan_done() -> None: + """astream() streams a StepDone per step and a terminal PlanDone.""" + from libtmux.experimental.ops import PlanDone, StepDone + + plan = _split_then_send() + + async def drain() -> list[object]: + return [event async for event in plan.astream(AsyncConcreteEngine())] + + events = asyncio.run(drain()) + assert [type(e).__name__ for e in events] == ["StepDone", "StepDone", "PlanDone"] + assert isinstance(events[-1], PlanDone) + assert isinstance(events[0], StepDone) + # the terminal PlanDone carries the same result aexecute() would return + assert events[-1].result.ok + + +def test_astream_last_result_matches_aexecute() -> None: + """The terminal PlanDone.result equals what aexecute() returns.""" + from libtmux.experimental.ops import PlanDone + + async def both() -> tuple[bool, bool]: + streamed = [e async for e in _split_then_send().astream(AsyncConcreteEngine())] + direct = await _split_then_send().aexecute(AsyncConcreteEngine()) + last = streamed[-1] + assert isinstance(last, PlanDone) + return last.result.ok, direct.ok + + stream_ok, direct_ok = asyncio.run(both()) + assert stream_ok == direct_ok From 5bc95ba6299226bf5ce228bc6db1c7909def011f Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 4 Jul 2026 13:27:05 -0500 Subject: [PATCH 06/13] Fluent(feat): Add sleep/wait host boundaries why: some builds need a real pause between dispatches -- a fixed delay, or waiting for a pane's shell prompt before sending the next command. A host step is a true blocker: it can't fold into a tmux chain, so the fold must break around it. what: - PlanBuilder.sleep(seconds): pause after the last recorded op - PlanBuilder.wait(pane): poll the pane's cursor until its prompt draws - run()/arun() bound the planner (BoundedPlanner) at each host boundary and run the recorded step from the on_step hook, sync and async --- src/libtmux/experimental/fluent.py | 118 +++++++++++++++++++++++++++-- tests/experimental/test_fluent.py | 48 ++++++++++++ 2 files changed, 160 insertions(+), 6 deletions(-) diff --git a/src/libtmux/experimental/fluent.py b/src/libtmux/experimental/fluent.py index 5641287aa..b0e6bf671 100644 --- a/src/libtmux/experimental/fluent.py +++ b/src/libtmux/experimental/fluent.py @@ -31,22 +31,47 @@ from __future__ import annotations +import asyncio +import time import typing as t from dataclasses import dataclass, field from libtmux.experimental.ops import ( + BoundedPlanner, + DisplayMessage, LazyPlan, MarkedPlanner, NameRef, NewSession, NewWindow, + arun, + run, ) -from libtmux.experimental.query import ForwardPaneRef +from libtmux.experimental.ops.plan import _resolve +from libtmux.experimental.query import ForwardPaneRef, _PaneRefBase if t.TYPE_CHECKING: from libtmux.experimental.engines.base import AsyncTmuxEngine, TmuxEngine - from libtmux.experimental.ops import Planner, PlanResult - from libtmux.experimental.ops._types import SlotRef + from libtmux.experimental.ops import Planner, PlanResult, StepReport + from libtmux.experimental.ops._types import SlotRef, Target + +_CURSOR_FMT = "#{cursor_x},#{cursor_y}" +_WAIT_PANE_POLLS = 40 +_WAIT_PANE_INTERVAL = 0.05 + + +def _pane_ready(cursor: str) -> bool: + """Whether a pane's cursor has left the origin (its shell prompt drew).""" + return bool(cursor) and cursor != "0,0" + + +@dataclass(frozen=True) +class _HostAction: + """A host-side pause recorded after an operation (a hard fold boundary).""" + + kind: t.Literal["sleep", "wait"] + seconds: float = 0.0 + pane: Target | None = None @dataclass(frozen=True) @@ -116,6 +141,7 @@ class PlanBuilder: """A fluent recorder over a :class:`LazyPlan`; :meth:`run` folds by default.""" plan: LazyPlan = field(default_factory=LazyPlan) + _host_after: dict[int, list[_HostAction]] = field(default_factory=dict) def new_session(self, name: str) -> SessionRef: """Create a session, capturing its first pane for forward refs. @@ -132,6 +158,53 @@ def new_session(self, name: str) -> SessionRef: slot = self.plan.add(NewSession(session_name=name, capture_panes=True)) return SessionRef(self.plan, name, slot) + def sleep(self, seconds: float) -> PlanBuilder: + """Pause *seconds* after the last recorded op (a hard fold boundary). + + A host step never folds into a ``tmux`` dispatch, so the chain breaks + before and after it; the pause runs between dispatches at build time. + + Examples + -------- + >>> from libtmux.experimental.engines.concrete import ConcreteEngine + >>> p = plan() + >>> pane = p.new_session("dev").window().pane() + >>> _ = pane.do(lambda c: c.send_keys("slow-start")) + >>> p.sleep(0.0).run(ConcreteEngine()).ok + True + """ + self._record_host(_HostAction("sleep", seconds=seconds)) + return self + + def wait(self, pane: _PaneRefBase) -> PlanBuilder: + """Wait for *pane*'s shell prompt before the next dispatch (anti-race). + + Polls the pane's cursor until it leaves the origin, so a follow-up + command isn't sent before the shell is ready. A hard fold boundary. + + Examples + -------- + >>> p = plan() + >>> pane = p.new_session("dev").window().pane() + >>> p.wait(pane) is p + True + """ + self._record_host(_HostAction("wait", pane=pane.target)) + return self + + def _record_host(self, action: _HostAction) -> None: + """Record *action* after the plan's current last operation.""" + index = len(self.plan.operations) - 1 + if index >= 0: + self._host_after.setdefault(index, []).append(action) + + def _planner(self, planner: Planner | None) -> Planner: + """Return the base planner, bounded by host-step boundaries if any.""" + base = planner or MarkedPlanner() + if self._host_after: + return BoundedPlanner(base, frozenset(self._host_after)) + return base + def run( self, engine: TmuxEngine, @@ -149,10 +222,26 @@ def run( >>> p.run(ConcreteEngine()).ok True """ + + def on_step(report: StepReport) -> None: + for action in self._host_after.get(report.step.indices[-1], ()): + if action.kind == "sleep": + time.sleep(action.seconds) + elif action.pane is not None: + op = _resolve( + DisplayMessage(target=action.pane, message=_CURSOR_FMT), + report.bindings, + ) + for _ in range(_WAIT_PANE_POLLS): + if _pane_ready(run(op, engine, version=version).text): + break + time.sleep(_WAIT_PANE_INTERVAL) + return self.plan.execute( engine, version=version, - planner=planner or MarkedPlanner(), + planner=self._planner(planner), + on_step=on_step, ) async def arun( @@ -162,11 +251,28 @@ async def arun( version: str | None = None, planner: Planner | None = None, ) -> PlanResult: - """Async twin of :meth:`run` (same fold, ``await``ed).""" + """Async twin of :meth:`run` (same fold and host steps, ``await``ed).""" + + async def on_step(report: StepReport) -> None: + for action in self._host_after.get(report.step.indices[-1], ()): + if action.kind == "sleep": + await asyncio.sleep(action.seconds) + elif action.pane is not None: + op = _resolve( + DisplayMessage(target=action.pane, message=_CURSOR_FMT), + report.bindings, + ) + for _ in range(_WAIT_PANE_POLLS): + result = await arun(op, engine, version=version) + if _pane_ready(result.text): + break + await asyncio.sleep(_WAIT_PANE_INTERVAL) + return await self.plan.aexecute( engine, version=version, - planner=planner or MarkedPlanner(), + planner=self._planner(planner), + on_step=on_step, ) def preview(self, *, version: str | None = None) -> list[tuple[str, ...] | None]: diff --git a/tests/experimental/test_fluent.py b/tests/experimental/test_fluent.py index 8e596b7a0..f126d328d 100644 --- a/tests/experimental/test_fluent.py +++ b/tests/experimental/test_fluent.py @@ -87,3 +87,51 @@ def test_build_session_live(session: Session) -> None: built = [s for s in session.server.sessions if s.session_name == "fluentdev"] assert len(built) == 1 assert len(built[0].windows[0].panes) == 2 + + +class _HostCase(t.NamedTuple): + """A host boundary recorded on the builder and its recorded action kind.""" + + test_id: str + record: t.Callable[[PlanBuilder, ForwardPaneRef], object] + kind: str + + +_HOST_CASES: tuple[_HostCase, ...] = ( + _HostCase("sleep", lambda p, _pane: p.sleep(0.0), "sleep"), + _HostCase("wait", lambda p, pane: p.wait(pane), "wait"), +) + + +@pytest.mark.parametrize("case", _HOST_CASES, ids=[c.test_id for c in _HOST_CASES]) +def test_host_step_recorded_after_last_op(case: _HostCase) -> None: + """sleep()/wait() record a host action keyed to the current last op.""" + p = plan() + pane = p.new_session("dev").window().pane() + case.record(p, pane) + assert list(p._host_after) == [0] # after new_session, the only op so far + assert len(p._host_after[0]) == 1 + assert p._host_after[0][0].kind == case.kind + + +def test_host_boundary_prevents_fold_across_it() -> None: + """No dispatch step may span a recorded host boundary (a true blocker).""" + p = plan() + pane = p.new_session("dev").window().pane() + pane.do(lambda c: c.send_keys("a")) # op 1 + p.sleep(0.0) # boundary after op 1 + pane.do(lambda c: c.send_keys("b")) # op 2 + + steps = p._planner(None).plan(p.plan.operations) + spanning = [s for s in steps if min(s.indices) <= 1 < max(s.indices)] + assert not spanning # nothing folds across the boundary at index 1 + + +def test_sleep_runs_offline() -> None: + """A build with a sleep boundary resolves and runs over the in-memory engine.""" + p = plan() + pane = p.new_session("dev").window().pane() + pane.do(lambda c: c.send_keys("vim")) + p.sleep(0.0) + pane.split().do(lambda c: c.send_keys("htop")) + assert p.run(ConcreteEngine()).ok From e73dff4bfe4fb45fc322243e00bfa357710f216b Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 4 Jul 2026 13:33:08 -0500 Subject: [PATCH 07/13] Mcp(feat): Add explain_plan tool why: an agent composing a plan over MCP could preview argv and query a result schema, but couldn't see why a planner folds or breaks the plan into the dispatches it does -- the fold hides that structure. what: - plan_tools.explain_plan(plan, planner): per-step boundary reasons (marked-fold/folded/created-id/capture/single), pure over a serialized plan - Register a readonly explain_plan tool in the fastmcp adapter - Export explain_plan from the mcp package --- src/libtmux/experimental/mcp/__init__.py | 2 ++ .../experimental/mcp/fastmcp_adapter.py | 12 +++++++++ src/libtmux/experimental/mcp/plan_tools.py | 27 +++++++++++++++++++ tests/experimental/mcp/test_mcp_projection.py | 14 ++++++++++ 4 files changed, 55 insertions(+) diff --git a/src/libtmux/experimental/mcp/__init__.py b/src/libtmux/experimental/mcp/__init__.py index 03e33ab6f..074e1254a 100644 --- a/src/libtmux/experimental/mcp/__init__.py +++ b/src/libtmux/experimental/mcp/__init__.py @@ -35,6 +35,7 @@ aexecute_plan, build_workspace, execute_plan, + explain_plan, preview_plan, result_schema, ) @@ -286,6 +287,7 @@ def main(argv: Sequence[str] | None = None) -> None: "default_async_server", "default_server", "execute_plan", + "explain_plan", "kill_pane", "kill_session", "kill_window", diff --git a/src/libtmux/experimental/mcp/fastmcp_adapter.py b/src/libtmux/experimental/mcp/fastmcp_adapter.py index 19589017a..0c2155604 100644 --- a/src/libtmux/experimental/mcp/fastmcp_adapter.py +++ b/src/libtmux/experimental/mcp/fastmcp_adapter.py @@ -459,6 +459,17 @@ def preview_plan( "argv": [list(item) if item is not None else None for item in preview.argv], } + def explain_plan( + operations: list[dict[str, t.Any]], + planner: str = "marked", + ) -> dict[str, t.Any]: + """Explain why *planner* folds or breaks a serialized plan (pure).""" + explanation = _plan.explain_plan( + _plan_from_dicts(operations), + planner=_planner(planner), + ) + return {"steps": explanation.steps} + def result_schema(kind: str) -> dict[str, t.Any]: """Report what an operation kind returns, for planning forward refs.""" schema = _plan.result_schema(reg, kind) @@ -471,6 +482,7 @@ def result_schema(kind: str) -> dict[str, t.Any]: tools: list[tuple[Callable[..., t.Any], str]] = [ (preview_plan, "readonly"), + (explain_plan, "readonly"), (result_schema, "readonly"), ] diff --git a/src/libtmux/experimental/mcp/plan_tools.py b/src/libtmux/experimental/mcp/plan_tools.py index 49da2edee..fb503af08 100644 --- a/src/libtmux/experimental/mcp/plan_tools.py +++ b/src/libtmux/experimental/mcp/plan_tools.py @@ -45,6 +45,33 @@ def preview_plan(plan: LazyPlan, *, version: str | None = None) -> PlanPreview: ) +@dataclass(frozen=True) +class PlanExplanation: + """Why a planner breaks a plan into its dispatch steps: one dict per step. + + Each entry carries ``indices`` (the operation indices in the step), + ``kinds`` (their operation kinds), and ``reason`` (the boundary reason -- + ``marked-fold`` / ``folded`` / ``created-id`` / ``capture`` / ``single``), so + an agent can see why a chain folds or breaks before it runs. + """ + + steps: list[dict[str, t.Any]] + + +def explain_plan(plan: LazyPlan, *, planner: Planner | None = None) -> PlanExplanation: + """Explain a plan's dispatch grouping under *planner* (pure, no engine).""" + return PlanExplanation( + steps=[ + { + "indices": list(entry.step.indices), + "kinds": list(entry.kinds), + "reason": entry.reason, + } + for entry in plan.explain(planner) + ], + ) + + @dataclass(frozen=True) class PlanOutcome: """The result of executing a plan: per-op result dicts + a bindings map.""" diff --git a/tests/experimental/mcp/test_mcp_projection.py b/tests/experimental/mcp/test_mcp_projection.py index 4bb50d2e1..8adfab6a3 100644 --- a/tests/experimental/mcp/test_mcp_projection.py +++ b/tests/experimental/mcp/test_mcp_projection.py @@ -13,12 +13,14 @@ OperationToolRegistry, build_workspace, execute_plan, + explain_plan, preview_plan, resolve_target, result_schema, ) from libtmux.experimental.ops import ( LazyPlan, + MarkedPlanner, NewSession, SendKeys, SplitWindow, @@ -107,6 +109,18 @@ def test_preview_plan_marks_unresolved_forward_refs() -> None: assert preview.ok is False +def test_explain_plan_reports_boundary_reasons() -> None: + """explain_plan annotates each dispatch step with why it can't fold further.""" + plan = LazyPlan() + pane = plan.add(SplitWindow(target=WindowId("@1"))) + plan.add(SendKeys(target=pane, keys="vim", enter=True)) + steps = explain_plan(plan, planner=MarkedPlanner()).steps + assert len(steps) == 1 + assert steps[0]["indices"] == [0, 1] + assert steps[0]["kinds"] == ["split_window", "send_keys"] + assert steps[0]["reason"] == "marked-fold" + + def test_execute_plan_returns_bindings() -> None: """execute_plan resolves forward refs and returns a JSON bindings map.""" plan = LazyPlan() From 1b01b3ea5dfcea3ff7e587521c782a6ed05c68f4 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 4 Jul 2026 13:50:39 -0500 Subject: [PATCH 08/13] Ops(feat): Add conditional find-or-create via ensure() why: rebuilding a workspace should reuse a session/window that already exists rather than error or duplicate it. Doing that in the pure, serializable plan model means the "does it exist?" branch has to live at execution time, not build time. what: - LazyPlan.ensure(index, probe): mark a create conditional -- the driver runs the probe (a read returning the object's capture format), and on success binds the slot to the found ids and skips the create - Reuse the create op's own build_result to parse the probe's ids, so a found object binds the same self/window/pane subrefs a created one would - Carry the probe through to_list/from_list so the conditional round-trips - Cover found-reuses / absent-creates and the serialization round-trip --- src/libtmux/experimental/ops/plan.py | 56 +++++++++++++++++-- tests/experimental/ops/test_plan.py | 80 +++++++++++++++++++++++++++- 2 files changed, 130 insertions(+), 6 deletions(-) diff --git a/src/libtmux/experimental/ops/plan.py b/src/libtmux/experimental/ops/plan.py index 83f25ba11..f7bd21b6d 100644 --- a/src/libtmux/experimental/ops/plan.py +++ b/src/libtmux/experimental/ops/plan.py @@ -245,6 +245,7 @@ class LazyPlan: def __init__(self) -> None: self._operations: list[Operation[t.Any]] = [] + self._ensures: dict[int, Operation[t.Any]] = {} def add(self, operation: Operation[t.Any]) -> SlotRef: """Record an operation; return a :class:`SlotRef` to its eventual id. @@ -255,6 +256,20 @@ def add(self, operation: Operation[t.Any]) -> SlotRef: self._operations.append(operation) return SlotRef(len(self._operations) - 1) + def ensure(self, index: int, probe: Operation[t.Any]) -> None: + """Make the create at *index* conditional: probe first, create only if absent. + + At execution the driver runs *probe* (a read that returns the object's + capture format -- e.g. ``display-message`` yielding ``#{session_id} ...``); + if it succeeds the create is *skipped* and the slot binds to the found + ids, so a find-or-create build reuses an existing object. The plan stays a + flat, serializable list of operations -- the branch lives in the driver. + The create at *index* must be a non-chainable create (so it is its own + dispatch step); *probe* must render the same capture format the create + captures, so :meth:`Operation.build_result` parses the found ids. + """ + self._ensures[index] = probe + @property def operations(self) -> tuple[Operation[t.Any], ...]: """The recorded operations, in order.""" @@ -269,14 +284,30 @@ def __iter__(self) -> Iterator[Operation[t.Any]]: return iter(self._operations) def to_list(self) -> list[dict[str, t.Any]]: - """Serialize the whole plan to a list of plain operation dicts.""" - return [operation_to_dict(operation) for operation in self._operations] + """Serialize the whole plan to a list of plain operation dicts. + + A find-or-create op (see :meth:`ensure`) carries its probe under an + ``"ensure"`` key so the conditional survives the round-trip. + """ + out: list[dict[str, t.Any]] = [] + for index, operation in enumerate(self._operations): + item = operation_to_dict(operation) + probe = self._ensures.get(index) + if probe is not None: + item["ensure"] = operation_to_dict(probe) + out.append(item) + return out @classmethod def from_list(cls, data: t.Sequence[t.Mapping[str, t.Any]]) -> LazyPlan: - """Reconstruct a plan from :meth:`to_list` output.""" + """Reconstruct a plan from :meth:`to_list` output (probes included).""" plan = cls() - plan._operations = [operation_from_dict(item) for item in data] + for index, item in enumerate(data): + rest = {key: value for key, value in item.items() if key != "ensure"} + plan._operations.append(operation_from_dict(rest)) + probe = item.get("ensure") + if probe is not None: + plan._ensures[index] = operation_from_dict(probe) return plan def add_chain(self, chain: OpChain) -> None: @@ -388,7 +419,22 @@ def _drive( bindings[create_idx] = new_id elif len(step.indices) == 1: index = step.indices[0] - result = yield _Single(_resolve(self._operations[index], bindings)) + probe = self._ensures.get(index) + if probe is not None: + found = yield _Single(_resolve(probe, bindings)) + if found.ok and found.text.strip(): + # The object exists: bind to its ids, skip the create. + result = self._operations[index].build_result( + returncode=0, + stdout=(found.text,), + version=version, + ) + else: + result = yield _Single( + _resolve(self._operations[index], bindings), + ) + else: + result = yield _Single(_resolve(self._operations[index], bindings)) results[index] = result if result.created_id is not None: bindings[index] = result.created_id diff --git a/tests/experimental/ops/test_plan.py b/tests/experimental/ops/test_plan.py index 43ce28ebc..e311ed2c9 100644 --- a/tests/experimental/ops/test_plan.py +++ b/tests/experimental/ops/test_plan.py @@ -8,22 +8,28 @@ import pytest from libtmux.experimental.engines import AsyncConcreteEngine, ConcreteEngine +from libtmux.experimental.engines.base import CommandResult from libtmux.experimental.ops import ( BreakPane, + DisplayMessage, JoinPane, LazyPlan, MarkedPlanner, MovePane, + NewSession, SendKeys, SequentialPlanner, SplitWindow, StepReport, SwapPane, ) -from libtmux.experimental.ops._types import PaneId, SlotRef, WindowId +from libtmux.experimental.ops._types import NameRef, PaneId, SlotRef, WindowId from libtmux.experimental.ops.exc import ForwardCaptureError, OperationError if t.TYPE_CHECKING: + from collections.abc import Sequence + + from libtmux.experimental.engines.base import CommandRequest from libtmux.experimental.ops.operation import Operation @@ -283,3 +289,75 @@ async def both() -> tuple[bool, bool]: stream_ok, direct_ok = asyncio.run(both()) assert stream_ok == direct_ok + + +class _FindEngine: + """A fake engine where the probe reports found-or-not and the create makes one.""" + + def __init__(self, *, found: bool) -> None: + self.found = found + self.calls: list[tuple[str, ...]] = [] + + def run(self, request: CommandRequest) -> CommandResult: + """Answer a display-message probe, or a new-session create.""" + self.calls.append(request.args) + cmd = ("tmux", *request.args) + if request.args[0] == "display-message": + if self.found: + return CommandResult(cmd=cmd, stdout=("$9 @9 %9",), returncode=0) + return CommandResult(cmd=cmd, stderr=("no session",), returncode=1) + return CommandResult(cmd=cmd, stdout=("$1 @1 %1",), returncode=0) + + def run_batch(self, requests: Sequence[CommandRequest]) -> list[CommandResult]: + """Run each request in order.""" + return [self.run(req) for req in requests] + + +class _EnsureCase(t.NamedTuple): + """Whether the probe finds the object, and the id + create-count expected.""" + + test_id: str + found: bool + session_id: str + creates: int + + +_ENSURE_CASES: tuple[_EnsureCase, ...] = ( + _EnsureCase("found_reuses", found=True, session_id="$9", creates=0), + _EnsureCase("absent_creates", found=False, session_id="$1", creates=1), +) + + +@pytest.mark.parametrize("case", _ENSURE_CASES, ids=[c.test_id for c in _ENSURE_CASES]) +def test_ensure_probes_then_creates_only_if_absent(case: _EnsureCase) -> None: + """An ensured create binds a found object's ids, or creates when absent.""" + plan = LazyPlan() + slot = plan.add(NewSession(session_name="dev", capture_panes=True)) + plan.ensure( + slot.slot, + DisplayMessage(target=NameRef("dev"), message="#{session_id}"), + ) + engine = _FindEngine(found=case.found) + result = plan.execute(engine) + + assert result.ok + assert result.bindings[0] == case.session_id + assert result.bindings[0, "pane"].startswith("%") # first-pane subref bound + creates = [call for call in engine.calls if call[0] == "new-session"] + assert len(creates) == case.creates # created only when the probe found nothing + + +def test_ensure_survives_serialization_round_trip() -> None: + """to_list/from_list carry an ensured op's probe, so the conditional persists.""" + plan = LazyPlan() + slot = plan.add(NewSession(session_name="dev", capture_panes=True)) + plan.ensure( + slot.slot, DisplayMessage(target=NameRef("dev"), message="#{session_id}") + ) + + revived = LazyPlan.from_list(plan.to_list()) + + assert revived.operations == plan.operations + engine = _FindEngine(found=True) + assert revived.execute(engine).bindings[0] == "$9" + assert not [call for call in engine.calls if call[0] == "new-session"] From f2dec63f8d8471fcd357728326b3b0ae5dda5c53 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 4 Jul 2026 13:50:46 -0500 Subject: [PATCH 09/13] Fluent(feat): Add find_or_create_session why: give the fluent builder an idempotent session entry, so re-running a build reuses the live session instead of duplicating it -- the tmuxp load -a shape, in the forward-ref builder. what: - PlanBuilder.find_or_create_session(name): record the same create as new_session, made conditional via LazyPlan.ensure with a display-message probe that captures the session's id, first window, and first pane - Cover the recorded shape and a live idempotent double-build --- src/libtmux/experimental/fluent.py | 30 ++++++++++++++++++++++++++++++ tests/experimental/test_fluent.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+) diff --git a/src/libtmux/experimental/fluent.py b/src/libtmux/experimental/fluent.py index b0e6bf671..265426f4c 100644 --- a/src/libtmux/experimental/fluent.py +++ b/src/libtmux/experimental/fluent.py @@ -58,6 +58,10 @@ _CURSOR_FMT = "#{cursor_x},#{cursor_y}" _WAIT_PANE_POLLS = 40 _WAIT_PANE_INTERVAL = 0.05 +#: The probe format for a session find-or-create -- the ids +#: ``NewSession(capture_panes=True)`` captures, so a found session binds the +#: same self/window/pane slots a created one would. +_SESSION_PROBE = "#{session_id} #{window_id} #{pane_id}" def _pane_ready(cursor: str) -> bool: @@ -158,6 +162,32 @@ def new_session(self, name: str) -> SessionRef: slot = self.plan.add(NewSession(session_name=name, capture_panes=True)) return SessionRef(self.plan, name, slot) + def find_or_create_session(self, name: str) -> SessionRef: + """Reach session *name*, creating it only if it does not exist. + + At build time this records the same create as :meth:`new_session`, but + makes it conditional (see :meth:`~..ops.plan.LazyPlan.ensure`): at + execution the plan probes for *name* and reuses the live session when it + is already there, so a re-run is idempotent instead of a duplicate. + + Examples + -------- + >>> from libtmux.experimental.engines.concrete import ConcreteEngine + >>> p = plan() + >>> _ = p.find_or_create_session("dev").window().pane() + >>> [op.kind for op in p.plan.operations] + ['new_session'] + >>> p.run(ConcreteEngine()).ok + True + """ + create = NewSession(session_name=name, capture_panes=True) + slot = self.plan.add(create) + self.plan.ensure( + slot.slot, + DisplayMessage(target=NameRef(name), message=_SESSION_PROBE), + ) + return SessionRef(self.plan, name, slot) + def sleep(self, seconds: float) -> PlanBuilder: """Pause *seconds* after the last recorded op (a hard fold boundary). diff --git a/tests/experimental/test_fluent.py b/tests/experimental/test_fluent.py index f126d328d..742a4d690 100644 --- a/tests/experimental/test_fluent.py +++ b/tests/experimental/test_fluent.py @@ -135,3 +135,32 @@ def test_sleep_runs_offline() -> None: p.sleep(0.0) pane.split().do(lambda c: c.send_keys("htop")) assert p.run(ConcreteEngine()).ok + + +def test_find_or_create_session_records_a_conditional_create() -> None: + """find_or_create_session records one create, made conditional via ensure.""" + p = plan() + pane = p.find_or_create_session("dev").window().pane() + pane.do(lambda c: c.send_keys("vim")) + assert [op.kind for op in p.plan.operations] == ["new_session", "send_keys"] + assert 0 in p.plan._ensures # the create is conditional + + +def test_find_or_create_session_is_idempotent_live(session: Session) -> None: + """Building the same session name twice reuses it instead of duplicating.""" + from libtmux.experimental.engines.subprocess import SubprocessEngine + + engine = SubprocessEngine.for_server(session.server) + + def build() -> None: + p = plan() + p.find_or_create_session("fluent-idem").window().pane().do( + lambda c: c.send_keys("echo hi", enter=False), + ) + p.run(engine).raise_for_status() + + build() + build() # second run must find the existing session, not create a duplicate + + named = [s for s in session.server.sessions if s.session_name == "fluent-idem"] + assert len(named) == 1 From cf7574d14cd3626f5804ce542235cca7db51febe Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 4 Jul 2026 14:22:32 -0500 Subject: [PATCH 10/13] Test(fix): Clean up sessions in fluent live tests why: the fluent live tests created sessions on the session-scoped shared server and never removed them. On tmux 3.3a/3.5 under xdist, that leaked session perturbed the phantom-reap tests, which assert an exact global session count -- CI failed there though the local gate (newer tmux) did not. what: - Kill the created session in a finally block (test_build_session_live, test_find_or_create_session_is_idempotent_live) via a _kill_named helper --- tests/experimental/test_fluent.py | 50 +++++++++++++++++++++---------- 1 file changed, 35 insertions(+), 15 deletions(-) diff --git a/tests/experimental/test_fluent.py b/tests/experimental/test_fluent.py index 742a4d690..28b930bd4 100644 --- a/tests/experimental/test_fluent.py +++ b/tests/experimental/test_fluent.py @@ -11,6 +11,7 @@ from libtmux.experimental.query import ForwardPaneRef if t.TYPE_CHECKING: + from libtmux.server import Server from libtmux.session import Session @@ -72,21 +73,36 @@ def test_window_pane_is_forward_handle() -> None: assert not hasattr(ref, "pane_id") +def _kill_named(server: Server, name: str) -> None: + """Kill every session named *name* so a live test leaves the shared server clean. + + The ``server`` fixture is session-scoped, so a leaked session would perturb + later tests that measure global session counts (e.g. the phantom-reap tests). + """ + for sess in server.sessions: + if sess.session_name == name: + sess.kill() + + def test_build_session_live(session: Session) -> None: """A fluent build creates a real session with the declared panes.""" from libtmux.experimental.engines.subprocess import SubprocessEngine - engine = SubprocessEngine.for_server(session.server) - p = plan() - pane = p.new_session("fluentdev").window().pane() - pane.do(lambda c: c.send_keys("echo top", enter=False)).split().do( - lambda c: c.send_keys("echo bottom", enter=False), - ) - p.run(engine).raise_for_status() + server = session.server + engine = SubprocessEngine.for_server(server) + try: + p = plan() + pane = p.new_session("fluentdev").window().pane() + pane.do(lambda c: c.send_keys("echo top", enter=False)).split().do( + lambda c: c.send_keys("echo bottom", enter=False), + ) + p.run(engine).raise_for_status() - built = [s for s in session.server.sessions if s.session_name == "fluentdev"] - assert len(built) == 1 - assert len(built[0].windows[0].panes) == 2 + built = [s for s in server.sessions if s.session_name == "fluentdev"] + assert len(built) == 1 + assert len(built[0].windows[0].panes) == 2 + finally: + _kill_named(server, "fluentdev") class _HostCase(t.NamedTuple): @@ -150,7 +166,8 @@ def test_find_or_create_session_is_idempotent_live(session: Session) -> None: """Building the same session name twice reuses it instead of duplicating.""" from libtmux.experimental.engines.subprocess import SubprocessEngine - engine = SubprocessEngine.for_server(session.server) + server = session.server + engine = SubprocessEngine.for_server(server) def build() -> None: p = plan() @@ -159,8 +176,11 @@ def build() -> None: ) p.run(engine).raise_for_status() - build() - build() # second run must find the existing session, not create a duplicate + try: + build() + build() # second run must find the existing session, not create a duplicate - named = [s for s in session.server.sessions if s.session_name == "fluent-idem"] - assert len(named) == 1 + named = [s for s in server.sessions if s.session_name == "fluent-idem"] + assert len(named) == 1 + finally: + _kill_named(server, "fluent-idem") From ce0c9475edafa588f2caff9c68434d6d43510dcb Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 4 Jul 2026 14:50:49 -0500 Subject: [PATCH 11/13] docs(experimental): Document the fluent plan() builder why: the fluent forward-ref build tier had only method/module doctests; the experimental page covered the Core ops/engines/plans but not the declarative surface a user reaches for first. what: - Add a "Building fluently with plan()" section to the experimental page, proportional to the other sections: forward-ref handles, fold-unless- true-blocker, and find_or_create_session, with runnable ConcreteEngine doctests --- docs/experimental.md | 39 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/docs/experimental.md b/docs/experimental.md index 0cc6fbe65..83ebcf439 100644 --- a/docs/experimental.md +++ b/docs/experimental.md @@ -115,6 +115,45 @@ many times tmux is invoked: True ``` +## Building fluently with `plan()` + +{func}`~libtmux.experimental.fluent.plan` is a fluent builder over a plan: you +name a session, walk down to a pane, and record what each pane runs, without +threading the new ids through yourself. Nothing touches tmux until +{meth}`~libtmux.experimental.fluent.PlanBuilder.run`, which folds the whole +description into a few dispatches (its async twin is ``arun``): + +```python +>>> from libtmux.experimental.fluent import plan +>>> from libtmux.experimental.engines import ConcreteEngine +>>> p = plan() +>>> pane = p.new_session("dev").window().pane() +>>> _ = pane.do(lambda c: c.send_keys("vim")).split().do(lambda c: c.send_keys("htop")) +>>> p.run(ConcreteEngine()).ok +True +``` + +``.split()`` makes a new pane that does not exist yet, so it comes back as a +*forward* handle: you keep building on it, but reading its id is a static type +error (the concrete {class}`~libtmux.experimental.query.PaneRef` has +``.pane_id``; the {class}`~libtmux.experimental.query.ForwardPaneRef` does not), +resolved against the captured id only when the plan runs. + +``run()`` folds by default and breaks the fold only at a true blocker -- a +created id a later op needs, or a host pause recorded by ``sleep()``/``wait()``. +{meth}`~libtmux.experimental.fluent.PlanBuilder.find_or_create_session` makes the +create conditional, so re-running a build reuses a live session instead of +duplicating it: + +```python +>>> from libtmux.experimental.fluent import plan +>>> from libtmux.experimental.engines import ConcreteEngine +>>> p = plan() +>>> _ = p.find_or_create_session("dev").window().pane() +>>> p.run(ConcreteEngine()).ok +True +``` + ## Operation catalog The catalog below is generated from the operation registry, so it always matches From ef4c79a18ddebd76e463f2a5fd409e09b9d18cd2 Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 4 Jul 2026 15:12:44 -0500 Subject: [PATCH 12/13] Mcp(fix): Preserve ensure across plan-tool serialization why: the MCP plan tools rebuilt a serialized plan with add() + operation_from_dict, which drops the `ensure` probe that to_list emits -- so a find-or-create plan round-tripped through MCP silently became an unconditional create (a duplicate-session footgun). what: - Route _plan_from_dicts through LazyPlan.from_list, which carries the ensure probe, so a conditional create survives the MCP round-trip --- src/libtmux/experimental/mcp/fastmcp_adapter.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/src/libtmux/experimental/mcp/fastmcp_adapter.py b/src/libtmux/experimental/mcp/fastmcp_adapter.py index 0c2155604..e35d77c9c 100644 --- a/src/libtmux/experimental/mcp/fastmcp_adapter.py +++ b/src/libtmux/experimental/mcp/fastmcp_adapter.py @@ -425,7 +425,6 @@ def register_plan_tools( Planner, SequentialPlanner, ) - from libtmux.experimental.ops.serialize import operation_from_dict reg = registry if registry is not None else OperationToolRegistry() planners: dict[str, type[Planner]] = { @@ -435,10 +434,9 @@ def register_plan_tools( } def _plan_from_dicts(operations: list[dict[str, t.Any]]) -> LazyPlan: - plan = LazyPlan() - for data in operations: - plan.add(operation_from_dict(data)) - return plan + # from_list (not add) so a serialized find-or-create `ensure` probe + # survives the round-trip instead of being silently dropped. + return LazyPlan.from_list(operations) def _planner(name: str) -> Planner: chosen = planners.get(name) From d361a2b15bb9f27ab1d04b25f7a52aeea4843c8b Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 4 Jul 2026 15:12:44 -0500 Subject: [PATCH 13/13] Ops(test): Make the ensure probe test format-honest why: the fake engine returned all three ids for any display-message, so the ensure test asserted a pane binding while probing only #{session_id} -- it passed on ids the probe never requested and would not catch a real probe/capture format mismatch. what: - Make _FindEngine format-aware (return only the ids the probe requests) - Probe the full capture format in test_ensure_probes_then_creates - Add test_ensure_probe_must_match_create_capture: a session-only probe binds no pane subref, guarding the format contract --- tests/experimental/ops/test_plan.py | 43 +++++++++++++++++++++++++---- 1 file changed, 38 insertions(+), 5 deletions(-) diff --git a/tests/experimental/ops/test_plan.py b/tests/experimental/ops/test_plan.py index e311ed2c9..61055c378 100644 --- a/tests/experimental/ops/test_plan.py +++ b/tests/experimental/ops/test_plan.py @@ -299,13 +299,22 @@ def __init__(self, *, found: bool) -> None: self.calls: list[tuple[str, ...]] = [] def run(self, request: CommandRequest) -> CommandResult: - """Answer a display-message probe, or a new-session create.""" + """Answer a display-message probe, or a new-session create. + + The probe is *format-aware*: it returns only the ids the probe's format + actually requests, so a probe that omits ``#{pane_id}`` yields no pane id + -- mirroring real tmux, so a test cannot pass on ids the probe never asked + for. + """ self.calls.append(request.args) cmd = ("tmux", *request.args) if request.args[0] == "display-message": - if self.found: - return CommandResult(cmd=cmd, stdout=("$9 @9 %9",), returncode=0) - return CommandResult(cmd=cmd, stderr=("no session",), returncode=1) + if not self.found: + return CommandResult(cmd=cmd, stderr=("no session",), returncode=1) + fmt = request.args[-1] # the -p value + ids = {"session_id": "$9", "window_id": "@9", "pane_id": "%9"} + text = " ".join(v for key, v in ids.items() if f"#{{{key}}}" in fmt) + return CommandResult(cmd=cmd, stdout=(text,), returncode=0) return CommandResult(cmd=cmd, stdout=("$1 @1 %1",), returncode=0) def run_batch(self, requests: Sequence[CommandRequest]) -> list[CommandResult]: @@ -333,9 +342,14 @@ def test_ensure_probes_then_creates_only_if_absent(case: _EnsureCase) -> None: """An ensured create binds a found object's ids, or creates when absent.""" plan = LazyPlan() slot = plan.add(NewSession(session_name="dev", capture_panes=True)) + # The probe renders the SAME capture format the create captures, so a found + # session binds the same self/window/pane subrefs a created one would. plan.ensure( slot.slot, - DisplayMessage(target=NameRef("dev"), message="#{session_id}"), + DisplayMessage( + target=NameRef("dev"), + message="#{session_id} #{window_id} #{pane_id}", + ), ) engine = _FindEngine(found=case.found) result = plan.execute(engine) @@ -347,6 +361,25 @@ def test_ensure_probes_then_creates_only_if_absent(case: _EnsureCase) -> None: assert len(creates) == case.creates # created only when the probe found nothing +def test_ensure_probe_must_match_create_capture() -> None: + """A probe that omits the pane id binds no pane subref (the format contract). + + This guards the ensure() footgun: the probe must render the create's capture + format. A session-only probe finds the session but yields no pane id, so a + downstream ``.pane`` forward-ref would fail closed rather than mis-bind. + """ + plan = LazyPlan() + slot = plan.add(NewSession(session_name="dev", capture_panes=True)) + plan.ensure( + slot.slot, DisplayMessage(target=NameRef("dev"), message="#{session_id}") + ) + + result = plan.execute(_FindEngine(found=True)) + + assert result.bindings[0] == "$9" # the session bound + assert (0, "pane") not in result.bindings # but no pane id -- probe omitted it + + def test_ensure_survives_serialization_round_trip() -> None: """to_list/from_list carry an ensured op's probe, so the conditional persists.""" plan = LazyPlan()