diff --git a/src/polygateway/ocr.py b/src/polygateway/ocr.py index a0f862c..f734f06 100644 --- a/src/polygateway/ocr.py +++ b/src/polygateway/ocr.py @@ -39,13 +39,17 @@ from polygateway.middleware.breaker import BreakerGate from polygateway.middleware.ratelimit import QuotaGate from polygateway.middleware.retry import StallClock, _failure_reason, backoff_delay from polygateway.middleware.telemetry import TelemetryEmitter, emit_terminal_once -from polygateway.ports import OutcomeAwareSelector +from polygateway.ports import OcrDocumentTransport, OutcomeAwareSelector from polygateway.types import ( CallStats, ChatRequest, LLMResponse, + OcrDocumentResult, + OcrDocumentTransportResult, OcrLayoutResult, + OcrLayoutTransportResult, OcrTextResult, + OcrTextTransportResult, TelemetryStatus, Usage, _CallContext, @@ -69,15 +73,18 @@ if TYPE_CHECKING: from polygateway.types import ( BackpressurePolicy, CallOperation, - OcrLayoutTransportResult, - OcrTextTransportResult, RetryPolicy, SourceConfig, ) _RESPONSE_TEXT_CAP = 200 # 遥测行 text 截断长度(与 embedding 口径一致) -_OcrKind = Literal["text", "layout"] +_OcrKind = Literal["text", "layout", "document"] + +# 输入校验文案里的名字。旧两个入口(text/layout)的报错文本保持不变,文档入口自称 document +_PAYLOAD_NAME: dict[_OcrKind, str] = {"text": "image", "layout": "image", "document": "document"} + +_TransportResult = OcrTextTransportResult | OcrLayoutTransportResult | OcrDocumentTransportResult @dataclass(frozen=True) @@ -88,7 +95,7 @@ class _FailedAttempt: @dataclass(frozen=True) class _AttemptOutcome: - result: OcrTextTransportResult | OcrLayoutTransportResult + result: _TransportResult source: SourceConfig call_id: str latency_ms: int @@ -254,6 +261,53 @@ class OcrClient: call_stats=call_stats, ) + async def parse_document( + self, + document: bytes, + *, + session_id: str | None = None, + parent_call_id: str | None = None, + tenant_id: str | None = None, + meta: Mapping[str, Any] | None = None, + call_deadline_s: float | None = None, + ) -> OcrDocumentResult: + """一次治理整份文档解析(/parse → ZIP);PDF 或图片,逐块带类型/页码/bbox/文字。 + + 上传类型由 transport 按内容判定,调用方无需也无法声明。结果的 `markdown` 是服务 + 生成的整篇原文,含悬空的图片引用;服务端会丢弃脚注、页眉页脚与页码。 + + `tenant_id` 与 `meta` 是调用方自定义维度,只进遥测(issue #11)。 + `call_deadline_s` 同 `recognize_text`(issue #22): `None` = 继承装配值。长文档的 + 硬上限靠它——源的 `timeout_s` 是逐次网络操作的超时,不是总耗时上界。 + """ + # 校验早于链路,理由同 recognize_text;transport 能力校验也在统计边界之外 + dimension_tenant_id, dimensions = validate_caller_dimensions( + tenant_id, meta, origin="parse_document(tenant_id=..., meta=...)" + ) + self._document_transport() + outcome, call_stats = await self._call( + "document", + "parse_document", + document, + session_id, + parent_call_id, + dimension_tenant_id, + dimensions, + call_deadline_s, + ) + result = outcome.result + return OcrDocumentResult( + elements=result.elements, + page_sizes=result.page_sizes, + markdown=result.markdown, + source_name=outcome.source.name, + usage=Usage(0, 0), + latency_ms=outcome.latency_ms, + call_id=outcome.call_id, + raw=result.raw, + call_stats=call_stats, + ) + async def check_health(self) -> dict[str, bool]: """逐源并发健康预检(R10);探测失败=False 不上抛,取消穿透。""" names = [s.name for s in self._sources] @@ -272,37 +326,38 @@ class OcrClient: self, kind: _OcrKind, operation: CallOperation, - image: bytes, + payload: bytes, session_id: str | None, parent_call_id: str | None, tenant_id: str | None, meta: dict[str, Any], call_deadline_s: float | None = None, ) -> tuple[_AttemptOutcome, CallStats]: - if not isinstance(image, bytes): - raise TypeError("image 必须是 bytes(路径读取/批量拼帧留业务侧,D9)") - if not image: - raise ValueError("image 不能为空") - # 期限校验与 `image` 校验同列(仍在 `_CallContext` 之前、创建 awaitable 之前) + name = _PAYLOAD_NAME[kind] + if not isinstance(payload, bytes): + raise TypeError(f"{name} 必须是 bytes(路径读取/批量拼帧留业务侧,D9)") + if not payload: + raise ValueError(f"{name} 不能为空") + # 期限校验与 `payload` 校验同列(仍在 `_CallContext` 之前、创建 awaitable 之前) deadline = ( self._call_deadline_s if call_deadline_s is None else ensure_call_deadline(call_deadline_s, f"{operation}(call_deadline_s=...)") ) - # M1 例外: `image` 校验在 `_call` 内而非公开方法,故上下文在该校验 + # M1 例外: `payload` 校验在 `_call` 内而非公开方法,故上下文在该校验 # **通过之后**创建——这样设计 §3 的"校验在统计边界外"对 OCR 才成立 context = _CallContext(now=self._now) try: return await with_call_deadline( self._run( - kind, operation, image, session_id, parent_call_id, tenant_id, meta, context + kind, operation, payload, session_id, parent_call_id, tenant_id, meta, context ), deadline_s=deadline, scope=self._scope, ) except PolyGatewayError as exc: await self._emit_terminal( - kind, operation, image, session_id, parent_call_id, tenant_id, meta, context, exc + kind, operation, payload, session_id, parent_call_id, tenant_id, meta, context, exc ) raise except asyncio.CancelledError: @@ -310,7 +365,7 @@ class OcrClient: await self._emit_terminal( kind, operation, - image, + payload, session_id, parent_call_id, tenant_id, @@ -324,7 +379,7 @@ class OcrClient: self, kind: _OcrKind, operation: CallOperation, - image: bytes, + payload: bytes, session_id: str | None, parent_call_id: str | None, tenant_id: str | None, @@ -351,7 +406,7 @@ class OcrClient: outcome = await self._attempt( kind, operation, - image, + payload, *picked, reasons, session_id, @@ -377,7 +432,7 @@ class OcrClient: self, kind: _OcrKind, operation: CallOperation, - image: bytes, + payload: bytes, source: SourceConfig, permit: Permit, entry: GateDecision, @@ -398,7 +453,7 @@ class OcrClient: # 裸生成时间(1.3.7 H8): 计时只包 transport 调用本身(同 RetryMW 口径); # layout 的 POST + ZIP GET 在同一次 `_invoke` 内,属同一段生成耗时 gen_started = self._now() - result = await self._invoke(kind, image, source, call_id) + result = await self._invoke(kind, payload, source, call_id) gen_ms = int((self._now() - gen_started) * 1000) await self._record_quietly(self._breaker.record_success(entry)) await self._record_quietly(self._quota.mark_progress()) @@ -408,7 +463,7 @@ class OcrClient: await self._emit( kind, operation, - image, + payload, source, call_id, started, @@ -425,7 +480,7 @@ class OcrClient: await self._emit( kind, operation, - image, + payload, source, call_id, started, @@ -443,7 +498,7 @@ class OcrClient: await self._emit( kind, operation, - image, + payload, source, call_id, started, @@ -464,7 +519,7 @@ class OcrClient: await self._emit( kind, operation, - image, + payload, source, call_id, started, @@ -482,11 +537,31 @@ class OcrClient: await settle_and_release(permit, 0) async def _invoke( - self, kind: _OcrKind, image: bytes, source: SourceConfig, call_id: str - ) -> OcrTextTransportResult | OcrLayoutTransportResult: + self, kind: _OcrKind, payload: bytes, source: SourceConfig, call_id: str + ) -> _TransportResult: if kind == "text": - return await self._transport.recognize_text(image=image, source=source, call_id=call_id) - return await self._transport.parse_layout(image=image, source=source, call_id=call_id) + return await self._transport.recognize_text( + image=payload, source=source, call_id=call_id + ) + if kind == "document": + return await self._document_transport().parse_document( + document=payload, source=source, call_id=call_id + ) + return await self._transport.parse_layout(image=payload, source=source, call_id=call_id) + + def _document_transport(self) -> OcrDocumentTransport: + """文档入口所需的 transport 能力校验(唯一的 isinstance 点)。 + + `OcrDocumentTransport` 独立于 `OcrTransport`: 往旧 Protocol 加成员会让所有现有 + 实现(测试替身、下游自带 fake)一夜之间不合格。代价是这里必须在运行时校验—— + 公共入口在进入统计边界前调用一次,让装配错误当场报。 + """ + if not isinstance(self._transport, OcrDocumentTransport): + raise TypeError( + "parse_document 需要实现 OcrDocumentTransport.parse_document 的 transport: " + f"{type(self._transport).__name__} 不满足" + ) + return self._transport # —— 辅助(与 embedding 同口径)—— @@ -519,7 +594,7 @@ class OcrClient: def _request_for( self, kind: _OcrKind, - image: bytes, + payload: bytes, session_id: str | None, parent_call_id: str | None, tenant_id: str | None, @@ -533,7 +608,7 @@ class OcrClient: 维度恒为空、`logical_call_id` 恒为 NULL。 """ return ChatRequest( - messages=[{"role": "user", "content": f""}], + messages=[{"role": "user", "content": f""}], session_id=session_id, parent_call_id=parent_call_id, tenant_id=tenant_id, @@ -545,7 +620,7 @@ class OcrClient: self, kind: _OcrKind, operation: CallOperation, - image: bytes, + payload: bytes, session_id: str | None, parent_call_id: str | None, tenant_id: str | None, @@ -557,7 +632,7 @@ class OcrClient: await emit_terminal_once( self._emitter, request=self._request_for( - kind, image, session_id, parent_call_id, tenant_id, meta, context + kind, payload, session_id, parent_call_id, tenant_id, meta, context ), context=context, error=error, @@ -570,7 +645,7 @@ class OcrClient: self, kind: _OcrKind, operation: CallOperation, - image: bytes, + payload: bytes, source: SourceConfig, call_id: str, started: float, @@ -579,14 +654,14 @@ class OcrClient: tenant_id: str | None, meta: dict[str, Any], context: _CallContext, - result: OcrTextTransportResult | OcrLayoutTransportResult | None = None, + result: _TransportResult | None = None, error: PolyGatewayError | str | None = None, ) -> None: """逐尝试遥测(单一 Emitter): messages 占位摘要,图像 bytes 绝不入库。""" if self._emitter is None: return request = self._request_for( - kind, image, session_id, parent_call_id, tenant_id, meta, context + kind, payload, session_id, parent_call_id, tenant_id, meta, context ) latency_ms = int((self._now() - started) * 1000) response = None @@ -622,8 +697,18 @@ class OcrClient: ) @staticmethod - def _summarize(result: OcrTextTransportResult | OcrLayoutTransportResult) -> str: - if hasattr(result, "text"): + def _summarize(result: _TransportResult) -> str: + """遥测行 response 摘要。**永不写入文档正文**:文档只记规模(元素数/页数/md 字数)。 + + 用 isinstance 显式区分三类结果:旧的 `hasattr(result, "text")` 对没有顶层 `text` 的 + 新结果会静默套成 layout 摘要,漏记 `md_chars`。 + """ + if isinstance(result, OcrDocumentTransportResult): + return ( + f"" + ) + if isinstance(result, OcrTextTransportResult): return result.text[:_RESPONSE_TEXT_CAP] return f"" diff --git a/tests/unit/_document_fixtures.py b/tests/unit/_document_fixtures.py index e3f017a..8707788 100644 --- a/tests/unit/_document_fixtures.py +++ b/tests/unit/_document_fixtures.py @@ -69,3 +69,13 @@ def _package( for name, data in extra: zf.writestr(name, data) return buf.getvalue() + + +def _encrypted_package() -> bytes: + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + info = zipfile.ZipInfo("doc_middle.json") + info.flag_bits |= 0x1 # 加密位:读取时 zipfile 要求密码 + zf.writestr(info, b'{"pdf_info": []}') + zf.writestr("doc.md", "x") + return buf.getvalue() diff --git a/tests/unit/test_monkey_document.py b/tests/unit/test_monkey_document.py index ce77c84..73dfb04 100644 --- a/tests/unit/test_monkey_document.py +++ b/tests/unit/test_monkey_document.py @@ -23,6 +23,7 @@ from polygateway.types import OcrDocumentElement from tests.unit._document_fixtures import ( _BBOX, _body, + _encrypted_package, _image_block, _package, _page, @@ -408,16 +409,6 @@ def _patch_first_member_compression(zip_bytes: bytes, method: int) -> bytes: return bytes(data) -def _encrypted_zip() -> bytes: - buf = io.BytesIO() - with zipfile.ZipFile(buf, "w") as zf: - info = zipfile.ZipInfo("doc_middle.json") - info.flag_bits |= 0x1 # 加密位:读取时 zipfile 要求密码 - zf.writestr(info, b'{"pdf_info": []}') - zf.writestr("doc.md", "x") - return buf.getvalue() - - class TestExternalDataExceptionBoundary: """外部字节引起的一切已知异常都归 `ResultInvalidError`,绝不漏出原生异常。 @@ -427,7 +418,7 @@ class TestExternalDataExceptionBoundary: def test_encrypted_member(self): with pytest.raises(ResultInvalidError): - parse_document_package(_encrypted_zip()) + parse_document_package(_encrypted_package()) def test_unknown_compression_method(self): patched = _patch_first_member_compression(_package([_page([])]), 99) diff --git a/tests/unit/test_ocr_client.py b/tests/unit/test_ocr_client.py index 8660e33..97ae95f 100644 --- a/tests/unit/test_ocr_client.py +++ b/tests/unit/test_ocr_client.py @@ -6,6 +6,7 @@ retry_exhausted/circuit_open/stalled 三组断言即设计 §6 ③ 的 G1 契约 import asyncio +import httpx import pytest from loguru import logger @@ -21,20 +22,34 @@ from polygateway.errors import ( 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)], @@ -72,7 +87,11 @@ class ScriptedOcrTransport: if action == "hang": self.entered.set() await asyncio.Event().wait() - return _TEXT_OK if action == "text" else _LAYOUT_OK + 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) @@ -80,6 +99,9 @@ class ScriptedOcrTransport: 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 @@ -548,7 +570,7 @@ class TestOcrCallerDimensions: 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"]) + @pytest.mark.parametrize("method", ["recognize_text", "parse_layout", "parse_document"]) async def test_invalid_meta_rejected_before_any_telemetry(self, method): """校验必须早于遥测: 链路内的失败都被降级成 warning,放下游等于没有校验。""" recorder = _MemoryRecorder() @@ -558,6 +580,14 @@ class TestOcrCallerDimensions: 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) @@ -566,6 +596,334 @@ class TestOcrCallerDimensions: 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_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 + + +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 "" in row["messages"] + assert "RAW-BYTES" not in row["messages"] # 文档字节绝不入库 + assert row["response"] == "" + 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) == "" + assert OcrClient._summarize(_LAYOUT_OK) == "" + assert OcrClient._summarize(_TEXT_OK) == "LINE-1" + + 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 + + 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", diff --git a/tests/unit/test_telemetry.py b/tests/unit/test_telemetry.py index 8778747..9a488bf 100644 --- a/tests/unit/test_telemetry.py +++ b/tests/unit/test_telemetry.py @@ -37,6 +37,8 @@ from polygateway.types import ( EmbeddingTransportResult, GlobalLimits, LLMResponse, + OcrDocumentElement, + OcrDocumentTransportResult, OcrTextTransportResult, RetryPolicy, SourceConfig, @@ -2422,6 +2424,18 @@ class _StubOcrTransport: async def parse_layout(self, *, image, source, call_id): raise NotImplementedError + async def parse_document(self, *, document, source, call_id): + return OcrDocumentTransportResult( + elements=[ + OcrDocumentElement( + type="text", bbox=(0.0, 0.0, 1.0, 1.0), page_index=0, text="SYNTH-BODY" + ) + ], + page_sizes=[(612.0, 792.0)], + markdown="# SYNTH-MD", + raw={"success": True}, + ) + def _governance(scope, sources): """embed/OCR 两条链路共用的最小治理装配(真实内存后端,不 mock)。""" @@ -2478,6 +2492,20 @@ class TestTextCapCoversEmbedAndOcrChains: assert json.loads(row["messages"])[0]["content"] == "