"""OcrClient 治理循环测试(M3 设计 §5): 换源/熔断口径/stall/取消/G1 契约。

循环与 EmbeddingClient 同构(设计 §2.A 有限重复裁决);本文件的
retry_exhausted/circuit_open/stalled 三组断言即设计 §6 ③ 的 G1 契约钉。
"""

import asyncio
from types import SimpleNamespace

import httpx
import pytest
from loguru import logger

from polygateway.backends.memory.breaker import InMemoryGate
from polygateway.backends.memory.limiter import InMemoryLimiter
from polygateway.errors import (
    AllSourcesExhausted,
    CallDeadlineExceeded,
    CircuitOpenError,
    RequestRejectedError,
    ResultInvalidError,
    SourceDeadError,
    TransientError,
)
from polygateway.ocr import OcrClient
from polygateway.transports.monkey_ocr import MonkeyOcrTransport
from polygateway.types import (
    BackpressurePolicy,
    BreakerConfig,
    GlobalLimits,
    OcrDocumentElement,
    OcrDocumentTransportResult,
    OcrLayoutElement,
    OcrLayoutTransportResult,
    OcrTextTransportResult,
    RetryPolicy,
    SourceConfig,
)
from tests.unit import _document_fixtures as fx

_BREAKER = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0)
_NO_GLOBAL = GlobalLimits(max_concurrency=0, rpm=0, tpm=0)
_TEXT_OK = OcrTextTransportResult(text="LINE-1", raw={"task_type": "text"})
_DOC_OK = OcrDocumentTransportResult(
    elements=[
        OcrDocumentElement(
            type="text", bbox=(1.0, 2.0, 3.0, 4.0), page_index=0, text="SYNTH-DOC-BODY"
        )
    ],
    page_sizes=[(612.0, 792.0)],
    markdown="# SYNTH-MD",
    raw={"success": True},
)
_LAYOUT_OK = OcrLayoutTransportResult(
    elements=[OcrLayoutElement(type="table", bbox=(41.0, 48.0, 218.0, 282.0), page_index=0)],
    page_sizes=[(759.0, 540.0)],
    raw={"success": True},
)


def _src(**overrides):
    base = {
        "name": "m1",
        "provider": "monkey",
        "base_url": "http://10.77.0.20:7866",
        "api_key": "none",
        "model": "monkey-ocr",
        "timeout_s": 120.0,
    }
    base.update(overrides)
    return SourceConfig(**base)


class ScriptedOcrTransport:
    """按脚本响应: Exception / "text" / "layout" / "hang";两方法共用一份脚本。"""

    def __init__(self, script):
        self.script = list(script)
        self.calls = []
        # 取消用例的确定性窗口: 进入 hang 分支即置位, 不用 sleep 撞窗口
        self.entered = asyncio.Event()

    async def _next(self, method, source, call_id):
        self.calls.append((method, source.name, call_id))
        action = self.script.pop(0)
        if isinstance(action, Exception):
            raise action
        if action == "hang":
            self.entered.set()
            await asyncio.Event().wait()
        if action == "text":
            return _TEXT_OK
        if action == "document":
            return _DOC_OK
        return _LAYOUT_OK

    async def recognize_text(self, *, image, source, call_id):
        return await self._next("text", source, call_id)

    async def parse_layout(self, *, image, source, call_id):
        return await self._next("layout", source, call_id)

    async def parse_document(self, *, document, source, call_id):
        return await self._next("document", source, call_id)

    async def check_health(self, *, source):
        raise NotImplementedError


class ClockAdvancingOcrTransport(ScriptedOcrTransport):
    """按脚本 [(推进秒数, 动作), ...] 在一次尝试内部推进时钟(issue #8)。

    stall 口径要区分"时间花在哪",故必须能让时钟只在 transport 内前进。
    """

    def __init__(self, script, clock):
        super().__init__([a for _, a in script])
        self._advances = [d for d, _ in script]
        self.clock = clock

    async def _next(self, method, source, call_id):
        self.clock.advance(self._advances.pop(0))
        return await super()._next(method, source, call_id)


class StaticSelector:
    def order(self, sources, stats):
        return list(sources)


class RecordingSelector(StaticSelector):
    def __init__(self):
        self.outcomes = []

    def record_outcome(self, source_name, ok):
        self.outcomes.append((source_name, ok))

    def health(self, source_name):
        return 1.0


class RecordingGate:
    """InMemoryGate 包装: 录 record_success 的 count_attempt 与 release_probe。"""

    def __init__(self, inner):
        self._inner = inner
        self.successes = []
        self.probe_releases = 0

    async def try_enter(self, source_name, owner):
        return await self._inner.try_enter(source_name, owner)

    async def record_success(self, entry, *, count_attempt=True):
        self.successes.append((entry.source_name, count_attempt))
        return await self._inner.record_success(entry, count_attempt=count_attempt)

    async def record_failure(self, entry, reason, force_open):
        return await self._inner.record_failure(entry, reason, force_open)

    async def release_probe(self, entry):
        self.probe_releases += 1
        return await self._inner.release_probe(entry)

    async def retry_after_s(self, sources):
        return await self._inner.retry_after_s(sources)


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

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


def _client(sources, script, **overrides):
    limiter = InMemoryLimiter(
        scope="ocr",
        sources={s.name: s for s in sources},
        global_limits=_NO_GLOBAL,
        lease_ttl_s=100.0,
    )
    gate = RecordingGate(InMemoryGate(config=_BREAKER))
    kwargs = {
        "scope": "ocr",
        "sources": sources,
        "selector": RecordingSelector(),
        "limiter": limiter,
        "breaker": gate,
        "transport": ScriptedOcrTransport(script),
        "retry": RetryPolicy(max_attempts=3, backoff_base_s=0.001, backoff_max_s=0.01),
        "backpressure": BackpressurePolicy(stall_window_s=300.0, poll_interval_s=0.001),
    }
    kwargs.update(overrides)
    return OcrClient(**kwargs), limiter, gate


