feat: model the thinking switch as shape plus capability (issue #5)

enable_thinking=False was a no-op for minimax and openai sources: both
profiles had empty dicts on each side, so the payload update injected
nothing while the caller believed reasoning had been turned off. A
downstream project was blocked on exactly this.

The root cause is that an empty dict meant two different things -- "no
injection needed" and "we do not know how this provider spells it" --
and that a provider-level table cannot express what turned out to be a
per-model property. Live testing showed MiniMax-M3 can disable
reasoning via reasoning_effort while M2.7 and M2.5 cannot be disabled
at all, which two external registries independently confirm.

So the shape stays at provider level and a capability table joins it at
model level. Unknown, unsupported and no-opinion are now three distinct
values, and resolve_thinking is the single place they meet: it raises at
assembly time when a model cannot honour the request, warns and injects
for unregistered models, and injects silently otherwise. Every registered
capability carries the evidence it was derived from.

enable_thinking also joins the cache fingerprint, since it now really
does change the request body.

src/polygateway/client.py

@@ -25,7 +25,7 @@ from polygateway.middleware.retry import RetryMW
 from polygateway.middleware.structured import StructuredMW
 from polygateway.middleware.telemetry import TelemetryEmitter, TelemetryMW
 from polygateway.pricing import PricingTable
-from polygateway.providers import get_provider
+from polygateway.providers import get_capability, get_provider, resolve_thinking
 from polygateway.sources import (
     AdaptivePacer,
     HealthAwareSelector,
@@ -51,7 +51,7 @@ if TYPE_CHECKING:
         TelemetryRecorder,
         Transport,
     )
-    from polygateway.providers import ProviderProfile
+    from polygateway.providers import ProviderProfile, ThinkingCapability
     from polygateway.types import (
         BackpressurePolicy,
         RetryPolicy,
@@ -61,22 +61,52 @@ if TYPE_CHECKING:
 _T = TypeVar("_T")
 
 
+def _guard_thinking(
+    sources: list[SourceConfig],
+    profiles: list[ProviderProfile],
+    capabilities: Mapping[str, ThinkingCapability] | None,
+) -> None:
+    """装配期把不可满足的推理开关炸掉,而不是留到运行时(issue #5)。
+
+    与 transport 内的同一次判定不是重复: 那里兜的是"构造函数全量注入"这条路
+    (CLAUDE.md §4.5 的第二条装配路),而工厂路占 90% 场景,配置错误应当在装配期
+    就带着指路信息炸掉。`get_provider` 现在就是同一形态的双点调用。
+    """
+    for source, profile in zip(sources, profiles, strict=True):
+        resolve_thinking(
+            profile,
+            get_capability(source.model, table=capabilities),
+            source.enable_thinking,
+            model=source.model,
+        )
+
+
+def _fingerprint_mark(source: SourceConfig) -> str:
+    """单源的指纹标记;`enable_thinking` 仅在**表态时**追加。
+
+    只在表态时追加不是省事: 这样只配了 `extra_body` 的存量源字面量与 issue #4
+    时期逐字相同,升级本版本不会给它们平白来一次全量缓存冷启动。
+    """
+    parts: list[Any] = [source.model, dict(source.extra_body)]
+    if source.enable_thinking is not None:
+        parts.append(source.enable_thinking)
+    return json.dumps(parts, sort_keys=True, ensure_ascii=False)
+
+
 def build_model_fingerprint(sources: Iterable[SourceConfig]) -> str:
     """缓存 key 的模型身份: 多源 scope = 排序去重的 model 合集。
 
     配置级采样参数(`extra_body`)必须参与,否则把 temperature 从 0 改成 1
-    后重启仍会读到旧缓存(issue #4 设计决策 C)。全源 `extra_body` 皆空时
+    后重启仍会读到旧缓存(issue #4 设计决策 C)。`enable_thinking` 同理
-    字面量与历史实现逐字相同,不触发存量缓存冷启动。
+    (issue #5): 它一旦真正改变请求体,"关掉推理后重启"就会读到开着推理时
+    缓存的旧响应。全源两者皆未表态时字面量与历史实现逐字相同,不触发存量
+    缓存冷启动。
     """
     fingerprint = ",".join(sorted({s.model for s in sources}))
-    # 按 (model, extra_body) 而非源名摘要: 语义是"本 scope 会用哪些
+    # 按 (model, extra_body[, enable_thinking]) 而非源名摘要: 语义是"本 scope
-    # (模型, 解码参数)组合",改源名不该误触全量冷启动
+    # 会用哪些(模型, 请求形态)组合",改源名不该误触全量冷启动
     marks = sorted(
-        {
+        {_fingerprint_mark(s) for s in sources if s.extra_body or s.enable_thinking is not None}
-            json.dumps([s.model, dict(s.extra_body)], sort_keys=True, ensure_ascii=False)
-            for s in sources
-            if s.extra_body
-        }
     )
     if marks:
         digest = hashlib.sha256("".join(marks).encode("utf-8")).hexdigest()
@@ -241,11 +271,13 @@ class GatewayClient:
         cache: CacheBackend | None = None,
         telemetry: TelemetryRecorder | None = None,
         registry: Mapping[str, ProviderProfile] | None = None,
+        capabilities: Mapping[str, ThinkingCapability] | None = None,
         rng: Any = random.random,
     ) -> GatewayClient:
         """按配置装配;显式传入的后端实例即共享(None 项按配置自建私有实例)。"""
         sources = list(settings.sources)
         profiles = [get_provider(s.provider, registry=registry) for s in sources]
+        _guard_thinking(sources, profiles, capabilities)
         strategy, escalation = _build_structured(profiles)
         return cls(
             scope=settings.scope,
@@ -253,7 +285,7 @@ class GatewayClient:
             selector=_build_selector(settings.selector, rng=rng),
             limiter=limiter or _build_limiter(settings, sources),
             breaker=breaker or _build_breaker(settings),
-            transport=OpenAICompatTransport(registry=registry),
+            transport=OpenAICompatTransport(registry=registry, capabilities=capabilities),
             retry=settings.retry,
             backpressure=settings.backpressure,
             quota_full=settings.quota_full,
@@ -279,6 +311,7 @@ class GatewayClient:
         cache: CacheBackend | None = None,
         telemetry: TelemetryRecorder | None = None,
         registry: Mapping[str, ProviderProfile] | None = None,
+        capabilities: Mapping[str, ThinkingCapability] | None = None,
         env: Mapping[str, str] | None = None,
     ) -> GatewayClient:
         """从 .env/环境变量装配一个 scope 的 client(键名清单见 .env.example)。"""
@@ -289,6 +322,7 @@ class GatewayClient:
             cache=cache,
             telemetry=telemetry,
             registry=registry,
+            capabilities=capabilities,
         )
 
 

