From 703ea7c0a4b2607a695433c44595bc71e914c99f Mon Sep 17 00:00:00 2001 From: Adityavardhan Agrawal Date: Tue, 13 Oct 2026 14:42:59 -0700 Subject: [PATCH] Persist use_colpali on the document record so requeue keeps it /ingest/requeue resolves use_colpali from the request, then from the document's system_metadata, and otherwise falls back to False. The flag was only written to system_metadata when ingestion completed, so a document that failed before ever completing was silently requeued without ColPali when the requeue request omitted the flag. Store the requested flag when the document record is created and when its content is replaced, so requeue keeps the original mode. Co-Authored-By: Claude Opus 5.5 --- core/services/ingestion_service.py | 3 + .../test_ingestion_service_metadata_update.py | 59 +++++++++++++++++++ 2 files changed, 62 insertions(+) diff --git a/core/services/ingestion_service.py b/core/services/ingestion_service.py index 5e5394a7..53e713f9 100644 --- a/core/services/ingestion_service.py +++ b/core/services/ingestion_service.py @@ -629,6 +629,8 @@ async def _ingest_file_content_locked( ) doc.system_metadata = self._reset_processing_metadata(doc.system_metadata) doc.system_metadata["ingestion_revision"] = 0 + # Persist the requested mode so a requeue of a never-completed document keeps it. + doc.system_metadata["use_colpali"] = bool(use_colpali) await self._verify_ingest_and_storage_limits(auth, len(file_content_bytes), doc.external_id) @@ -844,6 +846,7 @@ async def _queue_document_update_locked( doc.system_metadata = self._reset_processing_metadata(doc.system_metadata) doc.system_metadata["ingestion_revision"] = int(doc.system_metadata.get("ingestion_revision", 0)) + 1 + doc.system_metadata["use_colpali"] = bool(use_colpali) updates = { "metadata": doc.metadata, diff --git a/core/tests/unit/test_ingestion_service_metadata_update.py b/core/tests/unit/test_ingestion_service_metadata_update.py index 00b44e00..068661c7 100644 --- a/core/tests/unit/test_ingestion_service_metadata_update.py +++ b/core/tests/unit/test_ingestion_service_metadata_update.py @@ -301,3 +301,62 @@ async def no_stored_size(*args, **kwargs): assert queued["payload"]["file_key"] == "ingest_uploads/replacement/report.txt" assert queued["payload"]["_job_id"] == "ingest:doc-1:1" assert queued["payload"]["ingestion_revision"] == 1 + + +@pytest.mark.asyncio +async def test_file_ingest_persists_use_colpali_on_initial_record(): + doc = _document() + db = FakeStoreFailureDatabase(doc) + service = IngestionService(db, None, None, None, None) + + async def noop_limit_check(auth, content_length, document_id): + return None + + service._verify_ingest_and_storage_limits = noop_limit_check + service._resolve_content_type = lambda content, filename, content_type: "text/plain" + + with pytest.raises(HTTPException): + await service.ingest_file_content( + file_content_bytes=b"hello", + filename="report.txt", + content_type="text/plain", + metadata={}, + auth=_auth(), + redis=None, + metadata_types={}, + use_colpali=True, + external_id="doc-1", + ) + + # A requeue of a document that never completed reads this flag back. + assert db.store_calls[0]["document"].system_metadata["use_colpali"] is True + + +@pytest.mark.asyncio +async def test_queued_file_update_persists_use_colpali(): + doc = _document() + service, db = _service(doc) + + async def noop_limit_check(auth, content_length, document_id): + return None + + async def fake_upload(*, content_bytes, filename, content_type): + return "app-1", "ingest_uploads/replacement/report.txt", "report.txt" + + async def noop_record(*args, **kwargs): + return None + + service._verify_ingest_and_storage_limits = noop_limit_check + service._upload_content_bytes = fake_upload + service._record_storage_usage = noop_record + service._get_storage_object_size = noop_record + + await service.queue_document_update( + document_id="doc-1", + auth=_auth(), + redis=FakeRedis(), + content="replacement text", + use_colpali=True, + ) + + assert db.update_calls[-1]["updates"]["system_metadata"]["use_colpali"] is True