From a9bd92b482928982f996adf75ef45651959b8d04 Mon Sep 17 00:00:00 2001 From: Guolong Date: Mon, 16 Mar 2026 18:44:40 +0000 Subject: [PATCH 01/11] feat: add foundation for --windows layout mode - TmuxBackend: add new_window() and focus_pane() methods - TmuxBackend: extend create_pane() with layout_mode parameter - pane_registry: add get_layout_mode() accessor - ccb: add -w/--windows CLI flag and layout_mode to AILauncher - ccb_start_config: parse "layout" field from config JSON Co-Authored-By: Claude Opus 4.6 (1M context) --- ccb | 7 +++++ lib/ccb_start_config.py | 11 +++++++ lib/pane_registry.py | 5 ++++ lib/terminal.py | 64 +++++++++++++++++++++++++++++++++++++++-- 4 files changed, 85 insertions(+), 2 deletions(-) diff --git a/ccb b/ccb index 90d028c2..072acc9c 100755 --- a/ccb +++ b/ccb @@ -555,11 +555,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. @@ -3708,11 +3710,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() @@ -4889,6 +4895,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/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..f3d1ca24 100644 --- a/lib/terminal.py +++ b/lib/terminal.py @@ -561,6 +561,30 @@ 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 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 + # Log error and return empty string, matching split_pane error style + 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 set_pane_title(self, pane_id: str, title: str) -> None: if not pane_id: return @@ -718,6 +742,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 +841,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 +860,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 From e94c7bd0b7752279e6bf9f56b46c3c5da490808c Mon Sep 17 00:00:00 2001 From: Guolong Date: Mon, 16 Mar 2026 18:52:51 +0000 Subject: [PATCH 02/11] feat: implement windows layout logic in run_up() - run_up(): add windows-mode branch that creates new tmux windows per provider instead of split panes, with window renaming - WezTerm guard: exit with error if --windows used with non-tmux backend - cmd pane: always splits inside anchor window, even in windows mode - _start_provider/_start_*_tmux: thread layout_mode through to create_pane - test: update fake backend to accept layout_mode parameter Co-Authored-By: Claude Opus 4.6 (1M context) --- ccb | 103 +++++++++++++++++++++++++----------- test/test_ccb_tmux_split.py | 1 + 2 files changed, 73 insertions(+), 31 deletions(-) diff --git a/ccb b/ccb index 072acc9c..e27bc23e 100755 --- a/ccb +++ b/ccb @@ -1183,7 +1183,7 @@ class AILauncher: ) return True - def _start_provider(self, provider: str, *, parent_pane: str | None = None, direction: str | None = None) -> str | None: + def _start_provider(self, provider: str, *, parent_pane: str | None = None, direction: str | None = None, layout_mode: str = "panes") -> str | None: # Handle case when no terminal detected if self.terminal_type is None: print(f"❌ {t('no_terminal_backend')}") @@ -1222,13 +1222,13 @@ class AILauncher: print(f"🚀 {t('starting_backend', provider=provider.capitalize(), terminal='tmux')}") if provider == "codex": - return self._start_codex_tmux(parent_pane=parent_pane, direction=direction) + return self._start_codex_tmux(parent_pane=parent_pane, direction=direction, layout_mode=layout_mode) elif provider == "gemini": - return self._start_gemini_tmux(parent_pane=parent_pane, direction=direction) + return self._start_gemini_tmux(parent_pane=parent_pane, direction=direction, layout_mode=layout_mode) elif provider == "opencode": - return self._start_opencode_tmux(parent_pane=parent_pane, direction=direction) + return self._start_opencode_tmux(parent_pane=parent_pane, direction=direction, layout_mode=layout_mode) elif provider == "droid": - return self._start_droid_tmux(parent_pane=parent_pane, direction=direction) + return self._start_droid_tmux(parent_pane=parent_pane, direction=direction, layout_mode=layout_mode) else: print(f"❌ {t('unknown_provider', provider=provider)}") return None @@ -2098,6 +2098,7 @@ class AILauncher: *, parent_pane: str | None = None, direction: str | None = None, + layout_mode: str = "panes", ) -> str | None: runtime = self.runtime_dir / "codex" runtime.mkdir(parents=True, exist_ok=True) @@ -2134,7 +2135,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=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") @@ -2197,6 +2198,7 @@ class AILauncher: *, parent_pane: str | None = None, direction: str | None = None, + layout_mode: str = "panes", ) -> str | None: runtime = self.runtime_dir / "gemini" runtime.mkdir(parents=True, exist_ok=True) @@ -2225,7 +2227,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=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") @@ -2248,6 +2250,7 @@ class AILauncher: *, parent_pane: str | None = None, direction: str | None = None, + layout_mode: str = "panes", ) -> str | None: runtime = self.runtime_dir / "opencode" runtime.mkdir(parents=True, exist_ok=True) @@ -2279,7 +2282,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=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") @@ -2302,6 +2305,7 @@ class AILauncher: *, parent_pane: str | None = None, direction: str | None = None, + layout_mode: str = "panes", ) -> str | None: runtime = self.runtime_dir / "droid" runtime.mkdir(parents=True, exist_ok=True) @@ -2331,7 +2335,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=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") @@ -2355,6 +2359,7 @@ class AILauncher: parent_pane: str | None, direction: str | None, cmd_settings: dict, + layout_mode: str = "panes", ) -> str | None: if not cmd_settings.get("enabled"): return None @@ -2377,7 +2382,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=layout_mode) 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") @@ -3134,7 +3139,7 @@ class AILauncher: print(f"\n⚠️ {t('user_interrupted')}") return 130 - def _start_claude_pane(self, *, parent_pane: str | None, direction: str | None) -> str | None: + def _start_claude_pane(self, *, parent_pane: str | None, direction: str | None, layout_mode: str = "panes") -> str | None: print(f"🚀 {t('starting_claude')}") env_overrides = self._claude_env_overrides() @@ -3167,7 +3172,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=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") @@ -3340,6 +3345,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 @@ -3437,33 +3447,64 @@ class AILauncher: 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) - pane_id = self._start_provider(item, parent_pane=parent, direction=direction) + return self._start_claude_pane(parent_pane=parent, direction=direction, layout_mode=self.layout_mode) + pane_id = self._start_provider(item, parent_pane=parent, direction=direction, layout_mode=self.layout_mode) if pane_id: 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 + if self.layout_mode == "windows": + # Windows mode: each non-anchor provider gets its own tmux window. + # The "cmd" item splits inside the anchor window (panes behaviour). + _win_backend = TmuxBackend() + for item in spawn_items: + if item == "cmd": + pane_id = _start_item(item, parent=self.anchor_pane_id, direction="right") + else: + pane_id = _start_item(item, parent=self.anchor_pane_id, direction=None) + if not pane_id: + return 1 + # Label the new tmux window so the user can identify it. + if item != "cmd": + try: + _win_backend._tmux_run( + ["rename-window", "-t", pane_id, item.capitalize()], + check=False, + ) + except Exception: + pass + # Also label the anchor window. + try: + _win_backend._tmux_run( + ["rename-window", "-t", self.anchor_pane_id, self.anchor_provider.capitalize()], + check=False, + ) + except Exception: + pass + else: + 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 + 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 - 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 + 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 # Optional: start caskd after Codex session file exists (first startup convenience). if "codex" in self.providers and self.anchor_provider != "codex": 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}" From 0dcc02d53a43359e21aa0bfcb75f0506f1b92fe2 Mon Sep 17 00:00:00 2001 From: Guolong Date: Mon, 16 Mar 2026 18:59:20 +0000 Subject: [PATCH 03/11] feat: handle resume layout mismatch and persist layout_mode - Resume: detect stored vs current layout mode mismatch, skip pane reuse and launch fresh with warning when they differ - Registry: store layout_mode in all 6 provider upsert_registry calls - Optimize: reuse early config load to avoid duplicate file reads Co-Authored-By: Claude Opus 4.6 (1M context) --- ccb | 27 ++++++++++++++++++++++++--- 1 file changed, 24 insertions(+), 3 deletions(-) diff --git a/ccb b/ccb index e27bc23e..c3399f35 100755 --- a/ccb +++ b/ccb @@ -38,7 +38,7 @@ 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 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 @@ -1451,6 +1451,7 @@ class AILauncher: "ccb_project_id": compute_ccb_project_id(self.project_root), "work_dir": str(self.project_root), "terminal": terminal or self.terminal_type, + "layout_mode": self.layout_mode, "providers": { "claude": { "pane_id": pane_id, @@ -2701,6 +2702,7 @@ class AILauncher: "ccb_project_id": compute_ccb_project_id(self.project_root), "work_dir": str(self.project_root), "terminal": self.terminal_type, + "layout_mode": self.layout_mode, "providers": { "codex": { "pane_id": pane_id, @@ -2724,6 +2726,7 @@ class AILauncher: "ccb_project_id": compute_ccb_project_id(self.project_root), "work_dir": str(Path.cwd()), "terminal": self.terminal_type, + "layout_mode": self.layout_mode, "providers": { "claude": {"pane_id": claude_pane_id}, "codex": {"pane_id": codex_pane_id}, @@ -2778,6 +2781,7 @@ class AILauncher: "ccb_project_id": compute_ccb_project_id(self.project_root), "work_dir": str(self.project_root), "terminal": self.terminal_type, + "layout_mode": self.layout_mode, "providers": { "gemini": { "pane_id": pane_id, @@ -2827,6 +2831,7 @@ class AILauncher: "ccb_project_id": compute_ccb_project_id(self.project_root), "work_dir": str(self.project_root), "terminal": self.terminal_type, + "layout_mode": self.layout_mode, "providers": { "opencode": { "pane_id": pane_id, @@ -2893,6 +2898,7 @@ class AILauncher: "ccb_project_id": compute_ccb_project_id(self.project_root), "work_dir": str(self.project_root), "terminal": self.terminal_type, + "layout_mode": self.layout_mode, "providers": { "droid": { "pane_id": pane_id, @@ -3658,6 +3664,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: @@ -3671,6 +3683,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, "" @@ -3719,8 +3740,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") From b7c003bf4efc004544690d6d76f4a4639a861fab Mon Sep 17 00:00:00 2001 From: Guolong Date: Mon, 16 Mar 2026 18:59:26 +0000 Subject: [PATCH 04/11] test: add 19 unit tests for windows layout mode - Config parsing: layout field validation, defaults, case handling - Registry: get_layout_mode accessor, persistence via upsert - TmuxBackend: create_pane routing (new_window vs split_pane) - new_window: session/name flags, failure handling - focus_pane: window switching, error cases Co-Authored-By: Claude Opus 4.6 (1M context) --- test/test_windows_layout.py | 328 ++++++++++++++++++++++++++++++++++++ 1 file changed, 328 insertions(+) create mode 100644 test/test_windows_layout.py diff --git a/test/test_windows_layout.py b/test/test_windows_layout.py new file mode 100644 index 00000000..1842caa3 --- /dev/null +++ b/test/test_windows_layout.py @@ -0,0 +1,328 @@ +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 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) + return _cp(stdout="%77\n") + + 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" + assert len(calls) == 1 + 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" + 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 == "" + + +# --------------------------------------------------------------------------- +# 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 From 67c7b7ce9a9d32efb82e5305fb98f0d9f253f9ca Mon Sep 17 00:00:00 2001 From: Guolong Date: Mon, 16 Mar 2026 19:00:40 +0000 Subject: [PATCH 05/11] docs: document --windows flag and layout config option - README.md: add flag table entry, config example, usage tips - README_zh.md: matching Chinese translations Co-Authored-By: Claude Opus 4.6 (1M context) --- README.md | 5 +++++ README_zh.md | 12 ++++++++++++ 2 files changed, 17 insertions(+) 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。 ### 后续更新 From ef99a66f677d9f8cd5176cdeae47b2b017c0d21b Mon Sep 17 00:00:00 2001 From: Guolong Date: Mon, 16 Mar 2026 19:46:42 +0000 Subject: [PATCH 06/11] feat: create linked tmux sessions for independent provider viewports - new_window(): after creating window, also creates a linked session (tmux new-session -d -t {main} -s {main}-{provider}) and points it at the correct window for independent attach - destroy_linked_session(): cleanup method for linked sessions - AILauncher: track linked_sessions list, clean up on exit and kill - Registry: store linked_sessions field for all providers - Tests: 6 new tests for linked session create/destroy/failure Users can now: tmux attach -t {session}-{provider} from any terminal to view a provider independently. Co-Authored-By: Claude Opus 4.6 (1M context) --- ccb | 111 ++++++++++++++++++++------- lib/terminal.py | 54 +++++++++++++- test/test_windows_layout.py | 145 +++++++++++++++++++++++++++++++++++- 3 files changed, 280 insertions(+), 30 deletions(-) diff --git a/ccb b/ccb index c3399f35..f310040e 100755 --- a/ccb +++ b/ccb @@ -589,6 +589,7 @@ class AILauncher: self.tmux_panes = {} self.wezterm_panes = {} self.extra_panes = {} + self.linked_sessions: list[str] = [] self.processes = {} self.anchor_provider = None self.anchor_pane_id = None @@ -1445,24 +1446,25 @@ 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, - "layout_mode": self.layout_mode, - "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 = { + "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, + "layout_mode": self.layout_mode, + "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"), + } + }, + } + if self.linked_sessions: + _reg["linked_sessions"] = list(self.linked_sessions) + upsert_registry(_reg) except Exception: pass self._maybe_start_provider_daemon("claude") @@ -2697,7 +2699,7 @@ class AILauncher: print(err, file=sys.stderr) return False try: - upsert_registry({ + _reg = { "ccb_session_id": self.session_id, "ccb_project_id": compute_ccb_project_id(self.project_root), "work_dir": str(self.project_root), @@ -2710,7 +2712,10 @@ class AILauncher: "session_file": str(session_file), } }, - }) + } + if self.linked_sessions: + _reg["linked_sessions"] = list(self.linked_sessions) + upsert_registry(_reg) except Exception: pass self._maybe_start_provider_daemon("codex") @@ -2732,6 +2737,8 @@ class AILauncher: "codex": {"pane_id": codex_pane_id}, }, } + if self.linked_sessions: + record["linked_sessions"] = list(self.linked_sessions) ok = upsert_registry(record) if not ok: print("⚠️ Failed to update cpend registry", file=sys.stderr) @@ -2776,7 +2783,7 @@ class AILauncher: print(err, file=sys.stderr) return False try: - upsert_registry({ + _reg = { "ccb_session_id": self.session_id, "ccb_project_id": compute_ccb_project_id(self.project_root), "work_dir": str(self.project_root), @@ -2789,7 +2796,10 @@ class AILauncher: "session_file": str(session_file), } }, - }) + } + if self.linked_sessions: + _reg["linked_sessions"] = list(self.linked_sessions) + upsert_registry(_reg) except Exception: pass self._maybe_start_provider_daemon("gemini") @@ -2826,7 +2836,7 @@ class AILauncher: print(err, file=sys.stderr) return False try: - upsert_registry({ + _reg = { "ccb_session_id": self.session_id, "ccb_project_id": compute_ccb_project_id(self.project_root), "work_dir": str(self.project_root), @@ -2839,7 +2849,10 @@ class AILauncher: "session_file": str(session_file), } }, - }) + } + if self.linked_sessions: + _reg["linked_sessions"] = list(self.linked_sessions) + upsert_registry(_reg) except Exception: pass self._maybe_start_provider_daemon("opencode") @@ -2893,7 +2906,7 @@ class AILauncher: print(err, file=sys.stderr) return False try: - upsert_registry({ + _reg = { "ccb_session_id": self.session_id, "ccb_project_id": compute_ccb_project_id(self.project_root), "work_dir": str(self.project_root), @@ -2908,7 +2921,10 @@ class AILauncher: "droid_session_path": droid_session_path, } }, - }) + } + if self.linked_sessions: + _reg["linked_sessions"] = list(self.linked_sessions) + upsert_registry(_reg) except Exception: pass self._maybe_start_provider_daemon("droid") @@ -3240,6 +3256,16 @@ class AILauncher: except Exception: pass + # Kill linked tmux sessions (windows mode) before killing panes. + if kill_panes and self.linked_sessions: + _lb = TmuxBackend() + for ls_name in self.linked_sessions: + try: + _lb.destroy_linked_session(ls_name) + except Exception: + pass + self.linked_sessions.clear() + if kill_panes: if self.terminal_type == "wezterm": backend = WeztermBackend() @@ -3466,6 +3492,8 @@ class AILauncher: if self.layout_mode == "windows": # Windows mode: each non-anchor provider gets its own tmux window. # The "cmd" item splits inside the anchor window (panes behaviour). + # After creating each window we also create a linked tmux session so + # each provider can be independently attached from another terminal. _win_backend = TmuxBackend() for item in spawn_items: if item == "cmd": @@ -3476,13 +3504,34 @@ class AILauncher: return 1 # Label the new tmux window so the user can identify it. if item != "cmd": + win_name = item.capitalize() try: _win_backend._tmux_run( - ["rename-window", "-t", pane_id, item.capitalize()], + ["rename-window", "-t", pane_id, win_name], check=False, ) except Exception: pass + # Create linked session for this provider window. + try: + sn_cp = _win_backend._tmux_run( + ["display-message", "-p", "-t", pane_id, "#{session_name}"], + capture=True, + ) + main_sess = (sn_cp.stdout or "").strip() + if main_sess: + linked_name = f"{main_sess}-{win_name}" + _win_backend._tmux_run( + ["new-session", "-d", "-t", main_sess, "-s", linked_name], + check=True, capture=True, + ) + _win_backend._tmux_run( + ["select-window", "-t", f"{linked_name}:{win_name}"], + check=False, + ) + self.linked_sessions.append(linked_name) + except Exception: + pass # Also label the anchor window. try: _win_backend._tmux_run( @@ -3961,6 +4010,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: diff --git a/lib/terminal.py b/lib/terminal.py index f3d1ca24..58f1bbb8 100644 --- a/lib/terminal.py +++ b/lib/terminal.py @@ -562,7 +562,14 @@ def split_pane(self, parent_pane_id: str, direction: str, percent: int) -> str: return pane_id def new_window(self, session: str = "", window_name: str = "") -> str: - """Create a new tmux window and return its pane ID.""" + """Create a new tmux window and return its pane ID. + + When *window_name* is provided the method also creates a **linked + tmux session** named ``{main_session}-{window_name}`` so that each + provider window can be independently attached from another terminal. + Linked-session creation is best-effort -- failures are logged but do + not prevent the window itself from being used. + """ tmux_args = ["new-window", "-P", "-F", "#{pane_id}"] if session: tmux_args.extend(["-t", session]) @@ -583,8 +590,53 @@ def new_window(self, session: str = "", window_name: str = "") -> str: pane_id = (cp.stdout or "").strip() if not self._looks_like_pane_id(pane_id): return "" + + # --- Linked session creation (best-effort) --- + if window_name and pane_id: + try: + # Discover the main session name from the newly created pane. + sn_cp = self._tmux_run( + ["display-message", "-p", "-t", pane_id, "#{session_name}"], + capture=True, + ) + main_session = (sn_cp.stdout or "").strip() + if main_session: + linked_name = f"{main_session}-{window_name}" + # Create a linked session (shares the same window group). + self._tmux_run( + ["new-session", "-d", "-t", main_session, "-s", linked_name], + check=True, + capture=True, + ) + # Switch the linked session to the correct window. + self._tmux_run( + ["select-window", "-t", f"{linked_name}:{window_name}"], + check=False, + ) + except Exception: + import sys + print( + f"tmux linked-session creation failed for window {window_name!r} (non-fatal)", + file=sys.stderr, + ) + 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 diff --git a/test/test_windows_layout.py b/test/test_windows_layout.py index 1842caa3..854b19c6 100644 --- a/test/test_windows_layout.py +++ b/test/test_windows_layout.py @@ -230,14 +230,26 @@ def fake_tmux_run( timeout: float | None = None, ) -> subprocess.CompletedProcess[str]: calls.append(args) - return _cp(stdout="%77\n") + # new-window + if args and args[0] == "new-window": + return _cp(stdout="%77\n") + # session_name lookup for linked session creation + if len(args) >= 4 and "#{session_name}" in args: + return _cp(stdout="mysess\n") + # new-session (linked) + if args and args[0] == "new-session": + return _cp() + # select-window + if args and args[0] == "select-window": + return _cp() + 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" - assert len(calls) == 1 + # First call should be new-window. argv = calls[0] assert argv[0] == "new-window" assert "-P" in argv @@ -261,6 +273,8 @@ def fake_tmux_run( 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 @@ -279,6 +293,133 @@ def fake_tmux_run( pane_id = backend.new_window(session="s") assert pane_id == "" + def test_new_window_creates_linked_session(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Verify that new_window() with a window_name creates a linked session.""" + 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}) + # new-window + if args and args[0] == "new-window": + return _cp(stdout="%42\n") + # session_name lookup + if len(args) >= 4 and "#{session_name}" in args: + return _cp(stdout="main-sess\n") + # new-session (linked) + if args and args[0] == "new-session": + return _cp() + # select-window + if args and args[0] == "select-window": + return _cp() + 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 have: new-window, display-message (session_name), new-session, select-window + new_session_calls = [c for c in calls if c["args"] and c["args"][0] == "new-session"] + assert len(new_session_calls) == 1 + ns_args = new_session_calls[0]["args"] + assert "-d" in ns_args + assert "-t" in ns_args and "main-sess" in ns_args + assert "-s" in ns_args and "main-sess-Codex" in ns_args + + # select-window should target the linked session's window + sw_calls = [c for c in calls if c["args"] and c["args"][0] == "select-window"] + assert len(sw_calls) == 1 + assert "main-sess-Codex:Codex" in sw_calls[0]["args"] + + def test_new_window_linked_session_failure_nonfatal(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Linked session failure should not prevent new_window from returning the pane_id.""" + 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="%50\n") + # Fail the linked session creation + if args and args[0] == "new-session": + raise subprocess.CalledProcessError(1, ["tmux", *args]) + if len(args) >= 4 and "#{session_name}" in args: + return _cp(stdout="sess\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="sess", window_name="Claude") + # The window pane_id should still be returned. + assert pane_id == "%50" + + +# --------------------------------------------------------------------------- +# 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 From 74694f2e16d43876c431eb9e7651d9402b402cf1 Mon Sep 17 00:00:00 2001 From: Guolong Date: Mon, 16 Mar 2026 19:50:23 +0000 Subject: [PATCH 07/11] fix: create linked session for anchor provider in windows mode The anchor provider runs in-place and was skipped by the linked session creation loop. Now the anchor window also gets its own linked session so all providers are independently attachable. Co-Authored-By: Claude Opus 4.6 (1M context) --- ccb | 24 ++++++++++++++++++++++-- 1 file changed, 22 insertions(+), 2 deletions(-) diff --git a/ccb b/ccb index f310040e..32dfc60c 100755 --- a/ccb +++ b/ccb @@ -3532,14 +3532,34 @@ class AILauncher: self.linked_sessions.append(linked_name) except Exception: pass - # Also label the anchor window. + # Also label the anchor window and create its linked session. + anchor_win = self.anchor_provider.capitalize() try: _win_backend._tmux_run( - ["rename-window", "-t", self.anchor_pane_id, self.anchor_provider.capitalize()], + ["rename-window", "-t", self.anchor_pane_id, anchor_win], check=False, ) except Exception: pass + try: + sn_cp = _win_backend._tmux_run( + ["display-message", "-p", "-t", self.anchor_pane_id, "#{session_name}"], + capture=True, + ) + main_sess = (sn_cp.stdout or "").strip() + if main_sess: + linked_name = f"{main_sess}-{anchor_win}" + _win_backend._tmux_run( + ["new-session", "-d", "-t", main_sess, "-s", linked_name], + check=True, capture=True, + ) + _win_backend._tmux_run( + ["select-window", "-t", f"{linked_name}:{anchor_win}"], + check=False, + ) + self.linked_sessions.append(linked_name) + except Exception: + pass else: right_top: str | None = None if right_items: From e6ff70137a20e774e89fb164a24949d6375d4c3c Mon Sep 17 00:00:00 2001 From: Guolong Date: Mon, 16 Mar 2026 19:55:35 +0000 Subject: [PATCH 08/11] fix: capture main session name once before creating linked sessions tmux session groups cause #{session_name} to return a linked session name instead of the original after the first linked session is created. This caused chained names like 19-Claude-Codex instead of 19-Codex. Fix: query #{session_name} once before any linked sessions exist, then reuse that value for all providers. Move linked session creation out of new_window() into run_up() where the pre-captured name is available. Co-Authored-By: Claude Opus 4.6 (1M context) --- ccb | 46 ++++++++++++++++--------------- lib/terminal.py | 39 ++------------------------ test/test_windows_layout.py | 55 ++++--------------------------------- 3 files changed, 33 insertions(+), 107 deletions(-) diff --git a/ccb b/ccb index 32dfc60c..cfa851db 100755 --- a/ccb +++ b/ccb @@ -3495,6 +3495,18 @@ class AILauncher: # After creating each window we also create a linked tmux session so # each provider can be independently attached from another terminal. _win_backend = TmuxBackend() + # Capture main session name ONCE before creating any linked sessions, + # because linked sessions join the session group and later queries + # of #{session_name} may return a linked name instead of the original. + _main_sess = "" + try: + _ms_cp = _win_backend._tmux_run( + ["display-message", "-p", "-t", self.anchor_pane_id, "#{session_name}"], + capture=True, + ) + _main_sess = (_ms_cp.stdout or "").strip() + except Exception: + pass for item in spawn_items: if item == "cmd": pane_id = _start_item(item, parent=self.anchor_pane_id, direction="right") @@ -3513,16 +3525,11 @@ class AILauncher: except Exception: pass # Create linked session for this provider window. - try: - sn_cp = _win_backend._tmux_run( - ["display-message", "-p", "-t", pane_id, "#{session_name}"], - capture=True, - ) - main_sess = (sn_cp.stdout or "").strip() - if main_sess: - linked_name = f"{main_sess}-{win_name}" + if _main_sess: + try: + linked_name = f"{_main_sess}-{win_name}" _win_backend._tmux_run( - ["new-session", "-d", "-t", main_sess, "-s", linked_name], + ["new-session", "-d", "-t", _main_sess, "-s", linked_name], check=True, capture=True, ) _win_backend._tmux_run( @@ -3530,8 +3537,8 @@ class AILauncher: check=False, ) self.linked_sessions.append(linked_name) - except Exception: - pass + except Exception: + pass # Also label the anchor window and create its linked session. anchor_win = self.anchor_provider.capitalize() try: @@ -3541,16 +3548,11 @@ class AILauncher: ) except Exception: pass - try: - sn_cp = _win_backend._tmux_run( - ["display-message", "-p", "-t", self.anchor_pane_id, "#{session_name}"], - capture=True, - ) - main_sess = (sn_cp.stdout or "").strip() - if main_sess: - linked_name = f"{main_sess}-{anchor_win}" + if _main_sess: + try: + linked_name = f"{_main_sess}-{anchor_win}" _win_backend._tmux_run( - ["new-session", "-d", "-t", main_sess, "-s", linked_name], + ["new-session", "-d", "-t", _main_sess, "-s", linked_name], check=True, capture=True, ) _win_backend._tmux_run( @@ -3558,8 +3560,8 @@ class AILauncher: check=False, ) self.linked_sessions.append(linked_name) - except Exception: - pass + except Exception: + pass else: right_top: str | None = None if right_items: diff --git a/lib/terminal.py b/lib/terminal.py index 58f1bbb8..6c4d9eb7 100644 --- a/lib/terminal.py +++ b/lib/terminal.py @@ -564,11 +564,9 @@ def split_pane(self, parent_pane_id: str, direction: str, percent: int) -> str: def new_window(self, session: str = "", window_name: str = "") -> str: """Create a new tmux window and return its pane ID. - When *window_name* is provided the method also creates a **linked - tmux session** named ``{main_session}-{window_name}`` so that each - provider window can be independently attached from another terminal. - Linked-session creation is best-effort -- failures are logged but do - not prevent the window itself from being used. + Linked session creation is handled by the caller (run_up) to ensure + the main session name is captured once before any linked sessions + pollute the session group. """ tmux_args = ["new-window", "-P", "-F", "#{pane_id}"] if session: @@ -581,7 +579,6 @@ def new_window(self, session: str = "", window_name: str = "") -> str: err = (getattr(e, "stderr", "") or "").strip() out = (getattr(e, "stdout", "") or "").strip() msg = err or out - # Log error and return empty string, matching split_pane error style import sys print(f"tmux new-window failed (exit {e.returncode}): {msg or 'no stdout/stderr'}", file=sys.stderr) return "" @@ -590,36 +587,6 @@ def new_window(self, session: str = "", window_name: str = "") -> str: pane_id = (cp.stdout or "").strip() if not self._looks_like_pane_id(pane_id): return "" - - # --- Linked session creation (best-effort) --- - if window_name and pane_id: - try: - # Discover the main session name from the newly created pane. - sn_cp = self._tmux_run( - ["display-message", "-p", "-t", pane_id, "#{session_name}"], - capture=True, - ) - main_session = (sn_cp.stdout or "").strip() - if main_session: - linked_name = f"{main_session}-{window_name}" - # Create a linked session (shares the same window group). - self._tmux_run( - ["new-session", "-d", "-t", main_session, "-s", linked_name], - check=True, - capture=True, - ) - # Switch the linked session to the correct window. - self._tmux_run( - ["select-window", "-t", f"{linked_name}:{window_name}"], - check=False, - ) - except Exception: - import sys - print( - f"tmux linked-session creation failed for window {window_name!r} (non-fatal)", - file=sys.stderr, - ) - return pane_id def destroy_linked_session(self, session_name: str) -> bool: diff --git a/test/test_windows_layout.py b/test/test_windows_layout.py index 854b19c6..a610f840 100644 --- a/test/test_windows_layout.py +++ b/test/test_windows_layout.py @@ -293,8 +293,8 @@ def fake_tmux_run( pane_id = backend.new_window(session="s") assert pane_id == "" - def test_new_window_creates_linked_session(self, monkeypatch: pytest.MonkeyPatch) -> None: - """Verify that new_window() with a window_name creates a linked session.""" + 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( @@ -303,18 +303,8 @@ def fake_tmux_run( timeout: float | None = None, ) -> subprocess.CompletedProcess[str]: calls.append({"args": args, "check": check}) - # new-window if args and args[0] == "new-window": return _cp(stdout="%42\n") - # session_name lookup - if len(args) >= 4 and "#{session_name}" in args: - return _cp(stdout="main-sess\n") - # new-session (linked) - if args and args[0] == "new-session": - return _cp() - # select-window - if args and args[0] == "select-window": - return _cp() return _cp(stdout="") backend = terminal.TmuxBackend() @@ -323,44 +313,11 @@ def fake_tmux_run( pane_id = backend.new_window(session="main-sess", window_name="Codex") assert pane_id == "%42" - # Should have: new-window, display-message (session_name), new-session, select-window + # 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) == 1 - ns_args = new_session_calls[0]["args"] - assert "-d" in ns_args - assert "-t" in ns_args and "main-sess" in ns_args - assert "-s" in ns_args and "main-sess-Codex" in ns_args - - # select-window should target the linked session's window - sw_calls = [c for c in calls if c["args"] and c["args"][0] == "select-window"] - assert len(sw_calls) == 1 - assert "main-sess-Codex:Codex" in sw_calls[0]["args"] - - def test_new_window_linked_session_failure_nonfatal(self, monkeypatch: pytest.MonkeyPatch) -> None: - """Linked session failure should not prevent new_window from returning the pane_id.""" - 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="%50\n") - # Fail the linked session creation - if args and args[0] == "new-session": - raise subprocess.CalledProcessError(1, ["tmux", *args]) - if len(args) >= 4 and "#{session_name}" in args: - return _cp(stdout="sess\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="sess", window_name="Claude") - # The window pane_id should still be returned. - assert pane_id == "%50" + 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 # --------------------------------------------------------------------------- From dc468846a553c0415453edf28a4b36c3cc8a42cc Mon Sep 17 00:00:00 2001 From: Guolong Date: Mon, 16 Mar 2026 19:56:55 +0000 Subject: [PATCH 09/11] test: clean up dead stubs in test_new_window_returns_pane_id Remove session_name/new-session/select-window stubs that were added for linked session creation which has since been moved out of new_window(). Co-Authored-By: Claude Opus 4.6 (1M context) --- test/test_windows_layout.py | 10 ---------- 1 file changed, 10 deletions(-) diff --git a/test/test_windows_layout.py b/test/test_windows_layout.py index a610f840..2acef3a0 100644 --- a/test/test_windows_layout.py +++ b/test/test_windows_layout.py @@ -230,18 +230,8 @@ def fake_tmux_run( timeout: float | None = None, ) -> subprocess.CompletedProcess[str]: calls.append(args) - # new-window if args and args[0] == "new-window": return _cp(stdout="%77\n") - # session_name lookup for linked session creation - if len(args) >= 4 and "#{session_name}" in args: - return _cp(stdout="mysess\n") - # new-session (linked) - if args and args[0] == "new-session": - return _cp() - # select-window - if args and args[0] == "select-window": - return _cp() return _cp(stdout="") backend = terminal.TmuxBackend() From 30d11e3b6816144b9dd3dadd6345e2bed6f4629f Mon Sep 17 00:00:00 2001 From: Guolong Date: Mon, 16 Mar 2026 20:08:25 +0000 Subject: [PATCH 10/11] refactor: deduplicate layout_mode threading and registry boilerplate - Remove layout_mode parameter from 7 _start_* method signatures; read self.layout_mode directly (cmd pane hardcodes "panes") - Extract _create_linked_session() helper to deduplicate the new-session + select-window + append logic in run_up() - Extract _registry_base() helper to deduplicate the 6 identical registry dict constructions across _write_*_session methods No logic changes -- pure structural cleanup. 292 tests pass. Co-Authored-By: Claude Opus 4.6 (1M context) --- ccb | 232 ++++++++++++++++++++++++------------------------------------ 1 file changed, 92 insertions(+), 140 deletions(-) diff --git a/ccb b/ccb index cfa851db..2b004903 100755 --- a/ccb +++ b/ccb @@ -1184,7 +1184,7 @@ class AILauncher: ) return True - def _start_provider(self, provider: str, *, parent_pane: str | None = None, direction: str | None = None, layout_mode: str = "panes") -> str | None: + def _start_provider(self, provider: str, *, parent_pane: str | None = None, direction: str | None = None) -> str | None: # Handle case when no terminal detected if self.terminal_type is None: print(f"❌ {t('no_terminal_backend')}") @@ -1223,13 +1223,13 @@ class AILauncher: print(f"🚀 {t('starting_backend', provider=provider.capitalize(), terminal='tmux')}") if provider == "codex": - return self._start_codex_tmux(parent_pane=parent_pane, direction=direction, layout_mode=layout_mode) + return self._start_codex_tmux(parent_pane=parent_pane, direction=direction) elif provider == "gemini": - return self._start_gemini_tmux(parent_pane=parent_pane, direction=direction, layout_mode=layout_mode) + return self._start_gemini_tmux(parent_pane=parent_pane, direction=direction) elif provider == "opencode": - return self._start_opencode_tmux(parent_pane=parent_pane, direction=direction, layout_mode=layout_mode) + return self._start_opencode_tmux(parent_pane=parent_pane, direction=direction) elif provider == "droid": - return self._start_droid_tmux(parent_pane=parent_pane, direction=direction, layout_mode=layout_mode) + return self._start_droid_tmux(parent_pane=parent_pane, direction=direction) else: print(f"❌ {t('unknown_provider', provider=provider)}") return None @@ -1446,24 +1446,16 @@ class AILauncher: return if pane_id: try: - _reg = { - "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, - "layout_mode": self.layout_mode, - "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"), + } } - if self.linked_sessions: - _reg["linked_sessions"] = list(self.linked_sessions) upsert_registry(_reg) except Exception: pass @@ -2101,7 +2093,6 @@ class AILauncher: *, parent_pane: str | None = None, direction: str | None = None, - layout_mode: str = "panes", ) -> str | None: runtime = self.runtime_dir / "codex" runtime.mkdir(parents=True, exist_ok=True) @@ -2138,7 +2129,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, layout_mode=layout_mode) + 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") @@ -2201,7 +2192,6 @@ class AILauncher: *, parent_pane: str | None = None, direction: str | None = None, - layout_mode: str = "panes", ) -> str | None: runtime = self.runtime_dir / "gemini" runtime.mkdir(parents=True, exist_ok=True) @@ -2230,7 +2220,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, layout_mode=layout_mode) + 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") @@ -2253,7 +2243,6 @@ class AILauncher: *, parent_pane: str | None = None, direction: str | None = None, - layout_mode: str = "panes", ) -> str | None: runtime = self.runtime_dir / "opencode" runtime.mkdir(parents=True, exist_ok=True) @@ -2285,7 +2274,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, layout_mode=layout_mode) + 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") @@ -2308,7 +2297,6 @@ class AILauncher: *, parent_pane: str | None = None, direction: str | None = None, - layout_mode: str = "panes", ) -> str | None: runtime = self.runtime_dir / "droid" runtime.mkdir(parents=True, exist_ok=True) @@ -2338,7 +2326,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, layout_mode=layout_mode) + 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") @@ -2362,7 +2350,6 @@ class AILauncher: parent_pane: str | None, direction: str | None, cmd_settings: dict, - layout_mode: str = "panes", ) -> str | None: if not cmd_settings.get("enabled"): return None @@ -2385,7 +2372,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, layout_mode=layout_mode) + 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") @@ -2655,6 +2642,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.linked_sessions: + base["linked_sessions"] = list(self.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") @@ -2699,22 +2699,14 @@ class AILauncher: print(err, file=sys.stderr) return False try: - _reg = { - "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, - "layout_mode": self.layout_mode, - "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), + } } - if self.linked_sessions: - _reg["linked_sessions"] = list(self.linked_sessions) upsert_registry(_reg) except Exception: pass @@ -2724,21 +2716,13 @@ 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, - "layout_mode": self.layout_mode, - "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}, } - if self.linked_sessions: - record["linked_sessions"] = list(self.linked_sessions) ok = upsert_registry(record) if not ok: print("⚠️ Failed to update cpend registry", file=sys.stderr) @@ -2783,22 +2767,14 @@ class AILauncher: print(err, file=sys.stderr) return False try: - _reg = { - "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, - "layout_mode": self.layout_mode, - "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), + } } - if self.linked_sessions: - _reg["linked_sessions"] = list(self.linked_sessions) upsert_registry(_reg) except Exception: pass @@ -2836,22 +2812,14 @@ class AILauncher: print(err, file=sys.stderr) return False try: - _reg = { - "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, - "layout_mode": self.layout_mode, - "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), + } } - if self.linked_sessions: - _reg["linked_sessions"] = list(self.linked_sessions) upsert_registry(_reg) except Exception: pass @@ -2906,24 +2874,16 @@ class AILauncher: print(err, file=sys.stderr) return False try: - _reg = { - "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, - "layout_mode": self.layout_mode, - "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, + } } - if self.linked_sessions: - _reg["linked_sessions"] = list(self.linked_sessions) upsert_registry(_reg) except Exception: pass @@ -3161,7 +3121,7 @@ class AILauncher: print(f"\n⚠️ {t('user_interrupted')}") return 130 - def _start_claude_pane(self, *, parent_pane: str | None, direction: str | None, layout_mode: str = "panes") -> str | None: + def _start_claude_pane(self, *, parent_pane: str | None, direction: str | None) -> str | None: print(f"🚀 {t('starting_claude')}") env_overrides = self._claude_env_overrides() @@ -3194,7 +3154,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, layout_mode=layout_mode) + 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") @@ -3349,6 +3309,24 @@ class AILauncher: if not quiet: print(f"✅ {t('cleanup_complete')}") + def _create_linked_session(self, backend, main_sess: str, win_name: str) -> None: + """Create a linked tmux session for a provider window (best-effort).""" + if not main_sess: + return + try: + linked_name = f"{main_sess}-{win_name}" + backend._tmux_run( + ["new-session", "-d", "-t", main_sess, "-s", linked_name], + check=True, capture=True, + ) + backend._tmux_run( + ["select-window", "-t", f"{linked_name}:{win_name}"], + check=False, + ) + self.linked_sessions.append(linked_name) + except Exception: + pass + def run_up(self) -> int: git_info = _get_git_info() version_str = f"v{VERSION}" + (f" ({git_info})" if git_info else "") @@ -3483,8 +3461,8 @@ class AILauncher: # 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, layout_mode=self.layout_mode) - pane_id = self._start_provider(item, parent_pane=parent, direction=direction, layout_mode=self.layout_mode) + return self._start_claude_pane(parent_pane=parent, direction=direction) + pane_id = self._start_provider(item, parent_pane=parent, direction=direction) if pane_id: self._warmup_provider(item) return pane_id @@ -3525,20 +3503,7 @@ class AILauncher: except Exception: pass # Create linked session for this provider window. - if _main_sess: - try: - linked_name = f"{_main_sess}-{win_name}" - _win_backend._tmux_run( - ["new-session", "-d", "-t", _main_sess, "-s", linked_name], - check=True, capture=True, - ) - _win_backend._tmux_run( - ["select-window", "-t", f"{linked_name}:{win_name}"], - check=False, - ) - self.linked_sessions.append(linked_name) - except Exception: - pass + self._create_linked_session(_win_backend, _main_sess, win_name) # Also label the anchor window and create its linked session. anchor_win = self.anchor_provider.capitalize() try: @@ -3548,20 +3513,7 @@ class AILauncher: ) except Exception: pass - if _main_sess: - try: - linked_name = f"{_main_sess}-{anchor_win}" - _win_backend._tmux_run( - ["new-session", "-d", "-t", _main_sess, "-s", linked_name], - check=True, capture=True, - ) - _win_backend._tmux_run( - ["select-window", "-t", f"{linked_name}:{anchor_win}"], - check=False, - ) - self.linked_sessions.append(linked_name) - except Exception: - pass + self._create_linked_session(_win_backend, _main_sess, anchor_win) else: right_top: str | None = None if right_items: From 084eab51b897646f7e27bc4b4c520fe4800e4d9b Mon Sep 17 00:00:00 2001 From: Guolong Date: Mon, 16 Mar 2026 20:21:53 +0000 Subject: [PATCH 11/11] refactor: extract layout strategy pattern for windows/panes placement Move the orchestration logic from the big if/else in run_up() into PanesLayout and WindowsLayout strategy classes. Extract get_session_name, rename_window, and create_linked_session as proper TmuxBackend methods instead of raw _tmux_run calls in the main script. Linked session lifecycle is now encapsulated in WindowsLayout. Co-Authored-By: Claude Opus 4.6 (1M context) --- ccb | 110 +++------------- lib/layout.py | 131 ++++++++++++++++++ lib/terminal.py | 49 ++++++- test/test_windows_layout.py | 255 ++++++++++++++++++++++++++++++++++++ 4 files changed, 446 insertions(+), 99 deletions(-) create mode 100644 lib/layout.py diff --git a/ccb b/ccb index 2b004903..b8f59abb 100755 --- a/ccb +++ b/ccb @@ -39,6 +39,7 @@ from session_utils import ( safe_write_session, ) 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 @@ -589,7 +590,7 @@ class AILauncher: self.tmux_panes = {} self.wezterm_panes = {} self.extra_panes = {} - self.linked_sessions: list[str] = [] + self.layout_strategy = None # set in run_up() self.processes = {} self.anchor_provider = None self.anchor_pane_id = None @@ -2651,8 +2652,8 @@ class AILauncher: "terminal": terminal if terminal is not None else self.terminal_type, "layout_mode": self.layout_mode, } - if self.linked_sessions: - base["linked_sessions"] = list(self.linked_sessions) + 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): @@ -3216,15 +3217,12 @@ class AILauncher: except Exception: pass - # Kill linked tmux sessions (windows mode) before killing panes. - if kill_panes and self.linked_sessions: - _lb = TmuxBackend() - for ls_name in self.linked_sessions: - try: - _lb.destroy_linked_session(ls_name) - except Exception: - pass - self.linked_sessions.clear() + # 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": @@ -3309,24 +3307,6 @@ class AILauncher: if not quiet: print(f"✅ {t('cleanup_complete')}") - def _create_linked_session(self, backend, main_sess: str, win_name: str) -> None: - """Create a linked tmux session for a provider window (best-effort).""" - if not main_sess: - return - try: - linked_name = f"{main_sess}-{win_name}" - backend._tmux_run( - ["new-session", "-d", "-t", main_sess, "-s", linked_name], - check=True, capture=True, - ) - backend._tmux_run( - ["select-window", "-t", f"{linked_name}:{win_name}"], - check=False, - ) - self.linked_sessions.append(linked_name) - except Exception: - pass - def run_up(self) -> int: git_info = _get_git_info() version_str = f"v{VERSION}" + (f" ({git_info})" if git_info else "") @@ -3455,7 +3435,7 @@ 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. @@ -3468,72 +3448,14 @@ class AILauncher: return pane_id if self.layout_mode == "windows": - # Windows mode: each non-anchor provider gets its own tmux window. - # The "cmd" item splits inside the anchor window (panes behaviour). - # After creating each window we also create a linked tmux session so - # each provider can be independently attached from another terminal. _win_backend = TmuxBackend() - # Capture main session name ONCE before creating any linked sessions, - # because linked sessions join the session group and later queries - # of #{session_name} may return a linked name instead of the original. - _main_sess = "" - try: - _ms_cp = _win_backend._tmux_run( - ["display-message", "-p", "-t", self.anchor_pane_id, "#{session_name}"], - capture=True, - ) - _main_sess = (_ms_cp.stdout or "").strip() - except Exception: - pass - for item in spawn_items: - if item == "cmd": - pane_id = _start_item(item, parent=self.anchor_pane_id, direction="right") - else: - pane_id = _start_item(item, parent=self.anchor_pane_id, direction=None) - if not pane_id: - return 1 - # Label the new tmux window so the user can identify it. - if item != "cmd": - win_name = item.capitalize() - try: - _win_backend._tmux_run( - ["rename-window", "-t", pane_id, win_name], - check=False, - ) - except Exception: - pass - # Create linked session for this provider window. - self._create_linked_session(_win_backend, _main_sess, win_name) - # Also label the anchor window and create its linked session. - anchor_win = self.anchor_provider.capitalize() - try: - _win_backend._tmux_run( - ["rename-window", "-t", self.anchor_pane_id, anchor_win], - check=False, - ) - except Exception: - pass - self._create_linked_session(_win_backend, _main_sess, anchor_win) + self.layout_strategy = WindowsLayout(_win_backend, self.anchor_pane_id, self.anchor_provider) else: - 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 + 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": 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/terminal.py b/lib/terminal.py index 6c4d9eb7..1a4c7d29 100644 --- a/lib/terminal.py +++ b/lib/terminal.py @@ -561,13 +561,52 @@ 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 new_window(self, session: str = "", window_name: str = "") -> str: - """Create a new tmux window and return its 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*. - Linked session creation is handled by the caller (run_up) to ensure - the main session name is captured once before any linked sessions - pollute the session group. + 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]) diff --git a/test/test_windows_layout.py b/test/test_windows_layout.py index 2acef3a0..b2a9ac13 100644 --- a/test/test_windows_layout.py +++ b/test/test_windows_layout.py @@ -10,6 +10,7 @@ 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 @@ -310,6 +311,260 @@ def fake_tmux_run( 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 # ---------------------------------------------------------------------------