PolyGateway/tests/unit/test_client.py

"""GatewayClient 装配与端到端(fake 后端 + MockTransport)测试(设计 §2.4)。"""

import asyncio
import json
import sys
from pathlib import Path

import httpx
import pytest

from polygateway import (
    AllSourcesExhausted,
    GatewayClient,
    GatewaySettings,
    RequestRejectedError,
    ResultInvalidError,
    gather_bounded,
)
from polygateway.backends.memory.breaker import InMemoryGate
from polygateway.backends.memory.cache import InMemoryCache
from polygateway.backends.memory.limiter import InMemoryLimiter
from polygateway.sources import RoundRobinSelector
from polygateway.structured.json_repair import JsonRepairStrategy
from polygateway.structured.native_schema import NativeSchemaStrategy
from polygateway.transports.openai_compat import OpenAICompatTransport
from polygateway.types import (
    BackpressurePolicy,
    BreakerConfig,
    ChatRequest,
    Effort,
    GlobalLimits,
    RetryPolicy,
    SourceConfig,
)

_REPO = Path(__file__).resolve().parents[2]

_ENV = {
    "LLM__QWEN__1__BASE_URL": "https://gw.example/v1",
    "LLM__QWEN__1__API_KEY": "sk-a",
    "LLM__QWEN__1__MODEL": "qwen-max",
    "LLM__QWEN__1__TIMEOUT_S": "120",
    "LLM_MAX_RETRIES": "3",
    "LLM_RETRY_BASE_DELAY": "2.0",
    "LLM_RETRY_MAX_DELAY": "30.0",
    "LLM_CIRCUIT_BREAKER_THRESHOLD": "5",
    "LLM_CIRCUIT_BREAKER_COOLDOWN": "60",
    "PGW_CACHE_BACKEND": "none",
    "PGW_TELEMETRY_BACKEND": "none",
}

_OCR_ENV = {
    "OCR__MONKEY__1__BASE_URL": "http://10.77.0.20:7866",
    "OCR__MONKEY__1__API_KEY": "none",
    "OCR__MONKEY__1__MODEL": "monkey-ocr",
    "OCR__MONKEY__1__TIMEOUT_S": "120",
    "LLM_MAX_RETRIES": "3",
    "LLM_RETRY_BASE_DELAY": "2.0",
    "LLM_RETRY_MAX_DELAY": "30.0",
    "LLM_CIRCUIT_BREAKER_THRESHOLD": "5",
    "LLM_CIRCUIT_BREAKER_COOLDOWN": "60",
    "PGW_CACHE_BACKEND": "none",
    "PGW_TELEMETRY_BACKEND": "none",
}


def _sse(content='{"answer": 1}'):
    chunk = json.dumps({"choices": [{"delta": {"content": content}}]})
    usage = json.dumps({"choices": [], "usage": {"prompt_tokens": 3, "completion_tokens": 4}})
    body = f"data: {chunk}\n\ndata: {usage}\n\ndata: [DONE]\n\n"
    return httpx.Response(200, content=body.encode(), headers={"content-type": "text/event-stream"})


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


def _client(sources=None, handler=None, *, limiter=None, quota_full="wait", **overrides):
    sources = sources or [_source()]
    handler = handler or (lambda request: _sse())
    transport = OpenAICompatTransport(
        client_factory=lambda src: httpx.AsyncClient(transport=httpx.MockTransport(handler))
    )
    defaults = {
        "scope": "llm",
        "sources": sources,
        "selector": RoundRobinSelector(),
        "limiter": limiter
        or InMemoryLimiter(
            scope="llm",
            sources={s.name: s for s in sources},
            global_limits=GlobalLimits(0, 0, 0),
        ),
        "breaker": InMemoryGate(config=BreakerConfig(5, 60.0, 120.0)),
        "transport": transport,
        "retry": RetryPolicy(3, 2.0, 30.0),
        "backpressure": BackpressurePolicy(300.0, 0.01),
        "quota_full": quota_full,
        "structured_strategy": JsonRepairStrategy(),
    }
    defaults.update(overrides)
    return GatewayClient(**defaults)


class TestChatEndToEnd:
    async def test_plain_chat(self):
        async with _client() as client:
            resp = await client.chat([{"role": "user", "content": "hi"}])
        assert resp.content == '{"answer": 1}'
        assert resp.source_name == "qwen_1" and resp.prompt_tokens == 3

    async def test_structured_json_tier(self):
        async with _client() as client:
            resp = await client.chat([{"role": "user", "content": "hi"}], structured="json")
        assert resp.structured_data == {"answer": 1}

    async def test_cache_hit_roundtrip(self):
        client = _client(cache=InMemoryCache(), cache_namespace="proj", cache_ttl_s=3600)
        async with client:
            first = await client.chat([{"role": "user", "content": "hi"}])
            second = await client.chat([{"role": "user", "content": "hi"}])
        assert first.cache_hit is False and second.cache_hit is True

    async def test_structured_unavailable_fails_loudly(self):
        async with _client(structured_strategy=None) as client:
            with pytest.raises(ImportError, match="structured"):
                await client.chat([{"role": "user", "content": "hi"}], structured="json")


class TestSamplingOverlay:
    """调用级采样参数入口(issue #4 Task 3)。"""

    def _capturing_client(self, captured, **overrides):
        def handler(request):
            captured.append(json.loads(request.content))
            return _sse()

        return _client(handler=handler, **overrides)

    async def test_overlay_reaches_request_body(self):
        captured = []
        async with self._capturing_client(captured) as client:
            await client.chat([{"role": "user", "content": "hi"}], overlay={"seed": 42})
        assert captured[0]["seed"] == 42

    async def test_call_level_beats_config_level(self):
        """优先级: 调用级 > 配置级(设计决策 A)。"""
        captured = []
        source = _source(extra_body={"temperature": 0, "top_p": 0.9})
        async with self._capturing_client(captured, sources=[source]) as client:
            await client.chat([{"role": "user", "content": "hi"}], overlay={"temperature": 1})
        assert captured[0]["temperature"] == 1  # 调用级覆盖
        assert captured[0]["top_p"] == 0.9  # 配置级未被顶掉的键保留

    async def test_structured_injection_beats_call_level(self):
        """结构化注入优先级最高: 它关系到响应能否被解析(设计决策 A)。"""
        captured = []
        client = self._capturing_client(captured, structured_strategy=NativeSchemaStrategy())
        async with client:
            await client.chat(
                [{"role": "user", "content": "hi"}],
                structured="json",
                overlay={"response_format": {"type": "text"}},
            )
        assert captured[0]["response_format"] != {"type": "text"}

    async def test_protected_key_rejected_before_onion(self):
        """保护键在进洋葱之前就报错,transport 一次都不该被碰到。"""
        captured = []
        async with self._capturing_client(captured) as client:
            with pytest.raises(ValueError, match="stream"):
                await client.chat([{"role": "user", "content": "hi"}], overlay={"stream": False})
        assert captured == []

    async def test_unserializable_value_rejected_before_onion(self):
        """裸 TypeError 会在 CacheMW 的降级 try 之外炸且无遥测(设计决策 B)。"""
        captured = []
        async with self._capturing_client(captured) as client:
            with pytest.raises(ValueError, match="JSON"):
                await client.chat(
                    [{"role": "user", "content": "hi"}], overlay={"temperature": object()}
                )
        assert captured == []

    async def test_caller_dict_mutation_does_not_leak(self):
        """调用方逐次改 seed 复用同一 dict 是预期模式(设计决策 E)。"""
        captured = []
        caller_overlay = {"seed": 1}
        async with self._capturing_client(captured) as client:
            await client.chat([{"role": "user", "content": "hi"}], overlay=caller_overlay)
            caller_overlay["seed"] = 2
            await client.chat([{"role": "user", "content": "hi"}], overlay=caller_overlay)
        assert [c["seed"] for c in captured] == [1, 2]


