PolyGateway/tests/unit/test_thinking.py

"""推理裁定与对账的行为测试(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,
    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")