diff --git a/README.md b/README.md index f259f8f1..86a85c51 100644 --- a/README.md +++ b/README.md @@ -407,6 +407,7 @@ tmux tip: CCB's tmux status/pane theming is enabled only while CCB is running. tmux tip: press `Ctrl+b` then `Space` to cycle tmux layouts. You can press it repeatedly to keep switching layouts. Layout rule: the last provider runs in the current pane. Extras are ordered as `[cmd?, reversed providers]`; the first extra goes to the top-right, then the left column fills top-to-bottom, then the right column fills top-to-bottom. Examples: 4 panes = left2/right2, 5 panes = left2/right3. +Windows mode: use `ccb -w` or set `"layout": "windows"` in config. Each provider gets its own tmux window. Use `Ctrl+B w` to list and switch windows. The `cmd` (shell) pane stays in the anchor window. Windows mode is tmux-only (not supported with WezTerm). Note: `ccb up` is removed; use `ccb ...` or configure `ccb.config`. ``` @@ -415,6 +416,7 @@ Note: `ccb up` is removed; use `ccb ...` or configure `ccb.config`. | :--- | :--- | :--- | | `-r` | Resume previous session context | `ccb -r` | | `-a` | Auto-mode, skip permission prompts | `ccb -a` | +| `-w, --windows` | Launch each provider in its own tmux window instead of split panes | `ccb -w codex gemini` | | `-h` | Show help information | `ccb -h` | | `-v` | Show version and check for updates | `ccb -v` | @@ -437,12 +439,15 @@ Advanced JSON (optional, for flags or custom cmd pane): ```json { "providers": ["codex", "gemini", "opencode", "claude"], + "layout": "windows", "cmd": { "enabled": true, "title": "CCB-Cmd", "start_cmd": "bash" }, "flags": { "auto": false, "resume": false } } ``` Cmd pane participates in the layout as the first extra pane and does not change which AI runs in the current pane. +`layout` accepts `"panes"` (default, split panes) or `"windows"` (one tmux window per provider). CLI flag `--windows` overrides the config value. + ### Update ```bash ccb update # Update ccb to the latest version diff --git a/README_zh.md b/README_zh.md index 8c0fce81..1c2a97c7 100644 --- a/README_zh.md +++ b/README_zh.md @@ -363,6 +363,7 @@ tmux 提示:CCB 的 tmux 状态栏/窗格标题主题只会在 CCB 运行期 tmux 提示:在 tmux 内可以按 `Ctrl+b` 然后按 `Space` 来切换布局;可以连续按,多次循环切换不同布局。 布局规则:当前 pane 对应 providers 列表的最后一个。额外 pane 顺序为 `[cmd?, providers 反序]`;第一个额外 pane 在右上,其后先填满左列(从上到下),再填右列(从上到下)。例:4 个 pane 左2右2,5 个 pane 左2右3。 +窗口模式:使用 `ccb -w` 或在配置中设置 `"layout": "windows"`,每个 provider 占一个独立 tmux 窗口。使用 `Ctrl+B w` 列出并切换窗口。`cmd`(shell)pane 保留在锚定窗口中。窗口模式仅支持 tmux(不支持 WezTerm)。 提示:`ccb up` 已移除,请使用 `ccb ...` 或配置 `ccb.config`。 ``` @@ -371,6 +372,7 @@ tmux 提示:在 tmux 内可以按 `Ctrl+b` 然后按 `Space` 来切换布局 | :--- | :--- | :--- | | `-r` | 恢复上次会话上下文 | `ccb -r` | | `-a` | 全自动模式,跳过权限确认 | `ccb -a` | +| `-w, --windows` | 每个 provider 使用独立的 tmux 窗口而非分屏 | `ccb -w codex gemini` | | `-h` | 查看详细帮助信息 | `ccb -h` | | `-v` | 查看当前版本和检测更新 | `ccb -v` | @@ -389,6 +391,16 @@ codex,gemini,opencode,claude codex,gemini,opencode,claude,cmd ``` +高级 JSON 格式(可选,用于设置布局或自定义 cmd pane): +```json +{ + "providers": ["codex", "gemini"], + "layout": "windows" +} +``` + +`layout` 支持 `"panes"`(默认,分屏模式)或 `"windows"`(每个 provider 独立 tmux 窗口)。CLI 参数 `--windows` 会覆盖配置文件中的值。 + cmd pane 作为第一个额外 pane 参与布局,不会改变当前 pane 对应的 AI。 ### 后续更新 diff --git a/ccb b/ccb index 90d028c2..b8f59abb 100755 --- a/ccb +++ b/ccb @@ -38,7 +38,8 @@ from session_utils import ( resolve_project_config_dir, safe_write_session, ) -from pane_registry import upsert_registry, load_registry_by_project_id +from pane_registry import upsert_registry, load_registry_by_project_id, get_layout_mode +from layout import PanesLayout, WindowsLayout from project_id import compute_ccb_project_id from providers import CASK_CLIENT_SPEC, GASK_CLIENT_SPEC, OASK_CLIENT_SPEC, LASK_CLIENT_SPEC, DASK_CLIENT_SPEC from process_lock import ProviderLock @@ -555,11 +556,13 @@ class AILauncher: resume: bool = False, auto: bool = False, cmd_config: dict | None = None, + layout_mode: str = "panes", ): self.providers = providers or ["codex"] self.resume = resume self.auto = auto self.cmd_config = self._normalize_cmd_config(cmd_config) + self.layout_mode = layout_mode self.script_dir = Path(__file__).resolve().parent self.invocation_dir = Path.cwd() # Project root is strictly the current working directory. @@ -587,6 +590,7 @@ class AILauncher: self.tmux_panes = {} self.wezterm_panes = {} self.extra_panes = {} + self.layout_strategy = None # set in run_up() self.processes = {} self.anchor_provider = None self.anchor_pane_id = None @@ -1443,23 +1447,17 @@ class AILauncher: return if pane_id: try: - upsert_registry( - { - "ccb_session_id": self.session_id, - "ccb_project_id": compute_ccb_project_id(self.project_root), - "work_dir": str(self.project_root), - "terminal": terminal or self.terminal_type, - "providers": { - "claude": { - "pane_id": pane_id, - "pane_title_marker": pane_title_marker, - "session_file": str(path), - "claude_session_id": data.get("claude_session_id"), - "claude_session_path": data.get("claude_session_path"), - } - }, + _reg = self._registry_base(terminal=terminal or self.terminal_type) + _reg["providers"] = { + "claude": { + "pane_id": pane_id, + "pane_title_marker": pane_title_marker, + "session_file": str(path), + "claude_session_id": data.get("claude_session_id"), + "claude_session_path": data.get("claude_session_path"), } - ) + } + upsert_registry(_reg) except Exception: pass self._maybe_start_provider_daemon("claude") @@ -2132,7 +2130,7 @@ class AILauncher: except Exception: use_parent = None - pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent) + pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent, layout_mode=self.layout_mode) backend.respawn_pane(pane_id, cmd=start_cmd, cwd=str(Path.cwd()), remain_on_exit=True) backend.set_pane_title(pane_id, pane_title_marker) backend.set_pane_user_option(pane_id, "@ccb_agent", "Codex") @@ -2223,7 +2221,7 @@ class AILauncher: except Exception: use_parent = None - pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent) + pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent, layout_mode=self.layout_mode) backend.respawn_pane(pane_id, cmd=start_cmd, cwd=str(Path.cwd()), remain_on_exit=True) backend.set_pane_title(pane_id, pane_title_marker) backend.set_pane_user_option(pane_id, "@ccb_agent", "Gemini") @@ -2277,7 +2275,7 @@ class AILauncher: except Exception: use_parent = None - pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent) + pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent, layout_mode=self.layout_mode) backend.respawn_pane(pane_id, cmd=start_cmd, cwd=str(Path.cwd()), remain_on_exit=True) backend.set_pane_title(pane_id, pane_title_marker) backend.set_pane_user_option(pane_id, "@ccb_agent", "OpenCode") @@ -2329,7 +2327,7 @@ class AILauncher: except Exception: use_parent = None - pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent) + pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent, layout_mode=self.layout_mode) backend.respawn_pane(pane_id, cmd=start_cmd, cwd=str(Path.cwd()), remain_on_exit=True) backend.set_pane_title(pane_id, pane_title_marker) backend.set_pane_user_option(pane_id, "@ccb_agent", "Droid") @@ -2375,7 +2373,7 @@ class AILauncher: self.extra_panes["cmd"] = pane_id else: backend = TmuxBackend() - pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent) + pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent, layout_mode="panes") backend.respawn_pane(pane_id, cmd=full_cmd, cwd=str(Path.cwd()), remain_on_exit=True) backend.set_pane_title(pane_id, title) backend.set_pane_user_option(pane_id, "@ccb_agent", "Cmd") @@ -2645,6 +2643,19 @@ class AILauncher: print(f"❌ {t('unknown_provider', provider=provider)}") return 1 + def _registry_base(self, *, terminal: str | None = None, work_dir: str | None = None) -> dict: + """Return common registry fields shared by all provider upserts.""" + base = { + "ccb_session_id": self.session_id, + "ccb_project_id": compute_ccb_project_id(self.project_root), + "work_dir": work_dir if work_dir is not None else str(self.project_root), + "terminal": terminal if terminal is not None else self.terminal_type, + "layout_mode": self.layout_mode, + } + if self.layout_strategy is not None and self.layout_strategy.linked_sessions: + base["linked_sessions"] = self.layout_strategy.linked_sessions + return base + def _write_codex_session(self, runtime, tmux_session, input_fifo, output_fifo, pane_id=None, pane_title_marker=None, codex_start_cmd=None): session_file = self._project_session_file(".codex-session") @@ -2689,19 +2700,15 @@ class AILauncher: print(err, file=sys.stderr) return False try: - upsert_registry({ - "ccb_session_id": self.session_id, - "ccb_project_id": compute_ccb_project_id(self.project_root), - "work_dir": str(self.project_root), - "terminal": self.terminal_type, - "providers": { - "codex": { - "pane_id": pane_id, - "pane_title_marker": pane_title_marker, - "session_file": str(session_file), - } - }, - }) + _reg = self._registry_base() + _reg["providers"] = { + "codex": { + "pane_id": pane_id, + "pane_title_marker": pane_title_marker, + "session_file": str(session_file), + } + } + upsert_registry(_reg) except Exception: pass self._maybe_start_provider_daemon("codex") @@ -2710,17 +2717,12 @@ class AILauncher: def _write_cend_registry(self, claude_pane_id: str, codex_pane_id: str | None) -> bool: if not claude_pane_id: return False - record = { - "ccb_session_id": self.session_id, - "claude_pane_id": claude_pane_id, - "codex_pane_id": codex_pane_id, - "ccb_project_id": compute_ccb_project_id(self.project_root), - "work_dir": str(Path.cwd()), - "terminal": self.terminal_type, - "providers": { - "claude": {"pane_id": claude_pane_id}, - "codex": {"pane_id": codex_pane_id}, - }, + record = self._registry_base(work_dir=str(Path.cwd())) + record["claude_pane_id"] = claude_pane_id + record["codex_pane_id"] = codex_pane_id + record["providers"] = { + "claude": {"pane_id": claude_pane_id}, + "codex": {"pane_id": codex_pane_id}, } ok = upsert_registry(record) if not ok: @@ -2766,19 +2768,15 @@ class AILauncher: print(err, file=sys.stderr) return False try: - upsert_registry({ - "ccb_session_id": self.session_id, - "ccb_project_id": compute_ccb_project_id(self.project_root), - "work_dir": str(self.project_root), - "terminal": self.terminal_type, - "providers": { - "gemini": { - "pane_id": pane_id, - "pane_title_marker": pane_title_marker, - "session_file": str(session_file), - } - }, - }) + _reg = self._registry_base() + _reg["providers"] = { + "gemini": { + "pane_id": pane_id, + "pane_title_marker": pane_title_marker, + "session_file": str(session_file), + } + } + upsert_registry(_reg) except Exception: pass self._maybe_start_provider_daemon("gemini") @@ -2815,19 +2813,15 @@ class AILauncher: print(err, file=sys.stderr) return False try: - upsert_registry({ - "ccb_session_id": self.session_id, - "ccb_project_id": compute_ccb_project_id(self.project_root), - "work_dir": str(self.project_root), - "terminal": self.terminal_type, - "providers": { - "opencode": { - "pane_id": pane_id, - "pane_title_marker": pane_title_marker, - "session_file": str(session_file), - } - }, - }) + _reg = self._registry_base() + _reg["providers"] = { + "opencode": { + "pane_id": pane_id, + "pane_title_marker": pane_title_marker, + "session_file": str(session_file), + } + } + upsert_registry(_reg) except Exception: pass self._maybe_start_provider_daemon("opencode") @@ -2881,21 +2875,17 @@ class AILauncher: print(err, file=sys.stderr) return False try: - upsert_registry({ - "ccb_session_id": self.session_id, - "ccb_project_id": compute_ccb_project_id(self.project_root), - "work_dir": str(self.project_root), - "terminal": self.terminal_type, - "providers": { - "droid": { - "pane_id": pane_id, - "pane_title_marker": pane_title_marker, - "session_file": str(session_file), - "droid_session_id": droid_session_id, - "droid_session_path": droid_session_path, - } - }, - }) + _reg = self._registry_base() + _reg["providers"] = { + "droid": { + "pane_id": pane_id, + "pane_title_marker": pane_title_marker, + "session_file": str(session_file), + "droid_session_id": droid_session_id, + "droid_session_path": droid_session_path, + } + } + upsert_registry(_reg) except Exception: pass self._maybe_start_provider_daemon("droid") @@ -3165,7 +3155,7 @@ class AILauncher: self.wezterm_panes["claude"] = pane_id else: backend = TmuxBackend() - pane_id = backend.create_pane("", run_cwd, direction=use_direction, percent=50, parent_pane=use_parent) + pane_id = backend.create_pane("", run_cwd, direction=use_direction, percent=50, parent_pane=use_parent, layout_mode=self.layout_mode) backend.respawn_pane(pane_id, cmd=full_cmd, cwd=run_cwd, remain_on_exit=True) backend.set_pane_title(pane_id, "CCB-Claude") backend.set_pane_user_option(pane_id, "@ccb_agent", "Claude") @@ -3227,6 +3217,13 @@ class AILauncher: except Exception: pass + # Clean up layout strategy resources (e.g. linked sessions in windows mode). + if kill_panes and self.layout_strategy is not None: + try: + self.layout_strategy.cleanup() + except Exception: + pass + if kill_panes: if self.terminal_type == "wezterm": backend = WeztermBackend() @@ -3338,6 +3335,11 @@ class AILauncher: print(f" - {t('or_set_ccb_terminal')}", file=sys.stderr) return 2 + if self.layout_mode == "windows" and self.terminal_type != "tmux": + print("Error: --windows layout mode is only supported with tmux.", file=sys.stderr) + print("Tip: Run inside a tmux session or omit --windows.", file=sys.stderr) + return 2 + if not self._require_project_config_dir(): return 2 @@ -3433,8 +3435,10 @@ class AILauncher: except Exception: pass - def _start_item(item: str, *, parent: str | None, direction: str | None) -> str | None: + def _start_item(item: str, parent: str | None, direction: str | None) -> str | None: if item == "cmd": + # In windows mode, cmd splits inside the anchor window (panes behaviour) + # rather than getting its own window. return self._start_cmd_pane(parent_pane=parent, direction=direction, cmd_settings=cmd_settings) if item == "claude": return self._start_claude_pane(parent_pane=parent, direction=direction) @@ -3443,25 +3447,15 @@ class AILauncher: self._warmup_provider(item) return pane_id - right_top: str | None = None - if right_items: - right_top = _start_item(right_items[0], parent=self.anchor_pane_id, direction="right") - if not right_top: - return 1 - - last_left = self.anchor_pane_id - for item in left_items[1:]: - pane_id = _start_item(item, parent=last_left, direction="bottom") - if not pane_id: - return 1 - last_left = pane_id + if self.layout_mode == "windows": + _win_backend = TmuxBackend() + self.layout_strategy = WindowsLayout(_win_backend, self.anchor_pane_id, self.anchor_provider) + else: + self.layout_strategy = PanesLayout() - last_right = right_top - for item in right_items[1:]: - pane_id = _start_item(item, parent=last_right, direction="bottom") - if not pane_id: - return 1 - last_right = pane_id + rc = self.layout_strategy.place_providers(spawn_items, left_items, right_items, self.anchor_pane_id, _start_item) + if rc != 0: + return rc # Optional: start caskd after Codex session file exists (first startup convenience). if "codex" in self.providers and self.anchor_provider != "codex": @@ -3615,6 +3609,12 @@ def cmd_start(args): # Parse explicit provider args early so lock-collision path can reuse an existing pane. requested_providers, _requested_cmd_enabled = _parse_providers_with_cmd(args.providers or []) + # Compute layout_mode early so the lock-collision reuse path can detect mismatches. + _early_config = load_start_config(work_dir) + _early_config_data = _early_config.data if isinstance(_early_config.data, dict) else {} + _early_config_layout = str(_early_config_data.get("layout") or "").strip().lower() + _early_layout_mode = "windows" if args.windows else (_early_config_layout if _early_config_layout in ("windows", "panes") else "panes") + def _existing_provider_pane_for_project(project_work_dir: Path, provider: str) -> tuple[dict | None, str]: prov = (provider or "").strip().lower() if not prov: @@ -3628,6 +3628,15 @@ def cmd_start(args): record = load_registry_by_project_id(project_id, prov) if not isinstance(record, dict): return None, "" + # Check layout mode mismatch: if stored session used a different layout, + # skip reuse so a fresh launch can use the requested layout. + stored_layout = get_layout_mode(record) + if stored_layout != _early_layout_mode: + print( + f"Note: Previous session used '{stored_layout}' layout, current is '{_early_layout_mode}'. Launching fresh.", + file=sys.stderr, + ) + return None, "" providers_map = record.get("providers") if not isinstance(providers_map, dict): return record, "" @@ -3676,8 +3685,8 @@ def cmd_start(args): atexit.register(ccb_lock.release) providers, cmd_enabled = requested_providers, _requested_cmd_enabled - config = load_start_config(work_dir) - config_data = config.data if isinstance(config.data, dict) else {} + config = _early_config + config_data = _early_config_data if not providers: raw_providers = config_data.get("providers") @@ -3708,11 +3717,15 @@ def cmd_start(args): elif not cmd_config: cmd_config = True + config_layout = str(config_data.get("layout") or "").strip().lower() + layout_mode = "windows" if args.windows else (config_layout if config_layout in ("windows", "panes") else "panes") + launcher = AILauncher( providers=providers, resume=resume, auto=auto, cmd_config=cmd_config, + layout_mode=layout_mode, ) return launcher.run_up() @@ -3893,6 +3906,14 @@ def cmd_kill(args): backend.kill_pane(pane_id) elif pane_id and shutil.which("tmux"): backend = TmuxBackend() + # Best-effort cleanup of linked session for windows mode. + try: + tmux_session_for_linked = str(data.get("tmux_session") or "").strip() + if tmux_session_for_linked and not tmux_session_for_linked.startswith("%"): + linked_name = f"{tmux_session_for_linked}-{provider.capitalize()}" + backend.destroy_linked_session(linked_name) + except Exception: + pass if str(pane_id).startswith("%"): backend.kill_pane(str(pane_id)) else: @@ -4889,6 +4910,7 @@ def main(): ) start_parser.add_argument("-r", "--resume", "--restore", action="store_true", help="Resume context") start_parser.add_argument("-a", "--auto", action="store_true", help="Full auto permission mode") + start_parser.add_argument("-w", "--windows", action="store_true", help="Launch each provider in its own tmux window instead of split panes") args = start_parser.parse_args(argv) return cmd_start(args) diff --git a/lib/ccb_start_config.py b/lib/ccb_start_config.py index 58da17ae..f224e646 100644 --- a/lib/ccb_start_config.py +++ b/lib/ccb_start_config.py @@ -20,6 +20,7 @@ class StartConfig: _ALLOWED_PROVIDERS = {"codex", "gemini", "opencode", "claude", "droid"} +_ALLOWED_LAYOUTS = {"panes", "windows"} def _parse_tokens(raw: str) -> list[str]: @@ -76,6 +77,16 @@ def _parse_config_obj(obj: object) -> dict: data["providers"] = providers if cmd_enabled and "cmd" not in data: data["cmd"] = True + + raw_layout = data.get("layout") + if isinstance(raw_layout, str): + normalized = raw_layout.strip().lower() + if normalized in _ALLOWED_LAYOUTS: + data["layout"] = normalized + else: + data.pop("layout", None) + elif raw_layout is not None: + data.pop("layout", None) return data if isinstance(obj, list): diff --git a/lib/layout.py b/lib/layout.py new file mode 100644 index 00000000..df98950b --- /dev/null +++ b/lib/layout.py @@ -0,0 +1,131 @@ +"""Layout strategies for provider pane/window placement.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING, Callable, Optional + +if TYPE_CHECKING: + from terminal import TmuxBackend + +# Callback type: start_item(item, parent, direction) -> pane_id | None +StartItemFn = Callable[[str, Optional[str], Optional[str]], Optional[str]] + + +class LayoutStrategy(ABC): + """Base class for layout strategies controlling how providers are placed.""" + + @abstractmethod + def place_providers( + self, + spawn_items: list[str], + left_items: list[str], + right_items: list[str], + anchor_pane_id: str, + start_item: StartItemFn, + ) -> int: + """Place all providers. Returns 0 on success, 1 on failure.""" + ... + + @property + def linked_sessions(self) -> list[str]: + return [] + + def cleanup(self) -> None: + """Release resources (e.g. linked tmux sessions).""" + + +class PanesLayout(LayoutStrategy): + """Place providers in a left/right split-pane grid.""" + + def place_providers( + self, + spawn_items: list[str], + left_items: list[str], + right_items: list[str], + anchor_pane_id: str, + start_item: StartItemFn, + ) -> int: + # Panes mode uses left_items/right_items for grid layout; spawn_items unused. + right_top: str | None = None + if right_items: + right_top = start_item(right_items[0], anchor_pane_id, "right") + if not right_top: + return 1 + + last_left = anchor_pane_id + for item in left_items[1:]: + pane_id = start_item(item, last_left, "bottom") + if not pane_id: + return 1 + last_left = pane_id + + last_right = right_top + for item in right_items[1:]: + pane_id = start_item(item, last_right, "bottom") + if not pane_id: + return 1 + last_right = pane_id + + return 0 + + +class WindowsLayout(LayoutStrategy): + """Each non-anchor provider gets its own tmux window with a linked session.""" + + def __init__(self, backend: TmuxBackend, anchor_pane_id: str, anchor_provider: str): + self._backend = backend + self._anchor_provider = anchor_provider + # Capture main session name ONCE before any linked sessions are created. + # Linked sessions join the session group and can pollute later + # #{session_name} queries. + self._main_session = backend.get_session_name(anchor_pane_id) + self._linked: list[str] = [] + + @property + def linked_sessions(self) -> list[str]: + return list(self._linked) + + def place_providers( + self, + spawn_items: list[str], + left_items: list[str], + right_items: list[str], + anchor_pane_id: str, + start_item: StartItemFn, + ) -> int: + # Windows mode uses spawn_items; left_items/right_items unused. + for item in spawn_items: + if item == "cmd": + # cmd splits inside the anchor window (panes behaviour). + pane_id = start_item(item, anchor_pane_id, "right") + else: + pane_id = start_item(item, anchor_pane_id, None) + if not pane_id: + return 1 + if item != "cmd": + win_name = item.capitalize() + self._backend.rename_window(pane_id, win_name) + self._create_linked(win_name) + + # Label the anchor window and create its linked session. + anchor_win = self._anchor_provider.capitalize() + self._backend.rename_window(anchor_pane_id, anchor_win) + self._create_linked(anchor_win) + return 0 + + def cleanup(self) -> None: + for name in self._linked: + self._backend.destroy_linked_session(name) + self._linked.clear() + + def _create_linked(self, win_name: str) -> None: + if not self._main_session: + return + linked_name = f"{self._main_session}-{win_name}" + ok = self._backend.create_linked_session( + self._main_session, linked_name, + select_window=f"{linked_name}:{win_name}", + ) + if ok: + self._linked.append(linked_name) diff --git a/lib/pane_registry.py b/lib/pane_registry.py index 3ace8b8d..3af72224 100644 --- a/lib/pane_registry.py +++ b/lib/pane_registry.py @@ -178,6 +178,11 @@ def _provider_pane_alive(record: Dict[str, Any], provider: str) -> bool: return False +def get_layout_mode(record: dict) -> str: + """Return the layout mode for a registry record ('panes' or 'windows').""" + return record.get("layout_mode", "panes") + + def load_registry_by_session_id(session_id: str) -> Optional[Dict[str, Any]]: if not session_id: return None diff --git a/lib/terminal.py b/lib/terminal.py index be64eaa3..1a4c7d29 100644 --- a/lib/terminal.py +++ b/lib/terminal.py @@ -561,6 +561,88 @@ def split_pane(self, parent_pane_id: str, direction: str, percent: int) -> str: raise RuntimeError(f"tmux split-window did not return pane_id: {pane_id!r}") return pane_id + def get_session_name(self, pane_id: str) -> str: + """Return the tmux session name that owns *pane_id*.""" + if not pane_id: + return "" + try: + cp = self._tmux_run( + ["display-message", "-p", "-t", pane_id, "#{session_name}"], + capture=True, timeout=1.0, + ) + return (cp.stdout or "").strip() + except Exception: + return "" + + def rename_window(self, pane_id: str, name: str) -> None: + """Rename the tmux window that contains *pane_id*.""" + if not pane_id or not name: + return + try: + self._tmux_run(["rename-window", "-t", pane_id, name], check=False) + except Exception: + pass + + def create_linked_session( + self, target_session: str, linked_name: str, *, select_window: str = "", + ) -> bool: + """Create a linked tmux session attached to *target_session*. + + Returns True on success. + """ + if not target_session or not linked_name: + return False + try: + self._tmux_run( + ["new-session", "-d", "-t", target_session, "-s", linked_name], + check=True, capture=True, + ) + if select_window: + self._tmux_run( + ["select-window", "-t", select_window], check=False, + ) + return True + except Exception: + return False + + def new_window(self, session: str = "", window_name: str = "") -> str: + """Create a new tmux window and return its pane ID.""" + tmux_args = ["new-window", "-P", "-F", "#{pane_id}"] + if session: + tmux_args.extend(["-t", session]) + if window_name: + tmux_args.extend(["-n", window_name]) + try: + cp = self._tmux_run(tmux_args, check=True, capture=True) + except subprocess.CalledProcessError as e: + err = (getattr(e, "stderr", "") or "").strip() + out = (getattr(e, "stdout", "") or "").strip() + msg = err or out + import sys + print(f"tmux new-window failed (exit {e.returncode}): {msg or 'no stdout/stderr'}", file=sys.stderr) + return "" + except Exception: + return "" + pane_id = (cp.stdout or "").strip() + if not self._looks_like_pane_id(pane_id): + return "" + return pane_id + + def destroy_linked_session(self, session_name: str) -> bool: + """Kill a linked tmux session by name. + + Returns True on success, False on failure. + """ + if not session_name: + return False + try: + cp = self._tmux_run( + ["kill-session", "-t", session_name], check=False, capture=True, + ) + return cp.returncode == 0 + except Exception: + return False + def set_pane_title(self, pane_id: str, title: str) -> None: if not pane_id: return @@ -718,6 +800,26 @@ def activate(self, pane_id: str) -> None: return self._tmux_run(["attach", "-t", pane_id], check=False) + def focus_pane(self, pane_id: str) -> bool: + """Focus a pane, switching to its window first if needed.""" + if not pane_id: + return False + try: + cp = self._tmux_run( + ["display-message", "-p", "-t", pane_id, "#{window_id}"], + capture=True, timeout=1.0, + ) + window_id = (cp.stdout or "").strip() + if cp.returncode != 0 or not window_id: + return False + sw = self._tmux_run(["select-window", "-t", window_id], check=False, timeout=1.0) + if sw.returncode != 0: + return False + sp = self._tmux_run(["select-pane", "-t", pane_id], check=False, timeout=1.0) + return sp.returncode == 0 + except Exception: + return False + def respawn_pane(self, pane_id: str, *, cmd: str, cwd: str | None = None, stderr_log_path: str | None = None, remain_on_exit: bool = True) -> None: """ @@ -797,12 +899,13 @@ def save_crash_log(self, pane_id: str, crash_log_path: str, *, lines: int = 1000 p.write_text(text, encoding="utf-8") def create_pane(self, cmd: str, cwd: str, direction: str = "right", percent: int = 50, - parent_pane: Optional[str] = None) -> str: + parent_pane: Optional[str] = None, layout_mode: str = "panes") -> str: """ Create a new pane and run `cmd` inside it. - If `parent_pane` is provided (or we are inside tmux), split that pane. - If called outside tmux without `parent_pane`, create a detached session and return its root pane id. + - When `layout_mode` is ``"windows"``, create a new tmux window instead of splitting. """ cmd = (cmd or "").strip() cwd = (cwd or ".").strip() or "." @@ -815,7 +918,22 @@ def create_pane(self, cmd: str, cwd: str, direction: str = "right", percent: int base = None if base: - new_pane = self.split_pane(base, direction=direction, percent=percent) + if layout_mode == "windows": + # Derive the session name from the existing pane so the new window + # is created in the same session. + try: + cp = self._tmux_run( + ["display-message", "-p", "-t", base, "#{session_name}"], + capture=True, timeout=1.0, + ) + session_name = (cp.stdout or "").strip() + except Exception: + session_name = "" + new_pane = self.new_window(session=session_name) + if not new_pane: + raise RuntimeError("tmux new-window failed to create a window") + else: + new_pane = self.split_pane(base, direction=direction, percent=percent) if cmd: self.respawn_pane(new_pane, cmd=cmd, cwd=cwd) return new_pane diff --git a/test/test_ccb_tmux_split.py b/test/test_ccb_tmux_split.py index a5dc9d69..d95e686b 100644 --- a/test/test_ccb_tmux_split.py +++ b/test/test_ccb_tmux_split.py @@ -72,6 +72,7 @@ def create_pane( direction: str = "right", percent: int = 50, parent_pane: str | None = None, + layout_mode: str = "panes", ) -> str: self._created += 1 return f"%{10 + self._created}" diff --git a/test/test_windows_layout.py b/test/test_windows_layout.py new file mode 100644 index 00000000..b2a9ac13 --- /dev/null +++ b/test/test_windows_layout.py @@ -0,0 +1,671 @@ +from __future__ import annotations + +import json +import subprocess +import time +from pathlib import Path +from typing import Any + +import pytest + +import terminal +from ccb_start_config import _parse_config_obj +from layout import PanesLayout, WindowsLayout +from pane_registry import get_layout_mode, upsert_registry, registry_path_for_session + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _cp(*, stdout: str = "", returncode: int = 0) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess(args=["tmux"], returncode=returncode, stdout=stdout, stderr="") + + +# --------------------------------------------------------------------------- +# Config parsing tests +# --------------------------------------------------------------------------- + +class TestConfigLayoutParsing: + def test_config_layout_windows(self) -> None: + data = _parse_config_obj({"providers": ["codex"], "layout": "windows"}) + assert data["layout"] == "windows" + + def test_config_layout_panes_default(self) -> None: + data = _parse_config_obj({"providers": ["codex"]}) + assert "layout" not in data + + def test_config_layout_invalid(self) -> None: + data = _parse_config_obj({"providers": ["codex"], "layout": "stacked"}) + assert "layout" not in data + + def test_config_layout_panes_explicit(self) -> None: + data = _parse_config_obj({"providers": ["codex"], "layout": "panes"}) + assert data["layout"] == "panes" + + def test_config_layout_case_insensitive(self) -> None: + data = _parse_config_obj({"providers": ["codex"], "layout": "WINDOWS"}) + assert data["layout"] == "windows" + + def test_config_layout_non_string_stripped(self) -> None: + data = _parse_config_obj({"providers": ["codex"], "layout": 123}) + assert "layout" not in data + + +# --------------------------------------------------------------------------- +# Registry tests +# --------------------------------------------------------------------------- + +class TestRegistryLayoutMode: + def test_get_layout_mode_default(self) -> None: + record: dict[str, Any] = {"ccb_session_id": "s1"} + assert get_layout_mode(record) == "panes" + + def test_get_layout_mode_windows(self) -> None: + record: dict[str, Any] = {"ccb_session_id": "s1", "layout_mode": "windows"} + assert get_layout_mode(record) == "windows" + + def test_get_layout_mode_panes_explicit(self) -> None: + record: dict[str, Any] = {"ccb_session_id": "s1", "layout_mode": "panes"} + assert get_layout_mode(record) == "panes" + + def test_registry_stores_layout_mode(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + # Point the registry dir to tmp_path so we don't pollute the real home. + monkeypatch.setattr("pane_registry._registry_dir", lambda: tmp_path) + + record: dict[str, Any] = { + "ccb_session_id": "test-layout-001", + "layout_mode": "windows", + "terminal": "tmux", + "work_dir": str(tmp_path), + } + assert upsert_registry(record) is True + + written = registry_path_for_session("test-layout-001") + assert written.exists() + data = json.loads(written.read_text(encoding="utf-8")) + assert data["layout_mode"] == "windows" + + +# --------------------------------------------------------------------------- +# TmuxBackend mock tests +# --------------------------------------------------------------------------- + +class TestCreatePaneLayoutMode: + def test_create_pane_windows_mode_calls_new_window(self, monkeypatch: pytest.MonkeyPatch) -> None: + """When layout_mode='windows', create_pane should call new_window instead of split_pane.""" + calls: list[dict[str, Any]] = [] + + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + calls.append({"args": args, "check": check, "capture": capture}) + # get_current_pane_id queries + if args == ["display-message", "-p", "#{pane_id}"]: + return _cp(stdout="%0\n") + # pane_exists check + if args == ["display-message", "-p", "-t", "%0", "#{pane_dead}"]: + return _cp(stdout="0\n") + # session_name lookup for new_window + if len(args) >= 4 and "#{session_name}" in args: + return _cp(stdout="mysession\n") + # new-window call + if args and args[0] == "new-window": + return _cp(stdout="%99\n") + # respawn-pane (noop) + if args and args[0] == "respawn-pane": + return _cp() + # set-option for remain-on-exit + if args and args[0] == "set-option": + return _cp() + return _cp(stdout="") + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + pane_id = backend.create_pane(cmd="echo hello", cwd="/tmp", parent_pane="%0", layout_mode="windows") + assert pane_id == "%99" + + # Verify that new-window was called (not split-window). + new_window_calls = [c for c in calls if c["args"] and c["args"][0] == "new-window"] + split_calls = [c for c in calls if c["args"] and c["args"][0] == "split-window"] + assert len(new_window_calls) == 1 + assert len(split_calls) == 0 + + # Verify the session target was passed to new-window. + nw_args = new_window_calls[0]["args"] + assert "-t" in nw_args and "mysession" in nw_args + + def test_create_pane_panes_mode_calls_split(self, monkeypatch: pytest.MonkeyPatch) -> None: + """When layout_mode='panes' (default), create_pane should call split_pane.""" + calls: list[dict[str, Any]] = [] + + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + calls.append({"args": args, "check": check, "capture": capture}) + # pane_exists uses display-message #{pane_id} + if args == ["display-message", "-p", "-t", "%0", "#{pane_id}"]: + return _cp(stdout="%0\n") + # pane size for split_pane + if len(args) >= 4 and "#{pane_width}x#{pane_height}" in args: + return _cp(stdout="160x40\n") + # zoom check + if len(args) >= 4 and "#{window_zoomed_flag}" in args: + return _cp(stdout="0\n") + # split-window + if args and args[0] == "split-window": + return _cp(stdout="%55\n") + # respawn-pane + if args and args[0] == "respawn-pane": + return _cp() + # set-option for remain-on-exit + if args and args[0] == "set-option": + return _cp() + return _cp(stdout="") + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + pane_id = backend.create_pane(cmd="echo hello", cwd="/tmp", parent_pane="%0", layout_mode="panes") + assert pane_id == "%55" + + # Verify that split-window was called (not new-window). + split_calls = [c for c in calls if c["args"] and c["args"][0] == "split-window"] + new_window_calls = [c for c in calls if c["args"] and c["args"][0] == "new-window"] + assert len(split_calls) == 1 + assert len(new_window_calls) == 0 + + def test_create_pane_default_layout_is_panes(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Calling create_pane without layout_mode should behave as 'panes'.""" + calls: list[dict[str, Any]] = [] + + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + calls.append({"args": args}) + # pane_exists uses display-message #{pane_id} + if args == ["display-message", "-p", "-t", "%0", "#{pane_id}"]: + return _cp(stdout="%0\n") + if len(args) >= 4 and "#{pane_width}x#{pane_height}" in args: + return _cp(stdout="160x40\n") + if len(args) >= 4 and "#{window_zoomed_flag}" in args: + return _cp(stdout="0\n") + if args and args[0] == "split-window": + return _cp(stdout="%10\n") + if args and args[0] == "respawn-pane": + return _cp() + if args and args[0] == "set-option": + return _cp() + return _cp(stdout="") + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + pane_id = backend.create_pane(cmd="echo hi", cwd="/tmp", parent_pane="%0") + assert pane_id == "%10" + + split_calls = [c for c in calls if c["args"] and c["args"][0] == "split-window"] + new_window_calls = [c for c in calls if c["args"] and c["args"][0] == "new-window"] + assert len(split_calls) == 1 + assert len(new_window_calls) == 0 + + +# --------------------------------------------------------------------------- +# new_window unit tests +# --------------------------------------------------------------------------- + +class TestNewWindow: + def test_new_window_returns_pane_id(self, monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[list[str]] = [] + + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + calls.append(args) + if args and args[0] == "new-window": + return _cp(stdout="%77\n") + return _cp(stdout="") + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + pane_id = backend.new_window(session="sess1", window_name="my-win") + assert pane_id == "%77" + # First call should be new-window. + argv = calls[0] + assert argv[0] == "new-window" + assert "-P" in argv + assert "-F" in argv and "#{pane_id}" in argv + assert "-t" in argv and "sess1" in argv + assert "-n" in argv and "my-win" in argv + + def test_new_window_no_session_no_name(self, monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[list[str]] = [] + + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + calls.append(args) + return _cp(stdout="%80\n") + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + pane_id = backend.new_window() + assert pane_id == "%80" + # Without window_name, only new-window is called (no linked session). + assert len(calls) == 1 + argv = calls[0] + assert "-t" not in argv + assert "-n" not in argv + + def test_new_window_returns_empty_on_failure(self, monkeypatch: pytest.MonkeyPatch) -> None: + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + raise subprocess.CalledProcessError(1, ["tmux", *args]) + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + pane_id = backend.new_window(session="s") + assert pane_id == "" + + def test_new_window_does_not_create_linked_session(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Verify new_window() only creates the window; linked sessions are handled by run_up().""" + calls: list[dict[str, Any]] = [] + + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + calls.append({"args": args, "check": check}) + if args and args[0] == "new-window": + return _cp(stdout="%42\n") + return _cp(stdout="") + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + pane_id = backend.new_window(session="main-sess", window_name="Codex") + assert pane_id == "%42" + + # Should only have the new-window call -- no new-session or display-message + new_session_calls = [c for c in calls if c["args"] and c["args"][0] == "new-session"] + assert len(new_session_calls) == 0 + display_calls = [c for c in calls if c["args"] and "#{session_name}" in str(c["args"])] + assert len(display_calls) == 0 + + +# --------------------------------------------------------------------------- +# TmuxBackend helper method tests +# --------------------------------------------------------------------------- + +class TestBackendHelpers: + def test_get_session_name(self, monkeypatch: pytest.MonkeyPatch) -> None: + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + if "#{session_name}" in args: + return _cp(stdout="my-session\n") + return _cp() + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + assert backend.get_session_name("%0") == "my-session" + + def test_get_session_name_empty_pane(self, monkeypatch: pytest.MonkeyPatch) -> None: + backend = terminal.TmuxBackend() + assert backend.get_session_name("") == "" + + def test_rename_window(self, monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[list[str]] = [] + + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + calls.append(args) + return _cp() + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + backend.rename_window("%5", "Codex") + assert len(calls) == 1 + assert calls[0][0] == "rename-window" + assert "%5" in calls[0] + assert "Codex" in calls[0] + + def test_rename_window_empty_args(self, monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[list[str]] = [] + + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + calls.append(args) + return _cp() + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + backend.rename_window("", "Codex") + backend.rename_window("%5", "") + assert len(calls) == 0 + + def test_create_linked_session_success(self, monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[list[str]] = [] + + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + calls.append(args) + return _cp() + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + result = backend.create_linked_session("main", "main-Codex", select_window="main-Codex:Codex") + assert result is True + assert len(calls) == 2 + assert calls[0][0] == "new-session" + assert "main" in calls[0] and "main-Codex" in calls[0] + assert calls[1][0] == "select-window" + + def test_create_linked_session_empty_args(self, monkeypatch: pytest.MonkeyPatch) -> None: + backend = terminal.TmuxBackend() + assert backend.create_linked_session("", "linked") is False + assert backend.create_linked_session("main", "") is False + + def test_create_linked_session_failure(self, monkeypatch: pytest.MonkeyPatch) -> None: + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + raise subprocess.CalledProcessError(1, ["tmux"]) + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + assert backend.create_linked_session("main", "main-X") is False + + +# --------------------------------------------------------------------------- +# PanesLayout strategy tests +# --------------------------------------------------------------------------- + +class TestPanesLayout: + def test_places_right_then_left_then_right(self) -> None: + calls: list[tuple[str, str | None, str | None]] = [] + + def fake_start(item: str, parent: str | None, direction: str | None) -> str | None: + pane_id = f"%{len(calls) + 10}" + calls.append((item, parent, direction)) + return pane_id + + layout = PanesLayout() + rc = layout.place_providers( + spawn_items=["codex", "gemini"], + left_items=["claude", "codex"], + right_items=["gemini"], + anchor_pane_id="%0", + start_item=fake_start, + ) + assert rc == 0 + # First call: right_items[0] with anchor as parent, direction right + assert calls[0] == ("gemini", "%0", "right") + # Second call: left_items[1] (codex) with anchor as parent, direction bottom + assert calls[1] == ("codex", "%0", "bottom") + + def test_returns_1_on_failure(self) -> None: + def failing_start(item: str, parent: str | None, direction: str | None) -> str | None: + return None + + layout = PanesLayout() + rc = layout.place_providers(["codex"], ["claude"], ["codex"], "%0", failing_start) + assert rc == 1 + + def test_no_right_items(self) -> None: + calls: list[tuple[str, str | None, str | None]] = [] + + def fake_start(item: str, parent: str | None, direction: str | None) -> str | None: + calls.append((item, parent, direction)) + return f"%{len(calls) + 10}" + + layout = PanesLayout() + rc = layout.place_providers( + spawn_items=["codex"], + left_items=["claude", "codex"], + right_items=[], + anchor_pane_id="%0", + start_item=fake_start, + ) + assert rc == 0 + assert len(calls) == 1 + assert calls[0] == ("codex", "%0", "bottom") + + def test_linked_sessions_empty(self) -> None: + layout = PanesLayout() + assert layout.linked_sessions == [] + + def test_cleanup_noop(self) -> None: + layout = PanesLayout() + layout.cleanup() # should not raise + + +# --------------------------------------------------------------------------- +# WindowsLayout strategy tests +# --------------------------------------------------------------------------- + +class TestWindowsLayout: + @staticmethod + def _make_backend(monkeypatch: pytest.MonkeyPatch, session_name: str = "main") -> terminal.TmuxBackend: + calls: list[list[str]] = [] + + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + calls.append(args) + if "#{session_name}" in args: + return _cp(stdout=f"{session_name}\n") + return _cp() + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + backend._test_calls = calls # type: ignore[attr-defined] + return backend + + def test_captures_session_name(self, monkeypatch: pytest.MonkeyPatch) -> None: + backend = self._make_backend(monkeypatch, "my-session") + layout = WindowsLayout(backend, "%0", "claude") + assert layout._main_session == "my-session" + + def test_place_providers_creates_windows_and_linked(self, monkeypatch: pytest.MonkeyPatch) -> None: + backend = self._make_backend(monkeypatch) + layout = WindowsLayout(backend, "%0", "claude") + + started: list[str] = [] + + def fake_start(item: str, parent: str | None, direction: str | None) -> str | None: + started.append(item) + return f"%{len(started) + 20}" + + rc = layout.place_providers( + spawn_items=["codex", "gemini"], + left_items=[], right_items=[], + anchor_pane_id="%0", + start_item=fake_start, + ) + assert rc == 0 + assert started == ["codex", "gemini"] + # 2 providers + 1 anchor = 3 linked sessions + assert len(layout.linked_sessions) == 3 + + def test_cmd_gets_right_direction(self, monkeypatch: pytest.MonkeyPatch) -> None: + backend = self._make_backend(monkeypatch) + layout = WindowsLayout(backend, "%0", "claude") + + directions: list[str | None] = [] + + def fake_start(item: str, parent: str | None, direction: str | None) -> str | None: + directions.append(direction) + return f"%{len(directions) + 30}" + + layout.place_providers(["cmd", "codex"], [], [], "%0", fake_start) + assert directions[0] == "right" # cmd + assert directions[1] is None # codex (new window) + + def test_cleanup_destroys_linked_sessions(self, monkeypatch: pytest.MonkeyPatch) -> None: + backend = self._make_backend(monkeypatch) + layout = WindowsLayout(backend, "%0", "claude") + + def fake_start(item: str, parent: str | None, direction: str | None) -> str | None: + return "%50" + + layout.place_providers(["codex"], [], [], "%0", fake_start) + assert len(layout.linked_sessions) > 0 + + layout.cleanup() + assert layout.linked_sessions == [] + # Verify kill-session calls were made + kill_calls = [c for c in backend._test_calls if c and c[0] == "kill-session"] # type: ignore[attr-defined] + assert len(kill_calls) > 0 + + def test_returns_1_on_failure(self, monkeypatch: pytest.MonkeyPatch) -> None: + backend = self._make_backend(monkeypatch) + layout = WindowsLayout(backend, "%0", "claude") + + def failing_start(item: str, parent: str | None, direction: str | None) -> str | None: + return None + + rc = layout.place_providers(["codex"], [], [], "%0", failing_start) + assert rc == 1 + + +# --------------------------------------------------------------------------- +# destroy_linked_session unit tests +# --------------------------------------------------------------------------- + +class TestDestroyLinkedSession: + def test_destroy_linked_session_success(self, monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[list[str]] = [] + + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + calls.append(args) + return _cp(returncode=0) + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + result = backend.destroy_linked_session("main-sess-Codex") + assert result is True + assert len(calls) == 1 + assert calls[0][0] == "kill-session" + assert "-t" in calls[0] and "main-sess-Codex" in calls[0] + + def test_destroy_linked_session_failure(self, monkeypatch: pytest.MonkeyPatch) -> None: + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + return _cp(returncode=1) + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + result = backend.destroy_linked_session("nonexistent") + assert result is False + + def test_destroy_linked_session_empty_name(self, monkeypatch: pytest.MonkeyPatch) -> None: + backend = terminal.TmuxBackend() + assert backend.destroy_linked_session("") is False + + def test_destroy_linked_session_exception(self, monkeypatch: pytest.MonkeyPatch) -> None: + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + raise OSError("tmux not found") + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + result = backend.destroy_linked_session("sess-Codex") + assert result is False + + +# --------------------------------------------------------------------------- +# focus_pane unit tests +# --------------------------------------------------------------------------- + +class TestFocusPane: + def test_focus_pane_selects_window_then_pane(self, monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[list[str]] = [] + + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + calls.append(args) + if "#{window_id}" in args: + return _cp(stdout="@3\n") + return _cp() + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + result = backend.focus_pane("%5") + assert result is True + # Should have: display-message (get window), select-window, select-pane + assert len(calls) == 3 + assert calls[1][0] == "select-window" + assert "@3" in calls[1] + assert calls[2][0] == "select-pane" + assert "%5" in calls[2] + + def test_focus_pane_empty_id_returns_false(self, monkeypatch: pytest.MonkeyPatch) -> None: + backend = terminal.TmuxBackend() + assert backend.focus_pane("") is False + + def test_focus_pane_returns_false_on_error(self, monkeypatch: pytest.MonkeyPatch) -> None: + def fake_tmux_run( + self: terminal.TmuxBackend, args: list[str], *, check: bool = False, + capture: bool = False, input_bytes: bytes | None = None, + timeout: float | None = None, + ) -> subprocess.CompletedProcess[str]: + return _cp(returncode=1) + + backend = terminal.TmuxBackend() + monkeypatch.setattr(backend, "_tmux_run", fake_tmux_run.__get__(backend, terminal.TmuxBackend)) + + assert backend.focus_pane("%1") is False