From 393f2bf6170b4cc64c32bee9453213b16cdbb4d1 Mon Sep 17 00:00:00 2001 From: iomgaa Date: Wed, 9 Sep 2026 11:27:52 -0400 Subject: [PATCH] feat: record call observability columns and terminal failure rows Grow the telemetry contract from 26 to 36 fields and give every logical call a failure terminal row, so SQL can finally answer "how many calls failed" and "why did the whole pool die". Schema and port move together with the emitter writes in one commit: splitting them would ship columns that nothing populates. - schema: append 10 nullable columns (scope, operation, logical_call_id, event_kind, http_status_code, error_type, cause_type, error_body, attempts, total_latency_ms) to all five definition sites in one order - ports: 10 keyword-only parameters without defaults; the protocol signature is now the single source the assembly gate derives from - emitter: take domain exception objects instead of pre-flattened text and pin down the diagnostics in one helper; a relabelled 503 stays 503 and success rows leave all five columns NULL - emitter: reject recorders whose record_llm_call cannot accept the current field shape at assembly time, since _record would otherwise swallow the TypeError and drop every row while calls keep succeeding - clients: write at most one terminal row per logical call through a single shared exit, deduplicated by the call context; TelemetryMW stops writing terminals so the two sites cannot double count - clients: cancellation stays best effort and propagates, non-domain exceptions get no terminal row and keep their classification - transports: give _status_to_error an explicit operation and fix the historically mislabelled embedding HTTP failures - structured: promote the bounded error formatter so the reask feedback and the terminal explanation share one set of limits Terminal rows carry no cost and no tokens, so cost aggregation is unchanged; failure counts must now filter on event_kind. --- ...09-09-135-call-observability-validation.md | 141 ++++ src/polygateway/client.py | 27 +- src/polygateway/embedding.py | 90 ++- src/polygateway/middleware/retry.py | 15 +- src/polygateway/middleware/structured.py | 21 +- src/polygateway/middleware/telemetry.py | 317 ++++++-- src/polygateway/ocr.py | 149 +++- src/polygateway/ports.py | 21 +- src/polygateway/telemetry/schema.py | 62 +- src/polygateway/transports/openai_compat.py | 27 +- tests/unit/test_cache.py | 3 +- tests/unit/test_client.py | 148 +++- tests/unit/test_embedding.py | 7 +- tests/unit/test_ocr_client.py | 8 +- tests/unit/test_openai_compat.py | 3 +- tests/unit/test_ports.py | 53 +- tests/unit/test_pricing.py | 16 +- tests/unit/test_telemetry.py | 690 ++++++++++++++++-- tests/unit/test_usage_source_domain.py | 24 +- 19 files changed, 1628 insertions(+), 194 deletions(-) create mode 100644 research-wiki/findings/2026-09-09-135-call-observability-validation.md 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