class TestReasoningEffortPriority:
    """三层优先级: 请求级 > 源级 > `enable_thinking` 语法糖 > 不表态(设计 §4.2)。

    一律抓**真实请求体**而非只查 `ChatRequest` 字段: 档位的价值全在发出去的那几个
    字节上,只断言中间态会让"字段填了但一路没人读"这种缺口继续通过测试——issue #20
    的 `extra_body` 绕行正是这么长出来的。
    """

    def _capturing_client(self, captured, **overrides):
        def handler(request):
            captured.append(json.loads(request.content))
            return _sse()

        return _client(handler=handler, **overrides)

    def _zhipu(self, **overrides):
        # glm-5.3 的档位是 low/high/max(能力表已登记),zhipu 的 wire 三样俱全,
        # 是唯一能同时看清"开启形态"与"档位键"的组合
        overrides.setdefault("model", "glm-5.3")
        return _source(provider="zhipu", **overrides)

    async def test_request_effort_wins_over_source(self):
        captured = []
        source = self._zhipu(reasoning_effort=Effort.LOW)
        async with self._capturing_client(captured, sources=[source]) as client:
            await client.chat([{"role": "user", "content": "hi"}], reasoning_effort=Effort.MAX)
        assert captured[0]["reasoning_effort"] == "max"
        assert captured[0]["thinking"] == {"type": "enabled"}

    async def test_none_request_does_not_clear_source(self):
        """请求级不表态 ≠ 请求级要求"不推理": 前者必须让源级默认继续生效。"""
        captured = []
        source = self._zhipu(reasoning_effort=Effort.LOW)
        async with self._capturing_client(captured, sources=[source]) as client:
            await client.chat([{"role": "user", "content": "hi"}])
        assert captured[0]["reasoning_effort"] == "low"

    async def test_request_none_tier_is_an_opinion_not_an_absence(self):
        """请求级 `none` 是"要求不推理",不得被当成"没表态"而回落到源级档位。"""
        captured = []
        source = self._zhipu(model="glm-5.2", reasoning_effort=Effort.MAX)
        async with self._capturing_client(captured, sources=[source]) as client:
            await client.chat([{"role": "user", "content": "hi"}], reasoning_effort=Effort.NONE)
        assert captured[0]["thinking"] == {"type": "disabled"}
        assert "reasoning_effort" not in captured[0]

    @pytest.mark.parametrize(
        "provider,model,tier",
        [
            ("deepseek", "deepseek-v4-pro", "high"),
            ("zhipu", "glm-5.3", "low"),
            ("moonshot", "kimi-k3", "low"),
            ("minimax", "MiniMax-M3", "medium"),
        ],
    )
    async def test_legacy_auto_requires_explicit_migration(self, provider, model, tier):
        """旧糖配置明确拒绝，显式选择才能恢复可执行请求。"""
        from dataclasses import replace

        captured = []
        source = _source(provider=provider, model=model, enable_thinking=True)
        client = self._capturing_client(captured, sources=[source])
        try:
            with pytest.raises(RequestRejectedError):
                await client.chat([])
            assert captured == []
        finally:
            await client._transport.aclose()
        client = self._capturing_client(
            captured, sources=[replace(source, enable_thinking=None, reasoning_effort=tier)]
        )
        try:
            await client.chat([])
            assert captured[0]["reasoning_effort"] == tier
        finally:
            await client._transport.aclose()


class TestEffortFallbackWiring:
    """源级 `effort_fallback` 必须真的走到 `resolve_thinking`(issue #20)。

    只测配置值域与纯函数是不够的: 变异实测显示把 `fallback=source.effort_fallback`
    换成硬编码 `"error"`,全套单测依然全绿——`nearest` 是人类明确要求实现的功能,
    没有端到端用例它会在零告警下变成死代码(2026-09-05 独立验证查出)。

    **该形态有两道门,必须各测各的**: transport 那道决定发出去的字节,`_guard_thinking`
    那道决定装配能不能过。只钉住 transport,装配守卫退化成硬编码 `error` 时,配了
    `nearest` 的源会在**运行期本来跑得起来**的情况下于装配期当场被判死,而这半边
    第一轮修复时正是漏掉的那半边(M22)。
    """

    def _capturing_client(self, captured, **overrides):
        def handler(request):
            captured.append(json.loads(request.content))
            return _sse()

        return _client(handler=handler, **overrides)

    async def test_nearest_sends_the_mapped_tier(self):
        """glm-5.3 只有 low/high/max: 请求 `medium` 等距,按"取弱"落到 low。"""
        captured = []
        source = _source(provider="zhipu", model="glm-5.3", effort_fallback="nearest")
        async with self._capturing_client(captured, sources=[source]) as client:
            await client.chat([{"role": "user", "content": "hi"}], reasoning_effort=Effort.MEDIUM)
        assert captured[0]["reasoning_effort"] == "low"

    async def test_error_fallback_refuses_the_same_request(self):
        """默认 `error` 必须报错而不是映射: 一次静默的换档就是一笔没打招呼的账单。"""
        captured = []
        source = _source(provider="zhipu", model="glm-5.3")
        async with self._capturing_client(captured, sources=[source]) as client:
            with pytest.raises(RequestRejectedError, match="medium"):
                await client.chat(
                    [{"role": "user", "content": "hi"}], reasoning_effort=Effort.MEDIUM
                )
        assert captured == []  # 请求根本没发出去

    def test_nearest_survives_the_assembly_guard(self):
        """装配守卫也必须用源级 `fallback`: 运行期能映射的源不该在装配期被判死。

        源级 `medium` + glm-5.3(只有 low/high/max)是 `nearest` 唯一能被观测到的
        装配期形态——守卫硬编码 `error` 时,这次 `from_env` 会抛
        `ThinkingUnsupportedError` 而不是返回 client。
        """
        env = dict(_ENV)
        for key in list(env):
            if key.startswith("LLM__QWEN__"):
                del env[key]
        env.update(
            {
                "LLM__ZHIPU__1__BASE_URL": "https://gw.example/v1",
                "LLM__ZHIPU__1__API_KEY": "sk-a",
                "LLM__ZHIPU__1__MODEL": "glm-5.3",
                "LLM__ZHIPU__1__TIMEOUT_S": "120",
                "LLM__ZHIPU__1__REASONING_EFFORT": "medium",
                "LLM__ZHIPU__1__EFFORT_FALLBACK": "nearest",
            }
        )
        client = GatewayClient.from_env("LLM", env=env)
        assert isinstance(client, GatewayClient)


class TestAppliedTierReachesTheCaller:
    """`LLMResponse.applied_effort` 报的是**真正发出去的**那一档(计划 T8-5)。

    对下游是新能力(它终于能知道这次跑在哪档),对遥测是前置条件: 记请求档会让
    按档分组的压测把整行挂在一个从未发出过的档下,而那种数据错得看不出来。
    """

    async def test_response_carries_the_mapped_tier(self):
        """glm-5.3 无 `medium`: 开了 nearest 后实际跑的是 low,响应必须这么说。"""
        source = _source(provider="zhipu", model="glm-5.3", effort_fallback="nearest")
        async with _client(sources=[source]) as client:
            resp = await client.chat(
                [{"role": "user", "content": "hi"}], reasoning_effort=Effort.MEDIUM
            )
        assert resp.applied_effort is Effort.LOW

    async def test_no_statement_leaves_the_field_none(self):
        async with _client() as client:
            resp = await client.chat([{"role": "user", "content": "hi"}])
        assert resp.applied_effort is None


class TestUnsupportedTierIsRefusedNotRetried:
    """档位不可满足 = 请求本身的问题: 报 `RequestRejectedError`,不重试、不伤熔断。

    重试与换源都不会让它变对(设计 §10),而把它计进熔断更糟——一次配置错误会
    把一个健康的源关掉,拖垮与推理无关的所有调用。
    """

    async def test_tier_error_never_reaches_the_gateway_or_the_breaker(self):
        sent = []

        def handler(request):
            sent.append(request)
            return _sse()

        # 阈值取 1: 只要这次失败被计进熔断,门当场开路,断言立刻可见
        gate = InMemoryGate(config=BreakerConfig(1, 60.0, 120.0))
        source = _source(name="zp", provider="zhipu", model="glm-5.3")
        async with _client(sources=[source], handler=handler, breaker=gate) as client:
            with pytest.raises(RequestRejectedError, match="无法关闭推理"):
                await client.chat([{"role": "user", "content": "hi"}], reasoning_effort=Effort.NONE)
        assert sent == [], "请求根本不该发出去: 档位不可满足在组装期就已判定"
        assert (await gate.try_enter("zp", "w")).allowed, "配置错误不得计入熔断失败"


class TestRequestTierNormalization:
    """`chat(reasoning_effort=...)` 是公共入口,裸字符串必须在此归一(issue #20)。

    两条装配路(工厂 / 构造函数全量注入,CLAUDE.md §4.5)与 `.env` 路的口径必须
    一致——后者早已是"解析即归一"。不归一的后果不是"少个类型注解"那么轻: 档位
    一路要被 `is Effort.NONE` 身份比较,裸字符串会在 transport 的错误路径上抛
    `AttributeError`,而它不属错误四分类,会穿透 `except ThinkingUnsupportedError`
    与 RetryMW 的分类捕获,以未分类异常冒出 `chat()`(2026-09-05 独立验证实测)。
    """

    def _capturing_client(self, captured, **overrides):
        def handler(request):
            captured.append(json.loads(request.content))
            return _sse()

        return _client(handler=handler, **overrides)

    async def test_bare_string_tier_reaches_the_wire(self):
        captured = []
        source = _source(provider="zhipu", model="glm-5.3")
        async with self._capturing_client(captured, sources=[source]) as client:
            await client.chat([{"role": "user", "content": "hi"}], reasoning_effort="max")
        assert captured[0]["reasoning_effort"] == "max"

    async def test_illegal_tier_is_a_value_error_listing_the_vocabulary(self):
        """非法档位是调用方编程错误: 当场 `ValueError`,不进洋葱、不成为未分类异常。"""
        source = _source(provider="zhipu", model="glm-5.3")
        async with _client(sources=[source]) as client:
            with pytest.raises(ValueError) as exc:
                await client.chat([{"role": "user", "content": "hi"}], reasoning_effort="lowest")
        message = str(exc.value)
        assert "chat(reasoning_effort=...)" in message
        assert all(tier.value in message for tier in Effort)


