From 96e2590a3ccccdf234d41250eea9a82ed3084469 Mon Sep 17 00:00:00 2001
From: Michele Bigi <mikbigi@gmail.com>
Date: Fri, 28 Aug 2026 08:50:33 +0200
Subject: [PATCH] Detect a dead local pipeline worker after API restart, and
 flush SSE.

A docker rebuild left HubAudio status=running with no process, so the UI sat on a silent event stream. Persist the worker pid and disable proxy buffering on the SSE response.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 backend/routers/pipeline.py    | 10 ++++++-
 backend/services/job_runner.py | 54 +++++++++++++++++++++++++++-------
 tests/test_job_runner.py       | 11 +++++++
 3 files changed, 63 insertions(+), 12 deletions(-)
 create mode 100644 tests/test_job_runner.py

diff --git a/backend/routers/pipeline.py b/backend/routers/pipeline.py
index fd77ebd..fc718b2 100644
--- a/backend/routers/pipeline.py
+++ b/backend/routers/pipeline.py
@@ -471,7 +471,15 @@ async def events(project_id: str, request: Request):
             except (asyncio.CancelledError, Exception):
                 pass
 
-    return EventSourceResponse(event_generator())
+    return EventSourceResponse(
+        event_generator(),
+        ping=15,
+        headers={
+            "Cache-Control": "no-cache, no-transform",
+            "X-Accel-Buffering": "no",
+            "Connection": "keep-alive",
+        },
+    )
 
 
 @router.get("/pipeline/{project_id}/status")
diff --git a/backend/services/job_runner.py b/backend/services/job_runner.py
index 9545915..655cd36 100644
--- a/backend/services/job_runner.py
+++ b/backend/services/job_runner.py
@@ -16,6 +16,7 @@ import os
 import subprocess
 import sys
 import threading
+from pathlib import Path
 from typing import Literal
 
 from backend.config import settings
@@ -71,9 +72,9 @@ def _spawn_local_subprocess(
     proc = subprocess.Popen(
         [sys.executable, "-m", "backend.pipeline_worker"],
         env=env,
-        # Inherit stdout/stderr so logs appear in the dev terminal
         stdin=subprocess.DEVNULL,
     )
+    _write_pid(project_id, proc.pid)
     with _local_procs_lock:
         # Reap any old proc for the same project before tracking the new one.
         prior = _local_procs.pop(project_id, None)
@@ -87,20 +88,51 @@ def _spawn_local_subprocess(
     return name
 
 
+def _pid_path(project_id: str) -> Path:
+    return settings.data_dir / "workers" / f"{project_id}.pid"
+
+
+def _write_pid(project_id: str, pid: int) -> None:
+    path = _pid_path(project_id)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(str(pid))
+
+
+def _pid_alive(project_id: str) -> bool | None:
+    """True/False if a pid file exists; None if there is no file."""
+    path = _pid_path(project_id)
+    if not path.is_file():
+        return None
+    try:
+        pid = int(path.read_text().strip())
+    except ValueError:
+        return False
+    try:
+        os.kill(pid, 0)
+    except OSError:
+        return False
+    return True
+
+
 def _local_state(project_id: str) -> ExecutionState:
     with _local_procs_lock:
         proc = _local_procs.get(project_id)
-    if proc is None:
-        return "unknown"
-    rc = proc.poll()
-    if rc is None:
+    if proc is not None:
+        rc = proc.poll()
+        if rc is None:
+            return "running"
+        if rc == 0:
+            return "succeeded"
+        if rc < 0:
+            # Negative return = terminated by signal
+            return "cancelled"
+        return "failed"
+    alive = _pid_alive(project_id)
+    if alive is True:
         return "running"
-    if rc == 0:
-        return "succeeded"
-    if rc < 0:
-        # Negative return = terminated by signal
-        return "cancelled"
-    return "failed"
+    if alive is False:
+        return "failed"
+    return "unknown"
 
 
 def _local_cancel(project_id: str) -> None:
diff --git a/tests/test_job_runner.py b/tests/test_job_runner.py
new file mode 100644
index 0000000..3047807
--- /dev/null
+++ b/tests/test_job_runner.py
@@ -0,0 +1,11 @@
+"""Local worker pid file so a restarted API can see a dead subprocess."""
+
+from backend.services import job_runner
+
+
+def test_local_state_failed_when_pid_file_is_stale(tmp_path, monkeypatch):
+    monkeypatch.setattr(job_runner.settings, "data_dir", tmp_path)
+    pid_path = tmp_path / "workers" / "proj.pid"
+    pid_path.parent.mkdir(parents=True)
+    pid_path.write_text("99999999")
+    assert job_runner.get_execution_state("local/projects/proj") == "failed"