feat: collect reasoning_tokens from the provider usage payload (issue #6)

Reasoning tokens are already counted inside completion_tokens, so the
cost total was never wrong -- what was missing is the attribution: how
much of a call was spent thinking rather than answering.

LLMResponse and TransportResult each gain a trailing reasoning_tokens
field, and the telemetry port grows from 21 to 22 columns with the new
column appended in both backends so fresh and migrated schemas keep the
same physical order.

None means this particular call did not report the field, not that the
source never reports it: a relay that falls back to a local tokenizer
replaces the whole usage object and drops completion_tokens_details.
Downstream checks must therefore read "in (None, 0)"; no provider was
observed reporting a literal zero.

src/polygateway/middleware/retry.py

@@ -438,6 +438,7 @@ class RetryMW:
             usage_source=result.usage_source,
             cached_prompt_tokens=result.cached_prompt_tokens,
             model_reported=result.model_reported,
+            reasoning_tokens=result.reasoning_tokens,
         )
 
     async def _settle_and_release(self, permit: Permit, actual: int) -> None:

src/polygateway/middleware/telemetry.py

@@ -64,6 +64,7 @@ class TelemetryEmitter:
             error=error,
             cached_prompt_tokens=response.cached_prompt_tokens if response else None,
             model_reported=response.model_reported if response else None,
+            reasoning_tokens=response.reasoning_tokens if response else None,
             # 唯一有"生效源"的入口,故是唯一能并上 extra_body 的(设计决策 D)
             sampling=canonical_sampling_json(merge_sampling(source.extra_body, request.sampling)),
         )
@@ -90,6 +91,7 @@ class TelemetryEmitter:
             # 统计供应商缓存命中率必须带 WHERE cache_hit = false,否则重复计数。
             cached_prompt_tokens=response.cached_prompt_tokens,
             model_reported=response.model_reported,
+            reasoning_tokens=response.reasoning_tokens,
             # 由最外层 TelemetryMW 调用,手上没有 source。缓存命中行无损:
             # sampling 已进缓存 key,能命中即意味调用级参数与历史那次逐字相同
             sampling=canonical_sampling_json(request.sampling),
@@ -117,6 +119,7 @@ class TelemetryEmitter:
             error=error,
             cached_prompt_tokens=None,
             model_reported=None,
+            reasoning_tokens=None,
             # 无具体源,与 model/provider/source_name 置空同一先例(设计决策 D)
             sampling=canonical_sampling_json(request.sampling),
         )
@@ -142,6 +145,7 @@ class TelemetryEmitter:
         cached_prompt_tokens: int | None,
         model_reported: str | None,
         sampling: str | None,
+        reasoning_tokens: int | None,
     ) -> None:
         try:
             # 成本换算(M2 §6): 成功行按单价换算;缓存命中 0.0(未产生新调用);
@@ -182,6 +186,7 @@ class TelemetryEmitter:
                 cached_prompt_tokens=cached_prompt_tokens,
                 model_reported=model_reported,
                 sampling=sampling,
+                reasoning_tokens=reasoning_tokens,
             )
         except asyncio.CancelledError:
             raise

src/polygateway/ports.py

@@ -275,4 +275,5 @@ class TelemetryRecorder(Protocol):
         cached_prompt_tokens: int | None,
         model_reported: str | None,
         sampling: str | None,
+        reasoning_tokens: int | None,
     ) -> None: ...

src/polygateway/telemetry/postgres.py