class _MemoryRecorder:
    """收下遥测行原样存起来;断言"哪些行被写了"必须能看到零行的情形。"""

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

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


class TestCallerDimensions:
    """调用方自定义维度进遥测(issue #11 Task 4)。"""

    _MSG = [{"role": "user", "content": "hi"}]

    async def test_dimensions_reach_telemetry_row(self):
        recorder = _MemoryRecorder()
        async with _client(telemetry=recorder) as client:
            await client.chat(self._MSG, tenant_id="t1", meta={"batch": "b-42"})
        row = recorder.rows[-1]
        assert row["tenant_id"] == "t1"
        assert json.loads(row["meta"]) == {"batch": "b-42"}

    async def test_default_path_writes_sentinels(self):
        """不传两参数时落哨兵值而非 NULL(§4.4: NULL 在 RLS 下是永久不可见的黑洞)。"""
        recorder = _MemoryRecorder()
        async with _client(telemetry=recorder) as client:
            await client.chat(self._MSG)
        row = recorder.rows[-1]
        assert row["tenant_id"] == "" and row["meta"] == "{}"

    async def test_invalid_meta_key_rejected_before_any_telemetry(self):
        """校验早于遥测(§4.2 核心承诺): 放进洋葱就会被降级成 warning 而调用照常发出。"""
        recorder = _MemoryRecorder()
        async with _client(telemetry=recorder) as client:
            with pytest.raises(ValueError, match="meta"):
                await client.chat(self._MSG, meta={"BAD-KEY": 1})
        assert recorder.rows == []

    async def test_non_finite_float_rejected_before_any_telemetry(self):
        """nan 产出的是 PG 拒收的非法 JSON;放行等于把调用方 bug 变成静默丢遥测(§6)。"""
        recorder = _MemoryRecorder()
        async with _client(telemetry=recorder) as client:
            with pytest.raises(ValueError, match="nan"):
                await client.chat(self._MSG, meta={"k": float("nan")})
        assert recorder.rows == []

    async def test_meta_does_not_enter_cache_key(self):
        """仅 meta 不同必须仍命中缓存(F1): 进 key 会让存量缓存全量冷启动且不报错。

        带对照组: 只断言"命中"的话,缓存 key 退化成常量(忽略一切输入)时本用例
        照样绿——那是恒真断言。故再改一个**确实进 key** 的维度(namespace)断言
        miss,证明 key 仍在区分输入,"meta 不进 key"才是被测出来的结论。
        """
        cache = InMemoryCache()  # 两个 client 共用一份存储,否则对照组的 miss 是白来的
        client = _client(cache=cache, cache_namespace="proj", cache_ttl_s=3600)
        async with client:
            first = await client.chat(self._MSG, meta={"batch": "b-1"})
            second = await client.chat(self._MSG, meta={"batch": "b-2"})
        assert first.cache_hit is False and second.cache_hit is True

        other_ns = _client(cache=cache, cache_namespace="other", cache_ttl_s=3600)
        async with other_ns:
            assert (await other_ns.chat(self._MSG, meta={"batch": "b-1"})).cache_hit is False


class TestModelFingerprint:
    """配置级采样参数须进缓存身份,否则改 temperature 后仍读旧缓存(决策 C)。"""

    def test_empty_extra_body_keeps_legacy_fingerprint(self):
        """全源无 extra_body 时字面量与旧实现逐字相同,不触发存量缓存冷启动。"""
        from polygateway.client import build_model_fingerprint

        sources = [_source(), _source(name="qwen_2", model="qwen-plus")]
        assert build_model_fingerprint(sources) == "qwen-max,qwen-plus"

    def test_extra_body_changes_fingerprint(self):
        from polygateway.client import build_model_fingerprint

        plain = build_model_fingerprint([_source()])
        tuned = build_model_fingerprint([_source(extra_body={"temperature": 0})])
        assert plain != tuned
        assert tuned.startswith("qwen-max|")  # 旧字面量仍是前缀,便于人眼辨认

    def test_enable_thinking_changes_fingerprint(self):
        """issue #5 配套: thinking 一旦真正改变请求体,就必须进缓存身份。

        否则"关掉推理后重启"会读到开着推理时缓存的旧响应——issue #4 为
        temperature 写过逐字相同的理由。
        """
        from polygateway.client import build_model_fingerprint

        plain = build_model_fingerprint([_source()])
        off = build_model_fingerprint([_source(enable_thinking=False)])
        on = build_model_fingerprint([_source(enable_thinking=True)])
        assert len({plain, off, on}) == 3

    def test_extra_body_only_fingerprint_is_byte_identical_to_before(self):
        """只配 extra_body、不表态 thinking 的存量源不得触发冷启动。

        字面量在此硬编码: 这条断言的价值全在"逐字相同",改实现时必须先看见它红。
        """
        import hashlib
        import json

        from polygateway.client import build_model_fingerprint

        mark = json.dumps(["qwen-max", {"temperature": 0}], sort_keys=True, ensure_ascii=False)
        expected = "qwen-max|" + hashlib.sha256(mark.encode("utf-8")).hexdigest()
        assert build_model_fingerprint([_source(extra_body={"temperature": 0})]) == expected

    def test_source_rename_does_not_change_fingerprint(self):
        """指纹按 (model, extra_body) 而非源名: 改名不该误触全量冷启动。"""
        from polygateway.client import build_model_fingerprint

        a = build_model_fingerprint([_source(name="qwen_1", extra_body={"temperature": 0})])
        b = build_model_fingerprint([_source(name="renamed", extra_body={"temperature": 0})])
        assert a == b

    def test_differing_extra_body_across_sources_is_distinguished(self):
        from polygateway.client import build_model_fingerprint

        a = build_model_fingerprint([_source(extra_body={"temperature": 0})])
        b = build_model_fingerprint([_source(extra_body={"temperature": 1})])
        assert a != b

    def test_source_tier_enters_fingerprint(self):
        """源级 `reasoning_effort` 改变请求体,就必须改变缓存身份(与 issue #5 同理)。

        本用例同时守着一个易漏点: 只配 `REASONING_EFFORT`、既无 `extra_body` 也无
        `ENABLE_THINKING` 的源,必须能进入指纹的 marks 集合——否则 `_fingerprint_mark`
        改了也白改,四个指纹会全部相等。
        """
        from polygateway.client import build_model_fingerprint

        plain = build_model_fingerprint([_source()])
        low = build_model_fingerprint([_source(reasoning_effort=Effort.LOW)])
        max_ = build_model_fingerprint([_source(reasoning_effort=Effort.MAX)])
        off = build_model_fingerprint([_source(reasoning_effort=Effort.NONE)])
        assert len({plain, low, max_, off}) == 4

    def test_source_tier_is_distinguished_from_the_thinking_sugar(self):
        """`reasoning_effort=NONE` 与 `enable_thinking=False` 不得摘要成同一个指纹。

        两者语义等价但取值不同(`"none"` vs `false`),让它们撞车会把"两种写法"
        变成"一种缓存身份",日后任一侧语义微调都会静默复用另一侧的响应。
        """
        from polygateway.client import build_model_fingerprint

        by_tier = build_model_fingerprint([_source(reasoning_effort=Effort.NONE)])
        by_sugar = build_model_fingerprint([_source(enable_thinking=False)])
        assert by_tier != by_sugar

    def test_absent_tier_fingerprint_is_byte_identical_to_before(self):
        """不表态档位的存量源不得因本次升级平白冷启动: 字面量逐字相同。

        两条: 纯净源仍是裸 model 合集;只配 extra_body 的源仍是升级前那个摘要。
        """
        import hashlib
        import json

        from polygateway.client import build_model_fingerprint

        assert build_model_fingerprint([_source()]) == "qwen-max"
        mark = json.dumps(["qwen-max", {"temperature": 0}], sort_keys=True, ensure_ascii=False)
        expected = "qwen-max|" + hashlib.sha256(mark.encode("utf-8")).hexdigest()
        assert build_model_fingerprint([_source(extra_body={"temperature": 0})]) == expected

    def test_declared_tier_fingerprint_is_a_golden(self):
        """配了档位那一侧的指纹字面量也要钉死: 它变了就是该源整段缓存冷启动。

        存量(不表态)那侧由 `test_absent_tier_fingerprint_is_byte_identical_to_before`
        守着;本条守的是"档位怎么摘要进 mark"。字面量硬编码,不在测试里重算——
        重算等于把实现抄一遍,实现改了两边一起变,断言就白写了。
        """
        from polygateway.client import build_model_fingerprint

        assert build_model_fingerprint([_source(reasoning_effort=Effort.LOW)]) == (
            "qwen-max|76f2b3e419e1a727f7f31b6144da0b40d62a00ca93c91ce5901b4ebd48c0b8c0"
        )