class TestSuccessPaths:
    async def test_text_result_fields(self):
        client, _, gate = _client([_src()], ["text"])
        r = await client.recognize_text(b"jpg")
        assert r.text == "LINE-1"
        assert r.source_name == "m1"
        assert r.usage.prompt_tokens == 0 and r.usage.completion_tokens == 0
        assert r.latency_ms >= 0 and r.call_id and r.raw["task_type"] == "text"
        assert gate.successes == [("m1", True)]
        assert client._selector.outcomes == [("m1", True)]

    async def test_layout_result_fields(self):
        client, _, _ = _client([_src()], ["layout"])
        r = await client.parse_layout(b"jpg")
        assert r.elements[0].type == "table" and r.page_sizes == [(759.0, 540.0)]

    async def test_input_validation(self):
        client, _, _ = _client([_src()], [])
        with pytest.raises(TypeError):
            await client.recognize_text("not-bytes")
        with pytest.raises(ValueError):
            await client.parse_layout(b"")


class TestOcrGenerationMs:
    """OCR 裸生成时间单次覆盖(1.3.7 H8): 计时只包 transport 调用本身。"""

    async def test_generation_ms_single_transport_call(self):
        # 0.5s 为二进制可精确表示值: int 截断下非精确值会因浮点误差少 1ms
        clock = FakeClock()
        transport = ClockAdvancingOcrTransport([(0.5, "text")], clock)
        client, _, _ = _client([_src()], [], now=clock, transport=transport)
        r = await client.recognize_text(b"jpg")
        assert r.text == "LINE-1"
        assert r.call_stats is not None
        assert r.call_stats.attempts == 1
        assert r.call_stats.generation_ms == 500


class TestFailover:
    async def test_transient_retries_with_backoff(self):
        sleeps = []

        async def fake_sleep(delay):
            sleeps.append(delay)

        client, _, _ = _client(
            [_src()], [TransientError("boom", status_code=500), "text"], sleep=fake_sleep
        )
        r = await client.recognize_text(b"jpg")
        assert r.text == "LINE-1"
        assert any(s > 0 for s in sleeps)  # Transient 退避后重试
        assert client._selector.outcomes == [("m1", False), ("m1", True)]

    async def test_source_dead_switches_immediately(self):
        sleeps = []

        async def fake_sleep(delay):
            sleeps.append(delay)

        s1, s2 = _src(name="m1"), _src(name="m2", base_url="http://10.77.0.20:7867")
        client, _, _ = _client(
            [s1, s2], [SourceDeadError("401", status_code=401), "text"], sleep=fake_sleep
        )
        r = await client.recognize_text(b"jpg")
        assert r.source_name == "m2"
        assert not [s for s in sleeps if s > 0.0005]  # dead 立即换源不退避

    async def test_retry_exhausted_carries_g1_fields(self):
        client, _, _ = _client(
            [_src()],
            [TransientError("1", status_code=500)] * 3,
        )
        with pytest.raises(AllSourcesExhausted) as ei:
            await client.recognize_text(b"jpg")
        exc = ei.value
        assert exc.reason == "retry_exhausted"
        assert exc.per_source_reasons == {"m1": "network_error"}  # G1: 逐源原因
        assert exc.retry_after_s > 0  # G1: 可延期重投

    async def test_circuit_open_carries_g1_fields(self):
        client, _, _ = _client([_src()], [SourceDeadError("401", status_code=401)])
        with pytest.raises(CircuitOpenError) as ei:
            await client.recognize_text(b"jpg")
        exc = ei.value
        assert exc.per_source_reasons["m1"] == "circuit_open"
        assert exc.retry_after_s > 0


class TestTerminalOutcomes:
    async def test_result_invalid_counts_no_attempt_and_no_feed(self):
        client, _, gate = _client([_src()], [ResultInvalidError("bad zip")])
        with pytest.raises(ResultInvalidError):
            await client.parse_layout(b"jpg")
        assert gate.successes == [("m1", False)]  # 记成功但不入失败率窗
        assert client._selector.outcomes == []  # 坏结果 ≠ 坏服务,不喂健康

    async def test_rejected_with_status_counts_no_attempt(self):
        client, _, gate = _client([_src()], [RequestRejectedError("parse failed", status_code=200)])
        with pytest.raises(RequestRejectedError):
            await client.parse_layout(b"jpg")
        assert gate.successes == [("m1", False)]

    async def test_rejected_without_status_no_gate_success(self):
        client, _, gate = _client([_src()], [RequestRejectedError("local refuse")])
        with pytest.raises(RequestRejectedError):
            await client.recognize_text(b"jpg")
        assert gate.successes == []  # 服务未响应: 不记成功(非探针也不归还)


