diff --git a/ccb b/ccb index 259424b..a62e8c4 100755 --- a/ccb +++ b/ccb @@ -2,7 +2,7 @@ """ ccb (Claude Code Bridge) - 统一 AI 启动器 支持 Claude + Codex / Claude + Gemini / 三者同时 -支持 tmux 和 WezTerm 终端 +支持 tmux、WezTerm 和 iTerm2 终端 """ import sys @@ -23,7 +23,7 @@ from pathlib import Path script_dir = Path(__file__).resolve().parent sys.path.insert(0, str(script_dir / "lib")) -from terminal import TmuxBackend, WeztermBackend, detect_terminal, is_wsl, get_shell_type +from terminal import TmuxBackend, WeztermBackend, Iterm2Backend, detect_terminal, is_wsl, get_shell_type VERSION = "2.1" @@ -74,6 +74,7 @@ class AILauncher: self.terminal_type = self._detect_terminal_type() self.tmux_sessions = {} self.wezterm_panes = {} + self.iterm2_panes = {} self.processes = {} def _detect_terminal_type(self): @@ -85,6 +86,9 @@ class AILauncher: # 在 WezTerm pane 内时，强制使用 wezterm，完全不依赖 tmux if os.environ.get("WEZTERM_PANE"): return "wezterm" + # 只有在 iTerm2 环境中才用 iTerm2 分屏 + if os.environ.get("ITERM_SESSION_ID"): + return "iterm2" # 使用 detect_terminal() 自动检测（WezTerm 优先） detected = detect_terminal() @@ -120,6 +124,8 @@ class AILauncher: if self.terminal_type == "wezterm": print(f"🚀 启动 {provider.capitalize()} 后端 (wezterm)...") return self._start_provider_wezterm(provider) + elif self.terminal_type == "iterm2": + return self._start_provider_iterm2(provider) # tmux 模式：检查 tmux 是否可用 if not shutil.which("tmux"): @@ -179,6 +185,45 @@ class AILauncher: print(f"✅ {provider.capitalize()} 已启动 (wezterm pane: {pane_id})") return True + def _start_provider_iterm2(self, provider: str) -> bool: + runtime = self.runtime_dir / provider + runtime.mkdir(parents=True, exist_ok=True) + + start_cmd = self._get_start_cmd(provider) + # iTerm2 分屏里，进程退出会导致 pane 直接关闭；默认保持 pane 打开便于查看退出信息。 + keep_open = os.environ.get("CODEX_ITERM2_KEEP_OPEN", "1").lower() not in {"0", "false", "no", "off"} + if keep_open: + start_cmd = ( + f"{start_cmd}; " + f"code=$?; " + f'echo; echo \"[{provider}] exited with code $code. Press Enter to close...\"; ' + f"read -r _; " + f"exit $code" + ) + # Layout: first backend splits to the right of current pane, subsequent backends stack below + direction = "right" if not self.iterm2_panes else "bottom" + parent_pane = None + if direction == "bottom": + try: + parent_pane = next(iter(self.iterm2_panes.values())) + except StopIteration: + parent_pane = None + + backend = Iterm2Backend() + pane_id = backend.create_pane(start_cmd, str(Path.cwd()), direction=direction, percent=50, parent_pane=parent_pane) + self.iterm2_panes[provider] = pane_id + + if provider == "codex": + input_fifo = runtime / "input.fifo" + output_fifo = runtime / "output.fifo" + # iTerm2 模式通过 pane 注入文本，不强依赖 FIFO + self._write_codex_session(runtime, None, input_fifo, output_fifo, pane_id=pane_id) + else: + self._write_gemini_session(runtime, None, pane_id=pane_id) + + print(f"✅ {provider.capitalize()} 已启动 (iterm2 session: {pane_id})") + return True + def _work_dir_strings(self, work_dir: Path) -> list[str]: candidates: list[str] = [] env_pwd = os.environ.get("PWD") @@ -562,6 +607,8 @@ exec tmux attach -t "$TMUX_SESSION" env["CODEX_TERMINAL"] = self.terminal_type if self.terminal_type == "wezterm": env["CODEX_WEZTERM_PANE"] = self.wezterm_panes.get("codex", "") + elif self.terminal_type == "iterm2": + env["CODEX_ITERM2_PANE"] = self.iterm2_panes.get("codex", "") else: env["CODEX_TMUX_SESSION"] = self.tmux_sessions.get("codex", "") @@ -572,6 +619,8 @@ exec tmux attach -t "$TMUX_SESSION" env["GEMINI_TERMINAL"] = self.terminal_type if self.terminal_type == "wezterm": env["GEMINI_WEZTERM_PANE"] = self.wezterm_panes.get("gemini", "") + elif self.terminal_type == "iterm2": + env["GEMINI_ITERM2_PANE"] = self.iterm2_panes.get("gemini", "") else: env["GEMINI_TMUX_SESSION"] = self.tmux_sessions.get("gemini", "") @@ -624,6 +673,11 @@ exec tmux attach -t "$TMUX_SESSION" for provider, pane_id in self.wezterm_panes.items(): if pane_id: backend.kill_pane(pane_id) + elif self.terminal_type == "iterm2": + backend = Iterm2Backend() + for provider, pane_id in self.iterm2_panes.items(): + if pane_id: + backend.kill_pane(pane_id) else: for provider, tmux_session in self.tmux_sessions.items(): subprocess.run(["tmux", "kill-session", "-t", tmux_session], stderr=subprocess.DEVNULL) @@ -658,7 +712,7 @@ exec tmux attach -t "$TMUX_SESSION" signal.signal(signal.SIGTERM, lambda s, f: (self.cleanup(), sys.exit(0))) providers = list(self.providers) - if self.terminal_type == "wezterm": + if self.terminal_type in ("wezterm", "iterm2"): # Stable layout: codex on top, gemini on bottom (when both are present). order = {"codex": 0, "gemini": 1} providers.sort(key=lambda p: order.get(p, 99)) @@ -679,6 +733,10 @@ exec tmux attach -t "$TMUX_SESSION" pane = self.wezterm_panes.get(provider, "") if pane: print(f" {provider}: wezterm cli activate-pane --pane-id {pane}") + elif self.terminal_type == "iterm2": + pane = self.iterm2_panes.get(provider, "") + if pane: + print(f" {provider}: it2 session focus {pane}") else: tmux = self.tmux_sessions.get(provider, "") if tmux: @@ -717,12 +775,15 @@ def cmd_status(args): try: data = json.loads(session_file.read_text()) terminal = data.get("terminal", "tmux") - pane_id = data.get("pane_id") if terminal == "wezterm" else data.get("tmux_session", "") + pane_id = data.get("pane_id") if terminal in ("wezterm", "iterm2") else data.get("tmux_session", "") active = data.get("active", False) if terminal == "wezterm" and pane_id: backend = WeztermBackend() alive = backend.is_alive(pane_id) + elif terminal == "iterm2" and pane_id: + backend = Iterm2Backend() + alive = backend.is_alive(pane_id) elif pane_id and shutil.which("tmux"): result = subprocess.run(["tmux", "has-session", "-t", pane_id], capture_output=True) alive = result.returncode == 0 @@ -761,11 +822,14 @@ def cmd_kill(args): try: data = json.loads(session_file.read_text()) terminal = data.get("terminal", "tmux") - pane_id = data.get("pane_id") if terminal == "wezterm" else data.get("tmux_session", "") + pane_id = data.get("pane_id") if terminal in ("wezterm", "iterm2") else data.get("tmux_session", "") if terminal == "wezterm" and pane_id: backend = WeztermBackend() backend.kill_pane(pane_id) + elif terminal == "iterm2" and pane_id: + backend = Iterm2Backend() + backend.kill_pane(pane_id) elif pane_id and shutil.which("tmux"): subprocess.run(["tmux", "kill-session", "-t", pane_id], stderr=subprocess.DEVNULL) subprocess.run(["tmux", "kill-session", "-t", f"launcher-{pane_id}"], stderr=subprocess.DEVNULL) @@ -793,7 +857,7 @@ def cmd_restore(args): try: data = json.loads(session_file.read_text()) terminal = data.get("terminal", "tmux") - pane_id = data.get("pane_id") if terminal == "wezterm" else data.get("tmux_session", "") + pane_id = data.get("pane_id") if terminal in ("wezterm", "iterm2") else data.get("tmux_session", "") active = data.get("active", False) if terminal == "wezterm" and pane_id: @@ -801,6 +865,11 @@ def cmd_restore(args): if backend.is_alive(pane_id): backend.activate(pane_id) return 0 + elif terminal == "iterm2" and pane_id: + backend = Iterm2Backend() + if backend.is_alive(pane_id): + backend.activate(pane_id) + return 0 elif pane_id and shutil.which("tmux"): result = subprocess.run(["tmux", "has-session", "-t", pane_id], capture_output=True) if result.returncode == 0: diff --git a/install.sh b/install.sh index 1c58457..0900572 100755 --- a/install.sh +++ b/install.sh @@ -170,19 +170,138 @@ print_tmux_install_hint() { esac } +# 检测是否在 iTerm2 环境中运行 +is_iterm2_environment() { + # 检查 ITERM_SESSION_ID 环境变量 + if [[ -n "${ITERM_SESSION_ID:-}" ]]; then + return 0 + fi + # 检查 TERM_PROGRAM + if [[ "${TERM_PROGRAM:-}" == "iTerm.app" ]]; then + return 0 + fi + # macOS 上检查 iTerm2 是否正在运行 + if [[ "$(uname)" == "Darwin" ]] && pgrep -x "iTerm2" >/dev/null 2>&1; then + return 0 + fi + return 1 +} + +# 安装 it2 CLI +install_it2() { + echo + echo "📦 正在安装 it2 CLI..." + + # 检查 pip3 是否可用 + if ! command -v pip3 >/dev/null 2>&1; then + echo "❌ 未找到 pip3，无法自动安装 it2" + echo " 请手动运行: python3 -m pip install it2" + return 1 + fi + + # 安装 it2 + if pip3 install it2 --user 2>&1; then + echo "✅ it2 CLI 安装成功" + + # 检查是否在 PATH 中 + if ! command -v it2 >/dev/null 2>&1; then + local user_bin + user_bin="$(python3 -m site --user-base)/bin" + echo + echo "⚠️ it2 可能不在 PATH 中，请添加以下路径到你的 shell 配置文件：" + echo " export PATH=\"$user_bin:\$PATH\"" + fi + return 0 + else + echo "❌ it2 安装失败" + return 1 + fi +} + +# 显示 iTerm2 Python API 启用提示 +show_iterm2_api_reminder() { + echo + echo "================================================================" + echo "🔔 重要提示：请在 iTerm2 中启用 Python API" + echo "================================================================" + echo " 步骤：" + echo " 1. 打开 iTerm2" + echo " 2. 进入 Preferences (⌘ + ,)" + echo " 3. 选择 Magic 标签页" + echo " 4. 勾选 \"Enable Python API\"" + echo " 5. 确认警告对话框" + echo "================================================================" + echo +} + require_terminal_backend() { - # 检测 WezTerm（优先） local wezterm_override="${CODEX_WEZTERM_BIN:-${WEZTERM_BIN:-}}" + + # ============================================ + # 优先检测当前运行环境，确保使用正确的终端工具 + # ============================================ + + # 1. 如果在 WezTerm 环境中运行 + if [[ -n "${WEZTERM_PANE:-}" ]]; then + if [[ -n "${wezterm_override}" ]] && { command -v "${wezterm_override}" >/dev/null 2>&1 || [[ -f "${wezterm_override}" ]]; }; then + echo "✓ 检测到 WezTerm 环境 (${wezterm_override})" + return + fi + if command -v wezterm >/dev/null 2>&1 || command -v wezterm.exe >/dev/null 2>&1; then + echo "✓ 检测到 WezTerm 环境" + return + fi + fi + + # 2. 如果在 iTerm2 环境中运行 + if is_iterm2_environment; then + # 检查是否已安装 it2 + if command -v it2 >/dev/null 2>&1; then + echo "✓ 检测到 iTerm2 环境 (it2 CLI 已安装)" + echo " 💡 请确保已启用 iTerm2 Python API (Preferences > Magic > Enable Python API)" + return + fi + + # 未安装 it2，询问是否安装 + echo "🍎 检测到 iTerm2 环境，但未安装 it2 CLI" + echo + read -p "是否自动安装 it2 CLI？(Y/n): " -n 1 -r + echo + + if [[ ! $REPLY =~ ^[Nn]$ ]]; then + if install_it2; then + show_iterm2_api_reminder + return + fi + else + echo "跳过 it2 安装，将使用 tmux 作为后备方案" + fi + fi + + # 3. 如果在 tmux 环境中运行 + if [[ -n "${TMUX:-}" ]]; then + echo "✓ 检测到 tmux 环境" + return + fi + + # ============================================ + # 不在特定环境中，按可用性检测 + # ============================================ + + # 4. 检查 WezTerm 环境变量覆盖 if [[ -n "${wezterm_override}" ]]; then if command -v "${wezterm_override}" >/dev/null 2>&1 || [[ -f "${wezterm_override}" ]]; then echo "✓ 检测到 WezTerm (${wezterm_override})" return fi fi + + # 5. 检查 WezTerm 命令 if command -v wezterm >/dev/null 2>&1 || command -v wezterm.exe >/dev/null 2>&1; then echo "✓ 检测到 WezTerm" return fi + # WSL 场景：Windows PATH 可能未注入 WSL，尝试常见安装路径 if [[ -f "/proc/version" ]] && grep -qi microsoft /proc/version 2>/dev/null; then if [[ -x "/mnt/c/Program Files/WezTerm/wezterm.exe" ]] || [[ -f "/mnt/c/Program Files/WezTerm/wezterm.exe" ]]; then @@ -194,13 +313,31 @@ require_terminal_backend() { return fi fi - # 检测 tmux（备选） + + # 6. 检查 it2 CLI + if command -v it2 >/dev/null 2>&1; then + echo "✓ 检测到 it2 CLI" + return + fi + + # 7. 检查 tmux if command -v tmux >/dev/null 2>&1; then echo "✓ 检测到 tmux（建议同时安装 WezTerm 以获得更好体验）" return fi - echo "❌ 缺少依赖: WezTerm 或 tmux（推荐 WezTerm）" + + # 8. 没有找到任何可用的终端复用器 + echo "❌ 缺少依赖: WezTerm、tmux 或 it2 (至少需要安装其中一个)" echo " WezTerm 官网: https://wezfurlong.org/wezterm/" + + # macOS 上额外提示 iTerm2 + it2 选项 + if [[ "$(uname)" == "Darwin" ]]; then + echo + echo "💡 macOS 用户推荐选项：" + echo " - 如果你使用 iTerm2，可以安装 it2 CLI: pip3 install it2" + echo " - 或者安装 tmux: brew install tmux" + fi + print_tmux_install_hint exit 1 } @@ -410,7 +547,7 @@ with open('$claude_md', 'w', encoding='utf-8') as f: cat >> "$claude_md" << 'AI_RULES' ## Codex Collaboration Rules -Codex is another AI assistant running in a separate terminal session (WezTerm or tmux). When user intent involves asking/consulting/collaborating with Codex: +Codex is another AI assistant running in a separate terminal session (WezTerm, iTerm2 or tmux). When user intent involves asking/consulting/collaborating with Codex: Fast path (minimize latency): - If the user message starts with any of: `调取codex`, `@codex`, `codex:`, `codex：`, `问codex`, `让codex` then immediately run: @@ -435,7 +572,7 @@ Examples: - "don't wait for reply" → cask ## Gemini Collaboration Rules -Gemini is another AI assistant running in a separate terminal session (WezTerm or tmux). When user intent involves asking/consulting/collaborating with Gemini: +Gemini is another AI assistant running in a separate terminal session (WezTerm, iTerm2 or tmux). When user intent involves asking/consulting/collaborating with Gemini: Fast path (minimize latency): - If the user message starts with any of: `调取gemini`, `@gemini`, `gemini:`, `gemini：`, `问gemini`, `让gemini` then immediately run: @@ -562,20 +699,136 @@ install_all() { echo " 全局 settings.json 已添加权限" } +uninstall_claude_md_config() { + local claude_md="$HOME/.claude/CLAUDE.md" + + if [[ ! -f "$claude_md" ]]; then + return + fi + + if grep -qE "$RULE_MARKER|$LEGACY_RULE_MARKER|## Gemini" "$claude_md" 2>/dev/null; then + echo "正在移除 CLAUDE.md 中的协作规则..." + if command -v python3 >/dev/null 2>&1; then + python3 -c " +import re +with open('$claude_md', 'r', encoding='utf-8') as f: + content = f.read() +# Remove all collaboration rule sections +patterns = [ + r'## Codex Collaboration Rules.*?(?=\n## |\Z)', + r'## Codex 协作规则.*?(?=\n## |\Z)', + r'## Gemini Collaboration Rules.*?(?=\n## |\Z)', + r'## Gemini 协作规则.*?(?=\n## |\Z)', +] +for p in patterns: + content = re.sub(p, '', content, flags=re.DOTALL) +content = content.rstrip() + '\n' +with open('$claude_md', 'w', encoding='utf-8') as f: + f.write(content) +" + echo "已移除 CLAUDE.md 中的协作规则" + else + echo "⚠️ 需要 python3 来清理 CLAUDE.md，请手动移除协作规则部分" + fi + fi +} + +uninstall_settings_permissions() { + local settings_file="$HOME/.claude/settings.json" + + if [[ ! -f "$settings_file" ]]; then + return + fi + + local perms_to_remove=( + 'Bash(cask:*)' + 'Bash(cask-w:*)' + 'Bash(cpend)' + 'Bash(cping)' + 'Bash(gask:*)' + 'Bash(gask-w:*)' + 'Bash(gpend)' + 'Bash(gping)' + ) + + if command -v python3 >/dev/null 2>&1; then + local has_perms=0 + for perm in "${perms_to_remove[@]}"; do + if grep -q "$perm" "$settings_file" 2>/dev/null; then + has_perms=1 + break + fi + done + + if [[ $has_perms -eq 1 ]]; then + echo "正在移除 settings.json 中的权限配置..." + python3 -c " +import json +perms_to_remove = [ + 'Bash(cask:*)', + 'Bash(cask-w:*)', + 'Bash(cpend)', + 'Bash(cping)', + 'Bash(gask:*)', + 'Bash(gask-w:*)', + 'Bash(gpend)', + 'Bash(gping)', +] +with open('$settings_file', 'r') as f: + data = json.load(f) +if 'permissions' in data and 'allow' in data['permissions']: + data['permissions']['allow'] = [ + p for p in data['permissions']['allow'] + if p not in perms_to_remove + ] +with open('$settings_file', 'w') as f: + json.dump(data, f, indent=2) +" + echo "已移除 settings.json 中的权限配置" + fi + else + echo "⚠️ 需要 python3 来清理 settings.json，请手动移除相关权限" + fi +} + uninstall_all() { - rm -rf "$INSTALL_PREFIX" - for name in "${SCRIPTS_TO_LINK[@]}"; do - rm -f "$BIN_DIR/$name" + echo "🧹 开始卸载 ccb..." + + # 1. 移除项目目录 + if [[ -d "$INSTALL_PREFIX" ]]; then + rm -rf "$INSTALL_PREFIX" + echo "已移除项目目录: $INSTALL_PREFIX" + fi + + # 2. 移除 bin 链接 + for path in "${SCRIPTS_TO_LINK[@]}"; do + local name + name="$(basename "$path")" + if [[ -L "$BIN_DIR/$name" || -f "$BIN_DIR/$name" ]]; then + rm -f "$BIN_DIR/$name" + fi done for legacy in "${LEGACY_SCRIPTS[@]}"; do rm -f "$BIN_DIR/$legacy" done + echo "已移除 bin 链接: $BIN_DIR" + + # 3. 移除 Claude 命令文件 local claude_dir claude_dir="$(detect_claude_dir)" for doc in "${CLAUDE_MARKDOWN[@]}"; do rm -f "$claude_dir/$doc" done + echo "已移除 Claude 命令: $claude_dir" + + # 4. 移除 CLAUDE.md 中的协作规则 + uninstall_claude_md_config + + # 5. 移除 settings.json 中的权限配置 + uninstall_settings_permissions + echo "✅ 卸载完成" + echo " 💡 注意: 依赖项 (python3, tmux, wezterm, it2) 未被移除" } main() { diff --git a/lib/codex_comm.py b/lib/codex_comm.py index 02b9f1b..89237d9 100644 --- a/lib/codex_comm.py +++ b/lib/codex_comm.py @@ -260,14 +260,22 @@ def __init__(self): def _load_session_info(self): if "CODEX_SESSION_ID" in os.environ: + terminal = os.environ.get("CODEX_TERMINAL", "tmux") + # 根据终端类型获取正确的 pane_id + if terminal == "wezterm": + pane_id = os.environ.get("CODEX_WEZTERM_PANE", "") + elif terminal == "iterm2": + pane_id = os.environ.get("CODEX_ITERM2_PANE", "") + else: + pane_id = "" return { "session_id": os.environ["CODEX_SESSION_ID"], "runtime_dir": os.environ["CODEX_RUNTIME_DIR"], "input_fifo": os.environ["CODEX_INPUT_FIFO"], "output_fifo": os.environ.get("CODEX_OUTPUT_FIFO", ""), - "terminal": os.environ.get("CODEX_TERMINAL", "tmux"), + "terminal": terminal, "tmux_session": os.environ.get("CODEX_TMUX_SESSION", ""), - "pane_id": os.environ.get("CODEX_WEZTERM_PANE", ""), + "pane_id": pane_id, "_session_file": None, } @@ -310,13 +318,13 @@ def _check_session_health_impl(self, probe_terminal: bool): if not self.runtime_dir.exists(): return False, "运行时目录不存在" - # WezTerm 模式：没有 tmux wrapper，因此通常不会生成 codex.pid； + # WezTerm/iTerm2 模式：没有 tmux wrapper，因此通常不会生成 codex.pid； # 以 pane 存活作为健康判定（与 Gemini 逻辑一致）。 - if self.terminal == "wezterm": + if self.terminal in ("wezterm", "iterm2"): if not self.pane_id: - return False, "未找到 WezTerm pane_id" + return False, f"未找到 {self.terminal} pane_id" if probe_terminal and (not self.backend or not self.backend.is_alive(self.pane_id)): - return False, f"WezTerm pane 不存在: {self.pane_id}" + return False, f"{self.terminal} pane 不存在: {self.pane_id}" return True, "会话正常" # tmux 模式：依赖 wrapper 写入 codex.pid 与 FIFO @@ -353,8 +361,8 @@ def _send_message(self, content: str) -> Tuple[str, Dict[str, Any]]: state = self.log_reader.capture_state() - # tmux 模式优先通过 FIFO 驱动桥接器；WezTerm 模式则直接向 pane 注入文本 - if self.terminal == "wezterm": + # tmux 模式优先通过 FIFO 驱动桥接器；WezTerm/iTerm2 模式则直接向 pane 注入文本 + if self.terminal in ("wezterm", "iterm2"): self._send_via_terminal(content) else: with open(self.input_fifo, "w", encoding="utf-8") as fifo: diff --git a/lib/gemini_comm.py b/lib/gemini_comm.py index 00367b3..305cfb9 100755 --- a/lib/gemini_comm.py +++ b/lib/gemini_comm.py @@ -372,12 +372,20 @@ def _prime_log_binding(self) -> None: def _load_session_info(self): if "GEMINI_SESSION_ID" in os.environ: + terminal = os.environ.get("GEMINI_TERMINAL", "tmux") + # 根据终端类型获取正确的 pane_id + if terminal == "wezterm": + pane_id = os.environ.get("GEMINI_WEZTERM_PANE", "") + elif terminal == "iterm2": + pane_id = os.environ.get("GEMINI_ITERM2_PANE", "") + else: + pane_id = "" return { "session_id": os.environ["GEMINI_SESSION_ID"], "runtime_dir": os.environ["GEMINI_RUNTIME_DIR"], - "terminal": os.environ.get("GEMINI_TERMINAL", "tmux"), + "terminal": terminal, "tmux_session": os.environ.get("GEMINI_TMUX_SESSION", ""), - "pane_id": os.environ.get("GEMINI_WEZTERM_PANE", ""), + "pane_id": pane_id, "_session_file": None, } diff --git a/lib/terminal.py b/lib/terminal.py index e9c17c2..5068d6d 100644 --- a/lib/terminal.py +++ b/lib/terminal.py @@ -153,6 +153,98 @@ def create_pane(self, cmd: str, cwd: str, direction: str = "right", percent: int return session_name +class Iterm2Backend(TerminalBackend): + """iTerm2 后端，使用 it2 CLI (pip install it2)""" + _it2_bin: Optional[str] = None + + @classmethod + def _bin(cls) -> str: + if cls._it2_bin: + return cls._it2_bin + override = os.environ.get("CODEX_IT2_BIN") or os.environ.get("IT2_BIN") + if override: + cls._it2_bin = override + return override + cls._it2_bin = shutil.which("it2") or "it2" + return cls._it2_bin + + def send_text(self, session_id: str, text: str) -> None: + sanitized = text.replace("\r", "").strip() + if not sanitized: + return + # 类似 WezTerm 的方式：先发送文本，再发送回车 + # it2 session send 发送文本（不带换行） + subprocess.run( + [self._bin(), "session", "send", sanitized, "--session", session_id], + check=True, + ) + # 等待一点时间，让 TUI 处理输入 + time.sleep(0.01) + # 发送回车键（使用 \r） + subprocess.run( + [self._bin(), "session", "send", "\r", "--session", session_id], + check=True, + ) + + def is_alive(self, session_id: str) -> bool: + try: + result = subprocess.run( + [self._bin(), "session", "list", "--json"], + capture_output=True, text=True + ) + if result.returncode != 0: + return False + sessions = json.loads(result.stdout) + return any(s.get("id") == session_id for s in sessions) + except Exception: + return False + + def kill_pane(self, session_id: str) -> None: + subprocess.run( + [self._bin(), "session", "close", "--session", session_id, "--force"], + stderr=subprocess.DEVNULL + ) + + def activate(self, session_id: str) -> None: + subprocess.run([self._bin(), "session", "focus", session_id]) + + def create_pane(self, cmd: str, cwd: str, direction: str = "right", percent: int = 50, parent_pane: Optional[str] = None) -> str: + # iTerm2 分屏：vertical 对应 right，horizontal 对应 bottom + args = [self._bin(), "session", "split"] + if direction == "right": + args.append("--vertical") + # 如果有 parent_pane，指定目标 session + if parent_pane: + args.extend(["--session", parent_pane]) + + result = subprocess.run(args, capture_output=True, text=True, check=True) + # it2 输出格式: "Created new pane: " + output = result.stdout.strip() + if ":" in output: + new_session_id = output.split(":")[-1].strip() + else: + # 尝试从 stderr 或其他地方获取 + new_session_id = output + + # 在新 pane 中执行启动命令 + if new_session_id and cmd: + # 先 cd 到工作目录，再执行命令 + full_cmd = f"cd {shlex.quote(cwd)} && {cmd}" + time.sleep(0.2) # 等待 pane 就绪 + # 使用 send + 回车的方式，与 send_text 保持一致 + subprocess.run( + [self._bin(), "session", "send", full_cmd, "--session", new_session_id], + check=True + ) + time.sleep(0.01) + subprocess.run( + [self._bin(), "session", "send", "\r", "--session", new_session_id], + check=True + ) + + return new_session_id + + class WeztermBackend(TerminalBackend): _wezterm_bin: Optional[str] = None @@ -263,12 +355,22 @@ def create_pane(self, cmd: str, cwd: str, direction: str = "right", percent: int def detect_terminal() -> Optional[str]: + # 优先检测当前环境变量（已在某终端中运行） if os.environ.get("WEZTERM_PANE"): return "wezterm" + if os.environ.get("ITERM_SESSION_ID"): + return "iterm2" if os.environ.get("TMUX"): return "tmux" + # 检查配置的二进制覆盖或缓存路径 if _get_wezterm_bin(): return "wezterm" + override = os.environ.get("CODEX_IT2_BIN") or os.environ.get("IT2_BIN") + if override and Path(override).expanduser().exists(): + return "iterm2" + # 检查可用的终端工具 + if shutil.which("it2"): + return "iterm2" if shutil.which("tmux") or shutil.which("tmux.exe"): return "tmux" return None @@ -281,6 +383,8 @@ def get_backend(terminal_type: Optional[str] = None) -> Optional[TerminalBackend t = terminal_type or detect_terminal() if t == "wezterm": _backend_cache = WeztermBackend() + elif t == "iterm2": + _backend_cache = Iterm2Backend() elif t == "tmux": _backend_cache = TmuxBackend() return _backend_cache @@ -290,6 +394,8 @@ def get_backend_for_session(session_data: dict) -> Optional[TerminalBackend]: terminal = session_data.get("terminal", "tmux") if terminal == "wezterm": return WeztermBackend() + elif terminal == "iterm2": + return Iterm2Backend() return TmuxBackend() @@ -297,4 +403,6 @@ def get_pane_id_from_session(session_data: dict) -> Optional[str]: terminal = session_data.get("terminal", "tmux") if terminal == "wezterm": return session_data.get("pane_id") + elif terminal == "iterm2": + return session_data.get("pane_id") return session_data.get("tmux_session")