class TestFactories:
    def test_from_env_assembles(self):
        client = GatewayClient.from_env("LLM", env=_ENV)
        assert isinstance(client, GatewayClient)

    def test_from_env_unknown_provider_fails_at_assembly(self):
        env = dict(_ENV)
        for key in list(env):
            if key.startswith("LLM__QWEN__"):
                env[key.replace("QWEN", "GLM")] = env.pop(key)
        with pytest.raises(ValueError, match="glm"):
            GatewayClient.from_env("LLM", env=env)

    def test_from_env_builds_redis_governance_backends(self):
        """M2: 配置取 redis 时装配出 Redis 后端(构造不连库,unit 可测)。"""
        from polygateway.backends.redis.breaker import RedisGate
        from polygateway.backends.redis.limiter import RedisLimiter
        from polygateway.client import _build_breaker, _build_limiter

        env = dict(
            _ENV,
            PGW_LIMITER_BACKEND="redis",
            PGW_BREAKER_BACKEND="redis",
            REDIS_URL="redis://:pw@10.0.0.1:6379/3",
        )
        settings = GatewaySettings.from_env("LLM", env=env)
        assert isinstance(_build_limiter(settings, list(settings.sources)), RedisLimiter)
        assert isinstance(_build_breaker(settings), RedisGate)

    def test_from_settings_respects_injection(self):
        settings = GatewaySettings.from_env("LLM", env=_ENV)
        shared = InMemoryLimiter(
            scope="shared",
            sources={s.name: s for s in settings.sources},
            global_limits=GlobalLimits(0, 0, 0),
        )
        client = GatewayClient.from_settings(settings, limiter=shared)
        assert isinstance(client, GatewayClient)


class TestTelemetryTextCapWiring:
    """`PGW_TELEMETRY_TEXT_CAP` 必须走通全部三条 `from_settings` 装配路(issue #12)。

    三条链路写的是**同一张** `llm_calls` 表:只接通 chat,embed 与 OCR 的行就
    永远不受 cap 约束,同表内一半受控一半不受控——那正是本 issue 要消灭的状态。
    """

    _CAP_ENV = dict(_ENV, PGW_TELEMETRY_TEXT_CAP="8")
    _OCR_CAP_ENV = dict(_OCR_ENV, PGW_TELEMETRY_TEXT_CAP="8")

    def test_gateway_from_settings_wires_the_cap(self):
        settings = GatewaySettings.from_env("LLM", env=self._CAP_ENV)
        client = GatewayClient.from_settings(settings, telemetry=_MemoryRecorder())
        assert client._terminal._emitter._text_cap == 8
        # 对照组: 不设该键时 emitter 拿到的必须是 None,否则 8 可能是硬编码来的
        unset = GatewayClient.from_settings(
            GatewaySettings.from_env("LLM", env=_ENV), telemetry=_MemoryRecorder()
        )
        assert unset._terminal._emitter._text_cap is None

    def test_embedding_from_settings_wires_the_cap(self):
        from polygateway.config import EmbeddingSettings
        from polygateway.embedding import EmbeddingClient

        gateway = GatewaySettings.from_env("LLM", env=self._CAP_ENV)
        client = EmbeddingClient.from_settings(
            EmbeddingSettings(gateway=gateway, batch_size=2), telemetry=_MemoryRecorder()
        )
        assert client._emitter._text_cap == 8
        unset = EmbeddingClient.from_settings(
            EmbeddingSettings(gateway=GatewaySettings.from_env("LLM", env=_ENV), batch_size=2),
            telemetry=_MemoryRecorder(),
        )
        assert unset._emitter._text_cap is None

    def test_ocr_from_settings_wires_the_cap(self):
        from polygateway.config import OcrSettings
        from polygateway.ocr import OcrClient

        settings = OcrSettings.from_env("OCR", env=dict(self._OCR_CAP_ENV))
        client = OcrClient.from_settings(settings, telemetry=_MemoryRecorder())
        assert client._emitter._text_cap == 8
        no_cap = dict(self._OCR_CAP_ENV)
        no_cap.pop("PGW_TELEMETRY_TEXT_CAP")
        unset = OcrClient.from_settings(
            OcrSettings.from_env("OCR", env=no_cap), telemetry=_MemoryRecorder()
        )
        assert unset._emitter._text_cap is None

    async def test_capped_body_reaches_the_recorder_end_to_end(self, monkeypatch):
        """装配路通了还不够: 真跑一次 chat,落库的 messages 与 response 确已截断。

        `from_settings` 自建 transport(没有 client_factory 入口),故在装配点
        换掉该类以接上 MockTransport——洋葱其余各层仍是 `from_settings` 装的真件。
        """
        recorder = _MemoryRecorder()
        long_text = "甲乙丙丁戊己庚辛壬癸"  # 10 字,cap=8 → 略 2 字
        monkeypatch.setattr(
            "polygateway.client.OpenAICompatTransport",
            lambda **kwargs: OpenAICompatTransport(
                client_factory=lambda source: httpx.AsyncClient(
                    transport=httpx.MockTransport(lambda request: _sse(content=long_text))
                )
            ),
        )
        settings = GatewaySettings.from_env("LLM", env=self._CAP_ENV)
        async with GatewayClient.from_settings(settings, telemetry=recorder) as client:
            await client.chat([{"role": "user", "content": long_text}])
        row = recorder.rows[-1]
        assert json.loads(row["messages"])[0]["content"] == "甲乙丙丁戊己庚辛…(略 2 字)"
        assert row["response"] == "甲乙丙丁戊己庚辛…(略 2 字)"

    def test_non_positive_cap_rejected_on_the_direct_construction_path(self):
        """直接构造是库承诺的另一条公共装配路;cap=0 会让每条正文只剩省略标记。"""
        with pytest.raises(ValueError, match="text_cap"):
            _client(telemetry=_MemoryRecorder(), text_cap=0)


class TestSharedBackend:
    async def test_two_clients_share_global_concurrency_gate(self):
        """VT R5: 两个逻辑角色显式注入同一 limiter → 共享全局并发闸。"""
        src_a, src_b = _source("role_a_1"), _source("role_b_1")
        shared = InMemoryLimiter(
            scope="shared",
            sources={"role_a_1": src_a, "role_b_1": src_b},  # 共享后端持源并集
            global_limits=GlobalLimits(max_concurrency=1, rpm=0, tpm=0),
        )
        started = asyncio.Event()

        async def slow_handler(request):
            started.set()
            await asyncio.sleep(0.2)
            return _sse()

        client_a = _client([src_a], slow_handler, limiter=shared)
        client_b = _client([src_b], limiter=shared, quota_full="fail_fast")
        task = asyncio.ensure_future(client_a.chat([{"role": "user", "content": "x"}]))
        await started.wait()
        with pytest.raises(AllSourcesExhausted) as ei:
            await client_b.chat([{"role": "user", "content": "y"}])
        assert ei.value.reason == "quota_exhausted"  # 全局闸被 A 占满 → B 立即失败
        await task

    async def test_aclose_idempotent(self):
        client = _client()
        await client.aclose()
        await client.aclose()


class TestGatherBounded:
    async def test_order_preserved_and_concurrency_capped(self):
        peak = {"now": 0, "max": 0}

        async def work(i):
            peak["now"] += 1
            peak["max"] = max(peak["max"], peak["now"])
            await asyncio.sleep(0.01)
            peak["now"] -= 1
            return i

        results = await gather_bounded((work(i) for i in range(10)), concurrency=3)
        assert results == list(range(10))
        assert peak["max"] <= 3

    async def test_exception_propagates(self):
        async def boom():
            raise RuntimeError("x")

        async def ok():
            return 1

        with pytest.raises(RuntimeError):
            await gather_bounded([ok(), boom()], concurrency=2)

    async def test_invalid_concurrency(self):
        with pytest.raises(ValueError):
            await gather_bounded([], concurrency=0)


def _load_reference_protocol(insert_path: str, module: str):
    sys.path.insert(0, str(_REPO / insert_path))
    try:
        import importlib

        return importlib.import_module(module).LLMProvider
    finally:
        sys.path.pop(0)