class TestBackpressure:
    async def test_fail_fast_when_quota_full(self):
        src = _src(max_concurrency=1)
        client, limiter, _ = _client([src], ["text"], quota_full="fail_fast")
        permit = await limiter.acquire("m1", 0)  # 占满唯一并发位
        try:
            with pytest.raises(AllSourcesExhausted) as ei:
                await client.recognize_text(b"jpg")
            assert ei.value.reason == "quota_exhausted"
        finally:
            await permit.settle(0)
            await permit.release()

    async def test_wait_until_permit_freed(self):
        src = _src(max_concurrency=1)
        client, limiter, _ = _client([src], ["text"])
        permit = await limiter.acquire("m1", 0)

        async def free_later():
            await asyncio.sleep(0.01)
            await permit.settle(0)
            await permit.release()

        release_task = asyncio.create_task(free_later())
        r = await client.recognize_text(b"jpg")
        await release_task
        assert r.text == "LINE-1"  # quota_full=wait: 等到许可释放而非报错

    async def test_stalled_when_no_progress(self):
        src = _src(max_concurrency=1)
        client, limiter, _ = _client(
            [src],
            ["text"],
            backpressure=BackpressurePolicy(stall_window_s=0.01, poll_interval_s=0.001),
        )
        permit = await limiter.acquire("m1", 0)  # 永不释放且全局无进展
        try:
            with pytest.raises(AllSourcesExhausted) as ei:
                await client.recognize_text(b"jpg")
            assert ei.value.reason == "stalled"
            assert ei.value.per_source_reasons["m1"] == "rate_limited"
        finally:
            await permit.settle(0)
            await permit.release()

    async def test_single_timeout_does_not_exhaust_stall_budget(self):
        """issue #8: 一次耗满 timeout 的尝试不得吃掉 stall 预算而使重试失效。

        OCR 只有 `_on_no_runnable` 一处 stall 判定,故失效链条是"先超时一次
        (墙钟耗尽)→ 再遇到无可用源 → 判死"。此处正是这条路径。
        """
        clock = FakeClock()
        limiter = held = None
        rounds = []

        async def toggle_permit(_seconds):
            """首次退避占满 permit,迫使下一轮走 _on_no_runnable;之后放行。"""
            nonlocal held
            rounds.append(_seconds)
            if len(rounds) > 10:
                raise RuntimeError("超过 10 次轮询仍未判死/未获 permit")
            if len(rounds) == 1:
                held = await limiter.acquire("m1", 0)
            else:
                await held.settle(0)
                await held.release()

        transport = ClockAdvancingOcrTransport(
            [(300.1, TransientError("timeout", status_code=504)), (0.0, "text")], clock
        )
        client, limiter, _ = _client(
            [_src(max_concurrency=1)], [], now=clock, sleep=toggle_permit, transport=transport
        )
        r = await client.recognize_text(b"jpg")
        assert r.text == "LINE-1"
        assert len(transport.calls) == 2  # 第二次尝试确实发出了


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

    def __call__(self):
        return self.t

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


class TestCancellation:
    async def test_cancel_during_probe_returns_probe(self):
        # 探针取消归还(铁律分支;verifier I3): 开路→冷却过后半开探针→
        # transport 挂起中取消 → release_probe 必须发生,后续可再探
        clock = FakeClock()
        inner = InMemoryGate(config=_BREAKER, now=clock)  # 门与 client 共用注入钟
        gate = RecordingGate(inner)
        client, _, _ = _client([_src()], ["hang"], now=clock, breaker=gate)
        entry = await inner.try_enter("m1", "setup")
        await inner.record_failure(entry, "source_dead", True)  # force_open → OPEN
        clock.advance(61.0)  # 越过 cooldown_s=60 → HALF_OPEN
        task = asyncio.create_task(client.recognize_text(b"jpg"))
        await asyncio.sleep(0.05)
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
        assert gate.probe_releases >= 1  # 探针已归还,不悬挂
        probe = await inner.try_enter("m1", "again")
        assert probe.allowed and probe.is_probe  # 可再探 = 未悬挂的行为证据

    async def test_cancel_during_backoff_sleep_propagates(self):
        async def cancelling_sleep(delay):
            raise asyncio.CancelledError()

        client, limiter, _ = _client(
            [_src()], [TransientError("boom", status_code=500)], sleep=cancelling_sleep
        )
        with pytest.raises(asyncio.CancelledError):
            await client.recognize_text(b"jpg")
        stats = await limiter.source_stats("m1")
        assert stats.inflight == 0  # 退避期取消: permit 已在 finally 释放

    async def test_cancel_during_transport_releases_permit(self):
        src = _src(max_concurrency=1)
        client, limiter, _ = _client([src], ["hang"])
        task = asyncio.create_task(client.recognize_text(b"jpg"))
        await asyncio.sleep(0.01)
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
        stats = await limiter.source_stats("m1")
        assert stats.inflight == 0  # permit 在 finally 释放

    async def test_cancel_in_flight_still_settles_zero(self):
        """S6: OCR 的 0 token 是**事实**而非"未知", 取消也不得改成按 est 结算。"""
        transport = ScriptedOcrTransport(["hang"])
        client, limiter, _ = _client(
            [_src(max_concurrency=1, tpm=1000, est_tokens=400)], [], transport=transport
        )
        task = asyncio.create_task(client.recognize_text(b"jpg"))
        await transport.entered.wait()
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
        stats = await limiter.source_stats("m1")
        assert stats.tpm_used == 0 and stats.inflight == 0


class TestCheckHealth:
    class _HealthTransport(ScriptedOcrTransport):
        def __init__(self, mapping):
            super().__init__([])
            self.mapping = mapping

        async def check_health(self, *, source):
            result = self.mapping[source.name]
            if result == "hang":
                await asyncio.Event().wait()
            if isinstance(result, Exception):
                raise result
            return result

    async def test_per_source_dict(self):
        s1, s2 = _src(name="m1"), _src(name="m2", base_url="http://10.77.0.20:7867")
        client, _, _ = _client([s1, s2], [])
        client._transport = self._HealthTransport({"m1": True, "m2": False})
        assert await client.check_health() == {"m1": True, "m2": False}

    async def test_cancellation_passes_through(self):
        client, _, _ = _client([_src()], [])
        client._transport = self._HealthTransport({"m1": "hang"})
        task = asyncio.create_task(client.check_health())
        await asyncio.sleep(0.01)
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task


class TestExtraBodyStripped:
    """issue #4 决策 G: OCR 路径只发 multipart 表单,剥离 extra_body 并 warning。"""

    def test_stripped_with_warning_but_assembly_succeeds(self):
        messages: list[str] = []
        sink_id = logger.add(messages.append, level="WARNING")
        try:
            client, _, _ = _client([_src(extra_body={"temperature": 0})], ["text"])
        finally:
            logger.remove(sink_id)
        assert client._sources[0].extra_body == {}
        assert any("extra_body" in m for m in messages)

    async def test_telemetry_never_records_a_parameter_that_was_not_sent(self):
        """不剥离则审计表会显示这次 OCR 带了 temperature=0——数据造假。"""
        recorder = _MemoryRecorder()
        client, _, _ = _client([_src(extra_body={"temperature": 0})], ["text"], telemetry=recorder)
        await client.recognize_text(b"IMG")
        assert recorder.rows[0]["sampling"] is None


