feat(client): productionize proven sku reveal flow

client/src/cmbuyer_client/pdd/sku_reveal_spike.py

@@ -1,4 +1,4 @@
-"""T-103 尺码显示动作的一次性真机取证；不属于生产采购 Flow。"""
+"""T-103 尺码显示动作的一次性真机证据采集。"""
 
 from __future__ import annotations
 
@@ -20,32 +20,10 @@ from ..device.baseline import PDD_PACKAGE, _save_base64_screenshot, _sha256_file
 from .product_url import parse_product_url
 from .sku_selection import (
     EXPECTED_GOODS_ID,
-    _COLOR_ONLY_SUMMARY,
-    _PANEL_SURFACE,
-    _PanelProfile,
-    _REVEAL_NOT_PROVEN,
-    _S_SIZE_UI,
-    _TARGET_COLOR_UI,
-    _TARGET_SIZE_UI,
-    _action_bounds,
-    _clickable_before,
-    _descendant,
-    _exact_color_action,
-    _exact_inert,
-    _exact_live_layout,
-    _exact_readonly_text,
-    _exact_recycler,
-    _is_size_action_text,
-    _one,
     _parse_nodes,
     _require_color_only_panel,
-    _require_color_action_chain,
+    _require_safe_reveal_path,
-    _require_color_subtree,
+    _revealed_unselected_projection,
-    _require_exact_selected_set,
-    _require_panel_chain,
-    _require_size_wrapper,
-    _require_size_action_chain,
-    _spec_for,
     resolve_task_selection,
     SkuSelectionError,
     SkuSelectionFlow,
@@ -63,9 +41,6 @@ from .sku_selection_runner import (
 
 
 _TARGET_URL = f"https://mobile.yangkeduo.com/goods.html?goods_id={EXPECTED_GOODS_ID}"
-_REVEAL_START = (360, 1900)
-_REVEAL_END = (360, 1300)
-_REVEAL_STEPS = 30
 _REVEAL_FAILURE_STAGES = frozenset(
     (
         "reveal_precondition",
@@ -114,30 +89,6 @@ class SkuRevealSpikeResult:
     manifest_path: Path
 
 
-class _RevealEvidenceAdapter(UiautomatorSkuPanelAdapter):
-    """只为 spike 增加一个无参数、固定 profile 的一次性手势。"""
-
-    def __init__(self, device: Any, timeout_seconds: float) -> None:
-        super().__init__(device, timeout_seconds)
-        self._reveal_attempted = False
-
-    @property
-    def reveal_attempted(self) -> bool:
-        return self._reveal_attempted
-
-    def reveal_size_options_once(self) -> None:
-        if self._reveal_attempted:
-            raise SkuRevealSpikeError("规格显示动作已经尝试过，拒绝重试。")
-        # RPC 超时也可能表示手势已经送达，必须在调用前封存唯一机会。
-        self._reveal_attempted = True
-        self._call(
-            "jsonrpc_call",
-            "swipe",
-            [*_REVEAL_START, *_REVEAL_END, _REVEAL_STEPS],
-            timeout=self._timeout_seconds,
-        )
-
-
 class SkuRevealSpikeCapturer:
     """打开已取证面板、选一次目标颜色，再采集唯一 reveal 的前后证据。"""
 
@@ -165,7 +116,7 @@ class SkuRevealSpikeCapturer:
     ) -> SkuRevealSpikeResult:
         stage = "reveal_precondition"
         staging: Path | None = None
-        adapter: _RevealEvidenceAdapter | None = None
+        adapter: UiautomatorSkuPanelAdapter | None = None
         try:
             if type(goods_id) is not str or goods_id != EXPECTED_GOODS_ID:
                 raise SkuRevealSpikeError("商品不是 T-103 已取证目标，已停止取证。")
@@ -176,7 +127,7 @@ class SkuRevealSpikeCapturer:
             deadline = self._clock() + self._timeout_seconds
             inspection = self._adb_client.inspect(serial)
             _require_expected_device(inspection)
-            adapter = _RevealEvidenceAdapter(self._connector(serial), self._timeout_seconds)
+            adapter = UiautomatorSkuPanelAdapter(self._connector(serial), self._timeout_seconds)
             _require_expected_version(adapter.app_info(PDD_PACKAGE))
             if adapter.display_size() != EXPECTED_SCREEN_SIZE:
                 raise SkuRevealSpikeError("设备不是已取证的竖屏坐标空间，已停止取证。")
@@ -193,13 +144,10 @@ class SkuRevealSpikeCapturer:
                 sleep_function=self._sleep,
             )
             flow.open_sku_panel(link.canonical_url, pre_intent)
-            try:
+            # spike 只复用生产 Flow 的私有前置准备，不调用完整选择流程，避免重复 reveal/M。
-                flow.select_sku_options(resolve_task_selection("黑色CHA(纯棉)", "M(建议100-115)"))
+            flow._prepare_reveal_precondition(
-            except SkuSelectionError as error:
+                resolve_task_selection("黑色CHA(纯棉)", "M(建议100-115)")
-                if error.args != (_REVEAL_NOT_PROVEN,):
+            )
-                    raise
-            else:
-                raise SkuRevealSpikeError("规格流程未停在已取证的仅颜色状态。")
 
             before_hierarchy = adapter.dump_window_hierarchy()
             before_nodes = _parse_nodes(before_hierarchy)
@@ -231,7 +179,7 @@ class SkuRevealSpikeCapturer:
             after_directory = staging / "after"
             _capture_frame(adapter, after_directory, after_hierarchy)
             reverified = adapter.dump_window_hierarchy()
-            if _candidate_projection(_parse_nodes(reverified)) != projection:
+            if _revealed_unselected_projection(_parse_nodes(reverified)) != projection:
                 raise SkuRevealSpikeError("截图后候选状态漂移，未发布证据。")
             (after_directory / "hierarchy.xml").write_text(reverified, encoding="utf-8")
 
@@ -265,7 +213,7 @@ class SkuRevealSpikeCapturer:
 
     def _wait_for_candidate(
         self,
-        adapter: _RevealEvidenceAdapter,
+        adapter: UiautomatorSkuPanelAdapter,
         deadline: float,
     ) -> tuple[tuple[tuple[str, ...], ...], str]:
         stable: tuple[tuple[str, ...], ...] | None = None
@@ -276,7 +224,7 @@ class SkuRevealSpikeCapturer:
                 raise SkuRevealSpikeError("reveal 后拼多多不在前台，未发布证据。")
             hierarchy = adapter.dump_window_hierarchy()
             try:
-                projection = _candidate_projection(_parse_nodes(hierarchy))
+                projection = _revealed_unselected_projection(_parse_nodes(hierarchy))
             except SkuSelectionError:
                 projection = None
             if projection is not None and projection == stable:
@@ -288,315 +236,7 @@ class SkuRevealSpikeCapturer:
             self._sleep(min(0.2, remaining))
 
 
-def _require_safe_reveal_path(nodes: list[Any]) -> None:
+def _capture_frame(adapter: UiautomatorSkuPanelAdapter, directory: Path, hierarchy: str) -> None:
-    # 当前真机证据证明 x=360 是两列规格卡之间的空隙；必须验证完整线段，离散采样会漏掉窄浮层。
-    surface = _one(
-        [node for node in nodes if _exact_inert(node, "android.view.ViewGroup", _PANEL_SURFACE)],
-        "固定 reveal 通道无法绑定面板内容面。",
-    )
-    _require_panel_chain(surface)
-    content = surface.parent
-    action_root = content.parent if content is not None else None
-    action_parent = action_root.parent if action_root is not None else None
-    if (
-        content is None
-        or action_root is None
-        or action_parent is None
-        or not _exact_live_layout(action_root, "android.view.ViewGroup", "[0,366][1080,2328]")
-        or not _exact_live_layout(action_parent, "android.widget.LinearLayout", "[0,120][1080,2328]")
-    ):
-        raise SkuRevealSpikeError("固定 reveal 通道祖先身份漂移，未执行手势。")
-    allowed = {id(action_root.element), id(action_parent.element)}
-    occupants: set[int] = set()
-    for node in nodes:
-        if node.element.get("clickable") != "true":
-            continue
-        try:
-            left, top, right, bottom = _action_bounds(node.bounds)
-        except SkuSelectionError as error:
-            raise SkuRevealSpikeError("可点击节点坐标不可验证，未执行手势。") from error
-        if (
-            left <= _REVEAL_START[0] < right
-            and top <= _REVEAL_START[1]
-            and bottom > _REVEAL_END[1]
-        ):
-            occupants.add(id(node.element))
-    if occupants != allowed:
-        raise SkuRevealSpikeError("固定 reveal 通道被未取证可点击节点占用，未执行手势。")
-    if _REVEAL_START[1] >= 2079 or _REVEAL_END[1] >= 2079:
-        raise SkuRevealSpikeError("固定 reveal 通道越过规格内容区，未执行手势。")
-
-
-def _candidate_projection(nodes: list[Any]) -> tuple[tuple[str, ...], ...]:
-    surface = _one(
-        [node for node in nodes if _exact_inert(node, "android.view.ViewGroup", _PANEL_SURFACE)],
-        "候选面板内容面不唯一。",
-    )
-    _require_panel_chain(surface)
-    header = _one(
-        [node for node in nodes if node.parent is surface and _exact_inert(node, "android.widget.LinearLayout", "[0,366][1080,1000]")],
-        "候选面板头部不唯一。",
-    )
-    if len([child for child in header.element if child.tag == "node"]) != 4:
-        raise SkuSelectionError("候选面板头部子节点数量漂移。")
-    outer = _one(
-        [node for node in nodes if node.parent is surface and _exact_recycler(node, "[0,1000][1080,2079]")],
-        "候选维度容器不唯一。",
-    )
-    price_content, price_frame = _candidate_price_container(nodes, header)
-    price_nodes = _candidate_price_nodes(nodes, header, surface, price_frame)
-    summary = _one(
-        [
-            node
-            for node in nodes
-            if node.parent is price_content
-            and _exact_readonly_text(node, _COLOR_ONLY_SUMMARY, "[396,654][1053,716]")
-            and len(node.element) == 0
-        ],
-        "候选摘要不唯一。",
-    )
-    color_region = _one(
-        [node for node in nodes if _descendant(node, outer) and _exact_recycler(node, "[36,1000][1080,1483]")],
-        "候选颜色容器不唯一。",
-    )
-    color = _one(
-        [node for node in nodes if node.parent is color_region and _exact_color_action(node, "[372,1000][684,1024]", True)],
-        "候选目标颜色不唯一。",
-    )
-    rolled_spec = _spec_for(_PanelProfile.TARGETS_SELECTED)
-    selected_color_nodes = _require_color_subtree(color, rolled_spec)
-    _require_color_action_chain(color, color_region, outer, surface, rolled_spec)
-    size_label = _one(
-        [node for node in nodes if _descendant(node, outer) and _exact_readonly_text(node, "尺码", "[36,1506][114,1552]")],
-        "候选尺码标题不唯一。",
-    )
-    size_header = size_label.parent
-    if size_header is None or not _exact_live_layout(size_header, "android.widget.LinearLayout", "[36,1489][1044,1570]"):
-        raise SkuSelectionError("候选尺码标题父结构漂移。")
-    size_options = _one(
-        [node for node in nodes if _descendant(node, outer) and _exact_inert(node, "android.view.ViewGroup", "[36,1582][1044,1897]")],
-        "候选尺码 options 根不唯一。",
-    )
-    actions = [node for node in nodes if _descendant(node, size_options) and _is_size_action_text(node)]
-    s_action = _one(
-        [node for node in actions if node.text == _S_SIZE_UI and node.bounds == "[36,1582][409,1667]"],
-        "候选 S action 不唯一。",
-    )
-    m_action = _one(
-        [node for node in actions if node.text == _TARGET_SIZE_UI and node.bounds == "[439,1582][831,1667]"],
-        "候选 M action 不唯一。",
-    )
-    _require_size_wrapper(s_action, size_options)
-    _require_size_wrapper(m_action, size_options)
-    _require_size_action_chain(s_action, size_options, outer, surface)
-    _require_size_action_chain(m_action, size_options, outer, surface)
-    if any(node.element.get("selected") == "true" for node in actions):
-        raise SkuSelectionError("reveal 候选态已有尺码被选中。")
-    _require_exact_selected_set(nodes, surface, selected_color_nodes)
-    dangerous = [
-        node for node in nodes
-        if "提交订单" in node.text
-        and node.element.get("package") == PDD_PACKAGE
-        and node.element.get("class") == "android.widget.TextView"
-    ]
-    if len(dangerous) != 1 or _action_bounds(dangerous[0].bounds)[1] < 2079:
-        raise SkuSelectionError("提交硬拒绝区位置不唯一。")
-    return tuple(
-        (
-            node.element.get("class", ""),
-            node.bounds,
-            node.text,
-            node.desc,
-            node.element.get("selected", ""),
-            node.element.get("clickable", ""),
-        )
-        for node in (
-            surface,
-            header,
-            outer,
-            *price_nodes,
-            summary,
-            color_region,
-            color,
-            size_label,
-            size_options,
-            s_action,
-            m_action,
-            dangerous[0],
-        )
-    )
-
-
-def _candidate_price_nodes(
-    nodes: list[Any],
-    header: Any,
-    surface: Any,
-    price_frame: Any,
-) -> tuple[Any, Any, Any]:
-    matches: list[tuple[Any, Any, Any]] = []
-    for matcher in (_candidate_dual_price_nodes, _candidate_current_only_price_nodes):
-        try:
-            match = matcher(nodes, header, surface, price_frame)
-        except SkuSelectionError:
-            continue
-        matches.append(match)
-    if len(matches) != 1:
-        raise SkuSelectionError("候选价格布局不符合唯一完整取证变体。")
-    return matches[0]
-
-
-def _candidate_dual_price_nodes(
-    nodes: list[Any],
-    header: Any,
-    surface: Any,
-    price_frame: Any,
-) -> tuple[Any, Any, Any]:
-    price_row = _one(
-        [
-            node
-            for node in nodes
-            if node.parent is price_frame
-            and _exact_inert(node, "android.widget.LinearLayout", "[396,498][895,570]")
-        ],
-        "候选双价格行不唯一。",
-    )
-    if len([child for child in price_row.element if child.tag == "node"]) != 2:
-        raise SkuSelectionError("候选双价格行子节点数量漂移。")
-    current = _one(
-        [
-            node
-            for node in nodes
-            if node.parent is price_row
-            and _exact_readonly_text(node, "快卖完 ¥12.88", "[396,503][712,570]")
-            and len(node.element) == 0
-        ],
-        "候选双价格当前价不唯一。",
-    )
-    original = _one(
-        [
-            node
-            for node in nodes
-            if node.parent is price_row
-            and _exact_readonly_text(node, "¥29.88", "[730,503][895,570]")
-            and len(node.element) == 0
-        ],
-        "候选双价格原价不唯一。",
-    )
-    if _clickable_before(current, surface):
-        raise SkuSelectionError("候选双价格当前价位于可点击内容祖先下。")
-    if any(
-        _candidate_current_only_fragment(node)
-        for node in nodes
-        if _descendant(node, header)
-    ):
-        raise SkuSelectionError("候选双价格布局混入单价格残片。")
-    return price_row, current, original
-
-
-def _candidate_current_only_price_nodes(
-    nodes: list[Any],
-    header: Any,
-    surface: Any,
-    price_frame: Any,
-) -> tuple[Any, Any, Any]:
-    price_row = _one(
-        [
-            node
-            for node in nodes
-            if node.parent is price_frame
-            and _exact_inert(node, "android.widget.LinearLayout", "[396,498][693,570]")
-        ],
-        "候选单价格行父子关系漂移。",
-    )
-    if len([child for child in price_row.element if child.tag == "node"]) != 1:
-        raise SkuSelectionError("候选单价格行子节点数量漂移。")
-    current = _one(
-        [
-            node
-            for node in nodes
-            if node.parent is price_row
-            and _exact_readonly_text(node, "限1件 ¥12.88 ", "[396,503][675,570]")
-            and len(node.element) == 0
-        ],
-        "候选单价格当前价不唯一。",
-    )
-    if any(node.text in {"¥29.88", "券前¥29.88"} for node in nodes):
-        raise SkuSelectionError("候选单价格变体出现原价角色。")
-    if any(
-        _candidate_dual_fragment(node)
-        for node in nodes
-        if _descendant(node, header)
-    ):
-        raise SkuSelectionError("候选单价格布局混入双价格残片。")
-    if _clickable_before(current, surface):
-        raise SkuSelectionError("候选单价格当前价位于可点击内容祖先下。")
-    return price_frame, price_row, current
-
-
-def _candidate_price_container(nodes: list[Any], header: Any) -> tuple[Any, Any]:
-    content_frame = _one(
-        [
-            node
-            for node in nodes
-            if node.parent is header
-            and _exact_inert(node, "android.widget.FrameLayout", "[0,474][1080,863]")
-        ],
-        "候选价格内容外框不唯一。",
-    )
-    if len([child for child in content_frame.element if child.tag == "node"]) != 1:
-        raise SkuSelectionError("候选价格内容外框子节点数量漂移。")
-    relative = _one(
-        [
-            node
-            for node in nodes
-            if node.parent is content_frame
-            and _exact_inert(node, "android.widget.RelativeLayout", "[0,474][1080,863]")
-        ],
-        "候选价格 RelativeLayout 父子关系漂移。",
-    )
-    if len([child for child in relative.element if child.tag == "node"]) != 1:
-        raise SkuSelectionError("候选价格 RelativeLayout 子节点数量漂移。")
-    content = _one(
-        [
-            node
-            for node in nodes
-            if node.parent is relative
-            and _exact_inert(node, "android.view.ViewGroup", "[0,474][1080,863]")
-        ],
-        "候选价格内容 ViewGroup 父子关系漂移。",
-    )
-    if len([child for child in content.element if child.tag == "node"]) != 8:
-        raise SkuSelectionError("候选价格内容 ViewGroup 子节点数量漂移。")
-    price_frame = _one(
-        [
-            node
-            for node in nodes
-            if node.parent is content
-            and _exact_inert(node, "android.widget.FrameLayout", "[396,498][1053,570]")
-        ],
-        "候选价格 FrameLayout 父子关系漂移。",
-    )
-    if len([child for child in price_frame.element if child.tag == "node"]) != 1:
-        raise SkuSelectionError("候选价格 FrameLayout 子节点数量漂移。")
-    return content, price_frame
-
-
-def _candidate_current_only_fragment(node: Any) -> bool:
-    return (
-        _exact_inert(node, "android.widget.LinearLayout", "[396,498][693,570]")
-        or _exact_readonly_text(node, "限1件 ¥12.88 ", "[396,503][675,570]")
-    )
-
-
-def _candidate_dual_fragment(node: Any) -> bool:
-    return (
-        _exact_inert(node, "android.widget.LinearLayout", "[396,498][895,570]")
-        or _exact_readonly_text(node, "快卖完 ¥12.88", "[396,503][712,570]")
-        or _exact_readonly_text(node, "¥29.88", "[730,503][895,570]")
-    )
-
-
-def _capture_frame(adapter: _RevealEvidenceAdapter, directory: Path, hierarchy: str) -> None:
     directory.mkdir()
     hierarchy_path = directory / "hierarchy.xml"
     hierarchy_path.write_text(hierarchy, encoding="utf-8")

client/src/cmbuyer_client/pdd/sku_selection.py

@@ -18,6 +18,8 @@ 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()))
+_TARGET_COLOR_UNROLLED_BOUNDS = "[372,1188][684,1587]"
+_TARGET_SIZE_BOUNDS = "[439,1582][831,1667]"
 _ENTRY = "快要抢光 ¥ 12.88"
 _ENTRY_PROMOTION_LABEL = "快要抢光"
 _ENTRY_TEXT_BOUNDS = "[688,2184][1042,2253]"
