feat(update): add transactional external updater

app/updater_entry.py

@@ -0,0 +1,364 @@
+"""在主程序退出后执行程序根项目事务替换的独立更新器。"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import shutil
+import subprocess
+import tempfile
+import time
+import uuid
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+
+
+PACKAGE_FORMAT = "cmshopee-portable-v1"
+UPDATER_PROTOCOL = 1
+ALLOWED_ROOTS = {
+    "cmshopee.exe",
+    "_internal",
+    "version.txt",
+    "README.txt",
+    "package-manifest.json",
+    "cmshopee-updater.exe",
+}
+PROTECTED_ROOTS = {"data", ".cmshopee-update"}
+
+
+class UpdaterError(RuntimeError):
+    """独立更新器拒绝计划或无法安全完成事务。"""
+
+
+@dataclass(frozen=True)
+class UpdatePlan:
+    parent_pid: int
+    install_root: Path
+    staging_root: Path
+    target_version: str
+    transaction_id: str
+    log_path: Path
+
+
+def create_plan(install_root, staging_root, target_version, parent_pid):
+    install_root = Path(install_root).resolve()
+    transaction_id = uuid.uuid4().hex
+    update_root = install_root / ".cmshopee-update"
+    plan = UpdatePlan(
+        parent_pid=int(parent_pid),
+        install_root=install_root,
+        staging_root=Path(staging_root).resolve(),
+        target_version=str(target_version),
+        transaction_id=transaction_id,
+        log_path=(update_root / "logs" / (transaction_id + ".log")).resolve(),
+    )
+    validate_plan(plan)
+    plan_path = update_root / "plans" / (transaction_id + ".json")
+    _atomic_json(
+        plan_path,
+        {
+            "parent_pid": plan.parent_pid,
+            "install_root": str(plan.install_root),
+            "staging_root": str(plan.staging_root),
+            "target_version": plan.target_version,
+            "transaction_id": plan.transaction_id,
+            "log_path": str(plan.log_path),
+        },
+    )
+    return plan_path
+
+
+def _is_child(path, parent):
+    try:
+        Path(path).resolve().relative_to(Path(parent).resolve())
+        return True
+    except ValueError:
+        return False
+
+
+def _atomic_json(path, payload):
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    os.replace(str(temporary), str(path))
+
+
+def _append_log(path, message):
+    Path(path).parent.mkdir(parents=True, exist_ok=True)
+    with Path(path).open("a", encoding="utf-8") as stream:
+        stream.write("%s %s\n" % (time.strftime("%Y-%m-%d %H:%M:%S"), message))
+
+
+def load_plan(plan_path):
+    plan_path = Path(plan_path).resolve()
+    try:
+        payload = json.loads(plan_path.read_text(encoding="utf-8"))
+        plan = UpdatePlan(
+            parent_pid=int(payload["parent_pid"]),
+            install_root=Path(payload["install_root"]).resolve(),
+            staging_root=Path(payload["staging_root"]).resolve(),
+            target_version=str(payload["target_version"]),
+            transaction_id=str(payload["transaction_id"]),
+            log_path=Path(payload["log_path"]).resolve(),
+        )
+    except (OSError, ValueError, KeyError, TypeError) as exc:
+        raise UpdaterError("更新计划无效") from exc
+    validate_plan(plan, plan_path)
+    return plan
+
+
+def validate_plan(plan, plan_path=None):
+    if plan.parent_pid <= 0:
+        raise UpdaterError("更新计划缺少主程序进程编号")
+    if not plan.install_root.is_absolute() or not plan.install_root.is_dir():
+        raise UpdaterError("程序安装目录无效")
+    update_root = plan.install_root / ".cmshopee-update"
+    staging_parent = update_root / "staging"
+    if not plan.staging_root.is_dir() or plan.staging_root.parent != staging_parent.resolve():
+        raise UpdaterError("新版暂存目录越界")
+    if not re.fullmatch(r"\d+(?:\.\d+)*", plan.target_version):
+        raise UpdaterError("目标版本号格式不正确")
+    if not re.fullmatch(r"[A-Za-z0-9-]{8,64}", plan.transaction_id):
+        raise UpdaterError("更新事务编号无效")
+    if not _is_child(plan.log_path, update_root / "logs"):
+        raise UpdaterError("更新日志路径越界")
+    if plan_path is not None and not _is_child(plan_path, update_root):
+        raise UpdaterError("更新计划文件路径越界")
+    probe = update_root / (".write-probe-" + plan.transaction_id)
+    try:
+        probe.parent.mkdir(parents=True, exist_ok=True)
+        probe.write_bytes(b"")
+        probe.unlink()
+    except OSError as exc:
+        raise UpdaterError("程序安装目录不可写") from exc
+
+
+def _sha256(path):
+    digest = hashlib.sha256()
+    with Path(path).open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def validate_staging(plan):
+    manifest_path = plan.staging_root / "package-manifest.json"
+    try:
+        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    except (OSError, ValueError) as exc:
+        raise UpdaterError("新版程序清单无效") from exc
+    if (
+        manifest.get("package_format") != PACKAGE_FORMAT
+        or manifest.get("app_version") != plan.target_version
+        or manifest.get("entry_point") != "cmshopee.exe"
+        or int(manifest.get("updater_protocol") or 0) != UPDATER_PROTOCOL
+    ):
+        raise UpdaterError("新版程序清单与更新计划不一致")
+    roots = manifest.get("replace_roots")
+    if not isinstance(roots, list) or not roots:
+        raise UpdaterError("新版程序替换范围无效")
+    roots = list(dict.fromkeys(str(root) for root in roots))
+    if any(root not in ALLOWED_ROOTS or root in PROTECTED_ROOTS for root in roots):
+        raise UpdaterError("新版程序包含未授权替换范围")
+    if "cmshopee.exe" not in roots or "_internal" not in roots:
+        raise UpdaterError("新版程序缺少必要替换项目")
+
+    declared = {}
+    for item in manifest.get("files") or []:
+        relative = PurePosixPath(str(item.get("path") or ""))
+        if not relative.parts or ".." in relative.parts or relative.parts[0] not in roots:
+            raise UpdaterError("新版程序清单包含越界文件")
+        key = relative.as_posix().casefold()
+        if key in declared:
+            raise UpdaterError("新版程序清单包含重复文件")
+        declared[key] = (relative, int(item.get("size_bytes") or -1), str(item.get("sha256") or ""))
+    actual = {}
+    for path in plan.staging_root.rglob("*"):
+        if path.is_symlink():
+            raise UpdaterError("新版程序不允许符号链接")
+        if path.is_file() and path.name != "package-manifest.json":
+            relative = PurePosixPath(path.relative_to(plan.staging_root).as_posix())
+            actual[relative.as_posix().casefold()] = relative
+    if set(actual) != set(declared):
+        raise UpdaterError("新版程序文件与清单不一致")
+    for key, (relative, size_bytes, expected_hash) in declared.items():
+        path = plan.staging_root.joinpath(*relative.parts)
+        if path.stat().st_size != size_bytes or _sha256(path).lower() != expected_hash.lower():
+            raise UpdaterError("新版程序文件校验失败")
+    if (plan.staging_root / "version.txt").read_text(encoding="utf-8-sig").strip() != plan.target_version:
+        raise UpdaterError("新版程序版本文件不一致")
+    return roots
+
+
+def _pid_running(pid):
+    if pid <= 0:
+        return False
+    if os.name == "nt":
+        import ctypes
+
+        handle = ctypes.windll.kernel32.OpenProcess(0x1000, False, pid)
+        if handle:
+            ctypes.windll.kernel32.CloseHandle(handle)
+            return True
+        return False
+    try:
+        os.kill(pid, 0)
+        return True
+    except OSError:
+        return False
+
+
+def wait_for_parent_exit(pid, timeout=60, poll_seconds=0.2):
+    deadline = time.monotonic() + timeout
+    while _pid_running(pid):
+        if time.monotonic() >= deadline:
+            raise UpdaterError("等待旧版程序退出超时")
+        time.sleep(poll_seconds)
+
+
+class TransactionLock:
+    def __init__(self, path):
+        self.path = Path(path)
+        self.fd = None
+
+    def __enter__(self):
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            self.fd = os.open(str(self.path), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
+            os.write(self.fd, str(os.getpid()).encode("ascii"))
+        except FileExistsError as exc:
+            raise UpdaterError("已有更新程序正在处理当前安装目录") from exc
+        return self
+
+    def __exit__(self, *_args):
+        if self.fd is not None:
+            os.close(self.fd)
+        try:
+            self.path.unlink()
+        except FileNotFoundError:
+            pass
+
+
+def _default_launch(executable, args):
+    creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0)
+    return subprocess.Popen([str(executable), *args], close_fds=True, creationflags=creationflags)
+
+
+def apply_update(plan, *, wait_parent=wait_for_parent_exit, move=os.replace, launcher=_default_launch):
+    validate_plan(plan)
+    roots = validate_staging(plan)
+    update_root = plan.install_root / ".cmshopee-update"
+    old_version_path = plan.install_root / "version.txt"
+    old_version = old_version_path.read_text(encoding="utf-8-sig").strip() if old_version_path.exists() else "unknown"
+    backup_root = update_root / "backup" / (old_version + "-" + plan.transaction_id)
+    failed_root = update_root / "failed" / plan.transaction_id
+    journal_path = update_root / "transactions" / (plan.transaction_id + ".json")
+    operations = []
+
+    def journal(status, error=""):
+        _atomic_json(
+            journal_path,
+            {
+                "transaction_id": plan.transaction_id,
+                "target_version": plan.target_version,
+                "status": status,
+                "operations": operations,
+                "error": error,
+            },
+        )
+
+    with TransactionLock(update_root / "update.lock"):
+        wait_parent(plan.parent_pid)
+        backup_root.mkdir(parents=True, exist_ok=False)
+        journal("moving_old")
+        try:
+            for root in roots:
+                source = plan.install_root / root
+                if source.exists():
+                    destination = backup_root / root
+                    destination.parent.mkdir(parents=True, exist_ok=True)
+                    move(str(source), str(destination))
+                    operations.append({"kind": "old_to_backup", "root": root})
+                    journal("moving_old")
+            for root in roots:
+                source = plan.staging_root / root
+                if not source.exists():
+                    raise UpdaterError("新版程序缺少替换项目：%s" % root)
+                destination = plan.install_root / root
+                move(str(source), str(destination))
+                operations.append({"kind": "new_to_install", "root": root})
+                journal("moving_new")
+            launcher(
+                plan.install_root / "cmshopee.exe",
+                ["--update-transaction", plan.transaction_id, "--update-target-version", plan.target_version],
+            )
+            journal("launched")
+            _append_log(plan.log_path, "新版程序已替换并启动：%s" % plan.target_version)
+            return backup_root
+        except Exception as exc:
+            journal("rolling_back", str(exc))
+            failed_root.mkdir(parents=True, exist_ok=True)
+            rollback_errors = []
+            for operation in reversed(operations):
+                root = operation["root"]
+                try:
+                    if operation["kind"] == "new_to_install":
+                        source = plan.install_root / root
+                        if source.exists():
+                            destination = failed_root / root
+                            destination.parent.mkdir(parents=True, exist_ok=True)
+                            move(str(source), str(destination))
+                    else:
+                        source = backup_root / root
+                        if source.exists():
+                            move(str(source), str(plan.install_root / root))
+                except Exception as rollback_exc:
+                    rollback_errors.append(str(rollback_exc))
+            status = "rollback_failed" if rollback_errors else "rolled_back"
+            journal(status, "；".join(rollback_errors) or str(exc))
+            _append_log(plan.log_path, "更新失败，已尝试恢复旧版：%s" % exc)
+            if rollback_errors:
+                raise UpdaterError("更新失败且旧版恢复不完整，请联系管理员") from exc
+            raise UpdaterError("更新失败，已恢复旧版程序") from exc
+
+
+def copy_and_launch_updater(updater_path, plan_path):
+    updater_path = Path(updater_path).resolve()
+    if not updater_path.is_file():
+        raise UpdaterError("独立更新程序不存在")
+    temporary_dir = Path(tempfile.mkdtemp(prefix="cmshopee-update-"))
+    external_updater = temporary_dir / updater_path.name
+    shutil.copy2(str(updater_path), str(external_updater))
+    creationflags = getattr(subprocess, "CREATE_NO_WINDOW", 0) | getattr(
+        subprocess, "DETACHED_PROCESS", 0
+    )
+    return subprocess.Popen(
+        [str(external_updater), "--plan", str(Path(plan_path).resolve())],
+        close_fds=True,
+        creationflags=creationflags,
+    )
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description="蝦皮圈优化助手独立更新程序")
+    parser.add_argument("--plan", required=True)
+    args = parser.parse_args(argv)
+    try:
+        plan = load_plan(args.plan)
+        apply_update(plan)
+        return 0
+    except Exception as exc:
+        try:
+            if "plan" in locals():
+                _append_log(plan.log_path, "自动更新失败：%s" % exc)
+        except OSError:
+            pass
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

cmshopee-updater.spec

@@ -0,0 +1,28 @@
+# -*- mode: python ; coding: utf-8 -*-
+
+a = Analysis(
+    ["app/updater_entry.py"],
+    pathex=[],
+    binaries=[],
+    datas=[],
+    hiddenimports=[],
+    hookspath=[],
+    hooksconfig={},
+    runtime_hooks=[],
+    excludes=["PySide6"],
+    noarchive=False,
+)
+pyz = PYZ(a.pure)
+exe = EXE(
+    pyz,
+    a.scripts,
+    a.binaries,
+    a.datas,
+    [],
+    name="cmshopee-updater",
+    debug=False,
+    bootloader_ignore_signals=False,
+    strip=False,
+    upx=True,
+    console=False,
+)

docs/04-architecture.md

@@ -500,6 +500,7 @@ cmshopee/
 │   ├── editor.py / ai.py / prompts.py / gui.py / workers.py
 │   ├── release_manifest.py   # 发布包文件清单、zip哈希与服务端元数据模板
 │   ├── update_installer.py   # 自动升级安全下载、解压、manifest校验与同盘暂存
+│   ├── updater_entry.py      # 独立更新器入口、事务根项目切换、journal与回滚
 ├── main.py                    # GUI 启动入口：from app.gui import main
 ├── shopee待处理任务模板.xlsx   # 标准空 Excel 模板，可提交；业务填写后的副本不提交
 ├── data/                     # 用户本地数据根（整体 gitignore；打包更新时保留）
@@ -529,6 +530,7 @@ cmshopee/
 - 发布包只能包含程序根项目，`data/` 和 `.cmshopee-update/` 永远在替换边界之外。第一阶段预留签名字段，但 SHA-256 只负责传输完整性，不等同于发布者身份认证。
 - T-615 只提供可验证发布契约；启动门禁仍保持 T-544 的人工下载行为，直到后续下载、独立更新器、事务替换和失败熔断任务全部接入。
 - T-616 的下载暂存根固定为安装目录下 `.cmshopee-update/`，与 `data/` 完全隔离。远程zip必须经过HTTPS/受信任域名、声明大小、整包SHA-256、安全zip路径和包内manifest逐文件校验，才写 `pending.json`；此阶段不替换任何运行中程序文件。
+- T-617 的独立 `cmshopee-updater.exe` 必须先复制到系统临时目录运行，并等待主程序退出。替换粒度是manifest允许的程序根项目，旧根先整体移动到同盘backup，新根再整体移入；事务锁防止并发更新，journal记录每次移动，任一步失败逆序恢复。`data/`、更新管理目录与未知安装根项目永不进入替换清单。
 
 - CDP 交互事实变化同步第七节。
 - 正式代码只放 `app/` 包；根目录只保留 `main.py`、配置/数据目录、文档和原型目录，不新增正式业务模块。

docs/packaging.md

@@ -178,6 +178,8 @@ T-615 不改变客户端行为：T-544 仍只打开浏览器下载。客户端
 
 T-616 提供无Qt依赖的安全下载暂存层 `app/update_installer.py`。它只接受受信任域名的HTTPS，下载到 `<安装目录>/.cmshopee-update/downloads/`，完成大小与zip SHA-256校验后安全解压到同盘 `staging/`；路径穿越、符号链接、大小写重复、Windows保留名、ADS、异常压缩比、文件数/解压总量超限，以及manifest缺失或逐文件hash不一致都会拒绝。完整验证后才原子写 `pending.json`，当前程序目录和 `data/` 不变。
 
+T-617 增加独立 `cmshopee-updater.exe`。构建脚本用 `cmshopee-updater.spec` 生成无控制台单文件更新器并放进release和manifest。执行更新前，主程序把它复制到系统临时目录；更新器有上限地等待主程序退出，再按manifest白名单将旧程序根项目整体移动到 `.cmshopee-update/backup/`，把已验证暂存根项目移入安装目录。每一步写事务journal，移动或新版启动失败时逆序恢复；`data/` 与安装根未知文件不扫描、不移动、不删除。
+
 ## 四、绝不打包的本地数据
 
 发布包里不能包含以下本地数据、密钥、业务数据或登录态：

docs/tasks/T-617.md

@@ -3,7 +3,7 @@ id: T-617
 title: 独立更新器事务替换与失败恢复
 phase: 8
 deps: [T-615]
-status: TODO
+status: DONE
 created: 2026-07-13
 ---
 
@@ -45,4 +45,6 @@ Windows正在运行的 `cmshopee.exe` 和PyInstaller `_internal/` 可能被锁
 
 ## 执行记录
 
-（完成后记录实现、验证命令与结果。）
+- 2026-07-13：新增标准库独立更新器入口、结构化plan、父进程退出等待、安装目录事务锁、根项目整体切换、逐步journal、故障逆序回滚及临时目录隐藏启动能力。
+- 2026-07-13：新增无控制台单文件 `cmshopee-updater.spec`，构建脚本将更新器纳入release与manifest；Windows 10 / Python 3.10 / PyInstaller 6.11.1实际构建成功，产物约6.7MB且使用windowed bootloader。
+- 2026-07-13：干净worktree验证通过：ruff、compileall、完整unittest（408项）和 `git diff --check`。尚未在正式release上人工执行“锁住运行中主程序后自动退出、替换、重启”的端到端验收，该项留给T-618/T-619集成后执行。

docs/update-check.md

@@ -49,6 +49,8 @@
 
 T-616 已实现但尚未接入GUI的安全暂存层：仅接受受信任域名的 HTTPS 地址，流式下载到安装目录 `.cmshopee-update/`，校验zip大小和SHA-256，安全解压后再按包内manifest逐文件校验。任一步失败都不修改当前程序或 `data/`；GUI接入由T-618完成。
 
+T-617 已提供独立无控制台更新器和事务回滚能力：更新器从系统临时目录运行，旧主程序退出后才切换程序根项目，并在新版进程无法创建时恢复旧版。启动检查弹窗尚未调用该流程，直到T-618完成GUI编排。
+
 ## 五、发版约定（服务端据此控制）
 
 自动安装使用的全部字段必须放在同一个 `release` 对象内，不得把版本取自一个对象、hash 取自另一个对象。构建脚本生成的 `release/release-metadata.json` 是服务端录入模板；发布人员只补 HTTPS `download_url`、强制策略和中文发布说明，不得手工改写 hash、大小、包格式或协议版本。

scripts/build_exe.ps1

@@ -120,6 +120,18 @@ if ($LASTEXITCODE -ne 0) {
     throw "$(Decode-Utf8Base64 "UHlJbnN0YWxsZXIg5omT5YyF5aSx6LSl77yM6YCA5Ye656CB"): $LASTEXITCODE"
 }
 
+$updaterSpec = Join-Path $repoRoot "cmshopee-updater.spec"
+$updaterDist = Join-Path $repoRoot "dist\updater"
+$updaterWork = Join-Path $repoRoot "build\updater"
+& $pythonExe @pythonArgs -m PyInstaller --noconfirm --clean --distpath $updaterDist --workpath $updaterWork $updaterSpec
+if ($LASTEXITCODE -ne 0) {
+    throw "独立更新器打包失败，退出码: $LASTEXITCODE"
+}
+$updaterExe = Join-Path $updaterDist "cmshopee-updater.exe"
+if (-not (Test-Path -LiteralPath $updaterExe)) {
+    throw "独立更新器产物不存在: $updaterExe"
+}
+
 $distDir = Join-Path $repoRoot "dist\cmshopee"
 $exePath = Join-Path $distDir "cmshopee.exe"
 if (-not (Test-Path -LiteralPath $exePath)) {
@@ -164,6 +176,7 @@ foreach ($path in @($releaseDir, $portableZip, $releaseMetadata)) {
 
 New-Item -ItemType Directory -Force -Path $releaseDir | Out-Null
 Get-ChildItem -LiteralPath $distDir -Force | Copy-Item -Destination $releaseDir -Recurse -Force
+Copy-Item -LiteralPath $updaterExe -Destination (Join-Path $releaseDir "cmshopee-updater.exe") -Force
 Set-Content -LiteralPath (Join-Path $releaseDir "version.txt") -Value $appVersion -Encoding ascii -NoNewline
 
 $readme = @(

tests/test_packaging.py

@@ -96,6 +96,17 @@ class PackagingTests(unittest.TestCase):
         self.assertIn('-m app.release_manifest manifest', script)
         self.assertIn('-m app.release_manifest metadata', script)
         self.assertIn('release-metadata.json', script)
+        self.assertIn('cmshopee-updater.spec', script)
+        self.assertIn('cmshopee-updater.exe', script)
+
+    def test_updater_spec_builds_a_windowed_standalone_executable(self):
+        spec = self.read_text("cmshopee-updater.spec")
+        normalized = "".join(spec.split())
+
+        self.assertIn('app/updater_entry.py', spec)
+        self.assertIn('name="cmshopee-updater"', normalized)
+        self.assertIn('console=False', normalized)
+        self.assertNotIn("COLLECT(", spec)
 
     def make_release_dir(self, root, app_version="1.2.3"):
         release_dir = Path(root) / "release"

tests/test_updater_entry.py

@@ -0,0 +1,152 @@
+import hashlib
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+from app import release_manifest, updater_entry
+
+
+class UpdaterEntryTests(unittest.TestCase):
+    def make_trees(self, root):
+        install = Path(root) / "install"
+        staging = install / ".cmshopee-update" / "staging" / "2.0.0-test"
+        install.mkdir()
+        (install / "_internal").mkdir()
+        (install / "cmshopee.exe").write_bytes(b"old-exe")
+        (install / "_internal" / "old.dll").write_bytes(b"old")
+        (install / "version.txt").write_text("1.0.0", encoding="ascii")
+        (install / "README.txt").write_text("旧说明", encoding="utf-8")
+        (install / "package-manifest.json").write_text("{}", encoding="utf-8")
+        (install / "cmshopee-updater.exe").write_bytes(b"old-updater")
+        (install / "data").mkdir()
+        (install / "data" / "cmshopee.db").write_bytes(b"business-data")
+        (install / "operator-note.txt").write_text("保留", encoding="utf-8")
+
+        (staging / "_internal").mkdir(parents=True)
+        (staging / "cmshopee.exe").write_bytes(b"new-exe")
+        (staging / "_internal" / "new.dll").write_bytes(b"new")
+        (staging / "version.txt").write_text("2.0.0", encoding="ascii")
+        (staging / "README.txt").write_text("新说明", encoding="utf-8")
+        (staging / "cmshopee-updater.exe").write_bytes(b"new-updater")
+        release_manifest.write_package_manifest(staging, "2.0.0")
+        plan = updater_entry.UpdatePlan(
+            parent_pid=12345,
+            install_root=install.resolve(),
+            staging_root=staging.resolve(),
+            target_version="2.0.0",
+            transaction_id="transaction-1234",
+            log_path=(install / ".cmshopee-update/logs/update.log").resolve(),
+        )
+        return install, staging, plan
+
+    def test_transaction_replaces_roots_and_preserves_data_and_unknown_files(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            install, _staging, plan = self.make_trees(temp_dir)
+            before_hash = hashlib.sha256((install / "data/cmshopee.db").read_bytes()).hexdigest()
+            launched = []
+
+            backup = updater_entry.apply_update(
+                plan,
+                wait_parent=lambda _pid: None,
+                launcher=lambda executable, args: launched.append((executable, args)),
+            )
+
+            self.assertEqual(b"new-exe", (install / "cmshopee.exe").read_bytes())
+            self.assertTrue((install / "_internal/new.dll").is_file())
+            self.assertFalse((install / "_internal/old.dll").exists())
+            self.assertEqual("保留", (install / "operator-note.txt").read_text(encoding="utf-8"))
+            self.assertEqual(
+                before_hash,
+                hashlib.sha256((install / "data/cmshopee.db").read_bytes()).hexdigest(),
+            )
+            self.assertTrue((backup / "cmshopee.exe").is_file())
+            self.assertEqual(1, len(launched))
+
+    def test_move_failure_rolls_back_old_program(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            install, _staging, plan = self.make_trees(temp_dir)
+            calls = {"count": 0}
+
+            def failing_move(source, destination):
+                calls["count"] += 1
+                if calls["count"] == 8:
+                    raise OSError("injected move failure")
+                return os.replace(source, destination)
+
+            with self.assertRaisesRegex(updater_entry.UpdaterError, "已恢复旧版"):
+                updater_entry.apply_update(
+                    plan,
+                    wait_parent=lambda _pid: None,
+                    move=failing_move,
+                    launcher=lambda *_args: None,
+                )
+            self.assertEqual(b"old-exe", (install / "cmshopee.exe").read_bytes())
+            self.assertTrue((install / "_internal/old.dll").is_file())
+            self.assertEqual(b"business-data", (install / "data/cmshopee.db").read_bytes())
+
+    def test_parent_timeout_keeps_install_unchanged(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            install, _staging, plan = self.make_trees(temp_dir)
+
+            def timeout(_pid):
+                raise updater_entry.UpdaterError("等待旧版程序退出超时")
+
+            with self.assertRaisesRegex(updater_entry.UpdaterError, "退出超时"):
+                updater_entry.apply_update(plan, wait_parent=timeout)
+            self.assertEqual(b"old-exe", (install / "cmshopee.exe").read_bytes())
+
+    def test_new_process_launch_failure_rolls_back(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            install, _staging, plan = self.make_trees(temp_dir)
+
+            def fail_launch(*_args):
+                raise OSError("injected launch failure")
+
+            with self.assertRaisesRegex(updater_entry.UpdaterError, "已恢复旧版"):
+                updater_entry.apply_update(
+                    plan,
+                    wait_parent=lambda _pid: None,
+                    launcher=fail_launch,
+                )
+            self.assertEqual(b"old-exe", (install / "cmshopee.exe").read_bytes())
+            self.assertTrue((install / "_internal/old.dll").is_file())
+
+    def test_lock_blocks_concurrent_transaction(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            install, _staging, plan = self.make_trees(temp_dir)
+            lock_path = install / ".cmshopee-update/update.lock"
+            with updater_entry.TransactionLock(lock_path):
+                with self.assertRaisesRegex(updater_entry.UpdaterError, "已有更新程序"):
+                    updater_entry.apply_update(plan, wait_parent=lambda _pid: None)
+
+    def test_plan_path_and_staging_must_stay_inside_update_root(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            install, _staging, plan = self.make_trees(temp_dir)
+            escaped = updater_entry.UpdatePlan(
+                parent_pid=plan.parent_pid,
+                install_root=install,
+                staging_root=Path(temp_dir) / "outside",
+                target_version=plan.target_version,
+                transaction_id=plan.transaction_id,
+                log_path=plan.log_path,
+            )
+            escaped.staging_root.mkdir()
+            with self.assertRaisesRegex(updater_entry.UpdaterError, "暂存目录越界"):
+                updater_entry.validate_plan(escaped)
+
+    def test_create_and_load_plan_round_trip(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            install, staging, _plan = self.make_trees(temp_dir)
+            plan_path = updater_entry.create_plan(install, staging, "2.0.0", 4321)
+            loaded = updater_entry.load_plan(plan_path)
+            payload = json.loads(plan_path.read_text(encoding="utf-8"))
+
+            self.assertEqual("2.0.0", loaded.target_version)
+            self.assertEqual(4321, payload["parent_pid"])
+            self.assertNotIn("data", payload)
+
+
+if __name__ == "__main__":
+    unittest.main()