from __future__ import annotations

import base64
import datetime as dt
import json
import logging
import os
import plistlib
import shutil
import socket
import subprocess
import tempfile
import threading
import time
import uuid
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib import request as urlrequest
from urllib.parse import parse_qs, quote, urlparse

from .agent import NullTaskPlanner, OpenAITaskPlanner
from .codex_live_bridge import CodexLiveBridge, CodexLiveBridgeError
from .config import Settings
from .executor import LocalCommandExecutor
from .notifier import NullNotifier
from .state_store import StateStore
from .worker import WorkerController

logger = logging.getLogger(__name__)

_PROCESS_ENV_LOADED = False


def _load_process_env_file() -> None:
    global _PROCESS_ENV_LOADED
    if _PROCESS_ENV_LOADED:
        return
    _PROCESS_ENV_LOADED = True
    candidates = [
        Path.cwd() / ".env",
        Path(__file__).resolve().parents[2] / ".env",
    ]
    for env_path in candidates:
        if not env_path.exists():
            continue
        try:
            lines = env_path.read_text(encoding="utf-8").splitlines()
        except OSError:
            continue
        for line in lines:
            stripped = line.strip()
            if not stripped or stripped.startswith("#") or "=" not in stripped:
                continue
            key, value = stripped.split("=", 1)
            key = key.strip()
            if not key or key in os.environ:
                continue
            value = value.strip()
            if len(value) >= 2 and value[0] == value[-1] and value[0] in {"'", '"'}:
                value = value[1:-1]
            os.environ[key] = value
        return


def _env_first(*names: str, default: str = "") -> str:
    _load_process_env_file()
    for name in names:
        value = os.environ.get(name)
        if value:
            return value
    return default


def _upload_file_to_public(
    local_path: Path,
    filename: str,
    *,
    host_envs: tuple[str, ...],
    user_envs: tuple[str, ...],
    remote_dir_envs: tuple[str, ...],
    public_base_url_envs: tuple[str, ...],
    default_host: str,
    default_user: str,
    default_remote_dir: str,
    default_public_base_url: str,
    invalid_message: str,
    scp_timeout: int = 45,
) -> dict:
    remote_host = _env_first(
        *host_envs,
        default=default_host,
    )
    remote_user = _env_first(
        *user_envs,
        default=default_user,
    )
    remote_dir = _env_first(
        *remote_dir_envs,
        default=default_remote_dir,
    ).rstrip("/")
    public_base_url = _env_first(
        *public_base_url_envs,
        default=default_public_base_url,
    ).rstrip("/")

    if not local_path.exists() or local_path.stat().st_size <= 0:
        raise RuntimeError(invalid_message)
    if not remote_host or not public_base_url:
        host_names = " / ".join(host_envs)
        url_names = " / ".join(public_base_url_envs)
        raise RuntimeError(
            "未配置公開上傳目標，請設定公開上傳 host 和 public base URL 環境變數："
            f"{host_names} 與 {url_names}。"
        )

    remote = f"{remote_user}@{remote_host}"
    subprocess.run(
        ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=8", remote, f"mkdir -p {remote_dir}"],
        check=True,
        timeout=15,
    )
    subprocess.run(
        ["scp", "-o", "BatchMode=yes", "-o", "ConnectTimeout=8", str(local_path), f"{remote}:{remote_dir}/{filename}"],
        check=True,
        timeout=scp_timeout,
    )
    subprocess.run(
        ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=8", remote, f"chmod 0644 {remote_dir}/{filename}"],
        check=True,
        timeout=15,
    )
    return {
        "filename": filename,
        "local_path": str(local_path),
        "url": f"{public_base_url}/{filename}",
        "uploaded_at": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
    }


def _upload_png_to_public(local_path: Path, filename: str) -> dict:
    return _upload_file_to_public(
        local_path,
        filename,
        host_envs=("CODEXMOBILE_SCREENSHOT_VPS_HOST", "CODEXMOBILE_VPS_HOST", "CODEXBOT_VPS_HOST"),
        user_envs=("CODEXMOBILE_SCREENSHOT_VPS_USER", "CODEXMOBILE_VPS_USER", "CODEXBOT_VPS_USER"),
        remote_dir_envs=("CODEXMOBILE_SCREENSHOT_REMOTE_DIR",),
        public_base_url_envs=("CODEXMOBILE_SCREENSHOT_PUBLIC_BASE_URL",),
        default_host="",
        default_user="root",
        default_remote_dir="/var/www/html/codexmobile-shots",
        default_public_base_url="",
        invalid_message="圖片沒有產生有效文件。",
    )


def _upload_adhoc_artifact_to_public(local_path: Path, filename: str) -> dict:
    return _upload_file_to_public(
        local_path,
        filename,
        host_envs=("CODEXMOBILE_ADHOC_VPS_HOST", "CODEXMOBILE_VPS_HOST", "CODEXBOT_VPS_HOST"),
        user_envs=("CODEXMOBILE_ADHOC_VPS_USER", "CODEXMOBILE_VPS_USER", "CODEXBOT_VPS_USER"),
        remote_dir_envs=("CODEXMOBILE_ADHOC_REMOTE_DIR", "CODEXMOBILE_ARTIFACT_REMOTE_DIR"),
        public_base_url_envs=("CODEXMOBILE_ADHOC_PUBLIC_BASE_URL", "CODEXMOBILE_ARTIFACT_PUBLIC_BASE_URL"),
        default_host="",
        default_user="root",
        default_remote_dir="/var/www/html/codexmobile-artifacts",
        default_public_base_url="",
        invalid_message="Ad Hoc 打包文件沒有產生有效文件。",
        scp_timeout=600,
    )


def _capture_and_upload_desktop_screenshot() -> dict:
    timestamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%d-%H%M%S")
    filename = f"desktop-{timestamp}.png"
    local_path = Path(tempfile.gettempdir()) / filename

    subprocess.run(["/usr/sbin/screencapture", "-x", str(local_path)], check=True, timeout=20)
    try:
        uploaded = _upload_png_to_public(local_path, filename)
    except RuntimeError:
        uploaded = {
            "filename": filename,
            "local_path": str(local_path),
            "url": "",
            "uploaded_at": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
        }
    return {
        "ok": True,
        "message": "已截取 Mac 桌面截圖。",
        **uploaded,
    }


def _process_tail(value: str, limit: int = 4000) -> str:
    value = value.strip()
    if len(value) <= limit:
        return value
    return value[-limit:]


def _run_captured(args: list[str], *, cwd: Path | None = None, timeout: int = 60, check: bool = True) -> subprocess.CompletedProcess[str]:
    completed = subprocess.run(
        args,
        cwd=str(cwd) if cwd else None,
        text=True,
        capture_output=True,
        timeout=timeout,
        check=False,
    )
    if check and completed.returncode != 0:
        output = "\n".join(part for part in [completed.stdout, completed.stderr] if part).strip()
        raise RuntimeError(f"{' '.join(args)} failed with exit {completed.returncode}:\n{_process_tail(output)}")
    return completed


def _ios_adhoc_allow_provisioning_updates() -> bool:
    raw = _env_first("CODEXMOBILE_IOS_ALLOW_PROVISIONING_UPDATES", default="1").strip().lower()
    return raw not in {"0", "false", "no", "off"}


def _apple_developer_services_resolvable() -> bool:
    try:
        socket.getaddrinfo("developerservices3.apple.com", 443)
        return True
    except OSError:
        return False


def _local_ios_provisioning_profile_count() -> int:
    profiles_dir = Path.home() / "Library" / "MobileDevice" / "Provisioning Profiles"
    if not profiles_dir.exists():
        return 0
    return sum(1 for path in profiles_dir.glob("*") if path.is_file())


def _write_ios_adhoc_export_options(path: Path, *, team_id: str) -> dict:
    export_method = _env_first("CODEXMOBILE_IOS_EXPORT_METHOD", default="release-testing")
    payload = {
        "compileBitcode": False,
        "destination": "export",
        "manageAppVersionAndBuildNumber": False,
        "method": export_method,
        "stripSwiftSymbols": True,
    }
    signing_style = _env_first("CODEXMOBILE_IOS_SIGNING_STYLE", "CODEXBOT_IOS_SIGNING_STYLE")
    if signing_style:
        payload["signingStyle"] = signing_style
    if team_id:
        payload["teamID"] = team_id
    path.write_bytes(plistlib.dumps(payload, sort_keys=False))
    return payload


def _write_ios_adhoc_manifest(
    path: Path,
    *,
    ipa_url: str,
    bundle_id: str,
    bundle_version: str,
    title: str,
) -> dict:
    payload = {
        "items": [
            {
                "assets": [
                    {
                        "kind": "software-package",
                        "url": ipa_url,
                    }
                ],
                "metadata": {
                    "bundle-identifier": bundle_id,
                    "bundle-version": bundle_version,
                    "kind": "software",
                    "title": title,
                },
            }
        ]
    }
    path.write_bytes(plistlib.dumps(payload, sort_keys=False))
    return payload


def _safe_artifact_stem(value: str) -> str:
    cleaned = "".join(ch if ch.isalnum() or ch in {".", "-", "_"} else "-" for ch in value.strip())
    cleaned = "-".join(part for part in cleaned.split("-") if part)
    return cleaned or "ios-app"


