feat: 安全领取并分派采购任务 (#72)

2026-08-10 00:05:40 +08:00
parent 460619392f
commit f201bcc4bf
12 changed files with 789 additions and 76 deletions
@@ -47,10 +47,7 @@ class ClaimCapabilities:
    """Client 领取任务时声明的设备与执行能力。"""

    device: Optional[AndroidDeviceInfo] = None
-    supported_types: Tuple[TaskType, ...] = (
-        TaskType.COLLECT,
-        TaskType.PURCHASE,
-    )
+    supported_types: Tuple[TaskType, ...] = (TaskType.COLLECT,)
    purchase_mode: str = "dry_run"
    schema_versions: Tuple[int, ...] = (1,)

@@ -147,16 +147,17 @@ class HttpAdminGateway(AdminGateway):
        client: ClientInfo,
        capabilities: ClaimCapabilities,
    ) -> Optional[AdminTask]:
-        """领取一个采集任务；Admin 返回 204 时返回 ``None``。"""
+        """按调用方已安全确认的能力领取一个任务。"""

        request_id = str(uuid4())
        self._client_id = client.client_id.strip()
        payload = {
            "client": {"name": client.name.strip()},
-            # #30 只允许领取采集任务。采购能力必须由安全门禁工单开启。
-            "supported_types": [TaskType.COLLECT.value],
+            "supported_types": [
+                task_type.value for task_type in capabilities.supported_types
+            ],
            "capabilities": {
-                "purchase_mode": "dry_run",
+                "purchase_mode": capabilities.purchase_mode,
                "schema_versions": list(capabilities.schema_versions),
            },
        }
@@ -394,6 +395,8 @@ class HttpAdminGateway(AdminGateway):
                request_id,
            ) from exc

+        cls._validate_claim_payload(task_type, task_payload, request_id)
+
        return AdminTask(
            task_id=task_id,
            task_type=task_type,
@@ -404,6 +407,61 @@ class HttpAdminGateway(AdminGateway):
            updated_at=updated_at,
        )

+    @staticmethod
+    def _validate_claim_payload(
+        task_type: TaskType,
+        payload: Mapping[str, object],
+        request_id: str,
+    ) -> None:
+        """在采购任务进入本地执行前校验全部安全字段。"""
+
+        goods_url = payload.get("goods_url")
+        if not isinstance(goods_url, str) or not goods_url.strip():
+            raise AdminGatewayError(
+                "ADMIN_INVALID_RESPONSE",
+                "Admin 任务 payload.goods_url 不能为空",
+                False,
+                request_id,
+            )
+        if task_type is not TaskType.PURCHASE:
+            return
+        goods_id = payload.get("goods_id")
+        options = payload.get("options")
+        quantity = payload.get("quantity")
+        max_price_cent = payload.get("max_price_cent")
+        valid_options = (
+            isinstance(options, Mapping)
+            and bool(options)
+            and all(
+                isinstance(key, str)
+                and bool(key.strip())
+                and isinstance(value, str)
+                and bool(value.strip())
+                for key, value in options.items()
+            )
+        )
+        valid = (
+            isinstance(goods_id, str)
+            and bool(goods_id.strip())
+            and valid_options
+            and isinstance(quantity, int)
+            and not isinstance(quantity, bool)
+            and quantity > 0
+            and isinstance(max_price_cent, int)
+            and not isinstance(max_price_cent, bool)
+            and max_price_cent > 0
+        )
+        if not valid:
+            raise AdminGatewayError(
+                "ADMIN_INVALID_RESPONSE",
+                (
+                    "Admin 采购任务缺少有效的 goods_id、动态 options、"
+                    "quantity 或 max_price_cent"
+                ),
+                False,
+                request_id,
+            )
+
    @staticmethod
    def _decode_json(body: bytes, request_id: str) -> dict:
        try:
@@ -34,7 +34,6 @@ from qfluentwidgets import InfoBar, InfoBarPosition, MessageBox

 from .admin_gateway import (
    AdminGatewayError,
-    AdminTask,
    ClientInfo,
    AdminGateway,
 )
@@ -42,16 +41,20 @@ 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
+from .purchase_task_service import PurchaseAdapterFactory
 from .selected_android_device_service import SelectedAndroidDeviceService
 from .settings_repository import SettingsRepository
 from .task_models import (
-    NewClaimedTask,
    TaskFilters,
    TaskStatus,
    TaskSummary,
    TaskType,
 )
 from .task_repository import CollectRerunError, TaskRepository
+from .task_dispatcher import (
+    TaskDispatcher,
+    admin_task_to_new_claimed_task,
+)
 from .task_detail_view import TaskDetailWindow


@@ -88,42 +91,8 @@ TASK_STATUS_TEXT = {
 }


