marketing-shibata50/cmux
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add cmux claude-teams launcher (#1179)

Browse source 
* Add claude-teams CLI command

* Add claude-teams launcher regression test

* Exec claude-teams launcher in place

* Add existing-shim claude-teams regression test

* Reuse claude-teams shim and refresh dev CLI

* Add wrapper-selection claude-teams regression test

* Launch real claude binary for claude-teams

* Add claude-teams auto-mode launcher regression test

* Default claude-teams to fake tmux auto mode

* Build tagged reloads under DerivedData

* Add claude-teams tmux sequence regression test

* Fix claude-teams tmux teammate compatibility

* Add claude-teams split focus regression test

* Keep claude-teams leader pane focused

* Tighten claude-teams review fixes

* Pass claude-teams help through to Claude

* Use sentinel TERM_PROGRAM in claude-teams test
This commit is contained in:
Lawrence Chen 2026-03-11 02:42:33 -07:00 • committed by GitHub
parent 0cdeb451e8
commit 00587ed856
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 2055 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

1170
CLI/cmux.swift
View file

File diff suppressed because it is too large Load diff

17
Resources/Localizable.xcstrings
View file

@ -115,6 +115,23 @@
}
}
},
"cli.claude-teams.usage": {
"extractionState": "manual",
"localizations": {
"en": {
"stringUnit": {
"state": "translated",
"value": "Usage: cmux claude-teams [claude-args...]\n\nLaunch Claude Code with agent teams enabled.\n\nThis command:\n - defaults Claude teammate mode to auto\n - sets a tmux-like environment so Claude auto mode uses cmux splits\n - sets CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS=1\n - prepends a private tmux shim to PATH\n - forwards all remaining arguments to claude\n\nThe tmux shim translates supported tmux window/pane commands into cmux\nworkspace and split operations in the current cmux session.\n\nExamples:\n cmux claude-teams\n cmux claude-teams --continue\n cmux claude-teams --model sonnet"
}
},
"ja": {
"stringUnit": {
"state": "translated",
"value": "使い方: cmux claude-teams [claude-args...]\n\nエージェントチームを有効にした状態で Claude Code を起動します。\n\nこのコマンドは次を行います:\n - Claude の teammate mode を auto に設定\n - Claude の auto mode が cmux の split を使うよう tmux 風の環境を設定\n - CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS=1 を設定\n - 専用の tmux shim を PATH の先頭に追加\n - 残りの引数をそのまま claude に渡す\n\ntmux shim は、対応している tmux の window/pane コマンドを、現在の cmux セッション内の workspace と split 操作に変換します。\n\n例:\n cmux claude-teams\n cmux claude-teams --continue\n cmux claude-teams --model sonnet"
}
}
}
},
"applescript.error.disabled": {
"extractionState": "manual",
"localizations": {

37
scripts/reload.sh
View file

@ -45,6 +45,11 @@ sanitize_path() {
echo "$cleaned"
}
tagged_derived_data_path() {
local slug="$1"
echo "$HOME/Library/Developer/Xcode/DerivedData/cmux-${slug}"
}
print_tag_cleanup_reminder() {
local current_slug="$1"
local path=""
@ -53,7 +58,13 @@ print_tag_cleanup_reminder() {
local -a stale_tags=()
while IFS= read -r -d '' path; do
tag="${path#/tmp/cmux-}"
if [[ "$path" == /tmp/cmux-* ]]; then
tag="${path#/tmp/cmux-}"
elif [[ "$path" == "$HOME/Library/Developer/Xcode/DerivedData/cmux-"* ]]; then
tag="${path#$HOME/Library/Developer/Xcode/DerivedData/cmux-}"
else
continue
fi
if [[ "$tag" == "$current_slug" ]]; then
continue
fi
@ -66,7 +77,10 @@ print_tag_cleanup_reminder() {
fi
seen="${seen}${tag} "
stale_tags+=("$tag")
done < <(find /tmp -maxdepth 1 -type d -name 'cmux-*' -print0 2>/dev/null)
done < <(
find /tmp -maxdepth 1 -name 'cmux-*' -print0 2>/dev/null
find "$HOME/Library/Developer/Xcode/DerivedData" -maxdepth 1 -type d -name 'cmux-*' -print0 2>/dev/null
)
echo
echo "Tag cleanup status:"
@ -82,14 +96,14 @@ print_tag_cleanup_reminder() {
echo "Cleanup stale tags only:"
for tag in "${stale_tags[@]}"; do
echo " pkill -f \"cmux DEV ${tag}.app/Contents/MacOS/cmux DEV\""
echo " rm -rf \"/tmp/cmux-${tag}\" \"/tmp/cmux-debug-${tag}.sock\""
echo " rm -rf \"$(tagged_derived_data_path "$tag")\" \"/tmp/cmux-${tag}\" \"/tmp/cmux-debug-${tag}.sock\""
echo " rm -f \"/tmp/cmux-debug-${tag}.log\""
echo " rm -f \"$HOME/Library/Application Support/cmux/cmuxd-dev-${tag}.sock\""
done
fi
echo "After you verify current tag, cleanup command:"
echo " pkill -f \"cmux DEV ${current_slug}.app/Contents/MacOS/cmux DEV\""
echo " rm -rf \"/tmp/cmux-${current_slug}\" \"/tmp/cmux-debug-${current_slug}.sock\""
echo " rm -rf \"$(tagged_derived_data_path "$current_slug")\" \"/tmp/cmux-${current_slug}\" \"/tmp/cmux-debug-${current_slug}.sock\""
echo " rm -f \"/tmp/cmux-debug-${current_slug}.log\""
echo " rm -f \"$HOME/Library/Application Support/cmux/cmuxd-dev-${current_slug}.sock\""
}
@ -159,7 +173,7 @@ if [[ -n "$TAG" ]]; then
BUNDLE_ID="com.cmuxterm.app.debug.${TAG_ID}"
fi
if [[ "$DERIVED_SET" -eq 0 ]]; then
DERIVED_DATA="/tmp/cmux-${TAG_SLUG}"
DERIVED_DATA="$(tagged_derived_data_path "$TAG_SLUG")"
fi
fi
@ -230,6 +244,15 @@ if [[ -z "${APP_PATH}" || ! -d "${APP_PATH}" ]]; then
exit 1
fi
if [[ -n "${TAG_SLUG:-}" ]]; then
TMP_COMPAT_DERIVED_LINK="/tmp/cmux-${TAG_SLUG}"
if [[ "$DERIVED_DATA" != "$TMP_COMPAT_DERIVED_LINK" ]]; then
ABS_DERIVED_DATA="$(cd "$DERIVED_DATA" && pwd)"
rm -rf "$TMP_COMPAT_DERIVED_LINK"
ln -s "$ABS_DERIVED_DATA" "$TMP_COMPAT_DERIVED_LINK"
fi
fi
if [[ -n "$TAG" && "$APP_NAME" != "$SEARCH_APP_NAME" ]]; then
TAG_APP_PATH="$(dirname "$APP_PATH")/${APP_NAME}.app"
rm -rf "$TAG_APP_PATH"
@ -292,6 +315,10 @@ if [[ -x "$CMUXD_SRC" ]]; then
cp "$CMUXD_SRC" "$BIN_DIR/cmuxd"
chmod +x "$BIN_DIR/cmuxd"
fi
CLI_PATH="$APP_PATH/Contents/Resources/bin/cmux"
if [[ -x "$CLI_PATH" ]]; then
echo "$CLI_PATH" > /tmp/cmux-last-cli-path || true
fi
# Avoid inheriting cmux/ghostty environment variables from the terminal that
# runs this script (often inside another cmux instance), which can cause
# socket and resource-path conflicts.

22
tests/claude_teams_test_utils.py Normal file
View file

@ -0,0 +1,22 @@
#!/usr/bin/env python3
from __future__ import annotations
import os
from pathlib import Path
def resolve_cmux_cli() -> str:
explicit = os.environ.get("CMUX_CLI_BIN") or os.environ.get("CMUX_CLI")
if explicit and os.path.exists(explicit) and os.access(explicit, os.X_OK):
return explicit
recorded_path = Path("/tmp/cmux-last-cli-path")
if recorded_path.exists():
candidate = recorded_path.read_text(encoding="utf-8").strip()
if candidate and os.path.exists(candidate) and os.access(candidate, os.X_OK):
return candidate
raise RuntimeError(
"Unable to find cmux CLI binary. Set CMUX_CLI_BIN or run ./scripts/reload.sh --tag <tag> first."
)

192
tests/test_cli_claude_teams_env.py Normal file
View file

@ -0,0 +1,192 @@
#!/usr/bin/env python3
"""
Regression test: `cmux claude-teams` injects the tmux-style auto-mode env.
"""
from __future__ import annotations
import os
import subprocess
import tempfile
from pathlib import Path
from claude_teams_test_utils import resolve_cmux_cli
def make_executable(path: Path, content: str) -> None:
path.write_text(content, encoding="utf-8")
path.chmod(0o755)
def read_text(path: Path) -> str:
if not path.exists():
return ""
return path.read_text(encoding="utf-8").strip()
def main() -> int:
try:
cli_path = resolve_cmux_cli()
except Exception as exc:
print(f"FAIL: {exc}")
return 1
with tempfile.TemporaryDirectory(prefix="cmux-claude-teams-env-") as td:
tmp = Path(td)
real_bin = tmp / "real-bin"
real_bin.mkdir(parents=True, exist_ok=True)
env_log = tmp / "agent-teams.log"
tmux_log = tmp / "tmux-path.log"
cmux_bin_log = tmp / "cmux-bin.log"
argv_log = tmp / "argv.log"
tmux_env_log = tmp / "tmux-env.log"
tmux_pane_log = tmp / "tmux-pane.log"
term_log = tmp / "term.log"
term_program_log = tmp / "term-program.log"
socket_path_log = tmp / "socket-path.log"
socket_password_log = tmp / "socket-password.log"
fake_home = tmp / "home"
fake_home.mkdir(parents=True, exist_ok=True)
make_executable(
real_bin / "claude",
"""#!/usr/bin/env bash
set -euo pipefail
printf '%s\\n' "${CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS-__UNSET__}" > "$FAKE_AGENT_TEAMS_LOG"
command -v tmux > "$FAKE_TMUX_PATH_LOG"
printf '%s\\n' "${CMUX_CLAUDE_TEAMS_CMUX_BIN-__UNSET__}" > "$FAKE_CMUX_BIN_LOG"
printf '%s\\n' "$@" > "$FAKE_ARGV_LOG"
printf '%s\\n' "${TMUX-__UNSET__}" > "$FAKE_TMUX_ENV_LOG"
printf '%s\\n' "${TMUX_PANE-__UNSET__}" > "$FAKE_TMUX_PANE_LOG"
printf '%s\\n' "${TERM-__UNSET__}" > "$FAKE_TERM_LOG"
printf '%s\\n' "${TERM_PROGRAM-__UNSET__}" > "$FAKE_TERM_PROGRAM_LOG"
printf '%s\\n' "${CMUX_SOCKET_PATH-__UNSET__}" > "$FAKE_SOCKET_PATH_LOG"
printf '%s\\n' "${CMUX_SOCKET_PASSWORD-__UNSET__}" > "$FAKE_SOCKET_PASSWORD_LOG"
""",
)
env = os.environ.copy()
env["HOME"] = str(fake_home)
env["PATH"] = f"{real_bin}:/usr/bin:/bin"
env["FAKE_AGENT_TEAMS_LOG"] = str(env_log)
env["FAKE_TMUX_PATH_LOG"] = str(tmux_log)
env["FAKE_CMUX_BIN_LOG"] = str(cmux_bin_log)
env["FAKE_ARGV_LOG"] = str(argv_log)
env["FAKE_TMUX_ENV_LOG"] = str(tmux_env_log)
env["FAKE_TMUX_PANE_LOG"] = str(tmux_pane_log)
env["FAKE_TERM_LOG"] = str(term_log)
env["FAKE_TERM_PROGRAM_LOG"] = str(term_program_log)
env["FAKE_SOCKET_PATH_LOG"] = str(socket_path_log)
env["FAKE_SOCKET_PASSWORD_LOG"] = str(socket_password_log)
env["TMUX"] = "__HOST_TMUX__"
env["TMUX_PANE"] = "%999"
env["TERM"] = "xterm-256color"
env["TERM_PROGRAM"] = "__HOST_TERM_PROGRAM__"
explicit_socket_path = str(tmp / "explicit-cmux.sock")
explicit_socket_password = "topsecret"
proc = subprocess.run(
[
cli_path,
"--socket",
explicit_socket_path,
"--password",
explicit_socket_password,
"claude-teams",
"--version",
],
capture_output=True,
text=True,
check=False,
env=env,
timeout=30,
)
if proc.returncode != 0:
print("FAIL: `cmux claude-teams --version` exited non-zero")
print(f"exit={proc.returncode}")
print(f"stdout={proc.stdout.strip()}")
print(f"stderr={proc.stderr.strip()}")
return 1
agent_teams_value = read_text(env_log)
if agent_teams_value != "1":
print(f"FAIL: expected CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS=1, got {agent_teams_value!r}")
return 1
tmux_path = read_text(tmux_log)
if not tmux_path:
print("FAIL: fake claude did not observe a tmux binary in PATH")
return 1
tmux_name = Path(tmux_path).name
if tmux_name != "tmux":
print(f"FAIL: expected tmux shim path to end with 'tmux', got {tmux_path!r}")
return 1
if "claude-teams-bin" not in tmux_path:
print(f"FAIL: expected stable tmux shim path, got {tmux_path!r}")
return 1
if tmux_path.startswith(str(real_bin)):
print(f"FAIL: expected cmux tmux shim to shadow PATH, got {tmux_path!r}")
return 1
cmux_bin_value = read_text(cmux_bin_log)
if not cmux_bin_value or cmux_bin_value == "__UNSET__":
print("FAIL: missing CMUX_CLAUDE_TEAMS_CMUX_BIN")
return 1
if not os.path.exists(cmux_bin_value):
print(f"FAIL: CMUX_CLAUDE_TEAMS_CMUX_BIN does not exist: {cmux_bin_value!r}")
return 1
argv_lines = argv_log.read_text(encoding="utf-8").splitlines()
if argv_lines[:2] != ["--teammate-mode", "auto"]:
print(f"FAIL: expected launcher to prepend --teammate-mode auto, got {argv_lines!r}")
return 1
if "--version" not in argv_lines:
print(f"FAIL: expected launcher to preserve user args, got {argv_lines!r}")
return 1
tmux_env_value = read_text(tmux_env_log)
if tmux_env_value in {"", "__UNSET__"}:
print("FAIL: expected a fake TMUX env value")
return 1
tmux_pane_value = read_text(tmux_pane_log)
if tmux_pane_value in {"", "__UNSET__"} or not tmux_pane_value.startswith("%"):
print(f"FAIL: expected a fake TMUX_PANE value, got {tmux_pane_value!r}")
return 1
term_value = read_text(term_log)
if term_value != "screen-256color":
print(f"FAIL: expected TERM=screen-256color, got {term_value!r}")
return 1
term_program_value = read_text(term_program_log)
if term_program_value != "__UNSET__":
print(f"FAIL: expected TERM_PROGRAM to be unset, got {term_program_value!r}")
return 1
socket_path_value = read_text(socket_path_log)
if socket_path_value != explicit_socket_path:
print(f"FAIL: expected CMUX_SOCKET_PATH={explicit_socket_path!r}, got {socket_path_value!r}")
return 1
socket_password_value = read_text(socket_password_log)
if socket_password_value != explicit_socket_password:
print(
"FAIL: expected CMUX_SOCKET_PASSWORD to preserve the explicit CLI override, "
f"got {socket_password_value!r}"
)
return 1
print("PASS: cmux claude-teams injects the auto-mode tmux env and shim")
return 0
if __name__ == "__main__":
raise SystemExit(main())

90
tests/test_cli_claude_teams_existing_shim.py Normal file
View file

@ -0,0 +1,90 @@
#!/usr/bin/env python3
"""
Regression test: `cmux claude-teams` reuses an existing tmux shim.
"""
from __future__ import annotations
import os
import stat
import subprocess
import tempfile
from pathlib import Path
from claude_teams_test_utils import resolve_cmux_cli
def make_executable(path: Path, content: str) -> None:
path.write_text(content, encoding="utf-8")
path.chmod(0o755)
def main() -> int:
try:
cli_path = resolve_cmux_cli()
except Exception as exc:
print(f"FAIL: {exc}")
return 1
with tempfile.TemporaryDirectory(prefix="cmux-claude-teams-shim-") as td:
tmp = Path(td)
home = tmp / "home"
real_bin = tmp / "real-bin"
home.mkdir(parents=True, exist_ok=True)
real_bin.mkdir(parents=True, exist_ok=True)
shim_dir = home / ".cmuxterm" / "claude-teams-bin"
shim_dir.mkdir(parents=True, exist_ok=True)
shim_path = shim_dir / "tmux"
shim_path.write_text(
"#!/usr/bin/env bash\n"
"set -euo pipefail\n"
"exec \"${CMUX_CLAUDE_TEAMS_CMUX_BIN:-cmux}\" __tmux-compat \"$@\"\n",
encoding="utf-8",
)
shim_path.chmod(0o555)
shim_dir.chmod(0o555)
make_executable(
real_bin / "claude",
"""#!/usr/bin/env bash
set -euo pipefail
printf 'shim=%s\\n' "$(command -v tmux)"
""",
)
env = os.environ.copy()
env["HOME"] = str(home)
env["PATH"] = f"{real_bin}:/usr/bin:/bin"
proc = subprocess.run(
[cli_path, "claude-teams", "--version"],
capture_output=True,
text=True,
check=False,
env=env,
timeout=30,
)
shim_dir.chmod(0o755)
shim_path.chmod(0o755)
if proc.returncode != 0:
print("FAIL: `cmux claude-teams --version` failed with an existing shim")
print(f"exit={proc.returncode}")
print(f"stdout={proc.stdout.strip()}")
print(f"stderr={proc.stderr.strip()}")
return 1
expected = str(shim_path)
actual = proc.stdout.strip()
if actual != f"shim={expected}":
print(f"FAIL: expected existing shim path {expected!r}, got {actual!r}")
return 1
print("PASS: cmux claude-teams reuses an existing tmux shim")
return 0
if __name__ == "__main__":
raise SystemExit(main())

86
tests/test_cli_claude_teams_help_passthrough.py Normal file
View file

@ -0,0 +1,86 @@
#!/usr/bin/env python3
"""
Regression test: `cmux claude-teams --help` passes through to Claude.
"""
from __future__ import annotations
import os
import subprocess
import tempfile
from pathlib import Path
from claude_teams_test_utils import resolve_cmux_cli
def make_executable(path: Path, content: str) -> None:
path.write_text(content, encoding="utf-8")
path.chmod(0o755)
def main() -> int:
try:
cli_path = resolve_cmux_cli()
except Exception as exc:
print(f"FAIL: {exc}")
return 1
with tempfile.TemporaryDirectory(prefix="cmux-claude-teams-help-") as td:
tmp = Path(td)
home = tmp / "home"
real_bin = tmp / "real-bin"
home.mkdir(parents=True, exist_ok=True)
real_bin.mkdir(parents=True, exist_ok=True)
argv_log = tmp / "argv.log"
make_executable(
real_bin / "claude",
"""#!/usr/bin/env bash
set -euo pipefail
printf '%s\\n' "$@" > "$FAKE_ARGV_LOG"
""",
)
env = os.environ.copy()
env["HOME"] = str(home)
env["PATH"] = f"{real_bin}:/usr/bin:/bin"
env["FAKE_ARGV_LOG"] = str(argv_log)
proc = subprocess.run(
[cli_path, "claude-teams", "--help"],
capture_output=True,
text=True,
check=False,
env=env,
timeout=30,
)
if proc.returncode != 0:
print("FAIL: `cmux claude-teams --help` exited non-zero")
print(f"exit={proc.returncode}")
print(f"stdout={proc.stdout.strip()}")
print(f"stderr={proc.stderr.strip()}")
return 1
if not argv_log.exists():
print("FAIL: launcher intercepted --help instead of invoking Claude")
print(f"stdout={proc.stdout.strip()}")
print(f"stderr={proc.stderr.strip()}")
return 1
argv_lines = argv_log.read_text(encoding="utf-8").splitlines()
if argv_lines[:2] != ["--teammate-mode", "auto"]:
print(f"FAIL: expected launcher to prepend --teammate-mode auto, got {argv_lines!r}")
return 1
if "--help" not in argv_lines:
print(f"FAIL: expected --help to reach Claude, got {argv_lines!r}")
return 1
print("PASS: cmux claude-teams forwards --help to Claude")
return 0
if __name__ == "__main__":
raise SystemExit(main())

85
tests/test_cli_claude_teams_skips_wrapper_claude.py Normal file
View file

@ -0,0 +1,85 @@
#!/usr/bin/env python3
"""
Regression test: `cmux claude-teams` skips cmux wrapper scripts on PATH.
"""
from __future__ import annotations
import os
import subprocess
import tempfile
from pathlib import Path
from claude_teams_test_utils import resolve_cmux_cli
def make_executable(path: Path, content: str) -> None:
path.write_text(content, encoding="utf-8")
path.chmod(0o755)
def main() -> int:
try:
cli_path = resolve_cmux_cli()
except Exception as exc:
print(f"FAIL: {exc}")
return 1
with tempfile.TemporaryDirectory(prefix="cmux-claude-teams-wrapper-") as td:
tmp = Path(td)
wrapper_bin = tmp / "wrapper-bin"
real_bin = tmp / "real-bin"
logs = tmp / "logs"
wrapper_bin.mkdir(parents=True, exist_ok=True)
real_bin.mkdir(parents=True, exist_ok=True)
logs.mkdir(parents=True, exist_ok=True)
real_hit = logs / "real-hit.txt"
make_executable(
wrapper_bin / "claude",
"""#!/usr/bin/env bash
# cmux claude wrapper - injects hooks and session tracking
set -euo pipefail
echo WRAPPER_EXECUTED >&2
exit 91
""",
)
make_executable(
real_bin / "claude",
f"""#!/usr/bin/env bash
set -euo pipefail
printf 'REAL\\n' > {real_hit}
""",
)
env = os.environ.copy()
env["PATH"] = f"{wrapper_bin}:{real_bin}:/usr/bin:/bin"
proc = subprocess.run(
[cli_path, "claude-teams", "--version"],
capture_output=True,
text=True,
check=False,
env=env,
timeout=30,
)
if proc.returncode != 0:
print("FAIL: `cmux claude-teams --version` executed a wrapper instead of the real claude binary")
print(f"exit={proc.returncode}")
print(f"stdout={proc.stdout.strip()}")
print(f"stderr={proc.stderr.strip()}")
return 1
if not real_hit.exists():
print("FAIL: real claude binary was not reached")
return 1
print("PASS: cmux claude-teams skips cmux wrapper scripts on PATH")
return 0
if __name__ == "__main__":
raise SystemExit(main())

362
tests/test_cli_claude_teams_tmux_sequence.py Normal file
View file

@ -0,0 +1,362 @@
#!/usr/bin/env python3
"""
Regression test: `cmux claude-teams` supports Claude's tmux teammate flow.
"""
from __future__ import annotations
import json
import os
import socketserver
import subprocess
import tempfile
import threading
from pathlib import Path
from claude_teams_test_utils import resolve_cmux_cli
INITIAL_WORKSPACE_ID = "11111111-1111-4111-8111-111111111111"
INITIAL_WINDOW_ID = "22222222-2222-4222-8222-222222222222"
INITIAL_PANE_ID = "33333333-3333-4333-8333-333333333333"
INITIAL_SURFACE_ID = "44444444-4444-4444-8444-444444444444"
INITIAL_TAB_ID = "55555555-5555-4555-8555-555555555555"
NEW_PANE_ID = "66666666-6666-4666-8666-666666666666"
NEW_SURFACE_ID = "77777777-7777-4777-8777-777777777777"
def make_executable(path: Path, content: str) -> None:
path.write_text(content, encoding="utf-8")
path.chmod(0o755)
def read_text(path: Path) -> str:
if not path.exists():
return ""
return path.read_text(encoding="utf-8").strip()
class FakeCmuxState:
def __init__(self) -> None:
self.lock = threading.Lock()
self.requests: list[str] = []
self.workspace = {
"id": INITIAL_WORKSPACE_ID,
"ref": "workspace:1",
"index": 1,
"title": "demo-team",
}
self.window = {
"id": INITIAL_WINDOW_ID,
"ref": "window:1",
}
self.current_pane_id = INITIAL_PANE_ID
self.current_surface_id = INITIAL_SURFACE_ID
self.panes = [
{
"id": INITIAL_PANE_ID,
"ref": "pane:1",
"index": 7,
"surface_ids": [INITIAL_SURFACE_ID],
}
]
self.surfaces = [
{
"id": INITIAL_SURFACE_ID,
"ref": "surface:1",
"pane_id": INITIAL_PANE_ID,
"title": "leader",
}
]
def handle(self, method: str, params: dict[str, object]) -> dict[str, object]:
with self.lock:
self.requests.append(method)
if method == "system.identify":
return {
"socket_path": str(params.get("socket_path", "")),
"focused": {
"workspace_id": self.workspace["id"],
"workspace_ref": self.workspace["ref"],
"window_id": self.window["id"],
"window_ref": self.window["ref"],
"pane_id": self.current_pane_id,
"pane_ref": self._pane_ref(self.current_pane_id),
"surface_id": self.current_surface_id,
"surface_ref": self._surface_ref(self.current_surface_id),
"tab_id": INITIAL_TAB_ID,
"tab_ref": "tab:1",
"surface_type": "terminal",
"is_browser_surface": False,
},
}
if method == "workspace.current":
return {
"workspace_id": self.workspace["id"],
"workspace_ref": self.workspace["ref"],
}
if method == "workspace.list":
return {
"workspaces": [
{
"id": self.workspace["id"],
"ref": self.workspace["ref"],
"index": self.workspace["index"],
"title": self.workspace["title"],
}
]
}
if method == "window.list":
return {
"windows": [
{
"id": self.window["id"],
"ref": self.window["ref"],
"workspace_id": self.workspace["id"],
"workspace_ref": self.workspace["ref"],
}
]
}
if method == "pane.list":
return {
"panes": [
{
"id": pane["id"],
"ref": pane["ref"],
"index": pane["index"],
}
for pane in self.panes
]
}
if method == "pane.surfaces":
pane_id = str(params.get("pane_id") or "")
pane = self._pane_by_id(pane_id)
return {
"surfaces": [
{
"id": surface_id,
"selected": surface_id == self.current_surface_id,
}
for surface_id in pane["surface_ids"]
]
}
if method == "surface.current":
return {
"workspace_id": self.workspace["id"],
"workspace_ref": self.workspace["ref"],
"pane_id": self.current_pane_id,
"pane_ref": self._pane_ref(self.current_pane_id),
"surface_id": self.current_surface_id,
"surface_ref": self._surface_ref(self.current_surface_id),
}
if method == "surface.list":
return {
"surfaces": [
{
"id": surface["id"],
"ref": surface["ref"],
"title": surface["title"],
"pane_id": surface["pane_id"],
"pane_ref": self._pane_ref(surface["pane_id"]),
}
for surface in self.surfaces
]
}
if method == "surface.split":
self.panes.append(
{
"id": NEW_PANE_ID,
"ref": "pane:2",
"index": 8,
"surface_ids": [NEW_SURFACE_ID],
}
)
self.surfaces.append(
{
"id": NEW_SURFACE_ID,
"ref": "surface:2",
"pane_id": NEW_PANE_ID,
"title": "teammate",
}
)
return {
"surface_id": NEW_SURFACE_ID,
"pane_id": NEW_PANE_ID,
}
if method == "surface.focus":
self.current_surface_id = str(params.get("surface_id") or self.current_surface_id)
surface = self._surface_by_id(self.current_surface_id)
self.current_pane_id = surface["pane_id"]
return {"ok": True}
if method == "pane.resize":
return {"ok": True}
if method == "surface.send_text":
return {"ok": True}
raise RuntimeError(f"Unsupported fake cmux method: {method}")
def _pane_by_id(self, pane_id: str) -> dict[str, object]:
for pane in self.panes:
if pane["id"] == pane_id or pane["ref"] == pane_id:
return pane
raise RuntimeError(f"Unknown pane id: {pane_id}")
def _surface_by_id(self, surface_id: str) -> dict[str, object]:
for surface in self.surfaces:
if surface["id"] == surface_id or surface["ref"] == surface_id:
return surface
raise RuntimeError(f"Unknown surface id: {surface_id}")
def _pane_ref(self, pane_id: str) -> str:
return self._pane_by_id(pane_id)["ref"] # type: ignore[return-value]
def _surface_ref(self, surface_id: str) -> str:
return self._surface_by_id(surface_id)["ref"] # type: ignore[return-value]
class FakeCmuxUnixServer(socketserver.ThreadingUnixStreamServer):
allow_reuse_address = True
def __init__(self, socket_path: str, state: FakeCmuxState) -> None:
self.state = state
super().__init__(socket_path, FakeCmuxHandler)
class FakeCmuxHandler(socketserver.StreamRequestHandler):
def handle(self) -> None:
while True:
line = self.rfile.readline()
if not line:
return
request = json.loads(line.decode("utf-8"))
response = {
"ok": True,
"result": self.server.state.handle( # type: ignore[attr-defined]
request["method"],
request.get("params", {}),
),
"id": request.get("id"),
}
self.wfile.write((json.dumps(response) + "\n").encode("utf-8"))
self.wfile.flush()
def main() -> int:
try:
cli_path = resolve_cmux_cli()
except Exception as exc:
print(f"FAIL: {exc}")
return 1
with tempfile.TemporaryDirectory(prefix="cmux-claude-teams-seq-") as td:
tmp = Path(td)
home = tmp / "home"
home.mkdir(parents=True, exist_ok=True)
socket_path = tmp / "fake-cmux.sock"
state = FakeCmuxState()
server = FakeCmuxUnixServer(str(socket_path), state)
thread = threading.Thread(target=server.serve_forever, daemon=True)
thread.start()
real_bin = tmp / "real-bin"
real_bin.mkdir(parents=True, exist_ok=True)
tmux_pane_log = tmp / "tmux-pane.log"
tmux_socket_log = tmp / "tmux-socket.log"
window_target_log = tmp / "window-target.log"
split_pane_log = tmp / "split-pane.log"
pane_list_log = tmp / "pane-list.log"
make_executable(
real_bin / "claude",
"""#!/usr/bin/env bash
set -euo pipefail
printf '%s\\n' "${TMUX_PANE-__UNSET__}" > "$FAKE_TMUX_PANE_LOG"
printf '%s\\n' "${CMUX_SOCKET_PATH-__UNSET__}" > "$FAKE_SOCKET_LOG"
window_target="$(tmux display-message -t "${TMUX_PANE}" -p '#{session_name}:#{window_index}')"
printf '%s\\n' "$window_target" > "$FAKE_WINDOW_TARGET_LOG"
split_pane="$(tmux split-window -t "${TMUX_PANE}" -h -l 70% -P -F '#{pane_id}')"
printf '%s\\n' "$split_pane" > "$FAKE_SPLIT_PANE_LOG"
tmux select-layout -t "$window_target" main-vertical
tmux resize-pane -t "${TMUX_PANE}" -x 30%
tmux list-panes -t "$window_target" -F '#{pane_id}' > "$FAKE_PANE_LIST_LOG"
""",
)
env = os.environ.copy()
env["HOME"] = str(home)
env["PATH"] = f"{real_bin}:/usr/bin:/bin"
env["CMUX_SOCKET_PATH"] = str(socket_path)
env["FAKE_TMUX_PANE_LOG"] = str(tmux_pane_log)
env["FAKE_SOCKET_LOG"] = str(tmux_socket_log)
env["FAKE_WINDOW_TARGET_LOG"] = str(window_target_log)
env["FAKE_SPLIT_PANE_LOG"] = str(split_pane_log)
env["FAKE_PANE_LIST_LOG"] = str(pane_list_log)
try:
proc = subprocess.run(
[cli_path, "claude-teams", "--version"],
capture_output=True,
text=True,
check=False,
env=env,
timeout=30,
)
except subprocess.TimeoutExpired as exc:
print("FAIL: `cmux claude-teams --version` timed out")
print(f"cmd={exc.cmd!r}")
return 1
finally:
server.shutdown()
server.server_close()
thread.join(timeout=2)
if proc.returncode != 0:
print("FAIL: `cmux claude-teams --version` exited non-zero")
print(f"exit={proc.returncode}")
print(f"stdout={proc.stdout.strip()}")
print(f"stderr={proc.stderr.strip()}")
return 1
tmux_pane = read_text(tmux_pane_log)
if tmux_pane != f"%{INITIAL_PANE_ID}":
print(f"FAIL: expected TMUX_PANE=%{INITIAL_PANE_ID}, got {tmux_pane!r}")
return 1
socket_value = read_text(tmux_socket_log)
if socket_value != str(socket_path):
print(f"FAIL: expected CMUX_SOCKET_PATH={socket_path}, got {socket_value!r}")
return 1
window_target = read_text(window_target_log)
if window_target != "cmux:1":
print(f"FAIL: expected tmux window target 'cmux:1', got {window_target!r}")
return 1
split_pane = read_text(split_pane_log)
if split_pane != f"%{NEW_PANE_ID}":
print(f"FAIL: expected split-window to print %{NEW_PANE_ID}, got {split_pane!r}")
return 1
pane_lines = pane_list_log.read_text(encoding="utf-8").splitlines()
expected_panes = [f"%{INITIAL_PANE_ID}", f"%{NEW_PANE_ID}"]
if pane_lines != expected_panes:
print(f"FAIL: expected list-panes output {expected_panes!r}, got {pane_lines!r}")
return 1
if state.current_pane_id != INITIAL_PANE_ID:
print(
"FAIL: expected split-window to keep the leader pane focused, "
f"got current pane {state.current_pane_id!r}"
)
return 1
if "surface.send_text" in state.requests:
print("FAIL: split-window treated '-l 70%' like shell text and called surface.send_text")
print(f"requests={state.requests!r}")
return 1
print("PASS: cmux claude-teams supports Claude's tmux teammate flow")
return 0
if __name__ == "__main__":
raise SystemExit(main())