def _iter_ios_xcode_files(project_root: Path, suffix: str) -> list[Path]:
    ignored_parts = {"node_modules", "build", "DerivedData", ".git"}
    matches: list[Path] = []
    for root, dirs, _files in os.walk(project_root):
        dirs[:] = [name for name in dirs if name not in ignored_parts]
        for name in dirs:
            if not name.endswith(suffix) or name.startswith("Pods."):
                continue
            matches.append(Path(root) / name)
    return matches


def _xcode_file_sort_key(project_root: Path, path: Path) -> tuple[int, int, str]:
    relative = path.relative_to(project_root)
    parts = set(relative.parts)
    ios_bias = 0 if "ios" in parts else 1
    return (ios_bias, len(relative.parts), str(relative).casefold())


def _discover_ios_xcode_file(project_root: Path) -> tuple[str, Path]:
    if not project_root.exists() or not project_root.is_dir():
        raise RuntimeError(f"找不到項目資料夾：{project_root}")
    workspaces = sorted(_iter_ios_xcode_files(project_root, ".xcworkspace"), key=lambda path: _xcode_file_sort_key(project_root, path))
    if workspaces:
        return ("workspace", workspaces[0])
    projects = sorted(_iter_ios_xcode_files(project_root, ".xcodeproj"), key=lambda path: _xcode_file_sort_key(project_root, path))
    if projects:
        return ("project", projects[0])
    raise RuntimeError(f"當前項目不是可打包的 iOS/Xcode 項目，找不到 .xcworkspace 或 .xcodeproj：{project_root}")


def _xcode_schemes(kind: str, xcode_file: Path) -> list[str]:
    flag = "-workspace" if kind == "workspace" else "-project"
    completed = _run_captured(["xcodebuild", "-list", "-json", flag, str(xcode_file)], cwd=xcode_file.parent, timeout=120)
    payload = json.loads(completed.stdout or "{}")
    section = payload.get("workspace") if kind == "workspace" else payload.get("project")
    schemes = section.get("schemes") if isinstance(section, dict) else None
    if not isinstance(schemes, list):
        schemes = payload.get("schemes")
    return [str(item).strip() for item in schemes or [] if str(item or "").strip()]


def _select_xcode_scheme(kind: str, xcode_file: Path, project_root: Path, requested_scheme: str | None = None) -> str:
    schemes = _xcode_schemes(kind, xcode_file)
    if requested_scheme:
        if requested_scheme in schemes:
            return requested_scheme
        raise RuntimeError(f"找不到指定 scheme：{requested_scheme}。可用 schemes：{', '.join(schemes) or '無'}")
    if len(schemes) == 1:
        return schemes[0]
    preferred = [
        xcode_file.stem,
        project_root.name,
        xcode_file.parent.name,
    ]
    for name in preferred:
        if name in schemes:
            return name
    app_like = [scheme for scheme in schemes if not scheme.lower().startswith(("pods-", "test", "tests"))]
    if len(app_like) == 1:
        return app_like[0]
    raise RuntimeError(f"當前 Xcode 項目有多個 scheme，無法自動判斷要打包哪一個：{', '.join(schemes) or '無'}")


def _find_project_in_sidebar(projects: list[dict], project_id: str) -> dict | None:
    target = str(project_id or "").strip()
    if not target:
        return None
    for project in projects:
        if str(project.get("project_id") or "") == target:
            return project
        for child in project.get("children") or []:
            if str(child.get("project_id") or "") == target:
                merged = dict(child)
                if project.get("cwd"):
                    merged["_parent_cwd"] = project.get("cwd")
                if not merged.get("cwd"):
                    merged["cwd"] = project.get("cwd")
                if not merged.get("parent_project_title"):
                    merged["parent_project_title"] = project.get("title")
                return merged
    return None


def _resolve_ios_adhoc_project_root(
    live_bridge: CodexLiveBridge | None,
    *,
    thread_id: str | None = None,
    project_id: str | None = None,
    cwd: str | None = None,
) -> dict:
    candidates: list[tuple[str, str]] = []
    has_explicit_target = bool(str(cwd or "").strip() or str(project_id or "").strip() or str(thread_id or "").strip())
    if cwd:
        candidates.append(("request cwd", cwd))
    if project_id and project_id.startswith("cwd:"):
        candidates.append(("project_id cwd", project_id.removeprefix("cwd:")))
    if live_bridge is not None and project_id:
        project = _find_project_in_sidebar(live_bridge.list_projects(force=True), project_id)
        if project and project.get("cwd"):
            parent_cwd = str(project.get("_parent_cwd") or "").strip()
            if parent_cwd:
                candidates.append((f"parent project {project.get('parent_project_title') or project_id}", parent_cwd))
            candidates.append((f"project {project.get('title') or project_id}", str(project["cwd"])))
    if live_bridge is not None and thread_id:
        try:
            snapshot = live_bridge.get_thread_snapshot_payload(thread_id, limit=1)
            snapshot_cwd = ((snapshot.get("state") or {}).get("cwd") or "").strip()
            if snapshot_cwd:
                candidates.append((f"thread {thread_id}", snapshot_cwd))
        except Exception as exc:
            logger.warning("Unable to resolve Ad Hoc project from thread snapshot thread_id=%s: %s", thread_id, exc)
    if live_bridge is not None and not has_explicit_target:
        try:
            selected = live_bridge._get_selected_thread_fast()
            if selected.cwd:
                candidates.append(("selected desktop thread", selected.cwd))
        except Exception as exc:
            logger.warning("Unable to resolve Ad Hoc project from selected thread: %s", exc)

    seen: set[str] = set()
    rejected: list[str] = []
    for source, raw_cwd in candidates:
        text = str(raw_cwd or "").strip()
        if not text:
            continue
        path = Path(text).expanduser()
        if not path.is_absolute():
            path = (Path.cwd() / path).resolve(strict=False)
        else:
            path = path.resolve(strict=False)
        key = str(path)
        if key in seen:
            continue
        seen.add(key)
        if path.exists() and path.is_dir():
            try:
                _discover_ios_xcode_file(path)
            except RuntimeError as exc:
                rejected.append(f"{source}: {exc}")
                continue
            return {
                "cwd": str(path),
                "source": source,
            }
        rejected.append(f"{source}: 找不到項目資料夾：{path}")
    if rejected:
        raise RuntimeError("無法找到可打包的 iOS/Xcode 項目。已嘗試：" + "；".join(rejected))
    raise RuntimeError("無法判斷要打包哪個項目：手機端需要提供目前項目的 cwd / project_id / thread_id。")


def _archive_application_properties(archive_path: Path) -> dict:
    info_path = archive_path / "Info.plist"
    if not info_path.exists():
        return {}
    try:
        payload = plistlib.loads(info_path.read_bytes())
    except Exception:
        logger.exception("Unable to read archive Info.plist: %s", info_path)
        return {}
    properties = payload.get("ApplicationProperties")
    return properties if isinstance(properties, dict) else {}


def _run_ios_adhoc_package_build(
    live_bridge: CodexLiveBridge | None = None,
    *,
    thread_id: str | None = None,
    project_id: str | None = None,
    cwd: str | None = None,
    scheme: str | None = None,
) -> dict:
    project = _resolve_ios_adhoc_project_root(live_bridge, thread_id=thread_id, project_id=project_id, cwd=cwd)
    project_root = Path(str(project["cwd"]))
    xcode_kind, xcode_file = _discover_ios_xcode_file(project_root)
    requested_scheme = (scheme or _env_first("CODEXMOBILE_IOS_SCHEME", "CODEXBOT_IOS_SCHEME")).strip() or None
    scheme = _select_xcode_scheme(xcode_kind, xcode_file, project_root, requested_scheme=requested_scheme)
    team_id = _env_first("CODEXMOBILE_IOS_TEAM_ID", "CODEXBOT_IOS_TEAM_ID")
    bundle_id_override = _env_first("CODEXMOBILE_IOS_BUNDLE_ID", "CODEXBOT_IOS_BUNDLE_ID")
    marketing_version_override = _env_first("CODEXMOBILE_IOS_MARKETING_VERSION", "CODEXBOT_IOS_MARKETING_VERSION")
    build_version_override = _env_first("CODEXMOBILE_IOS_BUILD_VERSION", "CODEXBOT_IOS_BUILD_VERSION")
    timestamp = dt.datetime.now(dt.timezone.utc).strftime("%Y%m%d-%H%M%S")
    safe_scheme = _safe_artifact_stem(scheme)
    output_root = Path(tempfile.gettempdir()) / f"{safe_scheme}-ios-adhoc-{timestamp}"
    archive_path = output_root / f"{scheme}.xcarchive"
    export_dir = output_root / "export"
    export_options_path = output_root / "ExportOptions.plist"
    output_root.mkdir(parents=True, exist_ok=True)
    export_dir.mkdir(parents=True, exist_ok=True)

    _write_ios_adhoc_export_options(export_options_path, team_id=team_id)
    allow_updates = _ios_adhoc_allow_provisioning_updates()
    provisioning_profile_count = _local_ios_provisioning_profile_count()
    if allow_updates and provisioning_profile_count <= 0 and not _apple_developer_services_resolvable():
        raise RuntimeError(
            "Ad Hoc export 需要 Xcode 連到 developerservices3.apple.com 建立或更新 provisioning profile，"
            "但目前這台 Mac 無法解析 Apple Developer Services，且本機沒有可用 provisioning profile。"
            "請先恢復 Mac 到 Apple Developer Services 的網路連線後重試。"
        )
    xcode_flag = "-workspace" if xcode_kind == "workspace" else "-project"
    archive_command = [
        "xcodebuild",
        xcode_flag,
        str(xcode_file),
        "-scheme",
        scheme,
        "-configuration",
        "Release",
        "-destination",
        "generic/platform=iOS",
        "-archivePath",
        str(archive_path),
    ]
    if allow_updates:
        archive_command.append("-allowProvisioningUpdates")
    archive_command.append("archive")
    if team_id:
        archive_command.append(f"DEVELOPMENT_TEAM={team_id}")
    code_sign_style = _env_first("CODEXMOBILE_IOS_CODE_SIGN_STYLE", "CODEXBOT_IOS_CODE_SIGN_STYLE")
    if code_sign_style:
        archive_command.append(f"CODE_SIGN_STYLE={code_sign_style}")
    if bundle_id_override:
        archive_command.append(f"PRODUCT_BUNDLE_IDENTIFIER={bundle_id_override}")
    if marketing_version_override:
        archive_command.append(f"MARKETING_VERSION={marketing_version_override}")
    if build_version_override:
        archive_command.append(f"CURRENT_PROJECT_VERSION={build_version_override}")
    archive = _run_captured(archive_command, cwd=xcode_file.parent, timeout=1800)
    if not archive_path.exists():
        raise RuntimeError(f"xcodebuild archive 完成，但找不到 archive: {archive_path}")
    archive_properties = _archive_application_properties(archive_path)
    bundle_id = bundle_id_override or str(archive_properties.get("CFBundleIdentifier") or "").strip()
    marketing_version = marketing_version_override or str(archive_properties.get("CFBundleShortVersionString") or "").strip() or "1.0"
    build_version = build_version_override or str(archive_properties.get("CFBundleVersion") or "").strip() or "1"
    if not bundle_id:
        raise RuntimeError("xcodebuild archive 完成，但無法從 archive 讀取 bundle identifier。")

    export_command = [
        "xcodebuild",
        "-exportArchive",
        "-archivePath",
        str(archive_path),
        "-exportPath",
        str(export_dir),
        "-exportOptionsPlist",
        str(export_options_path),
    ]
    if allow_updates:
        export_command.append("-allowProvisioningUpdates")
    export = _run_captured(export_command, cwd=xcode_file.parent, timeout=1200)

    ipa_candidates = sorted(export_dir.glob("*.ipa"), key=lambda path: path.stat().st_mtime, reverse=True)
    if not ipa_candidates:
        raise RuntimeError(f"xcodebuild export 完成，但找不到 .ipa: {export_dir}")
    ipa_path = ipa_candidates[0]
    ipa_filename = f"{safe_scheme}-adhoc-{timestamp}.ipa"
    public_ipa_path = ipa_path.with_name(ipa_filename)
    shutil.copy2(ipa_path, public_ipa_path)
    ipa_upload = _upload_adhoc_artifact_to_public(public_ipa_path, ipa_filename)

    manifest_filename = f"{safe_scheme}-adhoc-{timestamp}.plist"
    manifest_path = output_root / manifest_filename
    _write_ios_adhoc_manifest(
        manifest_path,
        ipa_url=ipa_upload["url"],
        bundle_id=bundle_id,
        bundle_version=build_version,
        title=scheme,
    )
    manifest_upload = _upload_adhoc_artifact_to_public(manifest_path, manifest_filename)
    install_url = f"itms-services://?action=download-manifest&url={quote(manifest_upload['url'], safe='')}"

    return {
        "ok": True,
        "message": f"已完成 {scheme} Ad Hoc 打包，並上傳 IPA 與安裝 manifest。",
        "project_cwd": str(project_root),
        "project_source": project["source"],
        "xcode_kind": xcode_kind,
        "xcode_path": str(xcode_file),
        "scheme": scheme,
        "bundle_id": bundle_id,
        "team_id": team_id or archive_properties.get("Team"),
        "version": marketing_version,
        "build": build_version,
        "provisioning_profile_count": provisioning_profile_count,
        "archive_path": str(archive_path),
        "export_path": str(export_dir),
        "filename": ipa_filename,
        "ipa_url": ipa_upload["url"],
        "manifest_url": manifest_upload["url"],
        "install_url": install_url,
        "url": ipa_upload["url"],
        "uploaded_at": ipa_upload["uploaded_at"],
        "archive_output_tail": _process_tail("\n".join(part for part in [archive.stdout, archive.stderr] if part)),
        "export_output_tail": _process_tail("\n".join(part for part in [export.stdout, export.stderr] if part)),
    }


def _restart_codex_desktop(live_bridge: CodexLiveBridge | None = None, thread_id: str | None = None) -> dict:
    target_thread_id = (thread_id or "").strip()
    if live_bridge is not None and not target_thread_id:
        try:
            target_thread_id = live_bridge._get_selected_thread().thread_id
        except Exception:
            target_thread_id = ""

    # Codex shows a blocking "Quit Codex?" confirmation for normal AppleScript
    # quits when local threads are active. Kill the app process directly so the
    # mobile "restart desktop Codex" action does not get stuck behind that dialog.
    subprocess.run(["/usr/bin/pkill", "-x", "Codex"], check=False, timeout=8)
    time.sleep(0.8)
    subprocess.run(["/usr/bin/pkill", "-9", "-x", "Codex"], check=False, timeout=8)
    time.sleep(0.8)
    subprocess.run(["open", "-a", "Codex"], check=True, timeout=15)
    desktop_refresh = None
    refresh_error = None
    if live_bridge is not None and target_thread_id:
        for attempt in range(12):
            try:
                time.sleep(0.8 if attempt == 0 else 1.0)
                desktop_refresh = live_bridge.refresh_desktop_thread(target_thread_id, force=True)
                break
            except Exception as exc:
                refresh_error = str(exc)
    return {
        "ok": True,
        "message": "已重啟 Codex 桌面端，並嘗試回到目前手機端會話。" if desktop_refresh else "已重啟 Codex 桌面端，但尚未確認桌面端已切回目前手機端會話。",
        "restarted_at": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
        "thread_id": target_thread_id or thread_id,
        "desktop_refresh": desktop_refresh,
        "desktop_refresh_error": refresh_error,
    }


def _restart_desktop_connection() -> dict:
    label = "com.anthony.codexmobile.menuclient"
    plist_path = Path.home() / "Library" / "LaunchAgents" / f"{label}.plist"
    target = f"gui/{os.getuid()}/{label}"
    subprocess.run(["/usr/bin/pkill", "-x", "CodexMobileMenuClient"], check=False, timeout=8)
    time.sleep(0.6)
    print_result = subprocess.run(["launchctl", "print", target], check=False, capture_output=True, text=True, timeout=5)
    if print_result.returncode != 0 and plist_path.exists():
        subprocess.run(["launchctl", "bootstrap", f"gui/{os.getuid()}", str(plist_path)], check=False, capture_output=True, text=True, timeout=8)
    completed = subprocess.run(["launchctl", "kickstart", "-k", target], check=False, capture_output=True, text=True, timeout=15)
    if completed.returncode != 0:
        detail = (completed.stderr or completed.stdout or "").strip()
        raise RuntimeError(f"重啟桌面菜單欄連接失敗：{detail or f'launchctl exited {completed.returncode}'}")
    time.sleep(0.6)
    process_check = subprocess.run(
        ["/usr/bin/pgrep", "-x", "CodexMobileMenuClient"],
        check=False,
        capture_output=True,
        text=True,
        timeout=5,
    )
    pids = [line.strip() for line in process_check.stdout.splitlines() if line.strip()]
    return {
        "ok": True,
        "message": "已重啟桌面連接（Mac 菜單欄）。",
        "label": label,
        "process_count": len(pids),
        "pids": pids,
        "restarted_at": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
    }


def _restart_mac() -> dict:
    return {
        "ok": False,
        "error": "Mac restart disabled",
        "message": "重啟 Mac 功能已停用。",
        "requested_at": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
    }


def _parse_snapshot_limit(query: dict[str, list[str]], default: int = 180) -> int:
    raw = query.get("limit", [str(default)])[0]
    try:
        return max(0, min(220, int(raw)))
    except (TypeError, ValueError):
        return default

def build_local_worker(settings: Settings) -> WorkerController:
    planner = (
        OpenAITaskPlanner(settings.openai_api_key, settings.codexbot_agent_model)
        if settings.openai_api_key
        else NullTaskPlanner()
    )
    store = StateStore(settings.codexbot_state_path)
    executor = LocalCommandExecutor(
        workdir=settings.codexbot_workdir,
        allowed_commands=settings.allowed_commands,
        max_output_chars=settings.codexbot_max_output_chars,
    )
    return WorkerController(
        store=store,
        executor=executor,
        planner=planner,
        notifier=NullNotifier(),
        poll_seconds=settings.codexbot_worker_poll_seconds,
    )

