diff --git a/research-wiki/ARCHITECTURE.md b/research-wiki/ARCHITECTURE.md index 6eb6b3e..88512d0 100644 --- a/research-wiki/ARCHITECTURE.md +++ b/research-wiki/ARCHITECTURE.md @@ -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'` 同一纪律)。 diff --git a/research-wiki/index.md b/research-wiki/index.md index 471d247..7bf1519 100644 --- a/research-wiki/index.md +++ b/research-wiki/index.md @@ -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) diff --git a/research-wiki/log.md b/research-wiki/log.md index d0fa0ca..6abfff4 100644 --- a/research-wiki/log.md +++ b/research-wiki/log.md @@ -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 篇页面 diff --git a/src/polygateway/middleware/telemetry.py b/src/polygateway/middleware/telemetry.py index d728313..ef07fb4 100644 --- a/src/polygateway/middleware/telemetry.py +++ b/src/polygateway/middleware/telemetry.py @@ -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,15 +697,20 @@ async def emit_terminal_once( """ 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, - ) + try: + 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, + ) + 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) diff --git a/src/polygateway/telemetry/sqlite.py b/src/polygateway/telemetry/sqlite.py index 7781530..c3e61dd 100644 --- a/src/polygateway/telemetry/sqlite.py +++ b/src/polygateway/telemetry/sqlite.py @@ -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` 的 占位符同序——两者必须一起改,分开改就是把值写进错位的列。 diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 4213097..c9451ce 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -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", + )