sub2cli-inject

#!/usr/bin/env python3
"""sub2cli-inject — Codex 渠道切换器 (vendored from r266-tech/codex-provider-macos).

sub2cli 用 add-api 子命令配置 url + apikey 到本机 Codex CLI / Codex App.
单文件 Python, 无第三方依赖, MIT 许可.
"""

from __future__ import annotations

import argparse
import base64
import errno
import fcntl
import getpass
import json
import os
import re
import select
import shutil
import sqlite3
import subprocess
import sys
import termios
import time
import tty

try:
    import readline  # noqa: F401  启用 input() 的行编辑 (退格/CJK 宽度/Ctrl-U/方向键)
except ImportError:
    pass

from pathlib import Path
from urllib import error as urlerror
from urllib import parse as urlparse
from urllib import request as urlrequest

HOME = Path(os.environ.get("CODEX_PROVIDER_HOME", str(Path.home()))).expanduser()
CODEX_HOME = Path(os.environ.get("CODEX_HOME", str(HOME / ".codex"))).expanduser()
APP_SUPPORT = HOME / "Library" / "Application Support"
APP_PROFILE = APP_SUPPORT / "Codex"
AUTH_JSON = CODEX_HOME / "auth.json"
CONFIG_TOML = CODEX_HOME / "config.toml"
STATE_DB = CODEX_HOME / "state_5.sqlite"
SESSION_DIRS = [CODEX_HOME / "sessions", CODEX_HOME / "archived_sessions"]
SLOTS_FILE = CODEX_HOME / "provider-slots.json"
BACKUP_ROOT = CODEX_HOME / "provider-switch-backups"
LOCK_FILE = CODEX_HOME / ".sub2cli-inject.lock"
SNAPSHOT_FILENAME = "pre-switch-snapshot.json"
AUTO_ROLLBACK_MARKER = ".auto-rollback-done"

OAUTH_PROVIDER = "openai"
RELAY_PROVIDER = "OpenAI"
THREAD_PROVIDERS = (OAUTH_PROVIDER, RELAY_PROVIDER)
DEFAULT_MODEL = "gpt-5.5"
DEFAULT_API_BASE_URL = "https://api.openai.com/v1"
APP_VERSION = "0.1.0"
APP_URL = os.environ.get("CODEX_PROVIDER_URL", "https://github.com/r266-tech/codex-provider-macos")
APP_BANNER = r"""
  ____          _           ____                 _     _
 / ___|___   __| | _____  _|  _ \ _ __ _____   _(_) __| | ___ _ __
| |   / _ \ / _` |/ _ \ \/ / |_) | '__/ _ \ \ / / |/ _` |/ _ \ '__|
| |__| (_) | (_| |  __/>  <|  __/| | | (_) \ V /| | (_| |  __/ |
 \____\___/ \__,_|\___/_/\_\_|   |_|  \___/ \_/ |_|\__,_|\___|_|
""".strip("\n")


class ChineseHelpFormatter(argparse.HelpFormatter):
    def start_section(self, heading: str | None) -> None:
        translations = {
            "positional arguments": "命令",
            "optional arguments": "选项",
            "options": "选项",
        }
        super().start_section(translations.get(heading, heading))


def fail(message: str, code: int = 2) -> None:
    print(f"错误：{message}", file=sys.stderr)
    raise SystemExit(code)


# ==================== File lock ====================

class LockBusy(RuntimeError):
    """Another sub2cli-inject process is holding the lock."""


