From dfbb2f8f107d26cc09a961e511642ddafa9b2690 Mon Sep 17 00:00:00 2001
From: iomgaa <iomgaaycz@gmail.com>
Date: Tue, 21 Jul 2026 22:36:14 -0400
Subject: [PATCH] feat: add MonkeyOCR dual-endpoint transport

---
 src/polygateway/transports/monkey_ocr.py | 283 +++++++++++++++++++
 tests/unit/test_monkey_ocr.py            | 345 +++++++++++++++++++++++
 2 files changed, 628 insertions(+)
 create mode 100644 src/polygateway/transports/monkey_ocr.py
 create mode 100644 tests/unit/test_monkey_ocr.py

diff --git a/src/polygateway/transports/monkey_ocr.py b/src/polygateway/transports/monkey_ocr.py
new file mode 100644
index 0000000..8ebbdba
--- /dev/null
+++ b/src/polygateway/transports/monkey_ocr.py
@@ -0,0 +1,283 @@
+"""MonkeyOCR transport: 双端点协议细节(M3 设计 §4;不含任何治理)。
+
+蓝本: CHS `invokers.py:427-552`(两段协议 + 数值防御,逐段保真)与
+VT `adapters/ocr.py`(/ocr/text 请求形态)。已在设计声明的偏离: 全元素
+提取、`success!=true` 附 status_code=200、429 细分放弃、健康判定收紧 2xx、
+`content` 显式校验。服务无鉴权——不带 Authorization 头,api_key 是占位。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import math
+import zipfile
+from contextlib import contextmanager
+from io import BytesIO
+from typing import TYPE_CHECKING, Any
+
+import httpx
+
+from polygateway.errors import (
+    RequestRejectedError,
+    ResultInvalidError,
+    SourceDeadError,
+    TransientError,
+)
+from polygateway.types import (
+    OcrLayoutElement,
+    OcrLayoutTransportResult,
+    OcrTextTransportResult,
+)
+
+if TYPE_CHECKING:
+    from collections.abc import Callable, Iterator
+
+    from polygateway.types import SourceConfig
+
+_HEALTH_TIMEOUT_S = 5.0  # 探测常量(同 jitter 系数先例,不进配置)
+
+
+def _default_client_factory(source: SourceConfig) -> httpx.AsyncClient:
+    return httpx.AsyncClient(
+        base_url=source.base_url,
+        timeout=httpx.Timeout(source.timeout_s),
+        trust_env=source.trust_env,  # VT LAN 直连绕代理场景配 false(R9)
+    )
+
+
+@contextmanager
+def _translate_http_errors(source_name: str, operation: str) -> Iterator[None]:
+    """httpx 异常 → 错误四分类(CHS `_translate_http_errors` 语义主体保真)。
+
+    429 细分(insufficient_quota/Retry-After)有意放弃——MonkeyOCR 无鉴权
+    无计费(设计 §10.2);`from exc` 保留 __cause__ 供 RetryMW 归类 timeout。
+    """
+    try:
+        yield
+    except httpx.TimeoutException as exc:
+        raise TransientError(
+            f"{source_name} OCR {operation} 超时: {exc}",
+            source_name=source_name,
+            operation=operation,
+        ) from exc
+    except httpx.HTTPStatusError as exc:
+        raise _classify_status(exc, source_name, operation) from exc
+    except httpx.HTTPError as exc:
+        raise TransientError(
+            f"{source_name} OCR {operation} 网络错误: {exc}",
+            source_name=source_name,
+            operation=operation,
+        ) from exc
+
+
+def _classify_status(
+    exc: httpx.HTTPStatusError, source_name: str, operation: str
+) -> TransientError | SourceDeadError | RequestRejectedError:
+    status = exc.response.status_code
+    ctx: dict[str, Any] = {"source_name": source_name, "status_code": status, "operation": operation}
+    message = f"{source_name} OCR {operation} HTTP {status}"
+    if status >= 500 or status == 429:
+        return TransientError(message, **ctx)
+    if status in (401, 403):
+        return SourceDeadError(message, **ctx)
+    return RequestRejectedError(message, **ctx)
+
+
+def _finite_number(value: object) -> float:
+    """JSON 数值 → 有限浮点;布尔与非有限数拒绝(CHS invokers.py:427 逐字)。"""
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise ValueError("字段不是数值")
+    number = float(value)
+    if not math.isfinite(number):
+        raise ValueError("字段不是有限数值")
+    return number
+
+
+def _parse_page_size(page: dict) -> tuple[float, float]:
+    page_size = page.get("page_size")
+    if not isinstance(page_size, list) or len(page_size) != 2:
+        raise ValueError("page_size 无效")
+    width, height = (_finite_number(item) for item in page_size)
+    if width <= 0 or height <= 0:
+        raise ValueError("page_size 必须为正数")
+    return (width, height)
+
+
+def _parse_bbox(bbox: object) -> tuple[float, float, float, float]:
+    if not isinstance(bbox, list) or len(bbox) != 4:
+        raise ValueError("bbox 无效")
+    x1, y1, x2, y2 = (_finite_number(item) for item in bbox)
+    if x2 <= x1 or y2 <= y1:
+        raise ValueError("bbox 边界顺序无效")
+    # 库返回 float 原生 bbox;整数化退化校验专为 CHS shim 的 int() 裁剪路径
+    # 兜底(亚像素宽的框裁剪即空图)——不是死代码,勿删(设计 §4)
+    if int(x2) <= int(x1) or int(y2) <= int(y1):
+        raise ValueError("bbox 整数化后退化")
+    return (x1, y1, x2, y2)
+
+
+def _parse_block(block: object, page_index: int) -> OcrLayoutElement:
+    if not isinstance(block, dict):
+        raise ValueError("块结构无效")
+    block_type = block.get("type")
+    if not isinstance(block_type, str) or not block_type.strip():
+        raise ValueError("块缺少 type")
+    return OcrLayoutElement(
+        type=block_type, bbox=_parse_bbox(block.get("bbox")), page_index=page_index
+    )
+
+
+def _parse_pages(payload: object) -> tuple[list[OcrLayoutElement], list[tuple[float, float]]]:
+    if not isinstance(payload, dict) or not isinstance(payload.get("pdf_info"), list):
+        raise ValueError("_middle.json 缺少 pdf_info 列表")
+    elements: list[OcrLayoutElement] = []
+    page_sizes: list[tuple[float, float]] = []
+    for fallback_index, page in enumerate(payload["pdf_info"]):
+        if not isinstance(page, dict):
+            raise ValueError("页面结构无效")
+        page_sizes.append(_parse_page_size(page))
+        raw_index = page.get("page_idx", fallback_index)
+        page_index = raw_index if isinstance(raw_index, int) else fallback_index
+        blocks = page.get("para_blocks", [])
+        if not isinstance(blocks, list):
+            raise ValueError("para_blocks 无效")
+        elements.extend(_parse_block(block, page_index) for block in blocks)
+    return elements, page_sizes
+
+
+def _parse_middle_json(zip_bytes: bytes) -> tuple[list[OcrLayoutElement], list[tuple[float, float]]]:
+    """ZIP → (elements, page_sizes);一切形态异常归 ResultInvalid(坏图≠坏服务)。
+
+    数值防御全量下沉自 CHS `_parse_table_result`(invokers.py:437-479),
+    提取面从"首表"泛化为 para_blocks 全元素(设计 §1.2 取证证明无损)。
+    """
+    try:
+        with zipfile.ZipFile(BytesIO(zip_bytes)) as archive:
+            middle_files = [n for n in archive.namelist() if n.endswith("_middle.json")]
+            if not middle_files:
+                raise ValueError("结果包缺少 _middle.json")
+            with archive.open(middle_files[0]) as stream:
+                payload = json.load(stream)
+        return _parse_pages(payload)
+    except (ValueError, TypeError, KeyError, json.JSONDecodeError, zipfile.BadZipFile) as exc:
+        raise ResultInvalidError(f"OCR 结果包内容无效: {exc}") from exc
+
+
+class MonkeyOcrTransport:
+    """MonkeyOCR transport: 每源一个预配 httpx client,懒创建,aclose 统一释放。"""
+
+    def __init__(
+        self,
+        *,
+        client_factory: Callable[[SourceConfig], httpx.AsyncClient] | None = None,
+    ) -> None:
+        self._client_factory = client_factory or _default_client_factory
+        self._clients: dict[str, httpx.AsyncClient] = {}
+
+    def _client_for(self, source: SourceConfig) -> httpx.AsyncClient:
+        client = self._clients.get(source.name)
+        if client is None:
+            client = self._client_factory(source)
+            self._clients[source.name] = client
+        return client
+
+    @staticmethod
+    def _files(image: bytes) -> dict[str, tuple[str, bytes, str]]:
+        # 固定文件名/类型(CHS invokers.py:496 原样;服务按内容处理,35 样本实测)
+        return {"file": ("image.jpg", image, "image/jpeg")}
+
+    def _response_json(self, resp: httpx.Response, source: SourceConfig, operation: str) -> dict:
+        """响应体 JSON 化;坏 JSON/非 dict → Transient(设计 §4: 服务侧瞬时异常)。"""
+        try:
+            data = resp.json()
+        except (json.JSONDecodeError, ValueError) as exc:
+            raise TransientError(
+                f"{source.name} OCR {operation} 响应非合法 JSON",
+                source_name=source.name,
+                operation=operation,
+            ) from exc
+        if not isinstance(data, dict):
+            raise TransientError(
+                f"{source.name} OCR {operation} JSON 结构无效",
+                source_name=source.name,
+                operation=operation,
+            )
+        return data
+
+    async def recognize_text(
+        self, *, image: bytes, source: SourceConfig, call_id: str
+    ) -> OcrTextTransportResult:
+        """POST /ocr/text → 多行纯文本;content 缺失/非 str → ResultInvalid。"""
+        del call_id  # 协议无关,治理侧遥测使用
+        client = self._client_for(source)
+        with _translate_http_errors(source.name, "ocr_text"):
+            resp = await client.post("/ocr/text", files=self._files(image))
+            resp.raise_for_status()
+        data = self._response_json(resp, source, "ocr_text")
+        content = data.get("content")
+        if not isinstance(content, str):
+            # VT 的 .get("content","") 静默兜底有意替换为显式校验(P5)
+            raise ResultInvalidError(
+                f"{source.name} /ocr/text 响应 content 缺失或非字符串",
+                source_name=source.name,
+                operation="ocr_text",
+            )
+        return OcrTextTransportResult(
+            text=content, raw={k: v for k, v in data.items() if k != "content"}
+        )
+
+    async def parse_layout(
+        self, *, image: bytes, source: SourceConfig, call_id: str
+    ) -> OcrLayoutTransportResult:
+        """两段协议: POST /parse → GET download_url → ZIP → 全元素提取。"""
+        del call_id
+        client = self._client_for(source)
+        with _translate_http_errors(source.name, "parse"):
+            resp = await client.post("/parse", files=self._files(image))
+            resp.raise_for_status()
+        data = self._response_json(resp, source, "parse")
+        if data.get("success") is not True:
+            # 设计 §10.2 有意修复: 200 响应确证服务活着,附 status_code
+            # 让治理侧按"响应即健康"记熔断成功(CHS 原样会误走本地拒绝分支)
+            raise RequestRejectedError(
+                f"{source.name} OCR 解析失败: {data.get('message')}",
+                source_name=source.name,
+                status_code=resp.status_code,
+                operation="parse",
+            )
+        download_url = data.get("download_url")
+        if not isinstance(download_url, str) or not download_url.strip():
+            raise TransientError(
+                f"{source.name} /parse 响应缺少 download_url",
+                source_name=source.name,
+                operation="parse",
+            )
+        with _translate_http_errors(source.name, "download_result"):
+            # 实测 download_url 为相对路径(/static/*.zip),相对 base_url 解析;
+            # httpx 对绝对 URL 会忽略 base_url,两种形态天然兼容
+            result_resp = await client.get(download_url)
+            result_resp.raise_for_status()
+        elements, page_sizes = _parse_middle_json(result_resp.content)
+        return OcrLayoutTransportResult(
+            elements=elements,
+            page_sizes=page_sizes,
+            raw={"success": True, "download_url": download_url},
+        )
+
+    async def check_health(self, *, source: SourceConfig) -> bool:
+        """GET /health 探测: 2xx → True,失败 → False,取消穿透(R10)。"""
+        client = self._client_for(source)
+        try:
+            resp = await client.get("/health", timeout=_HEALTH_TIMEOUT_S)
+        except asyncio.CancelledError:
+            raise
+        except Exception:  # noqa: BLE001 — 探测不是调用,失败一律视为不健康
+            return False
+        return resp.is_success
+
+    async def aclose(self) -> None:
+        """幂等释放全部 per-source client。"""
+        clients, self._clients = self._clients, {}
+        for client in clients.values():
+            await client.aclose()
diff --git a/tests/unit/test_monkey_ocr.py b/tests/unit/test_monkey_ocr.py
new file mode 100644
index 0000000..56f65ba
--- /dev/null
+++ b/tests/unit/test_monkey_ocr.py
@@ -0,0 +1,345 @@
+"""MonkeyOcrTransport 测试(M3 设计 §4): 双端点协议、两段解析、数值防御、错误翻译。
+
+fixtures 按 2026-07-21 真实服务取证响应**脱敏二次构造**(设计 §11):
+保留结构骨架/数值/元素类型,OCR 识别文本一律替换为合成占位(零业务假设 +
+P5 医疗数据不入库)。
+"""
+
+import asyncio
+import io
+import json
+import zipfile
+
+import httpx
+import pytest
+
+from polygateway.errors import (
+    RequestRejectedError,
+    ResultInvalidError,
+    SourceDeadError,
+    TransientError,
+)
+from polygateway.transports.monkey_ocr import MonkeyOcrTransport, _parse_middle_json
+from polygateway.types import SourceConfig
+
+
+def _source(**overrides):
+    base = {
+        "name": "monkey_1",
+        "provider": "monkey",
+        "base_url": "http://10.77.0.20:7866",
+        "api_key": "none",  # MonkeyOCR 无鉴权,占位惯例(CHS .env 同款)
+        "model": "monkey-ocr",
+        "timeout_s": 120.0,
+    }
+    base.update(overrides)
+    return SourceConfig(**base)
+
+
+def _text_body(content="LINE-1\nLINE-2"):
+    # 真实形态: {"success":true,"task_type":"text","content":...,"message":...}
+    return {
+        "success": True,
+        "task_type": "text",
+        "content": content,
+        "message": "Text extraction completed successfully",
+    }
+
+
+def _parse_body(download_url="/static/sample_parsed_1.zip"):
+    # 真实形态含 output_dir/files;解析只消费 success 与 download_url
+    return {
+        "success": True,
+        "message": "image parsing (standard) completed successfully",
+        "output_dir": "/app/tmp/x",
+        "files": ["u_middle.json"],
+        "download_url": download_url,
+    }
+
+
+def _page(para_blocks, page_size=(759.0, 540.0), page_idx=0):
+    return {
+        "page_idx": page_idx,
+        "page_size": list(page_size),
+        "para_blocks": para_blocks,
+        "tables": [b for b in para_blocks if b.get("type") == "table"],
+    }
+
+
+def _block(type_, bbox):
+    body = {"type": type_, "bbox": list(bbox), "index": 0}
+    if type_ == "text":
+        body["lines"] = [{"spans": [{"content": "LINE-SYNTH"}]}]
+    return body
+
+
+_DEFAULT_PAGES = [
+    _page(
+        [
+            _block("table", (41, 48, 218, 282)),
+            _block("image", (240, 30, 700, 500)),
+            _block("text", (50, 300, 200, 340)),
+        ]
+    )
+]
+
+
+def _middle_bytes(pages=_DEFAULT_PAGES, top=None):
+    payload = {"pdf_info": pages, "_parse_type": "ocr", "_version_name": "0.0.1"}
+    if top is not None:
+        payload = top
+    return json.dumps(payload).encode()
+
+
+def _zip_bytes(middle=None, member="sample_middle.json"):
+    buf = io.BytesIO()
+    with zipfile.ZipFile(buf, "w") as zf:
+        zf.writestr(member, middle if middle is not None else _middle_bytes())
+        zf.writestr("sample.md", "# synth")
+    return buf.getvalue()
+
+
+def _transport_for(handler):
+    mock = httpx.MockTransport(handler)
+    return MonkeyOcrTransport(
+        client_factory=lambda src: httpx.AsyncClient(base_url=src.base_url, transport=mock)
+    )
+
+
+def _routes(text_resp=None, parse_resp=None, zip_resp=None, health_resp=None):
+    """按路径分发的 mock handler;None 项返回 500 便于暴露误路由。"""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        path = request.url.path
+        if path == "/ocr/text":
+            return text_resp or httpx.Response(500)
+        if path == "/parse":
+            return parse_resp or httpx.Response(500)
+        if path.startswith("/static/"):
+            return zip_resp or httpx.Response(500)
+        if path == "/health":
+            return health_resp or httpx.Response(500)
+        return httpx.Response(500)
+
+    return handler
+
+
+class TestRecognizeText:
+    async def test_normal(self):
+        t = _transport_for(_routes(text_resp=httpx.Response(200, json=_text_body())))
+        r = await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
+        assert r.text == "LINE-1\nLINE-2"
+        assert r.raw["task_type"] == "text"
+
+    async def test_empty_content_legal(self):
+        t = _transport_for(_routes(text_resp=httpx.Response(200, json=_text_body(""))))
+        r = await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
+        assert r.text == ""
+
+    async def test_content_missing_result_invalid(self):
+        body = _text_body()
+        del body["content"]
+        t = _transport_for(_routes(text_resp=httpx.Response(200, json=body)))
+        with pytest.raises(ResultInvalidError):
+            await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
+
+    async def test_content_non_str_result_invalid(self):
+        t = _transport_for(_routes(text_resp=httpx.Response(200, json=_text_body(123))))
+        with pytest.raises(ResultInvalidError):
+            await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
+
+    async def test_multipart_shape(self):
+        seen = {}
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            seen["content_type"] = request.headers.get("content-type", "")
+            seen["body"] = request.read()
+            return httpx.Response(200, json=_text_body())
+
+        t = _transport_for(handler)
+        await t.recognize_text(image=b"IMAGE-BYTES", source=_source(), call_id="c1")
+        assert seen["content_type"].startswith("multipart/form-data")
+        assert b"IMAGE-BYTES" in seen["body"]
+        assert b'filename="image.jpg"' in seen["body"]  # CHS invokers.py:496 固定名
+
+
+class TestParseLayout:
+    async def test_full_chain_relative_url(self):
+        t = _transport_for(
+            _routes(
+                parse_resp=httpx.Response(200, json=_parse_body()),
+                zip_resp=httpx.Response(200, content=_zip_bytes()),
+            )
+        )
+        r = await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
+        types = [e.type for e in r.elements]
+        assert types == ["table", "image", "text"]
+        assert r.elements[0].bbox == (41.0, 48.0, 218.0, 282.0)
+        assert r.elements[0].page_index == 0
+        assert r.page_sizes == [(759.0, 540.0)]
+
+    async def test_absolute_download_url(self):
+        url = "http://10.77.0.20:7866/static/abs.zip"
+        t = _transport_for(
+            _routes(
+                parse_resp=httpx.Response(200, json=_parse_body(url)),
+                zip_resp=httpx.Response(200, content=_zip_bytes()),
+            )
+        )
+        r = await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
+        assert len(r.elements) == 3
+
+    async def test_success_false_rejected_with_status_200(self):
+        body = {"success": False, "message": "unsupported"}
+        t = _transport_for(_routes(parse_resp=httpx.Response(200, json=body)))
+        with pytest.raises(RequestRejectedError) as ei:
+            await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
+        # 设计 §10.2 有意修复: 200 响应确证服务活着,熔断按"响应即健康"记成功
+        assert ei.value.status_code == 200
+
+    async def test_download_url_missing_transient(self):
+        body = _parse_body()
+        del body["download_url"]
+        t = _transport_for(_routes(parse_resp=httpx.Response(200, json=body)))
+        with pytest.raises(TransientError):
+            await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
+
+    async def test_non_json_response_transient(self):
+        t = _transport_for(_routes(parse_resp=httpx.Response(200, content=b"<html>")))
+        with pytest.raises(TransientError):
+            await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
+
+    async def test_no_table_page_legal(self):
+        pages = [_page([_block("text", (1, 2, 30, 40))])]
+        t = _transport_for(
+            _routes(
+                parse_resp=httpx.Response(200, json=_parse_body()),
+                zip_resp=httpx.Response(200, content=_zip_bytes(_middle_bytes(pages))),
+            )
+        )
+        r = await t.parse_layout(image=b"jpg", source=_source(), call_id="c1")
+        assert [e.type for e in r.elements] == ["text"]  # 无 table = 合法,不抛
+
+
+class TestMiddleJsonDefense:
+    """数值防御纯函数(CHS invokers.py:427-479 全量下沉)。"""
+
+    def _expect_invalid(self, zip_bytes):
+        with pytest.raises(ResultInvalidError):
+            _parse_middle_json(zip_bytes)
+
+    def test_bad_zip(self):
+        self._expect_invalid(b"not-a-zip")
+
+    def test_missing_middle_member(self):
+        self._expect_invalid(_zip_bytes(member="other.json"))
+
+    def test_pdf_info_not_list(self):
+        self._expect_invalid(_zip_bytes(_middle_bytes(top={"pdf_info": {}})))
+
+    def test_middle_not_json(self):
+        self._expect_invalid(_zip_bytes(b"{broken"))
+
+    @pytest.mark.parametrize(
+        "page_size", [[-1, 540], [float("nan"), 540], [True, 540], [759], "bad"]
+    )
+    def test_page_size_invalid(self, page_size):
+        page = _page([_block("table", (1, 2, 3, 4))])
+        page["page_size"] = page_size
+        self._expect_invalid(_zip_bytes(_middle_bytes([page])))
+
+    def test_bbox_order_invalid(self):
+        self._expect_invalid(_zip_bytes(_middle_bytes([_page([_block("table", (218, 48, 41, 282))])])))
+
+    def test_bbox_int_degenerate(self):
+        # float 合法但 int() 后宽度为零: 专为 CHS shim 的裁剪路径兜底
+        self._expect_invalid(_zip_bytes(_middle_bytes([_page([_block("table", (1.2, 1.2, 1.8, 5))])])))
+
+    def test_bbox_non_finite(self):
+        self._expect_invalid(_zip_bytes(_middle_bytes([_page([_block("table", (1, 2, float("inf"), 4))])])))
+
+    def test_type_missing(self):
+        block = {"bbox": [1, 2, 30, 40], "index": 0}
+        self._expect_invalid(_zip_bytes(_middle_bytes([_page([block])])))
+
+    def test_para_blocks_absent_defaults_empty(self):
+        page = _page([])
+        del page["para_blocks"]
+        elements, page_sizes = _parse_middle_json(_zip_bytes(_middle_bytes([page])))
+        assert elements == [] and page_sizes == [(759.0, 540.0)]
+
+    def test_page_idx_missing_falls_back_to_enumeration(self):
+        page = _page([_block("text", (1, 2, 30, 40))])
+        del page["page_idx"]
+        elements, _ = _parse_middle_json(_zip_bytes(_middle_bytes([page])))
+        assert elements[0].page_index == 0
+
+
+class TestErrorTranslation:
+    @pytest.mark.parametrize(
+        ("status", "exc_type"),
+        [(502, TransientError), (429, TransientError), (401, SourceDeadError),
+         (403, SourceDeadError), (404, RequestRejectedError)],
+    )
+    async def test_http_status(self, status, exc_type):
+        t = _transport_for(_routes(text_resp=httpx.Response(status)))
+        with pytest.raises(exc_type) as ei:
+            await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
+        assert ei.value.status_code == status
+
+    async def test_connect_error_transient(self):
+        def handler(request):
+            raise httpx.ConnectError("refused", request=request)
+
+        t = _transport_for(handler)
+        with pytest.raises(TransientError) as ei:
+            await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
+        assert ei.value.status_code is None
+
+    async def test_read_timeout_transient_with_cause(self):
+        def handler(request):
+            raise httpx.ReadTimeout("slow", request=request)
+
+        t = _transport_for(handler)
+        with pytest.raises(TransientError) as ei:
+            await t.recognize_text(image=b"jpg", source=_source(), call_id="c1")
+        # __cause__ 链保留,RetryMW._failure_reason 依赖它归类 timeout
+        assert isinstance(ei.value.__cause__, httpx.TimeoutException)
+
+
+class TestCheckHealth:
+    async def test_healthy(self):
+        t = _transport_for(_routes(health_resp=httpx.Response(200, json={"status": "healthy"})))
+        assert await t.check_health(source=_source()) is True
+
+    async def test_500_unhealthy(self):
+        t = _transport_for(_routes(health_resp=httpx.Response(500)))
+        assert await t.check_health(source=_source()) is False
+
+    async def test_3xx_unhealthy(self):
+        # 设计 §10.1 有意收紧: VT 的 resp.ok 含 3xx,库只认 2xx
+        t = _transport_for(_routes(health_resp=httpx.Response(302)))
+        assert await t.check_health(source=_source()) is False
+
+    async def test_connect_failure_false(self):
+        def handler(request):
+            raise httpx.ConnectError("refused", request=request)
+
+        t = _transport_for(handler)
+        assert await t.check_health(source=_source()) is False
+
+    async def test_cancellation_passes_through(self):
+        def handler(request):
+            raise asyncio.CancelledError()
+
+        t = _transport_for(handler)
+        with pytest.raises(asyncio.CancelledError):
+            await t.check_health(source=_source())
+
+
+class TestLifecycle:
+    async def test_aclose_idempotent(self):
+        t = _transport_for(_routes(health_resp=httpx.Response(200)))
+        await t.check_health(source=_source())
+        await t.aclose()
+        await t.aclose()