class TestTelemetry:
    async def test_success_and_failure_recorded_without_image_bytes(self):
        recorder = _MemoryRecorder()
        client, _, _ = _client(
            [_src()],
            [TransientError("boom", status_code=500), "text"],
            telemetry=recorder,
        )
        await client.recognize_text(b"RAW-IMAGE-BYTES")
        assert len(recorder.rows) == 2  # 失败尝试与成功尝试均必录
        for row in recorder.rows:
            assert "<ocr:text image_bytes=15>" in row["messages"]
            assert "RAW-IMAGE-BYTES" not in row["messages"]  # 图像字节绝不入库
        assert recorder.rows[0]["error"].startswith("TransientError:")  # 类名前缀口径
        assert recorder.rows[1]["error"] is None
        assert recorder.rows[1]["prompt_tokens"] == 0

    async def test_success_row_stays_measured_and_settles_zero(self):
        """OCR 的 0 token 是**事实**而非未知(est_tokens 解耦设计 §3.3 剔出决定)。

        三态化不得把 OCR 成功行改成 `unavailable`——那会灌水缺口度量
        `COUNT(*) WHERE usage_source='unavailable'`;settle 恒 0 的差异①同样不动。
        """
        recorder = _MemoryRecorder()
        client, limiter, _ = _client([_src(tpm=1000, est_tokens=400)], ["text"], telemetry=recorder)
        await client.recognize_text(b"jpg")
        row = recorder.rows[0]
        assert row["usage_source"] == "measured"
        assert row["prompt_tokens"] == 0 and row["completion_tokens"] == 0
        assert row["error"] is None
        assert (await limiter.source_stats("m1")).tpm_used == 0  # settle(0) 全额退回预扣


class TestOcrCallerDimensions:
    """issue #11: 调用方自定义维度必须沿 OCR 链四层透传到每一行遥测。

    OCR 行与 chat 行落在同一张 `llm_calls` 表: 不覆盖这条链会让同一张表里
    一部分行有租户归属、一部分永远空白,而"先启用后加列则归属无法还原"。
    """

    async def test_recognize_text_row_carries_dimensions(self):
        recorder = _MemoryRecorder()
        client, _, _ = _client([_src()], ["text"], telemetry=recorder)
        await client.recognize_text(b"jpg", tenant_id="t1", meta={"batch": "b-42"})
        assert recorder.rows[0]["tenant_id"] == "t1"
        assert recorder.rows[0]["meta"] == '{"batch": "b-42"}'

    async def test_parse_layout_row_carries_dimensions(self):
        """两个公共方法都是入口: 只测一个会漏掉另一个的透传缺口。"""
        recorder = _MemoryRecorder()
        client, _, _ = _client([_src()], ["layout"], telemetry=recorder)
        await client.parse_layout(b"jpg", tenant_id="t2", meta={"batch": "b-43"})
        assert recorder.rows[0]["tenant_id"] == "t2"
        assert recorder.rows[0]["meta"] == '{"batch": "b-43"}'

    async def test_failed_attempt_row_also_carries_dimensions(self):
        """失败行同样需要归属: 某租户的请求没被服务,正是审计最需要的一行。"""
        recorder = _MemoryRecorder()
        client, _, _ = _client(
            [_src()],
            [TransientError("boom", status_code=500), "text"],
            telemetry=recorder,
        )
        await client.recognize_text(b"jpg", tenant_id="t1", meta={"batch": "b-42"})
        assert len(recorder.rows) == 2  # 失败尝试 + 成功尝试
        assert [r["tenant_id"] for r in recorder.rows] == ["t1", "t1"]
        assert [r["meta"] for r in recorder.rows] == ['{"batch": "b-42"}'] * 2

    @pytest.mark.parametrize("method", ["recognize_text", "parse_layout", "parse_document"])
    async def test_invalid_meta_rejected_before_any_telemetry(self, method):
        """校验必须早于遥测: 链路内的失败都被降级成 warning,放下游等于没有校验。"""
        recorder = _MemoryRecorder()
        client, _, _ = _client([_src()], ["text"], telemetry=recorder)
        with pytest.raises(ValueError, match="meta"):
            await getattr(client, method)(b"jpg", meta={"Bad Key": 1})
        assert recorder.rows == []
        assert client._transport.calls == []  # 连调用都没发出

    async def test_parse_document_row_carries_dimensions(self):
        """第三个入口: 只测前两个会漏掉它的透传缺口。"""
        recorder = _MemoryRecorder()
        client, _, _ = _client([_src()], ["document"], telemetry=recorder)
        await client.parse_document(b"%PDF-1.4", tenant_id="t3", meta={"batch": "b-44"})
        assert recorder.rows[0]["tenant_id"] == "t3"
        assert recorder.rows[0]["meta"] == '{"batch": "b-44"}'

    async def test_defaults_land_as_sentinels(self):
        recorder = _MemoryRecorder()
        client, _, _ = _client([_src()], ["text"], telemetry=recorder)
        await client.recognize_text(b"jpg")
        assert recorder.rows[0]["tenant_id"] == ""  # 空串哨兵,不是 None
        assert recorder.rows[0]["meta"] == "{}"


class _LegacyOcrTransport:
    """只实现旧 `OcrTransport` 三方法的下游 fake: 不满足 `OcrDocumentTransport`。"""

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

    async def recognize_text(self, *, image, source, call_id):
        self.calls.append("text")
        return _TEXT_OK

    async def parse_layout(self, *, image, source, call_id):
        self.calls.append("layout")
        return _LAYOUT_OK

    async def check_health(self, *, source):
        return True