src/polygateway/providers.py

@@ -10,24 +10,37 @@ from dataclasses import dataclass
 from types import MappingProxyType
 from typing import Any
 
+from loguru import logger
+
 
 @dataclass(frozen=True)
 class ProviderProfile:
     """单个 provider 的能力与差异声明。
 
     thinking_on/thinking_off 分别是 `SourceConfig.enable_thinking` 为
-    True/False 时并入请求体的参数片段(None 时二者都不注入,用模型默认);
+    True/False 时并入请求体的参数片段(`enable_thinking` 为 None 时二者都不
-    strip_think_tags 声明响应 content 需剥离 ``<think>`` 标签(qwen 系);
+    注入,用模型默认);strip_think_tags 声明响应 content 需剥离 ``<think>``
-    supports_native_schema 供 D14 阶梯选择原生 response_format 策略。
+    标签(qwen 系);supports_native_schema 供 D14 阶梯选择原生 response_format。
 
-    注: 某个 provider 的两档若皆为空字典(如 openai/minimax),说明该 provider
+    两档各有三种取值,**语义互不重叠**(issue #5):
-    无已知的推理开关参数——此时 `enable_thinking` 对它**不产生任何效果**,
+
-    而非静默生效。需要下发自定义参数时用 `SourceConfig.extra_body`。
+    ==========  ==========================================================
+    ``{...}``   已知的注入片段
+    ``{}``      已知**无需注入**任何参数即处于该档
+    ``None``    **未知**: 本库不知道该 provider 如何表达这一档
+    ==========  ==========================================================
+
+    `None` 与 `{}` 必须分开: 二者曾同为空字典,导致 `enable_thinking=False`
+    对 minimax/openai 源静默失效——调用方以为关掉了推理,实际什么都没发生。
+    现在 `None` 会在装配期显式报错并指路 `register_provider` / `extra_body`。
+
+    注: 本类只声明**形态**(参数长什么样,按 provider 变);某个具体模型能否
+    关闭推理属**能力**(按 model 变),见 `ThinkingCapability`。
     """
 
     name: str
