feat(pipeline): retry separation on CPU when a GPU attempt fails

app/core/models.py

@@ -63,6 +63,10 @@ class Job:
     # Device the separation actually ran on ("cuda" / "mps" / "cpu"), recorded
     # per job for diagnostics -- settings may change between jobs.
     compute_device: str | None = None
+    # True when a GPU separation attempt failed and the job completed on the
+    # CPU fallback (#276) -- kept loud in state/metadata so the fallback is
+    # never silent (the #247 lesson).
+    gpu_fallback: bool = False
     # Wall-clock seconds per pipeline stage ({"download": 12.3, ...}); written
     # to metadata.json and the one-line completion summary in the log.
     stage_timings: dict[str, float] | None = None
@@ -101,6 +105,7 @@ def to_state(self) -> dict[str, Any]:
             "error": self.error,
             "error_detail": self.error_detail,
             "compute_device": self.compute_device,
+            "gpu_fallback": self.gpu_fallback,
             "stage_timings": self.stage_timings,
             "created_at": self.created_at,
         }

app/pipeline/runner.py

@@ -208,6 +208,7 @@ def _write_metadata(job: Job, job_dir: Path) -> None:
         "tags": job.tags,
         "has_video": job.has_video,
         "compute_device": job.compute_device,
+        "gpu_fallback": job.gpu_fallback,
         "stage_timings": job.stage_timings,
     }
     try:

app/pipeline/separate.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import re
+import shutil
 import subprocess
 import sys
 import threading
@@ -13,7 +14,7 @@
 from app.core.models import Job, JobCancelled, _set
 from app.core.registry import set_proc
 from app.core.settings import get_demucs_device
-from app.pipeline.errors import SeparationError
+from app.pipeline.errors import SeparationError, classify_failure
 
 logger = logging.getLogger("stemdeck.pipeline")
 
@@ -23,16 +24,10 @@
 # while still catching genuine hangs (GPU deadlock, OOM stall, etc.).
 
 
