From 232684e88167b005adae9a77ea1aa88a38e8e42f Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Mon, 15 Jun 2026 11:04:03 +0200 Subject: [PATCH] =?UTF-8?q?fix(ocr):=20round-4=20review=20=E2=80=94=20defe?= =?UTF-8?q?nsive=20poll=20+=20drop=20dead=20tracking=20columns?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round 4 review (PR #910), no blockers: - poll(): a 2xx body with no `status` now fails fast (logged) instead of being treated as perpetually pending until the deadline; defensive page index (`p.get("index", i)`) so a malformed page degrades rather than KeyError-ing. - Document on poll() that job_id is namespaced (embeds "/") so the gateway route must be a path-capture param (GET /v1/ocr/batch/{job_id:path}). - Drop the vestigial `status` + `updated_at` columns from batch_ocr_jobs: a row only ever exists while pending (terminal jobs are deleted) and the live status comes from a fresh poll, so a stored mirror was permanently "pending" / redundant with submitted_at. Simplifies the migration, store, and dataclass. - Tests: submit() ValueError on missing job_id; poll() missing-status → failed. 1653 unit tests pass; ruff + ty green. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../20260615_1200_008_add_batch_ocr_jobs.py | 9 +++---- .../embedding/gateway_batch_client.py | 25 ++++++++++++++++--- .../vector/batch_ocr_store.py | 15 +++++------ tests/unit/test_batch_ocr_store.py | 1 - tests/unit/test_gateway_batch_client.py | 16 ++++++++++++ 5 files changed, 50 insertions(+), 16 deletions(-) diff --git a/nextcloud_mcp_server/alembic/versions/20260615_1200_008_add_batch_ocr_jobs.py b/nextcloud_mcp_server/alembic/versions/20260615_1200_008_add_batch_ocr_jobs.py index af85c6c7..df092704 100644 --- a/nextcloud_mcp_server/alembic/versions/20260615_1200_008_add_batch_ocr_jobs.py +++ b/nextcloud_mcp_server/alembic/versions/20260615_1200_008_add_batch_ocr_jobs.py @@ -48,13 +48,12 @@ def upgrade() -> None: # The gateway's namespaced batch job id ("/") — # the only handle for polling (the gateway is stateless). sa.Column("job_id", sa.Text(), nullable=False), - # Gateway-normalised status mirror (pending|succeeded|failed). Kept for - # observability; the live decision always comes from a fresh poll. - sa.Column("status", sa.Text(), nullable=False), # Unix-epoch seconds. ``submitted_at`` anchors the poll deadline - # (DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS). + # (DOCUMENT_OCR_BATCH_MAX_WAIT_SECONDS). No status/updated_at column: a row + # only ever exists in the pending state (terminal jobs are deleted), and + # the live status always comes from a fresh poll — a stored mirror would + # be permanently "pending" and carry no information. sa.Column("submitted_at", sa.BigInteger(), nullable=False), - sa.Column("updated_at", sa.BigInteger(), nullable=False), sa.PrimaryKeyConstraint( "user_id", "doc_id", "doc_type", "etag", name="pk_batch_ocr_jobs" ), diff --git a/nextcloud_mcp_server/embedding/gateway_batch_client.py b/nextcloud_mcp_server/embedding/gateway_batch_client.py index 9991ee98..bb44143b 100644 --- a/nextcloud_mcp_server/embedding/gateway_batch_client.py +++ b/nextcloud_mcp_server/embedding/gateway_batch_client.py @@ -135,7 +135,14 @@ class GatewayBatchOcrClient: async def poll(self, job_id: str) -> BatchPollResult: """Poll a batch job. Raises on transport / non-2xx; maps a terminal job's - single-document result into :class:`BatchPollResult`.""" + single-document result into :class:`BatchPollResult`. + + ``job_id`` is the gateway's namespaced id (``/``), + so it embeds a ``/`` and the request path is multi-segment + (``/v1/ocr/batch/mistral/job-1``). The gateway declares this route with a + path-capture parameter (``GET /v1/ocr/batch/{job_id:path}``) so the slash + is captured whole — a plain single-segment ``{job_id}`` would 404 here. + """ async with httpx.AsyncClient( timeout=httpx.Timeout( _BATCH_REQUEST_TIMEOUT_SECONDS, connect=_BATCH_CONNECT_TIMEOUT_SECONDS @@ -146,7 +153,15 @@ class GatewayBatchOcrClient: ) resp.raise_for_status() body = resp.json() - status = body.get("status", _PENDING) + status = body.get("status") + if status is None: + # A well-formed gateway response always carries status. A 2xx without + # it is a contract violation: fail fast rather than silently treating + # it as pending and re-polling until the deadline. + logger.warning("gateway batch poll returned no status: %r", body) + return BatchPollResult( + status=_FAILED, pages=[], error="gateway returned no status" + ) if status != _SUCCEEDED: # pending: nothing to read yet. failed: surface the job-level error. return BatchPollResult(status=status, pages=[], error=body.get("error")) @@ -171,5 +186,9 @@ def _result_from_success(body: dict[str, Any]) -> BatchPollResult: item = results[0] if item.get("error") is not None or item.get("pages") is None: return BatchPollResult(status=_FAILED, pages=[], error=item.get("error")) - pages = [(p["index"], p.get("markdown", "")) for p in item["pages"]] + # Defensive on both fields (the page index falls back to position) so a + # malformed page object degrades rather than raising KeyError mid-parse. + pages = [ + (p.get("index", i), p.get("markdown", "")) for i, p in enumerate(item["pages"]) + ] return BatchPollResult(status=_SUCCEEDED, pages=pages) diff --git a/nextcloud_mcp_server/vector/batch_ocr_store.py b/nextcloud_mcp_server/vector/batch_ocr_store.py index 97867295..98e8382a 100644 --- a/nextcloud_mcp_server/vector/batch_ocr_store.py +++ b/nextcloud_mcp_server/vector/batch_ocr_store.py @@ -28,10 +28,11 @@ logger = logging.getLogger(__name__) @dataclass(frozen=True) class BatchOcrJob: - """A tracked in-flight batch OCR job.""" + """A tracked in-flight batch OCR job. A row exists only while pending (terminal + jobs are deleted), so there's no stored status — the live status comes from a + fresh ``GatewayBatchOcrClient.poll``. ``submitted_at`` anchors the deadline.""" job_id: str - status: str submitted_at: int @@ -67,14 +68,14 @@ class BatchOcrJobStore: """The in-flight job for this document+version, or ``None``.""" async with self._storage.acquire() as db: async with db.execute( - "SELECT job_id, status, submitted_at FROM batch_ocr_jobs " + "SELECT job_id, submitted_at FROM batch_ocr_jobs " "WHERE user_id = ? AND doc_id = ? AND doc_type = ? AND etag = ?", (user_id, doc_id, doc_type, etag), ) as cursor: row = await cursor.fetchone() if row is None: return None - return BatchOcrJob(job_id=row[0], status=row[1], submitted_at=int(row[2])) + return BatchOcrJob(job_id=row[0], submitted_at=int(row[1])) async def insert_pending( self, @@ -93,10 +94,10 @@ class BatchOcrJobStore: async with self._storage.acquire() as db: await db.execute( "INSERT INTO batch_ocr_jobs " - "(user_id, doc_id, doc_type, etag, job_id, status, submitted_at, updated_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?) " + "(user_id, doc_id, doc_type, etag, job_id, submitted_at) " + "VALUES (?, ?, ?, ?, ?, ?) " "ON CONFLICT (user_id, doc_id, doc_type, etag) DO NOTHING", - (user_id, doc_id, doc_type, etag, job_id, "pending", now, now), + (user_id, doc_id, doc_type, etag, job_id, now), ) await db.commit() diff --git a/tests/unit/test_batch_ocr_store.py b/tests/unit/test_batch_ocr_store.py index f161ec06..c7cc536d 100644 --- a/tests/unit/test_batch_ocr_store.py +++ b/tests/unit/test_batch_ocr_store.py @@ -35,7 +35,6 @@ async def test_insert_then_get(store): job = await store.get(**_DOC) assert job is not None assert job.job_id == "mistral/j1" - assert job.status == "pending" assert job.submitted_at > 0 diff --git a/tests/unit/test_gateway_batch_client.py b/tests/unit/test_gateway_batch_client.py index 2e2fd45b..a9bb64cd 100644 --- a/tests/unit/test_gateway_batch_client.py +++ b/tests/unit/test_gateway_batch_client.py @@ -79,6 +79,22 @@ async def test_submit_sends_bearer_when_token_provider(monkeypatch): assert seen[0].headers["Authorization"] == "Bearer tok-abc" +async def test_submit_raises_on_missing_job_id(monkeypatch): + # A 2xx with no job_id is a gateway contract violation -> actionable error. + _patch_transport(monkeypatch, lambda r: httpx.Response(202, json={})) + with pytest.raises(ValueError, match="no job_id"): + await gbc.GatewayBatchOcrClient("https://gw", "m").submit( + b"x", "application/pdf", custom_id="d" + ) + + +async def test_poll_missing_status_is_failed(monkeypatch): + # A 2xx body without a status field must fail fast, not poll forever. + _patch_transport(monkeypatch, lambda r: httpx.Response(200, json={"total": 1})) + result = await gbc.GatewayBatchOcrClient("https://gw", "m").poll("mistral/j") + assert result.is_failed + + async def test_poll_pending(monkeypatch): _patch_transport( monkeypatch,