feat: record call observability columns and terminal failure rows

Grow the telemetry contract from 26 to 36 fields and give every logical call a failure terminal row, so SQL can finally answer "how many calls failed" and "why did the whole pool die". Schema and port move together with the emitter writes in one commit: splitting them would ship columns that nothing populates. - schema: append 10 nullable columns (scope, operation, logical_call_id, event_kind, http_status_code, error_type, cause_type, error_body, attempts, total_latency_ms) to all five definition sites in one order - ports: 10 keyword-only parameters without defaults; the protocol signature is now the single source the assembly gate derives from - emitter: take domain exception objects instead of pre-flattened text and pin down the diagnostics in one helper; a relabelled 503 stays 503 and success rows leave all five columns NULL - emitter: reject recorders whose record_llm_call cannot accept the current field shape at assembly time, since _record would otherwise swallow the TypeError and drop every row while calls keep succeeding - clients: write at most one terminal row per logical call through a single shared exit, deduplicated by the call context; TelemetryMW stops writing terminals so the two sites cannot double count - clients: cancellation stays best effort and propagates, non-domain exceptions get no terminal row and keep their classification - transports: give _status_to_error an explicit operation and fix the historically mislabelled embedding HTTP failures - structured: promote the bounded error formatter so the reask feedback and the terminal explanation share one set of limits Terminal rows carry no cost and no tokens, so cost aggregation is unchanged; failure counts must now filter on event_kind.
2026-09-09 11:27:52 -04:00
parent 87c261bf73
commit 393f2bf617
19 changed files with 1628 additions and 194 deletions
@@ -0,0 +1,141 @@
+---
+type: finding
+node_id: finding:2026-09-09-135-call-observability-validation
+title: "1.3.5 T2/T3 验收证据：36 列遥测、失败终态与取消口径"
+date: 2026-09-09
+---
+
+# 1.3.5 T2/T3 验收证据
+
+> 范围：**仅 T2（遥测 10 列 / 诊断保真 / scope+operation / 装配闸）与 T3（终态行 / 取消 / 统一出口）**，
+> 外加计划 T4 中"签名与列数机械迁移"那一片（与 schema 同批完成，避免先提交 schema 却留写入缺键）。
+> **不含** T4 的 PG 集成、变异矩阵与文档同步——另任务承接，缺口见 §5。
+> 设计：`designs/2026-09-09-135-call-observability-design.md`；计划：`plans/2026-09-09-135-call-observability.md`。
+> 基线 HEAD `87c261b`（T1 已提交，1419 unit 全绿）。全部命令在 `PolyGateway` conda 环境执行。
+
+## 1. 红绿证据链
+
+TDD 纪律要求"先失败后通过"，且红必须是**行为红**而非 import 红。下表每行都对应本会话的真实工具输出。
+
+| # | 阶段 | 命令 | 结果 |
+| --- | --- | --- | --- |
+| 0 | 基线 | `pytest tests/unit -q` | **1419 passed**（对照底） |
+| 1 | 红（T2/T3 目标行为） | `pytest tests/unit/test_telemetry.py -k "RowLevelObservability or RecorderShapeGate"` | **26 failed**，证据 `tests/outputs/135/red-01-emitter-rows.txt` |
+| 2 | 实现后全量红面 | `pytest tests/unit -q` | **137 failed / 1308 passed**（机械迁移面暴露），`red-02-after-impl.txt` |
+| 3 | 迁移中 | 同上 | 60 → 25 → 8 → 4 failed（`red-03`/`red-04`/`red-05`） |
+| 4 | 绿 | `pytest tests/unit -q` | **1467 passed, 0 failed** |
+| 5 | 静态门 | `make check`（ruff lint+format + import-linter） | **Contracts: 1 kept, 0 broken** |
+
+第 1 步的红是行为红而非 import 红：新用例调用的是**已存在**的 `TelemetryEmitter`，
+失败形态是"缺 `scope=`/`operation=`/`stats=` 关键字"与"断言的列不存在"，不是模块导不进来。
+
+补测阶段另有两次真实红（均由 conda 实跑暴露、当场修正，非噪音）：
+`test_client.py` 的 `NameError: ResultInvalidError / ChatRequest`（漏 import）、
+`TestTerminalRowSqlSemantics` 的 `'coroutine' object has no attribute 'execute'`（async helper 漏 await）。
+
+## 2. 核心要求逐条对应
+
+### 2.1 36 列与 Emitter 完整一致（不留"schema 有列、写入缺键"）
+
+一次性同批改完五处列定义 + 端口签名 + Emitter 写入，故不存在中间态。
+
+| 判据 | 证据 |
+| --- | --- |
+| `len(COLUMNS) == 36`、物理列 37 | `test_telemetry.py::TestSchemaModule::test_columns_and_ddl_are_frozen`；`TestSQLiteSchemaMode` 断言 23 → 37 |
+| 五处列序一致（DDL/BACKFILL×2/COLUMNS） | 新建库与 ALTER 追加列序同为 `_EXPECTED_COLUMNS` |
+| 端口实测 36 字段、10 新参 keyword-only 且无默认值 | `test_ports.py::TestTelemetryRecorderSignature`（`inspect.signature` 实测，不凭记忆） |
+| **Emitter 实参键集合 == `schema.COLUMNS`** | `TestEmitterRecorderContract` 三入口逐个断言 `set(rows[0]) == set(COLUMNS)` |
+| 1.2.1 冻结 INSERT 未被改写 | 按 `_PRE_135_COLUMNS`（26 列）重现原文；全量 36 列另按占位符个数断言 |
+
+装配闸（C3）：`_assert_recorder_shape` 在 `TelemetryEmitter.__init__` 做一次 `signature.bind`，
+参数名**从协议签名派生**而非手抄第四份清单——`test_gate_derives_parameter_names_from_the_protocol`
+用 monkeypatch 换掉协议后闸自动跟随，证明没有硬编码。旧签名 recorder / 无该方法 / 不可 inspect
+一律装配期 `ValueError`（不是 warning：降级铁律管的是运行期写失败，不是配置错误）。
+
+### 2.2 三个 client 的**真实链路**失败终态
+
+不是只测 Emitter，而是驱动真实 client + MockTransport 到落库。
+
+| 链路 | 用例 | 断言 |
+| --- | --- | --- |
+| chat 重试耗尽 | `TestChatTerminalFailureRows::test_retry_exhaustion_writes_exactly_one_terminal_row` | 3 条 attempt + **恰 1 条**终态；终态 `attempts == 3` |
+| chat 结构化耗尽 | `test_structured_exhaustion_writes_the_only_failure_row` | attempt 行**全是成功行**，终态是唯一失败记录；error 含有界 `validation=`/`repair=` 且 < 1200 字符、不含 raw_text |
+| chat 400 直拒 | `test_request_rejected_now_has_both_an_attempt_and_a_terminal_row` | 尝试行 + 终态行**各 1**（已批准的行数翻倍）；两行共享同一 `logical_call_id` |
+| embedding | `test_embedding.py::TestReasonlessTelemetryContract::test_failed_attempts_still_have_null_effort` | attempt 数 == 脚本长度，终态恰 1 |
+| OCR 两方法 | `test_ocr_client.py` 同名用例（参数化 `recognize_text`/`parse_layout`） | 终态恰 1，且 `operation` 为**公开方法名** |
+
+去重：`test_terminal_row_is_written_once_per_logical_call` 连调出口 3 次，SQL 可见仍 1 条（`claim_terminal`）。
+非领域异常：`test_non_domain_exception_writes_no_terminal_row` —— `KeyError` 原样传播、**0 条**终态、分类不被改写。
+
+**双写已消除**：`TelemetryMW` 的两个终态分支删除，改由三个 client 的公开边界经
+`emit_terminal_once` 统一写出；`TestTelemetryMW::test_scope_level_failure_is_not_written_here_anymore`
+与 `test_cancellation_is_not_written_here_anymore` 锁死"本层不再写终态"，防回归双计。
+
+### 2.3 取消口径
+
+- chat 取消：`test_cancellation_writes_at_most_one_terminal_row` —— 恰 1 条，`error == "cancelled"`，
+  且 `error_type` 为 NULL（**字符串不解析猜诊断**）。
+- 三链路同策略"尽力写一条、允许 0"；`emit_terminal_once` **不 shield、不开后台任务**，
+  快照冻结是同步动作；写入 await 上再被取消则 `CancelledError` 原样传播（与 TelemetryMW 历史行为同款）。
+
+### 2.4 SQL 不双计（按真实 SQLite 落库断言）
+
+`TestTerminalRowSqlSemantics` 用真实 `SQLiteRecorder` 驱动一次失败 chat 后直接查表：
+
+| 迁移影响 | 断言 |
+| --- | --- |
+| 失败计数判据 | `error IS NOT NULL` 得 **3**（2 尝试 + 1 终态），`event_kind='terminal_failure'` 得 **1** |
+| 费用不双计 | 终态行 `cost IS NOT NULL` 计数为 **0**；且 `usage_source='unavailable'`、token 全 0 |
+| 时延分组 | 终态 `latency_ms` ≥ 任何单次尝试（含退避），故看板必须按 `event_kind` 分组 |
+| 双时钟微差 | 终态 `latency_ms == total_latency_ms`（同一份冻结快照） |
+| 逻辑两列归属 | 尝试行 `attempts`/`total_latency_ms` 恒 NULL |
+| §5 归因查询 | 同一 `logical_call_id` 同时给出整池 reason 文案与逐源 503 现场；终态 `http_status_code` 为 NULL（C1 不冒充） |
+
+### 2.5 诊断保真与 operation 修正
+
+- 中转把 529 改写成 503 → **记 503 不猜回 529**；直接 529 记 529。
+- 空 `str()` 的 Connect/Read/Write/PoolTimeout → `cause_type` 落对应 httpx 类名，`error` 退回类名。
+- 成功行五列全 NULL（**不统一填 200**）。
+- `_status_to_error` 增 keyword `operation`：`embed()` 非 200 改传 `"embedding"`（修正历史误标），
+  流式与非流式 chat 两处仍 `"chat"`（按实施计划 §2 归属表，未按设计行号误标）。
+- 新列 `operation` 恒为公开方法四值，**绝不读 `exc.operation`**：
+  `test_operation_is_given_by_the_call_site_not_the_exception` 用 `exc.operation="download_result"` 反证。
+- OCR `"类名: msg"` 前缀由出口的显式策略参数 `class_prefixed_error` 承载，不再三处各拼一遍。
+
+## 3. 有界 validation 说明的单一所有者
+
+`structured.py` 的 `_MAX_FEEDBACK_ERRORS`/`_MAX_ERROR_CHARS`/`_format_errors` 改名为
+公开的 `MAX_FEEDBACK_ERRORS`/`MAX_ERROR_CHARS`/`format_bounded_errors`，
+由"重问反馈"与"终态结构化说明"两个消费者共同引用，**数值只有一份**。
+行为逐字不变（`test_structured.py` 22 项全绿，含反馈文案用例）。
+
+## 4. 未改动确认（防越界）
+
+`errors.py`、`admission.py`、`ratelimit.py`、`breaker.py`、`sources.py`、`thinking.py`、
+`providers.py`、`telemetry/sqlite.py`、`telemetry/postgres.py`、`transports/monkey_ocr.py`
+一字未动——两个 recorder 靠 `**fields` + `schema.COLUMNS` 自动吃到新列。
+缓存 key 公式、重试预算与退避、429 免预算、stall 算法、取消结算、推理能力表均未触碰。
+唯一顺带修正：`RetryMW.__init__` 的 `emitter` 注解由 `object | None` 收紧为
+`TelemetryEmitter | None`（TYPE_CHECKING 导入，同层不破分层契约），因本轮改了它的 `_emit`。
+
+## 5. 明确缺口（交 T4 另任务，本轮不做）
+
+| 缺口 | 说明 |
+| --- | --- |
+| PG 存储兼容 | `tests/integration/test_postgres_telemetry.py` 的 `_EXPECTED_COLUMNS` 与字段字典**尚未补 10 列**；auto 追加 / manual 缺列裁剪 / 旧行 NULL / 新旧进程混写四项未跑（需真实 PG 沙箱） |
+| 变异证据 | 计划 §4 的七项变异（计数位置、上下文复制、缓存回放、提前压平、终态双计费用、去 `claim_terminal`、闸改 warning）未执行 |
+| 文档同步 | README「必录 26 字段」、`PGW_TELEMETRY_TEXT_CAP` 覆盖面、ARCHITECTURE §7.8、CHANGELOG、schemas/metrics 未更新 |
+| 独立验证 | 未派全新上下文 verifier（合并前硬门） |
+| slow / e2e | 未跑，属发布清单第 4 步 |
+
+## 6. 环境噪音记录（不是缺陷）
+
+自动检查器用**系统解释器 Python 3.13.9**（无 `redis`/`pydantic`/`httpx`/`loguru`/`asyncpg` 等依赖），
+持续报 `test_client.py` 2 项失败与大量 "Import could not be resolved"、`StrEnum is unknown import symbol`。
+已核实为环境问题、非本轮引入：
+
+- 失败根因是 `ModuleNotFoundError: No module named 'redis'`（optional extra），本轮 diff 对 redis 零改动；
+- 在**未改动的 HEAD** 上用同一系统解释器复跑，同样 2 failed / 90 passed；
+- 项目强制环境 `conda run -n PolyGateway`（Python 3.12.13）下：`test_client.py` 98 passed、全量 1467 passed。
+
+判据以 CLAUDE.md §2 规定的 conda 环境与 `make check` 为准。
@@ -20,11 +20,12 @@ from polygateway.backends.memory.breaker import InMemoryGate
 from polygateway.backends.memory.cache import InMemoryCache
 from polygateway.backends.memory.limiter import InMemoryLimiter
 from polygateway.config import GatewaySettings
