feat: 执行并提交 PDD 采集任务 (#32)

client/src/collect_task_service.py

@@ -0,0 +1,229 @@
+"""一条采集任务的应用层流程：领取、采集、落库、提交。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable, Optional
+
+from .admin_gateway import (
+    AdminGateway,
+    AdminGatewayError,
+    AndroidDeviceInfo,
+    ClaimCapabilities,
+    ClientInfo,
+)
+from .pdd_collect_service import PddCollectError, PddCollectService
+from .pdd_device_service import PddDeviceService
+from .db import data_dir
+from .task_models import OutboxEventRecord, OutboxEventType, TaskStatus, TaskType
+from .task_models import NewClaimedTask
+from .task_repository import DuplicateTaskError, TaskRepository
+
+
+@dataclass(frozen=True)
+class CollectTaskOutcome:
+    """工作线程返回给界面的简短结果。"""
+
+    kind: str
+    message: str
+    task_id: str = ""
+
+
+CollectServiceFactory = Callable[
+    [str, str, Callable[[], bool]], PddCollectService
+]
+
+
+class CollectTaskService:
+    """一次调用只处理一条本地工作或一个待提交事件。"""
+
+    def __init__(
+        self,
+        gateway: AdminGateway,
+        repository: TaskRepository,
+        client: ClientInfo,
+        device_address: str,
+        *,
+        cancelled: Callable[[], bool] = lambda: False,
+        collect_service_factory: Optional[CollectServiceFactory] = None,
+    ) -> None:
+        self._gateway = gateway
+        self._repository = repository
+        self._client = client
+        self._device_address = device_address
+        self._cancelled = cancelled
+        self._factory = collect_service_factory or self._default_factory
+
+    def execute_one(self) -> CollectTaskOutcome:
+        """先补交已有结果，再执行最早的本地任务，最后才领取新任务。"""
+
+        pending = self._repository.next_pending_outbox()
+        if pending is not None:
+            return self._submit(pending)
+
+        if not self._device_address.strip():
+            raise ValueError("请先在设置页选择并保存 Android 设备")
+
+        task = self._repository.next_collect_task()
+        if task is None:
+            remote = self._gateway.claim_next(
+                self._client,
+                ClaimCapabilities(
+                    device=AndroidDeviceInfo(self._device_address),
+                    supported_types=(TaskType.COLLECT,),
+                    purchase_mode="dry_run",
+                    schema_versions=(1,),
+                ),
+            )
+            if remote is None:
+                return CollectTaskOutcome("no_task", "暂无可领取的采集任务")
+            try:
+                self._repository.add_claimed_task(
+                    NewClaimedTask(
+                        remote_task_id=remote.task_id,
+                        task_type=remote.task_type,
+                        goods_url=str(remote.payload.get("goods_url") or ""),
+                        goods_id=(
+                            str(remote.payload["goods_id"])
+                            if remote.payload.get("goods_id") is not None
+                            else None
+                        ),
+                        priority=remote.priority,
+                        version=remote.version,
+                        admin_payload={
+                            "id": remote.task_id,
+                            "type": remote.task_type.value,
+                            "version": remote.version,
+                            "priority": remote.priority,
+                            "payload": dict(remote.payload),
+                            "created_at": remote.created_at,
+                            "updated_at": remote.updated_at,
+                        },
+                    )
+                )
+            except DuplicateTaskError:
+                pass
+            except Exception as exc:
+                raise RuntimeError(
+                    f"任务 {remote.task_id} 已领取，但本地保存失败：{exc}"
+                ) from exc
+            task = self._repository.get_task(remote.task_id)
+            if task is None:
+                raise RuntimeError(f"任务 {remote.task_id} 未能保存到本地")
+
+        if self._cancelled():
+            return CollectTaskOutcome("cancelled", "本次采集已取消", task.remote_task_id)
+
+        started = self._repository.start_collect_run(
+            task.remote_task_id, self._device_address
+        )
+        collector = self._factory(
+            self._device_address, self._client.client_id, self._cancelled
+        )
+        try:
+            result = collector.collect(started.task)
+            event = self._repository.save_collect_result(
+                task.remote_task_id, started.attempt_id, result.to_pdd_data()
+            )
+        except PddCollectError as exc:
+            status, retryable = self._classify_error(exc.code)
+            report_code = self._report_error_code(exc.code)
+            event = self._repository.save_collect_failure(
+                task.remote_task_id,
+                started.attempt_id,
+                status,
+                report_code,
+                exc.message,
+                retryable,
+                exc.diagnostics,
+            )
+        return self._submit(event)
+
+    def _submit(self, event: OutboxEventRecord) -> CollectTaskOutcome:
+        task_id = self._repository.outbox_task_id(event.id)
+        self._repository.mark_outbox_sending(event.id)
+        try:
+            if event.event_type is OutboxEventType.TASK_FAILURE:
+                receipt = self._gateway.submit_failure(
+                    task_id, event.idempotency_key, event.payload_json
+                )
+            else:
+                receipt = self._gateway.submit_result(
+                    task_id, event.idempotency_key, event.payload_json
+                )
+            if not receipt.accepted:
+                raise AdminGatewayError(
+                    "ADMIN_RESULT_NOT_ACCEPTED", "Admin 未确认接收结果", False
+                )
+        except AdminGatewayError as exc:
+            message = str(exc)
+            if exc.retryable:
+                self._repository.mark_outbox_retry(event.id, message)
+                return CollectTaskOutcome(
+                    "result_pending",
+                    f"任务 {task_id} 数据已保存在本地，等待重新提交 Admin：{message}",
+                    task_id,
+                )
+            self._repository.mark_outbox_failed(event.id, message)
+            return CollectTaskOutcome(
+                "manual_review",
+                f"任务 {task_id} 提交被 Admin 拒绝：{message}",
+                task_id,
+            )
+
+        self._repository.mark_outbox_sent(event.id)
+        if event.event_type is OutboxEventType.TASK_FAILURE:
+            return CollectTaskOutcome(
+                "failed",
+                f"任务 {task_id} 采集未完成，失败信息已提交 Admin",
+                task_id,
+            )
+        return CollectTaskOutcome(
+            "succeeded", f"任务 {task_id} 采集完成并已提交 Admin", task_id
+        )
+
+    @staticmethod
+    def _classify_error(code: str) -> tuple[TaskStatus, bool]:
+        if code == "PDD_CANCELLED":
+            return TaskStatus.CANCELLED, False
+        if code in {
+            "PDD_PAGE_LOGIN_REQUIRED",
+            "PDD_PAGE_CAPTCHA",
+            "PDD_DATA_SPEC_INCOMPLETE",
+            "PDD_DATA_TITLE_MISSING",
+            "PDD_DATA_PRICE_MISSING",
+            "PDD_DATA_SKU_NAME_TRUNCATED",
+        }:
+            return TaskStatus.MANUAL_REVIEW, False
+        if code.startswith("PDD_DATA_GOODS_"):
+            return TaskStatus.FAILED, False
+        return TaskStatus.RETRY_WAIT, True
+
+    @staticmethod
+    def _report_error_code(code: str) -> str:
+        """把内部解析代码映射为 Admin 可枚举的稳定代码。"""
+
+        return {
+            "PDD_DATA_SKU_NAME_TRUNCATED": "SKU_NAME_TRUNCATED",
+            "PDD_PAGE_SPEC_ENTRY_MISSING": "SKU_PANEL_NOT_FOUND",
+            "PDD_DATA_TITLE_MISSING": "TITLE_TOO_SHORT",
+            "PDD_PAGE_OVERALL_TIMEOUT": "COLLECT_TIMEOUT",
+            "DEVICE_OFFLINE": "DEVICE_DISCONNECTED",
+            "DEVICE_CONNECT_FAILED": "DEVICE_DISCONNECTED",
+            "DEVICE_DISCONNECTED": "DEVICE_DISCONNECTED",
+        }.get(code, code)
+
+    @staticmethod
+    def _default_factory(
+        device_address: str,
+        client_id: str,
+        cancelled: Callable[[], bool],
+    ) -> PddCollectService:
+        return PddCollectService(
+            PddDeviceService(),
+            device_address,
+            client_id,
+            cancelled=cancelled,
+            overall_timeout=600.0,
+            artifact_directory=data_dir() / "artifacts",
+        )

client/src/http_admin_gateway.py

@@ -3,20 +3,20 @@
 import json
 import socket
 from http.client import RemoteDisconnected
-from typing import Callable, Mapping, Optional
+from typing import Any, Callable, Mapping, Optional
 from urllib.error import HTTPError, URLError
 from urllib.parse import urlparse
 from urllib.request import ProxyHandler, Request, build_opener
 from uuid import uuid4
 
 from .admin_gateway import (
+    AdminGateway,
     AdminTask,
     AdminGatewayError,
     ClaimCapabilities,
     ClientInfo,
-    ClientRegistrationGateway,
     RegistrationReceipt,
-    TaskClaimGateway,
+    SubmissionReceipt,
 )
 from .task_models import TaskType
 
@@ -24,8 +24,8 @@ from .task_models import TaskType
 DEFAULT_ADMIN_BASE_URL = "http://127.0.0.1:8080"
 
 
-class HttpAdminGateway(ClientRegistrationGateway, TaskClaimGateway):
-    """通过 HTTP 登记 Client 和领取采集任务；令牌只保存在内存。"""
+class HttpAdminGateway(AdminGateway):
+    """通过 HTTP 登记、领取和提交任务；令牌只保存在内存。"""
 
     def __init__(
         self,
@@ -33,6 +33,7 @@ class HttpAdminGateway(ClientRegistrationGateway, TaskClaimGateway):
         token: str = "",
         timeout_seconds: float = 3.0,
         opener: Optional[Callable] = None,
+        client_id: str = "",
     ):
         normalized_url = base_url.strip().rstrip("/")
         parsed = urlparse(normalized_url)
@@ -44,6 +45,7 @@ class HttpAdminGateway(ClientRegistrationGateway, TaskClaimGateway):
         self._base_url = normalized_url
         self._token = token.strip()
         self._timeout_seconds = timeout_seconds
+        self._client_id = client_id.strip()
         # Admin 通常运行在本机或局域网。明确禁用环境代理，避免
         # HTTP_PROXY 把 127.0.0.1 请求错误转发到代理服务器。
         self._opener = opener or build_opener(ProxyHandler({})).open
@@ -54,6 +56,7 @@ class HttpAdminGateway(ClientRegistrationGateway, TaskClaimGateway):
         """调用独立登记接口，不领取或修改任务。"""
 
         request_id = str(uuid4())
+        self._client_id = client.client_id.strip()
         payload = {
             "client": {"name": client.name.strip()},
             "supported_types": [
@@ -146,6 +149,7 @@ class HttpAdminGateway(ClientRegistrationGateway, TaskClaimGateway):
         """领取一个采集任务；Admin 返回 204 时返回 ``None``。"""
 
         request_id = str(uuid4())
+        self._client_id = client.client_id.strip()
         payload = {
             "client": {"name": client.name.strip()},
             # #30 只允许领取采集任务。采购能力必须由安全门禁工单开启。
@@ -219,6 +223,104 @@ class HttpAdminGateway(ClientRegistrationGateway, TaskClaimGateway):
             )
         return self._parse_claim_response(body, request_id)
 
+    def submit_result(
+        self,
+        task_id: str,
+        idempotency_key: str,
+        result: Mapping[str, Any],
+    ) -> SubmissionReceipt:
+        """幂等提交成功结果。"""
+
+        return self._submit(task_id, idempotency_key, result, "result")
+
+    def submit_failure(
+        self,
+        task_id: str,
+        idempotency_key: str,
+        failure: Mapping[str, Any],
+    ) -> SubmissionReceipt:
+        """幂等提交失败或人工处理结果。"""
+
+        return self._submit(task_id, idempotency_key, failure, "failure")
+
+    def _submit(
+        self,
+        task_id: str,
+        idempotency_key: str,
+        payload: Mapping[str, Any],
+        endpoint: str,
+    ) -> SubmissionReceipt:
+        if not task_id.strip():
+            raise ValueError("task_id 不能为空")
+        if not idempotency_key.strip():
+            raise ValueError("Idempotency-Key 不能为空")
+        if not self._client_id:
+            raise AdminGatewayError(
+                "CLIENT_ID_MISSING",
+                "提交结果前必须先配置 Client 设备号",
+                False,
+            )
+        request_id = str(uuid4())
+        headers = {
+            "Content-Type": "application/json; charset=utf-8",
+            "Accept": "application/json",
+            "Idempotency-Key": idempotency_key.strip(),
+            "X-Request-Id": request_id,
+            "X-Client-Id": self._client_id,
+        }
+        if self._token:
+            headers["Authorization"] = f"Bearer {self._token}"
+        request = Request(
+            f"{self._base_url}/api/v1/client/tasks/{task_id.strip()}/{endpoint}",
+            data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
+            headers=headers,
+            method="POST",
+        )
+        try:
+            with self._opener(request, timeout=self._timeout_seconds) as response:
+                status = getattr(response, "status", None)
+                if status is None:
+                    status = response.getcode()
+                body = response.read()
+        except HTTPError as exc:
+            self._raise_http_error(exc, request_id, "提交任务")
+        except (
+            URLError,
+            RemoteDisconnected,
+            ConnectionError,
+            socket.timeout,
+            TimeoutError,
+        ) as exc:
+            reason = getattr(exc, "reason", exc)
+            code = "ADMIN_TIMEOUT" if isinstance(reason, (socket.timeout, TimeoutError)) else "ADMIN_UNAVAILABLE"
+            message = "Admin 提交请求超时，请稍后重试" if code == "ADMIN_TIMEOUT" else "无法连接 Admin，结果已保存在本地"
+            raise AdminGatewayError(code, message, True, request_id) from exc
+        if status not in (200, 201):
+            raise AdminGatewayError(
+                "ADMIN_UNEXPECTED_RESPONSE",
+                f"Admin 返回了未预期的状态码 {status}",
+                status >= 500,
+                request_id,
+            )
+        data = self._decode_json(body, request_id)
+        accepted = data.get("accepted")
+        result_id = data.get("result_id")
+        accepted_at = data.get("accepted_at")
+        if (
+            accepted is not True
+            or not isinstance(result_id, str)
+            or not result_id.strip()
+            or not isinstance(accepted_at, str)
+            or not accepted_at.strip()
+        ):
+            raise AdminGatewayError(
+                "ADMIN_INVALID_RESPONSE",
+                "Admin 提交响应字段不完整",
+                False,
+                request_id,
+            )
+        return SubmissionReceipt(True, result_id, accepted_at)
+
     @classmethod
     def _parse_claim_response(
         cls,

client/src/pdd_collect_service.py

@@ -14,6 +14,7 @@ import xml.etree.ElementTree as ET
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
+from pathlib import Path
 from typing import Any, Callable, Iterable, Mapping, Optional, Sequence
 from urllib.parse import parse_qs, urlparse
 
@@ -31,13 +32,32 @@ _CAPTCHA_MARKERS = ("请完成验证", "拖动滑块", "安全验证", "点击
 _READY_MARKERS = ("发起拼单", "立即购买", "单独购买", "免拼购买", "快要抢光")
 
 
+def _is_device_disconnect(error: BaseException) -> bool:
+    details = str(error).lower()
+    return any(
+        marker in details
+        for marker in (
+            "device not found",
+            "device offline",
+            "disconnected",
+            "closed transport",
+        )
+    )
+
+
 class PddCollectError(RuntimeError):
     """采集失败，并携带稳定错误码。"""
 
-    def __init__(self, code: str, message: str) -> None:
+    def __init__(
+        self,
+        code: str,
+        message: str,
+        diagnostics: Optional[Mapping[str, Any]] = None,
+    ) -> None:
         super().__init__(message)
         self.code = code
         self.message = message
+        self.diagnostics = dict(diagnostics or {})
 
 
 @dataclass(frozen=True)
@@ -85,11 +105,15 @@ class SkuResult:
     price_cent: Optional[int]
     available: bool
     raw_price: Optional[str]
+    price_observed_at: Mapping[str, str]
+    list_price_cent: Optional[int] = None
 
     def to_dict(self) -> dict[str, Any]:
         return {
             "options": dict(self.options),
             "price_cent": self.price_cent,
+            "list_price_cent": self.list_price_cent,
+            "price_observed_at": dict(self.price_observed_at),
             "currency": "CNY",
             "available": self.available,
             "raw_price": self.raw_price,
@@ -110,6 +134,7 @@ class SpecSnapshot:
     selected_text: Optional[str]
     price_cent: Optional[int]
     raw_price: Optional[str]
+    list_price_cent: Optional[int]
 
 
 @dataclass(frozen=True)
@@ -127,16 +152,16 @@ class CollectResult:
     captured_at: str
     client_id: str
     device_address: str
+    artifacts: tuple[Mapping[str, Any], ...] = ()
 
     def to_pdd_data(self) -> dict[str, Any]:
         return {
             "schema_version": 1,
-            "goods": {
-                "goods_id": self.goods_id,
-                "url": self.goods_url,
-                "title": self.title,
-            },
-            "shop": {"name": self.shop_name},
+            "goods_id": self.goods_id,
+            "goods_url": self.goods_url,
+            "title": self.title,
+            "shop_name": self.shop_name,
+            "price_granularity": "color",
             "metrics": {
                 "sales": self.sales.to_dict(),
                 "reviews": self.reviews.to_dict(),
@@ -150,7 +175,7 @@ class CollectResult:
                 "device_address": self.device_address,
                 "pdd_package": PDD_PACKAGE_NAME,
             },
-            "artifacts": [],
+            "artifacts": [dict(item) for item in self.artifacts],
         }
 
 
@@ -182,6 +207,14 @@ def _node_label(node: ET.Element) -> str:
     )
 
 
+def _preferred_node_label(node: ET.Element) -> str:
+    """规格名被 text 截断时，优先采用完整的无障碍描述。"""
+
+    text = (node.get("text") or "").strip()
+    description = (node.get("content-desc") or "").strip()
+    return description or text
+
+
 def _own_or_descendant_label(node: ET.Element) -> str:
     own = _node_label(node)
     if own:
@@ -193,6 +226,17 @@ def _own_or_descendant_label(node: ET.Element) -> str:
     return ""
 
 
+def _preferred_or_descendant_label(node: ET.Element) -> str:
+    own = _preferred_node_label(node)
+    if own:
+        return own
+    for child in node.iter("node"):
+        label = _preferred_node_label(child)
+        if label and label != "打开大图":
+            return label
+    return ""
+
+
 def _all_labels(root: ET.Element) -> list[str]:
     return [label for node in root.iter("node") if (label := _node_label(node))]
 
@@ -250,9 +294,21 @@ def parse_goods_page(xml_data: str | bytes) -> GoodsSnapshot:
                 title = candidate
                 break
     if title is None:
+        line_parts: dict[int, list[tuple[int, str]]] = {}
+        for node in root.iter("node"):
+            label = _preferred_node_label(node)
+            bounds = _parse_bounds(node.get("bounds", ""))
+            if not label or bounds is None:
+                continue
+            line_key = bounds[1] // 24
+            line_parts.setdefault(line_key, []).append((bounds[0], label))
+        joined_lines = [
+            "".join(text for _, text in sorted(parts))
+            for parts in line_parts.values()
+        ]
         candidates = [
             label
-            for label in labels
+            for label in labels + joined_lines
             if len(label) >= 12
             and not any(word in label for word in ("通知", "支付", "已拼", "评价"))
         ]
@@ -347,7 +403,7 @@ def _top_level_clickable_options(
 def _price_from_nodes(
     nodes: Iterable[ET.Element],
     spec_top: int,
-) -> tuple[Optional[int], Optional[str]]:
+) -> tuple[Optional[int], Optional[str], Optional[int]]:
     candidates: list[tuple[int, int, str]] = []
     for node in nodes:
         label = _node_label(node)
@@ -366,9 +422,12 @@ def _price_from_nodes(
             continue
         candidates.append((bounds[1], cents, match.group(0).replace(" ", "")))
     if not candidates:
-        return None, None
-    _, cents, raw = min(candidates, key=lambda item: (item[0], item[1]))
-    return cents, raw
+        return None, None, None
+    first_top = min(item[0] for item in candidates)
+    first_row = [item for item in candidates if abs(item[0] - first_top) <= 12]
+    _, cents, raw = min(first_row, key=lambda item: item[1])
+    list_prices = [item[1] for item in first_row if item[1] > cents]
+    return cents, raw, max(list_prices, default=None)
 
 
 def parse_spec_panel(xml_data: str | bytes) -> SpecSnapshot:
@@ -446,7 +505,12 @@ def parse_spec_panel(xml_data: str | bytes) -> SpecSnapshot:
         values: list[DimensionValue] = []
         seen: set[str] = set()
         for node in option_nodes:
-            text = _own_or_descendant_label(node).strip()
+            text = _preferred_or_descendant_label(node).strip()
+            if text.endswith(("…", "...")):
+                raise PddCollectError(
+                    "PDD_DATA_SKU_NAME_TRUNCATED",
+                    f"规格名称被截断，无法安全采集：{text}",
+                )
             if not text or text in seen:
                 continue
             seen.add(text)
@@ -458,8 +522,12 @@ def parse_spec_panel(xml_data: str | bytes) -> SpecSnapshot:
         dimensions.append(SpecDimension(key, name, tuple(values)))
 
     selected_text = next((label for label in labels if label.startswith("已选")), None)
-    price_cent, raw_price = _price_from_nodes(root.iter("node"), outer_bounds[1])
-    return SpecSnapshot(tuple(dimensions), selected_text, price_cent, raw_price)
+    price_cent, raw_price, list_price_cent = _price_from_nodes(
+        root.iter("node"), outer_bounds[1]
+    )
+    return SpecSnapshot(
+        tuple(dimensions), selected_text, price_cent, raw_price, list_price_cent
+    )
 
 
 def _ancestors(
@@ -561,9 +629,11 @@ class PddCollectService:
         now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
         cancelled: Callable[[], bool] = lambda: False,
         page_timeout: float = 30.0,
+        overall_timeout: float = 600.0,
         max_page_swipes: int = 12,
         max_spec_swipes: int = 12,
         max_sku_count: int = 200,
+        artifact_directory: Optional[Path] = None,
     ) -> None:
         self._device_service = device_service
         self._device_address = device_address
@@ -573,9 +643,15 @@ class PddCollectService:
         self._now = now
         self._cancelled = cancelled
         self._page_timeout = page_timeout
+        self._overall_timeout = overall_timeout
+        self._overall_deadline: Optional[float] = None
         self._max_page_swipes = max_page_swipes
         self._max_spec_swipes = max_spec_swipes
         self._max_sku_count = max_sku_count
+        self._artifact_directory = artifact_directory
+        self._last_goods_xml: Optional[str] = None
+        self._goods_screens_checked = 0
+        self._artifacts: list[Mapping[str, Any]] = []
 
     def collect(self, task: Any) -> CollectResult:
         """执行采集；``task`` 至少提供 ``goods_url`` 和可选 ``goods_id``。"""
@@ -585,6 +661,7 @@ class PddCollectService:
             raise PddCollectError("PDD_DATA_GOODS_URL_MISSING", "采集任务缺少商品链接")
         goods_id = str(getattr(task, "goods_id", "") or "").strip()
         goods_id = _validate_goods_url(goods_url, goods_id)
+        self._overall_deadline = self._monotonic() + self._overall_timeout
         self._check_cancelled()
 
         try:
@@ -593,12 +670,14 @@ class PddCollectService:
                 goods = self._collect_goods_details(device)
                 if not goods.title:
                     raise PddCollectError("PDD_DATA_TITLE_MISSING", "商品页没有可识别的标题")
-                if not goods.shop_name:
-                    raise PddCollectError("PDD_DATA_SHOP_MISSING", "商品页没有采集到店铺名称")
                 if not goods.sales.raw:
                     raise PddCollectError("PDD_DATA_SALES_MISSING", "商品页没有采集到已拼数量")
                 if not goods.reviews.raw:
                     raise PddCollectError("PDD_DATA_REVIEWS_MISSING", "商品页没有采集到评价数量")
+                if not goods.shop_name and self._last_goods_xml:
+                    artifact = self._save_xml("shop-not-found", self._last_goods_xml)
+                    if artifact:
+                        self._artifacts.append(artifact)
 
                 home_xml = device.dump_hierarchy()
                 coordinate = get_size_panel_coord(home_xml)
@@ -623,11 +702,23 @@ class PddCollectService:
                 )
                 if not skus or not has_available_price:
                     raise PddCollectError("PDD_DATA_PRICE_MISSING", "没有采集到可用 SKU 的价格")
-        except PddCollectError:
+        except PddCollectError as exc:
+            if not exc.diagnostics and self._last_goods_xml:
+                artifact = self._save_xml("collect-failed", self._last_goods_xml)
+                if artifact:
+                    exc.diagnostics = {
+                        "artifacts": [artifact],
+                        "goods_screens_checked": self._goods_screens_checked,
+                    }
             raise
         except PddDeviceError as exc:
             raise PddCollectError(exc.code, exc.message) from exc
         except Exception as exc:
+            if _is_device_disconnect(exc):
+                raise PddCollectError(
+                    "DEVICE_DISCONNECTED",
+                    f"Android 设备在采集过程中断开：{exc}",
+                ) from exc
             raise PddCollectError("PDD_PAGE_UNKNOWN", f"PDD 采集过程中发生未知错误：{exc}") from exc
 
         return CollectResult(
@@ -642,11 +733,17 @@ class PddCollectService:
             captured_at=self._now().astimezone(timezone.utc).isoformat().replace("+00:00", "Z"),
             client_id=self._client_id,
             device_address=self._device_address,
+            artifacts=tuple(self._artifacts),
         )
 
     def _check_cancelled(self) -> None:
         if self._cancelled():
             raise PddCollectError("PDD_CANCELLED", "采集任务已安全取消")
+        if (
+            self._overall_deadline is not None
+            and self._monotonic() >= self._overall_deadline
+        ):
+            raise PddCollectError("PDD_PAGE_OVERALL_TIMEOUT", "PDD 采集超过 10 分钟")
 
     def _open_goods(self, device: Any, goods_url: str) -> None:
         try:
@@ -659,6 +756,11 @@ class PddCollectService:
         except PddCollectError:
             raise
         except Exception as exc:
+            if _is_device_disconnect(exc):
+                raise PddCollectError(
+                    "DEVICE_DISCONNECTED",
+                    f"Android 设备在打开商品页时断开：{exc}",
+                ) from exc
             raise PddCollectError("DEVICE_APP_START_FAILED", f"无法打开 PDD 商品链接：{exc}") from exc
 
         deadline = self._monotonic() + self._page_timeout
@@ -686,13 +788,14 @@ class PddCollectService:
         for _ in range(self._max_page_swipes + 1):
             self._check_cancelled()
             xml_data = device.dump_hierarchy()
+            self._last_goods_xml = str(xml_data)
+            self._goods_screens_checked += 1
             root = _parse_xml(xml_data)
             labels = tuple(_all_labels(root))
             snapshots.append(parse_goods_page(xml_data))
             combined = _combine_goods_snapshots(snapshots)
             is_complete = (
                 combined.title
-                and combined.shop_name
                 and combined.sales.raw
                 and combined.reviews.raw
             )
@@ -720,6 +823,27 @@ class PddCollectService:
             self._sleep(0.35)
         return _combine_goods_snapshots(snapshots)
 
+    def _save_xml(self, label: str, xml_data: str) -> Optional[Mapping[str, Any]]:
+        """保存本地诊断 XML；测试未提供目录时不写文件。"""
+
+        if self._artifact_directory is None:
+            return None
+        try:
+            digest = hashlib.sha256(xml_data.encode("utf-8")).hexdigest()
+            directory = self._artifact_directory / self._client_id
+            directory.mkdir(parents=True, exist_ok=True)
+            path = directory / f"{label}-{digest[:12]}.xml"
+            if not path.exists():
+                path.write_text(xml_data, encoding="utf-8")
+            return {
+                "kind": "accessibility_xml",
+                "path": str(path.resolve()),
+                "sha256": digest,
+                "screens_checked": self._goods_screens_checked,
+            }
+        except OSError:
+            return None
+
     def _discover_dimensions(self, device: Any) -> list[SpecSnapshot]:
         snapshots: list[SpecSnapshot] = []
 
@@ -768,28 +892,66 @@ class PddCollectService:
                 f"规格组合共 {len(combinations)} 个，超过安全上限 {self._max_sku_count}",
             )
 
-        results: list[SkuResult] = []
-        for values in combinations:
+        color_index = next(
+            (index for index, item in enumerate(dimensions) if item.key == "color"),
+            None,
+        )
+        if color_index is None:
+            raise PddCollectError(
+                "PDD_DATA_SPEC_INCOMPLETE", "规格面板没有可识别的颜色分类"
+            )
+
+        samples: dict[
+            str,
+            tuple[Optional[int], Optional[str], Optional[int], dict[str, str]],
+        ] = {}
+        color_dimension = dimensions[color_index]
+        for color in color_dimension.values:
             self._check_cancelled()
-            options = {dimension.key: value.text for dimension, value in zip(dimensions, values)}
-            pairs = list(zip(dimensions, values))
-            selected = self._select_combination(device, pairs)
-            if not selected and len(pairs) > 1:
-                # 某个选项可能只是在当前搭配下禁用。反向选择一次，可以先改变
-                # 依赖维度，再重新判断目标组合是否真的缺货。
-                selected = self._select_combination(device, list(reversed(pairs)))
+            if not color.available:
+                samples[color.text] = (None, None, None, {"color": color.text})
+                continue
+            selected = self._select_option(device, color.text, color_dimension.key)
             if not selected:
-                results.append(SkuResult(options, None, False, None))
+                samples[color.text] = (None, None, None, {"color": color.text})
                 continue
             snapshot = parse_spec_panel(device.dump_hierarchy())
             summary = snapshot.selected_text or ""
-            confirmed = all(value.text in summary for value in values)
+            observed: dict[str, str] = {}
+            for dimension in dimensions:
+                match = next(
+                    (value.text for value in dimension.values if value.text in summary),
+                    None,
+                )
+                if match:
+                    observed[dimension.key] = match
+            confirmed = (
+                observed.get(color_dimension.key) == color.text
+                and len(observed) == len(dimensions)
+            )
+            samples[color.text] = (
+                snapshot.price_cent if confirmed else None,
+                snapshot.raw_price if confirmed else None,
+                snapshot.list_price_cent if confirmed else None,
+                observed,
+            )
+
+        results: list[SkuResult] = []
+        for values in combinations:
+            options = {
+                dimension.key: value.text
+                for dimension, value in zip(dimensions, values)
+            }
+            price, raw_price, list_price, observed = samples[values[color_index].text]
+            available = all(value.available for value in values) and price is not None
             results.append(
                 SkuResult(
                     options,
-                    snapshot.price_cent if confirmed else None,
-                    confirmed and snapshot.price_cent is not None,
-                    snapshot.raw_price if confirmed else None,
+                    price if available else None,
+                    available,
+                    raw_price if available else None,
+                    observed,
+                    list_price if available else None,
                 )
             )
         return tuple(results)
@@ -864,7 +1026,7 @@ class PddCollectService:
         ]
         screen_right = max(right_edges, default=0)
         for node in root.iter("node"):
-            if _own_or_descendant_label(node).strip() != target:
+            if _preferred_or_descendant_label(node).strip() != target:
                 continue
             bounds = _parse_bounds(node.get("bounds", ""))
             if node.get("clickable") != "true" or bounds is None:

client/src/pdd_ui_event.py

@@ -13,7 +13,7 @@
 - 窗口关闭时要断开信号并置标志位，否则迟到的后台结果会访问
   已经销毁的控件、直接崩溃。做法见同文档 §5.2。
 - 数据库读写走 Repository，**不要在这里拼业务 SQL**。
-- “获取任务”会真的去操作手机、可能下单；“搜索”只读本地数据库。
+- “获取任务”会真的去操作手机采集商品；“搜索”只读本地数据库。
   两者必须分开，不得共用入口。
 - 普通成功不弹窗，更新界面即可；可恢复错误用 `InfoBar`
   （模板见 `docs/client/05-ui-specification.md` §9.1）；
@@ -28,11 +28,10 @@ from qfluentwidgets import InfoBar, InfoBarPosition
 from .admin_gateway import (
     AdminGatewayError,
     AdminTask,
-    AndroidDeviceInfo,
-    ClaimCapabilities,
     ClientInfo,
-    TaskClaimGateway,
+    AdminGateway,
 )
+from .collect_task_service import CollectServiceFactory, CollectTaskService
 from .current_client_service import CurrentClientService
 from .http_admin_gateway import DEFAULT_ADMIN_BASE_URL, HttpAdminGateway
 from .pdd_ui import PDDTaskPage, TaskRow
@@ -45,7 +44,7 @@ from .task_models import (
     TaskSummary,
     TaskType,
 )
-from .task_repository import DuplicateTaskError, TaskRepository
+from .task_repository import TaskRepository
 
 
 TASK_TYPE_BY_TEXT = {
@@ -116,21 +115,23 @@ def admin_task_to_new_claimed_task(task: AdminTask) -> NewClaimedTask:
 
 
 class ClaimTaskWorker(QObject):
-    """在后台领取至多一个采集任务，并先写入本地数据库。"""
+    """在后台补交或执行至多一条采集任务。"""
 
     noTask = pyqtSignal()
     taskSaved = pyqtSignal(str)
     duplicateTask = pyqtSignal(str)
     localSaveFailed = pyqtSignal(str, str)
     failed = pyqtSignal(str)
+    outcome = pyqtSignal(str, str, str)
     completed = pyqtSignal()
 
     def __init__(
         self,
-        gateway: TaskClaimGateway,
+        gateway: AdminGateway,
         task_repository: TaskRepository,
         client_service: CurrentClientService,
         android_device_service: SelectedAndroidDeviceService,
+        collect_service_factory: Optional[CollectServiceFactory] = None,
     ) -> None:
         super().__init__()
         self._gateway = gateway
@@ -138,6 +139,7 @@ class ClaimTaskWorker(QObject):
         self._client_service = client_service
         self._android_device_service = android_device_service
         self._cancelled = False
+        self._collect_service_factory = collect_service_factory
 
     def cancel(self) -> None:
         """阻止尚未开始的领取；已领取的任务仍必须保存到本地。"""
@@ -146,7 +148,6 @@ class ClaimTaskWorker(QObject):
 
     @pyqtSlot()
     def run(self) -> None:
-        claimed_task: Optional[AdminTask] = None
         try:
             if self._cancelled:
                 return
@@ -155,42 +156,20 @@ class ClaimTaskWorker(QObject):
                 self.failed.emit("请先在设置页保存当前设备号和设备名")
                 return
             android_serial = self._android_device_service.load()
-            if not android_serial:
-                self.failed.emit("请先在设置页选择并保存 Android 设备")
-                return
 
-            capabilities = ClaimCapabilities(
-                device=AndroidDeviceInfo(android_serial),
-                supported_types=(TaskType.COLLECT,),
-                purchase_mode="dry_run",
-                schema_versions=(1,),
-            )
-            claimed_task = self._gateway.claim_next(
+            result = CollectTaskService(
+                self._gateway,
+                self._task_repository,
                 ClientInfo(
                     client_settings.client_id,
                     client_settings.client_name,
                 ),
-                capabilities,
-            )
-            if claimed_task is None:
-                if not self._cancelled:
-                    self.noTask.emit()
-                return
-
-            local_task = admin_task_to_new_claimed_task(claimed_task)
-            try:
-                self._task_repository.add_claimed_task(local_task)
-            except DuplicateTaskError:
-                if not self._cancelled:
-                    self.duplicateTask.emit(claimed_task.task_id)
-                return
-            except Exception as exc:
-                if not self._cancelled:
-                    self.localSaveFailed.emit(claimed_task.task_id, str(exc))
-                return
-
-            if not self._cancelled:
-                self.taskSaved.emit(claimed_task.task_id)
+                android_serial or "",
+                cancelled=lambda: self._cancelled,
+                collect_service_factory=self._collect_service_factory,
+            ).execute_one()
+            if not self._cancelled or result.kind == "cancelled":
+                self.outcome.emit(result.kind, result.message, result.task_id)
         except AdminGatewayError as exc:
             if not self._cancelled:
                 request_hint = (
@@ -199,10 +178,7 @@ class ClaimTaskWorker(QObject):
                 self.failed.emit(f"{exc}{request_hint}")
         except Exception as exc:
             if not self._cancelled:
-                if claimed_task is not None:
-                    self.localSaveFailed.emit(claimed_task.task_id, str(exc))
-                else:
-                    self.failed.emit(f"领取任务失败：{exc}")
+                self.failed.emit(f"执行采集任务失败：{exc}")
         finally:
             self.completed.emit()
 
@@ -215,8 +191,9 @@ class PDDTaskPageEvent(QObject):
         page: PDDTaskPage,
         repository: Optional[TaskRepository] = None,
         parent=None,
-        claim_gateway: Optional[TaskClaimGateway] = None,
+        claim_gateway: Optional[AdminGateway] = None,
         settings_repository: Optional[SettingsRepository] = None,
+        collect_service_factory: Optional[CollectServiceFactory] = None,
     ):
         super().__init__(parent or page)
         self._page = page
@@ -226,6 +203,13 @@ class PDDTaskPageEvent(QObject):
         self._claim_busy = False
         self._claim_thread: Optional[QThread] = None
         self._claim_worker: Optional[ClaimTaskWorker] = None
+        self._collect_service_factory = collect_service_factory
+
+        try:
+            self._repository.recover_interrupted_work()
+        except AttributeError:
+            # 测试用的只读 Repository 可以不实现恢复接口。
+            pass
 
         settings = settings_repository or SettingsRepository()
         self._client_service = CurrentClientService(settings)
@@ -242,6 +226,7 @@ class PDDTaskPageEvent(QObject):
                 self._claim_gateway = HttpAdminGateway(
                     base_url if isinstance(base_url, str) else "",
                     timeout_seconds=timeout_seconds,
+                    client_id=self._client_service.load().client_id,
                 )
             except (TypeError, ValueError) as exc:
                 self._claim_gateway_error = str(exc)
@@ -290,7 +275,7 @@ class PDDTaskPageEvent(QObject):
 
         self._claim_busy = True
         self._page.autoFetchButton.setEnabled(False)
-        self._page.set_engine_status("正在领取一个采集任务，请稍候…")
+        self._page.set_engine_status("正在处理一条采集任务，请稍候…")
 
         thread = QThread(self)
         worker = ClaimTaskWorker(
@@ -298,6 +283,7 @@ class PDDTaskPageEvent(QObject):
             self._repository,
             self._client_service,
             self._selected_android_device_service,
+            self._collect_service_factory,
         )
         worker.moveToThread(thread)
         thread.started.connect(worker.run)
@@ -306,6 +292,7 @@ class PDDTaskPageEvent(QObject):
         worker.duplicateTask.connect(self._on_duplicate_claimed_task)
         worker.localSaveFailed.connect(self._on_claimed_task_save_failed)
         worker.failed.connect(self._on_claim_failed)
+        worker.outcome.connect(self._on_collect_outcome)
         worker.completed.connect(thread.quit)
         worker.completed.connect(worker.deleteLater)
         thread.finished.connect(thread.deleteLater)
@@ -352,6 +339,15 @@ class PDDTaskPageEvent(QObject):
             self._page.set_engine_status(content)
             self._show_claim_error("领取任务失败", content)
 
+    @pyqtSlot(str, str, str)
+    def _on_collect_outcome(self, kind: str, message: str, _task_id: str) -> None:
+        if self._closing:
+            return
+        self._page.set_engine_status(message)
+        self._reload()
+        if kind in {"result_pending", "manual_review", "failed"}:
+            self._show_claim_error("采集任务需要处理", message)
+
     def _show_claim_error(self, title: str, content: str) -> None:
         """显示不会自动消失的可恢复错误，同时保留底部状态文字。"""
 
@@ -413,6 +409,7 @@ class PDDTaskPageEvent(QObject):
                         self._on_claimed_task_save_failed,
                     ),
                     (worker.failed, self._on_claim_failed),
+                    (worker.outcome, self._on_collect_outcome),
                 )
             except RuntimeError:
                 signal_slots = ()
@@ -424,7 +421,10 @@ class PDDTaskPageEvent(QObject):
 
         if thread is not None and thread.isRunning():
             thread.quit()
-            thread.wait(10_000)
+            # uiautomator2/ADB 的单次调用可能需要数秒才返回。先通过
+            # cancelled 标志让采集在下一个安全点退出，再等待工作线程收尾，
+            # 避免窗口销毁时出现 "QThread destroyed while running"。
+            thread.wait(60_000)
 
 
 def summary_to_row(summary: TaskSummary) -> TaskRow:

client/src/task_models.py

@@ -184,6 +184,15 @@ class OutboxEventRecord:
     sent_at: Optional[str]
 
 
+@dataclass(frozen=True)
+class StartedTaskRun:
+    """已进入执行状态的一次采集尝试。"""
+
+    task: TaskDetail
+    attempt_id: str
+    attempt_no: int
+
+
 @dataclass(frozen=True)
 class AppSettingRecord:
     """一条非敏感应用设置。"""

client/src/task_repository.py

@@ -8,10 +8,16 @@ import sqlite3
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Dict, List, Optional, Tuple, Union
+from uuid import uuid4
 
 from .db import initialize_database, open_database
 from .task_models import (
     NewClaimedTask,
+    OutboxEventRecord,
+    OutboxEventType,
+    OutboxStatus,
+    RunStatus,
+    StartedTaskRun,
     TaskDetail,
     TaskFilters,
     TaskStatus,
@@ -145,6 +151,353 @@ class TaskRepository:
             connection.close()
         return self._to_detail(row) if row is not None else None
 
+    def recover_interrupted_work(self) -> None:
+        """恢复上次异常退出留下的可重试状态。"""
+
+        now = utc_now_iso()
+        connection = open_database(self._db_path)
+        try:
+            with connection:
+                connection.execute(
+                    "UPDATE outbox_events SET status = 'pending', updated_at = ?"
+                    " WHERE status = 'sending'",
+                    (now,),
+                )
+                connection.execute(
+                    "UPDATE pdd_tasks SET status = 'retry_wait',"
+                    " current_step = 'interrupted', retry_count = retry_count + 1,"
+                    " last_error_code = 'CLIENT_INTERRUPTED',"
+                    " last_error_message = '客户端上次执行期间退出', updated_at = ?"
+                    " WHERE status = 'running' AND task_type = 'collect'",
+                    (now,),
+                )
+                connection.execute(
+                    "UPDATE task_runs SET run_status = 'failed',"
+                    " error_code = 'CLIENT_INTERRUPTED',"
+                    " error_message = '客户端上次执行期间退出',"
+                    " finished_at = ?, updated_at = ?"
+                    " WHERE run_status = 'running' AND irreversible_action_at IS NULL"
+                    " AND task_id IN (SELECT id FROM pdd_tasks WHERE task_type = 'collect')",
+                    (now, now),
+                )
+        finally:
+            connection.close()
+
+    def next_collect_task(self) -> Optional[TaskDetail]:
+        """返回最早的本地待执行采集任务。"""
+
+        connection = open_database(self._db_path)
+        try:
+            row = connection.execute(
+                "SELECT * FROM pdd_tasks"
+                " WHERE task_type = 'collect' AND status IN ('claimed', 'retry_wait')"
+                " ORDER BY received_at ASC, id ASC LIMIT 1"
+            ).fetchone()
+        finally:
+            connection.close()
+        return self._to_detail(row) if row is not None else None
+
+    def start_collect_run(
+        self, remote_task_id: str, device_address: str
+    ) -> StartedTaskRun:
+        """原子地把待执行任务改为执行中，并创建一次运行记录。"""
+
+        now = utc_now_iso()
+        attempt_id = str(uuid4())
+        connection = open_database(self._db_path)
+        try:
+            with connection:
+                row = connection.execute(
+                    "SELECT * FROM pdd_tasks WHERE remote_task_id = ?",
+                    (remote_task_id,),
+                ).fetchone()
+                if row is None:
+                    raise ValueError(f"任务 {remote_task_id} 不存在")
+                if row["task_type"] != TaskType.COLLECT.value:
+                    raise ValueError("当前只能执行采集任务")
+                if row["status"] not in {
+                    TaskStatus.CLAIMED.value,
+                    TaskStatus.RETRY_WAIT.value,
+                }:
+                    raise ValueError(f"任务状态 {row['status']} 不能开始采集")
+                attempt_no = int(
+                    connection.execute(
+                        "SELECT COALESCE(MAX(attempt_no), 0) + 1"
+                        " FROM task_runs WHERE task_id = ?",
+                        (row["id"],),
+                    ).fetchone()[0]
+                )
+                connection.execute(
+                    "UPDATE pdd_tasks SET status = 'running',"
+                    " current_step = 'collecting', started_at = COALESCE(started_at, ?),"
+                    " last_error_code = NULL, last_error_message = NULL, updated_at = ?"
+                    " WHERE id = ?",
+                    (now, now, row["id"]),
+                )
+                connection.execute(
+                    "INSERT INTO task_runs (task_id, attempt_id, attempt_no,"
+                    " device_address, run_status, current_step, started_at,"
+                    " created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                    (
+                        row["id"], attempt_id, attempt_no, device_address,
+                        RunStatus.RUNNING.value, "collecting", now, now, now,
+                    ),
+                )
+            task = self.get_task(remote_task_id)
+            assert task is not None
+            return StartedTaskRun(task, attempt_id, attempt_no)
+        finally:
+            connection.close()
+
+    def save_collect_result(
+        self,
+        remote_task_id: str,
+        attempt_id: str,
+        pdd_data: Dict[str, object],
+    ) -> OutboxEventRecord:
+        """在一个事务中保存采集结果并创建待提交事件。"""
+
+        now = utc_now_iso()
+        connection = open_database(self._db_path)
+        try:
+            with connection:
+                task = connection.execute(
+                    "SELECT id, version FROM pdd_tasks WHERE remote_task_id = ?",
+                    (remote_task_id,),
+                ).fetchone()
+                if task is None:
+                    raise ValueError(f"任务 {remote_task_id} 不存在")
+                payload = {
+                    "task_version": task["version"],
+                    "attempt_id": attempt_id,
+                    "result_type": "collect",
+                    "completed_at": now,
+                    "pdd_data": pdd_data,
+                }
+                idempotency_key = f"{remote_task_id}:{attempt_id}:result-v1"
+                connection.execute(
+                    "UPDATE pdd_tasks SET status = 'result_pending',"
+                    " current_step = 'submit_result', pdd_data = ?, goods_id = ?,"
+                    " title = ?, price_cent = ?, finished_at = ?, updated_at = ?"
+                    " WHERE id = ?",
+                    (
+                        json.dumps(pdd_data, ensure_ascii=False),
+                        pdd_data.get("goods_id"), pdd_data.get("title"),
+                        self._summary_price(pdd_data), now, now, task["id"],
+                    ),
+                )
+                connection.execute(
+                    "UPDATE task_runs SET run_status = 'succeeded',"
+                    " current_step = 'submit_result', finished_at = ?, updated_at = ?"
+                    " WHERE attempt_id = ?",
+                    (now, now, attempt_id),
+                )
+                cursor = connection.execute(
+                    "INSERT INTO outbox_events (task_id, event_type, idempotency_key,"
+                    " payload_json, status, created_at, updated_at)"
+                    " VALUES (?, 'collect_result', ?, ?, 'pending', ?, ?)",
+                    (
+                        task["id"], idempotency_key,
+                        json.dumps(payload, ensure_ascii=False), now, now,
+                    ),
+                )
+                event_id = int(cursor.lastrowid)
+            event = self.get_outbox_event(event_id)
+            assert event is not None
+            return event
+        finally:
+            connection.close()
+
+    def save_collect_failure(
+        self,
+        remote_task_id: str,
+        attempt_id: str,
+        status: TaskStatus,
+        error_code: str,
+        error_message: str,
+        retryable: bool,
+        diagnostics: Optional[Dict[str, object]] = None,
+    ) -> OutboxEventRecord:
+        """保存结构化失败，并可靠排队提交 Admin。"""
+
+        if status not in {
+            TaskStatus.RETRY_WAIT, TaskStatus.MANUAL_REVIEW,
+            TaskStatus.FAILED, TaskStatus.CANCELLED,
+        }:
+            raise ValueError("失败状态无效")
+        now = utc_now_iso()
+        connection = open_database(self._db_path)
+        try:
+            with connection:
+                task = connection.execute(
+                    "SELECT id, version FROM pdd_tasks WHERE remote_task_id = ?",
+                    (remote_task_id,),
+                ).fetchone()
+                if task is None:
+                    raise ValueError(f"任务 {remote_task_id} 不存在")
+                payload = {
+                    "task_version": task["version"],
+                    "attempt_id": attempt_id,
+                    "status": status.value,
+                    "error": {
+                        "code": error_code,
+                        "message": error_message,
+                        "retryable": retryable,
+                        "step": "collecting",
+                    },
+                    "diagnostics": diagnostics or {"artifacts": []},
+                    "reported_at": now,
+                }
+                idempotency_key = f"{remote_task_id}:{attempt_id}:failure-v1"
+                connection.execute(
+                    "UPDATE pdd_tasks SET status = ?, current_step = 'failed',"
+                    " retry_count = retry_count + ?, last_error_code = ?,"
+                    " last_error_message = ?, finished_at = ?, updated_at = ?"
+                    " WHERE id = ?",
+                    (
+                        status.value, 1 if status is TaskStatus.RETRY_WAIT else 0,
+                        error_code, error_message, now, now, task["id"],
+                    ),
+                )
+                run_status = {
+                    TaskStatus.CANCELLED: RunStatus.CANCELLED,
+                    TaskStatus.MANUAL_REVIEW: RunStatus.MANUAL_REVIEW,
+                }.get(status, RunStatus.FAILED)
+                connection.execute(
+                    "UPDATE task_runs SET run_status = ?, current_step = 'failed',"
+                    " error_code = ?, error_message = ?, finished_at = ?, updated_at = ?"
+                    " WHERE attempt_id = ?",
+                    (run_status.value, error_code, error_message, now, now, attempt_id),
+                )
+                cursor = connection.execute(
+                    "INSERT INTO outbox_events (task_id, event_type, idempotency_key,"
+                    " payload_json, status, created_at, updated_at)"
+                    " VALUES (?, 'task_failure', ?, ?, 'pending', ?, ?)",
+                    (
+                        task["id"], idempotency_key,
+                        json.dumps(payload, ensure_ascii=False), now, now,
+                    ),
+                )
+                event_id = int(cursor.lastrowid)
+            event = self.get_outbox_event(event_id)
+            assert event is not None
+            return event
+        finally:
+            connection.close()
+
+    def next_pending_outbox(self) -> Optional[OutboxEventRecord]:
+        connection = open_database(self._db_path)
+        try:
+            row = connection.execute(
+                "SELECT o.* FROM outbox_events o JOIN pdd_tasks t ON t.id = o.task_id"
+                " WHERE o.status = 'pending'"
+                " AND (o.next_retry_at IS NULL OR o.next_retry_at <= ?)"
+                " ORDER BY o.id ASC LIMIT 1",
+                (utc_now_iso(),),
+            ).fetchone()
+        finally:
+            connection.close()
+        return self._to_outbox(row) if row is not None else None
+
+    def get_outbox_event(self, event_id: int) -> Optional[OutboxEventRecord]:
+        connection = open_database(self._db_path)
+        try:
+            row = connection.execute(
+                "SELECT * FROM outbox_events WHERE id = ?", (event_id,)
+            ).fetchone()
+        finally:
+            connection.close()
+        return self._to_outbox(row) if row is not None else None
+
+    def outbox_task_id(self, event_id: int) -> str:
+        connection = open_database(self._db_path)
+        try:
+            row = connection.execute(
+                "SELECT t.remote_task_id FROM outbox_events o"
+                " JOIN pdd_tasks t ON t.id = o.task_id WHERE o.id = ?",
+                (event_id,),
+            ).fetchone()
+        finally:
+            connection.close()
+        if row is None:
+            raise ValueError(f"Outbox {event_id} 不存在")
+        return str(row[0])
+
+    def mark_outbox_sending(self, event_id: int) -> None:
+        self._update_outbox(event_id, "sending", None)
+
+    def mark_outbox_retry(self, event_id: int, message: str) -> None:
+        self._update_outbox(event_id, "pending", message, increment=True)
+
+    def mark_outbox_failed(self, event_id: int, message: str) -> None:
+        self._update_outbox(event_id, "failed", message, increment=True)
+
+    def mark_outbox_sent(self, event_id: int) -> None:
+        now = utc_now_iso()
+        connection = open_database(self._db_path)
+        try:
+            with connection:
+                row = connection.execute(
+                    "SELECT task_id, event_type FROM outbox_events WHERE id = ?",
+                    (event_id,),
+                ).fetchone()
+                if row is None:
+                    raise ValueError(f"Outbox {event_id} 不存在")
+                connection.execute(
+                    "UPDATE outbox_events SET status = 'sent', sent_at = ?,"
+                    " updated_at = ?, attempt_count = attempt_count + 1"
+                    " WHERE id = ?",
+                    (now, now, event_id),
+                )
+                if row["event_type"] == OutboxEventType.COLLECT_RESULT.value:
+                    connection.execute(
+                        "UPDATE pdd_tasks SET status = 'succeeded',"
+                        " current_step = 'completed', updated_at = ? WHERE id = ?",
+                        (now, row["task_id"]),
+                    )
+        finally:
+            connection.close()
+
+    def _update_outbox(
+        self, event_id: int, status: str, message: Optional[str], increment: bool = False
+    ) -> None:
+        connection = open_database(self._db_path)
+        try:
+            with connection:
+                cursor = connection.execute(
+                    "UPDATE outbox_events SET status = ?, last_error = ?, updated_at = ?,"
+                    f" attempt_count = attempt_count + {1 if increment else 0} WHERE id = ?",
+                    (status, message, utc_now_iso(), event_id),
+                )
+                if cursor.rowcount != 1:
+                    raise ValueError(f"Outbox {event_id} 不存在")
+        finally:
+            connection.close()
+
+    @staticmethod
+    def _summary_price(pdd_data: Dict[str, object]) -> Optional[int]:
+        skus = pdd_data.get("skus")
+        if not isinstance(skus, list):
+            return None
+        prices = [
+            item.get("price_cent") for item in skus
+            if isinstance(item, dict) and isinstance(item.get("price_cent"), int)
+        ]
+        return min(prices) if prices else None
+
+    @staticmethod
+    def _to_outbox(row: sqlite3.Row) -> OutboxEventRecord:
+        return OutboxEventRecord(
+            id=row["id"], task_id=row["task_id"],
+            event_type=OutboxEventType(row["event_type"]),
+            idempotency_key=row["idempotency_key"],
+            payload_json=TaskRepository._load_json_object(row["payload_json"]),
+            status=OutboxStatus(row["status"]), attempt_count=row["attempt_count"],
+            next_retry_at=row["next_retry_at"], last_error=row["last_error"],
+            created_at=row["created_at"], updated_at=row["updated_at"],
+            sent_at=row["sent_at"],
+        )
+
     @staticmethod
     def _validate_page(limit: int, offset: int) -> None:
         if not 1 <= limit <= MAX_PAGE_SIZE: