fix(gateway): document multi-item order intent schema

重试链路的关键信息是「升级路径」，而属性表达不了过程：同名属性后写覆盖先写，
三次尝试跑完只剩最后一次的状态码，前两次为什么失败、升到哪一级全被盖掉。

- site_session / rakuma_session：每次尝试各记一条 scrape.attempt event，带
  attempt / outcome / status_code / 截断后的错误串；outcome 区分 http_error、
  bad_status、challenge、validate_failed
- site_session 的每次升级各记一条 scrape.escalate：rewarm_on_home、浏览器兜底
  recovered / failed。兜底失败也要记——不然链路里只剩「最终失败」，看不出浏览器
  这一级试过没有，而「没装 playwright」和「装了也被挡」是两个查法（reason 取
  BrowserFallback.unavailable_reason，为 None 时由 add_event 跳过）
- 三处 span.record_exception 换成 record_error，失败的错误码与可重试位跟着落下来
- 失败页面快照带上 url 与 profile，省得回头对是哪条通道的哪个地址

测试：test_scrape_telemetry.py 补一条会话层用例，用 MockTransport 让所有请求都回
挑战页，断言三次尝试各留一条 event（而不是只剩最后一次）、升级路径依次是
rewarm_on_home → browser，以及浏览器那一级的 outcome=failed 与 reason。

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

README.md

@@ -139,7 +139,7 @@ PC UA 在搜索页、详情页、店铺页上都能拿到完整模板。因此
 | 接口 | 说明 |
 | --- | --- |
 | `GET /health` | 健康检查，含 worker 心跳、长时间无人领任务告警 |
-| `POST /api/orders` | 上游提交下单意图（幂等） |
+| `POST /api/orders` | 上游提交下单意图（幂等）；`intent.items` 支持一次购买多个商品，旧版 `intent.item_url` 仍兼容 |
 | `GET /api/orders/lease` | 本地 worker 长轮询领取（全局并发度 1） |
 | `POST /api/orders/{id}/renew` | 续租（worker 在长任务里每 60s 调一次） |
 | `POST /api/orders/{id}/report` | 本地回报订单状态（同 state 重复上报幂等） |
@@ -373,6 +373,20 @@ docker compose --profile gateway up -d
 
 搜索结果中的 `shop.shop_code` + `item_code` 可直接用作本接口入参。
 
+**下单前要决定的参数全在响应里**（规格与选项是两套东西，下单时走不同字段）：
+
+| 字段 | 下单时对应 | 说明 |
+| --- | --- | --- |
+| `sku.variants[].variant_id` | `intent.variant_id` | 规格组合（颜色 × 尺码等） |
+| `options[]` | `intent.choice` | 店铺自定义选项，格式「选项名:取值名」 |
+| `has_required_options` | — | `true` 时不给 `choice` 会被站点拒绝加购 |
+| `unfillable_required_options` | — | 必填但无法自动选值的选项名，**必须**由调用方给值 |
+
+`options[].values[].is_placeholder` 标出「選択してください」这类占位项——它们不是
+合法取值，拼 `choice` 时要跳过（`selectable_value_count` 已是剔除占位项后的数量）。
+`type="text"` 的选项是自由文本（如「【お名前】4文字まで」），站点不给候选值，
+必填时只能由调用方给值。
+
 部分官方店的商品页会跳转到独立子站，响应里的 `source` / `source_url` 会标明数据来源，
 字段覆盖差异见下一节。
 
@@ -432,6 +446,7 @@ docker compose --profile gateway up -d
 | `review` | ✅ | ✅ | ❌ 异步加载 | ❌ 异步加载 |
 | `sku.variants`（规格组合） | ✅ | — 图书无规格 | ✅ 颜色 × 尺码 | — 单一规格 |
 | `sku.attributes`（规格表） | ✅ | ✅ 出版社/ISBN 等 | 挂在各 variant 上 | ❌ |
+| `options`（店铺自定义选项） | ✅ | ❌ 未解析 | ❌ 未解析 | ❌ 未解析 |
 | `shipping` 运费明细 | ✅ | ❌ 仅库存措辞 | ❌ | 仅「是否含运费」 |
 | `shop.shop_id` | ✅ | ❌ | ✅ | ✅ |
 | `breadcrumbs[].url` | ✅ | ✅ | ❌ 站内分类编码，拼不出链接 | ✅ |
@@ -602,6 +617,8 @@ docker compose --profile gateway up -d
 | `purchase_unit` | 起订单位 |
 | `sku.inventory_type` | `multiple` 表示多规格，要看 `sku.variants[]` |
 | `sku.variants[].variant_id` | 多规格商品的规格 ID（trading 加购多规格时按此选择） |
+| `options[]` | 店铺自定义选项（必填项不给值站点拒绝加购），加购时走 `choice` 字段 |
+| `unfillable_required_options` | 必填但无法自动选值的选项名，必须由调用方显式给值 |
 
 交易服务对外接口（:31108，全部需要 Bearer token）：
 
@@ -611,8 +628,13 @@ docker compose --profile gateway up -d
 - `POST /api/cart/clear` — 清空购物车（UI 点击 `button[aria-label="削除"]`）
 - `POST /api/cart/remove` — 删除指定 `item_id`
 
-底层共用 `app/shared/purchase_contract.py` 的常量与字段构造（与 scraping 模型解耦）。
-trading 加购时的字段选择策略：多规格挑第一个非售罄的 variant；必填选项拼「名:值」。
+底层共用 `app/shared/purchase_contract.py` 的常量、字段构造与**选项解析**（与 scraping
+模型解耦）——`/api/item_detail` 对外暴露的 `options[]` 和 trading 自动填 `choice` 用的
+是同一份解析与占位项判定，避免「接口说能选的值」与「下单实际填的值」不一致。
+trading 加购时的字段选择策略：多规格挑第一个非售罄的 variant；必填选项拼「名:值」，
+取第一个**非占位**候选值（`values[0]` 往往是「選択してください」，填它等于没选）。
+必填项自动填不出来（自由文本项、候选值只剩占位项）且调用方没给 `choice` 时当场报错
+并点名是哪几项，不拿占位值凑数去撞站点。
 站点端点 `basketDomain` 逐商品不同（实测有 `sp.basket…` 与 `ts.sp.basket…`），不能写死。
 
 几点子站差异（仅信息，trading 不覆盖子站加购）：

app/gateway/models.py

@@ -6,10 +6,10 @@ intent 字段刻意保留成 `dict[str, Any]`——网关不解释下单意图
 """
 from __future__ import annotations
 
-from typing import Any
+from typing import Annotated, Any
 from urllib.parse import urlsplit
 
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field, WithJsonSchema, field_validator
 
 from app.shared.task_state import AccountQueryKind, OrderState, QueryStatus, TaskStatus
 
@@ -17,6 +17,79 @@ from app.shared.task_state import AccountQueryKind, OrderState, QueryStatus, Tas
 # ---- POST /api/orders ----
 
 
+# The gateway intentionally keeps intent open-ended at runtime. This schema documents
+# the stable trading fields without preventing future fields from being passed through.
+OrderIntent = Annotated[
+    dict[str, Any],
+    WithJsonSchema(
+        {
+            "type": "object",
+            "additionalProperties": True,
+            "description": (
+                "下单意图。推荐使用 items；旧版 item_url 等同级字段继续兼容。"
+            ),
+            "properties": {
+                "items": {
+                    "type": "array",
+                    "minItems": 1,
+                    "description": "本次购买的商品列表，按顺序加入同一购物车",
+                    "items": {
+                        "oneOf": [
+                            {
+                                "type": "object",
+                                "additionalProperties": True,
+                                "properties": {
+                                    "item_url": {
+                                        "type": "string",
+                                        "description": "商品页 URL",
+                                    },
+                                    "quantity": {
+                                        "type": "integer",
+                                        "minimum": 1,
+                                        "default": 1,
+                                    },
+                                    "variant_id": {"type": "string"},
+                                    "choice": {
+                                        "oneOf": [
+                                            {"type": "string"},
+                                            {
+                                                "type": "array",
+                                                "items": {"type": "string"},
+                                            },
+                                        ]
+                                    },
+                                },
+                                "required": ["item_url"],
+                            },
+                            {
+                                "type": "string",
+                                "description": "商品页 URL（简写）",
+                            },
+                        ]
+                    },
+                },
+                "item_url": {
+                    "type": "string",
+                    "description": "旧版单商品商品页 URL",
+                },
+                "quantity": {"type": "integer", "minimum": 1, "default": 1},
+                "variant_id": {"type": "string"},
+                "choice": {
+                    "oneOf": [
+                        {"type": "string"},
+                        {"type": "array", "items": {"type": "string"}},
+                    ]
+                },
+                "max_total_yen": {
+                    "type": "integer",
+                    "description": "本次订单允许的最高应付金额（日元）",
+                },
+            },
+        }
+    ),
+]
+
+
 class SubmitOrderRequest(BaseModel):
     """上游提交下单意图
 
@@ -34,19 +107,26 @@ class SubmitOrderRequest(BaseModel):
         description="站点标识。交易服务只覆盖乐天市场，固定 rakuten",
         examples=["rakuten"],
     )