class TestParseDocument:
    """#34 设计 §4.2/§6: 与 parse_layout 对称的整份文档入口,治理循环原样复用。"""

    async def test_result_fields(self):
        client, _, gate = _client([_src()], ["document"])
        r = await client.parse_document(b"%PDF-1.4")
        assert [e.text for e in r.elements] == ["SYNTH-DOC-BODY"]
        assert r.markdown == "# SYNTH-MD" and r.page_sizes == [(612.0, 792.0)]
        assert r.source_name == "m1"
        assert r.usage.prompt_tokens == 0 and r.usage.completion_tokens == 0
        assert r.latency_ms >= 0 and r.call_id and r.raw == {"success": True}
        assert r.call_stats is not None and r.call_stats.attempts == 1
        assert gate.successes == [("m1", True)]
        assert client._selector.outcomes == [("m1", True)]

    async def test_input_validation_names_the_document(self):
        client, _, _ = _client([_src()], [])
        with pytest.raises(TypeError, match="document 必须是 bytes"):
            await client.parse_document("not-bytes")
        with pytest.raises(ValueError, match="document 不能为空"):
            await client.parse_document(b"")

    async def test_legacy_entrances_keep_their_error_text(self):
        """私有形参改名为 payload 不得改变旧入口的报错文案(含 D9 的业务边界提示)。"""
        client, _, _ = _client([_src()], [])
        with pytest.raises(TypeError, match=r"image 必须是 bytes\(路径读取/批量拼帧留业务侧,D9\)"):
            await client.recognize_text("not-bytes")
        with pytest.raises(ValueError, match="image 不能为空"):
            await client.parse_layout(b"")

    async def test_transport_without_document_support_is_an_assembly_error(self):
        """装配错误当场报,且早于统计边界: 没有遥测行、没有任何 transport 调用。"""
        recorder = _MemoryRecorder()
        legacy = _LegacyOcrTransport()
        client, limiter, _ = _client([_src(rpm=100)], [], transport=legacy, telemetry=recorder)
        with pytest.raises(TypeError, match="OcrDocumentTransport"):
            await client.parse_document(b"%PDF-1.4")
        assert recorder.rows == [] and legacy.calls == []
        # 早于准入: 从未向限流后端申请过许可(否则 rpm 已被占用一次)
        assert (await limiter.source_stats("m1")).rpm_used == 0

    async def test_invalid_call_deadline_is_rejected_like_the_other_entrances(self):
        client, _, _ = _client([_src()], ["document"])
        with pytest.raises(ValueError, match="call_deadline_s"):
            await client.parse_document(b"%PDF-1.4", call_deadline_s=-1)

    async def test_transient_retries_then_succeeds(self):
        client, _, _ = _client(
            [_src()],
            [TransientError("boom", status_code=500), "document"],
            sleep=lambda _d: asyncio.sleep(0),
        )
        r = await client.parse_document(b"%PDF-1.4")
        assert r.markdown == "# SYNTH-MD"
        assert client._selector.outcomes == [("m1", False), ("m1", True)]

    async def test_source_dead_switches_to_the_second_source_immediately(self):
        s1, s2 = _src(name="m1"), _src(name="m2", base_url="http://10.77.0.20:7867")
        client, _, _ = _client(
            [s1, s2], [SourceDeadError("401", status_code=401), "document"], sleep=_no_sleep
        )
        r = await client.parse_document(b"%PDF-1.4")
        assert r.source_name == "m2"
        assert client._selector.outcomes == [("m1", False), ("m2", True)]

    async def test_result_invalid_counts_no_attempt_and_no_feed(self):
        client, _, gate = _client([_src()], [ResultInvalidError("bad package")])
        with pytest.raises(ResultInvalidError):
            await client.parse_document(b"%PDF-1.4")
        assert gate.successes == [("m1", False)]
        assert client._selector.outcomes == []

    async def test_deterministic_rejection_with_status_500_counts_as_a_live_service(self):
        """服务有响应(500 + 确定性坏输入标记)= 服务活着: 记成功不入窗口,不喂选源器。"""
        client, _, gate = _client(
            [_src()], [RequestRejectedError("Parsing failed: is no PDF", status_code=500)]
        )
        with pytest.raises(RequestRejectedError):
            await client.parse_document(b"%PDF-1.4")
        assert gate.successes == [("m1", False)]
        assert client._selector.outcomes == []

    async def test_local_rejection_does_not_touch_the_breaker(self):
        client, _, gate = _client([_src()], [RequestRejectedError("unrecognizable")])
        with pytest.raises(RequestRejectedError):
            await client.parse_document(b"%PDF-1.4")
        assert gate.successes == []

    async def test_rate_limited_responses_spend_retry_budget_but_not_breaker_failures(self):
        """429 是背压不是故障(熔断两通道均不计): 三次 429 耗尽重试后,熔断仍是 CLOSED。"""
        client, _, gate = _client(
            [_src()],
            [TransientError("busy", status_code=429)] * 3,
            sleep=lambda _d: asyncio.sleep(0),
        )
        with pytest.raises(AllSourcesExhausted):
            await client.parse_document(b"%PDF-1.4")
        entry = await gate._inner.try_enter("m1", "check")
        assert entry.allowed and not entry.is_probe

    async def test_read_timeouts_are_counted_as_source_failures(self):
        """对照: 文档路径的 httpx 超时仍是瞬时故障、计熔断(`timeout_s` 不是免责额度)。"""

        def timeout_exc():
            exc = TransientError("timed out", operation="parse")
            exc.__cause__ = httpx.ReadTimeout("read timed out")
            return exc

        client, _, gate = _client(
            [_src()], [timeout_exc() for _ in range(3)], sleep=lambda _d: asyncio.sleep(0)
        )
        with pytest.raises(AllSourcesExhausted) as ei:
            await client.parse_document(b"%PDF-1.4")
        assert ei.value.per_source_reasons == {"m1": "timeout"}
        entry = await gate._inner.try_enter("m1", "check")
        assert not entry.allowed  # 阈值 3 的熔断已开路

    async def test_cancel_releases_permit_and_probe(self):
        clock = FakeClock()
        inner = InMemoryGate(config=_BREAKER, now=clock)
        gate = RecordingGate(inner)
        transport = ScriptedOcrTransport(["hang"])
        client, limiter, _ = _client(
            [_src(max_concurrency=1)], [], now=clock, breaker=gate, transport=transport
        )
        entry = await inner.try_enter("m1", "setup")
        await inner.record_failure(entry, "source_dead", True)
        clock.advance(61.0)  # 越过冷却 → 半开探针
        task = asyncio.create_task(client.parse_document(b"%PDF-1.4"))
        await transport.entered.wait()
        task.cancel()
        with pytest.raises(asyncio.CancelledError):
            await task
        assert gate.probe_releases >= 1
        assert (await limiter.source_stats("m1")).inflight == 0


