from __future__ import annotations

import base64
import hashlib
import json
import logging
import mimetypes
import os
import re
import socket
import shutil
import sqlite3
import subprocess
import tempfile
import threading
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from urllib.parse import quote

from .config import Settings

try:
    import tomllib
except ModuleNotFoundError:  # pragma: no cover - Python < 3.11 fallback.
    tomllib = None


logger = logging.getLogger(__name__)

CODEX_GLOBAL_STATE_PATH = Path.home() / ".codex" / ".codex-global-state.json"
CODEXMOBILE_IMAGE_CACHE_DIR = Path.home() / ".codex" / "codexmobile-image-cache"
CODEXMOBILE_ATTACH_CACHE_DIR = Path.home() / ".codex" / "codexmobile-attachment-cache"
CODEX_APP_PATH = Path("/Applications/Codex.app")
CODEX_APP_BIN = CODEX_APP_PATH / "Contents" / "Resources" / "codex"


class CodexLiveBridgeError(RuntimeError):
    pass


@dataclass
class LiveThreadSelection:
    thread_id: str
    thread_name: str | None
    cwd: str | None
    updated_at: int | None
    status: str | None

    @property
    def project_title(self) -> str:
        if self.cwd:
            return Path(self.cwd).name
        if self.thread_name:
            return self.thread_name
        return "目前專案"


