test: verify the thinking switch against the live API (issue #5, #6)

A sixteen-row matrix over 127 real calls: disable and enable on MiniMax-M3 in both streaming and non-streaming mode, extra_body winning over the profile slot, qwen and deepseek still disabling correctly, a drift sentinel that re-derives every registered capability from live behaviour, and the assembly guard refusing the models that cannot comply. Two judgement criteria had to be corrected by the data they were meant to judge. Output length cannot separate the two regimes at all -- the disabled runs reach 46 tokens when the model narrates its working in the visible answer, and the enabled runs drop to 13 when medium effort barely thinks. reasoning_tokens separates them cleanly in both directions, which is precisely what issue #6 was collected for. A second anchor compares prompt_tokens between the two regimes: the vendor injects a reasoning instruction when thinking is on, so the input side grows, and comparing the two runs relatively avoids hardcoding any vendor number. Provider names are mapped explicitly rather than guessed from the model string; guessing had silently skipped the qwen row behind a "source unavailable" reason that was not true.
2026-08-02 06:55:38 -04:00
parent 82f4ec4910
commit 4c135075b3
6 changed files with 437 additions and 5 deletions
@@ -0,0 +1,385 @@
+"""真实 API 验证推理开关与 reasoning_tokens(issue #5 + #6)。
+
+本组用例**必须真跑**: 改动的正确性与具体模型强相关,mock 只能验证代码路径,
+验证不了"这个参数在这个模型上到底关没关掉推理"。
+
+两条判据纪律(来自 findings §4c 的实测教训):
+
+1. **判别量只能是 `reasoning_tokens`,不能是 `completion_tokens`。** 两档的输出
+   长度分布**是重叠的**: 实测关闭档最高 46 token(模型偶尔把解题过程写进正文),
+   开启档最低 13 token(medium 档想得少的那几轮),按长度阈值判两边都会误判。
+   而 `reasoning_tokens` 在同一批 30 轮里干净分开——关闭 15/15 为 None,
+   开启 15/15 大于 0。
+2. **另配一个不含魔数的确定性锚点**(见 L2b): 同一模型上,关闭档的
+   `prompt_tokens` 严格小于开启档——供应商在开启时注入了推理指令,输入侧
+   token 数随之变大。这是相对比较,不硬编码任何具体数值。
+3. **关闭方向要求每轮满足,开启方向只要求多数轮满足。** 中转在上游不返回
+   usage 时会本地补算并吃掉 `completion_tokens_details`(findings §4c),
+   开启方向因此可能偶尔观测不到;关闭方向不受影响。
+
+源不可用一律 `skip` 并在报告中记为「未覆盖」,**绝不静默计入通过**。
+"""
+
+import dataclasses
+import json
+import os
+from collections import Counter
+from datetime import datetime
+from pathlib import Path
+
+import pytest
+from dotenv import dotenv_values
+
+from polygateway import GatewayClient, GatewaySettings
+from polygateway.errors import RequestRejectedError
+from polygateway.providers import DEFAULT_CAPABILITIES, get_capability
+
+_ENV = {k: v for k, v in {**dotenv_values(".env"), **os.environ}.items() if v is not None}
+_HAS_SOURCE = any(k.split("__")[0] == "LLM" and k.endswith("__API_KEY") for k in _ENV)
+
+pytestmark = pytest.mark.skipif(
+    not _HAS_SOURCE, reason="需真实网关凭据: 在 .env 配置 LLM__{PROVIDER}__1__*(本组必须真跑)"
+)
+
+_OUT_DIR = Path("tests/outputs/e2e")
+_ROUNDS = int(os.environ.get("PGW_E2E_THINKING_ROUNDS", "10"))
+
+# 需要一点推理才能答对,但答案极短: 关掉推理时 completion 稳定在个位数,
+# 开着时则是几百——两档之间隔着一个数量级,判据不必卡在噪声里
+_PROMPT = "一个笼子里有若干鸡和兔,共 35 个头、94 只脚。鸡和兔各有多少只?只输出两个数字。"
+
+_ON_MIN_COMPLETION = 100
+"""仅用于 `reasoning_tokens` 被中转吃掉时的退路;关闭方向不设长度门(见 `_reasoning_off`)。"""
+
+_ROWS: list[dict] = []
+
+# 显式映射,不按模型名猜 provider —— 那正是 D11 要消灭的东西(providers.py 开篇)。
+# 漏登记会被 test_every_capability_has_a_provider_mapping 当场抓住,而不是
+# 在 L8 里被"源不可用"这个假理由吞掉
+_MODEL_PROVIDER = {
+    "MiniMax-M3": "minimax",
+    "MiniMax-M2.7": "minimax",
+    "MiniMax-M2.5": "minimax",
+    "qwen3.7-plus": "qwen",
+    "deepseek-v4-pro": "deepseek",
+}
+
+
+def _base_settings() -> GatewaySettings:
+    # 强制关缓存: 多轮测量要求每一轮都真的打到供应商,命中缓存会把后续轮次
+    # 变成对第一轮的回放,整组判据随之失效
+    return GatewaySettings.from_env("LLM", env={**_ENV, "PGW_CACHE_BACKEND": "none"})
+
+
+def _settings(**source_overrides) -> GatewaySettings:
+    base = _base_settings()
+    source = dataclasses.replace(base.sources[0], **source_overrides)
+    return dataclasses.replace(base, sources=(source,))
+
+
+async def _run_rounds(rounds: int, *, stream: bool = True, **source_overrides) -> list[dict]:
+    """跑 N 轮真实调用,返回逐轮观测;任一轮抛错即向上冒泡由用例决定处置。"""
+    client = GatewayClient.from_settings(_settings(**source_overrides))
+    observations = []
+    try:
+        for i in range(rounds):
+            resp = await client.chat(
+                [{"role": "user", "content": _PROMPT}],
+                stream=stream,
+                # 每轮独立 salt: 即便某层缓存意外开着也不会回放
+                cache_salt=f"thinking-live-{i}",
+            )
+            observations.append(
+                {
+                    "round": i + 1,
+                    "prompt_tokens": resp.prompt_tokens,
+                    "completion_tokens": resp.completion_tokens,
+                    "reasoning_tokens": resp.reasoning_tokens,
+                    "content": resp.content[:60],
+                }
+            )
+    finally:
+        await client.aclose()
+    return observations
+
+
+def _record(matrix_id: str, desc: str, status: str, detail, observations=None) -> None:
+    _ROWS.append(
+        {
+            "matrix": matrix_id,
+            "desc": desc,
+            "status": status,
+            "detail": detail,
+            "observations": observations or [],
+        }
+    )
+
+
+def _reasoning_off(obs: dict) -> bool:
+    """关闭方向: 只看 reasoning_tokens。
+
+    **刻意不设 completion_tokens 上限**: 实测关闭档偶尔会到 46 token(模型没照做
+    "只输出两个数字",把解题过程写进了正文),而那是正文不是推理。加长度门只会
+    把这种正常波动误判成"没关掉"。
+    """
+    return obs["reasoning_tokens"] in (None, 0)
+
+
+def _reasoning_on(obs: dict) -> bool:
+    """开启方向: 有 reasoning_tokens 就以它为准,它是本次改动引入的直接判据。
+
+    不能拿 completion_tokens 当开启方向的主判据: medium 档的推理量方差极大
+    (实测 15 轮跨 7-170 token),按长度阈值判会把"推理了但想得少"误判成没推理。
+    仅当中转吃掉了 ctd(reasoning_tokens is None)才退回长度判据。
+    """
+    reasoning = obs["reasoning_tokens"]
+    if reasoning is not None:
+        return reasoning > 0
+    return obs["completion_tokens"] > _ON_MIN_COMPLETION
+
+
+def _skip_if_unreachable(exc: Exception, matrix_id: str, desc: str):
+    """源不可用(渠道下线/模型未开通)→ 跳过并记为未覆盖,不伪装成通过。"""
+    _record(matrix_id, desc, "SKIP(源不可用)", str(exc)[:200])
+    pytest.skip(f"{matrix_id} 源不可用,已记为未覆盖: {str(exc)[:120]}")
+
+
+@pytest.fixture(scope="module", autouse=True)
+def _write_report():
+    yield
+    _OUT_DIR.mkdir(parents=True, exist_ok=True)
+    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+    path = _OUT_DIR / f"test_thinking_live_{ts}.md"
+    lines = [
+        "# 推理开关与 reasoning_tokens 真实 API 验证",
+        "",
+        f"- 时间: {ts}",
+        f"- 每档轮数: {_ROUNDS}",
+        "- 关闭判据: **每轮** reasoning_tokens in (None, 0);刻意不设输出长度上限"
+        "(两档的 completion 分布重叠: 实测关闭档最高 46、开启档最低 13)",
+        f"- 开启判据: **多数轮** reasoning_tokens > 0(被中转吃掉时退回 completion > {_ON_MIN_COMPLETION})",
+        "- 确定性锚点(L2b): 关闭档 prompt_tokens 最大值 < 开启档最小值,相对比较无魔数",
+        "",
+        "## 矩阵结论",
+        "",
+        "| 矩阵 | 场景 | 结论 | 说明 |",
+        "|---|---|---|---|",
+    ]
+    total_calls = 0
+    for row in _ROWS:
+        detail = str(row["detail"]).replace("|", "\\|").replace("\n", " ")[:160]
+        lines.append(f"| {row['matrix']} | {row['desc']} | {row['status']} | {detail} |")
+        total_calls += len(row["observations"])
+    lines += ["", f"**总真实调用次数: {total_calls}**", "", "## 逐轮原始观测", ""]
+    for row in _ROWS:
+        if not row["observations"]:
+            continue
+        lines += [f"### {row['matrix']} — {row['desc']}", "", "```json"]
+        lines.append(json.dumps(row["observations"], ensure_ascii=False, indent=2))
+        lines += ["```", ""]
+    uncovered = [r["matrix"] for r in _ROWS if r["status"].startswith("SKIP")]
+    if uncovered:
+        lines += ["## 未覆盖", "", f"以下矩阵行未跑到: {', '.join(uncovered)}", ""]
+    path.write_text("\n".join(lines), encoding="utf-8")
+    print(f"\n[e2e 报告] {path}")
+
+
+class TestMiniMaxM3:
+    """M3 是唯一实测可关闭推理的 MiniMax 模型,修复的地基压在它身上。"""
+
+    async def test_l1_disable_actually_disables(self):
+        obs = await _run_rounds(_ROUNDS, model="MiniMax-M3", enable_thinking=False)
+        offs = [o for o in obs if _reasoning_off(o)]
+        _record(
+            "L1",
+            "enable_thinking=False(流式)",
+            "PASS" if len(offs) == len(obs) else "FAIL",
+            f"{len(offs)}/{len(obs)} 轮确认未推理",
+            obs,
+        )
+        assert len(offs) == len(obs), f"关闭方向要求每轮满足: {obs}"
+
+    async def test_l2_enable_actually_enables(self):
+        obs = await _run_rounds(_ROUNDS, model="MiniMax-M3", enable_thinking=True)
+        ons = [o for o in obs if _reasoning_on(o)]
+        _record(
+            "L2",
+            "enable_thinking=True(流式,注入 medium)",
+            "PASS" if len(ons) * 2 > len(obs) else "FAIL",
+            f"{len(ons)}/{len(obs)} 轮观察到推理",
+            obs,
+        )
+        assert len(ons) * 2 > len(obs), f"开启方向要求多数轮满足: {obs}"
+
+    async def test_l2b_off_and_on_are_distinguishable_without_magic_numbers(self):
+        """确定性锚点: 开启档的 prompt_tokens 严格大于关闭档。
+
+        供应商在开启推理时会向模板注入推理指令,输入侧 token 数随之变大。这是
+        本组唯一不依赖输出侧噪声的证据,且是相对比较——不硬编码任何具体数值,
+        供应商改模板也不会让它假红。
+        """
+        rounds = max(3, _ROUNDS // 3)
+        off = await _run_rounds(rounds, model="MiniMax-M3", enable_thinking=False)
+        on = await _run_rounds(rounds, model="MiniMax-M3", enable_thinking=True)
+        off_max = max(o["prompt_tokens"] for o in off)
+        on_min = min(o["prompt_tokens"] for o in on)
+        _record(
+            "L2b",
+            "关闭/开启的 prompt_tokens 可分",
+            "PASS" if off_max < on_min else "FAIL",
+            f"关闭档最大 {off_max} < 开启档最小 {on_min}",
+            off + on,
+        )
+        assert off_max < on_min, (
+            f"两档的 prompt_tokens 未分开(关闭最大 {off_max},开启最小 {on_min}): 注入可能没到达模型"
+        )
+
+    async def test_l3_no_opinion_is_the_model_default(self):
+        obs = await _run_rounds(_ROUNDS, model="MiniMax-M3", enable_thinking=None)
+        _record("L3", "enable_thinking=None(不干预,基线)", "PASS", "仅记录基线,不断言方向", obs)
+        assert len(obs) == _ROUNDS
+
+    async def test_l4_extra_body_overrides_the_profile(self):
+        """profile 注入 none,extra_body 要求 high —— 后者必须赢(优先级不可调换)。
+
+        判据是行为而非报文: 若 extra_body 没赢,拿到的就是 none 的结果(不推理)。
+        """
+        rounds = max(3, _ROUNDS // 2)
+        obs = await _run_rounds(
+            rounds,
+            model="MiniMax-M3",
+            enable_thinking=False,
+            extra_body={"reasoning_effort": "high"},
+        )
+        ons = [o for o in obs if _reasoning_on(o)]
+        _record(
+            "L4",
+            "extra_body 覆盖 profile 注入",
+            "PASS" if len(ons) * 2 > len(obs) else "FAIL",
+            f"{len(ons)}/{len(obs)} 轮观察到推理(证明 high 生效而非 none)",
+            obs,
+        )
+        assert len(ons) * 2 > len(obs), f"extra_body 未能覆盖 profile: {obs}"
+
+    async def test_l5_non_stream_path_matches_stream(self):
+        """非流式快路径独立于流式实现,采集与注入都要各自验一遍。"""
+        rounds = max(3, _ROUNDS // 2)
+        off = await _run_rounds(rounds, stream=False, model="MiniMax-M3", enable_thinking=False)
+        on = await _run_rounds(rounds, stream=False, model="MiniMax-M3", enable_thinking=True)
+        offs = [o for o in off if _reasoning_off(o)]
+        ons = [o for o in on if _reasoning_on(o)]
+        ok = len(offs) == len(off) and len(ons) * 2 > len(on)
+        _record(
+            "L5",
+            "非流式路径重跑 L1/L2",
+            "PASS" if ok else "FAIL",
+            f"关闭 {len(offs)}/{len(off)} 轮,开启 {len(ons)}/{len(on)} 轮",
+            off + on,
+        )
+        assert len(offs) == len(off), f"非流式关闭方向未满足: {off}"
+        assert len(ons) * 2 > len(on), f"非流式开启方向未满足: {on}"
+
+
+class TestOtherProviders:
+    """qwen / deepseek 的 profile 是既有实现,本组防的是"改 minimax 时误伤它们"。"""
+
+    @pytest.mark.parametrize(
+        ("matrix", "provider", "model"),
+        [("L6", "qwen", "qwen3.7-plus"), ("L7", "deepseek", "deepseek-v4-pro")],
+    )
+    async def test_existing_profiles_still_disable(self, matrix, provider, model):
+        desc = f"{provider} enable_thinking=False"
+        try:
+            obs = await _run_rounds(_ROUNDS, provider=provider, model=model, enable_thinking=False)
+        except Exception as exc:  # 渠道未开通/下线: 记为未覆盖
+            _skip_if_unreachable(exc, matrix, desc)
+        offs = [o for o in obs if _reasoning_off(o)]
+        _record(
+            matrix,
+            desc,
+            "PASS" if len(offs) == len(obs) else "FAIL",
+            f"{len(offs)}/{len(obs)} 轮确认未推理",
+            obs,
+        )
+        assert len(offs) == len(obs), f"{provider} 关闭方向未满足: {obs}"
+
+
+class TestCapabilityDrift:
+    """L8 漂移哨兵: 能力表过期是必然事件,这里是它的过期告警。"""
+
+    def test_every_capability_has_a_provider_mapping(self):
+        """能力表新增条目必须同步本测试的映射,否则该行会被静默跳过。"""
+        missing = sorted(set(DEFAULT_CAPABILITIES) - set(_MODEL_PROVIDER))
+        assert not missing, f"这些模型缺 provider 映射,L8 会漏测: {missing}"
+
+    @pytest.mark.parametrize("model", sorted(DEFAULT_CAPABILITIES))
+    async def test_declared_capability_matches_reality(self, model):
+        cap = get_capability(model)
+        provider = _MODEL_PROVIDER[model]
+        rounds = max(3, _ROUNDS // 2)
+        desc = f"{model} 声明 can_disable={cap.can_disable}"
+        if not cap.can_disable:
+            # 声明关不掉: 装配期就该炸,炸了即与声明一致(不必真调用)
+            with pytest.raises(ValueError, match=model):
+                GatewayClient.from_settings(
+                    _settings(provider=provider, model=model, enable_thinking=False)
+                )
+            _record("L8", desc, "PASS", "装配期按声明拒绝,与实测一致")
+            return
+        try:
+            obs = await _run_rounds(rounds, provider=provider, model=model, enable_thinking=False)
+        except Exception as exc:
+            _skip_if_unreachable(exc, "L8", desc)
+        offs = [o for o in obs if _reasoning_off(o)]
+        verdict = Counter(_reasoning_off(o) for o in obs)
+        _record(
+            "L8",
+            desc,
+            "PASS" if len(offs) == len(obs) else "FAIL(能力表已漂移)",
+            f"实测 {dict(verdict)};声明 can_disable=True 要求每轮关闭",
+            obs,
+        )
+        assert len(offs) == len(obs), (
+            f"能力表漂移: {model} 声明可关闭推理,实测未关掉 —— 请复测后更新 DEFAULT_CAPABILITIES"
+        )
+
+
+class TestAssemblyGuardAgainstRealConfig:
+    """L9: 纯本地,但用的是 .env 里的真实配置形态,防"守卫只在合成配置上生效"。"""
+
+    def test_l9_m27_rejected_at_assembly(self):
+        with pytest.raises(ValueError, match="MiniMax-M2.7"):
+            GatewayClient.from_settings(
+                _settings(provider="minimax", model="MiniMax-M2.7", enable_thinking=False)
+            )
+        _record("L9", "M2.7 + enable_thinking=False", "PASS", "装配期报错,未发出任何请求")
+
+    def test_l9_unknown_shape_rejected_at_assembly(self):
+        with pytest.raises(ValueError, match="register_provider"):
+            GatewayClient.from_settings(
+                _settings(provider="openai", model="kimi-k3", enable_thinking=False)
+            )
+        _record("L9", "provider=openai 形态未知", "PASS", "装配期报错并指路")
+
+    async def test_transport_layer_rejects_when_guard_is_bypassed(self):
+        """构造函数全量注入这条路绕过装配守卫,transport 必须兜住并归四分类。"""
+        settings = _settings(provider="minimax", model="MiniMax-M2.7", enable_thinking=False)
+        client = GatewayClient.from_settings(
+            dataclasses.replace(
+                settings, sources=(dataclasses.replace(settings.sources[0], enable_thinking=None),)
+            )
+        )
+        try:
+            # 装配用 None 绕过守卫,再把源换成 False 直接喂给 transport
+            bad = dataclasses.replace(settings.sources[0], enable_thinking=False)
+            with pytest.raises(RequestRejectedError, match="MiniMax-M2.7"):
+                await client._terminal._transport.complete(
+                    messages=[{"role": "user", "content": _PROMPT}],
+                    source=bad,
+                    stream=True,
+                    overlay={},
+                    call_id="e2e-guard",
+                )
+        finally:
+            await client.aclose()
+        _record("L9", "绕过装配守卫时 transport 兜底", "PASS", "RequestRejectedError,属四分类")