PolyGateway/research-wiki/plans/2026-07-31-response-observability-fields.md

# 实现计划: 响应可观测字段扩展(Issue #3)

- **目标**: 让 `LLMResponse` 与遥测表如实暴露「供应商 prompt cache 命中的输入 token 数」与「API 实际返回的模型版本串」。
- **方案概述**: 报文解析留在 `transports/`(新增两个强类型字段随 `TransportResult` 上浮),`RetryMW` 只搬运;遥测端口由 18 字段扩到 20 并给两个后端加幂等补列;`PricingTable` 增加可选缓存单价档消除 cost 高估。缓存命中行按既有口径原样回放。
- **依据设计**: `research-wiki/designs/2026-07-31-response-observability-fields-design.md`(2026-07-31 已获人类批准,决策 A2/B1/C1/D1)。
- **涉及技术**: Python 3.11 frozen dataclass、httpx SSE 解析、sqlite3、asyncpg、pytest。
- **保真校验**: 本计划**不涉及** `reference/` 参考实现迁移,保真校验不适用。但遥测后端属 ARCHITECTURE §1.4 资产,T5 明确约束「不得改变既有降级语义」。

## 文件结构

| 文件 | 职责 | 本次改动 |
|---|---|---|
| `src/polygateway/types.py` | 冻结公共类型 | `LLMResponse` / `TransportResult` 各 +2 字段;`cache_hit` docstring 消歧 |
| `src/polygateway/transports/openai_compat.py` | OpenAI 兼容报文解析 | 防御解析 helper;SSE sink 采集 `model`;两处 `TransportResult` 构造填新字段 |
| `src/polygateway/middleware/retry.py` | 尝试循环 | `_build_response` 搬运两字段 |
| `src/polygateway/middleware/cache.py` | 响应缓存 | **零代码改动**(自动透传),仅补测试固化行为 |
| `src/polygateway/pricing.py` | 单价换算 | `ModelPrice` +可选档;`cost()` +可选参;`from_file` 校验 |
| `src/polygateway/ports.py` | 端口契约 | `TelemetryRecorder` 18 → 20 字段 |
| `src/polygateway/telemetry/{sqlite,postgres}.py` | 遥测后端 | DDL +2 列;`_COLUMNS` +2;初始化期幂等补列 |
| `src/polygateway/middleware/telemetry.py` | 遥测唯一调用点 | `_record` 与三个 `emit_*` 搬运两字段;cost 换算传入缓存 token |

字段定义(全库唯一权威,后续任务一律引用此处):

```python
# LLMResponse 与 TransportResult 尾部,同名同类型同默认值
cached_prompt_tokens: int | None = None  # 供应商 prompt cache 命中的输入 token;None = 该源未上报
model_reported: str | None = None        # API 响应体的 model 字段;None = 未上报
```

---

## T1. 类型层加字段

- [ ] **改**: `src/polygateway/types.py`

**行为**: 在 `LLMResponse` 尾部(`structured_data` 之后)与 `TransportResult` 尾部(`raw` 之后)各追加上面两个字段。`cache_hit` 的语义在 `LLMResponse` docstring 中写明是「**PolyGateway 自身响应缓存**命中,与供应商 prompt cache 无关,后者见 `cached_prompt_tokens`」。

**验收**: 前 11 个字段的顺序与名字一字不动;新字段有默认值,`LLMResponse(...)` 按前 11 位置参数构造仍成立;`TransportResult` 现有两处构造(`openai_compat.py:354/436`)不传新字段也能构造。

**测试**(`tests/unit/test_types.py`): ① 不传新字段时两个类型的新字段均为 `None`;② 按位置构造 `LLMResponse` 的前 11 字段仍可用(迁移兼容承诺)。

**验证**: `conda run -n PolyGateway --no-capture-output pytest tests/unit/test_types.py -v` → PASS。

**提交**: `feat: add cached prompt tokens and reported model to response types`

## T2. transport 采集与防御解析

- [ ] **改**: `src/polygateway/transports/openai_compat.py`

**行为**分三处:

1. 新增两个模块级防御 helper(网关返回一律不可信,解析失败**返回 None,不抛异常**):

```python
def _coerce_cached_tokens(usage: Any) -> int | None:
    """从 usage.prompt_tokens_details.cached_tokens 取非负整数;任何形态异常 → None。"""

def _coerce_model_reported(value: Any) -> str | None:
    """响应体 model 字段: 非空 str 才收,其余(含空串/非 str)→ None。"""
```

