"""推理裁定与对账的行为测试(issue #16/#17 设计 §4-§5)。 判据来自 2026-08-25 实测(findings): MiniMax-M3 在开启档流式路径下返回 185 字符 推理正文却不上报 `completion_tokens_details`,而 qwen/deepseek 两者都报。库因此 不能把任何单一信号当权威——本组用例逐条钉死"哪个信号该赢"。 """ import pytest from loguru import logger from polygateway.providers import get_provider from polygateway.thinking import ( DEFAULT_CAPABILITIES, ThinkingCapability, get_capability, observe_thinking, reconcile_thinking, register_capability, resolve_thinking, ) from polygateway.types import ThinkingObservation def _warnings(): """捕获库发出的 WARNING;loguru 不经标准 logging,pytest 的 caplog 抓不到。""" messages: list[str] = [] sink_id = logger.add(messages.append, level="WARNING") return messages, sink_id class TestObserveThinking: """三态裁定: 证据硬度决定优先级,无信号一律 UNKNOWN。""" def test_reasoning_text_alone_proves_it_happened(self): """推理正文是事实本身: 上游不报 token 数也照样成立(M3 流式实测形态)。""" assert ( observe_thinking(thinking="先解方程 x+y=35", reasoning_tokens=None) is ThinkingObservation.OBSERVED ) def test_blank_text_is_not_evidence(self): """纯空白正文不算证据: 网关响应是外部输入,truthy 判据会把空格计成推理(P5)。""" assert ( observe_thinking(thinking=" \n\t ", reasoning_tokens=None) is ThinkingObservation.UNKNOWN ) def test_positive_token_count_proves_it_happened(self): """无正文但上游报了推理用量(qwen 非流式形态)。""" assert observe_thinking(thinking="", reasoning_tokens=205) is ThinkingObservation.OBSERVED def test_zero_token_count_is_positive_evidence_of_absence(self): """`0` 是"上报了且为零",与"没上报"语义不同,故是 ABSENT 而非 UNKNOWN。""" assert observe_thinking(thinking="", reasoning_tokens=0) is ThinkingObservation.ABSENT def test_no_signal_at_all_stays_unknown(self): """M3 非流式开启档的真实形态: 推理已计费却既无正文也无 token 数。 判成 ABSENT 就是伪装成"没推理"——正是 issue #16/#17 的病根。 """ assert observe_thinking(thinking="", reasoning_tokens=None) is ThinkingObservation.UNKNOWN def test_text_outranks_a_zero_count(self): """转述与事实冲突时事实赢: 正文在,`reasoning_tokens=0` 不能翻案。""" assert ( observe_thinking(thinking="想了想", reasoning_tokens=0) is ThinkingObservation.OBSERVED ) class TestThinkingObservationEnum: def test_values_are_stable_strings(self): """取值进遥测落库,改名即历史数据断层。""" assert ThinkingObservation.OBSERVED == "observed" assert ThinkingObservation.ABSENT == "absent" assert ThinkingObservation.UNKNOWN == "unknown" def test_enum_lives_in_the_innermost_layer(self): """枚举必须定义在 `types.py`(最内层)。 它是 `LLMResponse` 的字段类型;定义在决策层 `thinking.py` 会让 `types.py` 反向 import 决策模块,违反 P7 依赖铁律(import-linter 契约执法)。 """ assert ThinkingObservation.__module__ == "polygateway.types" @pytest.mark.parametrize("bogus", ["", "OBSERVED", "yes", "none"]) def test_unknown_strings_are_rejected(bogus): """非法值必须抛 ValueError: 缓存回放靠它把污染数据挡成"未命中"(设计 §6)。""" with pytest.raises(ValueError): ThinkingObservation(bogus) class TestThinkingCapability: """issue #5: 能力按 model 登记——同一 provider 内部代际差异是决定性的。""" def test_registered_models_carry_evidence(self): """登记必须附实测证据: 表会过期,没有出处就无从判断该不该信。""" for model in ("MiniMax-M3", "MiniMax-M2.7", "MiniMax-M2.5"): cap = get_capability(model) assert cap is not None and cap.evidence.strip() def test_m3_can_disable_but_m2x_cannot(self): assert get_capability("MiniMax-M3").can_disable is True assert get_capability("MiniMax-M2.7").can_disable is False assert get_capability("MiniMax-M2.5").can_disable is False def test_unregistered_model_is_unknown(self): assert get_capability("some-brand-new-model") is None def test_register_capability_is_pure(self): table = register_capability("x-1", ThinkingCapability(True, "实测")) assert get_capability("x-1", table=table) is not None assert get_capability("x-1") is None # 默认表未被污染 def test_default_capabilities_mapping_is_read_only(self): with pytest.raises(TypeError): DEFAULT_CAPABILITIES["hack"] = None # type: ignore[index] class TestResolveThinking: """五条判定规则(顺序即语义);设计 §5 真值表。""" def test_rule1_none_injects_nothing(self): got = resolve_thinking(get_provider("minimax"), None, None, model="MiniMax-M3") assert got == {} @pytest.mark.parametrize("enable", [True, False]) def test_rule2_unknown_shape_raises_and_points_the_way(self, enable): with pytest.raises(ValueError, match="register_provider") as exc: resolve_thinking(get_provider("openai"), None, enable, model="kimi-k3") assert "extra_body" in str(exc.value) def test_rule3_unregistered_model_warns_but_passes(self): messages, sink_id = _warnings() try: got = resolve_thinking(get_provider("minimax"), None, False, model="MiniMax-M9") finally: logger.remove(sink_id) assert got == {"reasoning_effort": "none"} assert any("MiniMax-M9" in m for m in messages) def test_rule4_cannot_disable_raises_with_the_model_name(self): cap = get_capability("MiniMax-M2.7") with pytest.raises(ValueError, match="MiniMax-M2.7"): resolve_thinking(get_provider("minimax"), cap, False, model="MiniMax-M2.7") def test_rule4_only_blocks_the_off_direction(self): """关不掉 ≠ 开不了: M2.x 默认就在推理,开的方向不该被拦。""" cap = get_capability("MiniMax-M2.7") got = resolve_thinking(get_provider("minimax"), cap, True, model="MiniMax-M2.7") assert got == {"reasoning_effort": "medium"} def test_rule5_normal_path(self): cap = get_capability("MiniMax-M3") assert resolve_thinking(get_provider("minimax"), cap, False, model="MiniMax-M3") == { "reasoning_effort": "none" } def test_unknown_shape_beats_capability_check(self): """第 2 步先于第 4 步: 形态未知时无从注入,能力如何无关紧要。""" cap = ThinkingCapability(can_disable=False, evidence="构造") with pytest.raises(ValueError, match="register_provider"): resolve_thinking(get_provider("openai"), cap, False, model="whatever") class TestReconcileThinking: """声明 × 观测对账(设计 §5): 矛盾出文案,不表态出 None。 文案本身是被断言对象——判定与日志分离正是为此: 告警内容可直接比对,不必 去解析日志格式。 """ _CAP = ThinkingCapability( can_disable=True, evidence="2026-08-02 实测 reasoning_effort=none 可关闭" ) def test_off_but_observed_with_a_registered_capability_blames_the_table(self): """已登记却实测推理了 = 能力表漂移: 必须附 evidence 与更新指路。""" msg = reconcile_thinking( enable_thinking=False, observation=ThinkingObservation.OBSERVED, capability=self._CAP, model="MiniMax-M3", ) assert msg is not None assert "MiniMax-M3" in msg assert "2026-08-02 实测 reasoning_effort=none 可关闭" in msg assert "register_capability" in msg def test_off_but_observed_unregistered_never_claims_a_table_entry(self): """未登记模型没有"能力表声称"这回事——说它就是撒谎。""" msg = reconcile_thinking( enable_thinking=False, observation=ThinkingObservation.OBSERVED, capability=None, model="MiniMax-M9", ) assert msg is not None assert "MiniMax-M9" in msg assert "能力表" not in msg assert "register_capability" in msg def test_registered_and_unregistered_wordings_differ(self): registered = reconcile_thinking( enable_thinking=False, observation=ThinkingObservation.OBSERVED, capability=self._CAP, model="MiniMax-M3", ) unregistered = reconcile_thinking( enable_thinking=False, observation=ThinkingObservation.OBSERVED, capability=None, model="MiniMax-M3", ) assert registered != unregistered @pytest.mark.parametrize("capability", [None, _CAP]) def test_on_but_absent_is_a_contradiction(self, capability): """上游明确上报未推理: 这是唯一的正面证伪,与能力表登记与否无关。""" msg = reconcile_thinking( enable_thinking=True, observation=ThinkingObservation.ABSENT, capability=capability, model="qwen3.7-plus", ) assert msg is not None assert "qwen3.7-plus" in msg @pytest.mark.parametrize("capability", [None, _CAP]) def test_on_but_unknown_admits_it_cannot_confirm(self, capability): """issue #17 的诚实版本: 明说"我注入了,但我看不见结果"。""" msg = reconcile_thinking( enable_thinking=True, observation=ThinkingObservation.UNKNOWN, capability=capability, model="MiniMax-M3", ) assert msg is not None assert "MiniMax-M3" in msg def test_off_and_unknown_stays_silent(self): """UNKNOWN 没有证伪力: 拿它报警等于每次关闭调用都喊(M3 关闭档恒落此档)。""" assert ( reconcile_thinking( enable_thinking=False, observation=ThinkingObservation.UNKNOWN, capability=self._CAP, model="MiniMax-M3", ) is None ) @pytest.mark.parametrize( "observation", [ThinkingObservation.OBSERVED, ThinkingObservation.ABSENT, ThinkingObservation.UNKNOWN], ) def test_no_request_no_grievance(self, observation): """调用方不表态,就无从谈"违背"。""" assert ( reconcile_thinking( enable_thinking=None, observation=observation, capability=self._CAP, model="MiniMax-M3", ) is None ) def test_on_and_observed_is_exactly_what_was_asked_for(self): assert ( reconcile_thinking( enable_thinking=True, observation=ThinkingObservation.OBSERVED, capability=self._CAP, model="MiniMax-M3", ) is None )