feat(pipeline): failed-job quarantine + error classification + stage timings

app/core/config.py

@@ -79,6 +79,10 @@ def detect_torch_device() -> str:
 DEMUCS_MODEL = os.environ.get("STEMDECK_DEMUCS_MODEL", "htdemucs_6s").strip() or "htdemucs_6s"
 MAX_DURATION_SEC = max(60, _env_int("STEMDECK_MAX_DURATION_SEC", 1200))  # 20 min default
 JOB_TTL_SECONDS = max(300, _env_int("STEMDECK_JOB_TTL_SECONDS", 24 * 3600))  # 24 h default
+# TTL for quarantined failed-job dirs (jobs/failed/<id>, kept for diagnostics).
+# Swept unconditionally -- even deployments with a persistent library must not
+# accumulate failure evidence forever.
+FAILED_TTL_SECONDS = max(3600, _env_int("STEMDECK_FAILED_TTL_SECONDS", 7 * 24 * 3600))  # 7 d
 MAX_PENDING_JOBS = max(1, min(50, _env_int("STEMDECK_MAX_PENDING_JOBS", 3)))
 TIMEOUT_FFMPEG = _env_int("STEMDECK_TIMEOUT_FFMPEG", 300)
 TIMEOUT_ANALYZE = _env_int("STEMDECK_TIMEOUT_ANALYZE", 120)

app/core/models.py

@@ -56,6 +56,16 @@ class Job:
     # upload, enabling the "Export Mix (with video)" MP4 export.
     has_video: bool = False
     error: str | None = None
+    # Classified failure cause + last stderr line (e.g. "out-of-memory — ...").
+    # Shown by the UI as a secondary line under the generic error message so
+    # failures are actionable instead of uniformly opaque.
+    error_detail: str | None = None
+    # Device the separation actually ran on ("cuda" / "mps" / "cpu"), recorded
+    # per job for diagnostics -- settings may change between jobs.
+    compute_device: str | None = None
+    # 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
     # Set by POST /api/jobs/{id}/cancel; consumed by pipeline stages.
     # Not surfaced via to_state() -- it's internal control state.
     cancel_requested: bool = False
@@ -89,6 +99,9 @@ def to_state(self) -> dict[str, Any]:
             "source_url": self.source_url,
             "has_video": self.has_video,
             "error": self.error,
+            "error_detail": self.error_detail,
+            "compute_device": self.compute_device,
+            "stage_timings": self.stage_timings,
             "created_at": self.created_at,
         }
 

app/main.py

@@ -43,7 +43,7 @@
     set_port,
     set_video_max_height,
 )
-from app.pipeline.collect import sweep_old_jobs
+from app.pipeline.collect import sweep_failed_jobs, sweep_old_jobs
 
 # Show our INFO-level logs through uvicorn's root handler. Without this,
 # Python's default root level (WARNING) silently drops every
@@ -122,12 +122,17 @@ def _sweep_disabled() -> bool:
 
 
 async def _sweep_loop() -> None:
-    if _sweep_disabled():
+    # The job TTL sweep is disabled for persistent libraries, but the
+    # failed-job quarantine (jobs/failed/) expires unconditionally -- failure
+    # evidence is diagnostics, not library content, on every deployment.
+    persistent = _sweep_disabled()
+    if persistent:
         _log.info("job TTL sweep disabled (persistent library; user-managed)")
-        return
     while True:
         try:
-            await asyncio.to_thread(sweep_old_jobs, JOBS_DIR)
+            if not persistent:
+                await asyncio.to_thread(sweep_old_jobs, JOBS_DIR)
+            await asyncio.to_thread(sweep_failed_jobs, JOBS_DIR)
         except Exception:
             _log.warning("sweep failed", exc_info=True)
         await asyncio.sleep(3600)

app/pipeline/collect.py