`_coerce_cached_tokens` 需容忍:`usage` 为 None、`prompt_tokens_details` 缺失或非 dict、`cached_tokens` 为 `bool`/`str`/负数/浮点。`bool` 必须排除(Python 中 `isinstance(True, int)` 为真)。

2. 流式路径:`_sse_delta`(`:44-47`)当前只把 `usage` 旁路进 sink。补一条——chunk 里出现 `model` 时写 `usage_sink["model"]`(**首次写入即固定**,后续 chunk 不覆盖,避免末帧异常值污染)。`_stream_once` 的 `TransportResult` 构造(`:354`)填 `cached_prompt_tokens=_coerce_cached_tokens(sink.get("usage"))`、`model_reported=_coerce_model_reported(sink.get("model"))`。

3. 非流式路径:`TransportResult` 构造(`:436`)填 `_coerce_cached_tokens(body.get("usage"))` 与 `_coerce_model_reported(body.get("model"))`。

**验收**: `raw` 的内容保持原样不动(新字段是独立格子,不是杂物袋的扩充);OCR 与 embedding 的解析路径一行不改。

**测试**(`tests/unit/test_openai_compat.py`,用现有 fake 响应二次构造):

| 用例 | 期望 |
|---|---|
| 非流式 usage 含 `prompt_tokens_details.cached_tokens: 128` | `cached_prompt_tokens == 128` |
| 流式 usage 帧同上 | 同上 |
| 无 `prompt_tokens_details` / usage 帧缺失 | `None` |
| `cached_tokens` 为 `"abc"` / `-1` / `True` / `1.5` | `None`,且**不抛异常** |
| 非流式 body 含 `model: "MiniMax-Text-01-250321"` | `model_reported` 为该串 |
| 流式首个含 model 的 chunk 后又出现不同 model | 取**首个** |
| body 无 `model` / `model` 为 `""` | `None` |

**验证**: `conda run -n PolyGateway --no-capture-output pytest tests/unit/test_openai_compat.py -v` → PASS(新增用例先失败后通过)。

**提交**: `feat: collect provider cache tokens and reported model in transport`

## T3. RetryMW 搬运与缓存回放固化

- [ ] **改**: `src/polygateway/middleware/retry.py`

**行为**: `_build_response`(`:419-438`)追加 `cached_prompt_tokens=result.cached_prompt_tokens`、`model_reported=result.model_reported`。`model=source.model` **保持不变**——别名仍是主字段,真实版本是旁证(设计非目标 2)。

`middleware/cache.py` **不改一行**:`_RESPONSE_FIELDS` 由 `dataclasses.fields(LLMResponse)` 动态生成(`:26`)、`_serialize` 用 `asdict`(`:139`),新字段自动进出;决策 B1 要求命中时原样回放,而 `_rehydrate` 的覆写清单(`:113-119`)本就不含新字段,零改动即是正确行为。本任务用测试把它钉死。

**测试**:

- `tests/unit/test_retry.py`: transport 返回带两字段的 `TransportResult` → `chat()` 返回的 `LLMResponse` 上两字段一致;transport 未上报时为 `None`。
- `tests/unit/test_cache.py`: ① 带两字段的响应写入缓存再命中,回放值与原值相等且 `cache_hit=True`;② **旧格式兼容**——手工构造缺这两个键的缓存 JSON 塞进后端,命中后能正常 rehydrate 且两字段为 `None`(不得抛异常回源)。

**验证**: `conda run -n PolyGateway --no-capture-output pytest tests/unit/test_retry.py tests/unit/test_cache.py -v` → PASS。

**提交**: `feat: carry the new observability fields through retry and cache`

## T4. 缓存读取单价

- [ ] **改**: `src/polygateway/pricing.py`

**行为**:

```python
class ModelPrice:  # 追加第三档,可选
    cached_input_per_1m: float | None = None

def cost(self, model: str, prompt_tokens: int, completion_tokens: int,
         cached_prompt_tokens: int | None = None) -> float | None:
```

换算规则(设计 §4):配了缓存档**且** `cached_prompt_tokens` 为正 → `(prompt - cached) × input + cached × cached_input`;否则全额按 `input`(现状,逐位不变)。`cached > prompt` 时按 `prompt` 夹取并 `logger.warning` 一次(沿用 `_warned` 的去重思路,按 model 去重,防日志风暴),**绝不产生负成本**。

