docs(tasks): record T-110 architecture validation

client/requirements.txt

@@ -1,7 +1,9 @@
 # -*- coding: utf-8 -*-
 # 桌面界面（Qt 官方 Python 绑定）。
 PySide6
-# 后续真机取证会使用；本阶段不导入或连接设备。
+# T-101 基线取证使用；只连接显式 serial，不打开或操作拼多多页面。
 uiautomator2
+# T-101 直接使用当前 ADB server 的已列出设备对象交给 uiautomator2，禁止 WiFi 自动重连。
+adbutils>=2.11,<3
 # 后续截图完整性检查会使用。
 Pillow

client/scripts/capture_device_baseline.py

@@ -0,0 +1,76 @@
+"""采集指定 Android 设备的本地基线证据；不打开或操作拼多多页面。"""
+
+from __future__ import annotations
+
+import argparse
+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 (
+    BaselineCaptureError,
+    DeviceBaselineCapturer,
+    NoReconnectUiautomatorConnector,
+)
+
+
+def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="采集显式指定 Android 设备的本地基线证据。")
+    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、uiautomator2 RPC 与 ADB socket 超时（秒）。")
+    parser.add_argument("--adb", default="adb", help="adb 可执行文件路径。")
+    return parser.parse_args(argv)
+
+
+def validate_arguments(arguments: argparse.Namespace) -> None:
+    """在导入设备库前拒绝危险或无效输入，便于离线测试。"""
+
+    if not arguments.serial.strip():
+        raise ValueError("必须显式提供非空 --serial。")
+    if arguments.timeout <= 0:
+        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
+
+    client = AdbClient(SubprocessAdbRunner(arguments.adb), timeout_seconds=arguments.timeout)
+    connector = NoReconnectUiautomatorConnector(
+        adbutils.AdbClient(socket_timeout=arguments.timeout).device_list,
+        u2.connect,
+    )
+    capturer = DeviceBaselineCapturer(client, connector, timeout_seconds=arguments.timeout)
+    try:
+        result = capturer.capture(arguments.serial, arguments.output_dir)
+    except (DeviceConnectionError, BaselineCaptureError) as error:
+        # 错误类型只表达状态，不打印 ADB 输出、serial、XML 或页面正文。
+        print(f"基线取证失败：{error}", file=sys.stderr)
+        return 1
+    except OSError:
+        print("基线取证失败：无法创建或发布本地证据目录。", file=sys.stderr)
+        return 1
+
+    print(f"基线取证完成：{result.output_directory}")
+    print(f"manifest：{result.manifest_path}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

client/scripts/capture_product_open.py

@@ -0,0 +1,75 @@
+"""打开已验证的拼多多商品直链并采集只读本地证据。"""
+
+from __future__ import annotations
+
+import argparse
+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.product_open import ProductOpenCapturer, ProductOpenError
+from cmbuyer_client.pdd.product_url import ProductUrlError, parse_product_url
+
+
+def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="打开 canonical 拼多多商品链接并采集只读证据。")
+    parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。")
+    parser.add_argument("--url", required=True, help="唯一允许的 goods.html?goods_id= 直链。")
+    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 not arguments.serial.strip():
+        raise ValueError("必须显式提供非空 --serial。")
+    if arguments.timeout <= 0:
+        raise ValueError("--timeout 必须大于 0。")
+    parse_product_url(arguments.url)
+
+
+def main(argv: list[str] | None = None) -> int:
+    arguments = parse_arguments(argv)
+    try:
+        validate_arguments(arguments)
+        link = parse_product_url(arguments.url)
+    except (ValueError, ProductUrlError) 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
+
+    client = AdbClient(SubprocessAdbRunner(arguments.adb), timeout_seconds=arguments.timeout)
+    connector = NoReconnectUiautomatorConnector(
+        adbutils.AdbClient(socket_timeout=arguments.timeout).device_list,
+        u2.connect,
+    )
+    capturer = ProductOpenCapturer(client, connector, timeout_seconds=arguments.timeout)
+    try:
+        result = capturer.open_and_capture(arguments.serial, link.canonical_url, arguments.output_dir)
+    except (DeviceConnectionError, ProductOpenError) as error:
+        # 不打印 ADB 输出、serial、Activity、XML 或页面正文。
+        print(f"商品打开取证失败：{error}", file=sys.stderr)
+        return 1
+    except OSError:
+        print("商品打开取证失败：无法创建或发布本地证据目录。", file=sys.stderr)
+        return 1
+
+    print(f"商品打开取证完成：{result.output_directory}")
+    print(f"manifest：{result.manifest_path}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

client/scripts/capture_sku_panel_spike.py

@@ -0,0 +1,103 @@
+"""采集人工已停在规格面板的三种状态证据；不执行任何页面操作。"""
+
+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.product_url import ProductUrl, ProductUrlError, parse_product_url
+from cmbuyer_client.pdd.sku_panel_spike import (
+    HUMAN_DECLARED_STATES,
+    SkuPanelEvidenceCapturer,
+    SkuPanelEvidenceError,
+)
+
+
+def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="采集人工已打开的拼多多规格面板状态证据。")
+    parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。")
+    product_source = parser.add_mutually_exclusive_group(required=True)
+    product_source.add_argument("--url", help="唯一 canonical goods.html?goods_id= 直链。")
+    product_source.add_argument("--goods-id", help="纯数字商品标识；仅用于记录证据归属。")
+    parser.add_argument("--state", required=True, choices=sorted(HUMAN_DECLARED_STATES), help="人工声明的面板状态。")
+    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 resolve_product_url(arguments: argparse.Namespace) -> ProductUrl:
+    if isinstance(arguments.url, str):
+        return parse_product_url(arguments.url)
+    if isinstance(arguments.goods_id, str):
+        # 仅使用严格 parser 重新验证并构建，不把输入交给 ADB 或页面。
+        return parse_product_url(f"https://mobile.yangkeduo.com/goods.html?goods_id={arguments.goods_id}")
+    raise ValueError("必须提供 --url 或 --goods-id。")
+
+
+def validate_arguments(arguments: argparse.Namespace) -> ProductUrl:
+    if not isinstance(arguments.serial, str) or not arguments.serial.strip():
+        raise ValueError("必须显式提供非空 --serial。")
+    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 的有限数值。")
+    if arguments.state not in HUMAN_DECLARED_STATES:
+        raise ValueError("--state 必须是允许的人工声明状态。")
+    return resolve_product_url(arguments)
+
+
+def main(argv: list[str] | None = None) -> int:
+    arguments = parse_arguments(argv)
+    try:
+        link = validate_arguments(arguments)
+    except (ValueError, ProductUrlError) 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
+
+    client = AdbClient(SubprocessAdbRunner(arguments.adb), timeout_seconds=arguments.timeout)
+    connector = NoReconnectUiautomatorConnector(
+        adbutils.AdbClient(socket_timeout=arguments.timeout).device_list,
+        u2.connect,
+    )
+    capturer = SkuPanelEvidenceCapturer(client, connector, timeout_seconds=arguments.timeout)
+    try:
+        result = capturer.capture(
+            arguments.serial,
+            link.canonical_url,
+            arguments.state,
+            arguments.output_dir,
+        )
+    except (DeviceConnectionError, SkuPanelEvidenceError) as error:
+        # 不打印 ADB 输出、serial、Activity、XML 或页面正文。
+        print(f"规格面板证据采集失败：{error}", file=sys.stderr)
+        return 1
+    except OSError:
+        print("规格面板证据采集失败：无法创建或发布本地证据目录。", file=sys.stderr)
+        return 1
+
+    print(f"规格面板证据采集完成：{result.output_directory}")
+    print(f"manifest：{result.manifest_path}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

client/src/cmbuyer_client/device/__init__.py

@@ -0,0 +1,15 @@
+"""设备连接与基线取证边界。
+
+本包只提供显式设备选择、非敏感身份核验和本地基线采集；不包含任何采购页面或订单操作。
+"""
+
+from .adb import AdbClient, AdbDevice, CommandResult
+from .baseline import BaselineCaptureResult, DeviceBaselineCapturer
+
+__all__ = [
+    "AdbClient",
+    "AdbDevice",
+    "BaselineCaptureResult",
+    "CommandResult",
+    "DeviceBaselineCapturer",
+]

client/src/cmbuyer_client/device/adb.py

@@ -0,0 +1,282 @@
+"""ADB 设备清单与物理设备冲突的 fail-closed 边界。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+import subprocess
+from typing import Protocol, Sequence
+
+
+class DeviceConnectionError(RuntimeError):
+    """显式设备连接边界的基础错误，不携带命令输出或设备敏感内容。"""
+
+
+class SerialRequiredError(DeviceConnectionError):
+    """调用方没有明确指定设备 serial。"""
+
+
+class DeviceNotFoundError(DeviceConnectionError):
+    """指定 serial 不在 ADB 当前清单中。"""
+
+
+class DeviceOfflineError(DeviceConnectionError):
+    """指定设备处于 offline 状态。"""
+
+
+class DeviceUnauthorizedError(DeviceConnectionError):
+    """指定设备尚未授权此电脑。"""
+
+
+class DeviceStateError(DeviceConnectionError):
+    """指定设备处于其他不可用状态。"""
+
+
+class DeviceCommandTimeoutError(DeviceConnectionError):
+    """ADB 命令超过调用方指定的超时。"""
+
+
+class DeviceCommandError(DeviceConnectionError):
+    """ADB 命令失败；错误文本刻意不回显设备输出。"""
+
+
+class DeviceIdentityUnconfirmedError(DeviceConnectionError):
+    """多条在线通道无法完成同机身份判断，必须由人处理。"""
+
+
+class DuplicatePhysicalDeviceError(DeviceConnectionError):
+    """同一物理手机通过多个 ADB 通道同时在线。"""
+
+
+class IntentLaunchUnconfirmedError(DeviceConnectionError):
+    """`am start -W` 没有给出可确认的启动成功结果。"""
+
+
+@dataclass(frozen=True)
+class CommandResult:
+    """可注入命令执行器的最小、可离线构造结果。"""
+
+    stdout: str
+    stderr: str = ""
+    returncode: int = 0
+
+
+@dataclass(frozen=True)
+class IntentLaunchSummary:
+    """不含 Activity、页面内容或 ADB 输出的受限启动摘要。"""
+
+    status: str
+    returncode: int
+
+
+class CommandRunner(Protocol):
+    """运行 ADB 子命令的可替换边界。"""
+
+    def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
+        """运行参数，不得通过 shell 拼接。"""
+
+
+class SubprocessAdbRunner:
+    """使用 subprocess 的生产执行器，所有调用必须带超时。"""
+
+    def __init__(self, executable: str | Path = "adb") -> None:
+        self._executable = str(executable)
+
+    def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
+        try:
+            completed = subprocess.run(
+                [self._executable, *arguments],
+                check=False,
+                capture_output=True,
+                text=True,
+                encoding="utf-8",
+                errors="replace",
+                timeout=timeout_seconds,
+            )
+        except subprocess.TimeoutExpired as error:
+            raise DeviceCommandTimeoutError("ADB 命令超时，请检查设备连接后由人工重试。") from error
+        except OSError as error:
+            raise DeviceCommandError("无法启动 ADB，请检查 adb 路径与本机工具链。") from error
+
+        return CommandResult(
+            stdout=completed.stdout,
+            stderr=completed.stderr,
+            returncode=completed.returncode,
+        )
+
+
+@dataclass(frozen=True)
+class AdbDevice:
+    """`adb devices -l` 的单行非敏感传输元数据。"""
+
+    serial: str
+    state: str
+    product: str | None = None
+    model: str | None = None
+    device: str | None = None
+
+
+@dataclass(frozen=True)
+class DeviceInspection:
+    """选定通道的只读身份结果，原始硬件标识只在内存中参与比较。"""
+
+    device: AdbDevice
+    model: str
+    android_version: str
+
+
+def parse_adb_devices(output: str) -> list[AdbDevice]:
+    """解析 `adb devices -l`，忽略标题、空行和 adb 附加提示。"""
+
+    devices: list[AdbDevice] = []
+    for raw_line in output.splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("List of devices attached") or line.startswith("*"):
+            continue
+        fields = line.split()
+        if len(fields) < 2:
+            continue
+        details = {
+            key: value
+            for field in fields[2:]
+            if ":" in field
+            for key, value in [field.split(":", 1)]
+        }
+        devices.append(
+            AdbDevice(
+                serial=fields[0],
+                state=fields[1],
+                product=details.get("product"),
+                model=details.get("model"),
+                device=details.get("device"),
+            )
+        )
+    return devices
+
+
+class AdbClient:
+    """显式 serial 的 ADB 只读查询。
+
+    多个在线通道必须完成硬件身份比对。比对失败时不能用相同 model/product 猜测同一台手机，
+    因为那会把不确定性隐藏成错误的安全结论。
+    """
+
+    def __init__(self, runner: CommandRunner, timeout_seconds: float = 10.0) -> None:
+        if timeout_seconds <= 0:
+            raise ValueError("timeout_seconds 必须大于 0")
+        self._runner = runner
+        self._timeout_seconds = timeout_seconds
+
+    def inspect(self, serial: str) -> DeviceInspection:
+        """确认指定通道在线且不与另一在线通道指向同一物理设备。"""
+
+        selected_serial = _require_serial(serial)
+        devices = self.devices()
+        selected = next((device for device in devices if device.serial == selected_serial), None)
+        if selected is None:
+            raise DeviceNotFoundError("指定设备不在 ADB 清单中，请显式检查 serial。")
+        _raise_for_state(selected.state)
+
+        online_devices = [device for device in devices if device.state == "device"]
+        if len(online_devices) > 1:
+            identities: dict[str, frozenset[str]] = {}
+            for candidate in online_devices:
+                try:
+                    identities[candidate.serial] = self._physical_identity(candidate)
+                except DeviceConnectionError as error:
+                    raise DeviceIdentityUnconfirmedError(
+                        "存在多个在线 ADB 通道且身份无法确认，已拒绝选择设备。"
+                    ) from error
+
+            selected_identity = identities[selected.serial]
+            if any(
+                candidate_serial != selected.serial and selected_identity.intersection(candidate_identity)
+                for candidate_serial, candidate_identity in identities.items()
+            ):
+                raise DuplicatePhysicalDeviceError(
+                    "同一物理手机的多个 ADB 通道同时在线，已拒绝继续；请仅保留一个通道。"
+                )
+
+        model = self._getprop(selected.serial, "ro.product.model") or selected.model or "unknown"
+        android_version = self._getprop(selected.serial, "ro.build.version.release") or "unknown"
+        return DeviceInspection(device=selected, model=model, android_version=android_version)
+
+    def devices(self) -> list[AdbDevice]:
+        """读取并解析 ADB 设备清单。"""
+
+        result = self._run_checked(("devices", "-l"))
+        return parse_adb_devices(result.stdout)
+
+    def start_pdd_view_intent(self, serial: str, goods_id: str) -> IntentLaunchSummary:
+        """以参数数组启动唯一允许的拼多多 ACTION_VIEW Intent。
+
+        这里刻意不提供任意 shell 或任意 package 的执行接口。调用方必须先完成
+        ``inspect`` 和应用版本核验；本方法在本层从纯数字 ``goods_id`` 重建 URL，调用方不能
+        把另一个 URL 直接交给 ADB。本方法既不点击控件，也不解析 Activity 或页面文本。
+        """
+
+        selected_serial = _require_serial(serial)
+        if (
+            not isinstance(goods_id, str)
+            or not goods_id
+            or any(character < "0" or character > "9" for character in goods_id)
+        ):
+            raise ValueError("goods_id 必须是纯数字")
+        canonical_url = f"https://mobile.yangkeduo.com/goods.html?goods_id={goods_id}"
+        result = self._run_checked(
+            (
+                "-s",
+                selected_serial,
+                "shell",
+                "am",
+                "start",
+                "-W",
+                "-a",
+                "android.intent.action.VIEW",
+                "-d",
+                canonical_url,
+                "-p",
+                "com.xunmeng.pinduoduo",
+            )
+        )
+        if not any(line.strip() == "Status: ok" for line in result.stdout.splitlines()):
+            raise IntentLaunchUnconfirmedError("商品链接启动结果无法确认，已停止后续取证。")
+        return IntentLaunchSummary(status="ok", returncode=result.returncode)
+
+    def _physical_identity(self, device: AdbDevice) -> frozenset[str]:
+        serialno = self._getprop(device.serial, "ro.serialno")
+        boot_serialno = self._getprop(device.serial, "ro.boot.serialno")
+        identifiers = frozenset(value for value in (serialno, boot_serialno) if value)
+        if identifiers:
+            return identifiers
+        # model/product/device 只能作为展示元数据，不能证明两台同型号设备是同一物理机。
+        raise DeviceIdentityUnconfirmedError("无法读取设备硬件身份摘要。")
+
+    def _getprop(self, serial: str, property_name: str) -> str:
+        result = self._run_checked(("-s", serial, "shell", "getprop", property_name))
+        return result.stdout.strip()
+
+    def _run_checked(self, arguments: Sequence[str]) -> CommandResult:
+        try:
+            result = self._runner.run(arguments, self._timeout_seconds)
+        except subprocess.TimeoutExpired as error:
+            raise DeviceCommandTimeoutError("ADB 命令超时，请检查设备连接后由人工重试。") from error
+        if result.returncode != 0:
+            raise DeviceCommandError("ADB 命令失败，请检查设备连接或授权状态。")
+        return result
+
+
+def _require_serial(serial: str) -> str:
+    if not isinstance(serial, str) or not serial.strip():
+        raise SerialRequiredError("必须显式提供设备 serial，禁止自动选择设备。")
+    return serial.strip()
+
+
+def _raise_for_state(state: str) -> None:
+    if state == "device":
+        return
+    if state == "offline":
+        raise DeviceOfflineError("指定设备处于 offline 状态。")
+    if state == "unauthorized":
+        raise DeviceUnauthorizedError("指定设备尚未授权此电脑。")
+    raise DeviceStateError("指定设备不处于可用状态。")

client/src/cmbuyer_client/device/baseline.py

@@ -0,0 +1,238 @@
+"""只读设备基线取证，严格限制在元数据、截图和完整节点树。"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from hashlib import sha256
+import base64
+import binascii
+from io import BytesIO
+import json
+import os
+from pathlib import Path
+import shutil
+from typing import Any, Protocol
+from uuid import uuid4
+from xml.etree import ElementTree
+
+from adbutils.errors import AdbTimeout
+from PIL import Image, UnidentifiedImageError
+from uiautomator2.exceptions import HTTPTimeoutError
+
+from .adb import AdbClient, DeviceInspection
+
+
+PDD_PACKAGE = "com.xunmeng.pinduoduo"
+SCREENSHOT_PARAMS = [1, 80]
+HIERARCHY_PARAMS = [False, 50]
+_BASE64_ASCII_WHITESPACE = " \t\r\n"
+
+
+class BaselineCaptureError(RuntimeError):
+    """基线取证无法完整落盘时的失败，不会伪造成功产物。"""
+
+
+class BaselineCaptureTimeoutError(BaselineCaptureError):
+    """设备基线取证超时；底层异常文本不向 CLI 或日志泄露。"""
+
+
+class UiAutomatorDevice(Protocol):
+    """本任务唯一需要的 uiautomator2 只读能力。"""
+
+    def app_info(self, package_name: str) -> dict[str, Any]:
+        """读取已安装应用元数据。"""
+
+    def jsonrpc_call(self, method: str, params: Any = None, timeout: float = 10) -> Any:
+        """调用公开 uiautomator2 JSON-RPC 接口。"""
+
+
+class NoReconnectUiautomatorConnector:
+    """只把当前 ADB server 已列出的设备对象交给 uiautomator2。
+
+    uiautomator2 直接接收 IP serial 时会在内部尝试 adb disconnect/connect。这里先从已列出设备中
+    取对象再调用 ``u2.connect(device_object)``，避免连接阶段隐式重连已经掉线的 WiFi 通道。
+    """
+
+    def __init__(self, list_devices: Callable[[], list[Any]], connect: Callable[[Any], UiAutomatorDevice]) -> None:
+        self._list_devices = list_devices
+        self._connect = connect
+
+    def __call__(self, serial: str) -> UiAutomatorDevice:
+        device = next((item for item in self._list_devices() if item.serial == serial), None)
+        if device is None:
+            raise BaselineCaptureError("设备在连接前已从 ADB 清单消失，已拒绝自动重连。")
+        return self._connect(device)
+
+
+@dataclass(frozen=True)
+class BaselineCaptureResult:
+    """已原子发布的基线取证摘要，不包含页面正文或原始 serial。"""
+
+    output_directory: Path
+    manifest_path: Path
+    screenshot_path: Path
+    hierarchy_path: Path
+
+
+class DeviceBaselineCapturer:
+    """以先校验通道、后连接、最后原子发布的顺序采集基线。
+
+    截图和 XML 可能包含页面敏感内容，因此仅落在调用方明确指定的本地目录；manifest 只写
+    哈希、设备非敏感元数据和脱敏后的 serial 摘要，绝不嵌入 XML 或页面文本。
+    """
+
+    def __init__(
+        self,
+        adb_client: AdbClient,
+        connector: Callable[[str], UiAutomatorDevice],
+        timeout_seconds: float,
+    ) -> None:
+        if timeout_seconds <= 0:
+            raise ValueError("timeout_seconds 必须大于 0")
+        self._adb_client = adb_client
+        self._connector = connector
+        self._timeout_seconds = timeout_seconds
+
+    def capture(self, serial: str, output_directory: Path) -> BaselineCaptureResult:
+        """采集完整基线，任何一步失败均不发布 output_directory。"""
+
+        inspection = self._adb_client.inspect(serial)
+        target = Path(output_directory)
+        if target.exists():
+            raise BaselineCaptureError("输出目录已存在；为防止混入旧证据，拒绝覆盖。")
+        if not target.name:
+            raise BaselineCaptureError("输出目录必须是明确的新目录。")
+
+        target.parent.mkdir(parents=True, exist_ok=True)
+        staging = target.parent / f".{target.name}.staging-{uuid4().hex}"
+        staging.mkdir()
+        try:
+            device = self._connector(serial)
+            app_info = device.app_info(PDD_PACKAGE)
+            version = _extract_version(app_info)
+
+            screenshot_path = staging / "screenshot.png"
+            screenshot_base64 = device.jsonrpc_call(
+                "takeScreenshot",
+                SCREENSHOT_PARAMS,
+                timeout=self._timeout_seconds,
+            )
+            _save_base64_screenshot(screenshot_base64, screenshot_path)
+
+            hierarchy = device.jsonrpc_call(
+                "dumpWindowHierarchy",
+                HIERARCHY_PARAMS,
+                timeout=self._timeout_seconds,
+            )
+            _validate_hierarchy(hierarchy)
+            hierarchy_path = staging / "hierarchy.xml"
+            hierarchy_path.write_text(hierarchy, encoding="utf-8")
+
+            manifest_path = staging / "manifest.json"
+            manifest_path.write_text(
+                json.dumps(
+                    _manifest(inspection, serial, version, screenshot_path, hierarchy_path),
+                    ensure_ascii=False,
+                    indent=2,
+                    sort_keys=True,
+                )
+                + "\n",
+                encoding="utf-8",
+            )
+            os.replace(staging, target)
+        except BaselineCaptureError:
+            # 仅删除本次创建、名称带随机标识的暂存目录，绝不触碰调用方原有输出目录。
+            if staging.exists():
+                shutil.rmtree(staging)
+            raise
+        except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
+            if staging.exists():
+                shutil.rmtree(staging)
+            raise BaselineCaptureTimeoutError("设备基线取证超时，未发布任何证据产物。") from error
+        except Exception as error:
+            if staging.exists():
+                shutil.rmtree(staging)
+            # uiautomator2/adbutils 可能把 serial、路径或远端响应放入异常文本，不能直接传播到 CLI。
+            raise BaselineCaptureError("设备基线取证未完成，未发布任何证据产物。") from error
+
+        return BaselineCaptureResult(
+            output_directory=target,
+            manifest_path=target / "manifest.json",
+            screenshot_path=target / "screenshot.png",
+            hierarchy_path=target / "hierarchy.xml",
+        )
+
+
+def _extract_version(app_info: dict[str, Any]) -> str:
+    version = app_info.get("versionName") or app_info.get("version_name")
+    if not isinstance(version, str) or not version.strip():
+        raise BaselineCaptureError("无法读取拼多多版本，拒绝发布不完整取证。")
+    return version.strip()
+
+
+def _save_base64_screenshot(value: Any, target: Path) -> None:
+    """规范化常见 ASCII Base64 空白后严格解码；没有 adb screenshot fallback。"""
+
+    if not isinstance(value, str) or not value:
+        raise BaselineCaptureError("截图 RPC 未返回 base64 数据，拒绝发布不完整取证。")
+    try:
+        normalized = value.translate({ord(character): None for character in _BASE64_ASCII_WHITESPACE})
+        raw_image = base64.b64decode(normalized.encode("ascii"), validate=True)
+    except (UnicodeEncodeError, ValueError, binascii.Error) as error:
+        raise BaselineCaptureError("截图 RPC Base64 语法无效，拒绝发布不完整取证。") from error
+    try:
+        with Image.open(BytesIO(raw_image)) as image:
+            image.load()
+            image.save(target, format="PNG")
+    except (UnidentifiedImageError, OSError) as error:
+        raise BaselineCaptureError("截图 RPC 图像数据无效，拒绝发布不完整取证。") from error
+
+
+def _validate_hierarchy(value: Any) -> None:
+    """确认 RPC 返回的是完整节点树，不把原始 XML 放进错误或日志。"""
+
+    if not isinstance(value, str) or not value:
+        raise BaselineCaptureError("节点树导出为空，拒绝发布不完整取证。")
+    try:
+        root = ElementTree.fromstring(value)
+    except ElementTree.ParseError as error:
+        raise BaselineCaptureError("节点树不是有效 XML，拒绝发布不完整取证。") from error
+    if root.tag != "hierarchy":
+        raise BaselineCaptureError("节点树根节点无效，拒绝发布不完整取证。")
+
+
+def _manifest(
+    inspection: DeviceInspection,
+    serial: str,
+    pdd_version: str,
+    screenshot_path: Path,
+    hierarchy_path: Path,
+) -> dict[str, Any]:
+    """只序列化审计摘要；页面内容留在 XML 文件，不进入日志或 manifest。"""
+
+    return {
+        "schema_version": 1,
+        "captured_at": datetime.now(UTC).isoformat(),
+        "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": pdd_version,
+        },
+        "artifacts": [
+            {"path": screenshot_path.name, "sha256": _sha256_file(screenshot_path)},
+            {"path": hierarchy_path.name, "sha256": _sha256_file(hierarchy_path)},
+        ],
+    }
+
+
+def _sha256_file(path: Path) -> str:
+    digest = sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()

