From 1755b5c77bde232b3ca43ebdd6a507d27d16810a Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 20 Jun 2026 11:19:47 -0500 Subject: [PATCH 1/3] py(deps[libtmux]) Pin to sibling chainable-commands worktree why: Build the experimental ChainWorkspaceBuilder against the in-progress libtmux._experimental.chain API on the sibling libtmux worktree. what: - Add [tool.uv.sources] libtmux = { path = "../libtmux", editable = true } - Relock against the local editable checkout --- pyproject.toml | 5 +++++ uv.lock | 57 ++++++++++++++++++++++++++++++++++++++++++++------ 2 files changed, 56 insertions(+), 6 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 646caf9397..3bb4a278aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -113,6 +113,11 @@ lint = [ requires = ["hatchling"] build-backend = "hatchling.build" +[tool.uv.sources] +# Experiment: pin libtmux to the sibling chainable-commands worktree so the +# experimental libtmux._experimental.chain API is importable. Do not merge. +libtmux = { path = "../libtmux", editable = true } + [tool.uv.exclude-newer-package] # git-pull packages release in lockstep with their workspaces, so a # fresh release blocking on the 3-day cooldown blocks every diff --git a/uv.lock b/uv.lock index f594b2c089..b4047550a2 100644 --- a/uv.lock +++ b/uv.lock @@ -417,7 +417,7 @@ name = "exceptiongroup" version = "1.3.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "typing-extensions", marker = "python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" } wheels = [ @@ -620,10 +620,55 @@ wheels = [ [[package]] name = "libtmux" version = "0.58.1" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/c7/58/346776e0491ede33e1554a4bff9b545dbe9f3164e45abac483195938a1cf/libtmux-0.58.1.tar.gz", hash = "sha256:a294dd585aa419d4ecce36f3e55df656693743c97a0b5b5bb1e5fea31ada2482", size = 519541, upload-time = "2026-06-17T00:03:31.81Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/22/4d/e44ada32edfe947c40d4dfc596a6f5355400a16d08be06016bd754375e41/libtmux-0.58.1-py3-none-any.whl", hash = "sha256:ab0f47d03a59d674962bc23e36e188fcfa4a82b0f270d474afab519e3076839b", size = 113653, upload-time = "2026-06-17T00:03:30.48Z" }, +source = { editable = "../libtmux" } + +[package.metadata] + +[package.metadata.requires-dev] +coverage = [ + { name = "codecov" }, + { name = "coverage" }, + { name = "pytest-cov" }, +] +dev = [ + { name = "codecov" }, + { name = "coverage" }, + { name = "gp-libs" }, + { name = "gp-sphinx", specifier = "==0.0.1a31" }, + { name = "mypy" }, + { name = "pytest" }, + { name = "pytest-asyncio" }, + { name = "pytest-cov" }, + { name = "pytest-mock" }, + { name = "pytest-rerunfailures" }, + { name = "pytest-watcher" }, + { name = "pytest-xdist" }, + { name = "ruff" }, + { name = "sphinx-autobuild" }, + { name = "sphinx-autodoc-api-style", specifier = "==0.0.1a31" }, + { name = "sphinx-autodoc-pytest-fixtures", specifier = "==0.0.1a31" }, + { name = "types-docutils" }, + { name = "typing-extensions", marker = "python_full_version < '3.11'" }, +] +docs = [ + { name = "gp-sphinx", specifier = "==0.0.1a31" }, + { name = "sphinx-autobuild" }, + { name = "sphinx-autodoc-api-style", specifier = "==0.0.1a31" }, + { name = "sphinx-autodoc-pytest-fixtures", specifier = "==0.0.1a31" }, +] +lint = [ + { name = "mypy" }, + { name = "ruff" }, + { name = "typing-extensions", marker = "python_full_version < '3.11'" }, +] +testing = [ + { name = "gp-libs" }, + { name = "pytest" }, + { name = "pytest-asyncio" }, + { name = "pytest-mock" }, + { name = "pytest-rerunfailures" }, + { name = "pytest-watcher" }, + { name = "typing-extensions", marker = "python_full_version < '3.11'" }, ] [[package]] @@ -1685,7 +1730,7 @@ testing = [ [package.metadata] requires-dist = [ - { name = "libtmux", specifier = "~=0.58.1" }, + { name = "libtmux", editable = "../libtmux" }, { name = "pyyaml", specifier = ">=6.0" }, ] From d36730a31e33a88ff1f26b5a02db58a31d77f16b Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 20 Jun 2026 11:42:13 -0500 Subject: [PATCH 2/3] workspace(feat[chain]): add ChainWorkspaceBuilder why: Showcase libtmux's experimental chain API downstream: build a whole workspace's window/pane tree in the minimum tmux invocations instead of one subprocess per new-window / split-window. what: - Add ChainWorkspaceBuilder (subclass of WorkspaceBuilder): one ForwardPlan seeded from the session builds every window and pane, reusing the session's default window as window 1 (no orphan) via initial_window / initial_pane - send_keys_mode: readiness (default, classic per-pane shell-ready delivery) or batched (folded into the plan) - Reuse parent helpers (_wait_for_pane_ready, config_after_window, focus, plugins, before_script, options) - Tests: two/three-pane structure, focus, readiness commands land, batched --- src/tmuxp/workspace/chain_builder.py | 293 ++++++++++++++++++++++++++ tests/workspace/test_chain_builder.py | 110 ++++++++++ 2 files changed, 403 insertions(+) create mode 100644 src/tmuxp/workspace/chain_builder.py create mode 100644 tests/workspace/test_chain_builder.py diff --git a/src/tmuxp/workspace/chain_builder.py b/src/tmuxp/workspace/chain_builder.py new file mode 100644 index 0000000000..d3ae997425 --- /dev/null +++ b/src/tmuxp/workspace/chain_builder.py @@ -0,0 +1,293 @@ +"""Build a tmux workspace through libtmux's experimental chain API. + +``ChainWorkspaceBuilder`` is an alternative to +:class:`~tmuxp.workspace.builder.WorkspaceBuilder` that constructs the whole +window/pane tree with one +:class:`~libtmux._experimental.chain.ForwardPlan`, resolved over the minimum +number of tmux invocations instead of one subprocess per ``new-window`` / +``split-window``. Shell commands are delivered either readiness-aware (the +default, matching the classic builder) or batched into the plan. + +This is experimental and depends on the pinned ``libtmux._experimental.chain`` +worktree; do not ship to a release. +""" + +from __future__ import annotations + +import typing as t + +from libtmux._experimental.chain import ForwardPlan, ServerPlanRunner +from libtmux.pane import Pane +from libtmux.session import Session +from libtmux.window import Window + +from tmuxp import exc +from tmuxp.util import run_before_script +from tmuxp.workspace.builder import ( + WorkspaceBuilder, + _wait_for_pane_ready, + get_default_columns, + get_default_rows, +) + +if t.TYPE_CHECKING: + import collections.abc as cabc + + +Decorate: t.TypeAlias = "cabc.Callable[[t.Any], t.Any]" +SendKeysMode: t.TypeAlias = 't.Literal["readiness", "batched"]' + + +def _str_or_none(value: t.Any) -> str | None: + """Narrow an Any config value to a string or None.""" + return value if isinstance(value, str) else None + + +def _window_start_directory(window_config: dict[str, t.Any]) -> str | None: + """Resolve a window's start_directory (window, or its first pane).""" + panes = window_config.get("panes") or [{}] + return _str_or_none( + panes[0].get("start_directory", window_config.get("start_directory")), + ) + + +def _window_shell(window_config: dict[str, t.Any]) -> str | None: + """Resolve a window's launch command (window_shell, or its first pane shell).""" + panes = window_config.get("panes") or [{}] + return _str_or_none(panes[0].get("shell") or window_config.get("window_shell")) + + +def _pane_start_directory( + pane_config: dict[str, t.Any], window_config: dict[str, t.Any] +) -> str | None: + """Resolve a pane's start_directory (pane, then window).""" + return _str_or_none( + pane_config.get("start_directory", window_config.get("start_directory")), + ) + + +def _pane_shell( + pane_config: dict[str, t.Any], window_config: dict[str, t.Any] +) -> str | None: + """Resolve a pane's launch command (pane shell, then window_shell).""" + return _str_or_none(pane_config.get("shell", window_config.get("window_shell"))) + + +def _environment( + config: dict[str, t.Any], window_config: dict[str, t.Any] +) -> dict[str, str] | None: + """Resolve environment for a pane/window (own, then window default).""" + value = config.get("environment", window_config.get("environment")) + return value if isinstance(value, dict) else None + + +def _rename(name: str) -> Decorate: + """Return a decorate that renames a window.""" + return lambda handle: handle.window.rename(name) + + +def _set_window_option(key: str, value: t.Any) -> Decorate: + """Return a decorate that sets one window option.""" + return lambda handle: handle.window.set_option(key, str(value)) + + +def _select_layout(layout: str) -> Decorate: + """Return a decorate that applies a window layout.""" + return lambda handle: handle.window.select_layout(layout) + + +def _send_keys(command: str, *, enter: bool) -> Decorate: + """Return a decorate that sends keys into a pane (batched mode).""" + return lambda handle: handle.cmd.send_keys(command, enter=enter) + + +class ChainWorkspaceBuilder(WorkspaceBuilder): + """A :class:`WorkspaceBuilder` that builds the tree via the chain API. + + The session, then every window and pane, is described in one + :class:`~libtmux._experimental.chain.ForwardPlan` seeded from the live + session (reusing the session's default window as window 1, so nothing is + orphaned) and resolved in the fewest tmux invocations. Shell commands are + delivered per ``send_keys_mode``. + """ + + def __init__( + self, + *args: t.Any, + send_keys_mode: SendKeysMode = "readiness", + **kwargs: t.Any, + ) -> None: + super().__init__(*args, **kwargs) + self.send_keys_mode: SendKeysMode = send_keys_mode + + def build(self, session: Session | None = None, append: bool = False) -> None: + """Build the workspace, constructing the window/pane tree via the chain.""" + session = self._chain_create_session(session) + self._chain_setup_session(session) + + windows = self.session_config["windows"] + plan = ForwardPlan.from_session(session) + seed = plan.seed + for index, window_config in enumerate(windows): + first_window = index == 0 and not append + self._plan_window(plan, seed, window_config, first_window=first_window) + + plan.run_resolving(ServerPlanRunner(self.server)) + self._finalize(session, windows, append=append) + + # -- session creation + setup ------------------------------------------ + def _chain_create_session(self, session: Session | None) -> Session: + if session is not None: + self._session = session + return session + if not self.server: + msg = "ChainWorkspaceBuilder.build requires a server or a session" + raise exc.TmuxpException(msg) + + kwargs: dict[str, t.Any] = { + "x": get_default_columns(), + "y": get_default_rows(), + } + if "start_directory" in self.session_config: + kwargs["start_directory"] = self.session_config["start_directory"] + session = self.server.new_session( + session_name=self.session_config["session_name"], **kwargs + ) + assert session is not None + self._session = session + self.server = session.server + return session + + def _chain_setup_session(self, session: Session) -> None: + for plugin in self.plugins: + plugin.before_workspace_builder(session) + + if "before_script" in self.session_config: + cwd = self.session_config.get("start_directory") + try: + run_before_script( + self.session_config["before_script"], + cwd=cwd, + on_line=self.on_script_output, + ) + except Exception: + session.kill() + raise + + for option, value in self.session_config.get("options", {}).items(): + session.set_option(option, value) + for option, value in self.session_config.get("global_options", {}).items(): + session.set_option(option, value, global_=True) + for key, value in self.session_config.get("environment", {}).items(): + session.set_environment(key, value) + + # -- plan the window/pane tree ----------------------------------------- + def _plan_window( + self, + plan: ForwardPlan, + seed: t.Any, + window_config: dict[str, t.Any], + *, + first_window: bool, + ) -> None: + name = window_config.get("window_name") + if first_window: + window = seed.initial_window + first_pane = seed.initial_pane + if name: + window.do(_rename(name)) + else: + window = seed.new_window( + name=name, + start_directory=_window_start_directory(window_config), + environment=_environment(window_config, window_config), + window_shell=_window_shell(window_config), + ) + first_pane = window + + for key, value in window_config.get("options", {}).items(): + window.do(_set_window_option(key, value)) + + panes = window_config["panes"] + pane_handles = [first_pane] + previous = first_pane + for pane_config in panes[1:]: + handle = previous.split( + start_directory=_pane_start_directory(pane_config, window_config), + shell=_pane_shell(pane_config, window_config), + environment=_environment(pane_config, window_config), + ) + pane_handles.append(handle) + previous = handle + + if "layout" in window_config: + window.do(_select_layout(window_config["layout"])) + + if self.send_keys_mode == "batched": + for handle, pane_config in zip(pane_handles, panes, strict=False): + self._plan_pane_keys(handle, pane_config) + + def _plan_pane_keys(self, handle: t.Any, pane_config: dict[str, t.Any]) -> None: + enter = pane_config.get("enter", True) + for command in pane_config.get("shell_command", []): + handle.do(_send_keys(command["cmd"], enter=command.get("enter", enter))) + + # -- recover objects + deliver commands + focus ------------------------ + def _finalize( + self, session: Session, windows: list[dict[str, t.Any]], *, append: bool + ) -> None: + session.refresh() + # Windows in index order == config creation order (default window first, + # new windows appended); same for a window's panes. + built = ( + list(session.windows)[-len(windows) :] if append else list(session.windows) + ) + + focus_window: Window | None = None + for window, window_config in zip(built, windows, strict=False): + window.refresh() + for plugin in self.plugins: + plugin.on_window_create(window) + + focus_pane: Pane | None = None + for pane, pane_config in zip( + window.panes, window_config["panes"], strict=False + ): + if self.send_keys_mode == "readiness": + self._send_pane_commands(pane, pane_config, window_config) + if pane_config.get("focus"): + focus_pane = pane + + self.config_after_window(window, window_config) + for plugin in self.plugins: + plugin.after_window_finished(window) + if focus_pane is not None: + focus_pane.select() + if window_config.get("focus"): + focus_window = window + + if focus_window is not None: + focus_window.select() + + def _send_pane_commands( + self, + pane: Pane, + pane_config: dict[str, t.Any], + window_config: dict[str, t.Any], + ) -> None: + """Deliver a pane's shell commands readiness-aware (classic semantics).""" + if _pane_shell(pane_config, window_config) is None: + _wait_for_pane_ready(pane) + + if "suppress_history" in pane_config: + suppress = pane_config["suppress_history"] + else: + suppress = window_config.get("suppress_history", True) + + enter = pane_config.get("enter", True) + for command in pane_config.get("shell_command", []): + pane.send_keys( + command["cmd"], + suppress_history=command.get("suppress_history", suppress), + enter=command.get("enter", enter), + ) diff --git a/tests/workspace/test_chain_builder.py b/tests/workspace/test_chain_builder.py new file mode 100644 index 0000000000..9116bed307 --- /dev/null +++ b/tests/workspace/test_chain_builder.py @@ -0,0 +1,110 @@ +"""Tests for the chain-based workspace builder.""" + +from __future__ import annotations + +import typing as t + +from libtmux.test.retry import retry_until +from libtmux.window import Window + +from tests.fixtures import utils as test_utils +from tmuxp._internal.config_reader import ConfigReader +from tmuxp.workspace import loader +from tmuxp.workspace.chain_builder import ChainWorkspaceBuilder + +if t.TYPE_CHECKING: + from libtmux.session import Session + + +def _load(name: str) -> dict[str, t.Any]: + workspace = ConfigReader._from_file( + test_utils.get_workspace_file(f"workspace/builder/{name}"), + ) + workspace = loader.expand(workspace) + return loader.trickle(workspace) + + +def test_chain_split_windows(session: Session) -> None: + """The chain builder creates the windows and panes of a two-pane workspace.""" + workspace = _load("two_pane.yaml") + builder = ChainWorkspaceBuilder(session_config=workspace, server=session.server) + + builder.build(session=session) + + assert builder.session is session + assert [w.name for w in session.windows] == ["editor", "logging", "test"] + + editor = session.windows.get(window_name="editor") + assert isinstance(editor, Window) + editor.refresh() + assert len(editor.panes) == 2 + + +def test_chain_three_pane(session: Session) -> None: + """The chain builder splits a window into three panes.""" + workspace = _load("three_pane.yaml") + builder = ChainWorkspaceBuilder(session_config=workspace, server=session.server) + + builder.build(session=session) + + test_window = session.windows.get(window_name="test") + assert isinstance(test_window, Window) + test_window.refresh() + assert len(test_window.panes) == 3 + + +def test_chain_focus(session: Session) -> None: + """The chain builder honours window and pane focus.""" + workspace = _load("focus_and_pane.yaml") + builder = ChainWorkspaceBuilder(session_config=workspace, server=session.server) + + builder.build(session=session) + + assert session.active_window.name == "focused window" + + +def test_chain_commands_landed(session: Session) -> None: + """Readiness mode delivers each pane's shell command.""" + workspace = loader.trickle( + loader.expand( + { + "session_name": "cc_cmd", + "windows": [ + { + "window_name": "marker", + "panes": [{"shell_command": "echo CC_LANDED"}], + }, + ], + }, + ), + ) + builder = ChainWorkspaceBuilder(session_config=workspace, server=session.server) + builder.build(session=session) + + window = session.windows.get(window_name="marker") + assert isinstance(window, Window) + pane = window.active_pane + assert pane is not None + + def _landed() -> bool: + return any("CC_LANDED" in line for line in pane.capture_pane()) + + assert retry_until(_landed, 2) + + +def test_chain_batched_mode(session: Session) -> None: + """Batched mode folds send_keys into the plan and still builds the tree.""" + workspace = _load("two_pane.yaml") + builder = ChainWorkspaceBuilder( + session_config=workspace, + server=session.server, + send_keys_mode="batched", + ) + + builder.build(session=session) + + assert [w.name for w in session.windows] == ["editor", "logging", "test"] + editor = session.windows.get(window_name="editor") + assert isinstance(editor, Window) + editor.refresh() + assert len(editor.panes) == 2 From 0f297ebf653079328daffdc3c9e2569a97c00dfa Mon Sep 17 00:00:00 2001 From: Tony Narlock Date: Sat, 20 Jun 2026 11:46:00 -0500 Subject: [PATCH 3/3] cli(load[builder]): --builder=chain flag why: Make the experimental ChainWorkspaceBuilder selectable from the CLI without replacing the default builder. what: - Add --builder={default,chain} to `tmuxp load` (Env: TMUXP_BUILDER), threaded through command_load -> load_workspace as builder_name - Select ChainWorkspaceBuilder when chosen; default builder otherwise - Test the flag parsing (smoke-verified `tmuxp load --builder=chain` builds the expected window/pane tree) --- src/tmuxp/cli/load.py | 18 +++++++++++++++++- tests/workspace/test_chain_builder.py | 11 +++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/src/tmuxp/cli/load.py b/src/tmuxp/cli/load.py index 375cdb1b22..56631cba83 100644 --- a/src/tmuxp/cli/load.py +++ b/src/tmuxp/cli/load.py @@ -19,6 +19,7 @@ from tmuxp._internal.private_path import PrivatePath from tmuxp.workspace import loader from tmuxp.workspace.builder import WorkspaceBuilder +from tmuxp.workspace.chain_builder import ChainWorkspaceBuilder from tmuxp.workspace.finders import find_workspace_file, get_workspace_dir from ._colors import ColorMode, Colors, build_description, get_color_mode @@ -112,6 +113,7 @@ class CLILoadNamespace(argparse.Namespace): progress_format: str | None panel_lines: int | None no_progress: bool + builder: str def load_plugins( @@ -450,6 +452,7 @@ def load_workspace( progress_format: str | None = None, panel_lines: int | None = None, no_progress: bool = False, + builder_name: str = "default", ) -> Session | None: """Entrypoint for ``tmuxp load``, load a tmuxp "workspace" session via config file. @@ -578,8 +581,9 @@ def load_workspace( shutil.which("tmux") # raise exception if tmux not found # WorkspaceBuilder creation — outside spinner so plugin prompts are safe + builder_cls = ChainWorkspaceBuilder if builder_name == "chain" else WorkspaceBuilder try: - builder = WorkspaceBuilder( + builder = builder_cls( session_config=expanded_workspace, plugins=load_plugins(expanded_workspace, colors=cli_colors), server=t, @@ -820,6 +824,17 @@ def create_load_subparser(parser: argparse.ArgumentParser) -> argparse.ArgumentP help=("Disable the animated progress spinner. Env: TMUXP_PROGRESS=0"), ) + parser.add_argument( + "--builder", + dest="builder", + choices=["default", "chain"], + default=os.environ.get("TMUXP_BUILDER", "default"), + help=( + "Workspace builder. 'chain' (experimental) builds the window/pane " + "tree via libtmux's chain API. Env: TMUXP_BUILDER" + ), + ) + try: import shtab @@ -904,4 +919,5 @@ def command_load( progress_format=args.progress_format, panel_lines=args.panel_lines, no_progress=args.no_progress, + builder_name=args.builder, ) diff --git a/tests/workspace/test_chain_builder.py b/tests/workspace/test_chain_builder.py index 9116bed307..5f26117387 100644 --- a/tests/workspace/test_chain_builder.py +++ b/tests/workspace/test_chain_builder.py @@ -108,3 +108,14 @@ def test_chain_batched_mode(session: Session) -> None: assert isinstance(editor, Window) editor.refresh() assert len(editor.panes) == 2 + + +def test_cli_builder_flag() -> None: + """The load CLI exposes --builder with default and chain choices.""" + import argparse + + from tmuxp.cli.load import create_load_subparser + + parser = create_load_subparser(argparse.ArgumentParser()) + assert parser.parse_args(["ws.yaml", "--builder", "chain"]).builder == "chain" + assert parser.parse_args(["ws.yaml"]).builder == "default"