`from_file` 的 fail-loud 扩展:条目出现 `cached_input_per_1m` 键时必须可转 float 且非负,否则 `ValueError`;不出现该键 = 合法(旧价格表零改动)。`__post_init__` 同步校验非负。

顺带订正 `PricingTable` docstring(`pricing.py:36`)那句「cost() 是全库唯一换算点(经 TelemetryEmitter)」——实际有 `TelemetryEmitter`(`middleware/telemetry.py:137`)与 `embedding.py:419` 两个调用点(设计 §6 行为审计已声明)。**只改这一行注释,不做任何结构重构**。

**验收**: `embedding.py:419` 的三参调用形态**一行不改**仍可用;未配缓存档时,任意输入下 `cost()` 结果与改前逐位相等。

**测试**(`tests/unit/test_pricing.py`): ① 配缓存档 + 命中 → 成本严格低于全额且等于手算值;② 未配缓存档 + 命中 → 与不传该参数结果相等;③ `cached > prompt` → 结果等于全部按缓存价、非负、有 warning;④ `cached_prompt_tokens=None/0` → 全额;⑤ 三参旧调用签名可用;⑥ 价格表含 `cached_input_per_1m: -1` 或 `"x"` → `from_file` 抛 `ValueError`;⑦ 无该键的旧价格表照常加载。

**验证**: `conda run -n PolyGateway --no-capture-output pytest tests/unit/test_pricing.py -v` → PASS。

**提交**: `feat: support a cached input price tier in the pricing table`

## T5. 端口扩字段与后端补列

- [ ] **改**: `src/polygateway/ports.py`、`src/polygateway/telemetry/sqlite.py`、`src/polygateway/telemetry/postgres.py`

**行为**:

1. `TelemetryRecorder.record_llm_call`(`ports.py:250-271`)在 `cost` 之后追加 `cached_prompt_tokens: int | None` 与 `model_reported: str | None`,**不设默认值**(设计 §5:库外无第三方实现者)。同步该 Protocol 的「18 字段冻结」docstring。

2. 两个后端的 `_DDL` 加列(SQLite `INTEGER`/`TEXT`;PG `INTEGER`/`TEXT`,均可空、无默认值)。**新列在 DDL 里必须放在 `created_at` 之后(即表的最末尾),不得插在 `cost` 之后**——旧表走 `ALTER TABLE ADD COLUMN` 只能追加到末尾,若新建库把新列插在 `created_at` 前面,两条路径的物理列序就会分叉,而 `tests/integration/test_postgres_telemetry.py:117-128` 的 `test_schema_has_frozen_columns_in_order` 按 `ordinal_position` 逐位断言,且该表是与真实批跑共享的表、**严禁 DROP/TRUNCATE**(文件头隔离纪律),分叉后没有合规修法。

   `_COLUMNS` 在 `"cost"` 之后追加同名两项即可——`_INSERT` 是显式列名拼装(`sqlite.py:62-65`/`postgres.py:67-71`),`_COLUMNS` 只需与自身的 `row = tuple(...)` 自洽,**与 DDL 物理列序无关**。

3. **幂等补列**,按设计 D1 纪律执行:

- **SQLite**(`sqlite.py:74-84`):补列代码必须放在 `self._conn = conn` **之后**、用**独立 try**,且**首行必须守卫 `if self._conn is None: return`**——初始化 try 吞掉失败时 `self._conn` 仍是 `None`(局部 `conn` 甚至未绑定),无守卫的补列块会抛 `AttributeError`/`NameError`,这两者不被 `sqlite3.Error` 捕获,会直接逃出 `__init__`,打破「初始化失败静默降级」的对外契约(既有测试 `tests/unit/test_telemetry.py:132-135` `test_unwritable_path_degrades_silently` 会红)。守卫之后:`PRAGMA table_info(llm_calls)` 取现有列名集合,缺哪列补哪列;捕获 `sqlite3.Error` 时消息含 `duplicate column` 视为成功(多进程共库的 TOCTOU),其余记 warning。**绝不允许**因补列失败把 `self._conn` 置回 `None`——那会让整个 recorder 永久 no-op。
- **Postgres**(`postgres.py:_ensure_ready` 内、`_DDL` 执行之后):两条 `ALTER TABLE llm_calls ADD COLUMN IF NOT EXISTS ...`,共享既有 `_init_lock` 与 `except asyncio.CancelledError: raise` 结构。