class TestReferenceProtocolCompat:
    """结构兼容断言(只读 import reference;失败即公共承诺破裂)。"""

    def test_satisfies_govdoc_llm_provider(self):
        try:
            proto = _load_reference_protocol(
                "reference/GovDoc-SaaS/packages/docagent-core/src", "docagent_core.protocols"
            )
        except ImportError:
            # 兜底: 按 protocols.py:15-25 逐字复制的结构断言
            from typing import Any, Protocol, runtime_checkable

            @runtime_checkable
            class proto(Protocol):  # noqa: N801 — 复制自 GovDoc protocols.py:15-25
                async def chat(
                    self,
                    messages: list[dict[str, Any]],
                    *,
                    session_id: str | None = None,
                    parent_call_id: str | None = None,
                ): ...

        assert isinstance(_client(), proto)

    def test_satisfies_videotree_llm_provider(self):
        try:
            proto = _load_reference_protocol("reference/Video-Tree-TRM5", "core.protocols")
        except ImportError:
            from typing import Any, Protocol, runtime_checkable

            @runtime_checkable
            class proto(Protocol):  # noqa: N801 — 复制自 VT core/protocols.py:18-29
                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,
                ): ...

        assert isinstance(_client(), proto)


# —— 资源所有权纪律(issue #15 D 组): 谁建的谁关,注入的一律不碰 ——

_CACHE_ENV = dict(
    _ENV, PGW_CACHE_BACKEND="memory", PGW_CACHE_NAMESPACE="proj", PGW_CACHE_TTL_S="3600"
)


class _Closable:
    """记 close 次数的假组件;所有权纪律的唯一观测点。

    带 `record_llm_call(**fields)` 是因为它也被当作注入的 telemetry recorder 用:
    1.3.5 的装配闸在构造期就会拒掉不满足 `TelemetryRecorder` 的对象
    (否则下游升级后 100% 丢遥测而调用照常成功)。`**fields` 形态天然兼容。
    """

    def __init__(self):
        self.closed = 0

    async def record_llm_call(self, **fields):
        pass

    async def aclose(self):
        self.closed += 1


class _SyncClosable:
    """只有同步 close 的假 recorder(SQLiteRecorder 形态,收敛后的 helper 须探测到)。"""

    def __init__(self):
        self.closed = 0

    async def record_llm_call(self, **fields):
        pass

    def close(self):
        self.closed += 1


class _FalsyClosable(_Closable):
    """`bool()` 为假的组件(空容器形态的后端就长这样)。

    所有权判定必须写 `is None` / `is not None`,不得写 `or`(设计 §3.4,ARCH §4.5
    细则 2): 写 `or` 时注入这样一个后端会**悄悄走自建分支**,而所有权标志按
    `is None` 判成 False——于是既没用上注入的那个,自建的那个又没人关,正是本
    issue 要修的泄漏原地复活。判定与标志一漂移,两个 bug 一起回来。
    """

    def __bool__(self):
        return False


def _parts(*names):
    return {name: _Closable() for name in names}


def _falsy_parts(*names):
    return {name: _FalsyClosable() for name in names}


def _patch_builders(monkeypatch, built, *, transport_path):
    """把工厂的自建点换成可计数假件;transport 无注入入口,故恒自建。"""
    monkeypatch.setattr(transport_path, lambda **kwargs: built["transport"])
    monkeypatch.setattr("polygateway.client._build_limiter", lambda s, src: built["limiter"])
    monkeypatch.setattr("polygateway.client._build_breaker", lambda s: built["breaker"])
    monkeypatch.setattr("polygateway.client._build_telemetry", lambda s: built["telemetry"])
    if "cache" in built:
        monkeypatch.setattr("polygateway.client._build_cache", lambda s: built["cache"])


class TestGatewayClientOwnership:
    """`__init__` 是全量注入路径,经它传入的一切都归调用方(设计 §3.4)。"""

    _GATEWAY_TRANSPORT = "polygateway.client.OpenAICompatTransport"

    async def test_injected_components_are_never_closed(self):
        """共享 recorder/transport 被第一个关闭的 client 弄死,正是 R5 显式共享走不通的原因。"""
        injected = _parts(*("transport", "telemetry", "cache", "limiter", "breaker"))
        client = _client(
            transport=injected["transport"],
            telemetry=injected["telemetry"],
            cache=injected["cache"],
            cache_namespace="proj",
            cache_ttl_s=3600,
            limiter=injected["limiter"],
            breaker=injected["breaker"],
        )
        await client.aclose()
        assert {name: part.closed for name, part in injected.items()} == {
            "transport": 0,
            "telemetry": 0,
            "cache": 0,
            "limiter": 0,
            "breaker": 0,
        }

    async def test_factory_closes_every_component_it_built(self, monkeypatch):
        """泄漏钉子: 自建的 redis limiter/breaker 今天没人关,连引用都没留。"""
        built = _parts("transport", "telemetry", "cache", "limiter", "breaker")
        _patch_builders(monkeypatch, built, transport_path=self._GATEWAY_TRANSPORT)
        client = GatewayClient.from_settings(GatewaySettings.from_env("LLM", env=_CACHE_ENV))
        await client.aclose()
        assert {name: part.closed for name, part in built.items()} == {
            "transport": 1,
            "telemetry": 1,
            "cache": 1,
            "limiter": 1,
            "breaker": 1,
        }

    async def test_factory_keeps_hands_off_injected_components(self, monkeypatch):
        built = _parts("transport", "telemetry", "cache", "limiter", "breaker")
        _patch_builders(monkeypatch, built, transport_path=self._GATEWAY_TRANSPORT)
        injected = _parts("telemetry", "cache", "limiter", "breaker")
        client = GatewayClient.from_settings(
            GatewaySettings.from_env("LLM", env=_CACHE_ENV),
            limiter=injected["limiter"],
            breaker=injected["breaker"],
            cache=injected["cache"],
            telemetry=injected["telemetry"],
        )
        await client.aclose()
        assert all(part.closed == 0 for part in injected.values())
        assert built["transport"].closed == 1  # 工厂恒自建 transport,归 client

    async def test_falsy_injected_components_are_still_injected(self, monkeypatch):
        """`is not None` 是所有权判定成立的**必要条件**,不是风格偏好(设计 §3.4)。

        改回 `or` 时: 工厂拿自建件顶掉注入件(下游以为在共享,其实各跑各的),
        且自建件的 `_owns_*` 仍是 False —— redis 客户端就地泄漏。
        """
        built = _parts("transport", "telemetry", "cache", "limiter", "breaker")
        _patch_builders(monkeypatch, built, transport_path=self._GATEWAY_TRANSPORT)
        injected = _falsy_parts("telemetry", "cache", "limiter", "breaker")
        client = GatewayClient.from_settings(
            GatewaySettings.from_env("LLM", env=_CACHE_ENV),
            limiter=injected["limiter"],
            breaker=injected["breaker"],
            cache=injected["cache"],
            telemetry=injected["telemetry"],
        )
        assert client._limiter_backend is injected["limiter"]
        assert client._breaker_backend is injected["breaker"]
        assert client._cache is injected["cache"]
        assert client._telemetry is injected["telemetry"]
        owns = (client._owns_limiter, client._owns_breaker, client._owns_cache)
        assert owns == (False, False, False) and client._owns_telemetry is False
        await client.aclose()
        assert all(part.closed == 0 for part in injected.values())
        # 自建件根本不该被造出来更不该被关;只有恒自建的 transport 归 client
        assert [built[name].closed for name in ("telemetry", "cache", "limiter", "breaker")] == [
            0,
            0,
            0,
            0,
        ]

    async def test_aclose_is_idempotent(self, monkeypatch):
        built = _parts("transport", "telemetry", "cache", "limiter", "breaker")
        _patch_builders(monkeypatch, built, transport_path=self._GATEWAY_TRANSPORT)
        client = GatewayClient.from_settings(GatewaySettings.from_env("LLM", env=_CACHE_ENV))
        await client.aclose()
        await client.aclose()
        assert all(part.closed == 1 for part in built.values())

    async def test_sync_only_recorder_is_closed(self, monkeypatch):
        """SQLiteRecorder 只有同步 `close()`;收敛成 helper 之后这条分支不得丢。"""
        built = _parts("transport", "cache", "limiter", "breaker")
        recorder = _SyncClosable()
        built["telemetry"] = recorder
        _patch_builders(monkeypatch, built, transport_path=self._GATEWAY_TRANSPORT)
        client = GatewayClient.from_settings(GatewaySettings.from_env("LLM", env=_CACHE_ENV))
        await client.aclose()
        assert recorder.closed == 1


def _embedding_client(**overrides):
    from polygateway.embedding import EmbeddingClient

    defaults = {
        "scope": "embed",
        "sources": [_source()],
        "selector": RoundRobinSelector(),
        "limiter": _Closable(),
        "breaker": _Closable(),
        "transport": _Closable(),
        "retry": RetryPolicy(3, 2.0, 30.0),
        "backpressure": BackpressurePolicy(300.0, 0.01),
        "batch_size": 2,
    }
    defaults.update(overrides)
    return EmbeddingClient(**defaults)


