Persist use_colpali on the document record so requeue keeps it

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,

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