@@ -42,7 +42,8 @@ CREATE TABLE IF NOT EXISTS llm_calls (
     created_at         TIMESTAMPTZ NOT NULL DEFAULT now(),
     cached_prompt_tokens INTEGER,
     model_reported       TEXT,
-    sampling             TEXT
+    sampling             TEXT,
+    reasoning_tokens     INTEGER
 );
 """
 
@@ -51,6 +52,7 @@ _BACKFILL = (
     ("cached_prompt_tokens", "ALTER TABLE llm_calls ADD COLUMN cached_prompt_tokens INTEGER"),
     ("model_reported", "ALTER TABLE llm_calls ADD COLUMN model_reported TEXT"),
     ("sampling", "ALTER TABLE llm_calls ADD COLUMN sampling TEXT"),
+    ("reasoning_tokens", "ALTER TABLE llm_calls ADD COLUMN reasoning_tokens INTEGER"),
 )
 
 # 探测现有列;尊重 search_path(to_regclass 按当前 search_path 解析)
@@ -81,6 +83,7 @@ _COLUMNS = (
     "cached_prompt_tokens",
     "model_reported",
     "sampling",
+    "reasoning_tokens",
 )
 
 _INSERT = (

src/polygateway/telemetry/sqlite.py

@@ -37,7 +37,8 @@ CREATE TABLE IF NOT EXISTS llm_calls (
     created_at         TEXT    NOT NULL DEFAULT (datetime('now')),
     cached_prompt_tokens INTEGER,
     model_reported       TEXT,
-    sampling             TEXT
+    sampling             TEXT,
+    reasoning_tokens     INTEGER
 );
 """
 
