codex_chat_bridge.py

from __future__ import annotations

import csv
import json
import os
import re
import shutil
import sqlite3
import subprocess
import sys
import time
import tomllib
import uuid
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Iterable
from urllib.parse import urlparse

import tkinter as tk
from tkinter import messagebox, ttk


APP_NAME = "Codex Chat Bridge"
CHUNK_SIZE = 500
DEFAULT_LANGUAGE = "zh_CN"
LANGUAGE_LABELS = {
    "zh_CN": "中文",
    "en_US": "English",
}
RUNTIME_ERROR_KEYS = {
    "This bridge session is not open anymore.": "error.runtime.bridge_session_not_open",
    "Provider name cannot be empty.": "error.runtime.provider_name_empty",
    "Base URL cannot be empty.": "error.runtime.base_url_empty",
    "Base URL must start with http:// or https:// and include a host.": "error.runtime.base_url_invalid",
}
I18N = {
    "zh_CN": {
        "app_title": "Codex 聊天迁移器",
        "label.refresh": "刷新",
        "label.language": "语言",
        "label.codex_home": "Codex 目录：",
        "label.current_provider": "当前 Provider：",
        "label.codex_process": "Codex 进程：",
        "tab.chat_bridge": "聊天迁移",
        "tab.providers": "Provider 管理",
        "action.backup_full": "完整备份 .codex",
        "action.open_session_folder": "打开会话目录",
        "action.open_backup_folder": "打开备份目录",
        "section.log": "日志",
        "section.saved_configured_providers": "已保存与已配置的 Providers",
        "section.provider_details": "Provider 详情",
        "section.providers": "Providers",
        "section.start_bridge_session": "开始桥接会话",
        "section.direct_move": "直接迁移",
        "section.bridge_sessions": "桥接会话",
        "section.return_session": "回迁会话",
        "field.provider": "Provider",
        "field.base_url": "Base URL",
        "field.api_key": "API Key",
        "field.saved": "已保存",
        "field.config": "配置中",
        "field.current": "当前",
        "field.threads": "线程数",
        "field.provider_name": "Provider 名称",
        "field.in_config": "在配置中",
        "field.source_provider": "源 Provider",
        "field.target_provider": "目标 Provider",
        "field.session": "会话",
        "field.source": "源",
        "field.target": "目标",
        "field.status": "状态",
        "field.moved": "已迁移",
        "field.selected_session": "选中的会话",
        "checkbox.show_api_key": "显示 API Key",
        "checkbox.switch_current_to_target_after_start": "开始后把当前 Provider 切到目标",
        "checkbox.switch_current_to_target_after_move": "迁移后把当前 Provider 切到目标",
        "checkbox.move_new_target_threads": "把会话开始后在目标下新建的线程一起迁回",
        "checkbox.switch_current_back_after_return": "回迁后把当前 Provider 切回源",
        "button.save_update": "保存 / 更新",
        "button.save_switch": "保存并切换",
        "button.switch_form": "使用表单切换",
        "button.delete_saved_entry": "删除已保存项",
        "button.open_provider_store": "打开 Provider 存储",
        "button.load_current_provider": "载入当前 Provider",
        "button.start_session": "开始会话",
        "button.move_all_source_threads": "迁移源 Provider 全部线程",
        "button.return_selected_session": "回迁所选会话",
        "status.ready": "就绪",
        "status.checking_process": "正在检查 Codex 进程...",
        "status.refreshed": "已刷新",
        "status.app_started": "程序已启动",
        "misc.none": "未设置",
        "misc.no_process_detected": "未检测到 Codex 进程",
        "misc.yes": "是",
        "misc.no": "否",
        "misc.runtime_suffix": "运行中",
        "session_status.open": "进行中",
        "session_status.completed": "已完成",
        "note.provider_default": "已保存的 Provider 存在 provider_store.json 中。删除保存项不会删除聊天线程。",
        "note.provider_runtime_key": "这个 Key 当前来自活动 Provider 的 auth.json。若想以后快速切换时继续使用，点击“保存 / 更新”即可保存到本地。",
        "note.provider_saved_entry": "这个 Provider 已有本地保存项。删除它只会删除保存项，不会删除聊天历史，也不会删掉现有 config 块。",
        "note.provider_config_only": "这个 Provider 当前只存在于 Codex 配置中。如果你想记住它的 URL 和 Key 方便快速切换，可以在这里保存。",
        "note.bridge_hint": "说明：Codex 的聊天按 model_provider 分线。\n这个工具做的是“迁移/回迁”，不是物理复制两份线程。",
        "action_label.running_migration": "迁移",
        "action_label.switching_providers": "切换 Provider",
        "error.open_path": "无法打开路径：\n{path}\n\n{error}",
        "warn.codex_active": "Codex 似乎仍在运行：\n\n{processes}\n\n请先完全关闭 Codex Desktop，再执行“{action_label}”。",
        "error.missing_required_files": "缺少必要文件：\n\n{files}",
        "error.sqlite_not_ready": "SQLite 当前无法进行写入操作。\n\n{error}",
        "error.missing_required_config": "缺少必要配置：\n\n{path}",
        "error.enter_provider_and_url": "请同时填写 Provider 名称和 Base URL。",
        "error.save_provider_failed": "保存 Provider 失败。\n\n{error}",
        "error.choose_or_enter_provider": "请先选择或输入一个 Provider。",
        "error.enter_url_before_switch": "切换前请先填写 Base URL。",
        "error.switch_provider_failed": "切换 Provider 失败。\n\n{error}",
        "error.choose_provider_first": "请先选择一个 Provider。",
        "error.provider_has_no_saved_entry": "这个 Provider 没有本地保存项。\n\n当前版本只删除保存项，不会删除 config 块。",
        "error.saved_provider_missing": "找不到要删除的已保存项。",
        "error.full_backup_failed": "完整备份失败。\n\n{error}",
        "error.choose_source_target": "请选择源 Provider 和目标 Provider。",
        "error.source_target_different": "源 Provider 和目标 Provider 不能相同。",
        "error.start_bridge_session_failed": "开始桥接会话失败。\n\n{error}",
        "error.select_bridge_session_first": "请先选择一个桥接会话。",
        "error.bridge_session_completed": "这个桥接会话已经结束了。",
        "error.return_bridge_session_failed": "回迁桥接会话失败。\n\n{error}",
        "error.direct_move_failed": "迁移线程失败。\n\n{error}",
        "error.runtime.bridge_session_not_open": "这个桥接会话已经不能再回迁了。",
        "error.runtime.provider_name_empty": "Provider 名称不能为空。",
        "error.runtime.base_url_empty": "Base URL 不能为空。",
        "error.runtime.base_url_invalid": "Base URL 必须以 http:// 或 https:// 开头，并且要包含主机名。",
        "confirm.no_api_key_continue": "表单里没有 API Key。\n\n程序会更新 Provider 配置和当前 Provider，但不会改动 auth.json。\n\n要继续吗？",
        "confirm.switch_provider": "确认把当前 Provider 切换为“{provider}”？\n\nBase URL: {base_url}\n\n程序会先备份 config/auth/provider store。",
        "confirm.delete_saved_provider": "确认删除“{provider}”的已保存项？\n\n这不会删除聊天线程，也不会从 config.toml 移除现有的 Provider 配置块。",
        "confirm.full_backup_while_active": "Codex 似乎仍在运行。\n\n完整备份仍然可以执行，但复制过程中相关文件可能发生变化。\n\n仍要继续吗？",
        "confirm.start_bridge_session": "要开始桥接会话吗？\n\n当前属于“{source}”的全部线程会被迁移到“{target}”。\n程序会先创建关键备份。",
        "confirm.return_bridge_session": "要回迁这个桥接会话吗？\n\n记录在“{target}”下的线程会被迁回“{source}”。\n程序会先创建关键备份。",
        "confirm.direct_move": "要把“{source}”下的全部线程迁移到“{target}”吗？\n\n程序会先创建关键备份。",
        "success.saved_provider_entry": "已保存 Provider：\n{provider}\n\n当前 Provider 未切换。若要落盘切换，请点“保存并切换”或“使用表单切换”。",
        "success.provider_switched": "Provider 已切换并验证成功。\n\nProvider: {provider}\nAuth 已更新: {auth_updated}\n配置文件: {config_path}\n备份目录: {backup_dir}",
        "success.full_backup_created": "已创建完整备份：\n{target}",
        "success.bridge_session_started": "桥接会话已开始。\n\n会话: {session_id}\n已迁移线程: {moved_thread_count}",
        "success.bridge_session_returned": "桥接会话已回迁。\n\n恢复线程: {restored_thread_count}\n回迁新增线程: {new_thread_count}",
        "success.direct_move_finished": "直接迁移完成。\n\n已迁移线程: {changed}\n备份目录: {backup_dir}",
        "log.saved_provider": "已保存 Provider {provider}（{api_key_mode} API Key）",
        "log.switched_provider": "已切换到 Provider {provider}（{auth_mode}）",
        "log.deleted_saved_provider": "已删除 Provider {provider} 的本地保存项",
        "log.full_backup_created": "已创建完整备份：{target}",
        "log.started_bridge_session": "已开始桥接会话 {session_id}：迁移 {moved_thread_count} 个线程 {source} -> {target}",
        "log.completed_bridge_session": "已完成桥接会话 {session_id}：恢复 {restored_thread_count} 个线程，并回迁 {new_thread_count} 个新线程",
        "log.direct_move_finished": "直接迁移完成：已迁移 {changed} 个线程 {source} -> {target}",
        "phrase.with_api_key": "含",
        "phrase.without_api_key": "不含",
        "phrase.updated_auth": "已更新 auth",
        "phrase.config_only": "仅配置",
    },
    "en_US": {
        "app_title": "Codex Chat Bridge",
        "label.refresh": "Refresh",
        "label.language": "Language",
        "label.codex_home": "Codex Home:",
        "label.current_provider": "Current Provider:",
        "label.codex_process": "Codex Process:",
        "tab.chat_bridge": "Chat Bridge",
        "tab.providers": "Providers",
        "action.backup_full": "Backup Full .codex",
        "action.open_session_folder": "Open Session Folder",
        "action.open_backup_folder": "Open Backup Folder",
        "section.log": "Log",
        "section.saved_configured_providers": "Saved and Configured Providers",
        "section.provider_details": "Provider Details",
        "section.providers": "Providers",
        "section.start_bridge_session": "Start Bridge Session",
        "section.direct_move": "Direct Move",
        "section.bridge_sessions": "Bridge Sessions",
        "section.return_session": "Return Session",
        "field.provider": "Provider",
        "field.base_url": "Base URL",
        "field.api_key": "API Key",
        "field.saved": "Saved",
        "field.config": "Config",
        "field.current": "Current",
        "field.threads": "Threads",
        "field.provider_name": "Provider Name",
        "field.in_config": "In Config",
        "field.source_provider": "Source Provider",
        "field.target_provider": "Target Provider",
        "field.session": "Session",
        "field.source": "Source",
        "field.target": "Target",
        "field.status": "Status",
        "field.moved": "Moved",
        "field.selected_session": "Selected Session",
        "checkbox.show_api_key": "Show API key",
        "checkbox.switch_current_to_target_after_start": "Switch current provider to target after start",
        "checkbox.switch_current_to_target_after_move": "Switch current provider to target after move",
        "checkbox.move_new_target_threads": "Move new target threads created after the session started",
        "checkbox.switch_current_back_after_return": "Switch current provider back to source after return",
        "button.save_update": "Save / Update",
        "button.save_switch": "Save and Switch",
        "button.switch_form": "Switch Using Form",
        "button.delete_saved_entry": "Delete Saved Entry",
        "button.open_provider_store": "Open Provider Store",
        "button.load_current_provider": "Load Current Provider",
        "button.start_session": "Start Session",
        "button.move_all_source_threads": "Move All Source Threads",
        "button.return_selected_session": "Return Selected Session",
        "status.ready": "Ready",
        "status.checking_process": "Checking Codex process...",
        "status.refreshed": "Refreshed",
        "status.app_started": "App started",
        "misc.none": "(none)",
        "misc.no_process_detected": "No Codex process detected",
        "misc.yes": "Yes",
        "misc.no": "No",
        "misc.runtime_suffix": "runtime",
        "session_status.open": "Open",
        "session_status.completed": "Completed",
        "note.provider_default": "Saved providers live in provider_store.json. Deleting a saved entry will not remove chat threads.",
        "note.provider_runtime_key": "This key currently comes from auth.json for the active provider. Click Save / Update if you want to keep it for quick switching later.",
        "note.provider_saved_entry": "This provider has a saved local entry. Deleting it will only remove the saved entry, not chat history or the existing config block.",
        "note.provider_config_only": "This provider currently exists only in Codex config. Save it here if you want a remembered URL/key for quick switching.",
        "note.bridge_hint": "Note: Codex chats are separated by model_provider.\nThis tool does migration / return migration, not physical duplication of threads.",
        "action_label.running_migration": "running a migration",
        "action_label.switching_providers": "switching providers",
        "error.open_path": "Cannot open path:\n{path}\n\n{error}",
        "warn.codex_active": "Codex still looks active:\n\n{processes}\n\nPlease close Codex Desktop completely before {action_label}.",
        "error.missing_required_files": "Missing required files:\n\n{files}",
        "error.sqlite_not_ready": "SQLite is not ready for write operations.\n\n{error}",
        "error.missing_required_config": "Missing required config:\n\n{path}",
        "error.enter_provider_and_url": "Please enter both provider name and base URL.",
        "error.save_provider_failed": "Could not save provider entry.\n\n{error}",
        "error.choose_or_enter_provider": "Please choose or enter a provider first.",
        "error.enter_url_before_switch": "Please enter a base URL before switching.",
        "error.switch_provider_failed": "Could not switch provider.\n\n{error}",
        "error.choose_provider_first": "Please choose a provider first.",
        "error.provider_has_no_saved_entry": "This provider does not have a saved local entry.\n\nCurrent version only deletes saved entries and does not remove config blocks.",
        "error.saved_provider_missing": "The saved provider entry could not be found.",
        "error.full_backup_failed": "Full backup failed.\n\n{error}",
        "error.choose_source_target": "Please choose both source and target providers.",
        "error.source_target_different": "Source and target providers must be different.",
        "error.start_bridge_session_failed": "Could not start bridge session.\n\n{error}",
        "error.select_bridge_session_first": "Please select a bridge session first.",
        "error.bridge_session_completed": "That bridge session is already completed.",
        "error.return_bridge_session_failed": "Could not return bridge session.\n\n{error}",
        "error.direct_move_failed": "Could not move threads.\n\n{error}",
        "error.runtime.bridge_session_not_open": "This bridge session is not open anymore.",
        "error.runtime.provider_name_empty": "Provider name cannot be empty.",
        "error.runtime.base_url_empty": "Base URL cannot be empty.",
        "error.runtime.base_url_invalid": "Base URL must start with http:// or https:// and include a host.",
        "confirm.no_api_key_continue": "No API key is set in the form.\n\nThe provider block and current provider will be updated, but auth.json will be left unchanged.\n\nContinue?",
        "confirm.switch_provider": "Switch current provider to '{provider}'?\n\nBase URL: {base_url}\n\nA backup of config/auth/provider store will be created first.",
        "confirm.delete_saved_provider": "Delete the saved entry for '{provider}'?\n\nThis will not remove chat threads or the existing provider block from config.toml.",
        "confirm.full_backup_while_active": "Codex looks active.\n\nA full backup can still work, but files may change while copying.\n\nContinue anyway?",
        "confirm.start_bridge_session": "Start a bridge session?\n\nAll threads currently under '{source}' will be moved to '{target}'.\nA critical backup will be created first.",
        "confirm.return_bridge_session": "Return bridge session?\n\nThreads recorded in '{target}' will be moved back to '{source}'.\nA critical backup will be created first.",
        "confirm.direct_move": "Move all threads from '{source}' to '{target}'?\n\nA critical backup will be created first.",
        "success.saved_provider_entry": "Saved provider entry:\n{provider}\n\nThe current provider was not switched. Use Save and Switch or Switch Using Form to write the change to config.toml.",
        "success.provider_switched": "Provider switched and verified.\n\nProvider: {provider}\nAuth updated: {auth_updated}\nConfig path: {config_path}\nBackup: {backup_dir}",
        "success.full_backup_created": "Full backup created:\n{target}",
        "success.bridge_session_started": "Bridge session started.\n\nSession: {session_id}\nMoved threads: {moved_thread_count}",
        "success.bridge_session_returned": "Bridge session returned.\n\nRestored threads: {restored_thread_count}\nNew threads moved back: {new_thread_count}",
        "success.direct_move_finished": "Direct move finished.\n\nMoved threads: {changed}\nBackup: {backup_dir}",
        "log.saved_provider": "Saved provider {provider} ({api_key_mode} API key)",
        "log.switched_provider": "Switched provider to {provider} ({auth_mode})",
        "log.deleted_saved_provider": "Deleted saved provider entry for {provider}",
        "log.full_backup_created": "Created full backup at {target}",
        "log.started_bridge_session": "Started bridge session {session_id}: {moved_thread_count} threads moved {source} -> {target}",
        "log.completed_bridge_session": "Completed bridge session {session_id}: restored {restored_thread_count} and moved {new_thread_count} new threads",
        "log.direct_move_finished": "Direct move finished: {changed} threads moved {source} -> {target}",
        "phrase.with_api_key": "with",
        "phrase.without_api_key": "without",
        "phrase.updated_auth": "updated auth",
        "phrase.config_only": "config only",
    },
}