-    intent: dict[str, Any] = Field(
+    intent: OrderIntent = Field(
         description=(
             "下单意图原文。网关不解释内容，原样存库并透传给本地 worker，结构由 trading 侧定义："
-            "item_url（必填，商品页 URL）；quantity（可选，默认 1）；"
+            "推荐使用 items（非空数组，每项含 item_url，及可选 quantity/variant_id/choice）；"
+            "为兼容已发布客户端，也支持单商品 item_url（及同级 quantity/variant_id/choice）；"
+            "quantity（可选，默认 1）；"
             "variant_id（多规格商品必填，取自 /api/item_detail 的 variants，不传时 worker 自动选第一个非售罄规格）；"
             "choice（可选，商品选项，如 \"颜色:赤\"，可传字符串或字符串列表）；"
             "max_total_yen（可选，本次金额上限：确认页实际应付超过即中止并报 needs_human，"
             "缺省用服务端 RAKUTEN_ORDER_MAX_TOTAL_YEN）"
         ),
         examples=[{
-            "item_url": "https://item.rakuten.co.jp/shop/code/",
-            "quantity": 1,
-            "variant_id": "1001",
+            "items": [{
+                "item_url": "https://item.rakuten.co.jp/shop/code/",
+                "quantity": 1,
+                "variant_id": "1001",
+            }, {
+                "item_url": "https://item.rakuten.co.jp/shop/another-code/",
+                "quantity": 2,
+            }],
             "max_total_yen": 30000,
         }],
     )

app/scraping/models/scrape.py

@@ -338,6 +338,43 @@ class SkuInfo(BaseModel):
     variant_count: int = Field(default=0, description="SKU 组合总数；不受 include_sku_variants 影响，始终为真实组合数")
 
 
+class ItemOptionValue(BaseModel):
+    """店铺自定义选项的一个候选取值"""
+
+    value_id: int | None = Field(default=None, description="站点 values[].id 原值；给不出数字时为 null")
+    name: str = Field(default="", description="取值展示名，下单时 choice 里要用这个原文")
+    is_placeholder: bool = Field(
+        default=False,
+        description="是否为「選択してください」这类占位项。占位项不是合法取值，"
+        "提交它等于没选，构造 choice 时必须跳过",
+    )
+
+
+class ItemOption(BaseModel):
+    """一个店铺自定义选项（下单必填项的来源）
+
+    与 SKU 规格不是一回事：规格走 `sku.variants[].variant_id`，选项走下单接口的
+    `choice` 字段。店铺用它承载「名入れ文字」「配送方式确认」「レビュー依頼」等，
+    `is_required=true` 的选项不给值时站点会直接拒绝加购。
+    """
+
+    option_id: int | None = Field(default=None, description="站点 options[].id 原值")
+    name: str = Field(default="", description="选项名，下单时 choice 的「名」部分要用这个原文")
+    type: str = Field(
+        default="",
+        description="站点原值：select（下拉，看 values）/ text（自由文本，values 为空）",
+    )
+    is_required: bool = Field(default=False, description="是否必填；必填项不给值时站点拒绝加购")
+    values: list[ItemOptionValue] = Field(
+        default_factory=list, description="候选取值；type=text 时为空"
+    )
+    selectable_value_count: int = Field(
+        default=0,
+        description="剔除占位项后真正可提交的候选数。为 0 且 is_required=true 时"
+        "无法自动选值，必须由调用方在 choice 里显式给出",
+    )
+
+
 class ShippingInfo(BaseModel):
     """配送与运费信息"""
 
@@ -365,7 +402,8 @@ class ItemDetailData(BaseModel):
 
     加购（构造 cart 请求）不在本服务范围内——加购需要已登录的乐天账号会话，
     归 trading 服务（app.trading）。本响应只描述「商品状态」：能不能买
-    （purchase_condition / is_sold_out）、规格（sku.variants）、起订单位等。
+    （purchase_condition / is_sold_out）、规格（sku.variants）、起订单位、
+    店铺自定义选项（options）等，即「下单前需要先决定哪些参数」。
     """
 
     source: str = Field(default="ichiba", description="数据来源站点：ichiba / books / brandavenue / biccamera")
@@ -390,6 +428,21 @@ class ItemDetailData(BaseModel):
     breadcrumbs: list[Breadcrumb] = Field(default_factory=list, description="分类面包屑")
     shipping: ShippingInfo = Field(default_factory=ShippingInfo, description="该商品的配送与运费信息")
     sku: SkuInfo = Field(default_factory=SkuInfo, description="SKU 信息")
+    options: list[ItemOption] = Field(
+        default_factory=list,
+        description="店铺自定义选项（站点 purchase.information.options）。与 SKU 规格不同："
+        "规格选 sku.variants[].variant_id，选项走下单接口的 choice 字段。"
+        "空列表表示该商品页没有选项；子站（books / brandavenue / biccamera）暂不解析",
+    )
+    has_required_options: bool = Field(
+        default=False,
+        description="是否存在必填选项。为 true 时下单必须给 choice，否则站点拒绝加购",
+    )
+    unfillable_required_options: list[str] = Field(
+        default_factory=list,
+        description="必填但无法自动选值的选项名（自由文本项，或候选值只有占位项）。"
+        "非空时**必须**由调用方在下单 intent.choice 里显式给出这些项的取值",
+    )
 
 
 class HealthData(BaseModel):

app/scraping/parsers/item.py

@@ -11,9 +11,13 @@ from __future__ import annotations
 from typing import Any
 
 from app.shared.errors import ScrapeParseError
+from app.shared.purchase_contract import ItemOption as SharedItemOption
+from app.shared.purchase_contract import auto_choice_for, parse_options
 from app.scraping.models.scrape import (
     Breadcrumb,
     ItemDetailData,
+    ItemOption,
+    ItemOptionValue,
     ReviewSummary,
     ShippingInfo,
     ShopSummary,
@@ -31,6 +35,33 @@ _PURCHASABLE_CONDITION = "enabled"
 # 库存类型 → 加购表单里的 inventory_flag（常量与基础字段构造在 app.shared.purchase_contract）
 
 
+def _to_item_options(options: list[SharedItemOption]) -> list[ItemOption]:
+    """把共用契约的 ItemOption dataclass 搬成对外的 pydantic 模型
+
+    解析与占位项判定都在 `app.shared.purchase_contract.parse_options`——trading
+    下单时用同一份逻辑挑 choice 取值，两侧对「哪个取值是合法的」必须完全一致，
+    否则本接口告诉上游能选的值、下单时却填了别的。本函数只做搬运，不加判断。
+    """
+    return [
+        ItemOption(
+            option_id=option.option_id,
+            name=option.name,
+            type=option.type,
+            is_required=option.is_required,
+            values=[
+                ItemOptionValue(
+                    value_id=value.value_id,
+                    name=value.name,
+                    is_placeholder=value.is_placeholder,
+                )
+                for value in option.values
+            ],
+            selectable_value_count=len(option.selectable_values),
+        )
+        for option in options
+    ]
+
+
 def _parse_attributes(raw: Any) -> list[SkuAttribute]:
     return [
         SkuAttribute(title=as_str(attr.get("title")), value=as_str(attr.get("value")))
@@ -115,6 +146,13 @@ def parse_item_detail(
     sell_type = _pick_sell_type(as_dict(purchase.get("sellType")))
     purchase_condition = as_str(sell_type.get("purchaseCondition"))
 
+    purchase_information = as_dict(purchase.get("information"))
+    # 只解析一次，两个派生结果都从这份结果来
+    shared_options = parse_options(purchase_information)
+    # 无法自动选值的必填项：与 trading 自动填 choice 时的判定同源，上游据此知道
+    # 「哪些项必须自己给值」，而不是等下单时才被站点拒绝
+    _, unfillable_required = auto_choice_for(shared_options)
+
     raw_sku = as_dict(purchase.get("sku"))
     variants = _parse_variants(raw_sku.get("variants"))
     sku = SkuInfo(
@@ -188,7 +226,7 @@ def parse_item_detail(
         purchase_condition=purchase_condition,
         # purchaseCondition 是站点判定能否下单的直接依据；缺失时不臆断为售罄
         is_sold_out=bool(purchase_condition) and purchase_condition != _PURCHASABLE_CONDITION,
-        purchase_unit=as_int(as_dict(purchase.get("information")).get("unit")),
+        purchase_unit=as_int(purchase_information.get("unit")),
         images=images,
         shop=shop,
         review=review,
@@ -196,4 +234,7 @@ def parse_item_detail(
         breadcrumbs=breadcrumbs,
         shipping=shipping,
         sku=sku,
+        options=_to_item_options(shared_options),
+        has_required_options=any(option.is_required for option in shared_options),
+        unfillable_required_options=unfillable_required,
     )

app/scraping/services/rakuma_client.py

@@ -37,7 +37,7 @@ from app.scraping.utils.rakuma_urls import (
     split_item_url,
     split_shop_url,
 )
-from app.shared.telemetry import snapshot
+from app.shared.telemetry import record_parse_failure
 
 logger = logging.getLogger(__name__)
 tracer = trace.get_tracer(__name__)
@@ -77,10 +77,11 @@ class RakumaClient:
                     url, len(result.items), result.total_count,
                 )
                 return result
-            except Exception:
-                span.record_exception()
-                span.set_attribute("parse.fail_reason", "parse_error")
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+            except Exception as exc:
+                record_parse_failure(
+                    span, exc, html=html,
+                    max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
+                )
                 raise
 
     async def categories(self, payload: RakumaCategoryRequest) -> RakumaCategoryData:
@@ -110,10 +111,11 @@ class RakumaClient:
                     data.category_id, data.name, len(data.children), data.total_count,
                 )
                 return data
-            except Exception:
-                span.record_exception()
-                span.set_attribute("parse.fail_reason", "parse_error")
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+            except Exception as exc:
+                record_parse_failure(
+                    span, exc, html=html,
+                    max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
+                )
                 raise
 
     async def item_detail(self, payload: RakumaItemDetailRequest) -> RakumaItemDetailData:
@@ -139,10 +141,11 @@ class RakumaClient:
                     url, detail.item_name[:40], detail.price, detail.is_sold_out,
                 )
                 return detail
-            except Exception:
-                span.record_exception()
-                span.set_attribute("parse.fail_reason", "parse_error")
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+            except Exception as exc:
+                record_parse_failure(
+                    span, exc, html=html,
+                    max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
+                )
                 raise
 
     async def shop_detail(self, payload: RakumaShopDetailRequest) -> RakumaShopDetailData:
@@ -177,10 +180,11 @@ class RakumaClient:
                     shop_id, detail.shop_name, detail.item_count, detail.review_count,
                 )
                 return detail
-            except Exception:
-                span.record_exception()
-                span.set_attribute("parse.fail_reason", "parse_error")
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+            except Exception as exc:
+                record_parse_failure(
+                    span, exc, html=html,
+                    max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
+                )
                 raise
 
     async def shop_items(self, payload: RakumaShopItemsRequest) -> RakumaShopItemsData:
@@ -205,10 +209,11 @@ class RakumaClient:
                     shop_id, len(result.items), result.total_count,
                 )
                 return result
-            except Exception:
-                span.record_exception()
-                span.set_attribute("parse.fail_reason", "parse_error")
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+            except Exception as exc:
+                record_parse_failure(
+                    span, exc, html=html,
+                    max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
+                )
                 raise
 
     @staticmethod

app/scraping/services/rakuma_session.py

@@ -29,10 +29,14 @@ from app.shared.errors import (
     UpstreamBlockedError,
     UpstreamRequestError,
 )
-from app.shared.telemetry import snapshot
+from app.shared.telemetry import add_event, record_error, snapshot
 
 logger = logging.getLogger(__name__)
 
+# 逐次尝试 event 里错误串的截断长度：httpx 的异常消息可能很长（带完整 URL 与
+# 底层 socket 错误），而这里只是用来区分「这次是怎么失败的」，前半句就够
+_ERROR_MAX_CHARS = 200
+
 
 class RakumaSession:
     """ラクマ 站点抓取会话，管理并发限流与失败重试"""
@@ -108,7 +112,7 @@ class RakumaSession:
                 )
             except TimeoutError as exc:
                 err = ResourceBusyError()
-                span.record_exception(err)
+                record_error(span, err)
                 span.set_attribute("scrape.fail_reason", "ResourceBusyError")
                 raise err from exc
 
@@ -119,6 +123,13 @@ class RakumaSession:
                         response = await client.get(url)
                     except httpx.HTTPError as exc:
                         last_error = f"{type(exc).__name__}: {exc}"
+                        # 每次尝试各记一条 event：属性同名后写覆盖先写，三次尝试
+                        # 跑完只剩最后一次的状态，中间那两次为什么失败全被盖掉
+                        add_event(span, "scrape.attempt", {
+                            "attempt": attempt,
+                            "outcome": "http_error",
+                            "error": last_error[:_ERROR_MAX_CHARS],
+                        })
                         logger.warning(
                             "ラクマ 抓取请求异常：url=%s attempt=%s/%s err=%s",
                             url, attempt, max_attempts, last_error,
@@ -127,7 +138,7 @@ class RakumaSession:
 
                     if response.status_code == 404:
                         err = ItemNotFoundError(f"Page not found: {url}")
-                        span.record_exception(err)
+                        record_error(span, err)
                         span.set_attribute("scrape.fail_reason", "ItemNotFoundError")
                         raise err
 
@@ -145,6 +156,12 @@ class RakumaSession:
                     last_text = response.text
                     span.set_attribute("scrape.last_status_code", response.status_code)
                     span.set_attribute("scrape.html_bytes", len(response.text))
+                    add_event(span, "scrape.attempt", {
+                        "attempt": attempt,
+                        "outcome": "bad_status",
+                        "status_code": response.status_code,
+                        "error": last_error[:_ERROR_MAX_CHARS],
+                    })
                     logger.warning(
                         "ラクマ 抓取结果异常：url=%s attempt=%s/%s %s",
                         url, attempt, max_attempts, last_error,
@@ -154,9 +171,13 @@ class RakumaSession:
                     err = UpstreamRequestError(f"Upstream request failed: {last_error}")
                 else:
                     err = UpstreamBlockedError(f"Failed to fetch {url}: {last_error}")
-                span.record_exception(err)
+                record_error(span, err)
                 span.set_attribute("scrape.fail_reason", type(err).__name__)
-                snapshot(span, "scrape.failed_html", last_text, self._settings.otel_snapshot_max_bytes)
+                snapshot(
+                    span, "scrape.failed_html", last_text,
+                    self._settings.otel_snapshot_max_bytes,
+                    extra={"scrape.url": url, "scrape.profile": "rakuma"},
+                )
                 raise err
             finally:
                 self._semaphore.release()

app/scraping/services/rakuten_client.py

@@ -45,7 +45,7 @@ from app.scraping.utils.urls import (
     split_item_url,
     split_shop_url,
 )
-from app.shared.telemetry import snapshot
+from app.shared.telemetry import record_parse_failure
 
 logger = logging.getLogger(__name__)
 tracer = trace.get_tracer(__name__)
@@ -87,10 +87,11 @@ class RakutenClient:
                     url, len(result.items), result.ad_count, result.total_count,
                 )
                 return result
-            except Exception:
-                span.record_exception()
-                span.set_attribute("parse.fail_reason", "parse_error")
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+            except Exception as exc:
+                record_parse_failure(
+                    span, exc, html=html,
+                    max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
+                )
                 raise
 
     async def genres(self, payload: GenreRequest) -> GenreData:
@@ -113,10 +114,11 @@ class RakutenClient:
                     result.genre_id or "root", result.name, len(result.children),
                 )
                 return result
-            except Exception:
-                span.record_exception()
-                span.set_attribute("parse.fail_reason", "parse_error")
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+            except Exception as exc:
+                record_parse_failure(
+                    span, exc, html=html,
+                    max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
+                )
                 raise
 
     async def shop_detail(self, payload: ShopDetailRequest) -> ShopDetailData:
@@ -142,10 +144,11 @@ class RakutenClient:
                     result.shop_code, result.shop_id, result.shop_name, result.review_count,
                 )
                 return result
-            except Exception:
-                span.record_exception()
-                span.set_attribute("parse.fail_reason", "parse_error")
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+            except Exception as exc:
+                record_parse_failure(
+                    span, exc, html=html,
+                    max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
+                )
                 raise
 
     async def shop_items(self, payload: ShopItemsRequest) -> SearchResultData:
@@ -170,9 +173,11 @@ class RakutenClient:
                 span.set_attribute("parse.items", len(result.items))
                 span.set_attribute("parse.total", result.total_count)
                 return result
-            except Exception:
-                span.record_exception()
-                span.set_attribute("parse.fail_reason", "parse_error")
+            except Exception as exc:
+                # 本方法自己不抓页面，失败一定发生在它转调的 shop_detail / search
+                # 里（那两个 span 已各自记了自己的失败页面），所以显式标 delegate、
+                # 不落快照：按 html 推断只会得出「fetch 失败」的错误结论。
+                record_parse_failure(span, exc, stage="delegate")
                 raise
 
     async def item_detail(self, payload: ItemDetailRequest) -> ItemDetailData:
@@ -226,8 +231,9 @@ class RakutenClient:
                     url, detail.source, detail.item_name[:40], detail.price, detail.sku.variant_count,
                 )
                 return detail
-            except Exception:
-                span.record_exception()
-                span.set_attribute("parse.fail_reason", "parse_error")
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+            except Exception as exc:
+                record_parse_failure(
+                    span, exc, html=html,
+                    max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
+                )
                 raise

app/scraping/services/site_session.py

@@ -38,10 +38,14 @@ from app.shared.errors import (
 )
 from app.scraping.parsers.state import PageValidator, require_state_marker
 from app.scraping.services.browser_fallback import BrowserFallback
-from app.shared.telemetry import snapshot
+from app.shared.telemetry import add_event, record_error, snapshot
 
 logger = logging.getLogger(__name__)
 
+# 逐次尝试 event 里错误串的截断长度：httpx 异常消息可能很长（带完整 URL 与底层
+# socket 错误），这里只用来区分「这次是怎么失败的」，前半句够了
+_ERROR_MAX_CHARS = 200
+
 
 @dataclass(slots=True)
 class FetchedPage:
@@ -191,7 +195,7 @@ class SiteSession:
                 )
             except TimeoutError as exc:
                 err = ResourceBusyError()
-                span.record_exception(err)
+                record_error(span, err)
                 span.set_attribute("scrape.fail_reason", "ResourceBusyError")
                 raise err from exc
 
@@ -204,6 +208,14 @@ class SiteSession:
                         response = await profile.client.get(url)
                     except httpx.HTTPError as exc:
                         last_error = f"{type(exc).__name__}: {exc}"
+                        # 每次尝试各记一条 event。属性同名后写覆盖先写，三次尝试跑完
+                        # 只剩最后一次的状态码，前两次为什么失败、升级到哪一级全被
+                        # 盖掉——而这条链路的关键信息恰恰是「升级路径」。
+                        add_event(span, "scrape.attempt", {
+                            "attempt": attempt,
+                            "outcome": "http_error",
+                            "error": last_error[:_ERROR_MAX_CHARS],
+                        })
                         logger.warning(
                             "抓取请求异常：url=%s profile=%s attempt=%s/%s err=%s",
                             url, profile.name, attempt, max_attempts, last_error,
@@ -216,7 +228,7 @@ class SiteSession:
 
                     if response.status_code == 404:
                         err = ItemNotFoundError(f"Page not found: {url}")
-                        span.record_exception(err)
+                        record_error(span, err)
                         span.set_attribute("scrape.fail_reason", "ItemNotFoundError")
                         raise err
 
@@ -232,11 +244,21 @@ class SiteSession:
                         if reason is None:
                             return FetchedPage(html=text, url=final_url)
                         last_error = self._refine_failure(reason, text)
-                        if last_error.startswith("challenge page detected"):
+                        challenged = last_error.startswith("challenge page detected")
+                        if challenged:
                             span.set_attribute("scrape.challenge_detected", True)
+                        outcome = "challenge" if challenged else "validate_failed"
                     else:
                         last_error = self._describe_error_status(response.status_code)
+                        outcome = "bad_status"
                     last_text = text
+                    add_event(span, "scrape.attempt", {
+                        "attempt": attempt,
+                        "outcome": outcome,
+                        "status_code": response.status_code,
+                        "html_bytes": len(text),
+                        "error": last_error[:_ERROR_MAX_CHARS],
+                    })
                     logger.warning(
                         "抓取结果异常：url=%s profile=%s attempt=%s/%s %s",
                         url, profile.name, attempt, max_attempts, last_error,
@@ -250,20 +272,39 @@ class SiteSession:
                     if attempt == 1:
                         await self._rewarm_on_home(profile)
                         span.set_attribute("scrape.rewarmed_on_home", True)
+                        add_event(span, "scrape.escalate", {
+                            "attempt": attempt, "to": "rewarm_on_home",
+                        })
                     else:
                         page = await self._escalate_to_browser(profile, url, validate)
                         if page is not None:
                             span.set_attribute("scrape.fell_back_to_browser", True)
                             span.set_attribute("scrape.final_url", page.url)
+                            add_event(span, "scrape.escalate", {
+                                "attempt": attempt, "to": "browser", "outcome": "recovered",
+                            })
                             return page
+                        # 兜底没救回来（浏览器不可用，或取回的页面仍不合格）。不记
+                        # 这条的话链路里只看得到「最终失败」，看不出浏览器这一级到底
+                        # 试过没有——而「没装 playwright」和「装了也被挡」要分开查。
+                        add_event(span, "scrape.escalate", {
+                            "attempt": attempt,
+                            "to": "browser",
+                            "outcome": "failed",
+                            "reason": self._browser.unavailable_reason,
+                        })
 
                 if self._is_server_error(last_error):
                     err = UpstreamRequestError(f"Upstream request failed: {last_error}")
                 else:
                     err = UpstreamBlockedError(f"Blocked while fetching {url}: {last_error}")
-                span.record_exception(err)
+                record_error(span, err)
                 span.set_attribute("scrape.fail_reason", type(err).__name__)
-                snapshot(span, "scrape.failed_html", last_text, self._settings.otel_snapshot_max_bytes)
+                snapshot(
+                    span, "scrape.failed_html", last_text,
+                    self._settings.otel_snapshot_max_bytes,
+                    extra={"scrape.url": url, "scrape.profile": profile.name},
+                )
                 raise err
             finally:
                 self._semaphore.release()

app/shared/api.py

@@ -16,10 +16,12 @@ from typing import Any, Generic, TypeVar
 from fastapi import Depends, FastAPI, Request
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
+from opentelemetry import trace
 from pydantic import BaseModel, Field, ValidationError
 from starlette.exceptions import HTTPException as StarletteHTTPException
 
 from app.shared.errors import AppError, AuthenticationError
+from app.shared.telemetry import record_envelope, record_error
 
 logger = logging.getLogger(__name__)
 
@@ -98,16 +100,43 @@ def jsonable_errors(errors: list[dict]) -> list[dict]:
     return [{key: value for key, value in error.items() if key != "ctx"} for error in errors]
 
 
+def _trace_failure(
+    exc: BaseException | None, *, err_code: int, msg: str, status_code: int
+) -> None:
+    """把一次失败响应记到当前 server span 上（FastAPI 自动 instrumentation 建的那个）
+
+    这些处理器是所有对外失败的**唯一出口**，也是链路上唯一还知道「异常长什么样」
+    的地方：它们把异常吃掉换成 200/4xx 的信封响应，异常不再向上冒，自动
+    instrumentation 只看得到一个 HTTP 状态码。尤其 AppError 默认 status_code=400、
+    信封里 `success=false`，在 trace 里跟正常返回几乎分不出来——不在这里记一次，
+    上游报「调用失败了」时链路里根本找不到对应的错误。
+
+    span 没在录（otel 关闭、或 /health 这类被 excluded_urls 排除的路径）时
+    `set_attributes` / `record_error` 都作用在 NonRecordingSpan 上，是 noop，
+    不必额外判断。
+    """
+    span = trace.get_current_span()
+    if exc is not None:
+        record_error(span, exc)
+    record_envelope(span, success=False, err_code=err_code, msg=msg, status_code=status_code)
+
+
 def register_exception_handlers(app: FastAPI) -> None:
     """给应用挂上全套异常处理器
 
     两个入口都调用它，保证抓取失败与下单失败返回的错误结构完全一致，
     上游只需要按 code 分支，不必区分是哪个服务回的。
+
+    每个处理器除了构造响应，还把这次失败记到当前 server span 上（见
+    `_trace_failure`）——处理器是失败的唯一出口，不记就等于链路里没有这次失败。
     """
 
     @app.exception_handler(AppError)
     async def app_error_handler(_: Request, exc: AppError) -> JSONResponse:
         """业务异常处理器：返回结构化的错误响应"""
+        _trace_failure(
+            exc, err_code=exc.err_code, msg=exc.message, status_code=exc.status_code
+        )
         return JSONResponse(
             status_code=exc.status_code,
             content=ApiResponse[None](
@@ -123,11 +152,15 @@ def register_exception_handlers(app: FastAPI) -> None:
     async def validation_error_handler(_: Request, exc: RequestValidationError) -> JSONResponse:
         """请求参数校验异常处理器"""
         errors = exc.errors()
+        msg = _format_validation_msg(errors)
+        # 校验失败不记异常本身（pydantic 的 ValidationError 栈很长且没有诊断价值），
+        # 只留错误码与整理后的字段消息——排查要看的是「哪个字段不合法」
+        _trace_failure(None, err_code=1002, msg=msg, status_code=422)
         return JSONResponse(
             status_code=422,
             content=ApiResponse[object](
                 success=False,
-                msg=_format_validation_msg(errors),
+                msg=msg,
                 data=jsonable_errors(errors),
                 code=1002,
             ).model_dump(),
@@ -137,11 +170,13 @@ def register_exception_handlers(app: FastAPI) -> None:
     async def pydantic_validation_error_handler(_: Request, exc: ValidationError) -> JSONResponse:
         """Pydantic 模型校验异常处理器"""
         errors = exc.errors()
+        msg = _format_validation_msg(errors)
+        _trace_failure(None, err_code=1002, msg=msg, status_code=422)
         return JSONResponse(
             status_code=422,
             content=ApiResponse[object](
                 success=False,
-                msg=_format_validation_msg(errors),
+                msg=msg,
                 data=jsonable_errors(errors),
                 code=1002,
             ).model_dump(),
@@ -152,11 +187,13 @@ def register_exception_handlers(app: FastAPI) -> None:
         """HTTP 异常处理器（404、500 等）"""
         status_code = int(getattr(exc, "status_code", 500) or 500)
         err_code = 1404 if status_code == 404 else 1500
+        detail = str(getattr(exc, "detail", "HTTP error"))
+        _trace_failure(None, err_code=err_code, msg=detail, status_code=status_code)
         return JSONResponse(
             status_code=status_code,
             content=ApiResponse[None](
                 success=False,
-                msg=str(getattr(exc, "detail", "HTTP error")),
+                msg=detail,
                 data=None,
                 code=err_code,
             ).model_dump(),
@@ -167,6 +204,10 @@ def register_exception_handlers(app: FastAPI) -> None:
     async def unhandled_exception_handler(_: Request, exc: Exception) -> JSONResponse:
         """兜底异常处理器：捕获所有未处理的异常"""
         logger.exception("未处理异常：%s", exc)
+        # 这里最需要 record_error：对外只回一句无信息量的 "Internal server error"，
+        # 真正的异常类型与栈只在本进程日志里。记到 span 上，链路里就能直接看到
+        # 是什么炸了，不必再去捞日志按时间对。
+        _trace_failure(exc, err_code=1500, msg="Internal server error", status_code=500)
         return JSONResponse(
             status_code=500,
             content=ApiResponse[None](

app/shared/purchase_contract.py

@@ -9,11 +9,13 @@
 - `inventory_flag_for(inventory_type)` — 多规格判定
 - `basket_domain_of(sell_type)` — 抽 basketDomain + 反转义 `\\u002F`
 - `base_form_fields(shop_id, item_id, inventory_flag)` — 加购表单四件套
+- `parse_options(information)` — 店铺自定义选项（`purchase.information.options[]`）
+  的结构化解析，含占位值识别；scraping 用它对外暴露选项，trading 用它自动填 choice
+- `auto_choice_for(options)` / `format_choice(...)` — choice 表单值的构造
 
 **不**放这里：
 - `PurchaseInfo` pydantic 模型 — 是 scraping 的对外契约，4 个子站共用，留在 scraping/models
-- variant_id / choice 自动选择策略 — trading 独有（scraping 把决策权留给上游）
-- options 结构化解析 — scraping 独有（trading 用原始 dict）
+- variant_id 自动选择策略 — trading 独有（scraping 把决策权留给上游）
 - 子站（books/biccamera/brandavenue）的加购契约 — 各自独立，不走这里
 
 依赖约束：仅 `typing.Any`（标准库），不 import scraping / trading / gateway，
@@ -21,6 +23,7 @@
 """
 from __future__ import annotations
 
+from dataclasses import dataclass, field
 from typing import Any
 
 # 站点前端构造加购表单时固定带的事件标识（普通购买 normalPurchase）
@@ -76,3 +79,172 @@ def base_form_fields(
         "inventory_flag": inventory_flag,
         "__event": NORMAL_PURCHASE_EVENT,
     }
+
+
+# ---- 店铺自定义选项（purchase.information.options[]）----
+#
+# 「規格（SKU）」与「選項（option）」是两套完全不同的东西，加购表单里也走不同字段：
+# - 規格：sku.variants[].variantId → 表单 variant_id
+# - 選項：店铺自己配的下拉/文本框（名入れ文字、配送方式确认、レビュー依頼 等）
+#   → 表单 choice，格式「名:值」，多项用「,」连接
+#
+# 2026-08-28 用 .probe/ 下 3 份真实商品页样本 + tests/fixtures/item_with_options_state.json
+# 核对过结构，得到两条**结构性**判据（比按日文文案猜稳）：
+#
+# 1. `values[].id == 0` 是「請選擇」占位项，不是可提交的取值。三份真实样本里所有
+#    必填 select 的 values[0] 都是 id=0 的「選択してください」，真实可选值从 id=200
+#    起编号。此前 trading 侧自动填 choice 时取的就是 values[0]，等于把「選択して
+#    ください」当答案提交上去（.probe/checkout/probe-v2.txt 里两条 "成功" 记录的
+#    choice 值就是它）——站点当时收下了，但那是店铺没做校验，不代表填对了。
+# 2. `type == "text"` 的选项没有 values[]（自由文本，如「【お名前】4文字まで」），
+#    **无法**自动填。必填且为 text 时只能交由调用方给值，不猜。
+#
+# 占位项文案本身（「選択してください」）作为辅助判据一起保留：id 编号规则是从 4 份
+# 样本归纳的，万一某店铺不按 200 起编号，文案还能兜一层。两条判据命中任一即占位。
+_PLACEHOLDER_VALUE_ID: int = 0
+_PLACEHOLDER_VALUE_NAMES: frozenset[str] = frozenset({"選択してください", "選択して下さい"})
+
+# 选项类型：下拉（有候选值）与自由文本（无候选值）
+OPTION_TYPE_SELECT: str = "select"
+OPTION_TYPE_TEXT: str = "text"
+
+# choice 表单值的分隔符：项间用「,」，名与值之间用「:」
+_CHOICE_PAIR_SEPARATOR: str = ","
+_CHOICE_NAME_VALUE_SEPARATOR: str = ":"
+
+
+@dataclass(slots=True)
+class ItemOptionValue:
+    """选项的一个候选取值
+
+    `is_placeholder=True` 表示这是「選択してください」这类占位项，提交它等于没选。
+    """
+
+    value_id: int | None
+    name: str
+    is_placeholder: bool = False
+
+
+@dataclass(slots=True)
+class ItemOption:
+    """一个店铺自定义选项
+
+    `is_free_text` 为 True 时 values 必然为空（站点 type="text"），取值只能由
+    调用方给；`selectable_values` 是剔掉占位项后真正可提交的候选。
+    """
+
+    option_id: int | None
+    name: str
+    type: str
+    is_required: bool
+    values: list[ItemOptionValue] = field(default_factory=list)
+
+    @property
+    def is_free_text(self) -> bool:
+        return self.type == OPTION_TYPE_TEXT
+
+    @property
+    def selectable_values(self) -> list[ItemOptionValue]:
+        return [value for value in self.values if not value.is_placeholder]
+
+    @property
+    def can_auto_fill(self) -> bool:
+        """能否在不问调用方的情况下自动给出一个合法取值"""
+        return bool(self.selectable_values)
+
+
+def _coerce_option_id(raw: Any) -> int | None:
+    """选项/取值的 id 收敛为 int；给不出数字时返回 None（不编造 0——0 有含义）"""
+    if isinstance(raw, bool) or not isinstance(raw, (int, float, str)):
+        return None
+    try:
+        return int(raw)
+    except (TypeError, ValueError):
+        return None
+
+
+def _parse_option_value(raw: Any) -> ItemOptionValue | None:
+    """解析单个候选取值；不是 dict 或没有名字的直接丢掉"""
+    if not isinstance(raw, dict):
+        return None
+    name = raw.get("name")
+    if not isinstance(name, str) or not name:
+        return None
+    value_id = _coerce_option_id(raw.get("id"))
+    is_placeholder = value_id == _PLACEHOLDER_VALUE_ID or name.strip() in _PLACEHOLDER_VALUE_NAMES
+    return ItemOptionValue(value_id=value_id, name=name, is_placeholder=is_placeholder)
+
+
+def parse_options(information: Any) -> list[ItemOption]:
+    """从 `purchase.information` 解析店铺自定义选项列表
+
+    `information` 传站点原始 dict（不是 options 数组本身）；结构不符合预期时返回
+    空列表——**没有选项与解析不出选项在站点数据上长得一样**（缺 options 键），
+    这一层区分不出来，也不该假装能区分。
+    """
+    if not isinstance(information, dict):
+        return []
+    raw_options = information.get("options")
+    if not isinstance(raw_options, list):
+        return []
+
+    options: list[ItemOption] = []
+    for raw in raw_options:
+        if not isinstance(raw, dict):
+            continue
+        name = raw.get("name")
+        if not isinstance(name, str) or not name:
+            continue
+        raw_type = raw.get("type")
+        option_type = raw_type if isinstance(raw_type, str) else ""
+        values = [
+            value
+            for value in (_parse_option_value(item) for item in raw.get("values") or [])
+            if value is not None
+        ]
+        options.append(
+            ItemOption(
+                option_id=_coerce_option_id(raw.get("id")),
+                name=name,
+                type=option_type,
+                is_required=bool(raw.get("isRequired")),
+                values=values,
+            )
+        )
+    return options
+
+
+def format_choice(pairs: Any) -> str:
+    """把「名:值」序列拼成站点 choice 表单值
+
+    接受 str（原样透传，调用方自己拼好的）或可迭代的字符串序列。
+    """
+    if isinstance(pairs, str):
+        return pairs
+    return _CHOICE_PAIR_SEPARATOR.join(str(pair) for pair in pairs)
+
+
+def auto_choice_for(options: list[ItemOption]) -> tuple[str, list[str]]:
+    """为必填选项自动挑取值，返回 (choice 表单值, 无法自动填的必填项名)
+
+    策略：每个必填选项取**第一个非占位**候选值，拼成「名:值」。自由文本必填项
+    （type="text"，没有候选值）与候选值全是占位项的选项都自动填不了，其名字进
+    第二个返回值——调用方据此决定报错还是要求上游显式给 choice，**不拿占位值
+    凑数**（那正是本次修复的问题）。
+
+    只处理必填项：非必填项站点不强制，替上游擅自选（比如「置き配を希望する」）
+    等于替人做了业务决定。
+    """
+    pairs: list[str] = []
+    unfillable: list[str] = []
+    for option in options:
+        if not option.is_required:
+            continue
+        selectable = option.selectable_values
+        if not selectable:
+            unfillable.append(option.name)
+            continue
+        pairs.append(
+            f"{option.name}{_CHOICE_NAME_VALUE_SEPARATOR}{selectable[0].name}"
+        )
+    return format_choice(pairs), unfillable

app/shared/telemetry.py

@@ -34,7 +34,10 @@ from opentelemetry import trace
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
 from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
-from opentelemetry.instrumentation.utils import suppress_instrumentation
+from opentelemetry.instrumentation.utils import (
+    is_instrumentation_enabled,
+    suppress_instrumentation,
+)
 from opentelemetry.sdk.resources import SERVICE_NAME, Resource
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
@@ -52,6 +55,10 @@ R = TypeVar("R")
 
 logger = logging.getLogger(__name__)
 
+# 错误信息类属性的截断长度。站点错误页抽出来的 msg 可能很长（_extract_error_message
+# 拼两条提示），而属性值过长会把 OTLP 请求撑大；排查看的是前半句，够了。
+_MSG_MAX_CHARS = 512
+
 # 全局 provider 引用，用于 instrument_app / shutdown 时判断当前是否已初始化。
 # 显式持有比依赖 trace.get_tracer_provider() 的类型判断更稳——后者在测试场景
 # 下可能被其它用例改动全局状态。
@@ -147,11 +154,21 @@ def shutdown_telemetry() -> None:
     _provider = None
 
 
-def snapshot(span: Span, name: str, html: str | None, max_bytes: int) -> None:
+def snapshot(
+    span: Span,
+    name: str,
+    html: str | None,
+    max_bytes: int,
+    *,
+    extra: Mapping[str, AttributeValue | None] | None = None,
+) -> None:
     """把 HTML 作为 span event 上报，超 max_bytes 截断并标注。
 
     用于解析失败时复现页面：span 自身只放结构化指标（items 数、source 等），
     完整 HTML 体量大、含商品/价格内容，仅在失败分支通过 event 携带。
+
+    `extra` 用来带上「这份 HTML 是哪来的」——落地 URL、页面标题、证据文件路径
+    之类。光有一坨 HTML 还得自己回头对是哪一步的产物，附在同一条 event 上省事。
     """
     if html is None or not html:
         return
@@ -164,6 +181,9 @@ def snapshot(span: Span, name: str, html: str | None, max_bytes: int) -> None:
     }
     if truncated:
         attributes["snapshot.truncated"] = True
