feat: add sampling overlay validation and source extra_body

Three pure helpers in the innermost layer plus ChatRequest.sampling as a
cross-layer snapshot, so cache keys and telemetry read one stable value.

tests/unit/test_types.py

@@ -305,3 +305,92 @@ class TestOcrTypes:
 
         with pytest.raises(TypeError):
             OcrTextResult(text="x")  # 溯源件不可省略
+
+
+class TestSamplingValidation:
+    """采样参数覆盖层的构造期校验(issue #4 设计决策 B)。"""
+
+    @pytest.mark.parametrize("key", ["model", "messages", "stream", "stream_options"])
+    def test_protected_keys_rejected(self, key):
+        """保护键会击穿治理: 成本算错/口径失真/绕过看门狗与 usage 帧。"""
+        from polygateway.types import validate_request_overlay
+
+        with pytest.raises(ValueError) as exc:
+            validate_request_overlay({key: "x"}, origin="chat(overlay=...)")
+        assert key in str(exc.value)
+        assert "chat(overlay=...)" in str(exc.value)  # 信息须能定位来源
+
+    def test_non_str_key_reports_key_problem(self):
+        """非 str 键须报"键必须是 str",不能被 sort_keys 的比较错误误报成不可序列化。"""
+        from polygateway.types import validate_request_overlay
+
+        with pytest.raises(ValueError, match="str"):
+            validate_request_overlay({1: "a", "b": 2}, origin="test")
+
+    def test_unserializable_value_becomes_value_error(self):
+        """裸 TypeError 会逃出 CacheMW 的降级 try 且一行遥测都没有(设计决策 B)。"""
+        from polygateway.types import validate_request_overlay
+
+        with pytest.raises(ValueError, match="JSON"):
+            validate_request_overlay({"temperature": object()}, origin="test")
+
+    def test_returns_independent_copy(self):
+        """调用方逐次改 seed 复用同一 dict 是预期模式,不拷贝会有竞态(决策 E)。"""
+        from polygateway.types import validate_request_overlay
+
+        caller_dict = {"temperature": 0, "seed": 42}
+        validated = validate_request_overlay(caller_dict, origin="test")
+        caller_dict["seed"] = 43
+        assert validated == {"temperature": 0, "seed": 42}
+
+    def test_merge_prefers_call_level(self):
+        """优先级: 调用级 > 配置级(设计决策 A)。"""
+        from polygateway.types import merge_sampling
+
+        merged = merge_sampling({"temperature": 0, "top_p": 1}, {"temperature": 1})
+        assert merged == {"temperature": 1, "top_p": 1}
+
+    def test_canonical_json_is_key_order_stable(self):
+        """缓存 key 与遥测列共用同一序列化口径,键序不得影响结果。"""
+        from polygateway.types import canonical_sampling_json
+
+        assert canonical_sampling_json({"b": 1, "a": 2}) == canonical_sampling_json(
+            {"a": 2, "b": 1}
+        )
+        assert canonical_sampling_json({}) is None
+
+
+class TestSourceConfigExtraBody:
+    """配置级采样参数(issue #4 设计决策 A/E)。"""
+
+    def test_defaults_to_empty_and_is_read_only(self):
+        source = _make_source()
+        assert source.extra_body == {}
+        with pytest.raises(TypeError):
+            source.extra_body["temperature"] = 0  # MappingProxyType 只读
+
+    def test_protected_key_rejected_at_construction(self):
+        """装配期报错,不放到运行时才炸(CLAUDE.md §4.5)。"""
+        with pytest.raises(ValueError, match="model"):
+            _make_source(extra_body={"model": "sneaky"})
+
+    def test_accepts_sampling_params(self):
+        source = _make_source(extra_body={"temperature": 0})
+        assert source.extra_body["temperature"] == 0
+
+    def test_replace_rebuilds_proxy(self):
+        """决策 G 的剥离依赖 replace 能重跑 __post_init__ 且不递归。"""
+        source = _make_source(extra_body={"temperature": 0})
+        stripped = dataclasses.replace(source, extra_body={})
+        assert stripped.extra_body == {}
+        with pytest.raises(TypeError):
+            stripped.extra_body["x"] = 1
+
+    def test_no_longer_hashable_is_intentional(self):
+        """加 mapping 字段的固有代价(裸 dict 亦然),库内无调用点会踩。
+
+        锁定为有意行为: 将来踩到的人不应把它当 bug"修"回去——要可变副本用
+        dict(source.extra_body),要改字段用 dataclasses.replace(设计 Task 1)。
+        """
+        with pytest.raises(TypeError):
+            hash(_make_source())