fix: harden the observability fields against the verifier findings

research-wiki/index.md

@@ -1,6 +1,6 @@
 # Research Wiki 索引
 
-> 自动生成，更新时间：2026-07-31 11:11 UTC
+> 自动生成，更新时间：2026-07-31 12:25 UTC
 
 ## design (18)
 - [2026-07-20-m1-core-design](designs/2026-07-20-m1-core-design.md) `design:2026-07-20-m1-core-design`
@@ -52,7 +52,7 @@
 - [响应可观测字段扩展实现计划](plans/response-observability-fields.md) `plan:response-observability-fields`
 
 ## schema (1)
-- [表结构: llm_calls(遥测 18 字段)](schemas/llm-calls.md) `schema:llm-calls`
+- [表结构: llm_calls(遥测 20 字段)](schemas/llm-calls.md) `schema:llm-calls`
 
 ## metric (2)
 - [OCR 治理调用成功率与错误分类分布](metrics/ocr-call-success.md) `metric:ocr-call-success`

research-wiki/log.md

@@ -63,3 +63,4 @@
 - [2026-07-31 11:10 UTC] 新增边: plan:response-observability-fields --implements--> design:response-observability-fields
 - [2026-07-31 11:10 UTC] 重建索引: 46 篇页面
 - [2026-07-31 11:11 UTC] 重建索引: 46 篇页面
+- [2026-07-31 12:25 UTC] 重建索引: 46 篇页面

research-wiki/schemas/llm-calls.md

@@ -1,11 +1,11 @@
 ---
 type: schema
 node_id: schema:llm-calls
-title: "表结构: llm_calls(遥测 18 字段)"
+title: "表结构: llm_calls(遥测 20 字段)"
 date: 2026-07-20
 ---
 
-# 表结构: llm_calls(遥测 18 字段)
+# 表结构: llm_calls(遥测 20 字段)
 
 
 ## 列定义(冻结,M1 设计 §4.4 / ARCH §7.8)
@@ -24,6 +24,8 @@ date: 2026-07-20
 | error | TEXT | 异常信息;取消记 "cancelled" |
 | cost | REAL | M2 起 pricing 换算;`usage_source='unavailable'` 的真实调用行为 NULL(缓存命中行例外,仍为 0.0) |
 | created_at | TEXT NOT NULL DEFAULT (datetime('now')) | 落库时刻 |
