ccb

#!/usr/bin/env python3
"""
ccb (Claude Code Bridge) - Unified AI Launcher
Supports Claude + Codex / Claude + Gemini / all three simultaneously
Supports tmux and WezTerm
"""

import sys
import os
import json
import time
import subprocess
import signal
import atexit
import argparse
import uuid
import getpass
import platform
import tempfile
import re
import shutil
import posixpath
import shlex
import threading
from pathlib import Path

script_dir = Path(__file__).resolve().parent
sys.path.insert(0, str(script_dir / "lib"))
from terminal import TmuxBackend, WeztermBackend, detect_terminal, is_wsl, get_shell_type, get_backend_for_session
from compat import setup_windows_encoding
from ccb_config import get_backend_env
from ccb_start_config import DEFAULT_PROVIDERS, ensure_default_start_config, load_start_config
from session_utils import (
    check_session_writable,
    find_project_session_file,
    legacy_project_config_dir,
    project_config_dir,
    resolve_project_config_dir,
    safe_write_session,
)
from pane_registry import upsert_registry, load_registry_by_project_id
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
from askd_rpc import shutdown_daemon, read_state
from askd_runtime import state_file_path
from i18n import t

setup_windows_encoding()

backend_env = get_backend_env()
if backend_env and not os.environ.get("CCB_BACKEND_ENV"):
    os.environ["CCB_BACKEND_ENV"] = backend_env

VERSION = "5.2.8"
GIT_COMMIT = "c539e79"
GIT_DATE = "2026-02-25"

_WIN_DRIVE_RE = re.compile(r"^[A-Za-z]:([/\\\\]|$)")
_MNT_DRIVE_RE = re.compile(r"^/mnt/([A-Za-z])/(.*)$")
_MSYS_DRIVE_RE = re.compile(r"^/([A-Za-z])/(.*)$")


def _looks_like_windows_path(value: str) -> bool:
    s = value.strip()
    if not s:
        return False
    if _WIN_DRIVE_RE.match(s):
        return True
    if s.startswith("\\\\") or s.startswith("//"):
        return True
    return False


def _normalize_path_for_match(value: str) -> str:
    """
    Normalize a path-like string for loose matching across Windows/WSL/MSYS variations.
    This is used only for selecting a session for *current* cwd, so favor robustness.
    """
    s = (value or "").strip()
    if not s:
        return ""

    # Expand "~" early (common in shell-originated values). If expansion fails, keep original.
    if s.startswith("~"):
        try:
            s = os.path.expanduser(s)
        except Exception:
            pass

    # If the path is relative, absolutize it against current cwd for matching purposes only.
    # This reduces false negatives when upstream tools record a relative cwd.
    # NOTE: treat Windows-like absolute paths as absolute even on non-Windows hosts.
    try:
        preview = s.replace("\\", "/")
        is_abs = (
            preview.startswith("/")
            or preview.startswith("//")
            or bool(_WIN_DRIVE_RE.match(preview))
            or preview.startswith("\\\\")
        )
        if not is_abs:
            s = str((Path.cwd() / Path(s)).absolute())
    except Exception:
        pass

    s = s.replace("\\", "/")

    # Map WSL drive mount to Windows-style drive path for comparison.
    m = _MNT_DRIVE_RE.match(s)
    if m:
        drive = m.group(1).lower()
        rest = m.group(2)
        s = f"{drive}:/{rest}"
    else:
        # Map MSYS /c/... to c:/... (Git-Bash/MSYS2 environments on Windows).
        m = _MSYS_DRIVE_RE.match(s)
        if m and ("MSYSTEM" in os.environ or os.name == "nt"):
            drive = m.group(1).lower()
            rest = m.group(2)
            s = f"{drive}:/{rest}"

    # Collapse redundant separators and dot segments using POSIX semantics (we forced "/").
    # Preserve UNC double-slash prefix.
    if s.startswith("//"):
        prefix = "//"
        rest = s[2:]
        rest = posixpath.normpath(rest)
        s = prefix + rest.lstrip("/")
    else:
        s = posixpath.normpath(s)

    # Normalize Windows drive letter casing (c:/..., not C:/...).
    if _WIN_DRIVE_RE.match(s):
        s = s[0].lower() + s[1:]

    # Drop trailing slash (but keep "/" and "c:/").
    if len(s) > 1 and s.endswith("/"):
        s = s.rstrip("/")
        if _WIN_DRIVE_RE.match(s) and not s.endswith("/"):
            # Ensure drive root keeps trailing slash form "c:/".
            if len(s) == 2:
                s = s + "/"

    # On Windows-like paths, compare case-insensitively to avoid drive letter/case issues.
    if _looks_like_windows_path(s):
        s = s.casefold()

    return s


def _work_dir_match_keys(work_dir: Path) -> set[str]:
    keys: set[str] = set()
    candidates: list[str] = []
    for raw in (os.environ.get("PWD"), str(work_dir)):
        if raw:
            candidates.append(raw)
    try:
        candidates.append(str(work_dir.resolve()))
    except Exception:
        pass
    for candidate in candidates:
        normalized = _normalize_path_for_match(candidate)
        if normalized:
            keys.add(normalized)
    return keys


def _normpath_within(child_norm: str, parent_norm: str) -> bool:
    """
    Return True if normalized path `child_norm` is equal to or inside `parent_norm`.

    Both args must be normalized via `_normalize_path_for_match` (or equivalent).
    """
    if not child_norm or not parent_norm:
        return False
    if child_norm == parent_norm:
        return True
    prefix = parent_norm if parent_norm.endswith("/") else (parent_norm + "/")
    return child_norm.startswith(prefix)


def _extract_session_work_dir_norm(session_data: dict) -> str:
    """Extract a normalized work dir marker from a session file payload."""
    if not isinstance(session_data, dict):
        return ""
    raw_norm = session_data.get("work_dir_norm")
    if isinstance(raw_norm, str) and raw_norm.strip():
        return _normalize_path_for_match(raw_norm)
    raw = session_data.get("work_dir")
    if isinstance(raw, str) and raw.strip():
        return _normalize_path_for_match(raw)
    return ""


def _get_git_info() -> str:
    try:
        result = subprocess.run(
            ["git", "-C", str(script_dir), "log", "-1", "--format=%h %ci"],
            capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=2
        )
        if result.returncode == 0:
            return result.stdout.strip()
    except Exception:
        pass
    return ""


def _build_keep_open_cmd(provider: str, start_cmd: str) -> str:
    if get_shell_type() == "powershell":
        return (
            f'{start_cmd}; '
            f'$code = $LASTEXITCODE; '
            f'Write-Host "`n[{provider}] exited with code $code. Press Enter to close..."; '
            f'Read-Host; '
            f'exit $code'
        )
    return (
        f'{start_cmd}; '
        f'code=$?; '
        f'echo; echo "[{provider}] exited with code $code. Press Enter to close..."; '
        f'read -r _; '
        f'exit $code'
    )


def _build_pane_title_cmd(marker: str) -> str:
    if get_shell_type() == "powershell":
        safe_sq = marker.replace("'", "''")
        safe_dq = marker.replace('"', '`"')
        return (
            "$esc=[char]27; "
            f"[Console]::Write(\"$esc]0;{safe_dq}`a\"); "
            f"$Host.UI.RawUI.WindowTitle = '{safe_sq}'; "
        )
    return f"printf '\\033]0;{marker}\\007'; "

def _build_export_path_cmd(bin_dir: Path) -> str:
    """
    Ensure CCB's `bin/` is available inside the started pane/session.

    This allows running `oask`/`gask` from within Codex/Gemini/OpenCode environments consistently
    across WezTerm/tmux (and PowerShell shells on Windows).
    """
    bin_s = str(bin_dir)
    if get_shell_type() == "powershell":
        safe = bin_s.replace("'", "''")
        # Prefer a fully materialized PATH to avoid cases where the spawned shell inherits a stale
        # PATH (e.g. tmux server started before Homebrew paths were available).
        current = (os.environ.get("PATH") or "").replace("'", "''")
        if current:
            return f"$env:Path = '{safe};{current}'; "
        return f"$env:Path = '{safe};' + $env:Path; "

    # Materialize PATH from the current CCB process instead of relying on the spawned shell's `$PATH`.
    # This fixes macOS/Homebrew setups where `python3` exists in the interactive shell, but a spawned
    # pane/session (tmux/Terminal.app) starts with a minimal PATH and `/usr/bin/env python3` fails.
    current = os.environ.get("PATH") or ""
    if current:
        return f"export PATH={shlex.quote(bin_s)}{os.pathsep}{shlex.quote(current)}; "
    return f"export PATH={shlex.quote(bin_s)}{os.pathsep}$PATH; "


def _build_cd_cmd(work_dir: Path) -> str:
    if get_shell_type() == "powershell":
        safe = str(work_dir).replace("'", "''")
        return f"Set-Location -Path '{safe}'; "
    return f"cd {shlex.quote(str(work_dir))}; "


def _env_bool(name: str, default: bool) -> bool:
    raw = os.environ.get(name)
    if raw is None or raw == "":
        return default
    v = raw.strip().lower()
    if v in {"1", "true", "yes", "on"}:
        return True
    if v in {"0", "false", "no", "off"}:
        return False
    return default


def _env_float(name: str, default: float) -> float:
    raw = os.environ.get(name)
    if raw is None or raw == "":
        return default
    try:
        return float(raw)
    except Exception:
        return default


def _cleanup_tmpclaude_artifacts() -> int:
    """
    Best-effort cleanup for leftover Claude temp markers like `tmpclaude-xxxx-cwd`.

    Deletion is conservative: only removes entries older than `CCB_TMPCLAUDE_MIN_AGE_S`.
    Controls:
      - `CCB_TMPCLAUDE_CLEAN` (default: true)
      - `CCB_TMPCLAUDE_CLEAN_CWD` (default: true)
      - `CCB_TMPCLAUDE_MIN_AGE_S` (default: 300)
      - `CCB_TMPCLAUDE_DIRS` (extra dirs, split by `os.pathsep`)
      - `CCB_TMPCLAUDE_PATTERNS` (comma-separated globs; default: tmpclaude-*-cwd)
    """
    if not _env_bool("CCB_TMPCLAUDE_CLEAN", True):
        return 0

    patterns_raw = (os.environ.get("CCB_TMPCLAUDE_PATTERNS") or "").strip()
    patterns = [p.strip() for p in patterns_raw.split(",") if p.strip()] if patterns_raw else ["tmpclaude-*-cwd"]

    min_age_s = max(0.0, float(_env_float("CCB_TMPCLAUDE_MIN_AGE_S", 300.0)))

    dirs: list[Path] = []
    if _env_bool("CCB_TMPCLAUDE_CLEAN_CWD", True):
        dirs.append(Path.cwd())
    try:
        dirs.append(Path(tempfile.gettempdir()))
    except Exception:
        pass

    extra = (os.environ.get("CCB_TMPCLAUDE_DIRS") or "").strip()
    if extra:
        for part in extra.split(os.pathsep):
            p = part.strip()
            if not p:
                continue
            try:
                dirs.append(Path(p).expanduser())
            except Exception:
                continue

    seen_dirs: set[str] = set()
    unique_dirs: list[Path] = []
    for d in dirs:
        key = str(d)
        if key in seen_dirs:
            continue
        seen_dirs.add(key)
        unique_dirs.append(d)

    now = time.time()
    removed = 0
    for base in unique_dirs:
        try:
            if not base.exists() or not base.is_dir():
                continue
        except Exception:
            continue
        for pat in patterns:
            try:
                candidates = list(base.glob(pat))
            except Exception:
                candidates = []
            for path in candidates:
                try:
                    st = path.stat()
                    if min_age_s and (now - float(st.st_mtime)) < min_age_s:
                        continue
                    if path.is_dir():
                        shutil.rmtree(path, ignore_errors=True)
                    else:
                        path.unlink(missing_ok=True)
                    removed += 1
                except Exception:
                    continue
    return removed


def _is_pid_alive(pid: int) -> bool:
    if pid <= 0:
        return False
    if os.name == "nt":
        try:
            import ctypes

            kernel32 = ctypes.windll.kernel32
            SYNCHRONIZE = 0x00100000
            handle = kernel32.OpenProcess(SYNCHRONIZE, False, int(pid))
            if handle:
                kernel32.CloseHandle(handle)
                return True
            return False
        except Exception:
            return True
    try:
        os.kill(int(pid), 0)
        return True
    except ProcessLookupError:
        # Process doesn't exist
        return False
    except PermissionError:
        # Process exists but no permission to check
        return True
    except Exception:
        # Other errors - assume dead for safety
        return False


def _runtime_base_dir() -> Path:
    try:
        base = Path(tempfile.gettempdir())
    except Exception:
        base = Path("/tmp")
    return base / f"claude-ai-{getpass.getuser()}"


def _cleanup_stale_runtime_dirs(*, exclude: Path | None = None) -> int:
    """
    Best-effort garbage collection for stale CCB runtime dirs under `$TMP/claude-ai-<user>/ai-*`.

    Normal exits already remove the current `runtime_dir`. This targets leftovers from crashes
    or hard kills (e.g. SIGKILL).
    """
    if not _env_bool("CCB_RUNTIME_GC", True):
        return 0

    min_age_s = max(0.0, float(_env_float("CCB_RUNTIME_GC_MIN_AGE_S", 24 * 3600.0)))
    base = _runtime_base_dir()
    try:
        if not base.exists() or not base.is_dir():
            return 0
    except Exception:
        return 0

    exclude_resolved: str | None = None
    if exclude is not None:
        try:
            exclude_resolved = str(Path(exclude).resolve())
        except Exception:
            exclude_resolved = str(exclude)

    now = time.time()
    removed = 0
    try:
        candidates = sorted(base.glob("ai-*"), key=lambda p: p.stat().st_mtime if p.exists() else 0.0)
    except Exception:
        candidates = []

    for session_dir in candidates:
        try:
            if not session_dir.is_dir():
                continue
        except Exception:
            continue

        try:
            if exclude_resolved and str(session_dir.resolve()) == exclude_resolved:
                continue
        except Exception:
            if exclude_resolved and str(session_dir) == exclude_resolved:
                continue

        try:
            st = session_dir.stat()
            if min_age_s and (now - float(st.st_mtime)) < min_age_s:
                continue
        except Exception:
            continue

        # If any recorded PID is alive, don't delete.
        alive = False
        try:
            for pid_file in session_dir.glob("**/*.pid"):
                try:
                    raw = pid_file.read_text(encoding="utf-8", errors="ignore").strip()
                    if raw.isdigit() and _is_pid_alive(int(raw)):
                        alive = True
                        break
                except Exception:
                    continue
        except Exception:
            pass
        if alive:
            continue

        try:
            shutil.rmtree(session_dir, ignore_errors=True)
            removed += 1
        except Exception:
            continue

    return removed


def _shrink_ccb_logs() -> int:
    """
    Best-effort log slimming for daemon logs to avoid disk bloat.

    - Current daemons log under `~/.cache/ccb` (or `$XDG_CACHE_HOME/ccb`).
    - Older installs may have left logs under `~/.ccb/run`.
    """
    if not _env_bool("CCB_LOG_SHRINK", True):
        return 0

    try:
        max_bytes = max(0, int(_env_float("CCB_LOG_MAX_BYTES", 2 * 1024 * 1024)))
    except Exception:
        max_bytes = 2 * 1024 * 1024
    if max_bytes <= 0:
        return 0

    # Cache directory (matches lib/askd_runtime.py defaults)
    cache_dir: Path | None = None
    xdg_cache = (os.environ.get("XDG_CACHE_HOME") or "").strip()
    if xdg_cache:
        cache_dir = Path(xdg_cache) / "ccb"
    else:
        cache_dir = Path.home() / ".cache" / "ccb"

    legacy_dir = Path.home() / ".ccb" / "run"

    def _shrink_file(path: Path) -> bool:
        try:
            if not path.exists() or not path.is_file():
                return False
            size = path.stat().st_size
            if size <= max_bytes:
                return False
            with path.open("rb") as handle:
                handle.seek(-max_bytes, os.SEEK_END)
                tail = handle.read()
            tmp = path.with_suffix(path.suffix + ".tmp")
            tmp.write_bytes(tail)
            os.replace(tmp, path)
            return True
        except Exception:
            try:
                tmp = path.with_suffix(path.suffix + ".tmp")
                if tmp.exists():
                    tmp.unlink()
            except Exception:
                pass
            return False

    removed = 0
    for base in (cache_dir, legacy_dir):
        try:
            if not base.exists() or not base.is_dir():
                continue
        except Exception:
            continue
        try:
            for log_file in base.glob("*.log"):
                if _shrink_file(log_file):
                    removed += 1
        except Exception:
            continue
    return removed


