From e5e8c42966c46ae24ec5c0a4ad2f425aadc0c82a Mon Sep 17 00:00:00 2001
From: Michele Bigi <mikbigi@gmail.com>
Date: Fri, 28 Aug 2026 03:10:16 +0200
Subject: [PATCH] Let Reprocess stop a live or stuck running pipeline, then
 enqueue.

A docker rebuild can leave status=running with no worker; the old 409 blocked retry.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 backend/routers/pipeline.py | 49 ++++++++++++++++++++++++++++++++++---
 tests/test_reprocess.py     | 31 +++++++++++++++++------
 2 files changed, 69 insertions(+), 11 deletions(-)

diff --git a/backend/routers/pipeline.py b/backend/routers/pipeline.py
index 626e072..fd77ebd 100644
--- a/backend/routers/pipeline.py
+++ b/backend/routers/pipeline.py
@@ -204,11 +204,20 @@ async def reprocess(project_id: str, request: Request, req: ReprocessRequest | N
     owner_user_id, meta = await resolve_or_404(request, project_id)
     if not meta.has_bom or not meta.has_netlist:
         raise HTTPException(400, "Upload BOM and netlist before reprocessing")
-    if meta.status not in _REPROCESS_OK_FROM:
+
+    if _project_active(meta):
+        await _interrupt_active_pipeline(storage, owner_user_id, project_id)
+        meta = proj_svc.get_project(storage, owner_user_id, project_id) or meta
+
+    if meta.status == proj_svc.STATUS_PAUSED:
+        allowed = _REPROCESS_OK_FROM | {proj_svc.STATUS_PAUSED}
+    else:
+        allowed = _REPROCESS_OK_FROM
+
+    if meta.status not in allowed and not _project_active(meta):
         raise HTTPException(
             409,
-            f"Reprocess is for complete / error / cancelled runs "
-            f"(status={meta.status}). Pause uses resume; a live run uses cancel.",
+            f"Cannot reprocess from status={meta.status}.",
         )
 
     body = req or ReprocessRequest()
@@ -221,7 +230,9 @@ async def reprocess(project_id: str, request: Request, req: ReprocessRequest | N
     try:
         proj_svc.transition_status(
             storage, owner_user_id, project_id,
-            from_status=_REPROCESS_OK_FROM,
+            from_status=allowed | {
+                proj_svc.STATUS_QUEUED, proj_svc.STATUS_RUNNING,
+            },
             to_status=proj_svc.STATUS_QUEUED,
             cancel_requested=False,
             execution_name=None,
@@ -480,6 +491,36 @@ async def status(project_id: str, request: Request):
 # ---------------------------------------------------------------------------
 
 
+async def _interrupt_active_pipeline(
+    storage, user_id: str, project_id: str,
+) -> None:
+    """Cancel a queued/running worker so a new run can be enqueued.
+
+    If the worker is already dead (docker rebuild, OOM) the status can
+    stay ``running``; force it to cancelled after a short wait.
+    """
+    proj_svc.request_cancel(storage, user_id, project_id)
+    await _await_terminal(storage, user_id, project_id, timeout_s=10.0)
+    meta = proj_svc.get_project(storage, user_id, project_id)
+    if meta is None:
+        return
+    if _project_active(meta) and meta.execution_name:
+        job_runner.cancel_execution(meta.execution_name)
+        await _await_terminal(storage, user_id, project_id, timeout_s=5.0)
+        meta = proj_svc.get_project(storage, user_id, project_id) or meta
+    if _project_active(meta):
+        try:
+            proj_svc.transition_status(
+                storage, user_id, project_id,
+                from_status={proj_svc.STATUS_RUNNING, proj_svc.STATUS_QUEUED},
+                to_status=proj_svc.STATUS_CANCELLED,
+                pipeline_state={"error": "Superseded by reprocess"},
+                cancel_requested=False,
+            )
+        except proj_svc.StatusConflict:
+            pass
+
+
 async def _await_terminal(
     storage, user_id: str, project_id: str, *, timeout_s: float,
 ) -> None:
diff --git a/tests/test_reprocess.py b/tests/test_reprocess.py
index c6970a2..15d60b6 100644
--- a/tests/test_reprocess.py
+++ b/tests/test_reprocess.py
@@ -96,17 +96,34 @@ def test_reprocess_all_clears_kept_refs(tmp_path, monkeypatch):
     assert fresh.completed_review_refs == []
 
 
-def test_reprocess_rejects_running(tmp_path, monkeypatch):
-    monkeypatch.setattr(
-        "backend.services.job_runner.enqueue_pipeline",
-        lambda *a, **k: "x",
-    )
+def test_reprocess_stops_running_then_enqueues(tmp_path, monkeypatch):
+    captured: dict = {}
+
+    async def fake_await(*a, **k):
+        return None
+
+    monkeypatch.setattr("backend.routers.pipeline._await_terminal", fake_await)
+    monkeypatch.setattr("backend.services.job_runner.cancel_execution", lambda *a, **k: None)
+
+    def fake_enqueue(project_id, user_id, *, resume=False, free=False):
+        captured["resume"] = resume
+        return "local/projects/x"
+
+    monkeypatch.setattr("backend.services.job_runner.enqueue_pipeline", fake_enqueue)
+
     client = _client(tmp_path)
     meta = client.post("/api/projects", json={"name": "board"}).json()
     pid = meta["id"]
     storage = client.app.state.storage
     proj_svc.update_project(
-        storage, "local", pid, status="running", has_bom=True, has_netlist=True,
+        storage, "local", pid,
+        status="running",
+        has_bom=True,
+        has_netlist=True,
+        completed_review_refs=["U1"],
     )
     resp = client.post(f"/api/pipeline/{pid}/reprocess", json={"mode": "failed"})
-    assert resp.status_code == 409
+    assert resp.status_code == 202, resp.text
+    assert captured["resume"] is True
+    fresh = proj_svc.get_project(storage, "local", pid)
+    assert fresh.status == "queued"