"""OpenAICompatTransport 测试(M1 设计 §4.4/§6):SSE 解析、错误翻译、缺 DONE 语义。

SSE 帧样本按三项目真实网关响应形态二次构造(OpenAI 兼容 chunk 结构)。
"""

import json

import httpx
import pytest
from loguru import logger

from polygateway.errors import (
    RequestRejectedError,
    SourceDeadError,
    TransientError,
)
from polygateway.middleware.telemetry import TelemetryEmitter
from polygateway.pricing import ModelPrice, PricingTable
from polygateway.providers import ProviderProfile, ThinkingWire, register_provider
from polygateway.transports._http_errors import summarize_body
from polygateway.transports.openai_compat import (
    OpenAICompatTransport,
    _iter_sse_deltas,
    _sse_data_payload,
)
from polygateway.types import ChatRequest, LLMResponse, SourceConfig, ThinkingObservation


def _source(**overrides):
    base = {
        "name": "qwen_1",
        "provider": "qwen",
        "base_url": "https://gw.example/v1",
        "api_key": "sk-test",
        "model": "qwen-max",
        "timeout_s": 5.0,
    }
    base.update(overrides)
    return SourceConfig(**base)


def _chunk(content=None, reasoning=None, usage=None, model=None):
    delta = {}
    if content is not None:
        delta["content"] = content
    if reasoning is not None:
        delta["reasoning_content"] = reasoning
    body = {"choices": [{"delta": delta}]} if (delta or usage is None) else {"choices": []}
    if usage is not None:
        body["usage"] = usage
    if model is not None:
        body["model"] = model
    return f"data: {json.dumps(body)}\n\n"


_USAGE = {"prompt_tokens": 11, "completion_tokens": 7}


def _sse_stream(*frames, done=True):
    text = "".join(frames) + ("data: [DONE]\n\n" if done else "")
    return httpx.Response(200, content=text.encode(), headers={"content-type": "text/event-stream"})


def _transport_for(handler, *, registry=None):
    mock = httpx.MockTransport(handler)
    return OpenAICompatTransport(
        client_factory=lambda src: httpx.AsyncClient(base_url=src.base_url, transport=mock),
        registry=registry,
    )


async def _complete(transport, source, *, stream=True, overlay=None):
    return await transport.complete(
        messages=[{"role": "user", "content": "hi"}],
        source=source,
        stream=stream,
        overlay=overlay or {},
        call_id="cid-1",
    )


# 单价刻意取"输出贵于输入"的真实形态: est_tokens 兜底把整估值塞进 completion
# 时,虚高才显形(设计 §1 的 26 倍算例即此单价)。
_PRICING = PricingTable({"qwen-max": ModelPrice(input_per_1m=1.0, output_per_1m=8.0)})


class _MemoryRecorder:
    def __init__(self):
        self.rows = []

    async def record_llm_call(self, **fields):
        self.rows.append(fields)


async def _recorded_cost(result, source):
    """把 transport 产物走一遍真实计费路径,返回落库的 cost。

    `unavailable` → cost=None 的判定在 `TelemetryEmitter` 里(设计 §3.2 #5),
    直接调 `PricingTable.cost` 对 `0/0` 只会得到 `0.0`——那正是本组用例要防的
    假金额,故断言必须穿过 emitter 而不是单测 pricing。
    """
    recorder = _MemoryRecorder()
    response = LLMResponse(
        content=result.content,
        thinking=result.thinking,
        model=source.model,
        provider=source.provider,
        prompt_tokens=result.prompt_tokens,
        completion_tokens=result.completion_tokens,
        latency_ms=1,
        ttft_ms=result.ttft_ms,
        max_inter_token_ms=result.max_inter_token_ms,
        cache_hit=False,
        call_id="cid-1",
        source_name=source.name,
        usage_source=result.usage_source,
    )
    await TelemetryEmitter(recorder, pricing=_PRICING, text_cap=None).emit_attempt(
        request=ChatRequest(messages=[{"role": "user", "content": "hi"}]),
        source=source,
        call_id="cid-1",
        latency_ms=1,
        response=response,
        error=None,
    )
    return recorder.rows[0]["cost"]


