test(ingest): cover ocr-incluster routing + fix scan-gate & batch guard

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) <noreply@anthropic.com>

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")

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()

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"
+    )

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"