"""OpenAI 兼容 transport(D2 默认): 手写 httpx + SSE 解析 + 错误翻译。

SSE 纯函数移植 VT `adapters/llm.py:51-124`;错误翻译移植 CHS
`app/providers/invokers.py:127-227`。职责只到"一次原始调用"——重试/限流/
缓存归中间件。看门狗活性口径: content 与 reasoning_content 增量都作为流
元素产出,思考流天然刷新计时(CHS 迁移约束 R1)。
"""

from __future__ import annotations

import json
import re
import time
from typing import TYPE_CHECKING, Any

import httpx
from loguru import logger

from polygateway.errors import (
    PolyGatewayError,
    RequestRejectedError,
    ResultInvalidError,
    SourceDeadError,
    TransientError,
)
from polygateway.providers import ProviderProfile, get_provider
from polygateway.streaming import StreamLivenessTimeout, stream_with_liveness_timeouts
from polygateway.thinking import (
    ThinkingCapability,
    ThinkingUnsupportedError,
    get_capability,
    observe_thinking,
    reconcile_thinking,
    resolve_thinking,
)
from polygateway.transports._http_errors import compose_message, summarize_body
from polygateway.types import (
    Effort,
    EmbeddingTransportResult,
    SourceConfig,
    TransportResult,
)

if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Callable, Mapping

_THINK_PATTERN = re.compile(r"<think>(.*?)</think>", re.DOTALL)

# —— SSE 纯函数(VT llm.py 同款)——


def _sse_data_payload(raw: str) -> str | None:
    """提取 SSE data 行载荷;ping(: 开头)/空行/非 data 行返回 None 跳过。"""
    line = raw.strip()
    if not line or line.startswith(":") or not line.startswith("data:"):
        return None
    return line[len("data:") :].strip()


def _sse_delta(chunk: dict[str, Any], usage_sink: dict[str, Any]) -> tuple[bool, str] | None:
    """从 chunk 提取增量: (True, content) 或 (False, reasoning);usage 帧旁路进 sink。"""
    if chunk.get("usage"):
        usage_sink["usage"] = chunk["usage"]
    if "model" not in usage_sink:
        # 首个**有效**值即固定: 末帧的异常值不得覆盖它;但首帧报空串也不能锁死
        # sink——否则后续真实版本会丢(issue #3)
        reported = _coerce_model_reported(chunk.get("model"))
        if reported is not None:
            usage_sink["model"] = reported
    choices = chunk.get("choices") or []
    if not choices:
        return None
    delta = choices[0].get("delta") or {}
    content = delta.get("content")
    if content:
        return (True, content)
    reasoning = delta.get("reasoning_content")
    if reasoning:
        return (False, reasoning)
    return None


async def _iter_sse_deltas(
    lines: AsyncIterator[str], usage_sink: dict[str, Any]
) -> AsyncIterator[tuple[bool, str]]:
    """逐行解析 SSE 流;[DONE] 置 sink["done"];畸形 JSON 帧 → 瞬时错误(可重试)。"""
    async for raw in lines:
        data = _sse_data_payload(raw)
        if data is None:
            continue
        if data == "[DONE]":
            usage_sink["done"] = True
            return
        try:
            chunk = json.loads(data)
        except json.JSONDecodeError as exc:
            raise TransientError(
                f"SSE 帧畸形(malformed_json): {data[:80]!r}", operation="chat"
            ) from exc
        delta = _sse_delta(chunk, usage_sink)
        if delta is not None:
            yield delta


# —— 错误翻译(CHS invokers.py 同款)——


def _parse_retry_after(raw: str | None) -> float | None:
    """解析 Retry-After 头;仅支持秒数形态,HTTP-date 返回 None(CHS 同款)。"""
    if raw is None:
        return None
    try:
        seconds = float(raw.strip())
    except ValueError:
        return None
    return seconds if seconds > 0 else None


def _translate_429(
    source: SourceConfig, body_text: str, headers: Mapping[str, str], ctx: dict[str, Any]
) -> Exception:
    """429 细分。`body_text` 必须是**未截断的原文**——`ctx["body_text"]` 是摘要,
    头尾保留会破坏 JSON 结构,拿它解析会让超长 body 的配额耗尽退化成普通限速
    (该源不再 force_open),把一个诊断改进变成治理 bug(issue #10 实现红线)。
    """
    try:
        err_type = json.loads(body_text).get("error", {}).get("type", "")
    except (json.JSONDecodeError, AttributeError):
        err_type = ""
    summary = ctx["body_text"]
    if err_type == "insufficient_quota":
        return SourceDeadError(
            compose_message(f"{source.name} 配额耗尽(insufficient_quota)", summary), **ctx
        )
    return TransientError(
        compose_message(f"{source.name} 限速: 429", summary),
        retry_after_s=_parse_retry_after(headers.get("retry-after")),
        **ctx,
    )


