feat(observability): 失败响应与网关信封进链路，手工埋点尊重 suppressed

失败此前在 trace 里近乎不可见：异常处理器把异常吃掉换成信封响应，自动
instrumentation 只看到一个 HTTP 状态码，而 AppError 默认 400、信封里
success=false，跟正常返回分不出来。

- api.py：四个异常处理器（对外失败的唯一出口）各记一次 span；兜底处理器额外
  record_error——对外只回一句无信息量的错误文案，异常类型与栈只在本地日志里
- telemetry.py：新增 record_envelope / record_parse_failure /
  span_unless_suppressed；record_error 补 error.message / retryable /
  status_code；snapshot 支持 extra 带上「这份 HTML 是哪来的」
- worker/client.py：_request 自建 span，活到解信封之后。httpx 那个 CLIENT span
  在 request() 返回时就结束，此时信封还没解——success=false code=6002（租约
  无效）在它看来是完成的 200 请求
- span_unless_suppressed：suppress_instrumentation 只被 instrumentation 库尊重，
  手工 span 不看它，lease 空转长轮询会从这个口子把孤立 trace 放回来
- 两个 scraping client 的解析失败分支收拢到 record_parse_failure；shop_items
  显式标 stage=delegate 且不落快照（它自己不抓页面，按 html 推断只会得出
  「fetch 失败」的错误结论）

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

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:
+            except Exception as exc:
-                span.record_exception()
+                record_parse_failure(
-                span.set_attribute("parse.fail_reason", "parse_error")
+                    span, exc, html=html,
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+                    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:
+            except Exception as exc:
-                span.record_exception()
+                record_parse_failure(
-                span.set_attribute("parse.fail_reason", "parse_error")
+                    span, exc, html=html,
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+                    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:
+            except Exception as exc:
-                span.record_exception()
+                record_parse_failure(
-                span.set_attribute("parse.fail_reason", "parse_error")
+                    span, exc, html=html,
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+                    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:
+            except Exception as exc:
-                span.record_exception()
+                record_parse_failure(
-                span.set_attribute("parse.fail_reason", "parse_error")
+                    span, exc, html=html,
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+                    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:
+            except Exception as exc:
-                span.record_exception()
+                record_parse_failure(
-                span.set_attribute("parse.fail_reason", "parse_error")
+                    span, exc, html=html,
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+                    max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
+                )
                 raise
 
     @staticmethod

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:
+            except Exception as exc:
-                span.record_exception()
+                record_parse_failure(
-                span.set_attribute("parse.fail_reason", "parse_error")
+                    span, exc, html=html,
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+                    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:
+            except Exception as exc:
-                span.record_exception()
+                record_parse_failure(
-                span.set_attribute("parse.fail_reason", "parse_error")
+                    span, exc, html=html,
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+                    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:
+            except Exception as exc:
-                span.record_exception()
+                record_parse_failure(
-                span.set_attribute("parse.fail_reason", "parse_error")
+                    span, exc, html=html,
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+                    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:
+            except Exception as exc:
-                span.record_exception()
+                # 本方法自己不抓页面，失败一定发生在它转调的 shop_detail / search
-                span.set_attribute("parse.fail_reason", "parse_error")
+                # 里（那两个 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:
+            except Exception as exc:
-                span.record_exception()
+                record_parse_failure(
-                span.set_attribute("parse.fail_reason", "parse_error")
+                    span, exc, html=html,
-                snapshot(span, "parse.failed_html", html, self._settings.otel_snapshot_max_bytes)
+                    max_bytes=self._settings.otel_snapshot_max_bytes, url=url,
+                )
                 raise

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/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__)
+    retryable = getattr(exc, "retryable", None)
-    err_code = getattr(exc, "err_code", None)
+    set_attributes(
-    if isinstance(err_code, int):
+        span,
-        span.set_attribute("error.code", err_code)
+        {
+            "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"""
+        """发起请求并解信封。失败（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
 
-        if not body.get("success"):
+        整段套一个自己的 span，而不是依赖 httpx 自动 instrumentation 那个 CLIENT
-            raise AppError(
+        span：**网关的失败在信封里，不在 HTTP 状态码上**。httpx 那个 span 在
-                message=body.get("msg", "网关返回失败"),
+        `request()` 返回时就结束了，此时信封还没解——一次 `success=false, code=6002`
-                code="GATEWAY_ERROR",
+        （租约无效）的调用在它看来是完成的 200 请求，链路里跟成功毫无区别。
-                err_code=int(body.get("code", 1500)),
+        本 span 活到解信封之后，所以能把「这次调用的结论」记下来。
-                retryable=False,
+        """
-                status_code=response.status_code,
+        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
 
     # ---- 接口 ----
 

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(