+| cached_prompt_tokens | INTEGER | 供应商 prompt cache 命中的输入 token(2026-07-31,issue #3);NULL = 该源未上报,`0` = 上报了真实零命中,两者不可混同 |
+| model_reported | TEXT | API 响应体实际返回的 model;NULL = 未上报。与 `model`(配置别名)可能分叉 |
 
 ## usage/成本口径(2026-07-30,est_tokens 解耦)
 
@@ -35,6 +37,19 @@ date: 2026-07-20
 
 `SUM(cost)` 天然跳过 NULL,故账单汇总不再被虚构的估值污染;账目缺口的度量口径固定为 `WHERE usage_source = 'unavailable' AND cache_hit = false`。**`cache_hit` 限定不可省**:缓存命中行未产生新调用,cost 是事实上的 `0.0` 而非未知,本无账目缺口,漏掉该条件会让缺口度量偏高。
 
+## 供应商 prompt cache 口径(2026-07-31,issue #3)
+
+新增两列排在 `created_at` **之后**——旧表只能经 `ALTER TABLE ADD COLUMN` 追加到末尾,DDL 里若插在前面,新建库与升级库的物理列序会分叉(列序断言无合规修法)。两个后端在初始化期幂等补列:`CREATE TABLE IF NOT EXISTS` 不会给旧表加列,不补则每行写入被逐行 warning 丢弃、遥测静默全失;补列失败只降级为逐行丢弃,绝不让 recorder 整体失能。
+
+`cache_hit` 指 **PolyGateway 自身响应缓存**,与供应商 prompt cache 是两回事。缓存命中行的这两列是**原样回放**的历史值(与 `model`/`prompt_tokens` 同一口径),故命中率度量口径固定为:
+
+```sql
+SELECT SUM(cached_prompt_tokens)::float / NULLIF(SUM(prompt_tokens), 0)
+FROM llm_calls WHERE cache_hit = false AND cached_prompt_tokens IS NOT NULL;
+```
+
+`WHERE cache_hit = false` 不可省,理由与上面 cost 缺口口径同源:回放行计入即重复计数。
+
 ## 埋点位置(单一 helper 铁律)
 
 - `middleware/telemetry.py::TelemetryEmitter` 是全库**唯一** `record_llm_call` 调用点;

src/polygateway/pricing.py

@@ -99,7 +99,8 @@ class PricingTable:
                 logger.warning("pricing 表无 model {!r} 的单价,cost 记 None", model)
             return None
         billed_input = prompt_tokens / 1_000_000 * price.input_per_1m
-        if price.cached_input_per_1m is not None and cached_prompt_tokens:
+        # 负数按"无命中"处理: cost() 是公共方法,不能假定调用方已过 transport 的校验
+        if price.cached_input_per_1m is not None and (cached_prompt_tokens or 0) > 0:
             cached = self._clamp_cached(model, prompt_tokens, cached_prompt_tokens)
             billed_input = (prompt_tokens - cached) / 1_000_000 * price.input_per_1m + (
                 cached / 1_000_000 * price.cached_input_per_1m

src/polygateway/transports/openai_compat.py

@@ -45,9 +45,12 @@ def _sse_delta(chunk: dict[str, Any], usage_sink: dict[str, Any]) -> tuple[bool,
     """从 chunk 提取增量: (True, content) 或 (False, reasoning);usage 帧旁路进 sink。"""
     if chunk.get("usage"):
         usage_sink["usage"] = chunk["usage"]
-    if "model" not in usage_sink and chunk.get("model") is not None:
+    if "model" not in usage_sink:
-        # 首次写入即固定: 末帧的异常值不得覆盖首帧报的真实版本(issue #3)
+        # 首个**有效**值即固定: 末帧的异常值不得覆盖它;但首帧报空串也不能锁死
-        usage_sink["model"] = chunk["model"]
+        # sink——否则后续真实版本会丢(issue #3)
+        reported = _coerce_model_reported(chunk.get("model"))
+        if reported is not None:
+            usage_sink["model"] = reported
     choices = chunk.get("choices") or []
     if not choices:
         return None

tests/integration/test_postgres_telemetry.py

@@ -119,6 +119,105 @@ async def _fetch(dsn: str, sql: str, *args):
         await conn.close()
 
 
+_LEGACY_DDL = """
+CREATE TABLE {schema}.llm_calls (
+    call_id            TEXT PRIMARY KEY,
+    parent_call_id     TEXT,
+    session_id         TEXT,
+    model              TEXT NOT NULL,
+    provider           TEXT NOT NULL,
+    source_name        TEXT NOT NULL,
+    messages           TEXT NOT NULL,
+    response           TEXT NOT NULL,
+    thinking           TEXT NOT NULL DEFAULT '',
+    prompt_tokens      INTEGER NOT NULL,
+    completion_tokens  INTEGER NOT NULL,
+    usage_source       TEXT NOT NULL,
+    latency_ms         INTEGER NOT NULL,
+    ttft_ms            DOUBLE PRECISION,
+    max_inter_token_ms DOUBLE PRECISION,
+    cache_hit          BOOLEAN NOT NULL DEFAULT FALSE,
+    error              TEXT,
+    cost               DOUBLE PRECISION,
+    created_at         TIMESTAMPTZ NOT NULL DEFAULT now()
+)
+"""
+
+
+@pytest.fixture
+async def legacy_schema(dsn):
+    """在**自建的临时 schema** 里造一张 18 列旧表,验证补列(issue #3)。
+
+    绝不碰共享的 public.llm_calls: 用 search_path 把 recorder 指向临时 schema,
+    teardown 只 DROP 自己建的 schema。
+    """
+    import asyncpg
+
+    name = f"pgwtest_{uuid4().hex[:8]}"
+    conn = await asyncpg.connect(dsn, timeout=10)
+    try:
+        await conn.execute(f"CREATE SCHEMA {name}")
+        await conn.execute(_LEGACY_DDL.format(schema=name))
+    finally:
+        await conn.close()
+    sep = "&" if "?" in dsn else "?"
+    yield f"{dsn}{sep}options=-csearch_path%3D{name}", name
+    conn = await asyncpg.connect(dsn, timeout=10)
+    try:
+        await conn.execute(f"DROP SCHEMA {name} CASCADE")
+    finally:
+        await conn.close()
+
+
+class TestObservabilityColumns:
+    """issue #3: 两列写入可回读,且已存在的 18 列旧表会被自动补列。"""
+
+    async def test_values_round_trip(self, dsn):
+        recorder = PostgresRecorder(dsn)
+        try:
+            await _record_minimal(recorder, call_id=_cid("hit"), cached_prompt_tokens=64)
+            await _record_minimal(recorder, call_id=_cid("zero"), cached_prompt_tokens=0)
+            await _record_minimal(recorder, call_id=_cid("model"), model_reported="MiniMax-01")
+            rows = await _fetch(
+                dsn,
+                "SELECT call_id, cached_prompt_tokens, model_reported FROM llm_calls "
+                "WHERE call_id LIKE $1",
+                f"{_RUN_PREFIX}-%",
+            )
+            by_id = {r["call_id"]: r for r in rows}
+            assert by_id[_cid("hit")]["cached_prompt_tokens"] == 64
+            assert by_id[_cid("zero")]["cached_prompt_tokens"] == 0  # 真实零命中 ≠ NULL
+            assert by_id[_cid("model")]["cached_prompt_tokens"] is None
+            assert by_id[_cid("model")]["model_reported"] == "MiniMax-01"
+        finally:
+            await recorder.aclose()
+
+    async def test_legacy_table_is_upgraded_in_place(self, legacy_schema):
+        """18 列旧表不补列的话,每行写入都会被逐行 warning 丢弃(遥测静默全失)。"""
+        schema_dsn, schema = legacy_schema
+        recorder = PostgresRecorder(schema_dsn)
+        try:
+            await _record_minimal(
+                recorder, call_id=_cid("legacy"), cached_prompt_tokens=7, model_reported="m-real"
+            )
+            cols = await _fetch(
+                schema_dsn,
+                "SELECT column_name FROM information_schema.columns "
+                "WHERE table_schema = $1 AND table_name = 'llm_calls' ORDER BY ordinal_position",
+                schema,
+            )
+            # ALTER 只能追加到末尾: 与新建库的列序一致才不会分叉
+            assert [r["column_name"] for r in cols] == _EXPECTED_COLUMNS
+            rows = await _fetch(
+                schema_dsn,
+                "SELECT cached_prompt_tokens, model_reported FROM llm_calls WHERE call_id = $1",
+                _cid("legacy"),
+            )
+            assert (rows[0]["cached_prompt_tokens"], rows[0]["model_reported"]) == (7, "m-real")
+        finally:
+            await recorder.aclose()
+
+
 class TestSchema:
     async def test_schema_has_frozen_columns_in_order(self, dsn):
         recorder = PostgresRecorder(dsn)

tests/unit/test_openai_compat.py

@@ -360,6 +360,19 @@ class TestObservabilityFields:
         result = await _complete(_transport_for(handler), _source())
         assert result.model_reported == "MiniMax-Text-01-250321"
 
+    async def test_empty_first_model_does_not_block_a_later_real_one(self):
+        """首帧报空串不得锁死 sink: 守卫按"有效值"判断,否则真实版本会丢。"""
+
+        def handler(request):
+            return _sse_stream(
+                _chunk(content="a", model=""),
+                _chunk(content="b", model="MiniMax-Text-01-250321"),
+                _chunk(usage=_USAGE),
+            )
+
+        result = await _complete(_transport_for(handler), _source())
+        assert result.model_reported == "MiniMax-Text-01-250321"
+
     @pytest.mark.parametrize("bad", [None, "", "   ", 123, {}])
     async def test_missing_or_malformed_model_is_none(self, bad):
         def handler(request):

tests/unit/test_pricing.py

@@ -77,6 +77,10 @@ class TestCachedInputTier:
     def test_no_hit_is_billed_in_full(self, cached):
         assert self._CACHED.cost("m", 1_000_000, 0, cached) == pytest.approx(10.0)
 
+    def test_negative_cached_is_billed_in_full(self):
+        """负数命中数不得抬高成本: cost() 是公共方法,外部输入须校验后使用(P5)。"""
+        assert self._CACHED.cost("m", 1_000_000, 0, -500_000) == pytest.approx(10.0)
+
     def test_cached_over_prompt_is_clamped_and_never_negative(self):
         """网关口径异常时按输入总数夹取: 全部按缓存价,不得算出负成本。"""
         clamped = self._CACHED.cost("m", 1_000_000, 0, 5_000_000)

tests/unit/test_telemetry.py

@@ -202,8 +202,9 @@ class TestSQLiteColumnBackfill:
     async def test_backfill_failure_keeps_the_recorder_usable(self, tmp_path):
         """补列失败只能逐行降级,绝不能把 recorder 整体变成 no-op(设计 D1 纪律)。
 
-        把 llm_calls 建成 view: 表不存在故 CREATE TABLE IF NOT EXISTS 会撞名失败,
+        把 llm_calls 建成同名 view: `CREATE TABLE IF NOT EXISTS` 遇 view 静默
-        ALTER 也无从谈起——这是最坏路径。
+        no-op(不抛),随后的 ALTER 才抛 "Cannot add a column to a view"——正是
+        补列失败这条分支。`_conn` 必须保持非 None,否则整个 recorder 永久失能。
         """
         db = tmp_path / "view.db"
         conn = sqlite3.connect(db)
@@ -213,6 +214,7 @@ class TestSQLiteColumnBackfill:
         conn.close()
 
         recorder = SQLiteRecorder(db)  # 不得抛
+        assert recorder._conn is not None  # 补列失败 ≠ recorder 失能(D1 纪律)
         await _record_minimal(recorder)  # 不得抛
         recorder.close()