"""GatewayClient: 唯一组装层(D1 组装点)+ from_env/from_settings 工厂 + gather_bounded。

90% 用户三行起步: `client = GatewayClient.from_env(); resp = await client.chat(messages)`。
多逻辑角色 = 每角色调一次 `from_env(scope=...)`;共享限流/熔断状态 = 自建
后端实例(源表取并集)显式传入多次调用——共享必须显式,禁止隐式全局
(VT `evolve_llm = llm` 教训;ARCH §7.7 R5)。
"""

from __future__ import annotations

import asyncio
import dataclasses
import hashlib
import json
import random
import time
from typing import TYPE_CHECKING, Any, Literal

from polygateway.backends.memory.breaker import InMemoryGate
from polygateway.backends.memory.cache import InMemoryCache
from polygateway.backends.memory.limiter import InMemoryLimiter
from polygateway.config import GatewaySettings, check_hedge_assembly
from polygateway.deadline import ensure_call_deadline, with_call_deadline
from polygateway.errors import PolyGatewayError
from polygateway.middleware.base import compose
from polygateway.middleware.cache import CacheMW
from polygateway.middleware.retry import RetryMW
from polygateway.middleware.structured import StructuredMW
from polygateway.middleware.telemetry import TelemetryEmitter, TelemetryMW, emit_terminal_once
from polygateway.ports import TelemetryStatusProvider
from polygateway.pricing import PricingTable
from polygateway.providers import get_provider
from polygateway.sources import (
    AdaptivePacer,
    HealthAwareSelector,
    LeastInflightSelector,
    RoundRobinSelector,
    SourceCooldownMemo,
)
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,
    Effort,
    LLMResponse,
    TelemetryStatus,
    _CallContext,
    coerce_effort,
    validate_caller_dimensions,
    validate_request_overlay,
)

if TYPE_CHECKING:
    from collections.abc import Awaitable, Iterable, Mapping

    from pydantic import BaseModel

    from polygateway.ports import (
        CacheBackend,
        Middleware,
        ProviderGate,
        RateLimiter,
        SourceSelector,
        StructuredOutputStrategy,
        TelemetryRecorder,
        Transport,
    )
    from polygateway.providers import ProviderProfile
    from polygateway.thinking import ThinkingCapability
    from polygateway.types import (
        BackpressurePolicy,
        RetryPolicy,
        SourceConfig,
    )


def _guard_thinking(
    sources: list[SourceConfig],
    profiles: list[ProviderProfile],
    capabilities: Mapping[str, ThinkingCapability] | None,
) -> None:
    """装配期把不可满足的推理开关炸掉,而不是留到运行时(issue #5)。

    与 transport 内的同一次判定不是重复: 那里兜的是"构造函数全量注入"这条路
    (CLAUDE.md §4.5 的第二条装配路),而工厂路占 90% 场景,配置错误应当在装配期
    就带着指路信息炸掉。`get_provider` 现在就是同一形态的双点调用。
    """
    for source, profile in zip(sources, profiles, strict=True):
        effort = 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:
    """单源的指纹标记;`enable_thinking` 与 `reasoning_effort` 仅在**表态时**追加。

    只在表态时追加不是省事: 这样只配了 `extra_body` 的存量源字面量与 issue #4
    时期逐字相同,升级本版本不会给它们平白来一次全量缓存冷启动。

    `reasoning_effort`(issue #20)与 `enable_thinking` 同规则、同理由: 它一旦真正
    改变请求体,"把源级档位从 low 改成 max 后重启"就会读到 low 档时缓存的旧响应。
    两者取值域不相交(`"none"`/`"low"`… vs `true`/`false`),故追加进同一个列表也
    不会把两种写法摘要成同一身份。
    """
    parts: list[Any] = [source.model, dict(source.extra_body)]
    if source.enable_thinking is not None:
        parts.append(source.enable_thinking)
    if source.reasoning_effort is not None:
        parts.append(source.reasoning_effort)
    return json.dumps(parts, sort_keys=True, ensure_ascii=False)