**验收(降级语义不得改变)**: SQLite 侧 `except` 不得加宽(取消天然穿透);PG 侧 `CancelledError` 分支保持在最前;写入失败仍是逐行 warning 丢弃,不冒泡。

**必须同步改的测试(共 5 处)**:

| 位置 | 内容 | 漏改会怎样 |
|---|---|---|
| `tests/unit/test_telemetry.py:76` 起 `_record_minimal` | 手写 18 键 dict | **红**(`KeyError`) |
| `tests/integration/test_postgres_telemetry.py:81-105` `_record_minimal` | 同上 | **红** |
| `tests/unit/test_telemetry.py:18-40` `_EXPECTED_COLUMNS` | 19 项列序断言(含 `created_at`) | **红**;新列追加到 `created_at` **之后** |
| `tests/integration/test_postgres_telemetry.py:22-41` `_EXPECTED_COLUMNS` | 同上 | **红**;同上 |
| `tests/unit/test_ports.py:96` `_DummyRecorder` | 唯一写全签名的 fake | **不会红**(它只被 `:131` 的 `isinstance` 使用,`runtime_checkable` Protocol 只校验方法名不校验签名),但仍应同步以免误导后来者 |

前四处是本次仅有的天然拦截点;端口加参数**不会**带来编译期保护(本仓无 mypy,其余 8 个 fake 全是 `**fields`)。

**测试**:

- `tests/unit/test_telemetry.py`(SQLite):① 20 字段写入后可读回两个新列的值(含 `None`);② **旧表升级**——先用 18 列 DDL 手工建表,再实例化 `SQLiteRecorder`,写入成功且新列有值;③ **补列失败路径**(设计 §8 第 ③ 条,最危险的分支,不可用成功路径顶替)——构造一个 ALTER 必然失败的场景(把 `llm_calls` 建成同名 view,或注入在 ALTER 上抛 `sqlite3.OperationalError` 的连接),断言构造**不抛异常**、`recorder._conn` 仍非 `None`、后续 `record_llm_call` 不抛(降级为逐行 warning);④ 初始化路径不可写时仍静默降级(`test_unwritable_path_degrades_silently` 保持绿)。
- `tests/integration/test_postgres_telemetry.py`:① 20 字段写入 PG 并 `SELECT` 回读;② 18 列旧表经初始化后自动补列并写入成功。

**验证**: `conda run -n PolyGateway --no-capture-output pytest tests/unit/test_telemetry.py tests/unit/test_ports.py -v` → PASS;PG 部分 `conda run -n PolyGateway --no-capture-output pytest tests/integration/test_postgres_telemetry.py -v` → PASS。**PG/Redis 属共享后端,严禁与其他会话或钩子测试并跑**,起跑前确认无并发占用。

**提交**: **与 T6 合并为一次提交**,不得单独落地。理由:T5 落地后 emitter 仍只传 18 键,后端的 `row = tuple(fields[col] for col in _COLUMNS)` 会抛 `KeyError`,被 `_record` 的 `except Exception`(`middleware/telemetry.py:164-165`)吞成 warning → **该 commit 处于全量遥测静默丢失的状态**,且现有测试无一能捕获。提交信息见 T6。

## T6. Emitter 搬运与契约测试(与 T5 同一次提交)

- [ ] **改**: `src/polygateway/middleware/telemetry.py`

**行为**: `_record`(`:108-125`)新增两个参数并透传给 `record_llm_call`;三个入口各自提供取值——

| 入口 | `cached_prompt_tokens` | `model_reported` |
|---|---|---|
| `emit_attempt` | `response.cached_prompt_tokens if response else None` | 同左 |
| `emit_cache_hit` | `response.cached_prompt_tokens`(B1 原样回放) | 同左 |
| `emit_terminal_failure` | `None` | `None` |

cost 换算(`:137`)改为把 `cached_prompt_tokens` 传进 `self._pricing.cost(...)`。`cache_hit → 0.0` 与 `usage_source == "unavailable" → None` 两条短路的**先后顺序一字不动**(ARCHITECTURE §5.1 cost 口径不变式)。

**测试**(`tests/unit/test_telemetry.py`):