class Sub2cliLock:
    """Advisory exclusive file lock — prevents concurrent ~/.codex mutations.

    Wrap any cmd_switch / cmd_relay / cmd_oauth / cmd_init / cmd_rollback in
    `with Sub2cliLock(): ...`. Raises LockBusy if another process holds it.
    """

    _depth = 0

    def __init__(self, path: Path = LOCK_FILE, *, timeout: float = 0.0) -> None:
        self.path = path
        self.timeout = timeout
        self._fd: int | None = None
        self._nested = False

    def __enter__(self) -> "Sub2cliLock":
        if Sub2cliLock._depth > 0:
            Sub2cliLock._depth += 1
            self._nested = True
            return self
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self._fd = os.open(str(self.path), os.O_CREAT | os.O_WRONLY, 0o600)
        deadline = time.time() + self.timeout if self.timeout > 0 else None
        while True:
            try:
                fcntl.flock(self._fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
                os.ftruncate(self._fd, 0)
                os.write(self._fd, f"pid={os.getpid()} t={int(time.time())}\n".encode())
                Sub2cliLock._depth = 1
                return self
            except OSError as exc:
                if exc.errno not in (errno.EAGAIN, errno.EACCES):
                    os.close(self._fd)
                    self._fd = None
                    raise
                if deadline is None or time.time() >= deadline:
                    os.close(self._fd)
                    self._fd = None
                    raise LockBusy(
                        f"another sub2cli-inject is running (lock: {self.path})"
                    )
                time.sleep(0.1)

    def __exit__(self, *_exc: object) -> None:
        if self._nested:
            Sub2cliLock._depth = max(0, Sub2cliLock._depth - 1)
            return
        Sub2cliLock._depth = 0
        if self._fd is not None:
            try:
                fcntl.flock(self._fd, fcntl.LOCK_UN)
            finally:
                os.close(self._fd)
                self._fd = None


def atomic_write_json(path: Path, data: dict, *, mode: int = 0o600) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(f"{path.name}.{os.getpid()}.{time.time_ns()}.tmp")
    try:
        tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n")
        os.chmod(tmp, mode)
        tmp.replace(path)
    except Exception:
        if tmp.exists() or tmp.is_symlink():
            try:
                tmp.unlink()
            except OSError:
                pass
        raise


def atomic_write_text(path: Path, text: str, *, mode: int = 0o600) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(f"{path.name}.{os.getpid()}.{time.time_ns()}.tmp")
    try:
        tmp.write_text(text)
        os.chmod(tmp, mode)
        tmp.replace(path)
    except Exception:
        if tmp.exists() or tmp.is_symlink():
            try:
                tmp.unlink()
            except OSError:
                pass
        raise


def atomic_write_bytes(path: Path, data: bytes, *, mode: int = 0o600) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(f"{path.name}.{os.getpid()}.{time.time_ns()}.tmp")
    try:
        tmp.write_bytes(data)
        os.chmod(tmp, mode)
        tmp.replace(path)
    except Exception:
        if tmp.exists() or tmp.is_symlink():
            try:
                tmp.unlink()
            except OSError:
                pass
        raise


def atomic_symlink(link_path: Path, target: Path) -> None:
    """Replace link_path with a symlink to target. Used for APP_PROFILE.

    AUTH_JSON is no longer symlinked — codex CLI/App atomic-writes auth.json
    when refreshing tokens, which would replace any symlink with a real file
    (cat-and-mouse). Use copy_auth_atomic() instead for auth.json.
    """
    if link_path.exists() and not link_path.is_symlink():
        fail(f"{link_path} is a real file/dir, not a symlink. Run init first or move it aside manually.")
    link_path.parent.mkdir(parents=True, exist_ok=True)
    target.parent.mkdir(parents=True, exist_ok=True)
    tmp = link_path.with_name(f"{link_path.name}.{os.getpid()}.{time.time_ns()}.tmp")
    if tmp.exists() or tmp.is_symlink():
        tmp.unlink()
    tmp.symlink_to(target)
    os.replace(tmp, link_path)


# ==================== auth.json as a real file (no longer symlinked) ====================

def read_auth_bytes() -> bytes | None:
    """Return AUTH_JSON bytes regardless of whether it's a real file or symlink.

    Returns None if AUTH_JSON doesn't exist or can't be read.
    """
    if not AUTH_JSON.exists():
        return None
    try:
        return AUTH_JSON.read_bytes()
    except OSError:
        return None


def copy_auth_atomic(src: Path, *, mode: int = 0o600) -> None:
    """Atomic-write src's content to AUTH_JSON as a real file (replacing any
    existing symlink or file). Used by cmd_switch when changing slots.
    """
    if not src.exists():
        fail(f"slot auth file missing: {src}")
    data = src.read_bytes()
    AUTH_JSON.parent.mkdir(parents=True, exist_ok=True)
    tmp = AUTH_JSON.with_name(f"{AUTH_JSON.name}.{os.getpid()}.{time.time_ns()}.tmp")
    try:
        tmp.write_bytes(data)
        os.chmod(tmp, mode)
        os.replace(tmp, AUTH_JSON)
    except Exception:
        if tmp.exists() or tmp.is_symlink():
            try:
                tmp.unlink()
            except OSError:
                pass
        raise


def flush_auth_to_slot(slot_auth_file: Path, *, mode: int = 0o600) -> bool:
    """Save current AUTH_JSON content back to its slot's auth_file so that
    in-flight OAuth refresh tokens written by codex aren't lost on switch.

    Returns True if a flush was performed (content changed), False if no-op.
    """
    cur = read_auth_bytes()
    if cur is None:
        return False
    slot_auth_file.parent.mkdir(parents=True, exist_ok=True)
    if slot_auth_file.exists():
        try:
            if slot_auth_file.read_bytes() == cur:
                return False
        except OSError:
            pass
    tmp = slot_auth_file.with_name(f"{slot_auth_file.name}.{os.getpid()}.{time.time_ns()}.tmp")
    try:
        tmp.write_bytes(cur)
        os.chmod(tmp, mode)
        os.replace(tmp, slot_auth_file)
    except Exception:
        if tmp.exists() or tmp.is_symlink():
            try:
                tmp.unlink()
            except OSError:
                pass
        raise
    return True


def auth_content_equals(slot_auth_file: Path) -> bool:
    """Check if AUTH_JSON content matches a slot's auth_file content."""
    if not AUTH_JSON.exists() or not slot_auth_file.exists():
        return False
    try:
        return AUTH_JSON.read_bytes() == slot_auth_file.read_bytes()
    except OSError:
        return False


def symlink_points_to(link_path: Path, target: Path) -> bool:
    if not link_path.is_symlink():
        return False
    try:
        return Path(os.readlink(link_path)).expanduser().resolve(strict=False) == target.resolve(strict=False)
    except OSError:
        return False


def path_present(path: Path) -> bool:
    return path.exists() or path.is_symlink()


def lexical_abs(path: Path) -> Path:
    return Path(os.path.abspath(os.path.expanduser(str(path))))


def path_key(path: Path) -> str:
    return str(lexical_abs(path))


def is_within(path: Path, base: Path) -> bool:
    try:
        lexical_abs(path).relative_to(lexical_abs(base))
        return True
    except ValueError:
        return False


def safe_unlink_file(path: Path) -> bool:
    if not path_present(path):
        return False
    if path.is_dir() and not path.is_symlink():
        fail(f"refuse to unlink directory as file: {path}")
    path.unlink()
    return True


def safe_remove_profile_dir(path: Path) -> bool:
    if not path_present(path):
        return False
    if not is_within(path, APP_SUPPORT):
        fail(f"refuse to remove profile outside Application Support: {path}")
    if not lexical_abs(path).name.startswith("Codex."):
        fail(f"refuse to remove non-slot Codex profile: {path}")
    if path.is_symlink() or path.is_file():
        path.unlink()
        return True
    if path.is_dir():
        shutil.rmtree(path)
        return True
    return False


def toml_quote(value: str) -> str:
    return json.dumps(str(value), ensure_ascii=False)[1:-1]


def normalize_base_url(value: str) -> str:
    base_url = value.strip().rstrip("/")
    if base_url.lower().endswith("/v1"):
        base_url = base_url[:-3].rstrip("/")
    if not base_url.startswith(("http://", "https://")):
        fail(f"base_url must start with http:// or https://: {value!r}")
    return base_url


def slot_from_url(base_url: str) -> str:
    host = urlparse.urlparse(base_url).hostname or ""
    host = host.removeprefix("www.").removeprefix("api.")
    slot = host.split(".")[0] if host else ""
    if not slot:
        fail(f"could not derive a slot name from {base_url!r}; pass --slot")
    return validate_slot(slot)


def validate_slot(slot: str) -> str:
    slot = slot.strip().lower()
    if not re.fullmatch(r"[a-z0-9][a-z0-9_-]{0,30}", slot):
        fail(f"invalid slot {slot!r}; use lowercase a-z, 0-9, _ or -, max 31 chars")
    return slot


def load_slots(required: bool = True) -> dict:
    if not SLOTS_FILE.exists():
        if required:
            fail(f"{SLOTS_FILE} not found. Run: codex-provider init")
        return {"version": 1, "current": None, "app_history_slot": None, "slots": {}}
    try:
        data = json.loads(SLOTS_FILE.read_text())
    except json.JSONDecodeError as exc:
        fail(f"{SLOTS_FILE} is invalid JSON: {exc}")
    data.setdefault("version", 1)
    data.setdefault("slots", {})
    return data


def save_slots(data: dict) -> None:
    atomic_write_json(SLOTS_FILE, data)


def codex_running() -> bool:
    return subprocess.run(["pgrep", "-x", "Codex"], capture_output=True).returncode == 0


def quit_codex(timeout: float = 8.0) -> bool:
    if not codex_running():
        return False
    try:
        subprocess.run(
            ["osascript", "-e", 'tell application id "com.openai.codex" to quit'],
            check=False,
            timeout=timeout,
            capture_output=True,
            text=True,
        )
    except subprocess.TimeoutExpired:
        return True
    deadline = time.time() + timeout
    while time.time() < deadline:
        if not codex_running():
            return True
        time.sleep(0.25)
    fail("Codex.app 退出超时。请手动退出 Codex App 后重试。")
    return True


def launch_codex() -> None:
    subprocess.Popen(["open", "-a", "Codex"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)


def decode_auth_email(auth_file: Path) -> str | None:
    try:
        data = json.loads(auth_file.read_text())
        token = data.get("tokens", {}).get("id_token", "")
        if token.count(".") < 2:
            return None
        payload = token.split(".")[1]
        payload += "=" * (-len(payload) % 4)
        return json.loads(base64.urlsafe_b64decode(payload)).get("email")
    except Exception:
        return None


def _migrate_legacy_auth_symlink(*, quiet: bool = False) -> bool:
    """If AUTH_JSON is a legacy symlink (v0.2.0 layout), break it into a real
    file in place. Idempotent — safe to call from any cmd_init entry path,
    including upgrade users who already have populated slots.

    Returns True iff a migration was performed.
    """
    if not AUTH_JSON.is_symlink():
        return False
    legacy_target = Path(os.readlink(AUTH_JSON)).expanduser()
    if not legacy_target.is_absolute():
        legacy_target = (AUTH_JSON.parent / legacy_target).resolve(strict=False)
    if legacy_target.exists():
        try:
            content = legacy_target.read_bytes()
        except OSError:
            content = b""
        AUTH_JSON.unlink()
        if content:
            AUTH_JSON.write_bytes(content)
            os.chmod(AUTH_JSON, 0o600)
        if not quiet:
            print(f"  ↳ migrated legacy auth.json symlink → real file ({len(content)} bytes)")
    else:
        AUTH_JSON.unlink()
        if not quiet:
            print(f"  ↳ removed dangling legacy auth.json symlink ({legacy_target} missing)")
    return True


def cmd_init(slot: str = "local", display: str | None = None, *, quiet: bool = False) -> int:
    if Sub2cliLock._depth == 0:
        try:
            with Sub2cliLock():
                return cmd_init(slot=slot, display=display, quiet=quiet)
        except LockBusy as exc:
            fail(str(exc), code=4)
    slot = validate_slot(slot)
    data = load_slots(required=False)

    # Idempotent legacy migration: run BEFORE the early-return so upgrade
    # users (v0.2.0 layout: slots already populated, auth.json still a symlink)
    # get their symlink broken into a real file. Without this, codex's next
    # token-refresh atomic-write would replace the symlink anyway — but later
    # switches would race the symlink check against codex's writes.
    legacy_migrated = _migrate_legacy_auth_symlink(quiet=quiet)

    # Already initialized if slots are configured and APP_PROFILE is a symlink.
    # AUTH_JSON is now intentionally a real file (no longer symlinked).
    if data.get("current") and data.get("slots") and APP_PROFILE.is_symlink():
        if not quiet:
            print(f"already initialized: current={data['current']}")
            print(f"slots file: {SLOTS_FILE}")
        return 0

    quit_codex()
    CODEX_HOME.mkdir(parents=True, exist_ok=True)
    APP_SUPPORT.mkdir(parents=True, exist_ok=True)

    auth_slot = CODEX_HOME / f"auth.{slot}.json"
    profile_slot = APP_SUPPORT / f"Codex.{slot}"

    if AUTH_JSON.exists():
        # Real file already (typical post-codex-refresh state). Preserve content
        # into auth_slot if absent, then leave AUTH_JSON in place.
        if not auth_slot.exists():
            auth_slot.parent.mkdir(parents=True, exist_ok=True)
            auth_slot.write_bytes(AUTH_JSON.read_bytes())
            os.chmod(auth_slot, 0o600)
        elif auth_slot.read_bytes() != AUTH_JSON.read_bytes():
            # Existing auth_slot differs — keep the newer (AUTH_JSON) so live
            # tokens are preserved. Back up the old slot file.
            bak = auth_slot.with_suffix(auth_slot.suffix + f".pre-init-{int(time.time())}.bak")
            bak.write_bytes(auth_slot.read_bytes())
            os.chmod(bak, 0o600)
            auth_slot.write_bytes(AUTH_JSON.read_bytes())
            os.chmod(auth_slot, 0o600)

    if APP_PROFILE.is_symlink():
        profile_slot = Path(os.readlink(APP_PROFILE)).expanduser()
    elif APP_PROFILE.exists():
        if profile_slot.exists():
            fail(f"{profile_slot} already exists; refusing to overwrite it")
        APP_PROFILE.rename(profile_slot)
    else:
        profile_slot.mkdir(parents=True, exist_ok=True)

    atomic_symlink(APP_PROFILE, profile_slot)

    email = decode_auth_email(auth_slot) if auth_slot.exists() else None
    display_name = display or (f"Codex - {email}" if email else f"Codex - {slot}")
    data["slots"][slot] = {
        "display_name": display_name,
        "mode": "oauth",
        "auth_file": str(auth_slot),
        "app_profile_dir": str(profile_slot),
    }
    data["current"] = slot
    data["app_history_slot"] = slot
    save_slots(data)

    if not quiet:
        print(f"initialized: {slot}")
        print(f"  auth:    {auth_slot}")
        print(f"  profile: {profile_slot}")
        print(f"  slots:   {SLOTS_FILE}")
    return 0


def ensure_initialized() -> None:
    if not SLOTS_FILE.exists():
        cmd_init(quiet=True)


def probe_relay(base_url: str, api_key: str, timeout: float = 12.0) -> tuple[bool, str]:
    req = urlrequest.Request(
        base_url.rstrip("/") + "/v1/models",
        headers={
            "Authorization": f"Bearer {api_key}",
            "Accept": "application/json",
            "User-Agent": "codex-provider-macos/1.0",
        },
    )
    try:
        with urlrequest.urlopen(req, timeout=timeout) as resp:
            body = resp.read(8192)
        parsed = json.loads(body)
        if isinstance(parsed, dict) and isinstance(parsed.get("data"), list):
            return True, f"/v1/models 可访问（{len(parsed['data'])} 个模型）"
        return False, "中转有响应，但返回格式不像 OpenAI 兼容接口"
    except urlerror.HTTPError as exc:
        if exc.code == 401:
            return False, "401 Unauthorized - API key 不正确"
        if exc.code == 403:
            return False, "403 Forbidden - key 权限不足，或中转拒绝当前网络"
        if exc.code == 404:
            return False, "404 Not Found - base URL 可能不对，通常不要带 /v1"
        return False, f"HTTP {exc.code} {exc.reason}"
    except urlerror.URLError as exc:
        return False, f"无法访问中转：{exc.reason}"
    except TimeoutError:
        return False, f"{timeout:.0f}s 后超时"
    except Exception as exc:
        return False, f"{type(exc).__name__}: {exc}"


def write_apikey_auth(path: Path, api_key: str) -> bool:
    desired = {"OPENAI_API_KEY": api_key, "auth_mode": "apikey"}
    if path.exists():
        try:
            if json.loads(path.read_text()) == desired:
                return False
        except (OSError, json.JSONDecodeError):
            pass
    atomic_write_json(path, desired)
    return True


def split_toml_head(text: str) -> tuple[str, str]:
    match = re.search(r"(?m)^\s*\[", text)
    if not match:
        return text, ""
    return text[: match.start()], text[match.start() :]


def set_top_level_config(text: str, updates: dict[str, str]) -> str:
    head, tail = split_toml_head(text)
    keys = "|".join(re.escape(key) for key in updates)
    if keys:
        head = re.sub(rf'(?m)^\s*({keys})\s*=.*(?:\n)?', "", head)
    block = "".join(f'{key} = "{toml_quote(value)}"\n' for key, value in updates.items())
    return block + head.lstrip("\n") + tail


def strip_relay_provider_block(text: str) -> str:
    return re.sub(
        rf"(?ms)^\s*\[model_providers\.{re.escape(RELAY_PROVIDER)}\]\s*\n.*?(?=^\s*\[|\Z)",
        "",
        text,
        flags=re.IGNORECASE,
    ).rstrip() + "\n"


def patch_config(*, mode: str, model: str, relay_base_url: str | None = None) -> bool:
    text = CONFIG_TOML.read_text() if CONFIG_TOML.exists() else ""
    provider = RELAY_PROVIDER if mode == "relay" else OAUTH_PROVIDER
    new_text = set_top_level_config(
        text,
        {
            "model": model,
            "model_provider": provider,
            "api_base_url": DEFAULT_API_BASE_URL,
        },
    )
    new_text = strip_relay_provider_block(new_text)
    if mode == "relay":
        if not relay_base_url:
            fail("relay_base_url is required for relay mode")
        block = (
            f"\n[model_providers.{RELAY_PROVIDER}]\n"
            f'name = "{RELAY_PROVIDER}"\n'
            f'base_url = "{toml_quote(relay_base_url.rstrip("/"))}"\n'
            'wire_api = "responses"\n'
            "requires_openai_auth = true\n"
        )
        new_text = new_text.rstrip() + "\n" + block
    if new_text == text:
        return False
    atomic_write_text(CONFIG_TOML, new_text)
    return True


def top_level_value(text: str, key: str) -> str | None:
    head, _ = split_toml_head(text)
    match = re.search(rf'(?m)^\s*{re.escape(key)}\s*=\s*"([^"]*)"', head)
    return match.group(1) if match else None


def relay_block_base_url(text: str) -> str | None:
    match = re.search(
        rf"(?ms)^\s*\[model_providers\.{re.escape(RELAY_PROVIDER)}\]\s*\n(.*?)(?=^\s*\[|\Z)",
        text,
        flags=re.IGNORECASE,
    )
    if not match:
        return None
    body = match.group(1)
    url_match = re.search(r'(?m)^\s*base_url\s*=\s*"([^"]*)"', body)
    return url_match.group(1).rstrip("/") if url_match else None


def config_clean_for_slot(cfg: dict) -> bool:
    try:
        text = CONFIG_TOML.read_text()
    except OSError:
        return False
    mode = cfg.get("mode", "oauth")
    expected_provider = RELAY_PROVIDER if mode == "relay" else OAUTH_PROVIDER
    expected_model = cfg.get("model", DEFAULT_MODEL)
    if top_level_value(text, "model_provider") != expected_provider:
        return False
    if top_level_value(text, "model") != expected_model:
        return False
    if top_level_value(text, "api_base_url") != DEFAULT_API_BASE_URL:
        return False
    relay_url = relay_block_base_url(text)
    if mode == "relay":
        return relay_url == str(cfg.get("base_url", "")).rstrip("/")
    return relay_url is None


def new_backup_dir(label: str) -> Path:
    safe = re.sub(r"[^A-Za-z0-9_.-]+", "-", label)
    path = BACKUP_ROOT / f"{time.strftime('%Y%m%d-%H%M%S')}-{time.time_ns()}-{safe}"
    path.mkdir(parents=True, exist_ok=False)
    return path


def backup_state_db(backup_dir: Path) -> Path | None:
    if not STATE_DB.exists():
        return None
    target = backup_dir / STATE_DB.name
    with sqlite3.connect(f"file:{STATE_DB}?mode=ro", uri=True, timeout=30) as src:
        src.execute("PRAGMA busy_timeout=30000")
        with sqlite3.connect(target) as dst:
            src.backup(dst)
    os.chmod(target, 0o600)
    return target


def thread_provider_counts() -> dict[str, int]:
    if not STATE_DB.exists():
        return {}
    try:
        with sqlite3.connect(f"file:{STATE_DB}?mode=ro", uri=True, timeout=10) as conn:
            conn.execute("PRAGMA busy_timeout=10000")
            rows = conn.execute("SELECT model_provider, COUNT(*) FROM threads GROUP BY model_provider").fetchall()
    except sqlite3.Error:
        return {}
    return {str(provider): int(count) for provider, count in rows}


def normalize_state_db(target_provider: str, *, backup_dir: Path | None, dry_run: bool) -> dict:
    stats = {"exists": STATE_DB.exists(), "changed": 0, "backup": None, "error": None}
    if not STATE_DB.exists():
        return stats
    try:
        with sqlite3.connect(STATE_DB, timeout=30) as conn:
            conn.execute("PRAGMA busy_timeout=30000")
            pending = conn.execute(
                "SELECT COUNT(*) FROM threads WHERE model_provider IN (?, ?) AND model_provider != ?",
                (*THREAD_PROVIDERS, target_provider),
            ).fetchone()[0]
            stats["changed"] = int(pending)
            if dry_run or pending == 0:
                return stats
            if backup_dir is not None:
                stats["backup"] = str(backup_state_db(backup_dir))
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "UPDATE threads SET model_provider = ? WHERE model_provider IN (?, ?) AND model_provider != ?",
                (target_provider, *THREAD_PROVIDERS, target_provider),
            )
            conn.commit()
    except sqlite3.Error as exc:
        stats["error"] = f"{type(exc).__name__}: {exc}"
    return stats


def extract_rollout_thread_id(line: bytes) -> str | None:
    match = re.search(rb'"id":"([^"]+)"', line)
    if not match:
        return None
    try:
        return match.group(1).decode("utf-8")
    except UnicodeDecodeError:
        return None


def rollout_provider_records(path: Path, target_provider: str, *, dry_run: bool) -> list[dict]:
    tokens = [(provider, f'"model_provider":"{provider}"'.encode()) for provider in THREAD_PROVIDERS]
    records: list[dict] = []
    try:
        offset = 0
        with path.open("rb") as handle:
            for line_no, line in enumerate(handle, start=1):
                if b'"type":"session_meta"' not in line or b'"model_provider":"' not in line:
                    offset += len(line)
                    continue
                matches = [(line.find(token), provider, token) for provider, token in tokens]
                matches = [(pos, provider, token) for pos, provider, token in matches if pos >= 0]
                if not matches:
                    offset += len(line)
                    continue
                pos, provider, _ = min(matches, key=lambda item: item[0])
                if provider != target_provider:
                    records.append(
                        {
                            "path": str(path),
                            "line": line_no,
                            "offset": offset + pos + len(b'"model_provider":"'),
                            "thread_id": extract_rollout_thread_id(line[:4096]),
                            "from": provider,
                            "to": target_provider,
                        }
                    )
                offset += len(line)
    except OSError as exc:
        return [{"path": str(path), "error": f"{type(exc).__name__}: {exc}"}]

    if dry_run or not records:
        return records

    try:
        before = path.stat()
        with path.open("r+b") as handle:
            for record in records:
                provider = str(record["from"])
                handle.seek(int(record["offset"]))
                current = handle.read(len(provider))
                if current != provider.encode():
                    record["error"] = "provider bytes changed before write"
                    continue
                handle.seek(int(record["offset"]))
                handle.write(target_provider.encode())
            handle.flush()
            os.fsync(handle.fileno())
        os.utime(path, ns=(before.st_atime_ns, before.st_mtime_ns))
    except OSError as exc:
        for record in records:
            record["error"] = f"{type(exc).__name__}: {exc}"
    return records


def normalize_rollouts(target_provider: str, *, backup_dir: Path | None, dry_run: bool) -> dict:
    stats = {"scanned": 0, "changed": 0, "errors": 0, "audit": None}
    audit_handle = None
    try:
        if backup_dir is not None and not dry_run:
            audit_path = backup_dir / "rollout-provider-changes.jsonl"
            audit_handle = audit_path.open("w", encoding="utf-8")
            stats["audit"] = str(audit_path)
        for root in SESSION_DIRS:
            if not root.exists():
                continue
            for path in root.rglob("rollout-*.jsonl"):
                stats["scanned"] += 1
                for record in rollout_provider_records(path, target_provider, dry_run=dry_run):
                    if record.get("error"):
                        stats["errors"] += 1
                    else:
                        stats["changed"] += 1
                    if audit_handle is not None:
                        audit_handle.write(json.dumps(record, ensure_ascii=False) + "\n")
    finally:
        if audit_handle is not None:
            audit_handle.close()
    return stats


def normalize_sessions(
    target_provider: str,
    *,
    dry_run: bool = False,
    backup_dir: Path | None = None,
) -> dict:
    """Normalize sessions DB + rollouts to target_provider. If caller passes a
    `backup_dir` (e.g. the switch-level backup), reuse it so a later rollback
    finds the state_5.sqlite snapshot and rollout-provider-changes audit
    inside the same dir it restores config.toml / auth.json from.
    """
    if target_provider not in THREAD_PROVIDERS:
        fail(f"unsupported target provider {target_provider!r}; expected openai or OpenAI")
    preview = {
        "target_provider": target_provider,
        "dry_run": dry_run,
        "backup_dir": None,
        "state_db": None,
        "rollouts": None,
    }
    if not dry_run and backup_dir is None:
        backup_dir = new_backup_dir(f"thread-provider-{target_provider}")
    if dry_run:
        backup_dir = None  # don't write anything in dry-run
    preview["backup_dir"] = str(backup_dir) if backup_dir is not None else None
    preview["state_db"] = normalize_state_db(target_provider, backup_dir=backup_dir, dry_run=dry_run)
    preview["rollouts"] = normalize_rollouts(target_provider, backup_dir=backup_dir, dry_run=dry_run)
    if backup_dir is not None:
        # Name the manifest so it doesn't collide with switch_pre_switch's
        # other manifest files inside the same backup_dir.
        atomic_write_json(backup_dir / "thread-provider-manifest.json", preview)
    return preview


def normalize_sessions_if_needed(
    target_provider: str,
    *,
    backup_dir: Path | None = None,
) -> dict:
    preview = normalize_sessions(target_provider, dry_run=True)
    db_changed = int(preview["state_db"].get("changed", 0))
    db_error = preview["state_db"].get("error")
    rollouts = preview["rollouts"]
    rollout_changed = int(rollouts.get("changed", 0))
    rollout_errors = int(rollouts.get("errors", 0))
    if db_changed == 0 and not db_error and rollout_changed == 0 and rollout_errors == 0:
        preview["dry_run"] = False
        preview["backup_dir"] = None
        preview["skipped"] = True
        return preview
    return normalize_sessions(target_provider, dry_run=False, backup_dir=backup_dir)


def sessions_need_normalize(target_provider: str) -> bool:
    preview = normalize_sessions(target_provider, dry_run=True)
    return bool(
        preview["state_db"].get("error")
        or int(preview["state_db"].get("changed", 0))
        or int(preview["rollouts"].get("changed", 0))
        or int(preview["rollouts"].get("errors", 0))
    )


def print_session_summary(stats: dict) -> None:
    action = "将聚合" if stats.get("dry_run") else "已聚合"
    db_stats = stats.get("state_db") or {}
    rollout_stats = stats.get("rollouts") or {}
    print(
        f"  历史 session {action}: provider={stats.get('target_provider')} "
        f"(DB 行: {db_stats.get('changed', 0)}, rollout 头: {rollout_stats.get('changed', 0)})"
    )
    if db_stats.get("error"):
        print(f"  state DB 错误: {db_stats['error']}")
    if rollout_stats.get("errors"):
        print(f"  rollout 错误: {rollout_stats['errors']}")
    if stats.get("backup_dir"):
        print(f"  备份: {stats['backup_dir']}")


def resolve_slot(name: str, slots: dict) -> str:
    query = name.strip().lower()
    if query in slots:
        return query
    matches = [key for key, cfg in slots.items() if query in key or query in cfg.get("display_name", "").lower()]
    if len(matches) == 1:
        return matches[0]
    if not matches:
        fail(f"no slot matches {name!r}. Available: {', '.join(sorted(slots)) or '(none)'}")
    fail(f"{name!r} is ambiguous: {', '.join(matches)}")
    return query


def history_profile_target(data: dict, target_slot: str) -> Path:
    slots = data.get("slots", {})
    history_slot = data.get("app_history_slot")
    if history_slot in slots:
        return Path(slots[history_slot]["app_profile_dir"]).expanduser()
    return Path(slots[target_slot]["app_profile_dir"]).expanduser()


def slot_thread_provider(cfg: dict) -> str:
    return RELAY_PROVIDER if cfg.get("mode") == "relay" else OAUTH_PROVIDER


def slot_is_clean(data: dict, target_slot: str) -> bool:
    cfg = data["slots"][target_slot]
    profile_target = history_profile_target(data, target_slot)
    if data.get("current") != target_slot:
        return False
    if not symlink_points_to(APP_PROFILE, profile_target):
        return False
    # AUTH_JSON is a real file. For relay we check exact content; for oauth we
    # check auth_mode and size (token refresh changes bytes, so we can't compare
    # byte-equal with the slot's auth_file).
    if cfg.get("mode") == "relay":
        api_key = str(cfg.get("api_key") or "")
        try:
            if json.loads(AUTH_JSON.read_text()) != {"OPENAI_API_KEY": api_key, "auth_mode": "apikey"}:
                return False
        except (OSError, json.JSONDecodeError):
            return False
    else:
        # OAuth: full token JSON is several KB; API-key JSON is ~120 bytes.
        # Catch relay→oauth switch failures (auth.json still holds API-key JSON).
        if not AUTH_JSON.exists() or AUTH_JSON.stat().st_size < 1000:
            return False
        try:
            if json.loads(AUTH_JSON.read_text()).get("auth_mode") == "apikey":
                return False
        except (OSError, json.JSONDecodeError):
            return False
    return config_clean_for_slot(cfg) and not sessions_need_normalize(slot_thread_provider(cfg))


# ==================== Dry-run plan + snapshot/restore ====================

def collect_slot_auth_paths(data: dict) -> list[Path]:
    paths: list[Path] = []
    for cfg in (data.get("slots") or {}).values():
        raw = str((cfg or {}).get("auth_file") or "").strip()
        if raw:
            paths.append(Path(raw).expanduser())
    return paths


def collect_slot_profile_paths(data: dict) -> list[Path]:
    paths: list[Path] = []
    for cfg in (data.get("slots") or {}).values():
        raw = str((cfg or {}).get("app_profile_dir") or "").strip()
        if raw:
            paths.append(Path(raw).expanduser())
    return paths


def managed_existing_paths(data: dict | None = None) -> list[str]:
    paths: list[Path] = [CONFIG_TOML, AUTH_JSON, SLOTS_FILE, APP_PROFILE]
    if data:
        paths.extend(collect_slot_auth_paths(data))
        paths.extend(collect_slot_profile_paths(data))
    try:
        paths.extend(CODEX_HOME.glob("auth.*.json"))
    except OSError:
        pass
    return sorted({path_key(p) for p in paths if path_present(p)})


def backup_existing_slot_auth_files(backup_dir: Path, data: dict | None = None) -> dict[str, str]:
    candidates: list[Path] = []
    if data:
        candidates.extend(collect_slot_auth_paths(data))
    try:
        candidates.extend(CODEX_HOME.glob("auth.*.json"))
    except OSError:
        pass

    backups: dict[str, str] = {}
    auth_backup_dir = backup_dir / "slot-auth"
    seen: set[str] = set()
    for raw in candidates:
        path = raw.expanduser()
        key = path_key(path)
        if key in seen or not path_present(path):
            continue
        seen.add(key)
        if path.is_dir() and not path.is_symlink():
            continue
        target = auth_backup_dir / f"{len(backups):04d}.bin"
        target.parent.mkdir(parents=True, exist_ok=True)
        try:
            target.write_bytes(path.read_bytes())
        except OSError:
            continue
        os.chmod(target, 0o600)
        backups[key] = str(target)
    return backups


def snapshot_pre_switch(backup_dir: Path) -> dict:
    """Capture rollback-able state of ~/.codex into backup_dir.

    Captures: full config.toml content, AUTH_JSON file content (as bytes),
    current APP_PROFILE symlink target. state_5.sqlite + rollout JSONLs are
    backed up separately by backup_state_db() / rollout audit.

    Schema 3: auth.json is a real file (not symlinked), and rollback also
    removes managed config/auth/profile artifacts that did not exist before.
    """
    slots_state: dict | None = None
    if SLOTS_FILE.exists():
        try:
            slots_state = json.loads(SLOTS_FILE.read_text())
        except (OSError, json.JSONDecodeError):
            slots_state = None

    snap: dict = {
        "schema": 3,
        "timestamp": int(time.time()),
        "config_toml_existed": path_present(CONFIG_TOML),
        "config_toml_backup": None,
        "auth_json_existed": path_present(AUTH_JSON),
        "auth_json_backup": None,
        "app_profile_existed": path_present(APP_PROFILE),
        "app_profile_target": None,
        "provider_slots_backup": None,
        "provider_slots_existed": SLOTS_FILE.exists(),
        "current_slot": None,
        "app_history_slot": None,
        "managed_existing_paths": managed_existing_paths(slots_state),
        "slot_auth_backups": backup_existing_slot_auth_files(backup_dir, slots_state),
    }
    if CONFIG_TOML.exists():
        target = backup_dir / "config.toml"
        target.write_text(CONFIG_TOML.read_text())
        os.chmod(target, 0o600)
        snap["config_toml_backup"] = str(target)
    if AUTH_JSON.exists():
        target = backup_dir / "auth.json"
        target.write_bytes(AUTH_JSON.read_bytes())
        os.chmod(target, 0o600)
        snap["auth_json_backup"] = str(target)
    if APP_PROFILE.is_symlink():
        snap["app_profile_target"] = os.readlink(APP_PROFILE)
    if SLOTS_FILE.exists():
        target = backup_dir / SLOTS_FILE.name
        target.write_text(SLOTS_FILE.read_text())
        os.chmod(target, 0o600)
        snap["provider_slots_backup"] = str(target)
    # Record provider-slots.json pointers so rollback can revert `current` /
    # `app_history_slot` and `status` reads as "consistent" again.
    if slots_state:
        try:
            snap["current_slot"] = slots_state.get("current")
            snap["app_history_slot"] = slots_state.get("app_history_slot")
        except AttributeError:
            pass
    atomic_write_json(backup_dir / SNAPSHOT_FILENAME, snap)
    return snap


def compute_switch_plan(data: dict, target_slot: str) -> dict:
    """Compute what cmd_switch would change — read-only, no mutations."""
    cfg = data["slots"][target_slot]
    mode = cfg.get("mode", "oauth")
    auth_target = Path(cfg["auth_file"]).expanduser()
    profile_target = history_profile_target(data, target_slot)
    target_provider = slot_thread_provider(cfg)

    plan: dict = {
        "target_slot": target_slot,
        "mode": mode,
        "display_name": cfg.get("display_name", target_slot),
        "is_clean": slot_is_clean(data, target_slot),
        "actions": [],
    }
    if plan["is_clean"]:
        plan["actions"].append({"type": "noop", "reason": "已是当前渠道, 状态一致"})
        return plan

    # AUTH_JSON should be a real file post-migration. Path.stat() follows
    # legacy symlinks (size = target's size), so this works for both layouts.
    cur_size = AUTH_JSON.stat().st_size if AUTH_JSON.exists() else None
    cur_kind = "symlink" if AUTH_JSON.is_symlink() else "real-file"
    plan["actions"].append({
        "type": "auth_copy",
        "path": str(AUTH_JSON),
        "from_size": cur_size,
        "from_kind": cur_kind,
        "from_slot": data.get("current"),
        "to": str(auth_target),
    })
    cur_profile = os.readlink(APP_PROFILE) if APP_PROFILE.is_symlink() else None
    plan["actions"].append({
        "type": "symlink",
        "path": str(APP_PROFILE),
        "from": cur_profile,
        "to": str(profile_target),
    })

    if mode == "relay":
        plan["actions"].append({
            "type": "write",
            "path": str(auth_target),
            "what": "API key auth.json (apikey mode)",
        })
        plan["actions"].append({
            "type": "patch_toml",
            "path": str(CONFIG_TOML),
            "section": f"[model_providers.{RELAY_PROVIDER}]",
            "base_url": cfg.get("base_url"),
            "model": cfg.get("model", DEFAULT_MODEL),
        })
    else:
        plan["actions"].append({
            "type": "patch_toml",
            "path": str(CONFIG_TOML),
            "section": "(remove relay block; oauth mode)",
            "model": cfg.get("model", DEFAULT_MODEL),
        })

    if sessions_need_normalize(target_provider):
        norm_preview = normalize_sessions(target_provider, dry_run=True)
        plan["actions"].append({
            "type": "normalize_sessions",
            "target_provider": target_provider,
            "state_db_changes": int(norm_preview["state_db"].get("changed", 0)),
            "rollout_changes": int(norm_preview["rollouts"].get("changed", 0)),
        })

    if codex_running():
        plan["actions"].append({"type": "quit_codex"})
        plan["actions"].append({"type": "launch_codex"})
    else:
        plan["actions"].append({"type": "launch_codex", "note": "not running"})

    return plan


def print_switch_plan(plan: dict) -> None:
    print(f"DRY-RUN: 将切换到 {plan['display_name']} ({plan['target_slot']}) [{plan['mode']}]")
    if plan["is_clean"]:
        print("  当前状态一致, 无需修改")
        return
    print(f"  计划改动 ({len(plan['actions'])}):")
    for i, a in enumerate(plan["actions"], 1):
        t = a["type"]
        if t == "symlink":
            print(f"    {i}. symlink: {a['path']}")
            print(f"       from: {a['from']}")
            print(f"       to:   {a['to']}")
        elif t == "auth_copy":
            print(f"    {i}. copy:    {a['path']}  (real file, atomic write)")
            from_desc = a.get("from_slot") or "(none)"
            from_size = a.get("from_size")
            from_kind = a.get("from_kind", "")
            if from_size is not None:
                tag = f" ({from_kind})" if from_kind else ""
                print(f"       from: slot={from_desc} (current auth.json {from_size} bytes{tag})")
            else:
                print(f"       from: slot={from_desc} (no current auth.json)")
            print(f"       src:  {a['to']}")
        elif t == "write":
            print(f"    {i}. write:  {a['path']}  ({a['what']})")
        elif t == "patch_toml":
            print(f"    {i}. patch:  {a['path']}  {a['section']}")
            if a.get("base_url"):
                print(f"       base_url = {a['base_url']}")
            print(f"       model = {a['model']}")
        elif t == "normalize_sessions":
            print(
                f"    {i}. normalize sessions → provider={a['target_provider']}"
                f"  (DB rows: {a['state_db_changes']}, rollout headers: {a['rollout_changes']})"
            )
        elif t == "quit_codex":
            print(f"    {i}. quit Codex App")
        elif t == "launch_codex":
            note = f" [{a['note']}]" if a.get("note") else ""
            print(f"    {i}. launch Codex App{note}")
        elif t == "noop":
            print(f"    {i}. noop: {a['reason']}")
    print()
    print("  (no changes applied; rerun without --dry-run to execute)")


def restore_rollouts_from_audit(audit_path: Path) -> dict:
    """Reverse-apply a rollout-provider-changes.jsonl audit: at each recorded
    offset, write the `from` provider bytes back over `to`. Used by rollback
    so that codex.app's session sidebar sees the original provider tags.

    Returns: {scanned, reverted, errors, missing} stats.
    """
    stats = {"scanned": 0, "reverted": 0, "errors": 0, "missing": 0}
    if not audit_path.exists():
        return stats
    by_path: dict[str, list[dict]] = {}
    with audit_path.open("r", encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            try:
                rec = json.loads(line)
            except json.JSONDecodeError:
                continue
            if rec.get("error") or "offset" not in rec or "from" not in rec or "to" not in rec:
                continue
            by_path.setdefault(rec["path"], []).append(rec)
    for raw_path, records in by_path.items():
        path = Path(raw_path)
        if not path.exists():
            stats["missing"] += len(records)
            continue
        try:
            before = path.stat()
            with path.open("r+b") as handle:
                for record in records:
                    stats["scanned"] += 1
                    from_provider = str(record["from"]).encode()
                    to_provider = str(record["to"]).encode()
                    # provider names in THREAD_PROVIDERS happen to all be 6 bytes
                    # (`openai` / `OpenAI`) so byte-offset reverse is safe.
                    if len(from_provider) != len(to_provider):
                        stats["errors"] += 1
                        continue
                    handle.seek(int(record["offset"]))
                    current = handle.read(len(to_provider))
                    if current != to_provider:
                        stats["errors"] += 1
                        continue
                    handle.seek(int(record["offset"]))
                    handle.write(from_provider)
                    stats["reverted"] += 1
                handle.flush()
                os.fsync(handle.fileno())
            # Preserve original mtime so codex.app's session-list sort doesn't change.
            os.utime(path, ns=(before.st_atime_ns, before.st_mtime_ns))
        except OSError:
            stats["errors"] += len(records)
    return stats


def restore_from_backup(backup_dir: Path) -> int:
    """Restore config.toml + AUTH_JSON content + APP_PROFILE symlink + sqlite
    from a backup_dir snapshot. Supports schema 1 (legacy symlink), schema 2
    (real-file auth.json), and schema 3 cleanup metadata.
    """
    snap_path = backup_dir / SNAPSHOT_FILENAME
    if not snap_path.exists():
        fail(f"snapshot not found: {snap_path}")
    snap = json.loads(snap_path.read_text())
    schema = snap.get("schema")
    if schema not in (1, 2, 3):
        fail(f"unknown snapshot schema {schema!r} in {snap_path} (expected 1, 2 or 3)")
    quit_codex()
    restored: list[str] = []
    warnings: list[str] = []
    try:
        post_slots_state = load_slots(required=False) if SLOTS_FILE.exists() else {"slots": {}}
    except SystemExit as exc:
        post_slots_state = {"slots": {}}
        warnings.append(f"current provider-slots unreadable before restore: {exc}")
    if snap.get("config_toml_backup"):
        src = Path(snap["config_toml_backup"])
        if src.exists():
            atomic_write_text(CONFIG_TOML, src.read_text())
            restored.append(f"config.toml ← {src.name}")
        else:
            warnings.append(f"config.toml backup missing: {src}")
    elif snap.get("config_toml_existed") is False and path_present(CONFIG_TOML):
        try:
            safe_unlink_file(CONFIG_TOML)
            restored.append("config.toml removed")
        except SystemExit as exc:
            warnings.append(f"config.toml was NOT removed: {exc}")
    # Schema 2/3: restore auth.json from saved bytes.
    if snap.get("auth_json_backup"):
        src = Path(snap["auth_json_backup"])
        if src.exists():
            copy_auth_atomic(src)
            restored.append(f"auth.json ← {src.name}")
        else:
            warnings.append(f"auth.json backup missing: {src} — auth.json was NOT restored")
    # Schema 1 (legacy): restore auth.json by copying the target's content
    # as a real file (no longer symlinking).
    elif snap.get("auth_json_target"):
        target = Path(snap["auth_json_target"]).expanduser()
        if not target.is_absolute():
            target = (AUTH_JSON.parent / target).resolve(strict=False)
        if target.exists():
            copy_auth_atomic(target)
            restored.append(f"auth.json ← {target} (legacy)")
        else:
            warnings.append(f"legacy auth.json target missing: {target} — auth.json was NOT restored")
    elif snap.get("auth_json_existed") is False and path_present(AUTH_JSON):
        try:
            safe_unlink_file(AUTH_JSON)
            restored.append("auth.json removed")
        except SystemExit as exc:
            warnings.append(f"auth.json was NOT removed: {exc}")
    for key, src_name in (snap.get("slot_auth_backups") or {}).items():
        src = Path(src_name)
        dst = Path(key)
        if src.exists():
            try:
                atomic_write_bytes(dst, src.read_bytes())
                restored.append(f"{dst.name} ← {src.parent.name}/{src.name}")
            except OSError as exc:
                warnings.append(f"slot auth was NOT restored: {dst}: {exc}")
        else:
            warnings.append(f"slot auth backup missing: {src}")
    if snap.get("app_profile_target"):
        target = Path(snap["app_profile_target"]).expanduser()
        try:
            atomic_symlink(APP_PROFILE, target)
            restored.append(f"App profile → {target}")
        except SystemExit:
            warnings.append(f"App profile symlink refused (target may be a real dir): {target}")
    elif snap.get("app_profile_existed") is False and path_present(APP_PROFILE):
        try:
            safe_unlink_file(APP_PROFILE)
            restored.append("App profile removed")
        except SystemExit as exc:
            warnings.append(f"App profile was NOT removed: {exc}")
    state_db_backup = backup_dir / STATE_DB.name
    if state_db_backup.exists():
        # SQLite may run in WAL mode (state_5.sqlite-wal / -shm sidecars).
        # If we only replace the main file, the WAL is replayed at open time
        # and silently re-applies the post-switch UPDATE — the rollback looks
        # successful but threads come back tagged with the new provider.
        # Wipe main + WAL + SHM together, then copy backup's main file.
        wal = STATE_DB.parent / f"{STATE_DB.name}-wal"
        shm = STATE_DB.parent / f"{STATE_DB.name}-shm"
        for p in (STATE_DB, wal, shm):
            if p.exists():
                p.unlink()
        shutil.copyfile(state_db_backup, STATE_DB)
        os.chmod(STATE_DB, 0o600)
        restored.append(f"state DB ← {state_db_backup.name}")
    # Restore the full slot registry. Switching and add-api mutate more than
    # just `current`, so pointer-only rollback can leave supposedly removed
    # relay/API-key slots behind.
    if snap.get("provider_slots_backup"):
        src = Path(snap["provider_slots_backup"])
        if src.exists():
            atomic_write_json(SLOTS_FILE, json.loads(src.read_text()))
            restored.append(f"{SLOTS_FILE.name} ← {src.name}")
        else:
            warnings.append(f"provider-slots backup missing: {src}")
    elif snap.get("provider_slots_existed") is False and SLOTS_FILE.exists():
        SLOTS_FILE.unlink()
        restored.append(f"{SLOTS_FILE.name} removed")
    elif snap.get("current_slot"):
        # Legacy snapshot fallback.
        try:
            slots_state = load_slots(required=False)
            if slots_state.get("current") != snap["current_slot"]:
                slots_state["current"] = snap["current_slot"]
                if snap.get("app_history_slot"):
                    slots_state["app_history_slot"] = snap["app_history_slot"]
                save_slots(slots_state)
                restored.append(f"current → {snap['current_slot']}")
        except OSError as exc:
            warnings.append(f"could not revert slot pointer: {exc}")
    baseline_paths = set(snap.get("managed_existing_paths") or [])
    for path in collect_slot_auth_paths(post_slots_state):
        if path_key(path) not in baseline_paths and path_present(path):
            if not is_within(path, CODEX_HOME):
                warnings.append(f"created slot auth outside CODEX_HOME was NOT removed: {path}")
                continue
            try:
                safe_unlink_file(path)
                restored.append(f"{path.name} removed")
            except SystemExit as exc:
                warnings.append(f"created slot auth was NOT removed: {path}: {exc}")
    for path in collect_slot_profile_paths(post_slots_state):
        if path_key(path) not in baseline_paths and path_present(path):
            try:
                safe_remove_profile_dir(path)
                restored.append(f"{path.name} removed")
            except SystemExit as exc:
                warnings.append(f"created profile was NOT removed: {path}: {exc}")
    # Reverse-apply rollout provider tag changes (in-place edits to rollout
    # JSONLs). Without this, codex.app's session sidebar filters by stale
    # provider tag and "loses" history visible-but-not-deleted.
    rollout_audit = backup_dir / "rollout-provider-changes.jsonl"
    if rollout_audit.exists():
        roll_stats = restore_rollouts_from_audit(rollout_audit)
        if roll_stats["reverted"]:
            restored.append(
                f"rollout headers ← {rollout_audit.name} "
                f"(reverted {roll_stats['reverted']}/{roll_stats['scanned']})"
            )
        if roll_stats["errors"]:
            warnings.append(
                f"rollout revert had {roll_stats['errors']} error(s) — "
                f"some session_meta tags may still show post-switch provider"
            )
        if roll_stats["missing"]:
            warnings.append(
                f"{roll_stats['missing']} rollout file(s) gone since switch; cannot revert tags"
            )
    for w in warnings:
        print(f"⚠️  {w}", file=sys.stderr)
    if not restored:
        print("没有可恢复的项 (snapshot 为空)")
        return 1
    print("已恢复:")
    for r in restored:
        print(f"  · {r}")
    print(f"  备份来源: {backup_dir}")
    return 1 if warnings else 0


def auto_rollback_after_failure(backup_dir: Path | None, context: str) -> None:
    if not backup_dir:
        return
    snap_path = backup_dir / SNAPSHOT_FILENAME
    if not snap_path.exists():
        return
    marker = backup_dir / AUTO_ROLLBACK_MARKER
    if marker.exists():
        return
    print(f"  {context}失败，正在自动回滚到配置前状态…", file=sys.stderr)
    try:
        rc = restore_from_backup(backup_dir)
    except BaseException as exc:
        print(f"  自动回滚失败: {type(exc).__name__}: {exc}", file=sys.stderr)
        return
    try:
        marker.write_text(f"rc={rc} time={int(time.time())}\n")
        os.chmod(marker, 0o600)
    except OSError:
        pass
    if rc == 0:
        print("  自动回滚完成", file=sys.stderr)
    else:
        print(f"  自动回滚完成但有警告 (rc={rc})", file=sys.stderr)


def cmd_switch(
    name: str,
    *,
    no_restart: bool = False,
    dry_run: bool = False,
    backup_dir: Path | None = None,
    recover_missing_oauth_auth: bool = True,
) -> int:
    if dry_run:
        # Read-only; no lock needed. Computed from a one-shot snapshot.
        data = load_slots()
        slots = data.get("slots", {})
        target_slot = resolve_slot(name, slots)
        cfg = slots[target_slot]
        mode = cfg.get("mode", "oauth")
        if mode not in ("oauth", "relay"):
            fail(f"slot {target_slot!r} has invalid mode {mode!r}")
        plan = compute_switch_plan(data, target_slot)
        print_switch_plan(plan)
        return 0

    # Acquire lock BEFORE reasoning about state, so a second invocation
    # entering at the same time re-loads inside the lock and sees the
    # committed result of the first. (The old order load→clean-check→lock
    # could act on stale data.)
    try:
        lock_ctx = Sub2cliLock()
        lock_ctx.__enter__()
    except LockBusy as exc:
        fail(str(exc), code=4)
        return 4
    try:
        ensure_initialized()
        data = load_slots()
        slots = data.get("slots", {})
        target_slot = resolve_slot(name, slots)
        cfg = slots[target_slot]
        mode = cfg.get("mode", "oauth")
        if mode not in ("oauth", "relay"):
            fail(f"slot {target_slot!r} has invalid mode {mode!r}")

        if slot_is_clean(data, target_slot):
            print(f"当前已是：{cfg.get('display_name', target_slot)} ({target_slot}) [{mode}]")
            print("  无需修改；Codex App 未重启")
            return 0

        printed_rollback = False
        if backup_dir is None:
            backup_dir = new_backup_dir(f"switch-to-{target_slot}")
            snapshot_pre_switch(backup_dir)
            print(f"  回滚命令: sub2cli-inject rollback {backup_dir.name}")
            printed_rollback = True

        if not no_restart:
            quit_codex()

        auth_target = Path(cfg["auth_file"]).expanduser()
        profile_target = history_profile_target(data, target_slot)
        auth_target.parent.mkdir(parents=True, exist_ok=True)
        profile_target.mkdir(parents=True, exist_ok=True)

        # Before switching, flush the live AUTH_JSON content back to the
        # previous slot's auth_file, so any in-flight OAuth refresh tokens
        # written by codex CLI/App since the last switch aren't lost.
        previous = data.get("current")
        if previous and previous != target_slot:
            prev_cfg = data.get("slots", {}).get(previous) or {}
            prev_auth = Path(prev_cfg.get("auth_file", "")).expanduser()
            prev_mode = prev_cfg.get("mode", "oauth")
            # Only flush for oauth — relay's API-key JSON is static and rewritten below.
            if prev_mode == "oauth" and prev_auth and str(prev_auth):
                try:
                    flush_auth_to_slot(prev_auth)
                except OSError:
                    pass

        if mode == "relay":
            api_key = str(cfg.get("api_key") or "").strip()
            base_url = str(cfg.get("base_url") or "").strip().rstrip("/")
            if not api_key or not base_url:
                fail(f"relay slot {target_slot!r} is missing api_key/base_url")
            write_apikey_auth(auth_target, api_key)
            patch_config(mode="relay", model=cfg.get("model", DEFAULT_MODEL), relay_base_url=base_url)
        else:
            patch_config(mode="oauth", model=cfg.get("model", DEFAULT_MODEL))

        # Install target slot's auth as the live AUTH_JSON (real file).
        # Cases:
        #   1. auth_target exists → copy it onto AUTH_JSON.
        #   2. auth_target missing BUT current AUTH_JSON looks like a valid
        #      oauth token → promote it into auth_target (recover from a
        #      manually-deleted slot file without losing the live login).
        #   3. otherwise → clear AUTH_JSON so codex CLI sees ENOENT and
        #      prompts the user to `codex login` for this slot.
        if auth_target.exists():
            copy_auth_atomic(auth_target)
        else:
            cur = read_auth_bytes()
            recovered = False
            if cur and len(cur) >= 1000 and mode == "oauth" and recover_missing_oauth_auth:
                try:
                    parsed = json.loads(cur)
                    if parsed.get("auth_mode") != "apikey":
                        auth_target.parent.mkdir(parents=True, exist_ok=True)
                        auth_target.write_bytes(cur)
                        os.chmod(auth_target, 0o600)
                        # AUTH_JSON already holds the right content; no overwrite needed.
                        recovered = True
                        print(f"  ↳ 自当前 auth.json 恢复 oauth token 到 {auth_target}")
                except (ValueError, json.JSONDecodeError):
                    pass
            if not recovered:
                if AUTH_JSON.exists() or AUTH_JSON.is_symlink():
                    AUTH_JSON.unlink()
        atomic_symlink(APP_PROFILE, profile_target)
        session_stats = normalize_sessions_if_needed(slot_thread_provider(cfg), backup_dir=backup_dir)

        data["current"] = target_slot
        data.setdefault("app_history_slot", target_slot)
        save_slots(data)

        verb = "已修复当前渠道" if previous == target_slot else "已切换到"
        print(f"{verb}：{cfg.get('display_name', target_slot)} ({target_slot}) [{mode}]")
        if AUTH_JSON.exists():
            print(f"  auth.json   <- {auth_target}")
        else:
            print(f"  auth.json   已清空；等待 `codex login` 写入 {auth_target}")
        print(f"  App profile -> {profile_target}（历史保留）")
        if mode == "relay":
            print(f"  base_url    -> {cfg.get('base_url')}")
            print(f"  model       -> {cfg.get('model', DEFAULT_MODEL)}")
        elif not auth_target.exists() or auth_target.stat().st_size < 10:
            print(f"  尚未登录：请运行 `codex login` 写入 {auth_target}")
        print_session_summary(session_stats)
        if not no_restart:
            launch_codex()
            print("  Codex App 已重新打开")
        else:
            print("  --no-restart：Codex App 未重新打开")
        if not printed_rollback:
            print(f"  回滚命令: sub2cli-inject rollback {backup_dir.name}")
        return 0
    except BaseException:
        auto_rollback_after_failure(backup_dir, "切换")
        raise
    finally:
        lock_ctx.__exit__(None, None, None)


def cmd_relay(
    base_url: str,
    api_key: str,
    *,
    slot: str | None = None,
    model: str = DEFAULT_MODEL,
    display: str | None = None,
    skip_probe: bool = False,
    no_switch: bool = False,
    no_restart: bool = False,
    dry_run: bool = False,
) -> int:
    base_url = normalize_base_url(base_url)
    api_key = api_key.strip()
    if not api_key:
        fail("API key 不能为空")
    slot = validate_slot(slot) if slot else slot_from_url(base_url)

    if not skip_probe:
        ok, message = probe_relay(base_url, api_key)
        if not ok:
            print(f"中转检查失败：{message}", file=sys.stderr)
            print("渠道未保存。请换正确的 URL/key 后重试，或加 --skip-check 跳过检查。", file=sys.stderr)
            return 3
        print(f"检查通过：{message}")

    if dry_run:
        existing_data = load_slots(required=False)
        existing_slots = existing_data.get("slots", {}) or {}
        verb = "update" if slot in existing_slots else "add"
        print(f"DRY-RUN: 将 {verb} API 渠道")
        print(f"  slot:     {slot}")
        print(f"  base_url: {base_url}")
        print(f"  model:    {model}")
        if no_switch:
            print("  --no-switch: 不会切换")
            return 0
        # Simulate post-add state for the switch plan.
        simulated = dict(existing_data)
        simulated_slots = dict(existing_slots)
        simulated_slots[slot] = {
            "display_name": display or (existing_slots.get(slot) or {}).get("display_name") or f"Codex - {slot} relay",
            "mode": "relay",
            "auth_file": str(CODEX_HOME / f"auth.{slot}.json"),
            "app_profile_dir": str(APP_SUPPORT / f"Codex.{slot}"),
            "base_url": base_url,
            "model": model,
            "api_key": api_key,
        }
        simulated["slots"] = simulated_slots
        print()
        print_switch_plan(compute_switch_plan(simulated, slot))
        return 0

    with Sub2cliLock():
        backup_dir = new_backup_dir(f"add-api-{slot}")
        snapshot_pre_switch(backup_dir)
        print(f"  回滚命令: sub2cli-inject rollback {backup_dir.name}")
        try:
            ensure_initialized()

            data = load_slots()
            slots = data.setdefault("slots", {})
            existing = slots.get(slot)
            if existing and existing.get("mode") != "relay":
                fail(f"渠道 {slot!r} 已存在，但不是 API 渠道")

            slots[slot] = {
                "display_name": display or (existing or {}).get("display_name") or f"Codex - {slot} relay",
                "mode": "relay",
                "auth_file": str(CODEX_HOME / f"auth.{slot}.json"),
                "app_profile_dir": str(APP_SUPPORT / f"Codex.{slot}"),
                "base_url": base_url,
                "model": model,
                "api_key": api_key,
            }
            save_slots(data)
            print(f"{'已更新' if existing else '已添加'} API 渠道：{slot}")
            print(f"  base_url: {base_url}")
            print(f"  model:    {model}")

            if no_switch:
                print(f"  稍后切换：codex-provider use {slot}")
                print(f"  回滚命令: sub2cli-inject rollback {backup_dir.name}")
                return 0
            return cmd_switch(slot, no_restart=no_restart, backup_dir=backup_dir)
        except BaseException:
            auto_rollback_after_failure(backup_dir, "配置 API 渠道")
            raise


def read_api_key_secure(*, from_stdin: bool) -> str:
    if from_stdin or not sys.stdin.isatty():
        return sys.stdin.read().strip()
    if sys.stdout.isatty():
        return getpass.getpass("API key（隐藏输入）: ").strip()
    return ""


def reject_positional_api_key(argv: list[str]) -> None:
    """Reject legacy `add-api <url> <key>` before argparse can echo the key."""
    value_options = {"--name", "--slot", "--model", "--display"}
    positionals = 0
    i = 0
    while i < len(argv):
        arg = argv[i]
        if arg == "--":
            positionals += max(0, len(argv) - i - 1)
            break
        if arg.startswith("--"):
            name = arg.split("=", 1)[0]
            if name in value_options and "=" not in arg:
                i += 2
                continue
            i += 1
            continue
        positionals += 1
        if positionals >= 2:
            fail("出于安全原因，API key 不能作为命令行参数传入；请使用 --api-key-stdin 或交互式隐藏输入")
        i += 1


def cmd_oauth(
    slot: str,
    *,
    display: str | None = None,
    auth_file: str | None = None,
    no_switch: bool = False,
    no_restart: bool = False,
    dry_run: bool = False,
) -> int:
    slot = validate_slot(slot)
    auth_source = Path(auth_file).expanduser() if auth_file else None
    if auth_source and not auth_source.exists():
        fail(f"auth file not found: {auth_source}")
    if auth_source:
        try:
            parsed = json.loads(auth_source.read_text())
        except Exception as exc:
            fail(f"auth file is not valid JSON: {exc}")
        if parsed.get("auth_mode") == "apikey":
            fail(f"auth file is API-key auth, not a Codex OAuth account: {auth_source}")
        if not decode_auth_email(auth_source):
            fail(f"auth file does not contain a readable Codex account email: {auth_source}")

    if dry_run:
        existing_data = load_slots(required=False)
        existing_slots = existing_data.get("slots", {}) or {}
        verb = "update" if slot in existing_slots else "add"
        print(f"DRY-RUN: 将 {verb} 账号渠道")
        print(f"  slot: {slot}")
        if no_switch:
            print("  --no-switch: 不会切换")
            return 0
        simulated = dict(existing_data)
        simulated_slots = dict(existing_slots)
        simulated_slots[slot] = {
            "display_name": display or (existing_slots.get(slot) or {}).get("display_name") or f"Codex - {slot}",
            "mode": "oauth",
            "auth_file": str(CODEX_HOME / f"auth.{slot}.json"),
            "app_profile_dir": str(APP_SUPPORT / f"Codex.{slot}"),
        }
        simulated["slots"] = simulated_slots
        print()
        print_switch_plan(compute_switch_plan(simulated, slot))
        return 0

    with Sub2cliLock():
        backup_dir = new_backup_dir(f"add-account-{slot}")
        snapshot_pre_switch(backup_dir)
        print(f"  回滚命令: sub2cli-inject rollback {backup_dir.name}")
        try:
            ensure_initialized()

            data = load_slots()
            slots = data.setdefault("slots", {})
            existing = slots.get(slot)
            if existing and existing.get("mode") != "oauth":
                fail(f"渠道 {slot!r} 已存在，但不是账号渠道")
            auth_target = CODEX_HOME / f"auth.{slot}.json"
            if auth_source:
                existing_email = decode_auth_email(Path(existing.get("auth_file", "")).expanduser()) if existing else None
                source_email = decode_auth_email(auth_source)
                if existing_email and source_email and existing_email.lower() != source_email.lower():
                    fail(f"渠道 {slot!r} 已属于 {existing_email}, 不能覆盖为 {source_email}")
                if auth_source.resolve(strict=False) != auth_target.resolve(strict=False):
                    auth_target.parent.mkdir(parents=True, exist_ok=True)
                    tmp = auth_target.with_name(f"{auth_target.name}.{os.getpid()}.{time.time_ns()}.tmp")
                    try:
                        tmp.write_bytes(auth_source.read_bytes())
                        os.chmod(tmp, 0o600)
                        os.replace(tmp, auth_target)
                    except Exception:
                        if tmp.exists() or tmp.is_symlink():
                            try:
                                tmp.unlink()
                            except OSError:
                                pass
                        raise

            slots[slot] = {
                "display_name": display or (existing or {}).get("display_name") or f"Codex - {slot}",
                "mode": "oauth",
                "auth_file": str(auth_target),
                "app_profile_dir": str(APP_SUPPORT / f"Codex.{slot}"),
            }
            save_slots(data)
            print(f"{'已更新' if existing else '已添加'}账号渠道：{slot}")
            if no_switch:
                print(f"  稍后切换：codex-provider use {slot}")
                print("  切过去后运行：codex login")
                print(f"  回滚命令: sub2cli-inject rollback {backup_dir.name}")
                return 0
            rc = cmd_switch(
                slot,
                no_restart=no_restart,
                backup_dir=backup_dir,
                recover_missing_oauth_auth=auth_source is not None,
            )
            auth_target = Path(slots[slot]["auth_file"]).expanduser()
            if not auth_target.exists() or auth_target.stat().st_size < 10:
                print("  下一步：运行 `codex login`，登录这个账号")
            return rc
        except BaseException:
            auto_rollback_after_failure(backup_dir, "配置账号渠道")
            raise


def cmd_list() -> int:
    data = load_slots(required=False)
    current = data.get("current")
    slots = data.get("slots", {})
    if not slots:
        print("还没有渠道")
        print("  添加 API：  codex-provider add-api https://relay.example.com sk-...")
        print("  添加账号： codex-provider add-account work")
        return 0
    for key, cfg in slots.items():
        marker = "*" if key == current else " "
        mode = cfg.get("mode", "oauth")
        label = cfg.get("display_name", key)
        extra = cfg.get("base_url", "") if mode == "relay" else "(账号登录)"
        print(f"{marker} {key:<16} [{mode:<5}] {label:<32} {extra}")
    return 0


def cmd_status() -> int:
    data = load_slots(required=False)
    current = data.get("current")
    slots = data.get("slots", {})
    if not current or current not in slots:
        print("当前没有渠道")
        print("  运行：codex-provider")
        return 0
    cfg = slots[current]
    provider = slot_thread_provider(cfg)
    print(f"当前渠道：{cfg.get('display_name', current)} ({current}) [{cfg.get('mode', 'oauth')}]")
    print(f"  渠道文件:    {SLOTS_FILE}")
    if AUTH_JSON.is_symlink():
        # Legacy: leftover symlink from pre-schema-2 install.
        print(f"  auth.json:   -> {os.readlink(AUTH_JSON)} (legacy symlink)")
    elif AUTH_JSON.exists():
        cur_auth_file = Path(cfg.get("auth_file", "")).expanduser()
        match = "matches" if auth_content_equals(cur_auth_file) else "diverged"
        print(f"  auth.json:   {AUTH_JSON} ({AUTH_JSON.stat().st_size} bytes, {match} {cur_auth_file.name})")
    else:
        print(f"  auth.json:   {AUTH_JSON} (missing)")
    print(f"  App profile: {'-> ' + os.readlink(APP_PROFILE) if APP_PROFILE.is_symlink() else str(APP_PROFILE)}")
    print(f"  App 运行中:  {'是' if codex_running() else '否'}")
    counts = thread_provider_counts()
    if counts:
        summary = ", ".join(f"{key}:{value}" for key, value in sorted(counts.items()))
        print(f"  sessions:    {summary}")
    preview = normalize_sessions(provider, dry_run=True)
    print_session_summary(preview)
    if not slot_is_clean(data, current):
        print(f"  状态不一致：运行 `codex-provider use {current}` 修复")
    return 0


def cmd_remove(slot: str, *, hard: bool = False) -> int:
    if Sub2cliLock._depth == 0:
        try:
            with Sub2cliLock():
                return cmd_remove(slot, hard=hard)
        except LockBusy as exc:
            fail(str(exc), code=4)
    data = load_slots()
    slot = resolve_slot(slot, data.get("slots", {}))
    if slot == data.get("current"):
        fail(f"{slot!r} is current; switch away before removing it")
    cfg = data["slots"].pop(slot)
    save_slots(data)
    print(f"removed slot: {slot}")
    if hard:
        auth_path = Path(cfg.get("auth_file", "")).expanduser()
        profile_path = Path(cfg.get("app_profile_dir", "")).expanduser()
        if auth_path.is_file() or auth_path.is_symlink():
            if CODEX_HOME.resolve(strict=False) in auth_path.resolve(strict=False).parents:
                auth_path.unlink()
                print(f"  removed {auth_path}")
        if profile_path.exists() and not profile_path.is_symlink():
            if APP_SUPPORT.resolve(strict=False) in profile_path.resolve(strict=False).parents:
                import shutil

                shutil.rmtree(profile_path)
                print(f"  removed {profile_path}")
    return 0


def cmd_normalize_sessions(target: str | None, *, dry_run: bool) -> int:
    if not dry_run and Sub2cliLock._depth == 0:
        try:
            with Sub2cliLock():
                return cmd_normalize_sessions(target, dry_run=dry_run)
        except LockBusy as exc:
            fail(str(exc), code=4)
    if target in (None, "", "current"):
        data = load_slots()
        current = data.get("current")
        if not current:
            fail("当前没有渠道")
        target_provider = slot_thread_provider(data["slots"][current])
    elif target in THREAD_PROVIDERS:
        target_provider = target
    else:
        data = load_slots()
        slot = resolve_slot(target, data.get("slots", {}))
        target_provider = slot_thread_provider(data["slots"][slot])
    stats = normalize_sessions(target_provider, dry_run=dry_run)
    print_session_summary(stats)
    return 1 if stats.get("rollouts", {}).get("errors") else 0


def cmd_rollback(target: str | None = None, *, list_only: bool = False) -> int:
    """Restore the most recent (or named) backup. --list to show available ones."""
    if not BACKUP_ROOT.exists():
        if list_only:
            print(f"(no backups: {BACKUP_ROOT} doesn't exist)")
            return 0
        fail(f"备份目录不存在: {BACKUP_ROOT}")
    backups = sorted(
        [p for p in BACKUP_ROOT.iterdir() if p.is_dir()],
        key=lambda p: p.name,
        reverse=True,
    )
    if list_only:
        if not backups:
            print("(no backups)")
            return 0
        print(f"备份 ({len(backups)} 个, 最新在上):")
        for p in backups[:20]:
            snap = p / SNAPSHOT_FILENAME
            tag = "snapshot" if snap.exists() else "no-snapshot"
            print(f"  {p.name}  [{tag}]")
        if len(backups) > 20:
            print(f"  ... 还有 {len(backups) - 20} 个")
        return 0
    if not backups:
        fail("没有任何备份可恢复")
    if target in (None, "", "latest"):
        chosen = backups[0]
    else:
        exact = [p for p in backups if p.name == target]
        if exact:
            chosen = exact[0]
        else:
            partial = [p for p in backups if target in p.name]
            if len(partial) == 1:
                chosen = partial[0]
            elif not partial:
                head = ", ".join(p.name for p in backups[:5])
                fail(f"未找到匹配的备份: {target}\n  候选: {head}")
                return 2
            else:
                head = ", ".join(p.name for p in partial)
                fail(f"备份名歧义: {head}")
                return 2
    print(f"将从备份恢复: {chosen.name}")
    try:
        lock_ctx = Sub2cliLock()
        lock_ctx.__enter__()
    except LockBusy as exc:
        fail(str(exc), code=4)
        return 4
    try:
        return restore_from_backup(chosen)
    finally:
        lock_ctx.__exit__(None, None, None)


def clear_screen() -> None:
    if sys.stdout.isatty():
        print("\033[2J\033[H", end="")


def pause() -> None:
    if sys.stdin.isatty():
        input("\n按 Enter 继续...")


def read_key() -> str:
    fd = sys.stdin.fileno()
    old = termios.tcgetattr(fd)
    try:
        tty.setraw(fd)
        ch = os.read(fd, 1)
        if ch == b"\x1b":
            key = ch
            while select.select([sys.stdin], [], [], 0.02)[0]:
                key += os.read(fd, 1)
            if key in (b"\x1b[A", b"\x1b[D"):
                return "up"
            if key in (b"\x1b[B", b"\x1b[C"):
                return "down"
            return "back"
        if ch in (b"\r", b"\n"):
            return "enter"
        if ch in (b"q", b"Q", b"\x03"):
            return "quit"
        if ch in (b"b", b"B", b"\x7f"):
            return "back"
        if ch in (b"v", b"V"):
            return "version"
        if ch in (b"m", b"M"):
            return "more"
    finally:
        termios.tcsetattr(fd, termios.TCSADRAIN, old)
    return ""


def select_menu(
    title: str,
    options: list[tuple[str, str]],
    *,
    subtitle: str | None = None,
    branded: bool = False,
) -> str:
    if not options:
        return "back"
    if not (sys.stdin.isatty() and sys.stdout.isatty()):
        print(title)
        if subtitle:
            print(subtitle)
        for idx, (label, _) in enumerate(options, start=1):
            print(f"{idx}. {label}")
        choice = input("> ").strip()
        if not choice:
            return "back"
        try:
            return options[int(choice) - 1][1]
        except (ValueError, IndexError):
            return "back"

    selected = 0
    while True:
        clear_screen()
        if branded:
            print(APP_BANNER)
            print(f"  {APP_URL}")
            print(f"  {title}")
            if subtitle:
                print(f"  {subtitle}")
            print("")
        else:
            print(title)
            if subtitle:
                print(subtitle)
            print("")
        for idx, (label, _) in enumerate(options):
            marker = "➤" if idx == selected else " "
            print(f"{marker} {idx + 1}. {label}")
        if branded:
            print("\n↑↓  |  Enter 确认  |  Q 退出")
        else:
            print("\n↑↓  |  Enter 确认  |  B 返回  |  Q 退出")
        key = read_key()
        if key == "up":
            selected = (selected - 1) % len(options)
        elif key == "down":
            selected = (selected + 1) % len(options)
        elif key == "enter":
            return options[selected][1]
        elif key in ("back", "quit"):
            return key


def prompt_line(label: str, *, default: str | None = None, required: bool = False) -> str:
    suffix = f" [{default}]" if default is not None else ""
    while True:
        value = input(f"{label}{suffix}: ").strip()
        if value:
            return value
        if default is not None:
            return default
        if not required:
            return ""
        print("必填。")


def yes_no(label: str, *, default: bool = True) -> bool:
    marker = "Y/n" if default else "y/N"
    value = input(f"{label} [{marker}]: ").strip().lower()
    if not value:
        return default
    return value in ("y", "yes")


def menu_status() -> None:
    clear_screen()
    cmd_status()
    pause()


def menu_init() -> None:
    clear_screen()
    print("初始化当前 Codex 配置\n")
    slot = prompt_line("渠道名", default="local", required=True)
    display = prompt_line("显示名称", default="")
    clear_screen()
    cmd_init(slot=slot, display=display or None)
    pause()


def menu_add_relay() -> None:
    clear_screen()
    print("新增 API 渠道\n")
    base_url = prompt_line("中转 base URL，不要带 /v1", required=True)
    api_key = getpass.getpass("API key（隐藏输入）: ").strip()
    slot = prompt_line("渠道名（空着则自动从域名取）", default="")
    switch_now = yes_no("现在切到这个渠道", default=True)
    clear_screen()
    cmd_relay(
        base_url,
        api_key,
        slot=slot or None,
        no_switch=not switch_now,
    )
    pause()


def menu_add_oauth() -> None:
    clear_screen()
    print("新增账号渠道\n")
    slot = prompt_line("渠道名，例如 personal 或 work", required=True)
    switch_now = yes_no("现在切到这个账号", default=True)
    clear_screen()
    cmd_oauth(slot, no_switch=not switch_now)
    pause()


def menu_switch() -> None:
    data = load_slots(required=False)
    slots = data.get("slots", {})
    current = data.get("current")
    if not slots:
        clear_screen()
        print("还没有渠道。请先新增账号或 API。")
        pause()
        return
    options = []
    for key, cfg in slots.items():
        marker = "*" if key == current else " "
        mode = cfg.get("mode", "oauth")
        label = cfg.get("display_name", key)
        extra = cfg.get("base_url", "账号登录") if mode == "relay" else "账号登录"
        options.append((f"{marker} {key:<16} [{mode}] {label}  {extra}", key))
    options.append(("← 返回", "back"))
    choice = select_menu("切换渠道", options)
    if choice in ("back", "quit"):
        return
    clear_screen()
    cmd_switch(choice)
    pause()


def menu_remove() -> None:
    data = load_slots(required=False)
    slots = data.get("slots", {})
    current = data.get("current")
    removable = [(key, cfg) for key, cfg in slots.items() if key != current]
    if not removable:
        clear_screen()
        print("没有可删除的渠道。当前渠道不能删除。")
        pause()
        return
    options = [
        (f"{key:<16} [{cfg.get('mode', 'oauth')}] {cfg.get('display_name', key)}", key)
        for key, cfg in removable
    ]
    options.append(("← 返回", "back"))
    choice = select_menu("删除渠道", options)
    if choice in ("back", "quit"):
        return
    if not yes_no(f"从列表中删除渠道 {choice!r}", default=False):
        return
    clear_screen()
    cmd_remove(choice)
    pause()


def menu_normalize() -> None:
    clear_screen()
    print("Session normalize dry-run\n")
    cmd_normalize_sessions("current", dry_run=True)
    pause()


def cmd_version() -> int:
    print(f"codex-provider {APP_VERSION}")
    print(APP_URL)
    return 0


def cmd_more() -> int:
    data = load_slots(required=False)
    print("Codex Provider 路径")
    print(f"  渠道文件:    {SLOTS_FILE}")
    print(f"  Codex 目录:  {CODEX_HOME}")
    print(f"  配置:        {CONFIG_TOML}")
    print(f"  认证:        {AUTH_JSON}")
    print(f"  App profile: {APP_PROFILE}")
    print(f"  备份:        {BACKUP_ROOT}")
    if data.get("slots"):
        print("\n渠道")
        for key, cfg in data["slots"].items():
            marker = "*" if key == data.get("current") else " "
            print(f" {marker} {key:<16} {cfg.get('mode', 'oauth'):<5} {cfg.get('display_name', key)}")
    return 0


def menu_version() -> None:
    clear_screen()
    cmd_version()
    pause()


def menu_more() -> None:
    clear_screen()
    cmd_more()
    pause()


def cmd_menu() -> int:
    while True:
        data = load_slots(required=False)
        current = data.get("current")
        slots = data.get("slots", {})
        if current in slots:
            cfg = slots[current]
            current_label = f"{cfg.get('display_name', current)} ({current}) [{cfg.get('mode', 'oauth')}]"
        else:
            current_label = "未初始化"
        choice = select_menu(
            "Codex 渠道切换",
            [
                ("切换渠道      选择已保存的渠道", "switch"),
                ("新增 API      添加第三方 API key", "add-relay"),
                ("新增账号      添加 Codex 登录账号", "add-oauth"),
                ("当前状态      查看正在使用的渠道", "status"),
                ("退出          关闭工具", "quit"),
            ],
            subtitle=f"当前：{current_label}",
            branded=True,
        )
        if choice in ("quit", "back"):
            clear_screen()
            return 0
        if choice == "status":
            menu_status()
        elif choice == "add-relay":
            menu_add_relay()
        elif choice == "add-oauth":
            menu_add_oauth()
        elif choice == "switch":
            menu_switch()


def run_internal_command(command: str, argv: list[str]) -> int:
    parser = argparse.ArgumentParser(prog=f"codex-provider {command}")

    if command == "menu":
        parser.parse_args(argv)
        return cmd_menu()

    if command == "version":
        parser.parse_args(argv)
        return cmd_version()

    if command == "init":
        parser.add_argument("--slot", default="local")
        parser.add_argument("--display")
        args = parser.parse_args(argv)
        return cmd_init(slot=args.slot, display=args.display)

    if command == "more":
        parser.parse_args(argv)
        return cmd_more()

    if command == "status":
        parser.parse_args(argv)
        return cmd_status()

    if command == "ls":
        parser.parse_args(argv)
        return cmd_list()

    if command == "relay":
        reject_positional_api_key(argv)
        parser.add_argument("base_url")
        parser.add_argument("--api-key-stdin", action="store_true")
        parser.add_argument("--slot")
        parser.add_argument("--model", default=DEFAULT_MODEL)
        parser.add_argument("--display")
        parser.add_argument("--skip-probe", action="store_true")
        parser.add_argument("--no-switch", action="store_true")
        parser.add_argument("--no-restart", action="store_true")
        args = parser.parse_args(argv)
        api_key = read_api_key_secure(from_stdin=args.api_key_stdin)
        if not api_key:
            fail("API key 不能为空；请用 --api-key-stdin 从 stdin 传入，或在交互终端隐藏输入")
        return cmd_relay(
            args.base_url,
            api_key,
            slot=args.slot,
            model=args.model,
            display=args.display,
            skip_probe=args.skip_probe,
            no_switch=args.no_switch,
            no_restart=args.no_restart,
        )

    if command == "oauth":
        parser.add_argument("slot")
        parser.add_argument("--display")
        parser.add_argument("--auth-file")
        parser.add_argument("--no-switch", action="store_true")
        parser.add_argument("--no-restart", action="store_true")
        args = parser.parse_args(argv)
        return cmd_oauth(
            args.slot,
            display=args.display,
            auth_file=args.auth_file,
            no_switch=args.no_switch,
            no_restart=args.no_restart,
        )

    if command == "switch":
        parser.add_argument("slot")
        parser.add_argument("--no-restart", action="store_true")
        args = parser.parse_args(argv)
        return cmd_switch(args.slot, no_restart=args.no_restart)

    if command == "remove":
        parser.add_argument("slot")
        parser.add_argument("--hard", action="store_true")
        args = parser.parse_args(argv)
        return cmd_remove(args.slot, hard=args.hard)

    if command == "normalize-sessions":
        parser.add_argument("target", nargs="?", default="current")
        parser.add_argument("--dry-run", action="store_true")
        args = parser.parse_args(argv)
        return cmd_normalize_sessions(args.target, dry_run=args.dry_run)

    if command == "rollback":
        parser.add_argument("target", nargs="?", default=None,
                            help="备份名 (latest 或具体目录名), 不传=latest")
        parser.add_argument("--list", dest="list_only", action="store_true",
                            help="只列出可用备份")
        args = parser.parse_args(argv)
        return cmd_rollback(args.target, list_only=args.list_only)

    fail(f"unknown internal command: {command}")
    return 1


def main(argv: list[str] | None = None) -> int:
    argv = list(sys.argv[1:] if argv is None else argv)
    internal_commands = {
        "menu",
        "version",
        "init",
        "more",
        "status",
        "ls",
        "relay",
        "oauth",
        "switch",
        "remove",
        "normalize-sessions",
        "rollback",
    }
    if argv and argv[0] in internal_commands:
        return run_internal_command(argv[0], argv[1:])
    if argv and argv[0] == "add-api":
        reject_positional_api_key(argv[1:])

    parser = argparse.ArgumentParser(
        prog="codex-provider",
        description="在已保存的 Codex 账号和 API 渠道之间切换。",
        formatter_class=ChineseHelpFormatter,
        add_help=False,
    )
    parser.add_argument("-h", "--help", action="help", help="显示帮助并退出")
    sub = parser.add_subparsers(dest="command", metavar="command")

    p_add_api = sub.add_parser("add-api", help="添加 API key 渠道，并切过去")
    p_add_api.add_argument("base_url")
    p_add_api.add_argument("--api-key-stdin", action="store_true")
    p_add_api.add_argument("--name", "--slot", dest="slot")
    p_add_api.add_argument("--model", default=DEFAULT_MODEL)
    p_add_api.add_argument("--display")
    p_add_api.add_argument("--skip-check", "--skip-probe", dest="skip_probe", action="store_true")
    p_add_api.add_argument("--no-switch", action="store_true")
    p_add_api.add_argument("--no-restart", action="store_true")
    p_add_api.add_argument("--dry-run", action="store_true",
                           help="只打印计划改动, 不写入")

    p_add_account = sub.add_parser("add-account", help="添加 Codex 登录账号，并切过去")
    p_add_account.add_argument("name")
    p_add_account.add_argument("--display")
    p_add_account.add_argument("--auth-file",
                               help="从已有 Codex OAuth auth.json 导入账号登录文件")
    p_add_account.add_argument("--no-switch", action="store_true")
    p_add_account.add_argument("--no-restart", action="store_true")
    p_add_account.add_argument("--dry-run", action="store_true",
                               help="只打印计划改动, 不写入")

    p_use = sub.add_parser("use", help="切换到已保存的渠道")
    p_use.add_argument("name")
    p_use.add_argument("--no-restart", action="store_true")
    p_use.add_argument("--dry-run", action="store_true",
                       help="只打印计划改动, 不写入")

    p_rollback = sub.add_parser("rollback", help="从备份恢复 ~/.codex 状态")
    p_rollback.add_argument("target", nargs="?", default=None,
                            help="备份名 (latest 或具体目录名), 不传=latest")
    p_rollback.add_argument("--list", dest="list_only", action="store_true",
                            help="只列出可用备份")

    sub.add_parser("list", help="列出已保存渠道")
    sub.add_parser("current", help="查看当前渠道")

    args = parser.parse_args(argv)
    if args.command is None:
        if sys.stdin.isatty() and sys.stdout.isatty():
            return cmd_menu()
        return cmd_status()
    if args.command == "current":
        return cmd_status()
    if args.command == "list":
        return cmd_list()
    if args.command == "add-api":
        api_key = read_api_key_secure(from_stdin=args.api_key_stdin)
        if not api_key:
            fail("API key 不能为空；请用 --api-key-stdin 从 stdin 传入，或在交互终端隐藏输入")
        return cmd_relay(
            args.base_url,
            api_key,
            slot=args.slot,
            model=args.model,
            display=args.display,
            skip_probe=args.skip_probe,
            no_switch=args.no_switch,
            no_restart=args.no_restart,
            dry_run=args.dry_run,
        )
    if args.command == "add-account":
        return cmd_oauth(
            args.name,
            display=args.display,
            auth_file=args.auth_file,
            no_switch=args.no_switch,
            no_restart=args.no_restart,
            dry_run=args.dry_run,
        )
    if args.command == "use":
        return cmd_switch(args.name, no_restart=args.no_restart, dry_run=args.dry_run)
    if args.command == "rollback":
        return cmd_rollback(args.target, list_only=args.list_only)
    parser.print_help()
    return 1


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except KeyboardInterrupt:
        print("已取消", file=sys.stderr)
        raise SystemExit(130)
    except SystemExit:
        raise
    except Exception as exc:
        print(f"错误：{type(exc).__name__}: {exc}", file=sys.stderr)
        raise SystemExit(1)