def build_model_fingerprint(sources: Iterable[SourceConfig]) -> str:
    """缓存 key 的模型身份: 多源 scope = 排序去重的 model 合集。

    配置级采样参数(`extra_body`)必须参与,否则把 temperature 从 0 改成 1
    后重启仍会读到旧缓存(issue #4 设计决策 C)。`enable_thinking`(issue #5)与
    源级 `reasoning_effort`(issue #20)同理: 它们一旦真正改变请求体,"关掉推理后
    重启"就会读到开着推理时缓存的旧响应。全源三者皆未表态时字面量与历史实现逐字
    相同,不触发存量缓存冷启动。

    注意本指纹是**装配期**算出的**集合级**身份,覆盖不到逐次调用变化的请求级档位
    ——后者由 `build_cache_key` 的 `reasoning_effort` 参数单独承担(ARCH §7.5)。
    """
    fingerprint = ",".join(sorted({s.model for s in sources}))
    # 按 (model, extra_body[, enable_thinking][, reasoning_effort]) 而非源名摘要:
    # 语义是"本 scope 会用哪些(模型, 请求形态)组合",改源名不该误触全量冷启动。
    # 过滤条件必须与 `_fingerprint_mark` 追加的字段逐项对齐: 漏掉一项,只配了该项
    # 的源根本进不了 marks,`_fingerprint_mark` 改了也白改
    marks = sorted(
        {
            _fingerprint_mark(s)
            for s in sources
            if s.extra_body or s.enable_thinking is not None or s.reasoning_effort is not None
        }
    )
    if marks:
        digest = hashlib.sha256("".join(marks).encode("utf-8")).hexdigest()
        fingerprint = f"{fingerprint}|{digest}"
    return fingerprint


async def _aclose_component(component: object | None) -> None:
    """关闭一个**自建**组件: 优先 `aclose`,退到同步 `close`,两者皆无则跳过。

    退到 `close` 是给 SQLiteRecorder 的(它只有同步收尾);内存后端两者皆无,
    探测后静默跳过。三个 client 曾各持一份逐字复制的探测代码,收敛为一处是
    所有权纪律能被维持的前提——复制即是下一个 bug 的种子(设计 §3.4)。
    """
    if component is None:
        return
    aclose = getattr(component, "aclose", None)
    if aclose is not None:
        await aclose()
        return
    close = getattr(component, "close", None)
    if close is not None:
        close()


def _telemetry_status_of(telemetry: TelemetryRecorder | None) -> TelemetryStatus | None:
    """三个 client 共用的状态取值点: 不提供状态的 recorder 一律返回 None。

    判定写成 `isinstance(可选端口)` 而不是裸 `getattr`: 两者运行时都是结构检查
    (`@runtime_checkable` 按属性存在性判定),差别在**契约有没有名字**——端口是
    写进 `ports.py` 的公开承诺,下游可以照着实现;散落的 `getattr` 不是,而
    `aclose` 当年正是被复制成三份鸭子类型探测才漂移出越权关闭(设计 §3.3/§3.4)。
    """
    if isinstance(telemetry, TelemetryStatusProvider):
        return telemetry.telemetry_status
    return None


def _mark_owned_components(
    client: Any,
    *,
    limiter: RateLimiter | None,
    breaker: ProviderGate | None,
    telemetry: TelemetryRecorder | None,
) -> None:
    """工厂置位所有权(三个 client 共用): 传进来的是 None,就说明这一件是工厂自建的。

    与 `RedisLimiter.from_url` 逐字同款——私有属性由工厂标记,公共 API 面不变。
    transport 单列: 三处工厂都没有 transport 注入入口,它恒是自建的。
    """
    client._owns_transport = True
    client._owns_limiter = limiter is None
    client._owns_breaker = breaker is None
    client._owns_telemetry = telemetry is None


