diff --git a/backend/services/projects.py b/backend/services/projects.py index d8b0a1a..d0c6894 100644 --- a/backend/services/projects.py +++ b/backend/services/projects.py @@ -308,46 +308,68 @@ def mark_stale_running( def heal_if_pipeline_finished( storage: StorageBackend, user_id: str, project_id: str, ) -> ProjectMeta | None: - """If meta says queued/running but events already ended with - ``pipeline_complete``, flip status to ``complete``. + """Unstick analysis ``queued``/``running`` when the worker is gone. - Covers zombies where the worker wrote the terminal event (and often - the report) then died before the meta transition — e.g. container - rebuild mid-shutdown. Returns updated meta, or ``None`` if no heal. + - Last event ``pipeline_complete`` → ``complete`` + - Dead worker + ``report.json`` → ``complete`` + - Dead worker otherwise → ``error`` (so PCB/placement can start) """ meta = get_project(storage, user_id, project_id) if meta is None or meta.status not in (STATUS_RUNNING, STATUS_QUEUED): return None - events_prefix = f"{_project_prefix(user_id, project_id)}/events/" + prefix = _project_prefix(user_id, project_id) + events_prefix = f"{prefix}/events/" + last = None try: - keys = storage.list_prefix(events_prefix) + event_keys = sorted( + k for k in storage.list_prefix(events_prefix) + if k.endswith(".json") and "/events/" in k + ) + if event_keys: + last = storage.read_json(event_keys[-1]) except Exception: - return None - event_keys = sorted( - k for k in keys if k.endswith(".json") and "/events/" in k - ) - if not event_keys: - return None + last = None + + if (last or {}).get("event") == "pipeline_complete": + summary = (last.get("data") or {}).get("summary") + try: + return transition_status( + storage, user_id, project_id, + from_status={STATUS_RUNNING, STATUS_QUEUED}, + to_status=STATUS_COMPLETE, + summary=summary if isinstance(summary, dict) else meta.summary, + cancel_requested=False, + pipeline_state=None, + ) + except StatusConflict: + return None + + from backend.services import job_runner + + exec_name = meta.execution_name or f"local/projects/{project_id}" try: - last = storage.read_json(event_keys[-1]) + state = job_runner.get_execution_state(exec_name) except Exception: - return None - if (last or {}).get("event") != "pipeline_complete": + state = "unknown" + if state in ("pending", "running"): return None - summary = (last.get("data") or {}).get("summary") - try: - return transition_status( - storage, user_id, project_id, - from_status={STATUS_RUNNING, STATUS_QUEUED}, - to_status=STATUS_COMPLETE, - summary=summary if isinstance(summary, dict) else meta.summary, - cancel_requested=False, - pipeline_state=None, - ) - except StatusConflict: - return None + if storage.exists(f"{prefix}/report.json"): + try: + return transition_status( + storage, user_id, project_id, + from_status={STATUS_RUNNING, STATUS_QUEUED}, + to_status=STATUS_COMPLETE, + cancel_requested=False, + pipeline_state=None, + ) + except StatusConflict: + return None + return mark_stale_running( + storage, user_id, project_id, + f"Analysis worker terminated ({state})", + ) def heal_if_placement_stuck( diff --git a/tests/test_pcb_review.py b/tests/test_pcb_review.py index e6c3c5e..835b134 100644 --- a/tests/test_pcb_review.py +++ b/tests/test_pcb_review.py @@ -170,7 +170,27 @@ def test_pcb_busy_helpers(): assert draft.status in analysis -def test_pcb_start_requires_board(tmp_path: Path): +def test_heal_dead_analysis_worker_unblocks_pcb(tmp_path: Path, monkeypatch): + from backend.services import job_runner, projects as proj_svc + from backend.services.storage import LocalStorageBackend + + storage = LocalStorageBackend(tmp_path) + meta = proj_svc.create_project(storage, "usr_jwt", "board") + proj_svc.update_project( + storage, "usr_jwt", meta.id, + status=STATUS_RUNNING, + has_pcb=True, + has_bom=True, + has_netlist=True, + pcb_status="draft", + ) + monkeypatch.setattr(job_runner, "get_execution_state", lambda _n: "failed") + healed = proj_svc.heal_if_pipeline_finished(storage, "usr_jwt", meta.id) + assert healed is not None + assert healed.status == "error" + from backend.services.pcb_pipeline import analysis_busy + + assert not analysis_busy(healed) from fastapi.testclient import TestClient from backend.main import app from backend.services.storage import LocalStorageBackend