async def _no_sleep(_delay):
    await asyncio.sleep(0)


class TestNoLatencyJudgement:
    """长文档不会被判慢(设计 §7): 库内没有任何基于延迟的熔断/选源判定。

    时钟只在 transport 内前进(`ClockAdvancingOcrTransport`),门与 client 共用同一注入钟。
    probe_ttl_s 取极大值,避免把探针租约接管混进"慢"的断言。
    """

    _CFG = BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=1e12)

    async def test_slow_success_leaves_breaker_closed_with_no_failures(self):
        clock = FakeClock()
        gate = RecordingGate(InMemoryGate(config=self._CFG, now=clock))
        transport = ClockAdvancingOcrTransport([(100_000.0, "document")], clock)
        client, _, _ = _client(
            [_src(timeout_s=120.0)], [], now=clock, breaker=gate, transport=transport
        )
        r = await client.parse_document(b"%PDF-1.4")
        assert r.call_stats.attempts == 1 and r.call_stats.generation_ms >= 100_000_000
        assert gate._inner._gates["m1"].fails == 0
        assert gate.successes == [("m1", True)]

    async def test_slow_success_clears_pre_existing_failure_count(self):
        """前置状态不是初始 CLOSED: 已有 2 次失败(阈值 3),一次很慢的成功也只会清零。"""
        clock = FakeClock()
        inner = InMemoryGate(config=self._CFG, now=clock)
        for _ in range(2):
            entry = await inner.try_enter("m1", "setup")
            await inner.record_failure(entry, "network_error", False)
        assert inner._gates["m1"].fails == 2
        transport = ClockAdvancingOcrTransport([(100_000.0, "document")], clock)
        client, _, _ = _client(
            [_src()], [], now=clock, breaker=RecordingGate(inner), transport=transport
        )
        await client.parse_document(b"%PDF-1.4")
        assert inner._gates["m1"].fails == 0

    async def test_slow_success_on_the_half_open_probe_closes_the_breaker(self):
        clock = FakeClock()
        inner = InMemoryGate(config=self._CFG, now=clock)
        entry = await inner.try_enter("m1", "setup")
        await inner.record_failure(entry, "source_dead", True)  # OPEN
        clock.advance(61.0)  # → HALF_OPEN
        transport = ClockAdvancingOcrTransport([(100_000.0, "document")], clock)
        client, _, _ = _client(
            [_src()], [], now=clock, breaker=RecordingGate(inner), transport=transport
        )
        await client.parse_document(b"%PDF-1.4")
        after = await inner.try_enter("m1", "check")
        assert after.allowed and not after.is_probe  # 探针成功 → CLOSED,而不是被慢耗尽


class TestDocumentTelemetry:
    async def test_success_row(self):
        recorder = _MemoryRecorder()
        client, _, _ = _client([_src()], ["document"], telemetry=recorder)
        await client.parse_document(b"%PDF-RAW-BYTES")
        (row,) = recorder.rows
        assert row["operation"] == "parse_document"
        assert "<ocr:document image_bytes=14>" in row["messages"]
        assert "RAW-BYTES" not in row["messages"]  # 文档字节绝不入库
        assert row["response"] == "<document elements=1 pages=1 md_chars=10>"
        assert "SYNTH-MD" not in row["response"] and "SYNTH-DOC-BODY" not in row["response"]
        assert row["reasoning_effort"] is None  # `reasoning_applies=False` 的落库形态
        assert row["error"] is None and row["prompt_tokens"] == 0

    async def test_summary_distinguishes_the_three_result_kinds(self):
        """新结果类型必须在 `_summarize` 里显式区分,否则会被静默套成 layout 摘要而漏记 md_chars。"""
        assert OcrClient._summarize(_DOC_OK) == "<document elements=1 pages=1 md_chars=10>"
        assert OcrClient._summarize(_LAYOUT_OK) == "<elements n=1 pages=1>"
        assert OcrClient._summarize(_TEXT_OK) == "LINE-1"

    def test_summary_keeps_structural_dispatch_for_duck_typed_results(self):
        """旧实现按结构分派(`hasattr(result, "text")`)。下游测试替身可能返回非库类型的鸭子对象,
        改成 isinstance 会让它们在开启遥测后以原生 AttributeError 漏出——无登记的回归。"""
        assert OcrClient._summarize(SimpleNamespace(text="hello", raw={})) == "hello"
        assert (
            OcrClient._summarize(SimpleNamespace(elements=[1, 2], page_sizes=[(1.0, 1.0)]))
            == "<elements n=2 pages=1>"
        )
        assert (
            OcrClient._summarize(
                SimpleNamespace(elements=[1], page_sizes=[(1.0, 1.0)], markdown="abc")
            )
            == "<document elements=1 pages=1 md_chars=3>"
        )

    async def test_failed_attempt_row_carries_the_class_prefix(self):
        recorder = _MemoryRecorder()
        client, _, _ = _client([_src()], [ResultInvalidError("bad package")], telemetry=recorder)
        with pytest.raises(ResultInvalidError):
            await client.parse_document(b"%PDF-1.4")
        assert any(r["error"].startswith("ResultInvalidError:") for r in recorder.rows)
        assert {r["operation"] for r in recorder.rows} == {"parse_document"}


def _real_transport(handler):
    return MonkeyOcrTransport(
        client_factory=lambda src: httpx.AsyncClient(
            base_url=src.base_url, transport=httpx.MockTransport(handler)
        )
    )


def _doc_handler(package: bytes | None = None, *, download=None):
    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/parse":
            return httpx.Response(
                200,
                json={"success": True, "message": "ok", "download_url": "/static/doc.zip"},
            )
        if download is not None:
            return download()
        return httpx.Response(200, content=package)

    return handler