client/src/cmbuyer_client/pdd/__init__.py

@@ -0,0 +1,15 @@
+"""拼多多链接的受限打开与只读取证。
+
+此包不提供页面选择器、输入、滑动、下单或支付能力。
+"""
+
+from .product_open import ProductOpenCapturer, ProductOpenResult
+from .product_url import ProductUrl, ProductUrlError, parse_product_url
+
+__all__ = [
+    "ProductOpenCapturer",
+    "ProductOpenResult",
+    "ProductUrl",
+    "ProductUrlError",
+    "parse_product_url",
+]

client/src/cmbuyer_client/pdd/product_open.py

@@ -0,0 +1,262 @@
+"""安全打开 canonical 商品链接后的只读取证。"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from hashlib import sha256
+import json
+from math import isfinite
+import os
+from pathlib import Path
+import shutil
+from time import monotonic, sleep
+from typing import Any, Protocol
+from uuid import uuid4
+
+from adbutils.errors import AdbTimeout
+from uiautomator2.exceptions import HTTPTimeoutError
+
+from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection, IntentLaunchSummary
+from ..device.baseline import (
+    HIERARCHY_PARAMS,
+    PDD_PACKAGE,
+    SCREENSHOT_PARAMS,
+    _save_base64_screenshot,
+    _sha256_file,
+    _validate_hierarchy,
+)
+from .product_url import ProductUrl, parse_product_url
+
+
+EXPECTED_PDD_VERSION = "8.17.0"
+
+
+class ProductOpenError(RuntimeError):
+    """商品打开或证据发布未完整完成。"""
+
+
+class ProductVersionMismatchError(ProductOpenError):
+    """运行时拼多多版本不是经取证允许的版本。"""
+
+
+class ProductPackageMismatchError(ProductOpenError):
+    """Intent 后在有限时间内未观察到拼多多前台包。"""
+
+
+class ProductOpenTimeoutError(ProductOpenError):
+    """商品打开后的只读取证超时。"""
+
+
+class ProductScreenshotCaptureError(ProductOpenError):
+    """Intent 后截图不能作为完整 PNG 证据保存。"""
+
+
+class ProductHierarchyCaptureError(ProductOpenError):
+    """Intent 后完整节点树不能作为有效 XML 证据保存。"""
+
+
+class ProductOpenUiDevice(Protocol):
+    """本任务所需的只读 uiautomator2 接口；故意没有任何 UI 操作方法。"""
+
+    def app_info(self, package_name: str) -> dict[str, Any]:
+        """读取应用元数据。"""
+
+    def app_current(self) -> dict[str, Any]:
+        """读取当前前台应用元数据。"""
+
+    def jsonrpc_call(self, method: str, params: Any = None, timeout: float = 10) -> Any:
+        """调用只读取证所需的公开 JSON-RPC 方法。"""
+
+
+@dataclass(frozen=True)
+class ProductOpenResult:
+    """已原子发布的商品打开证据位置。"""
+
+    output_directory: Path
+    manifest_path: Path
+    screenshot_path: Path
+    hierarchy_path: Path
+
+
+class ProductOpenCapturer:
+    """以 fail-closed 顺序打开已重建链接，并在打开后只读留证。
+
+    本类不判断商品页、Activity、文案或控件；打开后只确认当前 package，随后采集截图与
+    完整节点树。任何失败都不会发布半成品证据目录。
+    """
+
+    def __init__(
+        self,
+        adb_client: AdbClient,
+        connector: Callable[[str], ProductOpenUiDevice],
+        timeout_seconds: float,
+        foreground_poll_interval_seconds: float = 0.2,
+        monotonic_clock: Callable[[], float] = monotonic,
+        sleep_function: Callable[[float], None] = sleep,
+    ) -> None:
+        if not _is_positive_finite(timeout_seconds):
+            raise ValueError("timeout_seconds 必须是大于 0 的有限数值")
+        if not _is_positive_finite(foreground_poll_interval_seconds):
+            raise ValueError("foreground_poll_interval_seconds 必须是大于 0 的有限数值")
+        self._adb_client = adb_client
+        self._connector = connector
+        self._timeout_seconds = timeout_seconds
+        self._foreground_poll_interval_seconds = foreground_poll_interval_seconds
+        self._monotonic_clock = monotonic_clock
+        self._sleep_function = sleep_function
+
+    def open_and_capture(self, serial: str, product_url: str, output_directory: Path) -> ProductOpenResult:
+        """完成唯一允许的 Intent 打开及其后的只读取证。"""
+
+        # 公共入口只接收原始字符串并每次重新解析，不能由调用方构造不一致的值对象伪造 manifest。
+        link = parse_product_url(product_url)
+        target = Path(output_directory)
+        _validate_new_target(target)
+
+        staging: Path | None = None
+        try:
+            # inspect 必须先于连接和 Intent，复用 T-101 的显式 serial、重复物理设备拒绝逻辑。
+            inspection = self._adb_client.inspect(serial)
+            device = self._connector(serial)
+            pdd_version = _require_expected_version(device.app_info(PDD_PACKAGE))
+
+            # 版本精确匹配是 Intent 的前置条件，失败时绝不调用 start_pdd_view_intent。
+            intent = self._adb_client.start_pdd_view_intent(serial, link.goods_id)
+            self._wait_for_pdd_foreground(device)
+
+            target.parent.mkdir(parents=True, exist_ok=True)
+            staging = target.parent / f".{target.name}.staging-{uuid4().hex}"
+            staging.mkdir()
+            screenshot_path = staging / "screenshot.png"
+            try:
+                _save_base64_screenshot(
+                    device.jsonrpc_call("takeScreenshot", SCREENSHOT_PARAMS, timeout=self._timeout_seconds),
+                    screenshot_path,
+                )
+            except (AdbTimeout, HTTPTimeoutError, TimeoutError):
+                raise
+            except Exception as error:
+                raise ProductScreenshotCaptureError("商品打开后截图取证失败，未发布任何证据产物。") from error
+
+            try:
+                hierarchy = device.jsonrpc_call(
+                    "dumpWindowHierarchy",
+                    HIERARCHY_PARAMS,
+                    timeout=self._timeout_seconds,
+                )
+                _validate_hierarchy(hierarchy)
+            except (AdbTimeout, HTTPTimeoutError, TimeoutError):
+                raise
+            except Exception as error:
+                raise ProductHierarchyCaptureError("商品打开后节点树取证失败，未发布任何证据产物。") from error
+            hierarchy_path = staging / "hierarchy.xml"
+            hierarchy_path.write_text(hierarchy, encoding="utf-8")
+
+            manifest_path = staging / "manifest.json"
+            manifest_path.write_text(
+                json.dumps(
+                    _manifest(inspection, serial, link, pdd_version, intent, screenshot_path, hierarchy_path),
+                    ensure_ascii=False,
+                    indent=2,
+                    sort_keys=True,
+                )
+                + "\n",
+                encoding="utf-8",
+            )
+            os.replace(staging, target)
+        except (ProductOpenError, DeviceConnectionError):
+            _clean_staging(staging)
+            raise
+        except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
+            _clean_staging(staging)
+            raise ProductOpenTimeoutError("商品打开后的只读取证超时，未发布任何证据产物。") from error
+        except Exception as error:
+            _clean_staging(staging)
+            # 底层异常可能含 serial、路径或远端页面内容，不能直接向 CLI 或日志传播。
+            raise ProductOpenError("商品打开或只读取证未完成，未发布任何证据产物。") from error
+
+        return ProductOpenResult(
+            output_directory=target,
+            manifest_path=target / "manifest.json",
+            screenshot_path=target / "screenshot.png",
+            hierarchy_path=target / "hierarchy.xml",
+        )
+
+    def _wait_for_pdd_foreground(self, device: ProductOpenUiDevice) -> None:
+        """只轮询当前 package，直到 deadline；Activity 和节点树均不参与本判据。"""
+
+        deadline = self._monotonic_clock() + self._timeout_seconds
+        while True:
+            if _is_pdd_foreground(device.app_current()):
+                return
+            remaining = deadline - self._monotonic_clock()
+            if remaining <= 0:
+                raise ProductPackageMismatchError(
+                    "商品链接打开后未在限定时间内进入拼多多，已停止后续取证。"
+                )
+            # 每个失败观察后都等待正的、受 deadline 约束的时长，避免 busy-loop。
+            self._sleep_function(min(self._foreground_poll_interval_seconds, remaining))
+
+
+def _validate_new_target(target: Path) -> None:
+    if target.exists():
+        raise ProductOpenError("输出目录已存在；为防止混入旧证据，拒绝覆盖。")
+    if not target.name:
+        raise ProductOpenError("输出目录必须是明确的新目录。")
+
+
+def _clean_staging(staging: Path | None) -> None:
+    if staging is not None and staging.exists():
+        # staging 仅在本次调用中创建，删除前不解析或扩展任何调用方提供的路径。
+        shutil.rmtree(staging)
+
+
+def _require_expected_version(app_info: dict[str, Any]) -> str:
+    if not isinstance(app_info, dict):
+        raise ProductVersionMismatchError("拼多多版本与已取证版本不一致，已停止打开商品链接。")
+    version = app_info.get("versionName") or app_info.get("version_name")
+    if not isinstance(version, str) or version != EXPECTED_PDD_VERSION:
+        raise ProductVersionMismatchError("拼多多版本与已取证版本不一致，已停止打开商品链接。")
+    return version
+
+
+def _is_positive_finite(value: object) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0 and isfinite(value)
+
+
+def _is_pdd_foreground(current: object) -> bool:
+    return isinstance(current, dict) and current.get("package") == PDD_PACKAGE
+
+
+def _manifest(
+    inspection: DeviceInspection,
+    serial: str,
+    link: ProductUrl,
+    pdd_version: str,
+    intent: IntentLaunchSummary,
+    screenshot_path: Path,
+    hierarchy_path: Path,
+) -> dict[str, Any]:
+    """只写审计摘要；原始 serial、Activity、ADB 输出和页面正文均不进入 manifest。"""
+
+    return {
+        "schema_version": 1,
+        "captured_at": datetime.now(UTC).isoformat(),
+        "product": {"goods_id": link.goods_id, "canonical_url": link.canonical_url},
+        "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": pdd_version,
+        },
+        "intent": {"status": intent.status, "returncode": intent.returncode},
+        "current_package": PDD_PACKAGE,
+        "artifacts": [
+            {"path": screenshot_path.name, "sha256": _sha256_file(screenshot_path)},
+            {"path": hierarchy_path.name, "sha256": _sha256_file(hierarchy_path)},
+        ],
+    }

client/src/cmbuyer_client/pdd/product_url.py

@@ -0,0 +1,61 @@
+"""唯一允许交给 Android Intent 的商品链接。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from urllib.parse import parse_qsl, urlsplit
+
+
+_SCHEME = "https"
+_HOST = "mobile.yangkeduo.com"
+_PATH = "/goods.html"
+
+
+class ProductUrlError(ValueError):
+    """输入不是可安全重建的 canonical 商品链接。"""
+
+
+@dataclass(frozen=True)
+class ProductUrl:
+    """经验证的商品标识及由它重建的 canonical URL。"""
+
+    goods_id: str
+    canonical_url: str
+
+
+def parse_product_url(value: str) -> ProductUrl:
+    """只接受一个 ASCII 数字 ``goods_id`` 的拼多多商品直链。
+
+    解析结果绝不原样透传：Intent 使用的 URL 必须从 ``goods_id`` 重新构建，以排除
+    短链、额外参数、userinfo、fragment 和 URL 解析器的边缘表示。
+    """
+
+    if not isinstance(value, str):
+        raise ProductUrlError("商品链接必须是字符串。")
+    try:
+        parsed = urlsplit(value)
+        port = parsed.port
+        query_pairs = parse_qsl(parsed.query, keep_blank_values=True, strict_parsing=True)
+    except ValueError as error:
+        raise ProductUrlError("商品链接格式无效。") from error
+
+    if (
+        parsed.scheme != _SCHEME
+        or parsed.hostname != _HOST
+        or parsed.username is not None
+        or parsed.password is not None
+        or port is not None
+        or parsed.path != _PATH
+        or parsed.fragment
+    ):
+        raise ProductUrlError("商品链接不是允许的拼多多商品直链。")
+    if len(query_pairs) != 1 or query_pairs[0][0] != "goods_id":
+        raise ProductUrlError("商品链接必须且只能包含一个 goods_id 参数。")
+
+    goods_id = query_pairs[0][1]
+    if not goods_id or any(character < "0" or character > "9" for character in goods_id):
+        raise ProductUrlError("goods_id 必须是纯数字。")
+    canonical_url = f"{_SCHEME}://{_HOST}{_PATH}?goods_id={goods_id}"
+    if value != canonical_url:
+        raise ProductUrlError("商品链接必须使用唯一 canonical 表示。")
+    return ProductUrl(goods_id=goods_id, canonical_url=canonical_url)

client/src/cmbuyer_client/pdd/sku_panel_spike.py