class AILauncher:
    def __init__(
        self,
        providers: list,
        resume: bool = False,
        auto: bool = False,
        cmd_config: dict | None = None,
    ):
        self.providers = providers or ["codex"]
        self.resume = resume
        self.auto = auto
        self.cmd_config = self._normalize_cmd_config(cmd_config)
        self.script_dir = Path(__file__).resolve().parent
        self.invocation_dir = Path.cwd()
        # Project root is strictly the current working directory.
        # Do NOT traverse upwards to infer a different root.
        try:
            self.project_root = self.invocation_dir.resolve()
        except Exception:
            self.project_root = self.invocation_dir.absolute()
        self.session_id = f"ai-{int(time.time())}-{os.getpid()}"
        self.ccb_pid = os.getpid()
        self.project_id = compute_ccb_project_id(self.project_root)
        project_hash = (self.project_id or "")[:16] or "unknown"
        self.project_run_dir = (Path.home() / ".cache" / "ccb" / "projects" / project_hash)
        self.temp_base = Path(tempfile.gettempdir())
        self.runtime_dir = self.temp_base / f"claude-ai-{getpass.getuser()}" / self.session_id
        self.runtime_dir.mkdir(parents=True, exist_ok=True)
        self._cleaned = False
        self._askd_checked = False
        self._watchdog_thread = None
        self._watchdog_stop_event = None
        self._daemon_proc = None  # Track daemon Popen object for reaping
        self._daemon_proc_lock = threading.Lock()  # Protect daemon_proc access
        self.terminal_type = self._detect_terminal_type()
        self.tmux_sessions = {}
        self.tmux_panes = {}
        self.wezterm_panes = {}
        self.extra_panes = {}
        self.processes = {}
        self.anchor_provider = None
        self.anchor_pane_id = None
        self._migrate_legacy_project_files()
        os.environ["CCB_MANAGED"] = "1"
        os.environ["CCB_PARENT_PID"] = str(self.ccb_pid)
        os.environ.setdefault("CCB_RUN_DIR", str(self.project_run_dir))

    def _managed_env_overrides(self) -> dict:
        env = {
            "CCB_MANAGED": "1",
            "CCB_PARENT_PID": str(self.ccb_pid),
        }
        if os.environ.get("CCB_RUN_DIR"):
            env["CCB_RUN_DIR"] = os.environ["CCB_RUN_DIR"]
        return env

    def _provider_env_overrides(self, provider: str) -> dict:
        """Managed env + explicit caller marker for the pane/provider process."""
        env = self._managed_env_overrides()
        prov = (provider or "").strip().lower()
        if prov in {"claude", "codex", "gemini", "opencode", "droid", "email", "manual"}:
            env["CCB_CALLER"] = prov
        return env

    def _project_config_dir(self) -> Path:
        return resolve_project_config_dir(self.project_root)

    def _project_session_file(self, filename: str) -> Path:
        cfg = self._project_config_dir()
        return cfg / filename

    def _migrate_legacy_project_files(self) -> None:
        """
        Move legacy project dotfiles from the project root into the project config dir
        (`.ccb/` or legacy `.ccb_config/`).

        This keeps the project root clean while preserving backwards-compatible lookup
        (see `lib/session_utils.py:find_project_session_file`).
        """
        cfg = self._project_config_dir()
        if not cfg.is_dir():
            return

        for name in (".codex-session", ".gemini-session", ".opencode-session", ".claude-session", ".droid-session"):
            legacy = self.project_root / name
            if not legacy.exists():
                continue
            try:
                target = cfg / name
                if not target.exists():
                    legacy.replace(target)
                    continue
                # Keep both, but move the legacy one under `.ccb/` with a suffix.
                suffix = time.strftime("%Y%m%d%H%M%S")
                legacy.replace(cfg / f"{name}.legacy.{suffix}")
            except Exception:
                pass

    def _normalize_cmd_config(self, raw: dict | None) -> dict:
        if raw is None or raw is False:
            return {"enabled": False}
        if isinstance(raw, bool):
            return {"enabled": bool(raw)}
        if isinstance(raw, str):
            return {"enabled": True, "start_cmd": raw.strip()}
        if isinstance(raw, dict):
            enabled = raw.get("enabled")
            if enabled is None:
                enabled = True
            start_cmd = raw.get("start_cmd") or raw.get("command") or raw.get("cmd") or ""
            title = raw.get("title") or raw.get("name") or "CCB-Cmd"
            return {
                "enabled": bool(enabled),
                "start_cmd": str(start_cmd).strip(),
                "title": str(title).strip() or "CCB-Cmd",
            }
        return {"enabled": False}

    def _cmd_settings(self) -> dict:
        cfg = self.cmd_config or {}
        if not cfg or not cfg.get("enabled"):
            return {"enabled": False}
        title = (cfg.get("title") or "CCB-Cmd").strip() or "CCB-Cmd"
        start_cmd = (cfg.get("start_cmd") or "").strip()
        if not start_cmd:
            start_cmd = self._default_cmd_start_cmd()
        return {"enabled": True, "title": title, "start_cmd": start_cmd}

    def _default_cmd_start_cmd(self) -> str:
        if get_shell_type() == "powershell":
            return "pwsh" if shutil.which("pwsh") else "powershell"
        shell = (os.environ.get("SHELL") or "bash").strip() or "bash"
        if not shutil.which(shell):
            shell = "bash"
        return shell

    def _with_bin_path_env(self, env: dict | None = None) -> dict:
        base = dict(env or os.environ)
        bin_path = str(self.script_dir / "bin")
        current = base.get("PATH") or ""
        parts = current.split(os.pathsep) if current else []
        if bin_path not in parts:
            base["PATH"] = bin_path + (os.pathsep + current if current else "")
        return base

    def _current_pane_id(self) -> str:
        if self.terminal_type == "wezterm":
            return (os.environ.get("WEZTERM_PANE") or "").strip()
        try:
            backend = TmuxBackend()
            return backend.get_current_pane_id()
        except Exception:
            return (os.environ.get("TMUX_PANE") or "").strip()

    def _build_env_prefix(self, env: dict) -> str:
        if not env:
            return ""
        if get_shell_type() == "powershell":
            parts: list[str] = []
            for key, val in env.items():
                if val is None:
                    continue
                safe = str(val).replace("'", "''")
                parts.append(f"$env:{key} = '{safe}'; ")
            return "".join(parts)
        parts = []
        for key, val in env.items():
            if val is None:
                continue
            parts.append(f"export {key}={shlex.quote(str(val))}; ")
        return "".join(parts)

    def _provider_pane_id(self, provider: str) -> str:
        prov = (provider or "").strip().lower()
        anchor = (self.anchor_provider or "").strip().lower()
        if prov and prov == anchor and self.anchor_pane_id:
            return str(self.anchor_pane_id)
        if self.terminal_type == "wezterm":
            return str(self.wezterm_panes.get(prov, "") or "")
        return str(self.tmux_panes.get(prov, "") or "")

    def _set_current_pane_label(self, provider: str) -> None:
        if self.terminal_type != "tmux":
            return
        if not os.environ.get("TMUX"):
            return
        try:
            backend = TmuxBackend()
            pane_id = backend.get_current_pane_id()
            title = f"CCB-{provider.capitalize()}"
            backend.set_pane_title(pane_id, title)
            backend.set_pane_user_option(pane_id, "@ccb_agent", provider.capitalize())
        except Exception:
            pass

    def _run_shell_command(self, cmd: str, *, env: dict | None = None, cwd: str | None = None) -> int:
        cmd = cmd or ""
        env = self._with_bin_path_env(env)
        if get_shell_type() == "powershell":
            shell = "pwsh" if shutil.which("pwsh") else "powershell"
            return subprocess.run([shell, "-Command", cmd], env=env, cwd=cwd).returncode
        shell = (os.environ.get("SHELL") or "bash").strip() or "bash"
        if not shutil.which(shell):
            shell = "bash"
        return subprocess.run([shell, "-lc", cmd], env=env, cwd=cwd).returncode

    def _maybe_start_caskd(self) -> None:
        self._maybe_start_provider_daemon("codex")

    def _maybe_start_unified_askd(self, *, quiet: bool = False) -> None:
        """Start unified askd daemon (provider-agnostic)."""
        # Try to start for any enabled provider that uses askd (including claude)
        for provider in ["codex", "gemini", "opencode", "droid", "claude"]:
            if provider in [p.lower() for p in self.providers]:
                # Try to start and check if successful
                self._maybe_start_provider_daemon(provider, quiet=quiet)
                # Verify daemon actually started by pinging
                try:
                    from askd_runtime import state_file_path
                    from askd.daemon import ping_daemon
                    state_file = state_file_path("askd.json")
                    if ping_daemon(timeout_s=0.5, state_file=state_file):
                        return  # Successfully started
                except Exception:
                    pass
                # If not successful, continue to next provider

    def _maybe_start_provider_daemon(self, provider: str, *, quiet: bool = False) -> None:
        def _bool_from_env(name: str):
            raw = os.environ.get(name)
            if raw is None or raw == "":
                return None
            v = raw.strip().lower()
            if v in {"0", "false", "no", "off"}:
                return False
            if v in {"1", "true", "yes", "on"}:
                return True
            return None

        def _emit(msg: str, *, err: bool = False) -> None:
            if quiet and not _env_bool("CCB_DEBUG", False):
                return
            print(msg, file=sys.stderr if err else sys.stdout)

        provider = (provider or "").strip().lower()
        specs = {
            "codex": CASK_CLIENT_SPEC,
            "gemini": GASK_CLIENT_SPEC,
            "opencode": OASK_CLIENT_SPEC,
            "claude": LASK_CLIENT_SPEC,
            "droid": DASK_CLIENT_SPEC,
        }
        spec = specs.get(provider)
        if not spec:
            return
        if provider not in [p.lower() for p in self.providers]:
            return

        autostart = _bool_from_env(spec.autostart_env_primary)
        if autostart is None:
            autostart = _bool_from_env(spec.autostart_env_legacy)
        if autostart is False:
            return
        if _bool_from_env(spec.enabled_env) is False:
            return

        try:
            from importlib import import_module
            daemon_module = import_module(spec.daemon_module)
            ping_daemon = getattr(daemon_module, "ping_daemon")
            read_state = getattr(daemon_module, "read_state", None)
            shutdown_daemon_fn = getattr(daemon_module, "shutdown_daemon", None)
        except Exception as exc:
            _emit(f"⚠️ Failed to import {spec.daemon_module}: {exc}")
            return

        def _owned_by_ccb(state: dict | None) -> bool:
            if not isinstance(state, dict):
                return False
            try:
                parent_pid = int(state.get("parent_pid") or 0)
            except Exception:
                parent_pid = 0
            managed = bool(state.get("managed"))
            return managed and parent_pid == self.ccb_pid

        state_file = None
        raw_state_file = (os.environ.get(spec.state_file_env) or "").strip()
        if raw_state_file:
            try:
                state_file = Path(raw_state_file).expanduser()
            except Exception:
                state_file = None

        if ping_daemon(state_file=state_file):
            st = read_state(state_file=state_file) or {} if callable(read_state) else {}
            if spec.daemon_bin_name == "askd" and not self._askd_checked:
                self._askd_checked = True
                if not _owned_by_ccb(st):
                    # Check if forced rebind is enabled (case-insensitive)
                    force_rebind = _env_bool("CCB_FORCE_REBIND", False)

                    # Check if foreign parent is still alive before forcing rebind
                    # Safely normalize parent_pid to int (handle None, non-int, etc.)
                    try:
                        foreign_parent_pid = int((st or {}).get("parent_pid") or 0)
                    except Exception:
                        foreign_parent_pid = 0

                    foreign_parent_alive = False
                    if not force_rebind and foreign_parent_pid > 0:
                        try:
                            foreign_parent_alive = _is_pid_alive(foreign_parent_pid)
                        except Exception:
                            foreign_parent_alive = False

                    if force_rebind or not foreign_parent_alive:
                        # Safe to rebind: either forced or foreign parent is dead/stale
                        if force_rebind:
                            _emit(f"⚠️ CCB_FORCE_REBIND=1 set, forcing askd rebind despite live parent (PID {foreign_parent_pid})...")
                        else:
                            _emit(f"⚠️ askd owned by dead parent (PID {foreign_parent_pid}), restarting to bind lifecycle...")
                        if callable(shutdown_daemon_fn):
                            try:
                                shutdown_daemon_fn(timeout_s=1.0, state_file=state_file)
                            except Exception:
                                pass
                    else:
                        # Foreign parent is still alive, don't force rebind
                        _emit(f"⚠️ askd owned by live parent (PID {foreign_parent_pid}), skipping rebind to avoid disruption")
                        _emit(f"   Set CCB_FORCE_REBIND=1 to override this safety check")
                        host = st.get("host") if isinstance(st, dict) else None
                        port = st.get("port") if isinstance(st, dict) else None
                        if host and port:
                            _emit(f"✅ {spec.daemon_bin_name} already running at {host}:{port}")
                        else:
                            _emit(f"✅ {spec.daemon_bin_name} already running")
                        return
                    deadline = time.time() + 2.0
                    while time.time() < deadline:
                        if not ping_daemon(timeout_s=0.2, state_file=state_file):
                            break
                        time.sleep(0.1)
                    if ping_daemon(timeout_s=0.2, state_file=state_file):
                        _emit("⚠️ askd restart skipped: existing daemon still running")
                        host = st.get("host") if isinstance(st, dict) else None
                        port = st.get("port") if isinstance(st, dict) else None
                        if host and port:
                            _emit(f"✅ {spec.daemon_bin_name} already running at {host}:{port}")
                        else:
                            _emit(f"✅ {spec.daemon_bin_name} already running")
                        return
                else:
                    host = st.get("host") if isinstance(st, dict) else None
                    port = st.get("port") if isinstance(st, dict) else None
                    if host and port:
                        _emit(f"✅ {spec.daemon_bin_name} already running at {host}:{port}")
                    else:
                        _emit(f"✅ {spec.daemon_bin_name} already running")
                    return
            else:
                host = st.get("host") if isinstance(st, dict) else None
                port = st.get("port") if isinstance(st, dict) else None
                if host and port:
                    _emit(f"✅ {spec.daemon_bin_name} already running at {host}:{port}")
                else:
                    _emit(f"✅ {spec.daemon_bin_name} already running")
                return

        daemon_script = self.script_dir / "bin" / spec.daemon_bin_name
        if not daemon_script.exists():
            _emit(f"⚠️ {spec.daemon_bin_name} not found (bin/{spec.daemon_bin_name}). Reinstall or update your checkout.")
            return

        kwargs = {
            "stdin": subprocess.DEVNULL,
            "stdout": subprocess.DEVNULL,
            "stderr": subprocess.DEVNULL,
            "close_fds": True,
        }
        if os.name == "nt":
            kwargs["creationflags"] = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) | getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000)
        else:
            kwargs["start_new_session"] = True

        try:
            env = os.environ.copy()
            env["CCB_PARENT_PID"] = str(os.getpid())
            proc = subprocess.Popen([sys.executable, str(daemon_script)], env=env, **kwargs)
            # Track daemon process for reaping (thread-safe)
            if spec.daemon_bin_name == "askd":
                with self._daemon_proc_lock:
                    self._daemon_proc = proc
        except Exception as exc:
            _emit(f"⚠️ Failed to start {spec.daemon_bin_name}: {exc}")
            return

        deadline = time.time() + 2.0
        while time.time() < deadline:
            if ping_daemon(timeout_s=0.2, state_file=state_file):
                st = read_state(state_file=state_file) or {} if callable(read_state) else {}
                host = st.get("host") if isinstance(st, dict) else None
                port = st.get("port") if isinstance(st, dict) else None
                if host and port:
                    _emit(f"✅ {spec.daemon_bin_name} started at {host}:{port}")
                else:
                    _emit(f"✅ {spec.daemon_bin_name} started")
                return
            time.sleep(0.1)
        _emit(f"⚠️ {spec.daemon_bin_name} start requested, but daemon not reachable yet")

    def _start_daemon_watchdog(self) -> None:
        """Start watchdog thread to monitor askd daemon health."""
        import threading

        # Restart if thread exists but is dead
        if self._watchdog_thread is not None:
            if not self._watchdog_thread.is_alive():
                self._watchdog_thread = None
                self._watchdog_stop_event = None
            else:
                return  # Already running

        self._watchdog_stop_event = threading.Event()
        self._watchdog_thread = threading.Thread(
            target=self._daemon_watchdog_loop,
            daemon=True,
            name="askd-watchdog"
        )
        self._watchdog_thread.start()

    def _daemon_watchdog_loop(self) -> None:
        """Watchdog loop to monitor and restart askd daemon if needed."""
        from askd_runtime import state_file_path
        from askd.daemon import ping_daemon, read_state

        verbose_watchdog = _env_bool("CCB_WATCHDOG_VERBOSE", False) or _env_bool("CCB_DEBUG", False)

        def _wd_log(msg: str) -> None:
            if verbose_watchdog:
                print(msg, file=sys.stderr)

        # Validate and clamp check interval
        try:
            check_interval = float(os.environ.get("CCB_WATCHDOG_INTERVAL_S", "10"))
            check_interval = max(1.0, min(check_interval, 300.0))  # Clamp to 1-300 seconds
        except (ValueError, TypeError):
            check_interval = 10.0

        consecutive_failures = 0
        max_failures = 3
        ownership_mismatch_count = 0
        max_ownership_mismatches = 3

        while not self._watchdog_stop_event.wait(check_interval):
            try:
                # Reap zombie child processes
                self._reap_zombie_children()

                # Check if tracked daemon process has exited
                with self._daemon_proc_lock:
                    if self._daemon_proc is not None:
                        exit_code = self._daemon_proc.poll()
                        if exit_code is not None:
                            # Daemon process has exited
                            _wd_log(f"⚠️ askd daemon process exited with code {exit_code}")
                            self._daemon_proc = None
                            # Will be restarted by health check below

                # Check askd daemon health
                state_file = state_file_path("askd.json")
                if not ping_daemon(timeout_s=0.5, state_file=state_file):
                    consecutive_failures += 1
                    if consecutive_failures >= max_failures:
                        # Daemon is unhealthy, try to restart
                        _wd_log(f"⚠️ askd daemon unhealthy (failed {consecutive_failures} checks), attempting restart...")
                        self._maybe_start_unified_askd(quiet=not verbose_watchdog)
                        consecutive_failures = 0
                else:
                    # Daemon is healthy, verify ownership
                    state = read_state(state_file=state_file)
                    if isinstance(state, dict):
                        parent_pid = int(state.get("parent_pid") or 0)
                        managed = bool(state.get("managed"))
                        if managed and parent_pid != self.ccb_pid:
                            ownership_mismatch_count += 1
                            _wd_log(f"⚠️ askd daemon ownership mismatch (parent_pid={parent_pid}, expected={self.ccb_pid}, count={ownership_mismatch_count})")
                            if ownership_mismatch_count >= max_ownership_mismatches:
                                # Check if forced rebind is enabled (case-insensitive)
                                force_rebind = _env_bool("CCB_FORCE_REBIND", False)

                                # Check if foreign parent is still alive before forcing rebind
                                foreign_parent_alive = False
                                if not force_rebind:
                                    # Use existing _is_pid_alive for consistent cross-platform check
                                    try:
                                        foreign_parent_alive = _is_pid_alive(parent_pid)
                                    except Exception:
                                        foreign_parent_alive = False

                                if force_rebind or not foreign_parent_alive:
                                    # Safe to rebind: either forced or foreign parent is dead/stale
                                    if force_rebind:
                                        _wd_log(f"⚠️ CCB_FORCE_REBIND=1 set, forcing rebind despite live parent (PID {parent_pid})...")
                                    else:
                                        _wd_log(f"⚠️ Foreign parent (PID {parent_pid}) is dead, attempting to rebind daemon...")
                                    try:
                                        from askd_rpc import shutdown_daemon
                                        shutdown_daemon("ask", timeout_s=2.0, state_file=state_file)
                                        time.sleep(1.0)  # Wait for shutdown
                                    except Exception:
                                        pass
                                    self._maybe_start_unified_askd(quiet=not verbose_watchdog)
                                    ownership_mismatch_count = 0
                                else:
                                    # Foreign parent is still alive, don't force rebind
                                    _wd_log(f"⚠️ Foreign parent (PID {parent_pid}) is still alive, skipping forced rebind to avoid disruption")
                                    _wd_log(f"   Set CCB_FORCE_REBIND=1 to override this safety check")
                                    ownership_mismatch_count = 0  # Reset to avoid repeated warnings
                        else:
                            ownership_mismatch_count = 0  # Reset on correct ownership
                    consecutive_failures = 0
            except Exception as e:
                # Silently continue on watchdog errors
                pass

    def _reap_zombie_children(self) -> None:
        """Reap tracked daemon process if it has exited."""
        # Only reap the tracked daemon process, not all children
        with self._daemon_proc_lock:
            if self._daemon_proc is not None:
                try:
                    exit_code = self._daemon_proc.poll()
                    if exit_code is not None:
                        # Process has exited, reap it by calling wait()
                        try:
                            self._daemon_proc.wait(timeout=0.1)
                        except Exception:
                            pass
                except Exception:
                    pass

    def _stop_daemon_watchdog(self) -> None:
        """Stop watchdog thread."""
        if self._watchdog_stop_event:
            self._watchdog_stop_event.set()
        if self._watchdog_thread:
            self._watchdog_thread.join(timeout=1.0)
            # Only clear if thread actually stopped
            if not self._watchdog_thread.is_alive():
                self._watchdog_thread = None
            else:
                # Thread still running, log warning
                pass  # Silently continue, daemon thread will exit with process

    def _detect_terminal_type(self):
        # Forced by environment variable
        forced = (os.environ.get("CCB_TERMINAL") or os.environ.get("CODEX_TERMINAL") or "").strip().lower()
        if forced in {"wezterm", "tmux"}:
            return forced

        detected = detect_terminal()
        if detected:
            return detected

        # Nothing found
        return None

    def _detect_launch_terminal(self):
        """Select terminal program for launching new windows (tmux mode only)"""
        # WezTerm mode doesn't need external terminal program
        if self.terminal_type == "wezterm":
            return None
        # tmux mode: select terminal
        terminals = ["gnome-terminal", "konsole", "alacritty", "xterm"]
        for term in terminals:
            if shutil.which(term):
                return term
        return "tmux"

    def _set_tmux_ui_active(self, active: bool) -> None:
        """
        Enable/disable CCB tmux UI theming for the *current tmux session*.

        This is session-scoped and reversible (saves/restores user options) via helper scripts
        installed to `~/.local/bin/`.
        """
        if self.terminal_type != "tmux":
            return
        if not os.environ.get("TMUX"):
            return
        script = Path.home() / ".local" / "bin" / ("ccb-tmux-on.sh" if active else "ccb-tmux-off.sh")
        if not script.exists():
            return
        try:
            debug = os.environ.get("CCB_DEBUG") in ("1", "true", "yes")
            cp = subprocess.run(
                [str(script)],
                check=False,
                capture_output=debug,
                text=True,
                encoding="utf-8",
                errors="replace",
            )
            if cp.returncode != 0 and debug:
                out = (cp.stdout or "").strip()
                err = (cp.stderr or "").strip()
                detail = "\n".join([s for s in [out, err] if s])
                if detail:
                    print(f"⚠️ ccb tmux ui script failed (rc={cp.returncode}):\n{detail}", file=sys.stderr)
        except Exception:
            return

    def _launch_script_in_macos_terminal(self, script_file: Path) -> bool:
        """macOS: Use Terminal.app to open new window for script (avoid tmux launcher nesting issues)"""
        if platform.system() != "Darwin":
            return False
        if not shutil.which("osascript"):
            return False
        env = os.environ.copy()
        env["CCB_WRAPPER_SCRIPT"] = str(script_file)
        subprocess.Popen(
            [
                "osascript",
                "-e",
                'tell application "Terminal" to do script "/bin/bash " & quoted form of (system attribute "CCB_WRAPPER_SCRIPT")',
                "-e",
                'tell application "Terminal" to activate',
            ],
            env=env,
        )
        return True

    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')}")
            print(f"   {t('solutions')}")
            print(f"   - {t('install_wezterm')}")
            print(f"   - {t('or_install_tmux')}")
            if (shutil.which("tmux") or shutil.which("tmux.exe")) and not (os.environ.get("TMUX") or os.environ.get("TMUX_PANE")):
                print(f"   - {t('tmux_installed_not_inside')}")
            print(f"   - {t('or_set_ccb_terminal')}")
            return None

        # WezTerm mode: no tmux dependency
        if self.terminal_type == "wezterm":
            print(f"🚀 {t('starting_backend', provider=provider.capitalize(), terminal='wezterm')}")
            return self._start_provider_wezterm(provider, parent_pane=parent_pane, direction=direction)

        # tmux mode: check if tmux is available
        if not shutil.which("tmux"):
            # Try fallback to WezTerm
            if detect_terminal() == "wezterm":
                self.terminal_type = "wezterm"
                print(f"🚀 {t('starting_backend', provider=provider.capitalize(), terminal='wezterm - tmux unavailable')}")
                return self._start_provider_wezterm(provider, parent_pane=parent_pane, direction=direction)
            else:
                print(f"❌ {t('tmux_not_installed')}")
                print(f"   {t('install_wezterm_or_tmux')}")
                return None

        # Require an existing tmux client session (no auto-launch).
        if not (os.environ.get("TMUX") or os.environ.get("TMUX_PANE")):
            print(f"❌ {t('tmux_installed_not_inside')}", file=sys.stderr)
            print(f"💡 Run: tmux", file=sys.stderr)
            print(f"   Then: {' '.join(['ccb', *sys.argv[1:]])}", file=sys.stderr)
            return None

        print(f"🚀 {t('starting_backend', provider=provider.capitalize(), terminal='tmux')}")

        if provider == "codex":
            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)
        elif provider == "opencode":
            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)
        else:
            print(f"❌ {t('unknown_provider', provider=provider)}")
            return None

    def _start_provider_wezterm(
        self,
        provider: str,
        *,
        parent_pane: str | None = None,
        direction: str | None = None,
    ) -> str | None:
        runtime = self.runtime_dir / provider
        runtime.mkdir(parents=True, exist_ok=True)

        start_cmd = self._get_start_cmd(provider)
        keep_open = os.environ.get("CODEX_WEZTERM_KEEP_OPEN", "1").lower() not in {"0", "false", "no", "off"}
        if keep_open:
            start_cmd = _build_keep_open_cmd(provider, start_cmd)
        use_direction = (direction or ("right" if not self.wezterm_panes else "bottom")).strip() or "right"
        use_parent = parent_pane
        if not use_parent and use_direction == "bottom":
            try:
                use_parent = next(reversed(self.wezterm_panes.values()))
            except StopIteration:
                use_parent = None

        pane_title_marker = f"CCB-{provider.capitalize()}"
        title_cmd = _build_pane_title_cmd(pane_title_marker)
        env_overrides = self._provider_env_overrides(provider)
        full_cmd = title_cmd + self._build_env_prefix(env_overrides) + _build_export_path_cmd(self.script_dir / "bin") + start_cmd
        backend = WeztermBackend()
        pane_id = backend.create_pane(full_cmd, str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent)
        self.wezterm_panes[provider] = pane_id

        if provider == "codex":
            input_fifo = runtime / "input.fifo"
            output_fifo = runtime / "output.fifo"
            # WezTerm mode injects text via pane, no strong FIFO dependency; Windows/WSL may not support mkfifo
            self._write_codex_session(
                runtime,
                None,
                input_fifo,
                output_fifo,
                pane_id=pane_id,
                pane_title_marker=pane_title_marker,
                codex_start_cmd=start_cmd,
            )
        elif provider == "gemini":
            self._write_gemini_session(runtime, None, pane_id=pane_id, pane_title_marker=pane_title_marker, start_cmd=start_cmd)
        elif provider == "opencode":
            self._write_opencode_session(runtime, None, pane_id=pane_id, pane_title_marker=pane_title_marker, start_cmd=start_cmd)
        elif provider == "droid":
            self._write_droid_session(runtime, None, pane_id=pane_id, pane_title_marker=pane_title_marker, start_cmd=start_cmd)
        else:
            print(f"❌ {t('unknown_provider', provider=provider)}")
            return None

        print(f"✅ {t('started_backend', provider=provider.capitalize(), terminal='wezterm pane', pane_id=pane_id)}")
        return pane_id

    def _work_dir_strings(self, work_dir: Path) -> list[str]:
        candidates: list[str] = []
        env_pwd = os.environ.get("PWD")
        if env_pwd:
            candidates.append(env_pwd)
        candidates.append(str(work_dir))
        try:
            candidates.append(str(work_dir.resolve()))
        except Exception:
            pass
        # de-dup while preserving order
        seen: set[str] = set()
        result: list[str] = []
        for candidate in candidates:
            if candidate not in seen:
                seen.add(candidate)
                result.append(candidate)
        return result

    def _read_json_file(self, path: Path) -> dict:
        try:
            if not path.exists():
                return {}
            # Session files are written as UTF-8; on Windows PowerShell 5.1 the default encoding
            # may not be UTF-8, so always decode explicitly and tolerate UTF-8 BOM.
            raw = path.read_text(encoding="utf-8-sig")
            data = json.loads(raw)
            return data if isinstance(data, dict) else {}
        except Exception:
            return {}

    def _write_json_file(self, path: Path, data: dict) -> None:
        try:
            if not path.parent.is_dir():
                raise FileNotFoundError(str(path.parent))
            path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
        except Exception:
            pass

    def _require_project_config_dir(self) -> bool:
        cfg = self._project_config_dir()
        if cfg.is_dir():
            return True
        print("❌ Missing required project config directory: .ccb", file=sys.stderr)
        print(f"   project_root: {self.project_root}", file=sys.stderr)
        print(f"   cwd:          {Path.cwd()}", file=sys.stderr)
        print(f"💡 Fix: mkdir -p {cfg}", file=sys.stderr)
        return False

    def _clear_codex_log_binding(self, data: dict) -> dict:
        try:
            if not isinstance(data, dict):
                return {}
            cleared = dict(data)
            for key in ("codex_session_path", "codex_session_id", "codex_start_cmd"):
                if key in cleared:
                    cleared.pop(key, None)
            if cleared.get("active") is False:
                cleared["active"] = True
            return cleared
        except Exception as exc:
            print(f"⚠️ codex_session_clear_failed: {exc}", file=sys.stderr)
            return data if isinstance(data, dict) else {}

    def _claude_session_file(self) -> Path:
        return self._project_session_file(".claude-session")

    def _backfill_claude_session_work_dir_fields(self) -> None:
        path = self._claude_session_file()
        if not path.exists():
            return
        data = self._read_json_file(path)
        if not isinstance(data, dict) or not data:
            return
        changed = False
        work_dir = str(self.project_root)
        work_dir_norm = _normalize_path_for_match(work_dir)
        if not str(data.get("work_dir") or "").strip():
            data["work_dir"] = work_dir
            changed = True
        if not str(data.get("work_dir_norm") or "").strip():
            data["work_dir_norm"] = work_dir_norm
            changed = True
        if not changed:
            return
        payload = json.dumps(data, ensure_ascii=False, indent=2)
        safe_write_session(path, payload)

    def _read_local_claude_session_id(self) -> str | None:
        data = self._read_json_file(self._claude_session_file())
        sid = data.get("claude_session_id")
        if not sid:
            legacy = data.get("session_id")
            if isinstance(legacy, str) and legacy.strip():
                try:
                    uuid.UUID(legacy)
                    sid = legacy
                except Exception:
                    sid = None
        if isinstance(sid, str) and sid.strip():
            # Guard against path-format mismatch (Windows case/slash differences, MSYS paths, etc.).
            recorded_norm = _extract_session_work_dir_norm(data)
            if not recorded_norm:
                # Old/foreign session file without a recorded work dir: refuse to resume to avoid cross-project reuse.
                return None
            current_keys = _work_dir_match_keys(Path.cwd())
            if current_keys and recorded_norm not in current_keys:
                return None
            return sid.strip()
        return None

    def _write_local_claude_session(
        self,
        session_id: str | None = None,
        *,
        active: bool = True,
        pane_id: str | None = None,
        pane_title_marker: str | None = None,
        terminal: str | None = None,
    ) -> None:
        path = self._claude_session_file()
        writable, reason, fix = check_session_writable(path)
        if not writable:
            print(f"❌ Cannot write {path.name}: {reason}", file=sys.stderr)
            print(f"💡 Fix: {fix}", file=sys.stderr)
            return

        data = self._read_json_file(path) if path.exists() else {}
        work_dir = self.project_root
        now = time.strftime("%Y-%m-%d %H:%M:%S")
        if session_id:
            data["claude_session_id"] = session_id
        data["session_id"] = self.session_id
        try:
            data["ccb_project_id"] = compute_ccb_project_id(work_dir)
        except Exception:
            data["ccb_project_id"] = data.get("ccb_project_id")
        data["work_dir"] = str(work_dir)
        data["work_dir_norm"] = _normalize_path_for_match(str(work_dir))
        data["start_dir"] = str(self.invocation_dir)
        data["terminal"] = terminal or self.terminal_type
        data["active"] = bool(active)
        data["started_at"] = data.get("started_at") or now
        data["updated_at"] = now
        if pane_id:
            data["pane_id"] = pane_id
        if pane_title_marker:
            data["pane_title_marker"] = pane_title_marker

        payload = json.dumps(data, ensure_ascii=False, indent=2)
        ok, err = safe_write_session(path, payload)
        if not ok:
            print(err, file=sys.stderr)
            return
        if pane_id:
            try:
                upsert_registry(
                    {
                        "ccb_session_id": self.session_id,
                        "ccb_project_id": compute_ccb_project_id(self.project_root),
                        "work_dir": str(self.project_root),
                        "terminal": terminal or self.terminal_type,
                        "providers": {
                            "claude": {
                                "pane_id": pane_id,
                                "pane_title_marker": pane_title_marker,
                                "session_file": str(path),
                                "claude_session_id": data.get("claude_session_id"),
                                "claude_session_path": data.get("claude_session_path"),
                            }
                        },
                    }
                )
            except Exception:
                pass
        self._maybe_start_provider_daemon("claude")

    def _get_latest_codex_session_id(self) -> tuple[str | None, bool]:
        """
        Returns (session_id, has_any_history_for_cwd).
        Session id is Codex CLI's UUID used by `codex resume <id>`.
        Always scans session logs to find the latest session for current cwd,
        then updates local .codex-session file.
        """
        # Always scan logs to find the latest session - don't use cached session file
        # as it may be stale (user may have started new sessions since last run).
        project_session = self._project_session_file(".codex-session")

        # Always scan Codex session logs for the latest session bound to this cwd.
        # This ensures we get the latest session even if user did /clear during run.
        root = Path(os.environ.get("CODEX_SESSION_ROOT") or (Path.home() / ".codex" / "sessions")).expanduser()
        if not root.exists():
            return None, False
        root_norm = _normalize_path_for_match(str(self.project_root))
        if not root_norm:
            return None, False
        try:
            logs = sorted(
                (p for p in root.glob("**/*.jsonl") if p.is_file()),
                key=lambda p: p.stat().st_mtime,
                reverse=True,
            )
        except Exception:
            logs = []

        scan_limit = 400
        raw_limit = (os.environ.get("CCB_CODEX_SCAN_LIMIT") or "").strip()
        if raw_limit:
            try:
                scan_limit = max(100, min(20000, int(raw_limit)))
            except Exception:
                scan_limit = 400

        for log_path in logs[:scan_limit]:
            try:
                with log_path.open("r", encoding="utf-8", errors="ignore") as handle:
                    head = [handle.readline().strip() for _ in range(30)]
            except OSError:
                continue
            for line in head:
                if not line:
                    continue
                try:
                    entry = json.loads(line)
                except Exception:
                    continue
                if not isinstance(entry, dict) or entry.get("type") != "session_meta":
                    continue
                payload = entry.get("payload") if isinstance(entry.get("payload"), dict) else {}
                cwd = payload.get("cwd")
                if not isinstance(cwd, str) or not cwd.strip():
                    continue
                cwd_norm = _normalize_path_for_match(cwd)
                # Accept sessions launched from any directory under the same project root.
                if not _normpath_within(cwd_norm, root_norm):
                    continue
                sid = payload.get("id")
                if isinstance(sid, str) and sid:
                    # Update local .codex-session file with latest session id
                    data = self._read_json_file(project_session) if project_session.exists() else {}
                    data.update({
                        "codex_session_id": sid,
                        "codex_session_path": str(log_path),
                        "work_dir": str(self.project_root),
                        "work_dir_norm": _normalize_path_for_match(str(self.project_root)),
                        "start_dir": str(self.invocation_dir),
                        "updated_at": time.strftime("%Y-%m-%d %H:%M:%S"),
                    })
                    self._write_json_file(project_session, data)
                    return sid, True
        return None, False

    def _ensure_codex_auto_approval(self) -> None:
        """Auto-configure codex approval policy for current directory when -a flag is used."""
        import re

        codex_config = Path.home() / ".codex" / "config.toml"
        if not codex_config.exists():
            return

        def _toml_unescape_basic_string(value: str) -> str:
            # Minimal TOML basic-string unescape for our use-case (paths).
            # https://toml.io/en/v1.0.0#string
            try:
                return (
                    (value or "")
                    .replace("\\\\", "\\")
                    .replace('\\"', '"')
                    .replace("\\t", "\t")
                    .replace("\\n", "\n")
                    .replace("\\r", "\r")
                )
            except Exception:
                return value or ""

        def _toml_escape_basic_string(value: str) -> str:
            return (value or "").replace("\\", "\\\\").replace('"', '\\"')

        cwd = str(Path.cwd())
        section_header = f'[projects."{_toml_escape_basic_string(cwd)}"]'

        ccb_begin = "# --- BEGIN CCB: codex auto-approval ---"
        ccb_end = "# --- END CCB: codex auto-approval ---"
        desired = [
            'trust_level = "trusted"',
            'approval_policy = "never"',
            'sandbox_mode = "danger-full-access"',
        ]

        try:
            content = codex_config.read_text(encoding="utf-8")

            def _migrate_legacy_full_auto(text: str) -> tuple[str, bool]:
                """
                Older CCB versions wrote `sandbox_mode = "full-auto"` into `~/.codex/config.toml`.
                Codex rejects this value (valid variants: read-only/workspace-write/danger-full-access),
                so we migrate it to `danger-full-access` to match CCB's `-a/--auto` intent.
                """
                lines = (text or "").splitlines(keepends=True)
                if not lines:
                    return text, False
                out: list[str] = []
                changed = False
                for line in lines:
                    stripped = line.strip()
                    if "sandbox_mode" in stripped and ("\"full-auto\"" in stripped or "'full-auto'" in stripped):
                        out.append(
                            line.replace('"full-auto"', '"danger-full-access"').replace("'full-auto'", '"danger-full-access"')
                        )
                        changed = True
                        continue
                    out.append(line)
                return "".join(out), changed

            content, migrated = _migrate_legacy_full_auto(content)

            def _dedupe_duplicate_project_tables(text: str) -> tuple[str, bool]:
                """
                Codex TOML rejects duplicate `[projects.\"<path>\"]` tables.
                Keep the last occurrence for each key and drop earlier duplicates.
                """
                lines = (text or "").splitlines(keepends=True)
                if not lines:
                    return text, False

                sections: list[tuple[str | None, list[str], str | None]] = []
                header: str | None = None
                body: list[str] = []

                def flush() -> None:
                    nonlocal header, body
                    if header is None and not body:
                        return
                    key: str | None = None
                    if header:
                        h = header.strip()
                        if h.startswith("[projects.") and h.endswith("]"):
                            inner = h[len("[projects.") : -1].strip()
                            if len(inner) >= 2 and inner[0] == inner[-1] and inner[0] in {'"', "'"}:
                                raw = inner[1:-1]
                                key = _toml_unescape_basic_string(raw) if inner[0] == '"' else raw
                    sections.append((header or "", body, key))
                    header = None
                    body = []

                for ln in lines:
                    # Start of a TOML table/array-of-table header (top-level only).
                    if ln.startswith("["):
                        flush()
                        header = ln
                        body = []
                        continue
                    body.append(ln)
                flush()

                last: dict[str, int] = {}
                for i, (_h, _b, k) in enumerate(sections):
                    if k is not None:
                        last[k] = i

                changed = False
                out: list[str] = []
                for i, (h, b, k) in enumerate(sections):
                    if k is not None and last.get(k) != i:
                        changed = True
                        continue
                    out.append(h)
                    out.extend(b)
                return "".join(out), changed

            content, deduped = _dedupe_duplicate_project_tables(content)

            def _section_bounds(text: str, header: str) -> tuple[int, int] | None:
                m = re.search(rf"(?m)^{re.escape(header)}\\s*$", text)
                if not m:
                    return None
                start = m.start()
                tail = text[m.end() :]
                n = re.search(r"(?m)^\\[", tail)
                end = m.end() + (n.start() if n else len(tail))
                return start, end

            def _has_key(section_text: str, key: str) -> bool:
                return re.search(rf"(?m)^\\s*{re.escape(key)}\\s*=", section_text) is not None

            def _ensure_block(section_text: str) -> tuple[str, bool]:
                lines = section_text.splitlines(keepends=True)
                changed = False

                try:
                    header_idx = next(i for i, ln in enumerate(lines) if ln.strip() == section_header)
                except StopIteration:
                    return section_text, False

                begin_idx = None
                end_idx = None
                for i, ln in enumerate(lines):
                    if ln.strip() == ccb_begin:
                        begin_idx = i
                    if ln.strip() == ccb_end:
                        end_idx = i
                        break

                missing = [ln for ln in desired if not _has_key(section_text, ln.split("=", 1)[0].strip())]
                if not missing and begin_idx is None and end_idx is None:
                    # If the keys already exist (likely added previously), wrap them with markers (best-effort)
                    # so future diffs stay localized. Keep it conservative: only wrap if span is small.
                    key_re = re.compile(r"^\\s*(trust_level|approval_policy|sandbox_mode)\\s*=")
                    key_lines = [i for i, ln in enumerate(lines) if key_re.search(ln)]
                    if key_lines:
                        span = max(key_lines) - min(key_lines)
                        if span <= 12:
                            insert_before = min(key_lines)
                            insert_after = max(key_lines) + 1
                            lines.insert(insert_before, ccb_begin + "\n")
                            lines.insert(insert_after + 1, ccb_end + "\n")
                            changed = True
                    return "".join(lines), changed

                if not missing:
                    return section_text, False

                block_lines = [ccb_begin + "\n", *[m + "\n" for m in missing], ccb_end + "\n"]

                if begin_idx is not None and end_idx is not None and begin_idx < end_idx:
                    # Insert missing keys right before end marker.
                    lines[end_idx:end_idx] = [m + "\n" for m in missing]
                    changed = True
                    return "".join(lines), changed

                # No existing block: append a new CCB-managed block at end of this section.
                insert_at = len(lines)
                while insert_at > 0 and lines[insert_at - 1].strip() == "":
                    insert_at -= 1
                if insert_at > 0 and lines[insert_at - 1].strip() != "":
                    block_lines.insert(0, "\n")
                lines[insert_at:insert_at] = block_lines
                changed = True
                return "".join(lines), changed

            changed_any = False
            bounds = _section_bounds(content, section_header)
            if bounds:
                start, end = bounds
                section = content[start:end]
                updated_section, changed = _ensure_block(section)
                if changed:
                    content = content[:start] + updated_section + content[end:]
                    changed_any = True
            else:
                # No existing section: append a new CCB-managed project section at end of file.
                entry = "\n".join([section_header, ccb_begin, *desired, ccb_end]) + "\n"
                content = content.rstrip() + "\n\n" + entry
                changed_any = True

            # De-dupe again after edits, in case older configs already had duplicates.
            content, deduped_after = _dedupe_duplicate_project_tables(content)
            deduped = deduped or deduped_after
            deduped = deduped or migrated

            if changed_any or deduped:
                codex_config.write_text(content, encoding="utf-8")
                print(f"✅ Codex auto-approval configured for: {cwd}")
        except Exception as e:
            print(f"⚠️ Failed to configure codex auto-approval: {e}")

    def _build_codex_start_cmd(self) -> str:
        if self.auto:
            self._ensure_codex_auto_approval()
        # NOTE: Codex CLI (codex-cli) does not support the legacy flag
        # `--dangerously-bypass-approvals-and-sandbox`. Auto-mode is implemented via config and
        # supported `-c key=value` overrides.
        cmd_parts = ["codex", "-c", "disable_paste_burst=true"]
        if self.auto:
            cmd_parts.extend([
                "-c", 'trust_level="trusted"',
                "-c", 'approval_policy="never"',
                "-c", 'sandbox_mode="danger-full-access"',
            ])
        cmd = " ".join(cmd_parts)
        codex_resumed = False
        if self.resume:
            session_id, has_history = self._get_latest_codex_session_id()
            if session_id:
                cmd = f"{cmd} resume {session_id}"
                print(f"🔁 {t('resuming_session', provider='Codex', session_id=session_id[:8])}")
                codex_resumed = True

            if not codex_resumed:
                print(f"ℹ️ {t('no_history_fresh', provider='Codex')}")
        return cmd

    def _get_latest_gemini_project_hash(self) -> tuple[str | None, bool, Path | None]:
        """
        Returns (project_hash, has_any_history_for_cwd).
        Gemini CLI stores sessions under ~/.gemini/tmp/<sha256(cwd)>/chats/.
        """
        import hashlib

        gemini_root = Path(os.environ.get("GEMINI_ROOT") or (Path.home() / ".gemini" / "tmp")).expanduser()

        candidates: list[str] = []
        try:
            candidates.append(str(Path.cwd().absolute()))
        except Exception:
            pass
        try:
            candidates.append(str(Path.cwd().resolve()))
        except Exception:
            pass
        try:
            candidates.append(str(self.project_root.absolute()))
        except Exception:
            candidates.append(str(self.project_root))
        try:
            candidates.append(str(self.project_root.resolve()))
        except Exception:
            pass
        env_pwd = (os.environ.get("PWD") or "").strip()
        if env_pwd:
            try:
                candidates.append(os.path.abspath(os.path.expanduser(env_pwd)))
            except Exception:
                candidates.append(env_pwd)

        seen: set[str] = set()
        for candidate in candidates:
            if not candidate or candidate in seen:
                continue
            seen.add(candidate)
            project_hash = hashlib.sha256(candidate.encode()).hexdigest()
            chats_dir = gemini_root / project_hash / "chats"
            if not chats_dir.exists():
                continue
            session_files = list(chats_dir.glob("session-*.json"))
            if session_files:
                resume_dir = None
                try:
                    p = Path(candidate)
                    if p.is_dir():
                        resume_dir = p
                except Exception:
                    resume_dir = None
                return project_hash, True, resume_dir

        return None, False, None

    def _build_gemini_start_cmd(self) -> str:
        cmd = "gemini --yolo" if self.auto else "gemini"
        if self.resume:
            _, has_history, resume_dir = self._get_latest_gemini_project_hash()
            if has_history:
                if resume_dir:
                    cmd = f"{_build_cd_cmd(resume_dir)}{cmd} --resume latest"
                else:
                    cmd = f"{cmd} --resume latest"
                print(f"🔁 {t('resuming_session', provider='Gemini', session_id='')}")
            else:
                print(f"ℹ️ {t('no_history_fresh', provider='Gemini')}")
        return cmd

    def _warmup_provider(self, provider: str, timeout: float = 8.0) -> bool:
        if provider == "gemini":
            return True
        if provider == "codex":
            ping_script = self.script_dir / "bin" / "cping"
        elif provider == "gemini":
            ping_script = self.script_dir / "bin" / "gping"
        elif provider == "opencode":
            ping_script = self.script_dir / "bin" / "oping"
        elif provider == "droid":
            ping_script = self.script_dir / "bin" / "dping"
        else:
            return False

        if not ping_script.exists():
            return False

        print(f"🔧 Warmup: {ping_script.name}")
        deadline = time.time() + timeout
        last_result: subprocess.CompletedProcess[str] | None = None
        sleep_s = 0.3
        while time.time() < deadline:
            last_result = subprocess.run(
                [sys.executable, str(ping_script)],
                cwd=str(Path.cwd()),
                capture_output=True,
                text=True,
                encoding='utf-8',
                errors='replace',
            )
            if last_result.returncode == 0:
                out = (last_result.stdout or "").strip()
                if out:
                    print(out)
                return True
            time.sleep(sleep_s)
            sleep_s = min(1.0, sleep_s * 1.5)

        if last_result:
            out = ((last_result.stdout or "") + "\n" + (last_result.stderr or "")).strip()
            if out:
                print(out)
        print(f"⚠️ Warmup failed: {provider}")
        return False

    def _get_start_cmd(self, provider: str) -> str:
        if provider == "codex":
            # NOTE: Codex TUI has paste-burst detection; terminal injection (wezterm send-text/tmux paste-buffer)
            # is often detected as "paste", causing Enter to only line-break not submit. Disable detection by default.
            return self._build_codex_start_cmd()
        elif provider == "gemini":
            return self._build_gemini_start_cmd()
        elif provider == "opencode":
            return self._build_opencode_start_cmd()
        elif provider == "droid":
            return self._build_droid_start_cmd()
        return ""

    def _opencode_resume_allowed(self) -> bool:
        try:
            from opencode_comm import OPENCODE_STORAGE_ROOT
        except Exception:
            return False

        root = Path(OPENCODE_STORAGE_ROOT)
        sessions_root = root / "session"
        if not sessions_root.exists():
            return False

        target_dir = _normalize_path_for_match(str(self.project_root))
        if not target_dir:
            return False

        def _load_json(path: Path) -> dict:
            try:
                raw = path.read_text(encoding="utf-8")
                data = json.loads(raw)
                return data if isinstance(data, dict) else {}
            except Exception:
                return {}

        def _session_dir_has_match(dir_path: Path) -> bool:
            if not dir_path.exists():
                return False
            try:
                session_files = list(dir_path.glob("ses_*.json"))
            except Exception:
                session_files = []
            for path in session_files:
                if not path.is_file():
                    continue
                payload = _load_json(path)
                directory = payload.get("directory")
                if not isinstance(directory, str) or not directory.strip():
                    continue
                if _normalize_path_for_match(directory) == target_dir:
                    return True
            return False

        # Prefer project metadata to limit the search scope.
        projects_root = root / "project"
        candidate_ids: list[str] = []
        if projects_root.exists():
            try:
                project_files = list(projects_root.glob("*.json"))
            except Exception:
                project_files = []
            for path in project_files:
                if not path.is_file():
                    continue
                payload = _load_json(path)
                worktree = payload.get("worktree")
                if not isinstance(worktree, str) or not worktree.strip():
                    continue
                if _normalize_path_for_match(worktree) != target_dir:
                    continue
                pid = payload.get("id") if isinstance(payload.get("id"), str) and payload.get("id") else path.stem
                if pid:
                    candidate_ids.append(pid)

        for pid in candidate_ids:
            if _session_dir_has_match(sessions_root / pid):
                return True

        # Non-git workdirs often land under "global"; check it explicitly.
        if _session_dir_has_match(sessions_root / "global"):
            return True

        # Fallback: scan all session dirs (best-effort).
        try:
            project_dirs = list(sessions_root.iterdir())
        except Exception:
            project_dirs = []
        for project_dir in project_dirs:
            if not project_dir.is_dir():
                continue
            if _session_dir_has_match(project_dir):
                return True

        return False

    def _build_opencode_start_cmd(self) -> str:
        # OpenCode CLI (TUI). Allow override via env for custom wrappers.
        cmd = (os.environ.get("OPENCODE_START_CMD") or "opencode").strip() or "opencode"

        # Auto mode: relax permissions and enable web tools via config file.
        if self.auto:
            self._ensure_opencode_auto_config()

        # Resume mode: continue last session
        if self.resume:
            if self._opencode_resume_allowed():
                cmd = f"{cmd} --continue"
                print(f"🔁 {t('resuming_session', provider='OpenCode', session_id='')}")
            else:
                print(f"ℹ️ {t('no_history_fresh', provider='OpenCode')}")

        return cmd

    def _get_latest_droid_session_id(self) -> tuple[str | None, bool, Path | None]:
        """
        Returns (session_id, has_any_history, resume_dir).
        - session_id: latest Droid session id if found (from session_start).
        - has_any_history: whether this project has any Droid sessions on disk.
        - resume_dir: cwd from session_start (if it exists on disk).
        """
        try:
            from droid_comm import DroidLogReader, read_droid_session_start
        except Exception:
            return None, False, None

        reader = DroidLogReader(work_dir=self.project_root)
        session_path = reader.current_session_path()
        if not session_path or not session_path.exists():
            return None, False, None

        cwd, session_id = read_droid_session_start(session_path)
        resume_dir = None
        if isinstance(cwd, str) and cwd.strip():
            try:
                candidate = Path(cwd)
                if candidate.is_dir():
                    resume_dir = candidate
            except Exception:
                resume_dir = None
        return session_id or None, True, resume_dir

    def _build_droid_start_cmd(self) -> str:
        cmd = (os.environ.get("DROID_START_CMD") or "droid").strip() or "droid"
        resume_cmd = (os.environ.get("DROID_RESUME_CMD") or "").strip()
        if self.resume:
            session_id, has_history, resume_dir = self._get_latest_droid_session_id()
            if has_history:
                if resume_cmd:
                    cmd = resume_cmd
                else:
                    if resume_dir:
                        cmd = f"{_build_cd_cmd(resume_dir)}{cmd}"
                    cmd = f"{cmd} -r"
                sid_hint = session_id[:8] if session_id else ""
                print(f"🔁 {t('resuming_session', provider='Droid', session_id=sid_hint)}")
            else:
                print(f"ℹ️ {t('no_history_fresh', provider='Droid')}")
        return cmd

    def _ensure_opencode_auto_config(self) -> None:
        """
        Ensure project-local opencode.json exists with full allow permissions (auto mode).
        Best-effort: never raise (should not block startup).
        """
        try:
            config_path = Path.cwd() / "opencode.json"

            desired = {
                "permission": {
                    "edit": "allow",
                    "bash": "allow",
                    "skill": "allow",
                    "webfetch": "allow",
                    "doom_loop": "allow",
                    "external_directory": "allow",
                }
            }

            current: dict = {}
            if config_path.exists():
                try:
                    current_raw = config_path.read_text(encoding="utf-8")
                    current_obj = json.loads(current_raw)
                    if isinstance(current_obj, dict):
                        current = current_obj
                except Exception:
                    current = {}

            current["permission"] = dict(desired["permission"])

            tmp_path = config_path.with_suffix(".tmp")
            tmp_path.write_text(json.dumps(current, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
            os.replace(tmp_path, config_path)
        except Exception as exc:
            # Non-fatal; OpenCode can still run with interactive permissions.
            print(f"⚠️ Failed to update OpenCode config: {exc}", file=sys.stderr)

    def _start_codex_tmux(
        self,
        *,
        parent_pane: str | None = None,
        direction: str | None = None,
    ) -> str | None:
        runtime = self.runtime_dir / "codex"
        runtime.mkdir(parents=True, exist_ok=True)

        input_fifo = runtime / "input.fifo"
        output_fifo = runtime / "output.fifo"

        if not input_fifo.exists():
            os.mkfifo(input_fifo, 0o600)
        if not output_fifo.exists():
            os.mkfifo(output_fifo, 0o644)

        env_overrides = self._provider_env_overrides("codex")
        start_cmd = self._build_env_prefix(env_overrides) + _build_export_path_cmd(self.script_dir / "bin") + self._build_codex_start_cmd()
        pane_title_marker = "CCB-Codex"

        backend = TmuxBackend()

        use_direction = (direction or ("right" if not self.tmux_panes else "bottom")).strip() or "right"
        use_parent = parent_pane
        if not use_parent:
            try:
                use_parent = backend.get_current_pane_id()
            except Exception:
                use_parent = None
        if not use_parent and use_direction == "bottom":
            try:
                use_parent = next(reversed(self.tmux_panes.values()))
            except StopIteration:
                use_parent = None
        try:
            if use_parent and str(use_parent).startswith("%") and not backend.pane_exists(str(use_parent)):
                use_parent = backend.get_current_pane_id()
        except Exception:
            use_parent = None

        pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent)
        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")

        self.tmux_panes["codex"] = pane_id

        bridge_script = self.script_dir / "lib" / "codex_dual_bridge.py"
        bridge_env = os.environ.copy()
        bridge_env["CODEX_TERMINAL"] = "tmux"
        bridge_env["CODEX_TMUX_SESSION"] = pane_id
        bridge_env["CODEX_SESSION_ID"] = self.session_id
        bridge_env["CODEX_RUNTIME_DIR"] = str(runtime)
        bridge_env["CODEX_INPUT_FIFO"] = str(input_fifo)
        bridge_env["CODEX_OUTPUT_FIFO"] = str(output_fifo)
        bridge_env["CODEX_TMUX_LOG"] = str(runtime / "bridge_output.log")
        bridge_env["PYTHONPATH"] = str(self.script_dir) + os.pathsep + bridge_env.get("PYTHONPATH", "")

        try:
            cp = subprocess.run(
                ["tmux", "display-message", "-p", "-t", pane_id, "#{pane_pid}"],
                capture_output=True,
                text=True,
                encoding="utf-8",
                errors="replace",
                check=True,
            )
            pane_pid = (cp.stdout or "").strip()
            if pane_pid.isdigit():
                (runtime / "codex.pid").write_text(pane_pid + "\n")
        except Exception:
            pass

        bridge_proc = subprocess.Popen(
            [sys.executable, str(bridge_script), "--runtime-dir", str(runtime), "--session-id", self.session_id],
            env=bridge_env,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
        try:
            (runtime / "bridge.pid").write_text(str(bridge_proc.pid), encoding="utf-8")
        except Exception:
            pass

        self._write_codex_session(
            runtime,
            None,
            input_fifo,
            output_fifo,
            pane_id=pane_id,
            pane_title_marker=pane_title_marker,
            codex_start_cmd=start_cmd,
        )

        print(f"✅ {t('started_backend', provider='Codex', terminal='tmux pane', pane_id=pane_id)}")
        return pane_id

    def _start_gemini_tmux(
        self,
        *,
        parent_pane: str | None = None,
        direction: str | None = None,
    ) -> str | None:
        runtime = self.runtime_dir / "gemini"
        runtime.mkdir(parents=True, exist_ok=True)

        env_overrides = self._provider_env_overrides("gemini")
        start_cmd = self._build_env_prefix(env_overrides) + _build_export_path_cmd(self.script_dir / "bin") + self._build_gemini_start_cmd()
        pane_title_marker = "CCB-Gemini"

        backend = TmuxBackend()

        use_direction = (direction or ("right" if not self.tmux_panes else "bottom")).strip() or "right"
        use_parent = parent_pane
        if not use_parent:
            try:
                use_parent = backend.get_current_pane_id()
            except Exception:
                use_parent = None
        if not use_parent and use_direction == "bottom":
            try:
                use_parent = next(reversed(self.tmux_panes.values()))
            except StopIteration:
                use_parent = None
        try:
            if use_parent and str(use_parent).startswith("%") and not backend.pane_exists(str(use_parent)):
                use_parent = backend.get_current_pane_id()
        except Exception:
            use_parent = None

        pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent)
        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")

        self.tmux_panes["gemini"] = pane_id

        self._write_gemini_session(
            runtime,
            None,
            pane_id=pane_id,
            pane_title_marker=pane_title_marker,
            start_cmd=start_cmd,
        )

        print(f"✅ {t('started_backend', provider='Gemini', terminal='tmux pane', pane_id=pane_id)}")
        return pane_id

    def _start_opencode_tmux(
        self,
        *,
        parent_pane: str | None = None,
        direction: str | None = None,
    ) -> str | None:
        runtime = self.runtime_dir / "opencode"
        runtime.mkdir(parents=True, exist_ok=True)

        env_overrides = self._provider_env_overrides("opencode")
        start_cmd = self._build_env_prefix(env_overrides) + _build_export_path_cmd(self.script_dir / "bin") + self._build_opencode_start_cmd()
        pane_title_marker = "CCB-OpenCode"

        backend = TmuxBackend()

        use_direction = (direction or ("right" if not self.tmux_panes else "bottom")).strip() or "right"
        use_parent = parent_pane
        if not use_parent:
            try:
                use_parent = backend.get_current_pane_id()
            except Exception:
                use_parent = None
        if not use_parent and use_direction == "bottom":
            try:
                use_parent = next(reversed(self.tmux_panes.values()))
            except StopIteration:
                use_parent = None

        # If the preferred parent pane disappeared (e.g. a fast-exiting backend), fall back
        # to a live pane instead of crashing.
        try:
            if use_parent and str(use_parent).startswith("%") and not backend.pane_exists(str(use_parent)):
                use_parent = backend.get_current_pane_id()
        except Exception:
            use_parent = None

        pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent)
        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")

        self.tmux_panes["opencode"] = pane_id

        self._write_opencode_session(
            runtime,
            None,
            pane_id=pane_id,
            pane_title_marker=pane_title_marker,
            start_cmd=start_cmd,
        )

        print(f"✅ {t('started_backend', provider='OpenCode', terminal='tmux pane', pane_id=pane_id)}")
        return pane_id

    def _start_droid_tmux(
        self,
        *,
        parent_pane: str | None = None,
        direction: str | None = None,
    ) -> str | None:
        runtime = self.runtime_dir / "droid"
        runtime.mkdir(parents=True, exist_ok=True)

        env_overrides = self._provider_env_overrides("droid")
        start_cmd = self._build_env_prefix(env_overrides) + _build_export_path_cmd(self.script_dir / "bin") + self._build_droid_start_cmd()
        pane_title_marker = "CCB-Droid"

        backend = TmuxBackend()

        use_direction = (direction or ("right" if not self.tmux_panes else "bottom")).strip() or "right"
        use_parent = parent_pane
        if not use_parent:
            try:
                use_parent = backend.get_current_pane_id()
            except Exception:
                use_parent = None
        if not use_parent and use_direction == "bottom":
            try:
                use_parent = next(reversed(self.tmux_panes.values()))
            except StopIteration:
                use_parent = None

        try:
            if use_parent and str(use_parent).startswith("%") and not backend.pane_exists(str(use_parent)):
                use_parent = backend.get_current_pane_id()
        except Exception:
            use_parent = None

        pane_id = backend.create_pane("", str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent)
        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")

        self.tmux_panes["droid"] = pane_id

        self._write_droid_session(
            runtime,
            None,
            pane_id=pane_id,
            pane_title_marker=pane_title_marker,
            start_cmd=start_cmd,
        )

        print(f"✅ {t('started_backend', provider='Droid', terminal='tmux pane', pane_id=pane_id)}")
        return pane_id

    def _start_cmd_pane(
        self,
        *,
        parent_pane: str | None,
        direction: str | None,
        cmd_settings: dict,
    ) -> str | None:
        if not cmd_settings.get("enabled"):
            return None
        title = (cmd_settings.get("title") or "CCB-Cmd").strip() or "CCB-Cmd"
        start_cmd = (cmd_settings.get("start_cmd") or "").strip() or self._default_cmd_start_cmd()
        full_cmd = (
            _build_pane_title_cmd(title)
            + self._build_env_prefix(self._provider_env_overrides("codex"))
            + _build_export_path_cmd(self.script_dir / "bin")
            + _build_cd_cmd(Path.cwd())
            + start_cmd
        )

        use_direction = (direction or "right").strip() or "right"
        use_parent = parent_pane

        if self.terminal_type == "wezterm":
            backend = WeztermBackend()
            pane_id = backend.create_pane(full_cmd, str(Path.cwd()), direction=use_direction, percent=50, parent_pane=use_parent)
            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)
            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")
            self.extra_panes["cmd"] = pane_id

        print(f"✅ Started cmd pane ({pane_id})")
        return pane_id

    def _start_codex_current_pane(self) -> int:
        runtime = self.runtime_dir / "codex"
        runtime.mkdir(parents=True, exist_ok=True)

        input_fifo = runtime / "input.fifo"
        output_fifo = runtime / "output.fifo"

        pane_id = self._current_pane_id()
        if not pane_id:
            print("❌ Unable to determine current pane id for Codex", file=sys.stderr)
            return 1

        pane_title_marker = "CCB-Codex"
        if self.terminal_type == "tmux":
            try:
                backend = TmuxBackend()
                backend.set_pane_title(pane_id, pane_title_marker)
                backend.set_pane_user_option(pane_id, "@ccb_agent", "Codex")
            except Exception:
                pass

        if self.terminal_type == "tmux":
            if not input_fifo.exists():
                os.mkfifo(input_fifo, 0o600)
            if not output_fifo.exists():
                os.mkfifo(output_fifo, 0o644)

        title_cmd = _build_pane_title_cmd(pane_title_marker) if self.terminal_type == "wezterm" else ""
        start_cmd = (
            title_cmd
            + self._build_env_prefix(self._provider_env_overrides("codex"))
            + _build_export_path_cmd(self.script_dir / "bin")
            + self._build_codex_start_cmd()
        )

        self._write_codex_session(
            runtime,
            None,
            input_fifo,
            output_fifo,
            pane_id=pane_id,
            pane_title_marker=pane_title_marker,
            codex_start_cmd=start_cmd,
        )

        # Optional: start caskd after Codex session file exists (first startup convenience).
        self._maybe_start_caskd()

        if self.terminal_type == "tmux":
            bridge_script = self.script_dir / "lib" / "codex_dual_bridge.py"
            bridge_env = os.environ.copy()
            bridge_env["CODEX_TERMINAL"] = "tmux"
            bridge_env["CODEX_TMUX_SESSION"] = pane_id
            bridge_env["CODEX_SESSION_ID"] = self.session_id
            bridge_env["CODEX_RUNTIME_DIR"] = str(runtime)
            bridge_env["CODEX_INPUT_FIFO"] = str(input_fifo)
            bridge_env["CODEX_OUTPUT_FIFO"] = str(output_fifo)
            bridge_env["CODEX_TMUX_LOG"] = str(runtime / "bridge_output.log")
            bridge_env["PYTHONPATH"] = str(self.script_dir) + os.pathsep + bridge_env.get("PYTHONPATH", "")

            bridge_proc = subprocess.Popen(
                [sys.executable, str(bridge_script), "--runtime-dir", str(runtime), "--session-id", self.session_id],
                env=bridge_env,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                start_new_session=True,
            )
            try:
                (runtime / "bridge.pid").write_text(str(bridge_proc.pid), encoding="utf-8")
            except Exception:
                pass

        # Windows 下全局安装的 codex-cli 常见形态是 npm 的 shim（`codex.cmd`）。
        # Python 的 subprocess 在传入 argv 列表时会走 CreateProcess，无法像 shell 一样解析 `.cmd`，
        # 因而会触发 [WinError 2] 找不到可执行文件。这里改为通过 shell 执行以确保可启动。
        if os.name == "nt":
            env = self._with_bin_path_env()
            env.update(self._provider_env_overrides("codex"))
            env["CODEX_SESSION_ID"] = self.session_id
            env["CODEX_RUNTIME_DIR"] = str(runtime)
            env["CODEX_INPUT_FIFO"] = str(input_fifo)
            env["CODEX_OUTPUT_FIFO"] = str(output_fifo)
            env["CODEX_TERMINAL"] = self.terminal_type
            if self.terminal_type == "wezterm":
                env["CODEX_WEZTERM_PANE"] = pane_id
            else:
                env["CODEX_TMUX_SESSION"] = pane_id
            return self._run_shell_command(self._build_codex_start_cmd(), env=env, cwd=str(Path.cwd()))

        cmd_parts = shlex.split(self._build_codex_start_cmd())
        env = self._with_bin_path_env()
        env.update(self._provider_env_overrides("codex"))
        env["CODEX_SESSION_ID"] = self.session_id
        env["CODEX_RUNTIME_DIR"] = str(runtime)
        env["CODEX_INPUT_FIFO"] = str(input_fifo)
        env["CODEX_OUTPUT_FIFO"] = str(output_fifo)
        env["CODEX_TERMINAL"] = self.terminal_type
        if self.terminal_type == "wezterm":
            env["CODEX_WEZTERM_PANE"] = pane_id
        else:
            env["CODEX_TMUX_SESSION"] = pane_id

        try:
            proc = subprocess.Popen(cmd_parts, env=env, cwd=str(Path.cwd()))
        except Exception as exc:
            print(f"❌ Failed to start Codex: {exc}", file=sys.stderr)
            return 1

        try:
            if self.terminal_type == "tmux":
                (runtime / "codex.pid").write_text(str(proc.pid) + "\n", encoding="utf-8")
        except Exception:
            pass

        return proc.wait()

    def _start_gemini_current_pane(self) -> int:
        runtime = self.runtime_dir / "gemini"
        runtime.mkdir(parents=True, exist_ok=True)

        pane_id = self._current_pane_id()
        if not pane_id:
            print("❌ Unable to determine current pane id for Gemini", file=sys.stderr)
            return 1

        pane_title_marker = "CCB-Gemini"
        if self.terminal_type == "tmux":
            try:
                backend = TmuxBackend()
                backend.set_pane_title(pane_id, pane_title_marker)
                backend.set_pane_user_option(pane_id, "@ccb_agent", "Gemini")
            except Exception:
                pass

        title_cmd = _build_pane_title_cmd(pane_title_marker) if self.terminal_type == "wezterm" else ""
        start_cmd = (
            title_cmd
            + self._build_env_prefix(self._provider_env_overrides("gemini"))
            + _build_export_path_cmd(self.script_dir / "bin")
            + self._build_gemini_start_cmd()
        )
        self._write_gemini_session(
            runtime,
            None,
            pane_id=pane_id,
            pane_title_marker=pane_title_marker,
            start_cmd=start_cmd,
        )

        return self._run_shell_command(start_cmd, cwd=str(Path.cwd()))

    def _start_opencode_current_pane(self) -> int:
        runtime = self.runtime_dir / "opencode"
        runtime.mkdir(parents=True, exist_ok=True)

        pane_id = self._current_pane_id()
        if not pane_id:
            print("❌ Unable to determine current pane id for OpenCode", file=sys.stderr)
            return 1

        pane_title_marker = "CCB-OpenCode"
        if self.terminal_type == "tmux":
            try:
                backend = TmuxBackend()
                backend.set_pane_title(pane_id, pane_title_marker)
                backend.set_pane_user_option(pane_id, "@ccb_agent", "OpenCode")
            except Exception:
                pass

        title_cmd = _build_pane_title_cmd(pane_title_marker) if self.terminal_type == "wezterm" else ""
        opencode_cmd = self._build_opencode_start_cmd()
        start_cmd = (
            title_cmd
            + self._build_env_prefix(self._provider_env_overrides("opencode"))
            + _build_export_path_cmd(self.script_dir / "bin")
            + opencode_cmd
        )
        self._write_opencode_session(
            runtime,
            None,
            pane_id=pane_id,
            pane_title_marker=pane_title_marker,
            start_cmd=start_cmd,
        )

        # tmux path: run OpenCode directly (without shell -lc) to avoid shell startup side-effects
        # that can make the TUI appear to hang before rendering.
        if self.terminal_type == "tmux" and os.name != "nt":
            env = self._with_bin_path_env()
            env.update(self._provider_env_overrides("opencode"))
            env["OPENCODE_SESSION_ID"] = self.session_id
            env["OPENCODE_RUNTIME_DIR"] = str(runtime)
            env["OPENCODE_TERMINAL"] = self.terminal_type
            env["OPENCODE_TMUX_SESSION"] = pane_id
            try:
                cmd_parts = shlex.split(opencode_cmd)
                if not cmd_parts:
                    print("❌ Empty OpenCode start command", file=sys.stderr)
                    return 1
                # Single-provider `ccb opencode`: fully hand over the pane process to OpenCode.
                # This mirrors manual `opencode` startup and avoids a long-lived Python parent.
                exec_raw = (os.environ.get("CCB_OPENCODE_EXEC") or "").strip().lower()
                use_exec = (exec_raw not in {"0", "false", "no", "off"}) and (len(self.providers) == 1)
                if use_exec:
                    os.execvpe(cmd_parts[0], cmd_parts, env)
                return subprocess.run(cmd_parts, env=env, cwd=str(Path.cwd())).returncode
            except Exception as exc:
                print(f"❌ Failed to start OpenCode: {exc}", file=sys.stderr)
                return 1

        return self._run_shell_command(start_cmd, cwd=str(Path.cwd()))

    def _start_droid_current_pane(self) -> int:
        runtime = self.runtime_dir / "droid"
        runtime.mkdir(parents=True, exist_ok=True)

        pane_id = self._current_pane_id()
        if not pane_id:
            print("❌ Unable to determine current pane id for Droid", file=sys.stderr)
            return 1

        pane_title_marker = "CCB-Droid"
        if self.terminal_type == "tmux":
            try:
                backend = TmuxBackend()
                backend.set_pane_title(pane_id, pane_title_marker)
                backend.set_pane_user_option(pane_id, "@ccb_agent", "Droid")
            except Exception:
                pass

        title_cmd = _build_pane_title_cmd(pane_title_marker) if self.terminal_type == "wezterm" else ""
        start_cmd = (
            title_cmd
            + self._build_env_prefix(self._provider_env_overrides("droid"))
            + _build_export_path_cmd(self.script_dir / "bin")
            + self._build_droid_start_cmd()
        )
        self._write_droid_session(
            runtime,
            None,
            pane_id=pane_id,
            pane_title_marker=pane_title_marker,
            start_cmd=start_cmd,
        )

        return self._run_shell_command(start_cmd, cwd=str(Path.cwd()))

    def _start_provider_in_current_pane(self, provider: str) -> int:
        if provider == "claude":
            return self._start_claude()
        if provider == "codex":
            return self._start_codex_current_pane()
        if provider == "gemini":
            return self._start_gemini_current_pane()
        if provider == "opencode":
            return self._start_opencode_current_pane()
        if provider == "droid":
            return self._start_droid_current_pane()
        print(f"❌ {t('unknown_provider', provider=provider)}")
        return 1

    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")

        # Pre-check permissions
        writable, reason, fix = check_session_writable(session_file)
        if not writable:
            print(f"❌ Cannot write {session_file.name}: {reason}", file=sys.stderr)
            print(f"💡 Fix: {fix}", file=sys.stderr)
            return False

        data = {}
        if session_file.exists():
            data = self._read_json_file(session_file)

        if not self.resume:
            data = self._clear_codex_log_binding(data)

        work_dir = self.project_root
        data.update({
            "session_id": self.session_id,
            "ccb_project_id": compute_ccb_project_id(work_dir),
            "runtime_dir": str(runtime),
            "input_fifo": str(input_fifo),
            "output_fifo": str(output_fifo),
            "terminal": self.terminal_type,
            "tmux_session": tmux_session,
            "pane_id": pane_id,
            "pane_title_marker": pane_title_marker,
            "tmux_log": str(runtime / "bridge_output.log"),
            "work_dir": str(work_dir),
            "work_dir_norm": _normalize_path_for_match(str(work_dir)),
            "start_dir": str(self.invocation_dir),
            "active": True,
            "started_at": time.strftime("%Y-%m-%d %H:%M:%S"),
        })
        if codex_start_cmd:
            data["codex_start_cmd"] = str(codex_start_cmd)
            data["start_cmd"] = str(codex_start_cmd)

        ok, err = safe_write_session(session_file, json.dumps(data, ensure_ascii=False, indent=2))
        if not ok:
            print(err, file=sys.stderr)
            return False
        try:
            upsert_registry({
                "ccb_session_id": self.session_id,
                "ccb_project_id": compute_ccb_project_id(self.project_root),
                "work_dir": str(self.project_root),
                "terminal": self.terminal_type,
                "providers": {
                    "codex": {
                        "pane_id": pane_id,
                        "pane_title_marker": pane_title_marker,
                        "session_file": str(session_file),
                    }
                },
            })
        except Exception:
            pass
        self._maybe_start_provider_daemon("codex")
        return True

    def _write_cend_registry(self, claude_pane_id: str, codex_pane_id: str | None) -> bool:
        if not claude_pane_id:
            return False
        record = {
            "ccb_session_id": self.session_id,
            "claude_pane_id": claude_pane_id,
            "codex_pane_id": codex_pane_id,
            "ccb_project_id": compute_ccb_project_id(self.project_root),
            "work_dir": str(Path.cwd()),
            "terminal": self.terminal_type,
            "providers": {
                "claude": {"pane_id": claude_pane_id},
                "codex": {"pane_id": codex_pane_id},
            },
        }
        ok = upsert_registry(record)
        if not ok:
            print("⚠️ Failed to update cpend registry", file=sys.stderr)
        return ok

    def _sync_cend_registry(self) -> None:
        if "codex" not in self.providers or "claude" not in self.providers:
            return
        codex_pane_id = self._provider_pane_id("codex")
        claude_pane_id = self._provider_pane_id("claude")
        if codex_pane_id and claude_pane_id:
            self._write_cend_registry(claude_pane_id, codex_pane_id)

    def _write_gemini_session(self, runtime, tmux_session, pane_id=None, pane_title_marker=None, start_cmd=None):
        session_file = self._project_session_file(".gemini-session")

        # Pre-check permissions
        writable, reason, fix = check_session_writable(session_file)
        if not writable:
            print(f"❌ Cannot write {session_file.name}: {reason}", file=sys.stderr)
            print(f"💡 Fix: {fix}", file=sys.stderr)
            return False

        data = {
            "session_id": self.session_id,
            "ccb_project_id": compute_ccb_project_id(self.project_root),
            "runtime_dir": str(runtime),
            "terminal": self.terminal_type,
            "tmux_session": tmux_session,
            "pane_id": pane_id,
            "pane_title_marker": pane_title_marker,
            "work_dir": str(self.project_root),
            "work_dir_norm": _normalize_path_for_match(str(self.project_root)),
            "start_dir": str(self.invocation_dir),
            "active": True,
            "started_at": time.strftime("%Y-%m-%d %H:%M:%S"),
            "start_cmd": str(start_cmd) if start_cmd else None,
        }

        ok, err = safe_write_session(session_file, json.dumps(data, ensure_ascii=False, indent=2))
        if not ok:
            print(err, file=sys.stderr)
            return False
        try:
            upsert_registry({
                "ccb_session_id": self.session_id,
                "ccb_project_id": compute_ccb_project_id(self.project_root),
                "work_dir": str(self.project_root),
                "terminal": self.terminal_type,
                "providers": {
                    "gemini": {
                        "pane_id": pane_id,
                        "pane_title_marker": pane_title_marker,
                        "session_file": str(session_file),
                    }
                },
            })
        except Exception:
            pass
        self._maybe_start_provider_daemon("gemini")
        return True

    def _write_opencode_session(self, runtime, tmux_session, pane_id=None, pane_title_marker=None, start_cmd=None):
        session_file = self._project_session_file(".opencode-session")

        writable, reason, fix = check_session_writable(session_file)
        if not writable:
            print(f"❌ Cannot write {session_file.name}: {reason}", file=sys.stderr)
            print(f"💡 Fix: {fix}", file=sys.stderr)
            return False

        data = {
            "session_id": self.session_id,
            "ccb_session_id": self.session_id,
            "ccb_project_id": compute_ccb_project_id(self.project_root),
            "runtime_dir": str(runtime),
            "terminal": self.terminal_type,
            "tmux_session": tmux_session,
            "pane_id": pane_id,
            "pane_title_marker": pane_title_marker,
            "work_dir": str(self.project_root),
            "work_dir_norm": _normalize_path_for_match(str(self.project_root)),
            "start_dir": str(self.invocation_dir),
            "active": True,
            "started_at": time.strftime("%Y-%m-%d %H:%M:%S"),
            "start_cmd": str(start_cmd) if start_cmd else None,
        }

        ok, err = safe_write_session(session_file, json.dumps(data, ensure_ascii=False, indent=2))
        if not ok:
            print(err, file=sys.stderr)
            return False
        try:
            upsert_registry({
                "ccb_session_id": self.session_id,
                "ccb_project_id": compute_ccb_project_id(self.project_root),
                "work_dir": str(self.project_root),
                "terminal": self.terminal_type,
                "providers": {
                    "opencode": {
                        "pane_id": pane_id,
                        "pane_title_marker": pane_title_marker,
                        "session_file": str(session_file),
                    }
                },
            })
        except Exception:
            pass
        self._maybe_start_provider_daemon("opencode")
        return True

    def _write_droid_session(self, runtime, tmux_session, pane_id=None, pane_title_marker=None, start_cmd=None):
        session_file = self._project_session_file(".droid-session")

        writable, reason, fix = check_session_writable(session_file)
        if not writable:
            print(f"❌ Cannot write {session_file.name}: {reason}", file=sys.stderr)
            print(f"💡 Fix: {fix}", file=sys.stderr)
            return False

        droid_session_path = None
        droid_session_id = None
        try:
            from droid_comm import DroidLogReader, read_droid_session_start

            reader = DroidLogReader(work_dir=self.project_root)
            session_path = reader.current_session_path()
            if session_path and session_path.exists():
                droid_session_path = str(session_path)
                _cwd, session_id = read_droid_session_start(session_path)
                if session_id:
                    droid_session_id = session_id
        except Exception:
            pass

        data = {
            "session_id": self.session_id,
            "ccb_session_id": self.session_id,
            "ccb_project_id": compute_ccb_project_id(self.project_root),
            "runtime_dir": str(runtime),
            "terminal": self.terminal_type,
            "tmux_session": tmux_session,
            "pane_id": pane_id,
            "pane_title_marker": pane_title_marker,
            "work_dir": str(self.project_root),
            "work_dir_norm": _normalize_path_for_match(str(self.project_root)),
            "start_dir": str(self.invocation_dir),
            "active": True,
            "started_at": time.strftime("%Y-%m-%d %H:%M:%S"),
            "start_cmd": str(start_cmd) if start_cmd else None,
            "droid_session_id": droid_session_id,
            "droid_session_path": droid_session_path,
        }

        ok, err = safe_write_session(session_file, json.dumps(data, ensure_ascii=False, indent=2))
        if not ok:
            print(err, file=sys.stderr)
            return False
        try:
            upsert_registry({
                "ccb_session_id": self.session_id,
                "ccb_project_id": compute_ccb_project_id(self.project_root),
                "work_dir": str(self.project_root),
                "terminal": self.terminal_type,
                "providers": {
                    "droid": {
                        "pane_id": pane_id,
                        "pane_title_marker": pane_title_marker,
                        "session_file": str(session_file),
                        "droid_session_id": droid_session_id,
                        "droid_session_path": droid_session_path,
                    }
                },
            })
        except Exception:
            pass
        self._maybe_start_provider_daemon("droid")
        return True

    def _claude_project_dir(self, work_dir: Path) -> Path:
        projects_root = Path.home() / ".claude" / "projects"
        # Claude Code uses a filesystem-friendly key derived from the working directory.
        # To handle symlinked paths (PWD) vs physical paths (resolve()), try multiple candidates.
        candidates: list[Path] = []
        env_pwd = os.environ.get("PWD")
        if env_pwd:
            try:
                candidates.append(Path(env_pwd))
            except Exception:
                pass
        candidates.extend([work_dir])
        try:
            candidates.append(work_dir.resolve())
        except Exception:
            pass

        for candidate in candidates:
            key = re.sub(r"[^A-Za-z0-9]", "-", str(candidate))
            project_dir = projects_root / key
            if project_dir.exists():
                return project_dir

        # Fallback to a best-effort key even if the directory doesn't exist yet.
        try:
            fallback_path = work_dir.resolve()
        except Exception:
            fallback_path = work_dir
        key = re.sub(r"[^A-Za-z0-9]", "-", str(fallback_path))
        return projects_root / key

    def _get_latest_claude_session_id(self) -> tuple[str | None, bool, Path | None]:
        """
        Returns (session_id, has_any_history).
        - session_id: latest UUID-like session id if found (for `claude --resume <id>`).
        - has_any_history: whether this project has any Claude sessions on disk.
        """
        candidates: list[Path] = []
        for candidate in (Path.cwd(), self.invocation_dir, self.project_root):
            try:
                p = candidate.resolve()
            except Exception:
                p = candidate.absolute()
            if p not in candidates:
                candidates.append(p)

        session_env_root = Path.home() / ".claude" / "session-env"
        best_uuid: Path | None = None
        best_any: Path | None = None
        has_any_history = False
        best_cwd: Path | None = None

        for work_dir in candidates:
            project_dir = self._claude_project_dir(work_dir)
            if not project_dir.exists():
                continue

            session_files = list(project_dir.glob("*.jsonl"))
            if not session_files:
                continue
            has_any_history = True

            try:
                best_in_dir = max(session_files, key=lambda p: p.stat().st_mtime)
                if best_any is None or best_in_dir.stat().st_mtime > best_any.stat().st_mtime:
                    best_any = best_in_dir
                    best_cwd = work_dir
            except Exception:
                pass

            for session_file in session_files:
                try:
                    uuid.UUID(session_file.stem)
                    if session_file.stat().st_size <= 0:
                        continue
                    if not (session_env_root / session_file.stem).exists():
                        continue
                except Exception:
                    continue
                if best_uuid is None:
                    best_uuid = session_file
                    best_cwd = work_dir
                    continue
                try:
                    if session_file.stat().st_mtime > best_uuid.stat().st_mtime:
                        best_uuid = session_file
                        best_cwd = work_dir
                except Exception:
                    continue

        if best_uuid is not None:
            return best_uuid.stem, True, best_cwd
        if has_any_history:
            return None, True, best_cwd
        return None, False, None

    def _find_claude_cmd(self) -> str:
        """Find Claude CLI executable"""
        if sys.platform == "win32":
            for cmd in ["claude.exe", "claude.cmd", "claude.bat", "claude"]:
                path = shutil.which(cmd)
                if path:
                    return path
            npm_paths = [
                Path(os.environ.get("APPDATA", "")) / "npm" / "claude.cmd",
                Path(os.environ.get("ProgramFiles", "")) / "nodejs" / "claude.cmd",
            ]
            for npm_path in npm_paths:
                if npm_path.exists():
                    return str(npm_path)
        else:
            path = shutil.which("claude")
            if path:
                return path
        raise FileNotFoundError(
            "❌ Claude CLI not found. Install: npm install -g @anthropic-ai/claude-code"
        )

    def _claude_env_overrides(self) -> dict:
        env: dict[str, str] = {}
        env.update(self._provider_env_overrides("claude"))
        if "codex" in self.providers:
            runtime = self.runtime_dir / "codex"
            env["CODEX_SESSION_ID"] = self.session_id
            env["CODEX_RUNTIME_DIR"] = str(runtime)
            env["CODEX_INPUT_FIFO"] = str(runtime / "input.fifo")
            env["CODEX_OUTPUT_FIFO"] = str(runtime / "output.fifo")
            env["CODEX_TERMINAL"] = self.terminal_type or ""
            pane_id = self._provider_pane_id("codex")
            if self.terminal_type == "wezterm":
                env["CODEX_WEZTERM_PANE"] = pane_id
            else:
                env["CODEX_TMUX_SESSION"] = pane_id

        if "gemini" in self.providers:
            runtime = self.runtime_dir / "gemini"
            env["GEMINI_SESSION_ID"] = self.session_id
            env["GEMINI_RUNTIME_DIR"] = str(runtime)
            env["GEMINI_TERMINAL"] = self.terminal_type or ""
            pane_id = self._provider_pane_id("gemini")
            if self.terminal_type == "wezterm":
                env["GEMINI_WEZTERM_PANE"] = pane_id
            else:
                env["GEMINI_TMUX_SESSION"] = pane_id

        if "opencode" in self.providers:
            runtime = self.runtime_dir / "opencode"
            env["OPENCODE_SESSION_ID"] = self.session_id
            env["OPENCODE_RUNTIME_DIR"] = str(runtime)
            env["OPENCODE_TERMINAL"] = self.terminal_type or ""
            pane_id = self._provider_pane_id("opencode")
            if self.terminal_type == "wezterm":
                env["OPENCODE_WEZTERM_PANE"] = pane_id
            else:
                env["OPENCODE_TMUX_SESSION"] = pane_id

        if "droid" in self.providers:
            runtime = self.runtime_dir / "droid"
            env["DROID_SESSION_ID"] = self.session_id
            env["DROID_RUNTIME_DIR"] = str(runtime)
            env["DROID_TERMINAL"] = self.terminal_type or ""
            pane_id = self._provider_pane_id("droid")
            if self.terminal_type == "wezterm":
                env["DROID_WEZTERM_PANE"] = pane_id
            else:
                env["DROID_TMUX_SESSION"] = pane_id

        return env

    def _build_claude_env(self) -> dict:
        env = self._with_bin_path_env()
        env.update(self._claude_env_overrides())
        return env

    def _claude_start_plan(self) -> tuple[list[str], str, bool]:
        claude_cmd = self._find_claude_cmd()
        cmd = [claude_cmd]
        if self.auto:
            cmd.append("--dangerously-skip-permissions")
        has_history = False
        resume_dir = None
        if self.resume:
            _, has_history, resume_dir = self._get_latest_claude_session_id()
            if has_history:
                cmd.append("--continue")
        run_cwd = str(self.project_root) if self.resume else str(Path.cwd())
        if self.resume and has_history and resume_dir:
            run_cwd = str(resume_dir)
        return cmd, run_cwd, has_history

    def _start_claude(self) -> int:
        print(f"🚀 {t('starting_claude')}")
        env = self._build_claude_env()

        try:
            cmd, run_cwd, has_history = self._claude_start_plan()
        except FileNotFoundError as e:
            print(str(e))
            return 1

        if self.resume:
            if has_history:
                print(f"🔁 {t('resuming_claude', session_id='')}")
            else:
                print(f"ℹ️ {t('no_claude_session')}")

        print(f"📋 Session ID: {self.session_id}")
        print(f"📁 Runtime dir: {self.runtime_dir}")
        print(f"🔌 Active backends: {', '.join(self.providers)}")
        print()
        print("🎯 Available commands:")
        if "codex" in self.providers:
            print("   cask/cping/cpend - Codex communication")
        if "gemini" in self.providers:
            print("   gask/gping/gpend - Gemini communication")
        if "opencode" in self.providers:
            print("   oask/oping/opend - OpenCode communication")
        if "droid" in self.providers:
            print("   dask/dping/dpend - Droid communication")
        print()
        print(f"Executing: {' '.join(cmd)}")

        try:
            # Let subprocess inherit stdio by default. Explicitly passing sys.stdin/out/err
            # can produce non-console handles on Windows (especially after re-wrapping
            # stdout/stderr for UTF-8), which may trigger issues in Node-based CLIs.
            return subprocess.run(cmd, env=env, cwd=run_cwd).returncode
        except KeyboardInterrupt:
            print(f"\n⚠️ {t('user_interrupted')}")
            return 130

    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()

        try:
            cmd_parts, run_cwd, has_history = self._claude_start_plan()
        except FileNotFoundError as e:
            print(str(e))
            return None

        if self.resume:
            if has_history:
                print(f"🔁 {t('resuming_claude', session_id='')}")
            else:
                print(f"ℹ️ {t('no_claude_session')}")

        start_cmd = " ".join(cmd_parts)
        full_cmd = (
            _build_pane_title_cmd("CCB-Claude")
            + self._build_env_prefix(env_overrides)
            + _build_export_path_cmd(self.script_dir / "bin")
            + start_cmd
        )

        use_direction = (direction or "right").strip() or "right"
        use_parent = parent_pane

        if self.terminal_type == "wezterm":
            backend = WeztermBackend()
            pane_id = backend.create_pane(full_cmd, run_cwd, direction=use_direction, percent=50, parent_pane=use_parent)
            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)
            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")
            self.tmux_panes["claude"] = pane_id

        try:
            self._write_local_claude_session(
                session_id=self._read_local_claude_session_id(),
                active=True,
                pane_id=str(pane_id or ""),
                pane_title_marker="CCB-Claude",
                terminal=self.terminal_type,
            )
        except Exception:
            pass

        print(f"✅ {t('started_backend', provider='Claude', terminal=f'{self.terminal_type} pane', pane_id=pane_id)}")
        return pane_id

    def cleanup(
        self,
        *,
        kill_panes: bool = True,
        clear_sessions: bool = True,
        remove_runtime: bool = True,
        quiet: bool = False,
    ):
        if self._cleaned:
            return
        self._cleaned = True

        # Stop watchdog thread first
        self._stop_daemon_watchdog()

        if not quiet:
            print(f"\n🧹 {t('cleaning_up')}")

        # Revert tmux UI theming early so the user's session returns to normal even if cleanup fails later.
        try:
            self._set_tmux_ui_active(False)
        except Exception:
            pass

        # Best-effort cleanup of leftover `tmpclaude-*-cwd` artifacts.
        try:
            _cleanup_tmpclaude_artifacts()
        except Exception:
            pass

        # Best-effort cleanup of stale runtime dirs in temp (crash leftovers).
        try:
            _cleanup_stale_runtime_dirs(exclude=self.runtime_dir)
        except Exception:
            pass

        # Best-effort daemon log slimming (cache/legacy dirs).
        try:
            _shrink_ccb_logs()
        except Exception:
            pass

        if kill_panes:
            if self.terminal_type == "wezterm":
                backend = WeztermBackend()
                for pane_id in list(self.wezterm_panes.values()) + list(self.extra_panes.values()):
                    if pane_id:
                        backend.kill_pane(pane_id)
            else:
                backend = TmuxBackend()
                for pane_id in list(self.tmux_panes.values()) + list(self.extra_panes.values()):
                    if pane_id:
                        backend.kill_pane(pane_id)

        if clear_sessions:
            for session_file in [
                self._project_session_file(".codex-session"),
                self._project_session_file(".gemini-session"),
                self._project_session_file(".opencode-session"),
                self._project_session_file(".claude-session"),
                self._project_session_file(".droid-session"),
            ]:
                if session_file.exists():
                    try:
                        data = self._read_json_file(session_file)
                        if not data:
                            continue
                        data["active"] = False
                        data["ended_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
                        safe_write_session(session_file, json.dumps(data, ensure_ascii=False, indent=2))
                    except Exception:
                        pass

        # Ensure unified askd daemon exits when CCB exits (with ownership safety).
        try:
            askd_state = state_file_path("askd.json")

            # Read askd state to check ownership
            askd_st = {}
            if askd_state and askd_state.exists():
                try:
                    with open(askd_state, "r", encoding="utf-8") as f:
                        askd_st = json.load(f)
                except Exception:
                    pass

            # Check if we own this daemon or if forced shutdown is enabled
            force_rebind = _env_bool("CCB_FORCE_REBIND", False)
            owned_by_us = False
            try:
                parent_pid = int((askd_st or {}).get("parent_pid") or 0)
                owned_by_us = (parent_pid == self.ccb_pid)
            except Exception:
                pass

            # Only shutdown if we own it, or force flag is set, or owner is dead
            should_shutdown = False
            if force_rebind:
                should_shutdown = True
            elif owned_by_us:
                should_shutdown = True
            else:
                # Check if foreign owner is still alive
                try:
                    parent_pid = int((askd_st or {}).get("parent_pid") or 0)
                    if parent_pid > 0:
                        foreign_alive = _is_pid_alive(parent_pid)
                        if not foreign_alive:
                            should_shutdown = True  # Owner is dead, safe to cleanup
                except Exception:
                    pass

            if should_shutdown:
                shutdown_daemon("ask", 1.0, askd_state)
        except Exception:
            pass

        if remove_runtime:
            import shutil
            if self.runtime_dir.exists():
                shutil.rmtree(self.runtime_dir, ignore_errors=True)

        if not quiet:
            print(f"✅ {t('cleanup_complete')}")

    def run_up(self) -> int:
        git_info = _get_git_info()
        version_str = f"v{VERSION}" + (f" ({git_info})" if git_info else "")
        print(f"🚀 Claude Code Bridge {version_str}")
        print(f"📅 {time.strftime('%Y-%m-%d %H:%M:%S')}")
        print(f"🔌 Backends: {', '.join(self.providers)}")
        print("=" * 50)

        # Require WezTerm pane or an existing tmux client session.
        # In plain terminals, users must run tmux first (no auto-launch).
        inside_tmux = bool(os.environ.get("TMUX") or os.environ.get("TMUX_PANE"))
        inside_wezterm = bool(os.environ.get("WEZTERM_PANE"))

        if self.terminal_type == "wezterm" and not inside_wezterm:
            self.terminal_type = None
        if self.terminal_type == "tmux" and not inside_tmux:
            self.terminal_type = None

        if self.terminal_type is None:
            print(f"❌ {t('no_terminal_backend')}", file=sys.stderr)
            print(f"   {t('solutions')}", file=sys.stderr)
            print(f"   - {t('install_wezterm')}", file=sys.stderr)
            print(f"   - {t('or_install_tmux')}", file=sys.stderr)
            if (shutil.which('tmux') or shutil.which('tmux.exe')):
                print(f"   - {t('tmux_installed_not_inside')}", file=sys.stderr)
            print(f"   - {t('or_set_ccb_terminal')}", file=sys.stderr)
            return 2

        if not self._require_project_config_dir():
            return 2

        self._backfill_claude_session_work_dir_fields()

        if not self.providers:
            print("❌ No providers configured. Define providers in ccb.config or pass them on the command line.", file=sys.stderr)
            return 2

        self.anchor_provider = self.providers[-1]
        self.anchor_pane_id = self._current_pane_id()
        if not self.anchor_pane_id:
            print("❌ Unable to determine current pane id. Run inside tmux or WezTerm.", file=sys.stderr)
            return 2

        cmd_settings = self._cmd_settings()
        spawn_items: list[str] = []
        if cmd_settings.get("enabled"):
            spawn_items.append("cmd")
        spawn_items.extend(list(reversed(self.providers[:-1])))
        total_panes = 1 + len(spawn_items)
        left_count = 1 if total_panes <= 1 else max(1, total_panes // 2)
        right_count = total_panes - left_count

        extras = list(spawn_items)
        right_top_item = extras[0] if right_count > 0 and extras else None
        remaining = extras[1:] if right_top_item else extras

        left_slots = max(0, left_count - 1)
        right_slots = max(0, right_count - (1 if right_top_item else 0))

        left_items = [self.anchor_provider]
        left_items.extend(remaining[:left_slots])

        right_items: list[str] = []
        if right_top_item:
            right_items.append(right_top_item)
        if right_slots:
            right_items.extend(remaining[left_slots:left_slots + right_slots])

        cleanup_kwargs = {}

        atexit.register(lambda: self.cleanup(**cleanup_kwargs))
        signal.signal(signal.SIGINT, lambda s, f: (self.cleanup(**cleanup_kwargs), sys.exit(0)))
        signal.signal(signal.SIGTERM, lambda s, f: (self.cleanup(**cleanup_kwargs), sys.exit(0)))

        # Best-effort cleanup of leftover `tmpclaude-*-cwd` artifacts (can be disabled via env).
        try:
            removed = _cleanup_tmpclaude_artifacts()
            if removed and os.environ.get("CCB_DEBUG") in ("1", "true", "yes"):
                print(f"🧹 Cleaned tmpclaude artifacts: {removed}")
        except Exception:
            pass

        # Best-effort cleanup of stale runtime dirs in temp (crash leftovers).
        try:
            removed_runtime = _cleanup_stale_runtime_dirs(exclude=self.runtime_dir)
            if removed_runtime and os.environ.get("CCB_DEBUG") in ("1", "true", "yes"):
                print(f"🧹 Cleaned stale runtime dirs: {removed_runtime}")
        except Exception:
            pass

        # Best-effort daemon log slimming (cache/legacy dirs).
        try:
            shrunk = _shrink_ccb_logs()
            if shrunk and os.environ.get("CCB_DEBUG") in ("1", "true", "yes"):
                print(f"🧹 Shrunk log files: {shrunk}")
        except Exception:
            pass

        # tmux-only: enable CCB UI theming for the current session while CCB is running.
        try:
            self._set_tmux_ui_active(True)
        except Exception:
            pass

        # tmux-only: label current pane for the anchor provider so titles are consistent.
        try:
            self._set_current_pane_label(self.anchor_provider)
        except Exception:
            pass

        # Mark current Claude pane/session as active when Claude is the anchor.
        if self.anchor_provider == "claude":
            try:
                self._write_local_claude_session(
                    session_id=self._read_local_claude_session_id(),
                    active=True,
                    pane_id=str(self.anchor_pane_id or ""),
                    pane_title_marker="CCB-Claude",
                    terminal=self.terminal_type,
                )
            except Exception:
                pass

        def _start_item(item: str, *, parent: str | None, direction: str | None) -> str | None:
            if item == "cmd":
                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)
            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

        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

        # Optional: start caskd after Codex session file exists (first startup convenience).
        if "codex" in self.providers and self.anchor_provider != "codex":
            self._maybe_start_caskd()

        # Start watchdog thread to monitor daemon health (provider-agnostic)
        self._start_daemon_watchdog()

        try:
            try:
                self._sync_cend_registry()
            except Exception:
                pass
            return self._start_provider_in_current_pane(self.anchor_provider)
        finally:
            self.cleanup(**cleanup_kwargs)


def cmd_start(args):
    # Enforce terminal environment requirement
    terminal = detect_terminal()
    if terminal is None:
        print("[ERROR] CCB must run inside tmux or WezTerm.", file=sys.stderr)
        print("", file=sys.stderr)
        print("Please start tmux first:", file=sys.stderr)
        print("  tmux", file=sys.stderr)
        print("", file=sys.stderr)
        print("Or use WezTerm terminal emulator.", file=sys.stderr)
        return 1

    def _env_truthy(name: str) -> bool:
        raw = os.environ.get(name)
        if raw is None:
            return False
        v = raw.strip().lower()
        if v in {"1", "true", "yes", "y", "on"}:
            return True
        if v in {"0", "false", "no", "n", "off"}:
            return False
        return False

    def _is_dangerous_root(cwd: Path) -> tuple[bool, str]:
        try:
            resolved = cwd.resolve()
        except Exception:
            resolved = cwd.absolute()

        try:
            home = Path.home().resolve()
        except Exception:
            try:
                home = Path.home().absolute()
            except Exception:
                home = None

        if home and resolved == home:
            return True, "$HOME"

        # POSIX root: "/"; Windows drive root: "C:\\"
        try:
            anchor = Path(resolved.anchor) if resolved.anchor else None
        except Exception:
            anchor = None
        if anchor and resolved == anchor:
            return True, "filesystem root"

        return False, ""

    def _find_parent_anchor_dir(cwd: Path) -> Path | None:
        """
        Find the nearest ancestor `.ccb/` (or legacy `.ccb_config/`) project anchor directory.

        Used only to block auto-creation when a parent anchor exists. Users can still
        create a local `.ccb/` manually to treat this directory as a new project.
        """
        try:
            start = cwd.resolve()
        except Exception:
            start = cwd.absolute()

        for parent in start.parents:
            candidate: Path | None = None
            primary = parent / ".ccb"
            if primary.is_dir():
                candidate = primary
            else:
                legacy = parent / ".ccb_config"
                if legacy.is_dir():
                    candidate = legacy
            if not candidate:
                continue
            # Ignore anchors at $HOME or filesystem root (too broad / likely accidental).
            is_dangerous, _reason = _is_dangerous_root(parent)
            if is_dangerous:
                continue
            return candidate
        return None

    work_dir = Path.cwd()
    primary_cfg = project_config_dir(work_dir)
    legacy_cfg = legacy_project_config_dir(work_dir)

    if primary_cfg.exists() and not primary_cfg.is_dir():
        print("❌ Invalid .ccb: exists but is not a directory", file=sys.stderr)
        print(f"   path: {primary_cfg}", file=sys.stderr)
        print("💡 Fix: remove it or rename it, then retry.", file=sys.stderr)
        return 2

    if legacy_cfg.exists() and not legacy_cfg.is_dir() and not primary_cfg.exists():
        print("❌ Invalid .ccb_config: exists but is not a directory", file=sys.stderr)
        print(f"   path: {legacy_cfg}", file=sys.stderr)
        print("💡 Fix: remove it or rename it, then retry.", file=sys.stderr)
        return 2

    if not primary_cfg.exists() and legacy_cfg.is_dir():
        try:
            legacy_cfg.replace(primary_cfg)
            print(f"✅ Migrated legacy config dir: {legacy_cfg} -> {primary_cfg}", file=sys.stderr)
        except Exception as exc:
            print(f"⚠️  Failed to migrate {legacy_cfg} -> {primary_cfg}: {exc}", file=sys.stderr)
            print("   Continuing with legacy .ccb_config directory.", file=sys.stderr)

    cfg = resolve_project_config_dir(work_dir)
    if not cfg.is_dir():
        is_dangerous, danger_reason = _is_dangerous_root(Path.cwd())
        parent_anchor = _find_parent_anchor_dir(Path.cwd())
        if parent_anchor:
            project_root = parent_anchor.parent
            print(
                "❌ Project config directory not found in current directory, but an existing project anchor was found in a parent directory.",
                file=sys.stderr,
            )
            print(f"   cwd:         {Path.cwd()}", file=sys.stderr)
            print(f"   project_root:{project_root}", file=sys.stderr)
            print("💡 Auto-create blocked to avoid accidental nesting.", file=sys.stderr)
            print("💡 If you want this directory to be a separate project, create .ccb here:", file=sys.stderr)
            print("   mkdir .ccb", file=sys.stderr)
            return 2

        if is_dangerous and not _env_truthy("CCB_INIT_PROJECT_DANGEROUS"):
            print(f"❌ Refusing to auto-create .ccb in {danger_reason}.", file=sys.stderr)
            print("💡 If you really intend to do this, set CCB_INIT_PROJECT_DANGEROUS=1 and retry.", file=sys.stderr)
            return 2
        try:
            cfg.mkdir(parents=False, exist_ok=True)
            print(f"✅ Created: {cfg}", file=sys.stderr)
        except Exception as exc:
            print(f"❌ Failed to create {cfg}: {exc}", file=sys.stderr)
            return 2

    # 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 [])

    def _existing_provider_pane_for_project(project_work_dir: Path, provider: str) -> tuple[dict | None, str]:
        prov = (provider or "").strip().lower()
        if not prov:
            return None, ""
        try:
            project_id = compute_ccb_project_id(project_work_dir)
        except Exception:
            return None, ""
        if not project_id:
            return None, ""
        record = load_registry_by_project_id(project_id, prov)
        if not isinstance(record, dict):
            return None, ""
        providers_map = record.get("providers")
        if not isinstance(providers_map, dict):
            return record, ""
        entry = providers_map.get(prov)
        if not isinstance(entry, dict):
            return record, ""
        pane_id = str(entry.get("pane_id") or "").strip()
        return record, pane_id

    # Enforce single ccb instance per directory.
    lock_cwd = str(Path.cwd().resolve())
    ccb_lock = ProviderLock("ccb", timeout=0.1, cwd=lock_cwd)
    if not ccb_lock.try_acquire():
        pid = ""
        try:
            pid = ccb_lock.lock_file.read_text(encoding="utf-8").strip()
        except Exception:
            pid = ""
        pid_msg = f" (pid {pid})" if pid else ""

        # Fast-path: if user requests a single provider, try activating its existing pane
        # in the already-running instance instead of hard failing.
        focus_provider = requested_providers[0] if len(requested_providers) == 1 else ""
        if focus_provider:
            record, pane_id = _existing_provider_pane_for_project(work_dir, focus_provider)
            if record and pane_id:
                try:
                    backend = get_backend_for_session(record)
                    if backend and backend.is_alive(pane_id):
                        if os.environ.get("TMUX") or sys.stdin.isatty():
                            backend.activate(pane_id)
                        print(
                            f"ℹ️ Reusing existing ccb instance for this directory{pid_msg}; "
                            f"activated {focus_provider} pane {pane_id}.",
                            file=sys.stderr,
                        )
                        return 0
                except Exception:
                    pass

        print(f"❌ Another ccb instance is already running for this directory{pid_msg}.", file=sys.stderr)
        print("💡 Only one ccb instance is allowed per directory.", file=sys.stderr)
        if focus_provider:
            print(f"💡 Try switching to the existing {focus_provider} pane in that ccb session.", file=sys.stderr)
        return 2
    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 {}

    if not providers:
        raw_providers = config_data.get("providers")
        if isinstance(raw_providers, str):
            raw_providers = [raw_providers]
        if not isinstance(raw_providers, list):
            raw_providers = []
        if not raw_providers:
            raw_providers = list(DEFAULT_PROVIDERS)
            if not config.path:
                created_path, created = ensure_default_start_config(work_dir)
                if created and created_path:
                    print(f"✅ Created default config: {created_path}", file=sys.stderr)
        providers = _parse_providers([str(p) for p in raw_providers])

    if not providers:
        print("❌ No providers configured. Define providers in ccb.config or pass them on the command line.", file=sys.stderr)
        return 2

    flags = config_data.get("flags") if isinstance(config_data.get("flags"), dict) else {}
    resume = bool(args.resume or flags.get("resume") or flags.get("restore") or flags.get("auto_resume"))
    auto = bool(args.auto or flags.get("auto") or flags.get("auto_mode"))
    cmd_config = config_data.get("cmd") if isinstance(config_data, dict) else None
    if cmd_enabled:
        if isinstance(cmd_config, dict):
            cmd_config = dict(cmd_config)
            cmd_config["enabled"] = True
        elif not cmd_config:
            cmd_config = True

    launcher = AILauncher(
        providers=providers,
        resume=resume,
        auto=auto,
        cmd_config=cmd_config,
    )
    return launcher.run_up()


def _find_all_zombie_sessions() -> list[dict]:
    """Find all zombie tmux sessions (CCB sessions whose parent process is dead)."""
    import re
    pattern = re.compile(r"^(codex|gemini|opencode|claude|droid)-(\d+)-")
    zombies = []

    if os.name == "nt" or not shutil.which("tmux"):
        return []

    try:
        result = subprocess.run(
            ["tmux", "list-sessions", "-F", "#{session_name}"],
            capture_output=True, text=True, timeout=5
        )
        if result.returncode != 0:
            return []
    except Exception:
        return []

    for session in result.stdout.strip().split("\n"):
        if not session:
            continue
        match = pattern.match(session)
        if not match:
            continue

        provider, parent_pid_str = match.groups()
        try:
            parent_pid = int(parent_pid_str)
        except ValueError:
            continue

        # Check if parent PID is alive
        if _is_pid_alive(parent_pid):
            continue  # Not a zombie

        zombies.append({
            "session": session,
            "provider": provider,
            "parent_pid": parent_pid
        })

    return zombies


def _kill_global_zombies(yes: bool = False) -> int:
    """Clean up all zombie tmux sessions globally."""
    zombies = _find_all_zombie_sessions()

    if not zombies:
        print("✅ No zombie sessions found")
        return 0

    # Show list
    print(f"Found {len(zombies)} zombie session(s):")
    for z in zombies:
        print(f"  - {z['session']} (parent PID {z['parent_pid']} exited)")

    # Confirm
    if not yes:
        try:
            reply = input("\nClean up these sessions? [y/N] ")
            if reply.lower() != 'y':
                print("❌ Cancelled")
                return 1
        except (EOFError, KeyboardInterrupt):
            print("\n❌ Cancelled")
            return 1

    # Kill sessions
    killed = 0
    failed = 0
    for z in zombies:
        try:
            result = subprocess.run(
                ["tmux", "kill-session", "-t", z["session"]],
                capture_output=True, timeout=5
            )
            if result.returncode == 0:
                killed += 1
            else:
                failed += 1
        except Exception:
            failed += 1

    if failed > 0:
        print(f"✅ Cleaned up {killed} zombie session(s), {failed} failed")
    else:
        print(f"✅ Cleaned up {killed} zombie session(s)")
    return 0


def _find_daemon_pids_by_name(daemon_name: str) -> list[int]:
    """Find all PIDs of processes matching the daemon name."""
    pids = []
    try:
        if os.name == "nt":
            result = subprocess.run(
                ["tasklist", "/FI", f"IMAGENAME eq python*", "/FO", "CSV", "/NH"],
                capture_output=True, text=True, encoding="utf-8", errors="replace"
            )
            # Windows: need to check command line, tasklist doesn't show it
            # Fall back to checking state files
            pass
        else:
            result = subprocess.run(
                ["pgrep", "-f", f"bin/{daemon_name}$"],
                capture_output=True, text=True, encoding="utf-8", errors="replace"
            )
            if result.returncode == 0:
                for line in result.stdout.strip().split("\n"):
                    line = line.strip()
                    if line:
                        try:
                            pids.append(int(line))
                        except ValueError:
                            pass
    except Exception:
        pass
    return pids


def _kill_pid(pid: int, force: bool = False) -> bool:
    """Kill a process by PID. Returns True if successful."""
    if pid <= 0:
        return False
    try:
        if os.name == "nt":
            if force:
                subprocess.run(["taskkill", "/F", "/PID", str(pid)], capture_output=True)
            else:
                subprocess.run(["taskkill", "/PID", str(pid)], capture_output=True)
        else:
            sig = signal.SIGKILL if force else signal.SIGTERM
            os.kill(pid, sig)
        return True
    except Exception:
        return False


def cmd_kill(args):
    force = getattr(args, "force", False)

    # Global zombie cleanup mode
    if force:
        yes = getattr(args, "yes", False)
        return _kill_global_zombies(yes=yes)

    # Project-level cleanup (original behavior)
    providers = _parse_providers(args.providers or ["codex", "gemini", "opencode", "claude", "droid"], allow_unknown=True)
    if not providers:
        return 2

    # Map provider names to their client specs to find daemon info
    specs = {
        "codex": CASK_CLIENT_SPEC,
        "gemini": GASK_CLIENT_SPEC,
        "opencode": OASK_CLIENT_SPEC,
        "claude": LASK_CLIENT_SPEC,
        "droid": DASK_CLIENT_SPEC,
    }

    for provider in providers:
        # 1. Kill UI sessions (tmux/wezterm)
        session_file = find_project_session_file(Path.cwd(), f".{provider}-session")
        if session_file and session_file.exists():
            try:
                data = json.loads(session_file.read_text(encoding="utf-8-sig"))
                terminal = data.get("terminal", "tmux")
                pane_id = data.get("pane_id") if terminal == "wezterm" else (data.get("pane_id") or data.get("tmux_session") or "")

                if terminal == "wezterm" and pane_id:
                    backend = WeztermBackend()
                    backend.kill_pane(pane_id)
                elif pane_id and shutil.which("tmux"):
                    backend = TmuxBackend()
                    if str(pane_id).startswith("%"):
                        backend.kill_pane(str(pane_id))
                    else:
                        tmux_session = str(data.get("tmux_session") or "").strip()
                        if tmux_session and not tmux_session.startswith("%"):
                            subprocess.run(["tmux", "kill-session", "-t", tmux_session], stderr=subprocess.DEVNULL)
                            subprocess.run(["tmux", "kill-session", "-t", f"launcher-{tmux_session}"], stderr=subprocess.DEVNULL)
                        else:
                            backend.kill_pane(str(pane_id))

                data["active"] = False
                data["ended_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
                safe_write_session(session_file, json.dumps(data, ensure_ascii=False, indent=2))
                print(f"✅ {provider.capitalize()} session terminated")
            except Exception as e:
                print(f"❌ {provider}: {e}")
        else:
             print(f"ℹ️  {provider}: No active session file found")

        # 2. Kill background daemon (graceful shutdown, then force kill)
        spec = specs.get(provider)
        if spec:
            daemon_name = spec.daemon_bin_name
            st_file = state_file_path(f"{daemon_name}.json")

            try:
                if shutdown_daemon(spec.protocol_prefix, 1.0, st_file):
                    print(f"✅ {daemon_name} daemon shutdown requested")
                else:
                    # Graceful shutdown failed, try force kill using PID from state file
                    st = read_state(st_file)
                    if st and st.get("pid"):
                        pid = int(st["pid"])
                        if _kill_pid(pid, force=True):
                            print(f"✅ {daemon_name} daemon force killed (pid={pid})")
                        else:
                            print(f"⚠️ {daemon_name} daemon could not be killed (pid={pid})")
            except Exception:
                pass

    return 0


def _split_provider_tokens(values: list[str]) -> list[str]:
    raw_parts: list[str] = []
    for item in (values or []):
        if item is None:
            continue
        for part in str(item).split(","):
            p = part.strip().lower()
            if p:
                raw_parts.append(p)
    return raw_parts


def _parse_providers(values: list[str], *, allow_unknown: bool = False) -> list[str]:
    """
    Parse providers from argv.

    Accept both:
      - space-separated: `ccb codex gemini opencode claude droid`
      - comma-separated: `ccb codex,gemini,opencode,claude,droid`

    Returns a de-duplicated list preserving order.
    """
    allowed = {"codex", "gemini", "opencode", "claude", "droid"}
    raw_parts = _split_provider_tokens(values)

    if not raw_parts:
        return []

    seen: set[str] = set()
    parsed: list[str] = []
    unknown: list[str] = []
    for p in raw_parts:
        if p in seen:
            continue
        seen.add(p)
        if p in allowed or allow_unknown:
            parsed.append(p)
        else:
            unknown.append(p)

    if unknown and not allow_unknown:
        print(f"❌ invalid provider(s): {', '.join(unknown)}", file=sys.stderr)
        print("💡 use: ccb codex gemini opencode claude droid  (spaces)  or  ccb codex,gemini,opencode,claude,droid  (commas)", file=sys.stderr)
        print("💡 allowed: codex, gemini, opencode, claude, droid", file=sys.stderr)
        return []

    return parsed


def _parse_providers_with_cmd(values: list[str]) -> tuple[list[str], bool]:
    """
    Parse providers from argv and treat "cmd" as a separate flag.
    Returns (providers, cmd_enabled).
    """
    allowed = {"codex", "gemini", "opencode", "claude", "droid"}
    raw_parts = _split_provider_tokens(values)
    if not raw_parts:
        return [], False

    seen: set[str] = set()
    parsed: list[str] = []
    unknown: list[str] = []
    cmd_enabled = False

    for p in raw_parts:
        if p == "cmd":
            cmd_enabled = True
            continue
        if p in seen:
            continue
        seen.add(p)
        if p in allowed:
            parsed.append(p)
        else:
            unknown.append(p)

    if unknown:
        print(f"❌ invalid provider(s): {', '.join(unknown)}", file=sys.stderr)
        print("💡 use: ccb codex gemini opencode claude droid cmd  (spaces)  or  ccb codex,gemini,opencode,claude,droid,cmd  (commas)", file=sys.stderr)
        print("💡 allowed: codex, gemini, opencode, claude, droid, cmd", file=sys.stderr)
        return [], cmd_enabled

    return parsed, cmd_enabled


def _get_version_info(dir_path: Path) -> dict:
    """Get commit hash, date and version from install directory"""
    info = {"commit": None, "date": None, "version": None}
    ccb_file = dir_path / "ccb"
    if ccb_file.exists():
        try:
            content = ccb_file.read_text(encoding='utf-8', errors='replace')
            for line in content.split('\n')[:60]:
                line = line.strip()
                if line.startswith('VERSION') and '=' in line:
                    info["version"] = line.split('=')[1].strip().strip('"').strip("'")
                elif line.startswith('GIT_COMMIT') and '=' in line:
                    val = line.split('=')[1].strip().strip('"').strip("'")
                    if val:
                        info["commit"] = val
                elif line.startswith('GIT_DATE') and '=' in line:
                    val = line.split('=')[1].strip().strip('"').strip("'")
                    if val:
                        info["date"] = val
        except Exception:
            pass
    if shutil.which("git") and (dir_path / ".git").exists():
        result = subprocess.run(
            ["git", "-C", str(dir_path), "log", "-1", "--format=%h|%ci"],
            capture_output=True, text=True, encoding='utf-8', errors='replace'
        )
        if result.returncode == 0 and result.stdout.strip():
            parts = result.stdout.strip().split("|")
            if len(parts) >= 2:
                info["commit"] = parts[0]
                info["date"] = parts[1].split()[0]
    return info


def _format_version_info(info: dict) -> str:
    """Format version info for display"""
    parts = []
    if info.get("version"):
        parts.append(f"v{info['version']}")
    if info.get("commit"):
        parts.append(info["commit"])
    if info.get("date"):
        parts.append(info["date"])
    return " ".join(parts) if parts else "unknown"


def _get_remote_version_info() -> dict | None:
    """Get latest version info from GitHub API"""
    import urllib.request
    import ssl

    api_url = "https://api.github.com/repos/bfly123/claude_code_bridge/commits/main"
    try:
        ctx = ssl.create_default_context()
        req = urllib.request.Request(api_url, headers={"User-Agent": "ccb"})
        with urllib.request.urlopen(req, context=ctx, timeout=5) as resp:
            data = json.loads(resp.read().decode('utf-8'))
            commit = data.get("sha", "")[:7]
            date_str = data.get("commit", {}).get("committer", {}).get("date", "")
            date = date_str[:10] if date_str else None
            return {"commit": commit, "date": date}
    except Exception:
        pass

    if shutil.which("curl"):
        result = subprocess.run(
            ["curl", "-fsSL", api_url],
            capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=10
        )
        if result.returncode == 0:
            try:
                data = json.loads(result.stdout)
                commit = data.get("sha", "")[:7]
                date_str = data.get("commit", {}).get("committer", {}).get("date", "")
                date = date_str[:10] if date_str else None
                return {"commit": commit, "date": date}
            except Exception:
                pass
    return None


def _get_available_versions() -> list[str]:
    """Get available version tags from GitHub API or git ls-remote"""
    import urllib.request
    import ssl

    repo_url = "https://github.com/bfly123/claude_code_bridge"
    api_url = "https://api.github.com/repos/bfly123/claude_code_bridge/tags"
    versions = []

    def parse_api_response(data):
        result = []
        for tag in data:
            name = tag.get("name", "")
            if name.startswith("v"):
                name = name[1:]
            if re.match(r"^\d+(\.\d+)*$", name):
                result.append(name)
        return result

    def parse_git_refs(output):
        result = []
        for line in output.strip().split('\n'):
            if not line:
                continue
            parts = line.split('\t')
            if len(parts) >= 2:
                ref = parts[1]
                if ref.startswith('refs/tags/v'):
                    name = ref.replace('refs/tags/v', '').rstrip('^{}')
                    if re.match(r"^\d+(\.\d+)*$", name):
                        result.append(name)
        return list(set(result))

    try:
        ctx = ssl.create_default_context()
        req = urllib.request.Request(api_url, headers={"User-Agent": "ccb"})
        with urllib.request.urlopen(req, context=ctx, timeout=10) as resp:
            data = json.loads(resp.read().decode('utf-8'))
            versions = parse_api_response(data)
    except Exception:
        pass

    if not versions and shutil.which("curl"):
        result = subprocess.run(
            ["curl", "-fsSL", api_url],
            capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=15
        )
        if result.returncode == 0:
            try:
                data = json.loads(result.stdout)
                versions = parse_api_response(data)
            except Exception:
                pass

    if not versions and shutil.which("git"):
        result = subprocess.run(
            ["git", "ls-remote", "--tags", repo_url],
            capture_output=True, text=True, encoding='utf-8', errors='replace', timeout=30
        )
        if result.returncode == 0:
            versions = parse_git_refs(result.stdout)

    return versions


def _find_matching_version(target: str, versions: list[str]) -> str | None:
    """Find the highest matching version for target prefix"""
    target_parts = target.split(".")

    def version_key(v: str):
        parts = v.split(".")
        return tuple(int(p) for p in parts if p.isdigit())

    matching = []
    for v in versions:
        v_parts = v.split(".")
        if len(v_parts) >= len(target_parts):
            if v_parts[:len(target_parts)] == target_parts:
                matching.append(v)

    if not matching:
        return None
    matching.sort(key=version_key, reverse=True)
    return matching[0]



def cmd_version(args):
    """Show version info and check for updates"""
    script_root = Path(__file__).resolve().parent
    # 候选目录列表（优先级递减）
    candidates = [
        script_root,  # 当前脚本目录
        Path(os.environ.get("CODEX_INSTALL_PREFIX", "")).expanduser() if os.environ.get("CODEX_INSTALL_PREFIX") else None,
        Path.home() / ".local/share/codex-dual",  # Linux/macOS
    ]

    # Windows 特定路径
    if platform.system() == "Windows":
        localappdata = os.environ.get("LOCALAPPDATA", "")
        if localappdata:
            candidates.extend([
                Path(localappdata) / "codex-dual",
                Path(localappdata) / "claude-code-bridge",
            ])
        candidates.append(Path.home() / "AppData/Local/codex-dual")

    # 选择第一个包含 ccb 文件的目录
    install_dir = None
    for candidate in candidates:
        if candidate and (candidate / "ccb").exists():
            install_dir = candidate
            break
    if not install_dir:
        install_dir = script_root

    local_info = _get_version_info(install_dir)
    local_str = _format_version_info(local_info)

    print(f"ccb (Claude Code Bridge) {local_str}")
    print(f"Install path: {install_dir}")

    print("\nChecking for updates...")
    remote_info = _get_remote_version_info()

    if remote_info is None:
        print("⚠️  Unable to check for updates (network error)")
    elif local_info.get("commit") and remote_info.get("commit"):
        if local_info["commit"] == remote_info["commit"]:
            print(f"✅ Up to date")
        else:
            remote_str = f"{remote_info['commit']} {remote_info.get('date', '')}"
            print(f"📦 Update available: {remote_str}")
            print(f"   Run: ccb update")
    else:
        print("⚠️  Unable to compare versions")

    return 0



def cmd_update(args):
    """Update ccb to latest or specified version"""
    import urllib.request
    import tarfile
    import tempfile

    # Prefer the directory where this script resides (installed copy), then fall back to env/default.
    script_root = Path(__file__).resolve().parent
    default_install_dir = Path.home() / ".local/share/codex-dual"
    install_dir = Path(os.environ.get("CODEX_INSTALL_PREFIX") or default_install_dir).expanduser()
    if (script_root / "install.sh").exists():
        install_dir = script_root
    repo_url = "https://github.com/bfly123/claude_code_bridge"

    # Parse version target
    target_version = None
    if hasattr(args, 'target') and args.target:
        target_spec = args.target.lstrip('v')
        if re.match(r"^\d+(\.\d+)*$", target_spec):
            print(f"🔍 Looking for version matching: {target_spec}")
            versions = _get_available_versions()
            if not versions:
                print("❌ Could not fetch available versions")
                return 1
            target_version = _find_matching_version(target_spec, versions)
            if not target_version:
                print(f"❌ No version found matching '{target_spec}'")
                print(f"   Available: {', '.join(sorted(versions, key=lambda v: [int(x) for x in v.split('.')], reverse=True)[:10])}")
                return 1
            print(f"📌 Target version: v{target_version}")
        else:
            print(f"❌ Invalid version format: {args.target}")
            print("   Examples: ccb update 4, ccb update 4.1, ccb update 4.1.3")
            return 1

    # Get current version info before update
    old_info = _get_version_info(install_dir)

    if target_version:
        print(f"🔄 Updating to v{target_version}...")
    else:
        print("🔄 Checking for updates...")

    # Method 1: Prefer git if available
    if shutil.which("git") and (install_dir / ".git").exists():
        if target_version:
            print(f"📦 Switching to v{target_version} via git...")
            subprocess.run(
                ["git", "-C", str(install_dir), "fetch", "--tags", "--force"],
                capture_output=True, text=True, encoding='utf-8', errors='replace'
            )
            result = subprocess.run(
                ["git", "-C", str(install_dir), "checkout", f"v{target_version}"],
                capture_output=True, text=True, encoding='utf-8', errors='replace'
            )
        else:
            print("📦 Updating via git pull...")
            result = subprocess.run(
                ["git", "-C", str(install_dir), "pull", "--ff-only"],
                capture_output=True, text=True, encoding='utf-8', errors='replace'
            )
        if result.returncode == 0:
            print(result.stdout.strip() if result.stdout.strip() else "Already up to date.")
            print("🔧 Reinstalling...")
            env = os.environ.copy()
            env["CCB_CLEAN_INSTALL"] = "1"
            # Windows: use install.ps1, Unix: use install.sh
            if platform.system() == "Windows":
                subprocess.run(
                    ["powershell", "-ExecutionPolicy", "Bypass", "-File",
                     str(install_dir / "install.ps1"), "install"],
                    env=env
                )
            else:
                subprocess.run([str(install_dir / "install.sh"), "install"], env=env)
            # Show upgrade info
            new_info = _get_version_info(install_dir)
            old_str = _format_version_info(old_info)
            new_str = _format_version_info(new_info)
            if old_info.get("commit") != new_info.get("commit"):
                print(f"✅ Updated: {old_str} → {new_str}")
            else:
                print(f"✅ Already up to date: {new_str}")
            return 0
        else:
            err_msg = "Git checkout failed" if target_version else "Git pull failed"
            print(f"⚠️ {err_msg}: {result.stderr.strip()}")
            print("Falling back to tarball download...")

    def _pick_temp_base_dir() -> Path:
        candidates: list[Path] = []
        for key in ("CCB_TMPDIR", "TMPDIR", "TEMP", "TMP"):
            value = (os.environ.get(key) or "").strip()
            if value:
                candidates.append(Path(value).expanduser())
        try:
            candidates.append(Path(tempfile.gettempdir()))
        except Exception:
            pass
        candidates.extend(
            [
                Path("/tmp"),
                Path("/var/tmp"),
                Path("/usr/tmp"),
                Path.home() / ".cache" / "ccb" / "tmp",
                install_dir / ".tmp",
                Path.cwd() / ".tmp",
            ]
        )

        for base in candidates:
            try:
                base.mkdir(parents=True, exist_ok=True)
                probe = base / f".ccb_tmp_probe_{os.getpid()}_{int(time.time() * 1000)}"
                probe.write_bytes(b"1")
                probe.unlink(missing_ok=True)
                return base
            except Exception:
                continue

        raise RuntimeError(
            "❌ No usable temporary directory found.\n"
            "Fix options:\n"
            "  - Create /tmp (Linux/WSL): sudo mkdir -p /tmp && sudo chmod 1777 /tmp\n"
            "  - Or set TMPDIR/CCB_TMPDIR to a writable path (e.g. export TMPDIR=$HOME/.cache/tmp)"
        )

    # Method 2: Download tarball
    if target_version:
        tarball_url = f"{repo_url}/archive/refs/tags/v{target_version}.tar.gz"
        extracted_name = f"claude_code_bridge-{target_version}"
    else:
        tarball_url = f"{repo_url}/archive/refs/heads/main.tar.gz"
        extracted_name = "claude_code_bridge-main"
    try:
        tmp_base = _pick_temp_base_dir()
    except Exception as exc:
        print(str(exc))
        return 1
    tmp_dir = tmp_base / "ccb_update"

    try:
        if target_version:
            print(f"📥 Downloading v{target_version}...")
        else:
            print(f"📥 Downloading latest version...")
        if tmp_dir.exists():
            shutil.rmtree(tmp_dir)
        tmp_dir.mkdir(parents=True, exist_ok=True)
        tarball_path = tmp_dir / "main.tar.gz"

        # Prefer curl/wget (better certificate handling)
        downloaded = False
        if shutil.which("curl"):
            result = subprocess.run(
                ["curl", "-fsSL", "-o", str(tarball_path), tarball_url],
                capture_output=True
            )
            downloaded = result.returncode == 0
        if not downloaded and shutil.which("wget"):
            result = subprocess.run(
                ["wget", "-q", "-O", str(tarball_path), tarball_url],
                capture_output=True
            )
            downloaded = result.returncode == 0
        if not downloaded:
            # Fallback to urllib (may have SSL issues)
            import ssl
            try:
                urllib.request.urlretrieve(tarball_url, tarball_path)
            except ssl.SSLError:
                print("⚠️ SSL certificate verification failed, trying to skip...")
                ctx = ssl.create_default_context()
                ctx.check_hostname = False
                ctx.verify_mode = ssl.CERT_NONE
                with urllib.request.urlopen(tarball_url, context=ctx) as resp:
                    tarball_path.write_bytes(resp.read())

        print("📂 Extracting...")
        def _safe_extract(tar: tarfile.TarFile, dest: Path) -> None:
            dest = dest.resolve()
            for member in tar.getmembers():
                member_path = (dest / member.name).resolve()
                if not str(member_path).startswith(str(dest) + os.sep):
                    raise RuntimeError(f"Unsafe tar member path: {member.name}")
            # Python 3.14+ requires filter argument
            try:
                tar.extractall(dest, filter='data')
            except TypeError:
                tar.extractall(dest)

        with tarfile.open(tarball_path, "r:gz") as tar:
            _safe_extract(tar, tmp_dir)

        extracted_dir = tmp_dir / extracted_name

        print("🔧 Installing...")
        env = os.environ.copy()
        env["CODEX_INSTALL_PREFIX"] = str(install_dir)
        env["CCB_CLEAN_INSTALL"] = "1"
        # Windows: use install.ps1, Unix: use install.sh
        if platform.system() == "Windows":
            ps1_script = extracted_dir / "install.ps1"
            subprocess.run(
                ["powershell", "-ExecutionPolicy", "Bypass", "-File", str(ps1_script), "install"],
                check=True, env=env
            )
        else:
            subprocess.run([str(extracted_dir / "install.sh"), "install"], check=True, env=env)

        # Show upgrade info
        new_info = _get_version_info(install_dir)
        old_str = _format_version_info(old_info)
        new_str = _format_version_info(new_info)
        if old_info.get("commit") != new_info.get("commit") or old_info.get("version") != new_info.get("version"):
            print(f"✅ Updated: {old_str} → {new_str}")
        else:
            print(f"✅ Already up to date: {new_str}")
        return 0

    except Exception as e:
        print(f"❌ Update failed: {e}")
        return 1

    finally:
        if tmp_dir.exists():
            shutil.rmtree(tmp_dir, ignore_errors=True)


_CLAUDE_COMMAND_DOCS = (
    "cpend.md",
    "cping.md",
    "gpend.md",
    "gping.md",
    "opend.md",
    "oping.md",
)


def _cleanup_claude_files() -> None:
    claude_dir = Path.home() / ".claude"
    claude_md = claude_dir / "CLAUDE.md"
    settings_json = claude_dir / "settings.json"
    command_dirs = [
        claude_dir / "commands",
        Path.home() / ".config" / "claude" / "commands",
        Path.home() / ".local" / "share" / "claude" / "commands",
    ]

    # Remove command docs.
    for cmd_dir in command_dirs:
        if not cmd_dir.is_dir():
            continue
        for name in _CLAUDE_COMMAND_DOCS:
            try:
                (cmd_dir / name).unlink(missing_ok=True)
            except Exception:
                pass

    # Remove CCB blocks and legacy collaboration rules.
    if claude_md.is_file():
        try:
            content = claude_md.read_text(encoding="utf-8", errors="replace")
        except Exception:
            content = ""
        if content:
            before = content
            content = re.sub(
                r"\n?<!-- CCB_CONFIG_START -->.*?<!-- CCB_CONFIG_END -->\n?",
                "\n",
                content,
                flags=re.DOTALL,
            )
            legacy_patterns = [
                r"## Codex Collaboration Rules.*?(?=\n## (?!Gemini)|\Z)",
                r"## Codex 协作规则.*?(?=\n## |\Z)",
                r"## Gemini Collaboration Rules.*?(?=\n## |\Z)",
                r"## Gemini 协作规则.*?(?=\n## |\Z)",
                r"## OpenCode Collaboration Rules.*?(?=\n## |\Z)",
                r"## OpenCode 协作规则.*?(?=\n## |\Z)",
            ]
            for pattern in legacy_patterns:
                content = re.sub(pattern, "", content, flags=re.DOTALL)
            content = content.strip()
            if content != before.strip():
                claude_md.write_text((content + "\n") if content else "", encoding="utf-8")

    # Remove CCB permissions from settings.json.
    if settings_json.is_file():
        try:
            data = json.loads(settings_json.read_text(encoding="utf-8", errors="replace"))
        except Exception:
            data = {}
        if isinstance(data, dict):
            perms = data.get("permissions")
            allow = perms.get("allow") if isinstance(perms, dict) else None
            if isinstance(allow, list):
                to_remove = {
                    "Bash(cask:*)",
                    "Bash(cpend)",
                    "Bash(cping)",
                    "Bash(gask:*)",
                    "Bash(gpend)",
                    "Bash(gping)",
                    "Bash(oask:*)",
                    "Bash(opend)",
                    "Bash(oping)",
                }
                new_allow = [entry for entry in allow if entry not in to_remove]
                if new_allow != allow:
                    perms["allow"] = new_allow
                    settings_json.write_text(
                        json.dumps(data, ensure_ascii=False, indent=2) + "\n",
                        encoding="utf-8",
                    )


def _find_install_dir() -> Path:
    script_root = Path(__file__).resolve().parent
    if (script_root / "install.sh").exists() or (script_root / "install.ps1").exists():
        return script_root

    candidates = []
    env_prefix = (os.environ.get("CODEX_INSTALL_PREFIX") or "").strip()
    if env_prefix:
        candidates.append(Path(env_prefix).expanduser())

    if platform.system() == "Windows":
        localappdata = os.environ.get("LOCALAPPDATA", "")
        if localappdata:
            candidates.append(Path(localappdata) / "codex-dual")
            candidates.append(Path(localappdata) / "claude-code-bridge")
        candidates.append(Path.home() / "AppData/Local/codex-dual")
    else:
        candidates.append(Path.home() / ".local/share/codex-dual")

    for candidate in candidates:
        if candidate and (candidate / "ccb").exists():
            return candidate
    return script_root


def _run_installer(action: str) -> int:
    install_dir = _find_install_dir()
    if platform.system() == "Windows":
        script = install_dir / "install.ps1"
        if not script.exists():
            print(f"❌ install.ps1 not found in {install_dir}", file=sys.stderr)
            return 1
        cmd = [
            "powershell",
            "-NoProfile",
            "-ExecutionPolicy",
            "Bypass",
            "-File",
            str(script),
            action,
            "-InstallPrefix",
            str(install_dir),
        ]
        return subprocess.run(cmd).returncode

    script = install_dir / "install.sh"
    if not script.exists():
        print(f"❌ install.sh not found in {install_dir}", file=sys.stderr)
        return 1
    env = os.environ.copy()
    env["CODEX_INSTALL_PREFIX"] = str(install_dir)
    return subprocess.run(["bash", str(script), action], env=env).returncode


def cmd_uninstall(_args) -> int:
    _cleanup_claude_files()
    return _run_installer("uninstall")


def cmd_reinstall(_args) -> int:
    _cleanup_claude_files()
    return _run_installer("install")


def _droid_server_path() -> Path:
    return script_dir / "mcp" / "ccb-delegation" / "server.py"


def cmd_droid_setup_delegation(args) -> int:
    server_path = _droid_server_path()
    if not server_path.exists():
        print(f"❌ MCP server not found: {server_path}", file=sys.stderr)
        return 2

    if args.force:
        subprocess.run(["droid", "mcp", "remove", "ccb-delegation"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

    cmd = [
        "droid",
        "mcp",
        "add",
        "ccb-delegation",
        "--type",
        "stdio",
        sys.executable,
        str(server_path),
    ]
    try:
        rc = subprocess.run(cmd).returncode
    except FileNotFoundError:
        print("❌ `droid` not found in PATH.", file=sys.stderr)
        return 2
    if rc != 0:
        print("❌ Failed to register MCP server. Ensure `droid` is installed and on PATH.", file=sys.stderr)
        return rc or 1
    print("✅ Registered MCP server: ccb-delegation")
    print("Next: run `ccb droid test-delegation` to verify tools are visible.")
    return 0


def _extract_tool_name(tool: dict) -> str:
    for key in ("id", "name", "toolName"):
        value = tool.get(key)
        if isinstance(value, str) and value:
            return value
    return ""


def cmd_droid_test_delegation(_args) -> int:
    cmd = ["droid", "exec", "--list-tools", "--output-format", "json"]
    try:
        res = subprocess.run(cmd, capture_output=True, text=True)
    except FileNotFoundError:
        print("❌ `droid` not found in PATH.", file=sys.stderr)
        return 2
    if res.returncode != 0:
        err = res.stderr.strip() or res.stdout.strip()
        print(f"❌ droid exec --list-tools failed: {err}", file=sys.stderr)
        return res.returncode or 1

    try:
        tools = json.loads(res.stdout)
    except Exception:
        print("❌ Failed to parse tool list JSON from droid.", file=sys.stderr)
        return 2

    if not isinstance(tools, list):
        print("❌ Tool list JSON is not an array.", file=sys.stderr)
        return 2

    names = {name for tool in tools if (name := _extract_tool_name(tool))}
    required = {
        "ccb_ask_codex",
        "ccb_ask_gemini",
        "ccb_ask_claude",
        "ccb_ask_opencode",
    }
    missing = sorted(required - names)
    if missing:
        print("❌ MCP delegation tools missing:", ", ".join(missing), file=sys.stderr)
        print("Hint: run `ccb droid setup-delegation`.", file=sys.stderr)
        return 2

    print("✅ MCP delegation tools detected.")
    return 0


def cmd_mail(argv: list[str]) -> int:
    """Handle ccb mail subcommands."""
    parser = argparse.ArgumentParser(prog="ccb mail", description="Mail service commands")
    subparsers = parser.add_subparsers(dest="subcommand", help="Mail subcommands")

    subparsers.add_parser("setup", help="Run mail configuration wizard")
    subparsers.add_parser("start", help="Start mail daemon")
    subparsers.add_parser("stop", help="Stop mail daemon")
    subparsers.add_parser("status", help="Show mail service status")
    subparsers.add_parser("test", help="Test mail connection")
    subparsers.add_parser("config", help="Show current configuration")

    args = parser.parse_args(argv)

    if args.subcommand == "setup":
        try:
            from mail_tui import run_wizard
            return 0 if run_wizard() else 1
        except ImportError as e:
            print(f"Error: {e}", file=sys.stderr)
            return 1

    if args.subcommand == "start":
        try:
            from maild_client import start_mail_service
            return 0 if start_mail_service() else 1
        except ImportError as e:
            print(f"Error: {e}", file=sys.stderr)
            return 1

    if args.subcommand == "stop":
        try:
            from maild_client import stop_mail_service
            return 0 if stop_mail_service() else 1
        except ImportError as e:
            print(f"Error: {e}", file=sys.stderr)
            return 1

    if args.subcommand == "status":
        try:
            from maild_client import get_mail_status
            status = get_mail_status()
            print("Mail Service Status")
            print("=" * 40)
            print(f"  Configured: {'Yes' if status['configured'] else 'No'}")
            print(f"  Enabled:    {'Yes' if status['enabled'] else 'No'}")
            print(f"  Email:      {status.get('email') or '-'}")
            print(f"  Provider:   {status.get('provider') or '-'}")
            daemon = status.get('daemon', {})
            if daemon.get('running'):
                print(f"  Daemon:     Running (PID: {daemon.get('pid')})")
            else:
                print("  Daemon:     Stopped")
            return 0
        except ImportError as e:
            print(f"Error: {e}", file=sys.stderr)
            return 1

    if args.subcommand == "test":
        try:
            from maild_client import test_mail_connection
            result = test_mail_connection()
            print("Mail Connection Test")
            print("=" * 40)
            imap = result.get('imap', {})
            smtp = result.get('smtp', {})
            print(f"  IMAP: {'✓' if imap.get('success') else '✗'} {imap.get('message', '')}")
            print(f"  SMTP: {'✓' if smtp.get('success') else '✗'} {smtp.get('message', '')}")
            return 0 if result.get('success') else 1
        except ImportError as e:
            print(f"Error: {e}", file=sys.stderr)
            return 1

    if args.subcommand == "config":
        try:
            from mail.config import load_config
            config = load_config()
            print("Mail Configuration")
            print("=" * 40)
            print(f"  Enabled:          {config.enabled}")
            print(f"  Provider:         {config.account.provider}")
            print(f"  Email:            {config.account.email or '-'}")
            print(f"  IMAP Host:        {config.account.imap.host}:{config.account.imap.port}")
            print(f"  SMTP Host:        {config.account.smtp.host}:{config.account.smtp.port}")
            print(f"  Routing Mode:     {config.routing.mode}")
            print(f"  Default Provider: {config.routing.default_provider}")
            print(f"  Poll Interval:    {config.polling.interval_seconds}s")
            # New fields
            allowed = config.routing.allowed_senders
            print(f"  Allowed Senders:  {', '.join(allowed) if allowed else '(all)'}")
            print(f"  Reply To:         {config.routing.reply_to_address or '(original sender)'}")
            return 0
        except ImportError as e:
            print(f"Error: {e}", file=sys.stderr)
            return 1

    parser.print_help()
    return 1


def cmd_droid_subcommand(argv: list[str]) -> int:
    parser = argparse.ArgumentParser(prog="ccb droid", description="Droid-specific commands")
    subparsers = parser.add_subparsers(dest="subcommand", help="Droid subcommands")

    setup_parser = subparsers.add_parser("setup-delegation", help="Register MCP delegation tools for Droid")
    setup_parser.add_argument("--force", action="store_true", help="Re-register MCP server")

    subparsers.add_parser("test-delegation", help="Verify MCP delegation tools are available")

    args = parser.parse_args(argv)
    if args.subcommand == "setup-delegation":
        return cmd_droid_setup_delegation(args)
    if args.subcommand == "test-delegation":
        return cmd_droid_test_delegation(args)
    parser.print_help()
    return 1


def main():
    argv = sys.argv[1:]
    if "--print-version" in argv:
        print(f"v{VERSION}")
        return 0

    if argv and argv[0] in {"-v", "--version"}:
        argv = ["version"]

    if argv and argv[0] == "up":
        print("❌ `ccb up` is no longer supported.", file=sys.stderr)
        print("💡 Use: ccb [providers...]  (or configure ccb.config)", file=sys.stderr)
        return 2

    if argv and argv[0] == "droid" and len(argv) > 1 and argv[1] in {"setup-delegation", "test-delegation"}:
        return cmd_droid_subcommand(argv[1:])

    if argv and argv[0] in {"kill", "update", "version", "uninstall", "reinstall"}:
        parser = argparse.ArgumentParser(description="Claude AI unified launcher", add_help=True)
        subparsers = parser.add_subparsers(dest="command", help="Subcommands")

        kill_parser = subparsers.add_parser("kill", help="Terminate session or clean up zombies")
        kill_parser.add_argument("providers", nargs="*", default=[], help="Backends to terminate (codex/gemini/opencode/claude/droid)")
        kill_parser.add_argument("-f", "--force", action="store_true", help="Clean up all zombie tmux sessions globally")
        kill_parser.add_argument("-y", "--yes", action="store_true", help="Skip confirmation prompt (with -f)")

        update_parser = subparsers.add_parser("update", help="Update to latest or specified version")
        update_parser.add_argument("target", nargs="?",
                                   help="version like '4', '4.1', '4.1.3'")

        subparsers.add_parser("version", help="Show version and check for updates")
        subparsers.add_parser("uninstall", help="Uninstall ccb and clean configs")
        subparsers.add_parser("reinstall", help="Reinstall ccb and refresh configs")

        args = parser.parse_args(argv)
        if args.command == "kill":
            return cmd_kill(args)
        if args.command == "update":
            return cmd_update(args)
        if args.command == "version":
            return cmd_version(args)
        if args.command == "uninstall":
            return cmd_uninstall(args)
        if args.command == "reinstall":
            return cmd_reinstall(args)
        parser.print_help()
        return 1

    # Handle 'ccb mail' subcommand
    if argv and argv[0] == "mail":
        return cmd_mail(argv[1:])

    start_parser = argparse.ArgumentParser(
        description="Claude AI unified launcher",
        add_help=True,
        epilog="Other commands: ccb update | ccb version | ccb kill | ccb uninstall | ccb reinstall | ccb droid setup-delegation | ccb mail setup",
    )
    start_parser.add_argument(
        "providers",
        nargs="*",
        help="Backends to start (space or comma separated): codex, gemini, opencode, claude, droid (add cmd for a shell pane)",
    )
    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")
    args = start_parser.parse_args(argv)
    return cmd_start(args)


if __name__ == "__main__":
    sys.exit(main())