+from polygateway.errors import PolyGatewayError
 from polygateway.middleware.base import compose
 from polygateway.middleware.cache import CacheMW
 from polygateway.middleware.retry import RetryMW
 from polygateway.middleware.structured import StructuredMW
-from polygateway.middleware.telemetry import TelemetryEmitter, TelemetryMW
+from polygateway.middleware.telemetry import TelemetryEmitter, TelemetryMW, emit_terminal_once
 from polygateway.ports import TelemetryStatusProvider
 from polygateway.pricing import PricingTable
 from polygateway.providers import get_provider
@@ -239,7 +240,7 @@ class GatewayClient:
        rng: Any = random.random,
    ) -> None:
        emitter = (
-            TelemetryEmitter(telemetry, pricing=pricing, text_cap=text_cap)
+            TelemetryEmitter(telemetry, scope=scope, pricing=pricing, text_cap=text_cap)
            if telemetry is not None
            else None
        )
@@ -393,7 +394,27 @@ class GatewayClient:
            meta=dimensions,
            call_context=context,
        )
+        try:
            response = await self._handler(request)
+        except PolyGatewayError as exc:
+            # 统计边界内的一切领域失败均尝试写一条终态行(1.3.5 设计 §6 I3),
+            # 包括已有 attempt 错误行的 RequestRejected / ResultInvalid——两类行描述
+            # 的不是同一件事(尝试 vs 逻辑终态),由 `event_kind` 区分
+            await emit_terminal_once(
+                self._emitter, request=request, context=context, error=exc, operation="chat"
+            )
+            raise
+        except asyncio.CancelledError:
+            # 尽力而为且**取消优先**: 不 shield、不开后台任务;写入那一次 await 上
+            # 再被取消则 `CancelledError` 照常传播(与 TelemetryMW 历史行为同款)
+            await emit_terminal_once(
+                self._emitter,
+                request=request,
+                context=context,
+                error="cancelled",
+                operation="chat",
+            )
+            raise
        # 快照在返回前冻结: 故它含缓存命中路径与已完成的内联遥测耗时
        return dataclasses.replace(response, call_stats=context.snapshot())

@@ -42,7 +42,7 @@ from polygateway.middleware.admission import SourceAdmission, settle_and_release
 from polygateway.middleware.breaker import BreakerGate
 from polygateway.middleware.ratelimit import QuotaGate
 from polygateway.middleware.retry import StallClock, _failure_reason, backoff_delay
-from polygateway.middleware.telemetry import TelemetryEmitter
+from polygateway.middleware.telemetry import TelemetryEmitter, emit_terminal_once
 from polygateway.types import (
    ChatRequest,
    EmbeddingResponse,
@@ -129,7 +129,9 @@ class EmbeddingClient:
        self._transport = transport
        self._retry = retry
        self._emitter = (
-            TelemetryEmitter(telemetry, pricing=pricing, text_cap=text_cap) if telemetry else None
+            TelemetryEmitter(telemetry, scope=self._scope, pricing=pricing, text_cap=text_cap)
+            if telemetry
+            else None
        )
        self._telemetry = telemetry
        # 限流/熔断后端在此之外只以 QuotaGate/BreakerGate 的形态存在,自持一份
@@ -203,6 +205,38 @@ class EmbeddingClient:
                source_name="",
                call_stats=context.snapshot(),
            )
+        try:
+            return await self._embed_all(
+                texts, session_id, parent_call_id, dimension_tenant_id, dimensions, context
+            )
+        except PolyGatewayError as exc:
+            await self._emit_terminal(
+                texts, session_id, parent_call_id, dimension_tenant_id, dimensions, context, exc
+            )
+            raise
+        except asyncio.CancelledError:
+            # 三条链路同一口径尽力写一条(允许 0 条);取消优先,不 shield
+            await self._emit_terminal(
+                texts,
+                session_id,
+                parent_call_id,
+                dimension_tenant_id,
+                dimensions,
+                context,
+                "cancelled",
+            )
+            raise
+
+    async def _embed_all(
+        self,
+        texts: list[str],
+        session_id: str | None,
+        parent_call_id: str | None,
+        tenant_id: str | None,
+        meta: dict[str, Any],
+        context: _CallContext,
+    ) -> EmbeddingResponse:
+        """切批串行执行并合并;无源的 raise 必须在本方法内——否则无源终态行写不出。"""
        if not self._sources:
            raise AllSourcesExhausted(scope=self._scope, reason="no_sources", retry_after_s=0.0)
        outcomes = []
@@ -212,14 +246,50 @@ class EmbeddingClient:
                    texts[start : start + self._batch_size],
                    session_id,
                    parent_call_id,
-                    dimension_tenant_id,
-                    dimensions,
+                    tenant_id,
+                    meta,
                    context,
                )
            )
        # 全批共享同一上下文,故分批是实现细节而非 N 次独立逻辑调用
        return dataclasses.replace(self._merge(outcomes), call_stats=context.snapshot())

+    async def _emit_terminal(
+        self,
+        texts: list[str],
+        session_id: str | None,
+        parent_call_id: str | None,
+        tenant_id: str | None,
+        meta: dict[str, Any],
+        context: _CallContext,
+        error: PolyGatewayError | str,
+    ) -> None:
+        """终态行的请求摘要(设计 §6 M4): 计数占位 + 第一批截断文本。
+
+        描述的是**本次调用的整体输入**但不扩大单行正文预算: 失败批的具体文本由同
+        `logical_call_id` 的 attempt 行给出,终态行不保存全量原输入。
+        """
+        batches = math.ceil(len(texts) / self._batch_size)
+        messages = [{"role": "user", "content": f"<embed texts={len(texts)} batches={batches}>"}]
+        # 与逐批行同款构造(至多 `batch_size` 条、每条 200 字符)
+        messages += [
+            {"role": "user", "content": t[:_TELEMETRY_TEXT_CAP]} for t in texts[: self._batch_size]
+        ]
+        await emit_terminal_once(
+            self._emitter,
+            request=ChatRequest(
+                messages=messages,
+                session_id=session_id,
+                parent_call_id=parent_call_id,
+                tenant_id=tenant_id,
+                meta=meta,
+                call_context=context,
+            ),
+            context=context,
+            error=error,
+            operation="embed",
+        )
+
    # —— 治理循环(与 RetryMW 同构;设计 §7.1 已声明的有限重复)——

    async def _embed_batch(
@@ -300,6 +370,7 @@ class EmbeddingClient:
                parent_call_id,
                tenant_id,
                meta,
+                context,
                result,
            )
            return _BatchOutcome(result, source, call_id, latency_ms)
@@ -314,6 +385,7 @@ class EmbeddingClient:
                parent_call_id,
                tenant_id,
                meta,
+                context,
                error=exc,
            )
            raise
@@ -329,6 +401,7 @@ class EmbeddingClient:
                parent_call_id,
                tenant_id,
                meta,
+                context,
                error="cancelled",
            )
            raise
@@ -349,6 +422,7 @@ class EmbeddingClient:
                parent_call_id,
                tenant_id,
                meta,
+                context,
                error=exc,
            )
            return _FailedBatch(exc, immediate=dead)
@@ -385,8 +459,9 @@ class EmbeddingClient:
        parent_call_id: str | None,
        tenant_id: str | None,
        meta: dict[str, Any],
+        context: _CallContext,
        result: EmbeddingTransportResult | None = None,
-        error: object | None = None,
+        error: PolyGatewayError | str | None = None,
    ) -> None:
        """逐批遥测(经同一 Emitter): messages=截断 texts、向量绝不入库。"""
        if self._emitter is None:
@@ -399,6 +474,7 @@ class EmbeddingClient:
            parent_call_id=parent_call_id,
            tenant_id=tenant_id,
            meta=meta,
+            call_context=context,
        )
        response = None
        if result is not None:
@@ -423,10 +499,12 @@ class EmbeddingClient:
            call_id=call_id,
            latency_ms=int((self._now() - started) * 1000),
            response=response,
-            error=None if error is None else str(error),
+            # 异常对象原样下传: 状态码/底层异常类型/网关正文在 Emitter 内定型
+            error=error,
            # embedding payload 硬编码 {model, input},从不带推理参数;源上即便
            # 误配了 ENABLE_THINKING,记一个档也是替这次调用声称它没做过的事
            reasoning_applies=False,