class TestEmbeddingClientOwnership:
    """三处必须各钉一次: 收敛成 helper 之后,有人把逻辑复制回去也得当场被发现。"""

    async def test_injected_components_are_never_closed(self):
        injected = _parts("transport", "telemetry", "limiter", "breaker")
        client = _embedding_client(
            transport=injected["transport"],
            telemetry=injected["telemetry"],
            limiter=injected["limiter"],
            breaker=injected["breaker"],
        )
        await client.aclose()
        assert all(part.closed == 0 for part in injected.values())

    async def test_factory_closes_every_component_it_built(self, monkeypatch):
        from polygateway.config import EmbeddingSettings
        from polygateway.embedding import EmbeddingClient

        built = _parts("transport", "telemetry", "limiter", "breaker")
        _patch_builders(
            monkeypatch,
            built,
            transport_path="polygateway.transports.openai_compat.OpenAICompatTransport",
        )
        settings = EmbeddingSettings(
            gateway=GatewaySettings.from_env("LLM", env=_ENV), batch_size=2
        )
        client = EmbeddingClient.from_settings(settings)
        await client.aclose()
        assert all(part.closed == 1 for part in built.values())

    async def test_factory_keeps_hands_off_injected_components(self, monkeypatch):
        from polygateway.config import EmbeddingSettings
        from polygateway.embedding import EmbeddingClient

        built = _parts("transport", "telemetry", "limiter", "breaker")
        _patch_builders(
            monkeypatch,
            built,
            transport_path="polygateway.transports.openai_compat.OpenAICompatTransport",
        )
        injected = _parts("telemetry", "limiter", "breaker")
        settings = EmbeddingSettings(
            gateway=GatewaySettings.from_env("LLM", env=_ENV), batch_size=2
        )
        client = EmbeddingClient.from_settings(
            settings,
            limiter=injected["limiter"],
            breaker=injected["breaker"],
            telemetry=injected["telemetry"],
        )
        await client.aclose()
        assert all(part.closed == 0 for part in injected.values())
        assert built["transport"].closed == 1

    async def test_falsy_injected_components_are_still_injected(self, monkeypatch):
        """三处工厂各写一遍 `is not None`,就是三处各有一次漂移回 `or` 的机会。"""
        from polygateway.config import EmbeddingSettings
        from polygateway.embedding import EmbeddingClient

        built = _parts("transport", "telemetry", "limiter", "breaker")
        _patch_builders(
            monkeypatch,
            built,
            transport_path="polygateway.transports.openai_compat.OpenAICompatTransport",
        )
        injected = _falsy_parts("telemetry", "limiter", "breaker")
        settings = EmbeddingSettings(
            gateway=GatewaySettings.from_env("LLM", env=_ENV), batch_size=2
        )
        client = EmbeddingClient.from_settings(
            settings,
            limiter=injected["limiter"],
            breaker=injected["breaker"],
            telemetry=injected["telemetry"],
        )
        assert client._limiter_backend is injected["limiter"]
        assert client._breaker_backend is injected["breaker"]
        assert client._telemetry is injected["telemetry"]
        assert (client._owns_limiter, client._owns_breaker, client._owns_telemetry) == (
            False,
            False,
            False,
        )
        await client.aclose()
        assert all(part.closed == 0 for part in injected.values())
        assert [built[name].closed for name in ("telemetry", "limiter", "breaker")] == [0, 0, 0]


def _ocr_client(**overrides):
    from polygateway.ocr import OcrClient

    defaults = {
        "scope": "ocr",
        "sources": [_source(name="m1", provider="monkey", model="monkey-ocr")],
        "selector": RoundRobinSelector(),
        "limiter": _Closable(),
        "breaker": _Closable(),
        "transport": _Closable(),
        "retry": RetryPolicy(3, 2.0, 30.0),
        "backpressure": BackpressurePolicy(300.0, 0.01),
    }
    defaults.update(overrides)
    return OcrClient(**defaults)


class TestOcrClientOwnership:
    async def test_injected_components_are_never_closed(self):
        injected = _parts("transport", "telemetry", "limiter", "breaker")
        client = _ocr_client(
            transport=injected["transport"],
            telemetry=injected["telemetry"],
            limiter=injected["limiter"],
            breaker=injected["breaker"],
        )
        await client.aclose()
        assert all(part.closed == 0 for part in injected.values())

    async def test_factory_closes_every_component_it_built(self, monkeypatch):
        from polygateway.config import OcrSettings
        from polygateway.ocr import OcrClient

        built = _parts("transport", "telemetry", "limiter", "breaker")
        _patch_builders(
            monkeypatch,
            built,
            transport_path="polygateway.transports.monkey_ocr.MonkeyOcrTransport",
        )
        client = OcrClient.from_settings(OcrSettings.from_env("OCR", env=dict(_OCR_ENV)))
        await client.aclose()
        assert all(part.closed == 1 for part in built.values())

    async def test_factory_keeps_hands_off_injected_components(self, monkeypatch):
        from polygateway.config import OcrSettings
        from polygateway.ocr import OcrClient

        built = _parts("transport", "telemetry", "limiter", "breaker")
        _patch_builders(
            monkeypatch,
            built,
            transport_path="polygateway.transports.monkey_ocr.MonkeyOcrTransport",
        )
        injected = _parts("telemetry", "limiter", "breaker")
        client = OcrClient.from_settings(
            OcrSettings.from_env("OCR", env=dict(_OCR_ENV)),
            limiter=injected["limiter"],
            breaker=injected["breaker"],
            telemetry=injected["telemetry"],
        )
        await client.aclose()
        assert all(part.closed == 0 for part in injected.values())
        assert built["transport"].closed == 1

    async def test_falsy_injected_components_are_still_injected(self, monkeypatch):
        from polygateway.config import OcrSettings
        from polygateway.ocr import OcrClient

        built = _parts("transport", "telemetry", "limiter", "breaker")
        _patch_builders(
            monkeypatch,
            built,
            transport_path="polygateway.transports.monkey_ocr.MonkeyOcrTransport",
        )
        injected = _falsy_parts("telemetry", "limiter", "breaker")
        client = OcrClient.from_settings(
            OcrSettings.from_env("OCR", env=dict(_OCR_ENV)),
            limiter=injected["limiter"],
            breaker=injected["breaker"],
            telemetry=injected["telemetry"],
        )
        assert client._limiter_backend is injected["limiter"]
        assert client._breaker_backend is injected["breaker"]
        assert client._telemetry is injected["telemetry"]
        assert (client._owns_limiter, client._owns_breaker, client._owns_telemetry) == (
            False,
            False,
            False,
        )
        await client.aclose()
        assert all(part.closed == 0 for part in injected.values())
        assert [built[name].closed for name in ("telemetry", "limiter", "breaker")] == [0, 0, 0]


class TestRedisCacheOwnership:
    """组件内部自建的连接归组件自己;照抄 RedisLimiter._owns_client 的正确先例。"""

    async def test_injected_client_is_not_closed(self):
        from polygateway.backends.redis_cache import RedisCache

        client = _Closable()
        await RedisCache(client).aclose()
        assert client.closed == 0

    async def test_self_built_client_is_closed_once(self, monkeypatch):
        from types import SimpleNamespace

        from polygateway.backends import redis_cache

        built = _Closable()
        monkeypatch.setattr(
            redis_cache, "aioredis", SimpleNamespace(from_url=lambda url, **kwargs: built)
        )
        cache = redis_cache.RedisCache.from_url("redis://localhost:6379/0")
        await cache.aclose()
        await cache.aclose()  # 幂等: 不重复关
        assert built.closed == 1


class TestTelemetryStatusExposure:
    """降级状态的只读出口: 一处 isinstance 判定,三个 client 各钉一次(设计 §3.3)。"""

    def _recorder(self, tmp_path):
        from polygateway.telemetry.sqlite import SQLiteRecorder

        return SQLiteRecorder(tmp_path / "telemetry.db", auto_migrate=True)

    def _assert_snapshot(self, status):
        from polygateway.types import TelemetryStatus

        assert isinstance(status, TelemetryStatus)
        assert status.degraded is False

    def test_gateway_client_without_telemetry_reports_none(self):
        assert _client().telemetry_status is None

    def test_gateway_client_with_foreign_recorder_reports_none(self):
        """注入的第三方 recorder 不提供状态 → None,绝不得抛 AttributeError。"""
        assert _client(telemetry=_Closable()).telemetry_status is None

    def test_gateway_client_with_builtin_recorder_reports_snapshot(self, tmp_path):
        self._assert_snapshot(_client(telemetry=self._recorder(tmp_path)).telemetry_status)

    def test_embedding_client_exposes_the_same_outlet(self, tmp_path):
        assert _embedding_client().telemetry_status is None
        assert _embedding_client(telemetry=_Closable()).telemetry_status is None
        self._assert_snapshot(
            _embedding_client(telemetry=self._recorder(tmp_path)).telemetry_status
        )

    def test_ocr_client_exposes_the_same_outlet(self, tmp_path):
        assert _ocr_client().telemetry_status is None
        assert _ocr_client(telemetry=_Closable()).telemetry_status is None
        self._assert_snapshot(_ocr_client(telemetry=self._recorder(tmp_path)).telemetry_status)