+    for key, value in (extra or {}).items():
+        if value is not None:
+            attributes[key] = value
     span.add_event(name, attributes=attributes)
 
 
@@ -174,25 +194,141 @@ def suppressed() -> Iterator[None]:
     给「空转的长轮询」用：worker 每 30 秒问一次网关有没有活干，绝大多数时候
     返回空。这些请求各自成为一条孤立 trace，量大且没有信息量——把观测后台刷满
     的正是它们。领到任务后的每一次网关调用都在任务根 span 底下，不受影响。
+
+    **只挡自动 instrumentation**：OTel 那个上下文标记是给 instrumentation 库看的，
+    手工 `start_as_current_span` 不看它，照样会建 span。所以在这个上下文里手工埋点
+    要走 `span_unless_suppressed`，否则空转长轮询会从另一个口子把孤立 trace 放回来。
     """
     with suppress_instrumentation():
         yield
 
 
+@contextmanager
+def span_unless_suppressed(
+    tracer: trace.Tracer, name: str, *, kind: SpanKind = SpanKind.INTERNAL
+) -> Iterator[Span]:
+    """同 `start_as_current_span`，但在 `suppressed()` 里退化成 noop span。
+
+    手工埋点与 `suppressed()` 的配套件。`suppress_instrumentation` 只被
+    instrumentation 库尊重，手工建的 span 不受它影响——worker 的 `lease` /
+    `lease_query` 正是在 `suppressed()` 里调 `GatewayClient._request` 的，那里若
+    无条件建 span，空转的长轮询就会每 30 秒产出一条孤立 trace，等于绕开了
+    `suppressed()` 本来要解决的问题。
+
+    退化时给的是 `INVALID_SPAN`（NonRecordingSpan）：`set_attributes` /
+    `record_error` / `record_envelope` 作用在它上面全是 noop，调用方不必分支。
+    """
+    if not is_instrumentation_enabled():
+        yield trace.INVALID_SPAN
+        return
+    with tracer.start_as_current_span(name, kind=kind) as span:
+        yield span
+
+
 def record_error(span: Span, exc: BaseException) -> None:
     """把异常记到 span 上并置 ERROR 状态。
 
-    单独抽出来是因为 AppError 带 `err_code`（对外错误码），排查时按码筛比按
-    异常类名筛更贴近上游看到的东西，值得单独落一个属性。
+    单独抽出来是因为 AppError 带的那几个字段（对外错误码 `err_code`、是否可重试
+    `retryable`、HTTP 状态码）正是排查时真正要看的东西：按码筛比按异常类名筛更
+    贴近上游看到的结果，而 `retryable` 直接决定这次失败该不该重来。异常消息也单独
+    落一个属性——`record_exception` 记的 event 在多数观测后台里要展开才看得到，
+    列表页按 `error.message` 筛不出来。
     """
     span.record_exception(exc)
-    span.set_attribute("error.type", type(exc).__name__)
-    err_code = getattr(exc, "err_code", None)
-    if isinstance(err_code, int):
-        span.set_attribute("error.code", err_code)
+    retryable = getattr(exc, "retryable", None)
+    set_attributes(
+        span,
+        {
+            "error.type": type(exc).__name__,
+            "error.message": str(exc)[:_MSG_MAX_CHARS],
+            "error.code": err_code if isinstance(err_code := getattr(exc, "err_code", None), int) else None,
+            "error.retryable": retryable if isinstance(retryable, bool) else None,
+            "error.status_code": sc if isinstance(sc := getattr(exc, "status_code", None), int) else None,
+        },
+    )
     span.set_status(Status(StatusCode.ERROR, f"{type(exc).__name__}: {exc}"))
 
 
+def record_envelope(
+    span: Span,
+    *,
+    success: bool,
+    err_code: int | None = None,
+    msg: str | None = None,
+    status_code: int | None = None,
+) -> None:
+    """把 `ApiResponse` 信封的结果记到 span 上。
+
+    自动 instrumentation 只看 HTTP 层，而本项目的失败**在信封里**：一次
+    `success=false, code=6002` 的回报，HTTP 层跟成功的调用长得一模一样（很多还
+    是 200）。于是 trace 里只剩「调用发生过」，「这次到底成没成、错在哪个码」
+    全在 body 里，不显式记就永远看不到——这正是「只知道调用了、不知道异常怎么
+    来的」的直接原因。
+
+    出入两侧都用它：服务端异常处理器往 server span 上记（见 `shared.api`），
+    worker 解信封时往 client span 上记（见 `trading.worker.client`），同一套
+    `api.*` 属性名，一条 trace 里两侧的结论可以直接对上。
+    """
+    set_attributes(
+        span,
+        {
+            "api.success": success,
+            "api.code": err_code,
+            "api.status_code": status_code,
+            "api.msg": msg[:_MSG_MAX_CHARS] if msg else None,
+        },
+    )
+    if not success:
+        span.set_status(Status(StatusCode.ERROR, msg or f"api.code={err_code}"))
+
+
+def add_event(
+    span: Span, name: str, attributes: Mapping[str, AttributeValue | None]
+) -> None:
+    """记一条 span event，跳过 None 值属性。
+
+    「过程」不能用属性表达：同名属性后写覆盖先写，三次抓取尝试写完只剩最后一次
+    的状态码，中间那两次为什么失败、升级到哪一级全被盖掉了。每次尝试各记一条
+    event，链路里才看得出升级路径。
+    """
+    span.add_event(
+        name,
+        attributes={key: value for key, value in attributes.items() if value is not None},
+    )
+
+
+def record_parse_failure(
+    span: Span,
+    exc: BaseException,
+    *,
+    html: str | None = None,
+    max_bytes: int = 0,
+    url: str | None = None,
+    stage: str | None = None,
+) -> None:
+    """页面类操作失败的统一记法：错误详情 + 失败阶段 + 失败页面快照。
+
+    抓取侧与 ラクマ 侧一共十个接口的失败分支原本各写一遍同样三步，且都漏了最关键
+    的一件事——**失败在哪一步**。`html` 是否已拿到恰好就是判据：还是空说明页面根本
+    没取回来（通道 / 反爬 / 上游 5xx），非空说明取回了但解析不出（多半站点改版）。
+    两者的排查方向完全相反，所以作为属性直接落下来，不让人对着一坨 HTML 猜。
+
+    `stage` 可显式覆盖：像 `shop_items` 那种「转调另外两个接口」的编排方法，失败
+    既不在自己的 fetch 也不在自己的 parse，据 html 推断只会给出错的结论。
+    """
+    record_error(span, exc)
+    set_attributes(
+        span,
+        {
+            "parse.stage": stage or ("fetch" if not html else "parse"),
+            # 保留原有属性名（观测后台的既有筛选条件），但值改成真实异常类名——
+            # 原先无论什么失败都写死 "parse_error"，把反爬阻断也说成解析失败。
+            "parse.fail_reason": type(exc).__name__,
+        },
+    )
+    snapshot(span, "parse.failed_html", html, max_bytes, extra={"parse.url": url})
+
+
 def traced(
     name: str,
     *,

app/trading/worker/client.py

@@ -15,15 +15,25 @@ import logging
 from typing import Any
 
 import httpx
+from opentelemetry import trace
+from opentelemetry.trace import SpanKind
 
 from app.shared.config import Settings
 from app.shared.errors import AppError
 from app.shared.proxy import httpx_client_options
 from app.shared.task_state import OrderState, TaskStatus
+from app.shared.telemetry import (
+    record_envelope,
+    record_error,
+    set_attributes,
+    span_unless_suppressed,
+)
 from app.trading.worker.models import LeaseTask, QueryTask
 
 logger = logging.getLogger(__name__)
 
+tracer = trace.get_tracer(__name__)
+
 
 class GatewayClient:
     """网关 HTTP 客户端
@@ -49,27 +59,58 @@ class GatewayClient:
     # ---- 基础封装 ----
 
     async def _request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]:
-        """发起请求并解信封。失败（success=False）抛 AppError"""
-        response = await self._client.request(method, path, **kwargs)
-        try:
-            body = response.json()
-        except ValueError as exc:
-            raise AppError(
-                message=f"网关响应不是合法 JSON：HTTP {response.status_code}",
-                code="GATEWAY_BAD_BODY",
-                err_code=3001,
-                retryable=True,
-            ) from exc
+        """发起请求并解信封。失败（success=False）抛 AppError
 
-        if not body.get("success"):
-            raise AppError(
-                message=body.get("msg", "网关返回失败"),
-                code="GATEWAY_ERROR",
-                err_code=int(body.get("code", 1500)),
-                retryable=False,
-                status_code=response.status_code,
+        整段套一个自己的 span，而不是依赖 httpx 自动 instrumentation 那个 CLIENT
+        span：**网关的失败在信封里，不在 HTTP 状态码上**。httpx 那个 span 在
+        `request()` 返回时就结束了，此时信封还没解——一次 `success=false, code=6002`
+        （租约无效）的调用在它看来是完成的 200 请求，链路里跟成功毫无区别。
+        本 span 活到解信封之后，所以能把「这次调用的结论」记下来。
+        """
+        with span_unless_suppressed(
+            tracer,
+            f"gateway.{path.strip('/').replace('/', '.')}",
+            kind=SpanKind.CLIENT,
+        ) as span:
+            set_attributes(span, {"gateway.method": method, "gateway.path": path})
+            response = await self._client.request(method, path, **kwargs)
+            try:
+                body = response.json()
+            except ValueError as exc:
+                err = AppError(
+                    message=f"网关响应不是合法 JSON：HTTP {response.status_code}",
+                    code="GATEWAY_BAD_BODY",
+                    err_code=3001,
+                    retryable=True,
+                )
+                record_error(span, err)
+                span.set_attribute("gateway.status_code", response.status_code)
+                raise err from exc
+
+            if not body.get("success"):
+                err_code = int(body.get("code", 1500))
+                msg = body.get("msg", "网关返回失败")
+                # 记成信封结果而不是只抛异常：上报被网关拒时，链路里能直接按
+                # api.code 筛出是哪一类拒绝（6002 租约无效 / 6003 状态不允许…），
+                # 不必回头翻 worker 日志。
+                record_envelope(
+                    span,
+                    success=False,
+                    err_code=err_code,
+                    msg=msg,
+                    status_code=response.status_code,
+                )
+                raise AppError(
+                    message=msg,
+                    code="GATEWAY_ERROR",
+                    err_code=err_code,
+                    retryable=False,
+                    status_code=response.status_code,
+                )
+            record_envelope(
+                span, success=True, status_code=response.status_code
             )
-        return body
+            return body
 
     # ---- 接口 ----
 

app/trading/worker/site_interact.py

@@ -94,7 +94,7 @@ import logging
 import re
 from dataclasses import dataclass, field
 from datetime import datetime
-from typing import TYPE_CHECKING, TypeVar
+from typing import TYPE_CHECKING, Any, TypeVar
 
 from opentelemetry.trace import SpanKind
 
@@ -109,9 +109,12 @@ from app.shared.errors import (
 from app.shared.purchase_contract import (
     INVENTORY_FLAG_DEFAULT,
     INVENTORY_FLAG_MULTIPLE,
+    auto_choice_for,
     base_form_fields,
     basket_domain_of,
+    format_choice,
     inventory_flag_for,
+    parse_options,
 )
 from app.shared.proxy import playwright_launch_proxy
 from app.shared.task_state import OrderState
@@ -646,9 +649,9 @@ class SiteInteractor:
       clear_cart() / remove_item(item_id)
     """
 
-    # 每任务保留的临时状态：task_id → {"item_id": str, "shop_bid": str}
-    # 用于 add_to_cart 把抓出来的 item_id / shop_bid 喂给 verify_cart
-    _per_task_state: dict[str, dict[str, str]]
+    # 每任务保留的临时状态：task_id → 首个商品字段及全部 item_ids。
+    # 旧调用方仍读取 item_id，新调用方由 item_ids 校验整组商品。
+    _per_task_state: dict[str, dict[str, Any]]
 
     # 每任务保留的下单确认页 Page：enter_checkout 落地后不关闭页面，存在这里，
     # submit_order / pay 复用同一个页面继续操作——下单确认页是服务端会话态，
@@ -981,40 +984,36 @@ class SiteInteractor:
     async def add_to_cart(self, task: LeaseTask) -> PageSnapshot:
         """加购（worker 入口）：从 task.intent 取字段，调 _add_to_cart_with_fields
 
-        调用方需在 task.intent 提供：
-        - item_url: 商品详情页 URL（必填）
-        - quantity: 数量，默认 1
-        - variant_id: 多规格商品的 variant_id；不传则从 sku.variants[] 自动选第一个非售罄
-        - choice: 必填选项的取值列表；不传则每个必填选项用第一个候选值（站点不严格校验）
+        调用方可在 task.intent 提供 `items` 商品数组；数组元素字段与旧版单商品
+        字段相同（item_url / quantity / variant_id / choice）。为兼容已发布的
+        调用方，也接受顶层 item_url 等旧字段并自动包装成单元素数组。
 
         Returns:
             PageSnapshot：加购响应的落地页 HTML + 商品页整页截图，供 runner 落证据。
 
         Raises:
-            InvalidRequestError: intent.item_url 缺失
+            InvalidRequestError: intent.item_url 缺失或 intent.items 格式非法
             NotLoggedInError: 登录态失效
             CartOperationError: 商品页打不开、state 解析失败、商品不可购买、加购返回错误页
         """
-        intent = task.intent or {}
-        item_url = intent.get("item_url")
-        if not item_url:
-            raise InvalidRequestError("intent.item_url 必填")
-        quantity = int(intent.get("quantity") or 1)
-        if quantity <= 0:
-            raise InvalidRequestError(f"intent.quantity 必须为正整数，收到 {quantity}")
+        items = _normalize_intent_items(task.intent or {})
 
         async with self._lock:
-            result = await self._add_to_cart_with_fields(
-                item_url=item_url,
-                quantity=quantity,
-                variant_id=intent.get("variant_id"),
-                choice=intent.get("choice"),
-            )
-        self._per_task_state[task.task_id] = {
-            "item_id": result["item_id"],
-            "shop_bid": result["shop_bid"],
-            "basket_domain": result["basket_domain"],
+            results = []
+            for item in items:
+                results.append(await self._add_to_cart_with_fields(**item))
+        first = results[0]
+        state: dict[str, Any] = {
+            # 保留旧字段，避免已有 verify/监控代码及外部桩失效。
+            "item_id": first["item_id"],
+            "shop_bid": first["shop_bid"],
+            "basket_domain": first["basket_domain"],
         }
+        if len(results) > 1:
+            state["item_ids"] = [result["item_id"] for result in results]
+            state["items"] = results
+        self._per_task_state[task.task_id] = state
+        result = results[-1]
         return PageSnapshot(
             html=result.get("response_html") or "",
             screenshot=result.get("screenshot") or b"",
@@ -1102,10 +1101,18 @@ class SiteInteractor:
                 raise CartOperationError(
                     "多规格商品未选 variant，且 sku.variants 全部售罄或为空"
                 )
-            # 必填选项要求填了 choice
-            if fields["has_required_options"] and not fields["form_fields"].get(fields["options_field"]):
+            # 必填选项要求填了 choice。调用方没给 choice 时，只要存在「自动填不了」
+            # 的必填项（自由文本项，或候选值只剩占位项）就当场失败并点名是哪几项——
+            # 这些项非人工给值不可能成功，继续 POST 只会拿站点的
+            # 「未選択の項目からどれか１つ選んでください。」错误页，排查成本更高
+            if fields["has_required_options"] and not fields["form_fields"].get(
+                fields["options_field"]
+            ):
+                unfillable = fields["unfillable_required_options"]
+                detail = f"：{unfillable}" if unfillable else ""
                 raise CartOperationError(
-                    "商品有必填选项但未提供 choice，且选项无候选值"
+                    "商品有必填选项但未提供 choice，且这些必填项无法自动选值"
+                    f"（需在 intent.choice 里按「选项名:取值名」显式给出）{detail}"
                 )
 
             payload = dict(fields["form_fields"])
@@ -1188,8 +1195,15 @@ class SiteInteractor:
             await self._ensure_context_ready()
 
             per_task = self._per_task_state.get(task.task_id, {})
-            item_id = (task.intent or {}).get("item_id") or per_task.get("item_id")
-            if not item_id:
+            intent_item_id = (task.intent or {}).get("item_id")
+            item_ids = (
+                [intent_item_id]
+                if intent_item_id
+                else (per_task.get("item_ids") or [])
+            )
+            if not item_ids and per_task.get("item_id"):
+                item_ids = [per_task["item_id"]]
+            if not item_ids:
                 raise CartOperationError(
                     "无法确定 item_id：intent 未提供且 add_to_cart 未记录"
                 )
@@ -1200,8 +1214,15 @@ class SiteInteractor:
                 raise CartOperationError("购物车为空，加购可能未生效")
             logger.info("cart count=%s task_id=%s", count, task.task_id)
 
-            # 2. 渲染 cart 页确认 item_id 在里面
-            return await self._verify_item_in_cart_html(item_id, label=f"task_id={task.task_id}")
+            # 2. 渲染一次 cart 页确认本任务的全部商品都在里面。
+            # 单商品继续走旧 helper，保留原有测试桩与内部调用契约。
+            if len(item_ids) == 1:
+                return await self._verify_item_in_cart_html(
+                    str(item_ids[0]), label=f"task_id={task.task_id}"
+                )
+            return await self._verify_items_in_cart_html(
+                [str(item_id) for item_id in item_ids], label=f"task_id={task.task_id}"
+            )
 
     @traced("site.cart_status", kind=SpanKind.CLIENT)
     async def cart_status(self) -> dict:
@@ -1425,6 +1446,12 @@ class SiteInteractor:
 
         返回渲染后的 cart 页 HTML + 整页截图（校验通过时），供调用方落证据。
         """
+        return await self._verify_items_in_cart_html([item_id], label=label)
+
+    async def _verify_items_in_cart_html(
+        self, item_ids: list[str], *, label: str
+    ) -> PageSnapshot:
+        """渲染一次 cart SPA，确认多个 item_id 都存在，避免多商品任务重复开页。"""
         page = await self._new_page()
         try:
             await page.goto(_CART_PAGE, wait_until="domcontentloaded", timeout=30_000)
@@ -1437,11 +1464,12 @@ class SiteInteractor:
                     site="rakuten",
                     detail="购物车页出现旧版未登录 marker",
                 )
-            if str(item_id) not in html:
+            missing = [item_id for item_id in item_ids if str(item_id) not in html]
+            if missing:
                 raise CartOperationError(
-                    f"购物车页未找到 item_id={item_id}（加购可能被服务端静默丢弃）"
+                    f"购物车页未找到 item_id={missing}（加购可能被服务端静默丢弃）"
                 )
-            logger.info("cart 校验通过：%s item_id=%s in cart HTML", label, item_id)
+            logger.info("cart 校验通过：%s item_ids=%s in cart HTML", label, item_ids)
             # 校验通过的 cart 页整页截图随结果带出；失败不掩盖校验结果
             screenshot = b""
             try:
@@ -2359,6 +2387,63 @@ class SiteInteractor:
 # ---- 模块级辅助函数（纯函数，便于单测）----
 
 
