Doc with images stranded at 55% when no vision provider is configured (caption task never chains to MRP)

[apjaphet]

Summary

A document with images gets stranded at 55% "Queuing compilation pipeline..." and never proceeds to compilation when no vision provider is configured.

Root cause

caption_images_task is what chains a source into the MRP pipeline (ingest_map_reduce_task), but only from its tail. When no vision provider is configured it early-returns before ever enqueuing the next task:

if not vision_provider:
    logger.info("caption_images_task: no vision provider configured, skipping")
    return   # <-- never chains into MRP

The if not image_records: return path has the same gap. Any doc with images is therefore stranded whenever vision is off. Text-only docs skip captioning entirely, so they were unaffected — which masked the bug.

Repro

1. Ensure no vision provider is configured.
2. Upload a PDF containing images.
3. Source reaches 55% "Queuing compilation pipeline..." and stays there indefinitely.

Fix applied locally

Added _chain_into_mrp(source_id) helper and called it on every early-return path in caption_images_task (no vision provider; no images) as well as the normal tail, so captioning always proceeds to compilation.

Verified: a stuck 12-image PDF, re-driven through the fixed path, advanced through MAP/REDUCE to plan_ready.

Diff

diff --git a/app/worker.py b/app/worker.py
index 3912059..a281504 100644
--- a/app/worker.py
+++ b/app/worker.py
@@ -65,6 +65,30 @@ async def enqueue_post_extraction_pipeline(source_id: str, has_images: bool) ->
     return job.job_id if job else None
 
 
+async def _chain_into_mrp(source_id: str) -> Optional[str]:
+    """Enqueue ingest_map_reduce_task and record the job_id on the source.
+
+    caption_images_task is responsible for chaining into MRP once captioning is
+    done. It has several early-return paths (no vision provider, no images) —
+    each MUST still chain, or a doc-with-images gets stranded at 55%
+    "Queuing compilation pipeline..." forever. Centralizing the hand-off here
+    guarantees every caption path proceeds to compilation.
+    """
+    from app.database import async_session_factory
+    from app.database.models import Source
+
+    pool = await get_arq_pool()
+    job = await pool.enqueue_job("ingest_map_reduce_task", source_id)
+    if job:
+        async with async_session_factory() as session:
+            source = await session.get(Source, uuid.UUID(source_id))
+            if source:
+                source.job_id = job.job_id
+                source.progress_message = "Extraction queued..."
+                await session.commit()
+    return job.job_id if job else None
+
+
 async def finalize_verbatim_source(session, source, tracker) -> dict:
     """Verbatim path: index raw chunks (no LLM) and mark the source ready.
 
@@ -1201,7 +1225,8 @@ async def caption_images_task(ctx: dict, source_id: str):
         registry = ProviderRegistry(session)
         vision_provider = await registry.get_vision()
         if not vision_provider:
-            logger.info("caption_images_task: no vision provider configured, skipping")
+            logger.info("caption_images_task: no vision provider configured, skipping to MRP")
+            await _chain_into_mrp(source_id)
             return
 
         rows = (await session.execute(
@@ -1212,6 +1237,7 @@ async def caption_images_task(ctx: dict, source_id: str):
         image_records = [(row.id, row.minio_key, row.content_type) for row in rows]
 
     if not image_records:
+        await _chain_into_mrp(source_id)
         return
 
     logger.info(f"caption_images_task: captioning {len(image_records)} images for {source_id}")
@@ -1278,15 +1304,7 @@ async def caption_images_task(ctx: dict, source_id: str):
                 logger.info(f"caption_images_task: refreshed full_text with {len(caption_by_id)} captions for {source_id}")
 
     # Chain into MAP-REDUCE (only now that captions are baked in).
-    pool = await get_arq_pool()
-    job = await pool.enqueue_job("ingest_map_reduce_task", source_id)
-    if job:
-        async with async_session_factory() as session:
-            source = await session.get(Source, sid)
-            if source:
-                source.job_id = job.job_id
-                source.progress_message = "Extraction queued..."
-                await session.commit()
+    await _chain_into_mrp(source_id)
     logger.info(f"caption_images_task: enqueued ingest_map_reduce_task for {source_id}")