@@ -0,0 +1,260 @@
+"""人工停留在规格面板后的只读取证。
+
+本模块不识别规格面板，不打开商品链接，也不读取价格；三种面板状态完全由现场人员声明。
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from hashlib import sha256
+import json
+from math import isfinite
+import os
+from pathlib import Path
+import shutil
+from typing import Any, Protocol
+from uuid import uuid4
+
+from adbutils.errors import AdbTimeout
+from uiautomator2.exceptions import HTTPTimeoutError
+
+from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection
+from ..device.baseline import (
+    HIERARCHY_PARAMS,
+    PDD_PACKAGE,
+    SCREENSHOT_PARAMS,
+    _save_base64_screenshot,
+    _sha256_file,
+    _validate_hierarchy,
+)
+from .product_open import EXPECTED_PDD_VERSION
+from .product_url import ProductUrl, parse_product_url
+
+
+HUMAN_DECLARED_STATES = frozenset(
+    {
+        "initial",
+        "one-dimension-selected",
+        "all-dimensions-selected",
+    }
+)
+
+
+class SkuPanelEvidenceError(RuntimeError):
+    """人工规格面板证据无法完整发布。"""
+
+
+class SkuPanelDeclaredStateError(SkuPanelEvidenceError):
+    """调用方没有提供允许的人工声明状态。"""
+
+
+class SkuPanelVersionMismatchError(SkuPanelEvidenceError):
+    """运行时拼多多版本不是已取证版本。"""
+
+
+class SkuPanelPackageMismatchError(SkuPanelEvidenceError):
+    """人工声明前台不是拼多多时仍试图留证。"""
+
+
+class SkuPanelEvidenceTimeoutError(SkuPanelEvidenceError):
+    """只读截图或节点树取证超时。"""
+
+
+class SkuPanelScreenshotError(SkuPanelEvidenceError):
+    """截图不能保存为严格有效的 PNG。"""
+
+
+class SkuPanelHierarchyError(SkuPanelEvidenceError):
+    """节点树不能保存为严格有效的 XML。"""
+
+
+class SkuPanelUiDevice(Protocol):
+    """人工面板证据所需的只读接口，故意没有任何页面操作方法。"""
+
+    def app_info(self, package_name: str) -> dict[str, Any]:
+        """读取应用元数据。"""
+
+    def app_current(self) -> dict[str, Any]:
+        """读取当前前台应用元数据。"""
+
+    def jsonrpc_call(self, method: str, params: Any = None, timeout: float = 10) -> Any:
+        """调用公开 JSON-RPC 的只读取证方法。"""
+
+
+@dataclass(frozen=True)
+class SkuPanelEvidenceResult:
+    """已原子发布的本地证据目录。"""
+
+    output_directory: Path
+    manifest_path: Path
+    screenshot_path: Path
+    hierarchy_path: Path
+
+
+class SkuPanelEvidenceCapturer:
+    """把人工已停留的面板状态留证，不对页面作任何自动结论。
+
+    状态字段命名为 ``human_declared_state``，防止消费者把本模块误解为自动面板/规格/价格识别。
+    """
+
+    def __init__(
+        self,
+        adb_client: AdbClient,
+        connector: Callable[[str], SkuPanelUiDevice],
+        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
+
+    def capture(
+        self,
+        serial: str,
+        product_url: str,
+        human_declared_state: str,
+        output_directory: Path,
+    ) -> SkuPanelEvidenceResult:
+        """采集人工已准备的状态；不会打开链接、面板或执行任何 UI 操作。"""
+
+        link = parse_product_url(product_url)
+        state = _validate_human_declared_state(human_declared_state)
+        target = Path(output_directory)
+        _validate_new_target(target)
+
+        staging: Path | None = None
+        try:
+            # 沿用 T-101 的显式 serial、在线状态与重复物理设备 fail-closed 核验。
+            inspection = self._adb_client.inspect(serial)
+            device = self._connector(serial)
+            pdd_version = _require_expected_version(device.app_info(PDD_PACKAGE))
+            _require_pdd_foreground(device.app_current())
+
+            target.parent.mkdir(parents=True, exist_ok=True)
+            staging = target.parent / f".{target.name}.staging-{uuid4().hex}"
+            staging.mkdir()
+            screenshot_path = staging / "screenshot.png"
+            try:
+                _save_base64_screenshot(
+                    device.jsonrpc_call("takeScreenshot", SCREENSHOT_PARAMS, timeout=self._timeout_seconds),
+                    screenshot_path,
+                )
+            except (AdbTimeout, HTTPTimeoutError, TimeoutError):
+                raise
+            except Exception as error:
+                raise SkuPanelScreenshotError("规格面板截图取证失败，未发布任何证据产物。") from error
+
+            try:
+                hierarchy = device.jsonrpc_call(
+                    "dumpWindowHierarchy",
+                    HIERARCHY_PARAMS,
+                    timeout=self._timeout_seconds,
+                )
+                _validate_hierarchy(hierarchy)
+            except (AdbTimeout, HTTPTimeoutError, TimeoutError):
+                raise
+            except Exception as error:
+                raise SkuPanelHierarchyError("规格面板节点树取证失败，未发布任何证据产物。") from error
+            hierarchy_path = staging / "hierarchy.xml"
+            hierarchy_path.write_text(hierarchy, encoding="utf-8")
+
+            manifest_path = staging / "manifest.json"
+            manifest_path.write_text(
+                json.dumps(
+                    _manifest(inspection, serial, link, state, pdd_version, screenshot_path, hierarchy_path),
+                    ensure_ascii=False,
+                    indent=2,
+                    sort_keys=True,
+                )
+                + "\n",
+                encoding="utf-8",
+            )
+            os.replace(staging, target)
+        except (SkuPanelEvidenceError, DeviceConnectionError):
+            _clean_staging(staging)
+            raise
+        except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
+            _clean_staging(staging)
+            raise SkuPanelEvidenceTimeoutError("规格面板只读取证超时，未发布任何证据产物。") from error
+        except Exception as error:
+            _clean_staging(staging)
+            # 第三方异常可能含 serial、Activity 或页面正文，不能直接向 CLI/日志传播。
+            raise SkuPanelEvidenceError("规格面板只读取证未完成，未发布任何证据产物。") from error
+
+        return SkuPanelEvidenceResult(
+            output_directory=target,
+            manifest_path=target / "manifest.json",
+            screenshot_path=target / "screenshot.png",
+            hierarchy_path=target / "hierarchy.xml",
+        )
+
+
+def _is_positive_finite(value: object) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0 and isfinite(value)
+
+
+def _validate_human_declared_state(value: object) -> str:
+    if not isinstance(value, str) or value not in HUMAN_DECLARED_STATES:
+        raise SkuPanelDeclaredStateError("必须提供允许的人工声明规格面板状态。")
+    return value
+
+
+def _validate_new_target(target: Path) -> None:
+    if target.exists():
+        raise SkuPanelEvidenceError("输出目录已存在；为防止混入旧证据，拒绝覆盖。")
+    if not target.name:
+        raise SkuPanelEvidenceError("输出目录必须是明确的新目录。")
+
+
+def _clean_staging(staging: Path | None) -> None:
+    if staging is not None and staging.exists():
+        # staging 仅在本次调用中创建，绝不删除调用方已存在的目录。
+        shutil.rmtree(staging)
+
+
+def _require_expected_version(app_info: object) -> str:
+    if not isinstance(app_info, dict):
+        raise SkuPanelVersionMismatchError("拼多多版本与已取证版本不一致，已停止取证。")
+    version = app_info.get("versionName") or app_info.get("version_name")
+    if not isinstance(version, str) or version != EXPECTED_PDD_VERSION:
+        raise SkuPanelVersionMismatchError("拼多多版本与已取证版本不一致，已停止取证。")
+    return version
+
+
+def _require_pdd_foreground(current: object) -> None:
+    if not isinstance(current, dict) or current.get("package") != PDD_PACKAGE:
+        raise SkuPanelPackageMismatchError("当前前台应用不是拼多多，已停止取证。")
+
+
+def _manifest(
+    inspection: DeviceInspection,
+    serial: str,
+    link: ProductUrl,
+    human_declared_state: str,
+    pdd_version: str,
+    screenshot_path: Path,
+    hierarchy_path: Path,
+) -> dict[str, Any]:
+    """仅记录人工声明与非敏感审计摘要，不写入 Activity 或页面内容。"""
+
+    return {
+        "schema_version": 1,
+        "captured_at": datetime.now(UTC).isoformat(),
+        "product": {"goods_id": link.goods_id, "canonical_url": link.canonical_url},
+        "human_declared_state": human_declared_state,
+        "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": pdd_version,
+        },
+        "artifacts": [
+            {"path": screenshot_path.name, "sha256": _sha256_file(screenshot_path)},
+            {"path": hierarchy_path.name, "sha256": _sha256_file(hierarchy_path)},
+        ],
+    }

client/tests/device/__init__.py

@@ -0,0 +1 @@
+"""设备连接与基线取证的离线测试。"""

client/tests/device/test_adb.py

@@ -0,0 +1,273 @@
+"""ADB 设备边界测试：所有命令执行器均为 mock，不连接真机。"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from pathlib import Path
+import subprocess
+import sys
+import unittest
+
+
+CLIENT_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(CLIENT_ROOT / "src"))
+
+from cmbuyer_client.device.adb import (
+    AdbClient,
+    CommandResult,
+    DeviceIdentityUnconfirmedError,
+    DeviceCommandError,
+    DeviceCommandTimeoutError,
+    DeviceNotFoundError,
+    DeviceOfflineError,
+    DeviceStateError,
+    DeviceUnauthorizedError,
+    DuplicatePhysicalDeviceError,
+    IntentLaunchUnconfirmedError,
+    SerialRequiredError,
+)
+
+
+USB_SERIAL = "3B65BD02H7F00000"
+WIFI_SERIAL = "192.168.0.173:5555"
+
+
+class FakeRunner:
+    def __init__(self, devices_output: str, properties: dict[tuple[str, str], CommandResult | str]) -> None:
+        self.devices_output = devices_output
+        self.properties = properties
+        self.calls: list[tuple[str, ...]] = []
+
+    def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
+        self.calls.append(tuple(arguments))
+        if tuple(arguments) == ("devices", "-l"):
+            return CommandResult(stdout=self.devices_output)
+        key = (arguments[1], arguments[-1])
+        value = self.properties.get(key, "")
+        return value if isinstance(value, CommandResult) else CommandResult(stdout=value)
+
+
+def _properties(serials: tuple[str, ...]) -> dict[tuple[str, str], str]:
+    values: dict[tuple[str, str], str] = {}
+    for serial in serials:
+        values[(serial, "ro.serialno")] = "physical-phone-1"
+        values[(serial, "ro.boot.serialno")] = "physical-phone-1"
+        values[(serial, "ro.product.model")] = "PKG110"
+        values[(serial, "ro.product.name")] = "PKG110"
+        values[(serial, "ro.product.device")] = "OP5D2BL1"
+        values[(serial, "ro.build.version.release")] = "16"
+    return values
+
+
+class AdbClientTests(unittest.TestCase):
+    def test_requires_explicit_serial(self) -> None:
+        runner = FakeRunner("List of devices attached\n", {})
+
+        with self.assertRaises(SerialRequiredError):
+            AdbClient(runner).inspect("   ")
+
+        self.assertEqual(runner.calls, [])
+
+    def test_missing_offline_and_unauthorized_are_distinct(self) -> None:
+        missing = AdbClient(FakeRunner("List of devices attached\n", {}))
+        with self.assertRaises(DeviceNotFoundError):
+            missing.inspect(USB_SERIAL)
+
+        offline = AdbClient(FakeRunner(f"List of devices attached\n{USB_SERIAL}\toffline\n", {}))
+        with self.assertRaises(DeviceOfflineError):
+            offline.inspect(USB_SERIAL)
+
+        unauthorized = AdbClient(FakeRunner(f"List of devices attached\n{USB_SERIAL}\tunauthorized\n", {}))
+        with self.assertRaises(DeviceUnauthorizedError):
+            unauthorized.inspect(USB_SERIAL)
+
+    def test_two_channels_with_same_physical_identity_fail_closed(self) -> None:
+        output = (
+            "List of devices attached\n"
+            f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
+            f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
+        )
+        runner = FakeRunner(output, _properties((WIFI_SERIAL, USB_SERIAL)))
+
+        with self.assertRaises(DuplicatePhysicalDeviceError):
+            AdbClient(runner).inspect(USB_SERIAL)
+
+        self.assertIn(("-s", WIFI_SERIAL, "shell", "getprop", "ro.serialno"), runner.calls)
+        self.assertIn(("-s", USB_SERIAL, "shell", "getprop", "ro.serialno"), runner.calls)
+
+    def test_multiple_online_devices_with_failed_identity_fail_closed(self) -> None:
+        output = (
+            "List of devices attached\n"
+            f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
+            f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
+        )
+        properties = _properties((WIFI_SERIAL, USB_SERIAL))
+        properties[(WIFI_SERIAL, "ro.serialno")] = CommandResult(stdout="", returncode=1)
+        runner = FakeRunner(output, properties)
+
+        with self.assertRaises(DeviceIdentityUnconfirmedError):
+            AdbClient(runner).inspect(USB_SERIAL)
+
+    def test_online_explicit_serial_reads_non_sensitive_metadata(self) -> None:
+        output = f"List of devices attached\n{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
+        inspection = AdbClient(FakeRunner(output, _properties((USB_SERIAL,)))).inspect(USB_SERIAL)
+
+        self.assertEqual(inspection.device.serial, USB_SERIAL)
+        self.assertEqual(inspection.model, "PKG110")
+        self.assertEqual(inspection.android_version, "16")
+
+    def test_single_online_device_does_not_require_hardware_identity(self) -> None:
+        output = f"List of devices attached\n{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
+        properties = _properties((USB_SERIAL,))
+        properties[(USB_SERIAL, "ro.serialno")] = ""
+        properties[(USB_SERIAL, "ro.boot.serialno")] = ""
+
+        inspection = AdbClient(FakeRunner(output, properties)).inspect(USB_SERIAL)
+
+        self.assertEqual(inspection.model, "PKG110")
+
+    def test_multiple_online_devices_without_hardware_identity_are_unconfirmed(self) -> None:
+        output = (
+            "List of devices attached\n"
+            f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
+            f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
+        )
+        properties = _properties((WIFI_SERIAL, USB_SERIAL))
+        properties[(WIFI_SERIAL, "ro.serialno")] = ""
+        properties[(WIFI_SERIAL, "ro.boot.serialno")] = ""
+
+        with self.assertRaises(DeviceIdentityUnconfirmedError):
+            AdbClient(FakeRunner(output, properties)).inspect(USB_SERIAL)
+
+    def test_multiple_online_devices_with_different_identity_keep_explicit_selection(self) -> None:
+        output = (
+            "List of devices attached\n"
+            f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
+            f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
+        )
+        properties = _properties((WIFI_SERIAL, USB_SERIAL))
+        properties[(WIFI_SERIAL, "ro.serialno")] = "physical-phone-2"
+        properties[(WIFI_SERIAL, "ro.boot.serialno")] = "physical-phone-2"
+
+        inspection = AdbClient(FakeRunner(output, properties)).inspect(USB_SERIAL)
+
+        self.assertEqual(inspection.device.serial, USB_SERIAL)
+
+    def test_shared_boot_serial_is_duplicate_even_when_ro_serial_differs(self) -> None:
+        output = (
+            "List of devices attached\n"
+            f"{WIFI_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
+            f"{USB_SERIAL}\tdevice product:PKG110 model:PKG110 device:OP5D2BL1\n"
+        )
+        properties = _properties((WIFI_SERIAL, USB_SERIAL))
+        properties[(WIFI_SERIAL, "ro.serialno")] = "wifi-transport-serial"
+        properties[(USB_SERIAL, "ro.serialno")] = "usb-transport-serial"
+        properties[(WIFI_SERIAL, "ro.boot.serialno")] = "shared-hardware-serial"
+        properties[(USB_SERIAL, "ro.boot.serialno")] = "shared-hardware-serial"
+
+        with self.assertRaises(DuplicatePhysicalDeviceError):
+            AdbClient(FakeRunner(output, properties)).inspect(USB_SERIAL)
+
+    def test_unknown_adb_state_is_rejected(self) -> None:
+        client = AdbClient(FakeRunner(f"List of devices attached\n{USB_SERIAL}\trecovery\n", {}))
+
+        with self.assertRaises(DeviceStateError):
+            client.inspect(USB_SERIAL)
+
+    def test_runner_timeout_is_a_distinct_connection_error(self) -> None:
+        class TimeoutRunner:
+            def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
+                raise subprocess.TimeoutExpired(arguments, timeout_seconds)
+
+        with self.assertRaises(DeviceCommandTimeoutError):
+            AdbClient(TimeoutRunner()).inspect(USB_SERIAL)
+
+    def test_product_intent_is_fixed_to_action_view_and_pdd_package(self) -> None:
+        class IntentRunner:
+            def __init__(self) -> None:
+                self.calls: list[tuple[str, ...]] = []
+
+            def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
+                self.calls.append(tuple(arguments))
+                return CommandResult(stdout="Status: ok\n")
+
+        runner = IntentRunner()
+        summary = AdbClient(runner).start_pdd_view_intent(
+            USB_SERIAL,
+            "123",
+        )
+
+        self.assertEqual(summary.status, "ok")
+        self.assertEqual(
+            runner.calls,
+            [
+                (
+                    "-s",
+                    USB_SERIAL,
+                    "shell",
+                    "am",
+                    "start",
+                    "-W",
+                    "-a",
+                    "android.intent.action.VIEW",
+                    "-d",
+                    "https://mobile.yangkeduo.com/goods.html?goods_id=123",
+                    "-p",
+                    "com.xunmeng.pinduoduo",
+                )
+            ],
+        )
+
+    def test_product_intent_without_explicit_success_is_rejected(self) -> None:
+        class UnknownIntentRunner:
+            def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
+                return CommandResult(stdout="Starting: Intent { ... }\n")
+
+        with self.assertRaises(IntentLaunchUnconfirmedError):
+            AdbClient(UnknownIntentRunner()).start_pdd_view_intent(
+                USB_SERIAL,
+                "123",
+            )
+
+    def test_product_intent_rejects_invalid_goods_id_before_runner(self) -> None:
+        class RecordingRunner:
+            def __init__(self) -> None:
+                self.calls: list[tuple[str, ...]] = []
+
+            def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
+                self.calls.append(tuple(arguments))
+                return CommandResult(stdout="Status: ok\n")
+
+        invalid_values: tuple[object, ...] = (
+            "",
+            "12a",
+            "１２３",
+            " 123",
+            "123 ",
+            "https://mobile.yangkeduo.com/goods.html?goods_id=123",
+            "am start -W -d anything",
+            123,
+            None,
+        )
+        for value in invalid_values:
+            with self.subTest(value=repr(value)):
+                runner = RecordingRunner()
+                with self.assertRaises(ValueError):
+                    AdbClient(runner).start_pdd_view_intent(USB_SERIAL, value)  # type: ignore[arg-type]
+                self.assertEqual(runner.calls, [])
+
+    def test_product_intent_nonzero_and_timeout_remain_distinct(self) -> None:
+        class FailedIntentRunner:
+            def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
+                return CommandResult(stdout="sensitive command output", returncode=1)
+
+        class TimeoutIntentRunner:
+            def run(self, arguments: Sequence[str], timeout_seconds: float) -> CommandResult:
+                raise subprocess.TimeoutExpired(arguments, timeout_seconds)
+
+        with self.assertRaises(DeviceCommandError) as command_error:
+            AdbClient(FailedIntentRunner()).start_pdd_view_intent(USB_SERIAL, "123")
+        self.assertNotIn("sensitive command output", str(command_error.exception))
+
+        with self.assertRaises(DeviceCommandTimeoutError):
+            AdbClient(TimeoutIntentRunner()).start_pdd_view_intent(USB_SERIAL, "123")

client/tests/device/test_baseline.py