class TestSsePureFunctions:
    def test_data_payload_filters_noise(self):
        assert _sse_data_payload("") is None
        assert _sse_data_payload(": ping") is None
        assert _sse_data_payload("event: x") is None
        assert _sse_data_payload("data: {}") == "{}"
        assert _sse_data_payload("data: [DONE]") == "[DONE]"

    async def test_iter_deltas_yields_and_flags_done(self):
        async def lines():
            for raw in [
                _chunk(content="he"),
                ": ping",
                _chunk(reasoning="think"),
                _chunk(usage=_USAGE),
                "data: [DONE]",
            ]:
                for line in raw.splitlines():
                    yield line

        sink = {}
        out = [d async for d in _iter_sse_deltas(lines(), sink)]
        assert out == [(True, "he"), (False, "think")]
        assert sink["done"] is True and sink["usage"] == _USAGE

    async def test_malformed_frame_raises_transient(self):
        async def lines():
            yield "data: {not-json"

        with pytest.raises(TransientError, match="malformed"):
            _ = [d async for d in _iter_sse_deltas(lines(), {})]


class TestStreamHappyPath:
    async def test_full_stream_with_usage(self):
        def handler(request):
            return _sse_stream(
                _chunk(reasoning="ponder"),
                _chunk(content="hello"),
                _chunk(content=" world"),
                _chunk(usage=_USAGE),
            )

        result = await _complete(_transport_for(handler), _source())
        assert result.content == "hello world"
        assert result.thinking == "ponder"
        assert result.prompt_tokens == 11 and result.completion_tokens == 7
        assert result.usage_source == "measured"
        assert result.ttft_ms is not None and result.ttft_ms >= 0

    async def test_qwen_think_tag_stripped(self):
        def handler(request):
            return _sse_stream(_chunk(content="<think>hmm</think>answer"), _chunk(usage=_USAGE))

        result = await _complete(_transport_for(handler), _source())
        assert result.content == "answer"
        assert result.thinking == "hmm"

    async def test_usage_missing_is_unavailable_with_null_cost(self):
        """usage 帧缺失 → 0/0 + unavailable + cost NULL(设计 §3.2 #3)。

        改前拿 `est_tokens` 当实测并整估值塞 completion,同一条调用记成
        `0/4000` → cost 0.032(设计 §1 的 26 倍虚高)。
        """

        def handler(request):
            return _sse_stream(_chunk(content="ok"))

        source = _source(tpm=1000, est_tokens=4000)
        result = await _complete(_transport_for(handler), source)
        assert result.usage_source == "unavailable"
        assert result.prompt_tokens == 0 and result.completion_tokens == 0
        assert await _recorded_cost(result, source) is None


class TestMissingDoneSemantics:
    def _no_done_handler(self, request):
        return _sse_stream(_chunk(content="partial"), _chunk(usage=_USAGE), done=False)

    async def test_default_retry_policy_raises_transient(self):
        with pytest.raises(TransientError, match="missing_done|truncated"):
            await _complete(_transport_for(self._no_done_handler), _source())

    async def test_salvage_with_usage_frame_degrades_to_estimated(self):
        """打捞且收到 usage 帧: 数字真实、可信度降级 → estimated 且照常计费。"""
        source = _source(missing_done="salvage", tpm=1000, est_tokens=4000)
        result = await _complete(_transport_for(self._no_done_handler), source)
        assert result.content == "partial"
        assert result.usage_source == "estimated"
        assert result.prompt_tokens == 11 and result.completion_tokens == 7
        assert await _recorded_cost(result, source) == pytest.approx(
            11 / 1_000_000 * 1.0 + 7 / 1_000_000 * 8.0
        )

    async def test_salvage_without_usage_frame_stays_unavailable(self):
        """打捞且 usage 帧缺失: 0/0 不得被洗成 estimated,否则算出假的 0.0(设计 §3.2 #4)。"""

        def handler(request):
            return _sse_stream(_chunk(content="partial"), done=False)

        source = _source(missing_done="salvage", tpm=1000, est_tokens=4000)
        result = await _complete(_transport_for(handler), source)
        assert result.content == "partial"
        assert result.usage_source == "unavailable"
        assert await _recorded_cost(result, source) is None

    async def test_early_eof_always_transient_even_under_salvage(self):
        def handler(request):
            return _sse_stream(done=False)  # 零内容提前断流

        with pytest.raises(TransientError, match="early_eof"):
            await _complete(_transport_for(handler), _source(missing_done="salvage"))