def normalize_language(value: str | None) -> str:
    candidate = str(value or "").strip()
    return candidate if candidate in I18N else DEFAULT_LANGUAGE


def language_label(code: str) -> str:
    return LANGUAGE_LABELS.get(normalize_language(code), LANGUAGE_LABELS[DEFAULT_LANGUAGE])


def language_code_from_label(label: str) -> str:
    raw = str(label or "").strip()
    for code, value in LANGUAGE_LABELS.items():
        if raw == value:
            return code
    return DEFAULT_LANGUAGE


def translate(language: str, key: str, **kwargs) -> str:
    lang = normalize_language(language)
    template = I18N.get(lang, {}).get(key) or I18N["en_US"].get(key) or key
    try:
        return template.format(**kwargs)
    except Exception:
        return template


@dataclass
class BridgePaths:
    codex_home: Path
    workspace_root: Path
    bridge_root: Path
    backup_root: Path
    session_root: Path
    log_root: Path
    settings_path: Path
    config_path: Path
    auth_path: Path
    provider_store_path: Path
    db_path: Path
    wal_path: Path
    shm_path: Path


def resolve_workspace_root() -> Path:
    if getattr(sys, "frozen", False):
        return Path(sys.executable).resolve().parent
    return Path(__file__).resolve().parent


def app_paths(codex_home: Path | None = None) -> BridgePaths:
    workspace_root = resolve_workspace_root()
    bridge_root = workspace_root / "artifacts" / "codex-chat-bridge"
    backup_root = bridge_root / "backups"
    session_root = bridge_root / "sessions"
    log_root = bridge_root / "logs"
    for folder in (bridge_root, backup_root, session_root, log_root):
        folder.mkdir(parents=True, exist_ok=True)
    actual_codex_home = codex_home or (Path.home() / ".codex")
    return BridgePaths(
        codex_home=actual_codex_home,
        workspace_root=workspace_root,
        bridge_root=bridge_root,
        backup_root=backup_root,
        session_root=session_root,
        log_root=log_root,
        settings_path=bridge_root / "app_settings.json",
        config_path=actual_codex_home / "config.toml",
        auth_path=actual_codex_home / "auth.json",
        provider_store_path=bridge_root / "provider_store.json",
        db_path=actual_codex_home / "state_5.sqlite",
        wal_path=actual_codex_home / "state_5.sqlite-wal",
        shm_path=actual_codex_home / "state_5.sqlite-shm",
    )