@@ -0,0 +1,281 @@
+"""基线取证测试：mock ADB/uiautomator2，不连接手机。"""
+
+from __future__ import annotations
+
+from pathlib import Path
+import base64
+from io import BytesIO
+import sys
+import tempfile
+import unittest
+
+from PIL import Image
+from uiautomator2.exceptions import HTTPTimeoutError
+
+
+CLIENT_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(CLIENT_ROOT / "src"))
+sys.path.insert(0, str(CLIENT_ROOT / "scripts"))
+
+from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
+from cmbuyer_client.device.baseline import (
+    BaselineCaptureError,
+    BaselineCaptureTimeoutError,
+    DeviceBaselineCapturer,
+    NoReconnectUiautomatorConnector,
+    PDD_PACKAGE,
+)
+from capture_device_baseline import parse_arguments, validate_arguments
+
+
+SERIAL = "USB-serial-for-test"
+
+
+class StaticAdbClient:
+    def __init__(self) -> None:
+        self.serials: list[str] = []
+
+    def inspect(self, serial: str) -> DeviceInspection:
+        self.serials.append(serial)
+        return DeviceInspection(
+            device=AdbDevice(serial=serial, state="device", model="Test Model"),
+            model="Test Model",
+            android_version="16",
+        )
+
+
+class FakeUiDevice:
+    def __init__(self, fail_dump: bool = False) -> None:
+        self.fail_dump = fail_dump
+        self.rpc_calls: list[tuple[str, object, float]] = []
+        self.app_info_calls: list[str] = []
+
+    def app_info(self, package_name: str) -> dict[str, str]:
+        self.app_info_calls.append(package_name)
+        return {"versionName": "8.17.0"}
+
+    def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+        self.rpc_calls.append((method, params, timeout))
+        if method == "takeScreenshot":
+            image_data = BytesIO()
+            Image.new("RGB", (1, 1), color="white").save(image_data, format="PNG")
+            return base64.b64encode(image_data.getvalue()).decode("ascii")
+        if method != "dumpWindowHierarchy":
+            raise AssertionError(f"unexpected method: {method}")
+        if self.fail_dump:
+            raise RuntimeError("mock dump failed")
+        return "<hierarchy><node text='page body must stay out of manifest'/></hierarchy>"
+
+
+class BaselineCaptureTests(unittest.TestCase):
+    def test_capture_writes_hashes_without_xml_or_raw_serial_in_manifest(self) -> None:
+        adb = StaticAdbClient()
+        device = FakeUiDevice()
+        capturer = DeviceBaselineCapturer(adb, lambda serial: device, timeout_seconds=7.5)
+
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "baseline"
+            result = capturer.capture(SERIAL, output)
+            manifest = result.manifest_path.read_text(encoding="utf-8")
+
+            self.assertEqual(adb.serials, [SERIAL])
+            self.assertEqual(device.app_info_calls, [PDD_PACKAGE])
+            self.assertEqual(
+                device.rpc_calls,
+                [
+                    ("takeScreenshot", [1, 80], 7.5),
+                    ("dumpWindowHierarchy", [False, 50], 7.5),
+                ],
+            )
+            self.assertTrue(result.screenshot_path.is_file())
+            self.assertTrue(result.hierarchy_path.is_file())
+            self.assertIn('"sha256"', manifest)
+            self.assertNotIn("page body must stay out of manifest", manifest)
+            self.assertNotIn(SERIAL, manifest)
+            self.assertIn('"channel": "usb"', manifest)
+
+    def test_capture_failure_cleans_staging_and_does_not_publish_partial_output(self) -> None:
+        device = FakeUiDevice(fail_dump=True)
+        capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: device, timeout_seconds=5)
+
+        with tempfile.TemporaryDirectory() as directory:
+            parent = Path(directory)
+            output = parent / "baseline"
+            with self.assertRaises(BaselineCaptureError) as raised:
+                capturer.capture(SERIAL, output)
+
+            self.assertFalse(output.exists())
+            self.assertEqual(list(parent.iterdir()), [])
+            self.assertNotIn("mock dump failed", str(raised.exception))
+
+    def test_existing_output_is_never_overwritten(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "baseline"
+            output.mkdir()
+            sentinel = output / "keep.txt"
+            sentinel.write_text("preserve", encoding="utf-8")
+            capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: FakeUiDevice(), timeout_seconds=5)
+
+            with self.assertRaises(BaselineCaptureError):
+                capturer.capture(SERIAL, output)
+            self.assertEqual(sentinel.read_text(encoding="utf-8"), "preserve")
+
+    def test_no_reconnect_connector_passes_only_current_adb_device_object(self) -> None:
+        class ListedDevice:
+            serial = SERIAL
+
+        listed = ListedDevice()
+        connected: list[object] = []
+
+        connector = NoReconnectUiautomatorConnector(lambda: [listed], lambda device: connected.append(device) or FakeUiDevice())
+        connector(SERIAL)
+
+        self.assertEqual(connected, [listed])
+
+    def test_no_reconnect_connector_refuses_disappeared_serial(self) -> None:
+        connector = NoReconnectUiautomatorConnector(lambda: [], lambda device: FakeUiDevice())
+
+        with self.assertRaises(BaselineCaptureError) as raised:
+            connector(SERIAL)
+        self.assertIn("拒绝自动重连", str(raised.exception))
+
+    def test_connector_exception_is_redacted_and_publishes_no_partial_output(self) -> None:
+        def failing_connector(serial: str) -> FakeUiDevice:
+            raise RuntimeError(f"third party leaked {serial}")
+
+        capturer = DeviceBaselineCapturer(StaticAdbClient(), failing_connector, timeout_seconds=5)
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "baseline"
+            with self.assertRaises(BaselineCaptureError) as raised:
+                capturer.capture(SERIAL, output)
+
+            self.assertNotIn(SERIAL, str(raised.exception))
+            self.assertFalse(output.exists())
+            self.assertEqual(list(Path(directory).iterdir()), [])
+
+    def test_invalid_screenshot_base64_syntax_fails_closed_without_partial_output(self) -> None:
+        class InvalidScreenshotDevice(FakeUiDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "takeScreenshot":
+                    valid = super().jsonrpc_call(method, params, timeout)
+                    return valid[:12] + "!" + valid[12:]
+                return super().jsonrpc_call(method, params, timeout)
+
+        capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: InvalidScreenshotDevice(), timeout_seconds=5)
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "baseline"
+            with self.assertRaises(BaselineCaptureError) as raised:
+                capturer.capture(SERIAL, output)
+
+            self.assertIn("Base64 语法无效", str(raised.exception))
+            self.assertNotIn("!", str(raised.exception))
+            self.assertFalse(output.exists())
+            self.assertEqual(list(Path(directory).iterdir()), [])
+
+    def test_invalid_padding_and_unapproved_ascii_whitespace_fail_closed(self) -> None:
+        invalid_insertions = {
+            "padding": lambda value: value[:-1],
+            "vertical-tab": lambda value: value[:12] + "\v" + value[12:],
+            "form-feed": lambda value: value[:12] + "\f" + value[12:],
+        }
+
+        for name, make_invalid in invalid_insertions.items():
+            with self.subTest(name=name), tempfile.TemporaryDirectory() as directory:
+                class InvalidScreenshotDevice(FakeUiDevice):
+                    def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                        value = super().jsonrpc_call(method, params, timeout)
+                        if method == "takeScreenshot":
+                            return make_invalid(value)
+                        return value
+
+                output = Path(directory) / "baseline"
+                capturer = DeviceBaselineCapturer(
+                    StaticAdbClient(),
+                    lambda serial: InvalidScreenshotDevice(),
+                    timeout_seconds=5,
+                )
+                with self.assertRaises(BaselineCaptureError) as raised:
+                    capturer.capture(SERIAL, output)
+
+                self.assertIn("Base64 语法无效", str(raised.exception))
+                self.assertNotIn(SERIAL, str(raised.exception))
+                self.assertFalse(output.exists())
+                self.assertEqual(list(Path(directory).iterdir()), [])
+
+    def test_base64_decoded_nonimage_fails_closed_without_partial_output(self) -> None:
+        class NonImageScreenshotDevice(FakeUiDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "takeScreenshot":
+                    return base64.b64encode(b"not an image").decode("ascii")
+                return super().jsonrpc_call(method, params, timeout)
+
+        capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: NonImageScreenshotDevice(), timeout_seconds=5)
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "baseline"
+            with self.assertRaises(BaselineCaptureError) as raised:
+                capturer.capture(SERIAL, output)
+
+            self.assertIn("图像数据无效", str(raised.exception))
+            self.assertNotIn("not an image", str(raised.exception))
+            self.assertFalse(output.exists())
+            self.assertEqual(list(Path(directory).iterdir()), [])
+
+    def test_ascii_base64_whitespace_is_normalized_before_strict_decode(self) -> None:
+        class WhitespaceScreenshotDevice(FakeUiDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                value = super().jsonrpc_call(method, params, timeout)
+                if method == "takeScreenshot":
+                    return value[:10] + " \t\r\n" + value[10:30] + "\n" + value[30:]
+                return value
+
+        capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: WhitespaceScreenshotDevice(), timeout_seconds=5)
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "baseline"
+            result = capturer.capture(SERIAL, output)
+
+            self.assertTrue(result.screenshot_path.is_file())
+            with Image.open(result.screenshot_path) as image:
+                self.assertEqual(image.size, (1, 1))
+
+    def test_invalid_or_non_hierarchy_xml_fails_closed_without_partial_output(self) -> None:
+        class InvalidHierarchyDevice(FakeUiDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "dumpWindowHierarchy":
+                    return "<not-hierarchy/>"
+                return super().jsonrpc_call(method, params, timeout)
+
+        capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: InvalidHierarchyDevice(), timeout_seconds=5)
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "baseline"
+            with self.assertRaises(BaselineCaptureError) as raised:
+                capturer.capture(SERIAL, output)
+
+            self.assertNotIn("not-hierarchy", str(raised.exception))
+            self.assertFalse(output.exists())
+            self.assertEqual(list(Path(directory).iterdir()), [])
+
+    def test_rpc_timeout_is_distinct_redacted_and_does_not_publish_partial_output(self) -> None:
+        class TimeoutRpcDevice(FakeUiDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                raise HTTPTimeoutError(f"raw serial={SERIAL} xml=<hierarchy/>")
+
+        capturer = DeviceBaselineCapturer(StaticAdbClient(), lambda serial: TimeoutRpcDevice(), timeout_seconds=5)
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "baseline"
+            with self.assertRaises(BaselineCaptureTimeoutError) as raised:
+                capturer.capture(SERIAL, output)
+
+            self.assertIn("超时", str(raised.exception))
+            self.assertNotIn(SERIAL, str(raised.exception))
+            self.assertNotIn("hierarchy", str(raised.exception))
+            self.assertFalse(output.exists())
+            self.assertEqual(list(Path(directory).iterdir()), [])
+
+    def test_cli_validation_rejects_empty_serial_and_nonpositive_timeout(self) -> None:
+        empty_serial = parse_arguments(["--serial", "", "--output-dir", "baseline"])
+        with self.assertRaisesRegex(ValueError, "非空 --serial"):
+            validate_arguments(empty_serial)
+
+        nonpositive_timeout = parse_arguments(["--serial", SERIAL, "--output-dir", "baseline", "--timeout", "0"])
+        with self.assertRaisesRegex(ValueError, "必须大于 0"):
+            validate_arguments(nonpositive_timeout)

client/tests/pdd/__init__.py

@@ -0,0 +1 @@
+"""拼多多受限打开模块的离线测试。"""

client/tests/pdd/test_product_open.py

@@ -0,0 +1,266 @@
+"""商品打开围栏的离线测试；所有设备和命令均为 fake。"""
+
+from __future__ import annotations
+
+import base64
+from io import BytesIO
+from pathlib import Path
+import sys
+from tempfile import TemporaryDirectory
+import unittest
+
+from PIL import Image
+
+
+CLIENT_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(CLIENT_ROOT / "src"))
+
+from cmbuyer_client.device.adb import AdbDevice, DeviceInspection, IntentLaunchSummary
+from cmbuyer_client.pdd.product_open import (
+    ProductOpenCapturer,
+    ProductOpenTimeoutError,
+    ProductOpenUiDevice,
+    ProductHierarchyCaptureError,
+    ProductPackageMismatchError,
+    ProductScreenshotCaptureError,
+    ProductVersionMismatchError,
+)
+from cmbuyer_client.pdd.product_url import ProductUrl, ProductUrlError
+
+
+SERIAL = "192.168.0.173:5555"
+URL = "https://mobile.yangkeduo.com/goods.html?goods_id=123"
+HIERARCHY = "<?xml version='1.0' encoding='UTF-8'?><hierarchy rotation='0'><node /></hierarchy>"
+
+
+def _png_base64() -> str:
+    image_data = BytesIO()
+    Image.new("RGB", (1, 1), color="white").save(image_data, format="PNG")
+    return base64.b64encode(image_data.getvalue()).decode("ascii")
+
+
+class FakeAdbClient:
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, str | None]] = []
+        self.inspection = DeviceInspection(
+            device=AdbDevice(serial=SERIAL, state="device", model="PKG110"),
+            model="PKG110",
+            android_version="16",
+        )
+
+    def inspect(self, serial: str) -> DeviceInspection:
+        self.calls.append(("inspect", serial))
+        return self.inspection
+
+    def start_pdd_view_intent(self, serial: str, goods_id: str) -> IntentLaunchSummary:
+        self.calls.append(("intent", goods_id))
+        return IntentLaunchSummary(status="ok", returncode=0)
+
+
+class FakeUiDevice:
+    def __init__(
+        self,
+        *,
+        version: str = "8.17.0",
+        current_package: str = "com.xunmeng.pinduoduo",
+        current_packages: list[str] | None = None,
+        hierarchy: str = HIERARCHY,
+        timeout_on_screenshot: bool = False,
+    ) -> None:
+        self.version = version
+        self.current_package = current_package
+        self.current_packages = list(current_packages) if current_packages is not None else None
+        self.hierarchy = hierarchy
+        self.timeout_on_screenshot = timeout_on_screenshot
+        self.calls: list[str] = []
+
+    def app_info(self, package_name: str) -> dict[str, str]:
+        self.calls.append("app_info")
+        return {"versionName": self.version}
+
+    def app_current(self) -> dict[str, str]:
+        self.calls.append("app_current")
+        if self.current_packages:
+            package = self.current_packages.pop(0)
+            self.current_package = package
+        return {"package": self.current_package, "activity": "sensitive.activity.name"}
+
+    def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+        self.calls.append(method)
+        if method == "takeScreenshot":
+            if self.timeout_on_screenshot:
+                raise TimeoutError("raw remote detail")
+            return _png_base64()
+        if method == "dumpWindowHierarchy":
+            return self.hierarchy
+        raise AssertionError(f"unexpected RPC {method}")
+
+
+class ProductOpenTests(unittest.TestCase):
+    def _capturer(self, adb: FakeAdbClient, device: FakeUiDevice, **kwargs: object) -> ProductOpenCapturer:
+        return ProductOpenCapturer(adb, lambda serial: device, timeout_seconds=2, **kwargs)
+
+    def test_success_uses_canonical_url_and_redacted_atomic_manifest(self) -> None:
+        adb = FakeAdbClient()
+        device = FakeUiDevice()
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "evidence"
+            result = self._capturer(adb, device).open_and_capture(SERIAL, URL, target)
+            manifest = result.manifest_path.read_text(encoding="utf-8")
+
+            self.assertTrue(result.screenshot_path.exists())
+            self.assertTrue(result.hierarchy_path.exists())
+            self.assertEqual(adb.calls, [("inspect", SERIAL), ("intent", "123")])
+            self.assertEqual(device.calls, ["app_info", "app_current", "takeScreenshot", "dumpWindowHierarchy"])
+            self.assertIn('"goods_id": "123"', manifest)
+            self.assertIn('"canonical_url": "https://mobile.yangkeduo.com/goods.html?goods_id=123"', manifest)
+            self.assertNotIn(SERIAL, manifest)
+            self.assertNotIn("sensitive.activity.name", manifest)
+            self.assertNotIn(HIERARCHY, manifest)
+
+    def test_version_mismatch_halts_before_intent(self) -> None:
+        adb = FakeAdbClient()
+        for version in ("8.17.1", " 8.17.0 "):
+            with self.subTest(version=version), TemporaryDirectory() as temporary:
+                target = Path(temporary) / "evidence"
+                with self.assertRaises(ProductVersionMismatchError):
+                    self._capturer(adb, FakeUiDevice(version=version)).open_and_capture(SERIAL, URL, target)
+
+                self.assertEqual(adb.calls[-1:], [("inspect", SERIAL)])
+                self.assertFalse(target.exists())
+
+    def test_public_entry_rejects_caller_constructed_url_value_object(self) -> None:
+        adb = FakeAdbClient()
+        with TemporaryDirectory() as temporary:
+            with self.assertRaises(ProductUrlError):
+                self._capturer(adb, FakeUiDevice()).open_and_capture(
+                    SERIAL,
+                    ProductUrl(goods_id="123", canonical_url="https://example.invalid/"),  # type: ignore[arg-type]
+                    Path(temporary) / "evidence",
+                )
+
+            self.assertEqual(adb.calls, [])
+
+    def test_foreground_package_mismatch_halts_before_capture(self) -> None:
+        adb = FakeAdbClient()
+        device = FakeUiDevice(current_package="com.example.other")
+        clock = FakeClock()
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "evidence"
+            with self.assertRaises(ProductPackageMismatchError):
+                self._capturer(
+                    adb,
+                    device,
+                    foreground_poll_interval_seconds=0.5,
+                    monotonic_clock=clock.monotonic,
+                    sleep_function=clock.sleep,
+                ).open_and_capture(SERIAL, URL, target)
+
+            self.assertEqual(adb.calls, [("inspect", SERIAL), ("intent", "123")])
+            self.assertEqual(device.calls, ["app_info", "app_current", "app_current", "app_current", "app_current", "app_current"])
+            self.assertEqual(clock.sleeps, [0.5, 0.5, 0.5, 0.5])
+            self.assertFalse(target.exists())
+
+    def test_foreground_package_poll_waits_for_pdd_before_reading_evidence(self) -> None:
+        adb = FakeAdbClient()
+        device = FakeUiDevice(current_packages=["com.example.other", "com.xunmeng.pinduoduo"])
+        clock = FakeClock()
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "evidence"
+            result = self._capturer(
+                adb,
+                device,
+                foreground_poll_interval_seconds=0.25,
+                monotonic_clock=clock.monotonic,
+                sleep_function=clock.sleep,
+            ).open_and_capture(SERIAL, URL, target)
+
+            self.assertTrue(result.manifest_path.exists())
+            self.assertEqual(clock.sleeps, [0.25])
+            self.assertEqual(
+                device.calls,
+                ["app_info", "app_current", "app_current", "takeScreenshot", "dumpWindowHierarchy"],
+            )
+
+    def test_foreground_package_poll_stops_at_deadline_without_evidence(self) -> None:
+        adb = FakeAdbClient()
+        device = FakeUiDevice(current_packages=["com.example.other", "", "com.example.other"])
+        clock = FakeClock()
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "evidence"
+            with self.assertRaises(ProductPackageMismatchError):
+                self._capturer(
+                    adb,
+                    device,
+                    foreground_poll_interval_seconds=0.8,
+                    monotonic_clock=clock.monotonic,
+                    sleep_function=clock.sleep,
+                ).open_and_capture(SERIAL, URL, target)
+
+            self.assertEqual(len(clock.sleeps), 3)
+            for actual, expected in zip(clock.sleeps, (0.8, 0.8, 0.4), strict=True):
+                self.assertAlmostEqual(actual, expected)
+            self.assertEqual(device.calls, ["app_info", "app_current", "app_current", "app_current", "app_current"])
+            self.assertFalse(target.exists())
+            self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
+
+    def test_foreground_poll_interval_must_be_positive_and_finite(self) -> None:
+        for interval in (0, -0.1, float("inf"), float("nan"), True):
+            with self.subTest(interval=interval):
+                with self.assertRaises(ValueError):
+                    ProductOpenCapturer(
+                        FakeAdbClient(),
+                        lambda serial: FakeUiDevice(),
+                        timeout_seconds=2,
+                        foreground_poll_interval_seconds=interval,  # type: ignore[arg-type]
+                    )
+
+    def test_timeout_and_invalid_hierarchy_leave_no_partial_evidence(self) -> None:
+        scenarios = (
+            (FakeUiDevice(timeout_on_screenshot=True), ProductOpenTimeoutError),
+            (FakeUiDevice(hierarchy="<not-hierarchy />"), ProductHierarchyCaptureError),
+        )
+        for device, error_type in scenarios:
+            with self.subTest(error_type=error_type.__name__), TemporaryDirectory() as temporary:
+                target = Path(temporary) / "evidence"
+                with self.assertRaises(error_type):
+                    self._capturer(FakeAdbClient(), device).open_and_capture(SERIAL, URL, target)
+
+                self.assertFalse(target.exists())
+                self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
+
+    def test_invalid_screenshot_is_a_distinct_redacted_failure(self) -> None:
+        class InvalidScreenshotDevice(FakeUiDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "takeScreenshot":
+                    self.calls.append(method)
+                    return "not valid base64!"
+                return super().jsonrpc_call(method, params, timeout)
+
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "evidence"
+            with self.assertRaises(ProductScreenshotCaptureError) as raised:
+                self._capturer(FakeAdbClient(), InvalidScreenshotDevice()).open_and_capture(SERIAL, URL, target)
+
+            self.assertNotIn("base64", str(raised.exception).lower())
+            self.assertFalse(target.exists())
+            self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
+
+    def test_read_only_protocol_has_no_ui_operation_methods(self) -> None:
+        forbidden = {"click", "swipe", "send_keys", "set_text", "press", "long_click"}
+
+        self.assertTrue(forbidden.isdisjoint(ProductOpenUiDevice.__dict__))
+        self.assertEqual(base64.b64decode(_png_base64())[:8], b"\x89PNG\r\n\x1a\n")
+
+
+class FakeClock:
+    def __init__(self) -> None:
+        self.value = 0.0
+        self.sleeps: list[float] = []
+
+    def monotonic(self) -> float:
+        return self.value
+
+    def sleep(self, seconds: float) -> None:
+        self.sleeps.append(seconds)
+        self.value += seconds

client/tests/pdd/test_product_url.py

@@ -0,0 +1,49 @@
+"""canonical 商品 URL 的离线解析测试。"""
+
+from __future__ import annotations
+
+from pathlib import Path
+import sys
+import unittest
+
+
+CLIENT_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(CLIENT_ROOT / "src"))
+
+from cmbuyer_client.pdd.product_url import ProductUrlError, parse_product_url
+
+
+class ProductUrlTests(unittest.TestCase):
+    def test_rebuilds_url_from_goods_id(self) -> None:
+        link = parse_product_url("https://mobile.yangkeduo.com/goods.html?goods_id=00123")
+
+        self.assertEqual(link.goods_id, "00123")
+        self.assertEqual(
+            link.canonical_url,
+            "https://mobile.yangkeduo.com/goods.html?goods_id=00123",
+        )
+
+    def test_rejects_noncanonical_and_ambiguous_urls(self) -> None:
+        rejected = (
+            "http://mobile.yangkeduo.com/goods.html?goods_id=123",
+            "https://other.example/goods.html?goods_id=123",
+            "https://mobile.yangkeduo.com/other.html?goods_id=123",
+            "https://user@mobile.yangkeduo.com/goods.html?goods_id=123",
+            "https://mobile.yangkeduo.com:8443/goods.html?goods_id=123",
+            "https://mobile.yangkeduo.com:443/goods.html?goods_id=123",
+            "https://mobile.yangkeduo.com/goods.html?goods_id=123#fragment",
+            "https://mobile.yangkeduo.com/goods.html",
+            "https://mobile.yangkeduo.com/goods.html?goods_id=123&goods_id=456",
+            "https://mobile.yangkeduo.com/goods.html?goods_id=123&source=share",
+            "https://mobile.yangkeduo.com/goods.html?goods_id=12a",
+            "https://mobile.yangkeduo.com/goods.html?goods_id=%EF%BC%91%EF%BC%92%EF%BC%93",
+            "https://mobile.yangkeduo.com/goods.html?goods_id=",
+            " https://mobile.yangkeduo.com/goods.html?goods_id=123",
+            "https://MOBILE.YANGKEDUO.COM/goods.html?goods_id=123",
+            "https://mobile.yangkeduo.com/goods.html?goods_id=%31%32%33",
+        )
+
+        for value in rejected:
+            with self.subTest(value=value):
+                with self.assertRaises(ProductUrlError):
+                    parse_product_url(value)

client/tests/pdd/test_sku_panel_spike.py

