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 diff --git a/src/libtmux/experimental/fluent.py b/src/libtmux/experimental/fluent.py new file mode 100644 index 000000000..265426f4c --- /dev/null +++ b/src/libtmux/experimental/fluent.py @@ -0,0 +1,332 @@ +"""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 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.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, StepReport + from libtmux.experimental.ops._types import SlotRef, Target + +_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: + """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) +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) + _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. + + 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 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). + + 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, + *, + 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 + """ + + 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=self._planner(planner), + on_step=on_step, + ) + + async def arun( + self, + engine: AsyncTmuxEngine, + *, + version: str | None = None, + planner: Planner | None = None, + ) -> PlanResult: + """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=self._planner(planner), + on_step=on_step, + ) + + 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/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..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) @@ -459,6 +457,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 +480,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/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..f7bd21b6d 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("%"): @@ -211,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. @@ -221,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.""" @@ -235,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: @@ -275,6 +340,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, @@ -315,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 @@ -413,3 +532,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/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/src/libtmux/experimental/workspace/__init__.py b/src/libtmux/experimental/workspace/__init__.py index 99b572142..1e7126c39 100644 --- a/src/libtmux/experimental/workspace/__init__.py +++ b/src/libtmux/experimental/workspace/__init__.py @@ -38,6 +38,12 @@ WindowCreated, WorkspaceBuilt, ) +from libtmux.experimental.workspace.expand import expand +from libtmux.experimental.workspace.freeze import ( + afreeze_server, + freeze, + freeze_server, +) from libtmux.experimental.workspace.ir import ( Command, Float, @@ -47,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", @@ -64,10 +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/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/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, + ) 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() diff --git a/tests/experimental/ops/test_plan.py b/tests/experimental/ops/test_plan.py index cb5a8f24b..61055c378 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 @@ -211,3 +217,180 @@ 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 + + +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. + + 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 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]: + """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)) + # 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} #{window_id} #{pane_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_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() + 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"] diff --git a/tests/experimental/test_fluent.py b/tests/experimental/test_fluent.py new file mode 100644 index 000000000..28b930bd4 --- /dev/null +++ b/tests/experimental/test_fluent.py @@ -0,0 +1,186 @@ +"""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.server import Server + 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 _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 + + 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 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): + """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 + + +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 + + server = session.server + engine = SubprocessEngine.for_server(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() + + try: + build() + build() # second run must find the existing session, not create a duplicate + + named = [s for s in server.sessions if s.session_name == "fluent-idem"] + assert len(named) == 1 + finally: + _kill_named(server, "fluent-idem") 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" 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()