-def admin_task_to_new_claimed_task(task: AdminTask) -> NewClaimedTask:
-    """把 Admin 任务显式映射为本地任务，避免字段名自动展开出错。"""
-
-    if task.task_type is not TaskType.COLLECT:
-        raise ValueError(f"任务 {task.task_id} 不是采集任务")
-
-    payload = dict(task.payload)
-    goods_url = payload.get("goods_url")
-    goods_id = payload.get("goods_id")
-    if not isinstance(goods_url, str) or not goods_url.strip():
-        raise ValueError("payload.goods_url 不能为空")
-    if goods_id is not None and not isinstance(goods_id, str):
-        raise ValueError("payload.goods_id 必须是文本")
-
-    original_task = {
-        "id": task.task_id,
-        "type": task.task_type.value,
-        "version": task.version,
-        "priority": task.priority,
-        "payload": payload,
-        "created_at": task.created_at,
-        "updated_at": task.updated_at,
-    }
-    return NewClaimedTask(
-        remote_task_id=task.task_id,
-        task_type=task.task_type,
-        goods_url=goods_url,
-        goods_id=goods_id.strip() if isinstance(goods_id, str) else None,
-        priority=task.priority,
-        version=task.version,
-        admin_payload=original_task,
-    )
-
-
 class ClaimTaskWorker(QObject):
-    """在后台补交或执行至多一条采集任务。"""
+    """在后台补交、领取或执行至多一条任务。"""

    noTask = pyqtSignal()
    taskSaved = pyqtSignal(str)
@@ -141,6 +110,7 @@ class ClaimTaskWorker(QObject):
        client_service: CurrentClientService,
        android_device_service: SelectedAndroidDeviceService,
        collect_service_factory: Optional[CollectServiceFactory] = None,
+        purchase_adapter_factory: Optional[PurchaseAdapterFactory] = None,
        selected_task_id: str = "",
    ) -> None:
        super().__init__()
@@ -150,6 +120,7 @@ class ClaimTaskWorker(QObject):
        self._android_device_service = android_device_service
        self._cancelled = False
        self._collect_service_factory = collect_service_factory
+        self._purchase_adapter_factory = purchase_adapter_factory
        self._selected_task_id = selected_task_id

    def cancel(self) -> None:
@@ -168,22 +139,31 @@ class ClaimTaskWorker(QObject):
                return
            android_serial = self._android_device_service.load()

+            client = ClientInfo(
+                client_settings.client_id,
+                client_settings.client_name,
+            )
+            if self._selected_task_id:
                service = CollectTaskService(
                    self._gateway,
                    self._task_repository,
-                ClientInfo(
-                    client_settings.client_id,
-                    client_settings.client_name,
-                ),
+                    client,
                    android_serial or "",
                    cancelled=lambda: self._cancelled,
                    collect_service_factory=self._collect_service_factory,
                )