@@ -37,10 +39,11 @@ _COLOR_ONLY_SUMMARY = "请选择： 尺码"
 _S_SIZE_UI = "S（建议80-100）"
 _S_SUMMARY = f"已选： {_TARGET_COLOR_UI} {_S_SIZE_UI}"
 _TARGET_SUMMARY = f"已选： {_TARGET_COLOR_UI} {_TARGET_SIZE_UI}"
-_REVEAL_NOT_PROVEN = "尺码仍在已取证视口外；受控显示动作尚未取证，已停止后续点击。"
 _BOUNDS = re.compile(r"^\[(\d+),(\d+)\]\[(\d+),(\d+)\]$")
 _ROLLED_CURRENT_PRICE = "快卖完 ¥12.88"
 _ROLLED_ORIGINAL_PRICE = "¥29.88"
+_REVEALED_CURRENT_PRICE = "限1件 ¥12.88 "
+_REVEAL_GESTURE = (360, 1900, 360, 1300, 30)
 _BAD_PRICE_ROLE = ("提交订单", "支付", "优惠", "券", "会员", "补贴", "区间", "实付", "到手", "原价", "划线价", "最低", "低至", "起价", "下单", "先用后付", "预估")
 
 
@@ -91,6 +94,7 @@ class SkuPanelDevice(Protocol):
     def dump_window_hierarchy(self) -> str: ...
     def tap_sku_entry(self, bounds: str) -> None: ...
     def tap_sku_option(self, bounds: str) -> None: ...
+    def reveal_size_options_once(self) -> None: ...
     def leave_sku_panel(self) -> None: ...
 
 
@@ -122,6 +126,7 @@ class _Node:
 class _PanelProfile(Enum):
     PANEL_OPEN_EMPTY = "panel_open_empty"
     COLOR_SELECTED_SIZE_HIDDEN = "color_selected_size_hidden"
+    SIZE_VISIBLE_UNSELECTED = "size_visible_unselected"
     SIZE_VISIBLE_NON_TARGET = "size_visible_non_target"
     TARGETS_SELECTED = "targets_selected"
 