@@ -12,6 +12,7 @@
 
 from app.core.config import (
     DEMUCS_MODEL,
+    FAILED_TTL_SECONDS,
     JOB_TTL_SECONDS,
     STEM_NAMES,
     TIMEOUT_FFMPEG,
@@ -238,6 +239,8 @@ def sweep_old_jobs(jobs_dir: Path) -> None:
     for d in jobs_dir.iterdir():
         if not d.is_dir():
             continue
+        if d.name == "failed":
+            continue  # the failure quarantine has its own TTL (sweep_failed_jobs)
         job = jobs.get(d.name)
         if job is not None:
             if job.status not in _TERMINAL:
@@ -251,3 +254,20 @@ def sweep_old_jobs(jobs_dir: Path) -> None:
         removed = True
     if removed:
         registry_persist(jobs_dir)
+
+
+def sweep_failed_jobs(jobs_dir: Path) -> None:
+    """Expire quarantined failure-evidence dirs (jobs/failed/<id>) older than
+    FAILED_TTL_SECONDS. Runs unconditionally from the hourly sweep loop --
+    unlike the job TTL sweep, which persistent-library deployments disable,
+    failure evidence must never accumulate forever (#277)."""
+    failed_root = jobs_dir / "failed"
+    if not failed_root.is_dir():
+        return
+    cutoff = time.time() - FAILED_TTL_SECONDS
+    for d in failed_root.iterdir():
+        try:
+            if d.is_dir() and d.stat().st_mtime < cutoff:
+                _rmtree(d)
+        except OSError:
+            logger.warning("failed-quarantine sweep could not stat %s", d, exc_info=True)

app/pipeline/errors.py

@@ -0,0 +1,88 @@
+"""Pipeline failure classification (#294) and the separation error type (#277).
+
+"Audio processing failed. Please try again." is shown for out-of-memory, a
+GPU fault, a bad input file, and a full disk alike. The classifier below maps
+the captured stderr/exception text to a small set of user-meaningful causes so
+the job can carry an actionable `error_detail` and the failed-job quarantine
+can record what actually happened. Phase 2's GPU->CPU fallback reuses it.
+"""
+
+from __future__ import annotations
+
+
+class SeparationError(RuntimeError):
+    """Demucs (or a later separation pass) failed.
+
+    Carries the stderr tail and the compute device so the runner's
+    quarantine can preserve the evidence that the old error path threw away.
+    """
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        tail: list[str] | None = None,
+        device: str | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.tail: list[str] = tail or []
+        self.device = device
+
+
+# Ordered, first match wins. Substring match against lowercased text.
+# Deliberately coarse: these route a user (or a bug report) to the right
+# next step, they don't diagnose. "unknown" is the honest default.
+_CAUSE_PATTERNS: tuple[tuple[str, tuple[str, ...]], ...] = (
+    (
+        "out-of-memory",
+        (
+            "cuda out of memory",
+            "mps backend out of memory",
+            "not enough memory",
+            "cannot allocate memory",
+            "out of memory",
+            "memoryerror",
+        ),
+    ),
+    (
+        "unsupported-device",
+        (
+            "no kernel image is available",
+            "invalid device",
+            "cuda driver version is insufficient",
+            "cudnn error",
+            "not currently implemented for the mps device",
+            "torch not compiled with cuda",
+        ),
+    ),
+    (
+        "disk-full",
+        (
+            "no space left on device",
+            "errno 28",
+            "disk quota exceeded",
+        ),
+    ),
+    (
+        "bad-input",
+        (
+            "invalid data found",
+            "could not open file",
+            "unable to open",
+            "no stems produced",
+            "failed to read",
+            "unsupported format",
+            "ffmpeg transcode failed",
+        ),
+    ),
+)
+
+
+def classify_failure(text: str) -> str:
+    """Map failure output to one of: out-of-memory, unsupported-device,
+    disk-full, bad-input, unknown."""
+    low = text.lower()
+    for cause, patterns in _CAUSE_PATTERNS:
+        if any(p in low for p in patterns):
+            return cause
+    return "unknown"

app/pipeline/runner.py

@@ -5,9 +5,11 @@
 import logging
 import shutil
 import subprocess
+import time
+from datetime import datetime, timezone
 from pathlib import Path
 
-from app.core.config import TIMEOUT_FFMPEG
+from app.core.config import DEMUCS_MODEL, TIMEOUT_FFMPEG
 from app.core.models import Job, JobCancelled, _set
 from app.core.registry import persist as persist_registry
 from app.pipeline.analyze import analyze, compute_stem_presence
@@ -19,6 +21,7 @@
     make_selected_mix,
 )
 from app.pipeline.download import download
+from app.pipeline.errors import classify_failure
 from app.pipeline.separate import separate
 
 logger = logging.getLogger("stemdeck.pipeline")
@@ -119,13 +122,27 @@ def _prepare_local_source(job: Job, source: Path, job_dir: Path) -> Path:
     return dest
 
 
+def _lap(job: Job, name: str, start: float) -> float:
+    """Record a stage duration in job.stage_timings; returns a new start mark.
+    Timings feed the one-line completion summary, metadata.json, and the
+    failure quarantine's error.txt (#293)."""
+    now = time.monotonic()
+    if job.stage_timings is None:
+        job.stage_timings = {}
+    job.stage_timings[name] = round(now - start, 1)
+    return now
+
+
 def _run_common(job: Job, source: Path, job_dir: Path) -> None:
     """Analyze → separate → collect → mix. Shared by both YouTube and local
     upload pipelines after their respective source acquisition steps."""
     _check_cancel(job)
+    mark = time.monotonic()
     analyze(job, source)
+    mark = _lap(job, "analyze", mark)
     _check_cancel(job)
     stems_root = separate(job, source, job_dir)
+    mark = _lap(job, "separate", mark)
     found = collect(job, stems_root, job_dir)
     stems_dir = job_dir / "stems"
     job.stem_presence = compute_stem_presence(stems_dir, found)
@@ -155,17 +172,22 @@ def _run_common(job: Job, source: Path, job_dir: Path) -> None:
     if mix_path is not None and mix_path.stem not in all_stem_names:
         all_stem_names.append(mix_path.stem)
     compute_stem_peaks(stems_dir, all_stem_names)
