From 43fd8ce66812d244389bb63d154383d5a5a0dbcd Mon Sep 17 00:00:00 2001
From: QiuSW <105186638@qq.com>
Date: Wed, 5 Aug 2026 17:25:27 +0800
Subject: [PATCH] feat(client): capture T-104 safe exit evidence

---
 client/scripts/capture_sku_exit_spike.py      |  85 ++++
 .../pdd/sku_selection_runner.py               | 307 +++++++++++-
 client/tests/pdd/test_sku_selection.py        | 441 +++++++++++++++++-
 3 files changed, 830 insertions(+), 3 deletions(-)
 create mode 100644 client/scripts/capture_sku_exit_spike.py

diff --git a/client/scripts/capture_sku_exit_spike.py b/client/scripts/capture_sku_exit_spike.py
new file mode 100644
index 0000000..4577fe6
--- /dev/null
+++ b/client/scripts/capture_sku_exit_spike.py
@@ -0,0 +1,85 @@
+"""采集 T-104 阶段 A 的一次 Back 后本机证据。"""
+
+from __future__ import annotations
+
+import argparse
+from math import isfinite
+from pathlib import Path
+import sys
+
+
+CLIENT_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(CLIENT_ROOT / "src"))
+
+from cmbuyer_client.device.adb import AdbClient, DeviceConnectionError, SubprocessAdbRunner
+from cmbuyer_client.device.baseline import NoReconnectUiautomatorConnector
+from cmbuyer_client.pdd.sku_selection import SkuSelectionError
+from cmbuyer_client.pdd.sku_selection_runner import (
+    SkuExitSpikeCapturer,
+    SkuSelectionRunError,
+)
+
+
+def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="采集 T-104 阶段 A 的一次 Back 后本机证据。")
+    parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。")
+    parser.add_argument("--output-dir", required=True, type=Path, help="全新本机证据目录；不得覆盖。")
+    parser.add_argument("--timeout", type=float, default=10.0, help="ADB 与设备 RPC 超时（秒）。")
+    parser.add_argument("--adb", default="adb", help="adb 可执行文件路径。")
+    return parser.parse_args(argv)
+
+
+def validate_arguments(arguments: argparse.Namespace) -> None:
+    if type(arguments.serial) is not str or not arguments.serial.strip():
+        raise ValueError("必须显式提供非空 --serial。")
+    if not isinstance(arguments.output_dir, Path) or not arguments.output_dir.name:
+        raise ValueError("--output-dir 必须是明确的全新目录。")
+    if (
+        not isinstance(arguments.timeout, (int, float))
+        or isinstance(arguments.timeout, bool)
+        or arguments.timeout <= 0
+        or not isfinite(arguments.timeout)
+    ):
+        raise ValueError("--timeout 必须是大于 0 的有限数值。")
+
+
+def main(argv: list[str] | None = None) -> int:
+    arguments = parse_arguments(argv)
+    try:
+        validate_arguments(arguments)
+    except ValueError as error:
+        print(f"失败：{error}", file=sys.stderr)
+        return 2
+
+    try:
+        import adbutils
+        import uiautomator2 as u2
+    except ImportError:
+        print("失败：缺少 uiautomator2；请在采购工具虚拟环境中运行。", file=sys.stderr)
+        return 2
+
+    capturer = SkuExitSpikeCapturer(
+        AdbClient(SubprocessAdbRunner(arguments.adb), timeout_seconds=arguments.timeout),
+        NoReconnectUiautomatorConnector(
+            adbutils.AdbClient(socket_timeout=arguments.timeout).device_list,
+            u2.connect,
+        ),
+        timeout_seconds=arguments.timeout,
+    )
+    try:
+        capturer.capture(arguments.serial, arguments.output_dir)
+    except (DeviceConnectionError, SkuSelectionError, SkuSelectionRunError):
+        # 不回显第三方异常、serial、页面正文或本机路径。
+        print("规格安全退出取证失败：已停止，未发布本地证据目录。", file=sys.stderr)
+        return 1
+    except OSError:
+        print("规格安全退出取证失败：无法发布本地证据目录。", file=sys.stderr)
+        return 1
+
+    print("规格安全退出取证完成。")
+    print("人工复核：请在指定目录检查退出后截图、XML、应用摘要和 manifest。")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/client/src/cmbuyer_client/pdd/sku_selection_runner.py b/client/src/cmbuyer_client/pdd/sku_selection_runner.py
index 763d99c..365458d 100644
--- a/client/src/cmbuyer_client/pdd/sku_selection_runner.py
+++ b/client/src/cmbuyer_client/pdd/sku_selection_runner.py
@@ -21,7 +21,13 @@ from adbutils.errors import AdbTimeout
 from uiautomator2.exceptions import HTTPTimeoutError
 
 from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection
-from ..device.baseline import PDD_PACKAGE, SCREENSHOT_PARAMS, _save_base64_screenshot, _sha256_file
+from ..device.baseline import (
+    PDD_PACKAGE,
+    SCREENSHOT_PARAMS,
+    _save_base64_screenshot,
+    _sha256_file,
+    _validate_hierarchy,
+)
 from .product_open import EXPECTED_PDD_VERSION
 from .product_url import ProductUrl, ProductUrlError, parse_product_url
 from .sku_selection import (
@@ -37,7 +43,9 @@ from .sku_selection import (
     _TARGET_SIZE_BOUNDS,
     _action_bounds,
     _annotate_sku_entry_failure,
+    _parse_nodes,
     _safe_sku_entry_failure_stage,
+    _unit_price,
     resolve_task_selection,
 )
 
@@ -86,6 +94,10 @@ class SkuSelectionDeviceAdapterError(SkuSelectionRunError):
     """第三方设备接口失败的脱敏映射。"""
 
 
+class SkuExitSpikeError(SkuSelectionRunError):
+    """T-104 阶段 A 未形成完整的本机退出证据。"""
+
+
 def safe_failure_stage(error: BaseException) -> str:
     """返回允许公开的固定阶段码，绝不回显异常正文。"""
 
@@ -138,6 +150,14 @@ class SkuSelectionRunResult:
     unit_price: str
 
 
+@dataclass(frozen=True)
+class SkuExitSpikeResult:
+    """已原子发布、仍需人工判断页面身份的 T-104 证据。"""
+
+    output_directory: Path
+    manifest_path: Path
+
+
 class UiautomatorSkuPanelAdapter(SkuPanelDevice):
     """把 uiautomator2 缩为 T-103 所需的读取与四种命名操作。
 
@@ -292,6 +312,210 @@ class UiautomatorSkuPanelAdapter(SkuPanelDevice):
             raise SkuSelectionDeviceAdapterError("规格面板设备操作失败，已停止操作。") from error
 
 
+class UiautomatorSkuExitAdapter:
+    """T-104 阶段 A 的窄设备边界：只读能力加唯一一次命名 Back。
+
+    取证脚本不能取得 T-103 的入口、规格选项或 reveal 方法。Back 的唯一机会在
+    JSON-RPC 前封存，因为超时无法证明事件没有送达，任何结果不明都不得重发。
+    """
+
+    def __init__(self, device: Any, timeout_seconds: float) -> None:
+        if not _is_positive_finite(timeout_seconds):
+            raise ValueError("timeout_seconds 必须是大于 0 的有限数值")
+        self._device = device
+        self._timeout_seconds = timeout_seconds
+        self._back_attempted = False
+        self._back_rpc_outcome = "not_attempted"
+
+    @property
+    def back_attempts(self) -> int:
+        return int(self._back_attempted)
+
+    @property
+    def back_rpc_outcome(self) -> str:
+        return self._back_rpc_outcome
+
+    def app_info(self, package_name: str) -> dict[str, Any]:
+        value = self._call("app_info", package_name)
+        if not isinstance(value, dict):
+            raise SkuSelectionDeviceAdapterError("无法读取应用版本，已停止取证。")
+        return value
+
+    def app_current(self) -> dict[str, Any]:
+        value = self._call("app_current")
+        if not isinstance(value, dict):
+            raise SkuSelectionDeviceAdapterError("无法读取前台应用，已停止取证。")
+        return value
+
+    def dump_window_hierarchy(self) -> str:
+        value = self._call(
+            "jsonrpc_call",
+            "dumpWindowHierarchy",
+            [False, 50],
+            timeout=self._timeout_seconds,
+        )
+        if not isinstance(value, str):
+            raise SkuSelectionDeviceAdapterError("节点树读取失败，已停止取证。")
+        return value
+
+    def capture_screenshot(self) -> str:
+        value = self._call(
+            "jsonrpc_call",
+            "takeScreenshot",
+            SCREENSHOT_PARAMS,
+            timeout=self._timeout_seconds,
+        )
+        if not isinstance(value, str):
+            raise SkuSelectionScreenshotError("退出后截图读取失败，未发布任何证据产物。")
+        return value
+
+    def display_size(self) -> tuple[int, int]:
+        value = self._call("window_size")
+        if (
+            not isinstance(value, tuple)
+            or len(value) != 2
+            or any(not isinstance(item, int) for item in value)
+        ):
+            raise SkuSelectionDeviceAdapterError("无法读取屏幕坐标空间，已停止取证。")
+        return value
+
+    def leave_sku_panel(self) -> None:
+        if self._back_attempted:
+            raise SkuSelectionDeviceAdapterError("本次取证已经尝试过返回，拒绝重试。")
+        self._back_attempted = True
+        self._back_rpc_outcome = "ambiguous"
+        self._call(
+            "jsonrpc_call",
+            "pressKey",
+            ["back"],
+            timeout=self._timeout_seconds,
+        )
+        self._back_rpc_outcome = "completed"
+
+    def _call(self, method: str, *args: Any, **kwargs: Any) -> Any:
+        try:
+            operation = getattr(self._device, method)
+            return operation(*args, **kwargs)
+        except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
+            raise SkuSelectionRunTimeoutError("T-104 设备操作超时，已停止操作。") from error
+        except SkuSelectionRunError:
+            raise
+        except Exception as error:
+            raise SkuSelectionDeviceAdapterError("T-104 设备操作失败，已停止操作。") from error
+
+
+class SkuExitSpikeCapturer:
+    """从人工停驻的已验证目标面板执行一次 Back，再只读采集阶段 A 证据。"""
+
+    def __init__(
+        self,
+        adb_client: AdbClient,
+        connector: Callable[[str], Any],
+        timeout_seconds: float,
+    ) -> None:
+        if not _is_positive_finite(timeout_seconds):
+            raise ValueError("timeout_seconds 必须是大于 0 的有限数值")
+        self._adb_client = adb_client
+        self._connector = connector
+        self._timeout_seconds = timeout_seconds
+        self._started = False
+
+    def capture(self, serial: str, output_directory: Path) -> SkuExitSpikeResult:
+        if self._started:
+            raise SkuExitSpikeError("同一取证器不可重复调用。")
+        self._started = True
+
+        staging: Path | None = None
+        adapter: UiautomatorSkuExitAdapter | None = None
+        try:
+            if type(serial) is not str or not serial.strip():
+                raise SkuExitSpikeError("必须显式提供非空设备通道。")
+            target = Path(output_directory)
+            _validate_new_target(target)
+            staging = _prepare_staging(target)
+
+            inspection = self._adb_client.inspect(serial)
+            _require_expected_device(inspection)
+            adapter = UiautomatorSkuExitAdapter(
+                self._connector(serial),
+                self._timeout_seconds,
+            )
+
+            # 第一次只读核验拒绝把任意页面带入 Back 边界；第二次紧邻 Back，覆盖核验期间漂移。
+            _require_t104_exit_precondition(adapter)
+            _require_t104_exit_precondition(adapter)
+
+            rpc_outcome = "completed"
+            try:
+                adapter.leave_sku_panel()
+            except SkuSelectionRunError:
+                if adapter.back_attempts != 1 or adapter.back_rpc_outcome != "ambiguous":
+                    raise
+                # RPC 失败不能证明 Back 未送达。只读采集可供人调和，但绝不再发第二次动作。
+                rpc_outcome = "ambiguous_reconciled"
+
+            if adapter.back_attempts != 1:
+                raise SkuExitSpikeError("返回动作审计不完整，未发布任何证据产物。")
+            if rpc_outcome == "completed" and adapter.back_rpc_outcome != "completed":
+                raise SkuExitSpikeError("返回动作结果不完整，未发布任何证据产物。")
+
+            # Back 后先确认仍是已取证版本的 PDD；不能先把其他前台应用写进本地证据。
+            initial_app = _require_t104_post_app(adapter)
+
+            screenshot_path = staging / "post_exit_screenshot.png"
+            _save_base64_screenshot(adapter.capture_screenshot(), screenshot_path)
+            _require_screenshot_size(screenshot_path)
+
+            hierarchy = adapter.dump_window_hierarchy()
+            _validate_hierarchy(hierarchy)
+            hierarchy_path = staging / "post_exit_hierarchy.xml"
+            hierarchy_path.write_text(hierarchy, encoding="utf-8")
+
+            current_app = _require_t104_post_app(adapter)
+            if current_app != initial_app:
+                raise SkuExitSpikeError("退出后应用摘要在采集期间漂移，未发布任何证据产物。")
+            app_path = staging / "post_exit_app.json"
+            app_path.write_text(
+                json.dumps(current_app, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+                encoding="utf-8",
+            )
+
+            manifest_path = staging / "manifest.json"
+            manifest_path.write_text(
+                json.dumps(
+                    _t104_exit_manifest(
+                        inspection,
+                        serial,
+                        rpc_outcome,
+                        screenshot_path,
+                        hierarchy_path,
+                        app_path,
+                    ),
+                    ensure_ascii=False,
+                    indent=2,
+                    sort_keys=True,
+                )
+                + "\n",
+                encoding="utf-8",
+            )
+            os.rename(staging, target)
+            staging = None
+        except (DeviceConnectionError, SkuSelectionError, SkuSelectionRunError):
+            _clean_staging(staging)
+            raise
+        except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
+            _clean_staging(staging)
+            raise SkuExitSpikeError("T-104 取证超时，未发布任何证据产物。") from error
+        except OSError as error:
+            _clean_staging(staging)
+            raise SkuExitSpikeError("T-104 证据目录无法发布，未发布任何证据产物。") from error
+        except Exception as error:
+            _clean_staging(staging)
+            raise SkuExitSpikeError("T-104 取证未完成，未发布任何证据产物。") from error
+
+        return SkuExitSpikeResult(target, target / "manifest.json")
+
+
 class SkuSelectionRunner:
     """只运行 T-103 目标规格恢复、价格确认、原始截图和一次安全退出。"""
 
@@ -470,6 +694,87 @@ def _require_expected_device(inspection: DeviceInspection) -> None:
         raise SkuSelectionRunError("设备型号或 Android 版本不是已取证组合，已停止操作。")
 
 
+def _require_t104_exit_precondition(adapter: UiautomatorSkuExitAdapter) -> None:
+    """fresh 核验 T-103 已人验的唯一 Back 前置，不产生任何页面动作。"""
+
+    _require_expected_version(adapter.app_info(PDD_PACKAGE))
+    current = adapter.app_current()
+    if current.get("package") != PDD_PACKAGE:
+        raise SkuExitSpikeError("Back 前拼多多不在前台，已停止取证。")
+    if adapter.display_size() != EXPECTED_SCREEN_SIZE:
+        raise SkuExitSpikeError("屏幕坐标空间不是已取证尺寸，已停止取证。")
+    if _unit_price(_parse_nodes(adapter.dump_window_hierarchy())) != EXPECTED_UNIT_PRICE:
+        raise SkuExitSpikeError("Back 前目标规格或当前价不匹配，已停止取证。")
+
+
+def _post_exit_app_evidence(value: object) -> dict[str, str]:
+    """只保留页面身份复核需要的应用字段，不把第三方返回整体写盘。"""
+
+    if not isinstance(value, dict):
+        raise SkuExitSpikeError("退出后应用摘要无效，未发布任何证据产物。")
+    package = value.get("package")
+    activity = value.get("activity")
+    if not isinstance(package, str) or not package.strip():
+        raise SkuExitSpikeError("退出后应用包名缺失，未发布任何证据产物。")
+    if not isinstance(activity, str) or not activity.strip():
+        raise SkuExitSpikeError("退出后 Activity 缺失，未发布任何证据产物。")
+    return {"package": package, "activity": activity}
+
+
+def _require_t104_post_app(adapter: UiautomatorSkuExitAdapter) -> dict[str, str]:
+    current = _post_exit_app_evidence(adapter.app_current())
+    if current["package"] != PDD_PACKAGE:
+        raise SkuExitSpikeError("Back 后拼多多不在前台，未采集页面证据。")
+    _require_expected_version(adapter.app_info(PDD_PACKAGE))
+    return current
+
+
+def _t104_exit_manifest(
+    inspection: DeviceInspection,
+    serial: str,
+    rpc_outcome: str,
+    screenshot_path: Path,
+    hierarchy_path: Path,
+    app_path: Path,
+) -> dict[str, Any]:
+    """阶段 A 只写事实与哈希；不声明退出成功，也不授予价格证据角色。"""
+
+    return {
+        "schema_version": 1,
+        "captured_at": datetime.now(UTC).isoformat(),
+        "operation": "t104-sku-exit-evidence",
+        "product": {"goods_id": EXPECTED_GOODS_ID},
+        "channel": "wifi" if ":" in serial else "usb",
+        "serial_sha256": sha256(serial.encode("utf-8")).hexdigest(),
+        "device": {
+            "model": inspection.model,
+            "android_version": inspection.android_version,
+            "pdd_package": PDD_PACKAGE,
+            "pdd_version": EXPECTED_PDD_VERSION,
+        },
+        "back_attempts": 1,
+        "rpc_outcome": rpc_outcome,
+        "post_exit_status": "human_review_required",
+        "artifacts": [
+            {
+                "path": screenshot_path.name,
+                "role": "post_exit_human_review_only",
+                "sha256": _sha256_file(screenshot_path),
+            },
+            {
+                "path": hierarchy_path.name,
+                "role": "post_exit_raw_hierarchy_local_only",
+                "sha256": _sha256_file(hierarchy_path),
+            },
+            {
+                "path": app_path.name,
+                "role": "post_exit_app_identity_human_review_only",
+                "sha256": _sha256_file(app_path),
+            },
+        ],
+    }
+
+
 def _require_screenshot_size(screenshot_path: Path) -> None:
     try:
         with Image.open(screenshot_path) as image:
diff --git a/client/tests/pdd/test_sku_selection.py b/client/tests/pdd/test_sku_selection.py
index e718ca3..ad5faba 100644
--- a/client/tests/pdd/test_sku_selection.py
+++ b/client/tests/pdd/test_sku_selection.py
@@ -2,9 +2,9 @@ from __future__ import annotations
 
 import ast
 import base64
-from contextlib import redirect_stderr
+from contextlib import redirect_stderr, redirect_stdout
 from functools import lru_cache
-from io import BytesIO
+from io import BytesIO, StringIO
 import importlib.util
 import json
 from pathlib import Path
@@ -28,9 +28,12 @@ from cmbuyer_client.pdd.sku_selection import (
     resolve_task_selection,
 )
 from cmbuyer_client.pdd.sku_selection_runner import (
+    SkuExitSpikeCapturer,
+    SkuExitSpikeError,
     SkuSelectionDeviceAdapterError,
     SkuSelectionRunError,
     SkuSelectionScreenshotError,
+    UiautomatorSkuExitAdapter,
     UiautomatorSkuPanelAdapter,
     safe_failure_stage,
 )
@@ -1666,6 +1669,338 @@ class SkuSelectionRunnerTests(unittest.TestCase):
         self.assertEqual(len(_actions(device, "pressKey")), 0)
 
 
+class _ExitEvidenceDevice(_RawDevice):
+    def __init__(self, hierarchy: str | None = None) -> None:
+        super().__init__(hierarchy or _M_FIXTURE.read_text(encoding="utf-8"))
+        self.activity = "com.xunmeng.pinduoduo.activity.NewPageActivity"
+
+    def app_current(self) -> dict[str, str]:
+        self.calls.append(("app_current",))
+        return {"package": self.package, "activity": self.activity}
+
+
+class SkuExitSpikeCapturerTests(unittest.TestCase):
+    def _capturer(self, adb: _FakeAdb, device: _ExitEvidenceDevice) -> SkuExitSpikeCapturer:
+        return SkuExitSpikeCapturer(adb, lambda serial: device, 0.03)
+
+    def test_completed_back_atomically_publishes_human_review_only_evidence(self) -> None:
+        adb = _FakeAdb()
+        device = _ExitEvidenceDevice()
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "exit-evidence"
+            result = self._capturer(adb, device).capture("192.168.0.173:5555", target)
+            manifest_text = result.manifest_path.read_text(encoding="utf-8")
+            manifest = json.loads(manifest_text)
+
+            self.assertEqual(result.output_directory, target)
+            self.assertEqual(manifest["product"], {"goods_id": "937122477375"})
+            self.assertEqual(manifest["channel"], "wifi")
+            self.assertEqual(manifest["back_attempts"], 1)
+            self.assertEqual(manifest["rpc_outcome"], "completed")
+            self.assertEqual(manifest["post_exit_status"], "human_review_required")
+            self.assertNotIn("safe_exit", manifest_text)
+            self.assertNotIn("SKU_PANEL_GATE_1", manifest_text)
+            self.assertNotIn("192.168.0.173:5555", manifest_text)
+            self.assertEqual(
+                [item["path"] for item in manifest["artifacts"]],
+                [
+                    "post_exit_screenshot.png",
+                    "post_exit_hierarchy.xml",
+                    "post_exit_app.json",
+                ],
+            )
+            self.assertEqual(
+                manifest["artifacts"][0]["role"],
+                "post_exit_human_review_only",
+            )
+            for artifact in manifest["artifacts"]:
+                self.assertTrue((target / artifact["path"]).is_file())
+                self.assertRegex(artifact["sha256"], r"^[0-9a-f]{64}$")
+            self.assertEqual(
+                json.loads((target / "post_exit_app.json").read_text(encoding="utf-8")),
+                {
+                    "activity": "com.xunmeng.pinduoduo.activity.NewPageActivity",
+                    "package": "com.xunmeng.pinduoduo",
+                },
+            )
+
+        self.assertEqual(adb.calls, [("inspect", "192.168.0.173:5555")])
+        self.assertEqual(len(_actions(device, "pressKey")), 1)
+        self.assertEqual(len(_actions(device, "takeScreenshot")), 1)
+        self.assertEqual(_actions(device, "click"), [])
+        self.assertEqual(_actions(device, "swipe"), [])
+
+    def test_precondition_mismatch_or_drift_never_sends_back_or_leaves_artifacts(self) -> None:
+        class WrongScreen(_ExitEvidenceDevice):
+            def window_size(self) -> tuple[int, int]:
+                self.calls.append(("window_size",))
+                return 1080, 1920
+
+        class HierarchyDrift(_ExitEvidenceDevice):
+            def __init__(self) -> None:
+                super().__init__()
+                self.reads = 0
+
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "dumpWindowHierarchy":
+                    self.reads += 1
+                    if self.reads == 2:
+                        self.hierarchy = _S_FIXTURE.read_text(encoding="utf-8")
+                return super().jsonrpc_call(method, params, timeout)
+
+        cases: list[tuple[str, _ExitEvidenceDevice]] = []
+        wrong_version = _ExitEvidenceDevice(); wrong_version.version = "8.18.0"
+        wrong_foreground = _ExitEvidenceDevice(); wrong_foreground.package = "other.package"
+        wrong_price = _ExitEvidenceDevice(); wrong_price.hierarchy = wrong_price.hierarchy.replace("快卖完 ¥12.88", "快卖完 ¥13.88")
+        cases.extend(
+            (
+                ("version", wrong_version),
+                ("foreground", wrong_foreground),
+                ("screen", WrongScreen()),
+                ("profile", _ExitEvidenceDevice(_S_FIXTURE.read_text(encoding="utf-8"))),
+                ("price", wrong_price),
+                ("drift", HierarchyDrift()),
+            )
+        )
+
+        for name, device in cases:
+            with self.subTest(name=name), TemporaryDirectory() as temporary:
+                target = Path(temporary) / "exit-evidence"
+                with self.assertRaises((SkuSelectionError, SkuSelectionRunError)):
+                    self._capturer(_FakeAdb(), device).capture("device-1", target)
+                self.assertEqual(_actions(device, "pressKey"), [])
+                self.assertEqual(_actions(device, "takeScreenshot"), [])
+                self.assertFalse(target.exists())
+                self.assertEqual(list(Path(temporary).glob(".exit-evidence.staging-*")), [])
+
+    def test_ambiguous_back_delivered_or_undelivered_publishes_once_for_human_review(self) -> None:
+        class AmbiguousBack(_ExitEvidenceDevice):
+            def __init__(self, delivered: bool) -> None:
+                super().__init__()
+                self.delivered = delivered
+
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "pressKey":
+                    self.calls.append(("jsonrpc", method, params, timeout))
+                    if self.delivered:
+                        self.hierarchy = "<hierarchy rotation=\"0\" />"
+                    raise TimeoutError("private RPC detail")
+                return super().jsonrpc_call(method, params, timeout)
+
+        for delivered in (True, False):
+            with self.subTest(delivered=delivered), TemporaryDirectory() as temporary:
+                device = AmbiguousBack(delivered)
+                target = Path(temporary) / "exit-evidence"
+                result = self._capturer(_FakeAdb(), device).capture("device-1", target)
+                manifest = json.loads(result.manifest_path.read_text(encoding="utf-8"))
+                self.assertEqual(manifest["back_attempts"], 1)
+                self.assertEqual(manifest["rpc_outcome"], "ambiguous_reconciled")
+                self.assertEqual(manifest["post_exit_status"], "human_review_required")
+                self.assertEqual(len(_actions(device, "pressKey")), 1)
+                self.assertEqual(len(_actions(device, "takeScreenshot")), 1)
+
+    def test_non_pdd_post_app_is_rejected_before_screenshot_for_completed_and_ambiguous_back(self) -> None:
+        class PostAppDrift(_ExitEvidenceDevice):
+            def __init__(self, ambiguous: bool) -> None:
+                super().__init__()
+                self.ambiguous = ambiguous
+
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "pressKey":
+                    self.calls.append(("jsonrpc", method, params, timeout))
+                    self.package = "external.app"
+                    if self.ambiguous:
+                        raise TimeoutError("private RPC detail")
+                    return ""
+                return super().jsonrpc_call(method, params, timeout)
+
+        for ambiguous in (False, True):
+            with self.subTest(ambiguous=ambiguous), TemporaryDirectory() as temporary:
+                device = PostAppDrift(ambiguous)
+                target = Path(temporary) / "exit-evidence"
+                with self.assertRaises(SkuExitSpikeError):
+                    self._capturer(_FakeAdb(), device).capture("device-1", target)
+                self.assertEqual(len(_actions(device, "pressKey")), 1)
+                self.assertEqual(_actions(device, "takeScreenshot"), [])
+                self.assertFalse(target.exists())
+                self.assertEqual(list(Path(temporary).glob(".exit-evidence.staging-*")), [])
+
+        class PostVersionDrift(_ExitEvidenceDevice):
+            def __init__(self) -> None:
+                super().__init__()
+                self.version_reads = 0
+
+            def app_info(self, package_name: str) -> dict[str, str]:
+                self.version_reads += 1
+                if self.version_reads == 3:
+                    return {"versionName": "8.18.0"}
+                return super().app_info(package_name)
+
+        class MissingPostActivity(_ExitEvidenceDevice):
+            def app_current(self) -> dict[str, str]:
+                value = super().app_current()
+                if _actions(self, "pressKey"):
+                    value.pop("activity")
+                return value
+
+        for name, device in (("version", PostVersionDrift()), ("app_summary", MissingPostActivity())):
+            with self.subTest(name=name), TemporaryDirectory() as temporary:
+                target = Path(temporary) / "exit-evidence"
+                with self.assertRaises(SkuSelectionRunError):
+                    self._capturer(_FakeAdb(), device).capture("device-1", target)
+                self.assertEqual(len(_actions(device, "pressKey")), 1)
+                self.assertEqual(_actions(device, "takeScreenshot"), [])
+                self.assertFalse(target.exists())
+                self.assertEqual(list(Path(temporary).glob(".exit-evidence.staging-*")), [])
+
+    def test_post_capture_app_drift_or_read_failures_clean_every_artifact_without_retry(self) -> None:
+        class AppDrift(_ExitEvidenceDevice):
+            def __init__(self) -> None:
+                super().__init__()
+                self.post_reads = 0
+
+            def app_current(self) -> dict[str, str]:
+                value = super().app_current()
+                if _actions(self, "pressKey"):
+                    self.post_reads += 1
+                    if self.post_reads == 2:
+                        value["activity"] = "com.xunmeng.pinduoduo.activity.OtherActivity"
+                return value
+
+        class BadXml(_ExitEvidenceDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "dumpWindowHierarchy" and _actions(self, "pressKey"):
+                    self.calls.append(("jsonrpc", method, params, timeout))
+                    return "not-xml"
+                return super().jsonrpc_call(method, params, timeout)
+
+        class ScreenshotTimeout(_ExitEvidenceDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "takeScreenshot":
+                    self.calls.append(("jsonrpc", method, params, timeout))
+                    raise TimeoutError("private RPC detail")
+                return super().jsonrpc_call(method, params, timeout)
+
+        bad_png = _ExitEvidenceDevice(); bad_png.screenshot = "not-image"
+        for name, device in (
+            ("app_drift", AppDrift()),
+            ("bad_xml", BadXml()),
+            ("bad_png", bad_png),
+            ("screenshot_timeout", ScreenshotTimeout()),
+        ):
+            with self.subTest(name=name), TemporaryDirectory() as temporary:
+                target = Path(temporary) / "exit-evidence"
+                with self.assertRaises(SkuSelectionRunError):
+                    self._capturer(_FakeAdb(), device).capture("device-1", target)
+                self.assertEqual(len(_actions(device, "pressKey")), 1)
+                self.assertFalse(target.exists())
+                self.assertEqual(list(Path(temporary).glob(".exit-evidence.staging-*")), [])
+
+    def test_existing_target_publish_race_preseal_failure_and_repeat_are_fail_closed(self) -> None:
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "existing"
+            target.mkdir()
+            sentinel = target / "keep"
+            sentinel.write_text("keep", encoding="utf-8")
+            adb = _FakeAdb(); device = _ExitEvidenceDevice()
+            with self.assertRaises(SkuSelectionRunError):
+                self._capturer(adb, device).capture("device-1", target)
+            self.assertEqual(adb.calls, [])
+            self.assertEqual(device.calls, [])
+            self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")
+
+            race_target = Path(temporary) / "race"
+            original_rename = runner_module.os.rename
+
+            def create_target_then_rename(source: str | Path, destination: str | Path) -> None:
+                Path(destination).mkdir()
+                (Path(destination) / "sentinel").write_text("keep", encoding="utf-8")
+                original_rename(source, destination)
+
+            race_device = _ExitEvidenceDevice()
+            with (
+                patch.object(runner_module.os, "rename", side_effect=create_target_then_rename),
+                self.assertRaises(SkuSelectionRunError),
+            ):
+                self._capturer(_FakeAdb(), race_device).capture("device-1", race_target)
+            self.assertEqual((race_target / "sentinel").read_text(encoding="utf-8"), "keep")
+            self.assertEqual(list(Path(temporary).glob(".race.staging-*")), [])
+            self.assertEqual(len(_actions(race_device, "pressKey")), 1)
+
+            preseal_target = Path(temporary) / "preseal"
+            preseal_device = _ExitEvidenceDevice()
+            with (
+                patch.object(UiautomatorSkuExitAdapter, "leave_sku_panel", side_effect=SkuSelectionRunError("private")),
+                self.assertRaises(SkuSelectionRunError),
+            ):
+                self._capturer(_FakeAdb(), preseal_device).capture("device-1", preseal_target)
+            self.assertEqual(_actions(preseal_device, "pressKey"), [])
+            self.assertFalse(preseal_target.exists())
+            self.assertEqual(list(Path(temporary).glob(".preseal.staging-*")), [])
+
+            repeated_device = _ExitEvidenceDevice()
+            capturer = self._capturer(_FakeAdb(), repeated_device)
+            capturer.capture("device-1", Path(temporary) / "once")
+            with self.assertRaises(SkuExitSpikeError):
+                capturer.capture("device-1", Path(temporary) / "twice")
+            self.assertEqual(len(_actions(repeated_device, "pressKey")), 1)
+
+    def test_static_t104_boundary_has_only_named_back_and_read_calls(self) -> None:
+        root = Path(__file__).resolve().parents[2]
+        runner_path = root / "src" / "cmbuyer_client" / "pdd" / "sku_selection_runner.py"
+        script_path = root / "scripts" / "capture_sku_exit_spike.py"
+        runner_tree = ast.parse(runner_path.read_text(encoding="utf-8"))
+        script_source = script_path.read_text(encoding="utf-8")
+        script_tree = ast.parse(script_source)
+        classes = {
+            node.name: node
+            for node in runner_tree.body
+            if isinstance(node, ast.ClassDef)
+        }
+        selected = (classes["UiautomatorSkuExitAdapter"], classes["SkuExitSpikeCapturer"])
+        called_attributes = {
+            node.func.attr
+            for selected_class in selected
+            for node in ast.walk(selected_class)
+            if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
+        }
+        forbidden_calls = {
+            "start_pdd_view_intent",
+            "tap_sku_entry",
+            "tap_sku_option",
+            "reveal_size_options_once",
+            "set_quantity_and_readback",
+            "go_to_order_confirm",
+            "create_submission_fence",
+            "submit_order_once",
+        }
+        self.assertTrue(forbidden_calls.isdisjoint(called_attributes))
+        self.assertTrue(forbidden_calls.isdisjoint(
+            node.attr for node in ast.walk(script_tree) if isinstance(node, ast.Attribute)
+        ))
+        self.assertTrue(
+            {"click", "swipe", "scroll", "intent", "quantity", "confirm", "fence", "submit", "payment"}.isdisjoint(
+                script_source.lower().split()
+            )
+        )
+        public_adapter_api = {
+            name for name in UiautomatorSkuExitAdapter.__dict__ if not name.startswith("_")
+        }
+        self.assertEqual(
+            public_adapter_api,
+            {
+                "app_info",
+                "app_current",
+                "dump_window_hierarchy",
+                "capture_screenshot",
+                "display_size",
+                "leave_sku_panel",
+                "back_attempts",
+                "back_rpc_outcome",
+            },
+        )
+
+
 class SkuSelectionCliTests(unittest.TestCase):
     def test_cli_accepts_only_target_url_and_task_values(self) -> None:
         script = _load_runner_script()
@@ -1725,6 +2060,98 @@ class SkuSelectionCliTests(unittest.TestCase):
             self.assertNotIn("page-body", output)
 
 
+class SkuExitSpikeCliTests(unittest.TestCase):
+    def test_cli_accepts_only_explicit_device_output_timeout_and_adb(self) -> None:
+        script = _load_exit_script()
+        arguments = script.parse_arguments(
+            [
+                "--serial", "device-1",
+                "--output-dir", "fresh-evidence",
+                "--timeout", "10",
+                "--adb", "adb.exe",
+            ]
+        )
+        script.validate_arguments(arguments)
+        self.assertEqual(
+            set(vars(arguments)),
+            {"serial", "output_dir", "timeout", "adb"},
+        )
+        for field, value in (("serial", ""), ("timeout", 0), ("timeout", float("inf"))):
+            with self.subTest(field=field), self.assertRaises(ValueError):
+                script.validate_arguments(
+                    type(
+                        "Arguments",
+                        (),
+                        {
+                            "serial": "device-1",
+                            "output_dir": Path("fresh-evidence"),
+                            "timeout": 10.0,
+                            "adb": "adb",
+                            field: value,
+                        },
+                    )()
+                )
+        with redirect_stderr(StringIO()), self.assertRaises(SystemExit):
+            script.parse_arguments(
+                [
+                    "--serial", "device-1",
+                    "--output-dir", "fresh-evidence",
+                    "--goods-id", "937122477375",
+                ]
+            )
+
+    def test_cli_never_echoes_serial_page_text_or_sensitive_path(self) -> None:
+        script = _load_exit_script()
+        secret = "SERIAL=192.168.0.173:5555 PATH=C:/Users/private <hierarchy>private</hierarchy>"
+
+        class FailingCapturer:
+            def __init__(self, *args: object, **kwargs: object) -> None:
+                pass
+
+            def capture(self, *args: object, **kwargs: object) -> object:
+                raise SkuExitSpikeError(secret)
+
+        stderr = BytesIO()
+        import io
+        text_stderr = io.TextIOWrapper(stderr, encoding="utf-8")
+        with patch.object(script, "SkuExitSpikeCapturer", FailingCapturer), redirect_stderr(text_stderr):
+            status = script.main(
+                [
+                    "--serial", "192.168.0.173:5555",
+                    "--output-dir", "C:/Users/private/evidence",
+                ]
+            )
+        text_stderr.flush()
+        output = stderr.getvalue().decode("utf-8")
+        self.assertEqual(status, 1)
+        self.assertNotIn(secret, output)
+        self.assertNotIn("192.168.0.173:5555", output)
+        self.assertNotIn("C:/Users/private", output)
+        self.assertNotIn("Traceback", output)
+
+        class SuccessfulCapturer:
+            def __init__(self, *args: object, **kwargs: object) -> None:
+                pass
+
+            def capture(self, *args: object, **kwargs: object) -> object:
+                return object()
+
+        stdout = BytesIO()
+        text_stdout = io.TextIOWrapper(stdout, encoding="utf-8")
+        with patch.object(script, "SkuExitSpikeCapturer", SuccessfulCapturer), redirect_stdout(text_stdout):
+            status = script.main(
+                [
+                    "--serial", "192.168.0.173:5555",
+                    "--output-dir", "C:/Users/private/evidence",
+                ]
+            )
+        text_stdout.flush()
+        output = stdout.getvalue().decode("utf-8")
+        self.assertEqual(status, 0)
+        self.assertNotIn("192.168.0.173:5555", output)
+        self.assertNotIn("C:/Users/private", output)
+
+
 def _load_runner_script() -> object:
     path = Path(__file__).resolve().parents[2] / "scripts" / "run_t103_sku_selection.py"
     specification = importlib.util.spec_from_file_location("run_t103_sku_selection_test", path)
@@ -1735,5 +2162,15 @@ def _load_runner_script() -> object:
     return module
 
 
+def _load_exit_script() -> object:
+    path = Path(__file__).resolve().parents[2] / "scripts" / "capture_sku_exit_spike.py"
+    specification = importlib.util.spec_from_file_location("capture_sku_exit_spike_test", path)
+    if specification is None or specification.loader is None:
+        raise RuntimeError("无法加载 T-104 阶段 A 脚本。")
+    module = importlib.util.module_from_spec(specification)
+    specification.loader.exec_module(module)
+    return module
+
+
 if __name__ == "__main__":
     unittest.main()