+            operation="embed",
        )

    def _merge(self, outcomes: list[_BatchOutcome]) -> EmbeddingResponse:
@@ -42,6 +42,7 @@ from polygateway.types import LLMResponse
 if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Awaitable, Callable

+    from polygateway.middleware.telemetry import TelemetryEmitter
    from polygateway.ports import (
        GateDecision,
        Permit,
@@ -181,7 +182,7 @@ class RetryMW:
        circuit_open: str = "fail_fast",
        cooldown_memo: SourceCooldownMemo | None = None,
        pacer: AdaptivePacer | None = None,
-        emitter: object | None = None,
+        emitter: TelemetryEmitter | None = None,
        now: Callable[[], float] = time.monotonic,
        sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
        rng: Callable[[], float] = random.random,
@@ -413,9 +414,13 @@ class RetryMW:
        started: float,
        *,
        response: LLMResponse | None = None,
-        error: object | None = None,
+        error: PolyGatewayError | str | None = None,
    ) -> None:
-        """逐次遥测(经注入的单一 Emitter);遥测失败不得影响调用(铁律)。"""
+        """逐次遥测(经注入的单一 Emitter);遥测失败不得影响调用(铁律)。
+
+        异常**对象原样下传**而非先 `str()` 压平(1.3.5 设计 §5): 状态码、底层异常
+        类型与网关响应体已经在异常上了,在这里压平就是把它们丢掉。
+        """
        if self._emitter is None:
            return
        try:
@@ -425,9 +430,11 @@ class RetryMW:
                call_id=call_id,
                latency_ms=int((self._now() - started) * 1000),
                response=response,
-                error=None if error is None else str(error),
+                error=error,
                # chat 路径是唯一带推理参数的路径,故实发档由这里的响应说了算
                reasoning_applies=True,
+                # 公开方法四值之一;本中间件只服务 chat 洋葱
+                operation="chat",
            )
        except asyncio.CancelledError:
            raise
@@ -14,6 +14,8 @@ from typing import TYPE_CHECKING
 from polygateway.errors import ResultInvalidError

 if TYPE_CHECKING:
+    from collections.abc import Sequence
+
    from polygateway.ports import CallNext, StructuredOutputStrategy
    from polygateway.types import ChatRequest, LLMResponse

@@ -22,12 +24,18 @@ _FEEDBACK_TEMPLATE = (
    "Your previous reply was not valid JSON matching the required schema. "
    "Errors: {errors}. Reply with ONLY the corrected JSON object."
 )
-_MAX_FEEDBACK_ERRORS = 3
-_MAX_ERROR_CHARS = 200
+MAX_FEEDBACK_ERRORS = 3
+MAX_ERROR_CHARS = 200


-def _format_errors(errors: list[str]) -> str:
-    clipped = [e[:_MAX_ERROR_CHARS] for e in errors[:_MAX_FEEDBACK_ERRORS]]
+def format_bounded_errors(errors: Sequence[str]) -> str:
+    """校验错误的有界拼装: 至多 3 条 × 每条 200 字符。
+
+    **本模块是这条规则的所有者**: 重问反馈文案与 1.3.5 终态行的结构化说明
+    两个消费者共用同一份实现与同一组数值——数值复制成两份必然漂移,而漂移后
+    "模型看到的错误"与"台账里记的错误"就不再是同一件事。行为与重命名前逐字相同。
+    """
+    clipped = [e[:MAX_ERROR_CHARS] for e in list(errors)[:MAX_FEEDBACK_ERRORS]]
    return "; ".join(clipped) if clipped else "output could not be parsed"


@@ -104,7 +112,10 @@ class StructuredMW:
        messages = [
            *current.messages,
            {"role": "assistant", "content": bad_content},
-            {"role": "user", "content": _FEEDBACK_TEMPLATE.format(errors=_format_errors(errors))},
+            {
+                "role": "user",
+                "content": _FEEDBACK_TEMPLATE.format(errors=format_bounded_errors(errors)),
+            },
        ]
        reask = dataclasses.replace(current, messages=messages)
        if self._escalation is not None:
@@ -2,13 +2,19 @@

 Emitter 是全库**唯一**调用 `record_llm_call` 的地方(三项目 4 处逐字复制
 15 参调用的教训)。分工: RetryMW 经 Emitter 逐次记录每次尝试;TelemetryMW
-(最外层)只记尝试层看不见的事件——缓存命中、scope 级失败、取消;
-RequestRejected/ResultInvalid 已被尝试层记录,最外层放行不重复记。
+(最外层)只记尝试层看不见的缓存命中;而**终态失败行**由三个 client 的公开
+边界经 `emit_terminal_once` 统一写出(1.3.5)——两处同时写就会双计。
+
+一行遥测属于三类事件之一(`event_kind`): `attempt`(一次尝试)、`cache_hit`
+(未产生网关调用)、`terminal_failure`(一次**逻辑调用**的失败终态)。后两者与
+前者**不是重复事实**,故统计失败调用次数只能取 `terminal_failure`,
+不得按 `error IS NOT NULL` 跨两类直接计数(设计 §6/§8)。
 """

 from __future__ import annotations

 import asyncio
+import inspect
 import json
 import time
 import uuid
@@ -17,12 +23,10 @@ from typing import TYPE_CHECKING

 from loguru import logger

-from polygateway.errors import (
-    GatewayUnavailableError,
-    GovernanceBackendError,
-    SourceNotConfiguredError,
-)
+from polygateway.errors import PolyGatewayError, ResultInvalidError
 from polygateway.middleware.cache import digest_messages
+from polygateway.middleware.structured import MAX_ERROR_CHARS, format_bounded_errors
+from polygateway.ports import TelemetryRecorder
 from polygateway.thinking import effective_effort
 from polygateway.types import Effort, ThinkingObservation, canonical_sampling_json, merge_sampling

@@ -30,9 +34,17 @@ if TYPE_CHECKING:
    from collections.abc import Callable, Mapping
    from typing import Any

-    from polygateway.ports import CallNext, TelemetryRecorder
+    from polygateway.ports import CallNext
    from polygateway.pricing import PricingTable
-    from polygateway.types import ChatRequest, LLMResponse, SourceConfig
+    from polygateway.types import (
+        CallOperation,
+        CallStats,
+        ChatRequest,
+        EventKind,
+        LLMResponse,
+        SourceConfig,
+        _CallContext,
+    )


 def _canonical_meta_json(meta: Mapping[str, Any]) -> str:
@@ -225,26 +237,159 @@ class _AttemptUsage:
        )


+@dataclass(frozen=True)
+class _ErrorFields:
+    """一行遥测的错误列;未知一律 `None`。
+
+    存在的理由是把"三种入参形态 × 两类行"的定型规则收敛到**一处**:
+    改前调用方先 `str(exc)` 压平,状态码、底层异常类型与网关正文全部丢失。
+    """
+
+    error: str | None = None
+    error_type: str | None = None
+    cause_type: str | None = None
+    http_status_code: int | None = None
+    error_body: str | None = None
+
+
+def _structured_detail(exc: ResultInvalidError) -> str:
+    """结构化阶梯耗尽的**有界**说明(设计 §5 C2)。
+
+    `ResultInvalidError("结构化输出阶梯耗尽")` 的 message 不含校验与修复错误,而该
+    失败发生在 StructuredMW 之上——RetryMW 侧的 attempt 行全是**成功行**,终态行是
+    唯一记录。故把说明并入现有 `error` 串。
+
+    **不含 `raw_text`**: 它是模型正文,attempt 行的 `response` 列已按 `text_cap` 记过
+    一份;再存一份等于绕过既有的正文预算。条数与限长复用 `structured.py` 的同一
+    套常量(重问反馈与本说明同一口径),数值只有一份。
+    """
+    parts: list[str] = []
+    if exc.repair_error:
+        parts.append(f"repair={exc.repair_error[:MAX_ERROR_CHARS]}")
+    if exc.validation_errors:
+        parts.append(f"validation={format_bounded_errors(exc.validation_errors)}")
+    return " | ".join(parts)
+
+
+def _error_fields(
+    error: PolyGatewayError | str | None,
+    *,
+    event_kind: EventKind,
+    class_prefixed: bool,
+) -> _ErrorFields:
+    """三种入参形态的唯一定型点(设计 §5/§6)。
+
+    - `None` → 全 None(成功行不统一填 200: 那会让"有状态码"不再等价于"失败了")。
+    - `str`(取消路径的 `"cancelled"`)→ 原样落 `error`,**不解析字符串猜诊断**。
+    - 领域异常 → 只读它既有的属性,不遍历任意对象、不猜正文。
+
+    **终态行的三列恒为 NULL(C1 红线)**: `GatewayUnavailableError` 家族从不携带
+    状态码与响应体,NULL 正是它自身的真实状态——把最后一次 attempt 的状态码与正文
+    搬上来,就是拿最后一个源冒充整池归因。逐源现场由同一 `logical_call_id` 的
+    attempt 行给出。
+    """
+    if error is None:
+        return _ErrorFields()
+    if isinstance(error, str):
+        return _ErrorFields(error=error)
+    name = type(error).__name__
+    # 空 `str()` 退回类名(httpx 的 Connect/Read/Write/PoolTimeout 文案就是空的);
+    # `class_prefixed` 是 OCR 的既有口径(按类名归组的 metric),故逐字保留它的拼法
+    text = f"{name}: {error}" if class_prefixed else (str(error) or name)
+    if event_kind == "terminal_failure":
+        if isinstance(error, ResultInvalidError):
+            detail = _structured_detail(error)
+            if detail:
+                text = f"{text} | {detail}"
+        return _ErrorFields(error=text, error_type=name)
+    cause = error.__cause__
+    return _ErrorFields(
+        error=text,
+        error_type=name,
+        cause_type=type(cause).__name__ if cause is not None else None,
+        # getattr 而非直读: 本函数在 `_record` 的降级 try **之外**求值,
+        # 一个非领域异常误传进来不得把一次真实失败换成 AttributeError
+        http_status_code=getattr(error, "status_code", None),
+        # 空串归 None: 既有 `body_text` 的缺省就是空串,而本列的语义是"未知"
+        error_body=getattr(error, "body_text", "") or None,
+    )
+
+
+def _assert_recorder_shape(recorder: TelemetryRecorder) -> None:
+    """装配期一次 `signature.bind` 形状校验: 不执行写入,只证明该形状能被接受。
+
+    `_record` 的 `except Exception` 会把旧 recorder 的 `TypeError` 吞成 warning,
+    后果是自定义 recorder 在下游升级后**100% 丢遥测且调用照常成功**——正是
+    "遥测必录"要防的形态,而文档级迁移清单挡不住它。故在装配期当场报错
+    (不是 warning: 降级方向的铁律管的是**运行期写失败**,不是装配错误)。
+
+    参数名从 `TelemetryRecorder.record_llm_call` 的协议签名**派生**(不手抄第四份
+    字段清单),绑定用哨兵 `None`,不读任何真实请求数据;`**kwargs`
+    (VAR_KEYWORD)自动通过。不可 inspect(C 实现等)同样按配置错误报错——宁可
+    装配不起来,不进入"运行期静默丢行"。
+
+    边界诚实声明: 它只证明该形状能被接受,**不能证明函数体真的落这些列**。
+
+    Raises:
+        ValueError: 签名不符、不可 inspect,或协议本身不可 inspect。
+    """
+    try:
+        # 模块全局查找而非常量快照: 协议改了,闸就跟着改(测试可据此机械验证)
+        protocol = inspect.signature(TelemetryRecorder.record_llm_call).parameters
+    except (TypeError, ValueError) as exc:  # pragma: no cover - 协议一向可 inspect
+        raise ValueError(f"TelemetryRecorder.record_llm_call 签名不可读取: {exc}") from exc
+    sentinels = {name: None for name in protocol if name != "self"}
+    label = type(recorder).__name__
+    method = getattr(recorder, "record_llm_call", None)
+    if method is None:
+        # 连方法都没有: 比旧签名更明确的配置错误。不让它以裸 AttributeError
+        # 逆流而上——那不属错误四分类,且现场离"注错了东西"这个真因很远
+        raise ValueError(
+            f"注入的遥测 recorder {label} 没有 record_llm_call 方法,不满足 TelemetryRecorder 端口"
+        )
+    try:
+        signature = inspect.signature(method)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(
+            f"遥测 recorder {label} 的 record_llm_call 不可 inspect(如 C 实现),"
+            "无法在装配期确认它接受当前字段形状;请换成 Python 实现或包一层"
+        ) from exc
+    try:
+        signature.bind(**sentinels)
+    except TypeError as exc:
+        raise ValueError(
+            f"遥测 recorder {label} 的 record_llm_call 签名与 TelemetryRecorder 不符"
+            f"(当前 {len(sentinels)} 个字段): {exc}。"
+            "这一条故意在装配期报错——放行的后果是每行遥测都被降级成 warning 后丢弃"
+        ) from exc
+
+
 class TelemetryEmitter:
-    """从请求与结果组装 26 字段并写入 recorder;一切写失败降级 warning。"""
+    """从请求与结果组装 36 字段并写入 recorder;一切写失败降级 warning。"""

    def __init__(
        self,
        recorder: TelemetryRecorder,
        *,
+        scope: str,
        pricing: PricingTable | None = None,
        text_cap: int | None,
    ) -> None:
-        """`text_cap` 无默认值是有意的: 它是关键行为参数,漏传即静默改变落库正文。
+        """`text_cap` 与 `scope` 无默认值是有意的: 两者都是关键行为参数。

+        `text_cap` 漏传即静默改变落库正文;`scope` 漏传则三类行都失去池名
+        ——终态失败可能根本没选出源,但 scope 始终已知,不拿 `source_name` 顶替。
        本类是库内部类,唯一构造者是三个公共 Client,必填能保证没有一处漏传。
-        同理,值域校验也放在这一处: 三个 Client 的 `text_cap` 全部汇流到这里,
+
+        同理,值域校验与**装配闸**都放在这一处: 三个 Client 全部汇流到这里,
        `GatewaySettings` 那道只管 env 一条路,而直接构造 Client 是库承诺的另一
-        条公共装配路——`text_cap=0` 会让每条正文只剩一个省略标记(P5 不得静默)。
+        条公共装配路(`text_cap=0` 会让每条正文只剩一个省略标记;P5 不得静默)。
        """
        if text_cap is not None and text_cap <= 0:
            raise ValueError(f"text_cap 必须 > 0(不截断请传 None): {text_cap}")