+    _lap(job, "post", mark)
 
 
 def _run_blocking(job: Job, url: str, job_dir: Path) -> None:
     _check_cancel(job)
+    mark = time.monotonic()
     source = download(job, url, job_dir)
+    _lap(job, "download", mark)
     _run_common(job, source, job_dir)
 
 
 def _run_local_blocking(job: Job, source_path: Path, job_dir: Path) -> None:
     _check_cancel(job)
+    mark = time.monotonic()
     source = _prepare_local_source(job, source_path, job_dir)
+    _lap(job, "prepare", mark)
     _run_common(job, source, job_dir)
 
 
@@ -185,13 +207,74 @@ def _write_metadata(job: Job, job_dir: Path) -> None:
         "stem_presence": job.stem_presence,
         "tags": job.tags,
         "has_video": job.has_video,
+        "compute_device": job.compute_device,
+        "stage_timings": job.stage_timings,
     }
     try:
         (job_dir / "metadata.json").write_text(json.dumps(meta, indent=2) + "\n", encoding="utf-8")
     except OSError:
         logger.warning("could not write metadata.json for job %s", job.id, exc_info=True)
 
 
+# Files worth keeping in a quarantined failure dir. Everything else (source
+# download, stem WAVs, video) is deleted first -- evidence must stay KB-scale,
+# not GB-scale.
+_QUARANTINE_KEEP = frozenset(("error.txt", "metadata.json"))
+
+
+def _quarantine_failed_job(job: Job, job_dir: Path, jobs_dir: Path, exc: Exception) -> None:
+    """Preserve failure evidence instead of destroying it (#277).
+
+    Writes error.txt (stage, device, model, timings, classified cause, stderr
+    tail), strips the heavy audio payloads, and moves the dir to
+    jobs/failed/<id> where sweep_failed_jobs expires it after FAILED_TTL.
+    Best-effort throughout: any step failing falls back to plain removal so a
+    pathological error can never leak disk."""
+    tail: list[str] = getattr(exc, "tail", None) or []
+    cause = classify_failure("\n".join([*tail, repr(exc)]))
+    detail = cause
+    if tail:
+        detail += f" — {tail[-1][:200]}"
+    job.error_detail = detail
+
+    try:
+        lines = [
+            f"time: {datetime.now(timezone.utc).isoformat(timespec='seconds')}",
+            f"job: {job.id}",
+            f"title: {job.title or '(unknown)'}",
+            f"source: {job.source_url or '(unknown)'}",
+            f"stage: {job.stage_message}",
+            f"device: {job.compute_device or getattr(exc, 'device', None) or '(not reached)'}",
+            f"model: {DEMUCS_MODEL}",
+            f"cause: {cause}",
+            f"timings: {json.dumps(job.stage_timings) if job.stage_timings else '(none)'}",
+            f"exception: {exc!r}",
+        ]
+        if tail:
+            lines += ["", "--- stderr tail ---", *tail]
+        (job_dir / "error.txt").write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+        # Strip heavy payloads: the quarantine keeps diagnostics, not audio.
+        for p in list(job_dir.iterdir()):
+            if p.name in _QUARANTINE_KEEP:
+                continue
+            if p.is_dir():
+                shutil.rmtree(p, ignore_errors=True)
+            else:
+                p.unlink(missing_ok=True)
+
+        failed_root = jobs_dir / "failed"
+        failed_root.mkdir(parents=True, exist_ok=True)
+        dest = failed_root / job.id
+        if dest.exists():
+            shutil.rmtree(dest, ignore_errors=True)
+        shutil.move(str(job_dir), str(dest))
+        logger.info("[%s] failure evidence kept at %s", job.id, dest)
+    except Exception:
+        logger.warning("[%s] quarantine failed; removing job dir", job.id, exc_info=True)
+        _rmtree(job_dir)
+
+
 async def _run_async(
     job: Job,
     job_dir: Path,
@@ -209,8 +292,8 @@ async def _run_async(
         if not isinstance(e, JobCancelled) and not job.cancel_requested:
             logger.exception("pipeline failed for job %s: %s", job.id, e)
             _set(job, status="error", stage="Error: Processing failed", error=error_msg)
+            _quarantine_failed_job(job, job_dir, jobs_dir, e)
             persist_registry(jobs_dir)
-            _rmtree(job_dir)
             return
         logger.info(
             "pipeline cancelled%s for job %s",
@@ -224,6 +307,17 @@ async def _run_async(
     _set(job, status="done", progress=1.0, stage="Done")
     _write_metadata(job, job_dir)
     persist_registry(jobs_dir)
+    # One-line per-job summary: the timing telemetry that makes performance
+    # regressions (and the CPU-vs-GPU question) answerable from logs (#293).
+    t = job.stage_timings or {}
+    logger.info(
+        "[%s] done device=%s model=%s %s total=%.1fs",
+        job.id,
+        job.compute_device or "n/a",
+        DEMUCS_MODEL,
+        " ".join(f"{k}={v}s" for k, v in t.items()),
+        sum(t.values()),
+    )
 
 
 async def run_pipeline(job: Job, url: str, jobs_dir: Path) -> None: