#!/usr/bin/env python3
"""Sand Stream Toolkit。

交互式运行：
    python sand_stream_installer.py

命令行运行：
    python sand_stream_installer.py install
    python sand_stream_installer.py uninstall
    python sand_stream_installer.py set-path <Cursor路径|auto>
"""

from __future__ import annotations

import argparse
import base64
import ctypes
import hashlib
import json
import os
import re
import shutil
import signal
import stat
import subprocess
import sys
import threading
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, Iterable, List, Mapping, Optional, Sequence, Set, Tuple, Union

import sand_license


TOOL_NAME = "Sand Stream Toolkit"
TOOL_VERSION = "2.1.0"
SUPPORTED_CURSOR_VERSION = "3.18.9"
CONFIG_VERSION = 1

SAND_CLIENT_MARKER = "/*SAND_CLIENT_MODE_V1*/"
SAND_CLIENT_EXISTING_MARKER = "/*SAND_CLIENT_EXISTING_V1*/"
SAND_ELIGIBILITY_MARKER = "/*SAND_ELIGIBILITY_MODE_V1*/"
SAND_MANAGED_LOCAL_ROUTE_MARKER = "/*SAND_MANAGED_LOCAL_ROUTE_V1*/"
SAND_DIRECT_STREAM_MARKER = "/*SAND_DIRECT_INFERENCE_STREAM_V1*/"
SAND_AGENT_HOST_ENABLEMENT_MARKER = "/*SAND_AGENT_HOST_ENABLEMENT_V1*/"
SAND_LOCAL_RUNTIME_LOAD_MARKER = "/*SAND_LOCAL_RUNTIME_LOAD_V1*/"
SAND_AGENT_HOST_IDENTITY_MARKER = "/*SAND_AGENT_HOST_IDENTITY_V1*/"
SAND_PLAN_MODE_MARKER = "/*SAND_PLAN_MODE_V1*/"
SAND_ACTION_GATE_MARKER = "/*SAND_ACTION_GATE_V1*/"
SAND_QUERY_KEY_MARKER = "/*SAND_QUERY_KEY_V1*/"
SAND_MOVE_EXEC_MARKER = "/*SAND_MOVE_EXEC_V1*/"
# 敏感补丁负载不再内置于程序，改为激活/校验通过后由授权服务器加密下发，
# 运行时由 load_payloads() 填充下列全局变量（见文件后部）。
MOVE_EXEC_ORIGINAL: Optional[str] = None
MOVE_EXEC_PATCHED: Optional[str] = None
LEGACY_SAND_CLIENT_MARKER = "/*K" + "C_SAND_CLIENT_V1*/"
LEGACY_SAND_ELIGIBILITY_MARKER = "/*K" + "C_SAND_ELIGIBILITY_V1*/"
CLIENT_MARKER_PATTERN = re.escape(SAND_CLIENT_MARKER)
CLIENT_EXISTING_MARKER_PATTERN = re.escape(SAND_CLIENT_EXISTING_MARKER)
ELIGIBILITY_MARKER_PATTERN = re.escape(SAND_ELIGIBILITY_MARKER)
LEGACY_CLIENT_MARKER_PATTERN = re.escape(LEGACY_SAND_CLIENT_MARKER)
LEGACY_ELIGIBILITY_MARKER_PATTERN = re.escape(LEGACY_SAND_ELIGIBILITY_MARKER)
CLIENT_MARKER_GUARD_PATTERN = r"/\*[A-Z0-9_]*SAND_CLIENT(?:_(?:MODE|EXISTING))?_V1\*/"
ELIGIBILITY_MARKER_GUARD_PATTERN = r"/\*[A-Z0-9_]*SAND_ELIGIBILITY(?:_MODE)?_V1\*/"
SAND_ONBOARDING_URL = "https://cursor.com/bot/onboarding?product=grok-bot"

ANSI_RESET = "\033[0m"
ANSI_BOLD = "\033[1m"
ANSI_RED = "\033[31m"
ANSI_GREEN = "\033[32m"
ANSI_YELLOW = "\033[33m"
ANSI_BLUE = "\033[36m"

_COLOR_ENABLED = True


TARGET_SPECS: Tuple[Tuple[str, Optional[str]], ...] = (
    ("out/main.js", None),
    ("out/vs/workbench/api/worker/extensionHostWorkerMain.js", None),
    ("out/vs/workbench/api/node/extensionHostProcess.js", None),
    ("out/vs/workbench/workbench.glass.main.js", None),
    ("out/vs/workbench/workbench.desktop.main.js", None),
    ("extensions/cursor-always-local/dist/main.js", "cursor-always-local"),
    (
        "extensions/cursor-local-agent-runtime/dist/main.js",
        "cursor-local-agent-runtime",
    ),
    ("extensions/cursor-agent-host/dist/main.js", "cursor-agent-host"),
    ("extensions/cursor-agent-exec/dist/main.js", "cursor-agent-exec"),
    ("extensions/cursor-agent-host/dist/657.js", None),
    ("extensions/cursor-agent-host/dist/675.js", None),
)

EXT_HOST_REL = "out/vs/workbench/api/node/extensionHostProcess.js"

ELIGIBILITY_PREFIXES: Tuple[str, ...] = ()  # 由 load_payloads() 填充


class SandToolError(RuntimeError):
    pass


@dataclass(frozen=True)
class CursorLayout:
    install_root: Path
    app_root: Path
    product_json: Path
    executable: Path
    target_paths: Tuple[Path, ...]
    ext_host_path: Optional[Path]
    version: str


@dataclass(frozen=True)
class PlannedFile:
    original: bytes
    next_bytes: bytes
    mode: int


@dataclass
class PatchStats:
    is_glass: int = 0
    object_header: int = 0
    set_header: int = 0
    eligibility: int = 0
    adopted_sand: int = 0
    migrated_client: int = 0
    migrated_eligibility: int = 0
    managed_local_route: int = 0
    local_runtime_load: int = 0
    direct_stream: int = 0
    agent_host_enablement: int = 0
    agent_host_identity: int = 0
    move_exec: int = 0
    plan_mode: int = 0
    action_gate: int = 0
    query_key: int = 0

    @property
    def total(self) -> int:
        return (
            self.is_glass
            + self.object_header
            + self.set_header
            + self.eligibility
            + self.migrated_client
            + self.migrated_eligibility
            + self.managed_local_route
            + self.local_runtime_load
            + self.direct_stream
            + self.agent_host_enablement
            + self.agent_host_identity
            + self.move_exec
            + self.plan_mode
            + self.action_gate
            + self.query_key
        )


@dataclass
class RemoveStats:
    client_type: int = 0
    eligibility: int = 0
    managed_local_route: int = 0
    local_runtime_load: int = 0
    direct_stream: int = 0
    agent_host_enablement: int = 0
    agent_host_identity: int = 0
    move_exec: int = 0
    plan_mode: int = 0
    action_gate: int = 0
    query_key: int = 0

    @property
    def total(self) -> int:
        return (
            self.client_type
            + self.eligibility
            + self.managed_local_route
            + self.local_runtime_load
            + self.direct_stream
            + self.agent_host_enablement
            + self.agent_host_identity
            + self.move_exec
            + self.plan_mode
            + self.action_gate
            + self.query_key
        )


@dataclass(frozen=True)
class PatchStatus:
    client_markers: int
    eligibility_markers: int
    ide_matches: int
    external_sand_matches: int
    external_marker_count: int
    legacy_client_markers: int
    legacy_eligibility_markers: int
    patched_files: Tuple[Path, ...]
    managed_local_route_markers: int
    local_runtime_load_markers: int
    direct_stream_markers: int
    agent_host_enablement_markers: int
    agent_host_identity_markers: int
    move_exec_markers: int
    plan_mode_markers: int = 0
    action_gate_markers: int = 0
    query_key_markers: int = 0

    @property
    def installed(self) -> bool:
        return (
            self.client_markers
            + self.eligibility_markers
            + self.legacy_client_markers
            + self.legacy_eligibility_markers
            + self.managed_local_route_markers
            + self.local_runtime_load_markers
            + self.direct_stream_markers
            + self.agent_host_enablement_markers
            + self.agent_host_identity_markers
            + self.plan_mode_markers
            + self.action_gate_markers
            + self.query_key_markers
            > 0
        )

    @property
    def stream_mode_installed(self) -> bool:
        return (
            self.managed_local_route_markers > 0
            and self.local_runtime_load_markers > 0
            and self.direct_stream_markers > 0
            and self.agent_host_enablement_markers > 0
            and self.agent_host_identity_markers > 0
        )


# ---------------------------------------------------------------------------
# 补丁负载：不内置于本程序，激活/校验通过后由授权服务器加密下发。
# load_payloads() 解密 JSON 后填充下列全局变量；未加载时执行 install 会报错。
# ---------------------------------------------------------------------------
CLIENT_RULES: Tuple[Tuple[str, re.Pattern[str]], ...] = ()
MANAGED_LOCAL_ROUTE_ORIGINAL: Optional[str] = None
MANAGED_LOCAL_ROUTE_PATCHED: Optional[str] = None
LOCAL_RUNTIME_LOAD_ORIGINAL: Optional[str] = None
LOCAL_RUNTIME_LOAD_PATCHED: Optional[str] = None
AGENT_HOST_IDENTITY_ORIGINAL: Optional[str] = None
AGENT_HOST_IDENTITY_PATCHED: Optional[str] = None
DIRECT_STREAM_ANCHOR: Optional[str] = None
AGENT_HOST_ENABLEMENT_RE: Optional[re.Pattern[str]] = None
AGENT_HOST_ENABLEMENT_PATCH_RE: Optional[re.Pattern[str]] = None
PLAN_MODE_GATE_RE: Optional[re.Pattern[str]] = None
PLAN_MODE_GATE_PATCH_RE: Optional[re.Pattern[str]] = None
ACTION_GATE_RE: Optional[re.Pattern[str]] = None
ACTION_GATE_PATCH_RE: Optional[re.Pattern[str]] = None
QUERY_KEY_ANCHOR: Optional[str] = None
QUERY_KEY_PATCHED: Optional[str] = None

_PAYLOADS_LOADED = False
_DIRECT_STREAM_INJECTION: Optional[str] = None
_ELIGIBILITY_ANCHOR: Optional[str] = None


def load_payloads(data: Dict[str, object]) -> None:
    """用授权服务器下发（并解密）的负载填充补丁规则。仅内存，不落盘。"""
    global CLIENT_RULES, ELIGIBILITY_PREFIXES
    global MANAGED_LOCAL_ROUTE_ORIGINAL, MANAGED_LOCAL_ROUTE_PATCHED
    global LOCAL_RUNTIME_LOAD_ORIGINAL, LOCAL_RUNTIME_LOAD_PATCHED
    global AGENT_HOST_IDENTITY_ORIGINAL, AGENT_HOST_IDENTITY_PATCHED
    global DIRECT_STREAM_ANCHOR, _DIRECT_STREAM_INJECTION
    global AGENT_HOST_ENABLEMENT_RE, AGENT_HOST_ENABLEMENT_PATCH_RE
    global PLAN_MODE_GATE_RE, PLAN_MODE_GATE_PATCH_RE
    global ACTION_GATE_RE, ACTION_GATE_PATCH_RE
    global QUERY_KEY_ANCHOR, QUERY_KEY_PATCHED
    global MOVE_EXEC_ORIGINAL, MOVE_EXEC_PATCHED, _PAYLOADS_LOADED, _ELIGIBILITY_ANCHOR

    ELIGIBILITY_PREFIXES = tuple(data["eligibility_prefixes"])
    _ELIGIBILITY_ANCHOR = data["eligibility_anchor"]
    CLIENT_RULES = tuple((key, re.compile(pat)) for key, pat in data["client_rules"])
    MANAGED_LOCAL_ROUTE_ORIGINAL = data["managed_local_route_original"]
    MANAGED_LOCAL_ROUTE_PATCHED = data["managed_local_route_patched"]
    LOCAL_RUNTIME_LOAD_ORIGINAL = data["local_runtime_load_original"]
    LOCAL_RUNTIME_LOAD_PATCHED = data["local_runtime_load_patched"]
    AGENT_HOST_IDENTITY_ORIGINAL = data["agent_host_identity_original"]
    AGENT_HOST_IDENTITY_PATCHED = data["agent_host_identity_patched"]
    DIRECT_STREAM_ANCHOR = data["direct_stream_anchor"]
    _DIRECT_STREAM_INJECTION = data["direct_stream_injection"]
    AGENT_HOST_ENABLEMENT_RE = re.compile(data["agent_host_enablement_re"])
    AGENT_HOST_ENABLEMENT_PATCH_RE = re.compile(data["agent_host_enablement_patch_re"])
    PLAN_MODE_GATE_RE = re.compile(data["plan_mode_gate_re"])
    PLAN_MODE_GATE_PATCH_RE = re.compile(data["plan_mode_gate_patch_re"])
    ACTION_GATE_RE = re.compile(data["action_gate_re"])
    ACTION_GATE_PATCH_RE = re.compile(data["action_gate_patch_re"])
    QUERY_KEY_ANCHOR = data["query_key_anchor"]
    QUERY_KEY_PATCHED = data["query_key_patched"]
    MOVE_EXEC_ORIGINAL = data["move_exec_original"]
    MOVE_EXEC_PATCHED = data["move_exec_patched"]
    _PAYLOADS_LOADED = True


def _ensure_payloads() -> None:
    if not _PAYLOADS_LOADED:
        raise SandToolError(
            "补丁数据缺失：请先联网激活/校验卡密（本工具的补丁规则由授权服务器下发）。"
        )


def _direct_stream_injection() -> str:
    if not _DIRECT_STREAM_INJECTION:
        raise SandToolError("补丁数据缺失，请先激活卡密")
    return _DIRECT_STREAM_INJECTION


def _platform_name() -> str:
    if sys.platform == "win32":
        return "windows"
    if sys.platform == "darwin":
        return "macos"
    if sys.platform.startswith("linux"):
        return "linux"
    raise SandToolError(f"当前不支持此操作系统：{sys.platform}")


def _enable_windows_ansi() -> bool:
    if sys.platform != "win32":
        return True
    try:
        kernel32 = ctypes.windll.kernel32
        for handle_id in (-11, -12):
            handle = kernel32.GetStdHandle(handle_id)
            if handle in (0, -1):
                continue
            mode = ctypes.c_uint32()
            if not kernel32.GetConsoleMode(handle, ctypes.byref(mode)):
                continue
            kernel32.SetConsoleMode(handle, mode.value | 0x0004)
        return True
    except Exception:
        return False


def _configure_console() -> None:
    global _COLOR_ENABLED
    for stream in (sys.stdout, sys.stderr):
        reconfigure = getattr(stream, "reconfigure", None)
        if callable(reconfigure):
            try:
                reconfigure(encoding="utf-8", errors="replace")
            except Exception:
                pass
    if os.environ.get("NO_COLOR"):
        _COLOR_ENABLED = False
        return
    _COLOR_ENABLED = _enable_windows_ansi() and sys.stdout.isatty()


def colorize(text: str, *codes: str) -> str:
    if not _COLOR_ENABLED or not codes:
        return text
    return "".join(codes) + text + ANSI_RESET


def print_warn(text: str) -> None:
    print(colorize(text, ANSI_YELLOW))


def print_error(text: str) -> None:
    print(colorize(text, ANSI_RED), file=sys.stderr)


def print_success(text: str) -> None:
    print(colorize(text, ANSI_GREEN, ANSI_BOLD))


class LoadingSpinner:
    def __init__(self, message: str = "处理中") -> None:
        self.message = message
        self._stop = threading.Event()
        self._thread: Optional[threading.Thread] = None

    def __enter__(self) -> "LoadingSpinner":
        if sys.stdout.isatty():
            self._thread = threading.Thread(target=self._run, daemon=True)
            self._thread.start()
        else:
            print(colorize(self.message + "...", ANSI_BLUE), flush=True)
        return self

    def __exit__(self, *_exc: object) -> None:
        self._stop.set()
        if self._thread is not None:
            self._thread.join()
            print("\r" + " " * 48 + "\r", end="", flush=True)

    def _run(self) -> None:
        frames = ("|", "/", "-", "\\")
        index = 0
        while not self._stop.wait(0.1):
            text = f"{frames[index % 4]} {self.message}"
            print("\r" + colorize(text, ANSI_BLUE), end="", flush=True)
            index += 1


def _config_dir() -> Path:
    if sys.platform == "win32":
        base = os.environ.get("LOCALAPPDATA") or os.environ.get("APPDATA")
        if base:
            return Path(base) / "SandClientModeStream" / "sand-client-cli"
        return (
            Path.home()
            / "AppData"
            / "Local"
            / "SandClientModeStream"
            / "sand-client-cli"
        )
    if sys.platform == "darwin":
        return (
            Path.home()
            / "Library"
            / "Application Support"
            / "SandClientModeStream"
            / "sand-client-cli"
        )
    return (
        Path.home()
        / ".config"
        / "SandClientModeStream"
        / "sand-client-cli"
    )


def _config_path() -> Path:
    return _config_dir() / "config.json"


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


def _path_key(path: Path) -> str:
    normalized = str(path.resolve())
    return os.path.normcase(normalized)


def _sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _product_checksum(data: bytes) -> str:
    digest = hashlib.sha256(data).digest()
    return base64.b64encode(digest).decode("ascii").rstrip("=")


def _atomic_write(path: Path, data: bytes, mode: Optional[int] = None) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temp = path.parent / (
        f".{path.name}.sand-client-{os.getpid()}-{time.time_ns()}.tmp"
    )
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
    fd: Optional[int] = None
    try:
        fd = os.open(str(temp), flags, 0o600)
        with os.fdopen(fd, "wb", closefd=True) as handle:
            fd = None
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
        if mode is not None:
            os.chmod(temp, stat.S_IMODE(mode))
        try:
            os.replace(temp, path)
        except PermissionError:
            original_mode: Optional[int] = None
            if path.exists():
                original_mode = stat.S_IMODE(path.stat().st_mode)
                os.chmod(path, original_mode | stat.S_IWRITE)
            try:
                os.replace(temp, path)
            except BaseException:
                if original_mode is not None and path.exists():
                    try:
                        os.chmod(path, original_mode)
                    except OSError:
                        pass
                raise
        if mode is not None:
            os.chmod(path, stat.S_IMODE(mode))
    finally:
        if fd is not None:
            os.close(fd)
        try:
            if temp.exists():
                temp.unlink()
        except OSError:
            pass


def _write_json_atomic(path: Path, value: Mapping[str, object]) -> None:
    data = (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
    _atomic_write(path, data, 0o600)


def _load_config() -> Mapping[str, object]:
    path = _config_path()
    if not path.exists():
        return {}
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except Exception as exc:
        raise SandToolError(
            f"配置文件损坏：{path}\n请运行 set-path auto 后重新检测"
        ) from exc
    if not isinstance(value, dict) or value.get("version") != CONFIG_VERSION:
        raise SandToolError(
            f"不支持的配置文件：{path}\n请运行 set-path auto 后重新检测"
        )
    return value


def _read_product(product_path: Path) -> Mapping[str, object]:
    try:
        size = product_path.stat().st_size
        if size <= 0 or size > 1024 * 1024:
            raise SandToolError(f"product.json 大小异常：{product_path}")
        raw = product_path.read_bytes()
        value = json.loads(raw.decode("utf-8-sig"))
    except SandToolError:
        raise
    except Exception as exc:
        raise SandToolError(f"无法读取 Cursor product.json：{product_path}") from exc
    if not isinstance(value, dict):
        raise SandToolError(f"Cursor product.json 格式错误：{product_path}")
    name = str(value.get("applicationName") or value.get("nameShort") or "")
    if name.casefold() != "cursor":
        raise SandToolError(f"所选目录不是 Cursor 安装：{product_path}")
    return value


def _find_app_bundle(app_root: Path) -> Optional[Path]:
    for item in (app_root, *app_root.parents):
        if item.name.casefold() == "cursor.app":
            return item
    return None


def _candidate_app_roots(raw_path: Path) -> Iterable[Path]:
    path = raw_path
    if path.is_file():
        if path.name.casefold() == "product.json":
            path = path.parent
        else:
            path = path.parent
    current = path
    for _ in range(8):
        yield current
        yield current / "resources" / "app"
        yield current / "Resources" / "app"
        yield current / "Contents" / "Resources" / "app"
        if current.parent == current:
            break
        current = current.parent


def _resolve_executable(app_root: Path) -> Tuple[Path, Path]:
    if sys.platform == "win32":
        if app_root.parent.name.casefold() == "resources":
            install_root = app_root.parent.parent
        else:
            install_root = app_root
        candidates = (
            install_root / "Cursor.exe",
            install_root / "cursor.exe",
        )
    elif sys.platform == "darwin":
        bundle = _find_app_bundle(app_root)
        if bundle is None:
            raise SandToolError("macOS Cursor 路径必须位于 Cursor.app 内")
        install_root = bundle
        candidates = (bundle / "Contents" / "MacOS" / "Cursor",)
    elif sys.platform.startswith("linux"):
        if app_root.parent.name.casefold() == "resources":
            install_root = app_root.parent.parent
        else:
            install_root = app_root
        candidates = (
            install_root / "cursor",
            install_root / "Cursor",
            install_root / "bin" / "cursor",
            install_root / "bin" / "cursor-server",
            app_root / "bin" / "cursor-server",
            install_root / "node",
        )
    else:
        raise SandToolError(f"当前不支持此操作系统：{sys.platform}")

    for executable in candidates:
        try:
            resolved = executable.resolve(strict=True)
        except (FileNotFoundError, OSError):
            continue
        if resolved.is_file() and _is_within(resolved, install_root.resolve()):
            return install_root.resolve(), resolved
    raise SandToolError(f"未找到 Cursor 可执行文件：{install_root}")


def layout_from_path(value: Union[str, Path]) -> CursorLayout:
    raw_text = str(value).strip().strip('"')
    if not raw_text:
        raise SandToolError("Cursor 路径不能为空")
    if sys.platform == "win32" and (
        raw_text.startswith("\\\\") or raw_text.startswith("\\\\?\\")
    ):
        raise SandToolError("不支持 UNC 或 Windows 设备路径")

    raw = Path(raw_text).expanduser()
    if not raw.is_absolute():
        raise SandToolError(f"Cursor 路径必须是绝对路径：{raw}")
    try:
        raw = raw.resolve(strict=True)
    except (FileNotFoundError, OSError) as exc:
        raise SandToolError(f"Cursor 路径不存在：{raw}") from exc

    seen: Set[str] = set()
    last_error: Optional[Exception] = None
    for candidate in _candidate_app_roots(raw):
        try:
            app_root = candidate.resolve(strict=True)
        except (FileNotFoundError, OSError):
            continue
        key = _path_key(app_root)
        if key in seen:
            continue
        seen.add(key)

        product_json = app_root / "product.json"
        if not product_json.is_file():
            continue
        try:
            product_real = product_json.resolve(strict=True)
            if not _is_within(product_real, app_root):
                raise SandToolError("product.json 符号链接逃逸出 Cursor app 目录")
            product = _read_product(product_real)
            install_root, executable = _resolve_executable(app_root)

            targets: List[Path] = []
            for rel, _extension_name in TARGET_SPECS:
                target = app_root.joinpath(*rel.split("/"))
                if not target.is_file():
                    continue
                target_real = target.resolve(strict=True)
                if not _is_within(target_real, app_root):
                    raise SandToolError(f"目标文件符号链接逃逸：{target}")
                targets.append(target_real)
            if not targets:
                raise SandToolError(
                    "Cursor 使用 app.asar 或当前版本没有可识别的 Sand 目标文件"
                )

            ext_host = app_root.joinpath(*EXT_HOST_REL.split("/"))
            ext_host_real = ext_host.resolve(strict=True) if ext_host.is_file() else None
            version = str(product.get("version") or product.get("commit") or "未知")
            return CursorLayout(
                install_root=install_root,
                app_root=app_root,
                product_json=product_real,
                executable=executable,
                target_paths=tuple(targets),
                ext_host_path=ext_host_real,
                version=version,
            )
        except SandToolError as exc:
            last_error = exc
            continue

    if last_error:
        raise SandToolError(f"Cursor 路径校验失败：{last_error}") from last_error
    raise SandToolError(f"路径中未找到 Cursor resources/app：{raw}")


def _powershell_executable() -> Optional[str]:
    return shutil.which("powershell.exe") or shutil.which("powershell") or shutil.which("pwsh")


def _windows_running_candidates() -> List[str]:
    powershell = _powershell_executable()
    if not powershell:
        return []
    script = (
        "$ErrorActionPreference='SilentlyContinue';"
        "[Console]::OutputEncoding=[System.Text.UTF8Encoding]::new();"
        "Get-CimInstance Win32_Process -Filter \"Name='Cursor.exe'\" | "
        "ForEach-Object { if ($_.ExecutablePath) { $_.ExecutablePath } }"
    )
    try:
        result = subprocess.run(
            [powershell, "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", script],
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=15,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        return []
    if result.returncode != 0:
        return []
    return [line.strip() for line in result.stdout.splitlines() if line.strip()]


def _windows_registry_candidates() -> List[str]:
    if sys.platform != "win32":
        return []
    try:
        import winreg
    except ImportError:
        return []

    candidates: List[str] = []
    roots = (winreg.HKEY_CURRENT_USER, winreg.HKEY_LOCAL_MACHINE)
    views = (winreg.KEY_WOW64_64KEY, winreg.KEY_WOW64_32KEY)
    uninstall = r"Software\Microsoft\Windows\CurrentVersion\Uninstall"
    for root in roots:
        for view in views:
            try:
                parent = winreg.OpenKey(root, uninstall, 0, winreg.KEY_READ | view)
            except OSError:
                continue
            with parent:
                index = 0
                while True:
                    try:
                        name = winreg.EnumKey(parent, index)
                    except OSError:
                        break
                    index += 1
                    try:
                        child = winreg.OpenKey(parent, name)
                    except OSError:
                        continue
                    with child:
                        def read(name_: str) -> str:
                            try:
                                return str(winreg.QueryValueEx(child, name_)[0] or "")
                            except OSError:
                                return ""

                        display_name = read("DisplayName").strip()
                        publisher = read("Publisher").strip()
                        if display_name.casefold() != "cursor" and "anysphere" not in publisher.casefold():
                            continue
                        install_location = read("InstallLocation").strip().strip('"')
                        display_icon = read("DisplayIcon").strip().strip('"')
                        if install_location:
                            candidates.append(install_location)
                        if display_icon:
                            icon_path = re.sub(r",\s*-?\d+$", "", display_icon).strip('"')
                            candidates.append(icon_path)
    return candidates


def _mac_process_paths(strict: bool = False) -> List[Tuple[int, Path]]:
    try:
        libproc = ctypes.CDLL("/usr/lib/libproc.dylib")
        proc_pidpath = libproc.proc_pidpath
        proc_pidpath.argtypes = [ctypes.c_int, ctypes.c_void_p, ctypes.c_uint32]
        proc_pidpath.restype = ctypes.c_int
        result = subprocess.run(
            ["ps", "-axo", "pid="],
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=10,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        if strict:
            raise SandToolError("无法读取 macOS 进程可执行路径") from exc
        return []
    if result.returncode != 0:
        if strict:
            raise SandToolError("无法读取 macOS 进程可执行路径")
        return []
    values: List[Tuple[int, Path]] = []
    for line in result.stdout.splitlines():
        try:
            pid = int(line.strip())
        except ValueError:
            continue
        buffer = ctypes.create_string_buffer(4096)
        length = proc_pidpath(pid, buffer, len(buffer))
        if length <= 0:
            continue
        try:
            executable = Path(os.fsdecode(buffer.value)).resolve(strict=False)
        except (OSError, ValueError):
            continue
        values.append((pid, executable))
    return values


def _bundle_for_executable(executable: Path) -> Optional[Path]:
    for item in (executable, *executable.parents):
        if item.name.casefold() == "cursor.app":
            return item
    return None


def _mac_running_candidates() -> List[str]:
    values: Dict[str, str] = {}
    for _pid, executable in _mac_process_paths():
        bundle = _bundle_for_executable(executable)
        if bundle is not None:
            values.setdefault(_path_key(bundle), str(bundle))
    return list(values.values())


def _mac_spotlight_candidates() -> List[str]:
    mdfind = shutil.which("mdfind")
    if not mdfind:
        return []
    try:
        result = subprocess.run(
            [
                mdfind,
                "kMDItemCFBundleIdentifier == 'com.todesktop.230313mzl4w4u92'",
            ],
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            timeout=15,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired):
        return []
    if result.returncode != 0:
        return []
    return [line.strip() for line in result.stdout.splitlines() if line.strip()]


def _linux_process_paths() -> List[str]:
    values: Dict[str, str] = {}
    proc_root = Path("/proc")
    try:
        processes = tuple(proc_root.iterdir())
    except OSError:
        return []
    for process in processes:
        if not process.name.isdigit() or process.name == str(os.getpid()):
            continue
        paths: List[Path] = []
        try:
            paths.append((process / "exe").resolve(strict=True))
        except (FileNotFoundError, OSError):
            pass
        try:
            with (process / "cmdline").open("rb") as handle:
                raw = handle.read(65536)
        except (FileNotFoundError, OSError):
            raw = b""
        for value in raw.split(b"\0"):
            if not value.startswith(b"/"):
                continue
            try:
                paths.append(Path(os.fsdecode(value)).resolve(strict=False))
            except (OSError, ValueError):
                continue
        for path in paths:
            values.setdefault(_path_key(path), str(path))
    return list(values.values())


def _linux_server_candidates() -> List[str]:
    roots: Dict[str, Path] = {}
    base = Path.home() / ".cursor-server" / "bin"
    for pattern in ("linux-*/*", "*"):
        try:
            candidates = base.glob(pattern)
        except OSError:
            continue
        for candidate in candidates:
            try:
                resolved = candidate.resolve(strict=True)
            except (FileNotFoundError, OSError):
                continue
            if resolved.is_dir() and (resolved / "product.json").is_file():
                roots.setdefault(_path_key(resolved), resolved)
    try:
        ordered = sorted(
            roots.values(),
            key=lambda item: item.stat().st_mtime,
            reverse=True,
        )
    except OSError:
        ordered = list(roots.values())
    return [str(item) for item in ordered]


def _default_candidate_groups() -> Iterable[Tuple[str, Sequence[str]]]:
    env_candidate = os.environ.get("SAND_CURSOR_INSTALL_DIR", "").strip()
    if env_candidate:
        yield "环境变量 SAND_CURSOR_INSTALL_DIR", (env_candidate,)

    if sys.platform == "win32":
        yield "运行中的 Cursor", _windows_running_candidates()
        yield "Windows 安装登记", _windows_registry_candidates()
        local = os.environ.get("LOCALAPPDATA", "")
        program_files = os.environ.get("ProgramFiles", r"C:\Program Files")
        program_files_x86 = os.environ.get("ProgramFiles(x86)", "")
        defaults = [
            str(Path(local) / "Programs" / "Cursor") if local else "",
            str(Path(local) / "Programs" / "cursor") if local else "",
            str(Path(local) / "Cursor") if local else "",
            str(Path(program_files) / "Cursor"),
            str(Path(program_files_x86) / "Cursor") if program_files_x86 else "",
        ]
        yield "Windows 默认目录", tuple(x for x in defaults if x)
    elif sys.platform == "darwin":
        yield "运行中的 Cursor", _mac_running_candidates()
        yield "macOS Spotlight", _mac_spotlight_candidates()
        yield "macOS 默认目录", (
            "/Applications/Cursor.app",
            str(Path.home() / "Applications" / "Cursor.app"),
        )
    elif sys.platform.startswith("linux"):
        yield "运行中的 Cursor", _linux_process_paths()
        yield "Cursor Server", _linux_server_candidates()
        yield "Linux 默认目录", (
            "/usr/share/cursor",
            "/usr/lib/cursor",
            "/opt/Cursor",
            "/opt/cursor",
            "/snap/cursor/current",
            str(Path.home() / ".local" / "share" / "cursor"),
        )

    path_cursor = shutil.which("cursor")
    if path_cursor:
        yield "PATH", (path_cursor,)


def _valid_layouts(values: Sequence[str]) -> List[CursorLayout]:
    layouts: Dict[str, CursorLayout] = {}
    for value in values:
        if not value:
            continue
        try:
            layout = layout_from_path(value)
        except SandToolError:
            continue
        layouts.setdefault(_path_key(layout.app_root), layout)
    return list(layouts.values())


def resolve_cursor_layout() -> CursorLayout:
    configured = _load_config().get("cursorInstallRoot")
    if isinstance(configured, str) and configured.strip():
        try:
            return layout_from_path(configured)
        except SandToolError as exc:
            raise SandToolError(
                f"已设置的 Cursor 路径失效：{configured}\n"
                "请运行 set-path <新路径>，或运行 set-path auto 恢复自动检测"
            ) from exc

    for source, values in _default_candidate_groups():
        layouts = _valid_layouts(tuple(values))
        if len(layouts) == 1:
            return layouts[0]
        if len(layouts) > 1:
            options = "\n".join(f"  - {item.install_root}" for item in layouts)
            raise SandToolError(
                f"{source}检测到多个 Cursor 安装，请先在菜单中选择 3 设置路径：\n{options}"
            )
    raise SandToolError(
        "未检测到 Cursor 安装，请在菜单中选择 3 设置 Cursor 路径"
        "（Cursor.exe、Cursor.app、cursor-server 或 resources/app）"
    )


def discover_cursor_layouts(include_configured: bool = True) -> List[CursorLayout]:
    """返回当前机器上可识别的 Cursor 安装，供图形界面选择。"""
    layouts: Dict[str, CursorLayout] = {}
    if include_configured:
        try:
            configured = _load_config().get("cursorInstallRoot")
        except SandToolError:
            configured = None
        if isinstance(configured, str) and configured.strip():
            try:
                layout = layout_from_path(configured)
                layouts.setdefault(_path_key(layout.app_root), layout)
            except SandToolError:
                pass
    for _source, values in _default_candidate_groups():
        for layout in _valid_layouts(tuple(values)):
            layouts.setdefault(_path_key(layout.app_root), layout)
    return list(layouts.values())


def save_cursor_path(value: str) -> Optional[CursorLayout]:
    if value.strip().casefold() in {"auto", "clear", "reset"}:
        _write_json_atomic(
            _config_path(),
            {
                "version": CONFIG_VERSION,
                "cursorInstallRoot": "",
                "lastVerifiedVersion": "",
                "updatedAt": datetime.now(timezone.utc).isoformat(),
            },
        )
        return None

    layout = layout_from_path(value)
    _write_json_atomic(
        _config_path(),
        {
            "version": CONFIG_VERSION,
            "cursorInstallRoot": str(layout.install_root),
            "lastVerifiedVersion": layout.version,
            "updatedAt": datetime.now(timezone.utc).isoformat(),
        },
    )
    return layout


def apply_patch_to_content(content: str) -> Tuple[str, PatchStats]:
    _ensure_payloads()
    stats = PatchStats()
    next_content = content
    legacy_client_re = re.compile(
        rf"([\"'])sand\1{LEGACY_CLIENT_MARKER_PATTERN}"
    )
    next_content, stats.migrated_client = legacy_client_re.subn(
        lambda match: match.group(1)
        + "sand"
        + match.group(1)
        + SAND_CLIENT_MARKER,
        next_content,
    )
    legacy_eligibility = "return!1;" + LEGACY_SAND_ELIGIBILITY_MARKER
    stats.migrated_eligibility = next_content.count(legacy_eligibility)
    next_content = next_content.replace(
        legacy_eligibility,
        "return!1;" + SAND_ELIGIBILITY_MARKER,
    )
    for key, rule in CLIENT_RULES:
        def replace_client(match: re.Match[str], stat_key: str = key) -> str:
            current = match.group(3)
            setattr(stats, stat_key, getattr(stats, stat_key) + 1)
            if current == "sand":
                stats.adopted_sand += 1
                marker = SAND_CLIENT_EXISTING_MARKER
            else:
                marker = SAND_CLIENT_MARKER
            return (
                match.group(1)
                + match.group(2)
                + "sand"
                + match.group(2)
                + marker
            )

        next_content = rule.sub(replace_client, next_content)

    for prefix in ELIGIBILITY_PREFIXES:
        count = next_content.count(prefix)
        if count == 0:
            continue
        patched = prefix.replace(
            _ELIGIBILITY_ANCHOR,
            "{return!1;" + SAND_ELIGIBILITY_MARKER + _ELIGIBILITY_ANCHOR[1:],
        )
        next_content = next_content.replace(prefix, patched)
        stats.eligibility += count

    route_count = next_content.count(MANAGED_LOCAL_ROUTE_ORIGINAL)
    if route_count:
        next_content = next_content.replace(
            MANAGED_LOCAL_ROUTE_ORIGINAL,
            MANAGED_LOCAL_ROUTE_PATCHED,
        )
        stats.managed_local_route += route_count

    runtime_load_count = next_content.count(LOCAL_RUNTIME_LOAD_ORIGINAL)
    if runtime_load_count:
        next_content = next_content.replace(
            LOCAL_RUNTIME_LOAD_ORIGINAL,
            LOCAL_RUNTIME_LOAD_PATCHED,
        )
        stats.local_runtime_load += runtime_load_count

    identity_count = next_content.count(AGENT_HOST_IDENTITY_ORIGINAL)
    if identity_count:
        next_content = next_content.replace(
            AGENT_HOST_IDENTITY_ORIGINAL,
            AGENT_HOST_IDENTITY_PATCHED,
        )
        stats.agent_host_identity += identity_count

    direct_injection = _direct_stream_injection()
    if (
        SAND_DIRECT_STREAM_MARKER not in next_content
        and DIRECT_STREAM_ANCHOR in next_content
    ):
        next_content = next_content.replace(
            DIRECT_STREAM_ANCHOR,
            DIRECT_STREAM_ANCHOR + direct_injection,
            1,
        )
        stats.direct_stream += 1

    if SAND_AGENT_HOST_ENABLEMENT_MARKER not in next_content:
        def enable_agent_host(match: re.Match[str]) -> str:
            variable = match.group(2)
            return (
                variable
                + "=!0;"
                + SAND_AGENT_HOST_ENABLEMENT_MARKER
                + match.group(1)
                + variable
                + match.group(3)
            )

        next_content, agent_host_count = AGENT_HOST_ENABLEMENT_RE.subn(
            enable_agent_host,
            next_content,
            count=1,
        )
        stats.agent_host_enablement += agent_host_count

    # 工具执行器修复（细节见服务端下发负载）
    if SAND_MOVE_EXEC_MARKER not in next_content:
        if MOVE_EXEC_ORIGINAL in next_content:
            next_content = next_content.replace(
                MOVE_EXEC_ORIGINAL, MOVE_EXEC_PATCHED, 1
            )
            stats.move_exec += 1

    # Plan 等非 Agent 模式路由修复
    if SAND_PLAN_MODE_MARKER not in next_content:
        next_content, plan_mode_count = PLAN_MODE_GATE_RE.subn(
            lambda match: SAND_PLAN_MODE_MARKER
            + "!1/*"
            + match.group(1)
            + "*/"
            + match.group(2),
            next_content,
            count=1,
        )
        stats.plan_mode += plan_mode_count

    # Build 按钮(executePlanAction)与后台任务回调(backgroundTaskCompletionAction)
    # Build/后台任务路由修复
    if SAND_ACTION_GATE_MARKER not in next_content:
        next_content, action_gate_count = ACTION_GATE_RE.subn(
            lambda match: SAND_ACTION_GATE_MARKER
            + "!1/*"
            + match.group(1)
            + "*/"
            + match.group(2),
            next_content,
            count=1,
        )
        stats.action_gate += action_gate_count

    # 交互队列键修复：同一 turn 内
    # 第二个问题会撞上第一个 pending 项（Plan 模式提问 → 创建计划）
    if (
        SAND_QUERY_KEY_MARKER not in next_content
        and QUERY_KEY_ANCHOR in next_content
    ):
        next_content = next_content.replace(
            QUERY_KEY_ANCHOR, QUERY_KEY_PATCHED, 1
        )
        stats.query_key += 1
    return next_content, stats


def remove_patch_from_content(content: str) -> Tuple[str, RemoveStats]:
    stats = RemoveStats()
    legacy_client_re = re.compile(
        rf"([\"'])sand\1{LEGACY_CLIENT_MARKER_PATTERN}"
    )
    next_content, legacy_client_count = legacy_client_re.subn(
        lambda match: match.group(1) + "ide" + match.group(1),
        content,
    )
    stats.client_type += legacy_client_count
    legacy_eligibility = "return!1;" + LEGACY_SAND_ELIGIBILITY_MARKER
    legacy_eligibility_count = next_content.count(legacy_eligibility)
    next_content = next_content.replace(legacy_eligibility, "")
    stats.eligibility += legacy_eligibility_count
    client_re = re.compile(rf"([\"'])sand\1{CLIENT_MARKER_PATTERN}")
    existing_re = re.compile(
        rf"([\"'])sand\1{CLIENT_EXISTING_MARKER_PATTERN}"
    )

    def remove_client(match: re.Match[str]) -> str:
        stats.client_type += 1
        return match.group(1) + "ide" + match.group(1)

    next_content = client_re.sub(remove_client, next_content)
    next_content, existing_count = existing_re.subn(
        lambda match: match.group(1) + "sand" + match.group(1),
        next_content,
    )
    stats.client_type += existing_count
    eligibility_re = re.compile(rf"return!1;{ELIGIBILITY_MARKER_PATTERN}")
    next_content, eligibility_count = eligibility_re.subn("", next_content)
    stats.eligibility += eligibility_count

    route_count = next_content.count(MANAGED_LOCAL_ROUTE_PATCHED)
    if route_count:
        next_content = next_content.replace(
            MANAGED_LOCAL_ROUTE_PATCHED,
            MANAGED_LOCAL_ROUTE_ORIGINAL,
        )
        stats.managed_local_route += route_count

    runtime_load_count = next_content.count(LOCAL_RUNTIME_LOAD_PATCHED)
    if runtime_load_count:
        next_content = next_content.replace(
            LOCAL_RUNTIME_LOAD_PATCHED,
            LOCAL_RUNTIME_LOAD_ORIGINAL,
        )
        stats.local_runtime_load += runtime_load_count

    identity_count = next_content.count(AGENT_HOST_IDENTITY_PATCHED)
    if identity_count:
        next_content = next_content.replace(
            AGENT_HOST_IDENTITY_PATCHED,
            AGENT_HOST_IDENTITY_ORIGINAL,
        )
        stats.agent_host_identity += identity_count

    direct_injection = _direct_stream_injection()
    direct_count = next_content.count(direct_injection)
    if direct_count:
        next_content = next_content.replace(direct_injection, "")
        stats.direct_stream += direct_count

    next_content, agent_host_count = AGENT_HOST_ENABLEMENT_PATCH_RE.subn(
        lambda match: match.group(2) + match.group(1) + match.group(3),
        next_content,
    )
    stats.agent_host_enablement += agent_host_count
    if MOVE_EXEC_PATCHED in next_content:
        next_content = next_content.replace(
            MOVE_EXEC_PATCHED, MOVE_EXEC_ORIGINAL, 1
        )
        stats.move_exec += 1
    next_content, plan_mode_count = PLAN_MODE_GATE_PATCH_RE.subn(
        lambda match: match.group(1) + match.group(2),
        next_content,
    )
    stats.plan_mode += plan_mode_count
    next_content, action_gate_count = ACTION_GATE_PATCH_RE.subn(
        lambda match: match.group(1) + match.group(2),
        next_content,
    )
    stats.action_gate += action_gate_count
    if QUERY_KEY_PATCHED in next_content:
        next_content = next_content.replace(
            QUERY_KEY_PATCHED, QUERY_KEY_ANCHOR, 1
        )
        stats.query_key += 1
    return next_content, stats


def _decode_js(data: bytes, path: Path) -> str:
    try:
        return data.decode("utf-8")
    except UnicodeDecodeError as exc:
        raise SandToolError(f"目标文件不是 UTF-8，拒绝修改：{path}") from exc


def _read_planned_file(path: Path) -> PlannedFile:
    original = path.read_bytes()
    return PlannedFile(
        original=original,
        next_bytes=original,
        mode=stat.S_IMODE(path.stat().st_mode),
    )


def _target_extension_name(layout: CursorLayout, file_path: Path) -> Optional[str]:
    for rel, extension_name in TARGET_SPECS:
        if not extension_name:
            continue
        candidate = layout.app_root.joinpath(*rel.split("/")).resolve()
        if candidate == file_path.resolve():
            return extension_name
    return None


def _update_extension_hashes(
    layout: CursorLayout,
    plan: Dict[Path, PlannedFile],
) -> None:
    changed_extensions: List[Tuple[str, bytes]] = []
    for file_path, planned in plan.items():
        extension_name = _target_extension_name(layout, file_path)
        if extension_name:
            changed_extensions.append((extension_name, planned.next_bytes))
    if not changed_extensions or layout.ext_host_path is None:
        return

    ext_path = layout.ext_host_path
    existing = plan.get(ext_path) or _read_planned_file(ext_path)
    next_content = _decode_js(existing.next_bytes, ext_path)
    original_content = _decode_js(existing.original, ext_path)

    for extension_name, next_main in changed_extensions:
        extension_id = "anysphere." + extension_name
        if f'"{extension_id}"' not in next_content:
            continue
        digest = hashlib.sha256(next_main).hexdigest()
        pattern = re.compile(
            rf'(\"{re.escape(extension_id)}\"\s*:\s*\{{[\s\S]{{0,2400}}?'
            rf'\"main\.js\"\s*:\s*\")[0-9a-f]{{64}}(\")'
        )
        next_content, count = pattern.subn(
            lambda match: match.group(1) + digest + match.group(2),
            next_content,
            count=1,
        )
        if count > 1:
            raise SandToolError(f"{extension_id} 的内嵌 main.js 哈希不唯一")

    if next_content != original_content:
        plan[ext_path] = PlannedFile(
            original=existing.original,
            next_bytes=next_content.encode("utf-8"),
            mode=existing.mode,
        )


def _sync_product_checksums(
    layout: CursorLayout,
    plan: Dict[Path, PlannedFile],
) -> None:
    product_file = _read_planned_file(layout.product_json)
    has_bom = product_file.original.startswith(b"\xef\xbb\xbf")
    try:
        product = json.loads(product_file.original.decode("utf-8-sig"))
    except Exception as exc:
        raise SandToolError("product.json 无法解析，拒绝提交补丁") from exc
    if not isinstance(product, dict):
        raise SandToolError("product.json 顶层必须是对象")
    checksums = product.get("checksums")
    if not isinstance(checksums, dict):
        return

    out_root = (layout.app_root / "out").resolve()
    changed = False
    for key in list(checksums.keys()):
        if not isinstance(key, str):
            continue
        parts = [part for part in re.split(r"[\\/]", key) if part]
        target = out_root.joinpath(*parts).resolve()
        if not _is_within(target, out_root):
            raise SandToolError(f"product.json checksum 路径逃逸：{key}")
        planned = plan.get(target)
        if planned is not None:
            data = planned.next_bytes
        elif target.is_file():
            data = target.read_bytes()
        else:
            continue
        digest = _product_checksum(data)
        if checksums.get(key) != digest:
            checksums[key] = digest
            changed = True

    if not changed:
        return
    text = json.dumps(product, ensure_ascii=False, indent="\t")
    next_bytes = text.encode("utf-8")
    if has_bom:
        next_bytes = b"\xef\xbb\xbf" + next_bytes
    plan[layout.product_json] = PlannedFile(
        original=product_file.original,
        next_bytes=next_bytes,
        mode=product_file.mode,
    )


def _planned_extension_names(
    layout: CursorLayout,
    plan: Mapping[Path, PlannedFile],
) -> Set[str]:
    names: Set[str] = set()
    for file_path in plan:
        extension_name = _target_extension_name(layout, file_path)
        if extension_name:
            names.add(extension_name)
    return names


def _verify_extension_hashes(
    layout: CursorLayout,
    extension_names: Iterable[str],
) -> None:
    names = set(extension_names)
    if layout.ext_host_path is None or not names:
        return
    ext_content = _decode_js(layout.ext_host_path.read_bytes(), layout.ext_host_path)
    for rel, extension_name in TARGET_SPECS:
        if not extension_name or extension_name not in names:
            continue
        main_path = layout.app_root.joinpath(*rel.split("/"))
        if not main_path.is_file():
            continue
        extension_id = "anysphere." + extension_name
        if f'"{extension_id}"' not in ext_content:
            continue
        pattern = re.compile(
            rf'\"{re.escape(extension_id)}\"\s*:\s*\{{[\s\S]{{0,2400}}?'
            rf'\"main\.js\"\s*:\s*\"([0-9a-f]{{64}})\"'
        )
        match = pattern.search(ext_content)
        if not match:
            continue
        expected = hashlib.sha256(main_path.read_bytes()).hexdigest()
        if match.group(1) != expected:
            raise SandToolError(f"{extension_id} 的内嵌哈希校验失败")


def _verify_product_checksums(layout: CursorLayout) -> int:
    product = json.loads(layout.product_json.read_bytes().decode("utf-8-sig"))
    checksums = product.get("checksums") if isinstance(product, dict) else None
    if not isinstance(checksums, dict):
        return 0
    out_root = (layout.app_root / "out").resolve()
    checked = 0
    for key, written in checksums.items():
        if not isinstance(key, str):
            continue
        parts = [part for part in re.split(r"[\\/]", key) if part]
        target = out_root.joinpath(*parts).resolve()
        if not _is_within(target, out_root) or not target.is_file():
            continue
        checked += 1
        if written != _product_checksum(target.read_bytes()):
            raise SandToolError(f"product.json 完整性哈希校验失败：{key}")
    return checked


def inspect_status(layout: CursorLayout) -> PatchStatus:
    client_markers = 0
    eligibility_markers = 0
    managed_local_route_markers = 0
    local_runtime_load_markers = 0
    direct_stream_markers = 0
    agent_host_enablement_markers = 0
    agent_host_identity_markers = 0
    move_exec_markers = 0
    plan_mode_markers = 0
    action_gate_markers = 0
    query_key_markers = 0
    legacy_client_markers = 0
    legacy_eligibility_markers = 0
    ide_matches = 0
    external_sand_matches = 0
    external_marker_count = 0
    patched_files: List[Path] = []
    for target in layout.target_paths:
        content = _decode_js(target.read_bytes(), target)
        client_count = content.count(SAND_CLIENT_MARKER) + content.count(
            SAND_CLIENT_EXISTING_MARKER
        )
        eligibility_count = content.count(SAND_ELIGIBILITY_MARKER)
        managed_local_route_count = content.count(SAND_MANAGED_LOCAL_ROUTE_MARKER)
        local_runtime_load_count = content.count(SAND_LOCAL_RUNTIME_LOAD_MARKER)
        direct_stream_count = content.count(SAND_DIRECT_STREAM_MARKER)
        agent_host_enablement_count = content.count(
            SAND_AGENT_HOST_ENABLEMENT_MARKER
        )
        agent_host_identity_count = content.count(
            SAND_AGENT_HOST_IDENTITY_MARKER
        )
        move_exec_count = content.count(SAND_MOVE_EXEC_MARKER)
        plan_mode_count = content.count(SAND_PLAN_MODE_MARKER)
        action_gate_count = content.count(SAND_ACTION_GATE_MARKER)
        query_key_count = content.count(SAND_QUERY_KEY_MARKER)
        legacy_client_count = len(
            re.findall(
                rf"([\"'])sand\1{LEGACY_CLIENT_MARKER_PATTERN}",
                content,
            )
        )
        legacy_eligibility_count = content.count(
            "return!1;" + LEGACY_SAND_ELIGIBILITY_MARKER
        )
        external_marker_count += max(
            0,
            len(re.findall(CLIENT_MARKER_GUARD_PATTERN, content))
            - client_count
            - legacy_client_count,
        )
        external_marker_count += max(
            0,
            len(re.findall(ELIGIBILITY_MARKER_GUARD_PATTERN, content))
            - eligibility_count
            - legacy_eligibility_count,
        )
        if (
            client_count
            + eligibility_count
            + legacy_client_count
            + legacy_eligibility_count
            + managed_local_route_count
            + local_runtime_load_count
            + direct_stream_count
            + agent_host_enablement_count
            + agent_host_identity_count
            + move_exec_count
            + plan_mode_count
            + action_gate_count
            + query_key_count
        ):
            patched_files.append(target)
        client_markers += client_count
        eligibility_markers += eligibility_count
        legacy_client_markers += legacy_client_count
        legacy_eligibility_markers += legacy_eligibility_count
        managed_local_route_markers += managed_local_route_count
        local_runtime_load_markers += local_runtime_load_count
        direct_stream_markers += direct_stream_count
        agent_host_enablement_markers += agent_host_enablement_count
        agent_host_identity_markers += agent_host_identity_count
        move_exec_markers += move_exec_count
        plan_mode_markers += plan_mode_count
        action_gate_markers += action_gate_count
        query_key_markers += query_key_count
        for _key, rule in CLIENT_RULES:
            for match in rule.finditer(content):
                if match.group(3) == "sand":
                    external_sand_matches += 1
                else:
                    ide_matches += 1
    return PatchStatus(
        client_markers=client_markers,
        eligibility_markers=eligibility_markers,
        ide_matches=ide_matches,
        external_sand_matches=external_sand_matches,
        external_marker_count=external_marker_count,
        legacy_client_markers=legacy_client_markers,
        legacy_eligibility_markers=legacy_eligibility_markers,
        patched_files=tuple(patched_files),
        managed_local_route_markers=managed_local_route_markers,
        local_runtime_load_markers=local_runtime_load_markers,
        direct_stream_markers=direct_stream_markers,
        agent_host_enablement_markers=agent_host_enablement_markers,
        agent_host_identity_markers=agent_host_identity_markers,
        move_exec_markers=move_exec_markers,
        plan_mode_markers=plan_mode_markers,
        action_gate_markers=action_gate_markers,
        query_key_markers=query_key_markers,
    )


def _create_backup(
    layout: CursorLayout,
    plan: Mapping[Path, PlannedFile],
    operation: str,
) -> Tuple[Path, Dict[str, object]]:
    app_hash = hashlib.sha256(str(layout.app_root).encode("utf-8")).hexdigest()[:16]
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%fZ")
    backup_dir = _config_dir() / "backups" / app_hash / f"{stamp}-{operation}"
    files_dir = backup_dir / "files"
    entries: List[Dict[str, object]] = []
    for path, planned in plan.items():
        try:
            relative = path.resolve().relative_to(layout.app_root.resolve())
        except ValueError as exc:
            raise SandToolError(f"计划文件逃逸出 Cursor app：{path}") from exc
        backup_file = files_dir / relative
        _atomic_write(backup_file, planned.original, planned.mode)
        entries.append(
            {
                "path": relative.as_posix(),
                "originalSha256": _sha256(planned.original),
                "nextSha256": _sha256(planned.next_bytes),
                "mode": planned.mode,
            }
        )
    manifest: Dict[str, object] = {
        "version": 1,
        "toolVersion": TOOL_VERSION,
        "operation": operation,
        "status": "prepared",
        "appRoot": str(layout.app_root),
        "cursorVersion": layout.version,
        "createdAt": datetime.now(timezone.utc).isoformat(),
        "files": entries,
    }
    _write_json_atomic(backup_dir / "manifest.json", manifest)
    return backup_dir, manifest


def _update_backup_manifest(
    backup_dir: Path,
    manifest: Dict[str, object],
    status_value: str,
    error: Optional[str] = None,
) -> None:
    manifest["status"] = status_value
    manifest["finishedAt"] = datetime.now(timezone.utc).isoformat()
    if error:
        manifest["error"] = error[:1000]
    _write_json_atomic(backup_dir / "manifest.json", manifest)


def _commit_plan(
    layout: CursorLayout,
    plan: Mapping[Path, PlannedFile],
    operation: str,
    validator,
) -> Tuple[Tuple[Path, ...], Path]:
    if not plan:
        raise SandToolError("内部错误：提交计划为空")
    for path, planned in plan.items():
        if _sha256(path.read_bytes()) != _sha256(planned.original):
            raise SandToolError(f"文件在计划生成后发生变化，已停止操作：{path}")
    backup_dir, manifest = _create_backup(layout, plan, operation)
    attempted: List[Path] = []
    written: List[Path] = []
    try:
        for path, planned in plan.items():
            if _sha256(path.read_bytes()) != _sha256(planned.original):
                raise SandToolError(f"文件在写入前发生变化，已停止操作：{path}")
            attempted.append(path)
            _atomic_write(path, planned.next_bytes, planned.mode)
            written.append(path)
        validator()
        for path, planned in plan.items():
            if _sha256(path.read_bytes()) != _sha256(planned.next_bytes):
                raise SandToolError(f"写入后哈希校验失败：{path}")
        _update_backup_manifest(backup_dir, manifest, "committed")
        return tuple(written), backup_dir
    except (Exception, KeyboardInterrupt) as exc:
        rollback_errors: List[str] = []
        for path in reversed(attempted):
            planned = plan[path]
            try:
                current_hash = _sha256(path.read_bytes())
                original_hash = _sha256(planned.original)
                next_hash = _sha256(planned.next_bytes)
                if current_hash == original_hash:
                    continue
                if current_hash != next_hash:
                    rollback_errors.append(f"{path}: 文件已被外部修改，未覆盖")
                    continue
                _atomic_write(path, planned.original, planned.mode)
            except Exception as rollback_exc:
                rollback_errors.append(f"{path}: {rollback_exc}")
        message = str(exc)
        if rollback_errors:
            message += "; rollback errors: " + " | ".join(rollback_errors)
        try:
            _update_backup_manifest(backup_dir, manifest, "rolled_back", message)
        except Exception:
            pass
        if rollback_errors:
            raise SandToolError(
                "补丁失败且有文件未能自动回滚，请保留备份目录："
                f"{backup_dir}\n{message}"
            ) from exc
        raise


def _windows_close_cursor(layout: CursorLayout) -> int:
    powershell = _powershell_executable()
    if not powershell:
        raise SandToolError("未找到 PowerShell，无法安全关闭 Cursor")
    script = r"""
$ErrorActionPreference = 'Stop'
[Console]::OutputEncoding = [System.Text.UTF8Encoding]::new()
$target = [System.IO.Path]::GetFullPath($env:SAND_CURSOR_EXE)
function Get-SandCursorTargets {
  @(Get-CimInstance Win32_Process -Filter "Name='Cursor.exe'" -ErrorAction Stop | Where-Object {
    $_.ExecutablePath -and [string]::Equals(
      [System.IO.Path]::GetFullPath($_.ExecutablePath),
      $target,
      [System.StringComparison]::OrdinalIgnoreCase
    )
  })
}
$before = @(Get-SandCursorTargets)
foreach ($item in $before) {
  try {
    $process = Get-Process -Id $item.ProcessId -ErrorAction Stop
    if ($process.MainWindowHandle -ne 0) { [void]$process.CloseMainWindow() }
  } catch {}
}
$deadline = [DateTime]::UtcNow.AddSeconds(12)
while ([DateTime]::UtcNow -lt $deadline -and @(Get-SandCursorTargets).Count -gt 0) {
  Start-Sleep -Milliseconds 250
}
$remaining = @(Get-SandCursorTargets)
foreach ($item in $remaining) {
  try { Stop-Process -Id $item.ProcessId -Force -ErrorAction Stop } catch {}
}
Start-Sleep -Milliseconds 500
if (@(Get-SandCursorTargets).Count -gt 0) { exit 3 }
Write-Output ("CLOSED=" + $before.Count)
""".strip()
    env = dict(os.environ)
    env["SAND_CURSOR_EXE"] = str(layout.executable)
    try:
        result = subprocess.run(
            [powershell, "-NoLogo", "-NoProfile", "-NonInteractive", "-Command", script],
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
            env=env,
            timeout=25,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        raise SandToolError("无法安全关闭所选 Cursor 进程，请手动退出后重试") from exc
    if result.returncode != 0:
        raise SandToolError(
            "无法安全关闭所选 Cursor 进程，请手动退出后重试"
        )
    match = re.search(r"CLOSED=(\d+)", result.stdout)
    return int(match.group(1)) if match else 0


def _mac_bundle_pids(layout: CursorLayout) -> List[int]:
    bundle = _find_app_bundle(layout.app_root)
    if bundle is None:
        return []
    contents = (bundle.resolve() / "Contents").resolve()
    pids: List[int] = []
    for pid, executable in _mac_process_paths(strict=True):
        if pid != os.getpid() and _is_within(executable, contents):
            pids.append(pid)
    return pids


def _wait_for_mac_exit(layout: CursorLayout, timeout_seconds: float) -> bool:
    deadline = time.monotonic() + timeout_seconds
    while time.monotonic() < deadline:
        if not _mac_bundle_pids(layout):
            return True
        time.sleep(0.25)
    return not _mac_bundle_pids(layout)


def _mac_close_cursor(layout: CursorLayout) -> int:
    before = _mac_bundle_pids(layout)
    if not before:
        return 0
    selected_bundle = _find_app_bundle(layout.app_root)
    running_bundles: Dict[str, Path] = {}
    for _pid, executable in _mac_process_paths(strict=True):
        bundle = _bundle_for_executable(executable)
        if bundle is not None:
            running_bundles.setdefault(_path_key(bundle), bundle)
    if selected_bundle is not None and len(running_bundles) == 1:
        osascript = shutil.which("osascript") or "/usr/bin/osascript"
        try:
            subprocess.run(
                [
                    osascript,
                    "-e",
                    'tell application id "com.todesktop.230313mzl4w4u92" to quit',
                ],
                capture_output=True,
                timeout=10,
                check=False,
            )
        except (OSError, subprocess.TimeoutExpired):
            pass
        if _wait_for_mac_exit(layout, 12):
            return len(before)

    for pid in _mac_bundle_pids(layout):
        try:
            os.kill(pid, signal.SIGTERM)
        except (ProcessLookupError, PermissionError):
            pass
    if _wait_for_mac_exit(layout, 3):
        return len(before)

    for pid in _mac_bundle_pids(layout):
        try:
            os.kill(pid, signal.SIGKILL)
        except (ProcessLookupError, PermissionError):
            pass
    if not _wait_for_mac_exit(layout, 2):
        raise SandToolError("无法安全关闭所选 Cursor 进程，请手动退出后重试")
    return len(before)


def _linux_layout_pids(layout: CursorLayout) -> List[int]:
    root = layout.install_root.resolve()
    pids: List[int] = []
    try:
        processes = tuple(Path("/proc").iterdir())
    except OSError as exc:
        raise SandToolError("无法读取 Linux 进程列表") from exc
    for process in processes:
        if not process.name.isdigit():
            continue
        pid = int(process.name)
        if pid == os.getpid():
            continue
        belongs_to_layout = False
        try:
            executable = (process / "exe").resolve(strict=True)
            belongs_to_layout = _is_within(executable, root)
        except (FileNotFoundError, OSError):
            pass
        if not belongs_to_layout:
            try:
                with (process / "cmdline").open("rb") as handle:
                    raw = handle.read(65536)
            except (FileNotFoundError, OSError):
                raw = b""
            for value in raw.split(b"\0"):
                if not value.startswith(b"/"):
                    continue
                try:
                    argument = Path(os.fsdecode(value)).resolve(strict=False)
                except (OSError, ValueError):
                    continue
                if _is_within(argument, root):
                    belongs_to_layout = True
                    break
        if belongs_to_layout:
            pids.append(pid)
    return pids


def _wait_for_linux_exit(layout: CursorLayout, timeout_seconds: float) -> bool:
    deadline = time.monotonic() + timeout_seconds
    while time.monotonic() < deadline:
        if not _linux_layout_pids(layout):
            return True
        time.sleep(0.25)
    return not _linux_layout_pids(layout)


def _linux_close_cursor(layout: CursorLayout) -> int:
    before = _linux_layout_pids(layout)
    if not before:
        return 0
    for pid in before:
        try:
            os.kill(pid, signal.SIGTERM)
        except ProcessLookupError:
            pass
        except PermissionError as exc:
            raise SandToolError("没有权限关闭 Cursor 进程，请使用 sudo 重试") from exc
    if _wait_for_linux_exit(layout, 12):
        return len(before)
    for pid in _linux_layout_pids(layout):
        try:
            os.kill(pid, signal.SIGKILL)
        except (ProcessLookupError, PermissionError):
            pass
    if not _wait_for_linux_exit(layout, 2):
        raise SandToolError("无法安全关闭所选 Cursor 进程，请手动退出后重试")
    return len(before)


def _is_cursor_server_layout(layout: CursorLayout) -> bool:
    return (
        layout.executable.name == "cursor-server"
        or ".cursor-server" in layout.install_root.parts
    )


def _manage_processes_for_layout(
    layout: CursorLayout,
    requested: bool = True,
) -> bool:
    return requested and not (
        sys.platform.startswith("linux") and _is_cursor_server_layout(layout)
    )


def _stream_mode_installed(layout: CursorLayout, status: PatchStatus) -> bool:
    base_markers_present = (
        status.managed_local_route_markers > 0
        and status.local_runtime_load_markers > 0
        and status.direct_stream_markers > 0
        and status.agent_host_identity_markers > 0
    )
    if not base_markers_present:
        return False
    if sys.platform.startswith("linux") and _is_cursor_server_layout(layout):
        # Cursor Server does not contain the desktop workbench's agent-host
        # enablement assignments. Its agent host is already part of
        # the server runtime, so the four shared Stream markers are complete.
        return status.agent_host_enablement_markers == 0
    return status.agent_host_enablement_markers > 0


def _expected_agent_host_enablement_markers(layout: CursorLayout) -> int:
    if sys.platform.startswith("linux") and _is_cursor_server_layout(layout):
        return 0
    return 2


def close_cursor(layout: CursorLayout) -> int:
    if sys.platform == "win32":
        return _windows_close_cursor(layout)
    if sys.platform == "darwin":
        return _mac_close_cursor(layout)
    if sys.platform.startswith("linux"):
        return _linux_close_cursor(layout)
    raise SandToolError(f"当前不支持此操作系统：{sys.platform}")


def start_cursor(layout: CursorLayout) -> bool:
    try:
        if sys.platform == "win32":
            startupinfo = subprocess.STARTUPINFO()
            startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
            subprocess.Popen(
                [str(layout.executable)],
                cwd=str(layout.install_root),
                stdin=subprocess.DEVNULL,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                startupinfo=startupinfo,
                creationflags=0x00000008 | 0x00000200,
            )
            return True
        if sys.platform == "darwin":
            bundle = _find_app_bundle(layout.app_root)
            if bundle is None:
                return False
            subprocess.run(
                [shutil.which("open") or "/usr/bin/open", "-a", str(bundle)],
                stdin=subprocess.DEVNULL,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                timeout=20,
                check=False,
            )
            return True
        if sys.platform.startswith("linux"):
            if _is_cursor_server_layout(layout):
                return False
            subprocess.Popen(
                [str(layout.executable)],
                cwd=str(layout.install_root),
                stdin=subprocess.DEVNULL,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                start_new_session=True,
            )
            return True
    except (OSError, subprocess.TimeoutExpired):
        return False
    return False


def _move_exec_main_js(layout: CursorLayout) -> Path:
    return layout.app_root / "extensions" / "cursor-agent-host" / "dist" / "main.js"


def move_exec_state(layout: CursorLayout) -> Tuple[bool, bool]:
    path = _move_exec_main_js(layout)
    if not path.is_file():
        return False, False
    content = _decode_js(path.read_bytes(), path)
    has_original = MOVE_EXEC_ORIGINAL is not None and MOVE_EXEC_ORIGINAL in content
    return (SAND_MOVE_EXEC_MARKER in content, has_original)


def cmd_move_exec_check(layout: CursorLayout) -> int:
    patched, original = move_exec_state(layout)
    print(f"[move-exec] 文件: {_move_exec_main_js(layout)}")
    if patched:
        print("[move-exec] 结论: 已强制 move_exec ON（已补丁）")
    elif original:
        print("[move-exec] 结论: 仍走 move_exec gate（未补丁，工具会崩）")
    else:
        print("[move-exec] 结论: 未匹配到目标代码（版本不同？）")
    return 0


def cmd_move_exec_apply(layout: CursorLayout, manage_process: bool = True) -> int:
    patched, original = move_exec_state(layout)
    if patched:
        print("[move-exec] 已补丁，跳过")
        return 0
    if not original:
        print("[move-exec] 错误: 未找到目标代码，无法补丁（版本不同？）")
        return 1
    path = _move_exec_main_js(layout)
    backup_dir = _config_dir() / "move_exec_backup"
    backup_dir.mkdir(parents=True, exist_ok=True)
    backup_file = backup_dir / "main.js"
    if not backup_file.exists():
        backup_file.write_bytes(path.read_bytes())
        print(f"[move-exec] 已备份到 {backup_file}")
    if manage_process:
        close_cursor(layout)
    content = _decode_js(path.read_bytes(), path)
    content = content.replace(MOVE_EXEC_ORIGINAL, MOVE_EXEC_PATCHED, 1)
    _atomic_write(path, content.encode("utf-8"), stat.S_IMODE(path.stat().st_mode))
    print("[move-exec] 补丁完成: move_exec 已强制 ON")
    if manage_process:
        start_cursor(layout)
    return 0


def cmd_move_exec_restore(layout: CursorLayout, manage_process: bool = True) -> int:
    patched, original = move_exec_state(layout)
    if not patched:
        print("[move-exec] 未打补丁，跳过")
        return 0
    path = _move_exec_main_js(layout)
    backup_file = _config_dir() / "move_exec_backup" / "main.js"
    if backup_file.exists():
        if manage_process:
            close_cursor(layout)
        _atomic_write(
            path, backup_file.read_bytes(), stat.S_IMODE(path.stat().st_mode)
        )
        print("[move-exec] 已从备份还原")
        if manage_process:
            start_cursor(layout)
        return 0
    if MOVE_EXEC_PATCHED in _decode_js(path.read_bytes(), path):
        if manage_process:
            close_cursor(layout)
        content = _decode_js(path.read_bytes(), path)
        content = content.replace(MOVE_EXEC_PATCHED, MOVE_EXEC_ORIGINAL, 1)
        _atomic_write(path, content.encode("utf-8"), stat.S_IMODE(path.stat().st_mode))
        print("[move-exec] 已反向替换还原")
        if manage_process:
            start_cursor(layout)
        return 0
    print("[move-exec] 错误: 既无备份又找不到补丁串，无法还原")
    return 1


def _build_install_plan(
    layout: CursorLayout,
) -> Tuple[Dict[Path, PlannedFile], PatchStats]:
    plan: Dict[Path, PlannedFile] = {}
    total = PatchStats()
    for target in layout.target_paths:
        original = _read_planned_file(target)
        content = _decode_js(original.original, target)
        next_content, stats = apply_patch_to_content(content)
        if next_content != content:
            plan[target] = PlannedFile(
                original=original.original,
                next_bytes=next_content.encode("utf-8"),
                mode=original.mode,
            )
        total.is_glass += stats.is_glass
        total.object_header += stats.object_header
        total.set_header += stats.set_header
        total.eligibility += stats.eligibility
        total.adopted_sand += stats.adopted_sand
        total.migrated_client += stats.migrated_client
        total.migrated_eligibility += stats.migrated_eligibility
        total.managed_local_route += stats.managed_local_route
        total.local_runtime_load += stats.local_runtime_load
        total.direct_stream += stats.direct_stream
        total.agent_host_enablement += stats.agent_host_enablement
        total.agent_host_identity += stats.agent_host_identity
        total.move_exec += stats.move_exec
        total.plan_mode += stats.plan_mode
        total.action_gate += stats.action_gate
        total.query_key += stats.query_key
    if plan:
        _update_extension_hashes(layout, plan)
        _sync_product_checksums(layout, plan)
    return plan, total


def require_license() -> None:
    """联网校验卡密授权并加载服务器下发的补丁负载。失败抛 SandToolError。"""
    try:
        result = sand_license.ensure_licensed(app_version=TOOL_VERSION, notify=print_warn)
    except sand_license.LicenseError as exc:
        raise SandToolError(f"授权校验未通过：{exc}") from exc
    payloads = result.get("payloads") if isinstance(result, dict) else None
    if not payloads:
        raise SandToolError("授权服务器未下发补丁数据，请重试或联系客服。")
    load_payloads(payloads)


def _build_uninstall_plan(
    layout: CursorLayout,
) -> Tuple[Dict[Path, PlannedFile], RemoveStats]:
    plan: Dict[Path, PlannedFile] = {}
    total = RemoveStats()
    for target in layout.target_paths:
        original = _read_planned_file(target)
        content = _decode_js(original.original, target)
        next_content, stats = remove_patch_from_content(content)
        if next_content != content:
            plan[target] = PlannedFile(
                original=original.original,
                next_bytes=next_content.encode("utf-8"),
                mode=original.mode,
            )
        total.client_type += stats.client_type
        total.eligibility += stats.eligibility
        total.managed_local_route += stats.managed_local_route
        total.local_runtime_load += stats.local_runtime_load
        total.direct_stream += stats.direct_stream
        total.agent_host_enablement += stats.agent_host_enablement
        total.agent_host_identity += stats.agent_host_identity
        total.move_exec += stats.move_exec
        total.plan_mode += stats.plan_mode
        total.action_gate += stats.action_gate
        total.query_key += stats.query_key
    if plan:
        _update_extension_hashes(layout, plan)
        _sync_product_checksums(layout, plan)
    return plan, total


def install(layout: CursorLayout, manage_process: bool = True) -> int:
    require_license()
    manage_process = _manage_processes_for_layout(layout, manage_process)
    if layout.version != SUPPORTED_CURSOR_VERSION:
        raise SandToolError(
            f"当前 Cursor 版本为 {layout.version}，"
            f"本工具仅适配 Cursor {SUPPORTED_CURSOR_VERSION}。"
            "请更换为适配版本后再安装"
        )
    before = inspect_status(layout)
    if before.external_marker_count:
        raise SandToolError(
            "检测到其他 Sand 模式标记，本脚本不会接管或覆盖它；"
            "请先用原安装方式卸载"
        )
    plan, _stats = _build_install_plan(layout)
    if not plan:
        if before.installed and _stream_mode_installed(layout, before):
            if manage_process:
                start_cursor(layout)
            return 0
        raise SandToolError("当前 Cursor 版本未匹配到 Sand 客户端模式规则")
    expected_agent_host = _expected_agent_host_enablement_markers(layout)
    if (
        before.managed_local_route_markers + _stats.managed_local_route != 1
        or (
            before.local_runtime_load_markers
            + _stats.local_runtime_load
            != 1
        )
        or (
            before.agent_host_identity_markers
            + _stats.agent_host_identity
            != 1
        )
        or before.direct_stream_markers + _stats.direct_stream != 1
        or (
            before.agent_host_enablement_markers
            + _stats.agent_host_enablement
            != expected_agent_host
        )
    ):
        raise SandToolError(
            "当前 Cursor 版本未完整匹配 Sand Stream 规则："
            f"route={before.managed_local_route_markers + _stats.managed_local_route}, "
            "runtimeLoad="
            f"{before.local_runtime_load_markers + _stats.local_runtime_load}, "
            "identity="
            f"{before.agent_host_identity_markers + _stats.agent_host_identity}, "
            "directStream="
            f"{before.direct_stream_markers + _stats.direct_stream}, "
            "agentHost="
            f"{before.agent_host_enablement_markers + _stats.agent_host_enablement}"
        )

    if manage_process:
        close_cursor(layout)
    changed_extensions = _planned_extension_names(layout, plan)

    def validate() -> None:
        status = inspect_status(layout)
        stream_mode_installed = _stream_mode_installed(layout, status)
        if (
            not status.installed
            or not stream_mode_installed
            or status.ide_matches != 0
            or status.external_marker_count != 0
            or status.legacy_client_markers != 0
            or status.legacy_eligibility_markers != 0
        ):
            raise SandToolError(
                "安装后状态校验失败："
                f"markers={status.client_markers + status.eligibility_markers}, "
                f"remainingIde={status.ide_matches}, "
                f"streamMode={stream_mode_installed}, "
                "remainingLegacy="
                f"{status.legacy_client_markers + status.legacy_eligibility_markers}"
            )
        _verify_extension_hashes(layout, changed_extensions)
        _verify_product_checksums(layout)

    _commit_plan(layout, plan, "install", validate)
    if manage_process:
        close_cursor(layout)
        start_cursor(layout)
    return 0


def uninstall(layout: CursorLayout, manage_process: bool = True) -> int:
    manage_process = _manage_processes_for_layout(layout, manage_process)
    before = inspect_status(layout)
    if before.external_marker_count:
        raise SandToolError(
            "检测到无法识别的 Sand 模式标记，拒绝修改；"
            "请先用原安装方式卸载"
        )
    plan, _stats = _build_uninstall_plan(layout)
    if not plan:
        if manage_process:
            start_cursor(layout)
        return 0

    if manage_process:
        close_cursor(layout)
    changed_extensions = _planned_extension_names(layout, plan)

    def validate() -> None:
        status = inspect_status(layout)
        if status.installed or status.external_marker_count:
            raise SandToolError(
                "卸载后仍有 Sand marker："
                f"{status.client_markers + status.eligibility_markers}，"
                f"external={status.external_marker_count}"
            )
        _verify_extension_hashes(layout, changed_extensions)
        _verify_product_checksums(layout)

    _commit_plan(layout, plan, "uninstall", validate)
    if manage_process:
        close_cursor(layout)
        start_cursor(layout)
    return 0


def _permission_hint() -> str:
    if getattr(sys, "frozen", False):
        script = Path(sys.executable).resolve()
    else:
        script = Path(__file__).resolve()
    if sys.platform == "win32":
        return "请右键以管理员身份打开 PowerShell/终端后重新运行命令。"
    return f'请使用管理员权限重试：sudo python3 "{script}" <命令>'


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description=f"{TOOL_NAME} - Cursor 客户端模式管理器",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "示例：\n"
            "  python sand_stream_installer.py install\n"
            "  python sand_stream_installer.py uninstall\n"
            "  python sand_stream_installer.py set-path \"E:\\Development\\IDE\\cursor\"\n"
            "  python3 sand_stream_installer.py set-path /Applications/Cursor.app\n"
            "  python3 sand_stream_installer.py set-path ~/.cursor-server/bin/linux-x64/<commit>\n"
            "  python sand_stream_installer.py set-path auto"
        ),
    )
    parser.add_argument("--version", action="version", version=f"%(prog)s {TOOL_VERSION}")
    commands = parser.add_subparsers(dest="command", required=True)
    commands.add_parser("install", help="安装/注入 Sand 客户端模式")
    commands.add_parser("uninstall", help="卸载 Sand 客户端模式")
    commands.add_parser("move-exec-check", help="检查 move_exec 补丁状态")
    commands.add_parser("move-exec-apply", help="单独应用 move_exec 补丁")
    commands.add_parser("move-exec-restore", help="单独还原 move_exec 补丁")
    commands.add_parser("license-status", help="查看本机卡密授权状态")
    activate = commands.add_parser("license-activate", help="输入/更换卡密并绑定本机")
    activate.add_argument("key", nargs="?", help="卡密，省略则交互输入")
    commands.add_parser("license-reset", help="清除本机保存的卡密")
    set_path = commands.add_parser("set-path", help="设置 Cursor 路径；auto 恢复自动检测")
    set_path.add_argument(
        "path",
        help="Cursor.exe、Cursor.app、cursor-server、resources/app、安装根目录，或 auto",
    )
    return parser


def collect_status_lines() -> List[Tuple[str, str]]:
    try:
        layout = resolve_cursor_layout()
        status = inspect_status(layout)
    except SandToolError as exc:
        return [(str(exc), ANSI_YELLOW)]

    lines: List[Tuple[str, str]] = [
        (f"[环境] Cursor {layout.version}  |  {layout.install_root}", ANSI_BLUE),
        (f"[适配] 仅支持 Cursor {SUPPORTED_CURSOR_VERSION}", ANSI_GREEN),
    ]
    if layout.version != SUPPORTED_CURSOR_VERSION:
        lines.append(
            (
                f"[警告] 当前版本 {layout.version} 不受支持，请勿安装",
                ANSI_RED,
            )
        )
    if status.installed:
        if _stream_mode_installed(layout, status):
            lines.append(("[状态] Stream 模式已启用", ANSI_GREEN))
        else:
            lines.append(("[状态] 检测到旧版客户端模式", ANSI_YELLOW))
    else:
        lines.append(("[状态] Cursor 原版模式", ANSI_YELLOW))
    if status.move_exec_markers:
        lines.append(("[move-exec] 工具执行补丁已启用", ANSI_GREEN))
    elif status.installed:
        lines.append(("[move-exec] 工具执行补丁未启用", ANSI_YELLOW))
    if status.plan_mode_markers:
        lines.append(("[plan-mode] Plan 模式 Sand 补丁已启用", ANSI_GREEN))
    elif status.installed:
        lines.append(("[plan-mode] Plan 模式 Sand 补丁未启用", ANSI_YELLOW))
    if status.action_gate_markers:
        lines.append(("[action-gate] Build/后台任务 Sand 补丁已启用", ANSI_GREEN))
    elif status.installed:
        lines.append(("[action-gate] Build/后台任务 Sand 补丁未启用", ANSI_YELLOW))
    if status.query_key_markers:
        lines.append(("[query-key] Plan 提问/多问题补丁已启用", ANSI_GREEN))
    elif status.installed:
        lines.append(("[query-key] Plan 提问/多问题补丁未启用", ANSI_YELLOW))
    if status.external_marker_count:
        lines.append(
            (f"[注意] 检测到其他工具标记：{status.external_marker_count} 处", ANSI_YELLOW)
        )
    return lines


def print_banner() -> None:
    width = 60
    print(colorize("=" * width, ANSI_BLUE))
    print(colorize(f"  {TOOL_NAME}  v{TOOL_VERSION}", ANSI_BOLD, ANSI_GREEN))
    print(colorize("  Cursor 客户端模式管理器", ANSI_BOLD))
    print(colorize("=" * width, ANSI_BLUE))
    for text, code in collect_status_lines():
        print(colorize(text, code))
    print(colorize(f"[授权] {sand_license.status_text()}", ANSI_BLUE))
    print(colorize("[账号] 请使用已开通 Sand 资格的 Cursor 账号", ANSI_YELLOW))
    print()


def apply_set_path(value: str) -> int:
    save_cursor_path(value)
    return 0


def _success_message(layout: CursorLayout, installed: bool) -> str:
    if sys.platform.startswith("linux") and _is_cursor_server_layout(layout):
        action = "Stream 模式已启用" if installed else "已恢复 Cursor 原版"
        return f"✓ {action}；请重新连接 Cursor Remote 使更改生效"
    if installed:
        return "✓ Stream 模式已启用，Cursor 已重新启动"
    return "✓ 已恢复 Cursor 原版，Cursor 已重新启动"


def print_menu() -> None:
    print(colorize("请选择操作：", ANSI_BOLD))
    print(colorize("  [1]", ANSI_BOLD, ANSI_GREEN) + " 启用 Stream 模式")
    print(colorize("  [2]", ANSI_BOLD, ANSI_GREEN) + " 恢复 Cursor 原版")
    print(colorize("  [3]", ANSI_BOLD, ANSI_GREEN) + " 指定 Cursor 路径")
    print(colorize("  [4]", ANSI_BOLD, ANSI_GREEN) + " 输入/更换卡密")
    print(colorize("  [0]", ANSI_BOLD, ANSI_BLUE) + " 退出")


def cmd_license_status() -> int:
    key = sand_license.load_saved_key()
    print(colorize(f"机器码：{sand_license.machine_code()}", ANSI_BLUE))
    print(colorize(f"平台：{sand_license.product_name()}    授权服务器：{sand_license.LICENSE_SERVER}", ANSI_BLUE))
    if not key:
        print_warn(f"本机尚未激活，购买地址：{sand_license.PURCHASE_URL}")
        return 1
    try:
        data = sand_license.validate(key, TOOL_VERSION)
    except sand_license.LicenseError as exc:
        print_error(f"卡密 {key} 校验失败：{exc}")
        return 2
    print_success(f"✓ 卡密 {key} 授权有效（激活于 {data.get('activated_at') or '—'}）")
    return 0


def cmd_license_activate(key: Optional[str]) -> int:
    if not key:
        try:
            key = input(colorize("请输入卡密> ", ANSI_BLUE)).strip()
        except (EOFError, KeyboardInterrupt):
            print()
            return 0
    if not key:
        print_warn("未输入卡密。")
        return 1
    try:
        sand_license.activate(key, TOOL_VERSION)
    except sand_license.LicenseError as exc:
        print_error(f"激活失败：{exc}")
        return 2
    print_success(f"✓ 激活成功，本机已绑定卡密 {sand_license.normalize_key(key)}")
    return 0


def cmd_license_reset() -> int:
    sand_license.clear_saved_key()
    print_success("✓ 已清除本机保存的卡密")
    return 0


def prompt_set_path() -> int:
    value = input(colorize("路径> ", ANSI_BLUE)).strip()
    if not value:
        return 0
    with LoadingSpinner("正在设置路径"):
        result = apply_set_path(value)
    print_success("✓ Cursor 路径已保存")
    return result


def run_choice(choice: str) -> Optional[int]:
    if choice == "1":
        layout = resolve_cursor_layout()
        with LoadingSpinner("正在启用 Stream 模式"):
            result = install(layout)
        print_success(_success_message(layout, True))
        return result
    if choice == "2":
        layout = resolve_cursor_layout()
        with LoadingSpinner("正在恢复 Cursor 原版"):
            result = uninstall(layout)
        print_success(_success_message(layout, False))
        return result
    if choice == "3":
        return prompt_set_path()
    if choice == "4":
        return cmd_license_activate(None)
    print_warn("无效选项，请输入 0-4。")
    return 0


def interactive_loop(skip_first_banner: bool = False) -> int:
    first = True
    while True:
        if not (first and skip_first_banner):
            print_banner()
        first = False
        print_menu()
        try:
            choice = input(colorize("请输入编号> ", ANSI_BLUE)).strip()
        except EOFError:
            print()
            return 0
        if choice == "0":
            print(colorize("已退出。", ANSI_BLUE))
            return 0
        try:
            run_choice(choice)
        except PermissionError as exc:
            print_error(f"错误：没有写入权限：{exc}")
            print_error(_permission_hint())
        except SandToolError as exc:
            print_error(f"错误：{exc}")
        except KeyboardInterrupt:
            print()
            return 0
        except Exception as exc:
            print_error(f"未预期错误：{exc}")
        print()


def main(argv: Optional[Sequence[str]] = None) -> int:
    _configure_console()
    args_list = list(sys.argv[1:] if argv is None else argv)
    try:
        _platform_name()
        if not args_list:
            print_banner()
            require_license()
            return interactive_loop(skip_first_banner=True)

        print_banner()
        args = build_parser().parse_args(args_list)
        if args.command == "license-status":
            return cmd_license_status()
        if args.command == "license-activate":
            return cmd_license_activate(args.key)
        if args.command == "license-reset":
            return cmd_license_reset()
        if args.command == "set-path":
            result = apply_set_path(args.path)
            print_success("✓ Cursor 路径已保存")
            return result

        require_license()
        layout = resolve_cursor_layout()
        if args.command == "install":
            result = install(layout)
            print_success(_success_message(layout, True))
            return result
        if args.command == "uninstall":
            result = uninstall(layout)
            print_success(_success_message(layout, False))
            return result
        if args.command == "move-exec-check":
            return cmd_move_exec_check(layout)
        if args.command == "move-exec-apply":
            return cmd_move_exec_apply(layout)
        if args.command == "move-exec-restore":
            return cmd_move_exec_restore(layout)
        raise SandToolError(f"未知命令：{args.command}")
    except PermissionError as exc:
        print_error(f"错误：没有写入权限：{exc}")
        print_error(_permission_hint())
        return 3
    except SandToolError as exc:
        print_error(f"错误：{exc}")
        return 2
    except KeyboardInterrupt:
        print_error("操作已取消。")
        return 130
    except Exception as exc:
        print_error(f"未预期错误：{exc}")
        return 1


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