diff --git a/agent_cli/dev/cli.py b/agent_cli/dev/cli.py index 79cc91061..bf20a36cc 100644 --- a/agent_cli/dev/cli.py +++ b/agent_cli/dev/cli.py @@ -618,11 +618,13 @@ def new( summary_lines.append( f"[bold]Agent Handle:[/bold] {agent_handle.handle} ({agent_handle.terminal_name})", ) - if agent_handle.session_name: + if agent_handle.session_name and agent_handle.terminal_name == "tmux": summary_lines.append(f"[bold]tmux Session:[/bold] {agent_handle.session_name}") summary_lines.append( f"[bold]Attach:[/bold] tmux attach -t {shlex.quote(agent_handle.session_name)}", ) + elif agent_handle.session_name and agent_handle.terminal_name == "cmux": + summary_lines.append(f"[bold]cmux Workspace:[/bold] {agent_handle.session_name}") console.print() console.print( @@ -1146,7 +1148,7 @@ def start_agent( f"{handle.terminal_name} handle: {handle.handle}" + ( f" (attach with: tmux attach -t {shlex.quote(handle.session_name)})" - if handle.session_name + if handle.session_name and handle.terminal_name == "tmux" else "" ), ) @@ -1280,7 +1282,7 @@ def list_terminals_cmd( ) -> None: """List available terminal multiplexers and their status. - Shows supported terminals: tmux, zellij, kitty, iTerm2, Terminal.app, + Shows supported terminals: tmux, zellij, cmux, kitty, iTerm2, Terminal.app, Warp, GNOME Terminal. These are used to open new tabs when launching AI agents with `dev new --start-agent`. diff --git a/agent_cli/dev/launch.py b/agent_cli/dev/launch.py index 08fe72542..11ebde79a 100644 --- a/agent_cli/dev/launch.py +++ b/agent_cli/dev/launch.py @@ -321,6 +321,36 @@ def _launch_in_tmux( return handle +def _launch_in_cmux( + path: Path, + agent: CodingAgent, + terminal: terminals.Terminal, + full_cmd: str, + tab_name: str, + repo_root: Path | None, +) -> TerminalHandle | None: + """Launch an agent in a cmux tab inside a workspace named after the repo.""" + from .terminals.cmux import Cmux # noqa: PLC0415 + + if not isinstance(terminal, Cmux): + warn("Could not open new tab in cmux") + return None + + workspace_name = (repo_root or path).name + handle = terminal.open_in_workspace( + path, + full_cmd, + tab_name=tab_name, + workspace_name=workspace_name, + ) + if handle is None: + warn("Could not open new tab in cmux") + return None + + success(f"Started {agent.name} in cmux workspace {workspace_name!r}") + return handle + + def _launch_in_terminal( path: Path, agent: CodingAgent, @@ -332,6 +362,10 @@ def _launch_in_terminal( tmux_session: str | None, ) -> tuple[bool, TerminalHandle | None]: """Launch an agent in the resolved terminal.""" + if terminal.name == "cmux": + handle = _launch_in_cmux(path, agent, terminal, full_cmd, tab_name, repo_root) + return handle is not None, handle + if terminal.name == "tmux": handle = _launch_in_tmux( path, diff --git a/agent_cli/dev/terminals/cmux.py b/agent_cli/dev/terminals/cmux.py new file mode 100644 index 000000000..75d634bfc --- /dev/null +++ b/agent_cli/dev/terminals/cmux.py @@ -0,0 +1,247 @@ +"""cmux terminal adapter. + +cmux (https://cmux.dev) is a Ghostty-based macOS terminal organized as +windows > workspaces > panes > surfaces (tabs), controlled via a Unix +socket through the bundled ``cmux`` CLI. + +agent-cli maps repos to workspaces: each repo gets a workspace named after +it, and each worktree launch opens a new tab inside that workspace. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import shlex +import shutil +import subprocess +from typing import TYPE_CHECKING + +from agent_cli.dev.worktree import get_main_repo_root + +from .base import Terminal, TerminalHandle + +if TYPE_CHECKING: + from pathlib import Path + +# Named colors accepted by `cmux workspace-action --action set-color`. +_WORKSPACE_COLORS = ( + "Red", + "Crimson", + "Orange", + "Amber", + "Olive", + "Green", + "Teal", + "Aqua", + "Blue", + "Navy", + "Indigo", + "Purple", + "Magenta", + "Rose", + "Brown", + "Charcoal", +) + + +class Cmux(Terminal): + """cmux - Ghostty-based terminal with workspaces for AI coding agents.""" + + name = "cmux" + + def detect(self) -> bool: + """Detect if running inside cmux via its auto-set environment variables.""" + return bool( + os.environ.get("CMUX_WORKSPACE_ID") or os.environ.get("CMUX_SURFACE_ID"), + ) + + def is_available(self) -> bool: + """Check if the cmux CLI is available.""" + return shutil.which("cmux") is not None + + def open_new_tab( + self, + path: Path, + command: str | None = None, + tab_name: str | None = None, + ) -> bool: + """Open a new tab in a workspace named after the repo containing ``path``.""" + repo_root = get_main_repo_root(path) + workspace_name = (repo_root or path).name + handle = self.open_in_workspace(path, command, tab_name, workspace_name=workspace_name) + return handle is not None + + def open_in_workspace( + self, + path: Path, + command: str | None = None, + tab_name: str | None = None, + *, + workspace_name: str, + ) -> TerminalHandle | None: + """Open a tab in the named cmux workspace, creating the workspace if needed. + + Workspace targets are always passed explicitly because the cmux CLI + otherwise defaults to the caller's workspace (``CMUX_WORKSPACE_ID``). + """ + if not self.is_available(): + return None + workspaces = self._list_workspaces() + if workspaces is None: + # The cmux CLI itself failed (e.g. app not running); creating a + # workspace would fail too, or duplicate one we could not see. + return None + workspace_ref = next( + (w.get("ref") for w in workspaces if w.get("title") == workspace_name), + None, + ) + if workspace_ref is None: + return self._create_workspace(path, command, tab_name, workspace_name=workspace_name) + return self._open_tab( + path, + command, + tab_name, + workspace_ref=workspace_ref, + workspace_name=workspace_name, + ) + + def _list_workspaces(self) -> list[dict] | None: + """List workspace dicts via the cmux CLI, or None when the CLI call fails.""" + stdout = self._run(["workspace", "list", "--json"]) + if stdout is None: + return None + try: + data = json.loads(stdout) + except json.JSONDecodeError: + return None + workspaces = data.get("workspaces", []) + return workspaces if isinstance(workspaces, list) else None + + def _create_workspace( + self, + path: Path, + command: str | None, + tab_name: str | None, + *, + workspace_name: str, + ) -> TerminalHandle | None: + """Create a workspace whose initial tab starts in ``path`` running ``command``.""" + cmd = ["workspace", "create", "--name", workspace_name, "--cwd", str(path)] + if command: + cmd.extend(["--command", command]) + workspace_ref = self._parse_ok_ref(self._run(cmd), "workspace:") + if workspace_ref is None: + return None + self._run( + [ + "workspace-action", + "--action", + "set-color", + "--workspace", + workspace_ref, + "--color", + self._workspace_color(workspace_name), + ], + ) + if tab_name: + # The freshly created workspace has a single tab, which is its + # focused tab, so rename-tab needs no explicit surface target. + self._run(["rename-tab", "--workspace", workspace_ref, "--title", tab_name]) + return TerminalHandle( + terminal_name=self.name, + handle=workspace_ref, + session_name=workspace_name, + ) + + def _open_tab( + self, + path: Path, + command: str | None, + tab_name: str | None, + *, + workspace_ref: str, + workspace_name: str, + ) -> TerminalHandle | None: + """Open a new tab in an existing workspace and run ``command`` in ``path``.""" + surface_ref = self._parse_ok_ref( + self._run(["new-surface", "--workspace", workspace_ref]), + "surface:", + ) + if surface_ref is None: + return None + if tab_name: + self._run( + [ + "rename-tab", + "--workspace", + workspace_ref, + "--surface", + surface_ref, + "--title", + tab_name, + ], + ) + # New surfaces don't accept a cwd/command, so type it into the shell. + # The terminal buffers the input until the shell is ready, and cmux + # turns the literal \n escape sequence into Enter. + shell_cmd = f"cd {shlex.quote(str(path))}" + if command: + shell_cmd += f" && {command}" + sent = self._run( + [ + "send", + "--workspace", + workspace_ref, + "--surface", + surface_ref, + "--", + shell_cmd + "\\n", + ], + ) + if sent is None: + # The command never reached the tab, so the agent is not running; + # remove the idle tab and report failure instead of a dead handle. + self._run(["close-surface", "--workspace", workspace_ref, "--surface", surface_ref]) + return None + return TerminalHandle( + terminal_name=self.name, + handle=surface_ref, + session_name=workspace_name, + ) + + @staticmethod + def _workspace_color(workspace_name: str) -> str: + """Pick a deterministic cmux named color for a workspace.""" + digest = hashlib.sha256(workspace_name.encode()).hexdigest() + return _WORKSPACE_COLORS[int(digest[:8], 16) % len(_WORKSPACE_COLORS)] + + @staticmethod + def _run(args: list[str]) -> str | None: + """Run a cmux CLI command and return its stdout, or None on failure.""" + env = {**os.environ, "CMUX_QUIET": "1"} + try: + result = subprocess.run( + ["cmux", *args], # noqa: S607 + check=True, + capture_output=True, + text=True, + env=env, + ) + except (subprocess.CalledProcessError, OSError): + return None + return result.stdout + + @staticmethod + def _parse_ok_ref(stdout: str | None, prefix: str) -> str | None: + """Extract the first ``N`` ref from an ``OK ...`` response line.""" + if not stdout: + return None + for line in stdout.splitlines(): + if not line.startswith("OK"): + continue + for token in line.split(): + if token.startswith(prefix): + return token + return None diff --git a/agent_cli/dev/terminals/registry.py b/agent_cli/dev/terminals/registry.py index e98140002..e72783df0 100644 --- a/agent_cli/dev/terminals/registry.py +++ b/agent_cli/dev/terminals/registry.py @@ -6,6 +6,7 @@ from .apple_terminal import AppleTerminal from .base import Terminal # noqa: TC001 +from .cmux import Cmux from .gnome import GnomeTerminal from .iterm2 import ITerm2 from .kitty import Kitty @@ -18,6 +19,7 @@ _TERMINALS: list[type[Terminal]] = [ Tmux, Zellij, + Cmux, ITerm2, Kitty, Warp, diff --git a/docs/commands/dev.md b/docs/commands/dev.md index 29216769d..be100e66a 100644 --- a/docs/commands/dev.md +++ b/docs/commands/dev.md @@ -578,6 +578,7 @@ agent-cli dev terminals [OPTIONS] |----------|-----------|-----------------| | tmux | `TMUX` env var | `tmux new-window -c ` | | Zellij | `ZELLIJ` env var | `zellij action new-tab --cwd ` | +| cmux | `CMUX_WORKSPACE_ID` env var | `cmux new-surface` in a workspace named after the repo (created on demand) | | Kitty | `KITTY_WINDOW_ID` | `kitten @ launch --type=tab` | | iTerm2 | `ITERM_SESSION_ID` | AppleScript | | Terminal.app | `TERM_PROGRAM=Apple_Terminal` | AppleScript + System Events * | diff --git a/tests/dev/test_launch.py b/tests/dev/test_launch.py index 781b34b11..56a3c523a 100644 --- a/tests/dev/test_launch.py +++ b/tests/dev/test_launch.py @@ -12,6 +12,7 @@ write_prompt_to_worktree, ) from agent_cli.dev.terminals import TerminalHandle +from agent_cli.dev.terminals.cmux import Cmux from agent_cli.dev.terminals.tmux import Tmux @@ -194,6 +195,35 @@ def test_uses_wrapper_script_for_requested_tmux(self, tmp_path: Path) -> None: session_name="agent-cli-repo-1234", ) + def test_cmux_uses_workspace_named_after_repo(self, tmp_path: Path) -> None: + """Detected cmux opens a tab in a workspace named after the repo.""" + agent = MagicMock() + agent.name = "codex" + agent.launch_command.return_value = ["codex"] + + cmux_terminal = Cmux() + handle = TerminalHandle("cmux", "surface:5", "repo") + + with ( + patch( + "agent_cli.dev.launch.terminals.detect_current_terminal", + return_value=cmux_terminal, + ), + patch.object(cmux_terminal, "is_available", return_value=True), + patch.object(cmux_terminal, "open_in_workspace", return_value=handle) as mock_open, + patch("agent_cli.dev.launch.worktree.get_main_repo_root", return_value=Path("/repo")), + patch("agent_cli.dev.launch.worktree.get_current_branch", return_value="feature"), + ): + result = launch_agent(tmp_path, agent) + + assert result == handle + mock_open.assert_called_once_with( + tmp_path, + "codex", + tab_name="feature", + workspace_name="repo", + ) + def test_non_tmux_success_does_not_fall_back_to_manual_instructions( self, tmp_path: Path, diff --git a/tests/dev/test_terminals.py b/tests/dev/test_terminals.py index 6d1434e86..4ed16407c 100644 --- a/tests/dev/test_terminals.py +++ b/tests/dev/test_terminals.py @@ -2,6 +2,7 @@ from __future__ import annotations +import shlex import subprocess from pathlib import Path from unittest.mock import MagicMock, patch @@ -10,11 +11,13 @@ from agent_cli.dev.terminals import ( Terminal, + TerminalHandle, detect_current_terminal, get_all_terminals, get_available_terminals, get_terminal, ) +from agent_cli.dev.terminals.cmux import Cmux from agent_cli.dev.terminals.kitty import Kitty from agent_cli.dev.terminals.tmux import Tmux, TmuxInventory, TmuxWindow from agent_cli.dev.terminals.zellij import Zellij @@ -402,6 +405,281 @@ def test_is_available(self) -> None: assert terminal.is_available() is True +class TestCmux: + """Tests for the cmux terminal. + + Evidence: `cmux --help` (cmux 0.64.14) and live verification on 2026-06-10. + cmux is controlled via a Unix socket through its bundled CLI, so all + operations are `cmux ` subprocess calls that work from any + terminal, not just inside cmux. + """ + + def test_detect_cmux(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Detect cmux via CMUX_WORKSPACE_ID environment variable. + + Evidence: `cmux --help` Environment section: "CMUX_WORKSPACE_ID + Auto-set in cmux terminals." (same for CMUX_SURFACE_ID). + """ + monkeypatch.delenv("CMUX_SURFACE_ID", raising=False) + monkeypatch.setenv("CMUX_WORKSPACE_ID", "AB56033C-F3AB-46DC-83D2-2891F13F47C5") + terminal = Cmux() + assert terminal.detect() is True + + def test_detect_cmux_not_set(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Not in cmux when its environment variables are not set.""" + monkeypatch.delenv("CMUX_WORKSPACE_ID", raising=False) + monkeypatch.delenv("CMUX_SURFACE_ID", raising=False) + terminal = Cmux() + assert terminal.detect() is False + + def test_is_available(self) -> None: + """Cmux is available if the CLI is in PATH. + + Evidence: the Homebrew cask installs the CLI binary at + /Applications/cmux.app/Contents/Resources/bin/cmux and cmux adds it + to PATH. + """ + terminal = Cmux() + with patch( + "shutil.which", return_value="/Applications/cmux.app/Contents/Resources/bin/cmux" + ): + assert terminal.is_available() is True + + def test_open_in_workspace_opens_tab_in_existing_workspace(self) -> None: + r"""An existing workspace (matched by title) gets a new named tab. + + Evidence (verified live against cmux 0.64.14): + - `cmux workspace list --json` returns {"workspaces": [{"ref": + "workspace:N", "title": ...}]}. + - `cmux new-surface --workspace ` creates a tab and prints + "OK surface:N pane:N workspace:N". + - `cmux rename-tab --workspace --surface --title ` + renames the tab. + - `cmux send` has no auto-Enter; per `cmux send --help` the literal + escape sequence \\n sends Enter. New surfaces don't accept a + cwd/command flag, so the launch command is typed into the shell; + the terminal buffers the input until the shell is ready (verified + live by sending immediately after surface creation). + """ + terminal = Cmux() + path = Path("/some/work tree") + workspaces_json = ( + '{"window_ref": "window:1", "workspaces": [' + '{"ref": "workspace:2", "title": "other"},' + '{"ref": "workspace:11", "title": "agent-cli"}]}' + ) + with ( + patch("shutil.which", return_value="/usr/local/bin/cmux"), + patch( + "subprocess.run", + side_effect=[ + MagicMock(stdout=workspaces_json), + MagicMock(stdout="OK surface:32 pane:25 workspace:11\n"), + MagicMock(stdout="OK action=rename tab=tab:32 workspace=workspace:11\n"), + MagicMock(stdout="OK surface:32 workspace:11\n"), + ], + ) as mock_run, + ): + handle = terminal.open_in_workspace( + path, + "echo hello", + tab_name="feature", + workspace_name="agent-cli", + ) + + assert handle is not None + assert handle.terminal_name == "cmux" + assert handle.handle == "surface:32" + assert handle.session_name == "agent-cli" + argvs = [call.args[0] for call in mock_run.call_args_list] + assert argvs == [ + ["cmux", "workspace", "list", "--json"], + ["cmux", "new-surface", "--workspace", "workspace:11"], + [ + "cmux", + "rename-tab", + "--workspace", + "workspace:11", + "--surface", + "surface:32", + "--title", + "feature", + ], + [ + "cmux", + "send", + "--workspace", + "workspace:11", + "--surface", + "surface:32", + "--", + f"cd {shlex.quote(str(path))} && echo hello\\n", + ], + ] + + def test_open_in_workspace_creates_missing_workspace(self) -> None: + """A missing workspace is created with cwd, command, and a stable color. + + Evidence (verified live against cmux 0.64.14): `cmux workspace create + --name --cwd --command ` prints "OK workspace:N", + starts the initial tab's shell in --cwd, and sends the command with + Enter after creation. `cmux workspace-action --action set-color + --color ` accepts the named colors listed in its --help. + `cmux rename-tab --workspace --title ` targets that + workspace's focused tab, which is the just-created single tab. + """ + terminal = Cmux() + path = Path("/some/path") + workspaces_json = '{"window_ref": "window:1", "workspaces": []}' + with ( + patch("shutil.which", return_value="/usr/local/bin/cmux"), + patch( + "subprocess.run", + side_effect=[ + MagicMock(stdout=workspaces_json), + MagicMock(stdout="OK workspace:13\n"), + MagicMock(stdout="OK workspace:13\n"), + MagicMock(stdout="OK action=rename tab=tab:33 workspace=workspace:13\n"), + ], + ) as mock_run, + ): + handle = terminal.open_in_workspace( + path, + "echo hello", + tab_name="feature", + workspace_name="my-repo", + ) + + assert handle is not None + assert handle.handle == "workspace:13" + assert handle.session_name == "my-repo" + argvs = [call.args[0] for call in mock_run.call_args_list] + assert argvs == [ + ["cmux", "workspace", "list", "--json"], + [ + "cmux", + "workspace", + "create", + "--name", + "my-repo", + "--cwd", + str(path), + "--command", + "echo hello", + ], + [ + "cmux", + "workspace-action", + "--action", + "set-color", + "--workspace", + "workspace:13", + "--color", + "Magenta", + ], + ["cmux", "rename-tab", "--workspace", "workspace:13", "--title", "feature"], + ] + + def test_workspace_color_is_deterministic(self) -> None: + """The same workspace name always maps to the same cmux named color.""" + assert Cmux._workspace_color("my-repo") == "Magenta" + assert Cmux._workspace_color("my-repo") == Cmux._workspace_color("my-repo") + + def test_open_tab_closes_surface_when_send_fails(self) -> None: + """A failed `send` closes the idle tab and reports failure. + + Without this, the caller would print a success message while the + agent never started in the new tab. + """ + terminal = Cmux() + send_error = subprocess.CalledProcessError(1, ["cmux", "send"]) + workspaces_json = ( + '{"window_ref": "window:1", "workspaces": [{"ref": "workspace:11", "title": "repo"}]}' + ) + with ( + patch("shutil.which", return_value="/usr/local/bin/cmux"), + patch( + "subprocess.run", + side_effect=[ + MagicMock(stdout=workspaces_json), + MagicMock(stdout="OK surface:32 pane:25 workspace:11\n"), + send_error, + MagicMock(stdout="OK surface:32\n"), + ], + ) as mock_run, + ): + handle = terminal.open_in_workspace( + Path("/some/path"), + "echo hello", + workspace_name="repo", + ) + + assert handle is None + assert mock_run.call_args_list[-1].args[0] == [ + "cmux", + "close-surface", + "--workspace", + "workspace:11", + "--surface", + "surface:32", + ] + + def test_open_new_tab_uses_repo_root_name(self, tmp_path: Path) -> None: + """The generic open_new_tab keeps the one-workspace-per-repo invariant. + + Worktree paths resolve to the main repo root, so tabs land in the + repo's workspace rather than one named after the worktree directory. + """ + terminal = Cmux() + handle = TerminalHandle("cmux", "surface:5", "repo") + with ( + patch( + "agent_cli.dev.terminals.cmux.get_main_repo_root", + return_value=Path("/repo"), + ), + patch.object(terminal, "open_in_workspace", return_value=handle) as mock_open, + ): + assert terminal.open_new_tab(tmp_path / "worktree", "echo hi", tab_name="t") is True + mock_open.assert_called_once_with( + tmp_path / "worktree", + "echo hi", + "t", + workspace_name="repo", + ) + + def test_run_sets_cmux_quiet(self) -> None: + """CLI calls silence cmux deprecation notices via CMUX_QUIET. + + Evidence: legacy command forms print "set CMUX_QUIET=1 to silence + this notice", which would corrupt parsed output. + """ + terminal = Cmux() + with patch("subprocess.run", return_value=MagicMock(stdout="PONG\n")) as mock_run: + assert terminal._run(["ping"]) == "PONG\n" + assert mock_run.call_args.kwargs["env"]["CMUX_QUIET"] == "1" + + def test_open_in_workspace_returns_none_on_cli_failure(self) -> None: + """A failing workspace listing aborts immediately, without a create attempt. + + Listing failure (e.g. app not running) is distinct from "workspace + not found": creating anyway would fail too, or duplicate a workspace + the listing could not see. + """ + terminal = Cmux() + error = subprocess.CalledProcessError(1, ["cmux", "workspace", "list", "--json"]) + with ( + patch("shutil.which", return_value="/usr/local/bin/cmux"), + patch("subprocess.run", side_effect=[error]) as mock_run, + ): + handle = terminal.open_in_workspace( + Path("/some/path"), + "echo hello", + workspace_name="my-repo", + ) + assert handle is None + assert mock_run.call_count == 1 + + class TestRegistry: """Tests for terminal registry functions.""" @@ -443,6 +721,23 @@ def test_detect_current_terminal_zellij(self, monkeypatch: pytest.MonkeyPatch) - assert terminal is not None assert terminal.name == "zellij" + def test_detect_current_terminal_cmux(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Detect current terminal as cmux when no multiplexer is running.""" + monkeypatch.delenv("TMUX", raising=False) + monkeypatch.delenv("ZELLIJ", raising=False) + monkeypatch.setenv("CMUX_WORKSPACE_ID", "AB56033C-F3AB-46DC-83D2-2891F13F47C5") + terminal = detect_current_terminal() + assert terminal is not None + assert terminal.name == "cmux" + + def test_detect_tmux_wins_inside_cmux(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Tmux running inside a cmux tab is detected as tmux (innermost wins).""" + monkeypatch.setenv("TMUX", "/run/user/1000/tmux") + monkeypatch.setenv("CMUX_WORKSPACE_ID", "AB56033C-F3AB-46DC-83D2-2891F13F47C5") + terminal = detect_current_terminal() + assert terminal is not None + assert terminal.name == "tmux" + def test_get_available_terminals(self) -> None: """Get available terminals returns list.""" terminals = get_available_terminals()