-            result = (
-                service.execute_selected(self._selected_task_id)
-                if self._selected_task_id
-                else service.execute_one()
+                result = service.execute_selected(self._selected_task_id)
+            else:
+                dispatcher = TaskDispatcher(
+                    self._gateway,
+                    self._task_repository,
+                    client,
+                    android_serial or "",
+                    collect_service_factory=self._collect_service_factory,
+                    purchase_adapter_factory=self._purchase_adapter_factory,
+                    cancelled=lambda: self._cancelled,
                )
+                result = dispatcher.execute_one()
            if not self._cancelled or result.kind == "cancelled":
                self.outcome.emit(result.kind, result.message, result.task_id)
        except AdminGatewayError as exc:
@@ -198,7 +178,7 @@ class ClaimTaskWorker(QObject):
                    self.failed.emit(message)
        except Exception as exc:
            if not self._cancelled:
-                self.failed.emit(f"执行采集任务失败：{exc}")
+                self.failed.emit(f"执行任务失败：{exc}")
        finally:
            self.completed.emit()

@@ -214,6 +194,7 @@ class PDDTaskPageEvent(QObject):
        claim_gateway: Optional[AdminGateway] = None,
        settings_repository: Optional[SettingsRepository] = None,
        collect_service_factory: Optional[CollectServiceFactory] = None,
+        purchase_adapter_factory: Optional[PurchaseAdapterFactory] = None,
        next_task_delay_ms: int = 500,
        no_task_delay_ms: int = 5_000,
        retry_delays_ms: tuple[int, ...] = (5_000, 10_000, 20_000, 30_000),
@@ -233,6 +214,7 @@ class PDDTaskPageEvent(QObject):
        self._retry_count = 0
        self._detail_windows: Dict[str, TaskDetailWindow] = {}
        self._collect_service_factory = collect_service_factory
+        self._purchase_adapter_factory = purchase_adapter_factory
        if next_task_delay_ms < 0 or no_task_delay_ms <= 0:
            raise ValueError("自动获取等待时间配置无效")
        if not retry_delays_ms or any(value <= 0 for value in retry_delays_ms):
@@ -476,15 +458,16 @@ class PDDTaskPageEvent(QObject):
        self._claim_busy = True
        self._cycle_next_delay_ms = None
        self._page.set_auto_fetch_state("running")
-        self._page.set_engine_status("自动获取：运行中 · 正在处理一条采集任务…")
+        self._page.set_engine_status("自动获取：运行中 · 正在处理一条任务…")

        thread = QThread(self)
        worker = ClaimTaskWorker(
-            self._claim_gateway,
-            self._repository,
-            self._client_service,
-            self._selected_android_device_service,
-            self._collect_service_factory,
+            gateway=self._claim_gateway,
+            task_repository=self._repository,
+            client_service=self._client_service,
+            android_device_service=self._selected_android_device_service,
+            collect_service_factory=self._collect_service_factory,
+            purchase_adapter_factory=self._purchase_adapter_factory,
        )
        worker.moveToThread(thread)
        thread.started.connect(worker.run)
@@ -630,7 +613,7 @@ class PDDTaskPageEvent(QObject):
            self._show_retry_paused(content)
            self._stop_after_current(content)
        elif kind in {"manual_review", "failed"}:
-            self._show_claim_error("采集任务需要处理", message)
+            self._show_claim_error("任务需要处理", message)
            self._stop_after_current(message)
        elif kind == "cancelled":
            self._stop_after_current(message or "自动获取已停止")
@@ -648,7 +631,11 @@ class PDDTaskPageEvent(QObject):
            task = self._repository.get_task(task_id)
        except Exception:
            return False
-        return task is not None and task.status is TaskStatus.RETRY_WAIT
+        return (
+            task is not None
+            and task.task_type is TaskType.COLLECT
+            and task.status is TaskStatus.RETRY_WAIT
+        )

    def _show_retry_paused(self, content: str) -> None:
        """用持久警告说明任务不会自行倒计时重试。"""
@@ -0,0 +1,272 @@
+"""串行领取、落库并按任务类型分派一条任务。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Callable, Mapping, Optional
+
+from .admin_gateway import (
+    AdminGateway,
+    AdminGatewayError,
+    AdminTask,
+    AndroidDeviceInfo,
+    ClaimCapabilities,
+    ClientInfo,
+)
+from .collect_task_service import CollectServiceFactory, CollectTaskService
+from .purchase_task_service import (
+    PurchaseAdapterFactory,
+    PurchaseTaskService,
+)
+from .task_models import (
+    NewClaimedTask,
+    OutboxEventRecord,
+    OutboxEventType,
+    TaskType,
+)
+from .task_repository import DuplicateTaskError, TaskRepository
+
+
+@dataclass(frozen=True)
+class TaskDispatchOutcome:
+    """一轮串行任务处理的结果。"""
+
+    kind: str
+    message: str
+    task_id: str = ""
+
+
+def admin_task_to_new_claimed_task(task: AdminTask) -> NewClaimedTask:
+    """严格校验 Admin 任务并映射为本地已领取任务。"""
+
+    payload = dict(task.payload)
+    goods_url = payload.get("goods_url")
+    if not isinstance(goods_url, str) or not goods_url.strip():
+        raise ValueError("payload.goods_url 不能为空")
+    goods_id = payload.get("goods_id")
+    if goods_id is not None and not isinstance(goods_id, str):
+        raise ValueError("payload.goods_id 必须是文本")
+
+    target_color = None
+    target_size = None
+    price_cent = None
+    quantity = None
+    if task.task_type is TaskType.PURCHASE:
+        if not isinstance(goods_id, str) or not goods_id.strip():
+            raise ValueError("采购任务 payload.goods_id 不能为空")
+        options = payload.get("options")
+        if not isinstance(options, Mapping) or not options:
+            raise ValueError("采购任务 payload.options 必须是非空对象")
+        normalized_options: dict[str, str] = {}
+        for key, value in options.items():
+            if not isinstance(key, str) or not key.strip():
+                raise ValueError("采购任务 options 的名称不能为空")
+            if not isinstance(value, str) or not value.strip():
+                raise ValueError("采购任务 options 的值必须是非空文本")
+            normalized_options[key.strip()] = value.strip()
+        payload["options"] = normalized_options
+        quantity = payload.get("quantity")
+        price_cent = payload.get("max_price_cent")
+        if (
+            isinstance(quantity, bool)
+            or not isinstance(quantity, int)
+            or quantity <= 0
+        ):
+            raise ValueError("采购任务 payload.quantity 必须是大于 0 的整数")
+        if (
+            isinstance(price_cent, bool)
+            or not isinstance(price_cent, int)
+            or price_cent <= 0
+        ):
+            raise ValueError(
+                "采购任务 payload.max_price_cent 必须是大于 0 的整数分"
+            )
+        target_color = normalized_options.get("color")
+        target_size = normalized_options.get("size")
+
+    original_task = {
+        "id": task.task_id,
+        "type": task.task_type.value,
+        "version": task.version,
+        "priority": task.priority,
+        "payload": payload,
+        "created_at": task.created_at,
+        "updated_at": task.updated_at,
+    }
+    return NewClaimedTask(
+        remote_task_id=task.task_id,
+        task_type=task.task_type,
+        goods_url=goods_url.strip(),
+        goods_id=goods_id.strip() if isinstance(goods_id, str) else None,
+        target_color=target_color,
+        target_size=target_size,
+        price_cent=price_cent,
+        quantity=quantity,
+        priority=task.priority,
+        version=task.version,
+        admin_payload=original_task,
+    )
+
+
+class TaskDispatcher:
+    """一次只补交、执行或领取并执行一条任务。"""
+
+    def __init__(
+        self,
+        gateway: AdminGateway,
+        repository: TaskRepository,
+        client: ClientInfo,
+        device_address: str,
+        *,
+        collect_service_factory: Optional[CollectServiceFactory] = None,
+        purchase_adapter_factory: Optional[PurchaseAdapterFactory] = None,
+        cancelled: Callable[[], bool] = lambda: False,
+    ) -> None:
+        self._gateway = gateway
+        self._repository = repository
+        self._client = client
+        self._device_address = str(device_address or "").strip()
+        self._collect_factory = collect_service_factory
+        self._purchase_factory = purchase_adapter_factory
+        self._cancelled = cancelled
+
+    @property
+    def purchase_ready(self) -> bool:
+        """只有显式提供采购演练适配器时才声明采购能力。"""
+
+        return self._purchase_factory is not None and bool(
+            self._device_address
+        )
+
+    def claim_capabilities(self) -> ClaimCapabilities:
+        """集中构造不会意外开放 live 的领取能力。"""
+
+        supported = [TaskType.COLLECT]
+        if self.purchase_ready:
+            supported.append(TaskType.PURCHASE)
+        device = (
+            AndroidDeviceInfo(self._device_address)
+            if self._device_address
+            else None
+        )
+        return ClaimCapabilities(
+            device=device,
+            supported_types=tuple(supported),
+            purchase_mode="dry_run",
+            schema_versions=(1,),
+        )
+
+    def execute_one(self) -> TaskDispatchOutcome:
+        """严格按 Outbox、本地任务、Admin 新任务的顺序处理。"""
+
+        pending = self._repository.next_pending_outbox()
+        if pending is not None:
+            return self._submit_pending(pending)
+        if not self._device_address:
+            raise ValueError("请先在设置页选择并保存 Android 设备")
+
+        if not self.purchase_ready:
+            pending_purchase = self._repository.next_purchase_task()
+            if pending_purchase is not None:
+                raise RuntimeError(
+                    f"本地采购任务 {pending_purchase.remote_task_id} 等待执行，"
+                    "但采购演练执行器未就绪；已停止领取新任务"
+                )
+
+        task = self._repository.next_runnable_task(
+            include_purchase=self.purchase_ready
+        )
+        if task is None:
+            remote = self._gateway.claim_next(
+                self._client, self.claim_capabilities()
+            )
+            if remote is None:
+                return TaskDispatchOutcome("no_task", "暂无可领取的任务")
+            try:
+                self._repository.add_claimed_task(
+                    admin_task_to_new_claimed_task(remote)
+                )
+            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 task.task_type is TaskType.COLLECT:
+            service = CollectTaskService(
+                self._gateway,
+                self._repository,
+                self._client,
+                self._device_address,
+                cancelled=self._cancelled,
+                collect_service_factory=self._collect_factory,
+            )
+            outcome = service.execute_selected(task.remote_task_id)
+        else:
+            if self._purchase_factory is None:
+                raise RuntimeError(
+                    "采购演练执行器未就绪，已停止领取以避免任务卡住"
+                )
+            purchase_service = PurchaseTaskService(
+                self._gateway,
+                self._repository,
+                self._client,
+                self._device_address,
+                self._purchase_factory,
+                cancelled=self._cancelled,
+            )
+            outcome = purchase_service.execute_selected(task.remote_task_id)
+        return TaskDispatchOutcome(
+            outcome.kind, outcome.message, outcome.task_id
+        )
+
+    def _submit_pending(
+        self, event: OutboxEventRecord
+    ) -> TaskDispatchOutcome:
+        """只补交已落库事件，不重新访问 PDD 页面。"""
+
+        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:
+            if exc.retryable:
+                self._repository.mark_outbox_retry(event.id, str(exc))
+                return TaskDispatchOutcome(
+                    "result_pending",
+                    f"任务 {task_id} 结果已保存在本地，等待提交 Admin：{exc}",
+                    task_id,
+                )
+            self._repository.mark_outbox_failed(event.id, str(exc))
+            return TaskDispatchOutcome(
+                "manual_review",
+                f"任务 {task_id} 结果被 Admin 拒绝：{exc}",
+                task_id,
+            )
+        self._repository.mark_outbox_sent(event.id)
+        if event.event_type is OutboxEventType.TASK_FAILURE:
+            return TaskDispatchOutcome(
+                "failed", f"任务 {task_id} 失败信息已提交 Admin", task_id
+            )
+        return TaskDispatchOutcome(
+            "succeeded", f"任务 {task_id} 结果已提交 Admin", task_id
+        )
@@ -224,6 +224,32 @@ class TaskRepository:
            connection.close()
        return self._to_detail(row) if row is not None else None

+    def next_runnable_task(
+        self, *, include_purchase: bool
+    ) -> Optional[TaskDetail]:
+        """按领取顺序返回当前执行器能够处理的最早本地任务。"""
+
+        if include_purchase:
+            where = (
+                "((task_type = 'collect'"
+                " AND status IN ('claimed', 'retry_wait'))"
+                " OR (task_type = 'purchase' AND status = 'claimed'))"
+            )
+        else:
+            where = (
+                "task_type = 'collect'"
+                " AND status IN ('claimed', 'retry_wait')"
+            )
+        connection = open_database(self._db_path)
+        try:
+            row = connection.execute(
+                f"SELECT * FROM pdd_tasks WHERE {where}"
+                " 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 validate_collect_rerun(self, remote_task_id: str) -> TaskDetail:
        """校验任务能否重新采集，成功时返回任务详情。"""

@@ -19,7 +19,8 @@ class MockAdminGatewayContractTest(unittest.TestCase):
        self.gateway = MockAdminGateway()
        self.client = ClientInfo("client-001")
        self.all_capabilities = ClaimCapabilities(
-            device=AndroidDeviceInfo("192.168.0.173:5555")
+            device=AndroidDeviceInfo("192.168.0.173:5555"),
+            supported_types=(TaskType.COLLECT, TaskType.PURCHASE),
        )

    @staticmethod
@@ -14,6 +14,7 @@ from src.admin_gateway import (
    ClientInfo,
 )
 from src.http_admin_gateway import HttpAdminGateway
+from src.task_models import TaskType


 class FakeResponse:
@@ -88,7 +89,7 @@ class HttpAdminGatewayTest(unittest.TestCase):
        self.assertEqual(headers["authorization"], "Bearer secret-token")
        body = json.loads(opener.request.data.decode("utf-8"))
        self.assertEqual(body["client"]["name"], "办公室电脑")
-        self.assertEqual(body["supported_types"], ["collect", "purchase"])
+        self.assertEqual(body["supported_types"], ["collect"])
        self.assertEqual(body["device"]["platform"], "android")
        self.assertEqual(body["capabilities"]["purchase_mode"], "dry_run")
        self.assertEqual(opener.timeout, 2.5)
@@ -218,7 +219,7 @@ class HttpAdminGatewayTest(unittest.TestCase):
        self.assertTrue(raised.exception.retryable)
        self.assertIn("可能已接收", str(raised.exception))

-    def test_claim_maps_real_admin_payload_and_only_reports_collect(self):
+    def test_claim_maps_payload_and_serializes_confirmed_capabilities(self):
        opener = RecordingOpener(
            FakeResponse(
                200,
@@ -267,6 +268,56 @@ class HttpAdminGatewayTest(unittest.TestCase):
        self.assertEqual(body["supported_types"], ["collect"])
        self.assertEqual(body["capabilities"]["purchase_mode"], "dry_run")

+    def test_claim_purchase_requires_all_safety_fields(self):
+        valid_task = {
+            "id": "PUR-001",
+            "type": "purchase",
+            "version": 1,
+            "priority": 0,
+            "payload": {
+                "goods_url": "https://example.test/goods/PUR-001",
+                "goods_id": "737116531267",
+                "options": {"color": "黑色", "size": "L"},
+                "quantity": 2,
+                "max_price_cent": 4200,
+            },
+            "created_at": "2026-08-09T08:00:00Z",
+            "updated_at": "2026-08-09T08:00:00Z",
+        }
+        gateway = HttpAdminGateway(
+            opener=RecordingOpener(FakeResponse(200, {"task": valid_task}))
+        )
+
+        task = gateway.claim_next(
+            ClientInfo("CLIENT-001"),
+            ClaimCapabilities(
+                supported_types=(TaskType.COLLECT, TaskType.PURCHASE),
+                purchase_mode="dry_run",
+            ),
+        )
+
+        self.assertEqual(task.task_type, TaskType.PURCHASE)
+        self.assertEqual(task.payload["max_price_cent"], 4200)
+
+        for missing in ("goods_id", "options", "quantity", "max_price_cent"):
+            with self.subTest(missing=missing):
+                invalid_task = dict(valid_task)
+                invalid_payload = dict(valid_task["payload"])
+                invalid_payload.pop(missing)
+                invalid_task["payload"] = invalid_payload
+                invalid_gateway = HttpAdminGateway(
+                    opener=RecordingOpener(
+                        FakeResponse(200, {"task": invalid_task})
+                    )
+                )
+                with self.assertRaises(AdminGatewayError) as raised:
+                    invalid_gateway.claim_next(
+                        ClientInfo("CLIENT-001"), self._capabilities()
+                    )
+                self.assertEqual(
+                    raised.exception.code, "ADMIN_INVALID_RESPONSE"
+                )
+
    def test_claim_204_returns_none(self):
        gateway = HttpAdminGateway(opener=RecordingOpener(FakeResponse(204, {})))

@@ -20,6 +20,7 @@ from src.pdd_ui_event import (
    summary_to_row,
 )
 from src.mock_admin_gateway import MockAdminGateway
+from src.pdd_purchase_adapter import PddPurchaseAdapter, PurchasePageState
 from src.settings_repository import SettingsRepository
 from src.task_models import NewClaimedTask, TaskStatus, TaskSummary, TaskType
 from src.task_repository import TaskRepository
@@ -45,6 +46,12 @@ class BrokenSaveRepository(BrokenRepository):
    def next_collect_task(self):
        return None

+    def next_runnable_task(self, *, include_purchase):
+        return None
+
+    def next_purchase_task(self):
+        return None
+

 class RecordingClaimGateway:
    """记录领取参数并返回预设结果。"""
@@ -148,6 +155,63 @@ def collect_admin_task(task_id="COL-001"):
    )


+def purchase_admin_task(task_id="PUR-001"):
+    return AdminTask(
+        task_id=task_id,
+        task_type=TaskType.PURCHASE,
+        version=1,
+        priority=10,
+        payload={
+            "goods_id": "737116531267",
+            "goods_url": "https://example.test/737116531267",
+            "options": {"color": "黑色", "size": "L"},
+            "quantity": 2,
+            "max_price_cent": 1200,
+        },
+        created_at="2026-08-09T08:00:00Z",
+        updated_at="2026-08-09T08:00:00Z",
+    )
+
+
+class FakePurchaseAdapter(PddPurchaseAdapter):
+    def __init__(self):
+        self.options = {}
+        self.quantity = 0
+        self.page_kind = "goods"
+
+    def open_goods(self, _goods_url):
+        pass
+
+    def read_state(self):
+        return PurchasePageState(
+            self.page_kind,
+            "737116531267",
+            dict(self.options),
+            self.quantity,
+            990,
+            1,
+        )
+
+    def select_options(self, options):
+        self.options = dict(options)
+
+    def set_quantity(self, quantity):
+        self.quantity = quantity
+
+    def enter_confirmation(self):
+        self.page_kind = "order_confirmation"
+
+    def stop_before_submit(self):
+        pass
+
+    def close(self):
+        pass
+
+
+def fake_purchase_factory(_address, _cancelled):
+    return FakePurchaseAdapter()
+
+
 def wait_until(application, predicate, timeout=3.0):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
@@ -442,6 +506,35 @@ class PDDTaskPageEventTest(unittest.TestCase):
        events.shutdown()
        page.deleteLater()

+    def test_ready_runtime_claims_and_dispatches_purchase_in_worker(self):
+        gateway = RecordingClaimGateway(purchase_admin_task())
+        page = PDDTaskPage()
+        events = PDDTaskPageEvent(
+            page,
+            self.repository,
+            claim_gateway=gateway,
+            settings_repository=self._saved_settings(),
+            collect_service_factory=fake_collect_factory,
+            purchase_adapter_factory=fake_purchase_factory,
+        )
+
+        page.autoFetchRequested.emit()
+
+        self.assertTrue(wait_until(self.app, lambda: not events._claim_busy))
+        _, capabilities = gateway.calls[0]
+        self.assertEqual(
+            capabilities.supported_types,
+            (TaskType.COLLECT, TaskType.PURCHASE),
+        )
+        self.assertEqual(capabilities.purchase_mode, "dry_run")
+        detail = self.repository.get_task("PUR-001")
+        assert detail is not None
+        self.assertEqual(detail.task_type, TaskType.PURCHASE)
+        self.assertEqual(detail.status, TaskStatus.SUCCEEDED)
+        self.assertIn("演练完成", page.statusLabel.text())
+        events.shutdown()
+        page.deleteLater()
+
    def test_no_task_schedules_next_claim_and_stop_cancels_timer(self):
        gateway = RecordingClaimGateway(None)
        page = PDDTaskPage()
@@ -0,0 +1,215 @@
+"""统一任务分派器测试；不连接真实 Admin 或手机。"""
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from src.admin_gateway import AdminTask, ClientInfo
+from src.mock_admin_gateway import MockAdminGateway
+from src.pdd_purchase_adapter import PddPurchaseAdapter, PurchasePageState
+from src.task_dispatcher import TaskDispatcher, admin_task_to_new_claimed_task
+from src.task_models import TaskStatus, TaskType
+from src.task_repository import TaskRepository
+
+
+class FakeCollectResult:
+    def to_pdd_data(self):
+        return {
+            "schema_version": 1,
+            "goods_id": "COL-GOODS",
+            "title": "采集测试商品",
+            "price_granularity": "color",
+            "dimensions": [],
+            "skus": [],
+        }
+
+
+class RecordingCollector:
+    def __init__(self, calls):
+        self.calls = calls
+
+    def collect(self, task):
+        self.calls.append(("collect", task.remote_task_id))
+        return FakeCollectResult()
+
+
+class ReadyPurchaseAdapter(PddPurchaseAdapter):
+    def __init__(self, calls):
+        self.calls = calls
+        self.options = {}
+        self.quantity = 0
+        self.page_kind = "goods"
+
+    def open_goods(self, goods_url):
+        self.calls.append(("purchase", "open", goods_url))
+
+    def read_state(self):
+        return PurchasePageState(
+            self.page_kind,
+            "PUR-GOODS",
+            dict(self.options),
+            self.quantity,
+            990,
+            1,
+        )
+
+    def select_options(self, options):
+        self.options = dict(options)
+
+    def set_quantity(self, quantity):
+        self.quantity = quantity
+
+    def enter_confirmation(self):
+        self.page_kind = "order_confirmation"
+
+    def stop_before_submit(self):
+        self.calls.append(("purchase", "stopped"))
+
+    def close(self):
+        pass
+
+
+def purchase_task(task_id="PUR-001"):
+    return AdminTask(
+        task_id=task_id,
+        task_type=TaskType.PURCHASE,
+        version=1,
+        priority=10,
+        payload={
+            "goods_url": "https://example.test/PUR-GOODS",
+            "goods_id": "PUR-GOODS",
+            "options": {
+                "color": "黑色",
+                "size": "L",
+                "bundle": "标准版",
+            },
+            "quantity": 2,
+            "max_price_cent": 1200,
+        },
+    )
+
+
+class TaskDispatcherTest(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.repository = TaskRepository(
+            Path(self.temporary.name) / "client.db"
+        )
+        self.gateway = MockAdminGateway()
+        self.client = ClientInfo("CLIENT-001", "测试客户端")
+        self.calls = []
+
+    def tearDown(self):
+        self.temporary.cleanup()
+
+    def _dispatcher(self, *, purchase_ready):
+        purchase_factory = None
+        if purchase_ready:
+            purchase_factory = (
+                lambda _address, _cancelled: ReadyPurchaseAdapter(self.calls)
+            )
+        return TaskDispatcher(
+            self.gateway,
+            self.repository,
+            self.client,
+            "USB-001",
+            collect_service_factory=(
+                lambda *_args: RecordingCollector(self.calls)
+            ),
+            purchase_adapter_factory=purchase_factory,
+        )
+
+    def test_capability_only_includes_purchase_when_adapter_is_ready(self):
+        collect_only = self._dispatcher(purchase_ready=False)
+        ready = self._dispatcher(purchase_ready=True)
+
+        self.assertEqual(
+            collect_only.claim_capabilities().supported_types,
+            (TaskType.COLLECT,),
+        )
+        self.assertEqual(
+            ready.claim_capabilities().supported_types,
+            (TaskType.COLLECT, TaskType.PURCHASE),
+        )
+        self.assertEqual(
+            ready.claim_capabilities().purchase_mode, "dry_run"
+        )
+
+    def test_claims_saves_then_dispatches_purchase_dry_run(self):
+        task = purchase_task()
+        self.gateway.enqueue_task(task, self.client.client_id)
+
+        outcome = self._dispatcher(purchase_ready=True).execute_one()
+
+        self.assertEqual(outcome.kind, "succeeded")
+        detail = self.repository.get_task(task.task_id)
+        assert detail is not None
+        self.assertEqual(detail.task_type, TaskType.PURCHASE)
+        self.assertEqual(detail.status, TaskStatus.SUCCEEDED)
+        self.assertEqual(
+            detail.admin_payload["payload"]["options"]["bundle"],
+            "标准版",
+        )
+        self.assertIn(("purchase", "stopped"), self.calls)
+
+    def test_purchase_is_not_claimed_when_runtime_is_not_ready(self):
+        self.gateway.enqueue_task(purchase_task(), self.client.client_id)
+
+        outcome = self._dispatcher(purchase_ready=False).execute_one()
+
+        self.assertEqual(outcome.kind, "no_task")
+        self.assertEqual(self.repository.count_tasks(), 0)
+
+    def test_local_purchase_stops_dispatch_when_runtime_is_not_ready(self):
+        task = purchase_task()
+        self.repository.add_claimed_task(
+            admin_task_to_new_claimed_task(task)
+        )
+
+        with self.assertRaisesRegex(RuntimeError, "停止领取新任务"):
+            self._dispatcher(purchase_ready=False).execute_one()
+
+        detail = self.repository.get_task(task.task_id)
+        assert detail is not None
+        self.assertEqual(detail.status, TaskStatus.CLAIMED)
+
+    def test_purchase_mapper_rejects_missing_or_invalid_safety_fields(self):
+        invalid_payloads = (
+            {"goods_url": "https://example.test"},
+            {
+                "goods_url": "https://example.test",
+                "goods_id": "G",
+                "options": {},
+                "quantity": 1,
+                "max_price_cent": 100,
+            },
+            {
+                "goods_url": "https://example.test",
+                "goods_id": "G",
+                "options": {"size": "L"},
+                "quantity": 0,
+                "max_price_cent": 100,
+            },
+            {
+                "goods_url": "https://example.test",
+                "goods_id": "G",
+                "options": {"size": "L"},
+                "quantity": 1,
+                "max_price_cent": 0,
+            },
+        )
+        for index, payload in enumerate(invalid_payloads):
+            with self.subTest(index=index):
+                task = AdminTask(
+                    f"PUR-BAD-{index}",
+                    TaskType.PURCHASE,
+                    1,
+                    0,
+                    payload,
+                )
+                with self.assertRaises(ValueError):
+                    admin_task_to_new_claimed_task(task)
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -89,7 +89,7 @@ client/
 | 目录分层 | domain / application / infrastructure / workers / ui | 平铺在 `src/` 下：`db.py`、`db_schema.py`、`task_repository.py`、`settings_repository.py`、`task_models.py` 等，另有 `src/util/`、`src/demo1/` |
 | 主按钮文案 | 「开始自动获取」⇄「停止自动获取」 | 已按持续串行模式实现 |
 | Admin 网关 | `AdminGateway` + Mock/HTTP 两实现 | 登记、领取、结果和失败提交已实现 |
-| 任务应用服务 | `CollectTaskService` / `PurchaseTaskService` | 采集已接入自动获取；采购已实现本地 `dry_run` 演练，领取与分派由后续工单接入 |
+| 任务应用服务 | `TaskDispatcher` / `CollectTaskService` / `PurchaseTaskService` | 自动获取先补交 Outbox，再按领取时间执行本地任务，最后按安全能力领取并分派新任务 |
 | Outbox 提交 | 从 `outbox_events` 取件重试 | 采集结果与失败已实现，重试不重复采集 |
 | PDD 自动化 | `infrastructure/pdd/` 适配层 | `pdd_device_service.py` 与 `pdd_collect_service.py` 已接入采集主链 |

@@ -330,6 +330,11 @@ Client 与 Admin 的任务交互只有三种调用：领一个任务、提交结
 6. Outbox 提交成功、Admin 返回 `accepted: true` 后标记 `succeeded`。
 7. 回到第 1 步领下一个任务。**同一时间只做一个任务。**

+`TaskDispatcher` 是能力声明的唯一入口。没有可用采购演练 Adapter、没有已保存
+Android 设备或本地持久化未准备好时，只声明 `collect`；条件满足时才声明
+`collect,purchase`，且 `purchase_mode` 永远是 `dry_run`。领取响应必须先完整校验并
+写入 SQLite，Repository 提交成功后才能分派，避免任务已在 Admin 领取却在本地丢失。
+
 中途 Admin 是否取消了这个任务、是否重派给了别人，Client 不查也不管，做完照样提交——
 Admin 侧必须无条件接受，见 [04](04-admin-api-contract.md) §6.1。

@@ -199,10 +199,16 @@ POST /api/v1/client/tasks/claim

 无可领取任务时返回 `204 No Content`。

-Client 的 `HttpAdminGateway.claim_next` 已实现本接口。当前只声明
-`supported_types: ["collect"]`，一次调用最多领取一个采集任务；采集结束后通过
+Client 的 `HttpAdminGateway.claim_next` 已实现本接口，并原样序列化应用层已经
+安全确认的 `ClaimCapabilities`。没有采购演练 Adapter 或设备未准备好时只声明
+`supported_types: ["collect"]`；条件满足时声明 `collect,purchase`，但
+`purchase_mode` 固定为 `dry_run`。一次调用最多领取一个任务，结果或失败通过
 本页 §6 或 §7 提交，完整请求会先进入本地 Outbox。

+采购响应在写入本地前必须校验非空 `goods_url`、`goods_id`、动态 `options`、
+正整数 `quantity` 和正整数分 `max_price_cent`。字段不完整时停止新的领取并向
+操作员显示协议错误；不得让不完整任务进入手机执行。
+
 `[必须]` **`204` 不是错误。** Client 要把它当"暂时没活干"处理，
 不要报错，也不要因此触发重试风暴。首次 claim 通常返回 204；如果该编号已经预先分配任务，也可以直接返回 200。

@@ -64,7 +64,8 @@
 才向 Admin 领取一条。领取、手机采集和提交都在单独的工作线程完成，一轮结束且
 线程完全退出后，由主线程的单次定时器安排下一轮。暂无任务时默认 5 秒后重试；
 可恢复的 Admin 错误按 5、10、20、30 秒退避。需要人工处理、不可恢复错误、设备
-或配置错误会停止自动获取。当前仍不执行采购任务。
+或配置错误会停止自动获取。采购演练执行器和 Android 设备都准备好时，可以领取
+采购任务；界面必须明确显示“演练”，任何路径都不得显示成真实下单。

 ### 4.2 搜索与筛选

@@ -86,6 +87,7 @@
 - 确认后只执行选中的稳定任务编号，不领取新任务，不先处理其他任务或 Outbox。
 - 重新采集在工作线程运行。执行期间禁用“获取任务”和“重新执行”，完成后刷新列表。
 - 每次重新采集创建新的执行记录和幂等键；当前结果更新，旧结果保存在历史执行记录中。
+- 采购任务始终不能通过“重新执行”入口启动；需要处理时由自动获取的安全恢复流程决定。
 - “等待重试”当前没有倒计时。自动获取因可恢复采集错误停止时，底部状态显示
  “重试已暂停”，并提示选择任务点击“重新执行”或重新启动获取任务。