-    thinking_on: dict[str, Any]
+    thinking_on: Mapping[str, Any] | None
-    thinking_off: dict[str, Any]
+    thinking_off: Mapping[str, Any] | None
     strip_think_tags: bool
     supports_native_schema: bool = False
 
@@ -47,26 +60,151 @@ DEFAULT_PROFILES: Mapping[str, ProviderProfile] = MappingProxyType(
             thinking_off={"thinking": {"type": "disabled"}},
             strip_think_tags=False,
         ),
-        # 两档皆空 ⇒ `enable_thinking` 对本 provider **不产生任何效果**(调用方
+        # OpenAI 兼容基线段名: 实践中被复用为**任意**兼容厂商的兜底(下游把
-        # 以为关掉了实际没关)。真需要控制推理时经 `SourceConfig.extra_body` 下发
+        # kimi-k3 挂在 provider=openai 下),故不能下发任何厂商方言参数——发给
+        # 不认识它的厂商会 400。两档标 None(未知): 配了 enable_thinking 即在
+        # 装配期报错并指路,真 OpenAI 推理模型的用户走 register_provider
         "openai": ProviderProfile(
             name="openai",
-            thinking_on={},
+            thinking_on=None,
-            thinking_off={},
+            thinking_off=None,
             strip_think_tags=False,
         ),
-        # OpenAI 兼容基线,无已知注入差异;reasoning_content 由 transport 通用处理。
+        # 注入形态出处: 2026-08-02 经自建 new-api 中转实测(findings §2),
-        # 同上: 两档皆空 ⇒ `enable_thinking` 对 MiniMax 源不产生任何效果
+        # **直连官方端点未验证**。实测 enable_thinking / thinking 两种写法均被
+        # 静默丢弃(prompt_tokens 恒定不变),reasoning_effort 才是真开关。
+        # "开"取 medium: qwen 的 enable_thinking:true 与 deepseek 的
+        # thinking:{enabled} 都不指定预算、由模型自定,medium 是五档里语义最接近
+        # "厂商正常强度"的一档;取 high 等于替下游做"加钱换质量"的业务判断。
+        # 要精确控制档位经 `SourceConfig.extra_body`(优先级高于本片段)
         "minimax": ProviderProfile(
             name="minimax",
-            thinking_on={},
+            thinking_on={"reasoning_effort": "medium"},
-            thinking_off={},
+            thinking_off={"reasoning_effort": "none"},
             strip_think_tags=False,
         ),
     }
 )
 
 
