From 87b8edd1399abfc17e3452a05b5ec79bfb8f1bba Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 18 Jun 2026 00:06:46 +0200 Subject: [PATCH] test(ingest): cover ocr-incluster routing + fix scan-gate & batch guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address claude-review on #922: Blocking — test coverage for the new tier2 rung: - test_registry_tiering.py: inline empty-text routes to ocr-incluster before ocr-upstream; only-incluster-enabled routes to incluster; disabled-incluster skips to upstream; evaluate_escalation empty_text hops to ocr-incluster (and falls through to upstream when incluster off); next_available_tier walks the full fast→structured→ocr-incluster→ocr-upstream ladder + ignore_ocr_enabled ideal-target. - test_escalation_signature.py: enabling document_ocr_incluster_enabled changes the dead-letter signature (independent of the upstream rung). - test_tiered_escalation_strategy.py: structured→ocr-incluster hops to INGEST_QUEUE_OCR_INCLUSTER; tier_for_queue covers the in-cluster queue. Important — real fixes: - registry.py: run scan detection (image_coverage_per_page) when EITHER OCR rung is enabled, not just the upstream one — a tenant with only in-cluster OCR on was missing image-coverage scan signals. - ocr.py: the gateway_only (in-cluster) processor never enters batch mode — the GPU is synchronous/low-latency; batch OCR is the upstream Mistral async path. _get_batch_client short-circuits to None. Covered by a new test. Nit: - cli.py: worker --tier help lists ocr-incluster/ocr-upstream as separate fleets. Left as-is: the lazy anyio.Lock init in OcrProcessor — instances ARE created at module import (document_processors/__init__.py), so deferring lock creation off import time is still required; moving it into __init__ would reintroduce the import-time-primitive issue the comment guards against. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/cli.py | 5 +- .../document_processors/ocr.py | 9 +- .../document_processors/registry.py | 8 +- tests/unit/test_escalation_signature.py | 8 ++ tests/unit/test_ocr_processor.py | 37 +++++ tests/unit/test_registry_tiering.py | 135 ++++++++++++++++++ .../vector/test_tiered_escalation_strategy.py | 11 ++ 7 files changed, 209 insertions(+), 4 deletions(-) diff --git a/nextcloud_mcp_server/cli.py b/nextcloud_mcp_server/cli.py index 7306ff8a..51c28511 100644 --- a/nextcloud_mcp_server/cli.py +++ b/nextcloud_mcp_server/cli.py @@ -354,8 +354,9 @@ def worker(concurrency: int | None, tier: str | None): \b With --tier the worker drains only that tier's queue (``ingest-``), so - a CPU-bound ``fast`` fleet, an in-cluster ``structured`` fleet, and a paid - ``ocr`` fleet scale independently. Without it, all tier queues are drained in + a CPU-bound ``fast`` fleet, an in-cluster ``structured`` fleet, an on-demand + GPU ``ocr-incluster`` fleet, and a paid ``ocr-upstream`` fleet scale + independently. Without it, all tier queues are drained in one process (handy for dev / a single Deployment). A low-quality parse hops the job to the next tier's queue automatically (see TieredEscalationStrategy). diff --git a/nextcloud_mcp_server/document_processors/ocr.py b/nextcloud_mcp_server/document_processors/ocr.py index bb6b771f..bb788f53 100644 --- a/nextcloud_mcp_server/document_processors/ocr.py +++ b/nextcloud_mcp_server/document_processors/ocr.py @@ -437,7 +437,14 @@ class OcrProcessor(DocumentProcessor): async def _get_batch_client(self) -> "GatewayBatchOcrClient | None": """Cached gateway batch client (or ``None`` when batch isn't applicable — provider=mistral / no gateway). Resolved once under the backend lock so the - token provider's M2M cache survives across documents.""" + token provider's M2M cache survives across documents. + + The in-cluster (``gateway_only``) rung never uses batch mode: it targets + the on-demand GPU, which is synchronous/low-latency, while batch OCR is the + upstream (Mistral) async-job path. So even with ``DOCUMENT_OCR_MODE=batch`` + set globally, the in-cluster tier stays on the synchronous backend.""" + if self._gateway_only: + return None if not self._batch_client_resolved: if self._batch_client_lock is None: self._batch_client_lock = anyio.Lock() diff --git a/nextcloud_mcp_server/document_processors/registry.py b/nextcloud_mcp_server/document_processors/registry.py index bda81b55..b5a59638 100644 --- a/nextcloud_mcp_server/document_processors/registry.py +++ b/nextcloud_mcp_server/document_processors/registry.py @@ -458,7 +458,13 @@ class ProcessorRegistry: return None try: image_coverage = None - if settings.document_ocr_enabled and settings.document_ocr_detect_scanned: + # Scan detection feeds either OCR rung (tier2 in-cluster or tier3 + # upstream), so run it whenever EITHER is enabled — a tenant with + # only in-cluster OCR on still needs image-coverage scan signals. + ocr_any_enabled = ( + settings.document_ocr_enabled or settings.document_ocr_incluster_enabled + ) + if ocr_any_enabled and settings.document_ocr_detect_scanned: try: image_coverage = image_coverage_per_page(content) except Exception: diff --git a/tests/unit/test_escalation_signature.py b/tests/unit/test_escalation_signature.py index 9a45674a..749a428f 100644 --- a/tests/unit/test_escalation_signature.py +++ b/tests/unit/test_escalation_signature.py @@ -42,6 +42,14 @@ def test_enabling_ocr_changes_signature() -> None: ) != escalation_tiers_signature(_settings(ocr=True)) +def test_enabling_ocr_incluster_changes_signature() -> None: + # Enabling the in-cluster (tier2) rung adds an escalation tier independently of + # the upstream rung -> previously dead-lettered scanned docs become retryable. + assert escalation_tiers_signature( + _settings(ocr=False, ocr_incluster=False) + ) != escalation_tiers_signature(_settings(ocr=False, ocr_incluster=True)) + + def test_tier1_engine_change_changes_signature() -> None: assert escalation_tiers_signature( _settings(ocr=False, engine="pypdfium2") diff --git a/tests/unit/test_ocr_processor.py b/tests/unit/test_ocr_processor.py index a9491f9d..cca2bdd7 100644 --- a/tests/unit/test_ocr_processor.py +++ b/tests/unit/test_ocr_processor.py @@ -435,6 +435,43 @@ def _wire_batch(monkeypatch, *, client, store, settings=None): monkeypatch.setattr(_bos.BatchOcrJobStore, "shared", classmethod(_shared)) +async def test_gateway_only_processor_never_uses_batch_mode(monkeypatch): + """The in-cluster (gateway_only) rung targets the synchronous GPU; batch mode + is the upstream Mistral async-job path. _get_batch_client returns None and + never builds a batch client even with DOCUMENT_OCR_MODE=batch set globally — + while the upstream (gateway_only=False) processor still resolves one.""" + called = {"n": 0} + + def _spy(settings, **kw): + called["n"] += 1 + return _FakeBatchClient() + + monkeypatch.setattr( + ocr, + "get_settings", + lambda: _settings( + document_ocr_mode="batch", + document_ocr_provider="gateway", + embedding_gateway_url="https://gw", + document_ocr_incluster_model="surya/surya-ocr-2", + ), + ) + monkeypatch.setattr(ocr, "build_gateway_batch_client", _spy) + + incluster = ocr.OcrProcessor( + name="ocr-incluster", + tier="ocr-incluster", + model_setting="document_ocr_incluster_model", + gateway_only=True, + ) + assert await incluster._get_batch_client() is None + assert called["n"] == 0 # short-circuited before building anything + + upstream = ocr.OcrProcessor() # gateway_only=False + assert await upstream._get_batch_client() is not None + assert called["n"] == 1 + + async def test_batch_first_run_submits_and_returns_pending_sentinel(monkeypatch): client = _FakeBatchClient() store = _FakeStore() diff --git a/tests/unit/test_registry_tiering.py b/tests/unit/test_registry_tiering.py index ca6bba4a..54acf67c 100644 --- a/tests/unit/test_registry_tiering.py +++ b/tests/unit/test_registry_tiering.py @@ -743,3 +743,138 @@ def test_evaluate_escalation_empty_suppressed_even_when_structured_registered( ) decision = r.evaluate_escalation(res, b"%PDF", "fast", _Settings(ocr=False)) assert decision == EscalationDecision("suppressed", "ocr-upstream", "empty_text") + + +# --- tier2 in-cluster OCR rung (Deck #353) ----------------------------------- + + +async def test_inline_empty_routes_to_ocr_incluster_before_upstream(monkeypatch): + """Both OCR rungs enabled + registered: an empty text layer hops to the + in-cluster (tier2) rung FIRST, never straight to the paid upstream rung.""" + monkeypatch.setattr( + reg_mod, "get_settings", lambda: _Settings(ocr=True, ocr_incluster=True) + ) + esc = MagicMock() + monkeypatch.setattr(reg_mod, "record_document_escalation", esc) + r = _registry( + (_Fake("fast", "fast", text=""), 20), + (_Fake("structured", "structured", text="should not run"), 10), + (_Fake("ocr-incluster", "ocr-incluster", text="incluster ocr text"), 6), + (_Fake("ocr-upstream", "ocr-upstream", text="upstream ocr text"), 5), + ) + res = await r.process(b"%PDF-1.7", "application/pdf") + assert res.processor == "ocr-incluster" + esc.assert_called_once_with("fast", "ocr-incluster", "empty_text") + + +async def test_inline_only_incluster_enabled_routes_to_incluster(monkeypatch): + """Tenant with ONLY in-cluster OCR on (upstream off): empty text still hops + to the in-cluster rung (its own enable flag gates it, independent of upstream).""" + monkeypatch.setattr( + reg_mod, "get_settings", lambda: _Settings(ocr=False, ocr_incluster=True) + ) + esc = MagicMock() + monkeypatch.setattr(reg_mod, "record_document_escalation", esc) + r = _registry( + (_Fake("fast", "fast", text=""), 20), + (_Fake("ocr-incluster", "ocr-incluster", text="incluster ocr text"), 6), + (_Fake("ocr-upstream", "ocr-upstream", text="upstream ocr text"), 5), + ) + res = await r.process(b"%PDF-1.7", "application/pdf") + assert res.processor == "ocr-incluster" + esc.assert_called_once_with("fast", "ocr-incluster", "empty_text") + + +async def test_inline_incluster_disabled_skips_to_upstream(monkeypatch): + """In-cluster registered but DISABLED, upstream enabled: the disabled tier2 + rung is skipped and the job escalates to the upstream rung.""" + monkeypatch.setattr( + reg_mod, "get_settings", lambda: _Settings(ocr=True, ocr_incluster=False) + ) + esc = MagicMock() + monkeypatch.setattr(reg_mod, "record_document_escalation", esc) + r = _registry( + (_Fake("fast", "fast", text=""), 20), + (_Fake("ocr-incluster", "ocr-incluster", text="incluster ocr text"), 6), + (_Fake("ocr-upstream", "ocr-upstream", text="upstream ocr text"), 5), + ) + res = await r.process(b"%PDF-1.7", "application/pdf") + assert res.processor == "ocr-upstream" + esc.assert_called_once_with("fast", "ocr-upstream", "empty_text") + + +def _empty_result() -> ProcessingResult: + return ProcessingResult( + text="", + metadata={ + "page_count": 1, + "page_boundaries": [{"page": 1, "start_offset": 0, "end_offset": 0}], + }, + processor="fast", + ) + + +def test_evaluate_escalation_empty_text_hops_to_incluster(monkeypatch): + """External path: empty text targets the cheapest OCR rung (in-cluster, tier2) + when it is enabled + registered.""" + monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock()) + r = _registry( + (_Fake("fast", "fast"), 20), + (_Fake("structured", "structured"), 10), + (_Fake("ocr-incluster", "ocr-incluster"), 6), + (_Fake("ocr-upstream", "ocr-upstream"), 5), + ) + decision = r.evaluate_escalation( + _empty_result(), b"%PDF", "fast", _Settings(ocr=True, ocr_incluster=True) + ) + assert decision == EscalationDecision("hop", "ocr-incluster", "empty_text") + + +def test_evaluate_escalation_incluster_disabled_hops_to_upstream(monkeypatch): + """External path: in-cluster disabled -> next_available_tier falls through to + the upstream rung (a real hop, not a suppression, since upstream is on).""" + monkeypatch.setattr(reg_mod, "record_document_classification", MagicMock()) + r = _registry( + (_Fake("fast", "fast"), 20), + (_Fake("ocr-incluster", "ocr-incluster"), 6), + (_Fake("ocr-upstream", "ocr-upstream"), 5), + ) + decision = r.evaluate_escalation( + _empty_result(), b"%PDF", "fast", _Settings(ocr=True, ocr_incluster=False) + ) + assert decision == EscalationDecision("hop", "ocr-upstream", "empty_text") + + +def test_next_available_tier_walks_full_four_rung_ladder(): + """next_available_tier walks fast -> structured -> ocr-incluster -> + ocr-upstream when every rung is registered + enabled.""" + r = _registry( + (_Fake("fast", "fast"), 20), + (_Fake("structured", "structured"), 10), + (_Fake("ocr-incluster", "ocr-incluster"), 6), + (_Fake("ocr-upstream", "ocr-upstream"), 5), + ) + s = _Settings(ocr=True, ocr_incluster=True) + assert r.next_available_tier("fast", s) == "structured" + assert r.next_available_tier("structured", s) == "ocr-incluster" + assert r.next_available_tier("ocr-incluster", s) == "ocr-upstream" + assert r.next_available_tier("ocr-upstream", s) is None + # minimum pins the floor: from fast with minimum=ocr-incluster skips structured. + assert r.next_available_tier("fast", s, minimum="ocr-incluster") == "ocr-incluster" + + +def test_next_available_tier_incluster_disabled_skips_to_upstream(): + """A disabled in-cluster rung is skipped; the walk lands on the upstream rung.""" + r = _registry( + (_Fake("fast", "fast"), 20), + (_Fake("structured", "structured"), 10), + (_Fake("ocr-incluster", "ocr-incluster"), 6), + (_Fake("ocr-upstream", "ocr-upstream"), 5), + ) + s = _Settings(ocr=True, ocr_incluster=False) + assert r.next_available_tier("structured", s) == "ocr-upstream" + # ...but the *ideal* target ignoring the enable gate is still in-cluster. + assert ( + r.next_available_tier("structured", s, ignore_ocr_enabled=True) + == "ocr-incluster" + ) diff --git a/tests/unit/vector/test_tiered_escalation_strategy.py b/tests/unit/vector/test_tiered_escalation_strategy.py index 8a389cb7..04147c9a 100644 --- a/tests/unit/vector/test_tiered_escalation_strategy.py +++ b/tests/unit/vector/test_tiered_escalation_strategy.py @@ -45,6 +45,7 @@ class TestLadder: assert TIER_LADDER == ("fast", "structured", "ocr-incluster", "ocr-upstream") def test_tier_for_queue(self): + assert pq.tier_for_queue(pq.INGEST_QUEUE_OCR_INCLUSTER) == "ocr-incluster" assert pq.tier_for_queue(pq.INGEST_QUEUE_OCR_UPSTREAM) == "ocr-upstream" assert pq.tier_for_queue(pq.INGEST_QUEUE_STRUCTURED) == "structured" # Legacy / unknown / None all fall back to the cheapest tier. @@ -64,6 +65,16 @@ class TestTieredEscalationStrategy: assert decision is not None assert decision.queue == pq.INGEST_QUEUE_OCR_UPSTREAM + def test_escalate_hops_to_incluster_queue(self): + # tier2 in-cluster (Deck #353): structured -> ocr-incluster lands on the + # in-cluster queue the GPU sentinel watches, not the paid upstream queue. + exc = EscalateError( + from_tier="structured", to_tier="ocr-incluster", reason="empty_text" + ) + decision = self._strategy().get_retry_decision(exception=exc, job=_job()) + assert decision is not None + assert decision.queue == pq.INGEST_QUEUE_OCR_INCLUSTER + def test_escalate_to_structured(self): exc = EscalateError( from_tier="fast", to_tier="structured", reason="low_confidence"