class MacServiceHandler(BaseHTTPRequestHandler):
    worker: WorkerController
    token: str
    settings: Settings
    live_bridge: CodexLiveBridge

    def _desktop_send_status_lock(self) -> threading.Lock:
        lock = getattr(self.server, "desktop_send_status_lock", None)
        if lock is None:
            lock = threading.Lock()
            setattr(self.server, "desktop_send_status_lock", lock)
        return lock

    def _desktop_send_statuses(self) -> list[dict]:
        statuses = getattr(self.server, "desktop_send_statuses", None)
        if not isinstance(statuses, list):
            statuses = []
            setattr(self.server, "desktop_send_statuses", statuses)
        return statuses

    def _set_desktop_send_status(self, request_id: str, **updates: object) -> dict:
        now = dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z")
        with self._desktop_send_status_lock():
            statuses = self._desktop_send_statuses()
            current = next((item for item in statuses if item.get("request_id") == request_id), None)
            if current is None:
                current = {"request_id": request_id, "created_at": now}
                statuses.append(current)
            current.update({key: value for key, value in updates.items() if value is not None})
            current["updated_at"] = now
            del statuses[:-20]
            return dict(current)

    def _latest_desktop_send_status(self) -> dict | None:
        def is_fresh(status: dict) -> bool:
            normalized = str(status.get("status") or "").lower()
            if normalized in {"queued", "sending", "retrying"}:
                return True
            timestamp = self._parse_iso_timestamp(status.get("updated_at") or status.get("created_at"))
            return bool(timestamp and (time.time() - timestamp) <= 300)

        with self._desktop_send_status_lock():
            statuses = self._desktop_send_statuses()
            fresh_statuses = [status for status in statuses if is_fresh(status)]
            if len(fresh_statuses) != len(statuses):
                statuses[:] = fresh_statuses[-20:]
            if fresh_statuses:
                return dict(fresh_statuses[-1])
        with self._desktop_send_queue_lock():
            self._expire_stale_desktop_send_jobs_locked()
            jobs = self._desktop_send_jobs()
            for job in reversed(jobs):
                public_job = self._public_desktop_send_job(job)
                if is_fresh(public_job):
                    return public_job
            return None

    def _adhoc_build_status_lock(self) -> threading.Lock:
        lock = getattr(self.server, "adhoc_build_status_lock", None)
        if lock is None:
            lock = threading.Lock()
            setattr(self.server, "adhoc_build_status_lock", lock)
        return lock

    def _adhoc_build_statuses(self) -> list[dict]:
        statuses = getattr(self.server, "adhoc_build_statuses", None)
        if not isinstance(statuses, list):
            statuses = []
            setattr(self.server, "adhoc_build_statuses", statuses)
        return statuses

    def _set_adhoc_build_status(self, request_id: str, **updates: object) -> dict:
        now = self._now_iso()
        with self._adhoc_build_status_lock():
            statuses = self._adhoc_build_statuses()
            current = next((item for item in statuses if item.get("request_id") == request_id), None)
            if current is None:
                current = {"request_id": request_id, "created_at": now}
                statuses.append(current)
            current.update({key: value for key, value in updates.items() if value is not None})
            current["updated_at"] = now
            del statuses[:-20]
            return dict(current)

    def _get_adhoc_build_status(self, request_id: str) -> dict | None:
        with self._adhoc_build_status_lock():
            current = next((item for item in self._adhoc_build_statuses() if item.get("request_id") == request_id), None)
            return dict(current) if current else None

    def _start_adhoc_build_job(self, job: dict) -> dict:
        request_id = str(job["request_id"])
        self._set_adhoc_build_status(
            request_id,
            status="queued",
            accepted=True,
            thread_id=job.get("thread_id"),
            project_id=job.get("project_id"),
            cwd=job.get("cwd"),
            scheme=job.get("scheme"),
        )

        def worker() -> None:
            self._set_adhoc_build_status(request_id, status="running", started_at=self._now_iso())
            try:
                payload = _run_ios_adhoc_package_build(
                    self.live_bridge,
                    thread_id=str(job.get("thread_id") or "").strip() or None,
                    project_id=str(job.get("project_id") or "").strip() or None,
                    cwd=str(job.get("cwd") or "").strip() or None,
                    scheme=str(job.get("scheme") or "").strip() or None,
                )
                self._set_adhoc_build_status(
                    request_id,
                    status="succeeded",
                    ok=True,
                    accepted=True,
                    completed_at=self._now_iso(),
                    result=payload,
                    project_cwd=payload.get("project_cwd"),
                    scheme=payload.get("scheme"),
                    install_url=payload.get("install_url"),
                    ipa_url=payload.get("ipa_url") or payload.get("url"),
                    manifest_url=payload.get("manifest_url"),
                    filename=payload.get("filename"),
                    message=payload.get("message"),
                )
            except Exception as exc:  # noqa: BLE001 - surfaced as job status.
                logger.exception("Async iOS Ad Hoc build failed")
                self._set_adhoc_build_status(
                    request_id,
                    status="failed",
                    ok=False,
                    accepted=False,
                    completed_at=self._now_iso(),
                    error=str(exc),
                )

        threading.Thread(target=worker, daemon=True).start()
        return self._get_adhoc_build_status(request_id) or {"request_id": request_id, "status": "queued", "accepted": True}

    @staticmethod
    def _desktop_send_error_text(exc: BaseException) -> str:
        text = str(exc).strip() or exc.__class__.__name__
        return text[:1600]

    @staticmethod
    def _run_daemon_call_with_timeout(func, timeout_seconds: float) -> tuple[bool, object | None, BaseException | None]:
        done = threading.Event()
        result: dict[str, object] = {}

        def runner() -> None:
            try:
                result["value"] = func()
            except BaseException as exc:  # noqa: BLE001 - propagated to caller as data.
                result["error"] = exc
            finally:
                done.set()

        threading.Thread(target=runner, daemon=True).start()
        if not done.wait(timeout_seconds):
            return False, None, None
        error = result.get("error")
        return True, result.get("value"), error if isinstance(error, BaseException) else None

    @staticmethod
    def _now_iso() -> str:
        return dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z")

    @staticmethod
    def _parse_iso_timestamp(value: object) -> float:
        if not isinstance(value, str) or not value:
            return 0.0
        try:
            return dt.datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
        except ValueError:
            return 0.0

    def _desktop_send_queue_path(self) -> Path:
        return self.settings.codexbot_desktop_send_queue_path

    def _desktop_send_queue_lock(self) -> threading.RLock:
        lock = getattr(self.server, "desktop_send_queue_lock", None)
        if lock is None:
            lock = threading.RLock()
            setattr(self.server, "desktop_send_queue_lock", lock)
        return lock

    def _desktop_send_jobs(self) -> list[dict]:
        jobs = getattr(self.server, "desktop_send_jobs", None)
        if isinstance(jobs, list):
            return jobs
        path = self._desktop_send_queue_path()
        loaded: list[dict] = []
        if path.exists():
            try:
                raw = json.loads(path.read_text(encoding="utf-8"))
                if isinstance(raw, dict) and isinstance(raw.get("jobs"), list):
                    loaded = [item for item in raw["jobs"] if isinstance(item, dict)]
                elif isinstance(raw, list):
                    loaded = [item for item in raw if isinstance(item, dict)]
            except Exception:
                logger.exception("Failed to load desktop send queue")
        setattr(self.server, "desktop_send_jobs", loaded[-100:])
        return getattr(self.server, "desktop_send_jobs")

    def _save_desktop_send_jobs_locked(self) -> None:
        path = self._desktop_send_queue_path()
        path.parent.mkdir(parents=True, exist_ok=True)
        jobs = self._desktop_send_jobs()[-100:]
        tmp = path.with_suffix(path.suffix + ".tmp")
        tmp.write_text(json.dumps({"version": 1, "jobs": jobs}, ensure_ascii=False, indent=2), encoding="utf-8")
        tmp.replace(path)

    def _public_desktop_send_job(self, job: dict) -> dict:
        return {
            key: value
            for key, value in job.items()
            if key
            not in {
                "image_data_urls",
                "local_image_paths",
                "text",
            }
        } | {"prompt_preview": str(job.get("text") or "")[:160]}

    def _publish_desktop_send_job_status_locked(self, job: dict) -> dict:
        request_id = str(job.get("request_id") or "")
        public_job = self._public_desktop_send_job(job)
        public_job.pop("request_id", None)
        if not request_id:
            return public_job
        return self._set_desktop_send_status(request_id, **public_job)

    def _enqueue_desktop_send_job(self, job: dict) -> dict:
        with self._desktop_send_queue_lock():
            jobs = self._desktop_send_jobs()
            jobs.append(job)
            del jobs[:-100]
            self._save_desktop_send_jobs_locked()
        self._set_desktop_send_status(
            str(job["request_id"]),
            status="queued",
            thread_id=job.get("thread_id"),
            attempts=0,
            max_attempts=job.get("max_attempts"),
            prompt_preview=str(job.get("text") or "")[:160],
        )
        self._ensure_desktop_send_worker()
        return dict(job)

    def _latest_desktop_send_job(self) -> dict | None:
        with self._desktop_send_queue_lock():
            self._expire_stale_desktop_send_jobs_locked()
            jobs = self._desktop_send_jobs()
            return self._public_desktop_send_job(jobs[-1]) if jobs else None

    def _desktop_send_queue_payload(self) -> dict:
        with self._desktop_send_queue_lock():
            self._expire_stale_desktop_send_jobs_locked()
            jobs = [self._public_desktop_send_job(job) for job in self._desktop_send_jobs()[-20:]]
        return {"ok": True, "jobs": jobs}

    def _wait_for_desktop_send_job(self, request_id: str, timeout_seconds: float = 8.0) -> dict | None:
        deadline = time.time() + max(0.0, timeout_seconds)
        while time.time() < deadline:
            with self._desktop_send_queue_lock():
                jobs = self._desktop_send_jobs()
                job = next((item for item in jobs if str(item.get("request_id") or "") == request_id), None)
                if job and str(job.get("status") or "") in {"sent", "failed", "cancelled"}:
                    return self._public_desktop_send_job(job)
            time.sleep(0.35)
        return None

    def _attach_latest_desktop_send_status(self, payload: dict) -> dict:
        latest_send = self._latest_desktop_send_status()
        if latest_send:
            payload["desktop_send_status"] = latest_send
            if isinstance(payload.get("state"), dict):
                payload["state"]["desktop_send_status"] = latest_send
        return payload

    def _ensure_desktop_send_worker(self) -> None:
        with self._desktop_send_queue_lock():
            self._expire_stale_desktop_send_jobs_locked()
            has_pending = any(
                str(item.get("status") or "") in {"queued", "retrying"}
                for item in self._desktop_send_jobs()
            )
        worker = getattr(self.server, "desktop_send_worker", None)
        if isinstance(worker, threading.Thread) and worker.is_alive():
            if not has_pending:
                return
            started_at = float(getattr(self.server, "desktop_send_worker_started_at", 0.0) or 0.0)
            # A daemon thread can be stuck inside the Codex app-server bridge. After
            # stale jobs are expired, allow a fresh worker to pick up queued work.
            if started_at and time.time() - started_at < 120.0:
                return
            logger.warning("Replacing stale desktop send worker")
        else:
            if not has_pending:
                return
        setattr(self.server, "desktop_send_worker_started_at", time.time())
        worker = threading.Thread(target=self._desktop_send_worker_loop, daemon=True)
        setattr(self.server, "desktop_send_worker", worker)
        worker.start()

    @staticmethod
    def _desktop_send_retry_delay(attempts: int) -> float:
        return min(90.0, max(2.0, 3.0 * (2 ** max(0, min(attempts - 1, 5)))))

    @staticmethod
    def _desktop_send_recoverable_error(error_text: str) -> bool:
        lowered = error_text.lower()
        markers = [
            "timeout",
            "timed out",
            "request timeout",
            "socket",
            "econnrefused",
            "connection refused",
            "connection reset",
            "onSocketClose".lower(),
            "app-server",
            "app server",
            "initialize",
            "bad request",
            "502",
            "504",
        ]
        return any(marker in lowered for marker in markers)

    def _next_desktop_send_job_locked(self) -> dict | None:
        self._expire_stale_desktop_send_jobs_locked()
        now = time.time()
        for job in self._desktop_send_jobs():
            status = str(job.get("status") or "")
            if status not in {"queued", "retrying"}:
                continue
            if self._parse_iso_timestamp(job.get("next_retry_at")) > now:
                continue
            return job
        return None

    def _update_desktop_send_job_locked(self, job: dict, **updates: object) -> None:
        job.update({key: value for key, value in updates.items() if value is not None})
        job["updated_at"] = self._now_iso()
        self._save_desktop_send_jobs_locked()
        self._publish_desktop_send_job_status_locked(job)

    def _expire_stale_desktop_send_jobs_locked(self) -> list[dict]:
        now = time.time()
        expired: list[dict] = []
        for job in self._desktop_send_jobs():
            if str(job.get("status") or "") != "sending":
                continue
            started = self._parse_iso_timestamp(job.get("started_at"))
            timeout_seconds = float(job.get("timeout_seconds") or 60.0)
            if started <= 0 or now - started <= timeout_seconds + 10.0:
                continue
            elapsed = int(now - started)
            error_text = f"Codex app-server 發送逾時（{elapsed} 秒），已停止等待；請重新發送。"
            job.update(
                {
                    "status": "failed",
                    "error": error_text,
                    "recoverable": False,
                    "failed_at": self._now_iso(),
                    "updated_at": self._now_iso(),
                }
            )
            expired.append(self._public_desktop_send_job(job))
            self._publish_desktop_send_job_status_locked(job)
        if expired:
            logger.warning("Expired stale desktop send jobs count=%s", len(expired))
            self._save_desktop_send_jobs_locked()
        return expired

    def _desktop_send_worker_loop(self) -> None:
        idle_ticks = 0
        while True:
            with self._desktop_send_queue_lock():
                job = self._next_desktop_send_job_locked()
                if job is not None:
                    attempts = int(job.get("attempts") or 0) + 1
                    self._update_desktop_send_job_locked(
                        job,
                        status="sending",
                        attempts=attempts,
                        started_at=self._now_iso(),
                        next_retry_at="",
                        error="",
                    )
                else:
                    self._expire_stale_desktop_send_jobs_locked()
                    active_left = any(
                        str(item.get("status") or "") in {"queued", "retrying", "sending"}
                        for item in self._desktop_send_jobs()
                    )
                    idle_ticks += 1
                    if not active_left and idle_ticks >= 5:
                        return
            if job is None:
                time.sleep(1.0)
                continue
            idle_ticks = 0
            self._process_desktop_send_job(job)

    def _process_desktop_send_job(self, job: dict) -> None:
        request_id = str(job["request_id"])
        attempts = int(job.get("attempts") or 1)
        max_attempts = int(job.get("max_attempts") or 1)
        try:
            send_done, send_value, send_error = self._run_daemon_call_with_timeout(
                lambda: self.live_bridge.send_message(
                    str(job.get("text") or ""),
                    thread_id=str(job.get("thread_id") or "").strip() or None,
                    model=str(job.get("model") or "").strip() or None,
                    effort=str(job.get("effort") or "").strip() or None,
                    permission=str(job.get("permission") or "").strip() or None,
                    image_data_urls=job.get("image_data_urls") if isinstance(job.get("image_data_urls"), list) else None,
                    local_image_paths=job.get("local_image_paths") if isinstance(job.get("local_image_paths"), list) else None,
                    wait_for_reply=False,
                ),
                float(job.get("timeout_seconds") or 60.0),
            )
            if not send_done:
                raise TimeoutError("Codex app-server 發送逾時。")
            if send_error:
                raise send_error
            payload = send_value if isinstance(send_value, dict) else {}
            with self._desktop_send_queue_lock():
                if str(job.get("status") or "") == "cancelled":
                    logger.info("Desktop message send result ignored after cancellation request_id=%s", request_id)
                    return
                self._update_desktop_send_job_locked(
                    job,
                    status="sent",
                    thread_id=payload.get("thread_id") or job.get("thread_id"),
                    message_id=payload.get("message_id"),
                    sent_at=self._now_iso(),
                    recoverable=False,
                )
            logger.info("Desktop message sent request_id=%s attempts=%s", request_id, attempts)
            return
        except Exception as exc:
            error_text = self._desktop_send_error_text(exc)
            recoverable = self._desktop_send_recoverable_error(error_text)
            logger.warning(
                "Desktop message send attempt failed request_id=%s attempts=%s recoverable=%s error=%s",
                request_id,
                attempts,
                recoverable,
                error_text,
            )
            with self._desktop_send_queue_lock():
                if attempts < max_attempts and recoverable:
                    delay = self._desktop_send_retry_delay(attempts)
                    next_retry = dt.datetime.now(dt.timezone.utc) + dt.timedelta(seconds=delay)
                    self._update_desktop_send_job_locked(
                        job,
                        status="retrying",
                        error=error_text,
                        recoverable=True,
                        next_retry_at=next_retry.isoformat().replace("+00:00", "Z"),
                    )
                else:
                    self._update_desktop_send_job_locked(
                        job,
                        status="failed",
                        error=error_text or "Codex app-server 發送失敗。",
                        recoverable=recoverable,
                        failed_at=self._now_iso(),
                    )

    def _cancel_desktop_send_jobs(self, thread_id: str | None = None, request_id: str | None = None) -> list[dict]:
        cancelled: list[dict] = []
        with self._desktop_send_queue_lock():
            for job in self._desktop_send_jobs():
                if str(job.get("status") or "") not in {"queued", "retrying", "sending"}:
                    continue
                if request_id and str(job.get("request_id") or "") != request_id:
                    continue
                if thread_id and str(job.get("thread_id") or "") != thread_id:
                    continue
                self._update_desktop_send_job_locked(
                    job,
                    status="cancelled",
                    cancelled_at=self._now_iso(),
                    recoverable=False,
                    error="已由手機端停止。",
                )
                cancelled.append(self._public_desktop_send_job(job))
        return cancelled

    def do_GET(self) -> None:  # noqa: N802
        if not self._authorize():
            return
        parsed = urlparse(self.path)
        query = parse_qs(parsed.query)
        if parsed.path == "/health":
            self._send_json({"ok": True})
            return
        if parsed.path == "/state":
            self._send_json({"state": self.worker.get_state().model_dump(mode="json")})
            return
        if parsed.path == "/task_state":
            task_ref = query.get("task_ref", [""])[0].strip()
            if not task_ref:
                self._send_json({"error": "Missing task_ref"}, status=HTTPStatus.BAD_REQUEST)
                return
            try:
                self._send_json({"state": self.worker.get_task_state(task_ref).model_dump(mode="json")})
            except FileNotFoundError as exc:
                self._send_json({"error": str(exc)}, status=HTTPStatus.NOT_FOUND)
            return
        if parsed.path == "/tasks":
            self._send_json({"text": self.worker.list_tasks()})
            return
        if parsed.path == "/task_summaries":
            self._send_json(
                {"tasks": [task.model_dump(mode="json") for task in self.worker.list_task_summaries()]}
            )
            return
        if parsed.path == "/notes":
            limit = int(query.get("limit", ["5"])[0])
            self._send_json({"text": self.worker.format_recent_notes(limit=limit)})
            return
        if parsed.path == "/queue":
            limit = int(query.get("limit", ["10"])[0])
            self._send_json({"text": self.worker.format_queue(limit=limit)})
            return
        if parsed.path == "/progress":
            self._send_json({"text": self.worker.progress_summary()})
            return
        if parsed.path == "/history":
            limit = int(query.get("limit", ["5"])[0])
            self._send_json({"text": self.worker.format_command_history(limit=limit)})
            return
        if parsed.path == "/pending_count":
            self._send_json({"count": self.worker.pending_count()})
            return
        if parsed.path == "/desktop/status":
            try:
                self._ensure_desktop_send_worker()
                self._send_json(self._attach_latest_desktop_send_status(self.live_bridge.get_status_payload()))
            except Exception as exc:
                self._send_json({"error": str(exc)}, status=HTTPStatus.BAD_GATEWAY)
            return
        if parsed.path == "/desktop/send_queue":
            self._ensure_desktop_send_worker()
            self._send_json(self._desktop_send_queue_payload())
            return
        if parsed.path == "/desktop/usage":
            try:
                self._send_json(self.live_bridge.get_usage_payload())
            except Exception as exc:
                self._send_json({"error": str(exc)}, status=HTTPStatus.BAD_GATEWAY)
            return
        if parsed.path == "/desktop/runtime":
            thread_id = query.get("thread_id", [""])[0].strip() or None
            try:
                self._send_json(self.live_bridge.get_runtime_payload(thread_id=thread_id))
            except Exception as exc:
                self._send_json({"error": str(exc)}, status=HTTPStatus.BAD_GATEWAY)
            return
        if parsed.path in {"/desktop/runtime_ready", "/desktop/runtime-ready"}:
            thread_id = query.get("thread_id", [""])[0].strip() or None
            try:
                self._send_json(self.live_bridge.get_runtime_ready_payload(thread_id=thread_id))
            except Exception as exc:
                self._send_json({"error": str(exc)}, status=HTTPStatus.BAD_GATEWAY)
            return
        if parsed.path == "/desktop/snapshot":
            try:
                self._send_json(self._attach_latest_desktop_send_status(self.live_bridge.get_snapshot_payload(limit=_parse_snapshot_limit(query))))
            except Exception as exc:
                logger.exception("desktop snapshot failed")
                self._send_json({"error": str(exc)}, status=HTTPStatus.BAD_GATEWAY)
            return
        if parsed.path == "/desktop/thread_snapshot":
            thread_id = query.get("thread_id", [""])[0].strip()
            if not thread_id:
                self._send_json({"error": "Missing thread_id"}, status=HTTPStatus.BAD_REQUEST)
                return
            try:
                self._send_json(self._attach_latest_desktop_send_status(self.live_bridge.get_thread_snapshot_payload(thread_id, limit=_parse_snapshot_limit(query))))
            except Exception as exc:
                self._send_json({"error": str(exc)}, status=HTTPStatus.BAD_GATEWAY)
            return
        if parsed.path == "/desktop/build_ios_adhoc_status":
            request_id = query.get("request_id", [""])[0].strip()
            if not request_id:
                self._send_json({"error": "Missing request_id"}, status=HTTPStatus.BAD_REQUEST)
                return
            status = self._get_adhoc_build_status(request_id)
            if not status:
                self._send_json({"error": f"Build job not found: {request_id}"}, status=HTTPStatus.NOT_FOUND)
                return
            self._send_json(status)
            return
        if parsed.path == "/desktop/file":
            path = query.get("path", [""])[0].strip()
            if not path:
                self._send_json({"error": "Missing path"}, status=HTTPStatus.BAD_REQUEST)
                return
            try:
                self._send_json(self.live_bridge.read_file(path, encoding=query.get("encoding", [""])[0].strip() or None))
            except FileNotFoundError:
                self._send_json({"error": f"File not found: {path}"}, status=HTTPStatus.NOT_FOUND)
            except PermissionError as exc:
                self._send_json({"error": str(exc)}, status=HTTPStatus.FORBIDDEN)
            except Exception as exc:
                self._send_json({"error": str(exc)}, status=HTTPStatus.BAD_GATEWAY)
            return
        if parsed.path == "/desktop/messages":
            try:
                self._send_json(self.live_bridge.get_messages_payload(limit=_parse_snapshot_limit(query)))
            except Exception as exc:
                self._send_json({"error": str(exc)}, status=HTTPStatus.BAD_GATEWAY)
            return
        if parsed.path == "/desktop/threads":
            try:
                self._send_json({"threads": self.live_bridge.list_threads()})
            except Exception as exc:
                self._send_json({"error": str(exc)}, status=HTTPStatus.BAD_GATEWAY)
            return
        if parsed.path == "/desktop/projects":
            try:
                force = (query.get("refresh") or query.get("force") or [""])[0] in {"1", "true", "yes"}
                self._send_json({"projects": self.live_bridge.list_projects(force=force)})
            except Exception as exc:
                self._send_json({"error": str(exc)}, status=HTTPStatus.BAD_GATEWAY)
            return
        self._send_json({"error": "Not found"}, status=HTTPStatus.NOT_FOUND)

    def do_POST(self) -> None:  # noqa: N802
        if not self._authorize():
            return
        data = self._read_json()
        path = urlparse(self.path).path
        try:
            if path == "/task/create":
                state = self.worker.create_task(str(data["title"]))
            elif path == "/task/switch":
                state = self.worker.switch_task(str(data["task_id"]))
            elif path == "/task/continue":
                state = self.worker.continue_task(data.get("instruction"))
            elif path == "/task/pause":
                state = self.worker.pause_task()
            elif path == "/task/note":
                state = self.worker.add_note(str(data["text"]), author=str(data.get("author", "telegram")))
            elif path == "/task/chat_entry":
                state = self.worker.append_chat_entry(str(data.get("author", "assistant")), str(data["text"]))
            elif path == "/task/run":
                state = self.worker.run_command(str(data["command"]))
            elif path == "/task/set_workdir":
                state = self.worker.set_workdir(str(data["workdir"]))
            elif path == "/task/set_title":
                state = self.worker.set_title(str(data["title"]))
            elif path == "/task/set_agent":
                state = self.worker.set_agent_enabled(bool(data["enabled"]))
            elif path == "/task/reject":
                state = self.worker.reject_pending_command()
            elif path == "/task/confirm_voice":
                state = self.worker.confirm_voice_command()
            elif path == "/task/cancel_voice":
                state = self.worker.cancel_voice_command()
            elif path == "/task/reset":
                state = self.worker.reset_state()
            elif path == "/task/approve":
                state = self.worker.approve()
            elif path in {"/desktop/inbox", "/desktop/message"}:
                text = str(data.get("text") or data.get("message") or "").strip()
                thread_id = str(data.get("thread_id", "")).strip() or None
                image_data_urls = [
                    str(item).strip()
                    for item in data.get("image_data_urls", [])
                    if str(item).strip()
                ] if isinstance(data.get("image_data_urls"), list) else None
                local_image_paths = [
                    str(item).strip()
                    for item in data.get("local_image_paths", [])
                    if str(item).strip()
                ] if isinstance(data.get("local_image_paths"), list) else None
                image_count = len(image_data_urls or []) + len(local_image_paths or [])
                if not text and image_count:
                    text = "請直接閱讀並分析這張圖片。" if image_count == 1 else "請直接閱讀並分析這些圖片。"
                async_mode = bool(data.get("async"))
                if not async_mode:
                    ready_payload = self.live_bridge.get_runtime_ready_payload(thread_id=thread_id)
                    if not ready_payload.get("ready"):
                        self._send_json(
                            {
                                "ok": False,
                                "accepted": False,
                                "status": ready_payload.get("status") or "not_ready",
                                "thread_id": ready_payload.get("thread_id") or thread_id,
                                "reason": ready_payload.get("reason") or "Codex app-server 尚未就緒。",
                                "error": ready_payload.get("reason") or "Codex app-server 尚未就緒。",
                                "recoverable": True,
                            },
                            status=HTTPStatus.CONFLICT,
                        )
                        return
                else:
                    ready_done = True
                    ready_error = None
                    ready_payload = {}
                    # Async mobile sends must not start with a readiness probe:
                    # when Codex app-server is slow, the probe itself can hang or
                    # trigger recovery. Queue once and let the bounded send path
                    # report sent/failed instead of leaving the phone spinning.

                def send_message() -> dict:
                    return self.live_bridge.send_message(
                        text,
                        thread_id=thread_id,
                        model=str(data.get("model", "")).strip() or None,
                        effort=str(data.get("effort", "")).strip() or None,
                        permission=str(data.get("permission", "")).strip() or None,
                        image_data_urls=image_data_urls,
                        local_image_paths=local_image_paths,
                    )

                if async_mode:
                    request_id = str(uuid.uuid4())
                    now = self._now_iso()
                    job = {
                        "request_id": request_id,
                        "status": "queued",
                        "thread_id": thread_id,
                        "text": text,
                        "model": str(data.get("model", "")).strip(),
                        "effort": str(data.get("effort", "")).strip(),
                        "permission": str(data.get("permission", "")).strip(),
                        "image_data_urls": image_data_urls or [],
                        "local_image_paths": local_image_paths or [],
                        "attempts": 0,
                        "max_attempts": int(data.get("max_attempts") or 1),
                        "timeout_seconds": float(data.get("timeout_seconds") or 180.0),
                        "created_at": now,
                        "updated_at": now,
                    }
                    self._enqueue_desktop_send_job(job)
                    immediate_status = self._wait_for_desktop_send_job(request_id, timeout_seconds=8.0)
                    if immediate_status:
                        response_payload = {
                            "ok": immediate_status.get("status") == "sent",
                            "accepted": immediate_status.get("status") == "sent",
                            "status": immediate_status.get("status"),
                            "request_id": request_id,
                            "thread_id": immediate_status.get("thread_id") or thread_id,
                            "message_id": immediate_status.get("message_id"),
                            "desktop_send_status": immediate_status,
                            "error": immediate_status.get("error") if immediate_status.get("status") != "sent" else None,
                            "reply": "已送達桌面 Codex。" if immediate_status.get("status") == "sent" else (immediate_status.get("error") or "桌面端發送失敗。"),
                        }
                        self._send_json(response_payload)
                        return
                    self._send_json(
                        {
                            "ok": True,
                            "accepted": True,
                            "status": "queued",
                            "request_id": request_id,
                            "thread_id": thread_id,
                            "desktop_send_status": self._public_desktop_send_job(job),
                            "reply": "已送出，正在等待桌面端處理。",
                        }
                    )
                    return
                self._send_json(send_message())
                return
            elif path == "/desktop/reply":
                self._send_json(
                    self.live_bridge.send_message(
                        str(data["text"]),
                        thread_id=str(data.get("thread_id", "")).strip() or None,
                        model=str(data.get("model", "")).strip() or None,
                        effort=str(data.get("effort", "")).strip() or None,
                        permission=str(data.get("permission", "")).strip() or None,
                        image_data_urls=[
                            str(item).strip()
                            for item in data.get("image_data_urls", [])
                            if str(item).strip()
                        ] if isinstance(data.get("image_data_urls"), list) else None,
                        local_image_paths=[
                            str(item).strip()
                            for item in data.get("local_image_paths", [])
                            if str(item).strip()
                        ] if isinstance(data.get("local_image_paths"), list) else None,
                    )
                )
                return
            elif path == "/desktop/mark_done":
                self._send_json(self.live_bridge.mark_done(str(data["message_id"])))
                return
            elif path == "/desktop/select_thread":
                self._send_json(self.live_bridge.select_thread(str(data["thread_id"])))
                return
            elif path == "/desktop/open_thread":
                self._send_json(self.live_bridge.open_thread(str(data["thread_id"])))
                return
            elif path == "/desktop/refresh_thread":
                thread_id = str(data.get("thread_id", "")).strip() or None
                force = bool(data.get("force", True))
                self._send_json(self.live_bridge.refresh_desktop_thread(thread_id=thread_id, force=force))
                return
            elif path == "/desktop/new_thread":
                self._send_json(
                    self.live_bridge.create_thread(
                        str(data.get("cwd", "")).strip() or None,
                        model=str(data.get("model", "")).strip() or None,
                        effort=str(data.get("effort", "")).strip() or None,
                        permission=str(data.get("permission", "")).strip() or None,
                    )
                )
                return
            elif path == "/desktop/rename_thread":
                thread_id = str(data.get("thread_id", "")).strip()
                title = str(data.get("title", data.get("name", ""))).strip()
                if not thread_id:
                    self._send_json({"error": "Missing thread_id"}, status=HTTPStatus.BAD_REQUEST)
                    return
                if not title:
                    self._send_json({"error": "Missing title"}, status=HTTPStatus.BAD_REQUEST)
                    return
                self._send_json(self.live_bridge.rename_thread(thread_id, title))
                return
            elif path == "/desktop/archive_thread":
                thread_id = str(data.get("thread_id", "")).strip()
                if not thread_id:
                    self._send_json({"error": "Missing thread_id"}, status=HTTPStatus.BAD_REQUEST)
                    return
                self._send_json(self.live_bridge.archive_thread(thread_id))
                return
            elif path == "/desktop/unarchive_thread":
                thread_id = str(data.get("thread_id", "")).strip()
                if not thread_id:
                    self._send_json({"error": "Missing thread_id"}, status=HTTPStatus.BAD_REQUEST)
                    return
                self._send_json(self.live_bridge.unarchive_thread(thread_id))
                return
            elif path == "/desktop/pin_thread":
                thread_id = str(data.get("thread_id", "")).strip()
                if not thread_id:
                    self._send_json({"error": "Missing thread_id"}, status=HTTPStatus.BAD_REQUEST)
                    return
                self._send_json(self.live_bridge.pin_thread(thread_id))
                return
            elif path == "/desktop/unpin_thread":
                thread_id = str(data.get("thread_id", "")).strip()
                if not thread_id:
                    self._send_json({"error": "Missing thread_id"}, status=HTTPStatus.BAD_REQUEST)
                    return
                self._send_json(self.live_bridge.unpin_thread(thread_id))
                return
            elif path == "/desktop/fork_thread":
                thread_id = str(data.get("thread_id", "")).strip()
                if not thread_id:
                    self._send_json({"error": "Missing thread_id"}, status=HTTPStatus.BAD_REQUEST)
                    return
                self._send_json(self.live_bridge.fork_thread(thread_id, str(data.get("cwd", "")).strip() or None))
                return
            elif path == "/desktop/new_project":
                cwd = str(data.get("cwd", "")).strip()
                if not cwd:
                    self._send_json({"error": "Missing cwd"}, status=HTTPStatus.BAD_REQUEST)
                    return
                self._send_json(self.live_bridge.create_project(cwd))
                return
            elif path == "/desktop/rename_project":
                cwd = str(data.get("cwd", "")).strip()
                title = str(data.get("title", data.get("name", ""))).strip()
                if not cwd:
                    self._send_json({"error": "Missing cwd"}, status=HTTPStatus.BAD_REQUEST)
                    return
                if not title:
                    self._send_json({"error": "Missing title"}, status=HTTPStatus.BAD_REQUEST)
                    return
                self._send_json(self.live_bridge.rename_project(cwd, title))
                return
            elif path == "/desktop/remove_project":
                cwd = str(data.get("cwd", "")).strip()
                if not cwd:
                    self._send_json({"error": "Missing cwd"}, status=HTTPStatus.BAD_REQUEST)
                    return
                self._send_json(self.live_bridge.remove_project(cwd))
                return
            elif path == "/desktop/interrupt":
                thread_id = str(data.get("thread_id", "")).strip() or None
                request_id = str(data.get("request_id", "")).strip() or None
                cancelled_jobs = self._cancel_desktop_send_jobs(thread_id=thread_id, request_id=request_id)
                if bool(data.get("async")):
                    def interrupt_in_background() -> None:
                        try:
                            self.live_bridge.interrupt_current_turn(thread_id)
                        except Exception:
                            logger.exception("Async desktop interrupt failed")

                    threading.Thread(target=interrupt_in_background, daemon=True).start()
                    self._send_json(
                        {
                            "ok": True,
                            "accepted": True,
                            "thread_id": thread_id,
                            "cancelled_jobs": cancelled_jobs,
                            "cancelled_count": len(cancelled_jobs),
                            "reply": "已送出停止請求。",
                        }
                    )
                    return
                payload = self.live_bridge.interrupt_current_turn(thread_id)
                if isinstance(payload, dict):
                    payload["cancelled_jobs"] = cancelled_jobs
                    payload["cancelled_count"] = len(cancelled_jobs)
                self._send_json(payload)
                return
            elif path == "/desktop/send_cancel":
                thread_id = str(data.get("thread_id", "")).strip() or None
                request_id = str(data.get("request_id", "")).strip() or None
                cancelled_jobs = self._cancel_desktop_send_jobs(thread_id=thread_id, request_id=request_id)
                self._send_json(
                    {
                        "ok": True,
                        "accepted": True,
                        "thread_id": thread_id,
                        "request_id": request_id,
                        "cancelled_jobs": cancelled_jobs,
                        "cancelled_count": len(cancelled_jobs),
                    }
                )
                return
            elif path == "/desktop/rollback":
                thread_id = str(data.get("thread_id", "")).strip() or None
                num_turns = int(data.get("num_turns") or 1)
                if bool(data.get("async")):
                    def rollback_in_background() -> None:
                        try:
                            self.live_bridge.rollback_turns(thread_id, num_turns)
                        except Exception:
                            logger.exception("Async rollback failed")

                    threading.Thread(target=rollback_in_background, daemon=True).start()
                    self._send_json(
                        {
                            "ok": True,
                            "accepted": True,
                            "thread_id": thread_id,
                            "num_turns": num_turns,
                            "reply": "已送出回退請求。",
                        }
                    )
                    return
                self._send_json(
                    self.live_bridge.rollback_turns(
                        thread_id,
                        num_turns,
                    )
                )
                return
            elif path == "/desktop/inject_items":
                items = data.get("items")
                if not isinstance(items, list):
                    self._send_json({"error": "Missing items"}, status=HTTPStatus.BAD_REQUEST)
                    return
                self._send_json(
                    self.live_bridge.inject_items(
                        items,
                        thread_id=str(data.get("thread_id", "")).strip() or None,
                    )
                )
                return
            elif path == "/desktop/computer_control":
                self._send_json(
                    self.live_bridge.request_computer_control(
                        instruction=str(data.get("instruction", "")).strip() or None,
                        thread_id=str(data.get("thread_id", "")).strip() or None,
                    )
                )
                return
            elif path == "/desktop/screenshot":
                self._send_json(_capture_and_upload_desktop_screenshot())
                return
            elif path == "/desktop/build_ios_adhoc":
                if bool(data.get("async")):
                    request_id = str(data.get("request_id") or uuid.uuid4())
                    self._send_json(
                        self._start_adhoc_build_job(
                            {
                                "request_id": request_id,
                                "thread_id": str(data.get("thread_id", "")).strip() or None,
                                "project_id": str(data.get("project_id", "")).strip() or None,
                                "cwd": str(data.get("cwd", "")).strip() or None,
                                "scheme": str(data.get("scheme", "")).strip() or None,
                            }
                        )
                    )
                    return
                self._send_json(
                    _run_ios_adhoc_package_build(
                        self.live_bridge,
                        thread_id=str(data.get("thread_id", "")).strip() or None,
                        project_id=str(data.get("project_id", "")).strip() or None,
                        cwd=str(data.get("cwd", "")).strip() or None,
                        scheme=str(data.get("scheme", "")).strip() or None,
                    )
                )
                return
            elif path == "/desktop/restart_codex":
                thread_id = str(data.get("thread_id", "")).strip() or None
                self._send_json(_restart_codex_desktop(self.live_bridge, thread_id=thread_id))
                return
            elif path == "/desktop/restart_desktop_connection":
                self._send_json(_restart_desktop_connection())
                return
            elif path == "/desktop/restart_mac":
                self._send_json(_restart_mac(), status=HTTPStatus.GONE)
                return
            elif path == "/desktop/upload_attachment":
                encoded_attachment = str(data.get("file_base64", "")).strip()
                filename = str(data.get("filename", "attachment.bin")).strip() or "attachment.bin"
                mime_type = str(data.get("mime_type", "application/octet-stream")).strip() or "application/octet-stream"
                thread_id = str(data.get("thread_id", "")).strip() or None
                if not encoded_attachment:
                    self._send_json({"error": "Missing file_base64"}, status=HTTPStatus.BAD_REQUEST)
                    return
                try:
                    file_bytes = base64.b64decode(encoded_attachment, validate=True)
                except Exception:
                    self._send_json({"error": "Invalid base64 attachment payload"}, status=HTTPStatus.BAD_REQUEST)
                    return
                uploads_dir = self.settings.codexbot_state_path.parent / "uploads"
                uploads_dir.mkdir(parents=True, exist_ok=True)
                safe_name = filename.replace("/", "_").replace("\\", "_").strip() or "attachment.bin"
                target = uploads_dir / safe_name
                if target.exists():
                    target = uploads_dir / f"{target.stem}-{len(file_bytes)}{target.suffix}"
                target.write_bytes(file_bytes)
                self._send_json(
                    {
                        "saved_path": str(target),
                        "attachment_path": str(target),
                        "filename": target.name,
                        "mime_type": mime_type,
                        "thread_id": thread_id,
                    }
                )
                return
            elif path == "/desktop/attachment":
                encoded_attachment = str(data.get("file_base64", "")).strip()
                filename = str(data.get("filename", "attachment.bin")).strip() or "attachment.bin"
                mime_type = str(data.get("mime_type", "application/octet-stream")).strip() or "application/octet-stream"
                thread_id = str(data.get("thread_id", "")).strip() or None
                if not encoded_attachment:
                    self._send_json({"error": "Missing file_base64"}, status=HTTPStatus.BAD_REQUEST)
                    return
                try:
                    file_bytes = base64.b64decode(encoded_attachment, validate=True)
                except Exception:
                    self._send_json({"error": "Invalid base64 attachment payload"}, status=HTTPStatus.BAD_REQUEST)
                    return
                uploads_dir = self.settings.codexbot_state_path.parent / "uploads"
                uploads_dir.mkdir(parents=True, exist_ok=True)
                safe_name = filename.replace("/", "_").replace("\\", "_").strip() or "attachment.bin"
                target = uploads_dir / safe_name
                if target.exists():
                    target = uploads_dir / f"{target.stem}-{len(file_bytes)}{target.suffix}"
                target.write_bytes(file_bytes)
                self._send_json(
                    self.live_bridge.analyze_attachment(
                        local_path=str(target),
                        filename=target.name,
                        mime_type=mime_type,
                        thread_id=thread_id,
                    )
                )
                return
            elif path == "/desktop/open_project":
                project_id = str(data.get("project_id", "")).strip()
                if not project_id:
                    self._send_json({"error": "Missing project_id"}, status=HTTPStatus.BAD_REQUEST)
                    return
                self._send_json(self.live_bridge.open_project(project_id))
                return
            else:
                self._send_json({"error": "Not found"}, status=HTTPStatus.NOT_FOUND)
                return
        except FileNotFoundError as exc:
            self._send_json({"error": str(exc)}, status=HTTPStatus.NOT_FOUND)
            return
        except CodexLiveBridgeError as exc:
            self._send_json({"error": str(exc)}, status=HTTPStatus.BAD_REQUEST)
            return
        except KeyError as exc:
            self._send_json({"error": f"Missing field: {exc}"}, status=HTTPStatus.BAD_REQUEST)
            return
        except Exception as exc:
            logger.exception("Mac service request failed path=%s", path)
            self._send_json({"error": str(exc)}, status=HTTPStatus.BAD_GATEWAY)
            return
        self._send_json({"state": state.model_dump(mode="json")})

    def _authorize(self) -> bool:
        expected = f"Bearer {self.token}"
        if self.headers.get("Authorization", "") != expected:
            self._send_json({"error": "Unauthorized"}, status=HTTPStatus.UNAUTHORIZED)
            return False
        return True

    def _read_json(self) -> dict:
        length = int(self.headers.get("Content-Length", "0") or "0")
        if length <= 0:
            return {}
        raw = self.rfile.read(length)
        return json.loads(raw.decode("utf-8"))

    def _send_json(self, payload: dict, status: HTTPStatus = HTTPStatus.OK) -> None:
        encoded = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Cache-Control", "no-store, no-cache, must-revalidate, max-age=0")
        self.send_header("Pragma", "no-cache")
        self.send_header("Expires", "0")
        self.send_header("Content-Length", str(len(encoded)))
        self.end_headers()
        try:
            self.wfile.write(encoded)
        except (BrokenPipeError, ConnectionResetError, socket.timeout):
            logger.debug("Client disconnected before mac service response could be written")

    def log_message(self, format: str, *args) -> None:  # noqa: A003
        logger.info("mac_service " + format, *args)


def main() -> None:
    settings = Settings()
    logging.basicConfig(
        level=getattr(logging, settings.codexbot_log_level.upper(), logging.INFO),
        format="%(asctime)s %(levelname)s %(name)s %(message)s",
    )
    worker = build_local_worker(settings)
    token = settings.codexbot_mac_service_token
    if not token:
        raise RuntimeError("CODEXMOBILE_MAC_SERVICE_TOKEN is required for CodexMobile Mac service")

    handler = type(
        "BoundMacServiceHandler",
        (MacServiceHandler,),
        {"worker": worker, "token": token, "settings": settings, "live_bridge": CodexLiveBridge(settings)},
    )
    server = ThreadingHTTPServer(
        (settings.codexbot_mac_service_host, settings.codexbot_mac_service_port),
        handler,
    )
    logger.info(
        "CodexMobile Mac service listening on %s:%s",
        settings.codexbot_mac_service_host,
        settings.codexbot_mac_service_port,
    )
    try:
        server.serve_forever()
    finally:
        worker.shutdown()
        server.server_close()


if __name__ == "__main__":
    main()