class TestEmptyCompletion:
    """空补全 → TransientError(2026-07-20 人类裁决;MiniMax 间歇形态,绝不缓存)。"""

    async def test_stream_zero_content_with_done_is_transient(self):
        def handler(request):
            return _sse_stream(_chunk(reasoning="only thinking"), _chunk(usage=_USAGE))

        with pytest.raises(TransientError, match="empty_completion"):
            await _complete(_transport_for(handler), _source())

    async def test_non_stream_empty_content_is_transient(self):
        def handler(request):
            return httpx.Response(
                200, json={"choices": [{"message": {"content": ""}}], "usage": _USAGE}
            )

        with pytest.raises(TransientError, match="empty_completion"):
            await _complete(_transport_for(handler), _source(), stream=False)

    async def test_think_only_content_after_strip_is_transient(self):
        def handler(request):
            return _sse_stream(_chunk(content="<think>hmm</think>"), _chunk(usage=_USAGE))

        with pytest.raises(TransientError, match="empty_completion"):
            await _complete(_transport_for(handler), _source())


class TestObservabilityFields:
    """issue #3: 供应商 prompt cache 命中数与 API 实际返回的模型版本串。

    网关报文一律不可信: 形态异常只归 None,绝不因一个可观测字段打断调用。
    """

    def _cached_usage(self, cached):
        return {**_USAGE, "prompt_tokens_details": {"cached_tokens": cached}}

    async def test_stream_reads_cached_tokens(self):
        def handler(request):
            return _sse_stream(_chunk(content="ok"), _chunk(usage=self._cached_usage(128)))

        result = await _complete(_transport_for(handler), _source())
        assert result.cached_prompt_tokens == 128

    async def test_non_stream_reads_cached_tokens(self):
        def handler(request):
            return httpx.Response(
                200,
                json={
                    "choices": [{"message": {"content": "42"}}],
                    "usage": self._cached_usage(128),
                },
            )

        result = await _complete(_transport_for(handler), _source(), stream=False)
        assert result.cached_prompt_tokens == 128

    async def test_zero_cached_tokens_is_a_real_zero(self):
        """0(真实零命中)与 None(该源未上报)必须可区分——issue #3 的核心诉求。"""

        def handler(request):
            return _sse_stream(_chunk(content="ok"), _chunk(usage=self._cached_usage(0)))

        result = await _complete(_transport_for(handler), _source())
        assert result.cached_prompt_tokens == 0

    async def test_usage_without_details_is_none(self):
        def handler(request):
            return _sse_stream(_chunk(content="ok"), _chunk(usage=_USAGE))

        result = await _complete(_transport_for(handler), _source())
        assert result.cached_prompt_tokens is None

    async def test_missing_usage_frame_is_none(self):
        def handler(request):
            return _sse_stream(_chunk(content="ok"))

        result = await _complete(_transport_for(handler), _source())
        assert result.cached_prompt_tokens is None

    @pytest.mark.parametrize("bad", ["abc", -1, True, 1.5, None, [], {"x": 1}])
    async def test_malformed_cached_tokens_degrade_to_none(self, bad):
        """`True` 必须排除: Python 里 isinstance(True, int) 为真。"""

        def handler(request):
            return _sse_stream(_chunk(content="ok"), _chunk(usage=self._cached_usage(bad)))

        result = await _complete(_transport_for(handler), _source())
        assert result.cached_prompt_tokens is None

    async def test_details_not_a_dict_is_none(self):
        def handler(request):
            usage = {**_USAGE, "prompt_tokens_details": "oops"}
            return _sse_stream(_chunk(content="ok"), _chunk(usage=usage))

        result = await _complete(_transport_for(handler), _source())
        assert result.cached_prompt_tokens is None

    async def test_non_stream_reads_reported_model(self):
        def handler(request):
            return httpx.Response(
                200,
                json={
                    "choices": [{"message": {"content": "42"}}],
                    "usage": _USAGE,
                    "model": "MiniMax-Text-01-250321",
                },
            )

        result = await _complete(_transport_for(handler), _source(), stream=False)
        assert result.model_reported == "MiniMax-Text-01-250321"

    async def test_stream_keeps_the_first_reported_model(self):
        """末帧异常值不得覆盖首帧: 首次写入即固定。"""

        def handler(request):
            return _sse_stream(
                _chunk(content="a", model="MiniMax-Text-01-250321"),
                _chunk(content="b", model="something-else"),
                _chunk(usage=_USAGE),
            )

        result = await _complete(_transport_for(handler), _source())
        assert result.model_reported == "MiniMax-Text-01-250321"

    async def test_empty_first_model_does_not_block_a_later_real_one(self):
        """首帧报空串不得锁死 sink: 守卫按"有效值"判断,否则真实版本会丢。"""

        def handler(request):
            return _sse_stream(
                _chunk(content="a", model=""),
                _chunk(content="b", model="MiniMax-Text-01-250321"),
                _chunk(usage=_USAGE),
            )

        result = await _complete(_transport_for(handler), _source())
        assert result.model_reported == "MiniMax-Text-01-250321"

    @pytest.mark.parametrize("bad", [None, "", "   ", 123, {}])
    async def test_missing_or_malformed_model_is_none(self, bad):
        def handler(request):
            body = {"choices": [{"message": {"content": "42"}}], "usage": _USAGE}
            if bad is not None:
                body["model"] = bad
            return httpx.Response(200, json=body)

        result = await _complete(_transport_for(handler), _source(), stream=False)
        assert result.model_reported is None

    async def test_raw_payload_is_unchanged(self):
        """新字段是独立格子,不改动 raw 的既有内容。"""

        def handler(request):
            return _sse_stream(_chunk(content="ok"), _chunk(usage=self._cached_usage(5)))

        result = await _complete(_transport_for(handler), _source())
        assert set(result.raw) == {"usage"}


