From 1ee74c35a8a30a3f7998f8f040d5114317a0939c Mon Sep 17 00:00:00 2001
From: iomgaa <iomgaaycz@gmail.com>
Date: Wed, 9 Sep 2026 01:24:54 -0400
Subject: [PATCH] fix: validate ownership of managed reasoning parameters

---
 src/polygateway/thinking.py | 57 +++++++++++++++++++++++++++++++++
 tests/unit/test_thinking.py | 64 +++++++++++++++++++++++++++++++++++++
 2 files changed, 121 insertions(+)

diff --git a/src/polygateway/thinking.py b/src/polygateway/thinking.py
index 15b6f26..da8af1f 100644
--- a/src/polygateway/thinking.py
+++ b/src/polygateway/thinking.py
@@ -441,6 +441,62 @@ def effective_effort(
     return Effort.AUTO if enable_thinking else Effort.NONE
 
 
+_RAW_THINKING_ROOTS = frozenset(
+    {
+        "reasoning_effort",
+        "enable_thinking",
+        "thinking",
+        "thinking_budget",
+        "reasoning",
+        "thinkingConfig",
+    }
+)
+
+
+def _has_output_effort(raw: Mapping[str, Any]) -> bool:
+    """标准嵌套强度只检查明确的路径，不猜私有方言。"""
+    output = raw.get("output_config")
+    return isinstance(output, Mapping) and "effort" in output
+
+
+def validate_thinking_wire(wire: ThinkingWire, *, model: str) -> None:
+    """拒绝开启片段代选强度，未知形态仍交由请求方向检查。"""
+    base = wire.on_base
+    if base is not None and (
+        "reasoning_effort" in base
+        or (wire.effort_key is not None and wire.effort_key in base)
+        or _has_output_effort(base)
+    ):
+        raise ThinkingUnsupportedError(
+            f"模型 {model!r} 的 on_base 不得包含强度档位；请移除强度并显式传 reasoning_effort"
+        )
+
+
+def validate_thinking_raw(
+    raw: Mapping[str, Any],
+    *,
+    effort: Effort | None,
+    wire: ThinkingWire | None,
+    origin: str,
+) -> None:
+    """受管推理只有一个来源；同值或被遮蔽的 raw 控制也拒绝。"""
+    if effort is None:
+        return
+    roots = set(_RAW_THINKING_ROOTS)
+    if wire is not None:
+        for fragment in (wire.on_base, wire.off):
+            if fragment is not None:
+                roots.update(fragment)
+        if wire.effort_key is not None:
+            roots.add(wire.effort_key)
+    if roots.intersection(raw) or _has_output_effort(raw):
+        # 不打印 raw 或自定义键名，防配置中夹带秘密。
+        raise ThinkingUnsupportedError(
+            f"{origin} 与受管 reasoning_effort={effort.value!r} 冲突；"
+            "请删除 raw 推理控制，或移除源级/请求级推理表态后仅用 raw"
+        )
+
+
 def resolve_thinking(
     profile: ProviderProfile,
     capability: ThinkingCapability | None,
@@ -490,6 +546,7 @@ def resolve_thinking(
     而是**静默判否**: Phase 2 按开启方向取字段、Phase 4 整条被绕过,最后在拼错误
     文案时才以 `AttributeError` 现形(一个未文档化、也不属四分类的异常)。
     """
+    validate_thinking_wire(profile.thinking, model=model)
     # Phase 0: 归一 —— 判据全是身份比较,入口不归一则后面每一关都在拿裸串比枚举
     if effort is not None:
         effort = coerce_effort(effort, origin=f"resolve_thinking(model={model!r})")
diff --git a/tests/unit/test_thinking.py b/tests/unit/test_thinking.py
index b5f630a..e3bb6bc 100644
--- a/tests/unit/test_thinking.py
+++ b/tests/unit/test_thinking.py
@@ -847,3 +847,67 @@ class TestEffectiveEffort:
         assert (
             effective_effort(request_effort=None, source_effort=None, enable_thinking=None) is None
         )
+
+
+class TestThinkingWireOwnership:
+    """开启片段只能表达开启，不能携带隐式强度。"""
+
+    @pytest.mark.parametrize(
+        "base",
+        [
+            {"reasoning_effort": "high"},
+            {"reasoning_effort": None},
+            {"depth": "auto"},
+            {"output_config": {"effort": "low"}},
+        ],
+    )
+    @pytest.mark.parametrize("effort", [None, Effort.AUTO, Effort.HIGH])
+    def test_on_base_cannot_hide_a_tier(self, base, effort):
+        profile = ProviderProfile(
+            name="custom",
+            thinking=ThinkingWire(off={"depth": "none"}, on_base=base, effort_key="depth"),
+            strip_think_tags=False,
+        )
+        with pytest.raises(ThinkingUnsupportedError, match="on_base"):
+            resolve_thinking(profile, None, effort, model="custom-model")
+
+
+class TestThinkingRawOwnership:
+    """纯规则覆盖标准、自定义根与无意图逃生口。"""
+
+    @pytest.mark.parametrize(
+        "raw",
+        [
+            {"reasoning_effort": "high"},
+            {"enable_thinking": True},
+            {"thinking": {}},
+            {"thinking_budget": 100},
+            {"reasoning": {}},
+            {"thinkingConfig": {}},
+            {"output_config": {"effort": None}},
+            {"depth.key": None},
+            {"off_control": {}},
+        ],
+    )
+    @pytest.mark.parametrize("effort", [Effort.NONE, Effort.AUTO, Effort.HIGH])
+    def test_control_roots_rejected_without_mutating_input(self, raw, effort):
+        from copy import deepcopy
+
+        from polygateway.thinking import validate_thinking_raw
+
+        wire = ThinkingWire(off={"off_control": False}, on_base={}, effort_key="depth.key")
+        before = deepcopy(raw)
+        with pytest.raises(ThinkingUnsupportedError):
+            validate_thinking_raw(raw, effort=effort, wire=wire, origin="test")
+        assert raw == before
+        validate_thinking_raw(raw, effort=None, wire=wire, origin="test")
+        assert raw == before
+
+    def test_output_format_is_not_effort_unless_wire_owns_root(self):
+        from polygateway.thinking import validate_thinking_raw
+
+        raw = {"output_config": {"format": "json"}, "temperature": 0, "seed": 7}
+        validate_thinking_raw(raw, effort=Effort.AUTO, wire=None, origin="test")
+        wire = ThinkingWire(off=None, on_base={"output_config": {"enabled": True}}, effort_key=None)
+        with pytest.raises(ThinkingUnsupportedError):
+            validate_thinking_raw(raw, effort=Effort.AUTO, wire=wire, origin="test")