@@ -47,6 +48,7 @@ _BACKFILL_COLUMNS = (
     ("cached_prompt_tokens", "INTEGER"),
     ("model_reported", "TEXT"),
     ("sampling", "TEXT"),
+    ("reasoning_tokens", "INTEGER"),
 )
 
 _COLUMNS = (
@@ -71,6 +73,7 @@ _COLUMNS = (
     "cached_prompt_tokens",
     "model_reported",
     "sampling",
+    "reasoning_tokens",
 )
 
 _INSERT = (

src/polygateway/transports/openai_compat.py

@@ -177,6 +177,25 @@ def _coerce_cached_tokens(usage: Any) -> int | None:
     return cached
 
 
+def _coerce_reasoning_tokens(usage: Any) -> int | None:
+    """取 usage.completion_tokens_details.reasoning_tokens(issue #6);形态异常一律 None。
+
+    与 `_coerce_cached_tokens` 逐条同构(两者是 OpenAI 兼容 usage 里对称的一对):
+    `0` 如实保留、负数与非整数归 None、`bool` 显式排除。差别只在语义——本字段
+    的 None 是"**本次调用**未上报"而非"该源不上报": 中转在上游不返回 usage 时
+    会本地补算并整体替换 usage 对象,把 details 一并吃掉(findings §4c)。
+    """
+    if not isinstance(usage, dict):
+        return None
+    details = usage.get("completion_tokens_details")
+    if not isinstance(details, dict):
+        return None
+    reasoning = details.get("reasoning_tokens")
+    if isinstance(reasoning, bool) or not isinstance(reasoning, int) or reasoning < 0:
+        return None
+    return reasoning
+
+
 def _coerce_model_reported(value: Any) -> str | None:
     """取响应体的 model 字段(issue #3);非 str 或空白串一律 None,收口时去空白。
 
@@ -400,6 +419,7 @@ class OpenAICompatTransport:
             raw={"usage": sink.get("usage")},
             cached_prompt_tokens=_coerce_cached_tokens(sink.get("usage")),
             model_reported=_coerce_model_reported(sink.get("model")),
+            reasoning_tokens=_coerce_reasoning_tokens(sink.get("usage")),
         )
 
     def _check_done(
@@ -484,6 +504,7 @@ class OpenAICompatTransport:
             raw={"usage": body.get("usage")},
             cached_prompt_tokens=_coerce_cached_tokens(body.get("usage")),
             model_reported=_coerce_model_reported(body.get("model")),
+            reasoning_tokens=_coerce_reasoning_tokens(body.get("usage")),
         )
 
     async def aclose(self) -> None:

src/polygateway/types.py

@@ -99,6 +99,15 @@ class LLMResponse:
     model_reported: str | None = None
     """API 响应体里的 model 字段;None = 未上报。与 `model`(配置别名)可能
     分叉——供应商把别名指向新权重时,实验复现必须认这个串。"""
+    reasoning_tokens: int | None = None
+    """推理消耗的输出 token 数(含在 `completion_tokens` 内,故不影响成本总额,
+    只补归因;issue #6)。
+
+    `None` = **本次调用**未上报,**不是**"该源不上报"——中转网关在上游不返回
+    usage 时会用本地 tokenizer 补算并整体替换 usage 对象,把
+    `completion_tokens_details` 一并吃掉(findings §4c 实测同一请求 10 轮呈
+    6:4 双峰)。实测三家供应商在未推理时都是整个 details 缺失、无人上报 `0`,
+    故下游判据须为 `in (None, 0)`,写 `== 0` 的条件永远不成立。"""
 
 
 @dataclass(frozen=True)
@@ -151,9 +160,10 @@ class TransportResult:
     ttft_ms: float | None
     max_inter_token_ms: float | None
     raw: dict[str, Any]
-    # —— 可观测字段(issue #3;带默认值,非 OpenAI 兼容的 transport 可不填)——
+    # —— 可观测字段(issue #3/#6;带默认值,非 OpenAI 兼容的 transport 可不填)——
     cached_prompt_tokens: int | None = None
     model_reported: str | None = None
+    reasoning_tokens: int | None = None
 
 
 @dataclass(frozen=True)

tests/integration/test_postgres_telemetry.py

@@ -43,6 +43,7 @@ _EXPECTED_COLUMNS = [
     "cached_prompt_tokens",
     "model_reported",
     "sampling",
+    "reasoning_tokens",
 ]
 
 # run 级前缀: 同库并存的其他运行(迁移批跑/另一开发机)互不可见
@@ -107,6 +108,7 @@ async def _record_minimal(
         "cached_prompt_tokens": None,
         "model_reported": None,
         "sampling": None,
+        "reasoning_tokens": None,
     }
     fields.update(overrides)
     await recorder.record_llm_call(**fields)

tests/unit/test_openai_compat.py

@@ -394,6 +394,81 @@ class TestObservabilityFields:
         assert set(result.raw) == {"usage"}
 
 
+class TestReasoningTokens:
+    """issue #6: 推理消耗的输出 token,与 issue #3 的 cached_tokens 对称。
+
+    实测三家供应商在"未推理"时是整个 completion_tokens_details 缺失,无人上报
+    0;且中转在上游不返回 usage 时会本地补算并吃掉该对象。故 None 的语义是
+    "本次调用未上报",不是"该源不上报"(findings §4c)。
+    """
+
+    def _reasoning_usage(self, reasoning):
+        return {**_USAGE, "completion_tokens_details": {"reasoning_tokens": reasoning}}
+
+    async def test_stream_reads_reasoning_tokens(self):
+        def handler(request):
+            return _sse_stream(_chunk(content="ok"), _chunk(usage=self._reasoning_usage(7)))
+
+        result = await _complete(_transport_for(handler), _source())
+        assert result.reasoning_tokens == 7
+
+    async def test_non_stream_reads_reasoning_tokens(self):
+        def handler(request):
+            return httpx.Response(
+                200,
+                json={
+                    "choices": [{"message": {"content": "42"}}],
+                    "usage": self._reasoning_usage(7),
+                },
+            )
+
+        result = await _complete(_transport_for(handler), _source(), stream=False)
+        assert result.reasoning_tokens == 7
+
+    async def test_zero_reasoning_tokens_is_a_real_zero(self):
+        """0(上报了且确实没推理)与 None(本次未上报)必须可区分。"""
+
+        def handler(request):
+            return _sse_stream(_chunk(content="ok"), _chunk(usage=self._reasoning_usage(0)))
+
+        result = await _complete(_transport_for(handler), _source())
+        assert result.reasoning_tokens == 0
+
+    async def test_usage_without_details_is_none(self):
+        def handler(request):
+            return _sse_stream(_chunk(content="ok"), _chunk(usage=_USAGE))
+
+        result = await _complete(_transport_for(handler), _source())
+        assert result.reasoning_tokens is None
+
+    @pytest.mark.parametrize("bad", ["abc", -1, True, 1.5, None, [], {"x": 1}])
+    async def test_malformed_reasoning_tokens_degrade_to_none(self, bad):
+        """`True` 必须排除: Python 里 isinstance(True, int) 为真。"""
+
+        def handler(request):
+            return _sse_stream(_chunk(content="ok"), _chunk(usage=self._reasoning_usage(bad)))
+
+        result = await _complete(_transport_for(handler), _source())
+        assert result.reasoning_tokens is None
+
+    async def test_details_not_a_dict_is_none(self):
+        def handler(request):
+            usage = {**_USAGE, "completion_tokens_details": "oops"}
+            return _sse_stream(_chunk(content="ok"), _chunk(usage=usage))
+
+        result = await _complete(_transport_for(handler), _source())
+        assert result.reasoning_tokens is None
+
+    async def test_salvage_path_records_none_not_zero(self):
+        """打捞路径拿不到 usage 帧: 记 None(未知)而非 0(确定没推理)。"""
+
+        def handler(request):
+            return _sse_stream(_chunk(content="ok"), done=False)
+
+        result = await _complete(_transport_for(handler), _source(missing_done="salvage"))
+        assert result.reasoning_tokens is None
+
+
 class TestNonStreamFastPath:
     async def test_non_stream_parses_message(self):
         def handler(request):

tests/unit/test_ports.py

@@ -117,6 +117,7 @@ class _DummyRecorder:
         cached_prompt_tokens,
         model_reported,
         sampling,
+        reasoning_tokens,
     ) -> None: ...
 
 

tests/unit/test_retry.py

@@ -209,11 +209,13 @@ class TestObservabilityPassthrough:
             raw={},
             cached_prompt_tokens=64,
             model_reported="MiniMax-Text-01-250321",
+            reasoning_tokens=7,
         )
         mw, *_ = _harness([_src("a")], [result])
         resp = await mw(_REQ)
         assert resp.cached_prompt_tokens == 64
         assert resp.model_reported == "MiniMax-Text-01-250321"
+        assert resp.reasoning_tokens == 7
         # model 仍是配置别名: 真实版本是旁证,不顶替溯源主字段
         assert resp.model == "m"
 
@@ -221,6 +223,7 @@ class TestObservabilityPassthrough:
         mw, *_ = _harness([_src("a")], [_ok()])
         resp = await mw(_REQ)
         assert resp.cached_prompt_tokens is None and resp.model_reported is None
+        assert resp.reasoning_tokens is None
 
 
 class TestRetryAndFailover:

tests/unit/test_telemetry.py

@@ -1,4 +1,4 @@
-"""遥测子系统测试: SQLiteRecorder(21 列)+ TelemetryEmitter 单一 helper + TelemetryMW。"""
+"""遥测子系统测试: SQLiteRecorder(22 列)+ TelemetryEmitter 单一 helper + TelemetryMW。"""
 
 import asyncio
 import json
@@ -39,6 +39,7 @@ _EXPECTED_COLUMNS = [
     "cached_prompt_tokens",
     "model_reported",
     "sampling",
+    "reasoning_tokens",
 ]
 
 
@@ -102,6 +103,7 @@ async def _record_minimal(recorder, call_id="c1", **overrides):
         "cached_prompt_tokens": None,
         "model_reported": None,
         "sampling": None,
+        "reasoning_tokens": None,
     }
     fields.update(overrides)
     await recorder.record_llm_call(**fields)
@@ -158,6 +160,22 @@ class TestSQLiteRecorder:
         assert rows["c-zero"] == 0  # 真实零命中,读回仍是 0 而非 NULL
         assert rows["c-none"] is None
 
+    async def test_reasoning_tokens_column_round_trip(self, tmp_path):
+        """issue #6: 7 / 0 / None 三种值各自如实落库,0 与 NULL 不得混同。"""
+        recorder = SQLiteRecorder(tmp_path / "t.db")
+        await _record_minimal(recorder, call_id="r-some", reasoning_tokens=7)
+        await _record_minimal(recorder, call_id="r-zero", reasoning_tokens=0)
+        await _record_minimal(recorder, call_id="r-none", reasoning_tokens=None)
+        recorder.close()
+        rows = dict(
+            sqlite3.connect(tmp_path / "t.db")
+            .execute("SELECT call_id, reasoning_tokens FROM llm_calls")
+            .fetchall()
+        )
+        assert rows["r-some"] == 7
+        assert rows["r-zero"] == 0  # 上报了且确实没推理
+        assert rows["r-none"] is None  # 本次调用未上报
+
     async def test_sampling_column_round_trips(self, tmp_path):
         """issue #4: 采样参数落库,否则事后无法证明某批数据跑在什么温度下。"""
         recorder = SQLiteRecorder(tmp_path / "t.db")
@@ -284,6 +302,7 @@ class TestPostgresBackfillDiscipline:
         "cached_prompt_tokens",
         "model_reported",
         "sampling",
+        "reasoning_tokens",
     ]
 
     def _recorder(self, conn):
@@ -384,11 +403,12 @@ class TestEmitterObservabilityFields:
             source=_source(),
             call_id="cid-1",
             latency_ms=42,
-            response=_resp(cached_prompt_tokens=64, model_reported="m-real"),
+            response=_resp(cached_prompt_tokens=64, model_reported="m-real", reasoning_tokens=7),
             error=None,
         )
         assert rec.rows[0]["cached_prompt_tokens"] == 64
         assert rec.rows[0]["model_reported"] == "m-real"
+        assert rec.rows[0]["reasoning_tokens"] == 7
 
     async def test_failed_attempt_has_no_provider_facts(self):
         rec = _MemoryRecorder()
@@ -402,16 +422,19 @@ class TestEmitterObservabilityFields:
         )
         assert rec.rows[0]["cached_prompt_tokens"] is None
         assert rec.rows[0]["model_reported"] is None
