Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions agent_cli/dev/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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 ""
),
)
Expand Down Expand Up @@ -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`.
Expand Down
34 changes: 34 additions & 0 deletions agent_cli/dev/launch.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand Down
247 changes: 247 additions & 0 deletions agent_cli/dev/terminals/cmux.py
Original file line number Diff line number Diff line change
@@ -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
Comment thread
greptile-apps[bot] marked this conversation as resolved.

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 ``<prefix>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
2 changes: 2 additions & 0 deletions agent_cli/dev/terminals/registry.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -18,6 +19,7 @@
_TERMINALS: list[type[Terminal]] = [
Tmux,
Zellij,
Cmux,
ITerm2,
Kitty,
Warp,
Expand Down
1 change: 1 addition & 0 deletions docs/commands/dev.md
Original file line number Diff line number Diff line change
Expand Up @@ -578,6 +578,7 @@ agent-cli dev terminals [OPTIONS]
|----------|-----------|-----------------|
| tmux | `TMUX` env var | `tmux new-window -c <path>` |
| Zellij | `ZELLIJ` env var | `zellij action new-tab --cwd <path>` |
| 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 * |
Expand Down
30 changes: 30 additions & 0 deletions tests/dev/test_launch.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -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,
Expand Down
Loading
Loading