class GatewayClient:
    """统一治理入口;构造函数全量注入(测试/高级),工厂覆盖 90% 场景。"""

    def __init__(
        self,
        *,
        scope: str,
        sources: list[SourceConfig],
        selector: SourceSelector,
        limiter: RateLimiter,
        breaker: ProviderGate,
        transport: Transport,
        retry: RetryPolicy,
        backpressure: BackpressurePolicy,
        quota_full: str = "wait",
        circuit_open: str = "fail_fast",
        telemetry: TelemetryRecorder | None = None,
        pricing: PricingTable | None = None,
        text_cap: int | None = None,
        cache: CacheBackend | None = None,
        cache_namespace: str | None = None,
        cache_ttl_s: int | None = None,
        structured_strategy: StructuredOutputStrategy | None = None,
        structured_escalation: StructuredOutputStrategy | None = None,
        structured_max_retries: int = 1,
        call_deadline_s: float | None = None,
        hedge_after_s: float | None = None,
        hedge_max_extra: int = 1,
        hedge_stall_after_s: float | None = None,
        stall_threshold_ms: int | None = None,
        now: Any = time.monotonic,
        sleep: Any = asyncio.sleep,
        rng: Any = random.random,
    ) -> None:
        # 入口即校: 装配错误当场报,不等到第一次调用才炸
        self._call_deadline_s = ensure_call_deadline(
            call_deadline_s, "GatewayClient(call_deadline_s=...)"
        )
        # 对冲守卫与 GatewaySettings 共用同一份(issue #24 H4 / 1.3.8 #28): 直接构造这条路
        # 不经过 settings,值域/交叉守卫若只挂在 settings 上就会被它绕过。
        # 返回契约(1.3.8 审查 B2): tuple(归一化 after_s, 归一化 stall_after_s)
        self._hedge_after_s, self._hedge_stall_after_s = check_hedge_assembly(
            hedge_after_s=hedge_after_s,
            hedge_max_extra=hedge_max_extra,
            sources=sources,
            call_deadline_s=self._call_deadline_s,
            origin="GatewayClient(hedge_after_s=...)",
            stall_after_s=hedge_stall_after_s,
            stall_origin="GatewayClient(hedge_stall_after_s=...)",
        )
        # 沉默计量阈值分发(1.3.8 设计 §6): None = 保留各 SourceConfig 自带值
        # (不 stomp 直构造调用方的 per-source 设置);非 None 经 replace 写进每个源
        # (replace 触发 SourceConfig.__post_init__ 重跑,值域校验复用同一处)
        if stall_threshold_ms is not None:
            sources = [
                dataclasses.replace(s, stall_threshold_ms=stall_threshold_ms) for s in sources
            ]
        emitter = (
            TelemetryEmitter(telemetry, scope=scope, pricing=pricing, text_cap=text_cap)
            if telemetry is not None
            else None
        )
        terminal = RetryMW(
            scope=scope,
            sources=sources,
            selector=selector,
            limiter=limiter,
            gate=breaker,
            transport=transport,
            retry=retry,
            backpressure=backpressure,
            quota_full=quota_full,
            circuit_open=circuit_open,
            cooldown_memo=SourceCooldownMemo(now=now),
            # AIMD ceiling 尊重源级静态并发上限(独立核验 I1: 不得静默钳制大于 64 的配置)
            pacer=AdaptivePacer(
                ceiling=float(max([64, *(s.max_concurrency for s in sources if s.max_concurrency)]))
            ),
            emitter=emitter,
            # max_extra 不下传(issue #24 H5): v1 编排固定单路对冲,无消费者
            hedge_after_s=self._hedge_after_s,
            # 中段沉默对冲阈值(1.3.8 #28): 独立于 hedge_after_s 生效,守卫归一化值
            stall_after_s=self._hedge_stall_after_s,
            now=now,
            sleep=sleep,
            rng=rng,
        )
        middlewares: list[Middleware] = []
        if emitter is not None:
            middlewares.append(TelemetryMW(emitter, now=now))
        if cache is not None:
            if cache_namespace is None or cache_ttl_s is None:
                raise ValueError("启用缓存必须提供 cache_namespace 与 cache_ttl_s")
            # 多源 scope 的 key 身份;源集合或其 extra_body 变化 → 一次性冷启动
            middlewares.append(
                CacheMW(
                    backend=cache,
                    model_fingerprint=build_model_fingerprint(sources),
                    default_namespace=cache_namespace,
                    ttl_s=cache_ttl_s,
                    strategy=structured_strategy,
                )
            )
        if structured_strategy is not None:
            middlewares.append(
                StructuredMW(
                    strategy=structured_strategy,
                    max_retries=structured_max_retries,
                    escalation=structured_escalation,
                )
            )
        self._structured_available = structured_strategy is not None
        self._terminal = terminal  # 内部引用: 装配自省/测试用
        self._handler = compose(middlewares, terminal)
        # 期限到期需要报出 scope(现之前只传给 RetryMW,未自存)
        self._scope = scope
        # 逻辑调用统计需要同一只注入钟(1.3.5);现之前只传给中间件未自存
        self._now = now
        # 终态行由公开边界统一写出(T3),故边界也需持有 emitter
        self._emitter = emitter
        self._transport = transport
        self._telemetry = telemetry
        self._cache = cache
        # limiter/breaker 交给 RetryMW 之后仍须自持引用,否则 aclose 触达不到
        # 自建的 redis 客户端(设计 §3.4 记录的现存泄漏)
        self._limiter_backend = limiter
        self._breaker_backend = breaker
        # 所有权默认"不拥有": `__init__` 是全量注入路径,经它传入的一切都是
        # 外部资源,关掉别人的连接会弄死共享同一后端的其他 client(ARCH §7.7 R5)。
        # 只有工厂在真正自建时才置 True
        self._owns_transport = False
        self._owns_telemetry = False
        self._owns_cache = False
        self._owns_limiter = False
        self._owns_breaker = False
        self._closed = False

    @property
    def telemetry_status(self) -> TelemetryStatus | None:
        """遥测后端的可写状态;无遥测或注入的 recorder 不提供状态时为 None。

        判定收敛在 `_telemetry_status_of` 一处(不是三处各自探测): 三个 client
        的 `aclose` 曾各持一份逐字复制,漂移的结果就是越权关闭(设计 §3.3/§3.4)。
        """
        return _telemetry_status_of(self._telemetry)

    async def chat(
        self,
        messages: list[dict[str, Any]],
        *,
        session_id: str | None = None,
        parent_call_id: str | None = None,
        cache_salt: str | None = None,
        cache_namespace: str | None = None,
        structured: type[BaseModel] | Literal["json"] | None = None,
        stream: bool = True,
        overlay: Mapping[str, Any] | None = None,
        reasoning_effort: Effort | str | None = None,
        tenant_id: str | None = None,
        meta: Mapping[str, Any] | None = None,
        call_deadline_s: float | None = None,
    ) -> LLMResponse:
        """一次治理调用(签名冻结,ARCH §5.2;与三项目 LLMProvider 协议兼容)。

        `overlay` 是采样参数覆盖层(`temperature`/`seed`/`max_tokens` 等),优先级
        高于源级 `extra_body`、低于结构化输出的注入。带默认值的 keyword-only
        参数不影响既有调用点(issue #4)。

        `reasoning_effort` 是本次调用的推理档位,优先级高于源级 `REASONING_EFFORT`
        与 `ENABLE_THINKING`(设计 §4.2)。`None` 是**不表态**(随源级配置),与
        `Effort.NONE`("要求不推理")严格区分。裸字符串(`"low"`)也收,在此归一成
        `Effort`,非法值当场 `ValueError`——与 `SourceConfig` 那条装配路同口径。

        `tenant_id` 与 `meta` 是调用方自定义维度,只进遥测、**不进缓存 key**
        (租户隔离由 `cache_namespace` 负责,ARCH §7.5);前者享有真实列待遇
        (可挂 RLS、可进复合索引),后者是任意 KV 容器(issue #11)。

        `call_deadline_s` 是本次调用的墙钟硬边界(issue #22): `None` = 继承装配值,
        正数 = 本次覆盖,**不提供"本次关闭"**。它治理的是**等待**: 到期抛
        `CallDeadlineExceeded`,但到期**不等于未产出、未计费**——在途请求可能已发出、
        已被上游计费,且清理仍在 `finally` 里完成,故返回时刻 = 期限 + 清理耗时。
        """
        if structured is not None and not self._structured_available:
            raise ImportError(
                "结构化输出未启用: 安装 pip install 'polygateway[structured]' 后重新装配"
            )
        # 进洋葱之前校验并拷贝: 保护键/不可序列化值在此收口(否则会在 CacheMW
        # 的降级 try 之外抛裸 TypeError);拷贝防调用方复用同一 dict 逐次改 seed
        # 造成的竞态。同一份快照填 overlay 与 sampling——前者会被结构化注入,
        # 后者跨层恒定,供缓存 key 与遥测读取(设计决策 A/B/E)
        sampling = validate_request_overlay(overlay or {}, origin="chat(overlay=...)")
        # 同理必须在洋葱之外: 洋葱内的一切失败都被遥测层降级成 warning(库铁律
        # 「遥测写失败降级不冒泡」),校验放里面等于没有校验——非法维度会变成
        # 静默丢失的遥测行,而调用照常发出(issue #11 §4.2)
        dimension_tenant_id, dimensions = validate_caller_dimensions(
            tenant_id, meta, origin="chat(tenant_id=..., meta=...)"
        )
        # 同样必须在洋葱之外归一: 档位一路要被 `is Effort.NONE` 身份比较,裸字符串
        # 进去会在 transport 的错误路径上抛 `AttributeError`——那不属错误四分类,
        # 会穿透 `except ThinkingUnsupportedError` 与 RetryMW 的分类捕获(库铁律
        # 「错误分类驱动」)。归一失败是调用方编程错误,抛裸 ValueError 不进洋葱
        effort = (
            None
            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")
        # 期限取值与校验必须在创建 awaitable **之前**: 否则非法值抛错时会遗留
        # 未 await 的协程(RuntimeWarning + 资源不释放)
        deadline = (
            self._call_deadline_s
            if call_deadline_s is None
            else ensure_call_deadline(call_deadline_s, "chat(call_deadline_s=...)")
        )
        # 三项校验均已通过 → 进入统计边界(设计 §3: 输入校验异常在边界之外,保持原行为)
        context = _CallContext(now=self._now)
        request = ChatRequest(
            messages=messages,
            session_id=session_id,
            parent_call_id=parent_call_id,
            cache_salt=cache_salt,
            cache_namespace=cache_namespace,
            structured=structured,
            stream=stream,
            overlay=sampling,
            sampling=sampling,
            reasoning_effort=effort,
            tenant_id=dimension_tenant_id,
            meta=dimensions,
            call_context=context,
        )
        try:
            response = await with_call_deadline(
                self._handler(request), deadline_s=deadline, scope=self._scope
            )
        except PolyGatewayError as exc:
            # 统计边界内的一切领域失败均尝试写一条终态行(1.3.5 设计 §6 I3),
            # 包括已有 attempt 错误行的 RequestRejected / ResultInvalid——两类行描述
            # 的不是同一件事(尝试 vs 逻辑终态),由 `event_kind` 区分
            await emit_terminal_once(
                self._emitter, request=request, context=context, error=exc, operation="chat"
            )
            raise
        except asyncio.CancelledError:
            # 尽力而为且**取消优先**: 不 shield、不开后台任务;写入那一次 await 上
            # 再被取消则 `CancelledError` 照常传播(与 TelemetryMW 历史行为同款)
            await emit_terminal_once(
                self._emitter,
                request=request,
                context=context,
                error="cancelled",
                operation="chat",
            )
            raise
        # 快照在返回前冻结: 故它含缓存命中路径与已完成的内联遥测耗时
        return dataclasses.replace(response, call_stats=context.snapshot())

    async def aclose(self) -> None:
        """幂等释放**自建**资源: transport、遥测、缓存、限流/熔断后端。

        注入的组件一律不碰——它们可能被别的 client 共享,关掉即越权。
        """
        if self._closed:
            return
        self._closed = True
        if self._owns_transport:
            await _aclose_component(self._transport)
        if self._owns_telemetry:
            await _aclose_component(self._telemetry)
        if self._owns_cache:
            await _aclose_component(self._cache)
        if self._owns_limiter:
            await _aclose_component(self._limiter_backend)
        if self._owns_breaker:
            await _aclose_component(self._breaker_backend)

    async def __aenter__(self) -> GatewayClient:
        return self

    async def __aexit__(self, *exc_info: object) -> None:
        await self.aclose()

    # —— 工厂 ——

    @classmethod
    def from_settings(
        cls,
        settings: GatewaySettings,
        *,
        limiter: RateLimiter | None = None,
        breaker: ProviderGate | None = None,
        cache: CacheBackend | None = None,
        telemetry: TelemetryRecorder | None = None,
        registry: Mapping[str, ProviderProfile] | None = None,
        capabilities: Mapping[str, ThinkingCapability] | None = None,
        rng: Any = random.random,
    ) -> GatewayClient:
        """按配置装配;显式传入的后端实例即共享(None 项按配置自建私有实例)。"""
        sources = list(settings.sources)
        profiles = [get_provider(s.provider, registry=registry) for s in sources]
        _guard_thinking(sources, profiles, capabilities)
        strategy, escalation = _build_structured(profiles)
        client = cls(
            scope=settings.scope,
            sources=sources,
            selector=_build_selector(settings.selector, rng=rng),
            limiter=limiter if limiter is not None else _build_limiter(settings, sources),
            breaker=breaker if breaker is not None else _build_breaker(settings),
            transport=OpenAICompatTransport(registry=registry, capabilities=capabilities),
            retry=settings.retry,
            backpressure=settings.backpressure,
            quota_full=settings.quota_full,
            circuit_open=settings.circuit_open,
            telemetry=telemetry if telemetry is not None else _build_telemetry(settings),
            pricing=PricingTable.from_file(settings.pricing_path)
            if settings.pricing_path is not None
            else None,
            text_cap=settings.telemetry_text_cap,
            cache=cache if cache is not None else _build_cache(settings),
            cache_namespace=settings.cache_namespace,
            cache_ttl_s=settings.cache_ttl_s,
            structured_strategy=strategy,
            structured_escalation=escalation,
            structured_max_retries=settings.structured_max_retries,
            call_deadline_s=settings.call_deadline_s,
            hedge_after_s=settings.hedge_after_s,
            hedge_max_extra=settings.hedge_max_extra,
            hedge_stall_after_s=settings.hedge_stall_after_s,
            # settings 值为权威,恒分发(1.3.8 §3.5): 含缺省 1000 在内
            stall_threshold_ms=settings.stall_threshold_ms,
        )
        _mark_owned_components(client, limiter=limiter, breaker=breaker, telemetry=telemetry)
        client._owns_cache = cache is None  # 缓存后端可以是 None(backend=none),helper 会跳过
        return client

    @classmethod
    def from_env(
        cls,
        scope: str = "LLM",
        *,
        limiter: RateLimiter | None = None,
        breaker: ProviderGate | None = None,
        cache: CacheBackend | None = None,
        telemetry: TelemetryRecorder | None = None,
        registry: Mapping[str, ProviderProfile] | None = None,
        capabilities: Mapping[str, ThinkingCapability] | None = None,
        env: Mapping[str, str] | None = None,
    ) -> GatewayClient:
        """从 .env/环境变量装配一个 scope 的 client(键名清单见 .env.example)。"""
        return cls.from_settings(
            GatewaySettings.from_env(scope, env=env),
            limiter=limiter,
            breaker=breaker,
            cache=cache,
            telemetry=telemetry,
            registry=registry,
            capabilities=capabilities,
        )


def _build_limiter(settings: GatewaySettings, sources: list[SourceConfig]) -> RateLimiter:
    if settings.limiter_backend == "redis":
        from polygateway.backends.redis.limiter import RedisLimiter

        assert settings.redis_url is not None  # 内部不变量: _validate_backends 已保证
        return RedisLimiter.from_url(
            settings.redis_url,
            scope=settings.scope,
            sources={s.name: s for s in sources},
            global_limits=settings.global_limits,
            lease_ttl_s=settings.lease_ttl_s,
        )
    return InMemoryLimiter(
        scope=settings.scope,
        sources={s.name: s for s in sources},
        global_limits=settings.global_limits,
        lease_ttl_s=settings.lease_ttl_s,
    )


def _build_breaker(settings: GatewaySettings) -> ProviderGate:
    if settings.breaker_backend == "redis":
        from polygateway.backends.redis.breaker import RedisGate

        assert settings.redis_url is not None  # 内部不变量: _validate_backends 已保证
        return RedisGate.from_url(settings.redis_url, config=settings.breaker, scope=settings.scope)
    return InMemoryGate(config=settings.breaker)


def _build_selector(name: str, *, rng: Any = random.random) -> SourceSelector:
    if name == "round_robin":
        return RoundRobinSelector()
    if name == "least_inflight":
        return LeastInflightSelector()
    return HealthAwareSelector(rng=rng)


def _build_cache(settings: GatewaySettings) -> CacheBackend | None:
    if settings.cache_backend == "none":
        return None
    if settings.cache_backend == "memory":
        return InMemoryCache()
    from polygateway.backends.redis_cache import RedisCache

    assert settings.redis_url is not None  # 内部不变量: _validate_backends 已保证
    return RedisCache.from_url(settings.redis_url)


def _build_telemetry(settings: GatewaySettings) -> TelemetryRecorder | None:
    if settings.telemetry_backend == "none":
        return None
    if settings.telemetry_backend == "postgres":
        from polygateway.telemetry.postgres import PostgresRecorder

        assert settings.telemetry_pg_dsn is not None  # 内部不变量: _validate_telemetry 已保证
        return PostgresRecorder(
            settings.telemetry_pg_dsn,
            auto_migrate=settings.telemetry_auto_migrate,
            pool_max=settings.telemetry_pg_pool_max,
            write_timeout_s=settings.telemetry_pg_write_timeout_s,
        )
    from polygateway.telemetry.sqlite import SQLiteRecorder

    assert settings.telemetry_sqlite_path is not None  # 内部不变量: _validate_telemetry 已保证
    return SQLiteRecorder(
        settings.telemetry_sqlite_path, auto_migrate=settings.telemetry_auto_migrate
    )


def _build_structured(
    profiles: list[ProviderProfile],
) -> tuple[StructuredOutputStrategy | None, StructuredOutputStrategy | None]:
    """按注册表能力选策略(设计 §5): 全员支持原生 schema 才用 NativeSchema。

    json_repair 缺失(未装 structured extra)→ 返回 (None, None),
    chat(structured=...) 时显式报缺 extra,绝不静默跳过校验。
    """
    try:
        from polygateway.structured.json_repair import JsonRepairStrategy
        from polygateway.structured.native_schema import NativeSchemaStrategy
    except ImportError:
        return None, None
    try:
        native_all = all(p.supports_native_schema for p in profiles)
        if native_all:
            return NativeSchemaStrategy(), NativeSchemaStrategy()
        return JsonRepairStrategy(), None
    except ImportError:
        return None, None


async def gather_bounded[T](aws: Iterable[Awaitable[T]], *, concurrency: int) -> list[T]:
    """有界并发 gather(D5 便利函数,替代 VT 手搓 semaphore+gather 样板)。

    语义与 `asyncio.gather` 默认一致: 结果保序、首个异常上抛;仅增加并发上限。
    期限计时从每次调用真正开始执行起算,信号量排队时长不在 `call_deadline_s` 之内。
    """
    if concurrency < 1:
        raise ValueError("concurrency 必须 ≥ 1")
    sem = asyncio.Semaphore(concurrency)

    async def _run(aw: Awaitable[T]) -> T:
        async with sem:
            return await aw

    return await asyncio.gather(*(_run(aw) for aw in aws))