- **契约测试(不可省)**: 用记录 kwargs 的 fake recorder 跑一次 `emit_attempt`,断言 `set(kwargs) == set(sqlite._COLUMNS) == set(postgres._COLUMNS)`。理由:`row = tuple(fields[col] for col in _COLUMNS)` 位于两个后端 try 之外(`sqlite.py:90`/`postgres.py:121`),emitter 漏传字段会抛 `KeyError` 并被 `_record` 的 `except Exception` 吞成 warning → 静默丢遥测;现有 8 个 `**fields` 形态的 fake 一个都拦不住。
- 三个入口各记一行,断言新字段取值符合上表。
- cost 回归:配了缓存档且响应带 `cached_prompt_tokens` → 落库 cost 低于全额;缓存命中行 cost 仍为 `0.0`;`unavailable` 行仍为 `None`。

**验证**: `conda run -n PolyGateway --no-capture-output pytest tests/unit/test_telemetry.py -v` → PASS;随后 `make ci` 全绿(含 ruff 与 import-linter)。

**提交**(含 T5 全部改动): `feat: record the observability fields end to end through telemetry`

## T7. 文档同步与发版

- [ ] **改**: `research-wiki/ARCHITECTURE.md`、`CHANGELOG.md`、`.env.example`、`pyproject.toml`、`src/polygateway/__init__.py`、四处「18 字段冻结」措辞点、Gitea Wiki 站

**行为**:

1. `ARCHITECTURE.md`:§5.1 新增字段表补两行;**§7.8「必录字段」的行内清单**(`:452`)补两项——该文件不含字面「18 字段冻结」,§7.8 与 D8(`:202`)才是遥测字段的落点;§7.8 末条「`pricing.py` 维护 model →(input 单价, output 单价)表」同步第三档。补一条度量口径警示(与 cost 缺口同款):**统计供应商缓存命中率必须带 `WHERE cache_hit = false`**,否则缓存回放行会被重复计入。
2. 代码里的「18 字段/18 列」措辞共 **6 处**,全部订正(`grep -rn "18 字段\|18 列" src/ tests/` 可复核):`ports.py:248`、`middleware/telemetry.py:31`、`pricing.py:6`、`telemetry/sqlite.py:87`、`telemetry/postgres.py:9`(「18 列 schema 与 SQLite 版同名同序」)、`tests/unit/test_telemetry.py:1`。
3. `.env.example:56` 是仓内**唯一**的价格表格式说明(无独立模板文件),补 `cached_input_per_1m` 可选档与「不填即全额计价、库不猜折扣率」的说明。
4. 版本 bump `1.0.3` → `1.1.0`,**两处必须同步**(`pyproject.toml:7` 与 `src/polygateway/__init__.py:34`;`tests/unit/test_package.py:11` 会断言二者相等)。
5. `CHANGELOG.md` 顶部新增 `## 1.1.0` 段,沿用既有写法(先讲问题、再讲变更、点明下游要读什么):两个新字段的语义与 `None`/`0` 之别、`cache_hit` 与供应商 prompt cache 的区分、遥测表新增两列与自动补列、价格表可选缓存档、度量口径的 `cache_hit = false` 约束。
6. Gitea Wiki 站(需单独 `git clone https://gitea.iomgaa.online/iomgaa/PolyGateway.wiki.git`)按 `docs-convention.md` §2 清单同步:`参考-公共API`(LLMResponse 字段表)、`参考-配置键`(价格表格式)、`指南-遥测与成本`(新列与成本校正口径)、`Home.md` 版本号与安装命令、`_Sidebar.md` 如有结构变化。

**验收**: 版本 bump 的提交**不允许单独存在**(docs-convention §2 门),必须与 wiki/CHANGELOG 同步在同一次交付内。

**验证**: `conda run -n PolyGateway --no-capture-output pytest tests/unit/test_package.py -v` → PASS;`make ci` 全绿。

**提交**: `chore: release 1.1.0 with the response observability fields`

---

## 合并前门(逐条对应 CLAUDE.md §3)

- [ ] 每个行为变更都有「先失败后通过」的测试证据(T1-T6 各自的新增用例)。
- [ ] `make ci` 全绿(ruff + import-linter + pytest + 覆盖率)。
- [ ] 派**全新上下文**的 verifier subagent 独立验证(跨多文件,`verification-before-completion` 强制档)。
- [ ] 合并前整分支代码审查(`requesting-code-review`)。
- [ ] Gitea Issue #3 的关闭说明:两个字段的最终名字与语义、缓存命中行的回放口径、遥测新列与补列行为。