Department change on compiled source wedges at 'processing' (map-reduce no-ops on commit-phase guard)

[apjaphet]

Summary

Reassigning an already-compiled source to a different department wedges it at processing / 56% permanently, with zero pages written into the new scope.

Root cause

The PATCH /sources/{id} department-change handler sets status=processing and enqueues ingest_map_reduce_task, but never resets pipeline_phase (still commit from the original run). run_mrp_pipeline hits its resume guard — "already in commit phase, skipping" — and returns in ~0.04s doing nothing. The source is left stuck at processing forever.

Repro

1. Ingest a document to completion (ready, pipeline_phase=commit).
2. Change its department assignment via the source edit dialog.
3. Source flips to processing/56% and never completes; no pages appear in the new department scope.

Fix applied locally

Route the re-ingest by pipeline phase, matching what /sources/{id}/retry already does: a source past plan_review (refine/verify/commit) is re-queued via ingest_refine_task, which resumes at COMMIT and rewrites the persisted page drafts into the new scope. Otherwise use ingest_map_reduce_task.

Verified: the wedged source resumed via ingest_refine_task, loaded 9 persisted drafts, re-verified, and wrote +9 pages into the new (Ops) department scope; source is now ready/100%.

Diff

diff --git a/app/routers/sources.py b/app/routers/sources.py
index 354495f..e764efc 100644
--- a/app/routers/sources.py
+++ b/app/routers/sources.py
@@ -545,7 +545,17 @@ async def update_source(
         await db.flush()
 
         pool = await get_arq_pool()
-        job = await pool.enqueue_job("ingest_map_reduce_task", str(source_id))
+        # Route by pipeline phase. A source that has already been compiled is
+        # past plan_review (phase refine/verify/commit); enqueuing
+        # ingest_map_reduce_task would hit the resume guard in run_mrp_pipeline
+        # ("already in commit phase, skipping") and no-op, leaving the source
+        # stuck at 'processing'. ingest_refine_task resumes at COMMIT and
+        # rewrites the persisted page drafts into the new scope.
+        if source.pipeline_phase in ("refine", "verify", "commit"):
+            reingest_task = "ingest_refine_task"
+        else:
+            reingest_task = "ingest_map_reduce_task"
+        job = await pool.enqueue_job(reingest_task, str(source_id))
         if job:
             source.job_id = job.job_id