def _classify(status: int) -> tuple[type[PolyGatewayError], str]:
    """状态码 → (错误类, message 标签);映射与 ARCH §6.2 逐条相同,本次零变更。"""
    if status in (401, 403):
        return SourceDeadError, "凭据失效/欠费"
    if status == 400:
        return RequestRejectedError, "请求被拒"
    if status >= 500:
        return TransientError, "瞬时错误"
    return RequestRejectedError, "客户端错误"


def _status_to_error(
    source: SourceConfig, status: int, body_text: str, headers: Mapping[str, str]
) -> Exception:
    """非 2xx → 领域错误,**全部分支**携带响应体摘要(issue #10)。

    摘要只算一次,message 与 `body_text` 共用同一份串: 两份不同长度会让"遥测里
    看到的"与"下游 catch 到的"对不上,排查时反而多一层困惑。
    """
    summary = summarize_body(body_text)
    ctx: dict[str, Any] = {
        "source_name": source.name,
        "status_code": status,
        "operation": "chat",
        "body_text": summary,
    }
    if status == 429:
        return _translate_429(source, body_text, headers, ctx)
    cls, label = _classify(status)
    return cls(compose_message(f"{source.name} {label}: {status}", summary), **ctx)


def _strip_think(content: str) -> tuple[str, str]:
    """剥离 <think> 标签(qwen 系),返回 (正文, 思考流)。VT llm.py:147-164 同款。"""
    match = _THINK_PATTERN.search(content)
    if match is None:
        return content, ""
    return _THINK_PATTERN.sub("", content).strip(), match.group(1).strip()


def _resolve_usage(usage: dict[str, Any]) -> tuple[int, int, str]:
    """usage 帧读取;缺失/非法记 0/0 并标 unavailable(est_tokens 解耦设计 §3.2 #3)。

    不再拿 `est_tokens` 兜底: 它按 CHS 定义是"最坏情形上界",拿上界当实测值
    只会系统性高估账单;宁可把用量记成显式的"不可得"(cost 随之为 NULL),
    让缺口可被统计,也不编一个看似有效的数字。用量口径自此不依赖源配置,
    故不再收 `SourceConfig`。
    """
    prompt, completion = usage.get("prompt_tokens"), usage.get("completion_tokens")
    if isinstance(prompt, int) and isinstance(completion, int) and prompt + completion > 0:
        return prompt, completion, "measured"
    return 0, 0, "unavailable"


def _coerce_cached_tokens(usage: Any) -> int | None:
    """取 usage.prompt_tokens_details.cached_tokens(issue #3);形态异常一律 None。

    `0` 与 `None` 必须可区分: 前者是"该源上报了一次真实零命中",后者是"该源
    不报这个数",下游对两者的处置不同(后者不可做缓存成本校正)。故只把
    **负数与非整数**归 None,`0` 如实保留。`bool` 显式排除——isinstance(True, int)
    在 Python 里为真,放行会把 `True` 记成 1 个命中 token。
    """
    if not isinstance(usage, dict):
        return None
    details = usage.get("prompt_tokens_details")
    if not isinstance(details, dict):
        return None
    cached = details.get("cached_tokens")
    if isinstance(cached, bool) or not isinstance(cached, int) or cached < 0:
        return None
    return cached


def _coerce_reasoning_tokens(usage: Any) -> int | None:
    """取 usage.completion_tokens_details.reasoning_tokens(issue #6);形态异常一律 None。

    与 `_coerce_cached_tokens` 逐条同构(两者是 OpenAI 兼容 usage 里对称的一对):
    `0` 如实保留、负数与非整数归 None、`bool` 显式排除。差别只在语义——本字段
    的 None 是"**本次调用**未上报"而非"该源不上报": 中转在上游不返回 usage 时
    会本地补算并整体替换 usage 对象,把 details 一并吃掉(findings §4c)。
    """
    if not isinstance(usage, dict):
        return None
    details = usage.get("completion_tokens_details")
    if not isinstance(details, dict):
        return None
    reasoning = details.get("reasoning_tokens")
    if isinstance(reasoning, bool) or not isinstance(reasoning, int) or reasoning < 0:
        return None
    return reasoning


def _coerce_model_reported(value: Any) -> str | None:
    """取响应体的 model 字段(issue #3);非 str 或空白串一律 None,收口时去空白。

    去空白不是洁癖: 下游拿这个串做实验快照的 key,`" m "` 与 `"m"` 会造成假分叉。
    """
    if not isinstance(value, str) or not value.strip():
        return None
    return value.strip()


def _resolve_stream_usage(sink: dict[str, Any], salvaged: bool) -> tuple[int, int, str]:
    """流式用量口径: 打捞路径把 measured 降级为 estimated,unavailable 原样保留。

    前置条件不可省(解耦设计 §3.2 #4): usage 帧本就缺失时 `0/0` 会被洗成
    `estimated`,进而按 token 换算出一个假的 `0.0` 成本。
    """
    prompt, completion, usage_source = _resolve_usage(sink.get("usage") or {})
    if salvaged and usage_source == "measured":
        # 收到 usage 帧但流被截断: 数字真实、可信度降级(M1 设计 §6)
        usage_source = "estimated"
    return prompt, completion, usage_source


def _extract_vectors(
    data: dict[str, Any], source: SourceConfig, expected_count: int, ctx: dict[str, Any]
) -> list[list[float]]:
    """按 data[].index 重排提取向量并校验条数/维度一致性。"""
    try:
        vectors = [
            [float(x) for x in item["embedding"]]
            for item in sorted(data["data"], key=lambda it: int(it["index"]))
        ]
    except (KeyError, TypeError, ValueError) as exc:
        raise ResultInvalidError(f"{source.name} embedding 响应形态异常: {exc}", **ctx) from exc
    if len(vectors) != expected_count:
        raise ResultInvalidError(
            f"{source.name} 返回 {len(vectors)} 条向量,与输入 {expected_count} 条不符", **ctx
        )
    if len({len(v) for v in vectors}) != 1 or not vectors[0]:
        raise ResultInvalidError(
            f"{source.name} 向量维度异常: {sorted({len(v) for v in vectors})}", **ctx
        )
    return vectors


def _resolve_embedding_usage(data: dict[str, Any]) -> tuple[int, str]:
    """usage 读取;缺失/非法记 0 并标 unavailable(与 chat 同口径,设计 §3.2 #3)。"""
    prompt = (data.get("usage") or {}).get("prompt_tokens")
    if isinstance(prompt, int) and prompt > 0:
        return prompt, "measured"
    return 0, "unavailable"


def _parse_embedding_payload(
    resp: httpx.Response, source: SourceConfig, expected_count: int
) -> EmbeddingTransportResult:
    """解析 /embeddings 响应;一切形态异常归 ResultInvalidError(坏结果≠坏服务)。"""
    ctx: dict[str, Any] = {"source_name": source.name, "operation": "embedding"}
    try:
        data = resp.json()
    except json.JSONDecodeError as exc:
        raise ResultInvalidError(f"{source.name} embedding 响应非 JSON: {exc}", **ctx) from exc
    vectors = _extract_vectors(data, source, expected_count, ctx)
    prompt_tokens, usage_source = _resolve_embedding_usage(data)
    return EmbeddingTransportResult(
        vectors=vectors,
        dim=len(vectors[0]),
        prompt_tokens=prompt_tokens,
        usage_source=usage_source,
        raw={"id": data.get("id")},
    )


def _default_client_factory(source: SourceConfig) -> httpx.AsyncClient:
    return httpx.AsyncClient(
        headers={"Authorization": f"Bearer {source.api_key}"},
        timeout=httpx.Timeout(source.timeout_s),
        trust_env=source.trust_env,
    )