+        _assert_recorder_shape(recorder)
        self._recorder = recorder
+        self._scope = scope
        self._pricing = pricing
        self._text_cap = text_cap

@@ -256,8 +401,10 @@ class TelemetryEmitter:
        call_id: str,
        latency_ms: int,
        response: LLMResponse | None,
-        error: str | None,
+        error: PolyGatewayError | str | None,
        reasoning_applies: bool,
+        operation: CallOperation,
+        class_prefixed_error: bool = False,
    ) -> None:
        """逐次尝试记录(三个 Client 的重试层调用);失败尝试无用量可言,记 0 并标 unavailable。

@@ -266,7 +413,11 @@ class TelemetryEmitter:
        共用同一个 `SourceConfig` 类型,一个误配了 `ENABLE_THINKING` 的 embedding 源
        会让下面的回落算出 `auto`,给一次从来不带推理参数的调用挂上一个从未发出过的
        档。**不设默认值**: 与 `TelemetryRecorder` 同一约定,库外无第三方调用者,漏传
-        当场 TypeError,好过被静默当成"没表态"。
+        当场 TypeError,好过被静默当成"没表态"。`operation` 同理且另有一层:
+        它只能由调用点给定,**绝不读 `exc.operation`**(后者是 HTTP 子操作)。
+
+        `error` 收**领域异常对象**而非预先 `str()` 压平的文本: 状态码/底层异常类型/
+        网关正文在此提取成四列(设计 §5)。取消路径仍传既有字符串 `"cancelled"`。
        """
        usage = _AttemptUsage.of(response)
        await self._record(
@@ -284,7 +435,7 @@ class TelemetryEmitter:
            ttft_ms=usage.ttft_ms,
            max_inter_token_ms=usage.max_inter_token_ms,
            cache_hit=False,
-            error=error,
+            errors=_error_fields(error, event_kind="attempt", class_prefixed=class_prefixed_error),
            cached_prompt_tokens=usage.cached_prompt_tokens,
            model_reported=usage.model_reported,
            reasoning_tokens=usage.reasoning_tokens,
@@ -296,10 +447,19 @@ class TelemetryEmitter:
            reasoning_effort=_attempt_effort(
                request=request, source=source, response=response, applies=reasoning_applies
            ),
+            operation=operation,
+            event_kind="attempt",
+            attempts=None,
+            total_latency_ms=None,
        )

-    async def emit_cache_hit(self, *, request: ChatRequest, response: LLMResponse) -> None:
-        """缓存命中记录: cache_hit=True、latency_ms=0(VT 同款)。"""
+    async def emit_cache_hit(
+        self, *, request: ChatRequest, response: LLMResponse, operation: CallOperation
+    ) -> None:
+        """缓存命中记录: cache_hit=True、latency_ms=0(VT 同款)。
+
+        逻辑计数两列恒 NULL: 本行描述的是"一次命中",不是一次逻辑调用的终态。
+        """
        await self._record(
            request=request,
            call_id=response.call_id,
@@ -315,7 +475,7 @@ class TelemetryEmitter:
            ttft_ms=None,
            max_inter_token_ms=None,
            cache_hit=True,
-            error=None,
+            errors=_ErrorFields(),
            # 决策 B1: 与 model/prompt_tokens 同一口径,原样回放历史值。
            # 统计供应商缓存命中率必须带 WHERE cache_hit = false,否则重复计数。
            cached_prompt_tokens=response.cached_prompt_tokens,
@@ -334,12 +494,28 @@ class TelemetryEmitter:
            # 与 sampling 同一口径: 命中行没有选中源,源级档位与 `nearest` 映射
            # 都无从谈起,只记调用方这次要的档(response 里那个是历史那次实发的)
            reasoning_effort=_normalize_effort(request.reasoning_effort),
+            operation=operation,
+            event_kind="cache_hit",
+            attempts=None,
+            total_latency_ms=None,
        )

    async def emit_terminal_failure(
-        self, *, request: ChatRequest, call_id: str, latency_ms: int, error: str
+        self,
+        *,
+        request: ChatRequest,
+        call_id: str,
+        error: PolyGatewayError | str,
+        operation: CallOperation,
+        stats: CallStats,
+        class_prefixed_error: bool = False,
    ) -> None:
-        """scope 级失败/取消记录: 无具体源,溯源字段置空标记。"""
+        """一次**逻辑调用**的失败终态: 无具体源,溯源字段置空标记。
+
+        `latency_ms` 与 `total_latency_ms` 同取**同一份冻结快照**,避免双时钟微差;
+        故本方法不再收 `latency_ms`。token 与 cost 一律不从 attempt 行复制
+        (费用聚合仍只由 attempt / cache_hit 行决定,口径不变)。
+        """
        await self._record(
            request=request,
            call_id=call_id,
@@ -351,11 +527,13 @@ class TelemetryEmitter:
            prompt_tokens=0,
            completion_tokens=0,
            usage_source="unavailable",
-            latency_ms=latency_ms,
+            latency_ms=stats.total_latency_ms,
            ttft_ms=None,
            max_inter_token_ms=None,
            cache_hit=False,
-            error=error,
+            errors=_error_fields(
+                error, event_kind="terminal_failure", class_prefixed=class_prefixed_error
+            ),
            cached_prompt_tokens=None,
            model_reported=None,
            reasoning_tokens=None,
@@ -368,6 +546,10 @@ class TelemetryEmitter:
            meta=request.meta,
            # 可能根本没选出源,故与 sampling 同样只取请求档
            reasoning_effort=_normalize_effort(request.reasoning_effort),
+            operation=operation,
+            event_kind="terminal_failure",
+            attempts=stats.attempts,
+            total_latency_ms=stats.total_latency_ms,
        )

    async def _record(
@@ -387,7 +569,9 @@ class TelemetryEmitter:
        ttft_ms: float | None,
        max_inter_token_ms: float | None,
        cache_hit: bool,
-        error: str | None,
+        # 1.3.5: 错误四列已由 `_error_fields` 定型(三种入参形态 × 两类行的唯一规则所有者),
+        # 本方法只搬运——拆成五个平铺参数就是把"一处定型"换回"三处各自拼"
+        errors: _ErrorFields,
        cached_prompt_tokens: int | None,
        model_reported: str | None,
        sampling: str | None,
@@ -403,6 +587,11 @@ class TelemetryEmitter:
        # 注释),本方法只搬运——把定型放这里就得再传一遍 response/source,等于把
        # "唯一 record_llm_call 调用点"换成"两处口径判断",那正是要避免的复制
        reasoning_effort: str | None,
+        # —— 1.3.5: 行形态与逻辑调用快照 ——
+        operation: CallOperation,
+        event_kind: EventKind,
+        attempts: int | None,
+        total_latency_ms: int | None,
    ) -> None:
        try:
            # 成本换算(M2 §6): 成功行按单价换算;缓存命中 0.0(未产生新调用);
@@ -413,7 +602,7 @@ class TelemetryEmitter:
                # 用量不可得: 宁可算不出成本,也不算错成本(解耦设计 §3.1 不变式)。
                # 必须排在 cache_hit 之后——缓存命中未产生新调用,0.0 是事实而非未知
                cost = None
-            elif error is None and model and self._pricing is not None:
+            elif errors.error is None and model and self._pricing is not None:
                cost = self._pricing.cost(
                    model, prompt_tokens, completion_tokens, cached_prompt_tokens
                )
@@ -425,6 +614,7 @@ class TelemetryEmitter:
                _cap_messages(digest_messages(request.messages), self._text_cap),
                ensure_ascii=False,
            )
