fix(ocr): round-3 review — guard unexpected batch status + tests/comments

Round 3 review (PR #910):
- Guard an unexpected terminal batch status in _process_batch: anything that
  isn't succeeded/failed (gateway version skew, a new lifecycle state) now marks
  the document parse-failed instead of falling through to _pages_to_text([]) — a
  0-chunk "success" that silently indexed empty text and re-submitted forever.
  Test added.
- gateway_batch_client.submit: raise an actionable ValueError on a 2xx response
  with no job_id (was a bare KeyError deep in the caller).
- Document that a _process_batch transport error intentionally propagates to
  procrastinate for retry rather than falling back to sync (opt-in batch wants
  the retry).
- Annotate _batch_client as GatewayBatchOcrClient | None (TYPE_CHECKING import
  already present); clarify the delete_stale_for_doc first-submit no-op comment.
- Add a parametrized build_gateway_batch_client test (the gateway-only invariant:
  mistral/none/no-URL -> None; gateway|auto + URL -> client).

1653 unit tests pass; ruff + ty green.

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

tests/unit/test_ocr_processor.py

@@ -76,6 +76,26 @@ def test_build_backend_auto_none_configured():
     assert ocr.build_ocr_backend(_settings()) is None
 
 
+@pytest.mark.parametrize(
+    "kw, expect_client",
+    [
+        # batch is gateway-only: the direct mistral backend never gets a client.
+        (dict(document_ocr_provider="mistral", mistral_api_key="k"), False),
+        # gateway selected but no URL -> no client (falls back to sync).
+        (dict(document_ocr_provider="gateway"), False),
+        (
+            dict(document_ocr_provider="gateway", embedding_gateway_url="https://gw"),
+            True,
+        ),
+        (dict(document_ocr_provider="auto", embedding_gateway_url="https://gw"), True),
+        (dict(document_ocr_provider="none", embedding_gateway_url="https://gw"), False),
+    ],
+)
+def test_build_gateway_batch_client_gateway_only(kw, expect_client):
+    client = ocr.build_gateway_batch_client(_settings(**kw))
+    assert (client is not None) is expect_client
+
+
 def test_build_backend_gateway_missing_m2m_raises():
     # client_id set but token_url/secret missing -> explicit ValueError (not a
     # stripped assert), surfaced on backend resolution.
@@ -391,6 +411,24 @@ async def test_batch_failed_marks_parse_error(monkeypatch):
     assert ("u1", "d1", "file", "v1") in store.deleted
 
 
+async def test_batch_unexpected_status_marks_failed_not_empty_success(monkeypatch):
+    # A terminal status that isn't succeeded/failed (gateway skew) must NOT
+    # produce a 0-chunk "success" that silently indexes empty text + loops.
+    preset = SimpleNamespace(job_id="mistral/j", status="pending", submitted_at=1000)
+    client = _FakeBatchClient(poll=BatchPollResult(status="cancelled", pages=[]))
+    store = _FakeStore(preset=preset)
+    _wire_batch(monkeypatch, client=client, store=store)
+
+    r = await ocr.OcrProcessor().process(
+        b"%PDF", "application/pdf", options=dict(_IDENTITY)
+    )
+
+    assert r.success is False
+    assert r.metadata["parse_failed_reason"] == "error"
+    assert "cancelled" in (r.error or "")
+    assert ("u1", "d1", "file", "v1") in store.deleted
+
+
 async def test_batch_deadline_exceeded_marks_timeout(monkeypatch):
     preset = SimpleNamespace(job_id="mistral/j", status="pending", submitted_at=1000)
     client = _FakeBatchClient(poll=BatchPollResult(status="pending", pages=[]))