fix: 对齐规格解析 Mock 契约 (#257)

client/test/test_admin_gateway_contract.py

@@ -2,6 +2,7 @@
 
 import hashlib
 import unittest
+from copy import deepcopy
 
 from src.admin_gateway import (
     AdminGateway,
@@ -17,6 +18,63 @@ from src.mock_admin_gateway import MockAdminGateway
 from src.task_models import TaskType
 
 
+def _frame(value: str) -> str:
+    return f"{len(value.encode('utf-8'))}:{value}"
+
+
+def _finalize_spec_observation(task_id: str, observation: dict) -> str:
+    candidates = observation["candidates"]
+    snapshot_values = [
+        "spec-resolution-v1",
+        observation["pdd_goods_id"],
+        observation["selected_color"],
+        str(len(candidates)),
+    ]
+    for candidate in candidates:
+        snapshot_values.extend(
+            (
+                candidate["candidate_id"],
+                candidate["raw_text"],
+                candidate["options"]["color"],
+                candidate["options"]["size"],
+            )
+        )
+    observation["candidate_snapshot_hash"] = hashlib.sha256(
+        "".join(_frame(value) for value in snapshot_values).encode("utf-8")
+    ).hexdigest()
+    identity_values = (
+        task_id,
+        observation["attempt_id"],
+        observation["candidate_snapshot_hash"],
+        "spec-resolution-v1",
+    )
+    return "spec-resolution-v1:" + hashlib.sha256(
+        "".join(_frame(value) for value in identity_values).encode("utf-8")
+    ).hexdigest()
+
+
+def _spec_observation(task_id: str = "PUR-SPEC") -> tuple[dict, str]:
+    observation = {
+        "schema_version": 1,
+        "task_version": 3,
+        "attempt_id": "attempt-001",
+        "pdd_goods_id": "737116531267",
+        "original_options": {"color": "黑色", "size": "60公斤"},
+        "selected_color": "黑色",
+        "target_size": "60公斤",
+        "candidates": [
+            {
+                "candidate_id": "c1",
+                "raw_text": "120斤",
+                "options": {"color": "黑色", "size": "120斤"},
+            }
+        ],
+        "candidate_snapshot_hash": "",
+        "observed_at": "2026-08-17T08:00:00Z",
+    }
+    return observation, _finalize_spec_observation(task_id, observation)
+
+
 class MockAdminGatewayContractTest(unittest.TestCase):
     def setUp(self):
         self.gateway = MockAdminGateway()
@@ -252,6 +310,115 @@ class MockAdminGatewayContractTest(unittest.TestCase):
         self.assertEqual(second, configured)
         self.assertEqual(self.gateway.spec_resolution_count, 1)
 
+    def test_spec_resolution_validation_matches_admin_contract(self):
+        task_id = "PUR-SPEC-VALIDATION"
+        task = AdminTask(
+            task_id=task_id,
+            task_type=TaskType.PURCHASE,
+            version=3,
+            priority=1,
+            payload={
+                "goods_id": "737116531267",
+                "options": {"color": "黑色", "size": "60公斤"},
+            },
+        )
+        self.gateway.enqueue_task(task, self.client.client_id)
+        self.gateway.claim_next(self.client, self.all_capabilities)
+        original, _key = _spec_observation(task_id)
+
+        cases = []
+
+        wrong_color = deepcopy(original)
+        wrong_color["selected_color"] = "白色"
+        wrong_color["candidates"][0]["options"]["color"] = "白色"
+        cases.append((
+            "已选颜色不是领取规格",
+            wrong_color,
+            _finalize_spec_observation(task_id, wrong_color),
+            "INVALID_SPEC_RESOLUTION_REQUEST",
+        ))
+
+        wrong_size = deepcopy(original)
+        wrong_size["target_size"] = "70公斤"
+        cases.append((
+            "目标尺码不是领取规格",
+            wrong_size,
+            _finalize_spec_observation(task_id, wrong_size),
+            "INVALID_SPEC_RESOLUTION_REQUEST",
+        ))
+
+        missing_timezone = deepcopy(original)
+        missing_timezone["observed_at"] = "2026-08-17T08:00:00"
+        cases.append((
+            "观测时间没有时区",
+            missing_timezone,
+            _finalize_spec_observation(task_id, missing_timezone),
+            "INVALID_SPEC_RESOLUTION_REQUEST",
+        ))
+
+        oversized_attempt = deepcopy(original)
+        oversized_attempt["attempt_id"] = "a" * 192
+        cases.append((
+            "执行尝试编号超长",
+            oversized_attempt,
+            _finalize_spec_observation(task_id, oversized_attempt),
+            "INVALID_SPEC_RESOLUTION_REQUEST",
+        ))
+
+        control_character = deepcopy(original)
+        control_character["target_size"] = "60\n公斤"
+        cases.append((
+            "字段包含控制字符",
+            control_character,
+            _finalize_spec_observation(task_id, control_character),
+            "INVALID_SPEC_RESOLUTION_REQUEST",
+        ))
+
+        oversized_body = deepcopy(original)
+        oversized_body["future_padding"] = "x" * (64 * 1024)
+        cases.append((
+            "请求体超过上限",
+            oversized_body,
+            _finalize_spec_observation(task_id, oversized_body),
+            "INVALID_BODY",
+        ))
+
+        for name, request, key, expected_code in cases:
+            with self.subTest(name=name):
+                with self.assertRaises(AdminGatewayError) as raised:
+                    self.gateway.resolve_purchase_spec(task_id, key, request)
+                self.assertEqual(raised.exception.code, expected_code)
+                self.assertFalse(raised.exception.retryable)
+
+    def test_spec_resolution_uses_admin_task_and_claim_error_codes(self):
+        observation, key = _spec_observation("PUR-MISSING")
+        with self.assertRaises(AdminGatewayError) as missing:
+            self.gateway.resolve_purchase_spec(
+                "PUR-MISSING", key, observation
+            )
+        self.assertEqual(missing.exception.code, "TASK_NOT_FOUND")
+
+        task_id = "PUR-UNCLAIMED"
+        self.gateway.enqueue_task(
+            AdminTask(
+                task_id=task_id,
+                task_type=TaskType.PURCHASE,
+                version=3,
+                priority=1,
+                payload={
+                    "goods_id": "737116531267",
+                    "options": {"color": "黑色", "size": "60公斤"},
+                },
+            ),
+            self.client.client_id,
+        )
+        observation, key = _spec_observation(task_id)
+        with self.assertRaises(AdminGatewayError) as unclaimed:
+            self.gateway.resolve_purchase_spec(task_id, key, observation)
+        self.assertEqual(
+            unclaimed.exception.code, "TASK_NOT_CLAIMED_BY_CLIENT"
+        )
+
     def test_same_idempotency_key_and_content_reuses_receipt(self):
         self.gateway.enqueue_task(
             self._task("TASK-001", TaskType.COLLECT), "client-001"