class TestReasoningTokens:
    """issue #6: 推理消耗的输出 token,与 issue #3 的 cached_tokens 对称。

    实测三家供应商在"未推理"时是整个 completion_tokens_details 缺失,无人上报
    0;且中转在上游不返回 usage 时会本地补算并吃掉该对象。故 None 的语义是
    "本次调用未上报",不是"该源不上报"(findings §4c)。
    """

    def _reasoning_usage(self, reasoning):
        return {**_USAGE, "completion_tokens_details": {"reasoning_tokens": reasoning}}

    async def test_stream_reads_reasoning_tokens(self):
        def handler(request):
            return _sse_stream(_chunk(content="ok"), _chunk(usage=self._reasoning_usage(7)))

        result = await _complete(_transport_for(handler), _source())
        assert result.reasoning_tokens == 7

    async def test_non_stream_reads_reasoning_tokens(self):
        def handler(request):
            return httpx.Response(
                200,
                json={
                    "choices": [{"message": {"content": "42"}}],
                    "usage": self._reasoning_usage(7),
                },
            )

        result = await _complete(_transport_for(handler), _source(), stream=False)
        assert result.reasoning_tokens == 7

    async def test_zero_reasoning_tokens_is_a_real_zero(self):
        """0(上报了且确实没推理)与 None(本次未上报)必须可区分。"""

        def handler(request):
            return _sse_stream(_chunk(content="ok"), _chunk(usage=self._reasoning_usage(0)))

        result = await _complete(_transport_for(handler), _source())
        assert result.reasoning_tokens == 0

    async def test_usage_without_details_is_none(self):
        def handler(request):
            return _sse_stream(_chunk(content="ok"), _chunk(usage=_USAGE))

        result = await _complete(_transport_for(handler), _source())
        assert result.reasoning_tokens is None

    @pytest.mark.parametrize("bad", ["abc", -1, True, 1.5, None, [], {"x": 1}])
    async def test_malformed_reasoning_tokens_degrade_to_none(self, bad):
        """`True` 必须排除: Python 里 isinstance(True, int) 为真。"""

        def handler(request):
            return _sse_stream(_chunk(content="ok"), _chunk(usage=self._reasoning_usage(bad)))

        result = await _complete(_transport_for(handler), _source())
        assert result.reasoning_tokens is None

    async def test_details_not_a_dict_is_none(self):
        def handler(request):
            usage = {**_USAGE, "completion_tokens_details": "oops"}
            return _sse_stream(_chunk(content="ok"), _chunk(usage=usage))

        result = await _complete(_transport_for(handler), _source())
        assert result.reasoning_tokens is None

    async def test_salvage_path_records_none_not_zero(self):
        """打捞路径拿不到 usage 帧: 记 None(未知)而非 0(确定没推理)。"""

        def handler(request):
            return _sse_stream(_chunk(content="ok"), done=False)

        result = await _complete(_transport_for(handler), _source(missing_done="salvage"))
        assert result.reasoning_tokens is None