+def _normalize_intent_items(intent: dict[str, Any]) -> list[dict[str, Any]]:
+    """把新旧下单意图统一成加购参数列表。
+
+    新格式是 ``{"items": [{"item_url": ..., "quantity": ...}, ...]}``；
+    旧格式的 ``item_url/quantity/variant_id/choice`` 仍直接支持。字符串元素
+    也接受，方便只传多个 URL 的调用方。
+    """
+    raw_items = intent.get("items")
+    legacy_single = raw_items is None
+    if raw_items is None:
+        if not intent.get("item_url"):
+            # 保持已发布的单商品错误契约不变。
+            raise InvalidRequestError("intent.item_url 必填")
+        raw_items = [intent]
+    if not isinstance(raw_items, list) or not raw_items:
+        raise InvalidRequestError("intent.items 必须是非空数组")
+
+    normalized: list[dict[str, Any]] = []
+    for index, raw in enumerate(raw_items):
+        if isinstance(raw, str):
+            raw = {"item_url": raw}
+        if not isinstance(raw, dict):
+            raise InvalidRequestError(f"intent.items[{index}] 必须是对象")
+        item_url = raw.get("item_url")
+        if not item_url:
+            raise InvalidRequestError(f"intent.items[{index}].item_url 必填")
+        try:
+            quantity = int(raw.get("quantity") or 1)
+        except (TypeError, ValueError) as exc:
+            field_name = (
+                "intent.quantity"
+                if legacy_single
+                else f"intent.items[{index}].quantity"
+            )
+            raise InvalidRequestError(
+                f"{field_name} 必须为正整数"
+            ) from exc
+        if quantity <= 0:
+            field_name = (
+                "intent.quantity"
+                if legacy_single
+                else f"intent.items[{index}].quantity"
+            )
+            raise InvalidRequestError(
+                f"{field_name} 必须为正整数，收到 {quantity}"
+            )
+        normalized.append(
+            {
+                "item_url": str(item_url),
+                "quantity": quantity,
+                "variant_id": raw.get("variant_id"),
+                "choice": raw.get("choice"),
+            }
+        )
+    return normalized
+
+
 def _parse_initial_state(html: str) -> dict | None:
     """从商品页 HTML 抽 window.__INITIAL_STATE__ 并解析为 dict"""
     m = re.search(
@@ -2421,22 +2506,18 @@ def _extract_purchase_fields(state: dict, *, intent_override: dict | None) -> di
     elif inventory_flag == INVENTORY_FLAG_DEFAULT and item.get("variantId"):
         form_fields["variant_id"] = str(item.get("variantId"))
 
-    # 必填选项：调用方覆盖 > 自动填第一个候选值
-    options = information.get("options") or []
-    required_options = [o for o in options if o.get("isRequired")]
-    has_required = bool(required_options)
+    # 必填选项：调用方覆盖 > 自动填第一个**非占位**候选值
+    # 解析与占位项判定走 app.shared.purchase_contract（与 scraping 的
+    # /api/item_detail 同源），旧实现直接取 values[0]，而必填 select 的 values[0]
+    # 恰恰是「選択してください」占位项，等于把「请选择」当答案提交上去。
+    options = parse_options(information)
+    has_required = any(option.is_required for option in options)
+    auto_choice, unfillable_required = auto_choice_for(options)
     if intent_override.get("choice"):
-        # 调用方给的可能是 list 或 str
-        c = intent_override["choice"]
-        form_fields["choice"] = ",".join(c) if isinstance(c, list) else str(c)
-    elif has_required:
-        pairs: list[str] = []
-        for opt in required_options:
-            values = opt.get("values") or []
-            if values:
-                pairs.append(f"{opt.get('name')}:{values[0].get('name')}")
-        if pairs:
-            form_fields["choice"] = ",".join(pairs)
+        # 调用方给的可能是 list 或 str，两种都交给共用的格式化
+        form_fields["choice"] = format_choice(intent_override["choice"])
+    elif auto_choice:
+        form_fields["choice"] = auto_choice
 
     return {
         "basket_domain": basket_domain,
@@ -2446,6 +2527,9 @@ def _extract_purchase_fields(state: dict, *, intent_override: dict | None) -> di
         "options_field": "choice" if options else "",
         "options": options,
         "has_required_options": has_required,
+        # 必填但自动填不了的选项名（自由文本项，或候选值只有占位项）。调用方没给
+        # choice 时这就是「非人工介入不可能成功」的直接依据，见 _add_to_cart_with_fields
+        "unfillable_required_options": unfillable_required,
         "inventory_flag": inventory_flag,
         "purchase_condition": sell_type.get("purchaseCondition"),
         "min_price": sell_type.get("minPrice"),

app/trading/worker/verify.py

@@ -15,7 +15,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 from enum import StrEnum
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 from urllib.parse import urlsplit
 
 from app.shared.errors import AppError
@@ -55,24 +55,44 @@ def _normalize_item_url(url: str | None) -> str | None:
     return f"{parts.scheme}://{parts.netloc}{parts.path.rstrip('/')}"
 
 
+def _intent_item_urls(intent: dict[str, Any]) -> list[str]:
+    """读取新旧意图中的商品 URL，供恢复核对使用。"""
+    raw_items = intent.get("items")
+    if raw_items is None:
+        raw_items = [intent]
+    if not isinstance(raw_items, list):
+        return []
+    urls: list[str] = []
+    for item in raw_items:
+        if isinstance(item, str):
+            url = item
+        elif isinstance(item, dict):
+            url = item.get("item_url")
+        else:
+            url = None
+        normalized = _normalize_item_url(url)
+        if normalized:
+            urls.append(normalized)
+    return urls
+
+
 async def verify_on_site(
     task: LeaseTask, *, gateway: "GatewayClient", site: "SiteInteractor"
 ) -> VerifyResult:
     """核对一笔任务是否已在站点上下过单
 
-    核对链路：intent.item_url → 查任务创建时间（GET /api/orders/{task_id}，
-    LeaseTask 本身不带 created_at）→ 拉「创建时间之后」的订单列表 → 按商品 URL
-    比对。任何一环拿不到足够信息都返回 UNKNOWN，不猜——尤其是 NOT_ORDERED，
+    核对链路：intent.items（或兼容的 intent.item_url）→ 查任务创建时间
+    （GET /api/orders/{task_id}，LeaseTask 本身不带 created_at）→ 拉「创建时间之后」
+    的订单列表 → 按商品 URL 比对。任何一环拿不到足够信息都返回 UNKNOWN，不猜——尤其是 NOT_ORDERED，
     只有在确认翻完了窗口内的全部订单后才允许返回，否则「没找到」可能只是没翻
     到那一页。
     """
     intent = task.intent or {}
-    item_url = intent.get("item_url")
-    if not item_url:
+    targets = set(_intent_item_urls(intent))
+    if not targets:
         return VerifyResult(
-            VerifyVerdict.UNKNOWN, detail="intent 缺 item_url，无法比对商品"
+            VerifyVerdict.UNKNOWN, detail="intent 缺商品 URL（item_url/items），无法比对商品"
         )
-    target = _normalize_item_url(item_url)
 
     try:
         task_detail = await gateway.get_task(task.task_id)
@@ -96,11 +116,16 @@ async def verify_on_site(
             detail=f"订单列表查询失败：{type(exc).__name__}: {exc}",
         )
 
-    matches = [
-        entry
-        for entry in window.entries
-        if any(_normalize_item_url(it.item_url) == target for it in entry.items)
-    ]
+    matches = []
+    for entry in window.entries:
+        entry_urls = {
+            normalized
+            for normalized in (_normalize_item_url(it.item_url) for it in entry.items)
+            if normalized
+        }
+        # 多商品任务必须在同一笔订单中全部命中，避免部分匹配误判为已下单。
+        if targets.issubset(entry_urls):
+            matches.append(entry)
     if len(matches) == 1:
         return VerifyResult(
             VerifyVerdict.ALREADY_ORDERED,

docs/order-gateway.md

@@ -112,16 +112,22 @@ CREATE TABLE workers (
   "task_id": "po-20260727-0001",   // 可选，上游自带的幂等键；不传则服务端生成
   "site": "rakuten",
   "intent": {                       // gateway 原样透传，结构由 trading 侧定义
-    "item_url": "https://item.rakuten.co.jp/shop/code/",
-    "quantity": 1,
-    "variant_id": "...",            // 多规格商品必填
-    "options": {},
+    "items": [{                     // 新格式：一次购买多个商品
+      "item_url": "https://item.rakuten.co.jp/shop/code/",
+      "quantity": 1,
+      "variant_id": "...",          // 多规格商品必填，取自 /api/item_detail 的 sku.variants[]
+      "choice": ["名入れ:希望する"]   // 店铺自定义必填选项，格式「选项名:取值名」
+    }],
     "max_total_yen": 30000          // 可选，覆盖本次的金额上限
   },
   "callback_url": "https://upstream.example.com/hooks/rakuten-order"  // 可选，终结类事件通知，见 §4.8
 }
 ```
 
+`items` 必须是非空数组，worker 会按顺序将每项加入同一购物车后再进入结算。
+为兼容已发布客户端，也可继续使用旧格式：`intent.item_url` 加同级的
+`quantity` / `variant_id` / `choice`，其语义等同于只有一个元素的 `items`。
+
 响应 `data`：`{"task_id": "...", "status": "queued", "created": true}`
 
 **幂等**：同一个 `task_id` 重复提交不新建任务，返回既有任务且 `created=false`。
@@ -614,4 +620,3 @@ collector 本身不回写「已经存在于下单任务表 `tasks` 的订单」
 - [x] 手动 trigger 立即派一轮 list，返回的单在查询队列里可见
 - [x] `GET /api/account/orders` 列编目；单笔不存在返回 6005
 - [x] collector 只读规范化字段，不解析 raw/raw_pages 站点原始 JSON
-

scripts/probe_option_choice.py

@@ -0,0 +1,245 @@
+"""必填选项 choice 取值的真账号验证：占位项修复是否真的被站点接受
+
+背景：2026-08-28 发现 trading 自动填 choice 时取 `values[0]`，而必填 select 的
+values[0] 恒为 id=0 的「選択してください」占位项，等于把「请选择」当答案提交。
+修复后取第一个非占位候选（见 app/shared/purchase_contract.py::auto_choice_for）。
+
+离线测试只能证明「我们填的值变了」，证明不了「站点接受这个值」——后者必须真账号
+实测。本探针就为这一件事。
+
+**刻意走生产代码路径**（SiteInteractor.add_to_cart_payload），不重写字段构造逻辑：
+上一版探针 scripts/probe_purchase_block_v2.py 自己抄了一遍 form 构造，结果那份
+抄写与生产代码一起用了 values[0]，两边同错就测不出问题。实际提交的 payload 从
+生产代码自己的 INFO 日志里抓（「加购请求：basket=... payload=...」），确保记录的
+就是真正发出去的东西。
+
+安全边界：
+- 只做「清空购物车 → 加购 → 校验 → 清空购物车」，**绝不进入结算/支付**
+- 加购完立即清空，账号购物车恢复原状（加购本身可逆，不产生订单、不扣款）
+- 不触碰 enter_checkout / submit_order / pay
+
+用法：
+    .venv/Scripts/python.exe scripts/probe_option_choice.py
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from app.shared.config import get_settings  # noqa: E402
+from app.shared.errors import AppError  # noqa: E402
+from app.shared.purchase_contract import auto_choice_for, parse_options  # noqa: E402
+from app.trading.services.auth_session import AuthSession  # noqa: E402
+from app.trading.worker.site_interact import (  # noqa: E402
+    SiteInteractor,
+    _parse_initial_state,
+)
+
+PROBE_DIR = Path(__file__).resolve().parent.parent / ".probe" / "options_choice"
+PROBE_DIR.mkdir(parents=True, exist_ok=True)
+
+# 目标商品：从 .probe/checkout/ 的历史样本里挑「有必填选项且占位项在 values[0]」的两个。
+# 这两个正是旧实现填「選択してください」也被站点收下的商品——它们不能证明旧值是对的
+# （店铺没做校验），但可以验证新值同样被接受，且是站点自己给的唯一非占位候选。
+TARGETS = (
+    "https://item.rakuten.co.jp/waabbit/088-4p",
+    "https://item.rakuten.co.jp/aoimorinomise/121",
+)
+
+
+class _PayloadCapture(logging.Handler):
+    """截获生产代码 `加购请求：basket=%s payload=%s` 这条 INFO 日志
+
+    目的是记录**真正发出去的 payload**，而不是探针自己再算一遍。
+    """
+
+    def __init__(self) -> None:
+        super().__init__(level=logging.INFO)
+        self.payloads: list[dict] = []
+
+    def emit(self, record: logging.LogRecord) -> None:
+        if record.msg == "加购请求：basket=%s payload=%s":
+            basket, payload = record.args
+            self.payloads.append({"basket_domain": basket, "payload": dict(payload)})
+
+
+def save(name: str, content: str) -> Path:
+    path = PROBE_DIR / name
+    path.write_text(content, encoding="utf-8")
+    print(f"    saved -> {path.name} ({len(content)} bytes)")
+    return path
+
+
+async def inspect_options(site: SiteInteractor, item_url: str) -> list[dict]:
+    """只读打开商品页，把解析出的选项结构落盘，供人工核对占位项判定是否符合真实页面"""
+    page = await site._context.new_page()
+    try:
+        await page.goto(item_url, wait_until="domcontentloaded", timeout=30_000)
+        await page.wait_for_function(
+            "() => window.__INITIAL_STATE__ && window.__INITIAL_STATE__.purchase",
+            timeout=10_000,
+        )
+        html = await page.content()
+    finally:
+        await page.close()
+
+    state = _parse_initial_state(html) or {}
+    information = (state.get("purchase") or {}).get("information") or {}
+    options = parse_options(information)
+    auto_choice, unfillable = auto_choice_for(options)
+
+    slug = item_url.rstrip("/").replace("https://item.rakuten.co.jp/", "").replace("/", "_")
+    save(
+        f"options-{slug}.json",
+        json.dumps(
+            {
+                "item_url": item_url,
+                "raw_options": information.get("options"),
+                "parsed": [
+                    {
+                        "name": option.name,
+                        "type": option.type,
+                        "is_required": option.is_required,
+                        "values": [
+                            {
+                                "value_id": value.value_id,
+                                "name": value.name,
+                                "is_placeholder": value.is_placeholder,
+                            }
+                            for value in option.values
+                        ],
+                    }
+                    for option in options
+                ],
+                "auto_choice": auto_choice,
+                "unfillable_required_options": unfillable,
+            },
+            ensure_ascii=False,
+            indent=1,
+        ),
+    )
+
+    for option in options:
+        if not option.is_required:
+            continue
+        names = [
+            f"{value.name}{'(占位)' if value.is_placeholder else ''}" for value in option.values
+        ]
+        print(f"    必填项「{option.name[:30]}」候选={names}")
+    print(f"    auto_choice = {auto_choice!r}")
+    if unfillable:
+        print(f"    !! 自动填不了的必填项：{unfillable}")
+    return [{"name": o.name, "is_required": o.is_required} for o in options]
+
+
+async def run() -> int:
+    settings = get_settings()
+    capture = _PayloadCapture()
+    logging.basicConfig(level=logging.WARNING)
+    # 必须显式把这个 logger 抬到 INFO：basicConfig 把 root 设成 WARNING，而目标
+    # logger 自身是 NOTSET，有效级别继承 root——INFO 记录会在到达 handler 之前
+    # 就被丢掉，capture 全程收不到东西（第一次跑就是这样，choice 打印成 None）
+    target_logger = logging.getLogger("app.trading.worker.site_interact")
+    target_logger.setLevel(logging.INFO)
+    target_logger.addHandler(capture)
+
+    auth = AuthSession(settings)
+    await auth.start()
+
+    print("=== 0. 登录态探测（storage_state 落盘于 2026-08-11，可能已过期）===")
+    status = await auth.check("rakuten")
+    print(f"    logged_in={status.logged_in} detail={status.detail}")
+    if not status.logged_in:
+        print("    登录态已失效——需要先跑 scripts/login.py 重新登录（可能要人工过验证码）")
+        await auth.close()
+        return 2
+
+    site = SiteInteractor(auth_session=auth, settings=settings)
+    await site.start()
+
+    results: list[dict] = []
+    try:
+        print("\n=== 1. 清空购物车（排除历史残留污染校验）===")
+        cleared = await site.clear_cart()
+        print(f"    removed={cleared['removed_count']} cart_count={cleared['cart_count']}")
+
+        for index, item_url in enumerate(TARGETS, start=1):
+            print(f"\n=== 2.{index} {item_url} ===")
+            print("  -- 只读核对选项结构 --")
+            try:
+                await inspect_options(site, item_url)
+            except Exception as exc:  # noqa: BLE001
+                print(f"    选项结构读取失败（不影响加购验证）：{type(exc).__name__}: {exc}")
+
+            print("  -- 走生产路径加购（add_to_cart_payload）--")
+            before = len(capture.payloads)
+            entry: dict = {"item_url": item_url}
+            try:
+                added = await site.add_to_cart_payload(item_url=item_url, quantity=1)
+                entry["ok"] = True
+                entry["item_id"] = added["item_id"]
+                entry["cart_count"] = added["cart_count"]
+                print(
+                    f"    加购成功 item_id={added['item_id']} cart_count={added['cart_count']}"
+                )
+                save(f"add-response-{index}.html", added.get("response_html") or "")
+            except AppError as exc:
+                entry["ok"] = False
+                entry["error"] = exc.message
+                print(f"    加购失败：{exc.message}")
+            except Exception as exc:  # noqa: BLE001
+                entry["ok"] = False
+                entry["error"] = f"{type(exc).__name__}: {exc}"
+                print(f"    加购异常：{type(exc).__name__}: {exc}")
+
+            # 真正发出去的 payload（从生产代码日志截获）
+            sent = capture.payloads[before:]
+            if not sent:
+                # 抓不到就等于这次验证什么也没证明——必须显眼报出来，不能让
+                # 「加购成功」把它盖过去（第一次跑就是 logger 级别没抬导致静默为 None）
+                entry["capture_failed"] = True
+                print("    !! 未截获到 payload 日志，本次无法确认实际提交的 choice")
+            else:
+                entry["sent_payload"] = sent[-1]["payload"]
+                choice = sent[-1]["payload"].get("choice")
+                print(f"    实际提交 choice = {choice!r}")
+                if choice is None:
+                    print("    !! payload 里没有 choice 字段")
+                elif "選択してください" in choice:
+                    print("    !! 提交值里仍含占位项——修复未生效")
+            results.append(entry)
+
+        print("\n=== 3. 购物车状态（确认商品真的进车了）===")
+        try:
+            status_after = await site.cart_status()
+            print(f"    {status_after}")
+        except AppError as exc:
+            print(f"    查询失败：{exc.message}")
+
+        print("\n=== 4. 清空购物车（恢复账号原状，不进入结算）===")
+        final = await site.clear_cart()
+        print(f"    removed={final['removed_count']} cart_count={final['cart_count']}")
+    finally:
+        await site.close()
+        await auth.close()
+
+    save("summary.json", json.dumps(results, ensure_ascii=False, indent=1))
+    print("\n========== 汇总 ==========")
+    for entry in results:
+        mark = "OK  " if entry.get("ok") else "FAIL"
+        choice = (entry.get("sent_payload") or {}).get("choice")
+        print(f"  {mark} {entry['item_url']}")
+        print(f"       choice={choice!r}")
+        if not entry.get("ok"):
+            print(f"       error={entry.get('error')}")
+    print(f"\n探针输出目录：{PROBE_DIR}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(asyncio.run(run()))

tests/test_api_telemetry.py

@@ -0,0 +1,154 @@
+"""对外失败响应的埋点测试：异常处理器是失败的唯一出口
+
+`register_exception_handlers` 里的处理器把异常吃掉、换成 `ApiResponse` 信封返回，
+异常不再向上冒——**自动 instrumentation 之后只看得到一个 HTTP 状态码**。而
+`AppError` 默认 `status_code=400`、信封里 `success=false`，在链路上跟正常返回几乎
+分不出来；兜底处理器更是只回一句无信息量的 "Internal server error"，真正的异常
+类型与栈只在本进程日志里。所以处理器必须把这次失败记到当前 server span 上。
+
+这里挂真实的 FastAPIInstrumentor 中间件（而不是手工造 span），断言的就是「一次
+真实请求打进来、失败返回之后，server span 上有什么」。
+"""
+from __future__ import annotations
+
+import pytest
+from fastapi import APIRouter, FastAPI
+from fastapi.testclient import TestClient
+from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+from opentelemetry.trace import StatusCode
+from pydantic import BaseModel
+
+from app.shared.api import ApiResponse, register_exception_handlers
+from app.shared.errors import ItemNotFoundError, UpstreamBlockedError
+
+
+class _Body(BaseModel):
+    keyword: str
+
+
+@pytest.fixture
+def app_and_spans():
+    """挂了异常处理器 + 真实 OTel ASGI 中间件的最小应用
+
+    provider 显式传给 instrument_app，避免碰全局 provider（只允许设置一次，
+    测试间共享会互相污染）。
+    """
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+
+    router = APIRouter()
+
+    @router.get("/api/blocked")
+    async def blocked() -> ApiResponse[None]:
+        raise UpstreamBlockedError("Blocked while fetching: challenge page detected")
+
+    @router.get("/api/missing")
+    async def missing() -> ApiResponse[None]:
+        raise ItemNotFoundError("Page not found: https://item.rakuten.co.jp/x/y/")
+
+    @router.get("/api/boom")
+    async def boom() -> ApiResponse[None]:
+        raise RuntimeError("unexpected explosion")
+
+    @router.post("/api/validated")
+    async def validated(_body: _Body) -> ApiResponse[None]:
+        return ApiResponse[None](success=True, msg="success", data=None, code=0)
+
+    app = FastAPI()
+    app.include_router(router)
+    register_exception_handlers(app)
+    FastAPIInstrumentor.instrument_app(app, tracer_provider=provider)
+    try:
+        yield app, exporter
+    finally:
+        FastAPIInstrumentor.uninstrument_app(app)
+
+
+def _server_span(exporter: InMemorySpanExporter):
+    spans = exporter.get_finished_spans()
+    assert spans, "没有产生 server span"
+    return spans[-1]
+
+
+def test_app_error_is_recorded_on_server_span(app_and_spans):
+    """业务异常：错误码、可重试、异常类型都要落在 span 上
+
+    AppError 默认 status_code=400，链路里光看状态码只知道「客户端错了」，
+    不知道是被反爬阻断（3002，可重试）还是别的什么。
+    """
+    app, exporter = app_and_spans
+    with TestClient(app) as http:
+        response = http.get("/api/blocked")
+    assert response.status_code == 400
+    assert response.json()["code"] == 3002
+
+    span = _server_span(exporter)
+    assert span.status.status_code is StatusCode.ERROR
+    assert span.attributes["error.type"] == "UpstreamBlockedError"
+    assert span.attributes["error.code"] == 3002
+    # 上游据此判断该不该重试，是失败分类里最要紧的一位
+    assert span.attributes["error.retryable"] is True
+    assert span.attributes["api.success"] is False
+    assert span.attributes["api.code"] == 3002
+    assert "challenge page" in span.attributes["api.msg"]
+    # 异常栈作为 event 保留，需要细看时能展开
+    assert any(event.name == "exception" for event in span.events)
+
+
+def test_not_found_records_its_own_code(app_and_spans):
+    """404 类业务异常与阻断类要能按码区分（4004 不可重试）"""
+    app, exporter = app_and_spans
+    with TestClient(app) as http:
+        assert http.get("/api/missing").status_code == 404
+
+    span = _server_span(exporter)
+    assert span.attributes["error.code"] == 4004
+    assert span.attributes["error.retryable"] is False
+
+
+def test_unhandled_exception_records_real_cause(app_and_spans):
+    """兜底分支最需要埋点：对外只回 "Internal server error"，真因只在日志里"""
+    app, exporter = app_and_spans
+    with TestClient(app, raise_server_exceptions=False) as http:
+        response = http.get("/api/boom")
+    assert response.status_code == 500
+    assert response.json()["msg"] == "Internal server error"
+
+    span = _server_span(exporter)
+    assert span.status.status_code is StatusCode.ERROR
+    # 响应体里查不到的真因，链路上能直接看到
+    assert span.attributes["error.type"] == "RuntimeError"
+    assert "unexpected explosion" in span.attributes["error.message"]
+    assert span.attributes["api.code"] == 1500
+
+
+def test_validation_error_records_field_message(app_and_spans):
+    """参数校验失败记错误码与字段消息，但不记 pydantic 那条没有诊断价值的长栈"""
+    app, exporter = app_and_spans
+    with TestClient(app) as http:
+        response = http.post("/api/validated", json={})
+    assert response.status_code == 422
+    assert response.json()["code"] == 1002
+
+    span = _server_span(exporter)
+    assert span.status.status_code is StatusCode.ERROR
+    assert span.attributes["api.code"] == 1002
+    # 排查要看的是「哪个字段不合法」
+    assert "keyword" in span.attributes["api.msg"]
+    assert "error.type" not in span.attributes
+
+
+def test_success_leaves_server_span_ok(app_and_spans):
+    """成功请求不被误标 ERROR——作为对照说明失败断言不是恒真"""
+    app, exporter = app_and_spans
+    with TestClient(app) as http:
+        assert http.post("/api/validated", json={"keyword": "switch"}).status_code == 200
+
+    span = _server_span(exporter)
+    assert span.status.status_code is not StatusCode.ERROR
+    assert "api.success" not in span.attributes
+    assert "error.type" not in span.attributes

tests/test_openapi_export.py

@@ -103,3 +103,16 @@ def test_all_refs_resolve(spec):
     names = set(spec["components"]["schemas"])
     refs = set(re.findall(r"#/components/schemas/([^\"]+)", json.dumps(spec)))
     assert not refs - names
+
+
+def test_submit_order_intent_schema_documents_multi_item_and_legacy_fields(spec):
+    """intent 保持透传对象，同时在 OpenAPI 中明确展示新旧两种商品格式。"""
+    intent = spec["components"]["schemas"]["SubmitOrderRequest"]["properties"]["intent"]
+    properties = intent["properties"]
+    assert intent["additionalProperties"] is True
+    assert properties["items"]["type"] == "array"
+    assert properties["items"]["minItems"] == 1
+    item_object = properties["items"]["items"]["oneOf"][0]
+    assert item_object["required"] == ["item_url"]
+    assert "item_url" in properties
+    assert "quantity" in properties

tests/test_parsers.py

@@ -142,6 +142,79 @@ def test_item_detail_falls_back_to_request_shop_code_when_state_lacks_it(item_st
     assert detail.shop.shop_code == "fallback"
 
 
+# ---- 店铺自定义选项（下单必填项）----
+
+def test_item_detail_exposes_shop_options_for_ordering(item_with_options_state):
+    """下单要用的选项必须能从接口拿到：名字、类型、是否必填、候选取值"""
+    detail = parse_item_detail(
+        item_with_options_state, item_url="https://x", shop_code="kizamu",
+        include_sku_variants=True,
+    )
+    assert detail.has_required_options is True
+    # 真实样本：1 个必填 select（名入れ）+ 1 个非必填 select + 2 个自由文本
+    assert [option.name for option in detail.options] == [
+        "名入れ",
+        " 【お名前】※漢字・かな文字：4文字まで",
+        "【定型】",
+        "【定型】※〇寿・〇婚式・〇長を選択の場合のみ(1文字まで)",
+    ]
+    required = detail.options[0]
+    assert required.is_required is True
+    assert required.type == "select"
+    assert required.option_id == 100
+
+
+def test_item_detail_marks_placeholder_option_values_as_unselectable(item_with_options_state):
+    """「選択してください」是占位项，不能作为下单取值——这是 choice 填错的根因"""
+    detail = parse_item_detail(
+        item_with_options_state, item_url="https://x", shop_code="kizamu",
+        include_sku_variants=True,
+    )
+    required = detail.options[0]
+    placeholder, first_real = required.values[0], required.values[1]
+    assert placeholder.name == "選択してください"
+    assert placeholder.is_placeholder is True
+    assert first_real.name == "希望する【次の項目で入力】"
+    assert first_real.is_placeholder is False
+    # 3 个候选里只有 2 个真正可提交
+    assert len(required.values) == 3
+    assert required.selectable_value_count == 2
+
+
+def test_item_detail_free_text_option_has_no_values(item_with_options_state):
+    """type=text 是自由文本，站点不给候选值，无法自动选"""
+    detail = parse_item_detail(
+        item_with_options_state, item_url="https://x", shop_code="kizamu",
+        include_sku_variants=True,
+    )
+    free_text = detail.options[1]
+    assert free_text.type == "text"
+    assert free_text.values == []
+    assert free_text.selectable_value_count == 0
+
+
+def test_item_detail_reports_required_options_that_cannot_be_auto_filled(item_with_options_state):
+    """必填自由文本项无法自动选值，必须点名交给调用方，不能静默蒙一个值"""
+    options = item_with_options_state["purchase"]["information"]["options"]
+    # 把那个自由文本项改成必填：店铺要求填「お名前」时就是这种形态
+    options[1]["isRequired"] = True
+    detail = parse_item_detail(
+        item_with_options_state, item_url="https://x", shop_code="kizamu",
+        include_sku_variants=True,
+    )
+    assert detail.unfillable_required_options == [" 【お名前】※漢字・かな文字：4文字まで"]
+
+
+def test_item_detail_without_options_reports_empty(item_state):
+    """没有选项的商品（真实样本 purchase.information 只有 unit）不应凭空造出选项"""
+    detail = parse_item_detail(
+        item_state, item_url="https://x", shop_code="s", include_sku_variants=False,
+    )
+    assert detail.options == []
+    assert detail.has_required_options is False
+    assert detail.unfillable_required_options == []
+
+
 def test_item_detail_rejects_state_without_item_node():
     with pytest.raises(ScrapeParseError):
         parse_item_detail({}, item_url="https://x", shop_code="s", include_sku_variants=True)

tests/test_scrape_telemetry.py

@@ -0,0 +1,296 @@
+"""抓取客户端失败分支的埋点测试：失败原因必须能从 span 上读出来
+
+这里钉住的核心是一个真实存在过的 bug：失败分支里写的是无参
+`span.record_exception()`，而该方法的 `exception` 是必填位置参数——**每一次抓取
+失败都会在记录异常时抛 TypeError**，把真正的失败原因（反爬阻断 / 解析失败 /
+404）整个替换掉。链路上只剩「调用发生过」，异常怎么来的完全看不到，失败页面
+快照也永远落不下来（抛错发生在 snapshot 之前）。
+
+用独立的 InMemory provider 断言 span，不碰全局 provider——OTel 的全局 provider
+只允许设置一次，测试间共享会互相污染（与 test_telemetry.py 同一套思路）。
+"""
+from __future__ import annotations
+
+import json
+
+import httpx
+import pytest
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+from opentelemetry.trace import StatusCode
+
+from app.scraping.models.scrape import (
+    ItemDetailRequest,
+    RakumaSearchRequest,
+    SearchRequest,
+    ShopItemsRequest,
+)
+from app.scraping.services import rakuma_client as rakuma_module
+from app.scraping.services import rakuten_client as rakuten_module
+from app.scraping.services import site_session as session_module
+from app.scraping.services.rakuma_client import RakumaClient
+from app.scraping.services.rakuten_client import RakutenClient
+from app.scraping.services.site_session import SiteSession
+from app.shared.config import Settings
+from app.shared.errors import AppError, ScrapeParseError, UpstreamBlockedError
+
+
+@pytest.fixture
+def spans(monkeypatch) -> InMemorySpanExporter:
+    """把两个抓取客户端模块级 tracer 换成写内存的，用于断言 span"""
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    tracer = provider.get_tracer("test")
+    monkeypatch.setattr(rakuten_module, "tracer", tracer)
+    monkeypatch.setattr(rakuma_module, "tracer", tracer)
+    return exporter
+
+
+def _settings() -> Settings:
+    return Settings(_env_file=None, otel_snapshot_max_bytes=2_000_000)
+
+
+class _StubSession:
+    """按需返回 HTML 或抛异常的会话桩；两站的 fetch_html 签名差异在这里吸收"""
+
+    def __init__(self, *, html: str = "", error: Exception | None = None):
+        self._html = html
+        self._error = error
+
+    async def fetch_html(self, url: str, *, mobile: bool = False) -> str:
+        if self._error is not None:
+            raise self._error
+        return self._html
+
+    async def fetch(self, url: str, *, mobile: bool = False, validator=None):
+        if self._error is not None:
+            raise self._error
+        raise AssertionError("本测试不该走到 fetch 的成功分支")
+
+
+def _span(exporter: InMemorySpanExporter, name: str):
+    return next(s for s in exporter.get_finished_spans() if s.name == name)
+
+
+async def test_parse_failure_preserves_original_exception(spans):
+    """回归：解析失败要原样抛出业务异常，不能被埋点自身的 TypeError 顶替
+
+    这是原 bug 最直接的后果——上游拿到的不再是 4001（解析失败），而是一个
+    TypeError 兜底成的 500，错误码表整个失效。
+    """
+    client = RakutenClient(_settings(), _StubSession(html="<html>no state</html>"))
+
+    with pytest.raises(ScrapeParseError) as excinfo:
+        await client.search(SearchRequest(keyword="switch"))
+    # 是业务异常而不是埋点炸出来的 TypeError
+    assert isinstance(excinfo.value, AppError)
+    assert excinfo.value.err_code == 4001
+
+
+async def test_parse_failure_records_reason_and_page_snapshot(spans):
+    """页面取回来了但解析不出：stage=parse，且失败页面 HTML 落成 event
+
+    「站点改版了」只能靠当时那份 HTML 判断，所以快照必须真的落下来——原 bug 里
+    记异常那步先抛了，snapshot 这行永远执行不到。
+    """
+    html = "<html>changed layout</html>"
+    client = RakutenClient(_settings(), _StubSession(html=html))
+
+    with pytest.raises(ScrapeParseError):
+        await client.search(SearchRequest(keyword="switch"))
+
+    span = _span(spans, "parse.rakuten.search")
+    assert span.status.status_code is StatusCode.ERROR
+    assert span.attributes["parse.stage"] == "parse"
+    # 真实异常类名，不是一律写死的 "parse_error"
+    assert span.attributes["parse.fail_reason"] == "ScrapeParseError"
+    assert span.attributes["error.code"] == 4001
+    assert span.attributes["error.retryable"] is False
+
+    snapshot = next(e for e in span.events if e.name == "parse.failed_html")
+    assert snapshot.attributes["snapshot.html"] == html
+    # 快照要能对上是哪个地址的页面
+    assert "search.rakuten.co.jp" in snapshot.attributes["parse.url"]
+
+
+async def test_fetch_failure_marks_fetch_stage_without_snapshot(spans):
+    """页面根本没取回来：stage=fetch，没有快照可落
+
+    与 parse 阶段的排查方向相反（通道/反爬/上游 5xx，而不是站点改版），
+    所以要能直接按 parse.stage 分流，而不是对着有没有 HTML 猜。
+    """
+    blocked = UpstreamBlockedError("Blocked while fetching: challenge page detected")
+    client = RakutenClient(_settings(), _StubSession(error=blocked))
+
+    with pytest.raises(UpstreamBlockedError):
+        await client.search(SearchRequest(keyword="switch"))
+
+    span = _span(spans, "parse.rakuten.search")
+    assert span.attributes["parse.stage"] == "fetch"
+    assert span.attributes["parse.fail_reason"] == "UpstreamBlockedError"
+    # 反爬阻断是可重试的，这一位直接决定上游要不要重来
+    assert span.attributes["error.retryable"] is True
+    assert not [e for e in span.events if e.name == "parse.failed_html"]
+
+
+async def test_item_detail_failure_records_reason(spans):
+    """详情接口走的是 fetch()（带校验器）而非 fetch_html，失败分支同样要记全"""
+    blocked = UpstreamBlockedError("challenge page detected")
+    client = RakutenClient(_settings(), _StubSession(error=blocked))
+
+    with pytest.raises(UpstreamBlockedError):
+        await client.item_detail(
+            ItemDetailRequest(shop_code="someshop", item_code="10000001")
+        )
+
+    span = _span(spans, "parse.rakuten.item_detail")
+    assert span.status.status_code is StatusCode.ERROR
+    assert span.attributes["parse.fail_reason"] == "UpstreamBlockedError"
+
+
+async def test_shop_items_marks_delegate_stage(spans):
+    """shop_items 自己不抓页面：失败在转调的 shop_detail / search 里
+
+    按 html 推断会得出「fetch 失败」的错误结论（它手里从来没有 html），
+    所以这里显式标 delegate，且不落快照——真正的现场在被转调那个 span 上。
+    """
+    blocked = UpstreamBlockedError("challenge page detected")
+    client = RakutenClient(_settings(), _StubSession(error=blocked))
+
+    with pytest.raises(UpstreamBlockedError):
+        await client.shop_items(ShopItemsRequest(shop_code="someshop"))
+
+    span = _span(spans, "parse.rakuten.shop_items")
+    assert span.attributes["parse.stage"] == "delegate"
+    assert not [e for e in span.events if e.name == "parse.failed_html"]
+    # 被转调的那一步才是现场所在，它自己落了快照
+    inner = _span(spans, "parse.rakuten.shop_detail")
+    assert inner.attributes["parse.fail_reason"] == "UpstreamBlockedError"
+
+
+async def test_rakuma_parse_failure_records_reason_and_snapshot(spans):
+    """ラクマ 侧五个接口是同一套失败分支，同样要能读出原因与现场"""
+    html = "<html>rakuma changed</html>"
+    client = RakumaClient(_settings(), _StubSession(html=html))
+
+    with pytest.raises(AppError):
+        await client.search(RakumaSearchRequest(keyword="switch"))
+
+    span = _span(spans, "parse.rakuma.search")
+    assert span.status.status_code is StatusCode.ERROR
+    assert span.attributes["parse.stage"] == "parse"
+    assert "error.code" in span.attributes
+    snapshot = next(e for e in span.events if e.name == "parse.failed_html")
+    assert snapshot.attributes["snapshot.html"] == html
+
+
+# ---- 会话层：逐次尝试与升级路径 ----
+#
+# 这一层的关键信息是「升级路径」：换 cookie → 浏览器兜底 → 放弃。属性表达不了过程
+# （同名后写覆盖先写，三次尝试跑完只剩最后一次），所以每次尝试与每次升级各记一条
+# event。site_session / rakuma_session 都在 fetch 内部 `trace.get_tracer(__name__)`，
+# 所以这里替 `trace.get_tracer` 本身（与 test_telemetry.py 的 spans 夹具同一手法）。
+
+
+@pytest.fixture
+def session_spans(monkeypatch) -> InMemorySpanExporter:
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    monkeypatch.setattr(session_module.trace, "get_tracer", provider.get_tracer)
+    return exporter
+
+
+class _FakeBrowser:
+    """浏览器兜底替身：visit 恒不可用，让升级链走完整条路"""
+
+    def __init__(self) -> None:
+        self.unavailable_reason = "playwright is not installed"
+
+    async def visit(self, url: str, *, mobile: bool):
+        return None
+
+    async def close(self) -> None:
+        pass
+
+
+async def _blocked_session() -> SiteSession:
+    """所有请求都回 Akamai 挑战页的会话，用 MockTransport 拦截"""
+    settings = Settings(
+        _env_file=None, http_max_attempts=3, request_timeout_seconds=5.0,
+        max_site_concurrency=4,
+    )
+    session = SiteSession(settings, _FakeBrowser())
+    await session.start()
+    for profile in session._profiles.values():
+        await profile.client.aclose()
+        profile.client = httpx.AsyncClient(
+            transport=httpx.MockTransport(
+                lambda _r: httpx.Response(
+                    200, text="<html>Access Denied. Reference #18.abc</html>"
+                )
+            ),
+            follow_redirects=True,
+        )
+    return session
+
+
+async def test_retry_attempts_and_escalation_are_recorded_as_events(session_spans):
+    """每次尝试与每次升级各留一条 event，链路上能读出完整升级路径
+
+    这些信息无法用属性表达：`scrape.attempts` 只剩最后一次的值，前两次为什么失败、
+    换过 cookie 没有、浏览器兜底试过没有全被覆盖掉。
+    """
+    session = await _blocked_session()
+    try:
+        with pytest.raises(UpstreamBlockedError):
+            await session.fetch_html(
+                "https://search.rakuten.co.jp/search/mall/x/", mobile=False
+            )
+    finally:
+        await session.close()
+
+    span = _span(session_spans, "scrape.fetch")
+    assert span.status.status_code is StatusCode.ERROR
+    assert span.attributes["scrape.fail_reason"] == "UpstreamBlockedError"
+    assert span.attributes["scrape.challenge_detected"] is True
+
+    # 三次尝试都留下了自己的记录，而不是只剩最后一次
+    attempts = [e for e in span.events if e.name == "scrape.attempt"]
+    assert [e.attributes["attempt"] for e in attempts] == [1, 2, 3]
+    assert {e.attributes["outcome"] for e in attempts} == {"challenge"}
+
+    # 升级路径：第 1 次失败后换 cookie，第 2 次失败后动用浏览器（本例不可用）
+    escalations = [e for e in span.events if e.name == "scrape.escalate"]
+    assert [e.attributes["to"] for e in escalations] == ["rewarm_on_home", "browser"]
+    browser_step = escalations[-1]
+    assert browser_step.attributes["outcome"] == "failed"
+    # 「没装 playwright」与「装了也被挡」要能分开查
+    assert browser_step.attributes["reason"] == "playwright is not installed"
+
+    # 最终失败页面的快照带上来源，便于确认是哪条通道的哪个地址
+    failed = next(e for e in span.events if e.name == "scrape.failed_html")
+    assert failed.attributes["scrape.profile"] == "pc"
+    assert "search.rakuten.co.jp" in failed.attributes["scrape.url"]
+
+
+async def test_successful_scrape_leaves_span_ok(spans, search_state):
+    """成功路径不被误标 ERROR，不落失败快照，且照常记结果指标
+
+    作为上面那些失败断言的对照：证明 ERROR 状态与 parse.fail_reason 是真的由失败
+    触发的，不是每条 span 都长这样。
+    """
+    html = f"<script>window.__INITIAL_STATE__ = {json.dumps(search_state)};</script>"
+    client = RakutenClient(_settings(), _StubSession(html=html))
+
+    result = await client.search(SearchRequest(keyword="switch"))
+
+    span = _span(spans, "parse.rakuten.search")
+    assert span.status.status_code is not StatusCode.ERROR
+    assert "parse.fail_reason" not in span.attributes
+    assert "error.type" not in span.attributes
+    assert not [e for e in span.events if e.name == "parse.failed_html"]
+    # 成功时记的是结果指标，与失败侧属性互不重叠
+    assert span.attributes["parse.items"] == len(result.items)

tests/test_site_interact.py

@@ -11,6 +11,7 @@ clear_cart 与 _dump_debug_snapshot 用不依赖 Playwright 的 fake page 覆盖
 """
 from __future__ import annotations
 
+import asyncio
 import json
 from pathlib import Path
 from typing import Any
@@ -39,6 +40,7 @@ from app.trading.worker.site_interact import (
     _DELETE_BUTTON_SELECTOR,
     _extract_error_message,
     _extract_purchase_fields,
+    _normalize_intent_items,
     _OrderListAccumulator,
     _parse_checkout_summary,
     _parse_initial_state,
@@ -70,6 +72,82 @@ def _wrap_state(state: dict[str, Any]) -> str:
     )
 
 
+# ---- 下单意图兼容 ----
+
+
+def test_normalize_intent_items_accepts_legacy_single_item():
+    assert _normalize_intent_items({
+        "item_url": "https://item.rakuten.co.jp/shop/x/",
+        "quantity": 2,
+    }) == [{
+        "item_url": "https://item.rakuten.co.jp/shop/x/",
+        "quantity": 2,
+        "variant_id": None,
+        "choice": None,
+    }]
+
+
+def test_normalize_intent_items_accepts_multiple_items_and_string_urls():
+    assert _normalize_intent_items({
+        "items": [
+            {"item_url": "https://item.rakuten.co.jp/shop/x/", "quantity": 2},
+            "https://item.rakuten.co.jp/shop/y/",
+        ]
+    }) == [
+        {
+            "item_url": "https://item.rakuten.co.jp/shop/x/",
+            "quantity": 2,
+            "variant_id": None,
+            "choice": None,
+        },
+        {
+            "item_url": "https://item.rakuten.co.jp/shop/y/",
+            "quantity": 1,
+            "variant_id": None,
+            "choice": None,
+        },
+    ]
+
+
+def test_normalize_intent_items_rejects_empty_items():
+    with pytest.raises(InvalidRequestError):
+        _normalize_intent_items({"items": []})
+
+
+async def test_add_to_cart_processes_all_items_and_keeps_legacy_state():
+    site = SiteInteractor.__new__(SiteInteractor)
+    site._lock = asyncio.Lock()
+    site._per_task_state = {}
+    calls: list[dict[str, Any]] = []
+
+    async def fake_add(**kwargs):
+        calls.append(kwargs)
+        index = len(calls)
+        return {
+            "item_id": str(index),
+            "shop_bid": "shop",
+            "basket_domain": "https://basket",
+            "response_html": f"html-{index}",
+            "screenshot": b"",
+        }
+
+    site._add_to_cart_with_fields = fake_add
+    snapshot = await site.add_to_cart(_make_task(intent={
+        "items": [
+            {"item_url": "https://item.rakuten.co.jp/shop/x/"},
+            {"item_url": "https://item.rakuten.co.jp/shop/y/", "quantity": 3},
+        ]
+    }))
+    assert [call["item_url"] for call in calls] == [
+        "https://item.rakuten.co.jp/shop/x/",
+        "https://item.rakuten.co.jp/shop/y/",
+    ]
+    assert calls[1]["quantity"] == 3
+    assert site._per_task_state["t1"]["item_id"] == "1"
+    assert site._per_task_state["t1"]["item_ids"] == ["1", "2"]
+    assert snapshot.html == "html-2"
+
+
 # ---- _parse_initial_state ----
 
 
@@ -180,6 +258,95 @@ def test_extract_fields_required_options_auto_picks_first_value():
     assert fields["form_fields"]["choice"] == "サイズ:S"
 
 
+def test_extract_fields_required_options_skips_placeholder_value():
+    """回归：必填 select 的 values[0] 是「選択してください」占位项，不能当答案提交
+
+    真实商品页（.probe/checkout/03-add-fail-2.html 等 3 份样本）里必填 select 的
+    第一个候选恒为 id=0 的占位项，真实取值从 id=200 起。旧实现直接取 values[0]，
+    等于把「請選擇」填进 choice 提交上去。
+    """
+    state = {
+        "item": {"itemId": 1, "variantId": "v"},
+        "purchase": {
+            "sku": {"inventoryType": "single"},
+            "sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}},
+            "information": {
+                "options": [
+                    {
+                        "id": 100,
+                        "name": "この商品は「トライタン製／プラスチック」です",
+                        "type": "select",
+                        "isRequired": True,
+                        "values": [
+                            {"id": 0, "name": "選択してください"},
+                            {"id": 200, "name": "確認した"},
+                        ],
+                    }
+                ]
+            },
+        },
+        "shop": {"information": {"shopId": 1}},
+    }
+    fields = _extract_purchase_fields(state, intent_override={})
+    assert fields["has_required_options"] is True
+    assert fields["form_fields"]["choice"] == "この商品は「トライタン製／プラスチック」です:確認した"
+    assert fields["unfillable_required_options"] == []
+
+
+def test_extract_fields_required_free_text_option_cannot_be_auto_filled():
+    """必填自由文本项（type=text，无候选值）自动填不了，要点名报出来"""
+    state = {
+        "item": {"itemId": 1, "variantId": "v"},
+        "purchase": {
+            "sku": {"inventoryType": "single"},
+            "sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}},
+            "information": {
+                "options": [
+                    {"id": 101, "name": "【お名前】", "type": "text", "isRequired": True},
+                ]
+            },
+        },
+        "shop": {"information": {"shopId": 1}},
+    }
+    fields = _extract_purchase_fields(state, intent_override={})
+    assert fields["has_required_options"] is True
+    # 填不出来就不填，也不拿占位/空值凑数
+    assert "choice" not in fields["form_fields"]
+    assert fields["unfillable_required_options"] == ["【お名前】"]
+
+
+def test_extract_fields_skips_optional_options_when_auto_filling():
+    """非必填项不替上游做业务决定（如「置き配を希望する」），只自动填必填项"""
+    state = {
+        "item": {"itemId": 1, "variantId": "v"},
+        "purchase": {
+            "sku": {"inventoryType": "single"},
+            "sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}},
+            "information": {
+                "options": [
+                    {
+                        "id": 100,
+                        "name": "必須確認",
+                        "type": "select",
+                        "isRequired": True,
+                        "values": [{"id": 0, "name": "選択してください"}, {"id": 200, "name": "了解"}],
+                    },
+                    {
+                        "id": 101,
+                        "name": "「置き配」希望について",
+                        "type": "select",
+                        "isRequired": False,
+                        "values": [{"id": 200, "name": "置き配を希望しない"}],
+                    },
+                ]
+            },
+        },
+        "shop": {"information": {"shopId": 1}},
+    }
+    fields = _extract_purchase_fields(state, intent_override={})
+    assert fields["form_fields"]["choice"] == "必須確認:了解"
+
+
 def test_extract_fields_intent_choice_override_accepts_list_and_str():
     state = {
         "item": {"itemId": 1, "variantId": "v"},

tests/test_telemetry.py

@@ -5,6 +5,9 @@
 2. enabled=true + endpoint 时 setup 注册真实 TracerProvider；shutdown 复位。
 3. instrument_app 在 setup 之前调用也要真的装上中间件（三个服务都是导入期打桩）。
 4. traced / set_attributes / record_error 的行为。
+5. 「采集返回结果」这一侧：record_envelope（信封失败在 HTTP 层看不出来）、
+   record_parse_failure（失败在 fetch 还是 parse）、snapshot（失败页面快照），
+   以及 span_unless_suppressed 与 suppressed() 的配套关系。
 
 不打真实网络：OTLPSpanExporter 创建时不发请求，BatchSpanProcessor 异步批量
 上报在没有 span 产生时也不会触发。span 断言用独立的 InMemory provider，不碰
@@ -27,7 +30,7 @@ from opentelemetry.util.http import parse_excluded_urls
 
 from app.shared import telemetry
 from app.shared.config import Settings
-from app.shared.errors import OrderGuardError
+from app.shared.errors import OrderGuardError, ScrapeParseError, UpstreamBlockedError
 from app.shared.telemetry import is_initialized, setup_telemetry, shutdown_telemetry
 
 
@@ -189,6 +192,153 @@ def test_set_attributes_skips_none(spans):
     assert "b" not in attributes
 
 
+def test_record_error_keeps_app_error_fields(spans):
+    """AppError 的排查字段（错误码/可重试/状态码/消息）都要落到属性上
+
+    只落 error.type 不够：上游看到的是错误码，「该不该重试」看 retryable，
+    而 record_exception 记的 event 在多数观测后台里要展开才看得到、列表页筛不出来。
+    """
+    tracer = telemetry.trace.get_tracer(__name__)
+    with tracer.start_as_current_span("unit.err") as span:
+        telemetry.record_error(span, UpstreamBlockedError("Akamai 挑战页"))
+
+    attributes = spans.get_finished_spans()[0].attributes
+    assert attributes["error.type"] == "UpstreamBlockedError"
+    assert attributes["error.code"] == UpstreamBlockedError().err_code
+    assert attributes["error.retryable"] is True
+    assert attributes["error.status_code"] == 400
+    assert "Akamai 挑战页" in attributes["error.message"]
+
+
+def test_record_error_on_plain_exception_omits_app_error_fields(spans):
+    """非 AppError 不应凭空长出 error.code / error.retryable 属性"""
+    tracer = telemetry.trace.get_tracer(__name__)
+    with tracer.start_as_current_span("unit.plain") as span:
+        telemetry.record_error(span, RuntimeError("boom"))
+
+    attributes = spans.get_finished_spans()[0].attributes
+    assert attributes["error.type"] == "RuntimeError"
+    assert "error.code" not in attributes
+    assert "error.retryable" not in attributes
+
+
+def test_record_envelope_failure_marks_span_error(spans):
+    """信封失败要置 ERROR 并落错误码——HTTP 层看不出这次调用失败了"""
+    tracer = telemetry.trace.get_tracer(__name__)
+    with tracer.start_as_current_span("unit.envelope") as span:
+        telemetry.record_envelope(
+            span, success=False, err_code=6002, msg="租约无效", status_code=409
+        )
+
+    finished = spans.get_finished_spans()[0]
+    assert finished.status.status_code is StatusCode.ERROR
+    assert finished.attributes["api.success"] is False
+    assert finished.attributes["api.code"] == 6002
+    assert finished.attributes["api.status_code"] == 409
+    assert finished.attributes["api.msg"] == "租约无效"
+
+
+def test_record_envelope_success_leaves_status_ok(spans):
+    tracer = telemetry.trace.get_tracer(__name__)
+    with tracer.start_as_current_span("unit.envelope_ok") as span:
+        telemetry.record_envelope(span, success=True, status_code=200)
+
+    finished = spans.get_finished_spans()[0]
+    assert finished.status.status_code is not StatusCode.ERROR
+    assert finished.attributes["api.success"] is True
+
+
+def test_snapshot_truncates_and_carries_extra(spans):
+    """超限 HTML 截断并标注，附带的来源信息（URL 等）也要落在同一条 event 上"""
+    tracer = telemetry.trace.get_tracer(__name__)
+    with tracer.start_as_current_span("unit.snapshot") as span:
+        telemetry.snapshot(
+            span, "parse.failed_html", "x" * 100, 10,
+            extra={"parse.url": "https://example.com/a", "parse.ignored": None},
+        )
+
+    event = spans.get_finished_spans()[0].events[0]
+    assert event.name == "parse.failed_html"
+    assert event.attributes["snapshot.html"] == "x" * 10
+    assert event.attributes["snapshot.original_bytes"] == 100
+    assert event.attributes["snapshot.truncated"] is True
+    assert event.attributes["parse.url"] == "https://example.com/a"
+    assert "parse.ignored" not in event.attributes
+
+
+def test_snapshot_skips_empty_html(spans):
+    """页面根本没取回来时不记空 event"""
+    tracer = telemetry.trace.get_tracer(__name__)
+    with tracer.start_as_current_span("unit.snapshot_empty") as span:
+        telemetry.snapshot(span, "parse.failed_html", None, 100)
+        telemetry.snapshot(span, "parse.failed_html", "", 100)
+
+    assert spans.get_finished_spans()[0].events == ()
+
+
+def test_record_parse_failure_distinguishes_fetch_from_parse(spans):
+    """失败阶段按「HTML 有没有拿到」区分：两者排查方向相反
+
+    html 为空＝页面没取回来（通道/反爬/上游 5xx）；非空＝取回了但解析不出
+    （多半站点改版）。fail_reason 要给真实异常类名，不能一律写死 parse_error。
+    """
+    tracer = telemetry.trace.get_tracer(__name__)
+    with tracer.start_as_current_span("unit.fetch_fail") as span:
+        telemetry.record_parse_failure(
+            span, UpstreamBlockedError("挑战页"), html=None, max_bytes=100,
+            url="https://example.com/a",
+        )
+    with tracer.start_as_current_span("unit.parse_fail") as span:
+        telemetry.record_parse_failure(
+            span, ScrapeParseError("没有 state"), html="<html/>", max_bytes=100,
+            url="https://example.com/b",
+        )
+    with tracer.start_as_current_span("unit.delegate_fail") as span:
+        telemetry.record_parse_failure(
+            span, ScrapeParseError("转调失败"), stage="delegate",
+        )
+
+    by_name = {s.name: s for s in spans.get_finished_spans()}
+    fetch = by_name["unit.fetch_fail"]
+    assert fetch.attributes["parse.stage"] == "fetch"
+    assert fetch.attributes["parse.fail_reason"] == "UpstreamBlockedError"
+    # 页面没取回来，没有快照可落
+    assert [e.name for e in fetch.events] == ["exception"]
+
+    parsed = by_name["unit.parse_fail"]
+    assert parsed.attributes["parse.stage"] == "parse"
+    assert parsed.attributes["parse.fail_reason"] == "ScrapeParseError"
+    snapshot_event = next(e for e in parsed.events if e.name == "parse.failed_html")
+    assert snapshot_event.attributes["snapshot.html"] == "<html/>"
+    assert snapshot_event.attributes["parse.url"] == "https://example.com/b"
+
+    # 显式 stage 覆盖推断：编排方法的失败既不在自己的 fetch 也不在自己的 parse
+    assert by_name["unit.delegate_fail"].attributes["parse.stage"] == "delegate"
+
+
+def test_span_unless_suppressed_is_noop_inside_suppressed(spans):
+    """`suppressed()` 里手工埋点必须退化成 noop，否则空转长轮询绕开抑制刷满后台
+
+    `suppress_instrumentation` 只被 instrumentation 库尊重，手工
+    `start_as_current_span` 不看它——worker 的 lease 正是在 suppressed() 里调
+    GatewayClient._request 的。
+    """
+    tracer = telemetry.trace.get_tracer(__name__)
+
+    with telemetry.suppressed():
+        with telemetry.span_unless_suppressed(tracer, "unit.suppressed") as span:
+            # 属性/异常写在 noop span 上不能报错，调用方不必分支
+            telemetry.set_attributes(span, {"a": 1})
+            telemetry.record_error(span, RuntimeError("boom"))
+            assert not span.is_recording()
+
+    assert spans.get_finished_spans() == ()
+
+    with telemetry.span_unless_suppressed(tracer, "unit.not_suppressed") as span:
+        assert span.is_recording()
+    assert [s.name for s in spans.get_finished_spans()] == ["unit.not_suppressed"]
+
+
 def test_enabled_initializes_and_shutdown_releases():
     """enabled=true 时 setup 注册 TracerProvider，shutdown 后 _provider 复位"""
     settings = Settings(

tests/test_verify.py

@@ -134,6 +134,54 @@ async def test_match_ignores_query_string_difference():
     assert result.verdict == verify.VerifyVerdict.ALREADY_ORDERED
 
 
+async def test_multiple_item_intent_requires_all_items_in_same_order():
+    gateway = FakeGateway()
+    site = FakeSite(
+        window=OrderListWindow(
+            entries=[
+                OrderListEntry(
+                    order_number="o1",
+                    order_date="2026-08-10T00:00:00Z",
+                    items=[
+                        OrderListItem(item_url="https://item.rakuten.co.jp/shop/x/"),
+                        OrderListItem(item_url="https://item.rakuten.co.jp/shop/y/"),
+                    ],
+                )
+            ],
+            window_fully_covered=True,
+        )
+    )
+    task = _make_task()
+    task.intent = {
+        "items": [
+            {"item_url": "https://item.rakuten.co.jp/shop/x/"},
+            {"item_url": "https://item.rakuten.co.jp/shop/y/"},
+        ]
+    }
+    result = await verify.verify_on_site(task, gateway=gateway, site=site)
+    assert result.verdict == verify.VerifyVerdict.ALREADY_ORDERED
+    assert result.site_order_id == "o1"
+
+
+async def test_multiple_item_intent_partial_order_match_is_not_ordered():
+    gateway = FakeGateway()
+    site = FakeSite(
+        window=OrderListWindow(
+            entries=[_entry("o1", "https://item.rakuten.co.jp/shop/x/")],
+            window_fully_covered=True,
+        )
+    )
+    task = _make_task()
+    task.intent = {
+        "items": [
+            {"item_url": "https://item.rakuten.co.jp/shop/x/"},
+            {"item_url": "https://item.rakuten.co.jp/shop/y/"},
+        ]
+    }
+    result = await verify.verify_on_site(task, gateway=gateway, site=site)
+    assert result.verdict == verify.VerifyVerdict.NOT_ORDERED
+
+
 # ---- 命中 0 笔且窗口确认覆盖完：未下单 ----
 
 

tests/test_worker_client_telemetry.py

@@ -0,0 +1,158 @@
+"""GatewayClient 埋点测试：信封里的失败必须在链路上看得见
+
+要钉住的问题：**网关的失败在响应信封里，不在 HTTP 状态码上**。一次
+`success=false, code=6002`（租约无效）的回报，httpx 自动 instrumentation 只看到
+一个完成了的请求——而且那个 CLIENT span 在 `request()` 返回时就结束了，此时信封
+还没解，所以它永远不可能带上这次调用的结论。链路里于是只剩「调用发生过」。
+
+另一半是抑制：worker 的空转长轮询在 `suppressed()` 里调 `_request`，而
+`suppress_instrumentation` 只被 instrumentation 库尊重，手工建的 span 不看它。
+手工埋点若不配合 `span_unless_suppressed`，孤立 trace 会从这个口子重新灌回来。
+
+全部用 httpx.MockTransport 拦截，不触达真实网关。
+"""
+from __future__ import annotations
+
+import httpx
+import pytest
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
+from opentelemetry.trace import SpanKind, StatusCode
+
+from app.shared.config import Settings
+from app.shared.errors import AppError
+from app.shared.task_state import OrderState
+from app.shared.telemetry import suppressed
+from app.trading.worker import client as worker_client
+from app.trading.worker.client import GatewayClient
+
+
+@pytest.fixture
+def spans(monkeypatch) -> InMemorySpanExporter:
+    """把 client 模块级 tracer 换成写内存的（不碰全局 provider）"""
+    exporter = InMemorySpanExporter()
+    provider = TracerProvider()
+    provider.add_span_processor(SimpleSpanProcessor(exporter))
+    monkeypatch.setattr(worker_client, "tracer", provider.get_tracer("test"))
+    return exporter
+
+
+def _client(handler) -> GatewayClient:
+    """构建 GatewayClient 并把它的 httpx 客户端换成 MockTransport 版本"""
+    client = GatewayClient(
+        "https://gateway.example", "token", settings=Settings(_env_file=None)
+    )
+    client._client = httpx.AsyncClient(
+        base_url="https://gateway.example",
+        transport=httpx.MockTransport(handler),
+    )
+    return client
+
+
+def _envelope(*, success: bool, code: int, msg: str, data=None, status: int = 200):
+    return httpx.Response(
+        status,
+        json={"success": success, "msg": msg, "data": data, "code": code},
+    )
+
+
+async def test_envelope_failure_is_recorded_on_span(spans):
+    """success=false 要把 span 置 ERROR 并落错误码——HTTP 层看这次是 200"""
+    gateway = _client(
+        lambda _r: _envelope(
+            success=False, code=6002, msg="租约无效：不是持有者", status=200
+        )
+    )
+    try:
+        with pytest.raises(AppError) as excinfo:
+            await gateway.report(
+                "task-1", "worker-1", state=OrderState.IN_CART, terminal=False
+            )
+        assert excinfo.value.err_code == 6002
+    finally:
+        await gateway.aclose()
+
+    span = spans.get_finished_spans()[0]
+    assert span.kind is SpanKind.CLIENT
+    assert span.status.status_code is StatusCode.ERROR
+    # 这次调用的结论——HTTP 状态码是 200，光看它分不出成败
+    assert span.attributes["api.success"] is False
+    assert span.attributes["api.code"] == 6002
+    assert span.attributes["api.status_code"] == 200
+    assert "租约无效" in span.attributes["api.msg"]
+    assert span.attributes["gateway.path"] == "/api/orders/task-1/report"
+    assert span.attributes["gateway.method"] == "POST"
+
+
+async def test_successful_envelope_leaves_span_ok(spans):
+    """success=true 不置 ERROR；作为对照说明上面的断言不是恒真"""
+    gateway = _client(
+        lambda _r: _envelope(success=True, code=0, msg="success", data={"recorded": True})
+    )
+    try:
+        await gateway.report("task-1", "worker-1", state=OrderState.IN_CART)
+    finally:
+        await gateway.aclose()
+
+    span = spans.get_finished_spans()[0]
+    assert span.status.status_code is not StatusCode.ERROR
+    assert span.attributes["api.success"] is True
+    assert "api.code" not in span.attributes
+
+
+async def test_non_json_body_is_recorded_as_error(spans):
+    """网关回了非 JSON（网关挂了/被反代拦了）：记成可重试的 3001"""
+    gateway = _client(lambda _r: httpx.Response(502, text="<html>502 Bad Gateway</html>"))
+    try:
+        with pytest.raises(AppError) as excinfo:
+            await gateway.renew("task-1", "worker-1")
+        assert excinfo.value.err_code == 3001
+    finally:
+        await gateway.aclose()
+
+    span = spans.get_finished_spans()[0]
+    assert span.status.status_code is StatusCode.ERROR
+    assert span.attributes["error.code"] == 3001
+    # 这类失败是可重试的，与信封里的业务拒绝（retryable=False）要能区分开
+    assert span.attributes["error.retryable"] is True
+    assert span.attributes["gateway.status_code"] == 502
+
+
+async def test_idle_long_poll_produces_no_span(spans):
+    """回归：空转长轮询在 suppressed() 里不能产生 span
+
+    worker 每 30 秒 lease 一次、绝大多数返回空。手工埋点若不看抑制标记，这些
+    调用会各自成为一条孤立 trace 把观测后台刷满——正是 suppressed() 要解决的问题。
+    """
+    gateway = _client(lambda _r: _envelope(success=True, code=0, msg="success", data=None))
+    try:
+        with suppressed():
+            assert await gateway.lease("worker-1", wait=0) is None
+        assert spans.get_finished_spans() == ()
+
+        # 抑制之外照常埋点，证明上面的空断言不是因为埋点根本没生效
+        assert await gateway.lease("worker-1", wait=0) is None
+    finally:
+        await gateway.aclose()
+
+    assert [s.attributes["gateway.path"] for s in spans.get_finished_spans()] == [
+        "/api/orders/lease"
+    ]
+
+
+async def test_leased_task_call_is_still_traced(spans):
+    """领到任务后的调用不受抑制影响：它们挂在任务根 span 底下，是要看的那部分"""
+    gateway = _client(
+        lambda _r: _envelope(
+            success=True, code=0, msg="success",
+            data={"task_id": "task-1", "site": "rakuten", "lease_count": 1},
+        )
+    )
+    try:
+        task = await gateway.lease("worker-1", wait=0)
+        assert task is not None and task.task_id == "task-1"
+    finally:
+        await gateway.aclose()
+
+    assert len(spans.get_finished_spans()) == 1