fix(client): wait for PDD foreground after intent

client/src/cmbuyer_client/pdd/product_open.py

@@ -7,9 +7,11 @@ from dataclasses import dataclass
 from datetime import UTC, datetime
 from hashlib import sha256
 import json
+from math import isfinite
 import os
 from pathlib import Path
 import shutil
+from time import monotonic, sleep
 from typing import Any, Protocol
 from uuid import uuid4
 
@@ -40,7 +42,7 @@ class ProductVersionMismatchError(ProductOpenError):
 
 
 class ProductPackageMismatchError(ProductOpenError):
-    """Intent 后当前前台包不是拼多多。"""
+    """Intent 后在有限时间内未观察到拼多多前台包。"""
 
 
 class ProductOpenTimeoutError(ProductOpenError):
@@ -90,12 +92,20 @@ class ProductOpenCapturer:
         adb_client: AdbClient,
         connector: Callable[[str], ProductOpenUiDevice],
         timeout_seconds: float,
+        foreground_poll_interval_seconds: float = 0.2,
+        monotonic_clock: Callable[[], float] = monotonic,
+        sleep_function: Callable[[float], None] = sleep,
     ) -> None:
-        if timeout_seconds <= 0:
-            raise ValueError("timeout_seconds 必须大于 0")
+        if not _is_positive_finite(timeout_seconds):
+            raise ValueError("timeout_seconds 必须是大于 0 的有限数值")
+        if not _is_positive_finite(foreground_poll_interval_seconds):
+            raise ValueError("foreground_poll_interval_seconds 必须是大于 0 的有限数值")
         self._adb_client = adb_client
         self._connector = connector
         self._timeout_seconds = timeout_seconds
+        self._foreground_poll_interval_seconds = foreground_poll_interval_seconds
+        self._monotonic_clock = monotonic_clock
+        self._sleep_function = sleep_function
 
     def open_and_capture(self, serial: str, product_url: str, output_directory: Path) -> ProductOpenResult:
         """完成唯一允许的 Intent 打开及其后的只读取证。"""
@@ -114,7 +124,7 @@ class ProductOpenCapturer:
 
             # 版本精确匹配是 Intent 的前置条件，失败时绝不调用 start_pdd_view_intent。
             intent = self._adb_client.start_pdd_view_intent(serial, link.goods_id)
-            _require_pdd_foreground(device.app_current())
+            self._wait_for_pdd_foreground(device)
 
             target.parent.mkdir(parents=True, exist_ok=True)
             staging = target.parent / f".{target.name}.staging-{uuid4().hex}"
@@ -174,6 +184,21 @@ class ProductOpenCapturer:
             hierarchy_path=target / "hierarchy.xml",
         )
 
+    def _wait_for_pdd_foreground(self, device: ProductOpenUiDevice) -> None:
+        """只轮询当前 package，直到 deadline；Activity 和节点树均不参与本判据。"""
+
+        deadline = self._monotonic_clock() + self._timeout_seconds
+        while True:
+            if _is_pdd_foreground(device.app_current()):
+                return
+            remaining = deadline - self._monotonic_clock()
+            if remaining <= 0:
+                raise ProductPackageMismatchError(
+                    "商品链接打开后未在限定时间内进入拼多多，已停止后续取证。"
+                )
+            # 每个失败观察后都等待正的、受 deadline 约束的时长，避免 busy-loop。
+            self._sleep_function(min(self._foreground_poll_interval_seconds, remaining))
+
 
 def _validate_new_target(target: Path) -> None:
     if target.exists():
@@ -197,9 +222,12 @@ def _require_expected_version(app_info: dict[str, Any]) -> str:
     return version
 
 
-def _require_pdd_foreground(current: dict[str, Any]) -> None:
-    if not isinstance(current, dict) or current.get("package") != PDD_PACKAGE:
-        raise ProductPackageMismatchError("商品链接打开后前台应用不是拼多多，已停止后续取证。")
+def _is_positive_finite(value: object) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool) and value > 0 and isfinite(value)
+
+
+def _is_pdd_foreground(current: object) -> bool:
+    return isinstance(current, dict) and current.get("package") == PDD_PACKAGE
 
 
 def _manifest(

client/tests/pdd/test_product_open.py

@@ -63,11 +63,13 @@ class FakeUiDevice:
         *,
         version: str = "8.17.0",
         current_package: str = "com.xunmeng.pinduoduo",
+        current_packages: list[str] | None = None,
         hierarchy: str = HIERARCHY,
         timeout_on_screenshot: bool = False,
     ) -> None:
         self.version = version
         self.current_package = current_package
+        self.current_packages = list(current_packages) if current_packages is not None else None
         self.hierarchy = hierarchy
         self.timeout_on_screenshot = timeout_on_screenshot
         self.calls: list[str] = []
@@ -78,6 +80,9 @@ class FakeUiDevice:
 
     def app_current(self) -> dict[str, str]:
         self.calls.append("app_current")
+        if self.current_packages:
+            package = self.current_packages.pop(0)
+            self.current_package = package
         return {"package": self.current_package, "activity": "sensitive.activity.name"}
 
     def jsonrpc_call(self, method: str, params: object = None, timeout: float = 10) -> str:
@@ -92,8 +97,8 @@ class FakeUiDevice:
 
 
 class ProductOpenTests(unittest.TestCase):
-    def _capturer(self, adb: FakeAdbClient, device: FakeUiDevice) -> ProductOpenCapturer:
-        return ProductOpenCapturer(adb, lambda serial: device, timeout_seconds=2)
+    def _capturer(self, adb: FakeAdbClient, device: FakeUiDevice, **kwargs: object) -> ProductOpenCapturer:
+        return ProductOpenCapturer(adb, lambda serial: device, timeout_seconds=2, **kwargs)
 
     def test_success_uses_canonical_url_and_redacted_atomic_manifest(self) -> None:
         adb = FakeAdbClient()
@@ -139,15 +144,77 @@ class ProductOpenTests(unittest.TestCase):
     def test_foreground_package_mismatch_halts_before_capture(self) -> None:
         adb = FakeAdbClient()
         device = FakeUiDevice(current_package="com.example.other")
+        clock = FakeClock()
         with TemporaryDirectory() as temporary:
             target = Path(temporary) / "evidence"
             with self.assertRaises(ProductPackageMismatchError):
-                self._capturer(adb, device).open_and_capture(SERIAL, URL, target)
+                self._capturer(
+                    adb,
+                    device,
+                    foreground_poll_interval_seconds=0.5,
+                    monotonic_clock=clock.monotonic,
+                    sleep_function=clock.sleep,
+                ).open_and_capture(SERIAL, URL, target)
 
             self.assertEqual(adb.calls, [("inspect", SERIAL), ("intent", "123")])
-            self.assertEqual(device.calls, ["app_info", "app_current"])
+            self.assertEqual(device.calls, ["app_info", "app_current", "app_current", "app_current", "app_current", "app_current"])
+            self.assertEqual(clock.sleeps, [0.5, 0.5, 0.5, 0.5])
             self.assertFalse(target.exists())
 
+    def test_foreground_package_poll_waits_for_pdd_before_reading_evidence(self) -> None:
+        adb = FakeAdbClient()
+        device = FakeUiDevice(current_packages=["com.example.other", "com.xunmeng.pinduoduo"])
+        clock = FakeClock()
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "evidence"
+            result = self._capturer(
+                adb,
+                device,
+                foreground_poll_interval_seconds=0.25,
+                monotonic_clock=clock.monotonic,
+                sleep_function=clock.sleep,
+            ).open_and_capture(SERIAL, URL, target)
+
+            self.assertTrue(result.manifest_path.exists())
+            self.assertEqual(clock.sleeps, [0.25])
+            self.assertEqual(
+                device.calls,
+                ["app_info", "app_current", "app_current", "takeScreenshot", "dumpWindowHierarchy"],
+            )
+
+    def test_foreground_package_poll_stops_at_deadline_without_evidence(self) -> None:
+        adb = FakeAdbClient()
+        device = FakeUiDevice(current_packages=["com.example.other", "", "com.example.other"])
+        clock = FakeClock()
+        with TemporaryDirectory() as temporary:
+            target = Path(temporary) / "evidence"
+            with self.assertRaises(ProductPackageMismatchError):
+                self._capturer(
+                    adb,
+                    device,
+                    foreground_poll_interval_seconds=0.8,
+                    monotonic_clock=clock.monotonic,
+                    sleep_function=clock.sleep,
+                ).open_and_capture(SERIAL, URL, target)
+
+            self.assertEqual(len(clock.sleeps), 3)
+            for actual, expected in zip(clock.sleeps, (0.8, 0.8, 0.4), strict=True):
+                self.assertAlmostEqual(actual, expected)
+            self.assertEqual(device.calls, ["app_info", "app_current", "app_current", "app_current", "app_current"])
+            self.assertFalse(target.exists())
+            self.assertEqual(list(Path(temporary).glob(".evidence.staging-*")), [])
+
+    def test_foreground_poll_interval_must_be_positive_and_finite(self) -> None:
+        for interval in (0, -0.1, float("inf"), float("nan"), True):
+            with self.subTest(interval=interval):
+                with self.assertRaises(ValueError):
+                    ProductOpenCapturer(
+                        FakeAdbClient(),
+                        lambda serial: FakeUiDevice(),
+                        timeout_seconds=2,
+                        foreground_poll_interval_seconds=interval,  # type: ignore[arg-type]
+                    )
+
     def test_timeout_and_invalid_hierarchy_leave_no_partial_evidence(self) -> None:
         scenarios = (
             (FakeUiDevice(timeout_on_screenshot=True), ProductOpenTimeoutError),
@@ -184,3 +251,16 @@ class ProductOpenTests(unittest.TestCase):
 
         self.assertTrue(forbidden.isdisjoint(ProductOpenUiDevice.__dict__))
         self.assertEqual(base64.b64decode(_png_base64())[:8], b"\x89PNG\r\n\x1a\n")
+
+
+class FakeClock:
+    def __init__(self) -> None:
+        self.value = 0.0
+        self.sleeps: list[float] = []
+
+    def monotonic(self) -> float:
+        return self.value
+
+    def sleep(self, seconds: float) -> None:
+        self.sleeps.append(seconds)
+        self.value += seconds