@@ -172,7 +177,7 @@ _PANEL_SPECS = (
             "券前¥29.88", "[693,580][912,647]",
         ),
         _EMPTY_SUMMARY, "[396,731][1053,793]", "[36,1106][192,1159]",
-        "[36,1188][1080,2046]", "[372,1188][684,1587]", False,
+        "[36,1188][1080,2046]", _TARGET_COLOR_UNROLLED_BOUNDS, False,
         "[36,2069][114,2079]", None,
     ),
     _PanelSpec(
@@ -183,7 +188,7 @@ _PANEL_SPECS = (
             "券前¥29.88", "[693,580][912,647]",
         ),
         _COLOR_ONLY_SUMMARY, "[396,731][1053,793]", "[36,1106][192,1159]",
-        "[36,1188][1080,2046]", "[372,1188][684,1587]", True,
+        "[36,1188][1080,2046]", _TARGET_COLOR_UNROLLED_BOUNDS, True,
         "[36,2069][114,2079]", None,
     ),
     _PanelSpec(
@@ -194,7 +199,7 @@ _PANEL_SPECS = (
             "限1件 ¥12.88 ", "[396,580][675,647]",
         ),
         _EMPTY_SUMMARY, "[396,731][1053,793]", "[36,1106][192,1159]",
-        "[36,1188][1080,2046]", "[372,1188][684,1587]", False,
+        "[36,1188][1080,2046]", _TARGET_COLOR_UNROLLED_BOUNDS, False,
         "[36,2069][114,2079]", None,
     ),
     _PanelSpec(
@@ -205,7 +210,7 @@ _PANEL_SPECS = (
             "限1件 ¥12.88 ", "[396,580][675,647]",
         ),
         _COLOR_ONLY_SUMMARY, "[396,731][1053,793]", "[36,1106][192,1159]",
-        "[36,1188][1080,2046]", "[372,1188][684,1587]", True,
+        "[36,1188][1080,2046]", _TARGET_COLOR_UNROLLED_BOUNDS, True,
         "[36,2069][114,2079]", None,
     ),
     _PanelSpec(
@@ -233,6 +238,19 @@ _PANEL_SPECS = (
 )
 
 
+_REVEALED_UNSELECTED_SPEC = _PanelSpec(
+    _PanelProfile.SIZE_VISIBLE_UNSELECTED,
+    "[0,366][1080,1000]", "[0,1000][1080,2079]",
+    _CurrentOnlyPriceLayout(
+        "[396,498][1053,570]", "[396,498][693,570]",
+        _REVEALED_CURRENT_PRICE, "[396,503][675,570]",
+    ),
+    _COLOR_ONLY_SUMMARY, "[396,654][1053,716]", None,
+    "[36,1000][1080,1483]", "[372,1000][684,1024]", True,
+    "[36,1506][114,1552]", None,
+)
+
+
 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,
@@ -241,9 +259,12 @@ class SkuSelectionFlow:
             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
+        self._pending: tuple[str, Callable[[list[_Node]], Any], bool] | None = None
+        self._terminal = False
+        self._color_selected_by_flow = False
 
     def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None:
+        self._require_active()
         try:
             if parse_product_url(product_url).goods_id != EXPECTED_GOODS_ID:
                 raise SkuSelectionError("商品不是已取证目标，已停止操作。")
@@ -265,90 +286,175 @@ class SkuSelectionFlow:
             raise
 
         try:
-            self._pending = (before, _require_empty_panel)
+            self._pending = (before, _require_empty_panel, False)
             self._device.tap_sku_entry(entry.bounds)
         except BaseException as error:
+            self._terminal = True
             _annotate_sku_entry_failure(error, "sku_entry_click")
             raise
 
         try:
             self._wait_after_action(before, _require_empty_panel)
         except BaseException as error:
+            self._terminal = True
             _annotate_sku_entry_failure(error, "sku_entry_panel_verify")
             raise
 
     def select_sku_options(self, selection: SkuSelection) -> None:
+        self._require_active()
+        if selection != SkuSelection(_TARGET_COLOR_UI, _TARGET_SIZE_UI):
+            raise SkuSelectionError("规格 UI 文案不是获准目标，已停止操作。")
+        nodes = self._verified_nodes()
+        initial_profile = _classify_panel(nodes).profile
+        if initial_profile is _PanelProfile.TARGETS_SELECTED:
+            return
+        if initial_profile is _PanelProfile.SIZE_VISIBLE_NON_TARGET:
+            self._require_foreground()
+            before = self._read_hierarchy()
+            nodes = _parse_nodes(before)
+            target = _target_size_action(nodes, selected=False)
+            _action_bounds(target.bounds)
+            _require_action_occupants(nodes, target)
+            self._perform_mutation(
+                before,
+                _require_target_panel,
+                partial(self._device.tap_sku_option, target.bounds),
+            )
+            return
+
+        self._prepare_reveal_precondition(selection)
+        self._require_foreground()
+        before = self._read_hierarchy()
+        nodes = _parse_nodes(before)
+        spec = _require_profile(nodes, _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN)
+        if not isinstance(spec.price_layout, _CurrentOnlyPriceLayout):
+            # 真机动作证据只覆盖 current-only；dual 仅颜色态必须在 reveal 前零动作停止。
+            raise SkuSelectionError("规格显示前置不是已取证单价格布局，已停止操作。")
+        _require_safe_reveal_path(nodes)
+        self._perform_mutation(
+            before,
+            _revealed_unselected_projection,
+            self._device.reveal_size_options_once,
+            require_stable=True,
+        )
+
+        self._require_foreground()
+        before = self._read_hierarchy()
+        nodes = _parse_nodes(before)
+        _revealed_unselected_projection(nodes)
+        target = _target_size_action(nodes, selected=False)
+        _action_bounds(target.bounds)
+        _require_action_occupants(nodes, target)
+        self._perform_mutation(
+            before,
+            _require_target_panel,
+            partial(self._device.tap_sku_option, target.bounds),
+        )
+
+    def _prepare_reveal_precondition(self, selection: SkuSelection) -> None:
+        """只把 current-only 初态推进到仅颜色态，供生产 Flow 与证据 spike 共用。"""
+
+        self._require_active()
         if selection != SkuSelection(_TARGET_COLOR_UI, _TARGET_SIZE_UI):
             raise SkuSelectionError("规格 UI 文案不是获准目标，已停止操作。")
         self._require_foreground()
         before = self._read_hierarchy()
         nodes = _parse_nodes(before)
         spec = _classify_panel(nodes)
-        profile = spec.profile
+        if spec.profile is _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN:
-
+            if not isinstance(spec.price_layout, _CurrentOnlyPriceLayout):
-        if profile is _PanelProfile.PANEL_OPEN_EMPTY:
+                raise SkuSelectionError("规格显示前置不是已取证单价格布局，已停止操作。")
-            target = _target_color_action(nodes, selected=False)
+            if not self._color_selected_by_flow:
-            _action_bounds(target.bounds)
+                raise SkuSelectionError("目标颜色不是本次 Flow 精确选中，未执行 reveal。")
-            _require_action_occupants(nodes, target)
-            color_postcondition = partial(
-                _require_profile_with_price_layout,
-                expected=_PanelProfile.COLOR_SELECTED_SIZE_HIDDEN,
-                expected_price_layout=spec.price_layout,
-            )
-            self._pending = (before, color_postcondition)
-            self._device.tap_sku_option(target.bounds)
-            self._wait_after_action(before, color_postcondition)
-            # 当前证据只证明颜色选择；尺码仍在视口外。没有动作证据时必须在此停住，
-            # 不能把一次通用 swipe 或下一次规格点击伪装成已验证流程。
-            raise SkuSelectionError(_REVEAL_NOT_PROVEN)
-        if profile is _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN:
-            raise SkuSelectionError(_REVEAL_NOT_PROVEN)
-        if profile is _PanelProfile.SIZE_VISIBLE_NON_TARGET:
-            target = _target_size_action(nodes, selected=False)
-            _action_bounds(target.bounds)
-            _require_action_occupants(nodes, target)
-            self._pending = (before, _require_target_panel)
-            self._device.tap_sku_option(target.bounds)
-            self._wait_after_action(before, _require_target_panel)
             return
-        if profile is _PanelProfile.TARGETS_SELECTED:
+        if spec.profile is not _PanelProfile.PANEL_OPEN_EMPTY or not isinstance(
-            return
+            spec.price_layout, _CurrentOnlyPriceLayout
-        raise SkuSelectionError("规格面板状态不属于已取证 profile，已停止操作。")
+        ):
+            raise SkuSelectionError("规格面板不属于获准的单价格初态，已停止操作。")
+        target = _target_color_action(nodes, selected=False)
+        _action_bounds(target.bounds)
+        _require_action_occupants(nodes, target)
+        color_postcondition = partial(
+            _require_profile_with_price_layout,
+            expected=_PanelProfile.COLOR_SELECTED_SIZE_HIDDEN,
+            expected_price_layout=spec.price_layout,
+        )
+        self._perform_mutation(
+            before,
+            color_postcondition,
+            partial(self._device.tap_sku_option, target.bounds),
+        )
+        self._color_selected_by_flow = True
 
     def read_sku_unit_price(self) -> str:
+        self._require_active()
         return _unit_price(self._verified_nodes())
 
     def verify_target_selection_and_read_price(self, selection: SkuSelection) -> str:
+        self._require_active()
         if selection != SkuSelection(_TARGET_COLOR_UI, _TARGET_SIZE_UI):
             raise SkuSelectionError("规格 UI 文案不是获准目标，已停止读取。")
         nodes = self._verified_nodes()
         return _unit_price(nodes)
 
     def exit_sku_panel_safely(self) -> None:
+        self._require_active()
         self._require_foreground()
         before = self._read_hierarchy()
-        _classify_panel(_parse_nodes(before))
+        # 截图重证与 Back 之间仍可能漂移；只有目标双规格和已验单价同时保持时才允许返回。
-        self._device.leave_sku_panel()
+        _unit_price(_parse_nodes(before))
+        try:
+            self._device.leave_sku_panel()
+        except BaseException:
+            self._terminal = True
+            raise
         deadline = self._clock() + self._entry_timeout
-        while True:
+        try:
-            self._require_foreground()
+            while True:
-            raw = self._read_hierarchy()
+                self._require_foreground()
-            if raw != before:
+                raw = self._read_hierarchy()
-                try:
+                if raw != before:
-                    _classify_panel(_parse_nodes(raw))
+                    try:
-                except SkuSelectionError:
+                        _classify_panel(_parse_nodes(raw))
-                    return
+                    except SkuSelectionError:
-            remaining = deadline - self._clock()
+                        self._terminal = True
-            if remaining <= 0:
+                        return
-                raise SkuSelectionError("安全退出后未确认离开规格面板，未重试返回。")
+                remaining = deadline - self._clock()
-            self._sleep(min(self._poll, remaining))
+                if remaining <= 0:
+                    raise SkuSelectionError("安全退出后未确认离开规格面板，未重试返回。")
+                self._sleep(min(self._poll, remaining))
+        except BaseException:
+            self._terminal = True
+            raise
 
     def reconcile_pending_action(self) -> None:
         """仅只读调和一次已发出但尚未得到后置条件确认的动作。"""
         if self._pending is None:
             return
-        before, condition = self._pending
+        before, condition, require_stable = self._pending
-        self._wait_after_action(before, condition)
+        self._wait_after_action(before, condition, require_stable=require_stable)
+
+    def _perform_mutation(
+        self,
+        before: str,
+        condition: Callable[[list[_Node]], Any],
+        mutation: Callable[[], None],
+        *,
+        require_stable: bool = False,
+    ) -> list[_Node]:
+        # 后置条件连同唯一动作机会必须在 RPC 前封存；超时可能表示已送达，不能重发。
+        self._pending = (before, condition, require_stable)
+        try:
+            mutation()
+            return self._wait_after_action(before, condition, require_stable=require_stable)
+        except BaseException:
+            # 结果不明时只允许 reconcile_pending_action 读回；即使调和成功，本 Flow 也不再继续。
+            self._terminal = True
+            raise
+
+    def _require_active(self) -> None:
+        if self._terminal:
+            raise SkuSelectionError("规格流程已进入不可重入终止态，已停止操作。")
 
     def _wait_for_entry(self, previous: str | None) -> tuple[_Node, str]:
         deadline, stable = self._clock() + self._entry_timeout, None
@@ -380,19 +486,31 @@ class SkuSelectionFlow:
                 raise SkuSelectionError("等待已取证规格入口超时，未执行点击。")
             self._sleep(min(self._poll, remaining))
 
-    def _wait_after_action(self, previous: str, condition: Callable[[list[_Node]], Any]) -> list[_Node]:
+    def _wait_after_action(
+        self,
+        previous: str,
+        condition: Callable[[list[_Node]], Any],
+        *,
+        require_stable: bool = False,
+    ) -> list[_Node]:
         deadline = self._clock() + self._entry_timeout
+        stable: Any = None
+        has_stable = False
         while True:
             self._require_foreground()
             raw = self._read_hierarchy()
             if raw != previous:
                 nodes = _parse_nodes(raw)
                 try:
-                    condition(nodes)
+                    projection = condition(nodes)
-                    self._pending = None
+                    if not require_stable or (has_stable and projection == stable):
-                    return nodes
+                        self._pending = None
+                        return nodes
+                    stable = projection
+                    has_stable = True
                 except SkuSelectionError:
-                    pass
+                    stable = None
+                    has_stable = False
             remaining = deadline - self._clock()
             if remaining <= 0:
                 raise SkuSelectionError("动作后页面未在限定时间内满足已取证后置条件，未重试动作。")
@@ -445,6 +563,12 @@ def _classify_panel(nodes: list[_Node]) -> _PanelSpec:
         except SkuSelectionError:
             continue
         matches.append(spec)
+    try:
+        _revealed_unselected_projection(nodes)
+    except SkuSelectionError:
+        pass
+    else:
+        matches.append(_REVEALED_UNSELECTED_SPEC)
     if len(matches) != 1:
         raise SkuSelectionError("规格面板不符合唯一完整取证 profile，已停止操作。")
     return matches[0]
@@ -574,6 +698,177 @@ def _match_panel_profile(nodes: list[_Node], spec: _PanelSpec) -> None:
     _require_exact_selected_set(nodes, surface, [*selected_nodes, expected_action])
 
 
+def _require_safe_reveal_path(nodes: list[_Node]) -> None:
+    """证明固定手势整条线段仍是已取证的两列卡片间安全通道。"""
+
+    surface = _one(
+        [node for node in nodes if _exact_inert(node, "android.view.ViewGroup", _PANEL_SURFACE)],
+        "固定 reveal 通道无法绑定面板内容面。",
+    )
+    _require_panel_chain(surface)
+    content = surface.parent
+    action_root = content.parent if content is not None else None
+    action_parent = action_root.parent if action_root is not None else None
+    if (
+        content is None
+        or action_root is None
+        or action_parent is None
+        or not _exact_live_layout(action_root, "android.view.ViewGroup", "[0,366][1080,2328]")
+        or not _exact_live_layout(action_parent, "android.widget.LinearLayout", "[0,120][1080,2328]")
+    ):
+        raise SkuSelectionError("固定 reveal 通道祖先身份漂移，未执行手势。")
+    start_x, start_y, end_x, end_y, _steps = _REVEAL_GESTURE
+    allowed = {id(action_root.element), id(action_parent.element)}
+    occupants: set[int] = set()
+    for node in nodes:
+        if node.element.get("clickable") != "true":
+            continue
+        try:
+            left, top, right, bottom = _action_bounds(node.bounds)
+        except SkuSelectionError as error:
+            raise SkuSelectionError("可点击节点坐标不可验证，未执行手势。") from error
+        if (
+            left <= start_x < right
+            and top <= start_y
+            and bottom > end_y
+        ):
+            occupants.add(id(node.element))
+    if occupants != allowed:
+        raise SkuSelectionError("固定 reveal 通道被未取证可点击节点占用，未执行手势。")
+    if start_y >= 2079 or end_y >= 2079:
+        raise SkuSelectionError("固定 reveal 通道越过规格内容区，未执行手势。")
+
+
+def _revealed_unselected_projection(nodes: list[_Node]) -> tuple[tuple[str, ...], ...]:
+    """唯一生产判据：current-only reveal 后，目标颜色已选且 S/M 均未选。"""
+
+    surface = _one(
+        [node for node in nodes if _exact_inert(node, "android.view.ViewGroup", _PANEL_SURFACE)],
+        "reveal 后面板内容面不唯一。",
+    )
+    _require_panel_chain(surface)
+    header = _one(
+        [node for node in nodes if node.parent is surface and _exact_inert(node, "android.widget.LinearLayout", "[0,366][1080,1000]")],
+        "reveal 后面板头部不唯一。",
+    )
+    if len([child for child in header.element if child.tag == "node"]) != 4:
+        raise SkuSelectionError("reveal 后面板头部子节点数量漂移。")
+    outer = _one(
+        [node for node in nodes if node.parent is surface and _exact_recycler(node, "[0,1000][1080,2079]")],
+        "reveal 后维度容器不唯一。",
+    )
+    content_frame = _one(
+        [node for node in nodes if node.parent is header and _exact_inert(node, "android.widget.FrameLayout", "[0,474][1080,863]")],
+        "reveal 后价格内容外框不唯一。",
+    )
+    if len([child for child in content_frame.element if child.tag == "node"]) != 1:
+        raise SkuSelectionError("reveal 后价格内容外框子节点数量漂移。")
+    relative = _one(
+        [node for node in nodes if node.parent is content_frame and _exact_inert(node, "android.widget.RelativeLayout", "[0,474][1080,863]")],
+        "reveal 后价格 RelativeLayout 父子关系漂移。",
+    )
+    if len([child for child in relative.element if child.tag == "node"]) != 1:
+        raise SkuSelectionError("reveal 后价格 RelativeLayout 子节点数量漂移。")
+    content = _one(
+        [node for node in nodes if node.parent is relative and _exact_inert(node, "android.view.ViewGroup", "[0,474][1080,863]")],
+        "reveal 后价格内容 ViewGroup 父子关系漂移。",
+    )
+    if len([child for child in content.element if child.tag == "node"]) != 8:
+        raise SkuSelectionError("reveal 后价格内容 ViewGroup 子节点数量漂移。")
+    price_frame = _one(
+        [node for node in nodes if node.parent is content and _exact_inert(node, "android.widget.FrameLayout", "[396,498][1053,570]")],
+        "reveal 后价格 FrameLayout 父子关系漂移。",
+    )
+    if len([child for child in price_frame.element if child.tag == "node"]) != 1:
+        raise SkuSelectionError("reveal 后价格 FrameLayout 子节点数量漂移。")
+    price_row = _one(
+        [node for node in nodes if node.parent is price_frame and _exact_inert(node, "android.widget.LinearLayout", "[396,498][693,570]")],
+        "reveal 后单价格行父子关系漂移。",
+    )
+    if len([child for child in price_row.element if child.tag == "node"]) != 1:
+        raise SkuSelectionError("reveal 后单价格行子节点数量漂移。")
+    current = _one(
+        [node for node in nodes if node.parent is price_row and _exact_readonly_text(node, _REVEALED_CURRENT_PRICE, "[396,503][675,570]") and len(node.element) == 0],
+        "reveal 后单价格当前价不唯一。",
+    )
+    if _clickable_before(current, surface):
+        raise SkuSelectionError("reveal 后单价格当前价位于可点击内容祖先下。")
+    if any(
+        node.text in {_ROLLED_CURRENT_PRICE, _ROLLED_ORIGINAL_PRICE, "券前¥29.88"}
+        or (node.bounds == "[396,498][895,570]" and node.element.get("class") == "android.widget.LinearLayout")
+        for node in nodes
+    ):
+        raise SkuSelectionError("reveal 后单价格布局混入双价格角色。")
+    summary = _one(
+        [node for node in nodes if node.parent is content and _exact_readonly_text(node, _COLOR_ONLY_SUMMARY, "[396,654][1053,716]") and len(node.element) == 0],
+        "reveal 后摘要不唯一。",
+    )
+    color_region = _one(
+        [node for node in nodes if _descendant(node, outer) and _exact_recycler(node, "[36,1000][1080,1483]")],
+        "reveal 后颜色容器不唯一。",
+    )
+    color = _one(
+        [node for node in nodes if node.parent is color_region and _exact_color_action(node, "[372,1000][684,1024]", True)],
+        "reveal 后目标颜色不唯一。",
+    )
+    selected_color_nodes = _require_color_subtree(color, _REVEALED_UNSELECTED_SPEC)
+    _require_color_action_chain(color, color_region, outer, surface, _REVEALED_UNSELECTED_SPEC)
+    size_label = _one(
+        [node for node in nodes if _descendant(node, outer) and _exact_readonly_text(node, _SIZE, "[36,1506][114,1552]")],
+        "reveal 后尺码标题不唯一。",
+    )
+    size_header = size_label.parent
+    if size_header is None or not _exact_live_layout(size_header, "android.widget.LinearLayout", "[36,1489][1044,1570]"):
+        raise SkuSelectionError("reveal 后尺码标题父结构漂移。")
+    size_options = _one(
+        [node for node in nodes if _descendant(node, outer) and _exact_inert(node, "android.view.ViewGroup", "[36,1582][1044,1897]")],
+        "reveal 后尺码 options 根不唯一。",
+    )
+    actions = [node for node in nodes if _descendant(node, size_options) and _is_size_action_text(node)]
+    s_action = _one(
+        [node for node in actions if node.text == _S_SIZE_UI and node.bounds == "[36,1582][409,1667]" and node.element.get("selected") == "false"],
+        "reveal 后 S action 不唯一。",
+    )
+    m_action = _one(
+        [node for node in actions if node.text == _TARGET_SIZE_UI and node.bounds == _TARGET_SIZE_BOUNDS and node.element.get("selected") == "false"],
+        "reveal 后 M action 不唯一。",
+    )
+    _require_size_wrapper(s_action, size_options)
+    _require_size_wrapper(m_action, size_options)
+    _require_size_action_chain(s_action, size_options, outer, surface)
+    _require_size_action_chain(m_action, size_options, outer, surface)
+    if any(node.element.get("selected") == "true" for node in actions):
+        raise SkuSelectionError("reveal 后已有尺码被选中。")
+    _require_exact_selected_set(nodes, surface, selected_color_nodes)
+    dangerous = [
+        node
+        for node in nodes
+        if node.element.get("package") == PDD_PACKAGE
+        and node.element.get("class") == "android.widget.TextView"
+        and "提交订单" in node.text
+    ]
+    danger = _one(dangerous, "提交硬拒绝区位置不唯一。")
+    if (
+        not _exact_readonly_text(danger, "选择尺码后，提交订单", "[285,2225][795,2284]")
+        or len(danger.element) != 0
+    ):
+        raise SkuSelectionError("提交硬拒绝区结构漂移。")
+    return tuple(
+        (
+            node.element.get("class", ""),
+            node.bounds,
+            node.text,
+            node.desc,
+            node.element.get("selected", ""),
+            node.element.get("clickable", ""),
+        )
+        for node in (
+            surface, header, outer, price_frame, price_row, current, summary,
+            color_region, color, size_label, size_options, s_action, m_action, danger,
+        )
+    )
+
+
 def _unit_price(nodes: list[_Node]) -> str:
     spec = _classify_panel(nodes)
     if spec.profile is not _PanelProfile.TARGETS_SELECTED:
@@ -601,11 +896,15 @@ def _target_color_action(nodes: list[_Node], *, selected: bool) -> _Node:
 
 def _target_size_action(nodes: list[_Node], *, selected: bool) -> _Node:
     profile = _classify_panel(nodes).profile
-    expected_profile = _PanelProfile.TARGETS_SELECTED if selected else _PanelProfile.SIZE_VISIBLE_NON_TARGET
+    expected_profiles = (
-    if profile is not expected_profile:
+        {_PanelProfile.TARGETS_SELECTED}
+        if selected
+        else {_PanelProfile.SIZE_VISIBLE_UNSELECTED, _PanelProfile.SIZE_VISIBLE_NON_TARGET}
+    )
+    if profile not in expected_profiles:
         raise SkuSelectionError("目标尺码 action 不属于预期 profile。")
     return _one(
-        [node for node in nodes if _is_size_action_text(node) and node.text == _TARGET_SIZE_UI and node.bounds == "[439,1582][831,1667]" and node.element.get("selected") == str(selected).lower()],
+        [node for node in nodes if _is_size_action_text(node) and node.text == _TARGET_SIZE_UI and node.bounds == _TARGET_SIZE_BOUNDS and node.element.get("selected") == str(selected).lower()],
         "目标尺码 action 不唯一。",
     )
 
@@ -825,7 +1124,7 @@ def _require_color_subtree(color: _Node, spec: _PanelSpec) -> list[_Node]:
     selected = str(spec.color_selected).lower()
     children = [child for child in color.element if child.tag == "node"]
     expected_selected: list[_Node] = [color] if spec.color_selected else []
-    if spec.color_bounds == "[372,1188][684,1587]":
+    if spec.color_bounds == _TARGET_COLOR_UNROLLED_BOUNDS:
         if len(children) != 4:
             raise SkuSelectionError("目标颜色完整卡片子树数量漂移。")
         child_nodes = [node for node in _walk_direct_children(color)]
@@ -869,7 +1168,7 @@ def _require_color_action_chain(
 ) -> None:
     if color.parent is not color_region:
         raise SkuSelectionError("目标颜色 action 不属于已取证颜色容器。")
-    if spec.color_bounds == "[372,1188][684,1587]":
+    if spec.color_bounds == _TARGET_COLOR_UNROLLED_BOUNDS:
         expected = (
             ("android.widget.FrameLayout", "[0,1188][1080,2046]"),
             ("android.widget.LinearLayout", "[0,1188][1080,2052]"),

client/src/cmbuyer_client/pdd/sku_selection_runner.py

@@ -31,6 +31,10 @@ from .sku_selection import (
     SkuSelectionError,
     SkuSelectionFlow,
     _SKU_ENTRY_FAILURE_STAGES,
+    _REVEAL_GESTURE,
+    _ENTRY_TEXT_BOUNDS,
+    _TARGET_COLOR_UNROLLED_BOUNDS,
+    _TARGET_SIZE_BOUNDS,
     _action_bounds,
     _annotate_sku_entry_failure,
     _safe_sku_entry_failure_stage,
@@ -135,9 +139,9 @@ class SkuSelectionRunResult:
 
 
 class UiautomatorSkuPanelAdapter(SkuPanelDevice):
-    """把 uiautomator2 缩为 T-103 所需的读取与三种命名操作。
+    """把 uiautomator2 缩为 T-103 所需的读取与四种命名操作。
 
-    ``tap_sku_entry``、``tap_sku_option`` 和 ``leave_sku_panel`` 是仅有的状态改变方法；
+    四个命名方法是仅有的状态改变入口；reveal 的手势参数固定且不向 Flow 暴露；
     坐标由 Flow 和本类双重检查后才计算中心点，每次调用只执行一次底层动作。
     """
 
@@ -147,7 +151,14 @@ class UiautomatorSkuPanelAdapter(SkuPanelDevice):
         self._device = device
         self._timeout_seconds = timeout_seconds
         self._entry_was_tapped = False
+        self._entry_bounds_attempted: str | None = None
+        self._entry_rpc_outcome = "not_attempted"
+        self._option_bounds_attempted: set[str] = set()
+        self._option_rpc_outcomes: dict[str, str] = {}
+        self._reveal_attempted = False
+        self._reveal_rpc_outcome = "not_attempted"
         self._left_panel = False
+        self._back_rpc_outcome = "not_attempted"
 
     @property
     def entry_was_tapped(self) -> bool:
@@ -159,6 +170,38 @@ class UiautomatorSkuPanelAdapter(SkuPanelDevice):
     def left_panel(self) -> bool:
         return self._left_panel
 
+    @property
+    def reveal_attempted(self) -> bool:
+        return self._reveal_attempted
+
+    @property
+    def entry_rpc_outcome(self) -> str:
+        return self._entry_rpc_outcome
+
+    @property
+    def entry_bounds_attempted(self) -> str | None:
+        return self._entry_bounds_attempted
+
+    @property
+    def option_rpc_outcomes(self) -> tuple[tuple[str, str], ...]:
+        return tuple(sorted(self._option_rpc_outcomes.items()))
+
+    @property
+    def reveal_rpc_outcome(self) -> str:
+        return self._reveal_rpc_outcome
+
+    @property
+    def option_attempts(self) -> int:
+        return len(self._option_bounds_attempted)
+
+    @property
+    def back_attempts(self) -> int:
+        return int(self._left_panel)
+
+    @property
+    def back_rpc_outcome(self) -> str:
+        return self._back_rpc_outcome
+
     def app_info(self, package_name: str) -> dict[str, Any]:
         value = self._call("app_info", package_name)
         if not isinstance(value, dict):
@@ -178,19 +221,46 @@ class UiautomatorSkuPanelAdapter(SkuPanelDevice):
         return value
 
     def tap_sku_entry(self, bounds: str) -> None:
+        if self._entry_was_tapped:
+            raise SkuSelectionDeviceAdapterError("规格入口已经尝试过，拒绝重试。")
         # 超时也可能表示底层事件已经送达；必须先封存 attempt，后续绝不重试该入口。
         self._entry_was_tapped = True
+        self._entry_bounds_attempted = bounds
+        self._entry_rpc_outcome = "ambiguous"
         self._tap_bounds_once(bounds)
+        self._entry_rpc_outcome = "completed"
 
     def tap_sku_option(self, bounds: str) -> None:
+        if bounds in self._option_bounds_attempted:
+            raise SkuSelectionDeviceAdapterError("同一规格选项已经尝试过，拒绝重试。")
+        # 规格 RPC 也可能送达后超时；按 exact bounds 封存本次唯一机会。
+        self._option_bounds_attempted.add(bounds)
+        self._option_rpc_outcomes[bounds] = "ambiguous"
         self._tap_bounds_once(bounds)
+        self._option_rpc_outcomes[bounds] = "completed"
+
+    def reveal_size_options_once(self) -> None:
+        if self._reveal_attempted:
+            raise SkuSelectionDeviceAdapterError("规格显示动作已经尝试过，拒绝重试。")
+        # 手势唯一机会在 RPC 前封存；生产 API 不接受方向、坐标或步数参数。
+        self._reveal_attempted = True
+        self._reveal_rpc_outcome = "ambiguous"
+        self._call(
+            "jsonrpc_call",
+            "swipe",
+            list(_REVEAL_GESTURE),
+            timeout=self._timeout_seconds,
+        )
+        self._reveal_rpc_outcome = "completed"
 
     def leave_sku_panel(self) -> None:
         if self._left_panel:
             raise SkuSelectionDeviceAdapterError("规格面板已经执行过返回，已停止操作。")
         # 底层调用即使报错也可能已把返回事件送达；先封存本次机会，finally 不得再次返回。
         self._left_panel = True
+        self._back_rpc_outcome = "ambiguous"
         self._call("jsonrpc_call", "pressKey", ["back"], timeout=self._timeout_seconds)
+        self._back_rpc_outcome = "completed"
 
     def capture_screenshot(self) -> str:
         value = self._call("jsonrpc_call", "takeScreenshot", SCREENSHOT_PARAMS, timeout=self._timeout_seconds)
@@ -312,9 +382,11 @@ class SkuSelectionRunner:
             stage = "safe_exit"
             flow.exit_sku_panel_safely()
 
+            _require_completed_action_audit(adapter)
+
             stage = "publish"
             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",
+                json.dumps(_manifest(inspection, serial, link, screenshot_path, task_color, task_size, adapter), ensure_ascii=False, indent=2, sort_keys=True) + "\n",
                 encoding="utf-8",
             )
             # Windows 的 rename 不替换既有目标；并发创建 target 时保留其内容并把本次运行判失败。
@@ -340,11 +412,10 @@ class SkuSelectionRunner:
             _annotate_mapped_failure(mapped, error, stage)
             raise mapped from error
         finally:
-            # 失败路径只能复用 Flow 的版本、前台和面板证明；证明不了便停止，绝不盲目返回。
+            # 结果不明只允许读回 pending；失败路径绝不继续后续动作或自动 Back。
-            if flow is not None and adapter is not None and adapter.entry_was_tapped and not adapter.left_panel:
+            if flow is not None:
                 try:
                     flow.reconcile_pending_action()
-                    flow.exit_sku_panel_safely()
                 except (SkuSelectionRunError, SkuSelectionError):
                     pass
 
@@ -411,9 +482,42 @@ def _require_screenshot_size(screenshot_path: Path) -> None:
         raise SkuSelectionScreenshotError("原始截图无效，未发布任何证据产物。") from error
 
 
-def _manifest(inspection: DeviceInspection, serial: str, link: ProductUrl, screenshot_path: Path, task_color: str, task_size: str) -> dict[str, Any]:
+def _require_completed_action_audit(adapter: UiautomatorSkuPanelAdapter) -> None:
+    """发布只接受完整正常链；测试替身或缺动作结果不能伪造真机闭环。"""
+
+    if (
+        not adapter.entry_was_tapped
+        or adapter.entry_bounds_attempted != _ENTRY_TEXT_BOUNDS
+        or adapter.entry_rpc_outcome != "completed"
+        or adapter.option_rpc_outcomes
+        != tuple(
+            sorted(
+                (
+                    (_TARGET_COLOR_UNROLLED_BOUNDS, "completed"),
+                    (_TARGET_SIZE_BOUNDS, "completed"),
+                )
+            )
+        )
+        or not adapter.reveal_attempted
+        or adapter.reveal_rpc_outcome != "completed"
+        or adapter.back_attempts != 1
+        or adapter.back_rpc_outcome != "completed"
+    ):
+        raise SkuSelectionRunError("规格动作审计链不完整，未发布任何证据产物。")
+
+
+def _manifest(
+    inspection: DeviceInspection,
+    serial: str,
+    link: ProductUrl,
+    screenshot_path: Path,
+    task_color: str,
+    task_size: str,
+    adapter: UiautomatorSkuPanelAdapter,
+) -> dict[str, Any]:
     """仅写可审计摘要；原始 serial、节点树、页面文案和实际截图内容均不写入 manifest。"""
 
+    option_outcomes = dict(adapter.option_rpc_outcomes)
     return {
         "schema_version": 1,
         "captured_at": datetime.now(UTC).isoformat(),
@@ -422,8 +526,36 @@ def _manifest(inspection: DeviceInspection, serial: str, link: ProductUrl, scree
         "target_selection": {"color": task_color, "size": task_size},
         "unit_price": EXPECTED_UNIT_PRICE,
         "selection_status": "restored",
-        "panel_status": "verified",
+        "panel_status": "verified_before_back",
-        "safe_exit": "completed",
+        "back_attempts": adapter.back_attempts,
+        "back_rpc_outcome": adapter.back_rpc_outcome,
+        "actions": {
+            "sku_entry": {
+                "attempts": int(adapter.entry_was_tapped),
+                "rpc_outcome": adapter.entry_rpc_outcome,
+            },
+            "target_color": {
+                "attempts": int(_TARGET_COLOR_UNROLLED_BOUNDS in option_outcomes),
+                "rpc_outcome": option_outcomes.get(
+                    _TARGET_COLOR_UNROLLED_BOUNDS, "not_attempted"
+                ),
+            },
+            "size_reveal": {
+                "attempts": int(adapter.reveal_attempted),
+                "rpc_outcome": adapter.reveal_rpc_outcome,
+            },
+            "target_size": {
+                "attempts": int(_TARGET_SIZE_BOUNDS in option_outcomes),
+                "rpc_outcome": option_outcomes.get(
+                    _TARGET_SIZE_BOUNDS, "not_attempted"
+                ),
+            },
+            "back": {
+                "attempts": adapter.back_attempts,
+                "rpc_outcome": adapter.back_rpc_outcome,
+            },
+        },
+        "post_exit_status": "human_review_required",
         "page_identity": "human_review_required",
         "channel": "wifi" if ":" in serial else "usb",
         "serial_sha256": sha256(serial.encode("utf-8")).hexdigest(),

client/tests/pdd/test_sku_reveal_spike.py

@@ -18,17 +18,19 @@ import cmbuyer_client.pdd.sku_reveal_spike as reveal_module
 from cmbuyer_client.device.adb import AdbDevice, DeviceInspection
 from cmbuyer_client.pdd.sku_reveal_spike import (
     _annotate_reveal_failure,
-    _candidate_projection,
-    _RevealEvidenceAdapter,
     safe_reveal_failure_stage,
     SkuRevealSpikeCapturer,
     SkuRevealSpikeError,
-    _require_safe_reveal_path,
 )
 from cmbuyer_client.pdd.sku_selection import (
     SkuSelectionError,
     _annotate_sku_entry_failure,
     _parse_nodes,
+    _require_safe_reveal_path,
+    _revealed_unselected_projection as _candidate_projection,
+)
+from cmbuyer_client.pdd.sku_selection_runner import (
+    UiautomatorSkuPanelAdapter as _RevealEvidenceAdapter,
 )
 
 
@@ -350,7 +352,7 @@ def _candidate_cross_variant_residue(variant: str, kind: str) -> str:
 
 
 def _candidate_with_color_unselected() -> str:
-    root = ElementTree.fromstring(_candidate_unselected())
+    root = ElementTree.fromstring(_candidate_current_only())
     target = next(
         node
         for node in root.iter("node")
@@ -364,7 +366,7 @@ def _candidate_with_color_unselected() -> str:
 
 
 def _candidate_with_submit_risk(kind: str) -> str:
-    root = ElementTree.fromstring(_candidate_unselected())
+    root = ElementTree.fromstring(_candidate_current_only())
     parents = {child: parent for parent in root.iter() for child in parent}
     submit = next(
         node
@@ -379,6 +381,13 @@ def _candidate_with_submit_risk(kind: str) -> str:
         parent.append(ElementTree.fromstring(ElementTree.tostring(submit, encoding="unicode")))
     elif kind == "moved_up":
         submit.set("bounds", "[285,1900][795,1959]")
+    elif kind in {"extra_altered", "extra_moved_up"}:
+        extra = ElementTree.fromstring(ElementTree.tostring(submit, encoding="unicode"))
+        if kind == "extra_altered":
+            extra.set("text", "异常提交订单文案")
+        else:
+            extra.set("bounds", "[285,1900][795,1959]")
+        parent.append(extra)
     else:
         raise AssertionError(kind)
     return ElementTree.tostring(root, encoding="unicode")
@@ -416,9 +425,9 @@ def _png() -> str:
 class _FakeDevice:
     def __init__(self) -> None:
         self.hierarchy = "<hierarchy />"
-        self.empty_hierarchy = _EMPTY
+        self.empty_hierarchy = _CURRENT_ONLY_EMPTY
-        self.color_hierarchy = _COLOR_ONLY
+        self.color_hierarchy = _CURRENT_ONLY_COLOR
-        self.after_hierarchy = _candidate_unselected()
+        self.after_hierarchy = _candidate_current_only()
         self.calls: list[tuple[object, ...]] = []
         self.swipe_error = False
 
@@ -545,8 +554,9 @@ class SkuRevealSpikeTests(unittest.TestCase):
             self.assertTrue((target / "after" / "hierarchy.xml").is_file())
 
     def test_candidate_price_variants_reject_hybrid_duplicate_unknown_and_parent_drift(self) -> None:
-        # 两个真实嵌套 variant 各自唯一命中；任意混搭、额外子节点或父子关系漂移均拒绝。
+        # 生产只放行 current-only；历史 dual 及任意混搭、额外节点、父链漂移均拒绝。
-        _candidate_projection(_parse_nodes(_candidate_unselected()))
+        with self.assertRaises(SkuSelectionError):
+            _candidate_projection(_parse_nodes(_candidate_unselected()))
         _candidate_projection(_parse_nodes(_candidate_current_only()))
         for variant in ("dual", "current_only"):
             kinds = ["hybrid", "duplicate", "unknown", "unknown_child", "parent_drift"]
@@ -683,7 +693,7 @@ class SkuRevealSpikeTests(unittest.TestCase):
         self._assert_after_rejected_once(_candidate_with_color_unselected())
 
     def test_submit_hard_reject_zone_risk_is_not_published_or_retried(self) -> None:
-        for kind in ("missing", "duplicate", "moved_up"):
+        for kind in ("missing", "duplicate", "moved_up", "extra_altered", "extra_moved_up"):
             with self.subTest(kind=kind):
                 self._assert_after_rejected_once(_candidate_with_submit_risk(kind))
 
@@ -739,7 +749,7 @@ class SkuRevealSpikeTests(unittest.TestCase):
         device.jsonrpc_call = drift  # type: ignore[method-assign]
         with TemporaryDirectory() as directory:
             target = Path(directory) / "evidence"
-            with self.assertRaises(SkuRevealSpikeError) as raised:
+            with self.assertRaises((SkuRevealSpikeError, SkuSelectionError)) as raised:
                 self._capturer(device).capture(
                     "192.168.0.173:5555",
                     "937122477375",
@@ -850,7 +860,7 @@ class SkuRevealSpikeTests(unittest.TestCase):
                 "not-a-bound",
             ),
         ):
-            with self.subTest(), self.assertRaises(SkuRevealSpikeError):
+            with self.subTest(), self.assertRaises((SkuRevealSpikeError, SkuSelectionError)):
                 _require_safe_reveal_path(_parse_nodes(hierarchy))
 
 

client/tests/pdd/test_sku_selection.py

@@ -3,8 +3,10 @@ from __future__ import annotations
 import ast
 import base64
 from contextlib import redirect_stderr
+from functools import lru_cache
 from io import BytesIO
 import importlib.util
+import json
 from pathlib import Path
 from tempfile import TemporaryDirectory
 import unittest
@@ -49,6 +51,14 @@ _TASK_SIZE = "M(建议100-115)"
 _PRODUCT_PAGE = _ENTRY_FIXTURE.read_text(encoding="utf-8")
 
 
+def _revealed_current_only() -> str:
+    # 与 spike 测试共用由已验 M fixture 机械恢复出的 current-only 未选态；生产判据只在源码一处。
+    from tests.pdd.test_sku_reveal_spike import _candidate_current_only
+
+    return _candidate_current_only()
+
+
+@lru_cache(maxsize=1)
 def _png_base64() -> str:
     image = Image.new("RGB", (1080, 2376), "white")
     raw = BytesIO()
@@ -59,8 +69,9 @@ def _png_base64() -> str:
 class _RawDevice:
     def __init__(self, hierarchy: str = _PRODUCT_PAGE, screenshot: str | None = None) -> None:
         self.hierarchy = hierarchy
-        self.panel_hierarchy = _EMPTY_FIXTURE.read_text(encoding="utf-8")
+        self.panel_hierarchy = _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8")
-        self.color_hierarchy = _COLOR_FIXTURE.read_text(encoding="utf-8")
+        self.color_hierarchy = _CURRENT_ONLY_COLOR_FIXTURE.read_text(encoding="utf-8")
+        self.revealed_hierarchy = _revealed_current_only()
         self.version = "8.17.0"
         self.package = "com.xunmeng.pinduoduo"
         self.screenshot = _png_base64() if screenshot is None else screenshot
@@ -89,6 +100,11 @@ class _RawDevice:
                 raise AssertionError(params)
             self._apply_tap(int(params[0]), int(params[1]))
             return ""
+        if method == "swipe":
+            if params != [360, 1900, 360, 1300, 30]:
+                raise AssertionError(params)
+            self.hierarchy = self.revealed_hierarchy
+            return ""
         raise AssertionError(method)
 
     def _apply_tap(self, x: int, y: int) -> None:
@@ -405,25 +421,223 @@ class SkuSelectionFlowTests(unittest.TestCase):
             flow.open_sku_panel(_TARGET_URL)
         self.assertEqual(_actions(device, "click"), [])
 
-    def test_target_mapping_stops_at_unproven_reveal_then_s_to_m_is_exact(self) -> None:
+    def test_current_only_exact_transition_reveals_then_selects_m_once(self) -> None:
         device = _RawDevice()
         flow = _flow(device)
 
         flow.open_sku_panel(_TARGET_URL)
-        with self.assertRaisesRegex(SkuSelectionError, "受控显示动作尚未取证"):
+        flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
-            flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
+        self.assertEqual(flow.read_sku_unit_price(), "12.88")
 
-        self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387)])
+        self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387), (635, 1624)])
+        self.assertEqual(len(_actions(device, "swipe")), 1)
         self.assertEqual(_actions(device, "pressKey"), [])
 
         restored = _RawDevice(_S_FIXTURE.read_text(encoding="utf-8"))
         restored_flow = _flow(restored)
         restored_flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
         self.assertEqual(restored_flow.read_sku_unit_price(), "12.88")
-        restored_flow.exit_sku_panel_safely()
-
         self.assertEqual(_tap_centers(restored), [(635, 1624)])
-        self.assertEqual(_actions(restored, "pressKey"), [("jsonrpc", "pressKey", ["back"], 10)])
+        self.assertEqual(_actions(restored, "swipe"), [])
+
+    def test_dual_color_only_is_zero_action_before_reveal(self) -> None:
+        for fixture in (_COLOR_FIXTURE, _CURRENT_ONLY_COLOR_FIXTURE):
+            with self.subTest(fixture=fixture.name):
+                device = _RawDevice(fixture.read_text(encoding="utf-8"))
+                with self.assertRaises(SkuSelectionError):
+                    _flow(device).select_sku_options(
+                        resolve_task_selection(_TASK_COLOR, _TASK_SIZE)
+                    )
+                self.assertEqual(_actions(device, "click"), [])
+                self.assertEqual(_actions(device, "swipe"), [])
+
+    def test_revealed_current_only_is_a_unique_profile_but_not_a_reentry_shortcut(self) -> None:
+        hierarchy = _revealed_current_only()
+        spec = _classify_panel(_parse_nodes(hierarchy))
+        self.assertIs(spec.profile, _PanelProfile.SIZE_VISIBLE_UNSELECTED)
+        self.assertEqual(spec.price_layout.__class__.__name__, "_CurrentOnlyPriceLayout")
+
+        device = _RawDevice(hierarchy)
+        with self.assertRaises(SkuSelectionError):
+            _flow(device).select_sku_options(
+                resolve_task_selection(_TASK_COLOR, _TASK_SIZE)
+            )
+        self.assertEqual(_actions(device, "click"), [])
+        self.assertEqual(_actions(device, "swipe"), [])
+
+    def test_revealed_exact_m_contract_rejects_missing_duplicate_disabled_near_and_other_selected(self) -> None:
+        base = _revealed_current_only()
+        for kind in ("missing", "duplicate", "disabled", "near", "other_selected"):
+            root = ElementTree.fromstring(base)
+            parents = {child: parent for parent in root.iter() for child in parent}
+            m_action = next(
+                node for node in root.iter("node")
+                if node.get("text") == "M（建议100-115）"
+            )
+            wrapper = parents[m_action]
+            if kind == "missing":
+                parents[wrapper].remove(wrapper)
+            elif kind == "duplicate":
+                parents[wrapper].append(
+                    ElementTree.fromstring(ElementTree.tostring(wrapper, encoding="unicode"))
+                )
+            elif kind == "disabled":
+                m_action.set("enabled", "false")
+            elif kind == "near":
+                m_action.set("text", "M（建议100-115）相近")
+            else:
+                s_action = next(
+                    node for node in root.iter("node")
+                    if node.get("text") == "S（建议80-100）"
+                )
+                s_action.set("selected", "true")
+            with self.subTest(kind=kind), self.assertRaises(SkuSelectionError):
+                _classify_panel(
+                    _parse_nodes(ElementTree.tostring(root, encoding="unicode"))
+                )
+
+    def test_reveal_requires_two_consecutive_stable_frames_and_terminal_blocks_reentry(self) -> None:
+        class UnstableRevealDevice(_RawDevice):
+            def __init__(self) -> None:
+                super().__init__(_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"))
+                self.after_reveal_reads = 0
+
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "dumpWindowHierarchy" and _actions(self, "swipe"):
+                    self.after_reveal_reads += 1
+                    candidate = _revealed_current_only()
+                    self.hierarchy = (
+                        candidate
+                        if self.after_reveal_reads % 2
+                        else candidate.replace("请选择： 尺码", "未知摘要")
+                    )
+                return super().jsonrpc_call(method, params, timeout)
+
+        device = UnstableRevealDevice()
+        flow = _flow(device)
+        with self.assertRaises(SkuSelectionError):
+            flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
+        action_count = len(_actions(device, "click")) + len(_actions(device, "swipe"))
+        self.assertEqual(len(_actions(device, "swipe")), 1)
+        self.assertEqual(_tap_centers(device), [(528, 1387)])
+        with self.assertRaisesRegex(SkuSelectionError, "不可重入终止态"):
+            flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
+        self.assertEqual(
+            len(_actions(device, "click")) + len(_actions(device, "swipe")),
+            action_count,
+        )
+
+    def test_reveal_timeout_is_read_only_reconciled_without_m_back_or_retry(self) -> None:
+        class RevealTimeoutDevice(_RawDevice):
+            def __init__(self, delivered: bool) -> None:
+                super().__init__(_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"))
+                self.delivered = delivered
+
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "swipe":
+                    if self.delivered:
+                        super().jsonrpc_call(method, params, timeout)
+                    else:
+                        self.calls.append(("jsonrpc", method, params, timeout))
+                    raise TimeoutError("private reveal timeout")
+                return super().jsonrpc_call(method, params, timeout)
+
+        for delivered in (False, True):
+            with self.subTest(delivered=delivered):
+                device = RevealTimeoutDevice(delivered)
+                flow = _flow(device)
+                with self.assertRaises(SkuSelectionRunError):
+                    flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
+                try:
+                    flow.reconcile_pending_action()
+                except SkuSelectionError:
+                    pass
+                self.assertEqual(len(_actions(device, "swipe")), 1)
+                self.assertEqual(_tap_centers(device), [(528, 1387)])
+                self.assertEqual(_actions(device, "pressKey"), [])
+                with self.assertRaisesRegex(SkuSelectionError, "不可重入终止态"):
+                    flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
+                self.assertEqual(len(_actions(device, "swipe")), 1)
+
+    def test_m_timeout_is_read_only_reconciled_without_back_or_retry(self) -> None:
+        class MTimeoutDevice(_RawDevice):
+            def __init__(self, delivered: bool) -> None:
+                super().__init__(_CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"))
+                self.delivered = delivered
+
+            def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                if method == "click" and params == [635, 1624]:
+                    if self.delivered:
+                        super().jsonrpc_call(method, params, timeout)
+                    else:
+                        self.calls.append(("jsonrpc", method, params, timeout))
+                    raise TimeoutError("private M timeout")
+                return super().jsonrpc_call(method, params, timeout)
+
+        for delivered in (False, True):
+            with self.subTest(delivered=delivered):
+                device = MTimeoutDevice(delivered)
+                flow = _flow(device)
+                with self.assertRaises(SkuSelectionRunError):
+                    flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
+                try:
+                    flow.reconcile_pending_action()
+                except SkuSelectionError:
+                    pass
+                self.assertEqual(len(_actions(device, "swipe")), 1)
+                self.assertEqual(_tap_centers(device), [(528, 1387), (635, 1624)])
+                self.assertEqual(_actions(device, "pressKey"), [])
+                with self.assertRaisesRegex(SkuSelectionError, "不可重入终止态"):
+                    flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
+                self.assertEqual(_tap_centers(device), [(528, 1387), (635, 1624)])
+
+    def test_foreground_is_freshly_rechecked_before_s_reveal_and_m_mutations(self) -> None:
+        class ForegroundDriftDevice(_RawDevice):
+            def __init__(self, hierarchy: str, flip_at: int) -> None:
+                super().__init__(hierarchy)
+                self.flip_at = flip_at
+                self.current_reads = 0
+
+            def app_current(self) -> dict[str, str]:
+                current = super().app_current()
+                self.current_reads += 1
+                if self.current_reads == self.flip_at:
+                    return {"package": "other"}
+                return current
+
+        cases = (
+            (
+                "s",
+                _S_FIXTURE.read_text(encoding="utf-8"),
+                2,
+                [],
+                0,
+            ),
+            (
+                "reveal",
+                _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"),
+                4,
+                [(528, 1387)],
+                0,
+            ),
+            (
+                "m",
+                _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"),
+                7,
+                [(528, 1387)],
+                1,
+            ),
+        )
+        for name, hierarchy, flip_at, expected_taps, expected_swipes in cases:
+            with self.subTest(name=name):
+                device = ForegroundDriftDevice(hierarchy, flip_at)
+                with self.assertRaises(SkuSelectionError):
+                    _flow(device).select_sku_options(
+                        resolve_task_selection(_TASK_COLOR, _TASK_SIZE)
+                    )
+                self.assertEqual(_tap_centers(device), expected_taps)
+                self.assertEqual(len(_actions(device, "swipe")), expected_swipes)
+                self.assertEqual(_actions(device, "pressKey"), [])
 
     def test_current_only_empty_to_color_only_uses_the_matching_exact_spec(self) -> None:
         device = _RawDevice()
@@ -432,15 +646,15 @@ class SkuSelectionFlowTests(unittest.TestCase):
         flow = _flow(device)
 
         flow.open_sku_panel(_TARGET_URL)
-        with self.assertRaisesRegex(SkuSelectionError, "受控显示动作尚未取证"):
+        flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
-            flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
 
-        self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387)])
+        self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387), (635, 1624)])
+        self.assertEqual(len(_actions(device, "swipe")), 1)
         matched = _classify_panel(_parse_nodes(device.hierarchy))
-        self.assertIs(matched.profile, _PanelProfile.COLOR_SELECTED_SIZE_HIDDEN)
+        self.assertIs(matched.profile, _PanelProfile.TARGETS_SELECTED)
         self.assertEqual(
             matched.price_layout.__class__.__name__,
-            "_CurrentOnlyPriceLayout",
+            "_DualPriceLayout",
         )
 
     def test_dual_and_current_only_variants_each_match_one_complete_spec(self) -> None:
@@ -585,7 +799,9 @@ class SkuSelectionFlowTests(unittest.TestCase):
                     flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
 
                 self.assertNotIn("受控显示动作尚未取证", str(raised.exception))
-                self.assertEqual(_tap_centers(device), [(865, 2218), (528, 1387)])
+                expected = [(865, 2218), (528, 1387)] if empty_fixture is _CURRENT_ONLY_EMPTY_FIXTURE else [(865, 2218)]
+                self.assertEqual(_tap_centers(device), expected)
+                self.assertEqual(_actions(device, "swipe"), [])
                 self.assertEqual(_actions(device, "pressKey"), [])
 
     def test_full_verified_entry_structure_taps_exact_text_child_once(self) -> None:
@@ -851,10 +1067,8 @@ class SkuSelectionFlowTests(unittest.TestCase):
         flow.open_sku_panel(_TARGET_URL)
         device.select_alternates()
         flow.select_sku_options(resolve_task_selection(_TASK_COLOR, _TASK_SIZE))
-        self.assertEqual(
+        self.assertEqual(_tap_centers(device), [(865, 2218), (635, 1624)])
-            _tap_centers(device),
+        self.assertEqual(_actions(device, "swipe"), [])
-            [(865, 2218), (635, 1624)],
-        )
 
     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"):
@@ -876,7 +1090,7 @@ class SkuSelectionFlowTests(unittest.TestCase):
             _flow(_RawDevice(clickable_parent)).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"}
+        forbidden = {"quantity", "confirm", "authorization", "fence", "submit", "payment", "click", "swipe", "scroll"}
         self.assertTrue(forbidden.isdisjoint(SkuSelectionFlow.__dict__))
         self.assertTrue(forbidden.isdisjoint(SkuPanelDevice.__dict__))
         self.assertTrue(forbidden.isdisjoint(pdd.__all__))
@@ -900,6 +1114,8 @@ class SkuSelectionFlowTests(unittest.TestCase):
         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)
+        swipe_calls = [node for node in ast.walk(runner_tree) if isinstance(node, ast.Constant) and node.value == "swipe"]
+        self.assertEqual(len(swipe_calls), 1)
 
     def test_entry_wait_rejects_unchanged_or_duplicate_page_without_click(self) -> None:
         now = [0.0]
@@ -947,22 +1163,24 @@ class SkuSelectionFlowTests(unittest.TestCase):
 
 
 class _CompletedFlow:
-    """仅隔离 runner 文件发布测试；生产 Flow 在 reveal 取证前仍必须停止。"""
+    """仅隔离 runner 文件发布测试，同时必须形成完整动作审计链。"""
 
     def __init__(self, device: object, *args: object, **kwargs: object) -> None:
         self.device = device
 
     def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None:
-        return None
+        self.device.tap_sku_entry("[688,2184][1042,2253]")
 
     def select_sku_options(self, selection: object) -> None:
-        return None
+        self.device.tap_sku_option("[372,1188][684,1587]")
+        self.device.reveal_size_options_once()
+        self.device.tap_sku_option("[439,1582][831,1667]")
 
     def verify_target_selection_and_read_price(self, selection: object) -> str:
         return "12.88"
 
     def exit_sku_panel_safely(self) -> None:
-        return None
+        self.device.leave_sku_panel()
 
     def reconcile_pending_action(self) -> None:
         return None
@@ -989,15 +1207,55 @@ class SkuSelectionRunnerTests(unittest.TestCase):
 
             self.assertEqual(result.unit_price, "12.88")
             manifest = result.manifest_path.read_text(encoding="utf-8")
+            manifest_data = json.loads(manifest)
             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('"panel_status": "verified_before_back"', manifest)
-            self.assertIn('"safe_exit": "completed"', manifest)
+            self.assertIn('"back_attempts": 1', manifest)
+            self.assertIn('"back_rpc_outcome": "completed"', manifest)
+            self.assertIn('"post_exit_status": "human_review_required"', manifest)
+            self.assertNotIn('"safe_exit"', manifest)
+            self.assertEqual(
+                manifest_data["actions"],
+                {
+                    "sku_entry": {"attempts": 1, "rpc_outcome": "completed"},
+                    "target_color": {"attempts": 1, "rpc_outcome": "completed"},
+                    "size_reveal": {"attempts": 1, "rpc_outcome": "completed"},
+                    "target_size": {"attempts": 1, "rpc_outcome": "completed"},
+                    "back": {"attempts": 1, "rpc_outcome": "completed"},
+                },
+            )
             self.assertFalse((target / "hierarchy.xml").exists())
+            self.assertEqual(len(_actions(device, "pressKey")), 1)
+
+    def test_publish_rejects_flow_stub_without_exact_action_audit_chain(self) -> None:
+        class IncompleteFlow(_CompletedFlow):
+            def open_sku_panel(self, product_url: str, pre_intent_hierarchy: str | None = None) -> None:
+                return None
+
+            def select_sku_options(self, selection: object) -> None:
+                return None
+
+            def exit_sku_panel_safely(self) -> None:
+                return None
+
+        device = _RawDevice()
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "result"
+            with (
+                patch.object(runner_module, "SkuSelectionFlow", IncompleteFlow),
+                self.assertRaises(SkuSelectionRunError) as raised,
+            ):
+                self._runner(_FakeAdb(), device).run(
+                    "device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target
+                )
+            self.assertEqual(safe_failure_stage(raised.exception), "safe_exit")
+            self.assertFalse(target.exists())
+            self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
             self.assertEqual(_actions(device, "pressKey"), [])
 
     def test_failure_stage_is_fixed_control_flow_metadata_without_error_text(self) -> None:
@@ -1223,6 +1481,34 @@ class SkuSelectionRunnerTests(unittest.TestCase):
         self.assertTrue(adapter.entry_was_tapped)
         self.assertEqual(_actions(adapter._device, "click"), [("jsonrpc", "click", [865, 2218], 10)])
 
+    def test_adapter_seals_every_named_mutation_before_rpc(self) -> None:
+        device = _RawDevice(_PRODUCT_PAGE)
+        adapter = UiautomatorSkuPanelAdapter(device, 10)
+        adapter.tap_sku_entry("[688,2184][1042,2253]")
+        with self.assertRaises(SkuSelectionDeviceAdapterError):
+            adapter.tap_sku_entry("[688,2184][1042,2253]")
+
+        adapter.tap_sku_option("[372,1188][684,1587]")
+        with self.assertRaises(SkuSelectionDeviceAdapterError):
+            adapter.tap_sku_option("[372,1188][684,1587]")
+
+        adapter.reveal_size_options_once()
+        with self.assertRaises(SkuSelectionDeviceAdapterError):
+            adapter.reveal_size_options_once()
+
+        adapter.leave_sku_panel()
+        with self.assertRaises(SkuSelectionDeviceAdapterError):
+            adapter.leave_sku_panel()
+
+        self.assertEqual(len(_actions(device, "click")), 2)
+        self.assertEqual(len(_actions(device, "swipe")), 1)
+        self.assertEqual(len(_actions(device, "pressKey")), 1)
+        self.assertEqual(adapter.option_attempts, 1)
+        self.assertEqual(adapter.entry_rpc_outcome, "completed")
+        self.assertEqual(adapter.reveal_rpc_outcome, "completed")
+        self.assertEqual(adapter.back_attempts, 1)
+        self.assertEqual(adapter.back_rpc_outcome, "completed")
+
     def test_entry_stability_interruptions_never_click(self) -> None:
         now = [0.0]
         class SequenceDevice(_RawDevice):
@@ -1276,8 +1562,48 @@ class SkuSelectionRunnerTests(unittest.TestCase):
         with self.assertRaises(SkuSelectionError): _flow(device).exit_sku_panel_safely()
         self.assertEqual(_actions(device, "pressKey"), [])
 
-    def test_unproven_reveal_publishes_nothing_and_safely_exits_once(self) -> None:
+    def test_back_precondition_rejects_every_non_target_profile_after_screenshot_reverify(self) -> None:
+        drifts = (
+            _CURRENT_ONLY_EMPTY_FIXTURE.read_text(encoding="utf-8"),
+            _CURRENT_ONLY_COLOR_FIXTURE.read_text(encoding="utf-8"),
+            _S_FIXTURE.read_text(encoding="utf-8"),
+            _revealed_current_only(),
+        )
+
+        for drift in drifts:
+            class DriftBeforeBackDevice(_RawDevice):
+                def __init__(self) -> None:
+                    super().__init__()
+                    self.after_screenshot_reads = 0
+                    self.screenshot_seen = False
+
+                def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
+                    if method == "takeScreenshot":
+                        value = super().jsonrpc_call(method, params, timeout)
+                        self.screenshot_seen = True
+                        return value
+                    if method == "dumpWindowHierarchy" and self.screenshot_seen:
+                        self.after_screenshot_reads += 1
+                        if self.after_screenshot_reads == 2:
+                            self.hierarchy = drift
+                    return super().jsonrpc_call(method, params, timeout)
+
+            with self.subTest(profile=_classify_panel(_parse_nodes(drift)).profile), TemporaryDirectory() as temporary:
+                device = DriftBeforeBackDevice()
+                target = Path(temporary) / "result"
+                with self.assertRaises(SkuSelectionError) as raised:
+                    self._runner(_FakeAdb(), device).run(
+                        "device-1", _TARGET_URL, _TASK_COLOR, _TASK_SIZE, target
+                    )
+                self.assertEqual(safe_failure_stage(raised.exception), "safe_exit")
+                self.assertEqual(_actions(device, "pressKey"), [])
+                self.assertFalse(target.exists())
+                self.assertEqual(list(Path(temporary).glob(".result.staging-*")), [])
+
+    def test_dual_precondition_publishes_nothing_without_reveal_or_back(self) -> None:
         device = _RawDevice()
+        device.panel_hierarchy = _EMPTY_FIXTURE.read_text(encoding="utf-8")
+        device.color_hierarchy = _COLOR_FIXTURE.read_text(encoding="utf-8")
         with TemporaryDirectory() as temporary:
             target = Path(temporary) / "out"
             with self.assertRaises(SkuSelectionError):
@@ -1285,7 +1611,8 @@ class SkuSelectionRunnerTests(unittest.TestCase):
             self.assertFalse(target.exists())
             self.assertFalse((target / "manifest.json").exists())
             self.assertEqual(list(Path(temporary).glob(".out.staging-*")), [])
-        self.assertEqual(len(_actions(device, "pressKey")), 1)
+        self.assertEqual(_actions(device, "swipe"), [])
+        self.assertEqual(len(_actions(device, "pressKey")), 0)
 
     def test_option_timeout_reconciliation_controls_back_once(self) -> None:
         class OptionTimeoutDevice(_RawDevice):
@@ -1300,7 +1627,7 @@ class SkuSelectionRunnerTests(unittest.TestCase):
                         raise TimeoutError("uncertain option")
                 return super().jsonrpc_call(method, params, timeout)
 
-        for delivered, expected_back in ((False, 0), (True, 1)):
+        for delivered, expected_back in ((False, 0), (True, 0)):
             with self.subTest(delivered=delivered), TemporaryDirectory() as temporary:
                 device = OptionTimeoutDevice(delivered)
                 adb = _FakeAdb(); device.hierarchy = "<hierarchy />"
@@ -1336,7 +1663,7 @@ class SkuSelectionRunnerTests(unittest.TestCase):
         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)
+        self.assertEqual(len(_actions(device, "pressKey")), 0)
 
 
 class SkuSelectionCliTests(unittest.TestCase):