-def separate(job: Job, source: Path, job_dir: Path) -> Path:
-    _set(job, status="separating", progress=0.0, stage="Separating stems...")
-
-    # Read the device fresh per job (not a frozen import) so a Settings change
-    # applies to the next separation without a restart. Recorded on the job for
-    # the completion summary / metadata / failure quarantine.
-    device = get_demucs_device()
-    job.compute_device = device
-    logger.info("[%s] separating on device=%s", job.id, device)
-    cmd = [
+def _demucs_cmd(device: str, source: Path, job_dir: Path) -> list[str]:
+    """Build the demucs CLI invocation. Module-level seam so tests can swap
+    in a stub executable without touching the process-management machinery."""
+    return [
         sys.executable,
         "-m",
         "demucs",
@@ -44,6 +39,13 @@ def separate(job: Job, source: Path, job_dir: Path) -> Path:
         str(job_dir),
         str(source),
     ]
+
+
+def _run_demucs(job: Job, source: Path, job_dir: Path, device: str) -> tuple[int, list[str]]:
+    """One demucs attempt on `device`: spawn, stream progress, watchdog stalls.
+
+    Returns (returncode, stderr_tail). Raises JobCancelled when the exit was
+    caused by POST /cancel. The retry policy lives in separate()."""
     env = os.environ.copy()
     try:
         import certifi
@@ -54,7 +56,7 @@ def separate(job: Job, source: Path, job_dir: Path) -> Path:
         pass
 
     proc = subprocess.Popen(
-        cmd,
+        _demucs_cmd(device, source, job_dir),
         stdout=subprocess.DEVNULL,
         stderr=subprocess.PIPE,
         text=True,
@@ -123,15 +125,69 @@ def _watchdog() -> None:
     # that into JobCancelled before the generic "demucs failed" path.
     if job.cancel_requested:
         raise JobCancelled()
-    if proc.returncode != 0:
+    return proc.returncode, tail
+
+
+def separate(job: Job, source: Path, job_dir: Path) -> Path:
+    """Run demucs on the configured device, falling back to CPU once when a
+    GPU attempt fails (#276).
+
+    The fallback is deliberately loud, never silent (the #247 lesson): the
+    stage line says so while it runs, the WARNING log carries the full stderr
+    tail, and gpu_fallback/compute_device persist to job state and metadata.
+    It applies even when the user forced cuda/mps in Settings -- a dead job
+    with no diagnostics is strictly worse for them than a slow one that
+    explains itself."""
+    _set(job, status="separating", progress=0.0, stage="Separating stems...")
+
+    # Read the device fresh per job (not a frozen import) so a Settings change
+    # applies to the next separation without a restart. Recorded on the job for
+    # the completion summary / metadata / failure quarantine.
+    device = get_demucs_device()
+    job.compute_device = device
+    logger.info("[%s] separating on device=%s", job.id, device)
+
+    rc, tail = _run_demucs(job, source, job_dir, device)
+
+    if rc != 0 and device != "cpu":
+        cause = classify_failure("\n".join(tail))
+        logger.warning(
+            "[%s] demucs failed on %s (exit %s, cause=%s); retrying on CPU. tail:\n%s",
+            job.id,
+            device,
+            rc,
+            cause,
+            "\n".join(tail[-15:]) or "(no stderr captured)",
+        )
+        # Partial output from the failed attempt must not be mistaken for
+        # results by collect(); CPU restarts from scratch, so does progress.
+        shutil.rmtree(job_dir / DEMUCS_MODEL, ignore_errors=True)
+        _set(job, progress=0.0, stage="GPU failed — retrying on CPU (slower)...")
+        job.gpu_fallback = True
+        job.compute_device = f"cpu (fallback from {device})"
+        first_tail = tail
+        rc, tail = _run_demucs(job, source, job_dir, "cpu")
+        if rc != 0:
+            combined = [
+                f"--- attempt on {device} ---",
+                *first_tail[-20:],
+                "--- cpu fallback attempt ---",
+                *tail[-20:],
+            ]
+            last = tail[-1] if tail else f"exit status {rc}"
+            logger.error("[%s] cpu fallback also failed (exit %s)", job.id, rc)
+            raise SeparationError(
+                f"demucs failed: {last}", tail=combined, device=f"{device}, then cpu"
+            )
+    elif rc != 0:
         detail = "\n".join(tail[-15:]) if tail else "(no stderr captured)"
-        logger.error("[%s] demucs exited %s; tail:\n%s", job.id, proc.returncode, detail)
-        last = tail[-1] if tail else f"exit status {proc.returncode}"
+        logger.error("[%s] demucs exited %s; tail:\n%s", job.id, rc, detail)
+        last = tail[-1] if tail else f"exit status {rc}"
         # SeparationError carries the stderr tail + device so the runner's
         # failure quarantine can preserve the evidence (#277).
         raise SeparationError(f"demucs failed: {last}", tail=tail[-40:], device=device)
 
     stems_root = job_dir / DEMUCS_MODEL / source.stem
     if not stems_root.is_dir():
-        raise SeparationError(f"demucs output not found at {stems_root}", device=device)
+        raise SeparationError(f"demucs output not found at {stems_root}", device=job.compute_device)
     return stems_root

tests/test_separate_fallback.py

@@ -0,0 +1,159 @@
+"""Tests for the GPU->CPU separation fallback (#276).
+
+The demucs invocation is swapped for stub Python one-liners via the
+_demucs_cmd seam, so the real process machinery (Popen, stderr streaming,
+watchdog, cancel translation) runs end-to-end without demucs or a GPU.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+from app.core.models import Job, JobCancelled
+from app.pipeline import separate as sep_mod
+from app.pipeline.errors import SeparationError
+
+
+def _stub_cmds(fail_devices: set[str], calls: list[str]):
+    """A _demucs_cmd replacement: fails with a CUDA-OOM message on the given
+    devices, succeeds (writing a stem WAV where demucs would) elsewhere."""
+
+    def fake_cmd(device: str, source: Path, job_dir: Path) -> list[str]:
+        calls.append(device)
+        if device in fail_devices:
+            code = (
+                "import sys;"
+                " sys.stderr.write('RuntimeError: CUDA out of memory. Tried 2 GiB\\n');"
+                " sys.exit(1)"
+            )
+        else:
+            out_dir = job_dir / sep_mod.DEMUCS_MODEL / source.stem
+            code = (
+                "import os, sys;"
+                f" d = {str(out_dir)!r};"
+                " os.makedirs(d, exist_ok=True);"
+                " open(os.path.join(d, 'vocals.wav'), 'wb').write(b'RIFF');"
+                " sys.stderr.write('100%|separated\\n')"
+            )
+        return [sys.executable, "-c", code]
+
+    return fake_cmd
+
+
+@pytest.fixture()
+def job(tmp_path: Path):
+    j = Job(id="abcdefabc276")
+    (tmp_path / "source.wav").write_bytes(b"RIFF")
+    return j
+
+
+def test_gpu_failure_falls_back_to_cpu(job, tmp_path, monkeypatch, caplog):
+    import logging
+
+    calls: list[str] = []
+    monkeypatch.setattr(sep_mod, "get_demucs_device", lambda: "cuda")
+    monkeypatch.setattr(sep_mod, "_demucs_cmd", _stub_cmds({"cuda"}, calls))
+
+    with caplog.at_level(logging.WARNING, logger="stemdeck.pipeline"):
+        stems_root = sep_mod.separate(job, tmp_path / "source.wav", tmp_path)
+
+    assert calls == ["cuda", "cpu"]
+    assert (stems_root / "vocals.wav").is_file()
+    assert job.gpu_fallback is True
+    assert job.compute_device == "cpu (fallback from cuda)"
+    # Loud, never silent: the warning names device, cause, and stderr.
+    warning = next(r.message for r in caplog.records if "retrying on CPU" in r.message)
+    assert "cause=out-of-memory" in warning
+    assert "CUDA out of memory" in warning
+
+
+def test_gpu_success_needs_no_fallback(job, tmp_path, monkeypatch):
+    calls: list[str] = []
+    monkeypatch.setattr(sep_mod, "get_demucs_device", lambda: "cuda")
+    monkeypatch.setattr(sep_mod, "_demucs_cmd", _stub_cmds(set(), calls))
+
+    stems_root = sep_mod.separate(job, tmp_path / "source.wav", tmp_path)
+
+    assert calls == ["cuda"]
+    assert stems_root.is_dir()
+    assert job.gpu_fallback is False
+    assert job.compute_device == "cuda"
+
+
+def test_cpu_failure_does_not_retry(job, tmp_path, monkeypatch):
+    calls: list[str] = []
+    monkeypatch.setattr(sep_mod, "get_demucs_device", lambda: "cpu")
+    monkeypatch.setattr(sep_mod, "_demucs_cmd", _stub_cmds({"cpu"}, calls))
+
+    with pytest.raises(SeparationError) as exc_info:
+        sep_mod.separate(job, tmp_path / "source.wav", tmp_path)
+
+    assert calls == ["cpu"]  # exactly one attempt
+    assert job.gpu_fallback is False
+    assert exc_info.value.device == "cpu"
+
+
+def test_both_attempts_failing_raises_with_both_tails(job, tmp_path, monkeypatch):
+    calls: list[str] = []
+    monkeypatch.setattr(sep_mod, "get_demucs_device", lambda: "mps")
+    monkeypatch.setattr(sep_mod, "_demucs_cmd", _stub_cmds({"mps", "cpu"}, calls))
+
+    with pytest.raises(SeparationError) as exc_info:
+        sep_mod.separate(job, tmp_path / "source.wav", tmp_path)
+
+    assert calls == ["mps", "cpu"]
+    err = exc_info.value
+    assert err.device == "mps, then cpu"
+    # The quarantine's error.txt gets both attempts' evidence.
+    joined = "\n".join(err.tail)
+    assert "--- attempt on mps ---" in joined
+    assert "--- cpu fallback attempt ---" in joined
+
+
+def test_cancel_during_gpu_attempt_skips_fallback(job, tmp_path, monkeypatch):
+    calls: list[str] = []
+    monkeypatch.setattr(sep_mod, "get_demucs_device", lambda: "cuda")
+    monkeypatch.setattr(sep_mod, "_demucs_cmd", _stub_cmds({"cuda"}, calls))
+    job.cancel_requested = True  # POST /cancel arrived before/mid attempt
+
+    with pytest.raises(JobCancelled):
+        sep_mod.separate(job, tmp_path / "source.wav", tmp_path)
+
+    assert calls == ["cuda"]  # no CPU retry after a cancel
+
+
+def test_partial_gpu_output_cleared_before_retry(job, tmp_path, monkeypatch):
+    """A failed GPU attempt's partial stems must not leak into the CPU run."""
+    calls: list[str] = []
+    marker = tmp_path / sep_mod.DEMUCS_MODEL / "partial-garbage.wav"
+
+    def fake_cmd(device: str, source: Path, job_dir: Path) -> list[str]:
+        calls.append(device)
+        if device == "cuda":
+            # Simulate demucs dying after writing partial output.
+            code = (
+                "import os, sys;"
+                f" os.makedirs(os.path.dirname({str(marker)!r}), exist_ok=True);"
+                f" open({str(marker)!r}, 'wb').write(b'junk');"
+                " sys.stderr.write('RuntimeError: CUDA error\\n');"
+                " sys.exit(1)"
+            )
+            return [sys.executable, "-c", code]
+        out_dir = tmp_path / sep_mod.DEMUCS_MODEL / source.stem
+        code = (
+            "import os;"
+            f" os.makedirs({str(out_dir)!r}, exist_ok=True);"
+            f" open(os.path.join({str(out_dir)!r}, 'vocals.wav'), 'wb').write(b'RIFF')"
+        )
+        return [sys.executable, "-c", code]
+
+    monkeypatch.setattr(sep_mod, "get_demucs_device", lambda: "cuda")
+    monkeypatch.setattr(sep_mod, "_demucs_cmd", fake_cmd)
+
+    sep_mod.separate(job, tmp_path / "source.wav", tmp_path)
+
+    assert calls == ["cuda", "cpu"]
+    assert not marker.exists(), "partial GPU output must be cleared before the CPU retry"