+@dataclass(frozen=True)
+class ThinkingCapability:
+    """某个**具体模型**能否关闭推理(issue #5);登记必须附实测证据与日期。
+
+    与 `ProviderProfile` 的分工: 后者声明**形态**(参数长什么样,按 provider 变,
+    数年不变一次),本类声明**能力**(按 model 变,同一 provider 每代都变)。二者
+    合一在 provider 级表达不了代际差异——实测 MiniMax-M3 可关闭推理,而同厂的
+    M2.7/M2.5 三种参数形态全部无效(findings §2.3),profile 一格管不住三个模型。
+
+    `evidence` 不是装饰: 能力表过期是必然事件,没有出处就无从判断该不该信它。
+    """
+
+    can_disable: bool
+    evidence: str
+
+
+DEFAULT_CAPABILITIES: Mapping[str, ThinkingCapability] = MappingProxyType(
+    {
+        "MiniMax-M3": ThinkingCapability(
+            can_disable=True,
+            evidence="2026-08-02 经 new-api 中转实测 N=10: reasoning_effort=none 稳定关闭,零跳变",
+        ),
+        "MiniMax-M2.7": ThinkingCapability(
+            can_disable=False,
+            evidence=(
+                "2026-08-02 实测 reasoning_effort=none / thinking:{disabled} / thinking:{adaptive} "
+                "各 N=3 全部无效;OpenRouter 注册表登记 mandatory:true,models.dev 登记无控制手段"
+            ),
+        ),
+        "MiniMax-M2.5": ThinkingCapability(
+            can_disable=False,
+            evidence="2026-08-02 实测同 M2.7: 三种形态各 N=3 全部无效;外部注册表同样登记为强制推理",
+        ),
+        "qwen3.7-plus": ThinkingCapability(
+            can_disable=True,
+            evidence="2026-08-02 实测 enable_thinking=false 关闭(completion 5 token,无推理)",
+        ),
+        "deepseek-v4-pro": ThinkingCapability(
+            can_disable=True,
+            evidence="2026-08-02 实测 thinking:{type:disabled} 关闭(completion 3 token,无推理)",
+        ),
+    }
+)
+"""在用模型的推理能力登记(YAGNI: 不覆盖全世界,未登记走 `resolve_thinking` 退化)。"""
+
+
+def get_capability(
+    model: str, *, table: Mapping[str, ThinkingCapability] | None = None
+) -> ThinkingCapability | None:
+    """按模型名精确查找;未登记返回 None(= 能力未知,由调用方决定如何退化)。
+
+    与 `get_provider` 未注册即报错不同: provider 是配置里写死的少数几个值,
+    写错就是配置错误;而模型名千变万化,新模型上线不该被库挡住(设计 §5 R4)。
+    """
+    return (DEFAULT_CAPABILITIES if table is None else table).get(model)
+
+
+def register_capability(
+    model: str,
+    capability: ThinkingCapability,
+    *,
+    base: Mapping[str, ThinkingCapability] | None = None,
+) -> dict[str, ThinkingCapability]:
+    """纯函数注册: 返回 base(缺省 DEFAULT_CAPABILITIES)+ 新条目的新表,同名覆盖。"""
+    table = dict(DEFAULT_CAPABILITIES if base is None else base)
+    table[model] = capability
+    return table
+
+
+def resolve_thinking(
+    profile: ProviderProfile,
+    capability: ThinkingCapability | None,
+    enable_thinking: bool | None,
+    *,
+    model: str,
+) -> Mapping[str, Any]:
+    """三态 + 两层能力 → 请求体注入片段;不可满足时 ValueError。
+
+    调用点负责翻译: 装配期直接冒泡(配置错误),transport 内翻译为
+    `RequestRejectedError`(四分类之一)。判定顺序即语义,不可调换——形态未知时
+    无从注入,能力如何无关紧要,故 Phase 2 必须先于 Phase 4;未登记模型没有
+    `can_disable` 可读,故 Phase 3 必须先于 Phase 4。
+
+    `model` 只用于错误与告警文案: 报错能定位到具体模型才有可操作性,而
+    `capability` 为 None(未登记)时无从从别处取得模型名。
+    """
+    # Phase 1: 调用方不表态 —— 与 False 严格区分,用模型默认档
+    if enable_thinking is None:
+        return {}
+    slot = profile.thinking_on if enable_thinking else profile.thinking_off
+    direction = "thinking_on" if enable_thinking else "thinking_off"
+    # Phase 2: 形态未知 —— 提供了开关却不知道怎么发,静默放行就是欺骗调用方
+    if slot is None:
+        raise ValueError(
+            f"provider {profile.name!r} 的 {direction} 形态未知(模型 {model!r}): "
+            f"本库不知道该 provider 如何表达这一档。请用 register_provider 注册形态,"
+            f"或改用 SourceConfig.extra_body 直接下发供应商参数"
+        )
+    # Phase 3: 能力未登记 —— 新模型上线不该被库挡住,但也不该假装成功
+    if capability is None:
+        logger.warning(
+            "模型 {} 的推理能力未登记,按 provider {} 的形态尽力注入 {};"
+            "若该模型实际不支持这一档,本次设置将静默失效。实测后请用 register_capability 登记",
+            model,
+            profile.name,
+            dict(slot),
+        )
+        return slot
+    # Phase 4: 明确不支持关闭 —— 调用方要的是"不推理"的语义保证,给不了必须说
+    if enable_thinking is False and not capability.can_disable:
+        raise ValueError(
+            f"模型 {model!r} 无法关闭推理,enable_thinking=False 无法满足: "
+            f"{capability.evidence}。该模型的推理是固有属性,任何参数都关不掉——"
+            f"若实验需要关闭思维链,请换用支持关闭的模型"
+        )
+    return slot
+
+
 def get_provider(
     name: str, *, registry: Mapping[str, ProviderProfile] | None = None
 ) -> ProviderProfile:

src/polygateway/transports/openai_compat.py

@@ -21,7 +21,13 @@ from polygateway.errors import (
     SourceDeadError,
     TransientError,
 )
-from polygateway.providers import ProviderProfile, get_provider
+from polygateway.providers import (
+    ProviderProfile,
+    ThinkingCapability,
+    get_capability,
+    get_provider,
+    resolve_thinking,
+)
 from polygateway.streaming import StreamLivenessTimeout, stream_with_liveness_timeouts
 from polygateway.types import EmbeddingTransportResult, SourceConfig, TransportResult
 
@@ -284,9 +290,11 @@ class OpenAICompatTransport:
         self,
         *,
         registry: Mapping[str, ProviderProfile] | None = None,
+        capabilities: Mapping[str, ThinkingCapability] | None = None,
         client_factory: Callable[[SourceConfig], httpx.AsyncClient] | None = None,
     ) -> None:
         self._registry = registry
+        self._capabilities = capabilities
         self._client_factory = client_factory or _default_client_factory
         self._clients: dict[str, httpx.AsyncClient] = {}
 
@@ -309,10 +317,12 @@ class OpenAICompatTransport:
         payload: dict[str, Any] = {"model": source.model, "messages": messages, "stream": stream}
         if stream:
             payload["stream_options"] = {"include_usage": True}  # 强制 usage 帧(三项目同款)
-        if source.enable_thinking is True:
+        # 形态(provider 级)与能力(model 级)在此相遇;不可满足时 ValueError,
-            payload.update(profile.thinking_on)
+        # 由 complete() 翻译为四分类之一(issue #5)
-        elif source.enable_thinking is False:
+        capability = get_capability(source.model, table=self._capabilities)
-            payload.update(profile.thinking_off)
+        payload.update(
+            resolve_thinking(profile, capability, source.enable_thinking, model=source.model)
+        )
         # 顺序即优先级(issue #4 设计决策 A): 配置级 extra_body 在前,调用级
         # overlay(含结构化注入)在后覆盖之。两行不可调换
         payload.update(source.extra_body)
@@ -330,9 +340,17 @@ class OpenAICompatTransport:
     ) -> TransportResult:
         """一次原始调用;HTTP/线路/流式异常按 ARCH §6.2 翻译为领域错误。"""
         profile = get_provider(source.provider, registry=self._registry)
-        payload = self._build_payload(
+        try:
-            messages=messages, source=source, profile=profile, stream=stream, overlay=overlay
+            payload = self._build_payload(
-        )
+                messages=messages, source=source, profile=profile, stream=stream, overlay=overlay
+            )
+        except ValueError as exc:
+            # 推理开关不可满足是**请求本身**的问题: 换源重试都救不了它
+            raise RequestRejectedError(
+                f"{source.name} 推理开关无法满足: {exc}",
+                source_name=source.name,
+                operation="chat",
+            ) from exc
         url = source.base_url.rstrip("/") + "/chat/completions"
         client = self._client_for(source)
         ctx: dict[str, Any] = {"source_name": source.name, "operation": "chat"}

tests/unit/test_client.py

@@ -202,6 +202,33 @@ class TestModelFingerprint:
         assert plain != tuned
         assert tuned.startswith("qwen-max|")  # 旧字面量仍是前缀,便于人眼辨认
 
