diff --git a/research-wiki/findings/2026-09-09-135-call-observability-validation.md b/research-wiki/findings/2026-09-09-135-call-observability-validation.md new file mode 100644 index 0000000..a0a3270 --- /dev/null +++ b/research-wiki/findings/2026-09-09-135-call-observability-validation.md @@ -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` 为准。 diff --git a/src/polygateway/client.py b/src/polygateway/client.py index 9ac1a64..4534337 100644 --- a/src/polygateway/client.py +++ b/src/polygateway/client.py @@ -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, ) - response = await self._handler(request) + 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()) diff --git a/src/polygateway/embedding.py b/src/polygateway/embedding.py index 44ac701..7b3a728 100644 --- a/src/polygateway/embedding.py +++ b/src/polygateway/embedding.py @@ -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""}] + # 与逐批行同款构造(至多 `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: diff --git a/src/polygateway/middleware/retry.py b/src/polygateway/middleware/retry.py index 0a3a1ea..7f66cd5 100644 --- a/src/polygateway/middleware/retry.py +++ b/src/polygateway/middleware/retry.py @@ -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 diff --git a/src/polygateway/middleware/structured.py b/src/polygateway/middleware/structured.py index 2a70b63..6887676 100644 --- a/src/polygateway/middleware/structured.py +++ b/src/polygateway/middleware/structured.py @@ -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: diff --git a/src/polygateway/middleware/telemetry.py b/src/polygateway/middleware/telemetry.py index 3b4c0f6..d728313 100644 --- a/src/polygateway/middleware/telemetry.py +++ b/src/polygateway/middleware/telemetry.py @@ -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 + response = await call_next(request) 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 diff --git a/src/polygateway/ocr.py b/src/polygateway/ocr.py index 683649f..501ae18 100644 --- a/src/polygateway/ocr.py +++ b/src/polygateway/ocr.py @@ -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""}], + 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: + """终态行: 沿用 `` 占位,错误文本保留类名前缀。""" + 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""}], - 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 diff --git a/src/polygateway/ports.py b/src/polygateway/ports.py index ed5b20b..a8397d8 100644 --- a/src/polygateway/ports.py +++ b/src/polygateway/ports.py @@ -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: ... diff --git a/src/polygateway/telemetry/schema.py b/src/polygateway/telemetry/schema.py index 46ada8d..1186e36 100644 --- a/src/polygateway/telemetry/schema.py +++ b/src/polygateway/telemetry/schema.py @@ -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) diff --git a/src/polygateway/transports/openai_compat.py b/src/polygateway/transports/openai_compat.py index a76673a..55f79ea 100644 --- a/src/polygateway/transports/openai_compat.py +++ b/src/polygateway/transports/openai_compat.py @@ -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: diff --git a/tests/unit/test_cache.py b/tests/unit/test_cache.py index 6e86044..113e6dd 100644 --- a/tests/unit/test_cache.py +++ b/tests/unit/test_cache.py @@ -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"]) diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index bc77f2c..4213097 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -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 diff --git a/tests/unit/test_embedding.py b/tests/unit/test_embedding.py index 3384a75..f6ad5a3 100644 --- a/tests/unit/test_embedding.py +++ b/tests/unit/test_embedding.py @@ -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): diff --git a/tests/unit/test_ocr_client.py b/tests/unit/test_ocr_client.py index acac9c2..d4f03d0 100644 --- a/tests/unit/test_ocr_client.py +++ b/tests/unit/test_ocr_client.py @@ -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) diff --git a/tests/unit/test_openai_compat.py b/tests/unit/test_openai_compat.py index f54f796..b65f0c4 100644 --- a/tests/unit/test_openai_compat.py +++ b/tests/unit/test_openai_compat.py @@ -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"] diff --git a/tests/unit/test_ports.py b/tests/unit/test_ports.py index f772271..57beecc 100644 --- a/tests/unit/test_ports.py +++ b/tests/unit/test_ports.py @@ -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 diff --git a/tests/unit/test_pricing.py b/tests/unit/test_pricing.py index de031a1..ee6690d 100644 --- a/tests/unit/test_pricing.py +++ b/tests/unit/test_pricing.py @@ -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 diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index 49aacca..75999d7 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -8,13 +8,20 @@ import sqlite3 import subprocess from pathlib import Path +import httpx import pytest from loguru import logger from polygateway.backends.memory.breaker import InMemoryGate from polygateway.backends.memory.limiter import InMemoryLimiter from polygateway.embedding import EmbeddingClient -from polygateway.errors import CircuitOpenError, RequestRejectedError +from polygateway.errors import ( + AllSourcesExhausted, + CircuitOpenError, + RequestRejectedError, + ResultInvalidError, + TransientError, +) from polygateway.middleware.cache import digest_messages from polygateway.middleware.telemetry import TelemetryEmitter, TelemetryMW from polygateway.ocr import OcrClient @@ -24,6 +31,7 @@ from polygateway.telemetry.sqlite import SQLiteRecorder from polygateway.types import ( BackpressurePolicy, BreakerConfig, + CallStats, ChatRequest, Effort, EmbeddingTransportResult, @@ -37,6 +45,11 @@ from polygateway.types import ( _REQ = ChatRequest(messages=[{"role": "user", "content": "hi"}], session_id="sess-1") +# 终态行的快照入参(1.3.5): `emit_terminal_failure` 不再收 `latency_ms`, +# `latency_ms` 与 `total_latency_ms` 同取这一份冻结快照,避免双时钟微差。 +# 取 1ms 是为与迁移前逐字写死的 `latency_ms=1` 保持同值。 +_MIGRATED_STATS = CallStats(logical_call_id="lcid-mig", attempts=1, total_latency_ms=1) + _EXPECTED_COLUMNS = [ "call_id", "parent_call_id", @@ -65,8 +78,26 @@ _EXPECTED_COLUMNS = [ "meta", "thinking_observation", "reasoning_effort", + # —— 1.3.5 逻辑调用统计与结构化失败诊断的十列 —— + "scope", + "operation", + "logical_call_id", + "event_kind", + "http_status_code", + "error_type", + "cause_type", + "error_body", + "attempts", + "total_latency_ms", ] +# 1.3.5 之前的 26 个 INSERT 字段(旧表形态),供冻结 INSERT 语句与旧表补列用例复用 +_PRE_135_COLUMNS = [c for c in _EXPECTED_COLUMNS if c != "created_at"][:26] + +# `_PRE_TENANT_DDL` 那张旧表的 23 个物理列(22 个 INSERT 字段 + created_at)。 +# 写成固定切片而非 `[:-N]`: 后者会随每一次补列静默漂移到另一张表上去。 +_PRE_TENANT_PHYSICAL = _EXPECTED_COLUMNS[:23] + def _resp(**overrides): base = { @@ -136,6 +167,20 @@ async def _record_minimal(recorder, call_id="c1", **overrides): "thinking_observation": "unknown", # 同理: `Effort` 归一成裸 str,不表态则是 None(与 'low' 必须分得开) "reasoning_effort": None, + # —— 1.3.5 十列: 默认形态即"一次普通尝试行"—— + "scope": "LLM", + "operation": "chat", + # 库内现场构造的请求没有上下文 → NULL,不造 ID + "logical_call_id": None, + "event_kind": "attempt", + # 诊断四列只在失败的 attempt 行上非空;成功行不统一填 200 + "http_status_code": None, + "error_type": None, + "cause_type": None, + "error_body": None, + # 逻辑快照两列只属终态行 + "attempts": None, + "total_latency_ms": None, } fields.update(overrides) await recorder.record_llm_call(**fields) @@ -217,7 +262,7 @@ class TestSchemaModule: # COLUMNS 是 INSERT 字段序,不含数据库自填的 created_at assert list(COLUMNS) == [c for c in _EXPECTED_COLUMNS if c != "created_at"] - assert len(COLUMNS) == 26 + assert len(COLUMNS) == 36 # 两端 DDL 的列出现顺序 == 物理列序(created_at 在第 19 位) for ddl in (SQLITE_DDL, PG_DDL): assert _first_occurrence_order(ddl, _EXPECTED_COLUMNS) == _EXPECTED_COLUMNS @@ -231,17 +276,25 @@ class TestSchemaModule: "ALTER TABLE llm_calls ADD COLUMN cached_prompt_tokens INTEGER", ) assert PG_BACKFILL[-1] == ( - "reasoning_effort", - "ALTER TABLE llm_calls ADD COLUMN reasoning_effort TEXT", + "total_latency_ms", + "ALTER TABLE llm_calls ADD COLUMN total_latency_ms INTEGER", ) assert all("IF NOT EXISTS" not in stmt for _, stmt in PG_BACKFILL) def test_insert_sql_reproduces_the_frozen_statements(self): - """`insert_sql(backend, COLUMNS)` 与搬迁前的 `_INSERT` 一致(PG 侧去掉冲突目标)。""" + """搬迁前(1.2.1)的 26 字段 `_INSERT` 逐字可重现(PG 侧去掉冲突目标)。 + + 冻结串是"纯搬迁不改行为"的机械证据,故仍按**当时那 26 列**构造; + 1.3.5 补列后的全量语句另由下一条用例按占位符个数断言。 + """ from polygateway.telemetry.schema import COLUMNS, insert_sql - assert insert_sql("sqlite", COLUMNS) == _FROZEN_SQLITE_INSERT - assert insert_sql("postgres", COLUMNS) == _FROZEN_PG_INSERT + assert insert_sql("sqlite", _PRE_135_COLUMNS) == _FROZEN_SQLITE_INSERT + assert insert_sql("postgres", _PRE_135_COLUMNS) == _FROZEN_PG_INSERT + # 全量 36 列: 占位符随列数增长,且不留空洞 + assert insert_sql("postgres", COLUMNS).count("$") == 36 + assert "$36)" in insert_sql("postgres", COLUMNS) + assert insert_sql("sqlite", COLUMNS).count("?") == 36 # 裁剪列表按位置占位符重新编号,不留空洞 assert insert_sql("postgres", ["call_id", "model"]) == ( "INSERT INTO llm_calls (call_id, model) VALUES ($1, $2) ON CONFLICT DO NOTHING" @@ -318,7 +371,7 @@ class TestBackendColumnParity: """新列只能追加在末尾: 旧表经 ALTER 补列必落末尾,插在中间会让两条路径分叉。""" from polygateway.telemetry.schema import COLUMNS - assert COLUMNS[-4:] == ("tenant_id", "meta", "thinking_observation", "reasoning_effort") + assert COLUMNS[-4:] == ("cause_type", "error_body", "attempts", "total_latency_ms") class TestSQLiteRecorder: @@ -594,7 +647,7 @@ class TestSQLiteCallerDimensionsAcceptance: conn = sqlite3.connect(db) cols = [r[1] for r in conn.execute("PRAGMA table_info(llm_calls)")] - assert cols == _EXPECTED_COLUMNS # 22 → 26 个 recorder 字段(+ created_at 共 27 物理列) + assert cols == _EXPECTED_COLUMNS # 22 → 36 个 recorder 字段(+ created_at 共 37 物理列) rows = dict(conn.execute("SELECT call_id, tenant_id FROM llm_calls").fetchall()) assert rows["new-row"] == "tenant-a" assert rows["old-row"] == "" # 不是 None: NULL 会被 RLS 静默吞掉 @@ -636,7 +689,7 @@ class TestSQLiteCallerDimensionsAcceptance: stale = sqlite3.connect(db) assert [r[1] for r in stale.execute("PRAGMA table_info(llm_calls)")] == ( - _EXPECTED_COLUMNS[:-4] + _PRE_TENANT_PHYSICAL ) # 补列确实没成功,用例不是在只读库上空转 @@ -644,7 +697,7 @@ class TestSQLiteSchemaMode: """issue #13: `auto_migrate` 两档——auto 保持自动补列,manual 只裁剪写入不发 DDL。 列数断言一律按**物理列数**写: 旧表 22 个 INSERT 字段 + `created_at` = 23, - 补齐后 26 + `created_at` = 27。混用 INSERT 字段数与物理列数是本处最易错的地方。 + 补齐后 36 + `created_at` = 37。混用 INSERT 字段数与物理列数是本处最易错的地方。 """ def _physical_columns(self, db: Path) -> list[str]: @@ -683,7 +736,7 @@ class TestSQLiteSchemaMode: assert "ALTER TABLE" in message # 给出可直接执行的补列 SQL async def test_auto_mode_still_upgrades_the_legacy_table(self, tmp_path): - """auto + 同款旧表: 现状回归,补列后物理列数 23 → 27。""" + """auto + 同款旧表: 现状回归,补列后物理列数 23 → 37。""" db = tmp_path / "auto_legacy.db" _make_pre_tenant_db(db) @@ -692,10 +745,10 @@ class TestSQLiteSchemaMode: recorder.close() assert self._physical_columns(db) == _EXPECTED_COLUMNS - assert len(self._physical_columns(db)) == 27 + assert len(self._physical_columns(db)) == 37 async def test_manual_mode_still_creates_a_fresh_table(self, tmp_path): - """manual 只管 ALTER,不管 CREATE: 全新库照建,27 个物理列齐全(设计 §4.2)。""" + """manual 只管 ALTER,不管 CREATE: 全新库照建,37 个物理列齐全(设计 §4.2)。""" db = tmp_path / "manual_fresh.db" recorder = SQLiteRecorder(db, auto_migrate=False) await _record_minimal(recorder, call_id="c-fresh", tenant_id="tenant-a") @@ -921,6 +974,17 @@ class TestPostgresBackfillDiscipline: "meta", "thinking_observation", "reasoning_effort", + # 1.3.5 十列: 稳态的定义随补列一起前移,否则本用例会把"每进程首写抢锁"放行 + "scope", + "operation", + "logical_call_id", + "event_kind", + "http_status_code", + "error_type", + "cause_type", + "error_body", + "attempts", + "total_latency_ms", ] def _recorder(self, conn): @@ -1140,7 +1204,7 @@ class TestEmitterRecorderContract: from polygateway.telemetry.schema import COLUMNS rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(), call_id="cid-1", @@ -1148,6 +1212,7 @@ class TestEmitterRecorderContract: response=_resp(), error=None, reasoning_applies=True, + operation="chat", ) assert set(rec.rows[0]) == set(COLUMNS) @@ -1156,7 +1221,7 @@ class TestEmitterRecorderContract: from polygateway.telemetry.schema import COLUMNS rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, text_cap=None) + emitter = TelemetryEmitter(rec, text_cap=None, scope="LLM") if emit == "attempt": await emitter.emit_attempt( request=_REQ, @@ -1166,12 +1231,17 @@ class TestEmitterRecorderContract: response=None, error="boom", reasoning_applies=True, + operation="chat", ) elif emit == "cache_hit": - await emitter.emit_cache_hit(request=_REQ, response=_resp()) + await emitter.emit_cache_hit(request=_REQ, response=_resp(), operation="chat") else: await emitter.emit_terminal_failure( - request=_REQ, call_id="c", latency_ms=1, error="dead" + request=_REQ, + call_id="c", + error="dead", + operation="chat", + stats=_MIGRATED_STATS, ) assert set(rec.rows[0]) == set(COLUMNS) @@ -1187,7 +1257,7 @@ class TestEmitterThinkingObservation: async def test_attempt_carries_the_verdict_as_a_plain_string(self): rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(), call_id="c", @@ -1195,6 +1265,7 @@ class TestEmitterThinkingObservation: response=_resp(thinking_observation=ThinkingObservation.OBSERVED), error=None, reasoning_applies=True, + operation="chat", ) value = rec.rows[0]["thinking_observation"] assert value == "observed" @@ -1210,7 +1281,7 @@ class TestEmitterThinkingObservation: 是整行,正是 1.3.0 那次"19 次调用一行未落"的同款形态。 """ rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(), call_id="c", @@ -1218,6 +1289,7 @@ class TestEmitterThinkingObservation: response=_resp(thinking_observation="observed"), error=None, reasoning_applies=True, + operation="chat", ) assert len(rec.rows) == 1, "整行被吞了" value = rec.rows[0]["thinking_observation"] @@ -1236,7 +1308,7 @@ class TestEmitterThinkingObservation: messages: list[str] = [] sink_id = logger.add(messages.append, level="WARNING") try: - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(), call_id="c", @@ -1244,6 +1316,7 @@ class TestEmitterThinkingObservation: response=_resp(thinking_observation="OBSERVED"), # 大小写不符即域外 error=None, reasoning_applies=True, + operation="chat", ) finally: logger.remove(sink_id) @@ -1256,17 +1329,22 @@ class TestEmitterThinkingObservation: async def test_cache_hit_replays_the_recorded_verdict(self): """缓存命中回放历史那次的裁定: 与 model/prompt_tokens 同一口径。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_cache_hit( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_cache_hit( request=_REQ, response=_resp(cache_hit=True, thinking_observation=ThinkingObservation.ABSENT), + operation="chat", ) assert rec.rows[0]["thinking_observation"] == "absent" async def test_terminal_failure_records_unknown(self): """终态失败无响应可言,记 `unknown`——它恰好就是"观测不到",不撒谎。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure( - request=_REQ, call_id="c", latency_ms=1, error="dead" + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_terminal_failure( + request=_REQ, + call_id="c", + error="dead", + operation="chat", + stats=_MIGRATED_STATS, ) value = rec.rows[0]["thinking_observation"] assert value == "unknown" @@ -1275,7 +1353,7 @@ class TestEmitterThinkingObservation: async def test_failed_attempt_records_unknown(self): """失败尝试(response=None)同理: 默认视图即 UNKNOWN。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(), call_id="c", @@ -1283,6 +1361,7 @@ class TestEmitterThinkingObservation: response=None, error="boom", reasoning_applies=True, + operation="chat", ) assert rec.rows[0]["thinking_observation"] == "unknown" @@ -1303,7 +1382,7 @@ class TestEmitterReasoningEffort: 映射的源上恒等——本地跑不开映射的源永远看不出这个错。 """ rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=ChatRequest( messages=[{"role": "user", "content": "hi"}], reasoning_effort=Effort.MEDIUM ), @@ -1313,6 +1392,7 @@ class TestEmitterReasoningEffort: response=_resp(applied_effort=Effort.LOW), error=None, reasoning_applies=True, + operation="chat", ) value = rec.rows[0]["reasoning_effort"] assert value == "low" # 不是 medium: 那一档从未发出去过 @@ -1326,7 +1406,7 @@ class TestEmitterReasoningEffort: "哪一档配错了" 是有用信号,不该被过滤掉。 """ rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(reasoning_effort=Effort.HIGH), call_id="c", @@ -1334,13 +1414,14 @@ class TestEmitterReasoningEffort: response=None, error="boom", reasoning_applies=True, + operation="chat", ) assert rec.rows[0]["reasoning_effort"] == "high" async def test_failed_attempt_resolves_the_syntactic_sugar_too(self): """回落走 `effective_effort` 而非裸读字段: `enable_thinking` 也是表态。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(enable_thinking=True), call_id="c", @@ -1348,30 +1429,33 @@ class TestEmitterReasoningEffort: response=None, error="boom", reasoning_applies=True, + operation="chat", ) assert rec.rows[0]["reasoning_effort"] == "auto" async def test_cache_hit_records_the_request_tier_not_the_replayed_one(self): """命中行没有选中源,故记请求档;与 model/prompt_tokens 的回放口径相反。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_cache_hit( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_cache_hit( request=ChatRequest( messages=[{"role": "user", "content": "hi"}], reasoning_effort=Effort.MEDIUM ), response=_resp(cache_hit=True, applied_effort=Effort.LOW), + operation="chat", ) assert rec.rows[0]["reasoning_effort"] == "medium" async def test_terminal_failure_records_the_request_tier(self): """终态失败可能根本没选出源,源级档位无从谈起。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_terminal_failure( request=ChatRequest( messages=[{"role": "user", "content": "hi"}], reasoning_effort=Effort.XHIGH ), call_id="c", - latency_ms=1, error="dead", + operation="chat", + stats=_MIGRATED_STATS, ) value = rec.rows[0]["reasoning_effort"] assert value == "xhigh" @@ -1384,7 +1468,7 @@ class TestEmitterReasoningEffort: 库并不观测模型内部的默认档,记一个推定值等于把"没看见"说成"发生了"。 """ rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, text_cap=None) + emitter = TelemetryEmitter(rec, text_cap=None, scope="LLM") if emit == "attempt": await emitter.emit_attempt( request=_REQ, @@ -1394,12 +1478,19 @@ class TestEmitterReasoningEffort: response=_resp(), error=None, reasoning_applies=True, + operation="chat", ) elif emit == "cache_hit": - await emitter.emit_cache_hit(request=_REQ, response=_resp(cache_hit=True)) + await emitter.emit_cache_hit( + request=_REQ, response=_resp(cache_hit=True), operation="chat" + ) else: await emitter.emit_terminal_failure( - request=_REQ, call_id="c", latency_ms=1, error="dead" + request=_REQ, + call_id="c", + error="dead", + operation="chat", + stats=_MIGRATED_STATS, ) assert rec.rows[0]["reasoning_effort"] is None @@ -1410,7 +1501,7 @@ class TestEmitterReasoningEffort: 挂上 `auto` ——那一档从来没有、也不可能被发出去。 """ rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(enable_thinking=True), call_id="c", @@ -1418,6 +1509,7 @@ class TestEmitterReasoningEffort: response=None, error="boom", reasoning_applies=False, + operation="chat", ) assert rec.rows[0]["reasoning_effort"] is None @@ -1446,7 +1538,7 @@ class TestEmitterReasoningEffort: messages: list[str] = [] sink_id = logger.add(messages.append, level="WARNING") try: - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(), call_id="c", @@ -1454,6 +1546,7 @@ class TestEmitterReasoningEffort: response=_resp(applied_effort="lowest"), error=None, reasoning_applies=True, + operation="chat", ) finally: logger.remove(sink_id) @@ -1466,7 +1559,7 @@ class TestEmitterReasoningEffort: async def test_a_bare_string_tier_still_lands(self): """裸串在域内时照常归一并落库,整行不得丢失。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(), call_id="c", @@ -1474,6 +1567,7 @@ class TestEmitterReasoningEffort: response=_resp(applied_effort="max"), error=None, reasoning_applies=True, + operation="chat", ) assert len(rec.rows) == 1, "整行被吞了" value = rec.rows[0]["reasoning_effort"] @@ -1486,7 +1580,7 @@ class TestEmitterObservabilityFields: async def test_attempt_carries_the_response_values(self): rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(), call_id="cid-1", @@ -1494,6 +1588,7 @@ class TestEmitterObservabilityFields: response=_resp(cached_prompt_tokens=64, model_reported="m-real", reasoning_tokens=7), error=None, reasoning_applies=True, + operation="chat", ) assert rec.rows[0]["cached_prompt_tokens"] == 64 assert rec.rows[0]["model_reported"] == "m-real" @@ -1501,7 +1596,7 @@ class TestEmitterObservabilityFields: async def test_failed_attempt_has_no_provider_facts(self): rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(), call_id="cid-2", @@ -1509,6 +1604,7 @@ class TestEmitterObservabilityFields: response=None, error="boom", reasoning_applies=True, + operation="chat", ) assert rec.rows[0]["cached_prompt_tokens"] is None assert rec.rows[0]["model_reported"] is None @@ -1517,9 +1613,10 @@ class TestEmitterObservabilityFields: async def test_cache_hit_replays_the_recorded_values(self): """决策 B1: 命中行原样回放,故命中率统计必须带 WHERE cache_hit = false。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_cache_hit( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_cache_hit( request=_REQ, response=_resp(cached_prompt_tokens=64, model_reported="m-real", reasoning_tokens=7), + operation="chat", ) row = rec.rows[0] assert row["cache_hit"] is True @@ -1528,8 +1625,12 @@ class TestEmitterObservabilityFields: async def test_terminal_failure_records_none(self): rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure( - request=_REQ, call_id="c", latency_ms=1, error="dead" + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_terminal_failure( + request=_REQ, + call_id="c", + error="dead", + operation="chat", + stats=_MIGRATED_STATS, ) assert rec.rows[0]["cached_prompt_tokens"] is None assert rec.rows[0]["model_reported"] is None @@ -1552,7 +1653,7 @@ class TestEmitterSamplingColumn: async def test_attempt_merges_source_extra_body(self): rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=self._SAMPLED, source=_source(extra_body={"temperature": 0}), call_id="c", @@ -1560,13 +1661,14 @@ class TestEmitterSamplingColumn: response=_resp(), error=None, reasoning_applies=True, + operation="chat", ) assert json.loads(rec.rows[0]["sampling"]) == {"seed": 42, "temperature": 0} async def test_response_format_never_leaks_into_the_column(self): """三行都不得出现 response_format——它不是采样参数。""" rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, text_cap=None) + emitter = TelemetryEmitter(rec, text_cap=None, scope="LLM") await emitter.emit_attempt( request=self._SAMPLED, source=_source(), @@ -1575,10 +1677,15 @@ class TestEmitterSamplingColumn: response=_resp(), error=None, reasoning_applies=True, + operation="chat", ) - await emitter.emit_cache_hit(request=self._SAMPLED, response=_resp()) + await emitter.emit_cache_hit(request=self._SAMPLED, response=_resp(), operation="chat") await emitter.emit_terminal_failure( - request=self._SAMPLED, call_id="c", latency_ms=1, error="dead" + request=self._SAMPLED, + call_id="c", + error="dead", + operation="chat", + stats=_MIGRATED_STATS, ) assert len(rec.rows) == 3 for row in rec.rows: @@ -1588,19 +1695,23 @@ class TestEmitterSamplingColumn: async def test_sourceless_entries_record_call_level_only(self, emit): """两个最外层入口没有"生效源"可言,与 model/source_name 置空同一先例。""" rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, text_cap=None) + emitter = TelemetryEmitter(rec, text_cap=None, scope="LLM") if emit == "cache_hit": - await emitter.emit_cache_hit(request=self._SAMPLED, response=_resp()) + await emitter.emit_cache_hit(request=self._SAMPLED, response=_resp(), operation="chat") else: await emitter.emit_terminal_failure( - request=self._SAMPLED, call_id="c", latency_ms=1, error="dead" + request=self._SAMPLED, + call_id="c", + error="dead", + operation="chat", + stats=_MIGRATED_STATS, ) assert json.loads(rec.rows[0]["sampling"]) == {"seed": 42} async def test_absent_sampling_is_null(self): """无采样参数时为 NULL,而非空字符串或 "{}"——便于 SQL 过滤。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(), call_id="c", @@ -1608,6 +1719,7 @@ class TestEmitterSamplingColumn: response=_resp(), error=None, reasoning_applies=True, + operation="chat", ) assert rec.rows[0]["sampling"] is None @@ -1631,7 +1743,7 @@ class TestEmitterCallerDimensions: async def test_every_entry_point_carries_the_dimensions(self, emit): """三条路径写出的行都必须带维度: 漏掉任一条,该租户的账就永远对不上。""" rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, text_cap=None) + emitter = TelemetryEmitter(rec, text_cap=None, scope="LLM") if emit == "attempt": await emitter.emit_attempt( request=self._REQ_A, @@ -1641,12 +1753,19 @@ class TestEmitterCallerDimensions: response=_resp(), error=None, reasoning_applies=True, + operation="chat", ) elif emit == "cache_hit": - await emitter.emit_cache_hit(request=self._REQ_A, response=_resp(cache_hit=True)) + await emitter.emit_cache_hit( + request=self._REQ_A, response=_resp(cache_hit=True), operation="chat" + ) else: await emitter.emit_terminal_failure( - request=self._REQ_A, call_id="c", latency_ms=1, error="dead" + request=self._REQ_A, + call_id="c", + error="dead", + operation="chat", + stats=_MIGRATED_STATS, ) row = rec.rows[0] assert row["tenant_id"] == "tenant-a" @@ -1665,7 +1784,7 @@ class TestEmitterCallerDimensions: meta={"batch": "old-batch"}, ) rec = _MemoryRecorder() - mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None)) + mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None, scope="LLM")) async def terminal(request): # 缓存层回放的是历史那次的响应对象(其 call_id 属于 historical 那次) @@ -1687,7 +1806,7 @@ class TestEmitterCallerDimensions: JSON 函数直接查询,NULL 则要每条查询都额外判空。 """ rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_attempt( request=_REQ, # tenant_id=None, meta={} source=_source(), call_id="c", @@ -1695,6 +1814,7 @@ class TestEmitterCallerDimensions: response=_resp(), error=None, reasoning_applies=True, + operation="chat", ) row = rec.rows[0] assert row["tenant_id"] == "" @@ -1703,8 +1823,12 @@ class TestEmitterCallerDimensions: async def test_meta_is_serialized_with_sorted_keys(self): """键序固定,同一份维度在任意两行里字节一致,可直接做等值比对与去重。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure( - request=self._REQ_A, call_id="c", latency_ms=1, error="dead" + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_terminal_failure( + request=self._REQ_A, + call_id="c", + error="dead", + operation="chat", + stats=_MIGRATED_STATS, ) assert list(json.loads(rec.rows[0]["meta"])) == ["a_first", "m_mid", "z_last"] @@ -1712,8 +1836,12 @@ class TestEmitterCallerDimensions: """`ensure_ascii=False`: 中文维度按原文落库,而非 `\\uXXXX` 转义串。""" rec = _MemoryRecorder() req = ChatRequest(messages=[{"role": "user", "content": "hi"}], meta={"dept": "研发"}) - await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure( - request=req, call_id="c", latency_ms=1, error="dead" + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_terminal_failure( + request=req, + call_id="c", + error="dead", + operation="chat", + stats=_MIGRATED_STATS, ) assert "研发" in rec.rows[0]["meta"] @@ -1731,8 +1859,12 @@ class TestEmitterCallerDimensions: """ rec = _MemoryRecorder() req = ChatRequest(messages=[{"role": "user", "content": "hi"}], meta={"k": float("nan")}) - await TelemetryEmitter(rec, text_cap=None).emit_terminal_failure( - request=req, call_id="c", latency_ms=1, error="dead" + await TelemetryEmitter(rec, text_cap=None, scope="LLM").emit_terminal_failure( + request=req, + call_id="c", + error="dead", + operation="chat", + stats=_MIGRATED_STATS, ) assert rec.rows == [] @@ -1746,7 +1878,7 @@ class TestCostWithCachedTier: async def test_cached_hit_lowers_the_recorded_cost(self): rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, pricing=self._TABLE, text_cap=None) + emitter = TelemetryEmitter(rec, pricing=self._TABLE, text_cap=None, scope="LLM") full = _resp(prompt_tokens=1_000_000, completion_tokens=0) await emitter.emit_attempt( request=_REQ, @@ -1756,6 +1888,7 @@ class TestCostWithCachedTier: response=full, error=None, reasoning_applies=True, + operation="chat", ) await emitter.emit_attempt( request=_REQ, @@ -1767,6 +1900,7 @@ class TestCostWithCachedTier: ), error=None, reasoning_applies=True, + operation="chat", ) assert rec.rows[0]["cost"] == pytest.approx(10.0) assert rec.rows[1]["cost"] == pytest.approx(5.2) # 400k×10 + 600k×2 @@ -1774,15 +1908,16 @@ class TestCostWithCachedTier: async def test_cache_hit_row_still_costs_zero(self): """缓存命中未产生新调用 → cost 恒 0.0,该短路必须排在任何换算之前。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, pricing=self._TABLE, text_cap=None).emit_cache_hit( + await TelemetryEmitter(rec, pricing=self._TABLE, text_cap=None, scope="LLM").emit_cache_hit( request=_REQ, response=_resp(prompt_tokens=1_000_000, cached_prompt_tokens=600_000), + operation="chat", ) assert rec.rows[0]["cost"] == 0.0 async def test_unavailable_usage_still_costs_none(self): rec = _MemoryRecorder() - await TelemetryEmitter(rec, pricing=self._TABLE, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, pricing=self._TABLE, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(), call_id="c", @@ -1790,6 +1925,7 @@ class TestCostWithCachedTier: response=_resp(usage_source="unavailable", cached_prompt_tokens=5), error=None, reasoning_applies=True, + operation="chat", ) assert rec.rows[0]["cost"] is None @@ -1797,7 +1933,7 @@ class TestCostWithCachedTier: class TestEmitter: async def test_attempt_success_row(self): rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, text_cap=None) + emitter = TelemetryEmitter(rec, text_cap=None, scope="LLM") await emitter.emit_attempt( request=_REQ, source=_source(), @@ -1806,6 +1942,7 @@ class TestEmitter: response=_resp(), error=None, reasoning_applies=True, + operation="chat", ) row = rec.rows[0] assert row["call_id"] == "cid-1" and row["error"] is None @@ -1814,7 +1951,7 @@ class TestEmitter: async def test_attempt_failure_row(self): rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, text_cap=None) + emitter = TelemetryEmitter(rec, text_cap=None, scope="LLM") await emitter.emit_attempt( request=_REQ, source=_source(), @@ -1823,6 +1960,7 @@ class TestEmitter: response=None, error="TransientError: boom", reasoning_applies=True, + operation="chat", ) row = rec.rows[0] assert row["error"].startswith("TransientError") @@ -1832,8 +1970,14 @@ class TestEmitter: async def test_terminal_failure_row_is_unavailable(self): rec = _MemoryRecorder() - await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None).emit_terminal_failure( - request=_REQ, call_id="cid-t", latency_ms=5, error="cancelled" + await TelemetryEmitter( + rec, pricing=_PRICING, text_cap=None, scope="LLM" + ).emit_terminal_failure( + request=_REQ, + call_id="cid-t", + error="cancelled", + operation="chat", + stats=_MIGRATED_STATS, ) row = rec.rows[0] assert row["usage_source"] == "unavailable" and row["cost"] is None @@ -1845,7 +1989,7 @@ class TestEmitter: 参数第二组是改前兜底写出的 `0/4000` 形态: 那时换算出 0.032 的假金额。 """ rec = _MemoryRecorder() - await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(), call_id="cid-u", @@ -1855,13 +1999,14 @@ class TestEmitter: ), error=None, reasoning_applies=True, + operation="chat", ) assert rec.rows[0]["cost"] is None async def test_measured_row_still_priced(self): """对照组: 同一价格表下 measured 行照常换算,证明 None 不是价格表没接上。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None).emit_attempt( + await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None, scope="LLM").emit_attempt( request=_REQ, source=_source(), call_id="cid-m", @@ -1869,21 +2014,23 @@ class TestEmitter: response=_resp(prompt_tokens=0, completion_tokens=4000), error=None, reasoning_applies=True, + operation="chat", ) assert rec.rows[0]["cost"] == pytest.approx(0.032) async def test_cache_hit_keeps_zero_cost_even_when_unavailable(self): """缓存命中未产生新调用,0.0 是事实而非未知 → 短路必须排在 cache_hit 之后。""" rec = _MemoryRecorder() - await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None).emit_cache_hit( + await TelemetryEmitter(rec, pricing=_PRICING, text_cap=None, scope="LLM").emit_cache_hit( request=_REQ, response=_resp(cache_hit=True, usage_source="unavailable", completion_tokens=4000), + operation="chat", ) assert rec.rows[0]["cache_hit"] is True and rec.rows[0]["cost"] == 0.0 async def test_multimodal_messages_digested_before_storage(self): rec = _MemoryRecorder() - emitter = TelemetryEmitter(rec, text_cap=None) + emitter = TelemetryEmitter(rec, text_cap=None, scope="LLM") big = "data:image/png;base64," + "A" * 100_000 req = ChatRequest( messages=[ @@ -1903,6 +2050,7 @@ class TestEmitter: response=None, error="x", reasoning_applies=True, + operation="chat", ) assert len(rec.rows[0]["messages"]) < 500 # base64 不整段进库(VT R12) @@ -1911,7 +2059,7 @@ class TestEmitter: async def record_llm_call(self, **fields): raise OSError("disk full") - emitter = TelemetryEmitter(Broken(), text_cap=None) + emitter = TelemetryEmitter(Broken(), text_cap=None, scope="LLM") await emitter.emit_attempt( request=_REQ, source=_source(), @@ -1920,13 +2068,14 @@ class TestEmitter: response=_resp(), error=None, reasoning_applies=True, + operation="chat", ) # 不抛(降级不冒泡) class TestTelemetryMW: async def test_cache_hit_recorded(self): rec = _MemoryRecorder() - mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None)) + mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None, scope="LLM")) async def terminal(request): return _resp(cache_hit=True, latency_ms=0, call_id="cache-cid") @@ -1935,11 +2084,14 @@ class TestTelemetryMW: assert resp.cache_hit assert len(rec.rows) == 1 assert rec.rows[0]["cache_hit"] is True and rec.rows[0]["latency_ms"] == 0 + # 本层剩下的唯一职责就是这类行,故行形态在此钉死 + assert rec.rows[0]["event_kind"] == "cache_hit" + assert rec.rows[0]["operation"] == "chat" async def test_normal_success_not_double_recorded(self): """成功尝试由 RetryMW 逐次记录;最外层不得重复记。""" rec = _MemoryRecorder() - mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None)) + mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None, scope="LLM")) async def terminal(request): return _resp(cache_hit=False) @@ -1947,21 +2099,39 @@ class TestTelemetryMW: await mw(_REQ, terminal) assert rec.rows == [] - async def test_scope_level_failure_recorded(self): + async def test_scope_level_failure_is_not_written_here_anymore(self): + """1.3.5: 终态行改由 `GatewayClient.chat` 的公开边界写,本层不再写。 + + 两处同时写会让同一次失败出两条 `terminal_failure` 行,而下游正是按 + `WHERE event_kind = 'terminal_failure'` 计失败调用次数的——双写即双计。 + 异常本身仍须原样上抛。 + """ rec = _MemoryRecorder() - mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None)) + mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None, scope="LLM")) async def terminal(request): raise CircuitOpenError(scope="llm", retry_after_s=30.0) with pytest.raises(CircuitOpenError): await mw(_REQ, terminal) - assert len(rec.rows) == 1 and "circuit_open" in rec.rows[0]["error"] + assert rec.rows == [] + + async def test_cancellation_is_not_written_here_anymore(self): + """取消的终态行同样归公开边界;本层只负责不吞取消。""" + rec = _MemoryRecorder() + mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None, scope="LLM")) + + async def terminal(request): + raise asyncio.CancelledError + + with pytest.raises(asyncio.CancelledError): + await mw(_REQ, terminal) + assert rec.rows == [] async def test_attempt_level_failure_not_double_recorded(self): """RequestRejected 已被 RetryMW 逐次记录 → 最外层跳过。""" rec = _MemoryRecorder() - mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None)) + mw = TelemetryMW(TelemetryEmitter(rec, text_cap=None, scope="LLM")) async def terminal(request): raise RequestRejectedError("400") @@ -2009,7 +2179,7 @@ def _long_messages(): async def _emit_with_cap(messages, *, cap, response=_LONG, thinking=_LONG): rec = _MemoryRecorder() - await TelemetryEmitter(rec, text_cap=cap).emit_attempt( + await TelemetryEmitter(rec, text_cap=cap, scope="LLM").emit_attempt( request=ChatRequest(messages=messages, session_id="s"), source=_source(), call_id="c", @@ -2017,6 +2187,7 @@ async def _emit_with_cap(messages, *, cap, response=_LONG, thinking=_LONG): response=_resp(content=response, thinking=thinking), error=None, reasoning_applies=True, + operation="chat", ) return rec.rows[0] @@ -2091,7 +2262,7 @@ class TestTelemetryTextCap: """ for bad in (0, -1): with pytest.raises(ValueError, match="text_cap"): - TelemetryEmitter(_MemoryRecorder(), text_cap=bad) + TelemetryEmitter(_MemoryRecorder(), text_cap=bad, scope="LLM") class _StubEmbedTransport: @@ -2926,3 +3097,360 @@ class TestConcurrentReasoningPathContracts: assert attempts[0]["error"] finally: await client._transport.aclose() + + +# —— 1.3.5: 行级归因(scope/operation/event_kind)、诊断保真与终态行 —— + +_STATS = CallStats(logical_call_id="lcid-1", attempts=3, total_latency_ms=1234) + + +def _ctx_request(**overrides): + """带逻辑调用上下文的请求(库外部现场构造的请求恒无上下文)。""" + from polygateway.types import _CallContext + + base = {"messages": [{"role": "user", "content": "hi"}], "session_id": "sess-1"} + base.update(overrides) + return ChatRequest(call_context=_CallContext(now=lambda: 0.0), **base) + + +class TestRowLevelObservability: + """三类行的新列取值(1.3.5 设计 §5 表 + 实施计划 §3.3)。 + + 诊断列的存在理由是**归因**: 改前 `error` 是被 `str()` 压平的一列自由文本, + "哪个源回了什么状态码、网关正文说了什么"全部不可 SQL 化。 + """ + + def _emitter(self, rec, **kwargs): + return TelemetryEmitter(rec, scope="LLM", text_cap=None, **kwargs) + + async def _attempt_row(self, error, *, response=None, class_prefixed=False, request=_REQ): + rec = _MemoryRecorder() + await self._emitter(rec).emit_attempt( + request=request, + source=_source(), + call_id="c", + latency_ms=7, + response=response, + error=error, + reasoning_applies=True, + operation="chat", + class_prefixed_error=class_prefixed, + ) + return rec.rows[0] + + async def test_attempt_row_carries_scope_operation_and_kind(self): + row = await self._attempt_row(None, response=_resp()) + assert row["scope"] == "LLM" # 构造期注入,不拿 source_name 顶替 + assert row["operation"] == "chat" + assert row["event_kind"] == "attempt" + + async def test_success_row_leaves_every_diagnostic_null(self): + """成功行不统一填 200: 那会让"有状态码"不再等价于"这次失败了"。""" + row = await self._attempt_row(None, response=_resp()) + assert row["http_status_code"] is None + assert row["error_type"] is None and row["cause_type"] is None + assert row["error_body"] is None + # attempts / total_latency_ms 只属终态行 + assert row["attempts"] is None and row["total_latency_ms"] is None + + async def test_relabelled_status_is_recorded_as_received(self): + """中转把 529 改写成 503 → 如实记 503,**不猜回 529**(设计 §1)。""" + exc = TransientError( + "s1 瞬时错误: 503 | upstream said 529", + source_name="s1", + status_code=503, + operation="chat", + body_text="upstream said 529", + ) + exc.__cause__ = httpx.ConnectTimeout("") + row = await self._attempt_row(exc) + assert row["http_status_code"] == 503 + assert row["error_type"] == "TransientError" + assert row["cause_type"] == "ConnectTimeout" + assert row["error_body"] == "upstream said 529" + assert row["error"] == "s1 瞬时错误: 503 | upstream said 529" + + async def test_direct_529_is_recorded_as_529(self): + exc = TransientError("s1 瞬时错误: 529", status_code=529, body_text="overloaded") + row = await self._attempt_row(exc) + assert row["http_status_code"] == 529 + + @pytest.mark.parametrize( + "cause", + [ + httpx.ConnectTimeout(""), + httpx.ReadTimeout(""), + httpx.WriteTimeout(""), + httpx.PoolTimeout(""), + ], + ) + async def test_empty_timeout_text_still_yields_a_cause_type(self, cause): + """`str()` 为空的 httpx 超时: 类型必须留在 `cause_type` 里,否则无从分辨。""" + exc = TransientError("") + exc.__cause__ = cause + row = await self._attempt_row(exc) + assert row["cause_type"] == type(cause).__name__ + assert row["error"] == "TransientError" # 空 str() 退回类名,不落空串 + + async def test_error_body_is_the_bounded_summary_not_the_model_output(self): + """`error_body` 是**网关拒绝时说的话**(`summarize_body` 摘要),不是模型正文。""" + exc = ResultInvalidError("坏结果", raw_text="x" * 5000, body_text="") + row = await self._attempt_row(exc) + assert row["error_body"] is None # 未知一律 None,不拿 raw_text 顶替 + assert "x" * 100 not in (row["error"] or "") + + async def test_string_error_is_never_parsed_for_diagnostics(self): + """取消行沿用既有 `"cancelled"`;字符串不解析猜诊断(设计 §5)。""" + row = await self._attempt_row("cancelled") + assert row["error"] == "cancelled" + assert row["error_type"] is None + assert row["cause_type"] is None and row["http_status_code"] is None + assert row["error_body"] is None + + async def test_class_prefix_is_an_explicit_policy_parameter(self): + """OCR 的 `"类名: msg"` 口径由出口显式参数保留,不在三处复制拼装。""" + exc = RequestRejectedError("m1 请求被拒: 400") + assert (await self._attempt_row(exc))["error"] == "m1 请求被拒: 400" + prefixed = await self._attempt_row(exc, class_prefixed=True) + assert prefixed["error"] == "RequestRejectedError: m1 请求被拒: 400" + + async def test_logical_call_id_comes_from_the_context(self): + request = _ctx_request() + row = await self._attempt_row(None, response=_resp(), request=request) + assert row["logical_call_id"] == request.call_context.logical_call_id + + async def test_absent_context_lands_null_instead_of_a_fabricated_id(self): + """库内现场构造的请求没有上下文 → NULL,**不造 ID**(设计 §5 I5)。""" + row = await self._attempt_row(None, response=_resp()) + assert row["logical_call_id"] is None + + async def test_cache_hit_row_kind_and_null_logical_counters(self): + rec = _MemoryRecorder() + await self._emitter(rec).emit_cache_hit( + request=_REQ, response=_resp(cache_hit=True), operation="chat" + ) + row = rec.rows[0] + assert row["event_kind"] == "cache_hit" and row["scope"] == "LLM" + assert row["operation"] == "chat" + assert row["attempts"] is None and row["total_latency_ms"] is None + assert row["error_type"] is None and row["http_status_code"] is None + + async def test_terminal_row_carries_the_frozen_snapshot(self): + rec = _MemoryRecorder() + await self._emitter(rec).emit_terminal_failure( + request=_REQ, + call_id="c", + error=AllSourcesExhausted(scope="llm", reason="retry_exhausted", retry_after_s=2.0), + operation="chat", + stats=_STATS, + ) + row = rec.rows[0] + assert row["event_kind"] == "terminal_failure" + assert row["attempts"] == 3 + # 两列同取一份冻结快照,避免双时钟微差 + assert row["total_latency_ms"] == 1234 and row["latency_ms"] == 1234 + assert row["error_type"] == "AllSourcesExhausted" + assert row["error"] == "llm 网关暂时不可用: retry_exhausted" + + async def test_terminal_row_never_borrows_the_last_attempt_diagnostics(self): + """C1 红线: 终态三列保持 NULL——不拿最后一个源的现场冒充整池归因。""" + exc = AllSourcesExhausted(scope="llm", reason="retry_exhausted", retry_after_s=2.0) + exc.__cause__ = TransientError("s1 503", status_code=503, body_text="gateway said 529") + rec = _MemoryRecorder() + await self._emitter(rec).emit_terminal_failure( + request=_REQ, call_id="c", error=exc, operation="chat", stats=_STATS + ) + row = rec.rows[0] + assert row["http_status_code"] is None + assert row["cause_type"] is None + assert row["error_body"] is None + + async def test_terminal_row_costs_nothing(self): + """费用聚合口径不变: 终态行 cost 恒 NULL、usage `unavailable`、token 0。""" + rec = _MemoryRecorder() + await TelemetryEmitter( + rec, scope="LLM", pricing=_PRICING, text_cap=None + ).emit_terminal_failure( + request=_REQ, call_id="c", error="cancelled", operation="chat", stats=_STATS + ) + row = rec.rows[0] + assert row["cost"] is None and row["usage_source"] == "unavailable" + assert row["prompt_tokens"] == 0 and row["completion_tokens"] == 0 + + async def test_structured_exhaustion_terminal_explains_itself_within_bounds(self): + """C2: 结构化耗尽的终态是唯一记录,故有界说明并入 `error`,且不含 raw_text。""" + exc = ResultInvalidError( + "结构化输出阶梯耗尽", + raw_text="y" * 5000, + repair_error="r" * 500, + validation_errors=tuple("v" * 500 for _ in range(5)), + ) + rec = _MemoryRecorder() + await self._emitter(rec).emit_terminal_failure( + request=_REQ, call_id="c", error=exc, operation="chat", stats=_STATS + ) + error = rec.rows[0]["error"] + assert error.startswith("结构化输出阶梯耗尽") + assert "repair=" in error and "validation=" in error + assert "y" * 50 not in error # 模型正文预算已由 attempt 行承担,不重复落库 + # 至多 3 条 × 200 字符 + repair 200 字符,整体有界 + assert error.count("v" * 200) == 3 + assert len(error) < 1200 + + @pytest.mark.parametrize("operation", ["chat", "embed", "recognize_text", "parse_layout"]) + async def test_operation_is_given_by_the_call_site_not_the_exception(self, operation): + """新列 `operation` 与 `exc.operation` 是两个语义(设计 §5 I1/I2)。""" + exc = TransientError("boom", status_code=500, operation="download_result") + rec = _MemoryRecorder() + await self._emitter(rec).emit_attempt( + request=_REQ, + source=_source(), + call_id="c", + latency_ms=1, + response=None, + error=exc, + reasoning_applies=False, + operation=operation, + ) + assert rec.rows[0]["operation"] == operation + + +class TestRecorderShapeGate: + """C3 装配闸: 旧签名 recorder 必须在装配期报错,而不是运行期静默丢行。 + + `_record` 的 `except Exception` 会把旧 recorder 的 `TypeError` 吞成 warning, + 后果是自定义 recorder 在下游升级后 100% 丢遥测且调用照常成功。 + """ + + def test_old_signature_recorder_is_refused_at_assembly(self): + class Old: + async def record_llm_call(self, *, call_id, model, error) -> None: ... + + with pytest.raises(ValueError, match="record_llm_call"): + TelemetryEmitter(Old(), scope="LLM", text_cap=None) + + def test_kwargs_recorder_passes(self): + TelemetryEmitter(_MemoryRecorder(), scope="LLM", text_cap=None) # 不抛 + + def test_uninspectable_recorder_is_a_configuration_error(self): + """不可 inspect(C 实现等)按配置错误当场报错,不进入运行期静默丢行。""" + + class Opaque: + record_llm_call = print # 内置函数: inspect.signature 拿不到 + + with pytest.raises(ValueError): + TelemetryEmitter(Opaque(), scope="LLM", text_cap=None) + + def test_gate_derives_parameter_names_from_the_protocol(self, monkeypatch): + """参数名从协议签名派生,不手抄第四份清单——改协议,闸自动跟随。""" + import polygateway.middleware.telemetry as tele + + class NarrowProtocol: + async def record_llm_call(self, *, call_id, model) -> None: ... + + monkeypatch.setattr(tele, "TelemetryRecorder", NarrowProtocol) + + class NarrowRecorder: + async def record_llm_call(self, *, call_id, model) -> None: ... + + # 若闸内硬编码了 36 个名字,这里必然拒绝 + tele.TelemetryEmitter(NarrowRecorder(), scope="LLM", text_cap=None) + + +class TestTerminalRowSqlSemantics: + """§5 的归因查询与 §8 的迁移清单,按**真实 SQLite 落库**断言。 + + 这些断言的对象是下游真正会写的 SQL: 改前"计失败调用次数"只能按 + `error IS NOT NULL`,而那会同时命中尝试行与终态行。 + """ + + async def _failing_chat_db(self, tmp_path): + from tests.unit.test_client import _client + + recorder = SQLiteRecorder(tmp_path / "t.db", auto_migrate=True) + client = _client( + handler=lambda request: httpx.Response(503), + telemetry=recorder, + retry=RetryPolicy(2, 0.001, 0.01), + ) + try: + with pytest.raises(AllSourcesExhausted): + await client.chat([{"role": "user", "content": "hi"}]) + finally: + await client._transport.aclose() + recorder.close() + return sqlite3.connect(tmp_path / "t.db") + + async def test_attribution_query_gives_terminal_reason_and_per_source_scene(self, tmp_path): + """设计 §5 验收查询: 一条 `logical_call_id` 同时给出整池终态与逐源现场。""" + conn = await self._failing_chat_db(tmp_path) + lcid = conn.execute( + "SELECT logical_call_id FROM llm_calls WHERE event_kind='terminal_failure'" + ).fetchone()[0] + assert lcid is not None + rows = conn.execute( + "SELECT event_kind, source_name, http_status_code, error_type, error " + "FROM llm_calls WHERE logical_call_id = ? ORDER BY created_at", + (lcid,), + ).fetchall() + kinds = [r[0] for r in rows] + assert kinds.count("attempt") == 2 # 逐源现场 + assert kinds.count("terminal_failure") == 1 # 整池终态 + attempts = [r for r in rows if r[0] == "attempt"] + assert all(a[2] == 503 for a in attempts) # 每个源怎么死的 + terminal = next(r for r in rows if r[0] == "terminal_failure") + assert terminal[3] == "AllSourcesExhausted" + assert "retry_exhausted" in terminal[4] # 整池为何失败 + assert terminal[2] is None # C1: 终态不冒充逐源状态码 + + async def test_failure_count_must_come_from_terminal_rows_only(self, tmp_path): + """`error IS NOT NULL` 跨两类行,不再是"失败调用数"的判据(§8)。""" + conn = await self._failing_chat_db(tmp_path) + (by_error,) = conn.execute( + "SELECT COUNT(*) FROM llm_calls WHERE error IS NOT NULL" + ).fetchone() + (by_terminal,) = conn.execute( + "SELECT COUNT(*) FROM llm_calls WHERE event_kind = 'terminal_failure'" + ).fetchone() + assert by_error == 3 # 2 条尝试错误行 + 1 条终态行 + assert by_terminal == 1 # 一次逻辑调用 = 一次失败 + + async def test_terminal_rows_never_contribute_to_cost(self, tmp_path): + """费用聚合口径不变: 终态行 cost 恒 NULL、usage `unavailable`、token 0。""" + conn = await self._failing_chat_db(tmp_path) + (billable,) = conn.execute( + "SELECT COUNT(*) FROM llm_calls " + "WHERE event_kind = 'terminal_failure' AND cost IS NOT NULL" + ).fetchone() + assert billable == 0 + row = conn.execute( + "SELECT usage_source, prompt_tokens, completion_tokens FROM llm_calls " + "WHERE event_kind = 'terminal_failure'" + ).fetchone() + assert row == ("unavailable", 0, 0) + + async def test_latency_must_be_grouped_by_event_kind(self, tmp_path): + """终态行携带**逻辑总耗时**,量级大于单次尝试 → 时延看板必须分组(§8)。""" + conn = await self._failing_chat_db(tmp_path) + grouped = dict( + conn.execute( + "SELECT event_kind, MAX(latency_ms) FROM llm_calls GROUP BY event_kind" + ).fetchall() + ) + assert set(grouped) == {"attempt", "terminal_failure"} + # 终态是整次调用的耗时,含退避与两次尝试,故不小于任何单次尝试 + assert grouped["terminal_failure"] >= grouped["attempt"] + terminal = conn.execute( + "SELECT latency_ms, total_latency_ms, attempts FROM llm_calls " + "WHERE event_kind = 'terminal_failure'" + ).fetchone() + assert terminal[0] == terminal[1] # 同一份冻结快照,无双时钟微差 + assert terminal[2] == 2 + # 尝试行的逻辑两列恒 NULL + assert ( + conn.execute( + "SELECT COUNT(*) FROM llm_calls WHERE event_kind = 'attempt' " + "AND (attempts IS NOT NULL OR total_latency_ms IS NOT NULL)" + ).fetchone()[0] + == 0 + ) diff --git a/tests/unit/test_usage_source_domain.py b/tests/unit/test_usage_source_domain.py index 8c583a5..36fff7a 100644 --- a/tests/unit/test_usage_source_domain.py +++ b/tests/unit/test_usage_source_domain.py @@ -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