+            context = request.call_context
            await self._recorder.record_llm_call(
                call_id=call_id,
                parent_call_id=request.parent_call_id,
@@ -442,7 +632,7 @@ class TelemetryEmitter:
                ttft_ms=ttft_ms,
                max_inter_token_ms=max_inter_token_ms,
                cache_hit=cache_hit,
-                error=error,
+                error=errors.error,
                cost=cost,
                cached_prompt_tokens=cached_prompt_tokens,
                model_reported=model_reported,
@@ -457,6 +647,19 @@ class TelemetryEmitter:
                # Postgres 那一路悄悄少一列数据
                thinking_observation=_normalize_observation(thinking_observation),
                reasoning_effort=reasoning_effort,
+                # —— 1.3.5 十列 ——
+                scope=self._scope,
+                # 调用点给定的公开方法四值,**绝不读 `exc.operation`**(设计 §5 I1/I2)
+                operation=operation,
+                # 上下文缺席(库内现场构造的 ChatRequest)→ NULL,**不造 ID**(I5)
+                logical_call_id=None if context is None else context.logical_call_id,
+                event_kind=event_kind,
+                http_status_code=errors.http_status_code,
+                error_type=errors.error_type,
+                cause_type=errors.cause_type,
+                error_body=errors.error_body,
+                attempts=attempts,
+                total_latency_ms=total_latency_ms,
            )
        except asyncio.CancelledError:
            raise
@@ -464,8 +667,45 @@ class TelemetryEmitter:
            logger.warning("遥测记录失败(降级不冒泡): {}", exc)


+async def emit_terminal_once(
+    emitter: TelemetryEmitter | None,
+    *,
+    request: ChatRequest,
+    context: _CallContext,
+    error: PolyGatewayError | str,
+    operation: CallOperation,
+    class_prefixed_error: bool = False,
+) -> None:
+    """三个 client 共用的**终态唯一出口**: 去重 + 同步冻结快照 + best effort 写入。
+
+    去重由 `claim_terminal()` 承担(每逻辑调用至多一条终态行);`emitter is None`
+    或已写过 → 直接返回。写入侧异常按既有降级只落 warning(在 `_record` 内)。
+
+    **`CancelledError` 原样传播**(取消优先,不 shield、不开后台任务): 这一次
+    `await` 本身就是新的取消点,外部取消落在它上时调用方会看到 `CancelledError`
+    而非领域错误——与 TelemetryMW 的历史行为同款,已经人类批准(设计 §6/§10)。
+    快照冻结是**同步**动作,故终态行不含它自身的写入耗时。
+    """
+    if emitter is None or not context.claim_terminal():
+        return
+    stats = context.snapshot()
+    await emitter.emit_terminal_failure(
+        request=request,
+        call_id=str(uuid.uuid4()),
+        error=error,
+        operation=operation,
+        stats=stats,
+        class_prefixed_error=class_prefixed_error,
+    )
+
+
 class TelemetryMW:
-    """洋葱最外层: 观测尝试层看不见的路径,任何路径都留痕(遥测必录)。"""
+    """洋葱最外层: 只观测尝试层看不见的**缓存命中**。
+
+    1.3.5 起不再在此写终态失败行: 终态由 `GatewayClient.chat` 的公开边界经
+    `emit_terminal_once` 统一写出。两处同时写会让同一次失败出两条终态行,
+        而下游正是按 `event_kind = 'terminal_failure'` 计失败调用次数的。
+    """

    def __init__(
        self, emitter: TelemetryEmitter, now: Callable[[], float] = time.monotonic
@@ -474,26 +714,7 @@ class TelemetryMW:
        self._now = now

    async def __call__(self, request: ChatRequest, call_next: CallNext) -> LLMResponse:
-        started = self._now()
-        try:
        response = await call_next(request)
-        except (GatewayUnavailableError, GovernanceBackendError, SourceNotConfiguredError) as exc:
-            await self._emitter.emit_terminal_failure(
-                request=request,
-                call_id=str(uuid.uuid4()),
-                latency_ms=int((self._now() - started) * 1000),
-                error=str(exc),
-            )
-            raise
-        except asyncio.CancelledError:
-            # 尽力而为: 取消也留痕(§5.1 约定④);随后立即重抛
-            await self._emitter.emit_terminal_failure(
-                request=request,
-                call_id=str(uuid.uuid4()),
-                latency_ms=int((self._now() - started) * 1000),
-                error="cancelled",
-            )
-            raise
        if response.cache_hit:
-            await self._emitter.emit_cache_hit(request=request, response=response)
+            await self._emitter.emit_cache_hit(request=request, response=response, operation="chat")
        return response
@@ -37,7 +37,7 @@ from polygateway.middleware.admission import SourceAdmission, settle_and_release
 from polygateway.middleware.breaker import BreakerGate
 from polygateway.middleware.ratelimit import QuotaGate
 from polygateway.middleware.retry import StallClock, _failure_reason, backoff_delay
-from polygateway.middleware.telemetry import TelemetryEmitter
+from polygateway.middleware.telemetry import TelemetryEmitter, emit_terminal_once
 from polygateway.ports import OutcomeAwareSelector
 from polygateway.types import (
    CallStats,
@@ -67,6 +67,7 @@ if TYPE_CHECKING:
    )
    from polygateway.types import (
        BackpressurePolicy,
+        CallOperation,
        OcrLayoutTransportResult,
        OcrTextTransportResult,
        RetryPolicy,
@@ -128,7 +129,9 @@ class OcrClient:
        self._breaker = BreakerGate(breaker, scope=self._scope)
        self._transport = transport
        self._retry = retry
-        self._emitter = TelemetryEmitter(telemetry, text_cap=text_cap) if telemetry else None
+        self._emitter = (
+            TelemetryEmitter(telemetry, scope=self._scope, text_cap=text_cap) if telemetry else None
+        )
        self._telemetry = telemetry
        # 限流/熔断后端在此之外只以 QuotaGate/BreakerGate 的形态存在,自持一份
        # 引用才关得到自建的 redis 客户端(设计 §3.4)
@@ -179,7 +182,13 @@ class OcrClient:
            tenant_id, meta, origin="recognize_text(tenant_id=..., meta=...)"
        )
        outcome, call_stats = await self._call(
-            "text", image, session_id, parent_call_id, dimension_tenant_id, dimensions
+            "text",
+            "recognize_text",
+            image,
+            session_id,
+            parent_call_id,
+            dimension_tenant_id,
+            dimensions,
        )
        result = outcome.result
        return OcrTextResult(
@@ -210,7 +219,13 @@ class OcrClient:
            tenant_id, meta, origin="parse_layout(tenant_id=..., meta=...)"
        )
        outcome, call_stats = await self._call(
-            "layout", image, session_id, parent_call_id, dimension_tenant_id, dimensions
+            "layout",
+            "parse_layout",
+            image,
+            session_id,
+            parent_call_id,
+            dimension_tenant_id,
+            dimensions,
        )
        result = outcome.result
        return OcrLayoutResult(
@@ -241,6 +256,7 @@ class OcrClient:
    async def _call(
        self,
        kind: _OcrKind,
+        operation: CallOperation,
        image: bytes,
        session_id: str | None,
        parent_call_id: str | None,
@@ -254,6 +270,46 @@ class OcrClient:
        # M1 例外: `image` 校验在 `_call` 内而非公开方法,故上下文在该校验
        # **通过之后**创建——这样设计 §3 的"校验在统计边界外"对 OCR 才成立
        context = _CallContext(now=self._now)
+        try:
+            return await self._run(
+                kind, operation, image, session_id, parent_call_id, tenant_id, meta, context
+            )
+        except PolyGatewayError as exc:
+            await self._emit_terminal(
+                kind, operation, image, session_id, parent_call_id, tenant_id, meta, context, exc
+            )
+            raise
+        except asyncio.CancelledError:
+            # 三条链路同一口径尽力写一条(允许 0 条);取消优先,不 shield
+            await self._emit_terminal(
+                kind,
+                operation,
+                image,
+                session_id,
+                parent_call_id,
+                tenant_id,
+                meta,
+                context,
+                "cancelled",
+            )
+            raise
+
+    async def _run(
+        self,
+        kind: _OcrKind,
+        operation: CallOperation,
+        image: bytes,
+        session_id: str | None,
+        parent_call_id: str | None,
+        tenant_id: str | None,
+        meta: dict[str, Any],
+        context: _CallContext,
+    ) -> tuple[_AttemptOutcome, CallStats]:
+        """选源与重试循环。
+
+        无源的 raise 必须在本方法内(而非循环之前的调用方): 它得被 `_call` 的
+        `try` 包住,否则无源终态行根本写不出来(设计 §3.5)。
+        """
        if not self._sources:
            raise AllSourcesExhausted(scope=self._scope, reason="no_sources", retry_after_s=0.0)
        fails = 0
@@ -268,6 +324,7 @@ class OcrClient:
            async with clock.attempting():
                outcome = await self._attempt(
                    kind,
+                    operation,
                    image,
                    *picked,
                    reasons,
@@ -293,6 +350,7 @@ class OcrClient:
    async def _attempt(
        self,
        kind: _OcrKind,
+        operation: CallOperation,
        image: bytes,
        source: SourceConfig,
        permit: Permit,
@@ -318,6 +376,7 @@ class OcrClient:
            latency_ms = int((self._now() - started) * 1000)
            await self._emit(
                kind,
+                operation,
                image,
                source,
                call_id,
@@ -326,6 +385,7 @@ class OcrClient:
                parent_call_id,
                tenant_id,
                meta,
+                context,
                result,
            )
            return _AttemptOutcome(result, source, call_id, latency_ms)
@@ -333,6 +393,7 @@ class OcrClient:
            await self._gate_on_terminal(exc, entry)
            await self._emit(
                kind,
+                operation,
                image,
                source,
                call_id,
@@ -341,6 +402,7 @@ class OcrClient:
                parent_call_id,
                tenant_id,
                meta,
+                context,
                error=exc,
            )
            raise
@@ -349,6 +411,7 @@ class OcrClient:
                await self._record_quietly(self._breaker.release_probe(entry))
            await self._emit(
                kind,
+                operation,
                image,
                source,
                call_id,
@@ -357,6 +420,7 @@ class OcrClient:
                parent_call_id,
                tenant_id,
                meta,
+                context,
                error="cancelled",
            )
            raise
@@ -368,6 +432,7 @@ class OcrClient:
            self._feed_outcome(source.name, ok=False)
            await self._emit(
                kind,
+                operation,
                image,
                source,
                call_id,
@@ -376,6 +441,7 @@ class OcrClient:
                parent_call_id,
                tenant_id,
                meta,
+                context,
                error=exc,
            )
            return _FailedAttempt(exc, immediate=dead)
@@ -417,9 +483,60 @@ class OcrClient:
        except (GovernanceBackendError, SourceNotConfiguredError) as exc:
            logger.warning("OCR 治理记账写回降级(不冒泡): {}", exc)

+    def _request_for(
+        self,
+        kind: _OcrKind,
+        image: bytes,
+        session_id: str | None,
+        parent_call_id: str | None,
+        tenant_id: str | None,
+        meta: dict[str, Any],
+        context: _CallContext,
+    ) -> ChatRequest:
+        """OCR 行的现场 ChatRequest: 占位摘要,**图像 bytes 永不入库**。
+
+        尝试行与终态行共用同一个构造点: 占位字面量复制成两份就会漂移。
+        调用方维度与上下文必须显式填回(OCR 不走 chat 洋葱),否则 OCR 行的
+        维度恒为空、`logical_call_id` 恒为 NULL。
+        """
+        return ChatRequest(
+            messages=[{"role": "user", "content": f"<ocr:{kind} image_bytes={len(image)}>"}],
+            session_id=session_id,
+            parent_call_id=parent_call_id,
+            tenant_id=tenant_id,
+            meta=meta,
+            call_context=context,
+        )
+
+    async def _emit_terminal(
+        self,
+        kind: _OcrKind,
+        operation: CallOperation,
+        image: bytes,
+        session_id: str | None,
+        parent_call_id: str | None,
+        tenant_id: str | None,
+        meta: dict[str, Any],
+        context: _CallContext,
+        error: PolyGatewayError | str,
+    ) -> None:
+        """终态行: 沿用 `<ocr:{kind} image_bytes=…>` 占位,错误文本保留类名前缀。"""
+        await emit_terminal_once(
+            self._emitter,
+            request=self._request_for(
+                kind, image, session_id, parent_call_id, tenant_id, meta, context
+            ),
+            context=context,
+            error=error,
+            operation=operation,
+            # metric ocr-call-success 的注册口径是按类名归组,终态行同款保留
+            class_prefixed_error=True,
+        )
+
    async def _emit(
        self,
        kind: _OcrKind,
+        operation: CallOperation,
        image: bytes,
        source: SourceConfig,
        call_id: str,
@@ -428,20 +545,15 @@ class OcrClient:
        parent_call_id: str | None,
        tenant_id: str | None,
        meta: dict[str, Any],
+        context: _CallContext,
        result: OcrTextTransportResult | OcrLayoutTransportResult | None = None,
-        error: object | None = None,
+        error: PolyGatewayError | str | None = None,
    ) -> None:
        """逐尝试遥测(单一 Emitter): messages 占位摘要,图像 bytes 绝不入库。"""
        if self._emitter is None:
            return
-        # 这个 ChatRequest 只为复用同一个 Emitter 而现场构造(OCR 不走 chat 洋葱),
-        # 故调用方维度必须在这里显式填回,否则 OCR 行的维度恒为空
-        request = ChatRequest(
-            messages=[{"role": "user", "content": f"<ocr:{kind} image_bytes={len(image)}>"}],
-            session_id=session_id,
-            parent_call_id=parent_call_id,
-            tenant_id=tenant_id,
-            meta=meta,
+        request = self._request_for(
+            kind, image, session_id, parent_call_id, tenant_id, meta, context
        )
        latency_ms = int((self._now() - started) * 1000)
        response = None
@@ -461,20 +573,19 @@ class OcrClient:
                source_name=source.name,
                usage_source="measured",
            )
-        # 错误带异常类名前缀(metric ocr-call-success 注册口径: 按类名归组)
-        if error is None or isinstance(error, str):
-            error_text = error
-        else:
-            error_text = f"{type(error).__name__}: {error}"
+        # 错误文本的类名前缀现由出口的显式策略参数承担(设计 §6 I7):
+        # 三处各拼一遍才是下一次漂移的种子,而取消行传的是字符串,不受前缀影响
        await self._emitter.emit_attempt(
            request=request,
            source=source,
            call_id=call_id,
            latency_ms=latency_ms,
            response=response,
-            error=error_text,
+            error=error,
            # OCR 走 MonkeyOCR 自有端点,没有推理参数可言(理由同 embedding)
            reasoning_applies=False,
+            operation=operation,
+            class_prefixed_error=True,
        )

    @staticmethod
@@ -271,7 +271,7 @@ class TelemetryStatusProvider(Protocol):

@runtime_checkable
 class TelemetryRecorder(Protocol):
-    """遥测后端;26 字段冻结(M1 设计 §4.4 + issue #3/#4/#11/#16/#20),唯一调用点是 TelemetryEmitter。
+    """遥测后端;36 字段冻结(M1 设计 §4.4 + issue #3/#4/#11/#16/#20 + 1.3.5),唯一调用点是 TelemetryEmitter。

    新增参数不设默认值: 库外无第三方实现者(三项目迁移时删除了各自的同名
    Protocol),完整签名的成本为零,而少写一列会被 emitter 的降级吞成 warning。
@@ -286,6 +286,15 @@ class TelemetryRecorder(Protocol):

    recorder 只负责落库,不做任何语义判断,与 `sampling` 列由
    `canonical_sampling_json()` 在 emitter 侧定型是同一先例。
+
+    1.3.5 新增十列同理已在 emitter 侧定型: `operation` 是**公开方法**四值之一
+    (与 `PolyGatewayError.operation` 这个 HTTP 子操作是两个语义);`event_kind` 区分
+    attempt / cache_hit / terminal_failure 三类行;`attempts` 与 `total_latency_ms`
+    只在终态行非空;`error_body` 沿用 transport 侧 `summarize_body` 的上限,
+    **不进 `PGW_TELEMETRY_TEXT_CAP` 的覆盖面**。
+
+    **本签名是装配闸的唯一事实源**: `TelemetryEmitter.__init__` 按它派生参数名做
+    一次 `signature.bind` 形状校验(设计 §7),改本签名即改闸的判据。
    """

    async def record_llm_call(
@@ -317,4 +326,14 @@ class TelemetryRecorder(Protocol):
        meta: str,
        thinking_observation: str,
        reasoning_effort: str | None,
+        scope: str,
+        operation: str,
+        logical_call_id: str | None,
+        event_kind: str,
+        http_status_code: int | None,
+        error_type: str | None,
+        cause_type: str | None,
+        error_body: str | None,
+        attempts: int | None,
+        total_latency_ms: int | None,
    ) -> None: ...
@@ -5,8 +5,8 @@
 多处各存一份必然漂移,而漂移的表现是"下游照打印的 SQL 建完表,库仍报缺列"。

 **`COLUMNS` 是 INSERT 字段序,不是物理列序**: 数据库自填的 `created_at` 不在其中(它带
-`DEFAULT now()` / `datetime('now')`,库从不显式写它)。物理表列 = 26 个 INSERT 字段 +
-`created_at` = 27;列数断言一律按物理列数写,两套口径混用是最易错处。
+`DEFAULT now()` / `datetime('now')`,库从不显式写它)。物理表列 = 36 个 INSERT 字段 +
+`created_at` = 37;列数断言一律按物理列数写,两套口径混用是最易错处。

 本模块只依赖标准库: `telemetry/` 与 `backends/`、`transports/`、`structured/` 同层且
 互不依赖(import-linter 契约执法)。
@@ -52,7 +52,17 @@ CREATE TABLE IF NOT EXISTS llm_calls (
    tenant_id            TEXT    NOT NULL DEFAULT '',
    meta                 TEXT    NOT NULL DEFAULT '{}',
    thinking_observation TEXT,
-    reasoning_effort     TEXT
+    reasoning_effort     TEXT,
+    scope                TEXT,
+    operation            TEXT,
+    logical_call_id      TEXT,
+    event_kind           TEXT,
+    http_status_code     INTEGER,
+    error_type           TEXT,
+    cause_type           TEXT,
+    error_body           TEXT,
+    attempts             INTEGER,
+    total_latency_ms     INTEGER
 );
 """

@@ -84,7 +94,17 @@ CREATE TABLE IF NOT EXISTS llm_calls (
    tenant_id            TEXT NOT NULL DEFAULT '',
    meta                 JSONB NOT NULL DEFAULT '{}'::jsonb,
    thinking_observation TEXT,
-    reasoning_effort     TEXT
+    reasoning_effort     TEXT,
+    scope                TEXT,
+    operation            TEXT,
+    logical_call_id      TEXT,
+    event_kind           TEXT,
+    http_status_code     INTEGER,
+    error_type           TEXT,
+    cause_type           TEXT,
+    error_body           TEXT,
+    attempts             INTEGER,
+    total_latency_ms     INTEGER
 );
 """

@@ -105,6 +125,18 @@ SQLITE_BACKFILL = (
    # 同样可空,但这里 NULL 表达的是"调用方没表态"(issue #20): 它与 'none'
    # (明确要求不推理)是两回事,折叠成任一档都等于替上游声称了它没说过的事
    ("reasoning_effort", "TEXT"),
+    # 1.3.5 十列: 全部可空且无默认值——旧行的 NULL 表达的是"补列之前根本没记过
+    # 这件事",与任何哨兵值都不是一回事,故不回填(设计 §5)
+    ("scope", "TEXT"),
+    ("operation", "TEXT"),
+    ("logical_call_id", "TEXT"),
+    ("event_kind", "TEXT"),
+    ("http_status_code", "INTEGER"),
+    ("error_type", "TEXT"),
+    ("cause_type", "TEXT"),
+    ("error_body", "TEXT"),
+    ("attempts", "INTEGER"),
+    ("total_latency_ms", "INTEGER"),
 )

 # PG 补列的列定义。语句由此派生成两份文本(见下),使"库内执行的那份"与"打印给
@@ -120,6 +152,17 @@ _PG_BACKFILL_DECLS = (
    # 可空,理由同 SQLITE_BACKFILL 同名项
    ("thinking_observation", "TEXT"),
    ("reasoning_effort", "TEXT"),
+    # 1.3.5 十列,列序与 SQLITE_BACKFILL 逐项对齐(两条路径的物理列序不许分叉)
+    ("scope", "TEXT"),
+    ("operation", "TEXT"),
+    ("logical_call_id", "TEXT"),
+    ("event_kind", "TEXT"),
+    ("http_status_code", "INTEGER"),
+    ("error_type", "TEXT"),
+    ("cause_type", "TEXT"),
+    ("error_body", "TEXT"),
+    ("attempts", "INTEGER"),
+    ("total_latency_ms", "INTEGER"),
 )

 # 新列排在 created_at 之后: 与旧表 ALTER 追加的位置一致(见 SQLITE_BACKFILL 同款注释)。
@@ -158,6 +201,17 @@ COLUMNS = (
    "meta",
    "thinking_observation",
    "reasoning_effort",
+    # —— 1.3.5 逻辑调用统计与结构化失败诊断(issue #19/#23)——
+    "scope",
+    "operation",
+    "logical_call_id",
+    "event_kind",
+    "http_status_code",
+    "error_type",
+    "cause_type",
+    "error_body",
+    "attempts",
+    "total_latency_ms",
 )

 _COLUMN_SET = frozenset(COLUMNS)
@@ -154,18 +154,28 @@ def _classify(status: int) -> tuple[type[PolyGatewayError], str]:


 def _status_to_error(
-    source: SourceConfig, status: int, body_text: str, headers: Mapping[str, str]
+    source: SourceConfig,
+    status: int,
+    body_text: str,
+    headers: Mapping[str, str],
+    *,
+    operation: str,
 ) -> Exception:
    """非 2xx → 领域错误,**全部分支**携带响应体摘要(issue #10)。

    摘要只算一次,message 与 `body_text` 共用同一份串: 两份不同长度会让"遥测里
    看到的"与"下游 catch 到的"对不上,排查时反而多一层困惑。
+
+    `operation` 是 **HTTP 子操作**词表(`chat` / `embedding` / `ocr_text` / ...),由
+    调用点显式给定。它曾被硬编码成 `"chat"`,而 `embed()` 的非 200 分支也走它
+    ——于是现存所有 embedding HTTP 失败的 `exc.operation` 都是错的(1.3.5 设计 §5 I1)。
+    注意它与遥测新列 `operation`(公开方法四值)是**两个语义**,不做自动转换。
    """
    summary = summarize_body(body_text)
    ctx: dict[str, Any] = {
        "source_name": source.name,
        "status_code": status,
-        "operation": "chat",
+        "operation": operation,
        "body_text": summary,
    }
    if status == 429:
@@ -509,7 +519,10 @@ class OpenAICompatTransport:
        except httpx.TransportError as exc:
            raise TransientError(f"{source.name} 网络错误: {exc}", **ctx) from exc
        if resp.status_code != 200:
-            raise _status_to_error(source, resp.status_code, resp.text, resp.headers)
+            # 历史误标修正: 本分支属 `embed()`,与上方 ctx 同为 `"embedding"`
+            raise _status_to_error(
+                source, resp.status_code, resp.text, resp.headers, operation="embedding"
+            )
        return _parse_embedding_payload(resp, source, len(texts))

    async def _complete_stream(
@@ -524,7 +537,9 @@ class OpenAICompatTransport:
        async with client.stream("POST", url, json=payload) as resp:
            if resp.status_code != 200:
                body = (await resp.aread()).decode("utf-8", errors="replace")
-                raise _status_to_error(source, resp.status_code, body, resp.headers)
+                raise _status_to_error(
+                    source, resp.status_code, body, resp.headers, operation="chat"
+                )
            sink: dict[str, Any] = {}
            guarded = stream_with_liveness_timeouts(
                _iter_sse_deltas(resp.aiter_lines(), sink),
@@ -622,7 +637,9 @@ class OpenAICompatTransport:
        """非流式快路径(三项目均无,库新增): 单 JSON 响应,仅 total 超时。"""
        resp = await client.post(url, json=payload)
        if resp.status_code != 200:
-            raise _status_to_error(source, resp.status_code, resp.text, resp.headers)
+            raise _status_to_error(
+                source, resp.status_code, resp.text, resp.headers, operation="chat"
+            )
        try:
            body = resp.json()
        except json.JSONDecodeError as exc:
@@ -570,7 +570,7 @@ class TestTelemetryCapDoesNotPoisonTheCacheKey:
        before = build_cache_key("m", messages, "proj", None)

        rec = self._Rows()
-        await TelemetryEmitter(rec, text_cap=8).emit_attempt(
+        await TelemetryEmitter(rec, text_cap=8, scope="LLM").emit_attempt(
            request=ChatRequest(messages=messages),
            source=SourceConfig(
                name="s1",
@@ -585,6 +585,7 @@ class TestTelemetryCapDoesNotPoisonTheCacheKey:
            response=_resp(),
            error=None,
            reasoning_applies=True,
+            operation="chat",
        )
        # 截断确实发生了(否则本用例恒真)
        logged = json.loads(rec.rows[0]["messages"])
@@ -13,6 +13,7 @@ from polygateway import (
    GatewayClient,
    GatewaySettings,
    RequestRejectedError,
+    ResultInvalidError,
    gather_bounded,
 )
 from polygateway.backends.memory.breaker import InMemoryGate
@@ -25,6 +26,7 @@ from polygateway.transports.openai_compat import OpenAICompatTransport
 from polygateway.types import (
    BackpressurePolicy,
    BreakerConfig,
+    ChatRequest,
    Effort,
    GlobalLimits,
    RetryPolicy,
@@ -846,11 +848,19 @@ _CACHE_ENV = dict(


 class _Closable:
-    """记 close 次数的假组件;所有权纪律的唯一观测点。"""
+    """记 close 次数的假组件;所有权纪律的唯一观测点。
+
+    带 `record_llm_call(**fields)` 是因为它也被当作注入的 telemetry recorder 用:
+    1.3.5 的装配闸在构造期就会拒掉不满足 `TelemetryRecorder` 的对象
+    (否则下游升级后 100% 丢遥测而调用照常成功)。`**fields` 形态天然兼容。
+    """

    def __init__(self):
        self.closed = 0

+    async def record_llm_call(self, **fields):
+        pass
+
    async def aclose(self):
        self.closed += 1

@@ -861,6 +871,9 @@ class _SyncClosable:
    def __init__(self):
        self.closed = 0

+    async def record_llm_call(self, **fields):
+        pass
+
    def close(self):
        self.closed += 1

@@ -1499,3 +1512,136 @@ class TestLogicalCallStats:
        async with _client() as client:
            with pytest.raises(ValueError, match="meta"):
                await client.chat(self._MSG, meta={"BAD-KEY": 1})
+
+
+class TestChatTerminalFailureRows:
+    """chat 链路的**真实**终态行(1.3.5 设计 §6;补漏而非改口径)。
+
+    这些路径改前一条失败行都没有(结构化耗尽)或只有尝试行,
+    "这次调用到底失败了几次"因此 SQL 答不出来。
+    """
+
+    _MSG = [{"role": "user", "content": "hi"}]
+
+    def _rows(self, recorder, kind):
+        return [r for r in recorder.rows if r["event_kind"] == kind]
+
+    async def test_structured_exhaustion_writes_the_only_failure_row(self):
+        """结构化耗尽发生在 transport 成功之后: attempt 行全是成功行,终态是唯一记录。"""
+        from pydantic import BaseModel
+
+        class Answer(BaseModel):
+            answer: int
+
+        recorder = _MemoryRecorder()
+        async with _client(
+            handler=lambda request: _sse("not json at all"),
+            telemetry=recorder,
+            structured_max_retries=1,
+        ) as client:
+            with pytest.raises(ResultInvalidError):
+                await client.chat(self._MSG, structured=Answer)
+        attempts = self._rows(recorder, "attempt")
+        terminals = self._rows(recorder, "terminal_failure")
+        assert [a["error"] for a in attempts] == [None, None]  # 两次尝试都成功
+        assert len(terminals) == 1
+        row = terminals[0]
+        assert row["error_type"] == "ResultInvalidError"
+        # C2: 有界结构化说明并入 error,且不含 raw_text(正文预算已由 attempt 行承担)
+        assert "validation=" in row["error"] or "repair=" in row["error"]
+        assert len(row["error"]) < 1200
+
+    async def test_retry_exhaustion_writes_exactly_one_terminal_row(self):
+        """重试耗尽: 逐次 attempt 行之外只能有**一条**终态行。
+
+        终态行的 `attempts` 是整次逻辑调用的真实尝试数——这正是改前 SQL
+        答不出的"这次调用到底重试了几次"。
+        """
+        recorder = _MemoryRecorder()
+        async with _client(
+            handler=lambda request: httpx.Response(503),
+            telemetry=recorder,
+            retry=RetryPolicy(3, 0.001, 0.01),
+        ) as client:
+            with pytest.raises(AllSourcesExhausted):
+                await client.chat(self._MSG)
+        attempts = self._rows(recorder, "attempt")
+        terminals = self._rows(recorder, "terminal_failure")
+        assert len(attempts) == 3
+        assert len(terminals) == 1
+        row = terminals[0]
+        assert row["error_type"] == "AllSourcesExhausted"
+        assert row["attempts"] == 3  # 整次逻辑调用的尝试数
+        assert row["scope"] == "llm" and row["operation"] == "chat"
+        # 终态行的 latency_ms 与 total_latency_ms 同取一份冻结快照
+        assert row["latency_ms"] == row["total_latency_ms"]
+
+    async def test_request_rejected_now_has_both_an_attempt_and_a_terminal_row(self):
+        """已批准的下游可见变化: 400 密集负载的错误行翻倍,失败计数只能取终态。"""
+        recorder = _MemoryRecorder()
+        async with _client(
+            handler=lambda request: httpx.Response(400, json={"error": {"message": "bad"}}),
+            telemetry=recorder,
+        ) as client:
+            with pytest.raises(RequestRejectedError):
+                await client.chat(self._MSG)
+        attempts = self._rows(recorder, "attempt")
+        terminals = self._rows(recorder, "terminal_failure")
+        assert len(attempts) == 1 and attempts[0]["http_status_code"] == 400
+        assert len(terminals) == 1
+        # C1: 终态不搬运最后一次 attempt 的状态码与正文
+        assert terminals[0]["http_status_code"] is None
+        assert terminals[0]["error_body"] is None
+        # 两类行共享同一 logical_call_id,归因查询才连得起来
+        assert attempts[0]["logical_call_id"] == terminals[0]["logical_call_id"]
+
+    async def test_non_domain_exception_writes_no_terminal_row(self):
+        """编程错原样传播,本版**不承诺**任何统计或终态行,也不偷偷改分类。"""
+        recorder = _MemoryRecorder()
+
+        async def boom(request):
+            raise KeyError("programming error")
+
+        async with _client(telemetry=recorder) as client:
+            client._handler = boom
+            with pytest.raises(KeyError):
+                await client.chat(self._MSG)
+        assert self._rows(recorder, "terminal_failure") == []
+
+    async def test_cancellation_writes_at_most_one_terminal_row(self):
+        """取消尽力写一条(允许 0),且 `CancelledError` 类型与语义不变。"""
+        recorder = _MemoryRecorder()
+
+        async def hang(request):
+            raise asyncio.CancelledError
+
+        async with _client(telemetry=recorder) as client:
+            client._handler = hang
+            with pytest.raises(asyncio.CancelledError):
+                await client.chat(self._MSG)
+        terminals = self._rows(recorder, "terminal_failure")
+        assert len(terminals) == 1
+        assert terminals[0]["error"] == "cancelled"
+        # 字符串不解析猜诊断
+        assert terminals[0]["error_type"] is None
+
+    async def test_terminal_row_is_written_once_per_logical_call(self):
+        """`claim_terminal` 去重: 同一次调用即便出口被多次触达也只有一条。"""
+        from polygateway.middleware.telemetry import emit_terminal_once
+        from polygateway.types import _CallContext
+
+        recorder = _MemoryRecorder()
+        async with _client(telemetry=recorder) as client:
+            context = _CallContext(now=client._now)
+            request = ChatRequest(messages=self._MSG, call_context=context)
+            for _ in range(3):
+                await emit_terminal_once(
+                    client._emitter,
+                    request=request,
+                    context=context,
+                    error=AllSourcesExhausted(
+                        scope="llm", reason="retry_exhausted", retry_after_s=1.0
+                    ),
+                    operation="chat",
+                )
+        assert len(self._rows(recorder, "terminal_failure")) == 1
@@ -580,7 +580,12 @@ class TestReasonlessTelemetryContract:
        client, _ = _embed_client([_src(enable_thinking=True)], script, telemetry=recorder)
        with pytest.raises(AllSourcesExhausted if exhausted else RequestRejectedError):
            await client.embed(["text"])
-        assert len(recorder.rows) == len(script)
+        # 1.3.5: 逐次 attempt 行之外,本次逻辑调用另有**一条**终态行
+        attempts = [r for r in recorder.rows if r["event_kind"] == "attempt"]
+        terminals = [r for r in recorder.rows if r["event_kind"] == "terminal_failure"]
+        assert len(attempts) == len(script)
+        assert len(terminals) == 1
+        # 推理档在三类行上都必须是 NULL: embed payload 从不带推理参数
        assert all(r["error"] and r["reasoning_effort"] is None for r in recorder.rows)

    async def test_embedding_wire_ignores_reasoning_configuration(self):
@@ -613,7 +613,13 @@ class TestReasonlessTelemetryContract:
        client, _, _ = _client([_src(enable_thinking=True)], script, telemetry=recorder)
        with pytest.raises(AllSourcesExhausted if exhausted else RequestRejectedError):
            await getattr(client, method)(b"image")
-        assert len(recorder.rows) == len(script)
+        # 1.3.5: 逐次 attempt 行之外,本次逻辑调用另有**一条**终态行
+        attempts = [r for r in recorder.rows if r["event_kind"] == "attempt"]
+        terminals = [r for r in recorder.rows if r["event_kind"] == "terminal_failure"]
+        assert len(attempts) == len(script)
+        assert len(terminals) == 1
+        # 终态行的 operation 是**公开方法**名,与尝试行一致
+        assert terminals[0]["operation"] == method
        assert all(r["error"] and r["reasoning_effort"] is None for r in recorder.rows)


@@ -116,7 +116,7 @@ async def _recorded_cost(result, source):
        source_name=source.name,
        usage_source=result.usage_source,
    )
-    await TelemetryEmitter(recorder, pricing=_PRICING, text_cap=None).emit_attempt(
+    await TelemetryEmitter(recorder, pricing=_PRICING, text_cap=None, scope="LLM").emit_attempt(
        request=ChatRequest(messages=[{"role": "user", "content": "hi"}]),
        source=source,
        call_id="cid-1",
@@ -124,6 +124,7 @@ async def _recorded_cost(result, source):
        response=response,
        error=None,
        reasoning_applies=True,
+        operation="chat",
    )
    return recorder.rows[0]["cost"]

@@ -124,6 +124,18 @@ class _DummyRecorder:
        reasoning_tokens,
        tenant_id,
        meta,
+        thinking_observation,
+        reasoning_effort,
+        scope,
+        operation,
+        logical_call_id,
+        event_kind,
+        http_status_code,
+        error_type,
+        cause_type,
+        error_body,
+        attempts,
+        total_latency_ms,
    ) -> None: ...


@@ -275,8 +287,47 @@ class TestTelemetryRecorderSignature:
        params = inspect.signature(TelemetryRecorder.record_llm_call).parameters
        assert {"tenant_id", "meta"} <= set(params)

+    def test_call_observability_fields_are_declared(self):
+        """1.3.5 十列进协议(issue #19/#23);字段总数以实测为准不凭记忆。
+
+        本签名同时是装配闸的事实源(`_assert_recorder_shape` 按它派生参数名),
+        故它与实现一旦漂移,下游自定义 recorder 会在装配期就被拒。
+        """
+        import inspect
+
+        params = inspect.signature(TelemetryRecorder.record_llm_call).parameters
+        assert {
+            "scope",
+            "operation",
+            "logical_call_id",
+            "event_kind",
+            "http_status_code",
+            "error_type",
+            "cause_type",
+            "error_body",
+            "attempts",
+            "total_latency_ms",
+        } <= set(params)
+        assert len(params) - 1 == 36  # 减掉 self
+
    @pytest.mark.parametrize(
-        "name", ["tenant_id", "meta", "thinking_observation", "reasoning_effort"]
+        "name",
+        [
+            "tenant_id",
+            "meta",
+            "thinking_observation",
+            "reasoning_effort",
+            "scope",
+            "operation",
+            "logical_call_id",
+            "event_kind",
+            "http_status_code",
+            "error_type",
+            "cause_type",
+            "error_body",
+            "attempts",
+            "total_latency_ms",
+        ],
    )
    def test_caller_dimensions_have_no_default(self, name):
        import inspect
@@ -169,7 +169,7 @@ def _source(model="qwen-max"):
 class TestEmitterCost:
    async def test_success_row_costed(self):
        rec = _MemoryRecorder()
-        emitter = TelemetryEmitter(rec, pricing=_TABLE, text_cap=None)
+        emitter = TelemetryEmitter(rec, pricing=_TABLE, text_cap=None, scope="LLM")
        await emitter.emit_attempt(
            request=_REQ,
            source=_source(),
@@ -178,18 +178,19 @@ class TestEmitterCost:
            response=_resp(),
            error=None,
            reasoning_applies=True,
+            operation="chat",
        )
        assert rec.rows[0]["cost"] == pytest.approx(7.2)

    async def test_cache_hit_row_costs_zero(self):
        rec = _MemoryRecorder()
-        emitter = TelemetryEmitter(rec, pricing=_TABLE, text_cap=None)
-        await emitter.emit_cache_hit(request=_REQ, response=_resp(cache_hit=True))
+        emitter = TelemetryEmitter(rec, pricing=_TABLE, text_cap=None, scope="LLM")
+        await emitter.emit_cache_hit(request=_REQ, response=_resp(cache_hit=True), operation="chat")
        assert rec.rows[0]["cost"] == 0.0

    async def test_failure_row_cost_none(self):
        rec = _MemoryRecorder()
-        emitter = TelemetryEmitter(rec, pricing=_TABLE, text_cap=None)
+        emitter = TelemetryEmitter(rec, pricing=_TABLE, text_cap=None, scope="LLM")
        await emitter.emit_attempt(
            request=_REQ,
            source=_source(),
@@ -198,12 +199,13 @@ class TestEmitterCost:
            response=None,
            error="TransientError: boom",
            reasoning_applies=True,
+            operation="chat",
        )
        assert rec.rows[0]["cost"] is None

    async def test_unknown_model_none_without_blocking(self):
        rec = _MemoryRecorder()
-        emitter = TelemetryEmitter(rec, pricing=_TABLE, text_cap=None)
+        emitter = TelemetryEmitter(rec, pricing=_TABLE, text_cap=None, scope="LLM")
        await emitter.emit_attempt(
            request=_REQ,
            source=_source(model="mystery"),
@@ -212,13 +214,14 @@ class TestEmitterCost:
            response=_resp(model="mystery"),
            error=None,
            reasoning_applies=True,
+            operation="chat",
        )
        assert rec.rows[0]["cost"] is None

    async def test_no_pricing_keeps_none(self):
        """未注入价格表 = M1 现状: cost 恒 None(回归)。"""
        rec = _MemoryRecorder()
-        emitter = TelemetryEmitter(rec, text_cap=None)
+        emitter = TelemetryEmitter(rec, text_cap=None, scope="LLM")
        await emitter.emit_attempt(
            request=_REQ,
            source=_source(),
@@ -227,5 +230,6 @@ class TestEmitterCost:
            response=_resp(),
            error=None,
            reasoning_applies=True,
+            operation="chat",
        )
        assert rec.rows[0]["cost"] is None
@@ -35,6 +35,7 @@ from polygateway.types import (
    USAGE_SOURCES,
    BackpressurePolicy,
    BreakerConfig,
+    CallStats,
    ChatRequest,
    EmbeddingTransportResult,
    GlobalLimits,
@@ -44,6 +45,9 @@ from polygateway.types import (
    SourceConfig,
 )

+# 终态行的快照入参(1.3.5): `emit_terminal_failure` 不再收 `latency_ms`。
+_MIGRATED_STATS = CallStats(logical_call_id="lcid-mig", attempts=1, total_latency_ms=1)
+
 _REQ = ChatRequest(messages=[{"role": "user", "content": "hi"}])
 _DOMAIN = sorted(USAGE_SOURCES)

@@ -255,7 +259,7 @@ def _resp(usage_source):
@pytest.mark.parametrize("emitted", _DOMAIN)
 async def test_emit_attempt_success_stays_in_domain(emitted):
    recorder = _MemoryRecorder()
-    await TelemetryEmitter(recorder, text_cap=None).emit_attempt(
+    await TelemetryEmitter(recorder, text_cap=None, scope="LLM").emit_attempt(
        request=_REQ,
        source=_src(),
        call_id="cid",
@@ -263,6 +267,7 @@ async def test_emit_attempt_success_stays_in_domain(emitted):
        response=_resp(emitted),
        error=None,
        reasoning_applies=True,
+        operation="chat",
    )
    assert recorder.rows[0]["usage_source"] in USAGE_SOURCES

@@ -270,7 +275,7 @@ async def test_emit_attempt_success_stays_in_domain(emitted):
 async def test_emit_attempt_failed_attempt_stays_in_domain():
    """失败尝试无 response,`usage_source` 取 emitter 自己的字面量。"""
    recorder = _MemoryRecorder()
-    await TelemetryEmitter(recorder, text_cap=None).emit_attempt(
+    await TelemetryEmitter(recorder, text_cap=None, scope="LLM").emit_attempt(
        request=_REQ,
        source=_src(),
        call_id="cid",
@@ -278,6 +283,7 @@ async def test_emit_attempt_failed_attempt_stays_in_domain():
        response=None,
        error="boom",
        reasoning_applies=True,
+        operation="chat",
    )
    assert recorder.rows[0]["usage_source"] in USAGE_SOURCES

@@ -285,8 +291,10 @@ async def test_emit_attempt_failed_attempt_stays_in_domain():
@pytest.mark.parametrize("emitted", _DOMAIN)
 async def test_emit_cache_hit_stays_in_domain(emitted):
    recorder = _MemoryRecorder()
-    await TelemetryEmitter(recorder, text_cap=None).emit_cache_hit(
-        request=_REQ, response=_resp(emitted)
+    await TelemetryEmitter(recorder, text_cap=None, scope="LLM").emit_cache_hit(
+        request=_REQ,
+        response=_resp(emitted),
+        operation="chat",
    )
    assert recorder.rows[0]["usage_source"] in USAGE_SOURCES

@@ -294,7 +302,11 @@ async def test_emit_cache_hit_stays_in_domain(emitted):
 async def test_emit_terminal_failure_stays_in_domain():
    """终态失败无具体源,`usage_source` 同样取 emitter 字面量。"""
    recorder = _MemoryRecorder()
-    await TelemetryEmitter(recorder, text_cap=None).emit_terminal_failure(
-        request=_REQ, call_id="cid", latency_ms=10, error="cancelled"
+    await TelemetryEmitter(recorder, text_cap=None, scope="LLM").emit_terminal_failure(
+        request=_REQ,
+        call_id="cid",
+        error="cancelled",
+        operation="chat",
+        stats=_MIGRATED_STATS,
    )
    assert recorder.rows[0]["usage_source"] in USAGE_SOURCES