class TestThinkingObservationVerdict:
    """issue #16/#17: 两条组装路径都必须裁定"推理到底发生没发生"。

    流式与非流式各测一遍是刻意的——只填一条路径正是本 issue 的根因形态:
    库在其中一条路径上悄悄给出了不同的可观测性,下游无从分辨。
    """

    def _reasoning_usage(self, reasoning):
        return {**_USAGE, "completion_tokens_details": {"reasoning_tokens": reasoning}}

    async def test_stream_reasoning_content_is_observed(self):
        def handler(request):
            return _sse_stream(
                _chunk(reasoning="想一下"), _chunk(content="ok"), _chunk(usage=_USAGE)
            )

        result = await _complete(_transport_for(handler), _source())
        assert result.thinking_observation is ThinkingObservation.OBSERVED

    async def test_stream_without_any_signal_is_unknown(self):
        """无正文、无 details: 库不知道,就如实说不知道。"""

        def handler(request):
            return _sse_stream(_chunk(content="ok"), _chunk(usage=_USAGE))

        result = await _complete(_transport_for(handler), _source())
        assert result.thinking_observation is ThinkingObservation.UNKNOWN

    async def test_stream_zero_reasoning_tokens_is_absent(self):
        """上游明确上报 0 才算 ABSENT——这是唯一的"确实没推理"证据。"""

        def handler(request):
            return _sse_stream(_chunk(content="ok"), _chunk(usage=self._reasoning_usage(0)))

        result = await _complete(_transport_for(handler), _source())
        assert result.thinking_observation is ThinkingObservation.ABSENT

    async def test_non_stream_reasoning_content_is_observed(self):
        def handler(request):
            return httpx.Response(
                200,
                json={
                    "choices": [{"message": {"content": "42", "reasoning_content": "想一下"}}],
                    "usage": _USAGE,
                },
            )

        result = await _complete(_transport_for(handler), _source(), stream=False)
        assert result.thinking_observation is ThinkingObservation.OBSERVED

    async def test_non_stream_without_any_signal_is_unknown(self):
        """M3 非流式实测形态: 推理已计费却既不回传正文也不回传 details。"""

        def handler(request):
            return httpx.Response(
                200, json={"choices": [{"message": {"content": "42"}}], "usage": _USAGE}
            )

        result = await _complete(_transport_for(handler), _source(), stream=False)
        assert result.thinking_observation is ThinkingObservation.UNKNOWN

    async def test_non_stream_zero_reasoning_tokens_is_absent(self):
        def handler(request):
            return httpx.Response(
                200,
                json={
                    "choices": [{"message": {"content": "42"}}],
                    "usage": self._reasoning_usage(0),
                },
            )

        result = await _complete(_transport_for(handler), _source(), stream=False)
        assert result.thinking_observation is ThinkingObservation.ABSENT


class TestThinkingReconciliation:
    """对账告警按 (source, model, direction) 节流(设计 §5)。

    键的三段缺一不可,理由同源: 合并任意一段,都会让先出现的那一组把另一组
    永久静音——同一模型的开/关两档是两个独立的矛盾,同一模型的两个源背后是
    两个独立的账号/网关。
    """

    def _handler(self, request):
        payload = json.loads(request.content)
        if payload.get("reasoning_effort") == "none":
            # 关闭档却回了推理正文 → OBSERVED,与"要求关闭"矛盾
            return _sse_stream(
                _chunk(reasoning="偷偷想了"), _chunk(content="ok"), _chunk(usage=_USAGE)
            )
        # 开启档却零信号 → UNKNOWN,无法确认是否生效(M3 实测形态)
        return _sse_stream(_chunk(content="ok"), _chunk(usage=_USAGE))

    def _minimax(self, enable_thinking, name="mm"):
        return _source(
            name=name, provider="minimax", model="MiniMax-M3", enable_thinking=enable_thinking
        )

    async def test_same_model_and_direction_warns_only_once(self):
        transport = _transport_for(self._handler)
        source = self._minimax(False)
        messages: list[str] = []
        sink_id = logger.add(messages.append, level="WARNING")
        try:
            await _complete(transport, source)
            await _complete(transport, source)
        finally:
            logger.remove(sink_id)
        hits = [m for m in messages if "MiniMax-M3" in m]
        assert len(hits) == 1, f"同一 (model, direction) 应只告警一次,实得 {len(hits)} 次"

    async def test_each_source_gets_its_own_warning(self):
        """多源多账号是本库的核心场景: 同一 model 跨 N 个源不得只喊第一个。

        节流键漏掉源标识时,5 个共用同一模型的源里第一个出问题的喊完一次,其余
        四个**永久静音**——而每个源背后是独立的账号/网关,它们的行为互不代表。
        """
        transport = _transport_for(self._handler)
        messages: list[str] = []
        sink_id = logger.add(messages.append, level="WARNING")
        try:
            await _complete(transport, self._minimax(False, name="gw-a"))
            await _complete(transport, self._minimax(False, name="gw-b"))
        finally:
            logger.remove(sink_id)
        hits = [m for m in messages if "MiniMax-M3" in m]
        assert len(hits) == 2, f"两个源各应告警一次,实得 {len(hits)} 次"

    async def test_the_warning_names_the_source(self):
        """拿到告警的人得知道该查哪个网关: 只报模型名定位不到源。"""
        transport = _transport_for(self._handler)
        messages: list[str] = []
        sink_id = logger.add(messages.append, level="WARNING")
        try:
            await _complete(transport, self._minimax(False, name="gw-a"))
        finally:
            logger.remove(sink_id)
        hits = [m for m in messages if "MiniMax-M3" in m]
        assert len(hits) == 1
        assert "gw-a" in hits[0], f"告警未点名出问题的源: {hits[0]}"

    async def test_switching_direction_earns_a_second_warning(self):
        transport = _transport_for(self._handler)
        messages: list[str] = []
        sink_id = logger.add(messages.append, level="WARNING")
        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))
        finally:
            logger.remove(sink_id)
        hits = [m for m in messages if "MiniMax-M3" in m]
        assert len(hits) == 2, f"两个方向各应告警一次,实得 {len(hits)} 次"