class TestDocumentThroughTheRealTransport:
    """真实 `MonkeyOcrTransport` + `MockTransport` + 真实 `OcrClient`: 证明异常/期限贯穿整条链。

    MockTransport 不执行 httpcore 的逐次读超时,所以本组只证明**治理期限与异常分类**,
    不证明 read timeout 的真实行为(后者留给活体用例)。
    """

    async def test_happy_path_end_to_end(self):
        package = fx._package([fx._page([fx._text_block("SYNTH-E2E")])])
        client, _, _ = _client([_src()], [], transport=_real_transport(_doc_handler(package)))
        r = await client.parse_document(b"%PDF-1.4 synthetic")
        assert [e.text for e in r.elements] == ["SYNTH-E2E"]
        assert r.markdown == "# synth\n"

    @pytest.mark.parametrize("package_kind", ["huge-int", "encrypted"])
    async def test_parser_exceptions_become_result_invalid_with_full_bookkeeping(
        self, package_kind
    ):
        """真实解析器抛出的异常(OverflowError / RuntimeError)经真实客户端: 终态遥测行已写、
        半开探针已归还、熔断窗口不变——不能只用桩直接抛预制的 `ResultInvalidError`。"""
        if package_kind == "huge-int":
            block = fx._text_block()
            block["bbox"] = [0, 0, 10**400, 1]
            package = fx._package([fx._page([block])])
        else:
            package = fx._encrypted_package()
        clock = FakeClock()
        inner = InMemoryGate(
            config=BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0), now=clock
        )
        entry = await inner.try_enter("m1", "setup")
        await inner.record_failure(entry, "source_dead", True)
        clock.advance(61.0)  # 半开: 本次调用持有探针
        recorder = _MemoryRecorder()
        client, _, gate = _client(
            [_src()],
            [],
            now=clock,
            breaker=RecordingGate(inner),
            telemetry=recorder,
            transport=_real_transport(_doc_handler(package)),
        )
        with pytest.raises(ResultInvalidError):
            await client.parse_document(b"%PDF-1.4 synthetic")
        assert any(r["error"].startswith("ResultInvalidError:") for r in recorder.rows)
        after = await inner.try_enter("m1", "check")
        assert after.allowed and not after.is_probe  # 探针已按"坏结果≠坏服务"结算,没有悬挂
        assert inner._gates["m1"].fails == 0

    @staticmethod
    def _slow_download_handler():
        async def slow_body():
            for _ in range(200):
                await asyncio.sleep(0.01)
                yield b"x" * 8

        return _doc_handler(download=lambda: httpx.Response(200, content=slow_body()))

    async def test_per_call_deadline_overrides_the_assembly_value(self):
        client, _, _ = _client(
            [_src(timeout_s=120.0)], [], transport=_real_transport(self._slow_download_handler())
        )  # 装配值未设期限
        with pytest.raises(CallDeadlineExceeded):
            await client.parse_document(b"%PDF-1.4 synthetic", call_deadline_s=0.1)

    async def test_deadline_while_holding_the_half_open_probe_returns_it_and_writes_a_terminal_row(
        self,
    ):
        """期限到期取消在飞请求时: 探针必须归还,且终态遥测行必须写出(遥测必录)。"""
        clock = FakeClock()
        inner = InMemoryGate(
            config=BreakerConfig(fail_threshold=3, cooldown_s=60.0, probe_ttl_s=120.0), now=clock
        )
        entry = await inner.try_enter("m1", "setup")
        await inner.record_failure(entry, "source_dead", True)
        clock.advance(61.0)  # 半开: 本次调用持有探针
        gate = RecordingGate(inner)
        recorder = _MemoryRecorder()
        client, _, _ = _client(
            [_src(timeout_s=120.0)],
            [],
            now=clock,
            breaker=gate,
            telemetry=recorder,
            transport=_real_transport(self._slow_download_handler()),
            call_deadline_s=0.1,
        )
        with pytest.raises(CallDeadlineExceeded):
            await client.parse_document(b"%PDF-1.4 synthetic")
        assert gate.probe_releases >= 1
        again = await inner.try_enter("m1", "check")
        assert again.allowed and again.is_probe  # 探针没有悬挂,下一位调用方可以再探
        terminals = [r for r in recorder.rows if r["event_kind"] == "terminal_failure"]
        assert len(terminals) == 1 and terminals[0]["operation"] == "parse_document"

    async def test_sustained_slow_download_hits_the_call_deadline_not_a_transient_error(self):
        """每个分块都在 read 超时内、总时长超过 call_deadline_s: 是期限,不是瞬时故障。"""
        closed = []
        produced = []

        async def slow_body():
            try:
                for _ in range(200):
                    await asyncio.sleep(0.01)
                    produced.append(1)
                    yield b"x" * 8
            finally:
                closed.append(True)

        handler = _doc_handler(download=lambda: httpx.Response(200, content=slow_body()))
        client, limiter, gate = _client(
            [_src(max_concurrency=1, timeout_s=120.0)],
            [],
            transport=_real_transport(handler),
            call_deadline_s=0.1,
        )
        with pytest.raises(CallDeadlineExceeded):
            await client.parse_document(b"%PDF-1.4 synthetic")
        assert len(produced) >= 2  # 确实进入了多块下载,而不是一开始就超时
        assert closed == [True]  # 流已关闭
        assert (await limiter.source_stats("m1")).inflight == 0  # permit 释放
        assert gate.successes == []  # 期限不是服务故障也不是成功
        entry = await gate._inner.try_enter("m1", "check")
        assert entry.allowed and not entry.is_probe  # 熔断窗口不变


class TestAssembly:
    _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",
    }

    async def test_from_env_assembles(self):
        client = OcrClient.from_env("OCR", env=dict(self._ENV))
        try:
            assert client._scope == "ocr"
        finally:
            await client.aclose()

    async def test_non_monkey_provider_rejected(self):
        env = {k.replace("MONKEY", "GLM"): v for k, v in self._ENV.items()}
        with pytest.raises(ValueError, match="monkey"):
            OcrClient.from_env("OCR", env=env)

    async def test_aclose_idempotent(self):
        client = OcrClient.from_env("OCR", env=dict(self._ENV))
        await client.aclose()
        await client.aclose()


