fix: reject conflicting raw reasoning overrides before sending

src/polygateway/client.py

@@ -34,7 +34,12 @@ from polygateway.sources import (
     RoundRobinSelector,
     SourceCooldownMemo,
 )
-from polygateway.thinking import effective_effort, get_capability, resolve_thinking
+from polygateway.thinking import (
+    effective_effort,
+    get_capability,
+    resolve_thinking,
+    validate_thinking_raw,
+)
 from polygateway.transports.openai_compat import OpenAICompatTransport
 from polygateway.types import (
     ChatRequest,
@@ -82,21 +87,26 @@ def _guard_thinking(
     就带着指路信息炸掉。`get_provider` 现在就是同一形态的双点调用。
     """
     for source, profile in zip(sources, profiles, strict=True):
-        resolve_thinking(
+        effort = effective_effort(
-            profile,
-            get_capability(source.model, table=capabilities),
-            # 装配期看不见请求级档位(它逐次调用才产生),故只解源级两层;请求级
-            # 只能在运行期由 transport 校验(设计 §10 的装配期/运行期分工)
-            effective_effort(
             request_effort=None,
             source_effort=source.reasoning_effort,
             enable_thinking=source.enable_thinking,
-            ),
+        )
+        resolve_thinking(
+            profile,
+            get_capability(source.model, table=capabilities),
+            effort,
             model=source.model,
             # 与 transport 用同一个 fallback,否则配了 nearest 的源会在装配期就被
             # 判死,而它在运行期本来是能映射到最近档跑起来的
             fallback=source.effort_fallback,
         )
+        validate_thinking_raw(
+            source.extra_body,
+            effort=effort,
+            wire=profile.thinking,
+            origin=f"源 {source.name} extra_body",
+        )
 
 
 def _fingerprint_mark(source: SourceConfig) -> str:
@@ -359,6 +369,7 @@ class GatewayClient:
             if reasoning_effort is None
             else coerce_effort(reasoning_effort, origin="chat(reasoning_effort=...)")
         )
+        validate_thinking_raw(sampling, effort=effort, wire=None, origin="chat overlay")
         request = ChatRequest(
             messages=messages,
             session_id=session_id,

src/polygateway/transports/openai_compat.py

@@ -34,6 +34,7 @@ from polygateway.thinking import (
     observe_thinking,
     reconcile_thinking,
     resolve_thinking,
+    validate_thinking_raw,
 )
 from polygateway.transports._http_errors import compose_message, summarize_body
 from polygateway.types import (
@@ -371,20 +372,23 @@ class OpenAICompatTransport:
         self._warned_models.add(source.model)
         # 三层优先级在此汇合: 请求级 > 源级 > enable_thinking 语法糖(设计 §4.2)。
         # 判定与装配守卫共用同一个纯函数,两处分叉就会变成"装配期放行、运行期报错"
-        resolution = resolve_thinking(
+        effort = effective_effort(
-            profile,
-            capability,
-            effective_effort(
             request_effort=reasoning_effort,
             source_effort=source.reasoning_effort,
             enable_thinking=source.enable_thinking,
-            ),
+        )
+        resolution = resolve_thinking(
+            profile,
+            capability,
+            effort,
             model=source.model,
             # 源级 `EFFORT_FALLBACK` 必须真的走到这里: 硬编码 "error" 会让人类明确
             # 要求实现的 `nearest` 在零告警下变成死代码(2026-09-05 独立验证查出)
             fallback=source.effort_fallback,
             warn_unregistered=first_time,
         )
+        for raw, origin in ((source.extra_body, "source extra_body"), (overlay, "request overlay")):
+            validate_thinking_raw(raw, effort=effort, wire=profile.thinking, origin=origin)
         payload.update(resolution.payload)
         # 顺序即优先级(issue #4 设计决策 A): 配置级 extra_body 在前,调用级
         # overlay(含结构化注入)在后覆盖之。两行不可调换

tests/unit/test_client.py

@@ -247,52 +247,35 @@ class TestReasoningEffortPriority:
         assert "reasoning_effort" not in captured[0]
 
     @pytest.mark.parametrize(
-        ("provider", "model", "fragment"),
+        "provider,model,tier",
         [
-            ("qwen", "qwen-max", {"enable_thinking": True}),
+            ("deepseek", "deepseek-v4-pro", "high"),
-            ("deepseek", "deepseek-v4-pro", {"thinking": {"type": "enabled"}}),
+            ("zhipu", "glm-5.3", "low"),
-            ("zhipu", "glm-5.3", {"thinking": {"type": "enabled"}}),
+            ("moonshot", "kimi-k3", "low"),
-            ("moonshot", "kimi-k3", {"thinking": {"type": "enabled"}}),
+            ("minimax", "MiniMax-M3", "medium"),
         ],
     )
-    async def test_legacy_on_tier_matches_old_fragment(self, provider, model, fragment):
+    async def test_legacy_auto_requires_explicit_migration(self, provider, model, tier):
-        """存量 `ENABLE_THINKING=true` 的回归门: 发出去的字节逐字不变。
+        """旧糖配置明确拒绝，显式选择才能恢复可执行请求。"""
+        from dataclasses import replace
 
-        **只覆盖 `on_base` 自己就说全了"开"的四段**。openai/anthropic/google 的开档
-        旧版硬编码 `{"reasoning_effort": "medium"}`,新版不注入任何档位——那是设计
-        §4.2 声明过的**有意变更**(medium 在 GLM/kimi/deepseek 的档位表里根本不存在,
-        是库替下游做的档位判断),不是本门要守的不变量;这三家的模型经 OpenRouter
-        登记均为默认推理,不注入也仍是"开"。minimax 不在此列: 它的模型不满足该前提,
-        已按 issue #21 改回 medium,由下一条用例单独守。
-
-        qwen/deepseek 两条字面量逐字取自升级前的 `ProviderProfile.thinking_on`;
-        zhipu/moonshot 升级前没有对应段,断言的是它们 2026-09-04 登记的形态。
-        """
         captured = []
         source = _source(provider=provider, model=model, enable_thinking=True)
-        async with self._capturing_client(captured, sources=[source]) as client:
+        client = self._capturing_client(captured, sources=[source])
-            await client.chat([{"role": "user", "content": "hi"}])
+        try:
-        body = captured[0]
+            with pytest.raises(RequestRejectedError):
-        assert {k: body[k] for k in fragment} == fragment
+                await client.chat([])
-        # `auto` = 开启但不指定强度: 语法糖不得替调用方挑一个档
+            assert captured == []
-        assert "reasoning_effort" not in body
+        finally:
-
+            await client._transport.aclose()
-    async def test_legacy_minimax_on_tier_actually_turns_reasoning_on(self):
+        client = self._capturing_client(
-        """回归门(issue #21): minimax 段的存量 `ENABLE_THINKING=true` 必须真开推理。
+            captured, sources=[replace(source, enable_thinking=None, reasoning_effort=tier)]
-
+        )
-        本次换代一度把这段的开启形态改成 `on_base={}`(什么参数都不注入),依据是
+        try:
-        "这些模型默认就推理,不注入也仍是'开'"。T10 真实网关实测推翻了该前提:
+            await client.chat([])
-        MiniMax-M3 不带任何推理参数时 5/5 轮**不推理**(六个强度值则全部生效)。
+            assert captured[0]["reasoning_effort"] == tier
-        于是存量下游从"真开推理"静默变成"不推理",而 `resolve_thinking` 的 Phase 5
+        finally:
-        无条件放行 `auto`、能力表也堵不住这条路。
+            await client._transport.aclose()
-
-        断言落在**发出去的字节**上而非中间态: 静默不推理这件事只有在请求体里才看得见。
-        """
-        captured = []
-        source = _source(provider="minimax", model="MiniMax-M3", enable_thinking=True)
-        async with self._capturing_client(captured, sources=[source]) as client:
-            await client.chat([{"role": "user", "content": "hi"}])
-        assert captured[0]["reasoning_effort"] == "medium"
 
 
 class TestEffortFallbackWiring:
@@ -1280,3 +1263,56 @@ class TestTelemetryStatusExposure:
         assert _ocr_client().telemetry_status is None
         assert _ocr_client(telemetry=_Closable()).telemetry_status is None
         self._assert_snapshot(_ocr_client(telemetry=self._recorder(tmp_path)).telemetry_status)
+
+
+class TestManagedReasoningAdmission:
+    """入口前置与实际准入边界保持明确。"""
+
+    @pytest.mark.parametrize("factory", ["env", "settings"])
+    def test_factory_rejects_conflict_before_building_backends(self, monkeypatch, factory):
+        env = {
+            **_ENV,
+            "LLM__QWEN__1__MODEL": "qwen3.7-plus",
+            "LLM__QWEN__1__ENABLE_THINKING": "true",
+            "LLM__QWEN__1__EXTRA_BODY": '{"enable_thinking":true}',
+        }
+        built = []
+
+        def forbidden(*args, **kwargs):
+            built.append(True)
+            raise AssertionError("后端不得构造")
+
+        monkeypatch.setattr("polygateway.client._build_limiter", forbidden)
+        with pytest.raises(ValueError, match="冲突"):
+            if factory == "env":
+                GatewayClient.from_env(env=env)
+            else:
+                GatewayClient.from_settings(GatewaySettings.from_env(env=env))
+        assert not built
+
+    async def test_request_conflict_does_not_enter_onion(self):
+        client = _client()
+
+        async def forbidden(request):
+            raise AssertionError("不得进入洋葱")
+
+        client._handler = forbidden
+        try:
+            with pytest.raises(ValueError, match="冲突"):
+                await client.chat([], reasoning_effort="high", overlay={"reasoning_effort": "high"})
+        finally:
+            await client._transport.aclose()
+
+    async def test_source_conflict_releases_admitted_permit(self):
+        source = _source(
+            reasoning_effort="high", provider="openai", extra_body={"reasoning_effort": "high"}
+        )
+        sent = []
+        client = _client(sources=[source], handler=lambda request: sent.append(request) or _sse())
+        try:
+            with pytest.raises(RequestRejectedError, match="冲突"):
+                await client.chat([])
+            assert sent == []
+            assert (await client._limiter_backend.source_stats(source.name)).inflight == 0
+        finally:
+            await client._transport.aclose()

tests/unit/test_openai_compat.py

@@ -623,8 +623,8 @@ class TestThinkingReconciliation:
         try:
             await _complete(transport, self._minimax(False))
             await _complete(transport, self._minimax(False))
-            await _complete(transport, self._minimax(True))
+            await _complete(transport, self._minimax(True), reasoning_effort=Effort.MEDIUM)
-            await _complete(transport, self._minimax(True))
+            await _complete(transport, self._minimax(True), reasoning_effort=Effort.MEDIUM)
         finally:
             logger.remove(sink_id)
         hits = [m for m in messages if "MiniMax-M3" in m]
@@ -705,75 +705,44 @@ class TestNonStreamFastPath:
 
 
 class TestRequestShaping:
-    @pytest.mark.parametrize(
+    @pytest.mark.parametrize("tier", [Effort.MEDIUM, Effort.NONE])
-        ("enable_thinking", "expected"),
+    async def test_minimax_explicit_tier_is_sent(self, tier):
-        [(True, {"enable_thinking": True}), (False, {"enable_thinking": False}), (None, {})],
-    )
-    async def test_thinking_tri_state_injection(self, enable_thinking, expected):
         seen = {}
 
         def handler(request):
             seen.update(json.loads(request.content))
             return _sse_stream(_chunk(content="x"), _chunk(usage=_USAGE))
 
-        await _complete(_transport_for(handler), _source(enable_thinking=enable_thinking))
+        transport = _transport_for(handler)
-        assert {k: seen[k] for k in expected} == expected
+        try:
-        if enable_thinking is None:
+            result = await _complete(
+                transport, _source(provider="minimax", model="MiniMax-M3"), reasoning_effort=tier
+            )
+            assert seen["reasoning_effort"] == tier.value
+            assert result.applied_effort is tier
             assert "enable_thinking" not in seen
-        assert seen["stream_options"] == {"include_usage": True}
+        finally:
+            await transport.aclose()
 
-    @pytest.mark.parametrize(
+    async def test_raw_only_keeps_source_then_request_priority(self):
-        ("enable_thinking", "expected"),
-        # 本条断言反复过一次,记下原委以免第三次改回去:
-        # T2(2026-09-04)按"MiniMax 开启档本就无需参数"的**推定**把 medium 改成不注入;
-        # T10(2026-09-05)真实网关实测推翻该推定——M3 不发任何推理参数时 5/5 轮不推理,
-        # 故 medium 回归(issue #21 的权宜之计,正解是让 auto 受能力表约束)
-        [(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(
+        transport = _transport_for(handler)
-            name="mm", provider="minimax", model="MiniMax-M3", enable_thinking=enable_thinking
+        try:
+            result = await _complete(
+                transport,
+                _source(extra_body={"reasoning_effort": "low", "temperature": 0}),
+                overlay={"reasoning_effort": "high", "temperature": 1},
             )
-        await _complete(_transport_for(handler), source)
-        if expected is None:
-            assert "reasoning_effort" not in seen
-        else:
-            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,两行不可调换。
-
-        固定用 **zhipu + glm-5.3 + 源级 low** 这组: 判据必须落在一个 profile
-        **真的写了值**的键上,两边写同一个键才谈得上谁覆盖谁。不挑 minimax 是因为
-        它的 `on_base` 只写 `reasoning_effort` 一个键(issue #21 的权宜之计),
-        覆盖发生后看不见"profile 独有的那半边仍在",判据少一半。
-        """
-        seen = {}
-
-        def handler(request):
-            seen.update(json.loads(request.content))
-            return _sse_stream(_chunk(content="x"), _chunk(usage=_USAGE))
-
-        source = _source(
-            name="zp",
-            provider="zhipu",
-            model="glm-5.3",
-            reasoning_effort="low",
-            extra_body={"reasoning_effort": "high"},
-        )
-        await _complete(_transport_for(handler), source)
-        # profile 注入的是 low,extra_body 后写故发出去的是 high;顺序一调换就变 low,
-        # 即下游写在 extra_body 里的覆盖被库悄悄顶掉(issue #20 的成因形态)
             assert seen["reasoning_effort"] == "high"
-        assert seen["thinking"] == {"type": "enabled"}  # profile 独有的那半边仍在
+            assert seen["temperature"] == 1
+            assert result.applied_effort is None
+        finally:
+            await transport.aclose()
 
     async def test_model_that_cannot_disable_is_rejected_not_silently_ignored(self):
         """M2.x 关不掉推理: 必须是四分类之一的 RequestRejected,不是裸 ValueError。
@@ -1056,3 +1025,86 @@ class TestLifecycle:
         await _complete(transport, _source())
         await transport.aclose()
         await transport.aclose()
+
+
+class TestManagedReasoningOwnership:
+    """通过真实 transport 验证拒绝发生在 HTTP 之前。"""
+
+    @pytest.mark.parametrize(
+        "raw",
+        [
+            {"reasoning_effort": "high"},
+            {"enable_thinking": True},
+            {"thinking": {}},
+            {"thinking_budget": 100},
+            {"reasoning": {}},
+            {"thinkingConfig": {}},
+            {"output_config": {"effort": "low"}},
+        ],
+    )
+    @pytest.mark.parametrize("layer", ["source", "request", "shadowed"])
+    async def test_raw_control_is_rejected_before_http(self, raw, layer):
+        sent = []
+
+        def handler(request):
+            sent.append(request)
+            return _sse_stream(_chunk(content="ok"), _chunk(usage=_USAGE))
+
+        source = _source(
+            provider="openai",
+            reasoning_effort=Effort.HIGH,
+            extra_body=raw if layer != "request" else {},
+        )
+        overlay = raw if layer != "source" else {}
+        transport = _transport_for(handler)
+        try:
+            with pytest.raises(RequestRejectedError):
+                await _complete(transport, source, overlay=overlay)
+            assert sent == []
+        finally:
+            await transport.aclose()
+
+
+class TestDefaultClientFactory:
+    """直接检查生产 factory，不用取证测试的另一套 factory 代替。"""
+
+    @pytest.mark.parametrize("key", ["fake-a", "fake-b"])
+    async def test_authorization_uses_source_api_key(self, key):
+        from polygateway.transports.openai_compat import _default_client_factory
+
+        client = _default_client_factory(_source(api_key=key))
+        try:
+            assert client.headers["Authorization"] == f"Bearer {key}"
+            assert (
+                client.build_request("POST", "https://gw.example/v1/chat/completions").headers[
+                    "Authorization"
+                ]
+                == f"Bearer {key}"
+            )
+        finally:
+            await client.aclose()
+
+    @pytest.mark.parametrize("timeout", [17.0, 53.0])
+    async def test_timeout_uses_source_timeout_for_all_phases(self, timeout):
+        from polygateway.transports.openai_compat import _default_client_factory
+
+        client = _default_client_factory(_source(timeout_s=timeout))
+        try:
+            assert [
+                client.timeout.connect,
+                client.timeout.read,
+                client.timeout.write,
+                client.timeout.pool,
+            ] == [timeout] * 4
+        finally:
+            await client.aclose()
+
+    @pytest.mark.parametrize("trust_env", [True, False])
+    async def test_trust_env_uses_source_setting(self, trust_env):
+        from polygateway.transports.openai_compat import _default_client_factory
+
+        client = _default_client_factory(_source(trust_env=trust_env))
+        try:
+            assert client.trust_env is trust_env
+        finally:
+            await client.aclose()