class TestNonStreamFastPath:
    async def test_non_stream_parses_message(self):
        def handler(request):
            body = json.loads(request.content)
            assert body.get("stream") is False and "stream_options" not in body
            return httpx.Response(
                200,
                json={
                    "choices": [{"message": {"content": "42", "reasoning_content": "count"}}],
                    "usage": _USAGE,
                },
            )

        result = await _complete(_transport_for(handler), _source(), stream=False)
        assert result.content == "42" and result.thinking == "count"
        assert result.usage_source == "measured"
        assert result.ttft_ms is None


class TestRequestShaping:
    @pytest.mark.parametrize(
        ("enable_thinking", "expected"),
        [(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))
        assert {k: seen[k] for k in expected} == expected
        if enable_thinking is None:
            assert "enable_thinking" not in seen
        assert seen["stream_options"] == {"include_usage": True}

    @pytest.mark.parametrize(
        ("enable_thinking", "expected"),
        # 开档不再附 medium(2026-09-04): 那是替下游做的档位判断,且 medium 不在
        # GLM/kimi/deepseek 的档位表里。MiniMax 开启档本就无需参数,要强度请配档位
        [(True, None), (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)
        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,两行不可调换。"""
        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_unregistered_model_warns_only_once_per_source(self):
        """未登记模型的告警不能打在请求热路径上: 装配期已喊过,逐次再喊是刷屏。"""

        def handler(request):
            return _sse_stream(_chunk(content="x"), _chunk(usage=_USAGE))

        source = _source(name="mm", provider="minimax", model="MiniMax-M99", enable_thinking=False)
        transport = _transport_for(handler)
        messages: list[str] = []
        sink_id = logger.add(messages.append, level="WARNING")
        try:
            await _complete(transport, source)
            await _complete(transport, source)
            await _complete(transport, source)
        finally:
            logger.remove(sink_id)
        hits = [m for m in messages if "MiniMax-M99" in m]
        assert len(hits) == 1, f"三次调用应只告警一次,实得 {len(hits)} 次"

    async def test_unrelated_value_error_is_not_mislabelled(self, monkeypatch):
        """只捕 ThinkingUnsupportedError: 无关的 ValueError 不该被贴成推理开关的错。

        今天 `_build_payload` 里只有 resolve_thinking 会抛 ValueError,所以这条
        是防御未来 —— 但正因如此才要钉住: 将来谁在那里加一处校验,宽 catch 会
        把它的错误信息盖掉,而这个用例会先红。
        """

        def handler(request):  # pragma: no cover - 不该走到发请求
            raise AssertionError("请求不该发出")

        def _boom(*args, **kwargs):
            raise ValueError("故意的无关错误")

        monkeypatch.setattr("polygateway.transports.openai_compat.resolve_thinking", _boom)
        with pytest.raises(ValueError, match="故意的无关错误") as exc:
            await _complete(_transport_for(handler), _source(enable_thinking=False))
        assert "推理开关" not in str(exc.value)
        assert not isinstance(exc.value, RequestRejectedError)

    async def test_unknown_shape_is_rejected(self):
        def handler(request):  # pragma: no cover - 不该走到发请求
            raise AssertionError("请求不该发出")

        # 2026-09-04 起默认表 8 段全部有形态,守卫样本改为显式注册的未知段
        mystery = ProviderProfile(
            name="mystery",
            thinking=ThinkingWire(off=None, on_base=None, effort_key=None),
            strip_think_tags=False,
        )
        source = _source(name="k3", provider="mystery", model="kimi-k3", enable_thinking=False)
        with pytest.raises(RequestRejectedError, match="register_provider"):
            await _complete(_transport_for(handler, registry=register_provider(mystery)), source)

    async def test_overlay_merged_into_payload(self):
        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(),
            overlay={"response_format": {"type": "json_object"}},
        )
        assert seen["response_format"] == {"type": "json_object"}

    async def test_extra_body_merged_and_outranked_by_overlay(self):
        """顺序即优先级: thinking profile → extra_body → overlay(issue #4)。"""
        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(extra_body={"temperature": 0, "top_p": 0.9}),
            overlay={"temperature": 1},
        )
        assert seen["temperature"] == 1  # 调用级覆盖配置级
        assert seen["top_p"] == 0.9  # 未被顶掉的配置级键保留

    async def test_extra_body_cannot_break_governed_keys(self):
        """治理键由 payload 骨架拥有;extra_body 的保护键在构造期已被拦下。"""
        with pytest.raises(ValueError, match="stream"):
            _source(extra_body={"stream": False})


class TestErrorTranslation:
    @pytest.mark.parametrize(
        ("status", "body", "headers", "exc", "match"),
        [
            (401, "{}", {}, SourceDeadError, None),
            (403, "{}", {}, SourceDeadError, None),
            (400, "{}", {}, RequestRejectedError, None),
            (418, "{}", {}, RequestRejectedError, None),
            (500, "{}", {}, TransientError, None),
            (503, "{}", {}, TransientError, None),
            (429, json.dumps({"error": {"type": "insufficient_quota"}}), {}, SourceDeadError, None),
            (429, "{}", {"retry-after": "2.5"}, TransientError, None),
        ],
    )
    async def test_status_translation(self, status, body, headers, exc, match):
        def handler(request):
            return httpx.Response(status, content=body.encode(), headers=headers)

        with pytest.raises(exc) as ei:
            await _complete(_transport_for(handler), _source())
        assert ei.value.status_code == status
        assert ei.value.source_name == "qwen_1"
        if status == 429 and exc is TransientError:
            assert ei.value.retry_after_s == 2.5

    async def test_retry_after_http_date_ignored(self):
        def handler(request):
            return httpx.Response(
                429, content=b"{}", headers={"retry-after": "Wed, 21 Oct 2026 07:28:00 GMT"}
            )

        with pytest.raises(TransientError) as ei:
            await _complete(_transport_for(handler), _source())
        assert ei.value.retry_after_s is None  # 仅支持秒数形态(CHS 同款)

    async def test_httpx_timeout_becomes_transient(self):
        def handler(request):
            raise httpx.ConnectTimeout("boom")

        with pytest.raises(TransientError):
            await _complete(_transport_for(handler), _source())

    async def test_httpx_transport_error_becomes_transient(self):
        def handler(request):
            raise httpx.RemoteProtocolError("broken pipe")

        with pytest.raises(TransientError):
            await _complete(_transport_for(handler), _source())


# issue #10 原文给出的真实响应体(一字不改): 关键在于 code 收尾
_REJECT_BODY = (
    '{"error":{"message":"<400> ***.***.InvalidParameter: The image format is illegal '
    'and cannot be opened","type":"invalid_request_error","param":"",'
    '"code":"invalid_parameter_error"}}'
)


class TestErrorBodyRetention:
    """issue #10: 网关说了什么必须活着离开翻译层——message 与字段各留一份。"""

    @pytest.mark.parametrize(
        ("status", "exc"),
        [
            (400, RequestRejectedError),
            (401, SourceDeadError),
            (403, SourceDeadError),
            (404, RequestRejectedError),
            (500, TransientError),
            (503, TransientError),
        ],
    )
    async def test_every_non_2xx_branch_keeps_the_body(self, status, exc):
        def handler(request):
            return httpx.Response(status, content=_REJECT_BODY.encode())

        with pytest.raises(exc) as ei:
            await _complete(_transport_for(handler), _source())
        assert ei.value.body_text == _REJECT_BODY
        # message 与字段共用同一份串: 遥测里看到的与下游 catch 到的不得打架
        assert str(ei.value).endswith(f" | {_REJECT_BODY}")
        assert "invalid_parameter_error" in str(ei.value)

    async def test_rate_limited_429_keeps_body_and_retry_after(self):
        def handler(request):
            return httpx.Response(
                429,
                content=b'{"error":{"message":"per-minute cap 3"}}',
                headers={"retry-after": "2.5"},
            )

        with pytest.raises(TransientError) as ei:
            await _complete(_transport_for(handler), _source())
        assert "per-minute cap 3" in str(ei.value)
        assert ei.value.retry_after_s == 2.5  # 摘要不得干扰既有解析

    async def test_insufficient_quota_429_keeps_body(self):
        body = json.dumps(
            {"error": {"type": "insufficient_quota", "message": "daily budget spent"}}
        )

        def handler(request):
            return httpx.Response(429, content=body.encode())

        with pytest.raises(SourceDeadError) as ei:
            await _complete(_transport_for(handler), _source())
        assert "daily budget spent" in str(ei.value)

    async def test_oversized_insufficient_quota_still_classified_dead(self):
        """实现红线: 类型判定必须读**原文**。

        摘要会破坏 JSON 结构,若改用摘要解析,超长 body 的配额耗尽将退化成普通
        限速——配额已耗尽的源不再 force_open,一个诊断改进就变成了治理 bug。

        **填充必须是多个键**,不能是单个超长字符串值: 后者的截断点落在字符串
        *内部*,省略标记成了合法的字符串内容,而头尾保留又让尾部的 error 对象
        幸存——摘要照样解析得出 `insufficient_quota`,用例即告空转(2026-08-16
        verifier 变异测试发现: 按错误写法实现,全套件 824 项依然全绿)。多键
        填充让截断点落在结构记号之间,摘要才真正不可解析。
        """
        body = json.dumps(
            {**{f"k{i}": "v" * 10 for i in range(300)}, "error": {"type": "insufficient_quota"}}
        )
        assert len(body) > 2048
        with pytest.raises(json.JSONDecodeError):
            # 判别力的前提: 摘要确实不再是合法 JSON,读它必然拿不到 type
            json.loads(summarize_body(body))

        def handler(request):
            return httpx.Response(429, content=body.encode())

        with pytest.raises(SourceDeadError):
            await _complete(_transport_for(handler), _source())

    async def test_empty_body_leaves_no_dangling_separator(self):
        def handler(request):
            return httpx.Response(400, content=b"")

        with pytest.raises(RequestRejectedError) as ei:
            await _complete(_transport_for(handler), _source())
        assert str(ei.value) == "qwen_1 请求被拒: 400"
        assert ei.value.body_text == ""

    @pytest.mark.parametrize("body", [b"<html>gateway down</html>", b"\xff\xfe not utf-8"])
    async def test_non_json_and_non_utf8_bodies_do_not_explode(self, body):
        def handler(request):
            return httpx.Response(400, content=body)

        with pytest.raises(RequestRejectedError) as ei:
            await _complete(_transport_for(handler), _source())
        assert ei.value.status_code == 400  # 分类不受 body 形态影响

    async def test_non_stream_path_keeps_the_body(self):
        def handler(request):
            return httpx.Response(400, content=_REJECT_BODY.encode())

        with pytest.raises(RequestRejectedError) as ei:
            await _complete(_transport_for(handler), _source(), stream=False)
        assert ei.value.body_text == _REJECT_BODY

    async def test_embedding_path_keeps_the_body(self):
        def handler(request):
            return httpx.Response(400, content=_REJECT_BODY.encode())

        with pytest.raises(RequestRejectedError) as ei:
            await _transport_for(handler).embed(texts=["hi"], source=_source(), call_id="cid-embed")
        assert ei.value.body_text == _REJECT_BODY


class TestLifecycle:
    async def test_aclose_idempotent(self):
        def handler(request):
            return _sse_stream(_chunk(content="x"), _chunk(usage=_USAGE))

        transport = _transport_for(handler)
        await _complete(transport, _source())
        await transport.aclose()
        await transport.aclose()