class TestReasonlessTelemetryContract:
    """text/layout/document 三个入口分别验证错误行不受源级推理配置污染。

    独立验证发现文档入口原先不在矩阵里: 让 OCR 一律按 `reasoning_applies=True` 记账,
    15 个旧用例转红,但没有一个是 `parse_document` 的。"""

    @pytest.mark.parametrize(
        "method,action",
        [("recognize_text", "text"), ("parse_layout", "layout"), ("parse_document", "document")],
    )
    @pytest.mark.parametrize("config", [{"enable_thinking": True}, {"reasoning_effort": "high"}])
    @pytest.mark.parametrize("backend", ["memory", "sqlite"])
    async def test_failed_then_successful_attempts_have_null_effort(
        self, method, action, config, backend, tmp_path
    ):
        import sqlite3

        from polygateway.telemetry.sqlite import SQLiteRecorder

        path = tmp_path / "ocr.sqlite"
        recorder = (
            _MemoryRecorder() if backend == "memory" else SQLiteRecorder(path, auto_migrate=True)
        )
        client, _, _ = _client(
            [_src(**config)], [TransientError("retry"), action], telemetry=recorder
        )
        try:
            await getattr(client, method)(b"image", session_id="run", parent_call_id=method)
            if backend == "memory":
                rows = [(r["error"], r["reasoning_effort"]) for r in recorder.rows]
            else:
                with sqlite3.connect(path) as db:
                    rows = db.execute("SELECT error, reasoning_effort FROM llm_calls").fetchall()
            assert len(rows) == 2
            assert sum(bool(error) for error, _ in rows) == 1
            assert [tier for _, tier in rows] == [None, None]
        finally:
            await client.aclose()
            if backend == "sqlite":
                recorder.close()

    @pytest.mark.parametrize("method", ["recognize_text", "parse_layout", "parse_document"])
    @pytest.mark.parametrize("exhausted", [False, True])
    async def test_failed_attempts_still_have_null_effort(self, method, exhausted):
        script = (
            [TransientError("retry")] * 3 if exhausted else [RequestRejectedError("bad request")]
        )
        recorder = _MemoryRecorder()
        client, _, _ = _client([_src(enable_thinking=True)], script, telemetry=recorder)
        with pytest.raises(AllSourcesExhausted if exhausted else RequestRejectedError):
            await getattr(client, method)(b"image")
        # 1.3.5: 逐次 attempt 行之外,本次逻辑调用另有**一条**终态行
        attempts = [r for r in recorder.rows if r["event_kind"] == "attempt"]
        terminals = [r for r in recorder.rows if r["event_kind"] == "terminal_failure"]
        assert len(attempts) == len(script)
        assert len(terminals) == 1
        # 终态行的 operation 是**公开方法**名,与尝试行一致
        assert terminals[0]["operation"] == method
        assert all(r["error"] and r["reasoning_effort"] is None for r in recorder.rows)


class TestOcrLogicalCallStats:
    """OCR 两个公开方法各自拥有一次逻辑调用(1.3.5 设计 §3/§3.5)。"""

    async def test_text_success_counts_one_attempt(self):
        client, _, _ = _client([_src()], ["text"])
        r = await client.recognize_text(b"jpg")
        assert r.call_stats is not None and r.call_stats.attempts == 1

    async def test_layout_two_http_calls_count_as_one_attempt(self):
        """POST + ZIP GET 在同一次 transport 调用内,计 **1** 次尝试而非 2。

        `attempts` 的语义是"调用 transport 端口的次数",不是 HTTP 请求条数。
        """
        client, _, _ = _client([_src()], ["layout"])
        r = await client.parse_layout(b"jpg")
        assert r.call_stats is not None and r.call_stats.attempts == 1

    async def test_retry_counts_every_attempt(self):
        client, _, _ = _client([_src(), _src(name="m2")], [TransientError("t1"), "text"])
        r = await client.recognize_text(b"jpg")
        assert r.call_stats.attempts == 2

    async def test_input_validation_stays_outside_the_stats_boundary(self):
        """`image` 类型/空校验先于上下文创建(M1 例外),保持原异常行为。"""
        client, _, _ = _client([_src()], [])
        with pytest.raises(TypeError):
            await client.recognize_text("not-bytes")
        with pytest.raises(ValueError):
            await client.recognize_text(b"")


class TestOcrCallDeadline:
    """OCR 两个公开入口的期限与 per-call 校验(计划 §3.4/§5 批次 D/E)。"""

    async def test_expiry_on_a_hanging_transport(self):
        client, limiter, _ = _client([_src()], ["hang"])
        with pytest.raises(CallDeadlineExceeded) as exc:
            await client.recognize_text(b"jpg", call_deadline_s=0.05)
        assert exc.value.scope == "ocr" and exc.value.deadline_s == 0.05
        # 清理照常在 finally 完成: 在途计数必须归零(OCR 无 token,结算恒 0)
        stats = await limiter.source_stats("m1")
        assert stats.inflight == 0 and stats.tpm_used == 0

    @pytest.mark.parametrize("method", ["recognize_text", "parse_layout"])
    async def test_illegal_per_call_value_names_the_entry_it_came_from(self, method):
        """两个入口各自报自己的名字: 多入口部署里才定位得到是哪次调用传错了。"""
        transport = ScriptedOcrTransport([])
        client, _, _ = _client([_src()], [], transport=transport)
        with pytest.raises(ValueError, match=rf"{method}\(call_deadline_s"):
            await getattr(client, method)(b"jpg", call_deadline_s=float("inf"))
        assert transport.calls == []

    def test_illegal_constructor_value_is_rejected_at_assembly(self):
        with pytest.raises(ValueError, match=r"OcrClient\(call_deadline_s"):
            _client([_src()], [], call_deadline_s=0)