class CodexLiveBridge:
    def __init__(self, settings: Settings) -> None:
        self.settings = settings
        self.repo_root = Path(__file__).resolve().parents[2]
        self.probe_script = settings.codexbot_live_probe_script
        self.codex_bin = settings.codexbot_live_codex_bin
        self.node_bin = self._resolve_node_bin(settings.codexbot_live_node_bin)
        self.url = settings.codexbot_live_app_server_url
        self._active_app_server_url = self.url
        self._active_app_server_url_checked_at = 0.0
        self._active_app_server_url_ttl = 20.0
        self.thread_cwd = str(settings.codexbot_live_thread_cwd)
        self.project_root = settings.codexbot_project_root
        self.thread_name = settings.codexbot_live_thread_name
        self.spawn_server = settings.codexbot_live_spawn_app_server
        self.timeout = settings.codexbot_live_timeout_seconds
        self.snapshot_app_server_timeout = max(
            0.5,
            min(float(settings.codexbot_live_snapshot_app_server_timeout_seconds), self.timeout),
        )
        self.selection_path = settings.codexbot_live_selection_path
        self.projects_path = settings.codexbot_live_projects_path
        self._thread_cache: dict | None = None
        self._thread_cache_at = 0.0
        self._thread_cache_ttl = 5.0
        self._raw_threads_cache: list[dict] = []
        self._raw_threads_cache_at = 0.0
        self._raw_threads_cache_ttl = 600.0
        self._raw_threads_refresh_interval = 30.0
        self._raw_threads_failed_at = 0.0
        self._raw_threads_failed_backoff = 20.0
        self._raw_threads_lock = threading.Lock()
        self._runtime_cache: dict | None = None
        self._runtime_cache_at = 0.0
        self._runtime_cache_ttl = 60.0
        self._usage_cache: dict | None = None
        self._usage_cache_at = 0.0
        self._usage_cache_ttl = 300.0
        self._runtime_failed_at = 0.0
        self._runtime_failed_backoff = 20.0
        self._app_server_failed_at = 0.0
        self._app_server_failed_backoff = max(10.0, min(60.0, self.snapshot_app_server_timeout * 4))
        self._app_server_recovery_attempted_at = 0.0
        self._app_server_recovery_cooldown = 45.0
        self._generation_settings_cache: dict[str, tuple[float, dict[str, Any]]] = {}
        self._generation_settings_cache_ttl = 60.0
        self._generation_settings_failed_at: dict[str, float] = {}
        self._generation_settings_failed_backoff = 20.0
        self._snapshot_cache: dict[str, tuple[float, dict]] = {}
        self._snapshot_cache_ttl = 4.0
        self._snapshot_stale_ttl = 600.0
        self._snapshot_lock = threading.Lock()
        self._rollout_thread_cache: dict[str, tuple[float, tuple[int, int], dict[str, Any]]] = {}
        self._rollout_thread_cache_ttl = 20.0
        self._workspace_diff_cache: dict[str, tuple[float, dict[str, Any] | None]] = {}
        self._workspace_diff_cache_ttl = 10.0
        self._pending_selection_ttl = 120.0
        self._desktop_refresh_lock = threading.Lock()
        self._last_desktop_refresh_at = 0.0

    def get_status_payload(self) -> dict:
        thread = self._get_selected_thread()
        updated_at = self._iso_from_epoch(thread.updated_at)
        threads = self.list_threads()
        projects = self._projects_from_thread_projects(threads)
        project_title = self._project_title_for_selection(thread)
        generation_settings = self._cached_generation_settings_or_fallback(thread.thread_id)
        runtime = self._cached_runtime_or_fallback(thread.thread_id, "status-fast-path")
        return {
            "pending_count": 0,
            "last_message": None,
            "state": {
                "messages": [],
                **self._server_clock_payload(),
                "updated_at": updated_at,
                "thread_id": thread.thread_id,
                "thread_name": thread.thread_name,
                "cwd": thread.cwd,
                "thread_status": thread.status,
                "project_title": project_title,
                "generation_settings": generation_settings,
                "runtime": runtime,
                "capabilities": runtime.get("capabilities") if isinstance(runtime, dict) else None,
                "host_status": self._codex_host_status_payload(),
                "models": runtime.get("models") if isinstance(runtime, dict) else [],
                "threads": threads,
                "projects": projects,
            },
        }

    def get_usage_payload(self) -> dict:
        now = time.time()
        if self._usage_cache and now - self._usage_cache_at < self._usage_cache_ttl:
            return {
                **self._usage_cache,
                **self._server_clock_payload(),
                "stale": False,
            }
        try:
            payload = self._run_probe("usage", timeout=24.0, bypass_backoff=True)
            account = payload.get("account") if isinstance(payload, dict) else None
            rate_limits = payload.get("rateLimits") if isinstance(payload, dict) else None
            result = {
                "ok": True,
                **self._server_clock_payload(),
                "account": account if isinstance(account, dict) else None,
                "rate_limits": rate_limits if isinstance(rate_limits, dict) else None,
                "raw_usage": payload if isinstance(payload, dict) else None,
                "stale": False,
            }
            self._usage_cache = result
            self._usage_cache_at = now
            return result
        except Exception as exc:
            if self._usage_cache:
                return {
                    **self._usage_cache,
                    **self._server_clock_payload(),
                    "ok": True,
                    "stale": True,
                    "error": str(exc),
                }
            return {
                "ok": False,
                **self._server_clock_payload(),
                "account": None,
                "rate_limits": None,
                "raw_usage": None,
                "stale": False,
                "error": str(exc),
            }

    def get_messages_payload(self, limit: int = 500) -> dict:
        thread = self._read_selected_thread()
        messages = self._flatten_thread_messages(thread)
        messages = self._trim_timeline_for_mobile(messages, limit)
        project_title = self._project_title_for_thread(thread) or (
            Path(thread["cwd"]).name if thread.get("cwd") else (thread.get("name") or "目前專案")
        )
        return {
            "messages": messages,
            **self._server_clock_payload(),
            "thread_id": thread["id"],
            "thread_name": thread.get("name"),
            "cwd": thread.get("cwd"),
            "project_title": project_title,
        }

    def get_snapshot_payload(self, limit: int = 500) -> dict:
        cache_key = f"selected:{int(limit)}"
        cached = self._get_cached_snapshot(cache_key)
        if cached is not None:
            return cached
        if not self._snapshot_lock.acquire(blocking=False):
            cached = self._get_cached_snapshot(cache_key, allow_stale=True)
            if cached is None:
                cached = self._get_best_cached_snapshot(allow_stale=True)
            if cached is not None:
                return cached
            return self._build_minimal_snapshot_payload(limit=limit, reason="snapshot-refresh-busy")
        try:
            return self._store_snapshot_cache(cache_key, self._build_snapshot_payload(limit=limit))
        except CodexLiveBridgeError as exc:
            logger.warning("Codex selected snapshot refresh failed: %s", exc)
            cached = self._get_cached_snapshot(cache_key, allow_stale=True)
            if cached is None:
                cached = self._get_best_cached_snapshot(allow_stale=True)
            if cached is not None:
                return cached
            return self._build_minimal_snapshot_payload(limit=limit, reason="snapshot-refresh-failed")
        except Exception:
            logger.exception("Codex selected snapshot refresh crashed")
            cached = self._get_cached_snapshot(cache_key, allow_stale=True)
            if cached is None:
                cached = self._get_best_cached_snapshot(allow_stale=True)
            if cached is not None:
                return cached
            return self._build_minimal_snapshot_payload(limit=limit, reason="snapshot-refresh-crashed")
        finally:
            self._snapshot_lock.release()

    def _build_snapshot_payload(self, limit: int = 500) -> dict:
        selected = self._get_selected_thread_fast()
        rollout_limit = max(360, min(max(int(limit) * 3, int(limit) + 120), 520))
        thread = (
            self._read_thread_from_rollout(
                selected.thread_id,
                max_items=rollout_limit,
                max_bytes=128 * 1024 * 1024,
            )
            or self._read_thread_by_id(
                selected.thread_id,
                timeout=self.snapshot_app_server_timeout,
                allow_rollout_fallback=False,
            )
            or {
                "id": selected.thread_id,
                "name": selected.thread_name,
                "cwd": selected.cwd,
                "updatedAt": selected.updated_at,
                "status": {"type": selected.status or "idle"},
                "turns": [],
            }
        )
        thread_id = str(thread.get("id") or selected.thread_id)
        selection_data = self._load_selection_data()
        threads = self._cached_or_selected_threads(selection_data)
        projects = self._projects_from_thread_projects(threads)
        messages = self._flatten_thread_messages(thread)
        messages = self._trim_timeline_for_mobile(messages, limit)
        timeline_source = thread.get("_source") or "codex-app-server"
        if timeline_source == "codex-rollout-jsonl" and self._rollout_timeline_is_incomplete(messages):
            logger.warning(
                "Ignoring incomplete rollout snapshot for thread %s: %s item(s)",
                thread_id,
                len(messages),
            )
            messages = []
            timeline_source = "codex-rollout-incomplete"
        updated_at = self._iso_from_epoch(thread.get("updatedAt") or selected.updated_at)
        cwd = thread.get("cwd") or selected.cwd
        thread_name = thread.get("name") or selected.thread_name
        project_title = self._project_title_for_thread(thread) or self._project_title_for_selection(selected) or "目前專案"
        generation_settings = self._cached_generation_settings_or_fallback(thread_id)
        runtime = self._cached_runtime_or_fallback(thread_id, "snapshot-fast-path")
        return {
            "pending_count": 0,
            "last_message": None,
            "state": {
                "messages": messages,
                "timeline": messages,
                "timeline_source": timeline_source,
                **self._server_clock_payload(),
                "updated_at": updated_at,
                "thread_id": thread_id,
                "thread_name": thread_name,
                "cwd": cwd,
                "project_root": str(self.project_root),
                "mobile_default_cwd": self.thread_cwd,
                "thread_status": self._status_label(thread.get("status")) or selected.status,
                "project_title": project_title,
                "generation_settings": generation_settings,
                "runtime": runtime,
                "capabilities": runtime.get("capabilities") if isinstance(runtime, dict) else None,
                "host_status": self._codex_host_status_payload(),
                "models": runtime.get("models") if isinstance(runtime, dict) else [],
                "threads": threads,
                "projects": projects,
            },
        }

    def get_thread_snapshot_payload(self, thread_id: str, limit: int = 500) -> dict:
        cache_key = f"thread:{thread_id}:{int(limit)}"
        cached = self._get_cached_snapshot(cache_key)
        if cached is not None:
            return cached
        if not self._snapshot_lock.acquire(blocking=False):
            cached = self._get_cached_snapshot(cache_key, allow_stale=True)
            if cached is None:
                cached = self._get_best_cached_snapshot(thread_id=thread_id, allow_stale=True)
            if cached is not None:
                return cached
            return self._build_minimal_snapshot_payload(thread_id=thread_id, limit=limit, reason="thread-snapshot-refresh-busy")
        try:
            return self._store_snapshot_cache(cache_key, self._build_thread_snapshot_payload(thread_id, limit=limit))
        except CodexLiveBridgeError as exc:
            logger.warning("Codex thread snapshot refresh failed for %s: %s", thread_id, exc)
            cached = self._get_cached_snapshot(cache_key, allow_stale=True)
            if cached is None:
                cached = self._get_best_cached_snapshot(thread_id=thread_id, allow_stale=True)
            if cached is not None:
                return cached
            return self._build_minimal_snapshot_payload(thread_id=thread_id, limit=limit, reason="thread-snapshot-refresh-failed")
        except Exception:
            logger.exception("Codex thread snapshot refresh crashed for %s", thread_id)
            cached = self._get_cached_snapshot(cache_key, allow_stale=True)
            if cached is None:
                cached = self._get_best_cached_snapshot(thread_id=thread_id, allow_stale=True)
            if cached is not None:
                return cached
            return self._build_minimal_snapshot_payload(thread_id=thread_id, limit=limit, reason="thread-snapshot-refresh-crashed")
        finally:
            self._snapshot_lock.release()

    def _build_thread_snapshot_payload(self, thread_id: str, limit: int = 500) -> dict:
        selected = self._get_selected_thread_fast()
        rollout_limit = max(360, min(max(int(limit) * 3, int(limit) + 120), 520))
        thread = (
            self._read_thread_from_rollout(
                thread_id,
                max_items=rollout_limit,
                max_bytes=32 * 1024 * 1024,
            )
            or self._read_thread_by_id(
                thread_id,
                timeout=min(self.snapshot_app_server_timeout, 1.5),
                allow_rollout_fallback=False,
            )
            or self._synthetic_thread_from_local_state(thread_id)
        )
        if not thread:
            raise CodexLiveBridgeError(f"找不到指定的會話：{thread_id}")

        threads = self._cached_or_selected_threads(self._load_selection_data())
        selected_project_title = self._project_title_for_thread(thread) or thread.get("name") or "目前專案"
        normalized_threads: list[dict[str, Any]] = []
        matched = False
        for item in threads:
            same_id = item.get("thread_id") == thread_id
            same_title = item.get("project_title") == selected_project_title
            is_selected = bool(same_id or same_title)
            matched = matched or is_selected
            normalized_threads.append({**item, "selected": is_selected})

        if not matched:
            normalized_threads.insert(
                0,
                {
                    "thread_id": thread_id,
                    "thread_name": thread.get("name"),
                    "cwd": thread.get("cwd"),
                    "project_title": selected_project_title,
                    "updated_at": self._iso_from_epoch(thread.get("updatedAt")),
                    "status": self._status_label(thread.get("status")),
                    "selected": True,
                },
            )

        projects = self._projects_from_thread_projects(normalized_threads)
        messages = self._flatten_thread_messages(thread)
        messages = self._trim_timeline_for_mobile(messages, limit)
        timeline_source = thread.get("_source") or "codex-app-server"
        if timeline_source == "codex-rollout-jsonl" and self._rollout_timeline_is_incomplete(messages):
            logger.warning(
                "Ignoring incomplete rollout thread snapshot for thread %s: %s item(s)",
                thread_id,
                len(messages),
            )
            messages = []
            timeline_source = "codex-rollout-incomplete"
        updated_at = self._iso_from_epoch(thread.get("updatedAt") or selected.updated_at)
        generation_settings = self._cached_generation_settings_or_fallback(thread_id)
        runtime = self._cached_runtime_or_fallback(thread_id, "thread-snapshot-fast-path")
        return {
            "pending_count": 0,
            "last_message": None,
            "state": {
                "messages": messages,
                "timeline": messages,
                "timeline_source": timeline_source,
                **self._server_clock_payload(),
                "updated_at": updated_at,
                "thread_id": thread_id,
                "thread_name": thread.get("name"),
                "cwd": thread.get("cwd"),
                "project_root": str(self.project_root),
                "mobile_default_cwd": self.thread_cwd,
                "thread_status": self._status_label(thread.get("status")),
                "project_title": selected_project_title,
                "generation_settings": generation_settings,
                "runtime": runtime,
                "capabilities": runtime.get("capabilities") if isinstance(runtime, dict) else None,
                "host_status": self._codex_host_status_payload(),
                "models": runtime.get("models") if isinstance(runtime, dict) else [],
                "threads": normalized_threads,
                "projects": projects,
            },
        }

    @staticmethod
    def _trim_timeline_for_mobile(messages: list[dict], limit: int) -> list[dict]:
        if limit <= 0 or len(messages) <= limit:
            return messages

        included: set[int] = set()
        included_turns: set[str] = set()
        visible_count = 0

        # Count conversational/file-change items against the mobile budget, but
        # do not let command events evict user/assistant messages from snapshot.
        for index in range(len(messages) - 1, -1, -1):
            item = messages[index]
            if item.get("kind") == "commandExecution" or item.get("command_execution"):
                continue
            included.add(index)
            turn_id = str(item.get("turn_id") or "")
            if turn_id:
                included_turns.add(turn_id)
            visible_count += 1
            if visible_count >= limit:
                break

        if not included:
            return messages[-limit:]

        recent_user_turns = 0
        for index in range(len(messages) - 1, -1, -1):
            item = messages[index]
            if item.get("role") != "user":
                continue
            included.add(index)
            turn_id = str(item.get("turn_id") or "")
            if turn_id:
                included_turns.add(turn_id)
            recent_user_turns += 1
            if recent_user_turns >= 5:
                break

        max_commands_per_turn = max(8, min(24, max(limit // 4, 1)))
        command_counts: dict[str, int] = {}
        for index in range(len(messages) - 1, -1, -1):
            item = messages[index]
            if item.get("kind") != "commandExecution" and not item.get("command_execution"):
                continue
            turn_id = str(item.get("turn_id") or "")
            if not turn_id or turn_id not in included_turns:
                continue
            count = command_counts.get(turn_id, 0)
            if count >= max_commands_per_turn:
                continue
            included.add(index)
            command_counts[turn_id] = count + 1

        return [messages[index] for index in sorted(included)]

    def get_runtime_payload(self, thread_id: str | None = None, refresh: bool = False) -> dict:
        target_thread_id = thread_id or self._get_selected_thread().thread_id
        if (
            not refresh
            and self._runtime_cache
            and time.time() - self._runtime_cache_at < self._runtime_cache_ttl
            and self._runtime_cache.get("thread_id") == target_thread_id
        ):
            return self._runtime_cache
        runtime = self._runtime_payload_for_thread(target_thread_id, refresh=True)
        return runtime

    def get_runtime_ready_payload(self, thread_id: str | None = None) -> dict:
        selection = self._load_selection_data()
        target_thread_id = thread_id or (
            str(selection.get("thread_id")) if selection and selection.get("thread_id") else None
        )
        if not target_thread_id:
            try:
                target_thread_id = self._get_selected_thread_fast().thread_id
            except Exception:
                target_thread_id = None
        if not target_thread_id:
            return {
                "ok": False,
                "ready": False,
                "thread_id": None,
                "status": "missing_thread",
                "reason": "找不到目前桌面端會話。",
                "recoverable": True,
            }

        port_ready = self._is_app_server_available()
        if port_ready or self.spawn_server:
            runtime = self._runtime_payload_for_thread(target_thread_id, refresh=True)
            runtime_error = str(runtime.get("error") or "").strip()
            if (
                runtime_error
                and self._should_restart_app_server_for_probe("runtime", runtime_error)
                and self._recover_app_server_once(f"runtime ready failed: {runtime_error[:200]}", force=True)
            ):
                runtime = self._runtime_payload_for_thread(target_thread_id, refresh=True)
        else:
            runtime = self._cached_runtime_or_fallback(target_thread_id, "app-server-unavailable")
        capabilities = runtime.get("capabilities") if isinstance(runtime.get("capabilities"), dict) else {}
        runtime_error = str(runtime.get("error") or "").strip()
        can_inject = bool(capabilities.get("threadInjectItems") or capabilities.get("threadRealtime"))
        runtime_probe_ready = not runtime_error and bool(capabilities)
        ready = bool((port_ready or runtime_probe_ready) and (can_inject or not runtime_error))
        if ready:
            status = "ready"
            reason = ""
        elif not port_ready and not runtime_probe_ready:
            status = "app_server_unavailable"
            reason = "Codex app-server 尚未就緒，手機訊息暫時不會送出。"
        else:
            status = "runtime_degraded"
            reason = f"Codex app-server runtime 未就緒：{runtime_error or 'capability unavailable'}"
        return {
            "ok": ready,
            "ready": ready,
            "thread_id": target_thread_id,
            "status": status,
            "reason": reason,
            "recoverable": True,
            "runtime": runtime,
        }

    def _codex_host_status_payload(self) -> dict[str, Any]:
        codex_bin = self._codex_bin_path()
        codex_version = ""
        codex_cli_ok = False
        if codex_bin:
            try:
                result = subprocess.run(
                    [str(codex_bin), "--version"],
                    capture_output=True,
                    text=True,
                    check=False,
                    timeout=3,
                )
                codex_version = (result.stdout or result.stderr or "").strip()
                codex_cli_ok = result.returncode == 0 and bool(codex_version)
            except (OSError, subprocess.TimeoutExpired):
                codex_version = ""

        remote_control_supported = False
        if codex_bin:
            try:
                result = subprocess.run(
                    [str(codex_bin), "remote-control", "--help"],
                    capture_output=True,
                    text=True,
                    check=False,
                    timeout=3,
                )
                output = f"{result.stdout}\n{result.stderr}"
                remote_control_supported = result.returncode == 0 and "remote-control" in output
            except (OSError, subprocess.TimeoutExpired):
                remote_control_supported = False

        codex_app_installed = CODEX_APP_PATH.exists()
        codex_app_running = self._process_running("Codex")
        app_server_ready = self._is_app_server_available()
        chatgpt_remote_ready = codex_app_installed and codex_app_running
        codexmobile_ready = app_server_ready
        return {
            "connection_modes": {
                "official_chatgpt": {
                    "label": "官方 ChatGPT App",
                    "ready": chatgpt_remote_ready,
                    "reason": (
                        "Codex App 已運行，可按官方流程掃 QR 連接"
                        if chatgpt_remote_ready
                        else "需要 Mac 保持醒著、在線，並打開 Codex App"
                    ),
                },
                "codexmobile_self_hosted": {
                    "label": "CodexMobile 自托管",
                    "ready": codexmobile_ready,
                    "reason": "app-server 已就緒" if codexmobile_ready else "等待 Codex app-server / desktop bridge 就緒",
                },
            },
            "codex_cli_ok": codex_cli_ok,
            "codex_cli_path": str(codex_bin) if codex_bin else "",
            "codex_version": codex_version,
            "remote_control_supported": remote_control_supported,
            "codex_app_installed": codex_app_installed,
            "codex_app_running": codex_app_running,
            "chatgpt_remote_ready": chatgpt_remote_ready,
            "app_server_ready": app_server_ready,
            "mac_awake_hint": "Mac 需要保持醒著、在線，官方 ChatGPT 遠程連接和 CodexMobile 自托管才會穩定。",
        }

    @staticmethod
    def _process_running(process_name: str) -> bool:
        result = subprocess.run(
            ["/usr/bin/pgrep", "-x", process_name],
            capture_output=True,
            text=True,
            check=False,
        )
        return result.returncode == 0

    @staticmethod
    def _codex_bin_path() -> Path | None:
        candidates: list[Path] = []
        env_bin = (os.environ.get("CODEXMOBILE_LIVE_CODEX_BIN") or os.environ.get("CODEXBOT_LIVE_CODEX_BIN") or "").strip()
        if env_bin:
            candidates.append(Path(env_bin).expanduser())
        candidates.append(CODEX_APP_BIN)
        resolved = shutil.which("codex")
        if resolved:
            candidates.append(Path(resolved))
        for candidate in candidates:
            if candidate.exists():
                return candidate
        return None

    def read_file(self, path: str, encoding: str | None = None) -> dict:
        requested_path = Path(path).expanduser()
        if not requested_path.is_absolute():
            selected = self._get_selected_thread()
            base = Path(selected.cwd or self.thread_cwd)
            requested_path = base / requested_path
        resolved_path = requested_path.resolve()
        if not resolved_path.exists() or not resolved_path.is_file():
            raise FileNotFoundError(str(resolved_path))
        mime_type = mimetypes.guess_type(str(resolved_path))[0] or "application/octet-stream"
        wants_base64 = str(encoding or "").strip().lower() in {"base64", "binary"}
        allowed_roots = self._file_view_roots()
        allowed_project_file = any(self._is_relative_to(resolved_path, root) for root in allowed_roots)
        allowed_image_file = (
            wants_base64
            and mime_type.startswith("image/")
            and (
                self._is_relative_to(resolved_path, Path.home())
                or self._is_relative_to(resolved_path, Path(tempfile.gettempdir()).resolve(strict=False))
                or self._is_relative_to(resolved_path, Path("/tmp").resolve(strict=False))
                or self._is_relative_to(resolved_path, Path("/private/tmp").resolve(strict=False))
            )
        )
        if not allowed_project_file and not allowed_image_file:
            raise PermissionError("只能查看目前專案、APPtest 目錄或桌面端會話圖片附件。")
        if wants_base64:
            if resolved_path.stat().st_size > 12_000_000:
                raise PermissionError("圖片過大，暫時不在手機端直接預覽。")
            return {
                "path": str(resolved_path),
                "name": resolved_path.name,
                "mime_type": mime_type,
                "encoding": "base64",
                "content_base64": base64.b64encode(resolved_path.read_bytes()).decode("ascii"),
            }
        if resolved_path.stat().st_size > 1_500_000:
            raise PermissionError("文件過大，暫時不在手機端直接預覽。")
        return {
            "path": str(resolved_path),
            "name": resolved_path.name,
            "mime_type": mime_type,
            "content": resolved_path.read_text(encoding="utf-8", errors="replace"),
        }

    def send_message(
        self,
        text: str,
        thread_id: str | None = None,
        model: str | None = None,
        effort: str | None = None,
        permission: str | None = None,
        image_data_url: str | None = None,
        local_image_path: str | None = None,
        image_data_urls: list[str] | None = None,
        local_image_paths: list[str] | None = None,
        wait_for_reply: bool = True,
    ) -> dict:
        self._thread_cache = None
        self._thread_cache_at = 0.0
        thread = self._get_selected_thread_for_send(thread_id)
        target_thread_id = thread_id or (thread.thread_id if thread else None)
        if not target_thread_id:
            raise CodexLiveBridgeError("找不到可用的會話來發送訊息。請先在桌面端打開目標會話，或在手機端重新選擇會話。")
        if self._app_server_in_backoff():
            elapsed = int(time.time() - self._app_server_failed_at)
            recovered = self._recover_app_server_once(
                f"mobile send requested during app-server backoff after {elapsed}s",
                force=True,
            )
            if not recovered:
                raise CodexLiveBridgeError(
                    f"Codex app-server 暫時不可用，上一個協議請求 {elapsed}s 前逾時。"
                    "手機端已避免重複重啟 app-server；請稍後重試，或在 Mac 端確認 Codex 桌面端可正常回應。"
                )
        selection = self._load_selection_data()
        selected_thread = selection if selection and selection.get("thread_id") == target_thread_id else None
        target_cwd = (
            (thread.cwd if thread else None)
            or (selected_thread.get("cwd") if selected_thread else None)
            or self.thread_cwd
        )
        target_name = (
            (thread.thread_name if thread else None)
            or (selected_thread.get("thread_name") if selected_thread else None)
            or (selected_thread.get("name") if selected_thread else None)
        )
        self._ensure_app_server_ready_for_send()
        # Codex 0.131+ uses stdio:// instead of a long-lived ws:// listener.
        # For stdio transports the app-server process must stay alive until the
        # turn finishes, so do not use no-wait detached submission there.
        use_detached_submit = not str(self._active_app_server_url or self.url).startswith("stdio://")
        args = [
            "send",
            "--json",
            "--cwd",
            target_cwd,
            "--thread-id",
            target_thread_id,
            "--prompt",
            text,
        ]
        if use_detached_submit:
            args.insert(2, "--no-wait")
        if target_name:
            args.extend(["--name", target_name])
        for url in [*(image_data_urls or []), *([image_data_url] if image_data_url else [])]:
            if url:
                args.extend(["--image-data-url", url])
        for path in [*(local_image_paths or []), *([local_image_path] if local_image_path else [])]:
            if path:
                args.extend(["--local-image-path", self._codex_supported_local_image_path(path)])
        if model:
            args.extend(["--model", model])
        if effort:
            args.extend(["--effort", effort])
        if permission:
            args.extend(["--permission", permission])
        # Newer Codex app-server builds can close the websocket while the probe
        # waits for turn/completed. Submit the turn quickly, keep a spawned
        # listener alive, then read the assistant response through snapshot.
        send_timeout = min(max(self.timeout, 45.0), 75.0)
        try:
            payload = self._run_probe(*args, timeout=send_timeout)
        except CodexLiveBridgeError as exc:
            detail = str(exc)
            raise CodexLiveBridgeError(
                f"{detail}。已嘗試受控恢復 Codex app-server；"
                "如果仍失敗，請稍後重試或在 Mac 端確認 Codex 桌面端可正常回應。"
            ) from exc
        reply_text = None
        if wait_for_reply:
            reply_text = self._wait_for_submitted_turn_reply(
                str(payload.get("threadId") or target_thread_id),
                str(payload.get("turnId") or ""),
                timeout_seconds=min(max(self.timeout, 45.0), 90.0),
            )
        self._persist_selection(
            thread_id=payload["threadId"],
            thread_name=payload.get("threadName") or (selected_thread.get("name") if selected_thread else None) or (thread.thread_name if thread else None),
            cwd=payload.get("cwd") or (selected_thread.get("cwd") if selected_thread else None) or (thread.cwd if thread else None) or self.thread_cwd,
            status="idle",
            pending_materialization=False,
        )
        if payload.get("accepted") or payload.get("status") in {"started", "steered"}:
            self._schedule_post_turn_refresh(payload["threadId"])
        return {
            "ok": True,
            "message_id": payload["turnId"],
            "thread_id": payload["threadId"],
            "thread_name": payload.get("threadName"),
            "cwd": payload.get("cwd"),
            "project_title": self._project_title_for_thread(
                {
                    "id": payload["threadId"],
                    "name": payload.get("threadName"),
                    "cwd": payload.get("cwd"),
                }
            )
            or (payload.get("threadName") or "目前專案"),
            "reply": reply_text or payload.get("text") or "已送出，正在等待桌面端處理。",
            "accepted": bool(payload.get("accepted") or payload.get("status") in {"started", "steered", "completed"}),
            "status": payload.get("status"),
            "steered": payload.get("status") == "steered",
        }

    def _wait_for_submitted_turn_reply(
        self,
        thread_id: str,
        turn_id: str,
        timeout_seconds: float = 60.0,
    ) -> str | None:
        if not thread_id:
            return None
        deadline = time.time() + max(5.0, timeout_seconds)
        cache_key = f"thread:{thread_id}:80"
        last_assistant_text: str | None = None
        while time.time() < deadline:
            self._snapshot_cache.pop(cache_key, None)
            try:
                snapshot = self.get_thread_snapshot_payload(thread_id, limit=80)
            except Exception as exc:
                logger.debug("Unable to poll submitted Codex turn %s/%s: %s", thread_id, turn_id, exc)
                time.sleep(1.5)
                continue
            state = snapshot.get("state", snapshot) if isinstance(snapshot, dict) else {}
            timeline = state.get("timeline") or state.get("messages") or []
            if not isinstance(timeline, list):
                time.sleep(1.5)
                continue
            for item in reversed(timeline):
                if not isinstance(item, dict) or item.get("role") != "assistant":
                    continue
                text = str(item.get("text") or "").strip()
                if not text:
                    continue
                last_assistant_text = text
                if turn_id and str(item.get("turn_id") or item.get("turnId") or "") == turn_id:
                    return text
                # Snapshot fallback does not always expose turn_id; the first
                # assistant after a no-wait submit is the response mobile needs.
                return text
            time.sleep(2.0)
        return last_assistant_text

    def _get_selected_thread_for_send(self, thread_id: str | None = None) -> LiveThreadSelection | None:
        """Resolve the target thread for mobile send without app-server list/read.

        Sending must not block on Codex app-server discovery. The actual send
        probe already has its own short timeout; this resolver only uses local
        selection and SQLite state so the mobile async worker can fail fast
        instead of staying in "sending" forever.
        """
        selection = self._load_selection_data()
        selected_id = str(thread_id or (selection or {}).get("thread_id") or "").strip()
        if not selected_id:
            return None

        local_row = self._local_thread_row(selected_id)
        if local_row:
            return LiveThreadSelection(
                thread_id=str(local_row.get("id") or selected_id),
                thread_name=local_row.get("title") or (selection or {}).get("thread_name"),
                cwd=local_row.get("cwd") or (selection or {}).get("cwd") or self.thread_cwd,
                updated_at=int(local_row.get("updated_at") or (selection or {}).get("updated_at") or time.time()),
                status=(selection or {}).get("status") or "idle",
            )

        if selection and selected_id == str(selection.get("thread_id") or "").strip():
            return self._live_selection_from_data(selection)
        return LiveThreadSelection(
            thread_id=selected_id,
            thread_name=(selection or {}).get("thread_name") if selection else None,
            cwd=(selection or {}).get("cwd") if selection else self.thread_cwd,
            updated_at=int((selection or {}).get("updated_at") or time.time()) if selection else int(time.time()),
            status=(selection or {}).get("status") if selection else "idle",
        )

    def _ensure_app_server_ready_for_send(self) -> None:
        """Verify Codex app-server at the protocol layer before mobile send.

        A stale app-server can keep the TCP port open while JSON-RPC initialize
        hangs. Mobile sends must not trust the TCP-only availability check,
        otherwise the app shows "sent" while the daemon never actually submits
        the turn.
        """
        if not self.spawn_server:
            return
        selected_url = self._select_app_server_url(
            timeout=4.0,
            bypass_backoff=True,
            require_protocol=True,
            ignore_cached=True,
        )
        if selected_url:
            return
        # Do not fail before send when no healthy listener exists. The send
        # command itself is allowed to spawn a fresh app-server. If a TCP port
        # is open but protocol probing failed, clear it first so send does not
        # attach to a stale listener.
        if self._is_app_server_available():
            self._recover_app_server_once("send preflight found stale app-server listener", force=True)

    def request_computer_control(self, instruction: str | None = None, thread_id: str | None = None) -> dict:
        user_instruction = (instruction or "").strip()
        prompt = (
            "請在目前 Mac 桌面端 Codex live 會話中啟用「控制電腦」能力來協助我操作本機。\n"
            "請遵守這些規則：\n"
            "1. 先說明你準備操作的目標與步驟，再開始執行。\n"
            "2. 涉及點擊、輸入、系統權限、文件修改或不可逆操作時，必須等待我的明確批准。\n"
            "3. 如果我從手機端點擊停止，請立即停止目前操作並回報狀態。\n"
            "4. 所有操作結果都要回寫到這個 live 會話，方便手機端與桌面端銜接。\n"
        )
        if user_instruction:
            prompt += f"\n我希望你優先處理：{user_instruction}"
        return self.send_message(prompt, thread_id=thread_id)


    def _get_cached_snapshot(self, cache_key: str, *, allow_stale: bool = False) -> dict | None:
        cached = self._snapshot_cache.get(cache_key)
        if not cached:
            return None
        if self._is_snapshot_cache_fresh(cached[0], allow_stale=allow_stale):
            return cached[1]
        return None

    def _is_snapshot_cache_fresh(self, cached_at: float, *, allow_stale: bool = False) -> bool:
        age = time.time() - cached_at
        return age <= self._snapshot_cache_ttl or (allow_stale and age <= self._snapshot_stale_ttl)

    def _snapshot_thread_id(self, payload: dict) -> str:
        state = payload.get("state") if isinstance(payload, dict) else None
        if not isinstance(state, dict):
            return ""
        return str(state.get("thread_id") or "")

    def _snapshot_timeline_size(self, payload: dict) -> int:
        state = payload.get("state") if isinstance(payload, dict) else None
        if not isinstance(state, dict):
            return 0
        timeline = state.get("timeline")
        if isinstance(timeline, list):
            return len(timeline)
        messages = state.get("messages")
        return len(messages) if isinstance(messages, list) else 0

    @staticmethod
    def _rollout_timeline_is_incomplete(messages: list[dict]) -> bool:
        if not messages:
            return True
        has_user = any((item.get("role") or item.get("author")) == "user" for item in messages)
        has_file_change = any(bool(item.get("file_changes")) or item.get("kind") == "fileChange" for item in messages)
        has_command = any(bool(item.get("command_execution")) or item.get("kind") == "commandExecution" for item in messages)
        if has_user or has_file_change or has_command:
            return False
        # During an active desktop response, the rollout tail can contain only
        # CodexMobile progress updates. Treating that as a full snapshot erases
        # the user's visible desktop messages on mobile.
        if len(messages) <= 3:
            return True
        return False

    def _get_best_cached_snapshot(self, *, thread_id: str | None = None, allow_stale: bool = False) -> dict | None:
        candidates: list[tuple[float, float, dict]] = []
        for cache_key, (cached_at, payload) in self._snapshot_cache.items():
            if not self._is_snapshot_cache_fresh(cached_at, allow_stale=allow_stale):
                continue
            payload_thread_id = self._snapshot_thread_id(payload)
            timeline_size = self._snapshot_timeline_size(payload)
            if thread_id and payload_thread_id and payload_thread_id != thread_id:
                continue
            score = 0.0
            if thread_id and payload_thread_id == thread_id:
                score += 100.0
            elif cache_key.startswith("selected:"):
                score += 40.0
            if timeline_size:
                score += 20.0
            score += cached_at / 1_000_000_000.0
            candidates.append((score, cached_at, payload))
        if not candidates and thread_id:
            # Keep mobile stable during a busy refresh. An older non-empty
            # snapshot is safer than an empty payload that wipes the timeline.
            for _cache_key, (cached_at, payload) in self._snapshot_cache.items():
                if not self._is_snapshot_cache_fresh(cached_at, allow_stale=allow_stale):
                    continue
                if self._snapshot_timeline_size(payload) > 0:
                    candidates.append((cached_at / 1_000_000_000.0, cached_at, payload))
        if candidates:
            return max(candidates, key=lambda item: (item[0], item[1]))[2]
        return None

    def _store_snapshot_cache(self, cache_key: str, payload: dict) -> dict:
        cached_at = time.time()
        self._snapshot_cache[cache_key] = (cached_at, payload)
        thread_id = self._snapshot_thread_id(payload)
        if thread_id:
            self._snapshot_cache[f"thread:{thread_id}:latest"] = (cached_at, payload)
        if cache_key.startswith("selected:"):
            self._snapshot_cache["selected:latest"] = (cached_at, payload)
        return payload

    def _build_minimal_snapshot_payload(
        self,
        *,
        thread_id: str | None = None,
        limit: int = 500,
        reason: str = "snapshot-unavailable",
    ) -> dict:
        selection = self._load_selection_data()
        selected = self._live_selection_from_data(selection) if selection and selection.get("thread_id") else None
        target_thread_id = thread_id or (selected.thread_id if selected else "")
        cwd = selected.cwd if selected else self.thread_cwd
        thread_name = selected.thread_name if selected else self.thread_name
        project_title = self._project_title_for_selection(selected) if selected else (Path(cwd).name if cwd else "目前專案")
        updated_at = self._iso_from_epoch(selected.updated_at if selected else int(time.time()))
        thread_item = {
            "thread_id": target_thread_id,
            "thread_name": thread_name,
            "cwd": cwd,
            "project_title": project_title,
            "updated_at": updated_at,
            "status": selected.status if selected else "idle",
            "selected": True,
        }
        return {
            "pending_count": 0,
            "last_message": None,
            "state": {
                "messages": [],
                "timeline": [],
                "timeline_source": "codex-app-server-stale",
                **self._server_clock_payload(),
                "updated_at": updated_at,
                "thread_id": target_thread_id,
                "thread_name": thread_name,
                "cwd": cwd,
                "project_root": str(self.project_root),
                "mobile_default_cwd": self.thread_cwd,
                "thread_status": selected.status if selected else "idle",
                "project_title": project_title,
                "generation_settings": self._fallback_generation_settings(target_thread_id) if target_thread_id else None,
                "runtime": {
                    "thread_id": target_thread_id,
                    "reason": reason,
                    "capabilities": {
                        "turnStart": True,
                        "turnSteer": True,
                        "turnInterrupt": True,
                        "threadRealtime": False,
                    },
                    "models": [],
                },
                "host_status": self._codex_host_status_payload(),
                "capabilities": {
                    "turnStart": True,
                    "turnSteer": True,
                    "turnInterrupt": True,
                    "threadRealtime": False,
                },
                "models": [],
                "threads": [thread_item] if target_thread_id else [],
                "projects": [],
            },
        }

    def inject_items(self, items: list[dict[str, Any]], thread_id: str | None = None) -> dict:
        target_thread_id = thread_id or self._get_selected_thread().thread_id
        if not target_thread_id:
            raise CodexLiveBridgeError("找不到可用的會話來注入資料。")
        payload = self._run_probe(
            "inject",
            "--json",
            "--thread-id",
            target_thread_id,
            "--items-json",
            json.dumps(items, ensure_ascii=False),
        )
        return {
            "ok": True,
            "thread_id": target_thread_id,
            "result": payload.get("result") if isinstance(payload, dict) else payload,
        }

    def interrupt_current_turn(self, thread_id: str | None = None) -> dict:
        thread = self._get_selected_thread() if not thread_id else None
        target_thread_id = thread_id or (thread.thread_id if thread else None)
        if not target_thread_id:
            raise CodexLiveBridgeError("找不到可用的會話來停止處理。")
        payload = self._run_probe(
            "interrupt",
            "--json",
            "--thread-id",
            target_thread_id,
            timeout=min(max(self.timeout, 3.0), 5.0),
        )
        interrupted = bool(payload.get("interrupted"))
        return {
            "ok": True,
            "thread_id": payload.get("threadId") or target_thread_id,
            "thread_name": payload.get("threadName") or (thread.thread_name if thread else None),
            "interrupted": interrupted,
            "turn_id": payload.get("turnId"),
            "reason": payload.get("reason"),
            "reply": "已停止目前處理。" if interrupted else "目前沒有可停止的處理。",
        }

    def rollback_turns(self, thread_id: str | None = None, num_turns: int = 1) -> dict:
        thread = self._get_selected_thread() if not thread_id else None
        target_thread_id = thread_id or (thread.thread_id if thread else None)
        if not target_thread_id:
            raise CodexLiveBridgeError("找不到可用的會話來回退。")
        turns_to_drop = max(1, int(num_turns or 1))
        payload = self._run_probe(
            "rollback",
            "--json",
            "--thread-id",
            target_thread_id,
            "--num-turns",
            str(turns_to_drop),
        )
        selected_thread = self._read_thread_by_id(target_thread_id)
        self._save_selected_thread(
            LiveThreadSelection(
                thread_id=target_thread_id,
                thread_name=payload.get("threadName") or (selected_thread.get("name") if selected_thread else None) or (thread.thread_name if thread else None),
                cwd=payload.get("cwd") or (selected_thread.get("cwd") if selected_thread else None) or (thread.cwd if thread else None) or self.thread_cwd,
                status="idle",
                pending_materialization=False,
            )
        )
        return {
            "ok": True,
            "thread_id": payload.get("threadId") or target_thread_id,
            "thread_name": payload.get("threadName") or (thread.thread_name if thread else None),
            "num_turns": turns_to_drop,
            "reply": f"已回退最近 {turns_to_drop} 輪。",
        }

    def analyze_attachment(
        self,
        *,
        local_path: str,
        filename: str,
        mime_type: str,
        thread_id: str | None = None,
    ) -> dict:
        suffix = Path(filename).suffix.lower()
        kind = "圖片" if mime_type.startswith("image/") else "PDF" if mime_type == "application/pdf" or suffix == ".pdf" else "附件"
        if kind == "圖片":
            prompt = (
                "請直接閱讀並分析這張圖片。"
                "請用繁體中文回答，至少包含：畫面主體、關鍵細節、可辨識文字、可能情境。"
                f"\n檔案名稱：{filename}"
                "\n不要只回覆已收到，請直接給出分析結果。"
            )
        elif kind == "PDF":
            prompt = (
                "我剛附上一個 PDF 檔，請直接閱讀並分析內容。"
                "請用繁體中文回答，至少包含：主題摘要、重點條列、可執行資訊或需要注意的地方。"
                f"\n檔案名稱：{filename}"
                f"\n檔案路徑：{local_path}"
                "\n不要只回覆已收到，請直接給出分析結果。"
            )
        else:
            prompt = (
                "我剛附上一個檔案，請先嘗試閱讀並分析這個附件。"
                "請用繁體中文回答你能讀到的內容與重點。"
                f"\n檔案名稱：{filename}"
                f"\n檔案類型：{mime_type}"
                f"\n檔案路徑：{local_path}"
                "\n不要只回覆已收到，請直接給出分析結果。"
            )
        payload = self.send_message(
            prompt,
            thread_id=thread_id,
            local_image_path=local_path if kind == "圖片" else None,
        )
        payload["attachment_path"] = local_path
        payload["attachment_name"] = filename
        payload["attachment_kind"] = kind
        return payload

    def mark_done(self, message_id: str) -> dict:
        return {"ok": True, "message_id": message_id}

    def list_threads(self, limit: int = 200) -> list[dict]:
        raw_threads = self._list_raw_threads(limit)
        selection = self._load_selection_data()
        selection_trusted = self._selection_is_trusted(selection, raw_threads)
        selected_id = selection.get("thread_id") if selection_trusted and selection else None
        pending_selection = bool(
            selection
            and selection.get("pending_materialization")
            and not self._is_stale_pending_selection(selection)
        )
        selected = self._preferred_active_thread(raw_threads)
        if selected:
            selected_id = selected.get("id")
            if selected_id and (not selection or selection.get("thread_id") != selected_id):
                self._persist_selection(
                    thread_id=selected_id,
                    thread_name=selected.get("name"),
                    cwd=selected.get("cwd"),
                    status=self._status_label(selected.get("status")) or "idle",
                    pending_materialization=False,
                )
        elif selected_id:
            selected = next((thread for thread in raw_threads if thread.get("id") == selected_id), None)
            selected = self._prefer_visible_thread(raw_threads, selected, selection)
            if not selected and pending_selection:
                selected = self._read_thread_by_id(selected_id, timeout=min(self.timeout, 3.0))
        if not selected and not selected_id:
            selected = self._select_thread(raw_threads)
            selected_id = selected.get("id") if selected else None
        elif selected and selected.get("id") and selected.get("id") != selected_id:
            selected_id = selected["id"]
            self._persist_selection(
                thread_id=selected_id,
                thread_name=selected.get("name"),
                cwd=selected.get("cwd"),
                status=self._status_label(selected.get("status")) or "idle",
                pending_materialization=False,
            )
        selected_project_title = self._project_title_for_thread(selected) if selected else None
        selection_live = self._live_selection_from_data(selection) if selection and selection.get("thread_id") else None
        representative_map: dict[str, dict] = {}
        for thread in raw_threads:
            title = self._project_title_for_thread(thread)
            if not title:
                continue
            existing = representative_map.get(title)
            if not existing:
                representative_map[title] = thread
                continue
            existing_score = self._score_project_representative(existing, title)
            next_score = self._score_project_representative(thread, title)
            if next_score > existing_score:
                representative_map[title] = thread
                continue
            if next_score == existing_score and int(thread.get("updatedAt") or 0) > int(existing.get("updatedAt") or 0):
                representative_map[title] = thread

        if (
            selected_id
            and selection_trusted
            and selection
            and selection.get("thread_id") == selected_id
            and not self._is_stale_pending_selection(selection)
        ):
            selected_title = selected_project_title or (
                self._project_title_for_selection(selection_live) if selection_live else None
            )
            synthetic_selected = {
                "id": selected_id,
                "name": selection.get("thread_name"),
                "cwd": selection.get("cwd") or self.thread_cwd,
                "updatedAt": selection.get("updated_at") or int(time.time()),
                "status": selection.get("status") or "idle",
            }
            if selected_title:
                # While a brand new thread has not materialized into thread/list yet,
                # keep the selected synthetic entry visible so mobile sidebars don't
                # instantly snap back to an older representative thread.
                if pending_selection or selected_title not in representative_map:
                    representative_map[selected_title] = synthetic_selected
                else:
                    existing = representative_map.get(selected_title)
                    if existing and existing.get("id") == selected_id:
                        representative_map[selected_title] = synthetic_selected

        if selected_id and selected_project_title and selected_project_title not in representative_map and not (selection and not selection_trusted):
            synthetic = selected or selection
            if synthetic:
                representative_map[selected_project_title] = {
                    "id": selected_id,
                    "name": (synthetic.get("name") if isinstance(synthetic, dict) else None) or getattr(selected, "thread_name", None),
                    "cwd": (synthetic.get("cwd") if isinstance(synthetic, dict) else None) or getattr(selected, "cwd", None) or self.thread_cwd,
                    "updatedAt": (synthetic.get("updatedAt") if isinstance(synthetic, dict) else None) or (synthetic.get("updated_at") if isinstance(synthetic, dict) else None) or getattr(selected, "updated_at", None),
                    "status": (synthetic.get("status") if isinstance(synthetic, dict) else None) or getattr(selected, "status", None) or "idle",
                }

        items: list[dict] = []
        for title, thread in representative_map.items():
            cwd = thread.get("cwd")
            is_selected = False
            if selected_project_title:
                is_selected = title == selected_project_title
            elif selected_id:
                is_selected = thread.get("id") == selected_id
            items.append(
                {
                    "thread_id": thread["id"],
                    "thread_name": thread.get("name"),
                    "cwd": cwd,
                    "project_title": title,
                    "updated_at": self._iso_from_epoch(thread.get("updatedAt")),
                    "status": self._status_label(thread.get("status")),
                    "selected": is_selected,
                }
            )
        items.sort(key=lambda item: item.get("updated_at") or "", reverse=True)
        return items

    def list_projects(self, limit: int = 200, *, force: bool = False) -> list[dict]:
        thread_projects = self._list_raw_threads(limit, force=force)
        archived_threads = self._list_archived_raw_threads(limit)
        for item in archived_threads:
            item["archived"] = True
        thread_projects = self._merge_threads(thread_projects, archived_threads, self._local_state_project_threads())
        return self._projects_from_thread_projects(thread_projects, limit=limit)

    def _projects_from_thread_projects(self, thread_projects: list[dict], limit: int = 200) -> list[dict]:
        saved_projects = self._load_saved_projects()
        desktop_sidebar = self._load_desktop_sidebar_state()
        pinned_thread_order = {thread_id: index for index, thread_id in enumerate(desktop_sidebar.get("pinned_thread_ids") or [])}
        pinned_thread_ids = set(pinned_thread_order)
        projectless_thread_ids = set(desktop_sidebar.get("projectless_thread_ids") or [])
        workspace_roots = [str(item) for item in desktop_sidebar.get("workspace_roots") or [] if str(item or "").strip()]
        workspace_labels = desktop_sidebar.get("workspace_labels") or {}
        workspace_order = {self._cwd_key(cwd): index for index, cwd in enumerate(workspace_roots)}
        workspace_root_keys = set(workspace_order)
        pinned_project_keys = {self._cwd_key(item) for item in desktop_sidebar.get("pinned_project_ids") or [] if self._cwd_key(item)}
        selection = self._load_selection_data()
        selected_id = str(selection.get("thread_id") or "").strip() if selection else ""
        project_map: dict[str, dict[str, Any]] = {}
        conversation_map: dict[str, dict[str, Any]] = {}
        project_children_map: dict[str, dict[str, dict[str, Any]]] = {}
        saved_project_order = {
            str(item.get("title") or "").strip().casefold(): index
            for index, item in enumerate(saved_projects)
            if str(item.get("title") or "").strip()
        }
        saved_cwd_order = {
            self._cwd_key(item.get("cwd")): index
            for index, item in enumerate(saved_projects)
            if self._cwd_key(item.get("cwd"))
        }

        for item in thread_projects:
            thread_id = item.get("thread_id") or item.get("id")
            thread_id_text = str(thread_id or "").strip()
            is_selected = bool(item.get("selected")) or (bool(thread_id) and str(thread_id) == selected_id)
            updated_at = item.get("updated_at") or self._iso_from_epoch(item.get("updatedAt"))
            status = item.get("status")
            is_archived = bool(item.get("archived") or item.get("isArchived") or item.get("is_archived"))
            thread_name = str(item.get("thread_name") or item.get("name") or "").strip()
            cwd_key = self._cwd_key(item.get("cwd"))
            is_pinned = bool(thread_id_text and thread_id_text in pinned_thread_ids)
            if is_pinned:
                raw_title = str(item.get("project_title") or self._project_title_for_thread(item) or thread_name or "未命名項目").strip()
            elif cwd_key in workspace_root_keys:
                raw_title = str(workspace_labels.get(str(item.get("cwd") or "").rstrip("/")) or Path(str(item.get("cwd") or "")).name or "未命名項目").strip()
            else:
                raw_title = str(item.get("project_title") or self._project_title_for_thread(item) or "未命名項目").strip()
            raw_title = raw_title or "未命名項目"
            is_standalone_conversation = self._is_desktop_conversation_thread(item)
            is_projectless = bool(thread_id_text and thread_id_text in projectless_thread_ids)
            is_project_child = self._is_desktop_project_child_thread(item, raw_title)
            is_conversation = is_archived or is_standalone_conversation or is_projectless
            title = raw_title if is_pinned else (thread_name or raw_title)
            if is_pinned and thread_id:
                pinned_title = thread_name or raw_title or "未命名對話"
                conversation_map[f"pinned:{thread_id}"] = {
                    "project_id": f"thread:{thread_id}",
                    "title": pinned_title,
                    "cwd": item.get("cwd"),
                    "updated_at": updated_at,
                    "selected": is_selected,
                    "thread_id": thread_id,
                    "thread_name": pinned_title,
                    "status": self._status_label(status) or status,
                    "is_pinned": True,
                    "section": "pinned",
                    "archived": False,
                    "parent_project_title": raw_title if raw_title and pinned_title.casefold() != raw_title.casefold() else None,
                    "parent_project_id": self._project_id(raw_title, item.get("cwd")) if raw_title and pinned_title.casefold() != raw_title.casefold() else None,
                    "project_order": None,
                    "pinned_order": pinned_thread_order.get(thread_id_text),
                }
            if (
                not is_archived
                and not is_standalone_conversation
                and not is_projectless
                and not is_pinned
                and thread_id
                and thread_name
                and raw_title
                and thread_name.casefold() != raw_title.casefold()
            ):
                project_children_map.setdefault(raw_title.casefold(), {})[str(thread_id)] = {
                    "project_id": f"thread:{thread_id}",
                    "title": thread_name,
                    "cwd": item.get("cwd"),
                    "updated_at": updated_at,
                    "selected": is_selected,
                    "thread_id": thread_id,
                    "thread_name": thread_name,
                    "status": self._status_label(status) or status,
                    "parent_project_id": self._project_id(raw_title, item.get("cwd")),
                    "parent_project_title": raw_title,
                }
            if is_conversation and thread_id:
                conversation_map[f"conversation:{thread_id}"] = {
                    "project_id": f"thread:{thread_id}",
                    "title": title,
                    "cwd": item.get("cwd"),
                    "updated_at": updated_at,
                    "selected": is_selected,
                    "thread_id": thread_id,
                    "thread_name": thread_name or title,
                    "status": self._status_label(status) or status,
                    "is_pinned": False,
                    "section": "archived" if is_archived else "conversation",
                    "archived": is_archived,
                    "parent_project_title": None if (is_archived or is_standalone_conversation or is_projectless) else raw_title,
                    "parent_project_id": None if (is_archived or is_standalone_conversation or is_projectless) else self._project_id(raw_title, item.get("cwd")),
                    "project_order": None,
                }
            if is_archived or is_standalone_conversation or is_pinned:
                continue

            project_key = raw_title.casefold()
            is_desktop_workspace = cwd_key in workspace_root_keys
            if workspace_root_keys and not is_desktop_workspace:
                continue
            if not is_desktop_workspace and is_project_child:
                continue
            is_project_pinned = bool(cwd_key and cwd_key in pinned_project_keys)
            candidate = {
                "project_id": self._project_id(raw_title, item.get("cwd")),
                "title": raw_title,
                "cwd": item.get("cwd"),
                "updated_at": updated_at,
                "selected": is_selected and not is_project_child,
                "thread_id": thread_id,
                "thread_name": thread_name or None,
                "status": self._status_label(status) or status,
                "is_pinned": is_project_pinned,
                "section": "pinned" if is_project_pinned else "project",
                "project_order": workspace_order.get(cwd_key, saved_project_order.get(project_key, saved_cwd_order.get(cwd_key))),
            }
            existing = project_map.get(project_key)
            if not existing:
                project_map[project_key] = candidate
                continue
            if self._score_project_candidate(candidate) > self._score_project_candidate(existing):
                project_map[project_key] = candidate
                continue
            if self._score_project_candidate(candidate) == self._score_project_candidate(existing):
                if (candidate.get("updated_at") or "") > (existing.get("updated_at") or ""):
                    project_map[project_key] = candidate

        for index, cwd in enumerate(workspace_roots):
            cwd_key = self._cwd_key(cwd)
            if not cwd_key:
                continue
            title = str(workspace_labels.get(cwd) or Path(cwd).name or "未命名項目").strip()
            project_key = title.casefold()
            cwd_existing_key = next(
                (
                    key
                    for key, value in project_map.items()
                    if cwd_key and self._cwd_key(value.get("cwd")) == cwd_key
                ),
                None,
            )
            if cwd_existing_key and cwd_existing_key != project_key:
                existing = project_map.pop(cwd_existing_key)
                existing["project_id"] = self._project_id(title, cwd)
                existing["title"] = title
                existing["cwd"] = cwd
                existing["is_pinned"] = cwd_key in pinned_project_keys
                existing["section"] = "pinned" if existing["is_pinned"] else "project"
                existing["project_order"] = index
                project_map[project_key] = existing
                continue
            if project_key in project_map:
                existing = project_map[project_key]
                existing["project_order"] = existing.get("project_order") if isinstance(existing.get("project_order"), int) else index
                if cwd_key in pinned_project_keys:
                    existing["is_pinned"] = True
                    existing["section"] = "pinned"
                else:
                    existing["section"] = "pinned" if existing.get("is_pinned") else "project"
                continue
            is_project_pinned = cwd_key in pinned_project_keys
            project_map[project_key] = {
                "project_id": self._project_id(title, cwd),
                "title": title,
                "cwd": cwd,
                "updated_at": self._iso_from_epoch(int(time.time())),
                "selected": False,
                "thread_id": None,
                "thread_name": None,
                "status": "idle",
                "is_pinned": is_project_pinned,
                "section": "pinned" if is_project_pinned else "project",
                "project_order": index,
            }

        for item in saved_projects:
            title = str(item.get("title") or "").strip()
            cwd = str(item.get("cwd") or "").strip() or None
            if not title:
                continue
            if workspace_roots and self._cwd_key(cwd) not in workspace_root_keys:
                continue
            cwd_key = self._cwd_key(cwd)
            if cwd_key in workspace_root_keys:
                continue
            project_key = title.casefold()
            cwd_existing_key = next(
                (
                    key
                    for key, value in project_map.items()
                    if cwd_key and self._cwd_key(value.get("cwd")) == cwd_key
                ),
                None,
            )
            if cwd_existing_key and cwd_existing_key != project_key:
                existing = project_map.pop(cwd_existing_key)
                if existing.get("is_pinned"):
                    project_map[cwd_existing_key] = existing
                    cwd_existing_key = None
                else:
                    existing["project_id"] = self._project_id(title, cwd or existing.get("cwd"))
                    existing["title"] = title
                    existing["cwd"] = cwd or existing.get("cwd")
                    existing["is_pinned"] = cwd_key in pinned_project_keys
                    existing["section"] = "pinned" if existing["is_pinned"] else "project"
                    existing["project_order"] = saved_project_order.get(
                        project_key, saved_cwd_order.get(cwd_key)
                    )
                    project_map[project_key] = existing
                    continue
            if project_key in project_map:
                existing = project_map[project_key]
                if not existing.get("cwd") and cwd:
                    existing["cwd"] = cwd
                    existing["project_id"] = self._project_id(title, cwd)
                if cwd_key in pinned_project_keys:
                    existing["is_pinned"] = True
                    existing["section"] = "pinned"
                continue
            project_map[project_key] = {
                "project_id": self._project_id(title, cwd),
                "title": title,
                "cwd": cwd,
                "updated_at": self._iso_from_epoch(int(item.get("added_at") or time.time())),
                "selected": False,
                "thread_id": None,
                "thread_name": None,
                "status": "idle",
                "is_pinned": self._cwd_key(cwd) in pinned_project_keys,
                "section": "pinned" if self._cwd_key(cwd) in pinned_project_keys else "project",
                "project_order": saved_project_order.get(project_key, saved_cwd_order.get(cwd_key)),
            }

        for project_key, project in project_map.items():
            child_map = project_children_map.get(project_key)
            if not child_map:
                continue
            children = list(child_map.values())
            children.sort(key=lambda child: (-self._updated_at_timestamp(child.get("updated_at")), str(child.get("title") or "").casefold()))
            project["children"] = children
            if any(child.get("selected") for child in children):
                project["selected"] = True
                selected_child = next(child for child in children if child.get("selected"))
                project["thread_id"] = selected_child.get("thread_id")
                project["thread_name"] = selected_child.get("thread_name")
                project["updated_at"] = selected_child.get("updated_at") or project.get("updated_at")

        items = [*project_map.values(), *conversation_map.values()]
        items.sort(key=self._desktop_project_sort_key)
        return items

    @staticmethod
    def _is_desktop_conversation_thread(item: dict[str, Any]) -> bool:
        thread_id = str(item.get("thread_id") or item.get("id") or "").strip()
        if not thread_id:
            return False
        cwd = str(item.get("cwd") or "").strip()
        if not cwd:
            return True
        normalized_cwd = cwd.replace("\\", "/")
        path_name = Path(cwd).name
        # Codex creates per-conversation working folders for uploaded or mentioned
        # files. The desktop sidebar treats those as conversations, not projects.
        if "/Documents/Codex/" in normalized_cwd and path_name.startswith("files-mentioned-by-the-user"):
            return True
        return False

    def _is_desktop_project_child_thread(self, item: dict[str, Any], project_title: str | None = None) -> bool:
        thread_id = str(item.get("thread_id") or item.get("id") or "").strip()
        if not thread_id:
            return False
        if self._is_desktop_conversation_thread(item):
            return True
        title = str(project_title or item.get("project_title") or self._project_title_for_thread(item) or "").strip()
        thread_name = str(item.get("thread_name") or item.get("name") or "").strip()
        if not thread_name or not title:
            return False
        return thread_name.casefold() != title.casefold()

    @classmethod
    def _desktop_project_sort_key(cls, item: dict[str, Any]) -> tuple[int, int, float, str]:
        title = str(item.get("title") or "").strip()
        if item.get("is_pinned"):
            pinned_order = item.get("pinned_order")
            return (0, pinned_order if isinstance(pinned_order, int) else 10_000, 0.0, title.casefold())
        if item.get("section") == "archived":
            return (
                3,
                10_000,
                -cls._updated_at_timestamp(item.get("updated_at")),
                title.casefold(),
            )
        if item.get("section") == "conversation":
            return (
                2,
                10_000,
                -cls._updated_at_timestamp(item.get("updated_at")),
                title.casefold(),
            )
        project_order = item.get("project_order")
        if isinstance(project_order, int):
            return (1, project_order, 0.0, title.casefold())
        return (
            1,
            10_000,
            -cls._updated_at_timestamp(item.get("updated_at")),
            title.casefold(),
        )

    @staticmethod
    def _updated_at_timestamp(value: Any) -> float:
        if value is None:
            return 0.0
        if isinstance(value, (int, float)):
            return float(value)
        text = str(value).strip()
        if not text:
            return 0.0
        try:
            return datetime.fromisoformat(text.replace("Z", "+00:00")).timestamp()
        except ValueError:
            return 0.0

    @staticmethod
    def _cwd_key(cwd: Any) -> str:
        text = str(cwd or "").strip()
        if not text:
            return ""
        try:
            return str(Path(text).expanduser().resolve(strict=False)).rstrip("/")
        except Exception:
            return text.rstrip("/")

    def select_thread(self, thread_id: str) -> dict:
        payload = self._run_probe("list", "--limit", "50")
        selected = next((thread for thread in payload.get("data", []) if thread.get("id") == thread_id), None)
        if not selected:
            selection = self._load_selection_data()
            if selection and selection.get("thread_id") == thread_id and not self._is_stale_pending_selection(selection):
                self._persist_selection(
                    thread_id=thread_id,
                    thread_name=selection.get("thread_name"),
                    cwd=selection.get("cwd"),
                    status=self._status_label(selection.get("status")) or "idle",
                    pending_materialization=bool(selection.get("pending_materialization")),
                )
                return {
                    "ok": True,
                    "thread_id": thread_id,
                    "thread_name": selection.get("thread_name"),
                    "cwd": selection.get("cwd"),
                    "project_title": self._project_title_for_thread(
                        {
                            "id": thread_id,
                            "name": selection.get("thread_name"),
                            "cwd": selection.get("cwd"),
                        }
                    )
                    or (selection.get("thread_name") or "目前專案"),
                }
            raise CodexLiveBridgeError(f"找不到指定的會話：{thread_id}")
        self._persist_selection(
            thread_id=thread_id,
            thread_name=selected.get("name"),
            cwd=selected.get("cwd"),
            status=self._status_label(selected.get("status")) or "idle",
            pending_materialization=False,
        )
        self._thread_cache = None
        self._thread_cache_at = 0.0
        cwd = selected.get("cwd")
        return {
            "ok": True,
            "thread_id": thread_id,
            "thread_name": selected.get("name"),
            "cwd": cwd,
            "project_title": self._project_title_for_thread(selected) or (selected.get("name") or "目前專案"),
        }

    def _clear_thread_caches(self) -> None:
        self._thread_cache = None
        self._thread_cache_at = 0.0
        self._raw_threads_cache = []
        self._raw_threads_cache_at = 0.0
        self._rollout_thread_cache.clear()

    def create_thread(
        self,
        cwd: str | None = None,
        model: str | None = None,
        effort: str | None = None,
        permission: str | None = None,
    ) -> dict:
        current = self._select_thread(self._run_probe("list", "--limit", "50").get("data", []))
        target_cwd = cwd or (current.get("cwd") if current else self.thread_cwd)
        target_path = Path(str(target_cwd))
        if not target_path.exists() or not target_path.is_dir():
            raise CodexLiveBridgeError(f"找不到項目資料夾：{target_cwd}")
        args = ["start", "--cwd", str(target_cwd)]
        if model:
            args.extend(["--model", model])
        if effort:
            args.extend(["--effort", effort])
        if permission:
            args.extend(["--permission", permission])
        payload = self._run_probe(*args)
        thread_id = payload["threadId"]
        thread_cwd = payload.get("cwd") or str(target_cwd)
        self._persist_selection(
            thread_id=thread_id,
            thread_name=payload.get("threadName"),
            cwd=thread_cwd,
            status="idle",
            pending_materialization=True,
        )
        self._thread_cache = None
        self._thread_cache_at = 0.0
        return {
            "ok": True,
            "thread_id": thread_id,
            "thread_name": payload.get("threadName"),
            "cwd": thread_cwd,
            "project_title": self._project_title_for_thread(
                {
                    "id": thread_id,
                    "name": payload.get("threadName"),
                    "cwd": thread_cwd,
                }
            )
            or (payload.get("threadName") or "目前專案"),
        }

    def rename_thread(self, thread_id: str, title: str) -> dict:
        target_thread_id = str(thread_id).strip()
        next_title = str(title).strip()
        if not target_thread_id:
            raise CodexLiveBridgeError("Missing thread_id")
        if not next_title:
            raise CodexLiveBridgeError("Missing title")
        payload = self._run_probe(
            "name",
            "--thread-id",
            target_thread_id,
            "--name",
            next_title,
            timeout=min(self.timeout, 15.0),
        )
        self._clear_thread_caches()
        selection = self._load_selection_data()
        thread_name = payload.get("threadName") or next_title
        cwd = payload.get("cwd") or (selection or {}).get("cwd")
        if selection and selection.get("thread_id") == target_thread_id:
            self._persist_selection(
                thread_id=target_thread_id,
                thread_name=thread_name,
                cwd=cwd,
                status=self._status_label(selection.get("status")) or "idle",
                pending_materialization=bool(selection.get("pending_materialization")),
            )
        projects = self.list_projects(force=True)
        return {
            "ok": True,
            "thread_id": target_thread_id,
            "thread_name": thread_name,
            "cwd": cwd,
            "project_title": self._project_title_for_thread({"id": target_thread_id, "name": thread_name, "cwd": cwd}) or thread_name,
            "projects": projects,
            "desktop_refresh": self._deep_link_refresh_codex_desktop(target_thread_id, force=True),
        }

    def archive_thread(self, thread_id: str) -> dict:
        target_thread_id = str(thread_id).strip()
        if not target_thread_id:
            raise CodexLiveBridgeError("Missing thread_id")
        payload = self._run_probe(
            "archive",
            "--thread-id",
            target_thread_id,
            timeout=min(self.timeout, 15.0),
        )
        self._clear_thread_caches()
        projects = self.list_projects(force=True)
        selection = self._load_selection_data()
        replacement: dict | None = None
        desktop_refresh = None
        if selection and selection.get("thread_id") == target_thread_id:
            replacement = next(
                (
                    item
                    for item in self._list_raw_threads(50, force=True)
                    if item.get("id") and item.get("id") != target_thread_id and not (item.get("archived") or item.get("isArchived"))
                ),
                None,
            )
            if replacement and replacement.get("id"):
                self._persist_selection(
                    thread_id=str(replacement["id"]),
                    thread_name=replacement.get("name"),
                    cwd=replacement.get("cwd"),
                    status=self._status_label(replacement.get("status")) or "idle",
                    pending_materialization=False,
                )
                desktop_refresh = self._deep_link_refresh_codex_desktop(str(replacement["id"]), force=True)
        return {
            "ok": True,
            "thread_id": target_thread_id,
            "thread_name": payload.get("threadName"),
            "cwd": payload.get("cwd"),
            "archived": True,
            "replacement_thread_id": replacement.get("id") if replacement else None,
            "replacement_thread_name": replacement.get("name") if replacement else None,
            "projects": projects,
            "desktop_refresh": desktop_refresh,
        }

    def unarchive_thread(self, thread_id: str) -> dict:
        target_thread_id = str(thread_id).strip()
        if not target_thread_id:
            raise CodexLiveBridgeError("Missing thread_id")
        payload = self._run_probe(
            "unarchive",
            "--thread-id",
            target_thread_id,
            timeout=min(self.timeout, 15.0),
        )
        self._clear_thread_caches()
        projects = self.list_projects(force=True)
        return {
            "ok": True,
            "thread_id": target_thread_id,
            "thread_name": payload.get("threadName"),
            "cwd": payload.get("cwd"),
            "archived": False,
            "projects": projects,
            "desktop_refresh": self._deep_link_refresh_codex_desktop(target_thread_id, force=True),
        }

    def set_thread_pinned(self, thread_id: str, pinned: bool) -> dict:
        target_thread_id = str(thread_id).strip()
        if not target_thread_id:
            raise CodexLiveBridgeError("Missing thread_id")
        payload = self._load_codex_global_state()
        pinned_thread_ids = payload.get("pinned-thread-ids")
        if not isinstance(pinned_thread_ids, list):
            pinned_thread_ids = []
        normalized_ids = [str(item).strip() for item in pinned_thread_ids if str(item).strip() and str(item).strip() != target_thread_id]
        if pinned:
            normalized_ids.insert(0, target_thread_id)
        payload["pinned-thread-ids"] = normalized_ids
        self._write_codex_global_state(payload)
        self._clear_thread_caches()
        projects = self.list_projects(force=True)
        return {
            "ok": True,
            "thread_id": target_thread_id,
            "pinned": pinned,
            "projects": projects,
            "desktop_refresh": self._deep_link_refresh_codex_desktop(target_thread_id, force=True),
        }

    def pin_thread(self, thread_id: str) -> dict:
        return self.set_thread_pinned(thread_id, True)

    def unpin_thread(self, thread_id: str) -> dict:
        return self.set_thread_pinned(thread_id, False)

    def fork_thread(self, thread_id: str, cwd: str | None = None) -> dict:
        source_thread_id = str(thread_id).strip()
        if not source_thread_id:
            raise CodexLiveBridgeError("Missing thread_id")
        args = ["--thread-id", source_thread_id]
        if cwd:
            args.extend(["--cwd", str(cwd)])
        payload = self._run_probe(
            "fork",
            *args,
            timeout=min(max(self.timeout, 20.0), 45.0),
        )
        new_thread_id = str(payload.get("threadId") or "").strip()
        if not new_thread_id:
            raise CodexLiveBridgeError("Codex app-server 沒有回傳 fork 後的 threadId")
        thread_cwd = payload.get("cwd") or cwd
        thread_name = payload.get("threadName")
        self._clear_thread_caches()
        self._persist_selection(
            thread_id=new_thread_id,
            thread_name=thread_name,
            cwd=thread_cwd,
            status="idle",
            pending_materialization=False,
        )
        projects = self.list_projects(force=True)
        return {
            "ok": True,
            "thread_id": new_thread_id,
            "thread_name": thread_name,
            "cwd": thread_cwd,
            "forked_from_id": payload.get("forkedFromId") or source_thread_id,
            "project_title": self._project_title_for_thread({"id": new_thread_id, "name": thread_name, "cwd": thread_cwd}) or (thread_name or "目前專案"),
            "projects": projects,
            "desktop_refresh": self._deep_link_refresh_codex_desktop(new_thread_id, force=True),
        }

    def create_project(self, cwd: str) -> dict:
        raw_cwd = str(cwd).strip()
        input_path = Path(raw_cwd).expanduser()
        target_path = self._resolve_project_path(raw_cwd)
        if not target_path.exists() and raw_cwd and not input_path.is_absolute():
            target_path.mkdir(parents=True, exist_ok=True)
        if not target_path.exists() or not target_path.is_dir():
            raise CodexLiveBridgeError(f"找不到項目資料夾：{target_path}")
        title = target_path.name
        self._save_project(title=title, cwd=str(target_path))
        projects = self.list_projects()
        normalized_target = str(target_path)
        exact = next((item for item in projects if str(item.get("cwd") or "").strip() == normalized_target), None)
        if exact and exact.get("thread_id"):
            snapshot = self.open_project(str(exact["project_id"]))
            snapshot["project_id"] = str(exact["project_id"])
            return snapshot

        created = self.create_thread(str(target_path))
        created_thread_id = str(created["thread_id"])
        snapshot = self.open_thread(created_thread_id)
        snapshot["project_id"] = self._project_id(title, str(target_path))
        snapshot["state"]["project_title"] = title
        snapshot["state"]["cwd"] = str(target_path)
        return snapshot

    def rename_project(self, cwd: str, title: str) -> dict:
        raw_cwd = str(cwd).strip()
        next_title = str(title).strip()
        if not raw_cwd:
            raise CodexLiveBridgeError("Missing cwd")
        if not next_title:
            raise CodexLiveBridgeError("Missing title")
        target_cwd = str(Path(raw_cwd).expanduser()).rstrip("/")
        state = self._load_codex_global_state()
        labels = state.get("electron-workspace-root-labels")
        if not isinstance(labels, dict):
            labels = {}
        labels[target_cwd] = next_title
        state["electron-workspace-root-labels"] = labels
        roots = state.get("electron-saved-workspace-roots")
        if not isinstance(roots, list):
            roots = []
        if target_cwd not in [str(item).rstrip("/") for item in roots]:
            roots.append(target_cwd)
        state["electron-saved-workspace-roots"] = roots
        project_order = state.get("project-order")
        if isinstance(project_order, list):
            normalized_order = [str(item).rstrip("/") for item in project_order if str(item or "").strip()]
            if target_cwd not in normalized_order:
                normalized_order.append(target_cwd)
            state["project-order"] = normalized_order
        self._write_codex_global_state(state)
        self._save_project(title=next_title, cwd=target_cwd)
        self._clear_thread_caches()
        projects = self.list_projects(force=True)
        selection = self._load_selection_data()
        selected_thread_id = str((selection or {}).get("thread_id") or "").strip()
        return {
            "ok": True,
            "cwd": target_cwd,
            "project_title": next_title,
            "projects": projects,
            "desktop_refresh": self._deep_link_refresh_codex_desktop(selected_thread_id, force=True) if selected_thread_id else None,
        }

    def remove_project(self, cwd: str) -> dict:
        raw_cwd = str(cwd).strip()
        if not raw_cwd:
            raise CodexLiveBridgeError("Missing cwd")
        target_cwd = str(Path(raw_cwd).expanduser()).rstrip("/")
        state = self._load_codex_global_state()
        for key in ("electron-saved-workspace-roots", "project-order"):
            value = state.get(key)
            if isinstance(value, list):
                state[key] = [item for item in value if str(item).rstrip("/") != target_cwd]
        labels = state.get("electron-workspace-root-labels")
        if isinstance(labels, dict):
            labels.pop(target_cwd, None)
            state["electron-workspace-root-labels"] = labels
        self._write_codex_global_state(state)
        projects = [
            item
            for item in self._load_saved_projects()
            if str(item.get("cwd") or "").rstrip("/") != target_cwd
        ]
        self.projects_path.parent.mkdir(parents=True, exist_ok=True)
        self.projects_path.write_text(json.dumps(projects, ensure_ascii=False), encoding="utf-8")
        self._clear_thread_caches()
        selection = self._load_selection_data()
        selected_thread_id = str((selection or {}).get("thread_id") or "").strip()
        return {
            "ok": True,
            "cwd": target_cwd,
            "removed": True,
            "projects": self.list_projects(force=True),
            "desktop_refresh": self._deep_link_refresh_codex_desktop(selected_thread_id, force=True) if selected_thread_id else None,
        }

    def _resolve_project_path(self, raw_cwd: str) -> Path:
        normalized = raw_cwd.strip()
        if not normalized:
            return self.project_root
        candidate = Path(normalized).expanduser()
        if candidate.is_absolute():
            return candidate
        return (self.project_root / candidate).resolve()

    def open_project(self, project_id: str, limit: int = 500) -> dict:
        if project_id.startswith("thread:"):
            thread_id = project_id.removeprefix("thread:").strip()
            if not thread_id:
                raise CodexLiveBridgeError(f"找不到指定的項目：{project_id}")
            snapshot = self.open_thread(thread_id, limit=limit)
            snapshot["project_id"] = project_id
            return snapshot

        projects = self.list_projects(force=True)
        project = next((item for item in projects if item.get("project_id") == project_id), None)
        if not project:
            raise CodexLiveBridgeError(f"找不到指定的項目：{project_id}")

        thread_id = project.get("thread_id")
        if thread_id:
            snapshot = self.open_thread(str(thread_id), limit=limit)
            snapshot["project_id"] = project_id
            return snapshot

        cwd = project.get("cwd")
        if not cwd:
            raise CodexLiveBridgeError(f"項目尚未綁定工作目錄：{project.get('title') or project_id}")

        created = self.create_thread(str(cwd))
        created_thread_id = str(created["thread_id"])
        snapshot = self.open_thread(created_thread_id, limit=limit)
        snapshot["project_id"] = self._project_id(project.get("title"), cwd)
        return snapshot

    def open_thread(self, thread_id: str, limit: int = 500) -> dict:
        self.select_thread(thread_id)
        desktop_refresh = self._deep_link_refresh_codex_desktop(thread_id, force=True)
        self._schedule_post_turn_refresh(thread_id)
        for _ in range(6):
            snapshot = self.get_snapshot_payload(limit=limit)
            if snapshot["state"].get("thread_id") == thread_id:
                snapshot["ok"] = True
                snapshot["desktop_refresh"] = desktop_refresh
                return snapshot
            time.sleep(0.25)
        snapshot = self.get_thread_snapshot_payload(thread_id, limit=limit)
        snapshot["ok"] = True
        snapshot["desktop_refresh"] = desktop_refresh
        return snapshot

    def refresh_desktop_thread(self, thread_id: str | None = None, limit: int = 500, force: bool = True) -> dict:
        target_thread_id = str(thread_id or "").strip()
        if not target_thread_id:
            target_thread_id = self._get_selected_thread().thread_id
        selected = self.select_thread(target_thread_id)
        self._schedule_post_turn_refresh(target_thread_id)
        refresh = self._deep_link_refresh_codex_desktop(target_thread_id, force=force)
        state = selected.get("state", selected) if isinstance(selected, dict) else {}
        return {
            "ok": True,
            "state": {
                "thread_id": target_thread_id,
                "thread_name": state.get("thread_name"),
                "project_title": state.get("project_title") or state.get("thread_name") or "目前專案",
                "cwd": state.get("cwd"),
                "project_root": str(self.project_root),
                "mobile_default_cwd": self.thread_cwd,
            },
            "desktop_refresh": refresh,
        }

    def _activate_codex_desktop(self) -> None:
        try:
            subprocess.run(["open", "-a", "Codex"], timeout=3, check=False)
        except Exception as exc:
            logger.debug("Codex desktop activation skipped: %s", exc)

    def _deep_link_refresh_codex_desktop(self, thread_id: str, force: bool = True) -> dict[str, Any]:
        if not thread_id:
            return {"ok": False, "method": "codex_deeplink", "error": "missing thread id"}
        renderer_reload: dict[str, Any] | None = None
        try:
            self._activate_codex_desktop()
            alternate_thread_id = self._alternate_thread_id_for_desktop_refresh(thread_id) if force else None
            if force:
                if alternate_thread_id:
                    subprocess.run(["open", f"codex://threads/{quote(alternate_thread_id, safe='')}"], timeout=3, check=False)
                else:
                    subprocess.run(["open", f"codex://threads/{quote(thread_id, safe='')}"], timeout=3, check=False)
                time.sleep(0.45)
            subprocess.run(["open", f"codex://threads/{quote(thread_id, safe='')}"], timeout=3, check=False)
            if force:
                time.sleep(0.45)
                renderer_reload = self._reload_codex_renderer(thread_id)
            return {
                "ok": True,
                "method": "codex_deeplink",
                "thread_id": thread_id,
                "forced": force,
                "alternate_thread_id": alternate_thread_id,
                "renderer_reload": renderer_reload,
            }
        except Exception as exc:
            logger.debug("Codex desktop deep-link refresh failed for %s: %s", thread_id, exc)
            return {"ok": False, "method": "codex_deeplink", "thread_id": thread_id, "error": str(exc)}

    def _reload_codex_renderer(self, thread_id: str) -> dict[str, Any]:
        """Ask the already-open Codex window to remount its live view.

        Codex.app can keep showing stale thread content after app-server receives
        a mobile turn. Re-opening the same deep link is sometimes ignored by the
        renderer, so we trigger a normal Cmd+R reload and then open the target
        thread again. If Accessibility permissions are unavailable, this fails
        softly and the deep link path still runs.
        """
        script = """
        tell application "Codex" to activate
        delay 0.15
        tell application "System Events"
          keystroke "r" using command down
        end tell
        """
        try:
            result = subprocess.run(
                ["/usr/bin/osascript", "-e", script],
                timeout=4,
                check=False,
                capture_output=True,
                text=True,
            )
            time.sleep(0.6)
            subprocess.run(["open", f"codex://threads/{quote(thread_id, safe='')}"], timeout=3, check=False)
            return {
                "ok": result.returncode == 0,
                "method": "codex_renderer_reload",
                "returncode": result.returncode,
                "stderr": (result.stderr or "").strip()[:240],
            }
        except Exception as exc:
            logger.debug("Codex renderer reload failed for %s: %s", thread_id, exc)
            return {"ok": False, "method": "codex_renderer_reload", "error": str(exc)}

    def _alternate_thread_id_for_desktop_refresh(self, target_thread_id: str) -> str | None:
        try:
            payload = self._run_probe("list", "--limit", "50", timeout=min(self.timeout, 8.0))
        except Exception as exc:
            logger.debug("Unable to list alternate Codex threads for desktop refresh: %s", exc)
            return None
        threads = payload.get("data") if isinstance(payload, dict) else None
        if not isinstance(threads, list):
            return None
        candidates = [
            thread
            for thread in threads
            if isinstance(thread, dict) and thread.get("id") and thread.get("id") != target_thread_id
        ]
        if not candidates:
            return None
        candidates.sort(key=lambda thread: int(thread.get("updatedAt") or 0), reverse=True)
        return str(candidates[0]["id"])

    def _schedule_codex_desktop_refresh(self, thread_id: str, delay: float = 1.2) -> None:
        if not thread_id or not self.settings.codexbot_auto_refresh_codex_desktop:
            return

        def worker() -> None:
            time.sleep(delay)
            with self._desktop_refresh_lock:
                now = time.time()
                if now - self._last_desktop_refresh_at < 1.0:
                    return
                self._last_desktop_refresh_at = now
            self._deep_link_refresh_codex_desktop(thread_id, force=True)

        threading.Thread(target=worker, name=f"codexmobile-desktop-refresh-{thread_id[:8]}", daemon=True).start()

    def _schedule_post_turn_refresh(self, thread_id: str) -> None:
        if not thread_id:
            return

        def worker() -> None:
            desktop_refreshed = False
            for delay in (1.5, 3.5, 7.0, 14.0, 28.0):
                time.sleep(delay)
                try:
                    refresh_payload = self._run_probe(
                        "materialize",
                        "--json",
                        "--thread-id",
                        thread_id,
                        timeout=20,
                    )
                    if self._snapshot_lock.acquire(blocking=False):
                        try:
                            payload = self._build_thread_snapshot_payload(thread_id, limit=180)
                            self._store_snapshot_cache(f"thread:{thread_id}:180", payload)
                        finally:
                            self._snapshot_lock.release()
                    if refresh_payload.get("materialized") and not desktop_refreshed:
                        desktop_refreshed = True
                except Exception as exc:
                    logger.debug("post-turn thread refresh failed for %s: %s", thread_id, exc)

        threading.Thread(target=worker, name=f"codexmobile-refresh-{thread_id[:8]}", daemon=True).start()

    def _score_project_candidate(self, item: dict[str, Any]) -> int:
        score = 0
        cwd = str(item.get("cwd") or "")
        title = str(item.get("title") or "").strip().casefold()
        thread_name = str(item.get("thread_name") or "").strip().casefold()
        if item.get("selected"):
            score += 100
        if thread_name and thread_name == title:
            score += 60
        if cwd and "/.codex/worktrees/" not in cwd:
            score += 40
        if item.get("thread_id"):
            score += 20
        if item.get("status") == "idle":
            score += 10
        return score

    def _get_selected_thread(self) -> LiveThreadSelection:
        selection = self._load_selection_data()
        threads = self._refresh_raw_threads(limit=50, selection=selection, timeout=min(self.timeout, 2.0))
        selected = self._select_thread(threads)
        if not selected:
            selected_id = selection.get("thread_id") if selection else None
            if selected_id and self._selection_is_trusted(selection, threads):
                selected = self._read_thread_by_id(selected_id, timeout=min(self.timeout, 3.0))
            if not selected and selected_id:
                local_row = self._local_thread_row(str(selected_id))
                if local_row:
                    local_row = self._repair_stale_local_selection(selection or {}, local_row)
                    return self._live_selection_from_local_row(local_row, selection or {})
        if not selected and selection and self._selection_is_trusted(selection, threads):
            return self._live_selection_from_data(selection)
        if not selected:
            raise CodexLiveBridgeError(
                f"找不到符合條件的 Codex 會話。cwd={self.thread_cwd!r}, name={self.thread_name!r}"
            )
        self._persist_selection(
            thread_id=selected["id"],
            thread_name=selected.get("name"),
            cwd=selected.get("cwd"),
            status=self._status_label(selected.get("status")) or "idle",
            pending_materialization=False,
        )
        return LiveThreadSelection(
            thread_id=selected["id"],
            thread_name=selected.get("name"),
            cwd=selected.get("cwd"),
            updated_at=selected.get("updatedAt"),
            status=self._status_label(selected.get("status")),
        )

    def _get_selected_thread_fast(self) -> LiveThreadSelection:
        selection = self._load_selection_data()
        selected_id = str((selection or {}).get("thread_id") or "").strip()
        if selected_id:
            local_row = self._local_thread_row(selected_id)
            if local_row:
                local_row = self._repair_stale_local_selection(selection or {}, local_row)
                return self._live_selection_from_local_row(local_row, selection or {})
            cached_threads = self._cached_or_selected_threads(selection)
            if self._selection_is_trusted(selection, cached_threads):
                return self._live_selection_from_data(selection)
        return self._get_selected_thread()

    def _read_selected_thread(self) -> dict:
        if self._thread_cache and (time.time() - self._thread_cache_at) < self._thread_cache_ttl:
            return self._thread_cache
        thread = self._get_selected_thread()
        self._thread_cache = self._read_thread_by_id(thread.thread_id, timeout=min(self.timeout, 5.0)) or {
            "id": thread.thread_id,
            "name": thread.thread_name,
            "cwd": thread.cwd,
            "updatedAt": thread.updated_at,
            "status": {"type": thread.status or "idle"},
            "turns": [],
        }
        self._thread_cache_at = time.time()
        return self._thread_cache

    def _read_thread_by_id(
        self,
        thread_id: str,
        timeout: float | None = None,
        *,
        allow_rollout_fallback: bool = True,
    ) -> dict | None:
        try:
            payload = self._run_probe("read", "--thread-id", thread_id, timeout=timeout)
        except CodexLiveBridgeError:
            if allow_rollout_fallback:
                return self._read_thread_from_rollout(thread_id)
            return None
        thread = payload.get("thread")
        if isinstance(thread, dict) and thread.get("turns"):
            return thread
        # Codex app-server can occasionally return an empty/partial read while
        # the local rollout still has the full conversation. Use that as a
        # read-only fallback so mobile does not show an empty thread.
        if allow_rollout_fallback:
            return self._read_thread_from_rollout(thread_id) or thread
        return thread

    def _read_thread_from_rollout(
        self,
        thread_id: str,
        max_items: int = 220,
        max_bytes: int = 96 * 1024 * 1024,
    ) -> dict | None:
        row = self._local_thread_row(thread_id)
        if not row:
            return None
        rollout_path = Path(str(row.get("rollout_path") or "")).expanduser()
        if not rollout_path.exists() or not rollout_path.is_file():
            return None
        try:
            stat = rollout_path.stat()
        except OSError:
            return None

        fingerprint = (int(stat.st_size), int(stat.st_mtime_ns))
        cache_key = f"{thread_id}:{int(max_items)}:{int(max_bytes)}"
        cached = self._rollout_thread_cache.get(cache_key)
        if cached is not None:
            cached_at, cached_fingerprint, cached_thread = cached
            if cached_fingerprint == fingerprint and time.time() - cached_at < self._rollout_thread_cache_ttl:
                return cached_thread

        items: list[dict[str, Any]] = []
        file_change_texts: list[str] = []
        items_by_turn: dict[str, list[dict[str, Any]]] = {}
        turn_epochs: dict[str, int] = {}
        turn_durations_ms: dict[str, int] = {}
        row_cwd = str(row.get("cwd") or "")
        turn_context_cwds = self._rollout_turn_context_cwds(
            rollout_path,
            max_bytes=max_bytes,
        )
        excluded_turn_ids = self._rollout_turns_outside_thread_cwd(
            rollout_path,
            thread_cwd=row_cwd,
            target_items=max_items,
            max_bytes=max(max_bytes, 64 * 1024 * 1024),
        )
        excluded_turn_windows = self._rollout_turn_exclusion_windows(
            rollout_path,
            excluded_turn_ids,
            max_bytes=max(max_bytes, 64 * 1024 * 1024),
        )
        pending_agent_items: list[tuple[dict[str, Any], int]] = []
        current_reverse_turn_id: str | None = None
        seen: set[tuple[str, ...]] = set()
        latest_epoch = int(row.get("updated_at") or time.time())

        def add_item_to_turn(turn_id: str, item: dict[str, Any], epoch: int) -> None:
            items_by_turn.setdefault(turn_id, []).append(item)
            turn_epochs[turn_id] = max(turn_epochs.get(turn_id, 0), epoch)

        def flush_pending_agent_items(reason_index: int) -> None:
            nonlocal pending_agent_items
            if not pending_agent_items:
                return
            fallback_turn_id = f"{thread_id}-rollout-pending-agent-{reason_index}"
            for pending_item, pending_epoch in pending_agent_items:
                add_item_to_turn(fallback_turn_id, pending_item, pending_epoch)
            pending_agent_items = []

        for index, entry in enumerate(
            self._iter_rollout_tail_entries(
                rollout_path,
                target_items=max_items,
                max_bytes=max_bytes,
            )
        ):
            if not self._rollout_entry_matches_thread_cwd(
                entry,
                row_cwd,
                turn_context_cwds,
                excluded_turn_ids,
                excluded_turn_windows,
            ):
                continue
            latest_epoch = max(latest_epoch, self._epoch_from_iso(entry.get("timestamp")) or latest_epoch)
            payload = entry.get("payload")
            if (
                entry.get("type") == "event_msg"
                and isinstance(payload, dict)
                and payload.get("type") == "task_complete"
            ):
                raw_turn_id = self._rollout_entry_raw_turn_id(entry)
                duration_ms = self._duration_ms(payload.get("duration_ms"))
                if raw_turn_id and duration_ms is not None:
                    current_reverse_turn_id = raw_turn_id
                    turn_durations_ms[raw_turn_id] = duration_ms
                    completed_epoch = self._epoch_from_iso(entry.get("timestamp")) or self._duration_ms(payload.get("completed_at"))
                    if completed_epoch is not None:
                        turn_epochs[raw_turn_id] = max(turn_epochs.get(raw_turn_id, 0), int(completed_epoch))
            file_change_text = self._rollout_entry_file_change_text(entry)
            if file_change_text:
                file_change_texts.append(file_change_text)
            item = self._rollout_entry_to_thread_item(
                entry,
                index=index,
                seen=seen,
                cwd=row_cwd,
            )
            if item:
                items.append(item)
                item_epoch = self._epoch_from_iso(entry.get("timestamp")) or latest_epoch
                raw_turn_id = self._rollout_entry_raw_turn_id(entry)
                if raw_turn_id:
                    current_reverse_turn_id = raw_turn_id
                    add_item_to_turn(raw_turn_id, item, item_epoch)
                elif item.get("type") == "userMessage":
                    flush_pending_agent_items(index)
                    add_item_to_turn(f"{thread_id}-rollout-item-{index}", item, item_epoch)
                    current_reverse_turn_id = None
                elif item.get("type") in {"agentMessage", "commandExecution", "fileChange", "contextCompaction", "plan"}:
                    if current_reverse_turn_id:
                        add_item_to_turn(current_reverse_turn_id, item, item_epoch)
                    else:
                        pending_agent_items.append((item, item_epoch))
                else:
                    add_item_to_turn(f"{thread_id}-rollout-item-{index}", item, item_epoch)
                if len(items) >= max_items:
                    break

        if not items:
            return None
        if pending_agent_items:
            flush_pending_agent_items(max_items)
        items.reverse()
        inferred_change = self._rollout_inferred_file_change_item(
            items,
            str(row.get("cwd") or ""),
            extra_texts=file_change_texts,
        )
        if inferred_change and not any(item.get("type") == "fileChange" for item in items):
            items.append(inferred_change)
            items_by_turn.setdefault(f"{thread_id}-rollout-fallback", []).append(inferred_change)
            turn_epochs[f"{thread_id}-rollout-fallback"] = latest_epoch
        workspace_change = self._workspace_file_change_item(str(row.get("cwd") or ""))
        if workspace_change and not any(item.get("type") == "fileChange" for item in items):
            items.append(workspace_change)
            items_by_turn.setdefault(f"{thread_id}-rollout-fallback", []).append(workspace_change)
            turn_epochs[f"{thread_id}-rollout-fallback"] = latest_epoch

        turns: list[dict[str, Any]] = []
        for turn_id, turn_items in items_by_turn.items():
            turn_items.reverse()
            turn_items = self._dedupe_rollout_agent_messages(turn_items)
            turn_items = self._merge_rollout_file_change_items(turn_items)
            turn_items.sort(key=self._rollout_turn_item_display_rank)
            turns.append(
                {
                    "id": turn_id,
                    "durationMs": turn_durations_ms.get(turn_id),
                    "completedAt": turn_epochs.get(turn_id) or latest_epoch,
                    "items": turn_items,
                }
            )
        turns.sort(key=lambda turn: int(turn.get("completedAt") or 0))

        thread = {
            "id": thread_id,
            "name": row.get("title"),
            "cwd": row.get("cwd"),
            "updatedAt": latest_epoch,
            "status": {"type": "idle"},
            "_source": "codex-rollout-jsonl",
            "turns": turns,
        }
        self._rollout_thread_cache[cache_key] = (time.time(), fingerprint, thread)
        return thread

    @staticmethod
    def _rollout_turn_item_display_rank(item: dict[str, Any]) -> int:
        item_type = str(item.get("type") or "")
        if item_type == "userMessage":
            return 0
        # Keep assistant text and command events in their rollout order. Ranking
        # commands ahead of text makes mobile show every command at the top of
        # the processed section instead of interleaving them with the narrative.
        if item_type in {"agentMessage", "commandExecution", "contextCompaction", "plan"}:
            return 1
        if item_type == "fileChange":
            return 2
        return 3

    @classmethod
    def _dedupe_rollout_agent_messages(cls, turn_items: list[dict[str, Any]]) -> list[dict[str, Any]]:
        kept: list[dict[str, Any]] = []
        for item in turn_items:
            if item.get("type") != "agentMessage":
                kept.append(item)
                continue
            text = str(item.get("text") or "").strip()
            display_scope = str(item.get("displayScope") or item.get("display_scope") or "")
            normalized = cls._rollout_duplicate_text_key(text)
            if not normalized:
                continue
            duplicate_index: int | None = None
            replace_duplicate = False
            for index, existing in enumerate(kept):
                if existing.get("type") != "agentMessage":
                    continue
                existing_scope = str(existing.get("displayScope") or existing.get("display_scope") or "")
                if existing_scope != display_scope:
                    continue
                existing_text = str(existing.get("text") or "").strip()
                existing_normalized = cls._rollout_duplicate_text_key(existing_text)
                if not existing_normalized:
                    continue
                same_text = normalized == existing_normalized
                current_extends_existing = normalized.startswith(existing_normalized) and len(normalized) > len(existing_normalized)
                existing_extends_current = existing_normalized.startswith(normalized) and len(existing_normalized) > len(normalized)
                if same_text or current_extends_existing or existing_extends_current:
                    duplicate_index = index
                    replace_duplicate = current_extends_existing
                    break
            if duplicate_index is None:
                kept.append(item)
            elif replace_duplicate:
                kept[duplicate_index] = item
        return kept

    @staticmethod
    def _rollout_duplicate_text_key(text: str) -> str:
        return re.sub(r"\s+", " ", text).strip()

    @staticmethod
    def _merge_rollout_file_change_items(turn_items: list[dict[str, Any]]) -> list[dict[str, Any]]:
        file_items = [item for item in turn_items if item.get("type") == "fileChange"]
        if len(file_items) <= 1:
            return turn_items

        merged_by_path: dict[str, dict[str, Any]] = {}
        for item in file_items:
            changes = item.get("changes")
            if not isinstance(changes, list):
                continue
            for change in changes:
                if not isinstance(change, dict):
                    continue
                path = str(change.get("path") or "").strip()
                if not path:
                    continue
                existing = merged_by_path.get(path)
                additions = int(change.get("additions") or 0)
                deletions = int(change.get("deletions") or 0)
                diff = str(change.get("diff") or "")
                if existing:
                    # Desktop Codex summarizes a whole assistant turn. When the
                    # rollout fallback emits several fileChange items for the
                    # same path, accumulate distinct diffs instead of keeping
                    # only the last partial edit.
                    diff_key = diff.strip()
                    seen_diffs = existing.setdefault("_seen_diffs", set())
                    should_count = not diff_key or diff_key not in seen_diffs
                    if should_count:
                        existing["additions"] = int(existing.get("additions") or 0) + additions
                        existing["deletions"] = int(existing.get("deletions") or 0) + deletions
                        if diff_key:
                            seen_diffs.add(diff_key)
                    if diff:
                        previous_diff = str(existing.get("diff") or "")
                        if diff.strip() not in previous_diff:
                            existing["diff"] = f"{previous_diff}\n{diff}".strip() if previous_diff else diff
                else:
                    merged_by_path[path] = {
                        "path": path,
                        "additions": additions,
                        "deletions": deletions,
                        "diff": diff,
                        "_seen_diffs": {diff.strip()} if diff.strip() else set(),
                    }

        if not merged_by_path:
            return [item for item in turn_items if item.get("type") != "fileChange"]

        first_file_index = next(
            (index for index, item in enumerate(turn_items) if item.get("type") == "fileChange"),
            len(turn_items),
        )
        merged_item = {
            "id": "rollout-merged-file-change",
            "type": "fileChange",
            "changes": [
                {key: value for key, value in change.items() if key != "_seen_diffs"}
                for change in list(merged_by_path.values())[:20]
            ],
        }
        merged_turn_items = [item for item in turn_items if item.get("type") != "fileChange"]
        merged_turn_items.insert(min(first_file_index, len(merged_turn_items)), merged_item)
        return merged_turn_items

    def _rollout_inferred_file_change_item(
        self,
        items: list[dict[str, Any]],
        cwd: str,
        *,
        extra_texts: list[str] | None = None,
    ) -> dict[str, Any] | None:
        changes_by_path: dict[str, dict[str, Any]] = {}
        texts: list[str] = []
        for item in items[-160:]:
            if item.get("type") == "agentMessage":
                texts.append(str(item.get("text") or ""))
        texts.extend(extra_texts or [])

        for text in texts:
            for change in self._extract_patch_file_change_details(text, cwd):
                path = str(change.get("path") or "")
                if not path:
                    continue
                additions = int(change.get("additions") or 0)
                deletions = int(change.get("deletions") or 0)
                diff = str(change.get("diff") or "")
                existing = changes_by_path.get(path)
                if existing:
                    diff_key = diff.strip()
                    seen_diffs = existing.setdefault("_seen_diffs", set())
                    if not diff_key or diff_key not in seen_diffs:
                        existing["additions"] = int(existing.get("additions") or 0) + additions
                        existing["deletions"] = int(existing.get("deletions") or 0) + deletions
                        if diff_key:
                            seen_diffs.add(diff_key)
                    if diff:
                        previous_diff = str(existing.get("diff") or "")
                        if diff.strip() not in previous_diff:
                            existing["diff"] = f"{previous_diff}\n{diff}".strip() if previous_diff else diff
                else:
                    changes_by_path[path] = {
                        "path": path,
                        "additions": additions,
                        "deletions": deletions,
                        "diff": diff,
                        "_seen_diffs": {diff.strip()} if diff.strip() else set(),
                    }
            for path, additions, deletions in self._extract_file_change_refs(text, cwd):
                existing = changes_by_path.get(path)
                if existing and str(existing.get("diff") or "").strip():
                    existing["additions"] = max(int(existing.get("additions") or 0), additions)
                    existing["deletions"] = max(int(existing.get("deletions") or 0), deletions)
                else:
                    changes_by_path[path] = {
                        "path": path,
                        "additions": additions,
                        "deletions": deletions,
                        "diff": "",
                    }
        if not changes_by_path:
            return None
        return {
            "id": "rollout-inferred-file-change",
            "type": "fileChange",
            "changes": [
                {key: value for key, value in change.items() if key != "_seen_diffs"}
                for change in list(changes_by_path.values())[:20]
            ],
        }

    @classmethod
    def _extract_patch_file_change_refs(cls, text: str, cwd: str) -> list[tuple[str, int, int]]:
        return [
            (str(change["path"]), int(change["additions"]), int(change["deletions"]))
            for change in cls._extract_patch_file_change_details(text, cwd)
        ]

    @classmethod
    def _extract_patch_file_change_details(cls, text: str, cwd: str) -> list[dict[str, Any]]:
        if not text or "*** " not in text:
            return []

        header = re.compile(r"^\*\*\* (?P<kind>Update|Add|Delete) File: (?P<path>.+?)\s*$")
        changes: list[dict[str, Any]] = []
        current_path = ""
        diff_lines: list[str] = []

        def flush() -> None:
            nonlocal current_path, diff_lines
            if current_path and diff_lines:
                normalized = cls._normalize_rollout_file_path(current_path, cwd)
                diff = "\n".join(diff_lines).rstrip()
                additions = sum(1 for line in diff_lines if line.startswith("+") and not line.startswith("+++"))
                deletions = sum(1 for line in diff_lines if line.startswith("-") and not line.startswith("---"))
                if normalized and (additions or deletions):
                    changes.append(
                        {
                            "path": normalized,
                            "additions": additions,
                            "deletions": deletions,
                            "diff": diff,
                        }
                    )
            current_path = ""
            diff_lines = []

        for line in text.splitlines():
            match = header.match(line.strip())
            if match:
                flush()
                current_path = match.group("path")
                continue
            if not current_path:
                continue
            if line.startswith("*** "):
                continue
            if line.startswith("@@") or line.startswith("+") or line.startswith("-") or line.startswith(" "):
                diff_lines.append(line)
        flush()
        return changes[:20]

    @classmethod
    def _extract_file_change_refs(cls, text: str, cwd: str) -> list[tuple[str, int, int]]:
        if not text:
            return []
        file_ext = r"(?:tsx|ts|py|swift|m|h|js|jsx|json|plist|html|css|scss|sh|md|yml|yaml|toml|env)"
        path_pattern = rf"(?:`|\[|^|[\s:：])(?P<path>(?:/[^\s`'，。；：\]\)]+/)?(?:[A-Za-z0-9_.-]+/)*[A-Za-z0-9_.-]+\.{file_ext})(?:`|\]|\)|\s)"
        stat_pattern = r"(?P<add>\+\d+)\s*(?P<del>-\d+)"
        combined = re.compile(path_pattern + r"[^\n]{0,80}?" + stat_pattern, re.IGNORECASE)
        found: list[tuple[str, int, int]] = []
        seen: set[str] = set()
        for match in combined.finditer(text):
            raw_path = match.group("path")
            path = cls._normalize_rollout_file_path(raw_path, cwd)
            if not path or path in seen:
                continue
            try:
                additions = int(match.group("add").lstrip("+"))
                deletions = abs(int(match.group("del")))
            except (TypeError, ValueError):
                continue
            # Avoid turning generic status text into a file card.
            if additions == 0 and deletions == 0:
                continue
            seen.add(path)
            found.append((path, additions, deletions))
        return found

    @staticmethod
    def _normalize_rollout_file_path(path: str, cwd: str) -> str:
        cleaned = str(path or "").strip().strip("`'\"[](){}，。；：,.;:")
        if not cleaned:
            return ""
        cwd_text = str(cwd or "").rstrip("/")
        if cleaned.startswith("/"):
            if cwd_text and CodexLiveBridge._path_is_inside(cwd_text, cleaned):
                cleaned = cleaned[len(cwd_text) + 1 :] if cleaned.startswith(cwd_text + "/") else Path(cleaned).name
            elif cwd_text:
                return ""
        while cleaned.startswith("./"):
            cleaned = cleaned[2:]
        return cleaned

    def _local_thread_row(self, thread_id: str) -> dict[str, Any] | None:
        state_db = Path.home() / ".codex" / "state_5.sqlite"
        if not state_db.exists():
            return None
        try:
            with sqlite3.connect(f"file:{state_db}?mode=ro", uri=True, timeout=0.25) as conn:
                conn.row_factory = sqlite3.Row
                row = conn.execute(
                    "select id,rollout_path,title,cwd,updated_at,model,reasoning_effort,approval_mode,sandbox_policy from threads where id = ? limit 1",
                    (thread_id,),
                ).fetchone()
        except Exception as exc:
            logger.debug("Codex local state unavailable for thread fallback: %s", exc)
            return None
        return dict(row) if row else None

    def _local_latest_thread_for_cwd(self, cwd: str, current_thread_id: str | None = None) -> dict[str, Any] | None:
        cwd = str(cwd or "").strip()
        if not cwd:
            return None
        state_db = Path.home() / ".codex" / "state_5.sqlite"
        if not state_db.exists():
            return None
        try:
            with sqlite3.connect(f"file:{state_db}?mode=ro", uri=True, timeout=0.25) as conn:
                conn.row_factory = sqlite3.Row
                rows = conn.execute(
                    """
                    select id,rollout_path,title,cwd,updated_at,model,reasoning_effort,approval_mode,sandbox_policy
                    from threads
                    where cwd = ?
                    order by updated_at desc
                    limit 20
                    """,
                    (cwd,),
                ).fetchall()
        except Exception as exc:
            logger.debug("Codex local state unavailable for cwd selection repair: %s", exc)
            return None
        current_id = str(current_thread_id or "").strip()
        for row in rows:
            candidate = dict(row)
            if current_id and str(candidate.get("id") or "") == current_id:
                continue
            return candidate
        return None

    @staticmethod
    def _local_thread_has_broader_access(row: dict[str, Any] | None) -> bool:
        if not row:
            return False
        sandbox = row.get("sandbox_policy")
        sandbox_text = json.dumps(sandbox, ensure_ascii=False) if isinstance(sandbox, (dict, list)) else str(sandbox or "")
        return (
            str(row.get("approval_mode") or "").strip() == "never"
            or "danger-full-access" in sandbox_text
            or str(row.get("reasoning_effort") or "").strip() in {"high", "xhigh"}
        )

    def _repair_stale_local_selection(
        self,
        selection: dict[str, Any],
        local_row: dict[str, Any],
    ) -> dict[str, Any]:
        """Prefer the freshest local desktop thread when a stale mobile test thread polluted selection."""
        cwd = str(local_row.get("cwd") or selection.get("cwd") or "").strip()
        if not cwd:
            return local_row
        current_id = str(local_row.get("id") or selection.get("thread_id") or "").strip()
        latest = self._local_latest_thread_for_cwd(cwd, current_thread_id=current_id)
        if not latest:
            return local_row
        try:
            current_updated = int(local_row.get("updated_at") or 0)
            latest_updated = int(latest.get("updated_at") or 0)
        except (TypeError, ValueError):
            return local_row
        if latest_updated <= current_updated:
            return local_row
        if not self._local_thread_has_broader_access(latest) and self._local_thread_has_broader_access(local_row):
            return local_row
        logger.info(
            "Repairing stale Codex selection %s -> %s for cwd=%s",
            current_id,
            latest.get("id"),
            cwd,
        )
        self._persist_selection(
            thread_id=str(latest.get("id") or current_id),
            thread_name=latest.get("title") or selection.get("thread_name"),
            cwd=latest.get("cwd") or cwd,
            status=selection.get("status") or "idle",
            pending_materialization=False,
        )
        return latest

    def _live_selection_from_local_row(
        self,
        row: dict[str, Any],
        selection: dict[str, Any] | None = None,
    ) -> LiveThreadSelection:
        selection = selection or {}
        return LiveThreadSelection(
            thread_id=str(row.get("id") or selection.get("thread_id") or ""),
            thread_name=row.get("title") or selection.get("thread_name"),
            cwd=row.get("cwd") or selection.get("cwd") or self.thread_cwd,
            updated_at=int(row.get("updated_at") or selection.get("updated_at") or time.time()),
            status=selection.get("status") or "idle",
        )

    def _synthetic_thread_from_local_state(self, thread_id: str) -> dict[str, Any] | None:
        row = self._local_thread_row(thread_id)
        if not row:
            return None
        return {
            "id": str(row.get("id") or thread_id),
            "name": row.get("title"),
            "cwd": row.get("cwd"),
            "updatedAt": int(row.get("updated_at") or time.time()),
            "status": {"type": "idle"},
            "turns": [],
            "_source": "codex-local-state",
        }

    def _iter_rollout_tail_entries(
        self,
        path: Path,
        *,
        target_items: int,
        max_bytes: int,
        chunk_size: int = 512 * 1024,
    ):
        try:
            size = path.stat().st_size
        except OSError:
            return

        offset = size
        scanned = 0
        carry = b""
        yielded_items = 0
        yielded_weight = 0
        max_yielded_items = max(target_items * 4, target_items)

        try:
            with path.open("rb") as handle:
                while offset > 0 and scanned < max_bytes and yielded_weight < target_items and yielded_items < max_yielded_items:
                    read_size = min(chunk_size, offset, max_bytes - scanned)
                    offset -= read_size
                    handle.seek(offset)
                    data = handle.read(read_size)
                    scanned += read_size

                    block = data + carry
                    lines = block.split(b"\n")
                    carry = lines[0] if offset > 0 else b""
                    complete_lines = lines[1:] if offset > 0 else lines

                    for raw_line in reversed(complete_lines):
                        if not raw_line.strip():
                            continue
                        try:
                            entry = json.loads(raw_line.decode("utf-8", errors="replace"))
                        except json.JSONDecodeError:
                            continue
                        if not isinstance(entry, dict):
                            continue
                        payload = entry.get("payload") if isinstance(entry.get("payload"), dict) else {}
                        payload_type = str(payload.get("type") or "")
                        has_turn_duration = (
                            entry.get("type") == "event_msg"
                            and payload_type == "task_complete"
                            and payload.get("duration_ms") is not None
                        )
                        if not (
                            self._rollout_entry_has_visible_message(entry)
                            or self._rollout_entry_has_file_change_evidence(entry)
                            or has_turn_duration
                        ):
                            continue
                        yielded_items += 1
                        # Live rollouts can contain long runs of command events
                        # after the last assistant text. Do not let commands
                        # alone exhaust the tail budget, otherwise mobile sees a
                        # command-only/process-only timeline unlike desktop.
                        if payload_type != "exec_command_end":
                            yielded_weight += 1
                        yield entry
                        if yielded_weight >= target_items or yielded_items >= max_yielded_items:
                            break
        except OSError:
            return

    @staticmethod
    def _path_same_or_nested(left: str, right: str) -> bool:
        if not left or not right:
            return False
        left_path = Path(left).expanduser().resolve(strict=False)
        right_path = Path(right).expanduser().resolve(strict=False)
        if left_path == right_path:
            return True
        try:
            return left_path.is_relative_to(right_path) or right_path.is_relative_to(left_path)
        except ValueError:
            return False

    @staticmethod
    def _path_is_inside(base: str, path: str) -> bool:
        if not base or not path:
            return False
        base_path = Path(base).expanduser().resolve(strict=False)
        target_path = Path(path).expanduser().resolve(strict=False)
        if target_path == base_path:
            return True
        try:
            return target_path.is_relative_to(base_path)
        except ValueError:
            return False

    def _rollout_turn_context_cwds(self, path: Path, *, max_bytes: int) -> dict[str, str]:
        try:
            size = path.stat().st_size
        except OSError:
            return {}
        read_size = min(size, max_bytes)
        try:
            with path.open("rb") as handle:
                handle.seek(max(0, size - read_size))
                data = handle.read(read_size)
        except OSError:
            return {}
        turn_cwds: dict[str, str] = {}
        for raw_line in data.splitlines():
            if not raw_line.strip():
                continue
            try:
                entry = json.loads(raw_line.decode("utf-8", errors="replace"))
            except json.JSONDecodeError:
                continue
            if not isinstance(entry, dict) or entry.get("type") != "turn_context":
                continue
            payload = entry.get("payload")
            if not isinstance(payload, dict):
                continue
            turn_id = str(payload.get("turn_id") or "").strip()
            cwd = str(payload.get("cwd") or "").strip()
            if turn_id and cwd:
                turn_cwds[turn_id] = cwd
        return turn_cwds

    def _rollout_turns_outside_thread_cwd(
        self,
        path: Path,
        *,
        thread_cwd: str,
        target_items: int,
        max_bytes: int,
    ) -> set[str]:
        if not thread_cwd:
            return set()
        try:
            size = path.stat().st_size
        except OSError:
            return set()
        read_size = min(size, max_bytes)
        excluded: set[str] = set()
        try:
            with path.open("rb") as handle:
                handle.seek(max(0, size - read_size))
                data = handle.read(read_size)
        except OSError:
            return set()
        for raw_line in data.splitlines():
            if not raw_line.strip():
                continue
            try:
                entry = json.loads(raw_line.decode("utf-8", errors="replace"))
            except json.JSONDecodeError:
                continue
            if not isinstance(entry, dict):
                continue
            payload = entry.get("payload") if isinstance(entry.get("payload"), dict) else {}
            if entry.get("type") != "event_msg" or payload.get("type") != "exec_command_end":
                continue
            raw_turn_id = self._rollout_entry_raw_turn_id(entry)
            command_cwd = str(payload.get("cwd") or "").strip()
            if raw_turn_id and command_cwd and not self._path_same_or_nested(thread_cwd, command_cwd):
                excluded.add(raw_turn_id)
        return excluded

    def _rollout_turn_exclusion_windows(
        self,
        path: Path,
        excluded_turn_ids: set[str],
        *,
        max_bytes: int,
    ) -> list[tuple[int, int]]:
        if not excluded_turn_ids:
            return []
        try:
            size = path.stat().st_size
        except OSError:
            return []
        read_size = min(size, max_bytes)
        try:
            with path.open("rb") as handle:
                handle.seek(max(0, size - read_size))
                data = handle.read(read_size)
        except OSError:
            return []

        starts: dict[str, int] = {}
        ends: dict[str, int] = {}
        seen_epochs: dict[str, list[int]] = {}
        for raw_line in data.splitlines():
            if not raw_line.strip():
                continue
            try:
                entry = json.loads(raw_line.decode("utf-8", errors="replace"))
            except json.JSONDecodeError:
                continue
            if not isinstance(entry, dict):
                continue
            payload = entry.get("payload") if isinstance(entry.get("payload"), dict) else {}
            raw_turn_id = self._rollout_entry_raw_turn_id(entry)
            if not raw_turn_id or raw_turn_id not in excluded_turn_ids:
                continue
            epoch = self._epoch_from_iso(entry.get("timestamp"))
            if epoch is None:
                continue
            seen_epochs.setdefault(raw_turn_id, []).append(epoch)
            if entry.get("type") == "event_msg" and payload.get("type") == "task_started":
                starts[raw_turn_id] = min(starts.get(raw_turn_id, epoch), epoch)
            if entry.get("type") == "event_msg" and payload.get("type") in {"task_complete", "turn_aborted"}:
                ends[raw_turn_id] = max(ends.get(raw_turn_id, epoch), epoch)

        windows: list[tuple[int, int]] = []
        for raw_turn_id in excluded_turn_ids:
            epochs = seen_epochs.get(raw_turn_id) or []
            if not epochs:
                continue
            start = starts.get(raw_turn_id, min(epochs))
            end = ends.get(raw_turn_id, max(epochs))
            windows.append((start, end))
        return windows

    def _rollout_entry_matches_thread_cwd(
        self,
        entry: dict[str, Any],
        thread_cwd: str,
        turn_context_cwds: dict[str, str],
        excluded_turn_ids: set[str] | None = None,
        excluded_turn_windows: list[tuple[int, int]] | None = None,
    ) -> bool:
        if not thread_cwd:
            return True
        raw_turn_id = self._rollout_entry_raw_turn_id(entry)
        if not raw_turn_id:
            entry_epoch = self._epoch_from_iso(entry.get("timestamp"))
            if entry_epoch is not None:
                for start, end in excluded_turn_windows or []:
                    if start <= entry_epoch <= end:
                        return False
            return True
        if raw_turn_id in (excluded_turn_ids or set()):
            return False
        if not turn_context_cwds:
            return True
        turn_cwd = turn_context_cwds.get(raw_turn_id)
        if not turn_cwd:
            return True
        return self._path_same_or_nested(thread_cwd, turn_cwd)

    def _rollout_entry_has_visible_message(self, entry: dict[str, Any]) -> bool:
        payload = entry.get("payload")
        if not isinstance(payload, dict):
            return False
        entry_type = str(entry.get("type") or "")
        if entry_type == "event_msg":
            payload_type = str(payload.get("type") or "")
            if payload_type in {"user_message", "agent_message", "error"}:
                text = str(payload.get("message") or "").strip()
            elif payload_type == "exec_command_end":
                text = self._rollout_command_text(payload)
            else:
                return False
        elif entry_type == "response_item" and payload.get("type") == "message":
            text = self._response_message_text(payload)
        else:
            return False
        return bool(text and not self._is_internal_context_text(text))

    def _rollout_entry_has_file_change_evidence(self, entry: dict[str, Any]) -> bool:
        return bool(self._rollout_entry_file_change_text(entry))

    @staticmethod
    def _rollout_entry_raw_turn_id(entry: dict[str, Any]) -> str | None:
        payload = entry.get("payload")
        if isinstance(payload, dict):
            raw_turn_id = payload.get("turn_id") or payload.get("turnId")
            if raw_turn_id:
                return str(raw_turn_id)
        raw_turn_id = entry.get("turn_id") or entry.get("turnId")
        if raw_turn_id:
            return str(raw_turn_id)
        return None

    @staticmethod
    def _rollout_entry_turn_id(entry: dict[str, Any], thread_id: str) -> str:
        return CodexLiveBridge._rollout_entry_raw_turn_id(entry) or f"{thread_id}-rollout-fallback"

    def _rollout_entry_file_change_text(self, entry: dict[str, Any]) -> str:
        payload = entry.get("payload")
        if not isinstance(payload, dict):
            return ""

        entry_type = str(entry.get("type") or "")
        payload_type = str(payload.get("type") or "")
        if entry_type == "event_msg" and payload_type == "agent_message":
            text = str(payload.get("message") or "")
        elif entry_type == "response_item" and payload_type == "message":
            text = self._response_message_text(payload)
        elif entry_type == "response_item" and payload_type == "function_call":
            name = str(payload.get("name") or "")
            arguments = payload.get("arguments")
            text = arguments if isinstance(arguments, str) else json.dumps(arguments, ensure_ascii=False)
            if "apply_patch" not in name and "*** Begin Patch" not in text:
                return ""
        elif entry_type == "response_item" and payload_type == "custom_tool_call":
            name = str(payload.get("name") or "")
            tool_input = payload.get("input")
            text = tool_input if isinstance(tool_input, str) else json.dumps(tool_input, ensure_ascii=False)
            if "apply_patch" not in name and "*** Begin Patch" not in text:
                return ""
        elif entry_type == "event_msg" and payload_type == "patch_apply_end":
            parts: list[str] = []
            changes = payload.get("changes")
            if isinstance(changes, dict):
                for raw_path, change in changes.items():
                    if not isinstance(change, dict):
                        continue
                    kind = str(change.get("type") or "update").lower()
                    label = "Add" if kind == "add" else "Delete" if kind == "delete" else "Update"
                    parts.append(f"*** {label} File: {raw_path}")
                    if change.get("unified_diff"):
                        parts.append(str(change["unified_diff"]))
            text = "\n".join(parts)
        else:
            return ""

        if not text:
            return ""
        if (
            "*** Update File:" in text
            or "*** Add File:" in text
            or "*** Delete File:" in text
            or re.search(r"\+\d+\s*-\d+", text)
            or "文件已更改" in text
            or "檔案已更改" in text
            or "已更改" in text
            or "已編輯" in text
            or "已编辑" in text
        ):
            return text
        return ""

    def _rollout_patch_file_change_item(
        self,
        entry: dict[str, Any],
        *,
        index: int,
        cwd: str,
        seen: set[tuple[str, ...]],
    ) -> dict[str, Any] | None:
        text = self._rollout_entry_file_change_text(entry)
        changes = self._extract_patch_file_change_details(text, cwd)
        if not changes:
            return None
        payload = entry.get("payload") if isinstance(entry.get("payload"), dict) else {}
        dedupe_key = ("fileChange", str(payload.get("call_id") or index))
        if dedupe_key in seen:
            return None
        seen.add(dedupe_key)
        return {
            "id": f"rollout-file-change-{index}",
            "type": "fileChange",
            "changes": changes,
        }

    def _rollout_entry_to_thread_item(
        self,
        entry: dict[str, Any],
        *,
        index: int,
        seen: set[tuple[str, ...]],
        cwd: str = "",
    ) -> dict[str, Any] | None:
        payload = entry.get("payload")
        if not isinstance(payload, dict):
            return None

        entry_type = str(entry.get("type") or "")
        if entry_type == "event_msg":
            payload_type = str(payload.get("type") or "")
            if payload_type == "user_message":
                text = str(payload.get("message") or "").strip()
                role = "user"
                display_scope = None
            elif payload_type == "agent_message":
                text = str(payload.get("message") or "").strip()
                role = "assistant"
                display_scope = self._rollout_message_display_scope(payload)
            elif payload_type == "error":
                text = str(payload.get("message") or "").strip()
                role = "assistant"
                display_scope = "process"
            elif payload_type == "exec_command_end":
                return self._rollout_command_item(payload, index=index, seen=seen)
            elif payload_type == "patch_apply_end":
                return self._rollout_patch_file_change_item(entry, index=index, cwd=cwd, seen=seen)
            else:
                return None
        elif entry_type == "response_item" and payload.get("type") == "message":
            role = "assistant" if payload.get("role") == "assistant" else "user"
            text = self._response_message_text(payload) if role == "assistant" else self._response_user_message_text(payload)
            display_scope = self._rollout_message_display_scope(payload) if role == "assistant" else None
        elif entry_type == "response_item" and payload.get("type") == "custom_tool_call":
            # The following patch_apply_end event has the real turn_id and the
            # canonical diff. Rendering the custom_tool_call itself creates a
            # duplicate file card attached to the fallback turn.
            return None
        else:
            return None

        text = text.strip()
        if not text:
            return None
        if role == "assistant":
            text = self._strip_memory_citation_block(text)
            if not text:
                return None
        else:
            text = self._canonical_rollout_user_text(text)
            if not text:
                return None
        if self._is_internal_context_text(text):
            return None
        if role == "assistant" and self._assistant_text_is_file_change_only(text):
            return None
        response_user_images = (
            self._response_user_image_content(payload)
            if role == "user" and entry_type == "response_item"
            else []
        )
        dedupe_text = self._canonical_rollout_user_text(text) if role == "user" else self._canonical_rollout_assistant_text(text)
        image_dedupe_key = ""
        if response_user_images:
            image_dedupe_key = "|".join(
                self._image_source_from_chunk(chunk) for chunk in response_user_images if self._image_source_from_chunk(chunk)
            )
        dedupe_key = (role, dedupe_text, image_dedupe_key)
        if dedupe_key in seen:
            return None
        seen.add(dedupe_key)

        if role == "user":
            content = [{"type": "text", "text": text}]
            content.extend(response_user_images)
            return {
                "id": f"rollout-user-{index}",
                "type": "userMessage",
                "content": content,
            }
        return {
            "id": f"rollout-agent-{index}",
            "type": "agentMessage",
            "text": text,
            **({"displayScope": display_scope} if display_scope else {}),
        }

    @staticmethod
    def _rollout_command_text(payload: dict[str, Any]) -> str:
        command = CodexLiveBridge._rollout_command_string(payload)
        return command.strip()

    @staticmethod
    def _rollout_message_display_scope(payload: dict[str, Any]) -> str | None:
        phase = str(payload.get("phase") or "").strip()
        if phase == "commentary":
            return "process"
        if phase == "final_answer":
            return "message"
        return None

    @staticmethod
    def _rollout_command_string(payload: dict[str, Any]) -> str:
        parsed_cmd = payload.get("parsed_cmd")
        if isinstance(parsed_cmd, list) and parsed_cmd:
            first = parsed_cmd[0]
            if isinstance(first, dict) and first.get("cmd"):
                return str(first["cmd"])
        command = payload.get("command")
        if isinstance(command, list):
            if len(command) >= 3 and str(command[-2]) == "-lc":
                return str(command[-1])
            return " ".join(str(part) for part in command)
        return str(command or "")

    @staticmethod
    def _rollout_duration_ms(value: Any) -> int | None:
        if isinstance(value, dict):
            seconds = value.get("secs", 0)
            nanos = value.get("nanos", 0)
            try:
                return int(float(seconds) * 1000 + float(nanos) / 1_000_000)
            except (TypeError, ValueError):
                return None
        return CodexLiveBridge._duration_ms(value)

    def _rollout_command_item(
        self,
        payload: dict[str, Any],
        *,
        index: int,
        seen: set[tuple[str, ...]],
    ) -> dict[str, Any] | None:
        command = self._rollout_command_string(payload).strip()
        if not command:
            return None
        exit_code = payload.get("exit_code")
        dedupe_key = ("commandExecution", f"{payload.get('timestamp')}:{command}:{exit_code}")
        if dedupe_key in seen:
            return None
        seen.add(dedupe_key)
        parsed_cmd = payload.get("parsed_cmd")
        command_actions = parsed_cmd if isinstance(parsed_cmd, list) else []
        return {
            "id": f"rollout-command-{index}",
            "type": "commandExecution",
            "command": command,
            "cwd": payload.get("cwd"),
            "status": payload.get("status"),
            "exitCode": exit_code,
            "durationMs": self._rollout_duration_ms(payload.get("duration")),
            "commandActions": command_actions,
        }

    def _workspace_file_change_item(self, cwd: str) -> dict[str, Any] | None:
        if not cwd:
            return None
        cached = self._workspace_diff_cache.get(cwd)
        if cached and time.time() - cached[0] < self._workspace_diff_cache_ttl:
            return cached[1]
        try:
            proc = subprocess.run(
                ["git", "-C", cwd, "diff", "--numstat"],
                capture_output=True,
                text=True,
                timeout=1.5,
                check=False,
            )
        except Exception as exc:
            logger.debug("workspace diff fallback unavailable: %s", exc)
            self._workspace_diff_cache[cwd] = (time.time(), None)
            return None
        if proc.returncode != 0:
            self._workspace_diff_cache[cwd] = (time.time(), None)
            return None
        changes: list[dict[str, Any]] = []
        for line in proc.stdout.splitlines():
            parts = line.split("\t")
            if len(parts) < 3:
                continue
            raw_additions, raw_deletions, path = parts[0], parts[1], parts[2]
            try:
                additions = 0 if raw_additions == "-" else int(raw_additions)
                deletions = 0 if raw_deletions == "-" else int(raw_deletions)
            except ValueError:
                additions = deletions = 0
            changes.append(
                {
                    "path": path,
                    "additions": additions,
                    "deletions": deletions,
                    "diff": "",
                }
            )
            if len(changes) >= 20:
                break
        item = {
            "id": "rollout-worktree-file-change",
            "type": "fileChange",
            "changes": changes,
        } if changes else None
        self._workspace_diff_cache[cwd] = (time.time(), item)
        return item

    @staticmethod
    def _response_message_text(payload: dict[str, Any]) -> str:
        parts: list[str] = []
        for chunk in payload.get("content") or []:
            if not isinstance(chunk, dict):
                continue
            chunk_type = str(chunk.get("type") or "")
            if chunk_type in {"text", "input_text", "output_text"} and chunk.get("text"):
                parts.append(str(chunk["text"]))
        return "\n".join(part.rstrip() for part in parts).strip()

    @classmethod
    def _response_user_message_text(cls, payload: dict[str, Any]) -> str:
        parts: list[str] = []
        for chunk in payload.get("content") or []:
            if not isinstance(chunk, dict):
                continue
            chunk_type = str(chunk.get("type") or "")
            text = str(chunk.get("text") or "") if chunk_type in {"text", "input_text"} else ""
            if not text:
                continue
            stripped = text.strip()
            if stripped.startswith("<image name="):
                continue
            if stripped == "</image>":
                continue
            if stripped.startswith("Codex could not read the local image at "):
                continue
            parts.append(text.rstrip())
        return cls._canonical_rollout_user_text("\n".join(parts))

    @staticmethod
    def _response_user_image_content(payload: dict[str, Any]) -> list[dict[str, Any]]:
        content: list[dict[str, Any]] = []
        seen: set[str] = set()
        for chunk in payload.get("content") or []:
            if not isinstance(chunk, dict):
                continue
            raw_type = str(chunk.get("type") or "").strip().lower()
            if raw_type not in {"image", "localimage", "local_image", "input_image", "image_url", "inputimage"}:
                continue
            source = CodexLiveBridge._image_source_from_chunk(chunk)
            if not source or source in seen:
                continue
            seen.add(source)
            content.append({"type": "input_image", "image_url": source})
        return content

    @staticmethod
    def _canonical_rollout_user_text(text: str) -> str:
        parts: list[str] = []
        for line in text.splitlines():
            stripped = line.strip()
            if stripped.startswith("<image name="):
                continue
            if stripped == "</image>":
                continue
            if stripped.startswith("Codex could not read the local image at "):
                continue
            parts.append(line.rstrip())
        cleaned = "\n".join(parts).strip()
        for marker in ("## My request for Codex:", "## My request:"):
            marker_index = cleaned.find(marker)
            if marker_index >= 0:
                return cleaned[marker_index + len(marker):].strip()
        return cleaned

    @staticmethod
    def _assistant_text_is_file_change_only(text: str) -> bool:
        stripped = text.strip()
        if not stripped:
            return False
        if stripped.startswith(("檔案變更：", "文件變更：", "文件已更改", "檔案已更改")):
            return True
        return bool(re.match(r"^\d+\s*[個个]\s*文件已更改\b", stripped))

    @classmethod
    def _canonical_rollout_assistant_text(cls, text: str) -> str:
        return cls._strip_memory_citation_block(text).strip()

    @staticmethod
    def _strip_memory_citation_block(text: str) -> str:
        return re.sub(r"\n*<oai-mem-citation>.*?</oai-mem-citation>\s*$", "", text, flags=re.DOTALL).strip()

    @staticmethod
    def _is_internal_context_text(text: str) -> bool:
        stripped = text.lstrip()
        internal_prefixes = (
            "<permissions instructions>",
            "<environment_context>",
            "<app-context>",
            "<collaboration_mode>",
            "<personality_spec>",
            "<apps_instructions>",
            "<skills_instructions>",
            "<plugins_instructions>",
        )
        return stripped.startswith(internal_prefixes)

    @staticmethod
    def _epoch_from_iso(value: Any) -> int | None:
        if not value:
            return None
        try:
            return int(datetime.fromisoformat(str(value).replace("Z", "+00:00")).timestamp())
        except ValueError:
            return None

    def _cached_generation_settings_or_fallback(self, thread_id: str | None) -> dict[str, Any] | None:
        if not thread_id:
            return None
        cached = self._generation_settings_cache.get(thread_id)
        if cached and (time.time() - cached[0]) <= self._generation_settings_cache_ttl:
            return cached[1]
        return self._fallback_generation_settings(thread_id)

    def _cached_runtime_or_fallback(self, thread_id: str | None, reason: str) -> dict[str, Any]:
        if (
            thread_id
            and self._runtime_cache
            and time.time() - self._runtime_cache_at < self._runtime_cache_ttl
            and self._runtime_cache.get("thread_id") == thread_id
        ):
            return self._runtime_cache
        return self._fallback_runtime_payload(thread_id, reason)

    def _generation_settings_for_thread(self, thread_id: str | None) -> dict[str, Any] | None:
        if not thread_id:
            return None
        cached = self._generation_settings_cache.get(thread_id)
        if cached and (time.time() - cached[0]) <= self._generation_settings_cache_ttl:
            return cached[1]
        failed_at = self._generation_settings_failed_at.get(thread_id, 0.0)
        if failed_at and (time.time() - failed_at) <= self._generation_settings_failed_backoff:
            return self._fallback_generation_settings(thread_id)
        try:
            payload = self._run_probe("settings", "--thread-id", thread_id, timeout=min(self.timeout, 1.5))
        except CodexLiveBridgeError:
            self._generation_settings_failed_at[thread_id] = time.time()
            return self._fallback_generation_settings(thread_id)
        if payload.get("resumeUnavailable"):
            settings = self._fallback_generation_settings(thread_id)
            self._generation_settings_cache[thread_id] = (time.time(), settings)
            self._generation_settings_failed_at.pop(thread_id, None)
            return settings
        approval_policy = payload.get("approvalPolicy")
        sandbox = payload.get("sandbox") if isinstance(payload.get("sandbox"), dict) else None
        sandbox_text = json.dumps(sandbox or {}, ensure_ascii=False).lower()
        payload_permission = str(payload.get("permission") or "").strip()
        permission_is_full = (
            payload_permission.lower() in {"fullaccess", "full_access", "full-access", "dangerfullaccess", "danger_full_access", "danger-full-access"}
            or "full" in payload_permission.lower()
            or "danger" in payload_permission.lower()
            or approval_policy == "never"
            or "danger" in sandbox_text
            or "fullaccess" in sandbox_text
            or "full-access" in sandbox_text
            or "full_access" in sandbox_text
        )
        permission = "fullAccess" if permission_is_full else "default"
        fallback_settings = self._fallback_generation_settings(thread_id)
        if permission != "fullAccess" and fallback_settings.get("permission") == "fullAccess":
            permission = "fullAccess"
            approval_policy = "never"
            sandbox = fallback_settings.get("sandbox") if isinstance(fallback_settings.get("sandbox"), dict) else {"type": "dangerFullAccess"}
        settings = {
            "thread_id": payload.get("threadId") or thread_id,
            "model": payload.get("model"),
            "reasoning": payload.get("effort"),
            "permission": permission,
            "approval_policy": approval_policy,
            "sandbox": sandbox,
        }
        self._generation_settings_cache[thread_id] = (time.time(), settings)
        self._generation_settings_failed_at.pop(thread_id, None)
        return settings

    def _runtime_payload_for_thread(self, thread_id: str | None, refresh: bool = False) -> dict[str, Any]:
        if (
            thread_id
            and not refresh
            and self._runtime_cache
            and time.time() - self._runtime_cache_at < self._runtime_cache_ttl
            and self._runtime_cache.get("thread_id") == thread_id
        ):
            return self._runtime_cache
        if (
            thread_id
            and not refresh
            and self._runtime_failed_at
            and (time.time() - self._runtime_failed_at) <= self._runtime_failed_backoff
        ):
            return self._fallback_runtime_payload(thread_id, f"runtime-backoff:{int(time.time() - self._runtime_failed_at)}s")
        args = ["runtime"]
        if thread_id:
            args.extend(["--thread-id", thread_id])
        try:
            payload = self._run_probe(*args, timeout=min(self.timeout, 2.0), bypass_backoff=refresh)
            self._runtime_failed_at = 0.0
        except CodexLiveBridgeError as exc:
            detail = str(exc)
            if (
                refresh
                and self._should_restart_app_server_for_probe("runtime", detail)
                and self._recover_app_server_once(f"runtime probe failed: {detail[:200]}")
            ):
                try:
                    payload = self._run_probe(
                        *args,
                        timeout=min(max(self.timeout, 3.0), 4.0),
                        bypass_backoff=True,
                    )
                    self._runtime_failed_at = 0.0
                except CodexLiveBridgeError as retry_exc:
                    detail = str(retry_exc)
                    self._runtime_failed_at = time.time()
                    payload = self._fallback_runtime_payload(thread_id, detail)
            else:
                self._runtime_failed_at = time.time()
                payload = self._fallback_runtime_payload(thread_id, detail)
        if not isinstance(payload, dict):
            payload = {}
        models = payload.get("models")
        if not isinstance(models, list):
            models = []
        normalized_models = [
            {
                "id": str(item.get("id") or item.get("value") or item.get("name") or item.get("label") or "").strip(),
                "name": str(item.get("name") or item.get("label") or item.get("id") or item.get("value") or "").strip(),
                "label": str(item.get("label") or item.get("name") or item.get("id") or item.get("value") or "").strip(),
                "description": item.get("description"),
            }
            for item in models
            if isinstance(item, dict)
        ]
        normalized_models = [item for item in normalized_models if item["id"] or item["name"] or item["label"]]
        capabilities = payload.get("capabilities") if isinstance(payload.get("capabilities"), dict) else {}
        capabilities = {
            **capabilities,
            "desktopComputerControl": self._computer_use_plugin_available(),
        }
        runtime = {
            **payload,
            "thread_id": thread_id,
            "capabilities": capabilities,
            "models": normalized_models,
            "reasoning_options": [
                {"label": "低", "value": "low"},
                {"label": "中", "value": "medium"},
                {"label": "高", "value": "high"},
                {"label": "超高", "value": "xhigh"},
            ],
            "permission_options": [
                {"label": "默認權限", "value": "default"},
                {"label": "完全訪問權限", "value": "fullAccess"},
            ],
            "generation_settings": self._generation_settings_for_thread(thread_id) if thread_id else None,
        }
        if thread_id:
            self._runtime_cache = runtime
            self._runtime_cache_at = time.time()
        return runtime

    def _fallback_runtime_payload(self, thread_id: str | None, error: str | None = None) -> dict[str, Any]:
        return {
            "protocol": "codex-app-server-v2",
            "thread_id": thread_id,
            "error": error,
            "capabilities": {
                "threadRead": True,
                "threadInjectItems": False,
                "threadRealtime": False,
                "turnStart": True,
                "turnSteer": True,
                "turnInterrupt": True,
                "threadRollback": True,
                "modelList": False,
                "configRead": False,
                "rateLimits": False,
                "desktopComputerControl": self._computer_use_plugin_available(),
            },
            "models": [],
        }

    def _computer_use_plugin_available(self) -> bool:
        return (
            Path("/Applications/Codex.app/Contents/Resources/plugins/openai-bundled/plugins/computer-use/.mcp.json").exists()
            or Path("/Applications/Codex.app/Contents/Resources/plugins/computer-use/.mcp.json").exists()
        )

    def _fallback_generation_settings(self, thread_id: str) -> dict[str, Any]:
        permission = "default"
        thread_row = self._local_thread_row(thread_id)
        config = self._read_codex_config()
        model = str((thread_row or {}).get("model") or config.get("model") or "gpt-5.5").strip() or "gpt-5.5"
        reasoning = str(
            (thread_row or {}).get("reasoning_effort")
            or config.get("model_reasoning_effort")
            or config.get("reasoning_effort")
            or config.get("reasoning")
            or "medium"
        ).strip() or "medium"
        approval_mode = str((thread_row or {}).get("approval_mode") or "").strip()
        sandbox_raw = (thread_row or {}).get("sandbox_policy")
        sandbox: Any = None
        if isinstance(sandbox_raw, str) and sandbox_raw.strip():
            try:
                sandbox = json.loads(sandbox_raw)
            except json.JSONDecodeError:
                sandbox = sandbox_raw
        sandbox_text = json.dumps(sandbox, ensure_ascii=False).lower() if sandbox is not None else ""
        if approval_mode == "never" or "danger" in sandbox_text or "fullaccess" in sandbox_text or "full-access" in sandbox_text:
            permission = "fullAccess"
        state_path = Path.home() / ".codex" / ".codex-global-state.json"
        if permission != "fullAccess" and not thread_row:
            try:
                state = json.loads(state_path.read_text(encoding="utf-8"))
                persisted = state.get("electron-persisted-atom-state") if isinstance(state, dict) else {}
                agent_mode = str((persisted or {}).get("agent-mode") or "").lower()
                host_modes = (persisted or {}).get("agent-mode-by-host-id")
                host_mode_text = json.dumps(host_modes, ensure_ascii=False).lower() if isinstance(host_modes, dict) else ""
                if "full" in agent_mode or "danger" in agent_mode or "full" in host_mode_text or "danger" in host_mode_text:
                    permission = "fullAccess"
            except Exception:
                pass
        return {
            "thread_id": thread_id,
            "model": model,
            "reasoning": reasoning,
            "permission": permission,
            "approval_policy": approval_mode or ("never" if permission == "fullAccess" else None),
            "sandbox": sandbox if sandbox is not None else ({"type": "dangerFullAccess"} if permission == "fullAccess" else None),
            "source": "codex-sqlite-fallback" if thread_row else "codex-config-fallback",
        }

    def _read_codex_config(self) -> dict[str, Any]:
        path = Path.home() / ".codex" / "config.toml"
        if not path.exists():
            return {}
        try:
            raw = path.read_bytes()
            if tomllib is not None:
                data = tomllib.loads(raw.decode("utf-8"))
                return data if isinstance(data, dict) else {}
            text = raw.decode("utf-8", errors="ignore")
            parsed: dict[str, Any] = {}
            for key in ("model", "model_reasoning_effort", "reasoning_effort", "reasoning"):
                match = re.search(rf"(?m)^\s*{re.escape(key)}\s*=\s*['\"]([^'\"]+)['\"]", text)
                if match:
                    parsed[key] = match.group(1)
            return parsed
        except Exception:
            logger.debug("Failed to read Codex config fallback", exc_info=True)
            return {}

    def _list_raw_threads(self, limit: int, *, force: bool = False) -> list[dict]:
        # `thread/list` is a convenience path for the mobile sidebar, not a
        # critical-path operation. Keep it short so a slow Codex app-server
        # cannot block phone connection or project switching.
        return self._refresh_raw_threads(limit=limit, selection=self._load_selection_data(), timeout=min(self.timeout, 2.0), force=force)

    def _list_archived_raw_threads(self, limit: int) -> list[dict]:
        try:
            payload = self._run_probe(
                "list",
                "--limit",
                str(limit),
                "--archived",
                "true",
                timeout=min(self.timeout, 2.0),
            )
        except CodexLiveBridgeError as exc:
            logger.debug("Codex archived thread list unavailable: %s", exc)
            return []
        threads = payload.get("data", [])
        if not isinstance(threads, list):
            return []
        return [item for item in threads if isinstance(item, dict)]

    def _refresh_raw_threads(
        self,
        limit: int,
        selection: dict[str, Any] | None = None,
        timeout: float | None = None,
        *,
        force: bool = False,
    ) -> list[dict]:
        now = time.time()
        if not force and self._raw_threads_cache and (time.time() - self._raw_threads_cache_at) <= self._raw_threads_refresh_interval:
            return self._raw_threads_cache
        if not force and self._raw_threads_failed_at and (now - self._raw_threads_failed_at) <= self._raw_threads_failed_backoff:
            return self._cached_or_selected_threads(selection)
        if not self._raw_threads_lock.acquire(blocking=False):
            return self._cached_or_selected_threads(selection)
        try:
            payload = self._run_probe("list", "--limit", str(limit), timeout=timeout or min(self.timeout, 8.0))
            threads = payload.get("data", [])
            if isinstance(threads, list):
                self._raw_threads_cache = [thread for thread in threads if isinstance(thread, dict)]
                self._raw_threads_cache_at = time.time()
                self._raw_threads_failed_at = 0.0
                return self._raw_threads_cache
        except CodexLiveBridgeError as exc:
            self._raw_threads_failed_at = time.time()
            logger.debug("Codex thread list unavailable, using cached selection: %s", exc)
        finally:
            self._raw_threads_lock.release()
        return self._cached_or_selected_threads(selection)

    def _cached_or_selected_threads(self, selection: dict[str, Any] | None = None) -> list[dict]:
        fallback_threads = self._local_state_project_threads()
        if self._raw_threads_cache and (time.time() - self._raw_threads_cache_at) <= self._raw_threads_cache_ttl:
            return self._merge_threads(self._raw_threads_cache, fallback_threads)
        selection = selection or self._load_selection_data()
        items = list(fallback_threads)
        if selection and selection.get("thread_id"):
            items.append(
                {
                    "id": selection.get("thread_id"),
                    "name": selection.get("thread_name"),
                    "cwd": selection.get("cwd") or self.thread_cwd,
                    "updatedAt": selection.get("updated_at") or int(time.time()),
                    "status": selection.get("status") or "idle",
                }
            )
        return self._merge_threads(items)

    def _local_state_project_threads(self) -> list[dict]:
        state_db = Path.home() / ".codex" / "state_5.sqlite"
        if not state_db.exists():
            return []
        query = (
            "select id,title,cwd,updated_at,archived,archived_at from threads "
            "where coalesce(cwd, '') != '' "
            "order by updated_at desc "
            "limit 240"
        )
        try:
            with sqlite3.connect(f"file:{state_db}?mode=ro", uri=True, timeout=0.25) as conn:
                rows = conn.execute(query).fetchall()
        except Exception as exc:
            logger.debug("Codex local state unavailable for project fallback: %s", exc)
            return []
        items: list[dict] = []
        for thread_id, title, cwd, updated_at, archived, archived_at in rows:
            cwd_text = str(cwd or "").strip()
            title_text = str(title or "").strip()
            fallback_title = None
            if cwd_text:
                fallback_title = self._project_title_for_thread({"name": title_text, "cwd": cwd_text})
            is_archived = bool(archived)
            items.append(
                {
                    "id": str(thread_id),
                    # Preserve the conversation title. Project grouping is
                    # derived later from cwd/title aliases, matching desktop.
                    "name": title_text or fallback_title,
                    "cwd": cwd_text or None,
                    "updatedAt": int(updated_at or time.time()),
                    "status": "idle",
                    "archived": is_archived,
                    "isArchived": is_archived,
                    "archivedAt": int(archived_at) if archived_at else None,
                    "_source": "local-state",
                }
            )
        return self._merge_threads(items)

    @staticmethod
    def _merge_threads(*groups: list[dict]) -> list[dict]:
        merged: dict[str, dict] = {}
        for group in groups:
            for item in group:
                thread_id = item.get("id")
                if not thread_id:
                    continue
                existing = merged.get(str(thread_id))
                if not existing:
                    merged[str(thread_id)] = item
                    continue
                existing_updated_at = int(existing.get("updatedAt") or 0)
                item_updated_at = int(item.get("updatedAt") or 0)
                if item.get("_source") == "local-state" and existing.get("_source") != "local-state":
                    merged[str(thread_id)] = {
                        **item,
                        **existing,
                        "name": existing.get("name") or item.get("name"),
                        "cwd": existing.get("cwd") or item.get("cwd"),
                        "status": existing.get("status") or item.get("status"),
                        "updatedAt": max(existing_updated_at, item_updated_at),
                        "archived": bool(existing.get("archived") or item.get("archived")),
                        "isArchived": bool(existing.get("isArchived") or item.get("isArchived") or existing.get("archived") or item.get("archived")),
                        "archivedAt": item.get("archivedAt") or existing.get("archivedAt"),
                    }
                    continue
                if item_updated_at >= existing_updated_at:
                    merged[str(thread_id)] = {
                        **existing,
                        **item,
                        "archived": bool(existing.get("archived") or item.get("archived")),
                        "isArchived": bool(existing.get("isArchived") or item.get("isArchived") or existing.get("archived") or item.get("archived")),
                        "archivedAt": item.get("archivedAt") or existing.get("archivedAt"),
                    }
                    continue
                if item.get("archived") or item.get("isArchived"):
                    existing["archived"] = True
                    existing["isArchived"] = True
                    existing["archivedAt"] = existing.get("archivedAt") or item.get("archivedAt")
        items = list(merged.values())
        items.sort(key=lambda item: int(item.get("updatedAt") or 0), reverse=True)
        return items

    def _load_selection_data(self) -> dict[str, Any] | None:
        if not self.selection_path.exists():
            return None
        try:
            payload = json.loads(self.selection_path.read_text(encoding="utf-8"))
        except Exception:
            return None
        if not isinstance(payload, dict):
            return None
        return payload

    def _load_saved_projects(self) -> list[dict[str, Any]]:
        if not self.projects_path.exists():
            return []
        try:
            payload = json.loads(self.projects_path.read_text(encoding="utf-8"))
        except Exception:
            return []
        if not isinstance(payload, list):
            return []
        return [item for item in payload if isinstance(item, dict)]

    def _load_desktop_sidebar_state(self) -> dict[str, Any]:
        payload = self._load_codex_global_state()
        if not isinstance(payload, dict):
            return {}
        project_order = payload.get("project-order")
        saved_roots = payload.get("electron-saved-workspace-roots")
        workspace_roots = [
            str(item).rstrip("/")
            for item in (project_order if isinstance(project_order, list) and project_order else saved_roots if isinstance(saved_roots, list) else [])
            if str(item or "").strip()
        ]
        labels = payload.get("electron-workspace-root-labels")
        pinned_thread_ids = payload.get("pinned-thread-ids")
        pinned_project_ids = payload.get("pinned-project-ids")
        projectless_thread_ids = payload.get("projectless-thread-ids")
        return {
            "workspace_roots": workspace_roots,
            "workspace_labels": labels if isinstance(labels, dict) else {},
            "pinned_thread_ids": [str(item) for item in pinned_thread_ids] if isinstance(pinned_thread_ids, list) else [],
            "pinned_project_ids": [str(item) for item in pinned_project_ids] if isinstance(pinned_project_ids, list) else [],
            "projectless_thread_ids": [str(item) for item in projectless_thread_ids] if isinstance(projectless_thread_ids, list) else [],
        }

    def _load_codex_global_state(self) -> dict[str, Any]:
        try:
            payload = json.loads(CODEX_GLOBAL_STATE_PATH.read_text(encoding="utf-8"))
        except Exception:
            return {}
        return payload if isinstance(payload, dict) else {}

    def _write_codex_global_state(self, payload: dict[str, Any]) -> None:
        CODEX_GLOBAL_STATE_PATH.parent.mkdir(parents=True, exist_ok=True)
        temp_path = CODEX_GLOBAL_STATE_PATH.with_suffix(f"{CODEX_GLOBAL_STATE_PATH.suffix}.tmp")
        temp_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
        temp_path.replace(CODEX_GLOBAL_STATE_PATH)

    def _save_project(self, *, title: str, cwd: str) -> None:
        projects = self._load_saved_projects()
        normalized_title = title.strip()
        normalized_cwd = cwd.strip()
        for item in projects:
            if str(item.get("cwd") or "").strip() == normalized_cwd:
                item["title"] = normalized_title
                break
        else:
            projects.append(
                {
                    "title": normalized_title,
                    "cwd": normalized_cwd,
                    "added_at": int(time.time()),
                }
            )
        self.projects_path.parent.mkdir(parents=True, exist_ok=True)
        self.projects_path.write_text(json.dumps(projects, ensure_ascii=False), encoding="utf-8")

    @staticmethod
    def _project_id(title: str | None, cwd: str | None) -> str:
        normalized_title = (title or "未命名項目").strip()
        return f"title:{normalized_title.casefold()}"

    def _live_selection_from_data(self, data: dict[str, Any]) -> LiveThreadSelection:
        return LiveThreadSelection(
            thread_id=str(data.get("thread_id", "")),
            thread_name=data.get("thread_name"),
            cwd=data.get("cwd") or self.thread_cwd,
            updated_at=int(data.get("updated_at") or time.time()),
            status=self._status_label(data.get("status")) or "idle",
        )

    def _run_probe(
        self,
        command: str,
        *extra_args: str,
        timeout: float | None = None,
        bypass_backoff: bool = False,
    ) -> dict:
        read_only_commands = {"list", "read", "runtime", "settings", "usage"}
        mutating_commands = {"send", "interrupt", "rollback", "inject", "start", "name", "archive", "unarchive", "fork"}
        spawn_commands = mutating_commands | {"runtime", "usage"}
        require_protocol = command in mutating_commands or bypass_backoff
        if command in read_only_commands and not bypass_backoff and self._app_server_in_backoff():
            if not self._select_app_server_url(
                timeout=min(timeout or self.timeout, 4.0),
                bypass_backoff=True,
                require_protocol=True,
            ):
                elapsed = int(time.time() - self._app_server_failed_at)
                raise CodexLiveBridgeError(f"Codex app-server 暫時不可用，使用本機降級資料：{elapsed}s")

        selected_url = self._select_app_server_url(
            timeout=min(timeout or self.timeout, 4.0),
            bypass_backoff=bypass_backoff or command not in read_only_commands,
            require_protocol=require_protocol,
        )
        if require_protocol and not selected_url:
            if command in spawn_commands and self.spawn_server:
                selected_url = self._active_app_server_url or self.url
                cmd = self._build_probe_command(command, extra_args, force_spawn=True, url=selected_url)
            else:
                raise CodexLiveBridgeError("沒有可用的 Codex app-server 協議連線")
        else:
            cmd = self._build_probe_command(command, extra_args, url=selected_url)
        try:
            return self._execute_probe_command(command, cmd, timeout=timeout, read_only_commands=read_only_commands)
        except CodexLiveBridgeError as exc:
            if not self._should_restart_app_server_for_probe(command, str(exc)):
                raise
            logger.warning("Restarting stale Codex app-server after probe failure: %s", exc)
            force_recover = command in {"send", "interrupt", "rollback", "inject", "start", "name", "archive", "unarchive", "fork"}
            if not self._recover_app_server_once(str(exc), force=force_recover):
                raise
            # A stale Codex app-server can keep the websocket port open while
            # JSON-RPC initialize hangs. After a protocol-level failure and a
            # listener restart, never trust a TCP-only availability check here.
            selected_url = self._select_app_server_url(
                timeout=min(timeout or self.timeout, 4.0),
                bypass_backoff=True,
                require_protocol=True,
            )
            if not selected_url:
                raise
            retry_cmd = self._build_probe_command(command, extra_args, force_spawn=True, url=selected_url)
            return self._execute_probe_command(command, retry_cmd, timeout=timeout, read_only_commands=read_only_commands)

    def _build_probe_command(
        self,
        command: str,
        extra_args: tuple[str, ...],
        *,
        force_spawn: bool = False,
        url: str | None = None,
    ) -> list[str]:
        app_server_url = url or self._active_app_server_url or self.url
        cmd = [self.node_bin, str(self.probe_script), command, "--url", app_server_url]
        spawnable_commands = {"send", "interrupt", "rollback", "inject", "start", "name", "archive", "unarchive", "fork", "runtime", "usage"}
        if (
            self.spawn_server
            and command in spawnable_commands
            and (force_spawn or not self._is_app_server_available(app_server_url))
        ):
            cmd.append("--spawn")
            cmd.extend(["--codex-bin", str(self.codex_bin)])
        extra_args_list = list(extra_args)
        if command != "list" and "--cwd" not in extra_args_list:
            cmd.extend(["--cwd", self.thread_cwd])
            if self.thread_name:
                cmd.extend(["--name", self.thread_name])
        cmd.extend(extra_args_list)
        return cmd

    def _execute_probe_command(
        self,
        command: str,
        cmd: list[str],
        *,
        timeout: float | None,
        read_only_commands: set[str],
    ) -> dict:
        try:
            result = subprocess.run(
                cmd,
                cwd=self.repo_root,
                text=True,
                encoding="utf-8",
                errors="replace",
                capture_output=True,
                timeout=timeout or self.timeout,
                check=True,
            )
        except subprocess.CalledProcessError as exc:
            detail = exc.stderr.strip() or exc.stdout.strip() or str(exc)
            self._mark_app_server_failure(detail)
            raise CodexLiveBridgeError(f"Codex app-server 請求失敗：{detail}") from exc
        except subprocess.TimeoutExpired as exc:
            self._mark_app_server_failure("request timeout")
            raise CodexLiveBridgeError("Codex app-server 請求逾時") from exc

        stdout = result.stdout.strip()
        if command in read_only_commands or command == "send":
            self._app_server_failed_at = 0.0
        if not stdout:
            return {}
        try:
            return json.loads(stdout)
        except json.JSONDecodeError as exc:
            raise CodexLiveBridgeError(f"Codex app-server 回傳不是有效 JSON：{stdout[:200]}") from exc

    def _should_restart_app_server_for_probe(self, command: str, detail: str) -> bool:
        if command not in {"send", "interrupt", "rollback", "inject", "start", "name", "archive", "unarchive", "fork", "list", "read", "runtime", "usage", "settings"}:
            return False
        if not self.spawn_server:
            return False
        text = detail.lower()
        markers = (
            "initialize: request timeout",
            "connect timeout",
            "websocket closed",
            "websocket error",
            "onsocketclose",
            "socket close",
            "econnreset",
            "socket hang up",
            "not open",
            "request timeout",
            "請求逾時",
            "请求逾时",
        )
        return any(marker in text for marker in markers)

    def _restart_listen_app_server(self) -> bool:
        pids = self._app_server_listen_pids()
        if not pids:
            return True
        for pid in pids:
            self._terminate_pid(pid, force=False)
        deadline = time.time() + 3.0
        while time.time() < deadline and self._app_server_listen_pids():
            time.sleep(0.1)
        for pid in self._app_server_listen_pids():
            self._terminate_pid(pid, force=True)
        time.sleep(0.8)
        self._active_app_server_url_checked_at = 0.0
        self._runtime_cache = None
        self._runtime_cache_at = 0.0
        return True

    def _recover_app_server_once(self, reason: str, *, force: bool = False) -> bool:
        if not self.spawn_server:
            return False
        healthy_url = self._select_app_server_url(
            timeout=3.0,
            bypass_backoff=True,
            require_protocol=True,
            ignore_cached=True,
        )
        if healthy_url:
            logger.info("Using alternate healthy Codex app-server before restart: %s", healthy_url)
            return True
        now = time.time()
        if not force and now - self._app_server_recovery_attempted_at < self._app_server_recovery_cooldown:
            return False
        self._app_server_recovery_attempted_at = now
        logger.warning("Recovering stale Codex app-server before send: %s", reason)
        if not self._restart_listen_app_server():
            return False
        self._app_server_failed_at = 0.0
        self._runtime_failed_at = 0.0
        return True

    def _candidate_app_server_urls(self) -> list[str]:
        urls: list[str] = []

        def add(url: str | None) -> None:
            value = (url or "").strip()
            if (value.startswith("ws://") or value.startswith("stdio://")) and value not in urls:
                urls.append(value)

        add(self._active_app_server_url)
        add(self.url)
        try:
            output = subprocess.check_output(
                ["ps", "-axo", "command"],
                text=True,
                stderr=subprocess.DEVNULL,
                timeout=1.5,
            )
            for line in output.splitlines():
                if " app-server --listen " not in line:
                    continue
                match = re.search(r"--listen\s+(ws://[^\s]+)", line)
                if match:
                    add(match.group(1))
        except Exception:
            pass
        # Codex has used both ports across recent desktop app versions. Keep
        # them as fallback candidates so a stale configured port cannot strand
        # mobile sends when another local app-server is already healthy.
        add("ws://127.0.0.1:8790")
        add("ws://127.0.0.1:8791")
        # Codex 0.131+ removed ws:// from `codex app-server --listen` and
        # supports stdio:// / unix:// instead. Keep stdio as the final fallback
        # so existing customer configs keep working after Codex upgrades.
        add("stdio://")
        return urls

    def _select_app_server_url(
        self,
        *,
        timeout: float = 4.0,
        bypass_backoff: bool = False,
        require_protocol: bool = False,
        ignore_cached: bool = False,
    ) -> str:
        now = time.time()
        if (
            not ignore_cached
            and self._active_app_server_url
            and now - self._active_app_server_url_checked_at < self._active_app_server_url_ttl
            and self._is_app_server_available(self._active_app_server_url)
        ):
            if require_protocol:
                probe_timeout = max(1.0, min(timeout, 4.0))
                if not self._probe_app_server_url(self._active_app_server_url, timeout=probe_timeout):
                    logger.warning(
                        "Ignoring stale Codex app-server URL after protocol probe failed: %s",
                        self._active_app_server_url,
                    )
                    self._mark_app_server_failure("request timeout")
                else:
                    return self._active_app_server_url
            else:
                return self._active_app_server_url

        probe_timeout = max(1.0, min(timeout, 4.0))
        for url in self._candidate_app_server_urls():
            if not self._is_app_server_available(url):
                continue
            if not require_protocol and url == self._active_app_server_url:
                return url
            if self._probe_app_server_url(url, timeout=probe_timeout):
                if url != self._active_app_server_url:
                    logger.info("Using healthy Codex app-server URL: %s", url)
                self._active_app_server_url = url
                self._active_app_server_url_checked_at = time.time()
                self._app_server_failed_at = 0.0
                return url

        if require_protocol:
            return ""
        if bypass_backoff:
            return self._active_app_server_url or self.url
        return self._active_app_server_url or self.url

    def _probe_app_server_url(self, url: str, *, timeout: float) -> bool:
        if url.startswith("stdio://"):
            timeout = max(timeout, 8.0)
        cmd = [
            self.node_bin,
            str(self.probe_script),
            "models",
            "--url",
            url,
            "--json",
        ]
        try:
            result = subprocess.run(
                cmd,
                cwd=self.repo_root,
                text=True,
                encoding="utf-8",
                errors="replace",
                capture_output=True,
                timeout=timeout,
                check=True,
            )
        except Exception as exc:
            logger.debug("Codex app-server URL probe failed for %s: %s", url, exc)
            return False
        try:
            payload = json.loads(result.stdout.strip() or "{}")
        except json.JSONDecodeError:
            return False
        return isinstance(payload, dict) and isinstance(payload.get("data"), list)

    def _app_server_listen_pids(self, url: str | None = None) -> list[int]:
        url_text = str(url or self._active_app_server_url or self.url)
        if url_text.startswith("stdio://"):
            return []
        codex_bin_text = str(self.codex_bin)
        try:
            output = subprocess.check_output(["ps", "-axo", "pid=,command="], text=True, stderr=subprocess.DEVNULL, timeout=1.5)
        except Exception:
            return []
        pids: list[int] = []
        for line in output.splitlines():
            stripped = line.strip()
            if not stripped:
                continue
            pid_text, _, command = stripped.partition(" ")
            if (
                codex_bin_text not in command
                or " app-server --listen " not in command
                or url_text not in command
            ):
                continue
            try:
                pid = int(pid_text)
            except ValueError:
                continue
            if pid > 0:
                pids.append(pid)
        return pids

    @staticmethod
    def _terminate_pid(pid: int, *, force: bool) -> None:
        signal = "-KILL" if force else "-TERM"
        try:
            subprocess.run(["kill", signal, str(pid)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=1.0)
        except Exception:
            return

    def _app_server_in_backoff(self) -> bool:
        return bool(
            self._app_server_failed_at
            and (time.time() - self._app_server_failed_at) <= self._app_server_failed_backoff
        )

    def _mark_app_server_failure(self, detail: str | None = None) -> None:
        text = (detail or "").lower()
        protocol_failure_markers = (
            "initialize: request timeout",
            "request timeout",
            "connect timeout",
            "websocket",
            "onsocketclose",
            "econnreset",
            "socket hang up",
            "not open",
        )
        if any(marker in text for marker in protocol_failure_markers):
            self._app_server_failed_at = time.time()

    def _is_app_server_available(self, url: str | None = None) -> bool:
        app_server_url = url or self._active_app_server_url or self.url
        if app_server_url.startswith("stdio://"):
            return Path(self.codex_bin).exists()
        if not app_server_url.startswith("ws://"):
            return False
        try:
            host_port = app_server_url.removeprefix("ws://").split("/", 1)[0]
            host, port_text = host_port.rsplit(":", 1)
            port = int(port_text)
        except Exception:
            return False
        sock = socket.socket()
        sock.settimeout(0.35)
        try:
            sock.connect((host, port))
            return True
        except Exception:
            return False
        finally:
            sock.close()

    def _select_thread(self, threads: list[dict]) -> dict | None:
        visible_threads = [thread for thread in threads if not (thread.get("archived") or thread.get("isArchived"))]
        active = self._preferred_active_thread(visible_threads)
        if active:
            return active
        selected_id = self._load_selected_thread_id()
        selection = self._load_selection_data()
        if selected_id and self._selection_is_trusted(selection, visible_threads):
            exact = next((thread for thread in visible_threads if thread.get("id") == selected_id), None)
            exact = self._prefer_visible_thread(visible_threads, exact, selection)
            if exact:
                return exact
            if selection and selection.get("pending_materialization") and not self._is_stale_pending_selection(selection):
                return None
        filtered = [thread for thread in visible_threads if thread.get("cwd") == self.thread_cwd]
        if self.thread_name:
            query = self.thread_name.casefold()
            filtered = [thread for thread in filtered if query in str(thread.get("name", "")).casefold()]
        if not filtered:
            return None
        filtered.sort(key=lambda thread: int(thread.get("updatedAt") or 0), reverse=True)
        return filtered[0]

    def _preferred_active_thread(self, threads: list[dict]) -> dict | None:
        active = [thread for thread in threads if self._status_label(thread.get("status")) == "active"]
        if not active:
            return None

        def score(thread: dict) -> tuple[int, int, int]:
            cwd = str(thread.get("cwd") or "")
            title = self._project_title_for_thread(thread) or ""
            configured_cwd = 1 if cwd == self.thread_cwd else 0
            named_project = 1 if self._looks_like_project_label(title) else 0
            updated_at = int(thread.get("updatedAt") or 0)
            return (configured_cwd, named_project, updated_at)

        active.sort(key=score, reverse=True)
        return active[0]

    def _project_title_for_thread(self, thread: dict | None) -> str | None:
        if not thread:
            return None
        cwd_text = str(thread.get("cwd") or "").strip()
        name_text = str(thread.get("name") or "").strip()
        cwd_name = Path(cwd_text).name if cwd_text else None
        # For unknown cwd-backed conversations, the desktop sidebar groups by
        # folder/project first and keeps the thread title as a child row.
        if self._looks_like_project_label(cwd_name):
            return str(cwd_name).strip()
        if self._looks_like_project_label(name_text):
            return name_text
        if cwd_name:
            return cwd_name
        if thread.get("name"):
            return str(thread.get("name")).strip()
        return None

    def _project_title_for_selection(self, selection: LiveThreadSelection) -> str:
        synthetic = {
            "id": selection.thread_id,
            "name": selection.thread_name,
            "cwd": selection.cwd,
        }
        return self._project_title_for_thread(synthetic) or selection.project_title

    def _score_project_representative(self, thread: dict, title: str) -> int:
        normalized_title = title.strip().casefold()
        normalized_name = str(thread.get("name") or "").strip().casefold()
        normalized_cwd = Path(str(thread.get("cwd"))).name.strip().casefold() if thread.get("cwd") else ""
        if normalized_name and normalized_name == normalized_title:
            return 100
        if normalized_cwd and normalized_cwd == normalized_title:
            return 90
        if thread.get("name"):
            return 70
        return 10

    @staticmethod
    def _looks_like_project_label(label: Any) -> bool:
        if not label:
            return False
        normalized = str(label).strip()
        if not normalized:
            return False
        if len(normalized) > 40:
            return False
        if normalized.lower().endswith("-old"):
            return False
        if "旧版" in normalized or "舊版" in normalized:
            return False
        lowered = normalized.casefold()
        if "received image" in lowered or "attachment to project" in lowered or "copy new received" in lowered:
            return False
        if lowered.startswith(("recover ", "恢復", "恢复", "繼續", "继续", "請", "所以", "目前")):
            return False
        return True

    def _prefer_visible_thread(
        self,
        threads: list[dict],
        selected: dict | None,
        selection: dict[str, Any] | None,
    ) -> dict | None:
        if not selected:
            return None
        if selection and selection.get("pending_materialization"):
            return selected
        if selected.get("name"):
            return selected
        selected_cwd = selected.get("cwd")
        if not selected_cwd:
            return selected
        preferred = [
            thread
            for thread in threads
            if thread.get("cwd") == selected_cwd and thread.get("name")
        ]
        if not preferred:
            return selected
        preferred.sort(key=lambda thread: int(thread.get("updatedAt") or 0), reverse=True)
        return preferred[0]

    def _is_stale_pending_selection(self, selection: dict[str, Any]) -> bool:
        if not selection.get("pending_materialization"):
            return False
        updated_at = float(selection.get("updated_at") or 0)
        if not updated_at:
            return True
        return (time.time() - updated_at) > self._pending_selection_ttl

    def _selection_is_trusted(self, selection: dict[str, Any] | None, threads: list[dict]) -> bool:
        if not selection:
            return False
        selected_id = str(selection.get("thread_id") or "").strip()
        if not selected_id:
            return False
        if self._is_stale_pending_selection(selection):
            return False
        if any(str(thread.get("id") or "") == selected_id for thread in threads):
            return True
        # A newly-created thread may need a short moment before it appears in
        # thread/list. Only that explicit, fresh pending state may synthesize a
        # selected row; old selections must not resurrect stale projects.
        return bool(selection.get("pending_materialization"))

    def _flatten_thread_messages(self, thread: dict) -> list[dict]:
        turns = thread.get("turns", [])
        resource_item_types = {"resource", "resources", "attachment", "fileAttachment", "fileCitation", "urlCitation", "webCitation", "reference"}
        visible_item_types = {
            "userMessage",
            "agentMessage",
            "fileChange",
            "commandExecution",
            "contextCompaction",
            "plan",
            *resource_item_types,
        }
        total_items = sum(
            1
            for turn in turns
            for item in turn.get("items", [])
            if item.get("type") in visible_item_types
        )
        if total_items == 0:
            return []

        base_epoch = int(thread.get("updatedAt") or time.time()) - total_items
        index = 0
        messages: list[dict] = []
        for turn_index, turn in enumerate(turns):
            turn_id = str(turn.get("id") or f"turn-{turn_index}")
            turn_duration_ms = self._turn_duration_ms(turn)
            turn_duration_label = self._format_turn_duration(turn_duration_ms)
            turn_completed_at = self._iso_from_epoch(turn.get("completedAt")) if turn.get("completedAt") is not None else None
            turn_items = turn.get("items", [])
            last_process_activity_index = self._last_process_activity_index(turn_items)
            first_user_message_index = self._first_user_message_index(turn_items)
            for item_index, item in enumerate(turn_items):
                item_type = item.get("type")
                if item_type == "userMessage":
                    text = self._user_message_text(item)
                    author = "mobile"
                    user_attachments = self._user_message_attachments(item)
                elif item_type == "agentMessage":
                    text = str(item.get("text", "")).strip()
                    author = "assistant"
                    user_attachments = []
                elif item_type == "fileChange":
                    text = self._file_change_text(item)
                    author = "assistant"
                    user_attachments = []
                elif item_type == "commandExecution":
                    text = self._command_execution_text(item)
                    author = "assistant"
                    user_attachments = []
                elif item_type == "contextCompaction":
                    text = "背景資訊已自動壓縮"
                    author = "assistant"
                    user_attachments = []
                elif item_type == "plan":
                    text = str(item.get("text") or "").strip()
                    author = "assistant"
                    user_attachments = []
                elif item_type in resource_item_types:
                    text = str(item.get("text") or item.get("title") or "").strip()
                    author = "assistant"
                    user_attachments = self._item_resource_attachments(item)
                else:
                    continue
                generated_image_attachments = self._generated_image_notice_attachments(text)
                if generated_image_attachments:
                    text = "已生成圖片，可在下方預覽。"
                    author = "assistant"
                if not text and not user_attachments and not generated_image_attachments:
                    continue
                created_at = self._iso_from_epoch(base_epoch + index)
                index += 1
                display_scope = self._timeline_display_scope(item_type, text)
                if (
                    item_type == "userMessage"
                    and first_user_message_index is not None
                    and item_index != first_user_message_index
                ):
                    display_scope = "process"
                if (
                    item_type == "agentMessage"
                    and last_process_activity_index is not None
                    and item_index < last_process_activity_index
                ):
                    display_scope = "process"
                explicit_display_scope = str(item.get("displayScope") or item.get("display_scope") or "").strip()
                if explicit_display_scope in {"message", "process", "fileChange"}:
                    display_scope = explicit_display_scope
                message = {
                    "message_id": str(item.get("id") or f"turn-{turn.get('id')}-{index}"),
                    "turn_id": turn_id,
                    "item_id": str(item.get("id") or ""),
                    "item_index": item_index,
                    "sequence": index,
                    "source": thread.get("_source") or "codex-app-server",
                    "created_at": created_at,
                    "author": author,
                    "role": "assistant" if generated_image_attachments else "user" if item_type == "userMessage" else "assistant",
                    "text": text,
                    "kind": self._timeline_item_kind(item_type),
                    "item_type": item_type,
                    "display_scope": display_scope,
                    "status": "done",
                    "turn_duration_ms": turn_duration_ms,
                    "turn_duration_label": turn_duration_label,
                    "turn_completed_at": turn_completed_at,
                }
                if generated_image_attachments:
                    message["attachments"] = generated_image_attachments
                    message["attachment"] = generated_image_attachments[0]
                elif item_type == "userMessage":
                    if user_attachments:
                        message["attachments"] = user_attachments
                        # Keep the first image available for the current mobile renderer;
                        # the full list remains in `attachments` for future multi-image UI.
                        message["attachment"] = user_attachments[0]
                elif item_type == "agentMessage":
                    resource_attachments = self._item_resource_attachments(item)
                    if resource_attachments:
                        message["attachments"] = resource_attachments
                        message["attachment"] = resource_attachments[0]
                elif item_type in resource_item_types:
                    if user_attachments:
                        message["attachments"] = user_attachments
                        message["attachment"] = user_attachments[0]
                if item_type == "fileChange":
                    message["file_changes"] = self._file_change_summaries(item)
                if item_type == "commandExecution":
                    message["duration_ms"] = self._duration_ms(item.get("durationMs"))
                    message["command_execution"] = {
                        "command": item.get("command") or item.get("cmd") or item.get("shellCommand"),
                        "status": self._status_label(item.get("status")) or item.get("status"),
                        "cwd": item.get("cwd"),
                        "exit_code": item.get("exitCode") or item.get("exit_code"),
                        "duration_ms": self._duration_ms(item.get("durationMs")),
                    }
                if item_type == "plan":
                    message["plan"] = self._plan_summary_from_text(text)
                messages.append(message)
        return self._drop_text_only_user_image_duplicates(messages)

    @staticmethod
    def _drop_text_only_user_image_duplicates(messages: list[dict]) -> list[dict]:
        filtered: list[dict] = []
        total = len(messages)
        for index, message in enumerate(messages):
            if message.get("role") != "user":
                filtered.append(message)
                continue
            if message.get("attachment") or message.get("attachments"):
                filtered.append(message)
                continue
            text = CodexLiveBridge._canonical_rollout_user_text(str(message.get("text") or ""))
            if not text:
                filtered.append(message)
                continue
            next_message = messages[index + 1] if index + 1 < total else None
            previous_message = filtered[-1] if filtered else None
            if CodexLiveBridge._user_image_duplicate_matches(next_message, text):
                continue
            if CodexLiveBridge._user_image_duplicate_matches(previous_message, text):
                continue
            filtered.append(message)
        return filtered

    @staticmethod
    def _user_image_duplicate_matches(message: dict | None, text: str) -> bool:
        if not isinstance(message, dict):
            return False
        if message.get("role") != "user":
            return False
        if not (message.get("attachment") or message.get("attachments")):
            return False
        message_text = CodexLiveBridge._canonical_rollout_user_text(str(message.get("text") or ""))
        return message_text == text

    @staticmethod
    def _last_process_activity_index(turn_items: list[dict]) -> int | None:
        indexes: list[int] = []
        for index, item in enumerate(turn_items):
            item_type = item.get("type")
            if item_type in {"commandExecution", "contextCompaction", "plan"}:
                indexes.append(index)
        return indexes[-1] if indexes else None

    @staticmethod
    def _first_user_message_index(turn_items: list[dict]) -> int | None:
        for index, item in enumerate(turn_items):
            if item.get("type") != "userMessage":
                continue
            text = CodexLiveBridge._user_message_text(item)
            attachments = CodexLiveBridge._user_message_attachments(item)
            if text or attachments:
                return index
        return None

    @staticmethod
    def _user_message_text(item: dict) -> str:
        parts: list[str] = []
        for chunk in CodexLiveBridge._message_content_chunks(item):
            chunk_type = str(chunk.get("type") or "").strip()
            if chunk_type not in {"text", "input_text", "output_text"} or not chunk.get("text"):
                continue
            text = str(chunk["text"])
            stripped = text.strip()
            if stripped.startswith("<image name="):
                continue
            if stripped == "</image>":
                continue
            if stripped.startswith("Codex could not read the local image at "):
                continue
            parts.append(text)
        return "\n".join(part.rstrip() for part in parts).strip()

    @staticmethod
    def _user_message_attachments(item: dict) -> list[dict[str, Any]]:
        attachments: list[dict[str, Any]] = []
        seen: set[str] = set()
        image_names = CodexLiveBridge._image_marker_names(item)
        image_index = 0
        for chunk in CodexLiveBridge._message_content_chunks(item):
            raw_type = str(chunk.get("type") or "").strip().lower()
            if raw_type not in {"image", "localimage", "local_image", "input_image", "image_url", "inputimage"}:
                continue
            source = CodexLiveBridge._image_source_from_chunk(chunk)
            if not source or source in seen:
                continue
            seen.add(source)
            is_data_url = source.lower().startswith("data:image")
            is_http_url = source.lower().startswith(("http://", "https://"))
            marker_name = image_names[image_index] if image_index < len(image_names) else ""
            image_index += 1
            name = marker_name or f"圖片 {image_index}"
            saved_path = None
            source_url = None
            local_uri = None
            mime_type = "image/*"
            if not is_data_url:
                try:
                    name = Path(source).name or name
                except Exception:
                    pass
                source_url = source if is_http_url else None
                saved_path = None if is_http_url else source
            else:
                mime_type = source.split(";", 1)[0].replace("data:", "") or "image/*"
                saved_path = CodexLiveBridge._cache_data_image_url(source, name)
                if saved_path:
                    try:
                        name = Path(saved_path).name or name
                    except Exception:
                        pass
                else:
                    local_uri = source
                    source_url = source
            attachments.append(
                {
                    "kind": "image",
                    "name": name,
                    "mimeType": mime_type,
                    "localUri": local_uri,
                    "sourceUrl": source_url,
                    "savedPath": saved_path,
                    "status": "uploaded",
                    "progress": 100,
                }
            )
        return attachments

    @staticmethod
    def _message_content_chunks(item: dict[str, Any]) -> list[dict[str, Any]]:
        chunks: list[dict[str, Any]] = []

        def append_content(container: Any) -> None:
            if not isinstance(container, dict):
                return
            content = container.get("content")
            if isinstance(content, list):
                chunks.extend(chunk for chunk in content if isinstance(chunk, dict))

        append_content(item)
        for key in ("payload", "message", "data"):
            append_content(item.get(key))
        return chunks

    @staticmethod
    def _image_source_from_chunk(chunk: dict[str, Any]) -> str:
        for key in ("url", "imageUrl", "image_url", "sourceUrl", "path", "file_path", "localPath"):
            value = chunk.get(key)
            if isinstance(value, dict):
                value = value.get("url") or value.get("image_url") or value.get("path")
            source = str(value or "").strip()
            if source:
                return source
        return ""

    @staticmethod
    def _cache_data_image_url(source: str, display_name: str) -> str | None:
        match = re.match(r"^data:(image/[a-zA-Z0-9.+-]+);base64,(.*)$", source, flags=re.DOTALL)
        if not match:
            return None
        mime_type, encoded = match.group(1), match.group(2)
        try:
            digest = hashlib.sha256(source.encode("utf-8")).hexdigest()
            extension = mimetypes.guess_extension(mime_type) or ".img"
            safe_stem = re.sub(r"[^A-Za-z0-9._-]+", "_", Path(display_name).stem).strip("._-") or "image"
            cache_path = CODEXMOBILE_IMAGE_CACHE_DIR / f"{digest[:16]}-{safe_stem}{extension}"
            if not cache_path.exists():
                CODEXMOBILE_IMAGE_CACHE_DIR.mkdir(parents=True, exist_ok=True)
                cache_path.write_bytes(base64.b64decode(encoded, validate=True))
            return str(cache_path)
        except Exception as exc:
            logger.debug("failed to cache rollout data image: %s", exc)
            return None

    @staticmethod
    def _codex_supported_local_image_path(path: str) -> str:
        source_path = Path(path).expanduser()
        if not source_path.exists() or not source_path.is_file():
            return str(source_path)
        mime_type = mimetypes.guess_type(str(source_path))[0] or ""
        if mime_type == "image/png" or source_path.suffix.lower() == ".png":
            return str(source_path)
        if not mime_type.startswith("image/"):
            return str(source_path)

        try:
            digest = hashlib.sha256(source_path.read_bytes()).hexdigest()
            safe_stem = re.sub(r"[^A-Za-z0-9._-]+", "_", source_path.stem).strip("._-") or "image"
            target_path = CODEXMOBILE_ATTACH_CACHE_DIR / f"{digest[:16]}-{safe_stem}-max2048.png"
            if target_path.exists() and target_path.stat().st_size > 0:
                return str(target_path)
            CODEXMOBILE_ATTACH_CACHE_DIR.mkdir(parents=True, exist_ok=True)
            subprocess.run(
                ["/usr/bin/sips", "-Z", "2048", "-s", "format", "png", str(source_path), "--out", str(target_path)],
                capture_output=True,
                text=True,
                check=True,
                timeout=30,
            )
            if target_path.exists() and target_path.stat().st_size > 0:
                return str(target_path)
        except Exception as exc:
            logger.warning("Unable to convert image for Codex app-server: %s", exc)
        return str(source_path)

    @staticmethod
    def _image_marker_names(item: dict[str, Any]) -> list[str]:
        names: list[str] = []
        for chunk in CodexLiveBridge._message_content_chunks(item):
            chunk_type = str(chunk.get("type") or "").strip()
            if chunk_type not in {"text", "input_text", "output_text"}:
                continue
            text = str(chunk.get("text") or "")
            for match in re.finditer(r"<image\s+name=\[?([^>\]]+)\]?>", text):
                name = match.group(1).strip()
                if name:
                    names.append(name)
        return names

    @staticmethod
    def _item_resource_attachments(item: dict) -> list[dict[str, Any]]:
        attachments: list[dict[str, Any]] = []
        seen: set[str] = set()

        def candidate_dicts(value: Any) -> list[dict[str, Any]]:
            if isinstance(value, dict):
                return [value]
            if isinstance(value, list):
                return [entry for entry in value if isinstance(entry, dict)]
            return []

        candidates: list[dict[str, Any]] = [item]
        for key in ("attachments", "resources", "files", "citations", "references", "content"):
            candidates.extend(candidate_dicts(item.get(key)))
        message = item.get("message")
        if isinstance(message, dict):
            for key in ("attachments", "resources", "files", "citations", "references", "content"):
                candidates.extend(candidate_dicts(message.get(key)))

        for index, resource in enumerate(candidates):
            source = next(
                (
                    str(resource.get(key) or "").strip()
                    for key in (
                        "url",
                        "href",
                        "uri",
                        "sourceUrl",
                        "source_url",
                        "path",
                        "filePath",
                        "file_path",
                        "localPath",
                        "local_path",
                        "savedPath",
                        "saved_path",
                    )
                    if str(resource.get(key) or "").strip()
                ),
                "",
            )
            install_url = str(resource.get("installUrl") or resource.get("install_url") or "").strip()
            if not source and install_url:
                source = install_url
            if not source:
                continue
            key = source.casefold()
            if key in seen:
                continue
            seen.add(key)

            raw_type = str(resource.get("type") or resource.get("kind") or "").strip().lower()
            mime_type = str(resource.get("mimeType") or resource.get("mime_type") or "").strip()
            name = str(
                resource.get("name")
                or resource.get("filename")
                or resource.get("fileName")
                or resource.get("title")
                or ""
            ).strip()
            if not name:
                try:
                    name = Path(source).name
                except Exception:
                    name = ""
            if not mime_type:
                mime_type = mimetypes.guess_type(name or source)[0] or "application/octet-stream"

            lower_source = source.lower()
            is_data_image = lower_source.startswith("data:image")
            is_http = lower_source.startswith(("http://", "https://"))
            is_image = (
                is_data_image
                or mime_type.lower().startswith("image/")
                or "image" in raw_type
                or Path(name or source).suffix.lower() in {".png", ".jpg", ".jpeg", ".gif", ".webp", ".heic", ".heif"}
            )
            attachment: dict[str, Any] = {
                "kind": "image" if is_image else "file",
                "name": name or (f"圖片 {index + 1}" if is_image else f"附件 {index + 1}"),
                "mimeType": mime_type,
                "status": "uploaded",
                "progress": 100,
            }
            if is_data_image:
                attachment["localUri"] = source
                attachment["sourceUrl"] = source
            elif is_http:
                attachment["sourceUrl"] = source
            else:
                attachment["savedPath"] = source
            if install_url:
                attachment["installUrl"] = install_url
            size_label = str(resource.get("sizeLabel") or resource.get("size_label") or "").strip()
            if size_label:
                attachment["sizeLabel"] = size_label
            attachments.append(attachment)
        return attachments

    @staticmethod
    def _generated_image_notice_attachments(text: str | None) -> list[dict[str, Any]]:
        if not text:
            return []
        normalized = text.strip()
        if "Generated images are saved to " not in normalized and "/.codex/generated_images/" not in normalized:
            return []

        generated_root = (Path.home() / ".codex" / "generated_images").resolve(strict=False)
        image_extensions = {".png", ".jpg", ".jpeg", ".webp", ".gif"}
        candidates: list[Path] = []

        def clean_path(raw_path: str) -> str:
            return raw_path.strip().strip("`'\"<>.,)")

        for match in re.finditer(r"(/[^\s`'\"<>]+)", normalized):
            raw_candidate = clean_path(match.group(1))
            if "/.codex/generated_images/" not in raw_candidate:
                continue
            path = Path(raw_candidate).expanduser()
            try:
                resolved = path.resolve(strict=False)
            except Exception:
                continue
            if not CodexLiveBridge._is_relative_to(resolved, generated_root):
                continue
            if resolved.suffix.lower() in image_extensions:
                candidates.append(resolved)
            else:
                candidates.append(resolved)

        image_paths: list[Path] = []
        seen: set[str] = set()
        for candidate in candidates:
            if candidate.is_file() and candidate.suffix.lower() in image_extensions:
                key = str(candidate)
                if key not in seen:
                    seen.add(key)
                    image_paths.append(candidate)
                continue
            if candidate.is_dir():
                try:
                    newest_images = sorted(
                        (
                            child
                            for child in candidate.iterdir()
                            if child.is_file() and child.suffix.lower() in image_extensions
                        ),
                        key=lambda child: child.stat().st_mtime,
                        reverse=True,
                    )
                except OSError:
                    newest_images = []
                for child in newest_images[:1]:
                    key = str(child)
                    if key not in seen:
                        seen.add(key)
                        image_paths.append(child)

        attachments: list[dict[str, Any]] = []
        for index, image_path in enumerate(image_paths[:4]):
            mime_type = mimetypes.guess_type(str(image_path))[0] or "image/png"
            attachments.append(
                {
                    "kind": "image",
                    "name": image_path.name or f"生成圖片 {index + 1}",
                    "mimeType": mime_type,
                    "savedPath": str(image_path),
                    "status": "uploaded",
                    "progress": 100,
                }
            )
        return attachments

    @staticmethod
    def _file_change_text(item: dict) -> str:
        file_changes = CodexLiveBridge._file_change_summaries(item)
        if not file_changes:
            return "檔案變更：沒有可顯示的檔案內容。"
        summaries: list[str] = [f"檔案變更：{len(file_changes)} 個文件已更新"]
        for change in file_changes:
            summaries.append(f"- {change['path']} +{change['additions']} -{change['deletions']}")
            if change.get("diff"):
                summaries.append(f"```diff\n{change['diff']}\n```")
        return "\n".join(summaries).strip()

    @staticmethod
    def _file_change_summaries(item: dict) -> list[dict[str, Any]]:
        changes = item.get("changes", [])
        if not isinstance(changes, list):
            return []
        summaries: list[dict[str, Any]] = []
        for change in changes:
            if not isinstance(change, dict):
                continue
            path = str(change.get("path") or "未知文件")
            diff = str(change.get("diff") or "").rstrip()
            explicit_additions = change.get("additions")
            explicit_deletions = change.get("deletions")
            if explicit_additions is not None or explicit_deletions is not None:
                try:
                    additions = int(explicit_additions or 0)
                except (TypeError, ValueError):
                    additions = 0
                try:
                    deletions = int(explicit_deletions or 0)
                except (TypeError, ValueError):
                    deletions = 0
            elif diff:
                additions = sum(1 for line in diff.splitlines() if line.startswith("+") and not line.startswith("+++"))
                deletions = sum(1 for line in diff.splitlines() if line.startswith("-") and not line.startswith("---"))
            else:
                additions = 0
                deletions = 0
            summaries.append(
                {
                    "path": path,
                    "name": Path(path).name,
                    "additions": additions,
                    "deletions": deletions,
                    "diff": diff,
                }
            )
        return summaries

    @staticmethod
    def _command_execution_text(item: dict) -> str:
        command = str(item.get("command") or "").strip()
        status = str(item.get("status") or "").strip()
        cwd = str(item.get("cwd") or "").strip()
        exit_code = item.get("exitCode")
        duration_ms = item.get("durationMs")
        action_names = [
            str(action.get("name") or action.get("path") or action.get("command") or "").strip()
            for action in item.get("commandActions", [])
            if isinstance(action, dict)
        ]
        lines = ["命令執行"]
        if status:
            lines.append(f"狀態：{status}")
        if cwd:
            lines.append(f"工作目錄：{cwd}")
        if command:
            lines.append(f"$ {command}")
        if action_names:
            lines.append("涉及項目：" + "、".join(name for name in action_names if name))
        if exit_code is not None:
            lines.append(f"退出碼：{exit_code}")
        if duration_ms is not None:
            try:
                duration_seconds = float(duration_ms) / 1000
                lines.append(f"耗時：{duration_seconds:.1f}s")
            except (TypeError, ValueError):
                lines.append(f"耗時：{duration_ms}ms")
        return "\n".join(lines).strip()

    @staticmethod
    def _timeline_item_kind(item_type: str) -> str:
        if item_type == "fileChange":
            return "fileChange"
        if item_type == "commandExecution":
            return "commandExecution"
        if item_type == "contextCompaction":
            return "contextCompaction"
        if item_type == "plan":
            return "plan"
        return "message"

    @staticmethod
    def _timeline_display_scope(item_type: str, text: str) -> str:
        if item_type == "userMessage":
            return "message"
        if item_type == "fileChange":
            return "fileChange"
        if item_type in {"commandExecution", "contextCompaction", "plan"}:
            return "process"
        return "message"

    @staticmethod
    def _plan_summary_from_text(text: str) -> dict[str, Any]:
        steps: list[dict[str, str]] = []
        for raw_line in text.splitlines():
            line = raw_line.strip()
            if not line:
                continue
            status = "pending"
            if line.startswith("[x]") or line.startswith("☒") or line.startswith("✓"):
                status = "completed"
            elif line.startswith("[ ]") or line.startswith("☐") or line.startswith("○"):
                status = "pending"
            elif line.startswith("[~]") or line.startswith("◐") or line.startswith("●"):
                status = "inProgress"
            step = line
            for prefix in ("[x]", "[ ]", "[~]", "☒", "☐", "✓", "○", "◐", "●", "-", "*"):
                if step.startswith(prefix):
                    step = step[len(prefix):].strip()
                    break
            if step:
                steps.append({"step": step, "status": status})
        completed = sum(1 for step in steps if step["status"] == "completed")
        return {
            "steps": steps,
            "total": len(steps),
            "completed": completed,
        }

    @staticmethod
    def _duration_ms(value: Any) -> int | None:
        if value is None:
            return None
        if isinstance(value, dict):
            seconds = value.get("secs", 0)
            nanos = value.get("nanos", 0)
            try:
                return int(float(seconds) * 1000 + float(nanos) / 1_000_000)
            except (TypeError, ValueError):
                return None
        try:
            return int(float(value))
        except (TypeError, ValueError):
            return None

    @staticmethod
    def _turn_duration_ms(turn: dict) -> int | None:
        duration_ms = (
            CodexLiveBridge._duration_ms(turn.get("durationMs"))
            or CodexLiveBridge._duration_ms(turn.get("duration_ms"))
            or CodexLiveBridge._duration_ms(turn.get("duration"))
        )
        if duration_ms is not None:
            return duration_ms
        started_at = turn.get("startedAt")
        completed_at = turn.get("completedAt")
        if started_at is None:
            return None
        try:
            if completed_at is None:
                status = CodexLiveBridge._status_label(turn.get("status"))
                if status != "inProgress":
                    return None
                end_at = time.time()
            else:
                end_at = float(completed_at)
            return max(0, int((end_at - float(started_at)) * 1000))
        except (TypeError, ValueError):
            return None

    @staticmethod
    def _format_turn_duration(duration_ms: int | None) -> str | None:
        if duration_ms is None or duration_ms <= 0:
            return None
        total_seconds = max(1, round(duration_ms / 1000))
        if total_seconds < 60:
            return f"已處理 {total_seconds}s"
        minutes = total_seconds // 60
        seconds = total_seconds % 60
        return f"已處理 {minutes}m {seconds}s" if seconds else f"已處理 {minutes}m"

    def _file_view_roots(self) -> list[Path]:
        roots = [
            self.repo_root,
            Path(self.thread_cwd),
            Path(self.project_root),
        ]
        try:
            selected = self._get_selected_thread()
            if selected.cwd:
                roots.append(Path(selected.cwd))
        except Exception:
            pass

        resolved_roots: list[Path] = []
        for root in roots:
            try:
                resolved = root.expanduser().resolve()
            except Exception:
                continue
            if resolved not in resolved_roots:
                resolved_roots.append(resolved)
        return resolved_roots

    @staticmethod
    def _is_relative_to(path: Path, root: Path) -> bool:
        try:
            path.relative_to(root)
            return True
        except ValueError:
            return False

    @staticmethod
    def _iso_from_epoch(value: int | None) -> str:
        timestamp = value if value is not None else int(time.time())
        return datetime.fromtimestamp(timestamp, tz=timezone.utc).isoformat().replace("+00:00", "Z")

    @staticmethod
    def _server_clock_payload() -> dict[str, Any]:
        now = datetime.now().astimezone()
        offset = now.utcoffset()
        offset_minutes = int(offset.total_seconds() // 60) if offset else 0
        return {
            "server_time": now.isoformat(),
            "server_epoch_ms": int(now.timestamp() * 1000),
            "server_timezone_offset_minutes": offset_minutes,
        }

    @staticmethod
    def _status_label(status: dict | str | None) -> str | None:
        if not status:
            return None
        if isinstance(status, str):
            return status
        kind = status.get("type")
        if kind == "loaded":
            return "loaded"
        if kind == "notLoaded":
            return "notLoaded"
        return kind or None

    @staticmethod
    def _resolve_node_bin(configured: str) -> str:
        if "/" in configured and Path(configured).exists():
            return configured
        discovered = shutil.which(configured)
        if discovered:
            return discovered
        nvm_root = Path.home() / ".nvm" / "versions" / "node"
        candidates = sorted(nvm_root.glob("*/bin/node"), reverse=True)
        if candidates:
            return str(candidates[0])
        raise CodexLiveBridgeError("找不到可用的 node 執行檔，請設定 CODEXBOT_LIVE_NODE_BIN")

    def _load_selected_thread_id(self) -> str | None:
        payload = self._load_selection_data()
        if not payload:
            return None
        thread_id = str(payload.get("thread_id", "")).strip()
        return thread_id or None

    def _persist_selection(
        self,
        *,
        thread_id: str,
        thread_name: str | None,
        cwd: str | Path | None,
        status: str | None,
        pending_materialization: bool,
    ) -> None:
        self.selection_path.parent.mkdir(parents=True, exist_ok=True)
        self.selection_path.write_text(
            json.dumps(
                {
                    "thread_id": thread_id,
                    "thread_name": thread_name,
                    "cwd": str(cwd) if cwd else str(self.thread_cwd),
                    "updated_at": int(time.time()),
                    "status": status or "idle",
                    "pending_materialization": pending_materialization,
                },
                ensure_ascii=False,
            ),
            encoding="utf-8",
        )