@@ -0,0 +1,258 @@
+"""人工声明规格面板状态的离线只读取证测试。"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+from importlib.util import module_from_spec, spec_from_file_location
+from io import BytesIO
+from pathlib import Path
+import sys
+from tempfile import TemporaryDirectory
+import unittest
+
+from PIL import Image
+
+
+CLIENT_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(CLIENT_ROOT / "src"))
+
+from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
+from cmbuyer_client.pdd.product_url import ProductUrlError
+from cmbuyer_client.pdd.sku_panel_spike import (
+    HUMAN_DECLARED_STATES,
+    SkuPanelDeclaredStateError,
+    SkuPanelEvidenceCapturer,
+    SkuPanelEvidenceError,
+    SkuPanelEvidenceTimeoutError,
+    SkuPanelHierarchyError,
+    SkuPanelPackageMismatchError,
+    SkuPanelScreenshotError,
+    SkuPanelUiDevice,
+    SkuPanelVersionMismatchError,
+)
+
+
+SERIAL = "192.168.0.173:5555"
+URL = "https://mobile.yangkeduo.com/goods.html?goods_id=123"
+HIERARCHY = "<?xml version='1.0' encoding='UTF-8'?><hierarchy><node text='sensitive page text' /></hierarchy>"
+
+
+def _png_base64() -> str:
+    image_data = BytesIO()
+    Image.new("RGB", (1, 1), color="white").save(image_data, format="PNG")
+    return base64.b64encode(image_data.getvalue()).decode("ascii")
+
+
+class FakeAdbClient:
+    def __init__(self) -> None:
+        self.calls: list[str] = []
+        self.inspection = DeviceInspection(
+            device=AdbDevice(serial=SERIAL, state="device", model="PKG110"),
+            model="PKG110",
+            android_version="16",
+        )
+
+    def inspect(self, serial: str) -> DeviceInspection:
+        self.calls.append(serial)
+        return self.inspection
+
+
+class FakeUiDevice:
+    def __init__(
+        self,
+        *,
+        version: str = "8.17.0",
+        package: str = "com.xunmeng.pinduoduo",
+        screenshot: str | None = None,
+        hierarchy: str = HIERARCHY,
+    ) -> None:
+        self.version = version
+        self.package = package
+        self.screenshot = screenshot if screenshot is not None else _png_base64()
+        self.hierarchy = hierarchy
+        self.calls: list[str] = []
+
+    def app_info(self, package_name: str) -> dict[str, str]:
+        self.calls.append("app_info")
+        return {"versionName": self.version}
+
+    def app_current(self) -> dict[str, str]:
+        self.calls.append("app_current")
+        return {"package": self.package, "activity": "sensitive.activity.name"}
+
+    def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+        self.calls.append(method)
+        if method == "takeScreenshot":
+            return self.screenshot
+        if method == "dumpWindowHierarchy":
+            return self.hierarchy
+        raise AssertionError(f"unexpected RPC {method}")
+
+
+def _load_spike_script() -> object:
+    script_path = CLIENT_ROOT / "scripts" / "capture_sku_panel_spike.py"
+    spec = spec_from_file_location("capture_sku_panel_spike_for_test", script_path)
+    assert spec is not None and spec.loader is not None
+    module = module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class SkuPanelEvidenceTests(unittest.TestCase):
+    def _capturer(self, adb: FakeAdbClient, device: FakeUiDevice) -> SkuPanelEvidenceCapturer:
+        return SkuPanelEvidenceCapturer(adb, lambda serial: device, timeout_seconds=2)
+
+    def test_all_human_declared_states_publish_redacted_manifest(self) -> None:
+        for state in sorted(HUMAN_DECLARED_STATES):
+            with self.subTest(state=state), TemporaryDirectory() as temporary:
+                adb = FakeAdbClient()
+                device = FakeUiDevice()
+                target = Path(temporary) / "evidence"
+                result = self._capturer(adb, device).capture(SERIAL, URL, state, target)
+                manifest = result.manifest_path.read_text(encoding="utf-8")
+
+                self.assertEqual(adb.calls, [SERIAL])
+                self.assertEqual(device.calls, ["app_info", "app_current", "takeScreenshot", "dumpWindowHierarchy"])
+                self.assertIn(f'"human_declared_state": "{state}"', manifest)
+                self.assertIn('"goods_id": "123"', manifest)
+                self.assertIn('"canonical_url": "https://mobile.yangkeduo.com/goods.html?goods_id=123"', manifest)
+                self.assertNotIn("detected_state", manifest)
+                self.assertNotIn(SERIAL, manifest)
+                self.assertNotIn("sensitive.activity.name", manifest)
+                self.assertNotIn(HIERARCHY, manifest)
+
+    def test_invalid_state_and_url_fail_before_device_access(self) -> None:
+        adb = FakeAdbClient()
+        with TemporaryDirectory() as temporary:
+            with self.assertRaises(SkuPanelDeclaredStateError):
+                self._capturer(adb, FakeUiDevice()).capture(SERIAL, URL, "guessed", Path(temporary) / "state")
+            with self.assertRaises(ProductUrlError):
+                self._capturer(adb, FakeUiDevice()).capture(
+                    SERIAL,
+                    "https://mobile.yangkeduo.com/goods.html?goods_id=12x",
+                    "initial",
+                    Path(temporary) / "url",
+                )
+
+        self.assertEqual(adb.calls, [])
+
+    def test_version_or_foreground_package_mismatch_stops_before_artifacts(self) -> None:
+        scenarios = (
+            (FakeUiDevice(version="8.17.1"), SkuPanelVersionMismatchError, ["app_info"]),
+            (FakeUiDevice(package="com.example.other"), SkuPanelPackageMismatchError, ["app_info", "app_current"]),
+        )
+        for device, error_type, expected_calls in scenarios:
+            with self.subTest(error_type=error_type.__name__), TemporaryDirectory() as temporary:
+                target = Path(temporary) / "evidence"
+                with self.assertRaises(error_type):
+                    self._capturer(FakeAdbClient(), device).capture(SERIAL, URL, "initial", target)
+
+                self.assertEqual(device.calls, expected_calls)
+                self.assertFalse(target.exists())
+
+    def test_screenshot_and_xml_failure_leave_no_partial_evidence(self) -> None:
+        scenarios = (
+            (FakeUiDevice(screenshot="not valid base64!"), SkuPanelScreenshotError),
+            (FakeUiDevice(hierarchy="<not-hierarchy />"), SkuPanelHierarchyError),
+        )
+        for device, error_type in scenarios:
+            with self.subTest(error_type=error_type.__name__), TemporaryDirectory() as temporary:
+                target = Path(temporary) / "evidence"
+                with self.assertRaises(error_type):
+                    self._capturer(FakeAdbClient(), device).capture(SERIAL, URL, "initial", target)
+
+                self.assertFalse(target.exists())
+                self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
+
+    def test_screenshot_timeout_is_redacted_and_leaves_no_partial_evidence(self) -> None:
+        class TimeoutScreenshotDevice(FakeUiDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "takeScreenshot":
+                    self.calls.append(method)
+                    raise TimeoutError("adb 192.168.0.173:5555 raw detail")
+                return super().jsonrpc_call(method, params, timeout)
+
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "evidence"
+            with self.assertRaises(SkuPanelEvidenceTimeoutError) as raised:
+                self._capturer(FakeAdbClient(), TimeoutScreenshotDevice()).capture(SERIAL, URL, "initial", target)
+
+            self.assertNotIn(SERIAL, str(raised.exception))
+            self.assertNotIn("adb", str(raised.exception).lower())
+            self.assertFalse(target.exists())
+            self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
+
+    def test_existing_output_directory_is_not_overwritten_or_connected(self) -> None:
+        adb = FakeAdbClient()
+        device = FakeUiDevice()
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "evidence"
+            target.mkdir()
+            sentinel = target / "sentinel.txt"
+            sentinel.write_text("keep", encoding="utf-8")
+
+            with self.assertRaises(SkuPanelEvidenceError):
+                self._capturer(adb, device).capture(SERIAL, URL, "initial", target)
+
+            self.assertEqual(adb.calls, [])
+            self.assertEqual(device.calls, [])
+            self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")
+
+    def test_protocol_has_no_ui_or_purchase_operation_methods(self) -> None:
+        forbidden = {
+            "click",
+            "swipe",
+            "send_keys",
+            "set_text",
+            "press",
+            "open_product",
+            "open_sku_panel",
+            "set_quantity",
+            "go_to_order_confirm",
+            "submit_order",
+            "pay",
+        }
+        self.assertTrue(forbidden.isdisjoint(SkuPanelUiDevice.__dict__))
+        self.assertEqual({name for name in SkuPanelEvidenceCapturer.__dict__ if not name.startswith("_")}, {"capture"})
+
+
+class SkuPanelSpikeCliTests(unittest.TestCase):
+    def test_validate_arguments_rejects_invalid_serial_timeout_state_and_url(self) -> None:
+        script = _load_spike_script()
+        valid = {
+            "serial": SERIAL,
+            "url": URL,
+            "goods_id": None,
+            "state": "initial",
+            "output_dir": Path("evidence"),
+            "timeout": 10.0,
+            "adb": "adb",
+        }
+        invalid_values = (
+            ("serial", ""),
+            ("timeout", 0),
+            ("timeout", float("inf")),
+            ("state", "not-declared"),
+            ("url", "https://mobile.yangkeduo.com/goods.html?goods_id=bad"),
+        )
+        for field, value in invalid_values:
+            with self.subTest(field=field, value=value):
+                arguments = argparse.Namespace(**(valid | {field: value}))
+                with self.assertRaises((ValueError, ProductUrlError)):
+                    script.validate_arguments(arguments)  # type: ignore[attr-defined]
+
+    def test_goods_id_is_rebuilt_as_canonical_url(self) -> None:
+        script = _load_spike_script()
+        arguments = argparse.Namespace(
+            serial=SERIAL,
+            url=None,
+            goods_id="00123",
+            state="all-dimensions-selected",
+            output_dir=Path("evidence"),
+            timeout=10.0,
+            adb="adb",
+        )
+
+        link = script.validate_arguments(arguments)  # type: ignore[attr-defined]
+        self.assertEqual(link.canonical_url, "https://mobile.yangkeduo.com/goods.html?goods_id=00123")

docs/02-requirements.md

@@ -99,14 +99,20 @@ US-006（付款前核对）在 MVP 降级为：系统展示订单截图与授权
 - **F-005**：执行员启动会话后桌面端定时轮询，同时领取待试选与已授权两类任务。
   两个实例并发领取同一条时只有一个成功，另一个得到明确的「无可领任务」而不是报错。
   **关闭会话即停止轮询；连续失败达到阈值自动停止并提示原因。**
-- **F-006**：手机打开对应商品详情页，打开规格面板，按维度精确匹配颜色分类和尺码。
+- **F-006**：手机打开对应商品详情页，只点击与当前拼多多版本及本项目真机证据绑定的精确唯一
+  **受控规格面板入口**，打开规格面板后按维度精确匹配颜色分类和尺码。当前只确认拼多多
+  `8.17.0`、goods_id `937122477375` 上的“快要抢光”；其他入口文案必须分别取证，不能按购买
+  语义泛化。
   **任一维度找不到精确值即停止并转人工，不选相近选项。** 勾选后读取该 SKU 单价
   （闸门一），读不到即转人工，**不用商品详情页正文或搜索页的数字凑合**。
 - **F-006 释放要求**：试选完成后**必须退出商品页释放手机**，不得停在规格面板等待人工。
-- **F-006 硬边界**：试选阶段**绝不点击「现在买」或任何进入下单流程的入口**。
-  必须有测试证明该路径不调用任何下单语义动作。
+- **F-006 硬边界**：第一趟只把上述精确入口点击视为可逆的“打开规格面板”能力；进入面板后
+  不得调整数量、进入订单确认页、点击“提交订单”或任何支付/资金控件，也不得暴露通用任意点击能力。
+  必须有静态调用链和测试证明上述下单语义动作不可达。入口缺失、重复、版本失配或打开后不是已取证
+  面板时立即停止，不尝试“免拼购买 / 单独购买 / 直接拼成”等相似文案。
 - **F-007**：回传商品标题、实际勾选到的颜色分类与尺码、单价、合计（单价 × 数量）和
-  规格面板截图；任务转「等你确认」。
+  **自动脱敏后的**规格面板截图；任务转「等你确认」。原始 screenshot/XML 只能留在采购工具本机
+  隔离目录，不能上传、写日志或提交 Git。
 
 ### 决策与资金
 
@@ -138,7 +144,8 @@ US-006（付款前核对）在 MVP 降级为：系统展示订单截图与授权
 
 - **F-011**：失败必须可区分至少这些原因：设备未连接、商品页打不开、规格面板打不开、
   规格不匹配、单价读不到、单价与授权价不符、数量设置失败、金额超上限、提交控件不唯一、
-  页面识别失败、安全校验、外部支付交接、超时。每种都保留截图和页面快照。
+  页面识别失败、安全校验、外部支付交接、超时。原始截图/页面快照只留采购工具本机隔离目录；
+  脱敏成功时保留派生物，任何上传、远程审阅或服务端展示都只能使用自动复检通过的派生证据。
 - **版本失配**：运行时读取到的拼多多 App 版本与当前已取证版本不一致时，桌面端必须停止
   领取真机任务并提示重新取证；不得继续使用旧页面判据。
 - **付款收口（MVP 简化版）**：订单创建后任务转「待付款」，页面展示订单截图、商品、
@@ -177,8 +184,13 @@ US-006（付款前核对）在 MVP 降级为：系统展示订单截图与授权
   可配置的动作节奏，并在检测到安全校验时立即停止。
 - **自动化边界风险**：会自动点击并创建订单，属不可逆操作。必须支持 dry-run（跑到
   订单确认页停止）、服务端提交围栏和点击后调和；任一环节状态不明都不得继续点击。
-- **隐私风险**：订单确认页含收货地址和掩码手机号。**只读取非敏感摘要，不提取地址
-  原文、手机号或支付凭据**；上传服务端的证据需先脱敏。
+- **隐私风险**：规格面板和订单确认页都会显示收货地址和掩码手机号。允许原始 screenshot/XML
+  仅在采购工具本机隔离目录短链路落盘供确定性脱敏器消费；业务逻辑、日志、agent、fixture 与采购
+  服务**不得提取或接收地址原文、手机号或支付凭据**。只有脱敏成功且自动复检通过的派生证据才能
+  上传或进入开发材料，无法确认脱敏完整即 fail closed。
+- **受控规格入口风险**：T-103 已证明当前衣服商品只有购买语义按钮能打开规格面板。项目所有者批准
+  仅把经真机取证的精确唯一入口作为第一趟可逆导航；该批准不覆盖其他文案，不授权调整数量、提交订单、
+  进入确认页或付款。T-103 必须用 capability 隔离和不可达测试证明边界没有扩散。
 - **规格面板上的单价位置未取证（阻塞 F-006 闸门一）**：选中 SKU 后价格显示在哪个节点、
   是否带「券后」前缀、是否与原价并列，尚无本项目的真机证据。**T-103 必须一并取证。**
   若规格面板上无法可靠读到单价，闸门一要改为「只截图不判价」，确认页设计随之调整。

docs/03-tech-stack.md

@@ -43,7 +43,7 @@
 | 调用位置 | 采购工具 | 已定 | PC 有算力；改 prompt 不需要重新打包 |
 | provider | 待定 | **待定** | 需先确认预算与合规；不得由 agent 自行选定 |
 | 凭据存储 | 采购工具本机配置文件，不入库、不上传 | 已定 | 采购服务不保存、不代理、不下发任何模型凭据 |
-| 输入 | 完整节点树 XML + 页面截图 | 已定 | `dump_hierarchy(compressed=False)` 不丢节点 |
+| 输入 | 自动复检通过的脱敏派生 XML + 页面截图 | 已定 | 原始证据只允许本机确定性脱敏器消费，AI/agent 不读取原始地址或手机号 |
 
 ## 四、决策记录与演进
 
@@ -72,6 +72,90 @@
 | 测试 | `go test ./...` | `.\.venv\Scripts\python.exe -m unittest discover -s tests -t .` |
 | 静态检查 | `go vet ./...` | `.\.venv\Scripts\python.exe -m compileall -q src tests scripts` |
 
+### T-101 设备基线取证（2026-08-04 已完成人工双通道验收）
+
+`client/scripts/capture_device_baseline.py` 只允许对**手工明确填写**的 ADB serial 做连接前核验、
+设备型号 / Android / 拼多多版本读取、截图和 `dump_hierarchy(compressed=False)`。它不打开商品、
+不读取页面判据，也不执行采购、下单或付款动作。多个在线通道必须完成 `getprop` 物理身份比对：
+同一手机 USB + WiFi 同时在线，或任一在线通道的身份读取失败，都会 fail closed，不能随机继续。
+
+人工验收前，先由人把手机切换到不含收货地址、手机号、支付信息或其他无关隐私的安全页面，再在
+`adb devices -l` 中**手工复制**一个在线 serial；USB 和 WiFi 分别验收，且每次只保留一个通道在线。
+WiFi 通道必须由人先行建立；脚本禁止 `adb connect`、`adb disconnect` 或自动重连。以下命令中的尖括号
+必须替换为该次人工确认的实际 serial，不能省略或改成自动选择：
+
+```powershell
+# 仓库根目录；先手工确认设备状态，命令本身只读 ADB 清单
+D:\Portable\adb\adb.exe devices -l
+
+# USB：粘贴该次 devices -l 显示的 USB serial
+.\client\.venv\Scripts\python.exe client\scripts\capture_device_baseline.py --serial <USB_SERIAL> --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-101\usb-baseline" --timeout 10 --adb D:\Portable\adb\adb.exe
+
+# WiFi：由人先建立 WiFi ADB 通道、断开 USB 后，粘贴该次 devices -l 显示的 WiFi serial
+.\client\.venv\Scripts\python.exe client\scripts\capture_device_baseline.py --serial <WIFI_SERIAL> --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-101\wifi-baseline" --timeout 10 --adb D:\Portable\adb\adb.exe
+```
+
+成功时输出目录仅包含截图、完整 XML 与不含页面正文的 `manifest.json`（设备元数据、通道、时间、
+文件 SHA-256 和 serial 哈希）。`--timeout` 约束 ADB 命令、ADB socket 及 `takeScreenshot` /
+`dumpWindowHierarchy(compressed=False, max_depth=50)` 的公开 JSON-RPC 调用；uiautomator2 初始化仍有
+上游固定启动上限。截图 Base64 仅兼容 RPC 返回值中的空格、TAB、CR、LF，其余字符
+仍严格拒绝。XML 仅留在本机明确指定的证据目录；人工必须先在本地检查截图/XML，再只记录路径
+和哈希，不得把原始证据提交 Git。USB 与 WiFi 已分别由人完成取证和隐私检查，设备型号、Android、
+拼多多版本、产物路径及 SHA-256 已记录到 T-101；上述命令保留用于可审计的复现与排障。
+
+### T-102 按链接打开商品取证（2026-08-04 已完成人工真机验收）
+
+`client/scripts/capture_product_open.py` 只接受
+`https://mobile.yangkeduo.com/goods.html?goods_id=<纯数字>` 的唯一 canonical 表示。解析后由
+`goods_id` 再次重建 URL，并以参数数组执行显式限定 `com.xunmeng.pinduoduo` 的 Android `VIEW`
+intent；没有任意 URL、任意 shell 或其他 App 控件操作入口。运行拼多多版本必须精确等于 T-101 已
+取证的 `8.17.0`，版本失配会在 intent 前停止。
+
+```powershell
+# 仓库根目录；USB 或 WiFi 每次只保留一个通道在线，再手工复制该次 serial
+D:\Portable\adb\adb.exe devices -l
+.\client\.venv\Scripts\python.exe client\scripts\capture_product_open.py --serial <SERIAL> --url "https://mobile.yangkeduo.com/goods.html?goods_id=<GOODS_ID>" --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-102\product-open-<GOODS_ID>" --timeout 10 --adb D:\Portable\adb\adb.exe
+```
+
+脚本在 intent 成功后，以 `--timeout` 为明确上限只读轮询前台 package；只有观察到拼多多才采集截图与
+完整 XML，超时仍 fail closed。这一等待只解决 App 异步切换，不根据 Activity、节点文本或旧项目常量
+声称已到详情页。成功目录以原子方式发布，manifest 仅记录 `goods_id`、canonical URL、
+设备/App 非敏感元数据、受限命令摘要、文件路径和 SHA-256，不含原始 serial、Activity 或页面正文。
+必须由人本地确认截图对应目标商品并检查截图/XML 无地址、手机号、支付信息或其他无关隐私；原始
+证据不得提交 Git。T-102 已由人确认 goods_id `958756616606` 对应目标商品并完成截图/XML 隐私检查，
+设备、版本、证据路径与 SHA-256 已记录到任务执行记录。
+
+### T-103 规格面板三状态只读取证（T-110 边界调整后待重新验证）
+
+`client/scripts/capture_sku_panel_spike.py` 只采集人已在手机上准备好的规格面板截图和完整 XML。
+它不打开链接或规格面板，不识别面板，不点击、滑动、输入或选择规格，也不读取价格；接口只暴露
+`app_info`、`app_current` 和两个只读 JSON-RPC 方法。`initial`、`one-dimension-selected`、
+`all-dimensions-selected` 三种值写入 manifest 的字段名是 `human_declared_state`，明确表示人工声明，
+不得将其当作自动识别结果。运行拼多多版本必须精确为 `8.17.0`，前台 package 必须是拼多多，否则
+fail closed；目标目录已存在、截图/XML 无效或超时均不得覆盖已有内容或发布半成品。
+
+T-103 已证明当前衣服商品没有独立「规格/已选」入口。T-110 只批准拼多多 `8.17.0`、goods_id
+`937122477375` 上经真机确认的精确唯一 `快要抢光` 作为可逆的受控规格面板入口；“免拼购买 / 单独购买 /
+直接拼成”等其他文案不能凭人工经验复用，必须分别重新取证。只读取证 CLI 本身仍不执行点击；三种状态
+分别使用一个全新原始证据目录：
+
+```powershell
+.\client\.venv\Scripts\python.exe client\scripts\capture_sku_panel_spike.py --serial <SERIAL> --url "https://mobile.yangkeduo.com/goods.html?goods_id=<GOODS_ID>" --state initial --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-103\sku-panel-initial-<GOODS_ID>\raw" --timeout 10 --adb D:\Portable\adb\adb.exe
+.\client\.venv\Scripts\python.exe client\scripts\capture_sku_panel_spike.py --serial <SERIAL> --url "https://mobile.yangkeduo.com/goods.html?goods_id=<GOODS_ID>" --state one-dimension-selected --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-103\sku-panel-one-dimension-<GOODS_ID>\raw" --timeout 10 --adb D:\Portable\adb\adb.exe
+.\client\.venv\Scripts\python.exe client\scripts\capture_sku_panel_spike.py --serial <SERIAL> --url "https://mobile.yangkeduo.com/goods.html?goods_id=<GOODS_ID>" --state all-dimensions-selected --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-103\sku-panel-all-dimensions-<GOODS_ID>\raw" --timeout 10 --adb D:\Portable\adb\adb.exe
+```
+
+PDD 规格面板不可避免显示收货区域和掩码手机号。原始截图/XML 只能留在
+`%LOCALAPPDATA%\cmbuyer\artifacts\...\raw` 隔离目录，不供 agent、fixture、业务或 HTTP 上传读取；
+T-103 必须实现本机确定性脱敏器，把隐私区域从 screenshot/XML 同时移除并在 sibling `derived` 目录
+原子发布。脱敏器必须绑定 PDD 版本、设备分辨率与页面结构，派生 XML 仍命中手机号模式、隐私区域无法
+确认、版本/分辨率/结构失配或哈希不一致时拒绝发布。只有派生 manifest 和人工状态确认完成后，agent
+才能从派生物提取最小 fixture、编写判据。
+
+2026-08-04 的首轮证据只用于确认上述入口事实；其中 `initial` 不是规格面板，另两张原始截图含隐私
+区域，因此 XML 未读取、fixture 与选择器未生成。T-110 完成受控入口与脱敏契约后，T-103 重新取证；
+该边界调整不授权第一趟调整数量、进入确认页、点击“提交订单”或触碰任何支付控件。
+
 Windows 的标准入口是仓库根 `./init.ps1`。它要求 Go、两端目录及其哨兵文件存在；已有合规
 `client/.venv` 时，所有采购工具检查与 validator 都使用该解释器。只有 venv 不存在时，才从 `py -0p`
 枚举的版本中确定性选择最高的 Python 3.11+ 创建它；没有合规版本时明确失败，绝不回退默认 `python`。
@@ -85,7 +169,8 @@ compileall。既有 `client/.venv` 若不是 Python 3.11+ 会明确失败，不
 
 当前 Windows 默认 `python` 仍可能指向 Python 3.10，不满足采购工具的 Python 3.11+ 下限；
 不得把未加版本选择器的 `python` 当作采购工具命令。统一入口优先使用既有合规 venv，仅在需要创建时
-自动选择 Launcher 中最高的合规版本；本机现有 venv 实际为 Python 3.12。桌面 GUI 与真机流程不属于 T-003 验收范围。
+自动选择 Launcher 中最高的合规版本；本机 Python 3.12 与 3.14 均已验证，主工作区当前选择 Python
+3.14。桌面 GUI 与真机流程不属于 T-003 验收范围。
 
 Windows PowerShell 差异：
 

docs/04-architecture.md

@@ -97,7 +97,7 @@
         ┌──────────────── 第一趟：试选 ────────────────┐
         │ 采购工具轮询领取 PENDING 任务                    │
         │ 1. open_product(url)                          │
-        │ 2. 打开规格面板                                │
+        │ 2. 经版本绑定、精确唯一的受控入口打开规格面板       │
         │ 3. 按维度精确勾选颜色分类、尺码                  │
         │ 4. 【闸门一】读该 SKU 单价，算合计                │
         │ 5. 截图                                       │
@@ -139,6 +139,25 @@
 代价是同一商品走两遍，但第二趟很快，而且换来两个好处：手机可以在人思考时继续跑别的
 任务的试选；价格变动能被第二趟抓住。
 
+### 第一趟受控规格面板入口
+
+T-103 的真机证据推翻了“商品详情页存在独立规格入口”的假设：拼多多 8.17.0、goods_id
+`937122477375` 只能从“快要抢光”打开规格面板。项目所有者于 2026-08-04 批准把这一点击定义为
+**可逆且能力受限的规格面板导航**，不把它当作下单授权，也不把购买语义文案整体加入白名单。
+
+第一趟执行器只能持有以下 capability：
+
+1. 打开 canonical 商品链接。
+2. 点击与证据哈希、拼多多版本和页面状态绑定的精确唯一 `快要抢光` 入口。
+3. 在已确认的维度容器中精确选择规格并读回选中态。
+4. 从规格面板读取单价、生成脱敏派生证据、关闭面板并退出商品页。
+
+第一趟 capability **不得包含**通用 `click`、数量增减、进入订单确认页、提交订单或付款能力。
+“免拼购买 / 单独购买 / 直接拼成”等其他文案即使人工认为行为相同，也必须各自重新取证后才能评审；
+入口缺失、重复、版本失配、打开后面板判据不唯一，或出现“提交订单”以外的未知终态按钮时立即停止。
+第一趟的静态依赖检查必须证明 `set_quantity()`、`go_to_order_confirm()`、`submit_order()` 和任何
+支付函数不可达。
+
 ### 三道价格闸门
 
 | 闸门 | 位置 | 作用 | 不通过时 |
@@ -175,7 +194,7 @@ V2 实现时仍遵守：**图搜的唯一产出是 goods_id**，不在搜索结
 | 提交订单控件唯一 | 文本精确等于「提交订单」且可点击祖先唯一，否则停 | 点到未知控件 |
 | 数量必须复核 | 设置后读回确认精确等于要求值，否则停 | 买错数量 |
 | 价格三道闸门 | 见第三节。任一道读不到或不通过即停，**不用其他位置的数字凑合** | 超预算采购 |
-| 第一趟不下单 | 试选阶段只勾选规格和读价，**绝不点击「现在买」或任何进入下单流程的入口** | 无授权下单 |
+| 第一趟不下单 | 只允许点击证据/版本绑定的精确唯一受控入口打开规格面板，当前仅为 `快要抢光`；随后只选规格、读价、脱敏取证和返回。数量、确认页、`提交订单`、付款与通用点击能力均不可达 | 无授权下单 |
 | 外部支付页 | 检测到微信等外部支付交接立即停止、转人工、保留证据 | 凭据泄露 |
 | 安全校验 | 检测到验证码、风控、人脸、短信校验立即停止，不尝试绕过 | 封号 / 违规 |
 | 敏感信息 | 只读非敏感摘要，不提取收货地址原文、手机号、支付凭据 | 隐私泄露 |
@@ -368,25 +387,30 @@ PENDING_RETRIAL ─────────────┴─claim→ CLAIMED
 
 | 数据 | 位置 | 理由 |
 | --- | --- | --- |
-| 候选商品页 / 规格页截图 | 上传采购服务 | 管理员做授权决策必须看 |
-| 订单确认页截图 | 上传采购服务 | 授权后核对与审计必须留 |
-| 订单核对截图 | 上传采购服务 | 资金核对证据 |
-| 完整节点树 XML | **仅采购工具本地** | 体积大、含页面全文、只用于排障 |
+| 原始商品页 / 规格页 screenshot/XML | **仅采购工具本机隔离目录** | PDD 页面不可避免包含收货区域和掩码手机号；只允许确定性脱敏器读取，不供业务、agent、fixture 或上传消费 |
+| 脱敏派生商品页 / 规格页截图 | 上传采购服务 | 管理员做授权决策必须看；服务端只接收派生哈希和 sanitizer 版本，原始/派生哈希映射仅留本机 manifest |
+| 最小脱敏 XML fixture | 采购工具测试 / 可提交 Git | 只保留页面判据所需结构；自动复检无地址、手机号、支付凭据后才可发布 |
+| 脱敏派生订单确认页截图 | 上传采购服务 | 授权后核对与审计必须留；原始物仍只在本机隔离目录 |
+| 脱敏派生订单核对截图 | 上传采购服务 | 资金核对证据；原始物仍只在本机隔离目录 |
 | AI 调用记录（P1） | **仅采购工具本地** | 含 prompt / 响应全文，脱敏成本高 |
-| 失败现场快照 | 仅采购工具本地，可按需手工导出 | 同上 |
+| 失败现场快照 | 原始物仅采购工具本机；只能手工导出脱敏派生物 | 同上 |
 
-上传前必须脱敏：**不上传含收货地址、手机号、支付凭据的截图区域或文本。**
+原始目录不得被 HTTP sink、Vikunja 导出、日志或 fixture 构建器读取。脱敏器必须先验证设备分辨率、
+页面/App 版本和预期隐私区域，再同时生成派生 screenshot/XML；派生 XML 仍命中手机号模式、截图隐私
+区域无法确定、sanitizer 异常或任一哈希不一致时，不发布派生目录。上传端只接受 manifest 明确标记
+`privacy_tier=SANITIZED` 的派生截图，**不上传含收货地址、手机号、支付凭据的区域或文本。**
 
 ## 六、关键技术难点
 
 | 难点 | 说明 | 应对 |
 | --- | --- | --- |
 | 拼多多页面结构随版本变化 | 前序项目已观察到详情页无独立规格入口、价格节点拆分等变化 | **每条判据先做真机 spike 取证再写代码**；判据与 App 版本一并记录 |
+| 规格面板安全入口 | T-103 已在拼多多 8.17.0 真机确认衣服商品只能通过购买语义入口打开规格面板 | T-110 已批准仅使用当前证据证明的精确唯一 `快要抢光` 作为受控导航；其他文案不泛化，第一趟下单能力保持不可达 |
 | 规格面板上的价格位置 | 选中 SKU 后价格显示在哪、是否含券后前缀，未取证 | **T-103 必须一并取证**，闸门一依赖它；读不到就转人工，不用详情页数字凑合 |
 | 同一商品两趟结果不一致 | 第二趟价格变了、规格选项变了或商品下架 | 闸门二拦截；一律转人工，不自动放弃也不自动继续 |
 | 图搜结果含跨类目商品（V2） | 搜服装出现纸巾 | B 路径只产 goods_id 且限 5 个；后续用 VLM 看截图筛同款 |
 | WiFi ADB 稳定性 | 息屏、换网、DHCP 续租会断连 | 超时可配置；断连视为技术失败并保留现场，不重试点击 |
-| 同一手机 USB + WiFi 同时在线 | `adb devices` 列出两条，自动选设备会失败 | 设备档案必须显式指定 serial，不允许留空自动选 |
+| 同一手机 USB + WiFi 同时在线 | `adb devices` 列出两条，自动选设备会失败 | serial 必填；多在线通道必须读到 `ro.serialno` 或 `ro.boot.serialno` 才能比对。身份一致或任一身份读取失败时都 fail closed，不能以相同 model/product 猜测后继续 |
 | 不可逆动作的重试 | 点击「现在买」后超时，无法判断订单是否已创建 | 一律转人工并预留金额额度，**禁止自动重试点击** |
 | 双端契约漂移 | 两端独立演进会静默不兼容 | 契约改动必跑完整门禁；[api.md](api.md) 是唯一权威 |
 
@@ -395,8 +419,9 @@ PENDING_RETRIAL ─────────────┴─claim→ CLAIMED
 ## 七、推荐开发顺序
 
 1. **Phase 0 地基**：两端骨架、测试命令、`init` 脚本可运行。
-2. **Phase 1 真机取证**：WiFi ADB 连通；打开商品 → 打开规格面板 → 按维度精确勾选颜色
-   分类和尺码 → **读到该 SKU 单价** → 设数量 → 进订单确认页 → 读「实付款」。
+2. **Phase 1 真机取证**：WiFi ADB 连通；先验证第一趟的打开商品 → 受控打开规格面板 → 按维度
+   精确勾选颜色分类和尺码 → **读到该 SKU 单价** → 脱敏取证 → 退出；再以独立的第二趟 spike
+   验证设数量 → 进订单确认页 → 读「实付款」。第一趟 capability 不含后三项。
    **结论写入文档，判据带拼多多 App 版本。**
 3. **Phase 2 采购服务核心**：数据模型与状态机、手工建单、任务查询、试选结果接收、
    确认页与授权签发、**授权超时与放弃**。
@@ -408,6 +433,10 @@ PENDING_RETRIAL ─────────────┴─claim→ CLAIMED
 **不要在 Phase 1 结论出来之前写 Phase 2 的页面**——确认页要显示什么，取决于真机上
 究竟能读到什么。尤其是闸门一的单价，如果规格面板上读不可靠，整个确认页的设计要改。
 
+> 2026-08-04：Phase 1 证明已验证衣服商品没有独立规格入口。项目所有者随后批准 T-110 的受控入口
+> 方案：当前只允许证据绑定的精确唯一 `快要抢光` 打开规格面板，并以 capability 隔离保证数量、确认页、
+> 提交订单和付款在第一趟不可达。T-103 仍须先实现自动脱敏与新证据门禁，完成前 Phase 2 保持冻结。
+
 ## 八、项目结构
 
 ```text

docs/05-coding-rules.md

@@ -28,6 +28,11 @@
   已消费。
 - **第一趟试选的代码路径不得引用 `go_to_order_confirm()` 与 `submit_order()`**，
   必须有测试证明不可达。
+- 第一趟只可通过独立的 `open_trial_sku_panel()` capability 点击本项目真机证据与 App 版本绑定的
+  精确唯一入口；当前仅允许拼多多 `8.17.0` 上已取证的 `快要抢光`。不得向第一趟暴露通用 `click`、
+  `set_quantity()`、订单确认、提交或付款能力，不得把其他购买文案作为包含/同义匹配兜底。
+- 规格面板内即使可见“提交订单”、微信支付、先用后付或 0 元下单，第一趟也只能把它们作为硬拒绝
+  判据，绝不能返回可点击对象或尝试继续。
 
 ### 1.2 匹配纪律
 
@@ -51,11 +56,17 @@
 - 检测到外部支付交接立即停止，**不读取、不保存、不输入任何凭据**。
 - 只读非敏感摘要，**不提取收货地址原文、手机号、支付凭据**。
 - 上传服务端的证据必须先脱敏。
+- PDD 页面不可避免显示地址/掩码手机号时，原始 screenshot/XML 只允许写入采购工具本机隔离目录，
+  只由确定性脱敏器消费；业务代码、agent、fixture、日志和 HTTP sink 只能读取自动复检通过且 manifest
+  标记 `privacy_tier=SANITIZED` 的派生物。脱敏失败、分辨率/版本不符或派生 XML 仍命中手机号模式时
+  必须拒绝发布，不得用人工口头确认绕过。
 
 ### 1.5 页面判据
 
 - **不得从前序项目、旧文档或推理直接写页面判据。** 必须有本项目的真机取证。
 - 每条判据必须记录取证时的**拼多多 App 版本**。
+- 购买语义按钮不能按“作用相同”共享判据。`快要抢光`、`免拼购买`、`单独购买`、`直接拼成` 等
+  每个入口文案都必须分别取证；只允许精确唯一匹配，不允许包含、前缀、相似或坐标兜底。
 - 判据失效时先重新取证，不要靠加兜底分支硬扛。
 
 > 这些红线不是建议。[`04-architecture.md`](04-architecture.md) 第四节列出的每一条都必须有

docs/api.md

@@ -168,6 +168,27 @@
 - 无可领任务返回 `200` 且 `task` 为 `null`，**不是 404**。
 - 后续所有该任务的调用必须携带 `X-Claim-Token` 与匹配的 `claim_generation`。
 
+### `POST /api/v1/tasks/{id}/evidence`（只接收脱敏派生物）
+
+规格面板和订单确认页可能固定展示地址与掩码手机号。采购工具必须先在本机隔离目录保存原始证据，
+再由确定性脱敏器生成派生 screenshot/XML；本接口只接受派生截图及如下审计元数据：
+
+```json
+{
+  "kind": "SKU_PANEL_SCREENSHOT",
+  "privacy_tier": "SANITIZED",
+  "artifact_sha256": "<派生截图 SHA-256>",
+  "sanitizer_version": "sku-panel-pdd-8.17.0-v1"
+}
+```
+
+- `privacy_tier` 必须精确为 `SANITIZED`；缺失、其他值或 sanitizer 元数据不完整均拒绝。
+- 上传内容的 SHA-256 必须等于 `artifact_sha256`。服务端不接收原始文件、原始哈希、原始路径、
+  原始 XML、地址、手机号或支付凭据；原始/派生哈希映射只存在于采购工具本机 manifest。
+- 原始目录不得被 `HttpResultSink` 或证据上传器枚举；调用方必须显式传入已原子发布的派生目录。
+- 完整 XML 永不上传。经自动复检的最小脱敏 XML 只用于采购工具离线 fixture；仍命中手机号模式或
+  脱敏结果不确定时，客户端调用 `/needs-manual`，不得上传截图或继续试选。
+
 ### `POST /api/v1/tasks/{id}/spec-trial`（第一趟回传）
 
 ```json
@@ -187,7 +208,7 @@
 - `unit_price` 来自闸门一（规格面板）。**读不到时不要发这个接口**，改发
   `/needs-manual` 并带原因码 `UNIT_PRICE_UNREADABLE`。
 - `total_price` = `unit_price` × 任务数量，服务端会重算校验。
-- 证据须先经 `/evidence` 上传。
+- 证据须先经 `/evidence` 上传，且对应资产必须为 `privacy_tier=SANITIZED`。
 - 服务端接收后创建 `spec_trials` 记录，任务转 `WAITING_CONFIRMATION`。
 
 ### dry-run 与真实提交协议
@@ -282,25 +303,29 @@ class ResultSink(ABC):
 
 ### 真机流程模块
 
-`client/src/android/pdd_flow.py` 的公开入口，每个都不得越界：
+`client/src/android/pdd_flow.py` 的公开入口按 capability 分离，每个都不得越界：
 
-| 函数 | 输入 | 输出 | 副作用边界 |
-| --- | --- | --- | --- |
-| `open_product(url)` | 商品 URL | 页面快照路径 | 只打开页面，不点击购买 |
-| `open_sku_panel()` | - | 面板快照 | 只点规格入口，不提交 |
-| `select_sku_options(items)` | `{维度: 值}` | 选中证据 | 按维度精确匹配，找不到抛错 |
-| `set_quantity(n)` | 数量 | 读回值 | 必须复核等于 n |
-| `read_sku_unit_price(xml)` | 规格面板 XML | 单价或 `None` | **闸门一**；读不到返回 `None`，不猜 |
-| `leave_product()` | - | - | 第一趟结束时退出并释放手机 |
-| `go_to_order_confirm()` | - | 确认页摘要 | **可能创建订单**，需显式授权 |
-| `read_order_confirm_info(xml)` | 页面 XML | 非敏感摘要 | **闸门三**；不提取地址原文、手机号 |
-| `submit_order(auth, submission)` | 授权 + 已建立的提交围栏 | 提交结果 | **唯一创建真实订单入口**，四条件与围栏全通过后只点一次 |
+| 函数 | 可用趟次 | 输入 | 输出 | 副作用边界 |
+| --- | --- | --- | --- | --- |
+| `open_product(url)` | TRIAL / ORDER | 商品 URL | 页面快照路径 | 只打开页面，不点击控件 |
+| `open_trial_sku_panel(evidence_key)` | **仅 TRIAL** | 版本与证据绑定键 | 面板快照 | 只点击精确唯一、已取证的受控入口；当前仅 `快要抢光`，无通用 click |
+| `select_sku_options(items)` | TRIAL / ORDER | `{维度: 值}` | 选中证据 | 按维度精确匹配，找不到抛错 |
+| `sanitize_evidence(raw_manifest)` | TRIAL / ORDER | 本机隔离目录 manifest | 派生 manifest | 原子发布脱敏派生物；失败不发布，原始内容不进入日志/上传 |
+| `read_sku_unit_price(xml)` | TRIAL / ORDER | 脱敏规格面板 XML | 单价或 `None` | **闸门一 / 二**；读不到返回 `None`，不猜 |
+| `leave_product()` | TRIAL / ORDER | - | - | 第一趟结束时退出并释放手机 |
+| `set_quantity(n)` | **仅 ORDER** | 数量 | 读回值 | 必须复核等于 n；TRIAL capability 不暴露 |
+| `go_to_order_confirm()` | **仅 ORDER** | - | 确认页摘要 | 需显式授权；TRIAL capability 不暴露 |
+| `read_order_confirm_info(xml)` | **仅 ORDER** | 脱敏页面 XML | 非敏感摘要 | **闸门三**；不提取地址原文、手机号 |
+| `submit_order(auth, submission)` | **仅 ORDER** | 授权 + 已建立的提交围栏 | 提交结果 | **唯一创建真实订单入口**，四条件与围栏全通过后只点一次 |
 
-两个不可逆入口：
+能力隔离规则：
 
 - `go_to_order_confirm()` 必须校验授权存在；`submit_order()` 还必须校验授权已由服务端围栏
   且 `submission` 与当前任务、命令、授权完全一致。
-- **第一趟的代码路径不得引用这两个函数。** 必须有测试证明试选流程不可达它们。
+- 第一趟只能拿到 `TrialSkuFlow` 窄接口，接口中不得出现通用 `click`、`set_quantity()`、
+  `go_to_order_confirm()`、`submit_order()` 或支付能力；静态依赖测试必须证明试选流程不可达它们。
+- `open_trial_sku_panel()` 的点击是唯一批准的购买语义控件例外，只用于打开已取证规格面板；入口
+  缺失/重复、App 版本不符、面板判据不唯一或出现未知终态控件时停止。其他入口文案不得推断复用。
 - `search_by_image()` 属 B 路径，V2 再实现。
 
 ## 四、待实现时确认

docs/current-state.md

@@ -11,35 +11,39 @@
 
 ## 当前快照
 
-- 日期：2026-08-03
-- 阶段：**Phase 0 · 地基（两端骨架、核心数据模型与统一入口已完成，尚无真机采购业务代码）**
+- 日期：2026-08-04
+- 阶段：**Phase 1 · 真机可行性（T-110 已批准受控规格入口，T-103 重新执行）**
 - MVP 形态：手工填链接建单 → 批量开始试选 → 定时轮询 → **第一趟试选** → 人工确认 → **第二趟下单** → 待付款
 - 技术栈：已定。采购服务（`admin/`）使用 Go 1.23+ / gin / SQLite；采购工具（`client/`）
   使用 Python 3.11+ / uiautomator2 / PySide6。
   详见 [`03-tech-stack.md`](03-tech-stack.md)
 - 生产代码：`admin/` 已有最小 Go 服务、健康检查、核心领域模型、SQLite 迁移与任务状态机；
-  `client/` 已有 Python 包、PySide6 最小入口、运行目录与日志脱敏策略；尚无真机采购流程
-- 测试：采购服务已覆盖健康检查、核心模型、迁移与状态机等离线包级测试；采购工具 6 项离线单元测试
-  （不连接真机）
+  `client/` 已有 Python 包、PySide6 最小入口、运行目录与日志脱敏策略，以及显式 serial 的 ADB
+  连接边界、本地基线取证 CLI、受限商品链接打开取证 CLI 和人工声明规格面板状态的只读取证 CLI；
+  尚无规格选择、价格读取或下单流程
+- 测试：采购服务已覆盖健康检查、核心模型、迁移与状态机等离线包级测试；采购工具 55 项离线单元测试
+  （全部 mock，不连接真机）
 - 数据：SQLite 核心表与迁移已落成；无业务实例数据
 - 标准启动路径：Windows PowerShell 运行 `./init.ps1`，Unix shell 运行 `./init.sh`。Windows 入口
   优先使用合规的既有 venv；仅在其缺失时才从 Python Launcher 已安装版本中选择最高的 Python 3.11+，
   并且不覆盖低版本环境；成功后打印真实启动命令。
 - 标准验证路径：`./init.ps1` 已实际跑通 admin 的 mod download / test / vet / build、client 的
   editable install / 包导入 / unittest / compileall，以及仓库上下文校验。可单独运行两端命令诊断。
-- 当前 blocker：无外部 blocker。T-002、T-003、T-004 已完成；T-101 真机环境盘点已就绪，
-  但尚未真机验收。桌面 GUI 与真机流程尚未验收。
+- 当前设备门禁：人工已确认拼多多 8.17.0、goods_id `937122477375` 的衣服商品只能通过“快要抢光”
+  打开规格面板；T-110 已获项目所有者批准，只把该证据/版本绑定的精确唯一入口作为第一趟可逆导航，
+  数量、确认页、提交订单、付款与通用点击能力仍不可达。T-103 下一步先实现原始证据本机隔离和自动
+  脱敏派生物，再重新采集三态证据；在此之前不写页面判据，Phase 2 仍不能抢跑。
 
 ## 当前目录要点
 
 | 路径 | 状态 | 说明 |
 | --- | --- | --- |
 | `docs/` | 已有 | 项目规范化文档，本次已完整生成 |
-| `docs/tasks/` | 已有（T-001～T-004、T-005～T-009、T-101） | T-001～T-004 已完成；T-101 已就绪、待真机验收 |
+| `docs/tasks/` | 已有（T-001～T-004、T-005～T-009、T-101～T-110） | T-001～T-004、T-101～T-102、T-110 已完成；T-103 重新执行 |
 | `docs/design/` | 已有（6 个原型） | web 登录 / 建单 / 工作台 / 详情，desk 采购执行 / 配置；均已人工确认 |
 | `scripts/` | 已有 | 上下文门禁、Vikunja 单向导出与 MCP 启动包装 |
 | `admin/` | 已初始化 | Go 1.23+ / gin / SQLite，含核心模型、迁移与状态机；无真机采购执行 |
-| `client/` | 已初始化 | Python 3.11+ 包、依赖源、PySide6 最小入口、离线测试与 wheel 元数据检查；无真机或采购流程 |
+| `client/` | 已初始化 | Python 3.11+ 包、依赖源、PySide6 最小入口、显式 serial 的基线/商品打开/规格面板只读取证、离线测试与 wheel 元数据检查；无规格选择、价格读取或下单流程 |
 | `init.ps1` / `init.sh` | 已完成 | 统一安装与离线验证入口；PowerShell 优先复用合规 venv，缺失时自动选择最高的 Python 3.11+，Unix 缺工具链明确失败 |
 
 ## 任务状态
@@ -50,9 +54,11 @@
   源码目录契约）、T-008（Vikunja 任务权威与单向导出）、T-009（MVP 关键路径与并行波次），
   以及 T-001（采购服务 Go 骨架）。
 - 已完成：T-002（采购工具 Python 骨架）、T-003（双端统一初始化与验证入口）、
-  T-004（核心数据模型）。T-101（真机环境盘点）已建档并就绪，尚未真机验收，接续推进
-  T-101 → T-102 → T-103。
-- T-103 是当前最高优先级和 MVP 生死线。通过前不开发依赖真机可读字段的 Phase 2 生产页面。
+  T-004（核心数据模型）、T-101（真机环境盘点与 USB/WiFi 双通道人工验收）、T-102（canonical
+  链接打开与目标商品/隐私人工验收）。
+- 已完成 T-110（第一趟受控规格入口与隐私脱敏边界）。T-103 是当前最高优先级和 MVP 生死线，
+  已补充 T-110 依赖并恢复 `DOING`；自动脱敏派生证据和新真机验收完成前不开发依赖真机可读字段的
+  Phase 2 生产页面。
 - 已确认原型继续只作信息架构依据；原型假数据不调用真实接口、不驱动真机。真机结论改变
   可读字段时必须先回修原型与交互清单。
 
@@ -93,9 +99,59 @@ cd client
 ```
 
 `client/requirements.txt` 是唯一依赖来源，`client/pyproject.toml` 动态读取它生成 wheel 的
-`Requires-Dist`。Python 3.12 已验证 6 项离线测试、编译与 wheel 元数据；完整运行时依赖安装
+`Requires-Dist`。Python 3.12 与 3.14 均已验证离线测试、编译与 wheel 元数据；完整运行时依赖安装
 （`pip install -e .`）已通过。`init.ps1` 优先使用合规既有 venv，缺失时自动选择最高的 Python 3.11+；
-本机现有 venv 实际验证为 Python 3.12。桌面 GUI 与真机流程未作为 T-003 验收执行。
+本机 Python 3.12 与 3.14 均已验证，主工作区当前选择 Python 3.14。桌面 GUI 与真机流程未作为 T-003
+验收执行。
+
+T-101 的人工真机验收命令（先把手机切到不含收货地址、手机号、支付信息或其他无关隐私的安全页面；
+必须从 `adb devices -l` 手工复制在线 serial，不能留空或自动选择）：
+
+```powershell
+# 仓库根目录；USB 和 WiFi 分开执行，每次只保留一个通道在线
+D:\Portable\adb\adb.exe devices -l
+.\client\.venv\Scripts\python.exe client\scripts\capture_device_baseline.py --serial <USB_SERIAL> --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-101\usb-baseline" --timeout 10 --adb D:\Portable\adb\adb.exe
+# WiFi 必须由人先建立通道、断开 USB 后再手工粘贴在线 WiFi serial；脚本不自动 connect/reconnect
+.\client\.venv\Scripts\python.exe client\scripts\capture_device_baseline.py --serial <WIFI_SERIAL> --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-101\wifi-baseline" --timeout 10 --adb D:\Portable\adb\adb.exe
+```
+
+脚本只读取非敏感设备元数据、拼多多版本、截图和完整 XML；不会打开商品或写页面判据。它对同机双
+通道、身份读取失败、offline、unauthorized、超时或 serial 不存在均 fail closed。成功目录的
+`manifest.json` 只记录元数据、产物路径和 SHA-256，不记录 XML 页面正文或原始 serial。T-101 已由人
+完成两次验收并确认原始证据不含敏感信息；`--timeout` 约束 ADB 命令、ADB socket 以及截图/节点树的
+公开 JSON-RPC 调用，uiautomator2 初始化仍有上游固定启动上限。截图/XML 只保留在本地，执行记录只写
+路径和 SHA-256，原始证据不得提交 Git。
+
+T-102 的人工真机验收命令（只接受唯一 canonical 链接；USB 或 WiFi 每次只保留一个通道在线）：
+
+```powershell
+# 仓库根目录；从输出中手工复制本次在线 serial
+D:\Portable\adb\adb.exe devices -l
+.\client\.venv\Scripts\python.exe client\scripts\capture_product_open.py --serial <SERIAL> --url "https://mobile.yangkeduo.com/goods.html?goods_id=<GOODS_ID>" --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-102\product-open-<GOODS_ID>" --timeout 10 --adb D:\Portable\adb\adb.exe
+```
+
+脚本只允许 Android `VIEW` intent，并把 package 固定为 `com.xunmeng.pinduoduo`；intent 后会在
+`--timeout` 的有限窗口内只读轮询前台 package，解决 App 异步切换造成的一次性误判，超时仍会停止。
+它不点击、滑动、输入或判断商品页节点，也不打开规格、读取价格、进入下单或支付。运行拼多多版本必须精确为
+`8.17.0`，否则在 intent 前停止。成功后由人本地查看截图/XML，确认页面确为该 `goods_id` 对应商品并
+检查无地址、手机号、支付信息或其他无关隐私；只回报 manifest 路径及截图/XML SHA-256，原始证据
+不得提交 Git。T-102 已由人确认 goods_id `958756616606` 的目标商品及截图/XML 隐私，并完成验收。
+
+T-103 已确认当前衣服商品只能从精确文案“快要抢光”进入规格面板；T-110 只批准该证据/版本绑定入口，
+不授权“免拼购买 / 单独购买 / 直接拼成”等其他文案。现有只读取证脚本本身仍不打开面板、不点击或选择
+规格，也不读取价格；`--state` 只是人工声明，不能作为自动判据。T-103 将先把以下原始目录接入本机
+确定性脱敏器，再使用派生证据：
+
+```powershell
+# 仓库根目录；三次采集使用三个全新的 output-dir
+.\client\.venv\Scripts\python.exe client\scripts\capture_sku_panel_spike.py --serial <SERIAL> --url "https://mobile.yangkeduo.com/goods.html?goods_id=<GOODS_ID>" --state initial --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-103\sku-panel-initial-<GOODS_ID>\raw" --timeout 10 --adb D:\Portable\adb\adb.exe
+.\client\.venv\Scripts\python.exe client\scripts\capture_sku_panel_spike.py --serial <SERIAL> --url "https://mobile.yangkeduo.com/goods.html?goods_id=<GOODS_ID>" --state one-dimension-selected --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-103\sku-panel-one-dimension-<GOODS_ID>\raw" --timeout 10 --adb D:\Portable\adb\adb.exe
+.\client\.venv\Scripts\python.exe client\scripts\capture_sku_panel_spike.py --serial <SERIAL> --url "https://mobile.yangkeduo.com/goods.html?goods_id=<GOODS_ID>" --state all-dimensions-selected --output-dir "$env:LOCALAPPDATA\cmbuyer\artifacts\T-103\sku-panel-all-dimensions-<GOODS_ID>\raw" --timeout 10 --adb D:\Portable\adb\adb.exe
+```
+
+原始 screenshot/XML 可能包含 PDD 固定展示的地址和掩码手机号，只能留在本机 `raw` 目录，不能由
+agent、fixture、业务或上传端消费。脱敏器在 sibling `derived` 目录原子发布 screenshot/XML 与 manifest，
+自动复检通过且人确认三态对应性后，agent 才能读取派生物并提取最小 fixture、编写判据。
 
 ## 关键背景
 

docs/tasks/T-101.md

@@ -3,7 +3,7 @@ id: T-101
 title: 验证 ADB 与 uiautomator2 双通道连接
 phase: 1
 deps: [T-002]
-status: TODO
+status: DONE
 created: 2026-08-03
 vikunja_task_id: 21
 context_ref: 87591e8
@@ -13,6 +13,7 @@ needs_human_review: true
 write_paths:
   - docs/tasks/T-101.md
   - client/.gitignore
+  - client/requirements.txt
   - client/src/cmbuyer_client/device/**
   - client/tests/device/**
   - client/scripts/capture_device_baseline.py
@@ -22,7 +23,7 @@ write_paths:
   - docs/current-state.md
 ---
 
-<!-- BEGIN VIKUNJA EXPORT id=21 synced=2026-08-03T10:40:11Z sha256=e6a7a6c24df85d52435692fc120d3d3ddea3148b46ea3cf1e99c3debfe53321e -->
+<!-- BEGIN VIKUNJA EXPORT id=21 synced=2026-08-04T01:02:36Z sha256=b6458effd879f2e258c861b1b7b99d1162fdbea586e3fa467d3a65178c900453 -->
 ## 问题 / 背景
 
 T-002 已建立采购工具骨架，但本项目还没有对实际 Android 手机、ADB 通道或 uiautomator2 做过取证。后续所有拼多多页面判据都依赖稳定、显式且可审计的设备连接；若复用前序项目结论或自动猜设备，会把错误设备和旧页面事实带入生产流程。
@@ -51,7 +52,33 @@ T-002 已建立采购工具骨架，但本项目还没有对实际 Android 手
 
 ## 执行记录
 
-（暂无）
+### 2026-08-03T10:56:19Z · ila
+
+T-101 已领取，Git 状态将提交为 DOING。只读环境盘点：adb 位于 D:\Portable\adb\adb.exe；当前同时列出 WiFi serial 192.168.0.173:5555 与 USB serial 3B65BD02H7F00000，二者 product/model/device 均为 PKG110/PKG110/OP5D2BL1。该现场必须由实现识别为同机双通道并 fail closed。尚未截图、dump 或操作手机，本记录不是人工真机验收；needs_device 规则继续生效。
+
+### 2026-08-03T11:27:08Z · ila
+
+2026-08-03 离线实现与主审完成：新增显式 serial 的 ADB 边界、USB/WiFi 同机多通道 fail-closed、禁止 uiautomator2 隐式重连的连接器，以及只读基线 CLI（型号、Android、拼多多版本、截图、compressed=false XML、manifest SHA-256）。主 agent 三轮审查后补齐：截图/XML JSON-RPC 可配置超时、adbutils/uiautomator2 类型化超时、完整硬件身份集合比对、非法截图/XML fail-closed、异常脱敏、暂存清理，以及人工安全页面/原始证据不入 Git 的操作要求。独立验证：init.ps1 通过；admin go test/vet/build 通过；client 27 项 mock 单测、compileall、wheel METADATA、validator、bash -n 与 diff check 通过。未连接或操作真机，T-101 继续保持 DOING；等待人分别完成 USB 与 WiFi 取证并记录 manifest 中的设备型号、Android、拼多多版本、路径与 SHA-256。已知边界：ADB/ADB socket/截图与节点树 RPC 超时可配置，uiautomator2 初始化仍受上游固定启动上限约束。
+
+### 2026-08-04T00:36:03Z · ila
+
+2026-08-04 人工进展：用户确认 WiFi ADB 已连接，serial 为 192.168.0.173:5555。主 agent 随后仅执行 `adb devices -l` 复核，WiFi 与 USB 3B65BD02H7F00000 当前均为 device，product/model/device 均显示 PKG110/PKG110/OP5D2BL1；本机尚无 `%LOCALAPPDATA%\cmbuyer\artifacts\T-101` 取证产物。该结果只证明 WiFi 通道在线，不是 T-101 完整人工验收。按 fail-closed 边界，人工进行 WiFi 取证前须先断开 USB，只保留 WiFi；USB 取证时须断开 WiFi，只保留 USB。仍等待两次截图、compressed=false XML 与 manifest 路径/SHA-256，任务保持 DOING。
+
+### 2026-08-04T00:43:38Z · ila
+
+2026-08-04 人工 WiFi 取证失败：用户在仅 WiFi serial 192.168.0.173:5555 在线时运行基线 CLI，工具返回“截图 RPC 返回的数据不是有效图像，拒绝发布不完整取证”。主 agent 只读复核：ADB 仅列出该 WiFi 通道为 device，T-101 产物目录为空，说明暂存清理与 fail-closed 生效。核对本机 uiautomator2 3.7.0 源码后发现其高层 screenshot 使用允许 Base64 空白的 `base64.b64decode`，而当前实现直接 `validate=True`，可能错误拒绝真实 RPC 返回中的 CR/LF/空格。已退回 client agent 修复为“只规范化 ASCII Base64 空白，再严格 validate=True；其他非法字符与非图像继续拒绝”，并补回归测试。该失败不是验收通过，T-101 继续 DOING；修复、主审、提交后由人重新执行 WiFi 取证。
+
+### 2026-08-04T00:49:31Z · ila
+
+2026-08-04 WiFi 截图兼容性修复已完成主审：实现只删除 RPC Base64 返回值中的空格、TAB、CR、LF，再使用 `validate=True` 严格解码；非法字符、错误 padding、vertical-tab、form-feed、data URI 与解码后非图像仍 fail closed，不增加 adb screenshot fallback。错误安全地区分为“Base64 语法无效”和“图像数据无效”，均不输出原始内容，失败继续清理暂存目录。新增真实 PNG Base64 空白兼容、非法字符、错误 padding、未白名单空白及非图像测试；client 共 30 项离线测试。主 agent 独立复核并运行 client 测试/compileall、admin test/vet/build、完整 init.ps1、validator、bash 语法与 diff check，均通过；Python 3.12 和主工作区 3.14 均已验证。未由 agent 连接或截图真机，T-101 保持 DOING，等待人工重新执行 WiFi 取证确认现场问题是否解决。
+
+### 2026-08-04T00:55:14Z · ila
+
+2026-08-04 WiFi 人工取证技术验收通过：用户报告基线命令成功，manifest 位于 `%LOCALAPPDATA%\cmbuyer\artifacts\T-101\wifi-baseline\manifest.json`。主 agent 仅读取 manifest 元数据并计算文件哈希，未打开或展示截图/XML 内容。manifest schema=1、channel=wifi、设备型号 PKG110、Android 16、拼多多包 com.xunmeng.pinduoduo、版本 8.17.0；serial 仅保存 64 位 SHA-256。`screenshot.png` 存在，SHA-256=`8f86c94700d93380805b275469123a31749d592d99c0c4d2bcfd540784d1620b`；`hierarchy.xml` 存在，SHA-256=`708182200e25085c07cbb546c9b0adbc7310022bb75a7285c5581bd262661573`；两者均与 manifest 一致。仍等待人工确认原始截图/XML 不含敏感信息，并完成仅 USB 通道的同等取证；T-101 保持 DOING。
+
+### 2026-08-04T01:02:03Z · ila
+
+2026-08-04 USB 人工取证与隐私验收通过：用户报告仅 USB 通道基线命令成功，manifest 位于 `%LOCALAPPDATA%\cmbuyer\artifacts\T-101\usb-baseline\manifest.json`。主 agent 仅读取 manifest 元数据并计算哈希，未打开或展示截图/XML：schema=1、channel=usb、设备 PKG110、Android 16、拼多多 8.17.0；`screenshot.png` SHA-256=`da66e27ef6df1d4b99a64590e9aa383069c8c3995eb9b3695e5daec3a0e7bcf6`，`hierarchy.xml` SHA-256=`e0817f26c07e6861f2ab80052995aff2622ce7cdb61f18689d55c5591bb2045a`，均与 manifest 一致。用户随后明确确认 USB 与 WiFi 两组截图/XML 均不含地址、手机号、支付信息或其他无关隐私。至此 USB/WiFi 显式 serial、截图、compressed=false XML、设备/Android/拼多多版本、路径和 SHA-256 的人工验收全部满足；T-101 可关闭，后续进入 T-102。
 <!-- END VIKUNJA EXPORT -->
 
 ## 边界

docs/tasks/T-102.md

@@ -0,0 +1,99 @@
+---
+id: T-102
+title: 验证按链接打开商品详情页
+phase: 1
+deps: [T-101]
+status: DONE
+created: 2026-08-04
+vikunja_task_id: 22
+context_ref: 393f26d
+work_branch: task/t-102-open-product
+needs_device: true
+needs_human_review: true
+write_paths:
+  - docs/tasks/T-102.md
+  - client/src/cmbuyer_client/pdd/**
+  - client/src/cmbuyer_client/device/**
+  - client/tests/pdd/**
+  - client/tests/device/**
+  - client/scripts/capture_product_open.py
+  - docs/03-tech-stack.md
+  - docs/04-architecture.md
+  - docs/current-state.md
+---
+
+<!-- BEGIN VIKUNJA EXPORT id=22 synced=2026-08-04T01:50:40Z sha256=ee2405865a36aa01a3503e6d24d62a3dddd0f19b451845e1e2c3937734c31442 -->
+## 问题 / 背景
+
+T-101 已证明同一台 PKG110（Android 16、拼多多 8.17.0）可通过 USB/WiFi 显式 serial 完成只读截图与完整 XML 取证。MVP 下一风险是任务自带的合法拼多多链接能否由 Android intent 打开到对应商品详情页。此前项目和旧文档不能证明当前 App 的行为；若先写详情页判据，会把未经本项目真机验证的假设带入 T-103。
+
+## 关联需求与交互
+
+- 功能：F-006 的第一步 `open_product(url)`；仅覆盖链接打开，不覆盖规格面板、规格选择或价格。
+- 用户故事 / 交互：采购工具执行链接任务前的只读真机 spike；无生产 GUI。
+- 架构 / API：`docs/04-architecture.md` 第三节 A 路径；`docs/api.md` 的 canonical `product_url` / `goods_id` 示例；T-103 前置。
+
+## 方案
+
+1. 只接受 HTTPS canonical 链接 `https://mobile.yangkeduo.com/goods.html?goods_id=<纯数字>`；拒绝 userinfo、非默认端口、fragment、重复/缺失/非数字 goods_id、其他 host/path 和额外业务参数。解析后由代码按 goods_id 重建 canonical URL，绝不把任意输入拼入 shell。
+2. 复用 T-101 的显式 serial、同机多通道 fail-closed、no-reconnect、类型化超时和原子证据发布边界；运行拼多多版本必须精确等于本项目已取证的 8.17.0，否则在打开链接前停止。
+3. 通过参数数组执行只含 Android `VIEW` intent 语义的 `adb shell am start -W`，显式限定包 `com.xunmeng.pinduoduo`；不使用 shell 字符串、不点击/滑动/输入任何控件。Android 命令失败、超时、未解析 intent、未停留在拼多多包、截图/XML 失败分别给出脱敏错误。
+4. 启动后只读取 current app/package、截图和 `dumpWindowHierarchy(compressed=False)`，manifest 记录 goods_id、canonical URL、设备/App 元数据、命令结果摘要、路径与 SHA-256，不记录页面全文或原始 serial。原始截图/XML 仅存 `%LOCALAPPDATA%`，不提交 Git。
+5. 离线 mock 测试覆盖 URL 正反例、参数数组与超时、版本失配在 intent 前停止、包不匹配、无 UI 点击 API、证据原子性和异常脱敏。代码完成后由人使用一个明确 goods_id 在仅单通道在线的真机执行，人工确认截图确为对应商品详情页并检查 XML/截图无敏感信息；再把设备、Android、拼多多版本、goods_id、路径、SHA-256 与结论写入执行记录。
+
+## 验收要点
+
+- 合法 canonical 链接可在拼多多 8.17.0 上打开；人工确认到达与 goods_id 对应的商品详情页。
+- 非法链接、设备状态、App 版本失配、intent 失败/超时、落在非拼多多包、证据失败均有可区分且不泄露页面内容的错误。
+- 取证记录包含 PKG110、Android 16、连接方式、拼多多 8.17.0、goods_id、canonical URL、截图/XML 本地路径与 SHA-256。
+- client 全部单测、compileall、wheel metadata、上下文校验及完整 `init.ps1` 通过。
+- `needs_device=true`：agent 只能完成离线实现并保持 DOING；只有人完成真机链接打开、页面对应性与隐私检查后才能 DONE。
+
+## 执行记录
+
+### 2026-08-04T01:09:47Z · ila
+
+2026-08-04：已认领 T-102，工作分支 `task/t-102-open-product`，从任务定义提交 `4281b06` 开始实现。离线阶段只实现严格 canonical URL 校验、Android VIEW intent、版本/设备 fail-closed 与原子证据采集；不写商品详情页节点判据，不操作任何 App 控件。`needs_device=true`，离线实现完成后仍保持 DOING，等待人工提供明确商品链接并完成真机页面对应性和隐私验收。
+
+### 2026-08-04T01:27:08Z · ila
+
+2026-08-04：T-102 离线实现已由子 agent 完成并经主 agent 独立审查、两轮退回修正后通过，代码提交 `7040bb6`。实现包含：唯一 canonical 商品 URL 校验与重建、ADB 内层纯数字 goods_id 二次围栏、固定 package 的 Android VIEW intent、拼多多 8.17.0 intent 前版本围栏、前台 package 核验，以及截图/XML 原子取证与脱敏 manifest；无点击、滑动、输入、规格、价格、下单或支付能力，也未写商品详情页节点/Activity 判据。主审验证：43 项 client 离线测试、compileall、完整 init.ps1（含 admin test/vet/build）、wheel metadata、agent-context 与 diff-check 均通过；当前 wheel SHA-256 为 c675cd527849149ec25f213a0d67da31d0bb497cd8c5ed255d71154c103659ed（审计临时产物已清理）。未连接或操作真机，T-102 继续保持 DOING，等待人工使用明确 canonical 商品链接执行取证，并确认页面对应性与隐私。
+
+### 2026-08-04T01:32:31Z · ila
+
+2026-08-04 人工真机首轮：使用 WiFi serial `192.168.0.173:5555` 与 canonical 商品链接 `https://mobile.yangkeduo.com/goods.html?goods_id=958756616606` 执行 T-102 脚本。脚本通过 URL/intent 前置围栏，但在 intent 后返回“商品链接打开后前台应用不是拼多多”，按 fail-closed 规则停止，未发布完整取证。人在 Chrome 中可打开该 URL，只能证明网页链接有效，不能证明拼多多 App 已接管。T-102 保持 DOING；下一步由人只读取当前前台 package，区分瞬时切换/检查过早与实际落到 Chrome、系统解析器或其他包，在得到事实前不移除 package 围栏、不增加页面判据或 UI 兜底。
+
+### 2026-08-04T01:34:39Z · ila
+
+2026-08-04 人工追加诊断：T-102 首轮返回前台包不符后，人在未切换 App 的情况下立即执行只读 `dumpsys window`，得到 `mCurrentFocus` 与 `mFocusedApp` 均为 `com.xunmeng.pinduoduo/com.xunmeng.pinduoduo.activity.NewPageActivity`。该事实证明失败返回后前台已稳定到拼多多，支持“一次性 app_current 检查过早/存在异步切换窗口”的诊断；Activity 名只记录为诊断证据，不作为商品详情页判据。修复保持 package 围栏：改为有限时长、只读 package 的轮询，超时仍 fail closed，期间不点击、滑动、输入或读取页面节点。
+
+### 2026-08-04T01:40:42Z · ila
+
+2026-08-04：已按人工 foreground 证据完成最小修复，代码提交 `cb646b4`。保留 PDD package 围栏，将 intent 后的一次性判断改为以 CLI `--timeout` 为 deadline、默认 0.2 秒间隔的只读 package 轮询；只有观察到 `com.xunmeng.pinduoduo` 才采集截图/XML，超时仍 fail closed。轮询不读取 Activity/节点，不点击、滑动或输入。主 agent 已独立验证 46 项 client 测试、compileall、完整 init.ps1、wheel metadata、agent-context、diff-check；wheel SHA-256 `80fbc73785bb12b40856c6b1ef503bdb5cbc350fadd2b23cc6cc58358499bfe1`，临时产物已清理。未由 agent 连接真机，T-102 保持 DOING，等待人以 goods_id `958756616606` 重跑原命令并检查页面对应性与隐私。
+
+### 2026-08-04T01:46:35Z · ila
+
+2026-08-04 人工真机重跑已成功：人先退出到拼多多首页，再用 WiFi serial `192.168.0.173:5555`、goods_id `958756616606` 执行同一命令，证据原子发布到 `C:\Users\ila20\AppData\Local\cmbuyer\artifacts\T-102\product-open-958756616606`。manifest 记录 PKG110 / Android 16 / PDD 8.17.0 / current package `com.xunmeng.pinduoduo` / intent status ok；截图 SHA-256 `be9ea1f53b13fec82fd716875567870132ef0a1480cfbec5ad603436d774d81c`，XML SHA-256 `1449e78a154fddf7108e63a14a2e85bf74f162f4689c1fe95f2cb67fdb7b4fc5`。主 agent 只读取 manifest 并重新计算哈希，二者均匹配；未打开原始截图/XML。仍等待人明确确认截图对应目标商品且截图/XML 无地址、手机号、支付信息或其他无关隐私，确认前 T-102 保持 DOING。
+
+### 2026-08-04T01:49:55Z · ila
+
+2026-08-04 人工最终验收：人已查看本地 screenshot.png，确认对应 goods_id 958756616606；并确认截图与 hierarchy.xml 不含地址、手机号、支付信息或其他无关隐私。T-102 的 canonical 链接打开、PDD 8.17.0 前台 package、证据原子发布与人工页面对应性验收全部通过，允许关闭。
+<!-- END VIKUNJA EXPORT -->
+
+## 边界
+
+- 本任务只验证由显式 canonical 链接启动拼多多并采集本地证据；不编写或声称商品详情页节点判据，
+  页面是否为对应商品必须由人查看本项目新产物确认。
+- 只允许 Android `VIEW` intent；不点击、滑动、长按、输入任何 App 控件，不打开规格面板，不选择
+  颜色或尺码，不读取价格，不进入购买或订单页面。
+- 不引用或实现 `go_to_order_confirm()`、`submit_order()`、提交订单、付款、支付、授权或采购服务接口；
+  不把前序项目的 URL、activity、节点文本或页面结论当作事实。
+- 输入 URL 必须先严格解析并按纯数字 `goods_id` 重建，再以参数数组传给 ADB；禁止 shell 拼接、
+  任意 scheme/host/path、短链跳转、额外参数或自动猜测链接。
+- 运行拼多多版本与本项目已取证版本不一致时必须在 intent 前停止；证据失败、当前包不符或结果无法
+  由人确认时均不得声称成功，不增加自动点击或其他兜底路径。
+- 截图和完整 XML 只保存在明确的本地 T-102 目录，人工检查后只把路径、SHA-256 与非敏感元数据
+  写入执行记录；不得提交原始证据，不记录地址、手机号、支付信息或无关页面正文。
+- 不修改 `admin/`、业务 API、数据模型、生产 GUI 或 T-103 规格面板逻辑。
+- `needs_device: true`：agent 完成离线实现后保持 `DOING`；只有人使用明确 goods_id 完成真机打开、
+  确认对应商品详情页并完成隐私检查后才能标 `DONE`。

docs/tasks/T-103.md

@@ -0,0 +1,126 @@
+---
+id: T-103
+title: 验证规格面板精确选择与 SKU 单价
+phase: 1
+deps: [T-102, T-110]
+status: DOING
+created: 2026-08-04
+vikunja_task_id: 23
+context_ref: 1dc8308
+work_branch: task/t-103-sku-panel
+needs_device: true
+needs_human_review: true
+write_paths:
+  - docs/tasks/T-103.md
+  - client/src/cmbuyer_client/pdd/**
+  - client/src/cmbuyer_client/device/**
+  - client/tests/pdd/**
+  - client/tests/device/**
+  - client/scripts/capture_sku_panel_spike.py
+  - docs/02-requirements.md
+  - docs/03-tech-stack.md
+  - docs/04-architecture.md
+  - docs/api.md
+  - docs/current-state.md
+---
+
+<!-- BEGIN VIKUNJA EXPORT id=23 synced=2026-08-04T02:56:52Z sha256=2f5ce314d8ae55d466abe28d5d3a4c36e235e008af3a4d251b4ea362a4de1314 -->
+## 问题 / 背景
+
+T-102 已证明 canonical 链接可进入目标商品。T-103 随后在 PKG110 / Android 16 / 拼多多 8.17.0、衣服商品 goods_id `937122477375` 上确认：规格面板只能从详情页右下角精确文案“快要抢光”进入，面板固定显示收货区域和掩码手机号。T-110 经项目所有者批准，将该已取证点击定义为可逆、能力受限的规格面板导航；这不是通用购买入口豁免，不授权其他文案、数量、确认页、提交订单或支付。
+
+## 关联需求与交互
+
+- 功能：F-006 第一趟试选；覆盖自动脱敏、`open_trial_sku_panel()`、`select_sku_options()`、`read_sku_unit_price()` 的真机 spike。
+- 用户故事：US-003、US-004、US-008；本任务无生产 GUI。
+- 架构 / API：`docs/04-architecture.md` 第三、四、5.4、六节；`docs/api.md` 证据上传和 `TrialSkuFlow`；T-104、T-105 与 Phase 2 的前置。
+
+## 方案
+
+1. 先实现本机确定性脱敏器，不先写页面判据。原始 screenshot/XML 只保存到 `%LOCALAPPDATA%\cmbuyer\artifacts\T-103\...\raw`，只允许脱敏器消费；agent、fixture、业务、日志、HTTP sink、Vikunja 与 Git 均不得读取/上传原始内容。
+2. 脱敏器绑定 PKG110 分辨率、拼多多 8.17.0、goods_id、人工声明状态和本项目证据配置，同时从 screenshot/XML 移除收货区域、地址与手机号内容，在 sibling `derived` 目录原子发布派生物。派生 XML 仍命中手机号模式、隐私区域无法确认、结构/分辨率/版本失配、哈希不一致或目标目录存在时 fail closed，不发布半成品。服务端只接收派生截图哈希和 sanitizer 版本，不接收原始文件/哈希/路径/XML。
+3. 由人运行脱敏器并本地确认派生 screenshot/XML 的 initial、one-dimension-selected、all-dimensions-selected 三态对应性；主 agent 只在派生 manifest 与人工确认齐全后读取派生证据，提取最小 fixture。首轮 `initial` 详情页截图与含隐私面板截图不得复用为 fixture。
+4. 实现 `TrialSkuFlow` 窄 capability。`open_trial_sku_panel()` 只允许拼多多 8.17.0、goods_id `937122477375` 上证据绑定、精确唯一的 `快要抢光`；缺失、重复、版本失配或打开后面板判据不唯一时零后续点击。`免拼购买 / 单独购买 / 直接拼成` 等其他文案必须分别取证，不能包含、前缀、同义、OCR 或坐标兜底。
+5. 第一趟 capability 只含打开商品、受控打开面板、在维度容器内精确选择、读面板单价、脱敏取证和返回。接口不得暴露通用 `click`、`set_quantity()`、`go_to_order_confirm()`、`submit_order()` 或支付能力；规格面板中的“提交订单”、微信支付、先用后付和 0 元下单只作为硬拒绝判据，静态调用链测试证明不可达。
+6. 在脱敏派生 XML 上按维度容器隔离候选，文本等值匹配并读回选中态；缺失、重复、禁用、维度不明或选中态不唯一均停止。纯函数测试覆盖 `红`/`粉红`、`1`/`10` 等前缀碰撞。
+7. 单价只从规格面板证据确定的唯一节点读取，以十进制字符串返回，禁止浮点。记录真实文本是否含优惠前缀、货币符号是否拆节点、是否并列原价/区间价；语义不唯一、节点缺失、候选冲突或只能从详情页读取时返回 unreadable。
+8. 人最终在真机执行受控入口、精确选择和读价脚本，确认实际选择 `颜色分类=黑色 CHA（纯棉）`、`尺码=M（建议100-115）`（以派生 XML 精确文本为准）、面板单价语义、派生证据隐私与退出行为；agent 保持 DOING 直到人工验收。
+
+## 验收要点
+
+- 三组原始证据仅在本机 raw 目录；脱敏器离线测试覆盖地址/手机号移除、截图与 XML 同步处理、版本/分辨率/结构失配、手机号残留、原子性、异常脱敏和已有目录不覆盖。
+- 派生 manifest 记录 source/derived 本机哈希、sanitizer 版本、设备/App/goods_id/人工状态；原始页面正文、serial、地址和手机号不进入 manifest、日志、错误或上传。
+- 只从人工验收通过的派生证据编写页面判据与最小 fixture；判据绑定拼多多 8.17.0 和证据哈希。
+- `快要抢光` 精确唯一时才打开面板；其他文案、入口缺失/重复、面板不唯一、规格缺失/碰撞/禁用、价格不唯一均 fail closed。
+- 静态 API/导入测试证明 T-103 第一趟无通用点击、数量、确认页、提交订单或支付能力，“提交订单”及资金控件不可达。
+- 从规格面板可靠读取选中 SKU 单价为十进制字符串；无法唯一解释优惠价/原价/区间价时返回 unreadable。
+- client 全部单测、compileall、wheel metadata、上下文校验、完整 `init.ps1` 与 diff-check 通过。
+- `needs_device=true`：只有人完成受控入口、精确选择、单价读取、派生证据对应性/隐私和安全退出验收后才能 DONE。
+
+## 执行记录
+
+### 2026-08-04T01:58:01Z · ila
+
+2026-08-04：已认领 T-103，工作分支 `task/t-103-sku-panel`，任务定义基线 `27999c8`。先执行人工证据门禁，不立即编写页面判据：T-102 现有详情截图只证明目标商品，且当前视口可见“单独购买 / 免拼购买”“先用后付 / 0元下单”等禁触入口，没有证明独立安全规格入口。下一步由人手工寻找非购买的“规格 / 已选”入口；若不存在就记录 no-go，不点击购买按钮兜底。规格面板三态证据与隐私确认完成前，受托 agent 只允许做证据采集脚手架、纯函数边界和只读审计，T-103 保持 DOING。
+
+### 2026-08-04T02:09:48Z · ila
+
+2026-08-04 离线实现与主审记录：
+
+- 提交 `a92b8f6` 新增 T-103 人工声明三状态的只读取证 CLI、捕获模块与离线测试；脚本不打开链接/面板，不点击、滑动、选择规格或读取价格。
+- 失败闭合覆盖：严格 URL/状态/版本/前台包校验、截图/XML 有效性、超时脱敏、原子发布、已有输出目录不覆盖；协议无购买、下单、支付或数量操作。
+- 主 agent 独立复跑：client 55 项单测、compileall、wheel metadata、agent-context validator、diff-check 与完整 `init.ps1` 全部通过。
+- `needs_device=true`，任务保持 DOING；等待人通过独立安全规格入口分别采集 initial、one-dimension-selected、all-dimensions-selected 三组证据并完成对应性与隐私检查。若只有购买/下单/资金入口可打开面板，则记录 no-go，不放宽边界。
+
+### 2026-08-04T02:36:33Z · ila
+
+2026-08-04 真机证据主审退回：
+
+- goods_id `937122477375` 的三组 manifest、设备/App 元数据与文件 SHA-256 均一致。
+- 人工给出隐私确认后，主 agent 只查看三张 screenshot，未读取 XML。
+- `initial` 截图实际仍为商品详情页，不是规格面板初始态；`one-dimension-selected` 与 `all-dimensions-selected` 截图顶部含收货区域和掩码手机号，不满足任务的无地址/手机号证据边界。
+- 本组证据判定不合格：不提取 fixture、不编写页面判据、不提交原始证据；T-103 保持 DOING。
+- 下一步使用全新输出目录重采：人先在规格面板取消全部规格选择，并滚动到收货区域和手机号完全离开可见视口，同时保留规格面板价格与规格区域；三态逐次人工核对后再交主审。
+
+### 2026-08-04T02:41:12Z · ila
+
+2026-08-04 T-103 真机 no-go 结论：
+
+- 人工确认 goods_id `937122477375` 的规格面板是通过商品详情页右下角“快要抢光”进入；通常“免拼购买 / 快要抢光 / 单独购买 / 直接拼成”等购买语义按钮才会打开规格面板。
+- 这直接触发任务方案与边界中的 no-go 条件：第一趟不得点击任何购买、下单、提交或资金入口，也不得把购买按钮包装成规格入口。
+- 因此不读取本轮 XML、不提取 fixture、不实现选择器；已有截图仅用于判定入口不可行，原始证据不提交 Git。
+- 当前 MVP 的自动第一趟试选在现有安全边界下不可实现，T-103 转 BLOCKED。解除阻塞需要项目所有者选择新的架构方向，并先更新架构/任务边界；不得由 agent 擅自放宽。
+
+### 2026-08-04T02:56:36Z · ila
+
+2026-08-04：T-110 已获项目所有者授权并同步架构契约。T-103 方案已重写为“先自动脱敏派生证据，再实现受控 `快要抢光` 入口与精确规格/价格判据”；T-110 完成并转 Done 后，本任务恢复 Doing。
+<!-- END VIKUNJA EXPORT -->
+
+## 边界
+
+- 任何规格面板入口、维度容器、选项、选中态和价格判据都必须来自本项目新采集、自动复检通过的
+  拼多多 8.17.0 **脱敏派生** screenshot/XML；T-102 详情页证据只能证明已到目标商品，不能证明
+  规格面板结构。派生证据路径、SHA-256、goods_id、设备、Android、App 与 sanitizer 版本写入
+  执行记录之前，不得提交页面判据代码。
+- 第一趟只允许 T-110 批准的 `open_trial_sku_panel()` 窄能力点击证据/版本绑定、精确唯一的
+  `快要抢光`；当前事实仅覆盖 goods_id `937122477375`、拼多多 `8.17.0`。不得把“免拼购买 /
+  单独购买 / 直接拼成”等其他文案加入包含、前缀、同义或坐标兜底，分别取证前一律拒绝。
+- 规格选项只能在已确认的面板及对应维度容器内按文本精确唯一匹配；缺失、重复、禁用、维度不明、
+  选中态无法唯一读回或页面版本不符均 fail closed。不得前缀、包含、模糊、相似或跨维度匹配，
+  不得用 OCR / 坐标兜底猜选项。
+- 单价只允许从规格面板证据确认的节点读取，并以十进制字符串表达；不得使用浮点，不得从详情页、
+  搜索卡片或其他页面的数字补值。券后价、原价、区间价、货币符号拆分或多个候选的语义无法唯一
+  证明时必须返回 unreadable 并转人工。
+- T-103 代码路径不得引用或实现 `set_quantity()`、`go_to_order_confirm()`、`submit_order()`、
+  通用 `click`、创建订单、授权、提交围栏或任何支付能力；不得进入订单确认页，不得创建待付款订单。
+  规格面板中的“提交订单”、微信支付、先用后付、0 元下单只可作为硬拒绝判据，不能返回可点击对象。
+- 不把 `/mnt/d/chengma/cmroubao`、`/mnt/d/chengma/cmpdd` 或任何旧版本的节点、Activity、选择器、
+  坐标、文本形态和页面结论当作事实；前序项目只可用于理解为什么必须 fail closed。
+- 原始 screenshot/XML 只保存在 `%LOCALAPPDATA%\cmbuyer\artifacts\T-103\...\raw`，只允许本机
+  确定性脱敏器消费，不得由 agent、业务、fixture、HTTP sink、Vikunja 或 Git 读取/上传。脱敏器必须
+  校验设备分辨率、页面/App 版本和预期隐私区域，同时移除 screenshot/XML 中的地址与手机号内容；
+  派生 XML 仍命中手机号模式、结构/版本/分辨率失配或哈希不一致时不得发布 `derived`。
+- 只有自动复检通过的最小脱敏 fixture、派生路径/哈希、sanitizer 版本与非敏感结论可入库；服务端
+  不接收原始文件、原始路径或原始哈希。
+- `needs_device: true`：agent 只能完成离线实现并保持 `DOING`；只有人完成安全入口、精确规格选择、
+  SKU 单价语义、页面对应性与派生证据验收后才能标 `DONE`。

docs/tasks/T-110.md

@@ -0,0 +1,87 @@
+---
+id: T-110
+title: 调整第一趟受控规格入口与隐私脱敏边界
+phase: 1
+deps: [T-102]
+status: DONE
+created: 2026-08-04
+vikunja_task_id: 24
+context_ref: abeefbb
+work_branch: task/t-110-trial-sku-boundary
+needs_device: false
+needs_human_review: true
+write_paths:
+  - docs/tasks/T-110.md
+  - docs/tasks/T-103.md
+  - docs/02-requirements.md
+  - docs/03-tech-stack.md
+  - docs/04-architecture.md
+  - docs/05-coding-rules.md
+  - docs/api.md
+  - docs/current-state.md
+---
+
+<!-- BEGIN VIKUNJA EXPORT id=24 synced=2026-08-04T03:00:16Z sha256=dd7209dda3769397f96882556864c1836955507fd5ee293103295d4fd9f9a840 -->
+## 问题 / 背景
+
+T-103 在 PKG110 / Android 16 / 拼多多 8.17.0 真机确认：衣服商品 goods_id `937122477375` 的规格面板只能从详情页右下角“快要抢光”等购买语义按钮进入，不存在原架构假设的独立“规格 / 已选”入口。与此同时，规格面板固定展示收货区域与掩码手机号。项目所有者已批准调整第一趟入口边界，但“不付款、第一趟不可达提交订单、价格只从规格面板读取、敏感信息不进入业务数据/日志/上传/Git”等边界不变。
+
+## 关联需求与交互
+
+- 功能：F-006 第一趟试选；为 T-103 解阻，不实现页面选择器或真机点击代码。
+- 架构：`docs/04-architecture.md` 第一趟流程、安全边界、证据分层；`docs/05-coding-rules.md` 不可逆动作和敏感信息纪律。
+- API：设备试选证据只允许上传脱敏派生物；原始截图/XML 仅留采购工具本机隔离目录。
+
+## 方案
+
+1. 把“独立规格入口”改为“受控规格面板入口”：第一趟只可点击本项目真机证据证明、与拼多多版本绑定的唯一精确入口；当前仅确认 `快要抢光`。其他文案即使语义相近也不得推断复用，必须重新取证。
+2. 将点击入口与下单能力隔离。第一趟 capability 只允许打开受控面板、在维度内精确选择、读规格面板单价、生成脱敏证据和返回商品页；接口不得出现设置数量、进入确认页、提交订单、支付或通用任意点击能力。静态调用链和测试必须证明这些能力不可达。
+3. 规格面板内的“提交订单”以及微信支付、先用后付、0 元下单等控件全部列入硬拒绝区。入口缺失/重复、版本失配、打开后不是已取证面板、选择或价格不唯一时立即返回，不尝试相近按钮。
+4. 接受 PDD 页面会显示收货区域与掩码手机号这一事实，但不允许业务系统提取或传播。原始 screenshot/XML 仅写入 `%LOCALAPPDATA%` 隔离目录，不打印页面正文、不上传、不提交 Git；本地确定性脱敏器生成派生 screenshot/XML 和哈希，开发、fixture、服务端证据只消费派生物。
+5. 脱敏必须 fail closed：无法确认地址/手机号区域已移除、派生 XML 仍命中手机号模式、分辨率/结构不符合已取证版本，均拒绝发布派生证据。不得依赖人工口头“无需处理”绕过自动检查。
+6. 同步需求、架构、技术栈、编码规则、API、current-state 与 T-103。T-110 完成后把 T-103 依赖补为 T-110 并重新转 DOING；真正的脱敏器、入口判据和真机实现仍由 T-103 完成。
+
+## 验收要点
+
+- 文档明确区分“可逆的受控规格面板入口”与“不可逆的提交订单动作”，不把购买文案泛化为通用可点击入口。
+- 第一趟的 capability 列表与硬拒绝列表完整；提交订单、数量、确认页、付款和任意点击能力不可达。
+- 原始证据与脱敏派生证据的目录、生命周期、消费者和 fail-closed 条件清楚；服务端和 Git 永不接收原始隐私证据。
+- `docs/tasks/T-103.md` 在 T-110 完成后恢复 DOING，并要求先产出自动脱敏证据再写页面判据。
+- `python scripts/validate_agent_context.py`、Vikunja 导出检查与 `git diff --check` 通过。
+
+## 执行记录
+
+### 2026-08-04T02:48:25Z · ila
+
+2026-08-04：项目所有者明确批准受控规格面板入口方案；PDD 页面不可避免显示地址/手机号的事实被接受，但系统仍按本机原始证据隔离、自动脱敏派生物消费的方式收紧传播边界。任务已认领，分支 `task/t-110-trial-sku-boundary`，基线 `abeefbb`。
+
+### 2026-08-04T02:58:44Z · ila
+
+2026-08-04 完成记录：
+
+- 已同步 `docs/02-requirements.md`、`03-tech-stack.md`、`04-architecture.md`、`05-coding-rules.md`、`api.md`、`current-state.md` 与 T-103。
+- 第一趟只批准证据/版本绑定、精确唯一的 `快要抢光`；其他购买文案不泛化。`TrialSkuFlow` 不暴露通用点击、数量、确认页、提交订单或支付能力。
+- 原始 screenshot/XML 只留本机隔离目录；服务端只接收自动复检通过的派生截图哈希与 sanitizer 版本，不接收原始文件、哈希、路径或 XML。
+- Phase 1 已拆分第一趟规格 spike 与第二趟数量/确认页 spike；T-103 已增加 T-110 依赖并按新边界恢复 DOING。
+- `validate_agent_context.py` 与 `git diff --check` 通过；本任务没有实现或运行任何真机点击代码。
+
+### 2026-08-04T03:00:01Z · ila
+
+最终提交 `82f57c7`。主 agent 独立复跑完整 `init.ps1`：admin go test/vet/build、client editable install/55 项 unittest/compileall、agent-context validator 全部通过；`git diff --check` 通过。T-110 已完成且未实现或执行任何真机点击。
+<!-- END VIKUNJA EXPORT -->
+
+## 边界
+
+- 本任务只调整架构、需求、API、编码规则与后续任务契约，不实现或运行任何真机点击代码。
+- 仍然绝不点击支付、免密支付、先用后付、0 元下单或扣款控件；不创建订单，不进入订单确认页。
+- 第一趟只允许未来实现点击**本项目真机证据证明、与拼多多版本绑定、精确唯一**的规格面板入口；
+  当前只确认 goods_id `937122477375`、拼多多 `8.17.0` 上的“快要抢光”。不得把人工经验中的
+  “免拼购买 / 单独购买 / 直接拼成”等文案直接加入白名单，分别取证前一律拒绝。
+- 第一趟 capability 不得包含通用 `click`、数量调整、订单确认、提交订单或付款能力；规格面板中的
+  “提交订单”及任何支付提示始终属于硬拒绝区，静态调用链必须保持不可达。
+- 接受原始规格面板证据在手机本地可能包含收货区域和掩码手机号，但不得从中提取业务字段、打印、
+  上传或提交 Git。只有确定性脱敏并通过自动检查的派生 screenshot/XML 才能供 agent、fixture、
+  采购服务或人工远程评审使用；脱敏不确定即 fail closed。
+- 价格仍只允许从规格面板和订单确认页读取，使用十进制字符串；不得用详情页价格补齐。
+- 本任务不得把 T-103 标为 `DONE`。文档边界完成并获得本次用户明确授权后，只能把 T-103 重新置为
+  `DOING`，由 T-103 继续实现脱敏器、判据、离线测试和人工真机验收。