diff --git a/nextcloud_mcp_server/document_processors/ocr.py b/nextcloud_mcp_server/document_processors/ocr.py index 21a072b9..91265e86 100644 --- a/nextcloud_mcp_server/document_processors/ocr.py +++ b/nextcloud_mcp_server/document_processors/ocr.py @@ -277,7 +277,7 @@ class OcrProcessor(DocumentProcessor): # documents. ``_batch_fallback_warned`` rate-limits the "can't batch, # using sync" warning to once per pod. self._batch_client_resolved = False - self._batch_client: Any = None + self._batch_client: GatewayBatchOcrClient | None = None self._batch_client_lock: anyio.Lock | None = None self._batch_fallback_warned = False @@ -310,6 +310,11 @@ class OcrProcessor(DocumentProcessor): # "pending" sentinel) when handled, or None to fall back to the # synchronous path below (no gateway backend, or no per-doc identity — the # inline/memory pool can't defer a poll). + # + # A transport error from _process_batch (e.g. the gateway briefly down) + # is intentionally NOT caught here: it propagates to procrastinate for a + # durable retry rather than silently falling back to sync. If you've opted + # into batch mode you want the retry, not an unexpected sync transcription. if settings.document_ocr_mode == "batch": batch_result = await self._process_batch( content, content_type, filename, options, settings @@ -446,7 +451,8 @@ class OcrProcessor(DocumentProcessor): ) if job is None: # New submission. Drop any superseded-version rows for this doc first - # (a re-edited file changes etag), then submit + record. + # (a re-edited file changes etag) — a no-op on the very first submit, + # one cheap DELETE on a resubmit. Then submit + record. await store.delete_stale_for_doc( user_id=user_id, doc_id=doc_id, doc_type=doc_type, keep_etag=etag ) @@ -500,6 +506,24 @@ class OcrProcessor(DocumentProcessor): success=False, error=f"batch OCR failed: {result.error or 'unknown'}", ) + if not result.is_succeeded: + # Defensive: poll() maps anything that isn't "succeeded" to its raw + # status, and only pending/succeeded/failed are handled above. An + # unexpected terminal status (gateway version skew, a new lifecycle + # state) must NOT fall through to _pages_to_text([]) -> a 0-chunk + # "success" that silently indexes empty text and re-submits forever. + logger.warning( + "batch OCR job %s returned unexpected status %r; marking failed", + job.job_id, + result.status, + ) + return ProcessingResult( + text="", + metadata={"parse_failed_reason": "error"}, + processor=self.name, + success=False, + error=f"unexpected batch status: {result.status}", + ) text, boundaries = _pages_to_text(result.pages) return ProcessingResult( text=text, diff --git a/nextcloud_mcp_server/embedding/gateway_batch_client.py b/nextcloud_mcp_server/embedding/gateway_batch_client.py index 6208a5c1..9991ee98 100644 --- a/nextcloud_mcp_server/embedding/gateway_batch_client.py +++ b/nextcloud_mcp_server/embedding/gateway_batch_client.py @@ -120,7 +120,11 @@ class GatewayBatchOcrClient: ) resp.raise_for_status() body = resp.json() - job_id = body["job_id"] + job_id = body.get("job_id") + if not job_id: + # Contract violation (2xx without a job id) — fail with an actionable + # message rather than a bare KeyError deep in the caller. + raise ValueError(f"gateway batch submit returned no job_id: {body!r}") logger.info( "batch OCR submitted: job_id=%s custom_id=%s status=%s", job_id, diff --git a/tests/unit/test_ocr_processor.py b/tests/unit/test_ocr_processor.py index 2f12f6e6..a279ee3a 100644 --- a/tests/unit/test_ocr_processor.py +++ b/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=[]))