+        assert rec.rows[0]["reasoning_tokens"] is None
 
     async def test_cache_hit_replays_the_recorded_values(self):
         """决策 B1: 命中行原样回放,故命中率统计必须带 WHERE cache_hit = false。"""
         rec = _MemoryRecorder()
         await TelemetryEmitter(rec).emit_cache_hit(
-            request=_REQ, response=_resp(cached_prompt_tokens=64, model_reported="m-real")
+            request=_REQ,
+            response=_resp(cached_prompt_tokens=64, model_reported="m-real", reasoning_tokens=7),
         )
         row = rec.rows[0]
         assert row["cache_hit"] is True
         assert row["cached_prompt_tokens"] == 64 and row["model_reported"] == "m-real"
+        assert row["reasoning_tokens"] == 7  # 与 cached 同口径原样回放
 
     async def test_terminal_failure_records_none(self):
         rec = _MemoryRecorder()
@@ -420,6 +443,7 @@ class TestEmitterObservabilityFields:
         )
         assert rec.rows[0]["cached_prompt_tokens"] is None
         assert rec.rows[0]["model_reported"] is None
+        assert rec.rows[0]["reasoning_tokens"] is None
 
 
 class TestEmitterSamplingColumn:

tests/unit/test_types.py

@@ -54,6 +54,7 @@ class TestLLMResponse:
         resp = LLMResponse("c", "t", "m", "p", 1, 2, 3, None, None, False, "cid")
         assert resp.cached_prompt_tokens is None
         assert resp.model_reported is None
+        assert resp.reasoning_tokens is None  # issue #6: 本次调用未上报
         filled = LLMResponse(
             "c",
             "t",
@@ -68,9 +69,11 @@ class TestLLMResponse:
             "cid",
             cached_prompt_tokens=0,
             model_reported="MiniMax-Text-01-250321",
+            reasoning_tokens=0,
         )
         assert filled.cached_prompt_tokens == 0  # 真实零命中,不得与 None 混同
         assert filled.model_reported == "MiniMax-Text-01-250321"
+        assert filled.reasoning_tokens == 0  # 上报了且确实没推理,不得与 None 混同
 
     def test_frozen(self):
         resp = LLMResponse("c", "t", "m", "p", 1, 2, 3, None, None, False, "cid")
@@ -247,6 +250,7 @@ class TestAuxTypes:
         assert s.raw["id"] == "x"
         # issue #3: 新字段带默认值,不填也能构造(OCR 等其他 transport 零改动)
         assert s.cached_prompt_tokens is None and s.model_reported is None
+        assert s.reasoning_tokens is None
 
 
 class TestOcrTypes: