feat(ingest): split OCR into tier2 in-cluster (GPU, gateway-only) + tier3 upstream

Insert a configurable in-cluster OCR rung into the escalation ladder (Deck #353):
a tier2-eligible doc is OCR'd on the on-demand burst GPU before falling through to
paid upstream OCR. The in-cluster backend is reached ONLY via the embedding gateway
(model prefix routes to the GPU over the tailnet) and is a config value (default
surya/surya-ocr-2, swappable to e.g. lightonocr) — never hard-coded.

Ladder: fast -> structured -> ocr-incluster -> ocr-upstream
(queues ingest-ocr-incluster / ingest-ocr-upstream).

- escalation.py: 4-tier ladder; in-cluster flag folded into the dead-letter signature.
- ocr.py: OcrProcessor(name, tier, model_setting, gateway_only); build_ocr_backend(
  ..., model=, gateway_only=) — gateway_only forces the gateway backend (never the
  direct Mistral fallback), disabling the tier with a warning if no gateway URL.
- registry.py: per-rung enable map; scanned docs target minimum="ocr-incluster";
  inline path runs the cheapest available OCR rung.
- procrastinate.py: two OCR queues; legacy ingest-ocr kept as a drain target.
- config.py: DOCUMENT_OCR_INCLUSTER_ENABLED (off) + DOCUMENT_OCR_INCLUSTER_MODEL.
- __init__.py: register the two OCR instances; vector/processor.py: pages_ocr
  metered for the upstream (paid) rung only; cli.py: new --tier choices + legacy drain.
- metrics.py: zero the legacy ingest-ocr queue gauge during rollout.
- tests: migrated to the split ladder + new tests (gateway-only forcing, per-tier
  model incl. lightonocr override, no-hard-coded-surya guard). 1792 pass; ruff + ty green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

tests/unit/vector/test_processor_dead_letter.py

@@ -29,6 +29,7 @@ pytestmark = pytest.mark.unit
 def _settings(*, ocr_enabled: bool) -> SimpleNamespace:
     return SimpleNamespace(
         document_ocr_enabled=ocr_enabled,
+        document_ocr_incluster_enabled=False,
         document_tier1_engine="pypdfium2",
         get_collection_name=lambda: "c",
     )

tests/unit/vector/test_procrastinate_producer.py

@@ -115,7 +115,7 @@ class TestProcessDocumentTask:
         # Calling the Task runs its wrapped function in-process. The job is on the
         # ocr queue, so the queue-aware task must derive tier="ocr".
         await pq.process_document_task(
-            _ctx(pq.INGEST_QUEUE_OCR),
+            _ctx(pq.INGEST_QUEUE_OCR_UPSTREAM),
             user_id="alice",
             doc_id="42",
             doc_type="note",
@@ -131,7 +131,7 @@ class TestProcessDocumentTask:
         # Worker disables the in-process retry loop; durable retry is the queue's.
         assert captured["max_retries"] == 1
         # Tier is derived from the job's queue (escalation enabled by default).
-        assert captured["tier"] == "ocr"
+        assert captured["tier"] == "ocr-upstream"
         fake_client.close.assert_awaited_once()
 
     async def test_pipeline_error_propagates_and_closes_client(self, monkeypatch):

tests/unit/vector/test_tiered_escalation_strategy.py

@@ -36,15 +36,16 @@ def _job(queue: str = pq.INGEST_QUEUE_FAST, attempts: int = 1) -> Job:
 class TestLadder:
     def test_next_tier_ordering(self):
         assert next_tier("fast") == "structured"
-        assert next_tier("structured") == "ocr"
-        assert next_tier("ocr") is None  # terminal
+        assert next_tier("structured") == "ocr-incluster"
+        assert next_tier("ocr-incluster") == "ocr-upstream"
+        assert next_tier("ocr-upstream") is None  # terminal
         assert next_tier("unknown") is None
 
     def test_ladder_is_cheapest_first(self):
-        assert TIER_LADDER == ("fast", "structured", "ocr")
+        assert TIER_LADDER == ("fast", "structured", "ocr-incluster", "ocr-upstream")
 
     def test_tier_for_queue(self):
-        assert pq.tier_for_queue(pq.INGEST_QUEUE_OCR) == "ocr"
+        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.
         assert pq.tier_for_queue(pq.LEGACY_INGEST_QUEUE) == "fast"
@@ -56,10 +57,12 @@ class TestTieredEscalationStrategy:
         return pq.TieredEscalationStrategy(max_transient_attempts=max_transient)
 
     def test_escalate_hops_to_target_queue(self):
-        exc = EscalateError(from_tier="fast", to_tier="ocr", reason="empty_text")
+        exc = EscalateError(
+            from_tier="fast", to_tier="ocr-upstream", 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
+        assert decision.queue == pq.INGEST_QUEUE_OCR_UPSTREAM
 
     def test_escalate_to_structured(self):
         exc = EscalateError(
@@ -75,11 +78,13 @@ class TestTieredEscalationStrategy:
         assert decision is None
 
     def test_escalate_unwraps_exception_group(self):
-        exc = EscalateError(from_tier="fast", to_tier="ocr", reason="empty_text")
+        exc = EscalateError(
+            from_tier="fast", to_tier="ocr-upstream", reason="empty_text"
+        )
         group = ExceptionGroup("wrapped", [exc])
         decision = self._strategy().get_retry_decision(exception=group, job=_job())
         assert decision is not None
-        assert decision.queue == pq.INGEST_QUEUE_OCR
+        assert decision.queue == pq.INGEST_QUEUE_OCR_UPSTREAM
 
     def test_transient_retries_same_queue_under_cap(self):
         decision = self._strategy(max_transient=5).get_retry_decision(
@@ -126,7 +131,8 @@ class TestTieredEscalationStrategy:
         # Batch OCR re-poll (Deck #332): same-queue deferral after retry_in.
         before = datetime.now(timezone.utc)
         decision = self._strategy().get_retry_decision(
-            exception=BatchPending(retry_in=120), job=_job(queue=pq.INGEST_QUEUE_OCR)
+            exception=BatchPending(retry_in=120),
+            job=_job(queue=pq.INGEST_QUEUE_OCR_UPSTREAM),
         )
         after = datetime.now(timezone.utc)
         assert decision is not None