class TestManagedReasoningAdmission:
    """入口前置与实际准入边界保持明确。"""

    @pytest.mark.parametrize("factory", ["env", "settings"])
    def test_factory_rejects_conflict_before_building_backends(self, monkeypatch, factory):
        env = {
            **_ENV,
            "LLM__QWEN__1__MODEL": "qwen3.7-plus",
            "LLM__QWEN__1__ENABLE_THINKING": "true",
            "LLM__QWEN__1__EXTRA_BODY": '{"enable_thinking":true}',
        }
        built = []

        def forbidden(*args, **kwargs):
            built.append(True)
            raise AssertionError("后端不得构造")

        monkeypatch.setattr("polygateway.client._build_limiter", forbidden)
        with pytest.raises(ValueError, match="冲突"):
            if factory == "env":
                GatewayClient.from_env(env=env)
            else:
                GatewayClient.from_settings(GatewaySettings.from_env(env=env))
        assert not built

    async def test_request_conflict_does_not_enter_onion(self):
        client = _client()

        async def forbidden(request):
            raise AssertionError("不得进入洋葱")

        client._handler = forbidden
        try:
            with pytest.raises(ValueError, match="冲突"):
                await client.chat([], reasoning_effort="high", overlay={"reasoning_effort": "high"})
        finally:
            await client._transport.aclose()

    async def test_source_conflict_releases_admitted_permit(self):
        source = _source(
            reasoning_effort="high", provider="openai", extra_body={"reasoning_effort": "high"}
        )
        sent = []
        client = _client(sources=[source], handler=lambda request: sent.append(request) or _sse())
        try:
            with pytest.raises(RequestRejectedError, match="冲突"):
                await client.chat([])
            assert sent == []
            assert (await client._limiter_backend.source_stats(source.name)).inflight == 0
        finally:
            await client._transport.aclose()


async def test_managed_conflict_returns_half_open_probe_and_permit():
    """本地拒绝没有上游响应，不计故障且必须归还半开探针。"""
    from tests.contracts.conftest import FakeClock

    clock = FakeClock()
    source = _source(
        provider="openai", reasoning_effort="high", extra_body={"reasoning_effort": "high"}
    )
    gate = InMemoryGate(config=BreakerConfig(1, 1, 10), now=clock)
    entry = await gate.try_enter(source.name, "setup")
    await gate.record_failure(entry, "source_dead", True)
    clock.advance(2)
    client = _client(sources=[source], breaker=gate)
    try:
        with pytest.raises(RequestRejectedError, match="冲突"):
            await client.chat([])
        assert (await client._limiter_backend.source_stats(source.name)).inflight == 0
        next_entry = await gate.try_enter(source.name, "next")
        assert next_entry.allowed and next_entry.is_probe
        await gate.release_probe(next_entry)
    finally:
        await client._transport.aclose()


async def test_synthetic_runtime_protocol_and_legacy_call_signatures():
    """合成 Protocol 仅证明本库兼容契约，不冒充缺失下游实际验收。"""
    import inspect
    from typing import Protocol, runtime_checkable

    @runtime_checkable
    class Caller(Protocol):
        """旧调用点只依赖 chat 协议。"""

        async def chat(self, messages, **kwargs): ...

    client = GatewayClient.from_env("LLM", env=_ENV)
    try:
        assert isinstance(client, Caller)
        signature = inspect.signature(client.chat)
        signature.bind([], session_id="session", parent_call_id="step")
        signature.bind([], session_id="session", cache_salt="epoch-1")
        assert client._transport._clients == {}
    finally:
        await client.aclose()


class _StatsClock:
    """确定性单调钟;测试主动推进以断言"哪些区段计入了总耗时"。"""

    def __init__(self, start=1000.0):
        self.t = start

    def __call__(self):
        return self.t

    def advance(self, seconds):
        self.t += seconds


class _TickingCache:
    """假缓存后端: 每次 IO 推进注入钟。

    不推进时钟的替身会让"缓存 IO 计入总耗时"的断言退化成恒等于 0 的空转绿
    (计划 §T1 替身构造要求)。
    """

    def __init__(self, clock, tick=0.25):
        self._clock = clock
        self._tick = tick
        self._data = {}

    async def get(self, key):
        self._clock.advance(self._tick)
        return self._data.get(key)

    async def set(self, key, value, ttl_s):
        self._clock.advance(self._tick)
        self._data[key] = value


class _TickingRecorder:
    """假 recorder: 写入时推进注入钟,用于断言内联遥测收尾计入总耗时。"""

    def __init__(self, clock, tick=0.5):
        self._clock = clock
        self._tick = tick
        self.rows = []

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


class TestLogicalCallStats:
    """一次公开 chat 调用的统计(1.3.5 设计 §3)。"""

    _MSG = [{"role": "user", "content": "hi"}]

    async def test_success_reports_one_attempt(self):
        async with _client() as client:
            resp = await client.chat(self._MSG)
        assert resp.call_stats is not None
        assert resp.call_stats.attempts == 1
        assert resp.call_stats.logical_call_id

    async def test_concurrent_calls_do_not_share_counters_or_ids(self):
        """同一 client 并发两路必须各自计数与各自 ID(库铁律「纯 asyncio 中立」)。

        上下文若被提升成 client 实例属性,这条就会红——那正是 VT
        `evolve_llm = llm` 教训的同一形态。
        """
        async with _client() as client:
            a, b = await asyncio.gather(client.chat(self._MSG), client.chat(self._MSG))
        assert a.call_stats.logical_call_id != b.call_stats.logical_call_id
        assert a.call_stats.attempts == b.call_stats.attempts == 1

    async def test_cache_hit_is_zero_attempts_with_a_fresh_logical_id(self):
        """命中不产生网关调用 → 0 尝试;且是**新**逻辑调用,不回放历史统计。"""
        clock = _StatsClock()
        cache = _TickingCache(clock)
        async with _client(
            cache=cache, cache_namespace="proj", cache_ttl_s=600, now=clock
        ) as client:
            first = await client.chat(self._MSG)
            second = await client.chat(self._MSG)
        assert first.cache_hit is False and first.call_stats.attempts == 1
        assert second.cache_hit is True
        assert second.call_stats.attempts == 0
        assert second.call_stats.logical_call_id != first.call_stats.logical_call_id

    async def test_cache_io_counts_into_total_latency(self):
        """缓存读写是本次调用真实花掉的时间,必须进总耗时(设计 §3)。"""
        clock = _StatsClock()
        cache = _TickingCache(clock, tick=0.25)
        async with _client(
            cache=cache, cache_namespace="proj", cache_ttl_s=600, now=clock
        ) as client:
            hit = (await client.chat(self._MSG), await client.chat(self._MSG))[1]
        # 命中路径只有一次 get(0.25s),无网关调用
        assert hit.call_stats.attempts == 0
        assert hit.call_stats.total_latency_ms == 250

    async def test_inline_telemetry_teardown_counts_into_total_latency(self):
        """成功响应的快照含返回前已完成的内联遥测耗时(设计 §6)。"""
        clock = _StatsClock()
        recorder = _TickingRecorder(clock, tick=0.5)
        async with _client(telemetry=recorder, now=clock) as client:
            resp = await client.chat(self._MSG)
        assert recorder.rows  # 确实写了行,否则本断言空转
        assert resp.call_stats.total_latency_ms == 500

    async def test_milliseconds_not_seconds(self):
        """毫秒/秒不混用: 1.5s 必须是 1500 而不是 1 或 1.5。"""
        clock = _StatsClock()
        recorder = _TickingRecorder(clock, tick=1.5)
        async with _client(telemetry=recorder, now=clock) as client:
            resp = await client.chat(self._MSG)
        assert resp.call_stats.total_latency_ms == 1500

    async def test_stats_work_without_any_telemetry(self):
        """统计生效与否**不由 telemetry 是否启用决定**(设计 §3.5)。"""
        async with _client(telemetry=None) as client:
            resp = await client.chat(self._MSG)
        assert resp.call_stats is not None and resp.call_stats.attempts == 1

    async def test_failure_exception_carries_no_stats_attribute(self):
        """本版**不向异常对象附加统计**(设计 §3.1): 第三方可能复用同一异常实例。"""

        def reject(request):
            return httpx.Response(400, json={"error": {"message": "bad"}})

        async with _client(handler=reject) as client:
            with pytest.raises(RequestRejectedError) as exc:
                await client.chat(self._MSG)
        assert hasattr(exc.value, "call_stats") is False

    async def test_input_validation_stays_outside_the_stats_boundary(self):
        """校验异常保持原行为,发生在统计边界之外(设计 §3)。"""
        async with _client() as client:
            with pytest.raises(ValueError, match="meta"):
                await client.chat(self._MSG, meta={"BAD-KEY": 1})