+    def test_enable_thinking_changes_fingerprint(self):
+        """issue #5 配套: thinking 一旦真正改变请求体,就必须进缓存身份。
+
+        否则"关掉推理后重启"会读到开着推理时缓存的旧响应——issue #4 为
+        temperature 写过逐字相同的理由。
+        """
+        from polygateway.client import build_model_fingerprint
+
+        plain = build_model_fingerprint([_source()])
+        off = build_model_fingerprint([_source(enable_thinking=False)])
+        on = build_model_fingerprint([_source(enable_thinking=True)])
+        assert len({plain, off, on}) == 3
+
+    def test_extra_body_only_fingerprint_is_byte_identical_to_before(self):
+        """只配 extra_body、不表态 thinking 的存量源不得触发冷启动。
+
+        字面量在此硬编码: 这条断言的价值全在"逐字相同",改实现时必须先看见它红。
+        """
+        import hashlib
+        import json
+
+        from polygateway.client import build_model_fingerprint
+
+        mark = json.dumps(["qwen-max", {"temperature": 0}], sort_keys=True, ensure_ascii=False)
+        expected = "qwen-max|" + hashlib.sha256(mark.encode("utf-8")).hexdigest()
+        assert build_model_fingerprint([_source(extra_body={"temperature": 0})]) == expected
+
     def test_source_rename_does_not_change_fingerprint(self):
         """指纹按 (model, extra_body) 而非源名: 改名不该误触全量冷启动。"""
         from polygateway.client import build_model_fingerprint

tests/unit/test_config.py

@@ -460,6 +460,36 @@ class TestCrossFieldInvariants:
         with pytest.raises(ValueError, match="lease_ttl_s"):
             GatewayClient.from_settings(dataclasses.replace(base, lease_ttl_s=1.0))
 
+    # —— 推理开关的装配守卫(issue #5)——
+
+    def _thinking_sources(self, provider, model, enable_thinking):
+        base = self._base()
+        src = dataclasses.replace(
+            base.sources[0], provider=provider, model=model, enable_thinking=enable_thinking
+        )
+        return dataclasses.replace(base, sources=(src,))
+
+    def test_model_that_cannot_disable_thinking_fails_at_assembly(self):
+        """M2.x 关不掉推理: 配了 false 必须当场炸,而不是装出一个骗人的 client。"""
+        settings = self._thinking_sources("minimax", "MiniMax-M2.7", False)
+        with pytest.raises(ValueError, match="MiniMax-M2.7"):
+            GatewayClient.from_settings(settings)
+
+    def test_unknown_thinking_shape_fails_at_assembly(self):
+        """provider=openai 是任意兼容厂商的兜底段名,形态未知即报错并指路。"""
+        settings = self._thinking_sources("openai", "kimi-k3", False)
+        with pytest.raises(ValueError, match="register_provider"):
+            GatewayClient.from_settings(settings)
+
+    def test_supported_combination_assembles(self):
+        settings = self._thinking_sources("minimax", "MiniMax-M3", False)
+        assert GatewayClient.from_settings(settings) is not None
+
+    def test_not_taking_a_position_never_trips_the_guard(self):
+        """enable_thinking=None(不干预)对任何 provider 都不该被守卫拦下。"""
+        settings = self._thinking_sources("openai", "kimi-k3", None)
+        assert GatewayClient.from_settings(settings) is not None
+
     def test_ocr_settings_cannot_wrap_invalid_gateway(self):
         """OcrSettings/EmbeddingSettings 只是包一层 GatewaySettings,自动继承同一把关。"""
         base = self._base()

tests/unit/test_openai_compat.py

@@ -506,6 +506,65 @@ class TestRequestShaping:
             assert "enable_thinking" not in seen
         assert seen["stream_options"] == {"include_usage": True}
 
+    @pytest.mark.parametrize(
+        ("enable_thinking", "expected"),
+        [(True, "medium"), (False, "none")],
+    )
+    async def test_minimax_injects_reasoning_effort(self, enable_thinking, expected):
+        """issue #5: MiniMax 认的是 reasoning_effort,不是 enable_thinking。"""
+        seen = {}
+
+        def handler(request):
+            seen.update(json.loads(request.content))
+            return _sse_stream(_chunk(content="x"), _chunk(usage=_USAGE))
+
+        source = _source(
+            name="mm", provider="minimax", model="MiniMax-M3", enable_thinking=enable_thinking
+        )
+        await _complete(_transport_for(handler), source)
+        assert seen["reasoning_effort"] == expected
+        assert "enable_thinking" not in seen  # 旧形态实测被静默丢弃,不再下发
+
+    async def test_extra_body_overrides_the_profile_slot(self):
+        """注入顺序即优先级: profile → extra_body → overlay,两行不可调换。"""
+        seen = {}
+
+        def handler(request):
+            seen.update(json.loads(request.content))
+            return _sse_stream(_chunk(content="x"), _chunk(usage=_USAGE))
+
+        source = _source(
+            name="mm",
+            provider="minimax",
+            model="MiniMax-M3",
+            enable_thinking=True,
+            extra_body={"reasoning_effort": "high"},
+        )
+        await _complete(_transport_for(handler), source)
+        assert seen["reasoning_effort"] == "high"
+
+    async def test_model_that_cannot_disable_is_rejected_not_silently_ignored(self):
+        """M2.x 关不掉推理: 必须是四分类之一的 RequestRejected,不是裸 ValueError。
+
+        裸异常会逃出 chat() —— 它不属错误四分类、TelemetryMW 也不捕,结果是一行
+        遥测都没有就崩了(设计 §5.1)。
+        """
+
+        def handler(request):  # pragma: no cover - 不该走到发请求
+            raise AssertionError("请求不该发出")
+
+        source = _source(name="mm", provider="minimax", model="MiniMax-M2.7", enable_thinking=False)
+        with pytest.raises(RequestRejectedError, match="MiniMax-M2.7"):
+            await _complete(_transport_for(handler), source)
+
+    async def test_unknown_shape_is_rejected(self):
+        def handler(request):  # pragma: no cover - 不该走到发请求
+            raise AssertionError("请求不该发出")
+
+        source = _source(name="k3", provider="openai", model="kimi-k3", enable_thinking=False)
+        with pytest.raises(RequestRejectedError, match="register_provider"):
+            await _complete(_transport_for(handler), source)
+
     async def test_overlay_merged_into_payload(self):
         seen = {}
 

tests/unit/test_providers.py

@@ -1,12 +1,18 @@
 """providers.py 注册表测试(M1 设计 §7;register_provider 为纯函数,无可变全局)。"""
 
 import pytest
+from loguru import logger
 
 from polygateway.providers import (
+    DEFAULT_CAPABILITIES,
     DEFAULT_PROFILES,
     ProviderProfile,
+    ThinkingCapability,
+    get_capability,
     get_provider,
+    register_capability,
     register_provider,
+    resolve_thinking,
 )
 
 
@@ -24,14 +30,21 @@ class TestDefaultProfiles:
         assert p.thinking_off == {"thinking": {"type": "disabled"}}
         assert p.strip_think_tags is False
 
-    def test_openai_baseline_profile(self):
+    def test_openai_slots_are_unknown_not_empty(self):
+        """issue #5: 该段名实践中被复用为任意兼容厂商的兜底(下游把 kimi 挂在此),
+
+        故不能下发任何厂商方言参数。None = 形态未知 → 配了 enable_thinking 即报错,
+        而不是空字典那种"注入了个寂寞"的静默失效。
+        """
         p = get_provider("openai")
-        assert p.thinking_on == {} and p.thinking_off == {}
+        assert p.thinking_on is None and p.thinking_off is None
         assert p.strip_think_tags is False
 
-    def test_minimax_baseline_profile(self):
+    def test_minimax_profile_uses_reasoning_effort(self):
+        """2026-08-02 实测: reasoning_effort 才是 MiniMax 认的开关。"""
         p = get_provider("minimax")
-        assert p.thinking_on == {} and p.thinking_off == {}
+        assert p.thinking_off == {"reasoning_effort": "none"}
+        assert p.thinking_on == {"reasoning_effort": "medium"}
         assert p.strip_think_tags is False
 
     def test_unknown_provider_fails_loudly(self):
@@ -62,3 +75,83 @@ class TestPureFunctionRegistration:
     def test_default_profiles_mapping_is_read_only(self):
         with pytest.raises(TypeError):
             DEFAULT_PROFILES["hack"] = None  # type: ignore[index]
+
+
+def _warnings():
+    """捕获库发出的 WARNING;loguru 不经标准 logging,pytest 的 caplog 抓不到。"""
+    messages: list[str] = []
+    sink_id = logger.add(messages.append, level="WARNING")
+    return messages, sink_id
+
+
+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")