fix: degrade terminal telemetry failures instead of masking domain errors
终态出口 `emit_terminal_once` 此前只让 `_record` 内的 except 兜住落库,而 诊断字段的提取(`_error_fields` → `_structured_detail` → `format_bounded_errors`) 在降级 try **之外**求值。下游经公共端口(自实现 `StructuredOutputStrategy` 或 transport)构造出 `ResultInvalidError(validation_errors=(非 str,))` 时,提取期 抛的 `TypeError` 会顶替调用方本该收到的领域异常——错误四分类被击穿(下游 `except ResultInvalidError` 落空),且 `claim_terminal()` 已消耗故终态行照样丢, 同时违反"遥测写失败降级不冒泡"。 改法与 RetryMW 的 attempt 出口(`retry.py::_emit`)同款: 把快照冻结与 await 整段包进 try,`CancelledError` 原样上抛、其余落一条 warning。终态行按已批准的 best effort(兜底命中时该次逻辑调用 0 条终态行,不补写)。异常类型校验与 `ResultInvalidError` 的既有设计均未改动。 顺带同步审查报告的 Minor 项: SQLite recorder docstring 26 → 36 字段、 research-wiki 索引重建、ARCHITECTURE 的 `sampling` 段落终态调用点口径, 并删除 `TelemetryMW` 迁移后无读取点的 `self._now` 死字段(保留形参, 避免平白打断既有装配写法)。
This commit is contained in:
@@ -564,7 +564,7 @@ flowchart TB
|
||||
|
||||
**必录字段**(继承三项目 15 字段规范;当前 36 个 INSERT 字段,物理表列 37 = 36 + 数据库自填的 `created_at`,两套口径的区分见 `telemetry/schema.py` 模块 docstring): call_id、parent_call_id、session_id、model、provider、source_name、messages(JSON)、response、thinking、prompt_tokens、completion_tokens、usage_source、latency_ms、ttft_ms、max_inter_token_ms、cache_hit、error、**cost**、**cached_prompt_tokens**、**model_reported**、**sampling**、**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**。
|
||||
|
||||
**`sampling` 列(2026-07-31,issue #4,端口 20 → 21)**: 列语义 = 「调用方采样意图 ⊎ 生效源 `extra_body`」的 canonical JSON,空则 NULL。**不含**结构化注入的 `response_format`——列名是采样参数,schema 不是,且数 KB schema 逐行落库会让审计表无谓膨胀。三个 emit 入口口径必须各自定死,否则同一列在不同行含义不同: `emit_attempt`(RetryMW 调用,**唯一**有生效源者)并上 `source.extra_body`;`emit_cache_hit` / `emit_terminal_failure`(TelemetryMW 最外层调用)无 source 可言,只记调用级——与 `model`/`source_name` 在终态行置空是同一先例,且缓存命中行无损(`sampling` 已进缓存 key,能命中即意味调用级参数与历史那次逐字相同)。三者统一读 `request.sampling` 而非 `request.overlay`(后者在 RetryMW 处已被结构化注入污染、在 TelemetryMW 处未被污染,直接用必然三行分叉)。OCR/embedding 路径因决策 G 剥离 `extra_body`,该列恒 NULL。
|
||||
**`sampling` 列(2026-07-31,issue #4,端口 20 → 21)**: 列语义 = 「调用方采样意图 ⊎ 生效源 `extra_body`」的 canonical JSON,空则 NULL。**不含**结构化注入的 `response_format`——列名是采样参数,schema 不是,且数 KB schema 逐行落库会让审计表无谓膨胀。三个 emit 入口口径必须各自定死,否则同一列在不同行含义不同: `emit_attempt`(RetryMW 调用,**唯一**有生效源者)并上 `source.extra_body`;`emit_cache_hit` / `emit_terminal_failure`(前者由 TelemetryMW 最外层调用,后者自 1.3.5 起由三个 client 的公开边界经 `emit_terminal_once` 统一写出)无 source 可言,只记调用级——与 `model`/`source_name` 在终态行置空是同一先例,且缓存命中行无损(`sampling` 已进缓存 key,能命中即意味调用级参数与历史那次逐字相同)。三者统一读 `request.sampling` 而非 `request.overlay`(后者在 RetryMW 处已被结构化注入污染、在 TelemetryMW 处未被污染,直接用必然三行分叉)。OCR/embedding 路径因决策 G 剥离 `extra_body`,该列恒 NULL。
|
||||
|
||||
**`reasoning_tokens` 列(2026-08-11,issue #6,端口 21 → 22)**: 推理 token 已计入 `completion_tokens`,故成本总额一直是对的——这不是计费缺口而是**归因**缺口:缺了它,"这次调用花的钱里有多少花在推理上"无法区分,也就无从判断某个 scope 该不该关推理。供应商不报时记 NULL 而非 0(不可得 ≠ 为零,与 `usage_source='unavailable'` 同一纪律)。
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# Research Wiki 索引
|
||||
|
||||
> 自动生成，更新时间：2026-09-09 13:42 UTC
|
||||
> 自动生成，更新时间：2026-09-09 16:49 UTC
|
||||
|
||||
## design (43)
|
||||
|
||||
@@ -48,9 +48,10 @@
|
||||
- [调用方自定义维度设计(issue #11)](designs/issue11-caller-dimensions.md) `design:issue11-caller-dimensions`
|
||||
- [采样参数透传设计(issue #4)](designs/sampling-params.md) `design:sampling-params`
|
||||
|
||||
## finding (15)
|
||||
## finding (16)
|
||||
|
||||
- [1.3.4 推理契约验证与发布准备](findings/2026-09-09-134-thinking-contracts-validation.md) `finding:2026-09-09-134-thinking-contracts-validation`
|
||||
- [1.3.5 T2/T3/T4 验收证据：36 列遥测、失败终态、PG 存储兼容与变异矩阵](findings/2026-09-09-135-call-observability-validation.md) `finding:2026-09-09-135-call-observability-validation`
|
||||
- [2026-07-20-m2-soak-workload](findings/2026-07-20-m2-soak-workload.md) `finding:2026-07-20-m2-soak-workload`
|
||||
- [2026-07-21-m25-acceptance](findings/2026-07-21-m25-acceptance.md) `finding:2026-07-21-m25-acceptance`
|
||||
- [2026-07-21-p6-soak-baseline](findings/2026-07-21-p6-soak-baseline.md) `finding:2026-07-21-p6-soak-baseline`
|
||||
@@ -113,7 +114,7 @@
|
||||
|
||||
## schema (1)
|
||||
|
||||
- [表结构: llm_calls(遥测 26 字段)](schemas/llm-calls.md) `schema:llm-calls`
|
||||
- [表结构: llm_calls(遥测 36 字段)](schemas/llm-calls.md) `schema:llm-calls`
|
||||
|
||||
## metric (2)
|
||||
|
||||
|
||||
@@ -158,3 +158,4 @@
|
||||
- [2026-09-09 06:32 UTC] 重建索引: 98 篇页面
|
||||
- [2026-09-09 13:42 UTC] 新增边: plan:2026-09-09-135-call-observability --implements--> design:2026-09-09-135-call-observability-design
|
||||
- [2026-09-09 13:42 UTC] 重建索引: 100 篇页面
|
||||
- [2026-09-09 16:49 UTC] 重建索引: 101 篇页面
|
||||
|
||||
@@ -679,7 +679,16 @@ async def emit_terminal_once(
|
||||
"""三个 client 共用的**终态唯一出口**: 去重 + 同步冻结快照 + best effort 写入。
|
||||
|
||||
去重由 `claim_terminal()` 承担(每逻辑调用至多一条终态行);`emitter is None`
|
||||
或已写过 → 直接返回。写入侧异常按既有降级只落 warning(在 `_record` 内)。
|
||||
或已写过 → 直接返回。
|
||||
|
||||
**降级范围包含诊断字段的提取与构建**,不只是写入那一步: `_record` 内的
|
||||
`except Exception` 只兜住落库,而 `_error_fields` / `canonical_sampling_json`
|
||||
在它**之外**求值——下游经公共端口(自实现 `StructuredOutputStrategy` 或
|
||||
transport)构造出的 `ResultInvalidError(validation_errors=(非 str,))` 会让提取期
|
||||
抛 `TypeError` 顶替调用方本该收到的领域异常,错误四分类被击穿且终态行照样丢。
|
||||
故在此整段兜底,与 RetryMW 的 attempt 出口(`retry.py::_emit`)同款写法。
|
||||
终态行按 best effort: 兜底命中时本次逻辑调用**没有**终态行(`claim_terminal()`
|
||||
已消耗,不补写、不重试——重写一遍只会把同一个提取期异常再抛一次)。
|
||||
|
||||
**`CancelledError` 原样传播**(取消优先,不 shield、不开后台任务): 这一次
|
||||
`await` 本身就是新的取消点,外部取消落在它上时调用方会看到 `CancelledError`
|
||||
@@ -688,6 +697,7 @@ async def emit_terminal_once(
|
||||
"""
|
||||
if emitter is None or not context.claim_terminal():
|
||||
return
|
||||
try:
|
||||
stats = context.snapshot()
|
||||
await emitter.emit_terminal_failure(
|
||||
request=request,
|
||||
@@ -697,6 +707,10 @@ async def emit_terminal_once(
|
||||
stats=stats,
|
||||
class_prefixed_error=class_prefixed_error,
|
||||
)
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except Exception as exc:
|
||||
logger.warning("终态遥测记录失败(降级不冒泡): {}", exc)
|
||||
|
||||
|
||||
class TelemetryMW:
|
||||
@@ -711,7 +725,8 @@ class TelemetryMW:
|
||||
self, emitter: TelemetryEmitter, now: Callable[[], float] = time.monotonic
|
||||
) -> None:
|
||||
self._emitter = emitter
|
||||
self._now = now
|
||||
# `now` 自 1.3.5 起本类不再读取(终态行迁到公开边界后无耗时可测),但形参保留:
|
||||
# 删它会平白打断 `TelemetryMW(emitter, now=...)` 这一既有装配写法
|
||||
|
||||
async def __call__(self, request: ChatRequest, call_next: CallNext) -> LLMResponse:
|
||||
response = await call_next(request)
|
||||
|
||||
@@ -143,7 +143,7 @@ class SQLiteRecorder:
|
||||
logger.warning("SQLite 遥测补列失败(写入将逐行降级): {}", exc)
|
||||
|
||||
async def record_llm_call(self, **fields: object) -> None:
|
||||
"""写一行遥测;字段集合即 26 字段冻结签名(ports.TelemetryRecorder)。
|
||||
"""写一行遥测;字段集合即 36 字段冻结签名(ports.TelemetryRecorder)。
|
||||
|
||||
取值按 `self._columns`(manual 档可能已被裁剪),与 `self._insert` 的
|
||||
占位符同序——两者必须一起改,分开改就是把值写进错位的列。
|
||||
|
||||
@@ -1645,3 +1645,75 @@ class TestChatTerminalFailureRows:
|
||||
operation="chat",
|
||||
)
|
||||
assert len(self._rows(recorder, "terminal_failure")) == 1
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def captured_warnings():
|
||||
"""捕获库发出的 WARNING;loguru 不经标准 logging,pytest 的 caplog 抓不到。
|
||||
|
||||
名字避开裸 `warnings`: 那会遮蔽标准库模块名,本文件将来任何一次
|
||||
`import warnings` 都会与它静默互相顶掉,而报错点离真因很远。
|
||||
"""
|
||||
from loguru import logger
|
||||
|
||||
messages: list[str] = []
|
||||
sink_id = logger.add(messages.append, level="WARNING")
|
||||
yield messages
|
||||
logger.remove(sink_id)
|
||||
|
||||
|
||||
class TestTerminalEmitDegradation:
|
||||
"""终态出口的降级方向: 诊断字段的**提取**同样在降级范围内(铁律"遥测写失败降级不冒泡")。
|
||||
|
||||
改前 `_record` 的 except 只包住写入本身,而错误诊断列在它之外求值——
|
||||
经公共扩展点(自实现 `StructuredOutputStrategy`/transport)构造出的
|
||||
`ResultInvalidError(validation_errors=(非 str,))` 会让提取期抛 `TypeError`
|
||||
顶替调用方本该收到的领域异常,错误四分类当场被击穿。
|
||||
"""
|
||||
|
||||
_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_broken_validation_errors_keep_the_domain_error(self, captured_warnings):
|
||||
"""自实现策略给出非 str 的 `validation_errors`: 领域异常必须原样上抛。"""
|
||||
|
||||
class _BadStrategy:
|
||||
"""公共端口 `StructuredOutputStrategy` 的下游实现(库外没有类型执法)。"""
|
||||
|
||||
def request_overlay(self, schema):
|
||||
return {}
|
||||
|
||||
def parse(self, text):
|
||||
raise ResultInvalidError("模型返回不可解析", validation_errors=(object(),))
|
||||
|
||||
recorder = _MemoryRecorder()
|
||||
async with _client(telemetry=recorder, structured_strategy=_BadStrategy()) as client:
|
||||
with pytest.raises(ResultInvalidError): # 不是 TypeError
|
||||
await client.chat(self._MSG, structured="json")
|
||||
# 降级有声: 静默吞掉等于遥测缺口无人知道
|
||||
assert [m for m in captured_warnings if "终态遥测记录失败" in m]
|
||||
# 尝试行不受影响;终态行按 best effort 允许 0 条,但绝不能重复
|
||||
assert len(self._rows(recorder, "attempt")) == 1
|
||||
assert len(self._rows(recorder, "terminal_failure")) <= 1
|
||||
|
||||
async def test_cancellation_is_never_swallowed_by_the_degradation(self):
|
||||
"""降级不得吞取消: 写入那一次 await 上被取消,`CancelledError` 照常传播。"""
|
||||
from polygateway.middleware.telemetry import emit_terminal_once
|
||||
from polygateway.types import _CallContext
|
||||
|
||||
class _CancellingEmitter:
|
||||
async def emit_terminal_failure(self, **kwargs):
|
||||
raise asyncio.CancelledError
|
||||
|
||||
context = _CallContext(now=asyncio.get_running_loop().time)
|
||||
request = ChatRequest(messages=self._MSG, call_context=context)
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await emit_terminal_once(
|
||||
_CancellingEmitter(),
|
||||
request=request,
|
||||
context=context,
|
||||
error=AllSourcesExhausted(scope="llm", reason="retry_exhausted", retry_after_s=1.0),
|
||||
operation="chat",
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user