class TestChatTerminalFailureRows:
    """chat 链路的**真实**终态行(1.3.5 设计 §6;补漏而非改口径)。

    这些路径改前一条失败行都没有(结构化耗尽)或只有尝试行,
    "这次调用到底失败了几次"因此 SQL 答不出来。
    """

    _MSG = [{"role": "user", "content": "hi"}]

    def _rows(self, recorder, kind):
        return [r for r in recorder.rows if r["event_kind"] == kind]

    async def test_structured_exhaustion_writes_the_only_failure_row(self):
        """结构化耗尽发生在 transport 成功之后: attempt 行全是成功行,终态是唯一记录。"""
        from pydantic import BaseModel

        class Answer(BaseModel):
            answer: int

        recorder = _MemoryRecorder()
        async with _client(
            handler=lambda request: _sse("not json at all"),
            telemetry=recorder,
            structured_max_retries=1,
        ) as client:
            with pytest.raises(ResultInvalidError):
                await client.chat(self._MSG, structured=Answer)
        attempts = self._rows(recorder, "attempt")
        terminals = self._rows(recorder, "terminal_failure")
        assert [a["error"] for a in attempts] == [None, None]  # 两次尝试都成功
        assert len(terminals) == 1
        row = terminals[0]
        assert row["error_type"] == "ResultInvalidError"
        # C2: 有界结构化说明并入 error,且不含 raw_text(正文预算已由 attempt 行承担)
        assert "validation=" in row["error"] or "repair=" in row["error"]
        assert len(row["error"]) < 1200

    async def test_retry_exhaustion_writes_exactly_one_terminal_row(self):
        """重试耗尽: 逐次 attempt 行之外只能有**一条**终态行。

        终态行的 `attempts` 是整次逻辑调用的真实尝试数——这正是改前 SQL
        答不出的"这次调用到底重试了几次"。
        """
        recorder = _MemoryRecorder()
        async with _client(
            handler=lambda request: httpx.Response(503),
            telemetry=recorder,
            retry=RetryPolicy(3, 0.001, 0.01),
        ) as client:
            with pytest.raises(AllSourcesExhausted):
                await client.chat(self._MSG)
        attempts = self._rows(recorder, "attempt")
        terminals = self._rows(recorder, "terminal_failure")
        assert len(attempts) == 3
        assert len(terminals) == 1
        row = terminals[0]
        assert row["error_type"] == "AllSourcesExhausted"
        assert row["attempts"] == 3  # 整次逻辑调用的尝试数
        assert row["scope"] == "llm" and row["operation"] == "chat"
        # 终态行的 latency_ms 与 total_latency_ms 同取一份冻结快照
        assert row["latency_ms"] == row["total_latency_ms"]

    async def test_request_rejected_now_has_both_an_attempt_and_a_terminal_row(self):
        """已批准的下游可见变化: 400 密集负载的错误行翻倍,失败计数只能取终态。"""
        recorder = _MemoryRecorder()
        async with _client(
            handler=lambda request: httpx.Response(400, json={"error": {"message": "bad"}}),
            telemetry=recorder,
        ) as client:
            with pytest.raises(RequestRejectedError):
                await client.chat(self._MSG)
        attempts = self._rows(recorder, "attempt")
        terminals = self._rows(recorder, "terminal_failure")
        assert len(attempts) == 1 and attempts[0]["http_status_code"] == 400
        assert len(terminals) == 1
        # C1: 终态不搬运最后一次 attempt 的状态码与正文
        assert terminals[0]["http_status_code"] is None
        assert terminals[0]["error_body"] is None
        # 两类行共享同一 logical_call_id,归因查询才连得起来
        assert attempts[0]["logical_call_id"] == terminals[0]["logical_call_id"]

    async def test_non_domain_exception_writes_no_terminal_row(self):
        """编程错原样传播,本版**不承诺**任何统计或终态行,也不偷偷改分类。"""
        recorder = _MemoryRecorder()

        async def boom(request):
            raise KeyError("programming error")

        async with _client(telemetry=recorder) as client:
            client._handler = boom
            with pytest.raises(KeyError):
                await client.chat(self._MSG)
        assert self._rows(recorder, "terminal_failure") == []

    async def test_cancellation_writes_at_most_one_terminal_row(self):
        """取消尽力写一条(允许 0),且 `CancelledError` 类型与语义不变。"""
        recorder = _MemoryRecorder()

        async def hang(request):
            raise asyncio.CancelledError

        async with _client(telemetry=recorder) as client:
            client._handler = hang
            with pytest.raises(asyncio.CancelledError):
                await client.chat(self._MSG)
        terminals = self._rows(recorder, "terminal_failure")
        assert len(terminals) == 1
        assert terminals[0]["error"] == "cancelled"
        # 字符串不解析猜诊断
        assert terminals[0]["error_type"] is None

    async def test_terminal_row_is_written_once_per_logical_call(self):
        """`claim_terminal` 去重: 同一次调用即便出口被多次触达也只有一条。"""
        from polygateway.middleware.telemetry import emit_terminal_once
        from polygateway.types import _CallContext

        recorder = _MemoryRecorder()
        async with _client(telemetry=recorder) as client:
            context = _CallContext(now=client._now)
            request = ChatRequest(messages=self._MSG, call_context=context)
            for _ in range(3):
                await emit_terminal_once(
                    client._emitter,
                    request=request,
                    context=context,
                    error=AllSourcesExhausted(
                        scope="llm", reason="retry_exhausted", retry_after_s=1.0
                    ),
                    operation="chat",
                )
        assert len(self._rows(recorder, "terminal_failure")) == 1


@pytest.fixture
def captured_warnings():
    """捕获库发出的 WARNING;loguru 不经标准 logging,pytest 的 caplog 抓不到。

    名字避开裸 `warnings`: 那会遮蔽标准库模块名,本文件将来任何一次
    `import warnings` 都会与它静默互相顶掉,而报错点离真因很远。
    """
    from loguru import logger

    messages: list[str] = []
    sink_id = logger.add(messages.append, level="WARNING")
    yield messages
    logger.remove(sink_id)


class TestTerminalEmitDegradation:
    """终态出口的降级方向: 诊断字段的**提取**同样在降级范围内(铁律"遥测写失败降级不冒泡")。

    改前 `_record` 的 except 只包住写入本身,而错误诊断列在它之外求值——
    经公共扩展点(自实现 `StructuredOutputStrategy`/transport)构造出的
    `ResultInvalidError(validation_errors=(非 str,))` 会让提取期抛 `TypeError`
    顶替调用方本该收到的领域异常,错误四分类当场被击穿。
    """

    _MSG = [{"role": "user", "content": "hi"}]

    def _rows(self, recorder, kind):
        return [r for r in recorder.rows if r["event_kind"] == kind]

    async def test_broken_validation_errors_keep_the_domain_error(self, captured_warnings):
        """自实现策略给出非 str 的 `validation_errors`: 领域异常必须原样上抛。"""

        class _BadStrategy:
            """公共端口 `StructuredOutputStrategy` 的下游实现(库外没有类型执法)。"""

            def request_overlay(self, schema):
                return {}

            def parse(self, text):
                raise ResultInvalidError("模型返回不可解析", validation_errors=(object(),))

        recorder = _MemoryRecorder()
        async with _client(telemetry=recorder, structured_strategy=_BadStrategy()) as client:
            with pytest.raises(ResultInvalidError):  # 不是 TypeError
                await client.chat(self._MSG, structured="json")
        # 降级有声: 静默吞掉等于遥测缺口无人知道
        assert [m for m in captured_warnings if "终态遥测记录失败" in m]
        # 尝试行不受影响;终态行按 best effort 允许 0 条,但绝不能重复
        assert len(self._rows(recorder, "attempt")) == 1
        assert len(self._rows(recorder, "terminal_failure")) <= 1

    async def test_cancellation_is_never_swallowed_by_the_degradation(self):
        """降级不得吞取消: 写入那一次 await 上被取消,`CancelledError` 照常传播。"""
        from polygateway.middleware.telemetry import emit_terminal_once
        from polygateway.types import _CallContext

        class _CancellingEmitter:
            async def emit_terminal_failure(self, **kwargs):
                raise asyncio.CancelledError

        context = _CallContext(now=asyncio.get_running_loop().time)
        request = ChatRequest(messages=self._MSG, call_context=context)
        with pytest.raises(asyncio.CancelledError):
            await emit_terminal_once(
                _CancellingEmitter(),
                request=request,
                context=context,
                error=AllSourcesExhausted(scope="llm", reason="retry_exhausted", retry_after_s=1.0),
                operation="chat",
            )