docs(tasks): record T-103 entry audit

client/scripts/run_t103_sku_selection.py

@@ -0,0 +1,84 @@
+"""恢复 T-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 ProductUrlError, parse_product_url
+from cmbuyer_client.pdd.sku_selection import EXPECTED_GOODS_ID, SkuSelectionError, TASK_TO_UI_SELECTION
+from cmbuyer_client.pdd.sku_selection_runner import SkuSelectionRunError, SkuSelectionRunner
+
+
+def parse_arguments(argv: list[str] | None = None) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="恢复 T-103 已取证规格并保存本地原始截图。")
+    parser.add_argument("--serial", required=True, help="ADB device serial；禁止自动选择。")
+    parser.add_argument("--url", required=True, help="唯一 canonical goods.html?goods_id= 直链。")
+    parser.add_argument("--color", required=True, help="T-103 任务颜色值。")
+    parser.add_argument("--size", required=True, help="T-103 任务尺码值。")
+    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 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 的有限数值。")
+    link = parse_product_url(arguments.url)
+    if link.goods_id != EXPECTED_GOODS_ID:
+        raise ValueError("--url 不是 T-103 已取证商品。")
+    if (arguments.color, arguments.size) not in TASK_TO_UI_SELECTION:
+        raise ValueError("--color 与 --size 必须是 T-103 已取证任务值。")
+
+
+def main(argv: list[str] | None = None) -> int:
+    arguments = parse_arguments(argv)
+    try:
+        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
+
+    runner = SkuSelectionRunner(
+        AdbClient(SubprocessAdbRunner(arguments.adb), timeout_seconds=arguments.timeout),
+        NoReconnectUiautomatorConnector(adbutils.AdbClient(socket_timeout=arguments.timeout).device_list, u2.connect),
+        timeout_seconds=arguments.timeout,
+    )
+    try:
+        result = runner.run(arguments.serial, arguments.url, arguments.color, arguments.size, arguments.output_dir)
+    except (DeviceConnectionError, SkuSelectionRunError, SkuSelectionError) as error:
+        # Flow 可能来自测试替身或未来实现；CLI 不回显任何异常正文，避免泄露节点树或页面文本。
+        print("规格恢复失败：已停止，未发布本地证据目录。", file=sys.stderr)
+        return 1
+    except OSError:
+        print("规格恢复失败：无法创建或发布本地证据目录。", file=sys.stderr)
+        return 1
+
+    print(f"规格恢复完成：{result.output_directory}")
+    print(f"manifest：{result.manifest_path}")
+    print(f"目标规格：{arguments.color} / {arguments.size}")
+    print(f"确认单价：{result.unit_price}")
+    print("页面对应性：请人工核对本地原始截图。")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

client/src/cmbuyer_client/pdd/__init__.py

@@ -1,15 +1,23 @@
-"""拼多多链接的受限打开与只读取证。
+"""拼多多链接的受限打开、只读取证与经取证的规格面板选择。
 
-此包不提供页面选择器、输入、滑动、下单或支付能力。
+此包不提供通用页面选择器、输入、滑动或任何订单动作。
 """
 
 from .product_open import ProductOpenCapturer, ProductOpenResult
 from .product_url import ProductUrl, ProductUrlError, parse_product_url
+from .sku_selection import SkuSelection, SkuSelectionError, SkuSelectionFlow
+from .sku_selection_runner import SkuSelectionRunError, SkuSelectionRunResult, SkuSelectionRunner
 
 __all__ = [
     "ProductOpenCapturer",
     "ProductOpenResult",
     "ProductUrl",
     "ProductUrlError",
+    "SkuSelection",
+    "SkuSelectionError",
+    "SkuSelectionFlow",
+    "SkuSelectionRunError",
+    "SkuSelectionRunResult",
+    "SkuSelectionRunner",
     "parse_product_url",
 ]

client/src/cmbuyer_client/pdd/sku_selection.py

@@ -0,0 +1,420 @@
+"""T-103：仅限已取证 PDD 8.17.0 的规格面板恢复。"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+import re
+from time import monotonic, sleep
+from typing import Any, Callable, Protocol
+from xml.etree import ElementTree
+
+from ..device.baseline import PDD_PACKAGE
+from .product_open import EXPECTED_PDD_VERSION
+from .product_url import parse_product_url
+
+EXPECTED_GOODS_ID = "937122477375"
+EXPECTED_UNIT_PRICE = "12.88"
+# 任务值不是页面判据；右侧是 v5 取证的唯一 accessibility 文案（空格/全角括号均有意义）。
+TASK_TO_UI_SELECTION = {("黑色CHA(纯棉)", "M(建议100-115)"): ("黑色 CHA (纯棉）", "M（建议100-115）")}
+_TARGET_COLOR_UI, _TARGET_SIZE_UI = next(iter(TASK_TO_UI_SELECTION.values()))
+_ENTRY = "快要抢光"
+_ENTRY_TEXT_BOUNDS = "[900,1312][1056,1355]"
+_ENTRY_INNER_BOUNDS = "[712,1312][1056,1355]"
+_ENTRY_ACTION_BOUNDS = "[0,1256][1080,1355]"
+_SIZE = "尺码"
+_W, _H = 1080, 2376
+_PRICE_PARENT = "[396,498][895,570]"
+_CURRENT = "[396,503][712,570]"
+_ORIGINAL = "[730,503][895,570]"
+_SUMMARY = "[396,654][1053,716]"
+_COLOR_REGION = "[36,1000][1080,1631]"
+_SIZE_LABEL = "[36,1654][114,1700]"
+_SIZE_HEADER = "[36,1637][1044,1718]"
+_SIZE_OPTIONS = "[36,1730][1044,2045]"
+_BOUNDS = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$")
+_PRICE = re.compile(r"^[^0-9¥￥]*[¥￥]([1-9][0-9]*\.[0-9]{2})$")
+_ORIGINAL_PRICE = re.compile(r"^[¥￥][1-9][0-9]*\.[0-9]{2}$")
+_BAD_PRICE_ROLE = ("提交订单", "支付", "优惠", "券", "会员", "补贴", "区间", "实付", "到手", "原价", "划线价", "最低", "低至", "起价", "下单", "先用后付", "预估")
+
+
+class SkuSelectionError(RuntimeError):
+    """已取证判据不成立时的脱敏停止。"""
+
+
+class SkuPanelDevice(Protocol):
+    def app_info(self, package_name: str) -> dict[str, Any]: ...
+    def app_current(self) -> dict[str, Any]: ...
+    def dump_window_hierarchy(self) -> str: ...
+    def tap_sku_entry(self, bounds: str) -> None: ...
+    def tap_sku_option(self, bounds: str) -> None: ...
+    def leave_sku_panel(self) -> None: ...
+
+
+@dataclass(frozen=True)
+class SkuSelection:
+    color: str
+    size: str
+
+
+def resolve_task_selection(color: str, size: str) -> SkuSelection:
+    mapped = TASK_TO_UI_SELECTION.get((color, size))
+    if mapped is None:
+        raise SkuSelectionError("规格任务值不是已取证的唯一目标，已停止操作。")
+    return SkuSelection(*mapped)
+
+
+@dataclass(frozen=True)
+class _Node:
+    element: ElementTree.Element
+    parent: "_Node | None"
+    @property
+    def text(self) -> str: return self.element.get("text", "")
+    @property
+    def desc(self) -> str: return self.element.get("content-desc", "")
+    @property
+    def bounds(self) -> str: return self.element.get("bounds", "")
+
+
+class SkuSelectionFlow:
+    def __init__(self, device: SkuPanelDevice, entry_wait_timeout_seconds: float = 0.2,
+                 entry_poll_interval_seconds: float = 0.2, monotonic_clock: Callable[[], float] = monotonic,
+                 sleep_function: Callable[[float], None] = sleep) -> None:
+        if entry_wait_timeout_seconds < 0 or entry_poll_interval_seconds <= 0:
+            raise ValueError("入口等待参数无效。")
+        self._device, self._entry_timeout, self._poll = device, entry_wait_timeout_seconds, entry_poll_interval_seconds
+        self._clock, self._sleep = monotonic_clock, sleep_function
+        self._pending: tuple[str, Callable[[list[_Node]], Any]] | None = None
+
+    def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None:
+        if parse_product_url(product_url).goods_id != EXPECTED_GOODS_ID:
+            raise SkuSelectionError("商品不是已取证目标，已停止操作。")
+        if pre_intent_hierarchy is not None:
+            previous_nodes = _parse_nodes(pre_intent_hierarchy)
+            if _eligible_entries(previous_nodes):
+                raise SkuSelectionError("intent 前页面已出现规格入口，已拒绝旧商品误点。")
+        entry, before = self._wait_for_entry(pre_intent_hierarchy)
+        _action_bounds(entry.bounds)
+        self._pending = (before, _panel)
+        self._device.tap_sku_entry(entry.bounds)
+        self._wait_after_action(before, _panel)
+
+    def select_sku_options(self, selection: SkuSelection) -> None:
+        if selection not in {SkuSelection(*item) for item in TASK_TO_UI_SELECTION.values()}:
+            raise SkuSelectionError("规格 UI 文案不是获准目标，已停止操作。")
+        initial = self._verified_nodes()
+        _option(initial, "color", selection.color); _option(initial, "size", selection.size)
+        _selected_label(initial, "color"); _selected_label(initial, "size")
+        self._restore("color", selection.color)
+        self._restore("size", selection.size)
+
+    def read_sku_unit_price(self) -> str:
+        return _unit_price(self._verified_nodes())
+
+    def verify_target_selection_and_read_price(self, selection: SkuSelection) -> str:
+        nodes = self._verified_nodes()
+        _selected(nodes, "color", selection.color)
+        _selected(nodes, "size", selection.size)
+        return _unit_price(nodes)
+
+    def exit_sku_panel_safely(self) -> None:
+        self._require_foreground()
+        before = self._read_hierarchy()
+        _panel(_parse_nodes(before))
+        self._device.leave_sku_panel()
+        deadline = self._clock() + self._entry_timeout
+        while True:
+            self._require_foreground()
+            raw = self._read_hierarchy()
+            if raw != before:
+                try:
+                    _panel(_parse_nodes(raw))
+                except SkuSelectionError:
+                    return
+            remaining = deadline - self._clock()
+            if remaining <= 0:
+                raise SkuSelectionError("安全退出后未确认离开规格面板，未重试返回。")
+            self._sleep(min(self._poll, remaining))
+
+    def reconcile_pending_action(self) -> None:
+        """仅只读调和一次已发出但尚未得到后置条件确认的动作。"""
+        if self._pending is None:
+            return
+        before, condition = self._pending
+        self._wait_after_action(before, condition)
+
+    def _restore(self, dimension: str, expected: str) -> None:
+        self._require_foreground()
+        before = self._read_hierarchy()
+        nodes = _panel(_parse_nodes(before))
+        target = _option(nodes, dimension, expected)
+        if _selected_label(nodes, dimension) == expected:
+            return
+        _action_bounds(target.bounds)
+        condition: Callable[[list[_Node]], Any]
+        if dimension == "color":
+            condition = lambda refreshed: _post_color(refreshed, expected)
+        else:
+            condition = lambda refreshed: _post_all_targets(refreshed, expected)
+        self._pending = (before, condition)
+        self._device.tap_sku_option(target.bounds)
+        if dimension == "color":
+            self._wait_after_action(before, condition)
+        else:
+            self._wait_after_action(before, condition)
+
+    def _wait_for_entry(self, previous: str | None) -> tuple[_Node, str]:
+        deadline, stable = self._clock() + self._entry_timeout, None
+        while True:
+            self._require_version()
+            current = self._device.app_current()
+            if isinstance(current, dict) and current.get("package") == PDD_PACKAGE:
+                raw = self._read_hierarchy()
+                entries = _eligible_entries(_parse_nodes(raw))
+                if len(entries) > 1:
+                    raise SkuSelectionError("商品页规格入口不唯一，已停止操作。")
+                if len(entries) == 1 and raw != previous:
+                    if stable == raw:
+                        return entries[0], raw
+                    stable = raw
+                else:
+                    stable = None
+            else:
+                stable = None
+            remaining = deadline - self._clock()
+            if remaining <= 0:
+                raise SkuSelectionError("等待已取证规格入口超时，未执行点击。")
+            self._sleep(min(self._poll, remaining))
+
+    def _wait_after_action(self, previous: str, condition: Callable[[list[_Node]], Any]) -> list[_Node]:
+        deadline = self._clock() + self._entry_timeout
+        while True:
+            self._require_foreground()
+            raw = self._read_hierarchy()
+            if raw != previous:
+                nodes = _parse_nodes(raw)
+                try:
+                    condition(nodes)
+                    self._pending = None
+                    return nodes
+                except SkuSelectionError:
+                    pass
+            remaining = deadline - self._clock()
+            if remaining <= 0:
+                raise SkuSelectionError("动作后页面未在限定时间内满足已取证后置条件，未重试动作。")
+            self._sleep(min(self._poll, remaining))
+
+    def _verified_nodes(self) -> list[_Node]:
+        self._require_foreground()
+        return _panel(self._read_nodes())
+
+    def _require_version(self) -> None:
+        info = self._device.app_info(PDD_PACKAGE)
+        version = (info.get("versionName") or info.get("version_name")) if isinstance(info, dict) else None
+        if version != EXPECTED_PDD_VERSION:
+            raise SkuSelectionError("拼多多版本与已取证版本不一致，已停止操作。")
+
+    def _require_foreground(self) -> None:
+        self._require_version()
+        current = self._device.app_current()
+        if not isinstance(current, dict) or current.get("package") != PDD_PACKAGE:
+            raise SkuSelectionError("拼多多不在前台，已停止操作。")
+
+    def _read_hierarchy(self) -> str:
+        try: raw = self._device.dump_window_hierarchy()
+        except Exception as error: raise SkuSelectionError("节点树读取失败，已停止操作。") from error
+        if not isinstance(raw, str) or not raw: raise SkuSelectionError("节点树不可用，已停止操作。")
+        return raw
+
+    def _read_nodes(self) -> list[_Node]: return _parse_nodes(self._read_hierarchy())
+
+
+def _parse_nodes(raw: str) -> list[_Node]:
+    try: root = ElementTree.fromstring(raw)
+    except ElementTree.ParseError as error: raise SkuSelectionError("节点树格式无效，已停止操作。") from error
+    if root.tag != "hierarchy": raise SkuSelectionError("节点树根节点无效，已停止操作。")
+    result: list[_Node] = []
+    def visit(element: ElementTree.Element, parent: _Node | None) -> None:
+        node = _Node(element, parent); result.append(node)
+        for child in element: visit(child, node)
+    visit(root, None)
+    return result
+
+
+def _panel(nodes: list[_Node]) -> list[_Node]:
+    parent = _one([n for n in nodes if n.bounds == _PRICE_PARENT], "规格面板价格区域不唯一，已停止操作。")
+    _one([n for n in nodes if n.parent is parent and n.bounds == _ORIGINAL and _readonly(n) and _ORIGINAL_PRICE.fullmatch(n.text)], "规格面板原价槽位不唯一，已停止操作。")
+    _one([n for n in nodes if n.bounds == _SUMMARY and _readonly(n) and n.text.startswith("已选：")], "规格面板已选摘要不唯一，已停止操作。")
+    _color_container(nodes); _size_container(nodes)
+    return nodes
+
+
+def _unit_price(nodes: list[_Node]) -> str:
+    parent = _one([n for n in nodes if n.bounds == _PRICE_PARENT], "规格面板价格区域不唯一，已停止读取。")
+    money = [n for n in nodes if n.parent is parent and _readonly(n) and any(mark in n.text for mark in "¥￥")]
+    if len(money) != 2: raise SkuSelectionError("规格面板金额槽位不唯一，已停止读取。")
+    current = _one([n for n in money if n.bounds == _CURRENT and not _clickable_ancestor(n) and not any(word in n.text for word in _BAD_PRICE_ROLE) and _PRICE.fullmatch(n.text)], "规格面板现价不唯一或不符合已取证槽位，已停止读取。")
+    if not any(n.bounds == _ORIGINAL and _ORIGINAL_PRICE.fullmatch(n.text) for n in money):
+        raise SkuSelectionError("规格面板原价槽位无效，已停止读取。")
+    match = _PRICE.fullmatch(current.text)
+    if match is None: raise SkuSelectionError("规格面板现价格式失效，已停止读取。")
+    return match.group(1)
+
+
+def _option(nodes: list[_Node], dimension: str, expected: str) -> _Node:
+    _panel(nodes)
+    return _one([n for n in _options(nodes, dimension) if _label(n) == expected], "规格选项不唯一或不是精确匹配，已停止操作。")
+
+
+def _selected(nodes: list[_Node], dimension: str, expected: str) -> None:
+    if _selected_label(nodes, dimension) != expected:
+        raise SkuSelectionError("规格选择后读回的 selected 文案不一致，已停止操作。")
+
+
+def _selected_label(nodes: list[_Node], dimension: str) -> str:
+    selected = [n for n in _options(nodes, dimension) if n.element.get("selected") == "true"]
+    label = _label(_one(selected, "规格维度没有唯一 selected 状态，已停止操作。"))
+    if label is None: raise SkuSelectionError("规格维度 selected 文案无效，已停止操作。")
+    return label
+
+
+def _post_color(nodes: list[_Node], expected: str) -> None:
+    _panel(nodes)
+    _selected(nodes, "color", expected)
+    _selected_label(nodes, "size")
+
+
+def _post_all_targets(nodes: list[_Node], expected_size: str) -> None:
+    _panel(nodes)
+    _selected(nodes, "color", _TARGET_COLOR_UI)
+    _selected(nodes, "size", expected_size)
+
+
+def _options(nodes: list[_Node], dimension: str) -> list[_Node]:
+    container = _color_container(nodes) if dimension == "color" else _size_container(nodes) if dimension == "size" else None
+    if container is None: raise SkuSelectionError("未知规格维度，已停止操作。")
+    candidates = [n for n in nodes if _descendant(n, container) and _contained(n, container) and _choice(n) and _label(n) is not None]
+    return [n for n in candidates if not _labeled_ancestor(n, candidates)]
+
+
+def _color_container(nodes: list[_Node]) -> _Node:
+    return _one([n for n in nodes if n.element.get("package") == PDD_PACKAGE and n.element.get("class") == "androidx.recyclerview.widget.RecyclerView" and n.bounds == _COLOR_REGION], "规格面板颜色容器不唯一，已停止操作。")
+
+
+def _size_container(nodes: list[_Node]) -> _Node:
+    label = _one([n for n in nodes if n.text == _SIZE and n.bounds == _SIZE_LABEL and _readonly(n)], "规格面板尺码标签不唯一，已停止操作。")
+    header = label.parent
+    if header is None or header.element.get("package") != PDD_PACKAGE or header.element.get("class") != "android.widget.LinearLayout" or header.bounds != _SIZE_HEADER or header.parent is None:
+        raise SkuSelectionError("规格面板尺码标题容器不符合已取证结构，已停止操作。")
+    return _one([n for n in nodes if n.parent is header.parent and n.element.get("package") == PDD_PACKAGE and n.element.get("class") == "android.widget.LinearLayout" and n.bounds == _SIZE_OPTIONS], "规格面板尺码选项容器不唯一，已停止操作。")
+
+
+def _label(node: _Node) -> str | None:
+    values = {value for value in (node.text, node.desc) if value}
+    return values.pop() if len(values) == 1 else None
+
+
+def _labeled_ancestor(node: _Node, candidates: list[_Node]) -> bool:
+    ids, parent = {id(n.element) for n in candidates}, node.parent
+    while parent is not None:
+        if id(parent.element) in ids and _label(parent) is not None: return True
+        parent = parent.parent
+    return False
+
+
+def _descendant(node: _Node, ancestor: _Node) -> bool:
+    parent = node.parent
+    while parent is not None:
+        if parent.element is ancestor.element: return True
+        parent = parent.parent
+    return False
+
+
+def _contained(node: _Node, container: _Node) -> bool:
+    left, top, right, bottom = _action_bounds(node.bounds)
+    outer_left, outer_top, outer_right, outer_bottom = _action_bounds(container.bounds)
+    return outer_left <= left < right <= outer_right and outer_top <= top < bottom <= outer_bottom
+
+
+def _clickable_ancestor(node: _Node) -> bool:
+    parent = node.parent
+    while parent is not None:
+        if parent.element.get("clickable") == "true": return True
+        parent = parent.parent
+    return False
+
+
+def _readonly(node: _Node) -> bool:
+    return node.element.get("package") == PDD_PACKAGE and node.element.get("class") == "android.widget.TextView" and node.element.get("clickable") == "false" and node.element.get("enabled") == "true" and node.element.get("visible-to-user") == "true"
+
+
+def _live(node: _Node) -> bool:
+    return node.element.get("package") == PDD_PACKAGE and node.element.get("clickable") == "true" and node.element.get("enabled") == "true" and node.element.get("visible-to-user") == "true" and bool(node.bounds)
+
+
+def _choice(node: _Node) -> bool:
+    return _live(node) and node.element.get("class") == "android.view.ViewGroup" and node.element.get("selected") in {"true", "false"}
+
+
+def _eligible_entries(nodes: list[_Node]) -> list[_Node]:
+    # 入口文本本身不可点击：必须逐层证明它仍位于已取证的唯一可点击祖先中，但动作坐标继续
+    # 使用文本子节点的窄 bounds，避免把同一祖先内未知区域变成坐标兜底。“免拼购买”等底部
+    # 容器既不属于这条祖先链，也绝不能作为替代入口。
+    if any(node.bounds == _PRICE_PARENT for node in nodes): return []
+    entry_labels = [
+        node for node in nodes
+        if node.text == _ENTRY
+        and node.element.get("package") == PDD_PACKAGE
+        and node.element.get("class") == "android.widget.TextView"
+    ]
+    if len(entry_labels) != 1:
+        return entry_labels
+    action_ancestors = [
+        node for node in nodes
+        if _exact_entry_node(node, "android.view.ViewGroup", _ENTRY_ACTION_BOUNDS, "true")
+    ]
+    if len(action_ancestors) != 1:
+        return []
+    action_ancestor = action_ancestors[0]
+    entries: list[_Node] = []
+    for node in entry_labels:
+        if not _exact_entry_node(node, "android.widget.TextView", _ENTRY_TEXT_BOUNDS, "false"):
+            return []
+        inner = node.parent
+        switcher = inner.parent if inner is not None else None
+        frame = switcher.parent if switcher is not None else None
+        ancestor = frame.parent if frame is not None else None
+        if (
+            inner is not None
+            and _exact_entry_node(inner, "android.view.ViewGroup", _ENTRY_INNER_BOUNDS, "false")
+            and switcher is not None
+            and _exact_entry_node(switcher, "android.widget.ViewSwitcher", _ENTRY_INNER_BOUNDS, "false")
+            and frame is not None
+            and _exact_entry_node(frame, "android.widget.FrameLayout", _ENTRY_INNER_BOUNDS, "false")
+            and ancestor is action_ancestor
+        ):
+            entries.append(node)
+    return entries
+
+
+def _exact_entry_node(node: _Node, class_name: str, bounds: str, clickable: str) -> bool:
+    return (
+        node.element.get("package") == PDD_PACKAGE
+        and node.element.get("class") == class_name
+        and node.bounds == bounds
+        and node.element.get("clickable") == clickable
+        and node.element.get("enabled") == "true"
+        and node.element.get("visible-to-user") == "true"
+    )
+
+
+def _action_bounds(bounds: str) -> tuple[int, int, int, int]:
+    match = _BOUNDS.fullmatch(bounds)
+    if match is None: raise SkuSelectionError("规格节点坐标格式无效，已停止操作。")
+    left, top, right, bottom = (int(item) for item in match.groups())
+    if not (0 <= left < right <= _W and 0 <= top < bottom <= _H):
+        raise SkuSelectionError("规格节点坐标不在已取证屏幕范围内，已停止操作。")
+    return left, top, right, bottom
+
+
+def _one(nodes: list[_Node], message: str) -> _Node:
+    if len(nodes) != 1: raise SkuSelectionError(message)
+    return nodes[0]

client/src/cmbuyer_client/pdd/sku_selection_runner.py

@@ -0,0 +1,348 @@
+"""T-103 真机运行边界：窄适配器、原始截图和无页面正文的摘要。"""
+
+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
+from typing import Any
+from uuid import uuid4
+
+from PIL import Image, UnidentifiedImageError
+
+from adbutils.errors import AdbTimeout
+from uiautomator2.exceptions import HTTPTimeoutError
+
+from ..device.adb import AdbClient, DeviceConnectionError, DeviceInspection
+from ..device.baseline import PDD_PACKAGE, SCREENSHOT_PARAMS, _save_base64_screenshot, _sha256_file
+from .product_open import EXPECTED_PDD_VERSION
+from .product_url import ProductUrl, parse_product_url
+from .sku_selection import (
+    EXPECTED_GOODS_ID,
+    EXPECTED_UNIT_PRICE,
+    SkuPanelDevice,
+    SkuSelectionError,
+    SkuSelectionFlow,
+    _action_bounds,
+    resolve_task_selection,
+)
+
+
+EXPECTED_DEVICE_MODEL = "PKG110"
+EXPECTED_ANDROID_VERSION = "16"
+EXPECTED_SCREEN_SIZE = (1080, 2376)
+
+
+class SkuSelectionRunError(RuntimeError):
+    """T-103 运行未完整完成；错误文本不携带设备或页面原文。"""
+
+
+class SkuSelectionRunTimeoutError(SkuSelectionRunError):
+    """设备 RPC 或操作超时。"""
+
+
+class SkuSelectionScreenshotError(SkuSelectionRunError):
+    """原始截图无法作为完整 PNG 原子发布。"""
+
+
+class SkuSelectionUnexpectedPriceError(SkuSelectionRunError):
+    """取证面板现价不是本任务已确认值。"""
+
+
+class SkuSelectionDeviceAdapterError(SkuSelectionRunError):
+    """第三方设备接口失败的脱敏映射。"""
+
+
+@dataclass(frozen=True)
+class SkuSelectionRunResult:
+    """已发布的截图和无页面正文 manifest 摘要。"""
+
+    output_directory: Path
+    screenshot_path: Path
+    manifest_path: Path
+    unit_price: str
+
+
+class UiautomatorSkuPanelAdapter(SkuPanelDevice):
+    """把 uiautomator2 缩为 T-103 所需的读取与三种命名操作。
+
+    ``tap_sku_entry``、``tap_sku_option`` 和 ``leave_sku_panel`` 是仅有的状态改变方法；
+    坐标由 Flow 和本类双重检查后才计算中心点，每次调用只执行一次底层动作。
+    """
+
+    def __init__(self, device: Any, timeout_seconds: float) -> None:
+        if not _is_positive_finite(timeout_seconds):
+            raise ValueError("timeout_seconds 必须是大于 0 的有限数值")
+        self._device = device
+        self._timeout_seconds = timeout_seconds
+        self._entry_was_tapped = False
+        self._left_panel = False
+
+    @property
+    def entry_was_tapped(self) -> bool:
+        """仅供运行器决定故障后的单次尽力返回，不是页面操作。"""
+
+        return self._entry_was_tapped
+
+    @property
+    def left_panel(self) -> bool:
+        return self._left_panel
+
+    def app_info(self, package_name: str) -> dict[str, Any]:
+        value = self._call("app_info", package_name)
+        if not isinstance(value, dict):
+            raise SkuSelectionDeviceAdapterError("无法读取应用版本，已停止操作。")
+        return value
+
+    def app_current(self) -> dict[str, Any]:
+        value = self._call("app_current")
+        if not isinstance(value, dict):
+            raise SkuSelectionDeviceAdapterError("无法读取前台应用，已停止操作。")
+        return value
+
+    def dump_window_hierarchy(self) -> str:
+        value = self._call("jsonrpc_call", "dumpWindowHierarchy", [False, 50], timeout=self._timeout_seconds)
+        if not isinstance(value, str):
+            raise SkuSelectionDeviceAdapterError("节点树读取失败，已停止操作。")
+        return value
+
+    def tap_sku_entry(self, bounds: str) -> None:
+        # 超时也可能表示底层事件已经送达；必须先封存 attempt，后续绝不重试该入口。
+        self._entry_was_tapped = True
+        self._tap_bounds_once(bounds)
+
+    def tap_sku_option(self, bounds: str) -> None:
+        self._tap_bounds_once(bounds)
+
+    def leave_sku_panel(self) -> None:
+        if self._left_panel:
+            raise SkuSelectionDeviceAdapterError("规格面板已经执行过返回，已停止操作。")
+        # 底层调用即使报错也可能已把返回事件送达；先封存本次机会，finally 不得再次返回。
+        self._left_panel = True
+        self._call("jsonrpc_call", "pressKey", ["back"], timeout=self._timeout_seconds)
+
+    def capture_screenshot(self) -> str:
+        value = self._call("jsonrpc_call", "takeScreenshot", SCREENSHOT_PARAMS, timeout=self._timeout_seconds)
+        if not isinstance(value, str):
+            raise SkuSelectionScreenshotError("规格面板原始截图读取失败，未发布任何证据产物。")
+        return value
+
+    def display_size(self) -> tuple[int, int]:
+        value = self._call("window_size")
+        if not isinstance(value, tuple) or len(value) != 2 or any(not isinstance(item, int) for item in value):
+            raise SkuSelectionDeviceAdapterError("无法读取屏幕坐标空间，已停止操作。")
+        return value
+
+    def _tap_bounds_once(self, bounds: str) -> None:
+        left, top, right, bottom = _action_bounds(bounds)
+        center_x = left + (right - left) // 2
+        center_y = top + (bottom - top) // 2
+        self._call("jsonrpc_call", "click", [center_x, center_y], timeout=self._timeout_seconds)
+
+    def _call(self, method: str, *args: Any, **kwargs: Any) -> Any:
+        try:
+            operation = getattr(self._device, method)
+            return operation(*args, **kwargs)
+        except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
+            raise SkuSelectionRunTimeoutError("规格面板设备操作超时，已停止操作。") from error
+        except SkuSelectionRunError:
+            raise
+        except Exception as error:
+            raise SkuSelectionDeviceAdapterError("规格面板设备操作失败，已停止操作。") from error
+
+
+class SkuSelectionRunner:
+    """只运行 T-103 目标规格恢复、价格确认、原始截图和一次安全退出。"""
+
+    def __init__(
+        self,
+        adb_client: AdbClient,
+        connector: Callable[[str], Any],
+        timeout_seconds: float,
+        monotonic_clock: Callable[[], float] = monotonic,
+    ) -> None:
+        if not _is_positive_finite(timeout_seconds):
+            raise ValueError("timeout_seconds 必须是大于 0 的有限数值")
+        self._adb_client = adb_client
+        self._connector = connector
+        self._timeout_seconds = timeout_seconds
+        self._monotonic_clock = monotonic_clock
+
+    def run(
+        self,
+        serial: str,
+        product_url: str,
+        task_color: str,
+        task_size: str,
+        output_directory: Path,
+    ) -> SkuSelectionRunResult:
+        link = parse_product_url(product_url)
+        if link.goods_id != EXPECTED_GOODS_ID:
+            raise SkuSelectionRunError("商品不是 T-103 已取证目标，已停止操作。")
+        selection = resolve_task_selection(task_color, task_size)
+        target = Path(output_directory)
+        _validate_new_target(target)
+
+        adapter: UiautomatorSkuPanelAdapter | None = None
+        flow: SkuSelectionFlow | None = None
+        staging = _prepare_staging(target)
+        deadline = self._monotonic_clock() + self._timeout_seconds
+        try:
+            inspection = self._adb_client.inspect(serial)
+            _require_expected_device(inspection)
+            adapter = UiautomatorSkuPanelAdapter(self._connector(serial), self._timeout_seconds)
+            _require_expected_version(adapter.app_info(PDD_PACKAGE))
+            if adapter.display_size() != EXPECTED_SCREEN_SIZE:
+                raise SkuSelectionRunError("设备不是已取证的竖屏坐标空间，已停止操作。")
+            pre_intent_hierarchy = adapter.dump_window_hierarchy()
+            # 固定 ACTION_VIEW、固定 PDD package 和 canonical goods_id；不接受任意 URL 或 shell。
+            self._adb_client.start_pdd_view_intent(serial, link.goods_id)
+
+            remaining = deadline - self._monotonic_clock()
+            if remaining <= 0:
+                raise SkuSelectionRunTimeoutError("等待规格入口超时，未执行点击。")
+            flow = SkuSelectionFlow(adapter, entry_wait_timeout_seconds=remaining)
+            flow.open_sku_panel(link.canonical_url, pre_intent_hierarchy)
+            flow.select_sku_options(selection)
+            unit_price = flow.verify_target_selection_and_read_price(selection)
+            if unit_price != EXPECTED_UNIT_PRICE:
+                raise SkuSelectionUnexpectedPriceError("规格面板现价不是本任务已确认值，已停止操作。")
+
+            screenshot_path = staging / "screenshot.png"
+            try:
+                _save_base64_screenshot(adapter.capture_screenshot(), screenshot_path)
+                _require_screenshot_size(screenshot_path)
+            except SkuSelectionRunError:
+                raise
+            except Exception as error:
+                raise SkuSelectionScreenshotError("规格面板原始截图保存失败，未发布任何证据产物。") from error
+
+            manifest_path = staging / "manifest.json"
+            # 截图可能落在动态页面切换边界；发布前必须用一棵更新节点树同时重证两维和现价。
+            final_price = flow.verify_target_selection_and_read_price(selection)
+            if final_price != EXPECTED_UNIT_PRICE:
+                raise SkuSelectionUnexpectedPriceError("截图后规格面板现价不是本任务已确认值，已停止操作。")
+            # 正常路径仍经 Flow 做最后一次前台和面板判定；返回操作只发生一次。
+            flow.exit_sku_panel_safely()
+            manifest_path.write_text(
+                json.dumps(_manifest(inspection, serial, link, screenshot_path, task_color, task_size), ensure_ascii=False, indent=2, sort_keys=True) + "\n",
+                encoding="utf-8",
+            )
+            # Windows 的 rename 不替换既有目标；并发创建 target 时保留其内容并把本次运行判失败。
+            os.rename(staging, target)
+            staging = None
+        except (DeviceConnectionError, SkuSelectionRunError, SkuSelectionError):
+            _clean_staging(staging)
+            raise
+        except (AdbTimeout, HTTPTimeoutError, TimeoutError) as error:
+            _clean_staging(staging)
+            raise SkuSelectionRunTimeoutError("规格面板运行超时，未发布任何证据产物。") from error
+        except OSError as error:
+            _clean_staging(staging)
+            raise SkuSelectionRunError("规格面板证据目录无法创建或发布，未发布任何证据产物。") from error
+        except Exception as error:
+            _clean_staging(staging)
+            raise SkuSelectionRunError("规格面板运行未完成，未发布任何证据产物。") from error
+        finally:
+            # 失败路径只能复用 Flow 的版本、前台和面板证明；证明不了便停止，绝不盲目返回。
+            if flow is not None and adapter is not None and adapter.entry_was_tapped and not adapter.left_panel:
+                try:
+                    flow.reconcile_pending_action()
+                    flow.exit_sku_panel_safely()
+                except (SkuSelectionRunError, SkuSelectionError):
+                    pass
+
+        return SkuSelectionRunResult(
+            output_directory=target,
+            screenshot_path=target / "screenshot.png",
+            manifest_path=target / "manifest.json",
+            unit_price=EXPECTED_UNIT_PRICE,
+        )
+
+
+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_new_target(target: Path) -> None:
+    if target.exists():
+        raise SkuSelectionRunError("输出目录已存在；为防止覆盖旧证据，已停止操作。")
+    if not target.name:
+        raise SkuSelectionRunError("输出目录必须是明确的新目录。")
+
+
+def _prepare_staging(target: Path) -> Path:
+    staging: Path | None = None
+    try:
+        target.parent.mkdir(parents=True, exist_ok=True)
+        staging = target.parent / f".{target.name}.staging-{uuid4().hex}"
+        staging.mkdir()
+        probe = staging / ".write-probe"
+        probe.write_bytes(b"ok")
+        probe.unlink()
+        return staging
+    except OSError as error:
+        _clean_staging(staging)
+        raise SkuSelectionRunError("输出目录不可写，已停止操作。") from error
+
+
+def _clean_staging(staging: Path | None) -> None:
+    if staging is not None and staging.exists():
+        shutil.rmtree(staging)
+
+
+def _require_expected_version(app_info: object) -> str:
+    version = (app_info.get("versionName") or app_info.get("version_name")) if isinstance(app_info, dict) else None
+    if version != EXPECTED_PDD_VERSION:
+        raise SkuSelectionRunError("拼多多版本与已取证版本不一致，已停止操作。")
+    return version
+
+
+def _require_expected_device(inspection: DeviceInspection) -> None:
+    if inspection.model != EXPECTED_DEVICE_MODEL or inspection.android_version != EXPECTED_ANDROID_VERSION:
+        raise SkuSelectionRunError("设备型号或 Android 版本不是已取证组合，已停止操作。")
+
+
+def _require_screenshot_size(screenshot_path: Path) -> None:
+    try:
+        with Image.open(screenshot_path) as image:
+            image.load()
+            if image.size != EXPECTED_SCREEN_SIZE:
+                raise SkuSelectionScreenshotError("原始截图坐标空间不是已取证尺寸，未发布任何证据产物。")
+    except SkuSelectionRunError:
+        raise
+    except (UnidentifiedImageError, OSError) as error:
+        raise SkuSelectionScreenshotError("原始截图无效，未发布任何证据产物。") from error
+
+
+def _manifest(inspection: DeviceInspection, serial: str, link: ProductUrl, screenshot_path: Path, task_color: str, task_size: str) -> dict[str, Any]:
+    """仅写可审计摘要；原始 serial、节点树、页面文案和实际截图内容均不写入 manifest。"""
+
+    return {
+        "schema_version": 1,
+        "captured_at": datetime.now(UTC).isoformat(),
+        "operation": "t103-sku-selection",
+        "product": {"goods_id": link.goods_id, "canonical_url": link.canonical_url},
+        "target_selection": {"color": task_color, "size": task_size},
+        "unit_price": EXPECTED_UNIT_PRICE,
+        "selection_status": "restored",
+        "panel_status": "verified",
+        "safe_exit": "completed",
+        "page_identity": "human_review_required",
+        "channel": "wifi" if ":" in serial else "usb",
+        "serial_sha256": sha256(serial.encode("utf-8")).hexdigest(),
+        "device": {
+            "model": inspection.model,
+            "android_version": inspection.android_version,
+            "pdd_package": PDD_PACKAGE,
+            "pdd_version": EXPECTED_PDD_VERSION,
+        },
+        "artifacts": [{"path": screenshot_path.name, "sha256": _sha256_file(screenshot_path)}],
+    }

client/tests/pdd/fixtures/product_entry_8_17_0.xml

@@ -0,0 +1,14 @@
+<hierarchy>
+  <node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[0,1256][1080,1355]" clickable="true" enabled="true" visible-to-user="true">
+    <node package="com.xunmeng.pinduoduo" class="android.widget.FrameLayout" bounds="[712,1312][1056,1355]" clickable="false" enabled="true" visible-to-user="true">
+      <node package="com.xunmeng.pinduoduo" class="android.widget.ViewSwitcher" bounds="[712,1312][1056,1355]" clickable="false" enabled="true" visible-to-user="true">
+        <node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[712,1312][1056,1355]" clickable="false" enabled="true" visible-to-user="true">
+          <node text="快要抢光" package="com.xunmeng.pinduoduo" class="android.widget.TextView" bounds="[900,1312][1056,1355]" clickable="false" enabled="true" visible-to-user="true" />
+        </node>
+      </node>
+    </node>
+  </node>
+  <node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[446,2166][1080,2328]" clickable="true" enabled="true" visible-to-user="true">
+    <node text="免拼购买" package="com.xunmeng.pinduoduo" class="android.widget.TextView" bounds="[688,2256][856,2305]" clickable="false" enabled="true" visible-to-user="true" />
+  </node>
+</hierarchy>

client/tests/pdd/fixtures/sku_panel_8_17_0.xml

@@ -0,0 +1,26 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<hierarchy>
+  <node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[0,474][1080,2328]">
+    <node package="" class="android.view.ViewGroup" bounds="[396,498][895,570]">
+      <node text="快卖完 ¥12.88" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[396,503][712,570]" />
+      <node text="¥29.88" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[730,503][895,570]" />
+    </node>
+    <node text="已选： 黑色 CHA (纯棉） M（建议100-115）" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[396,654][1053,716]" />
+    <node package="com.xunmeng.pinduoduo" class="androidx.recyclerview.widget.RecyclerView" bounds="[36,1000][1080,1631]">
+      <node content-desc="黑色 CHA (纯棉）" package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]" />
+      <node content-desc="粉红" package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" selected="false" clickable="true" enabled="true" visible-to-user="true" bounds="[456,1000][690,1172]" />
+    </node>
+    <node package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" bounds="[36,1637][1044,2045]">
+      <node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" bounds="[36,1637][1044,1718]">
+        <node text="尺码" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[36,1654][114,1700]" />
+      </node>
+      <node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" bounds="[36,1730][1044,2045]">
+        <node text="M（建议100-115）" package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[439,1730][831,1815]" />
+        <node text="L（建议115-130）" package="com.xunmeng.pinduoduo" class="android.view.ViewGroup" selected="false" clickable="true" enabled="true" visible-to-user="true" bounds="[840,1730][1044,1815]" />
+      </node>
+    </node>
+    <node package="com.xunmeng.pinduoduo" class="android.widget.LinearLayout" clickable="true" enabled="true" visible-to-user="true" bounds="[357,2181][722,2328]">
+      <node text="提交订单 ¥12.88" package="com.xunmeng.pinduoduo" class="android.widget.TextView" clickable="false" enabled="true" visible-to-user="true" bounds="[369,2225][710,2284]" />
+    </node>
+  </node>
+</hierarchy>

client/tests/pdd/test_sku_selection.py

@@ -0,0 +1,689 @@
+from __future__ import annotations
+
+import ast
+import base64
+from contextlib import redirect_stderr
+from io import BytesIO
+import importlib.util
+from pathlib import Path
+from tempfile import TemporaryDirectory
+import unittest
+from unittest.mock import patch
+from xml.etree import ElementTree
+
+from PIL import Image
+
+import cmbuyer_client.pdd as pdd
+import cmbuyer_client.pdd.sku_selection_runner as runner_module
+from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
+from cmbuyer_client.pdd import SkuSelectionError, SkuSelectionFlow, SkuSelectionRunner
+from cmbuyer_client.pdd.sku_selection import SkuPanelDevice, _action_bounds, resolve_task_selection
+from cmbuyer_client.pdd.sku_selection_runner import (
+    SkuSelectionDeviceAdapterError,
+    SkuSelectionRunError,
+    SkuSelectionScreenshotError,
+    UiautomatorSkuPanelAdapter,
+)
+
+
+_FIXTURE = Path(__file__).with_name("fixtures") / "sku_panel_8_17_0.xml"
+_ENTRY_FIXTURE = Path(__file__).with_name("fixtures") / "product_entry_8_17_0.xml"
+_TARGET_URL = "https://mobile.yangkeduo.com/goods.html?goods_id=937122477375"
+_TASK_COLOR = "黑色CHA(纯棉)"
+_TASK_SIZE = "M(建议100-115)"
+_PRODUCT_PAGE = _ENTRY_FIXTURE.read_text(encoding="utf-8")
+
+
+def _png_base64() -> str:
+    image = Image.new("RGB", (1080, 2376), "white")
+    raw = BytesIO()
+    image.save(raw, format="PNG")
+    return base64.b64encode(raw.getvalue()).decode("ascii")
+
+
+class _RawDevice:
+    def __init__(self, hierarchy: str = _PRODUCT_PAGE, screenshot: str | None = None) -> None:
+        self.hierarchy = hierarchy
+        self.panel_hierarchy = _FIXTURE.read_text(encoding="utf-8")
+        self.version = "8.17.0"
+        self.package = "com.xunmeng.pinduoduo"
+        self.screenshot = _png_base64() if screenshot is None else screenshot
+        self.calls: list[tuple[object, ...]] = []
+        self.fail_color_readback = False
+
+    def app_info(self, package_name: str) -> dict[str, str]:
+        self.calls.append(("app_info", package_name))
+        return {"versionName": self.version}
+
+    def app_current(self) -> dict[str, str]:
+        self.calls.append(("app_current",))
+        return {"package": self.package}
+
+    def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+        self.calls.append(("jsonrpc", method, params, timeout))
+        if method == "dumpWindowHierarchy":
+            return self.hierarchy
+        if method == "takeScreenshot":
+            return self.screenshot
+        if method == "pressKey":
+            self.hierarchy = "<hierarchy />"
+            return ""
+        if method == "click":
+            if not isinstance(params, list) or len(params) != 2:
+                raise AssertionError(params)
+            self._apply_tap(int(params[0]), int(params[1]))
+            return ""
+        raise AssertionError(method)
+
+    def _apply_tap(self, x: int, y: int) -> None:
+        if "快要抢光" in self.hierarchy and "[396,498][895,570]" not in self.hierarchy:
+            self.hierarchy = self.panel_hierarchy
+            return
+        root = ElementTree.fromstring(self.hierarchy)
+        target = next(node for node in root.iter("node") if _center(node.get("bounds", "")) == (x, y))
+        color = target.get("bounds", "").endswith("][438,1172]")
+        for node in root.iter("node"):
+            if node.get("selected") is not None and ((color and ",1000]" in node.get("bounds", "")) or (not color and ",1730]" in node.get("bounds", ""))):
+                node.set("selected", "false")
+        if color and self.fail_color_readback:
+            next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true")
+        else:
+            target.set("selected", "true")
+        self.hierarchy = ElementTree.tostring(root, encoding="unicode")
+
+    def window_size(self) -> tuple[int, int]:
+        self.calls.append(("window_size",))
+        return 1080, 2376
+
+    def select_alternates(self) -> None:
+        root = ElementTree.fromstring(self.panel_hierarchy)
+        for node in root.iter("node"):
+            if node.get("selected") is not None:
+                node.set("selected", "false")
+        next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true")
+        next(node for node in root.iter("node") if node.get("text") == "L（建议115-130）").set("selected", "true")
+        self.panel_hierarchy = ElementTree.tostring(root, encoding="unicode")
+        if self.hierarchy != _PRODUCT_PAGE:
+            self.hierarchy = self.panel_hierarchy
+
+
+def _center(bounds: str) -> tuple[int, int]:
+    left_top, right_bottom = bounds.split("][")
+    left, top = (int(value) for value in left_top.removeprefix("[").split(","))
+    right, bottom = (int(value) for value in right_bottom.removesuffix("]").split(","))
+    return left + (right - left) // 2, top + (bottom - top) // 2
+
+
+def _entry_chain(root: ElementTree.Element) -> list[ElementTree.Element]:
+    parents = {child: parent for parent in root.iter() for child in parent}
+    child = next(node for node in root.iter("node") if node.get("text") == "快要抢光")
+    chain = [child]
+    for _ in range(4):
+        chain.append(parents[chain[-1]])
+    return chain
+
+
+def _mutate_entry(depth: int, attribute: str, value: str) -> str:
+    root = ElementTree.fromstring(_PRODUCT_PAGE)
+    _entry_chain(root)[depth].set(attribute, value)
+    return ElementTree.tostring(root, encoding="unicode")
+
+
+def _without_entry() -> str:
+    root = ElementTree.fromstring(_PRODUCT_PAGE)
+    root.remove(_entry_chain(root)[4])
+    return ElementTree.tostring(root, encoding="unicode")
+
+
+def _duplicate_entry() -> str:
+    root = ElementTree.fromstring(_PRODUCT_PAGE)
+    entry_root = _entry_chain(root)[4]
+    root.append(ElementTree.fromstring(ElementTree.tostring(entry_root, encoding="unicode")))
+    return ElementTree.tostring(root, encoding="unicode")
+
+
+def _actions(device: _RawDevice, method: str) -> list[tuple[object, ...]]:
+    return [call for call in device.calls if call[0] == "jsonrpc" and call[1] == method]
+
+
+def _tap_centers(device: _RawDevice) -> list[tuple[int, int]]:
+    return [tuple(call[2]) for call in _actions(device, "click")]  # type: ignore[misc]
+
+
+class _FakeAdb:
+    def __init__(self) -> None:
+        self.calls: list[tuple[object, ...]] = []
+        self.inspection = DeviceInspection(AdbDevice(serial="device-1", state="device"), "PKG110", "16")
+        self.on_intent: callable | None = None
+
+    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) -> object:
+        self.calls.append(("intent", serial, goods_id))
+        if self.on_intent is not None:
+            self.on_intent()
+        return object()
+
+
+class SkuSelectionFlowTests(unittest.TestCase):
+    def _assert_entry_rejected_without_click(self, hierarchy: str) -> None:
+        now = [0.0]
+        device = _RawDevice(hierarchy)
+        flow = SkuSelectionFlow(
+            UiautomatorSkuPanelAdapter(device, 10),
+            0.01,
+            0.01,
+            lambda: now[0],
+            lambda seconds: now.__setitem__(0, now[0] + seconds),
+        )
+        with self.assertRaises(SkuSelectionError):
+            flow.open_sku_panel(_TARGET_URL)
+        self.assertEqual(_actions(device, "click"), [])
+
+    def test_target_mapping_is_exact_and_success_path_restores_target(self) -> None:
+        device = _RawDevice()
+        adapter = UiautomatorSkuPanelAdapter(device, 10)
+        flow = SkuSelectionFlow(adapter)
+
+        flow.open_sku_panel(_TARGET_URL)
+        flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
+        self.assertEqual(flow.read_sku_unit_price(), "12.88")
+        flow.exit_sku_panel_safely()
+
+        self.assertEqual(_tap_centers(device), [(978, 1333)])
+        self.assertEqual(_actions(device, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)])
+
+    def test_full_verified_entry_structure_taps_exact_text_child_once(self) -> None:
+        device = _RawDevice()
+        SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
+
+        self.assertEqual(_tap_centers(device), [(978, 1333)])
+
+    def test_entry_child_and_every_ancestor_attribute_drift_never_clicks(self) -> None:
+        expected_clickable = ("false", "false", "false", "false", "true")
+        for depth in range(5):
+            changes = {
+                "package": "other.package",
+                "class": "android.widget.Button",
+                "bounds": "[1,1][2,2]",
+                "clickable": "true" if expected_clickable[depth] == "false" else "false",
+                "enabled": "false",
+                "visible-to-user": "false",
+            }
+            for attribute, value in changes.items():
+                with self.subTest(depth=depth, attribute=attribute):
+                    self._assert_entry_rejected_without_click(_mutate_entry(depth, attribute, value))
+
+    def test_duplicate_entry_and_forbidden_sibling_entry_never_click(self) -> None:
+        self._assert_entry_rejected_without_click(_duplicate_entry())
+        self._assert_entry_rejected_without_click(_without_entry())
+        self._assert_entry_rejected_without_click(_mutate_entry(0, "clickable", "true"))
+
+    def test_unknown_task_or_ui_variants_are_rejected_without_action(self) -> None:
+        for color, size in (("黑色 CHA (纯棉）", _TASK_SIZE), (_TASK_COLOR, "M（建议100-115）"), ("黑色CHA（纯棉）", _TASK_SIZE)):
+            with self.subTest(color=color, size=size), self.assertRaises(SkuSelectionError):
+                resolve_task_selection(color, size)
+
+        device = _RawDevice(_FIXTURE.read_text(encoding="utf-8"))
+        with self.assertRaises(SkuSelectionError):
+            SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).select_sku_options(
+                resolve_task_selection(_TASK_COLOR, _TASK_SIZE).__class__("粉红", "L（建议115-130）")
+            )
+        self.assertEqual(_actions(device, "click"), [])
+
+    def test_option_selected_and_container_drift_fail_closed_before_click(self) -> None:
+        base = _FIXTURE.read_text(encoding="utf-8")
+        cases = (
+            base.replace('selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"'),
+            base.replace('selected="true" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'selected="maybe" clickable="true" enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"'),
+            base.replace('bounds="[126,1000][438,1172]"', 'bounds="[1,1][20,20]"'),
+            base.replace('enabled="true" visible-to-user="true" bounds="[126,1000][438,1172]"', 'enabled="false" visible-to-user="true" bounds="[126,1000][438,1172]"'),
+        )
+        for hierarchy in cases:
+            with self.subTest(), self.assertRaises(SkuSelectionError):
+                SkuSelectionFlow(UiautomatorSkuPanelAdapter(_RawDevice(hierarchy), 10)).select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
+
+    def test_invalid_bounds_stop_before_action(self) -> None:
+        for bounds in ("", "[1,2][1,3]", "[1,2][3,2]", "[0,0][1081,1]", "[0,0][1,2377]", "[a,0][1,1]"):
+            with self.subTest(bounds=bounds), self.assertRaises(SkuSelectionError):
+                _action_bounds(bounds)
+
+        device = _RawDevice(_PRODUCT_PAGE.replace("[900,1312][1056,1355]", "[0,0][1081,1]"))
+        with self.assertRaises(SkuSelectionError):
+            SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
+        self.assertEqual(_actions(device, "click"), [])
+
+    def test_color_readback_failure_never_attempts_second_option(self) -> None:
+        device = _RawDevice()
+        device.fail_color_readback = True
+        flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10))
+        flow.open_sku_panel(_TARGET_URL)
+        device.select_alternates()
+        with self.assertRaises(SkuSelectionError):
+            flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
+        self.assertEqual(_tap_centers(device), [(978, 1333), (282, 1086)])
+
+    def test_non_target_selection_restores_each_dimension_once(self) -> None:
+        device = _RawDevice()
+        flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10))
+        flow.open_sku_panel(_TARGET_URL)
+        device.select_alternates()
+        flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
+        self.assertEqual(
+            _tap_centers(device),
+            [(978, 1333), (282, 1086), (635, 1772)],
+        )
+
+    def test_price_rejects_coupon_prefix_extra_amount_and_bottom_action(self) -> None:
+        for replacement in ("券后 ¥12.88", "会员补贴 ¥12.88", "到手 ¥12.88", "实付 ¥12.88", "区间 ¥12.88", "原价 ¥12.88", "划线价 ¥12.88", "最低 ¥12.88", "低至 ¥12.88", "起价 ¥12.88", "快卖完 1 ¥12.88", "快卖完 ¥12.88 ¥11.88"):
+            with self.subTest(replacement=replacement):
+                device = _RawDevice(_FIXTURE.read_text(encoding="utf-8").replace("快卖完 ¥12.88", replacement))
+                with self.assertRaises(SkuSelectionError):
+                    SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).read_sku_unit_price()
+        device = _RawDevice(_FIXTURE.read_text(encoding="utf-8").replace("快卖完 ¥12.88", "提交订单 ¥12.88"))
+        with self.assertRaises(SkuSelectionError):
+            SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).read_sku_unit_price()
+        clickable_parent = _FIXTURE.read_text(encoding="utf-8").replace(
+            '<node package="" class="android.view.ViewGroup" bounds="[396,498][895,570]">',
+            '<node package="" class="android.view.ViewGroup" clickable="true" bounds="[396,498][895,570]">',
+        )
+        with self.assertRaises(SkuSelectionError):
+            SkuSelectionFlow(UiautomatorSkuPanelAdapter(_RawDevice(clickable_parent), 10)).read_sku_unit_price()
+
+    def test_public_api_and_protocol_have_no_broad_or_order_operations(self) -> None:
+        forbidden = {"quantity", "confirm", "authorization", "fence", "submit", "payment", "click"}
+        self.assertTrue(forbidden.isdisjoint(SkuSelectionFlow.__dict__))
+        self.assertTrue(forbidden.isdisjoint(SkuPanelDevice.__dict__))
+        self.assertTrue(forbidden.isdisjoint(pdd.__all__))
+
+    def test_static_ast_boundary_limits_flow_runner_adapter_and_cli(self) -> None:
+        root = Path(__file__).resolve().parents[2]
+        files = (
+            root / "src" / "cmbuyer_client" / "pdd" / "sku_selection.py",
+            root / "src" / "cmbuyer_client" / "pdd" / "sku_selection_runner.py",
+            root / "scripts" / "run_t103_sku_selection.py",
+        )
+        forbidden = ("quantity", "confirm", "authorization", "fence", "submit_order", "payment")
+        for path in files:
+            source = path.read_text(encoding="utf-8")
+            with self.subTest(path=path.name):
+                self.assertTrue(all(token not in source.lower() for token in forbidden))
+                tree = ast.parse(source)
+                self.assertFalse(any(isinstance(node, ast.ImportFrom) and node.module in {"selenium", "requests"} for node in ast.walk(tree)))
+        runner_tree = ast.parse(files[1].read_text(encoding="utf-8"))
+        click_calls = [node for node in ast.walk(runner_tree) if isinstance(node, ast.Constant) and node.value == "click"]
+        self.assertEqual(len(click_calls), 1)
+
+    def test_entry_wait_rejects_unchanged_or_duplicate_page_without_click(self) -> None:
+        now = [0.0]
+        device = _RawDevice()
+        flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10), 0.01, 0.01, lambda: now[0], lambda seconds: now.__setitem__(0, now[0] + seconds))
+        with self.assertRaises(SkuSelectionError):
+            flow.open_sku_panel(_TARGET_URL, _PRODUCT_PAGE)
+        self.assertEqual(_actions(device, "click"), [])
+
+    def test_action_postcondition_wait_never_repeats_entry_click(self) -> None:
+        class NoPanelAfterEntry(_RawDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "click":
+                    self.calls.append(("jsonrpc", method, params, timeout))
+                    return ""
+                return super().jsonrpc_call(method, params, timeout)
+
+        device = NoPanelAfterEntry()
+        with self.assertRaises(SkuSelectionError):
+            SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
+        self.assertEqual(_tap_centers(device), [(978, 1333)])
+
+        duplicate = _duplicate_entry()
+        device = _RawDevice(duplicate)
+        with self.assertRaises(SkuSelectionError):
+            SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).open_sku_panel(_TARGET_URL)
+        self.assertEqual(_actions(device, "click"), [])
+
+    def test_fixture_contains_no_address_phone_or_payment_credentials(self) -> None:
+        for fixture in (_FIXTURE, _ENTRY_FIXTURE):
+            content = fixture.read_text(encoding="utf-8")
+            with self.subTest(fixture=fixture.name):
+                self.assertNotRegex(content, r"1[3-9]\d{9}")
+                for forbidden in ("地址", "收货", "支付", "银行卡", "身份证"):
+                    self.assertNotIn(forbidden, content)
+        content = _FIXTURE.read_text(encoding="utf-8")
+        root = ElementTree.fromstring(content)
+        leaf = next(node for node in root.iter("node") if node.get("text") == "提交订单 ¥12.88")
+        self.assertEqual(leaf.get("clickable"), "false")
+        self.assertEqual(leaf.get("bounds"), "[369,2225][710,2284]")
+
+
+class SkuSelectionRunnerTests(unittest.TestCase):
+    def _runner(self, adb: _FakeAdb, device: _RawDevice) -> SkuSelectionRunner:
+        device.hierarchy = "<hierarchy />"
+        adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("<hierarchy>", '<hierarchy post-intent="1">'))
+        return SkuSelectionRunner(adb, lambda serial: device, 10)
+
+    def test_runner_atomically_publishes_screenshot_and_redacted_manifest(self) -> None:
+        adb = _FakeAdb()
+        device = _RawDevice()
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "result"
+            result = self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
+
+            self.assertEqual(result.unit_price, "12.88")
+            manifest = result.manifest_path.read_text(encoding="utf-8")
+            self.assertTrue(result.screenshot_path.is_file())
+            self.assertNotIn("device-1", manifest)
+            self.assertNotIn("hierarchy", manifest)
+            self.assertNotIn("已选", manifest)
+            self.assertIn('"unit_price": "12.88"', manifest)
+            self.assertIn('"selection_status": "restored"', manifest)
+            self.assertIn('"panel_status": "verified"', manifest)
+            self.assertIn('"safe_exit": "completed"', manifest)
+            self.assertFalse((target / "hierarchy.xml").exists())
+            self.assertEqual(_actions(device, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)])
+
+    def test_target_created_during_publish_is_preserved_without_staging_residue(self) -> None:
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "result"
+            original_rename = runner_module.os.rename
+
+            def create_target_then_rename(source: str | Path, destination: str | Path) -> None:
+                Path(destination).mkdir()
+                (Path(destination) / "sentinel").write_text("keep", encoding="utf-8")
+                original_rename(source, destination)
+
+            with patch.object(runner_module.os, "rename", side_effect=create_target_then_rename), self.assertRaises(SkuSelectionRunError):
+                self._runner(_FakeAdb(), _RawDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
+            self.assertEqual((target / "sentinel").read_text(encoding="utf-8"), "keep")
+            self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
+
+    def test_bad_screenshot_or_existing_target_never_publishes_manifest(self) -> None:
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "result"
+            with self.assertRaises(SkuSelectionScreenshotError):
+                self._runner(_FakeAdb(), _RawDevice(screenshot="not-image")).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
+            self.assertFalse(target.exists())
+            self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
+
+            target = Path(temporary) / "write-failure"
+            with patch.object(runner_module, "_save_base64_screenshot", side_effect=OSError("private path")):
+                with self.assertRaises(SkuSelectionScreenshotError):
+                    self._runner(_FakeAdb(), _RawDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
+            self.assertFalse(target.exists())
+            self.assertEqual(list(Path(temporary).glob(".write-failure.staging-*")), [])
+
+            adb = _FakeAdb()
+            device = _RawDevice()
+            target.mkdir()
+            sentinel = target / "keep"
+            sentinel.write_text("keep", encoding="utf-8")
+            with self.assertRaises(SkuSelectionRunError):
+                self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
+            self.assertEqual(adb.calls, [])
+            self.assertEqual(device.calls, [])
+            self.assertEqual(sentinel.read_text(encoding="utf-8"), "keep")
+
+    def test_device_screen_and_output_preflight_fail_before_any_click(self) -> None:
+        with TemporaryDirectory() as temporary:
+            adb = _FakeAdb()
+            adb.inspection = DeviceInspection(AdbDevice(serial="device-1", state="device"), "wrong", "16")
+            device = _RawDevice()
+            with self.assertRaises(SkuSelectionRunError):
+                self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
+            self.assertEqual(device.calls, [])
+
+            class WrongScreenDevice(_RawDevice):
+                def window_size(self) -> tuple[int, int]:
+                    return 1080, 1920
+
+            device = WrongScreenDevice()
+            with self.assertRaises(SkuSelectionRunError):
+                self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "screen")
+            self.assertEqual(_actions(device, "click"), [])
+
+            parent_file = Path(temporary) / "not-a-directory"
+            parent_file.write_text("x", encoding="utf-8")
+            adb = _FakeAdb()
+            device = _RawDevice()
+            with self.assertRaises(SkuSelectionRunError):
+                self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, parent_file / "result")
+            self.assertEqual(adb.calls, [])
+            self.assertEqual(device.calls, [])
+
+    def test_small_but_valid_png_is_not_accepted(self) -> None:
+        image = Image.new("RGB", (1, 1), "white")
+        raw = BytesIO(); image.save(raw, format="PNG")
+        with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionScreenshotError):
+            self._runner(_FakeAdb(), _RawDevice(screenshot=base64.b64encode(raw.getvalue()).decode("ascii"))).run(
+                "device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result"
+            )
+
+    def test_failure_after_entry_attempts_one_safe_exit_and_hides_device_detail(self) -> None:
+        adb = _FakeAdb()
+        device = _RawDevice()
+        device.fail_color_readback = True
+        device.select_alternates()
+        with TemporaryDirectory() as temporary:
+            with self.assertRaises(SkuSelectionError):
+                self._runner(adb, device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
+        self.assertEqual(_actions(device, "pressKey"), [])
+
+        class FailingRawDevice(_RawDevice):
+            def app_info(self, package_name: str) -> dict[str, str]:
+                raise RuntimeError("device-1 <xml>private</xml>")
+
+        with self.assertRaises(SkuSelectionDeviceAdapterError) as raised:
+            UiautomatorSkuPanelAdapter(FailingRawDevice(), 10).app_info("com.xunmeng.pinduoduo")
+        self.assertNotIn("device-1", str(raised.exception))
+        self.assertNotIn("private", str(raised.exception))
+
+    def test_unverified_failure_never_sends_blind_back(self) -> None:
+        class InvalidAfterOptionDevice(_RawDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                value = super().jsonrpc_call(method, params, timeout)
+                if method == "click" and "[396,498][895,570]" in self.hierarchy:
+                    self.hierarchy = "<hierarchy />"
+                return value
+
+        device = InvalidAfterOptionDevice()
+        device.select_alternates()
+        with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionError):
+            self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
+        self.assertEqual(_actions(device, "pressKey"), [])
+
+    def test_adapter_timeout_is_mapped_without_third_party_detail(self) -> None:
+        class TimeoutRawDevice(_RawDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                raise TimeoutError("device-1 <hierarchy>private</hierarchy>")
+
+        with self.assertRaises(SkuSelectionRunError) as raised:
+            UiautomatorSkuPanelAdapter(TimeoutRawDevice(), 10).dump_window_hierarchy()
+        self.assertNotIn("device-1", str(raised.exception))
+        self.assertNotIn("private", str(raised.exception))
+
+    def test_entry_attempt_is_recorded_before_unconfirmed_click_and_not_retried(self) -> None:
+        class TimeoutTapDevice(_RawDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                self.calls.append(("jsonrpc", method, params, timeout))
+                if method == "click":
+                    raise TimeoutError("device detail")
+                return super().jsonrpc_call(method, params, timeout)
+
+        adapter = UiautomatorSkuPanelAdapter(TimeoutTapDevice(), 10)
+        with self.assertRaises(SkuSelectionRunError):
+            adapter.tap_sku_entry("[900,1312][1056,1355]")
+        self.assertTrue(adapter.entry_was_tapped)
+        self.assertEqual(_actions(adapter._device, "click"), [("jsonrpc", "click", [978, 1333], 10)])
+
+    def test_entry_stability_interruptions_never_click(self) -> None:
+        now = [0.0]
+        class SequenceDevice(_RawDevice):
+            def __init__(self) -> None:
+                super().__init__(); self.frames = [_PRODUCT_PAGE, "<hierarchy />", _PRODUCT_PAGE]
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "dumpWindowHierarchy" and self.frames:
+                    self.hierarchy = self.frames.pop(0)
+                return super().jsonrpc_call(method, params, timeout)
+        device = SequenceDevice()
+        flow = SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10), .02, .01, lambda: now[0], lambda x: now.__setitem__(0, now[0] + x))
+        with self.assertRaises(SkuSelectionError): flow.open_sku_panel(_TARGET_URL, "<hierarchy />")
+        self.assertEqual(_actions(device, "click"), [])
+
+    def test_screenshot_drift_and_foreground_drift_publish_nothing_and_never_back(self) -> None:
+        for drift in ("color", "size", "price"):
+            class DriftDevice(_RawDevice):
+                def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                    value = super().jsonrpc_call(method, params, timeout)
+                    if method == "takeScreenshot":
+                        if drift == "price":
+                            self.hierarchy = self.hierarchy.replace("快卖完 ¥12.88", "快卖完 ¥13.88")
+                        else:
+                            root = ElementTree.fromstring(self.hierarchy)
+                            if drift == "color":
+                                for node in root.iter("node"):
+                                    if node.get("selected") is not None and ",1000]" in node.get("bounds", ""):
+                                        node.set("selected", "false")
+                                next(node for node in root.iter("node") if node.get("content-desc") == "粉红").set("selected", "true")
+                            else:
+                                for node in root.iter("node"):
+                                    if node.get("selected") is not None and ",1730]" in node.get("bounds", ""):
+                                        node.set("selected", "false")
+                                next(node for node in root.iter("node") if node.get("text") == "L（建议115-130）").set("selected", "true")
+                            self.hierarchy = ElementTree.tostring(root, encoding="unicode")
+                    return value
+            with self.subTest(drift=drift), TemporaryDirectory() as temporary:
+                target = Path(temporary) / "out"
+                with self.assertRaises((SkuSelectionError, SkuSelectionRunError)):
+                    self._runner(_FakeAdb(), DriftDevice()).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
+                self.assertFalse(target.exists())
+                self.assertFalse((target / "manifest.json").exists())
+                self.assertEqual(list(Path(temporary).glob(".out.staging-*")), [])
+
+        device = _RawDevice(); device.select_alternates()
+        device.package = "other"
+        with self.assertRaises(SkuSelectionError): SkuSelectionFlow(UiautomatorSkuPanelAdapter(device, 10)).exit_sku_panel_safely()
+        self.assertEqual(_actions(device, "pressKey"), [])
+
+    def test_screenshot_then_foreground_drift_publishes_nothing_and_never_back(self) -> None:
+        class ForegroundDriftDevice(_RawDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                value = super().jsonrpc_call(method, params, timeout)
+                if method == "takeScreenshot": self.package = "other"
+                return value
+
+        device = ForegroundDriftDevice()
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "out"
+            with self.assertRaises(SkuSelectionError):
+                self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target)
+            self.assertFalse(target.exists())
+            self.assertFalse((target / "manifest.json").exists())
+            self.assertEqual(list(Path(temporary).glob(".out.staging-*")), [])
+        self.assertEqual(_actions(device, "pressKey"), [])
+
+    def test_option_timeout_reconciliation_controls_back_once(self) -> None:
+        class OptionTimeoutDevice(_RawDevice):
+            def __init__(self, delivered: bool) -> None:
+                super().__init__(); self.delivered = delivered; self.clicks = 0
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "click":
+                    self.clicks += 1
+                    if self.clicks == 2:
+                        if self.delivered: super().jsonrpc_call(method, params, timeout)
+                        else: self.calls.append(("jsonrpc", method, params, timeout))
+                        raise TimeoutError("uncertain option")
+                return super().jsonrpc_call(method, params, timeout)
+
+        for delivered, expected_back in ((False, 0), (True, 1)):
+            with self.subTest(delivered=delivered), TemporaryDirectory() as temporary:
+                device = OptionTimeoutDevice(delivered); device.select_alternates()
+                adb = _FakeAdb(); device.hierarchy = "<hierarchy />"
+                adb.on_intent = lambda: setattr(device, "hierarchy", _PRODUCT_PAGE.replace("<hierarchy>", '<hierarchy post-intent="1">'))
+                runner = SkuSelectionRunner(adb, lambda serial: device, .03)
+                with self.assertRaises(SkuSelectionRunError):
+                    runner.run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "out")
+                self.assertEqual(len(_actions(device, "click")), 2)
+                self.assertEqual(len(_actions(device, "pressKey")), expected_back)
+
+    def test_back_timeout_is_never_retried(self) -> None:
+        class BackTimeoutDevice(_RawDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "pressKey":
+                    super().jsonrpc_call(method, params, timeout)
+                    raise TimeoutError("back uncertain")
+                return super().jsonrpc_call(method, params, timeout)
+
+        device = BackTimeoutDevice()
+        with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError):
+            self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "out")
+        self.assertEqual(len(_actions(device, "pressKey")), 1)
+
+    def test_entry_click_timeout_reconciles_only_through_verified_flow_exit(self) -> None:
+        class DeliveredThenTimeout(_RawDevice):
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "click" and self.hierarchy != _FIXTURE.read_text(encoding="utf-8"):
+                    super().jsonrpc_call(method, params, timeout)
+                    raise TimeoutError("delivery uncertain")
+                return super().jsonrpc_call(method, params, timeout)
+
+        device = DeliveredThenTimeout()
+        with TemporaryDirectory() as temporary, self.assertRaises(SkuSelectionRunError):
+            self._runner(_FakeAdb(), device).run("device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, Path(temporary) / "result")
+        self.assertEqual(len(_actions(device, "click")), 1)
+        self.assertEqual(len(_actions(device, "pressKey")), 1)
+
+
+class SkuSelectionCliTests(unittest.TestCase):
+    def test_cli_accepts_only_target_url_and_task_values(self) -> None:
+        script = _load_runner_script()
+        valid = {
+            "serial": "device-1",
+            "url": _TARGET_URL,
+            "color": _TASK_COLOR,
+            "size": _TASK_SIZE,
+            "output_dir": Path("evidence"),
+            "timeout": 10.0,
+            "adb": "adb",
+        }
+        script.validate_arguments(type("Arguments", (), valid)())
+        for field, value in (("serial", ""), ("url", "https://mobile.yangkeduo.com/goods.html?goods_id=1"), ("color", "黑色 CHA (纯棉）"), ("size", "M（建议100-115）"), ("timeout", 0), ("timeout", float("inf"))):
+            with self.subTest(field=field, value=value), self.assertRaises((ValueError, SkuSelectionError)):
+                script.validate_arguments(type("Arguments", (), valid | {field: value})())
+
+    def test_cli_main_catches_flow_error_without_traceback_or_page_body(self) -> None:
+        script = _load_runner_script()
+
+        class FlowFailingRunner:
+            def __init__(self, *args: object, **kwargs: object) -> None: pass
+            def run(self, *args: object, **kwargs: object) -> object:
+                raise SkuSelectionError("<hierarchy>page-body</hierarchy>")
+
+        stderr = BytesIO()
+        # TextIOWrapper keeps the assertion independent from host console encoding.
+        import io
+        text_stderr = io.TextIOWrapper(stderr, encoding="utf-8")
+        with patch.object(script, "SkuSelectionRunner", FlowFailingRunner), redirect_stderr(text_stderr):
+            status = script.main([
+                "--serial", "device-1", "--url", _TARGET_URL, "--color", _TASK_COLOR,
+                "--size", _TASK_SIZE, "--output-dir", "evidence",
+            ])
+        text_stderr.flush()
+        output = stderr.getvalue().decode("utf-8")
+        self.assertEqual(status, 1)
+        self.assertNotIn("Traceback", output)
+        self.assertNotIn("page-body", output)
+
+
+def _load_runner_script() -> object:
+    path = Path(__file__).resolve().parents[2] / "scripts" / "run_t103_sku_selection.py"
+    specification = importlib.util.spec_from_file_location("run_t103_sku_selection_test", path)
+    if specification is None or specification.loader is None:
+        raise RuntimeError("无法加载 T-103 运行脚本。")
+    module = importlib.util.module_from_spec(specification)
+    specification.loader.exec_module(module)
+    return module
+
+
+if __name__ == "__main__":
+    unittest.main()

docs/tasks/T-103.md

@@ -17,15 +17,12 @@ write_paths:
   - client/tests/pdd/**
   - client/tests/device/**
   - client/scripts/capture_sku_panel_spike.py
+  - client/scripts/run_t103_sku_selection.py
   - client/scripts/sanitize_sku_panel_evidence.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-04T08:24:57Z sha256=19a0864d83c5e41e57a29002b69f7767b50429dcf0fbace02f8303f92b71bac6 -->
+<!-- BEGIN VIKUNJA EXPORT id=23 synced=2026-08-04T10:47:47Z sha256=5471bdfeed1fffe4f0abd25201b9d5bf7bf55fa9500c11aa297eb4c422da5f02 -->
 ## 问题 / 背景
 
 T-102 已证明 canonical 链接可进入目标商品。T-103 在 PKG110 / Android 16 / 拼多多 8.17.0、goods_id `937122477375` 上确认：规格面板由详情页精确唯一的“快要抢光”打开；T-110 已把该证据/版本绑定入口批准为受控导航。面板刚打开时目标颜色“黑色CHA(纯棉)”和尺码“M(建议100-115)”自动选中。
@@ -173,6 +170,18 @@ T-103 sanitizer v2 坐标修正与主审：提交 44c027a 将 screenshot space
 ### 2026-08-04T08:24:12Z · ila
 
 2026-08-04：T-111 单趟契约主审通过，T-103 解除架构阻塞。恢复后只实现 SkuSelectionFlow：受控入口、维度内精确选择、选中态读回、SKU 当前价唯一读取、原始截图和安全退出；不得包含数量、确认页、授权、提交围栏、提交订单或支付。客户端工作树现有未提交 v6 遮罩测试残留必须先撤销到已提交 v5 基线，再开始新实现。needs_device=true，离线实现后仍等待人工真机验收。
+
+### 2026-08-04T09:50:25Z · ila
+
+2026-08-04 离线实现与主审通过：提交 b49a9b4 实现仅限已取证 PDD 8.17.0 / goods_id 937122477375 的受控规格入口、颜色/尺码精确恢复、SKU 当前价读取、本机原始截图和单次安全退出；不包含数量、确认页、授权、提交围栏、提交订单或付款能力。独立审计 PASS，合入主分支 5f650f1；完整 init、109 项测试、Go vet/build、compileall、上下文校验及 diff-check 全部通过。needs_device=true，任务继续 DOING，等待项目所有者先把已记忆规格改为非目标值并返回 PDD 首页后运行真机验收。
+
+### 2026-08-04T10:01:50Z · ila
+
+2026-08-04 T-103 入口父容器追加真机证据：只读取证目录 C:\Users\ila20\AppData\Local\cmbuyer\artifacts\T-103\entry-parent-evidence-937122477375-20260804-175737；截图 screenshot.png，XML hierarchy.xml，manifest.json。设备 PKG110 / Android 16 / 拼多多 8.17.0 / Wi-Fi，goods_id 937122477375。项目所有者人工确认截图为目标商品详情页，且“快要抢光 12.88”与“免拼购买”的位置和手机当前画面一致。只读结构核对显示：精确“快要抢光”文本节点本身不可点击，但位于唯一、可见、启用、可点击的 PDD ViewGroup 祖先内；“免拼购买”属于另一底部可点击容器。后续只允许把入口判据收紧调整为“精确唯一快要抢光子节点 + 证据绑定唯一可点击祖先”，不得允许或点击“免拼购买”，必须先补 fixture/反例/超时不重试测试再真机运行。
+
+### 2026-08-04T10:47:38Z · ila
+
+2026-08-04 T-103 入口父链收紧完成：提交 44586fe，合入主分支 ce9d6ca。入口只允许精确唯一“快要抢光”子节点及五层已取证可点击 PDD 祖先链，实际点击仍取子节点中心；“免拼购买”独立容器为硬拒绝。独立审计复算证据 hash、验证 focused 32 / full 112 tests 和完整 init 均 PASS。最终真机运行曾安全失败且未发布输出目录；随后只读诊断发现系统通知栏覆盖，app_current 虽仍报告 PDD，但节点树 64 个节点全部属于 SystemUI。下一次必须由人先完全收起通知栏并停在 PDD 首页再运行；T-103 保持 DOING，不把这次失败或入口截图确认误记为完整验收。
 <!-- END VIKUNJA EXPORT -->
 
 ## 边界

docs/tasks/T-104.md

@@ -0,0 +1,82 @@
+---
+id: T-104
+title: 验证规格选择能力安全退出
+phase: 1
+deps: [T-103]
+status: TODO
+created: 2026-08-04
+vikunja_task_id: 32
+context_ref: 9a4d11f
+work_branch: task/t-104-safe-exit
+needs_device: true
+needs_human_review: true
+write_paths:
+  - docs/tasks/T-104.md
+  - client/src/cmbuyer_client/pdd/sku_selection.py
+  - client/src/cmbuyer_client/pdd/sku_selection_runner.py
+  - client/tests/pdd/test_sku_selection.py
+  - client/tests/pdd/fixtures/product_exit_8_17_0.xml
+  - client/scripts/capture_sku_exit_spike.py
+  - client/scripts/run_t103_sku_selection.py
+  - docs/current-state.md
+---
+
+<!-- BEGIN VIKUNJA EXPORT id=32 synced=2026-08-04T10:47:05Z sha256=d0fe6a30708d16440962b8ab9ec81ce33065f96363597bd55e0a1fb4b442b6fd -->
+## 问题 / 背景
+
+T-103 已实现受控规格选择、读价、截图和单次 Back，但当前 `exit_sku_panel_safely()` 在 hierarchy 变化且规格面板判据不再成立时就返回成功。2026-08-04 的只读诊断已证明：通知栏覆盖时 `app_current` 仍可能报告拼多多，而节点树全部属于 SystemUI。因此“已不在面板”不是安全退出的充分条件。T-104 用独立真机证据把后置条件收紧为“稳定回到同一目标商品详情页”。
+
+## 关联需求与交互
+
+- 功能：F-006 单趟流程中的安全退出边界。
+- 架构/API：复用 `SkuSelectionFlow.exit_sku_panel_safely()`；不改变服务端 API。
+- 依赖：T-103 完成真机规格选择/读价/人眼安全退出验收后才开工。
+
+## 方案
+
+1. 先新增窄取证脚本：只在已验证 PDD 8.17.0 目标规格面板上发送一次 Back，随后采集本机截图/XML/App/设备/goods_id 元数据；不得打开确认页、设置数量或点击任何页面控件。
+2. 由人确认 post-exit 截图确为 goods_id `937122477375` 商品详情、规格面板已关闭、未进入确认页/外部支付页，并确认截图/XML 对应；`needs_device=true`，agent 保持 DOING。
+3. 从本项目 post-exit XML 提取只含判据节点的最小 fixture。成功判据必须稳定命中该商品详情的证据绑定正结构；可复用 T-103 的“快要抢光”五层入口链，但必须由新的 post-exit 证据再次验证。
+4. 收紧 `exit_sku_panel_safely()`：版本和前台包正确、节点树仅含 PDD 受控页面、目标商品详情正判据精确唯一并连续稳定；面板仍在、SystemUI/锁屏、PDD 其他页、确认/提交/支付语义、入口缺失/重复或结构漂移均不得记成功。
+5. Back 最多发送一次。超时或响应不明不重试；故障调和也不能产生第二次 Back。失败不得发布 `safe_exit=completed` manifest。
+6. 复用 T-103 runner/flow，不新建第二套采购流程；可增加 post-exit 本机原始截图和无页面正文摘要，完整 XML 仍只留本机，不上传、Git 或日志。
+
+## 验收要点
+
+- 真机证据记录路径、截图/XML hash、PKG110、Android 16、Wi-Fi/USB、拼多多 8.17.0 和 goods_id；只有人能完成页面对应性与无订单创建确认。
+- 离线测试覆盖成功稳定详情页，以及面板未退、SystemUI/锁屏、PDD 其他页、确认/提交/支付页、入口缺失/重复/漂移、版本/前台漂移。
+- 所有失败分支 Back 总数最多 1；超时不重试；失败不发布 completed manifest。
+- 静态证明没有数量、确认页导航、提交订单、围栏或支付能力。
+- client 全量单测、compileall、完整 init、上下文校验与 diff-check 通过。
+
+## 执行记录
+
+等待 T-103 完成后认领。
+
+## 执行记录
+
+（暂无）
+<!-- END VIKUNJA EXPORT -->
+
+## 边界
+
+- T-104 必须等待 T-103 完成人工真机验收后才可转 `DOING`。本任务先取证、后写退出成功判据；不得从
+  当前实现、前序项目、旧 XML、Activity 名或推理直接声明商品详情页。
+- 取证动作只允许在已证明为拼多多 8.17.0、goods_id `937122477375` 的目标规格面板上发送一次
+  Android Back，随后只读采集截图/XML/App/设备摘要。不得点击关闭坐标、空白处、购买、数量、确认、
+  提交或支付控件，也不得在 Back 失败或结果不明时重试。
+- “规格面板已消失”不是退出成功。成功必须由本项目 post-exit 真机证据证明：版本和前台包正确、节点树
+  仅属于 PDD 受控页面、同一目标商品详情正判据精确唯一并连续稳定；SystemUI/锁屏、PDD 其他页面、
+  确认页、提交页、外部支付页、入口缺失/重复或结构漂移一律停止且不得发布 completed manifest。
+- 可以复用 T-103 已取证的“快要抢光”五层入口链作为候选结构，但必须在新的 post-exit 截图/XML 上
+  重新验证并由人确认页面对应性；不得使用相似文本、包含/前缀、OCR、裸坐标或其他购买文案兜底。
+- 只收紧现有 `SkuSelectionFlow` 与 runner 的退出后置条件，不新建第二套采购流程，不新增通用 `click`、
+  数量、确认页导航、授权、提交围栏、创建待付款订单或支付能力。第一趟与任何真实提交函数继续静态隔离。
+- 原始完整 XML 只保存在 `%LOCALAPPDATA%\cmbuyer\artifacts\T-104\...`，不得上传、提交 Git、写入
+  Vikunja 或日志。Git 中最小 fixture 只能保留退出判据所需结构，并由人确认不含地址、手机号或支付凭据。
+- `needs_device: true`：agent 不得自行标 `DONE`。只有人确认 post-exit 截图为同一目标商品详情、面板
+  已关闭、未进入确认/提交/支付页、未创建订单，且截图/XML 对应后，任务才可完成。
+- 本任务不实现、不调用 `set_quantity_and_readback()`、`go_to_order_confirm()`、
+  `create_submission_fence()`、`submit_order_once()`、支付、免密支付、先用后付或任何扣款能力；既有
+  三道价格闸门、服务端围栏、只点一次且不重试的约束不得放宽。
+

docs/tasks/T-203.md

@@ -3,7 +3,7 @@ id: T-203
 title: 表格查询与批量开始采购授权
 phase: 2
 deps: [T-202, T-209]
-status: DOING
+status: DONE
 created: 2026-08-04
 vikunja_task_id: 30
 context_ref: 1f20271
@@ -21,7 +21,7 @@ write_paths:
   - admin/README.md
 ---
 
-<!-- BEGIN VIKUNJA EXPORT id=30 synced=2026-08-04T09:16:48Z sha256=6c62ba368a760e96b8feb06f3f3ced8a2a40d10f7754a99471f7429c3862a80c -->
+<!-- BEGIN VIKUNJA EXPORT id=30 synced=2026-08-04T10:27:15Z sha256=540fb01c9da32148898851f6450b713103f1db40222e1ac0d6dbcc74b007bcb3 -->
 ## 问题 / 背景
 
 T-202 已完成手工 DRAFT 建单；T-209 将生产 schema/领域状态机迁移到单趟模型。项目所有者明确：管理员点击“开始采购（只创建待付款订单）”本身就是授权，不再增加试选后确认。T-203 负责采购服务查询和批量授权事务，使设备后续只能领取显式授权的 PENDING 任务。
@@ -54,6 +54,14 @@ F-004、F-008、F-018；US-003、US-005；IX-005；GET /tasks、POST /tasks/star
 ### 2026-08-04T09:16:39Z · ila
 
 2026-08-04 开始 T-203：依赖 T-209 已完成并合入 main。主 agent 已完成开工前只读审计，冻结 v2 schema 启动校验、单进程 writeGate、start_key 规范集合重放、Asia/Shanghai 到 UTC 半开区间、julianday 查询及 big.Int 分金额边界；本地状态转 DOING，分支 task/t-203-start-purchases。
+
+### 2026-08-04T10:18:10Z · ila
+
+2026-08-04 T-203 独立终审退回两项：未认证须按 docs/api.md 返回 401，已认证但 CSRF 错误才返回 403；原始 JSON 在 64KiB 边界内必须先做严格 UTF-8 校验。现有 auth.Manager 无法只读区分“有效管理会话 + 错 CSRF”与“无会话”，因此任务所有者批准把 write_paths 最小扩展为 admin/internal/auth/**，仅允许增加只读认证状态 API 及测试；不得创建/旋转会话或放宽 CSRF。修复、复审和完整门禁通过前 T-203 保持 DOING。
+
+### 2026-08-04T10:26:57Z · ila
+
+2026-08-04 T-203 完成：提交 5dcff4b，合入主分支 03a067e。实现传统任务表格、筛选、DRAFT 批量勾选和“开始采购（只创建待付款订单）”授权；1/100 条在单一事务内生成 ACTIVE 快照并原子转 PENDING，相同 start_key 原集合稳定重放。独立终审两轮后修复未认证 401/CSRF 403 与原始 UTF-8 严格校验。主 agent 合入后完整 init（client 112 tests）、Go 全量/竞态/vet/build、Node 7 tests、上下文和 diff-check 全部通过；未实现领取、真机、提交订单或付款。
 <!-- END VIKUNJA EXPORT -->
 
 ## 边界

docs/tasks/T-204.md

@@ -0,0 +1,101 @@
+---
+id: T-204
+title: 路由化任务详情与内部截图资产
+phase: 2
+deps: [T-203]
+status: DOING
+created: 2026-08-04
+vikunja_task_id: 31
+context_ref: 7928457
+work_branch: task/t-204-details-evidence
+needs_device: false
+needs_human_review: false
+write_paths:
+  - docs/tasks/T-204.md
+  - admin/migrations/00003_evidence_assets.sql
+  - admin/internal/migrations/migrations_test.go
+  - admin/internal/evidence/**
+  - admin/internal/storage/evidence/**
+  - admin/internal/taskdetail/**
+  - admin/internal/config/**
+  - admin/internal/server/**
+  - admin/internal/transport/webui/**
+  - admin/cmd/server/**
+  - admin/README.md
+  - docs/api.md
+  - docs/routes.md
+  - docs/04-architecture.md
+---
+
+<!-- BEGIN VIKUNJA EXPORT id=31 synced=2026-08-04T10:42:09Z sha256=2bd6675f2249b461d72b5e0a066da90e33af8023c4b97b52f0d77a8eb391c034 -->
+## 问题 / 背景
+
+T-203 已完成传统任务表格与批量“开始采购”授权。采购管理员还缺少可复制、可返回的任务详情，以及只供内部审计的规格面板/确认页原始截图。T-204 提供详情与截图资产底座；不实现 T-205 的 attempt/event 写入，不实现 T-306 的客户端上传调用。
+
+## 关联需求与交互
+
+- 功能：F-007、F-011。
+- 用户故事：US-002、US-004、US-007。
+- 交互：IX-003、IX-004；双击非控件区域或 Enter 打开同一 `/tasks/{id}`，直达显示完整页，列表增强为抽屉。
+- 架构/API：`GET /tasks/{id}`、`POST /api/v1/tasks/{id}/evidence`、`GET /evidence/{asset_id}`。
+
+## 方案
+
+1. 新增任务详情只读查询，返回任务要求、状态、版本、授权摘要和当前已有的 attempt/submission/evidence 摘要；缺少后续事实时显示明确空态，不伪造数据。
+2. `GET /tasks/{id}` 以同一数据模型渲染完整页或同源 HTML fragment。列表行双击非交互区域、键盘 Enter 打开抽屉并把 URL 推到同一路径；Esc、关闭或浏览器返回恢复筛选、滚动与触发行焦点。外部商品链接、复选框和批量按钮不得触发行详情。
+3. 详情是审计视图，不提供试选后确认、再次提交、付款自动化或围栏后重试。状态文案始终区分“订单已创建，系统尚未付款”。
+4. 新建 `evidence_assets` 迁移和存储边界：只接收明确的单个 PNG screenshot；attempt_id 必填且必须属于 URL 中的 task。元数据仅允许 upload_key、attempt_id、kind、privacy_tier=INTERNAL_RAW、sha256、captured_at；记录 MIME、字节数、宽高、存储键和上传主体。
+5. 上传先经注入的 DeviceAuthenticator；T-301 前生产默认拒绝，测试可用 fake principal 验证契约，不创建临时 token 或共享密钥。T-301 完成后再接真实 Bearer 身份。
+6. multipart 设严格总大小、字段/单文件限制；拒绝未知字段、重复字段、XML/manifest/目录/本机路径、非 PNG、魔数/DecodeConfig/尺寸/hash 不一致以及未批准的截图 kind。服务端文件名只由资产 id/存储键生成，不使用上传名。
+7. 文件同目录临时写、流式 SHA-256、校验、fsync、原子 rename 后才写 DB；故障只允许产生不可达孤儿，不允许 DB 指向缺失或半文件。upload_key 同任务同载荷幂等，载荷变化冲突，并发只生成一条可读资产。
+8. `GET /evidence/{asset_id}` 仅有效管理员会话可读，返回 `image/png`、`Cache-Control: no-store`、`X-Content-Type-Options: nosniff`；未登录不泄露资产是否存在，不经静态目录暴露。
+9. UI 采用现有采购服务高密度表格风格：可见关闭按钮、焦点圈、44px 交互目标、截图 width/height 预留和 lazy loading、响应式布局、reduced-motion；不得只有双击一种入口。
+10. T-204 明确定值单文件最大 10 MiB、解码后最大 16,777,216 像素；证据目录由显式配置提供。保留期限仍由后续部署任务定值，不在本任务自动删除资产。
+
+## 验收要点
+
+- 详情完整页与抽屉使用同一路由/数据；刷新/复制 URL、浏览器返回、Esc、焦点和滚动/筛选恢复均有测试。
+- 不存在、畸形 id、未登录、fragment 伪造请求均 fail closed，不泄露内部错误。
+- 上传覆盖认证矩阵、multipart 负例、PNG 魔数/解码/大小/尺寸/hash、未知/重复字段、任务/attempt 归属、路径穿越、幂等冲突/并发及原子故障注入。
+- 截图读取覆盖管理员会话、未登录、缺失资产和 no-store/nosniff；完整 XML、支付页和凭据不会进入接口或 Git。
+- 运行 Go 全量/竞态/vet/build、Node 测试与语法检查、完整 init、上下文校验和 diff-check。
+
+## 执行记录
+
+待认领。
+
+## 执行记录
+
+### 2026-08-04T10:42:01Z · ila
+
+2026-08-04 开始 T-204：依赖 T-203 已完成，任务定义提交 7928457。采用生产默认拒绝的 DeviceAuthenticator 接口，T-301 前不发明临时设备凭据；先实现同 URL 详情抽屉/完整页、INTERNAL_RAW PNG 原子存储与管理员 no-store 读取。工作分支 task/t-204-details-evidence。
+<!-- END VIKUNJA EXPORT -->
+
+## 边界
+
+- 本任务只实现管理员任务详情、内部 PNG 证据资产底座和受保护读取；不实现 T-205 的 attempt/event
+  写接口、闸门计算、失败分类、提交围栏或结果调和，不实现 T-306 的客户端截图与上传调用。
+- T-301 完成前不得发明临时设备 token、共享密钥或把管理员会话当设备身份。上传 handler 必须先调用窄
+  `DeviceAuthenticator`；生产默认实现一律拒绝，只有测试可以注入 fake principal。
+- 上传只接受一个调用方显式选择的 PNG；不得枚举目录，不接收 XML、manifest、本机路径、原文件名、
+  任意备注、Cookie、token、支付凭据或未批准的截图 kind。MVP 本任务只开放
+  `SKU_PANEL_GATE_1`，后续 kind 必须在对应任务中收紧扩展。
+- `attempt_id` 必填且必须以复合外键证明属于 URL 中的 task。`privacy_tier` 只能是
+  `INTERNAL_RAW`；截图可含页面已显示的地址/手机号，但服务端不得 OCR、提取、索引、搜索或写日志。
+- 单文件最多 10 MiB，单边最多 8192 px，解码后最多 16,777,216 像素；必须同时校验 multipart 类型、
+  PNG 魔数、解码结果、字节数、尺寸与调用方声明的 64 位小写 SHA-256，任一不符零发布。
+- 文件必须在显式配置的证据根目录内以服务端生成的内容地址落盘，不进入公开静态目录。先同目录临时写、
+  流式 hash、`fsync`、原子 rename，再写数据库；故障最多留下不可达孤儿，不得留下指向半文件或缺失
+  文件的可见数据库记录，也不得为清理孤儿而删除并发复用文件。
+- 同一设备主体与 `upload_key` 的相同规范请求只返回原资产；内容、任务、attempt 或元数据变化一律
+  `409`。不得把 SHA-256 当业务记录唯一键，因为不同合法证据可以复用相同物理内容。
+- `GET /evidence/{asset_id}` 只允许有效管理员会话，必须返回 `Cache-Control: no-store` 和
+  `X-Content-Type-Options: nosniff`；匿名请求先拒绝，不泄露资产是否存在，不提供公开 URL、目录浏览、
+  批量导出、删除或预签名链接。
+- `/tasks/{id}` 的抽屉与完整页必须共享同一数据和详情模板。双击不是唯一入口；Enter、可见关闭按钮、
+  Esc、浏览器前进/后退、焦点与列表滚动/筛选恢复均可用，外部商品链接、复选框、输入框和按钮不得误触
+  行详情。截图必须预留尺寸、响应式缩放、延迟加载，alt 不转录地址或手机号。
+- 详情是只读审计视图，不制造不存在的 attempt/event/闸门/提交事实，不出现“机器选对了吗”、围栏后
+  重试、再次提交或自动付款动作。截图不作为价格闸门通过或人工审批的唯一依据。
+- 本任务不实现、不调用通用真机点击、`submit_order_once()`、支付、免密支付、先用后付或任何扣款能力；
+  既有三道价格闸门、服务端提交围栏、唯一点击一次且不重试的规则不得放宽。