def iso_now() -> str:
    return datetime.now().isoformat(timespec="seconds")


def ts_now() -> str:
    return datetime.now().strftime("%Y%m%d-%H%M%S")


def load_text(path: Path) -> str:
    return path.read_text(encoding="utf-8") if path.exists() else ""


def atomic_write_text(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temp_path = path.with_name(f"{path.name}.tmp-{uuid.uuid4().hex}")
    try:
        temp_path.write_text(content, encoding="utf-8")
        os.replace(temp_path, path)
    finally:
        if temp_path.exists():
            temp_path.unlink()


def save_text(path: Path, content: str) -> None:
    atomic_write_text(path, content)


def load_json_file(path: Path, default):
    if not path.exists():
        return default
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return default


def save_json_file(path: Path, payload: object) -> None:
    save_text(path, json.dumps(payload, ensure_ascii=False, indent=2) + "\n")


def safe_filename_component(value: str) -> str:
    cleaned = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", str(value or "").strip())
    cleaned = cleaned.rstrip(" .")
    return cleaned or "item"


def load_app_settings(path: Path) -> dict[str, object]:
    payload = load_json_file(path, {})
    return payload if isinstance(payload, dict) else {}


def save_app_settings(path: Path, payload: dict[str, object]) -> None:
    save_json_file(path, payload)


def append_log(paths: BridgePaths, message: str) -> None:
    log_path = paths.log_root / f"{datetime.now():%Y-%m}.log"
    with log_path.open("a", encoding="utf-8") as fh:
        fh.write(f"[{iso_now()}] {message}\n")


def is_blocking_codex_process(image_name: str) -> bool:
    normalized = str(image_name or "").strip().lower()
    if not normalized:
        return False

    ignored_names = {
        "codexchatbridge.exe",
        "codex-chat-bridge.exe",
        Path(sys.executable).name.lower(),
        Path(sys.argv[0]).name.lower(),
    }
    if normalized in ignored_names:
        return False

    if "bridge" in normalized:
        return False

    return "codex" in normalized


def detect_codex_processes() -> list[str]:
    try:
        output = subprocess.check_output(
            ["tasklist", "/FO", "CSV", "/NH"],
            text=True,
            encoding="utf-8",
            errors="ignore",
            creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
        )
    except Exception:
        return []

    found: list[str] = []
    for row in csv.reader(output.splitlines()):
        if not row:
            continue
        image_name = row[0].strip()
        if is_blocking_codex_process(image_name):
            found.append(image_name)
    return sorted(set(found))


def ensure_sqlite_ready(db_path: Path) -> None:
    conn = sqlite3.connect(str(db_path), timeout=1.0)
    try:
        cur = conn.cursor()
        cur.execute("PRAGMA busy_timeout = 1000")
        cur.execute("BEGIN IMMEDIATE")
        conn.rollback()
    finally:
        conn.close()


def sqlite_connect(db_path: Path) -> sqlite3.Connection:
    conn = sqlite3.connect(str(db_path), timeout=5.0)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA busy_timeout = 5000")
    return conn


def get_thread_counts(db_path: Path) -> dict[str, int]:
    conn = sqlite_connect(db_path)
    try:
        cur = conn.cursor()
        cur.execute(
            "select model_provider, count(*) as cnt from threads group by model_provider order by model_provider"
        )
        return {str(row["model_provider"]): int(row["cnt"]) for row in cur.fetchall()}
    finally:
        conn.close()


def list_thread_ids(db_path: Path, provider: str) -> list[str]:
    conn = sqlite_connect(db_path)
    try:
        cur = conn.cursor()
        cur.execute(
            "select id from threads where model_provider = ? order by updated_at desc, id asc",
            (provider,),
        )
        return [str(row["id"]) for row in cur.fetchall()]
    finally:
        conn.close()


def list_new_thread_ids_since(db_path: Path, provider: str, started_at: int, exclude_ids: set[str]) -> list[str]:
    conn = sqlite_connect(db_path)
    try:
        cur = conn.cursor()
        cur.execute(
            """
            select id
            from threads
            where model_provider = ?
              and created_at >= ?
            order by created_at asc, id asc
            """,
            (provider, started_at),
        )
        result = []
        for row in cur.fetchall():
            thread_id = str(row["id"])
            if thread_id not in exclude_ids:
                result.append(thread_id)
        return result
    finally:
        conn.close()


def update_thread_provider(db_path: Path, thread_ids: Iterable[str], target_provider: str) -> int:
    ids = [thread_id for thread_id in thread_ids if thread_id]
    if not ids:
        return 0
    conn = sqlite_connect(db_path)
    changed = 0
    try:
        cur = conn.cursor()
        cur.execute("BEGIN IMMEDIATE")
        for start in range(0, len(ids), CHUNK_SIZE):
            chunk = ids[start:start + CHUNK_SIZE]
            placeholders = ",".join("?" for _ in chunk)
            sql = f"update threads set model_provider = ? where id in ({placeholders})"
            cur.execute(sql, [target_provider, *chunk])
            changed += cur.rowcount
        conn.commit()
        return changed
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def extract_provider_names_from_config(config_text: str) -> list[str]:
    try:
        data = tomllib.loads(config_text)
        providers = data.get("model_providers", {})
        if isinstance(providers, dict):
            return sorted(str(name).strip() for name in providers.keys() if str(name).strip())
    except Exception:
        pass

    names: list[str] = []
    for match in re.finditer(r'(?m)^\[model_providers\.(?:"([^"]+)"|([^\]\r\n]+))\]$', config_text):
        provider = match.group(1) or match.group(2) or ""
        provider = provider.strip()
        if provider:
            names.append(provider)
    return sorted(set(names))


def get_current_provider(config_path: Path) -> str:
    raw = load_text(config_path)
    try:
        data = tomllib.loads(raw)
        current = data.get("model_provider", "")
        return str(current).strip() if current else ""
    except Exception:
        match = re.search(r'(?m)^model_provider = "([^"]+)"$', raw)
        return match.group(1) if match else ""


def find_provider_block(config_text: str, provider: str) -> tuple[str | None, tuple[int, int] | None]:
    patterns = [
        rf'(?ms)^\[model_providers\.{re.escape(provider)}\]\r?\n.*?(?=^\[|\Z)',
        rf'(?ms)^\[model_providers\."{re.escape(provider)}"\]\r?\n.*?(?=^\[|\Z)',
    ]
    for pattern in patterns:
        match = re.search(pattern, config_text)
        if match:
            return match.group(0), match.span()
    return None, None


def provider_settings_from_block(block: str, provider: str) -> dict[str, object]:
    settings: dict[str, object] = {
        "provider": provider,
        "name": provider,
        "base_url": "",
        "wire_api": "responses",
        "requires_openai_auth": True,
    }
    for line in block.splitlines()[1:]:
        match = re.match(r'^\s*([A-Za-z0-9_]+)\s*=\s*(.+?)\s*$', line)
        if not match:
            continue
        key, value = match.group(1), match.group(2)
        if value.startswith('"') and value.endswith('"'):
            parsed_value: object = value[1:-1]
        elif value in ("true", "false"):
            parsed_value = value == "true"
        else:
            parsed_value = value
        settings[key] = parsed_value
    return settings


def load_config_provider_map(config_path: Path) -> dict[str, dict[str, object]]:
    raw = load_text(config_path)
    result: dict[str, dict[str, object]] = {}
    try:
        data = tomllib.loads(raw)
        providers = data.get("model_providers", {})
        if isinstance(providers, dict):
            for key, value in providers.items():
                provider_key = str(key).strip()
                if not provider_key:
                    continue
                if isinstance(value, dict):
                    result[provider_key] = {
                        "provider": provider_key,
                        "name": str(value.get("name") or provider_key),
                        "base_url": str(value.get("base_url") or ""),
                        "wire_api": str(value.get("wire_api") or "responses"),
                        "requires_openai_auth": bool(value.get("requires_openai_auth", True)),
                    }
                else:
                    result[provider_key] = {
                        "provider": provider_key,
                        "name": provider_key,
                        "base_url": "",
                        "wire_api": "responses",
                        "requires_openai_auth": True,
                    }
            return result
    except Exception:
        pass

    for provider in extract_provider_names_from_config(raw):
        block, _ = find_provider_block(raw, provider)
        if block:
            result[provider] = provider_settings_from_block(block, provider)
    return result


def render_provider_block(
    provider: str,
    base_url: str,
    existing_block: str | None = None,
    wire_api: str = "responses",
    requires_openai_auth: bool = True,
) -> str:
    line_ending = "\r\n" if existing_block and "\r\n" in existing_block else "\n"
    preserved_lines: list[str] = []
    if existing_block:
        for line in existing_block.splitlines()[1:]:
            if re.match(r'^\s*(name|base_url|wire_api|requires_openai_auth)\s*=', line):
                continue
            preserved_lines.append(line)

    lines = [
        f'[model_providers."{provider}"]',
        f'name = "{provider}"',
        f'base_url = "{base_url}"',
        f'wire_api = "{wire_api}"',
        f'requires_openai_auth = {"true" if requires_openai_auth else "false"}',
    ]
    lines.extend(preserved_lines)
    return line_ending.join(lines).rstrip() + line_ending


def minimal_provider_block(provider: str, base_url: str = "https://api.openai.com/v1") -> str:
    return (
        f'\n[model_providers."{provider}"]\n'
        f'name = "{provider}"\n'
        f'base_url = "{base_url}"\n'
        'wire_api = "responses"\n'
        'requires_openai_auth = true\n'
    )


def upsert_provider_in_config(
    config_path: Path,
    provider: str,
    base_url: str,
    wire_api: str = "responses",
    requires_openai_auth: bool = True,
) -> None:
    raw = load_text(config_path)
    block, span = find_provider_block(raw, provider)
    rendered = render_provider_block(provider, base_url, block, wire_api=wire_api, requires_openai_auth=requires_openai_auth)
    if span:
        updated = raw[:span[0]] + rendered + raw[span[1]:]
    else:
        updated = raw.rstrip() + "\n\n" + rendered.strip("\n") + "\n"
    save_text(config_path, updated)


def ensure_provider_exists(config_path: Path, source_provider: str, target_provider: str) -> None:
    raw = load_text(config_path)
    target_block, _ = find_provider_block(raw, target_provider)
    if target_block:
        return

    source_block, _ = find_provider_block(raw, source_provider)
    if source_block:
        source_settings = provider_settings_from_block(source_block, source_provider)
        base_url = str(source_settings.get("base_url") or "https://api.openai.com/v1")
        wire_api = str(source_settings.get("wire_api") or "responses")
        requires_openai_auth = bool(source_settings.get("requires_openai_auth", True))
    else:
        base_url = "https://api.openai.com/v1"
        wire_api = "responses"
        requires_openai_auth = True
    upsert_provider_in_config(
        config_path,
        target_provider,
        base_url,
        wire_api=wire_api,
        requires_openai_auth=requires_openai_auth,
    )


def set_current_provider(config_path: Path, provider: str) -> None:
    raw = load_text(config_path)
    if re.search(r'(?m)^model_provider = "([^"]+)"$', raw):
        updated = re.sub(r'(?m)^model_provider = "([^"]+)"$', f'model_provider = "{provider}"', raw, count=1)
    else:
        updated = f'model_provider = "{provider}"\n' + raw
    save_text(config_path, updated)


def load_provider_store(path: Path) -> dict[str, dict[str, object]]:
    raw = load_json_file(path, {})
    providers = raw.get("providers") if isinstance(raw, dict) else {}
    if not isinstance(providers, dict):
        providers = raw if isinstance(raw, dict) else {}
    result: dict[str, dict[str, object]] = {}
    for provider, entry in providers.items():
        provider_key = str(provider).strip()
        if not provider_key or not isinstance(entry, dict):
            continue
        result[provider_key] = {
            "base_url": str(entry.get("base_url") or ""),
            "api_key": str(entry.get("api_key") or ""),
            "updated_at_iso": str(entry.get("updated_at_iso") or ""),
        }
    return result


def save_provider_store(path: Path, providers: dict[str, dict[str, object]]) -> None:
    payload = {
        "version": 1,
        "providers": dict(sorted(providers.items(), key=lambda item: item[0].lower())),
    }
    save_json_file(path, payload)


def save_provider_entry(path: Path, provider: str, base_url: str, api_key: str) -> None:
    providers = load_provider_store(path)
    providers[provider] = {
        "base_url": base_url,
        "api_key": api_key,
        "updated_at_iso": iso_now(),
    }
    save_provider_store(path, providers)


def delete_provider_entry(path: Path, provider: str) -> bool:
    providers = load_provider_store(path)
    if provider not in providers:
        return False
    del providers[provider]
    save_provider_store(path, providers)
    return True


def load_auth_payload(path: Path) -> dict[str, object]:
    payload = load_json_file(path, {})
    return payload if isinstance(payload, dict) else {}


def get_runtime_api_key(path: Path) -> str:
    payload = load_auth_payload(path)
    value = payload.get("OPENAI_API_KEY", "")
    return str(value).strip() if value else ""


def write_runtime_api_key(path: Path, api_key: str) -> None:
    payload = load_auth_payload(path)
    payload["OPENAI_API_KEY"] = api_key
    save_json_file(path, payload)


def mask_secret(secret: str) -> str:
    if not secret:
        return ""
    if len(secret) <= 8:
        return "*" * len(secret)
    return f"{secret[:4]}...{secret[-4:]}"


def validate_provider_url(value: str) -> bool:
    parsed = urlparse(value.strip())
    return parsed.scheme in {"http", "https"} and bool(parsed.netloc)


def normalize_url_for_compare(value: str) -> str:
    return str(value or "").strip().rstrip("/")


def build_provider_catalog(paths: BridgePaths, include_threads: bool = True) -> dict[str, dict[str, object]]:
    config_entries = load_config_provider_map(paths.config_path)
    saved_entries = load_provider_store(paths.provider_store_path)
    thread_counts = get_thread_counts(paths.db_path) if include_threads and paths.db_path.exists() else {}
    current_provider = get_current_provider(paths.config_path)
    runtime_key = get_runtime_api_key(paths.auth_path)

    catalog: dict[str, dict[str, object]] = {}
    all_names = sorted(set(config_entries.keys()) | set(saved_entries.keys()) | set(thread_counts.keys()))
    for provider in all_names:
        config_entry = config_entries.get(provider, {})
        saved_entry = saved_entries.get(provider, {})
        saved_key = str(saved_entry.get("api_key") or "")
        runtime_key_for_provider = runtime_key if provider == current_provider and not saved_key else ""
        effective_base_url = str(saved_entry.get("base_url") or config_entry.get("base_url") or "")
        catalog[provider] = {
            "provider": provider,
            "base_url": effective_base_url,
            "config_base_url": str(config_entry.get("base_url") or ""),
            "saved_base_url": str(saved_entry.get("base_url") or ""),
            "api_key": saved_key or runtime_key_for_provider,
            "saved_api_key": saved_key,
            "key_source": "saved" if saved_key else ("runtime" if runtime_key_for_provider else ""),
            "in_config": provider in config_entries,
            "in_store": provider in saved_entries,
            "current": provider == current_provider,
            "threads": int(thread_counts.get(provider, 0)),
            "wire_api": str(config_entry.get("wire_api") or "responses"),
            "requires_openai_auth": bool(config_entry.get("requires_openai_auth", True)),
        }
    return catalog


def backup_selected_files(paths: BridgePaths, label: str, sources: Iterable[Path]) -> Path:
    target = paths.backup_root / f"{ts_now()}-{safe_filename_component(label)}"
    target.mkdir(parents=True, exist_ok=True)
    for src in sources:
        if src.exists():
            shutil.copy2(src, target / src.name)
    append_log(paths, f"Created backup at {target}")
    return target


def backup_critical_files(paths: BridgePaths, label: str) -> Path:
    return backup_selected_files(paths, label, (paths.config_path, paths.db_path, paths.wal_path, paths.shm_path))


def backup_provider_files(paths: BridgePaths, label: str) -> Path:
    return backup_selected_files(paths, label, (paths.config_path, paths.auth_path, paths.provider_store_path))


def backup_full_codex(paths: BridgePaths) -> Path:
    target = paths.backup_root / f"{ts_now()}-full-codex"
    shutil.copytree(paths.codex_home, target)
    append_log(paths, f"Created full Codex backup at {target}")
    return target


def session_file(paths: BridgePaths, session_id: str) -> Path:
    return paths.session_root / f"{session_id}.json"


def load_session_records(paths: BridgePaths) -> list[dict]:
    records: list[dict] = []
    for path in sorted(paths.session_root.glob("*.json"), reverse=True):
        try:
            data = json.loads(path.read_text(encoding="utf-8"))
            data["_file"] = str(path)
            records.append(data)
        except Exception:
            continue
    return records


def save_session_record(paths: BridgePaths, record: dict) -> Path:
    path = session_file(paths, record["session_id"])
    path.write_text(json.dumps(record, ensure_ascii=False, indent=2), encoding="utf-8")
    return path


def create_bridge_session(
    paths: BridgePaths,
    source_provider: str,
    target_provider: str,
    switch_current: bool,
) -> dict:
    ensure_sqlite_ready(paths.db_path)
    backup_dir = backup_critical_files(paths, f"bridge-start-{source_provider}-to-{target_provider}")
    started_at = int(time.time())
    moved_thread_ids = list_thread_ids(paths.db_path, source_provider)
    changed = update_thread_provider(paths.db_path, moved_thread_ids, target_provider)
    ensure_provider_exists(paths.config_path, source_provider, target_provider)
    if switch_current:
        set_current_provider(paths.config_path, target_provider)

    record = {
        "session_id": f"{datetime.now():%Y%m%d-%H%M%S}-{uuid.uuid4().hex[:8]}",
        "created_at_iso": iso_now(),
        "started_at": started_at,
        "source_provider": source_provider,
        "target_provider": target_provider,
        "switch_current_on_start": switch_current,
        "status": "open",
        "moved_thread_ids": moved_thread_ids,
        "moved_thread_count": changed,
        "backup_dir": str(backup_dir),
    }
    save_session_record(paths, record)
    append_log(
        paths,
        f"Started bridge session {record['session_id']}: moved {changed} threads from {source_provider} to {target_provider}",
    )
    return record


def complete_bridge_session(
    paths: BridgePaths,
    record: dict,
    switch_current_back: bool,
    include_new_target_threads: bool,
) -> dict:
    if record.get("status") != "open":
        raise RuntimeError("This bridge session is not open anymore.")

    source_provider = str(record["source_provider"])
    target_provider = str(record["target_provider"])
    started_at = int(record["started_at"])
    moved_thread_ids = [str(item) for item in record.get("moved_thread_ids", [])]

    ensure_sqlite_ready(paths.db_path)
    backup_dir = backup_critical_files(paths, f"bridge-return-{target_provider}-to-{source_provider}")

    restored_count = update_thread_provider(paths.db_path, moved_thread_ids, source_provider)

    new_thread_ids: list[str] = []
    new_thread_count = 0
    if include_new_target_threads:
        exclude_ids = set(moved_thread_ids)
        new_thread_ids = list_new_thread_ids_since(paths.db_path, target_provider, started_at, exclude_ids)
        new_thread_count = update_thread_provider(paths.db_path, new_thread_ids, source_provider)

    ensure_provider_exists(paths.config_path, target_provider, source_provider)
    if switch_current_back:
        set_current_provider(paths.config_path, source_provider)

    record["status"] = "completed"
    record["completed_at_iso"] = iso_now()
    record["switch_current_on_return"] = switch_current_back
    record["include_new_target_threads"] = include_new_target_threads
    record["restored_thread_count"] = restored_count
    record["new_thread_ids"] = new_thread_ids
    record["new_thread_count"] = new_thread_count
    record["return_backup_dir"] = str(backup_dir)
    save_session_record(paths, record)
    append_log(
        paths,
        f"Completed bridge session {record['session_id']}: restored {restored_count} threads and moved {new_thread_count} new threads from {target_provider} back to {source_provider}",
    )
    return record


def direct_move(
    paths: BridgePaths,
    source_provider: str,
    target_provider: str,
    switch_current: bool,
) -> dict:
    ensure_sqlite_ready(paths.db_path)
    backup_dir = backup_critical_files(paths, f"direct-move-{source_provider}-to-{target_provider}")
    thread_ids = list_thread_ids(paths.db_path, source_provider)
    changed = update_thread_provider(paths.db_path, thread_ids, target_provider)
    ensure_provider_exists(paths.config_path, source_provider, target_provider)
    if switch_current:
        set_current_provider(paths.config_path, target_provider)
    append_log(
        paths,
        f"Direct move: moved {changed} threads from {source_provider} to {target_provider}",
    )
    return {
        "backup_dir": str(backup_dir),
        "changed": changed,
        "source_provider": source_provider,
        "target_provider": target_provider,
    }


def save_provider_definition(paths: BridgePaths, provider: str, base_url: str, api_key: str) -> dict[str, object]:
    provider_name = provider.strip()
    url = base_url.strip()
    if not provider_name:
        raise RuntimeError("Provider name cannot be empty.")
    if not url:
        raise RuntimeError("Base URL cannot be empty.")
    if not validate_provider_url(url):
        raise RuntimeError("Base URL must start with http:// or https:// and include a host.")

    save_provider_entry(paths.provider_store_path, provider_name, url, api_key.strip())
    append_log(paths, f"Saved provider definition for {provider_name}")
    return {
        "provider": provider_name,
        "base_url": url,
        "has_api_key": bool(api_key.strip()),
        "store_path": str(paths.provider_store_path),
    }


def verify_provider_switch(config_path: Path, provider: str, base_url: str) -> dict[str, str]:
    current_provider = get_current_provider(config_path)
    if current_provider != provider:
        raise RuntimeError(
            f'Config write verification failed: model_provider in {config_path} is "{current_provider or "(none)"}", expected "{provider}".'
        )

    provider_entry = load_config_provider_map(config_path).get(provider)
    if not provider_entry:
        raise RuntimeError(
            f'Config write verification failed: provider "{provider}" is missing in {config_path}.'
        )

    actual_base_url = str(provider_entry.get("base_url") or "").strip()
    if normalize_url_for_compare(actual_base_url) != normalize_url_for_compare(base_url):
        raise RuntimeError(
            f'Config write verification failed: base_url for "{provider}" in {config_path} is "{actual_base_url}", expected "{base_url}".'
        )

    return {
        "current_provider": current_provider,
        "actual_base_url": actual_base_url,
    }


def switch_provider(paths: BridgePaths, provider: str, base_url: str, api_key: str = "") -> dict[str, object]:
    provider_name = provider.strip()
    url = base_url.strip()
    if not provider_name:
        raise RuntimeError("Provider name cannot be empty.")
    if not url:
        raise RuntimeError("Base URL cannot be empty.")
    if not validate_provider_url(url):
        raise RuntimeError("Base URL must start with http:// or https:// and include a host.")

    backup_dir = backup_provider_files(paths, f"provider-switch-{provider_name}")
    upsert_provider_in_config(paths.config_path, provider_name, url)
    set_current_provider(paths.config_path, provider_name)

    auth_updated = False
    clean_key = api_key.strip()
    if clean_key:
        write_runtime_api_key(paths.auth_path, clean_key)
        auth_updated = True
    verified = verify_provider_switch(paths.config_path, provider_name, url)

    append_log(
        paths,
        f"Switched current provider to {provider_name} (auth {'updated' if auth_updated else 'left unchanged'})",
    )
    return {
        "provider": provider_name,
        "base_url": url,
        "auth_updated": auth_updated,
        "backup_dir": str(backup_dir),
        "config_path": str(paths.config_path),
        "verified_base_url": verified["actual_base_url"],
    }


class BridgeApp:
    def __init__(self, root: tk.Tk):
        self.root = root
        self.paths = app_paths()
        self.app_settings = load_app_settings(self.paths.settings_path)
        self.language = normalize_language(self.app_settings.get("language"))
        self.root.geometry("1380x920")
        self.root.minsize(1180, 800)

        self.current_provider_name = ""
        self.status_var = tk.StringVar(value=self.t("status.ready"))
        self.current_provider_var = tk.StringVar(value=self.t("misc.none"))
        self.process_var = tk.StringVar(value=self.t("status.checking_process"))
        self.provider_note_var = tk.StringVar()
        self.lang_var = tk.StringVar(value=language_label(self.language))

        self.start_source_var = tk.StringVar()
        self.start_target_var = tk.StringVar(value="openai")
        self.start_switch_var = tk.BooleanVar(value=True)

        self.direct_source_var = tk.StringVar()
        self.direct_target_var = tk.StringVar(value="openai")
        self.direct_switch_var = tk.BooleanVar(value=True)

        self.return_session_var = tk.StringVar()
        self.return_switch_var = tk.BooleanVar(value=True)
        self.return_new_threads_var = tk.BooleanVar(value=True)

        self.manager_provider_var = tk.StringVar()
        self.manager_url_var = tk.StringVar()
        self.manager_key_var = tk.StringVar()
        self.manager_show_key_var = tk.BooleanVar(value=False)

        self.provider_values: list[str] = []
        self.provider_catalog: dict[str, dict[str, object]] = {}
        self.session_records: list[dict] = []
        self.localized_widgets: list[tuple[object, str]] = []
        self.localized_tabs: list[tuple[ttk.Notebook, object, str]] = []
        self.localized_headings: list[tuple[ttk.Treeview, str, str]] = []
        self.provider_note_key = "note.provider_default"
        self.provider_note_kwargs: dict[str, object] = {}

        self._configure_styles()
        self._build_ui()
        self.apply_locale()
        self.set_provider_note("note.provider_default")
        self.refresh_all()

    def t(self, key: str, **kwargs) -> str:
        return translate(self.language, key, **kwargs)

    def _configure_styles(self) -> None:
        self.style = ttk.Style(self.root)
        for theme_name in ("vista", "xpnative", "clam"):
            try:
                self.style.theme_use(theme_name)
                break
            except tk.TclError:
                continue

        base_font = ("Microsoft YaHei UI", 10)
        self.style.configure(".", font=base_font)
        self.style.configure("TButton", padding=(10, 6))
        self.style.configure("Header.TLabel", font=("Microsoft YaHei UI", 17, "bold"))
        self.style.configure("Muted.TLabel", foreground="#5A6472")
        self.style.configure("Treeview", rowheight=24, font=base_font)
        self.style.configure("Treeview.Heading", font=("Microsoft YaHei UI", 10, "bold"))
        self.style.configure("TNotebook.Tab", padding=(14, 8))

    def register_text(self, widget: object, key: str) -> object:
        self.localized_widgets.append((widget, key))
        if hasattr(widget, "configure"):
            widget.configure(text=self.t(key))
        return widget

    def register_tab(self, notebook: ttk.Notebook, tab_id: object, key: str) -> None:
        self.localized_tabs.append((notebook, tab_id, key))
        notebook.tab(tab_id, text=self.t(key))

    def register_heading(self, tree: ttk.Treeview, column: str, key: str) -> None:
        self.localized_headings.append((tree, column, key))
        tree.heading(column, text=self.t(key))

    def apply_locale(self) -> None:
        self.root.title(self.t("app_title"))
        self.lang_var.set(language_label(self.language))
        for widget, key in self.localized_widgets:
            if hasattr(widget, "configure"):
                widget.configure(text=self.t(key))
        for notebook, tab_id, key in self.localized_tabs:
            notebook.tab(tab_id, text=self.t(key))
        for tree, column, key in self.localized_headings:
            tree.heading(column, text=self.t(key))
        self.provider_note_var.set(self.t(self.provider_note_key, **self.provider_note_kwargs))
        if not self.current_provider_name:
            self.current_provider_var.set(self.t("misc.none"))

    def save_language(self) -> None:
        self.app_settings["language"] = self.language
        save_app_settings(self.paths.settings_path, self.app_settings)

    def set_language(self, code: str, persist: bool = True) -> None:
        next_language = normalize_language(code)
        if next_language == self.language and self.lang_var.get() == language_label(next_language):
            return
        self.language = next_language
        if persist:
            self.save_language()
        self.apply_locale()
        if hasattr(self, "provider_tree"):
            self.refresh_all()

    def on_language_selected(self, _event=None) -> None:
        self.set_language(language_code_from_label(self.lang_var.get()))

    def app_title(self) -> str:
        return self.t("app_title")

    def localize_error(self, exc: Exception) -> str:
        message = str(exc).strip()
        if message in RUNTIME_ERROR_KEYS:
            return self.t(RUNTIME_ERROR_KEYS[message])
        return message or repr(exc)

    def show_error(self, key: str, **kwargs) -> None:
        messagebox.showerror(self.app_title(), self.t(key, **kwargs))

    def show_info(self, key: str, **kwargs) -> None:
        messagebox.showinfo(self.app_title(), self.t(key, **kwargs))

    def show_warning(self, key: str, **kwargs) -> None:
        messagebox.showwarning(self.app_title(), self.t(key, **kwargs))

    def ask_yes_no(self, key: str, **kwargs) -> bool:
        return messagebox.askyesno(self.app_title(), self.t(key, **kwargs))

    def set_provider_note(self, key: str, **kwargs) -> None:
        self.provider_note_key = key
        self.provider_note_kwargs = dict(kwargs)
        self.provider_note_var.set(self.t(key, **kwargs))

    def yes_no(self, value: bool, blank_if_false: bool = False) -> str:
        if value:
            return self.t("misc.yes")
        return "" if blank_if_false else self.t("misc.no")

    def format_session_status(self, value: object) -> str:
        status = str(value or "").strip()
        if status in {"open", "completed"}:
            return self.t(f"session_status.{status}")
        return status

    def log_key(self, key: str, **kwargs) -> None:
        self.log(self.t(key, **kwargs))

    def _build_ui(self) -> None:
        self.root.columnconfigure(0, weight=1)
        self.root.rowconfigure(2, weight=3)
        self.root.rowconfigure(3, weight=2)

        top = ttk.Frame(self.root, padding=14)
        top.grid(row=0, column=0, sticky="ew")
        top.columnconfigure(1, weight=1)

        self.register_text(ttk.Label(top, style="Header.TLabel"), "app_title").grid(row=0, column=0, sticky="w")
        self.register_text(ttk.Label(top), "label.language").grid(row=0, column=2, sticky="e", padx=(8, 8))
        self.lang_combo = ttk.Combobox(
            top,
            textvariable=self.lang_var,
            state="readonly",
            values=[language_label("zh_CN"), language_label("en_US")],
            width=10,
        )
        self.lang_combo.grid(row=0, column=3, sticky="e")
        self.lang_combo.bind("<<ComboboxSelected>>", self.on_language_selected)
        self.register_text(ttk.Button(top, command=self.refresh_all), "label.refresh").grid(
            row=0, column=4, sticky="e", padx=(10, 0)
        )

        self.register_text(ttk.Label(top), "label.codex_home").grid(row=1, column=0, sticky="w", pady=(10, 0))
        ttk.Label(top, text=str(self.paths.codex_home)).grid(row=1, column=1, sticky="w", pady=(8, 0))
        self.register_text(ttk.Label(top), "label.current_provider").grid(row=2, column=0, sticky="w", pady=(6, 0))
        ttk.Label(top, textvariable=self.current_provider_var).grid(row=2, column=1, sticky="w", pady=(4, 0))
        self.register_text(ttk.Label(top), "label.codex_process").grid(row=3, column=0, sticky="w", pady=(6, 0))
        ttk.Label(top, textvariable=self.process_var).grid(row=3, column=1, sticky="w", pady=(4, 0))

        self.body_notebook = ttk.Notebook(self.root)
        self.body_notebook.grid(row=2, column=0, sticky="nsew", padx=12, pady=(0, 12))

        bridge_tab = ttk.Frame(self.body_notebook, padding=8)
        provider_tab = ttk.Frame(self.body_notebook, padding=8)
        self.body_notebook.add(bridge_tab)
        self.body_notebook.add(provider_tab)
        self.register_tab(self.body_notebook, bridge_tab, "tab.chat_bridge")
        self.register_tab(self.body_notebook, provider_tab, "tab.providers")

        self._build_bridge_tab(bridge_tab)
        self._build_provider_manager_tab(provider_tab)

        bottom = ttk.Frame(self.root, padding=(12, 0, 12, 12))
        bottom.grid(row=3, column=0, sticky="nsew")
        bottom.columnconfigure(0, weight=1)
        bottom.rowconfigure(1, weight=1)

        actions = ttk.Frame(bottom)
        actions.grid(row=0, column=0, sticky="ew")
        self.register_text(ttk.Button(actions, command=self.handle_full_backup), "action.backup_full").pack(side=tk.LEFT)
        self.register_text(
            ttk.Button(actions, command=lambda: self.open_path(self.paths.session_root)),
            "action.open_session_folder",
        ).pack(side=tk.LEFT, padx=(8, 0))
        self.register_text(
            ttk.Button(actions, command=lambda: self.open_path(self.paths.backup_root)),
            "action.open_backup_folder",
        ).pack(side=tk.LEFT, padx=(8, 0))
        ttk.Label(actions, textvariable=self.status_var, style="Muted.TLabel").pack(side=tk.RIGHT)

        log_frame = ttk.LabelFrame(bottom, padding=8)
        self.register_text(log_frame, "section.log")
        log_frame.grid(row=1, column=0, sticky="nsew", pady=(8, 0))
        log_frame.columnconfigure(0, weight=1)
        log_frame.rowconfigure(0, weight=1)
        self.log_text = tk.Text(
            log_frame,
            height=12,
            wrap="word",
            state="disabled",
            font=("Consolas", 10),
            relief="flat",
            borderwidth=0,
        )
        self.log_text.grid(row=0, column=0, sticky="nsew")
        log_scroll = ttk.Scrollbar(log_frame, orient="vertical", command=self.log_text.yview)
        log_scroll.grid(row=0, column=1, sticky="ns")
        self.log_text.configure(yscrollcommand=log_scroll.set)

    def _build_bridge_tab(self, parent: ttk.Frame) -> None:
        parent.columnconfigure(0, weight=1)
        parent.rowconfigure(0, weight=1)

        body = ttk.Panedwindow(parent, orient=tk.HORIZONTAL)
        body.grid(row=0, column=0, sticky="nsew")

        left = ttk.Frame(body, padding=8)
        center = ttk.Frame(body, padding=8)
        right = ttk.Frame(body, padding=8)
        body.add(left, weight=2)
        body.add(center, weight=3)
        body.add(right, weight=3)

        self._build_provider_panel(left)
        self._build_start_panel(center)
        self._build_return_panel(right)

    def _build_provider_manager_tab(self, parent: ttk.Frame) -> None:
        parent.columnconfigure(0, weight=3)
        parent.columnconfigure(1, weight=2)
        parent.rowconfigure(0, weight=1)

        list_frame = ttk.LabelFrame(parent, padding=10)
        self.register_text(list_frame, "section.saved_configured_providers")
        list_frame.grid(row=0, column=0, sticky="nsew", padx=(0, 8))
        list_frame.columnconfigure(0, weight=1)
        list_frame.rowconfigure(0, weight=1)

        columns = ("provider", "url", "key", "saved", "config", "current", "threads")
        self.provider_manager_tree = ttk.Treeview(list_frame, columns=columns, show="headings")
        for column, key, width, anchor in (
            ("provider", "field.provider", 140, "center"),
            ("url", "field.base_url", 300, "w"),
            ("key", "field.api_key", 130, "center"),
            ("saved", "field.saved", 60, "center"),
            ("config", "field.config", 60, "center"),
            ("current", "field.current", 60, "center"),
            ("threads", "field.threads", 70, "center"),
        ):
            self.register_heading(self.provider_manager_tree, column, key)
            self.provider_manager_tree.column(column, width=width, anchor=anchor)
        self.provider_manager_tree.grid(row=0, column=0, sticky="nsew")
        self.provider_manager_tree.bind("<<TreeviewSelect>>", self.on_provider_manager_select)

        manager_scroll = ttk.Scrollbar(list_frame, orient="vertical", command=self.provider_manager_tree.yview)
        manager_scroll.grid(row=0, column=1, sticky="ns")
        self.provider_manager_tree.configure(yscrollcommand=manager_scroll.set)

        detail = ttk.LabelFrame(parent, padding=10)
        self.register_text(detail, "section.provider_details")
        detail.grid(row=0, column=1, sticky="nsew")
        detail.columnconfigure(1, weight=1)

        self.register_text(ttk.Label(detail), "field.provider_name").grid(row=0, column=0, sticky="w")
        ttk.Entry(detail, textvariable=self.manager_provider_var).grid(row=0, column=1, sticky="ew", pady=(0, 6))

        self.register_text(ttk.Label(detail), "field.base_url").grid(row=1, column=0, sticky="w")
        ttk.Entry(detail, textvariable=self.manager_url_var).grid(row=1, column=1, sticky="ew", pady=(0, 6))

        self.register_text(ttk.Label(detail), "field.api_key").grid(row=2, column=0, sticky="w")
        self.manager_key_entry = ttk.Entry(detail, textvariable=self.manager_key_var, show="*")
        self.manager_key_entry.grid(row=2, column=1, sticky="ew", pady=(0, 6))

        self.register_text(
            ttk.Checkbutton(
                detail,
                variable=self.manager_show_key_var,
                command=self.update_key_visibility,
            ),
            "checkbox.show_api_key",
        ).grid(row=3, column=0, columnspan=2, sticky="w", pady=(0, 8))

        buttons = ttk.Frame(detail)
        buttons.grid(row=4, column=0, columnspan=2, sticky="ew")
        buttons.columnconfigure(0, weight=1)
        buttons.columnconfigure(1, weight=1)
        self.register_text(ttk.Button(buttons, command=self.handle_save_provider), "button.save_update").grid(
            row=0, column=0, sticky="ew", padx=(0, 4)
        )
        self.register_text(
            ttk.Button(buttons, command=self.handle_save_and_switch_provider), "button.save_switch"
        ).grid(row=0, column=1, sticky="ew", padx=(4, 0))
        self.register_text(ttk.Button(buttons, command=self.handle_switch_provider), "button.switch_form").grid(
            row=1, column=0, sticky="ew", padx=(0, 4), pady=(8, 0)
        )
        self.register_text(
            ttk.Button(buttons, command=self.handle_delete_saved_provider), "button.delete_saved_entry"
        ).grid(row=1, column=1, sticky="ew", padx=(4, 0), pady=(8, 0))

        extra = ttk.Frame(detail)
        extra.grid(row=5, column=0, columnspan=2, sticky="ew", pady=(10, 0))
        self.register_text(
            ttk.Button(extra, command=lambda: self.open_path(self.paths.provider_store_path.parent)),
            "button.open_provider_store",
        ).pack(side=tk.LEFT)
        self.register_text(
            ttk.Button(extra, command=self.load_current_provider_into_form),
            "button.load_current_provider",
        ).pack(side=tk.LEFT, padx=(8, 0))

        ttk.Label(
            detail,
            textvariable=self.provider_note_var,
            style="Muted.TLabel",
            wraplength=360,
            justify="left",
        ).grid(row=6, column=0, columnspan=2, sticky="w", pady=(12, 0))

    def update_key_visibility(self) -> None:
        self.manager_key_entry.configure(show="" if self.manager_show_key_var.get() else "*")

    def _build_provider_panel(self, parent: ttk.Frame) -> None:
        parent.columnconfigure(0, weight=1)
        parent.rowconfigure(1, weight=1)
        self.register_text(ttk.Label(parent, font=("Microsoft YaHei UI", 11, "bold")), "section.providers").grid(
            row=0, column=0, sticky="w"
        )

        columns = ("provider", "threads", "config", "current")
        self.provider_tree = ttk.Treeview(parent, columns=columns, show="headings", height=16)
        for column, key, width in (
            ("provider", "field.provider", 180),
            ("threads", "field.threads", 70),
            ("config", "field.in_config", 80),
            ("current", "field.current", 70),
        ):
            self.register_heading(self.provider_tree, column, key)
            self.provider_tree.column(column, width=width, anchor="center")
        self.provider_tree.grid(row=1, column=0, sticky="nsew", pady=(8, 0))

        scroll = ttk.Scrollbar(parent, orient="vertical", command=self.provider_tree.yview)
        scroll.grid(row=1, column=1, sticky="ns", pady=(8, 0))
        self.provider_tree.configure(yscrollcommand=scroll.set)

        self.register_text(ttk.Label(parent, style="Muted.TLabel"), "note.bridge_hint").grid(
            row=2, column=0, sticky="w", pady=(8, 0)
        )

    def _build_start_panel(self, parent: ttk.Frame) -> None:
        parent.columnconfigure(1, weight=1)

        start_frame = ttk.LabelFrame(parent, padding=10)
        self.register_text(start_frame, "section.start_bridge_session")
        start_frame.grid(row=0, column=0, sticky="ew")
        start_frame.columnconfigure(1, weight=1)

        self.register_text(ttk.Label(start_frame), "field.source_provider").grid(row=0, column=0, sticky="w")
        self.start_source_combo = ttk.Combobox(start_frame, textvariable=self.start_source_var, state="readonly")
        self.start_source_combo.grid(row=0, column=1, sticky="ew", pady=(0, 6))

        self.register_text(ttk.Label(start_frame), "field.target_provider").grid(row=1, column=0, sticky="w")
        self.start_target_combo = ttk.Combobox(start_frame, textvariable=self.start_target_var)
        self.start_target_combo.grid(row=1, column=1, sticky="ew", pady=(0, 6))

        self.register_text(ttk.Checkbutton(start_frame, variable=self.start_switch_var), "checkbox.switch_current_to_target_after_start").grid(
            row=2, column=0, columnspan=2, sticky="w", pady=(0, 8)
        )

        self.register_text(ttk.Button(start_frame, command=self.handle_start_session), "button.start_session").grid(
            row=3, column=0, columnspan=2, sticky="ew"
        )

        direct_frame = ttk.LabelFrame(parent, padding=10)
        self.register_text(direct_frame, "section.direct_move")
        direct_frame.grid(row=1, column=0, sticky="ew", pady=(12, 0))
        direct_frame.columnconfigure(1, weight=1)

        self.register_text(ttk.Label(direct_frame), "field.source_provider").grid(row=0, column=0, sticky="w")
        self.direct_source_combo = ttk.Combobox(direct_frame, textvariable=self.direct_source_var, state="readonly")
        self.direct_source_combo.grid(row=0, column=1, sticky="ew", pady=(0, 6))

        self.register_text(ttk.Label(direct_frame), "field.target_provider").grid(row=1, column=0, sticky="w")
        self.direct_target_combo = ttk.Combobox(direct_frame, textvariable=self.direct_target_var)
        self.direct_target_combo.grid(row=1, column=1, sticky="ew", pady=(0, 6))

        self.register_text(ttk.Checkbutton(direct_frame, variable=self.direct_switch_var), "checkbox.switch_current_to_target_after_move").grid(
            row=2, column=0, columnspan=2, sticky="w", pady=(0, 8)
        )

        self.register_text(
            ttk.Button(direct_frame, command=self.handle_direct_move),
            "button.move_all_source_threads",
        ).grid(row=3, column=0, columnspan=2, sticky="ew")

    def _build_return_panel(self, parent: ttk.Frame) -> None:
        parent.columnconfigure(0, weight=1)
        parent.rowconfigure(1, weight=1)

        self.register_text(ttk.Label(parent, font=("Microsoft YaHei UI", 11, "bold")), "section.bridge_sessions").grid(
            row=0, column=0, sticky="w"
        )

        columns = ("session", "source", "target", "status", "moved")
        self.session_tree = ttk.Treeview(parent, columns=columns, show="headings", height=14)
        for column, key, width in (
            ("session", "field.session", 180),
            ("source", "field.source", 100),
            ("target", "field.target", 100),
            ("status", "field.status", 80),
            ("moved", "field.moved", 70),
        ):
            self.register_heading(self.session_tree, column, key)
            self.session_tree.column(column, width=width, anchor="center")
        self.session_tree.grid(row=1, column=0, sticky="nsew", pady=(8, 0))
        self.session_tree.bind("<<TreeviewSelect>>", self.on_session_select)

        scroll = ttk.Scrollbar(parent, orient="vertical", command=self.session_tree.yview)
        scroll.grid(row=1, column=1, sticky="ns", pady=(8, 0))
        self.session_tree.configure(yscrollcommand=scroll.set)

        return_frame = ttk.LabelFrame(parent, padding=10)
        self.register_text(return_frame, "section.return_session")
        return_frame.grid(row=2, column=0, sticky="ew", pady=(12, 0))
        return_frame.columnconfigure(1, weight=1)

        self.register_text(ttk.Label(return_frame), "field.selected_session").grid(row=0, column=0, sticky="w")
        ttk.Entry(return_frame, textvariable=self.return_session_var, state="readonly").grid(row=0, column=1, sticky="ew", pady=(0, 6))

        self.register_text(ttk.Checkbutton(return_frame, variable=self.return_new_threads_var), "checkbox.move_new_target_threads").grid(
            row=1, column=0, columnspan=2, sticky="w"
        )

        self.register_text(ttk.Checkbutton(return_frame, variable=self.return_switch_var), "checkbox.switch_current_back_after_return").grid(
            row=2, column=0, columnspan=2, sticky="w", pady=(0, 8)
        )

        self.register_text(
            ttk.Button(return_frame, command=self.handle_return_session),
            "button.return_selected_session",
        ).grid(row=3, column=0, columnspan=2, sticky="ew")

    def log(self, message: str) -> None:
        append_log(self.paths, message)
        self.log_text.configure(state="normal")
        self.log_text.insert("end", f"[{iso_now()}] {message}\n")
        self.log_text.see("end")
        self.log_text.configure(state="disabled")
        self.status_var.set(message)

    def open_path(self, path: Path) -> None:
        try:
            os.startfile(str(path))  # type: ignore[attr-defined]
        except Exception as exc:
            self.show_error("error.open_path", path=path, error=exc)

    def ensure_codex_closed(self, action_label: str) -> bool:
        processes = detect_codex_processes()
        if processes:
            self.show_warning("warn.codex_active", processes="\n".join(processes), action_label=action_label)
            return False
        return True

    def ensure_migration_ready(self) -> bool:
        missing = []
        if not self.paths.config_path.exists():
            missing.append(str(self.paths.config_path))
        if not self.paths.db_path.exists():
            missing.append(str(self.paths.db_path))
        if missing:
            self.show_error("error.missing_required_files", files="\n".join(missing))
            return False
        if not self.ensure_codex_closed(self.t("action_label.running_migration")):
            return False
        try:
            ensure_sqlite_ready(self.paths.db_path)
        except Exception as exc:
            self.show_error("error.sqlite_not_ready", error=exc)
            return False
        return True

    def ensure_provider_switch_ready(self) -> bool:
        if not self.paths.config_path.exists():
            self.show_error("error.missing_required_config", path=self.paths.config_path)
            return False
        if not self.ensure_codex_closed(self.t("action_label.switching_providers")):
            return False
        return True

    def selected_provider_name(self) -> str:
        selection = self.provider_manager_tree.selection() if hasattr(self, "provider_manager_tree") else ()
        if selection:
            return str(selection[0])
        return self.manager_provider_var.get().strip()

    def select_provider_row(self, provider: str) -> None:
        provider_name = provider.strip()
        if not provider_name or not hasattr(self, "provider_manager_tree"):
            return
        if provider_name not in self.provider_manager_tree.get_children():
            return
        self.provider_manager_tree.selection_set(provider_name)
        self.provider_manager_tree.focus(provider_name)
        self.provider_manager_tree.see(provider_name)

    def populate_provider_form(self, provider: str) -> None:
        provider_name = provider.strip()
        if not provider_name:
            return
        entry = self.provider_catalog.get(provider_name)
        self.manager_provider_var.set(provider_name)
        if not entry:
            return

        self.manager_url_var.set(str(entry.get("saved_base_url") or entry.get("base_url") or ""))
        self.manager_key_var.set(str(entry.get("saved_api_key") or entry.get("api_key") or ""))

        key_source = str(entry.get("key_source") or "")
        if key_source == "runtime":
            self.set_provider_note("note.provider_runtime_key")
        elif entry.get("in_store"):
            self.set_provider_note("note.provider_saved_entry")
        else:
            self.set_provider_note("note.provider_config_only")

    def load_current_provider_into_form(self) -> None:
        current = self.current_provider_name.strip()
        if current:
            self.populate_provider_form(current)
            self.select_provider_row(current)

    def collect_provider_form(self) -> tuple[str, str, str]:
        return (
            self.manager_provider_var.get().strip(),
            self.manager_url_var.get().strip(),
            self.manager_key_var.get().strip(),
        )

    def refresh_all(self) -> None:
        previous_provider = self.selected_provider_name()
        previous_session = self.return_session_var.get().strip()

        self.provider_catalog = build_provider_catalog(self.paths)
        current_provider = get_current_provider(self.paths.config_path)
        self.current_provider_name = current_provider
        config_providers = {name for name, entry in self.provider_catalog.items() if entry.get("in_config")}
        all_providers = sorted(self.provider_catalog.keys())
        self.provider_values = all_providers
        self.current_provider_var.set(current_provider or self.t("misc.none"))

        processes = detect_codex_processes()
        self.process_var.set(", ".join(processes) if processes else self.t("misc.no_process_detected"))

        for item in self.provider_tree.get_children():
            self.provider_tree.delete(item)
        for provider in all_providers:
            entry = self.provider_catalog.get(provider, {})
            self.provider_tree.insert(
                "",
                "end",
                iid=provider,
                values=(
                    provider,
                    entry.get("threads", 0),
                    self.yes_no(provider in config_providers),
                    self.yes_no(provider == current_provider, blank_if_false=True),
                ),
            )

        for combo in (
            self.start_source_combo,
            self.start_target_combo,
            self.direct_source_combo,
            self.direct_target_combo,
        ):
            combo["values"] = all_providers

        if not self.start_source_var.get() and current_provider:
            self.start_source_var.set(current_provider)
        if not self.direct_source_var.get() and current_provider:
            self.direct_source_var.set(current_provider)

        for item in self.provider_manager_tree.get_children():
            self.provider_manager_tree.delete(item)
        for provider in all_providers:
            entry = self.provider_catalog.get(provider, {})
            key_display = mask_secret(str(entry.get("api_key") or ""))
            if key_display and entry.get("key_source") == "runtime":
                key_display += f" ({self.t('misc.runtime_suffix')})"
            self.provider_manager_tree.insert(
                "",
                "end",
                iid=provider,
                values=(
                    provider,
                    entry.get("base_url", ""),
                    key_display,
                    self.yes_no(bool(entry.get("in_store"))),
                    self.yes_no(bool(entry.get("in_config"))),
                    self.yes_no(bool(entry.get("current")), blank_if_false=True),
                    entry.get("threads", 0),
                ),
            )

        self.session_records = load_session_records(self.paths)
        for item in self.session_tree.get_children():
            self.session_tree.delete(item)
        for record in self.session_records:
            session_id = str(record.get("session_id", ""))
            self.session_tree.insert(
                "",
                "end",
                iid=session_id,
                values=(
                    session_id,
                    record.get("source_provider", ""),
                    record.get("target_provider", ""),
                    self.format_session_status(record.get("status", "")),
                    record.get("moved_thread_count", 0),
                ),
            )

        if previous_provider and previous_provider in self.provider_catalog:
            self.select_provider_row(previous_provider)
            self.populate_provider_form(previous_provider)
        elif current_provider:
            self.load_current_provider_into_form()
        else:
            self.set_provider_note("note.provider_default")

        if previous_session and previous_session in {str(record.get("session_id", "")) for record in self.session_records}:
            self.return_session_var.set(previous_session)
            self.session_tree.selection_set(previous_session)
            self.session_tree.focus(previous_session)

        self.status_var.set(self.t("status.refreshed"))

    def selected_session_record(self) -> dict | None:
        selected = self.return_session_var.get().strip()
        if not selected:
            return None
        for record in self.session_records:
            if record.get("session_id") == selected:
                return record
        return None

    def on_session_select(self, _event=None) -> None:
        selection = self.session_tree.selection()
        if selection:
            self.return_session_var.set(selection[0])

    def on_provider_manager_select(self, _event=None) -> None:
        provider = self.selected_provider_name()
        if provider:
            self.populate_provider_form(provider)

    def handle_save_provider(self) -> None:
        provider, base_url, api_key = self.collect_provider_form()
        if not provider or not base_url:
            self.show_error("error.enter_provider_and_url")
            return
        try:
            result = save_provider_definition(self.paths, provider, base_url, api_key)
            self.log_key(
                "log.saved_provider",
                provider=result["provider"],
                api_key_mode=self.t("phrase.with_api_key" if result["has_api_key"] else "phrase.without_api_key"),
            )
            self.refresh_all()
            self.select_provider_row(provider)
            self.show_info("success.saved_provider_entry", provider=provider)
        except Exception as exc:
            self.show_error("error.save_provider_failed", error=self.localize_error(exc))

    def handle_save_and_switch_provider(self) -> None:
        provider, base_url, api_key = self.collect_provider_form()
        if not provider or not base_url:
            self.show_error("error.enter_provider_and_url")
            return
        try:
            save_provider_definition(self.paths, provider, base_url, api_key)
        except Exception as exc:
            self.show_error("error.save_provider_failed", error=self.localize_error(exc))
            return
        self.handle_switch_provider()

    def handle_switch_provider(self) -> None:
        provider, base_url, api_key = self.collect_provider_form()
        if not provider:
            self.show_error("error.choose_or_enter_provider")
            return
        if not base_url:
            self.show_error("error.enter_url_before_switch")
            return
        if not self.ensure_provider_switch_ready():
            return
        if not api_key:
            proceed = self.ask_yes_no("confirm.no_api_key_continue")
            if not proceed:
                return

        proceed = self.ask_yes_no("confirm.switch_provider", provider=provider, base_url=base_url)
        if not proceed:
            return
        try:
            result = switch_provider(self.paths, provider, base_url, api_key)
            self.log_key(
                "log.switched_provider",
                provider=result["provider"],
                auth_mode=self.t("phrase.updated_auth" if result["auth_updated"] else "phrase.config_only"),
            )
            self.refresh_all()
            self.select_provider_row(provider)
            self.show_info(
                "success.provider_switched",
                provider=result["provider"],
                auth_updated=self.yes_no(bool(result["auth_updated"])),
                config_path=result["config_path"],
                backup_dir=result["backup_dir"],
            )
        except Exception as exc:
            self.show_error("error.switch_provider_failed", error=self.localize_error(exc))

    def handle_delete_saved_provider(self) -> None:
        provider = self.selected_provider_name()
        if not provider:
            self.show_error("error.choose_provider_first")
            return
        entry = self.provider_catalog.get(provider, {})
        if not entry.get("in_store"):
            self.show_error("error.provider_has_no_saved_entry")
            return
        proceed = self.ask_yes_no("confirm.delete_saved_provider", provider=provider)
        if not proceed:
            return

        removed = delete_provider_entry(self.paths.provider_store_path, provider)
        if not removed:
            self.show_error("error.saved_provider_missing")
            return
        self.log_key("log.deleted_saved_provider", provider=provider)
        if self.manager_provider_var.get().strip() == provider:
            self.manager_key_var.set("")
        self.refresh_all()

    def handle_full_backup(self) -> None:
        if detect_codex_processes():
            proceed = self.ask_yes_no("confirm.full_backup_while_active")
            if not proceed:
                return
        try:
            target = backup_full_codex(self.paths)
            self.log_key("log.full_backup_created", target=target)
            self.show_info("success.full_backup_created", target=target)
        except Exception as exc:
            self.show_error("error.full_backup_failed", error=self.localize_error(exc))

    def handle_start_session(self) -> None:
        source = self.start_source_var.get().strip()
        target = self.start_target_var.get().strip()
        if not source or not target:
            self.show_error("error.choose_source_target")
            return
        if source == target:
            self.show_error("error.source_target_different")
            return
        if not self.ensure_migration_ready():
            return
        proceed = self.ask_yes_no("confirm.start_bridge_session", source=source, target=target)
        if not proceed:
            return
        try:
            record = create_bridge_session(self.paths, source, target, self.start_switch_var.get())
            self.log_key(
                "log.started_bridge_session",
                session_id=record["session_id"],
                moved_thread_count=record["moved_thread_count"],
                source=source,
                target=target,
            )
            self.refresh_all()
            self.return_session_var.set(record["session_id"])
            self.show_info(
                "success.bridge_session_started",
                session_id=record["session_id"],
                moved_thread_count=record["moved_thread_count"],
            )
        except Exception as exc:
            self.show_error("error.start_bridge_session_failed", error=self.localize_error(exc))

    def handle_return_session(self) -> None:
        record = self.selected_session_record()
        if not record:
            self.show_error("error.select_bridge_session_first")
            return
        if not self.ensure_migration_ready():
            return
        if record.get("status") != "open":
            self.show_error("error.bridge_session_completed")
            return
        source = record.get("source_provider")
        target = record.get("target_provider")
        proceed = self.ask_yes_no("confirm.return_bridge_session", source=source, target=target)
        if not proceed:
            return
        try:
            updated = complete_bridge_session(
                self.paths,
                record,
                self.return_switch_var.get(),
                self.return_new_threads_var.get(),
            )
            self.log_key(
                "log.completed_bridge_session",
                session_id=updated["session_id"],
                restored_thread_count=updated.get("restored_thread_count", 0),
                new_thread_count=updated.get("new_thread_count", 0),
            )
            self.refresh_all()
            self.show_info(
                "success.bridge_session_returned",
                restored_thread_count=updated.get("restored_thread_count", 0),
                new_thread_count=updated.get("new_thread_count", 0),
            )
        except Exception as exc:
            self.show_error("error.return_bridge_session_failed", error=self.localize_error(exc))

    def handle_direct_move(self) -> None:
        source = self.direct_source_var.get().strip()
        target = self.direct_target_var.get().strip()
        if not source or not target:
            self.show_error("error.choose_source_target")
            return
        if source == target:
            self.show_error("error.source_target_different")
            return
        if not self.ensure_migration_ready():
            return
        proceed = self.ask_yes_no("confirm.direct_move", source=source, target=target)
        if not proceed:
            return
        try:
            result = direct_move(self.paths, source, target, self.direct_switch_var.get())
            self.log_key(
                "log.direct_move_finished",
                changed=result["changed"],
                source=source,
                target=target,
            )
            self.refresh_all()
            self.show_info(
                "success.direct_move_finished",
                changed=result["changed"],
                backup_dir=result["backup_dir"],
            )
        except Exception as exc:
            self.show_error("error.direct_move_failed", error=self.localize_error(exc))


def main() -> int:
    root = tk.Tk()
    app = BridgeApp(root)
    app.log_key("status.app_started")
    root.mainloop()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())