class OpenAICompatTransport:
    """默认 transport: 每源一个预配 httpx client,懒创建,aclose 统一释放。"""

    def __init__(
        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
        # 未登记模型只喊一次: 装配期已喊过,逐次调用再喊是日志洪水。
        # 实例级而非模块级 —— 模块级可变状态违反纯 asyncio 中立铁律
        self._warned_models: set[str] = set()
        # 对账告警独立节流,**不复用** `_warned_models`: 两者语义不同(那个 set 记
        # 的是"未登记能力已告警过",这个记的是"某源某方向的矛盾已告警过"),共用
        # 一个容器会让两种告警的生命周期纠缠在一起——将来任一侧想加清空/过期策略,
        # 都会连带改掉另一侧的行为。(键空间恰好不相交,故当下**不会**互相压制;
        # 分开维护的理由是语义,不是碰撞)
        self._warned_mismatches: set[tuple[str, str, bool | None]] = set()
        self._client_factory = client_factory or _default_client_factory
        self._clients: dict[str, httpx.AsyncClient] = {}

    def _client_for(self, source: SourceConfig) -> httpx.AsyncClient:
        client = self._clients.get(source.name)
        if client is None:
            client = self._client_factory(source)
            self._clients[source.name] = client
        return client

    def _build_payload(
        self,
        *,
        messages: list[dict[str, Any]],
        source: SourceConfig,
        profile: ProviderProfile,
        stream: bool,
        overlay: dict[str, Any],
    ) -> dict[str, Any]:
        payload: dict[str, Any] = {"model": source.model, "messages": messages, "stream": stream}
        if stream:
            payload["stream_options"] = {"include_usage": True}  # 强制 usage 帧(三项目同款)
        # 形态(provider 级)与能力(model 级)在此相遇;不可满足时 ValueError,
        # 由 complete() 翻译为四分类之一(issue #5)
        capability = get_capability(source.model, table=self._capabilities)
        first_time = source.model not in self._warned_models
        self._warned_models.add(source.model)
        # `enable_thinking` 的档位语法糖(True → auto,False → none,None 不表态);
        # 就地转换是过渡形态,T5 起由 thinking.effective_effort() 统一收口并接上
        # 源级/请求级档位(设计 §4.2)
        enabled = source.enable_thinking
        effort = None if enabled is None else (Effort.AUTO if enabled else Effort.NONE)
        payload.update(
            resolve_thinking(
                profile,
                capability,
                effort,
                model=source.model,
                warn_unregistered=first_time,
            ).payload
        )
        # 顺序即优先级(issue #4 设计决策 A): 配置级 extra_body 在前,调用级
        # overlay(含结构化注入)在后覆盖之。两行不可调换
        payload.update(source.extra_body)
        payload.update(overlay)
        return payload

    async def complete(
        self,
        *,
        messages: list[dict[str, Any]],
        source: SourceConfig,
        stream: bool,
        overlay: dict[str, Any],
        call_id: str,
    ) -> TransportResult:
        """一次原始调用;HTTP/线路/流式异常按 ARCH §6.2 翻译为领域错误。"""
        profile = get_provider(source.provider, registry=self._registry)
        try:
            payload = self._build_payload(
                messages=messages, source=source, profile=profile, stream=stream, overlay=overlay
            )
        except ThinkingUnsupportedError as exc:
            # 推理开关不可满足是**请求本身**的问题: 换源重试都救不了它。只捕这个
            # 专用类型而非宽 catch ValueError —— 后者会把序列化等无关错误误贴标签
            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"}
        try:
            if stream:
                result = await self._complete_stream(client, url, payload, source, profile)
            else:
                result = await self._complete_once(client, url, payload, source, profile)
        except StreamLivenessTimeout as exc:
            raise TransientError(f"{source.name} 流活性超时({exc.kind})", **ctx) from exc
        except httpx.TimeoutException as exc:
            raise TransientError(f"{source.name} 超时: {exc}", **ctx) from exc
        except httpx.TransportError as exc:
            # VT 宽集: 覆盖断连/协议错误/读写失败(设计 §9 行 8)
            raise TransientError(f"{source.name} 网络错误: {exc}", **ctx) from exc
        # 此处是唯一同时握有请求方向与响应结果的地方,对账只能落在这里
        self._warn_on_thinking_mismatch(source, result)
        return result

    def _warn_on_thinking_mismatch(self, source: SourceConfig, result: TransportResult) -> None:
        """声明与观测矛盾即 warning;按 (source, model, direction) 节流,同组合只喊一次。

        三段缺一不可。**方向**: 同一模型的开、关两档是两个独立的矛盾。**源名**:
        多源多账号是本库的核心场景,同一 model 跨 N 个源是常态,而每个源背后是
        独立的账号/网关,一个源的行为不代表另一个——漏掉源名,5 个源里第一个出
        问题的喊完一次,其余四个永久静音。逐次调用刷屏会把告警变成噪声,噪声等于
        没有告警。

        **先判键再对账**: `reconcile_thinking` 会拼含完整 `evidence` 的长字符串,
        而非流式档每次调用都命中这一分支,节流后再拼是纯粹的热路径浪费。
        """
        key = (source.name, source.model, source.enable_thinking)
        if key in self._warned_mismatches:
            return
        message = reconcile_thinking(
            enable_thinking=source.enable_thinking,
            observation=result.thinking_observation,
            capability=get_capability(source.model, table=self._capabilities),
            model=source.model,
        )
        if message is None:
            return
        self._warned_mismatches.add(key)
        # 源名拼在调用点而不是加进 `reconcile_thinking` 的签名: 那是纯判定函数,
        # 输入只该含判定依据(声明/观测/能力/模型),源名是**定位信息**,进不了判据。
        # 单参数传入 loguru: 文案里带 `thinking:{type:disabled}` 这类字面花括号
        # (能力表 evidence),将来有人给这行加个格式化参数就会炸在成功调用的返回
        # 路径上(与 telemetry/sqlite.py 的缺列告警同一先例)
        logger.warning("源 {} —— {}", source.name, message)

    async def embed(
        self, *, texts: list[str], source: SourceConfig, call_id: str
    ) -> EmbeddingTransportResult:
        """一次原始 embedding 调用(M2 §7): POST /embeddings,错误翻译同 chat。

        响应按 data[].index 重排保序(GovDoc embedding.py:149 / VT :164 同款);
        空 data/长度不符/维度不一致 → ResultInvalidError(坏结果不熔断)。
        """
        if not texts:
            raise ValueError("texts 不能为空(空输入由 EmbeddingClient 短路)")
        url = source.base_url.rstrip("/") + "/embeddings"
        client = self._client_for(source)
        ctx: dict[str, Any] = {"source_name": source.name, "operation": "embedding"}
        try:
            resp = await client.post(url, json={"model": source.model, "input": texts})
        except httpx.TimeoutException as exc:
            raise TransientError(f"{source.name} 超时: {exc}", **ctx) from exc
        except httpx.TransportError as exc:
            raise TransientError(f"{source.name} 网络错误: {exc}", **ctx) from exc
        if resp.status_code != 200:
            raise _status_to_error(source, resp.status_code, resp.text, resp.headers)
        return _parse_embedding_payload(resp, source, len(texts))

    async def _complete_stream(
        self,
        client: httpx.AsyncClient,
        url: str,
        payload: dict[str, Any],
        source: SourceConfig,
        profile: ProviderProfile,
    ) -> TransportResult:
        started = time.monotonic()
        async with client.stream("POST", url, json=payload) as resp:
            if resp.status_code != 200:
                body = (await resp.aread()).decode("utf-8", errors="replace")
                raise _status_to_error(source, resp.status_code, body, resp.headers)
            sink: dict[str, Any] = {}
            guarded = stream_with_liveness_timeouts(
                _iter_sse_deltas(resp.aiter_lines(), sink),
                ttft_s=source.ttft_timeout_s or source.timeout_s,
                inter_token_s=source.inter_token_timeout_s or source.timeout_s,
                total_s=source.timeout_s,
            )
            content_parts: list[str] = []
            thinking_parts: list[str] = []
            ttft_ms: float | None = None
            last = started
            max_gap = 0.0
            async for is_content, text in guarded:
                now = time.monotonic()
                if ttft_ms is None:
                    ttft_ms = (now - started) * 1000
                else:
                    max_gap = max(max_gap, (now - last) * 1000)
                last = now
                (content_parts if is_content else thinking_parts).append(text)
        salvaged = self._check_done(sink, content_parts, thinking_parts, source)
        content, thinking = self._finalize_text(content_parts, thinking_parts, profile)
        self._reject_empty_completion(content, source)
        prompt, completion, usage_source = _resolve_stream_usage(sink, salvaged)
        reasoning_tokens = _coerce_reasoning_tokens(sink.get("usage"))
        return TransportResult(
            content=content,
            thinking=thinking,
            prompt_tokens=prompt,
            completion_tokens=completion,
            usage_source=usage_source,
            ttft_ms=ttft_ms,
            max_inter_token_ms=(max_gap if ttft_ms is not None else None),
            raw={"usage": sink.get("usage")},
            cached_prompt_tokens=_coerce_cached_tokens(sink.get("usage")),
            model_reported=_coerce_model_reported(sink.get("model")),
            reasoning_tokens=reasoning_tokens,
            # 两条组装路径必须同口径裁定: 只在一条路径上给结论,下游就得靠
            # "这次是不是流式"去猜可观测性,那正是 issue #16/#17 的根因形态
            thinking_observation=observe_thinking(
                thinking=thinking, reasoning_tokens=reasoning_tokens
            ),
        )

    def _check_done(
        self,
        sink: dict[str, Any],
        content_parts: list[str],
        thinking_parts: list[str],
        source: SourceConfig,
    ) -> bool:
        """缺 [DONE] 语义(设计 §6): 零内容恒 retry;有内容按 missing_done 策略。"""
        if sink.get("done"):
            return False
        ctx: dict[str, Any] = {"source_name": source.name, "operation": "chat"}
        if not content_parts and not thinking_parts:
            raise TransientError(f"{source.name} SSE early_eof: 零内容断流", **ctx)
        if source.missing_done == "retry":
            raise TransientError(f"{source.name} SSE missing_done: 截断且无 [DONE]", **ctx)
        return True

    def _reject_empty_completion(self, content: str, source: SourceConfig) -> None:
        """空补全 → 瞬时错误(2026-07-20 人类裁决,M1 验证发现)。

        服务 200 且流程完整([DONE]/usage 正常)但 content 为空——MiniMax 等
        网关的间歇异常形态。视为服务抖动: 退避重试/换源,**绝不缓存空响应**;
        承 CHS "VLM 零 content"归瞬时的先例(invokers.py:309)。
        """
        if not content.strip():
            raise TransientError(
                f"{source.name} 空补全(empty_completion): 流程完整但零内容",
                source_name=source.name,
                operation="chat",
            )

    def _finalize_text(
        self, content_parts: list[str], thinking_parts: list[str], profile: ProviderProfile
    ) -> tuple[str, str]:
        content = "".join(content_parts)
        thinking = "".join(thinking_parts)
        if profile.strip_think_tags and "<think>" in content:
            content, tag_thinking = _strip_think(content)
            if tag_thinking:
                thinking = tag_thinking
        return content, thinking

    async def _complete_once(
        self,
        client: httpx.AsyncClient,
        url: str,
        payload: dict[str, Any],
        source: SourceConfig,
        profile: ProviderProfile,
    ) -> TransportResult:
        """非流式快路径(三项目均无,库新增): 单 JSON 响应,仅 total 超时。"""
        resp = await client.post(url, json=payload)
        if resp.status_code != 200:
            raise _status_to_error(source, resp.status_code, resp.text, resp.headers)
        try:
            body = resp.json()
        except json.JSONDecodeError as exc:
            raise TransientError(
                f"{source.name} 非流式响应非法 JSON", source_name=source.name, operation="chat"
            ) from exc
        choices = body.get("choices") or []
        if not choices:
            raise TransientError(
                f"{source.name} 响应缺 choices", source_name=source.name, operation="chat"
            )
        message = choices[0].get("message") or {}
        content, thinking = self._finalize_text(
            [message.get("content") or ""], [message.get("reasoning_content") or ""], profile
        )
        self._reject_empty_completion(content, source)
        prompt, completion, usage_source = _resolve_usage(body.get("usage") or {})
        reasoning_tokens = _coerce_reasoning_tokens(body.get("usage"))
        return TransportResult(
            content=content,
            thinking=thinking,
            prompt_tokens=prompt,
            completion_tokens=completion,
            usage_source=usage_source,
            ttft_ms=None,
            max_inter_token_ms=None,
            raw={"usage": body.get("usage")},
            cached_prompt_tokens=_coerce_cached_tokens(body.get("usage")),
            model_reported=_coerce_model_reported(body.get("model")),
            reasoning_tokens=reasoning_tokens,
            # 本路径的裁定多半落 UNKNOWN(M3 实测: 推理已计费却正文与 details 双
            # 缺)。如实标记"观测不到",好过让下游误读成"没推理"
            thinking_observation=observe_thinking(
                thinking=thinking, reasoning_tokens=reasoning_tokens
            ),
        )

    async def aclose(self) -> None:
        """幂等关闭全部源 client。"""
        clients, self._clients = self._clients, {}
        for client in clients.values():
            await client.aclose()