From 49e915e3e7bc3c1f2fdca704e6f2ea7537f41efa Mon Sep 17 00:00:00 2001 From: Thales <> Date: Thu, 16 Jul 2026 17:38:04 +0100 Subject: [PATCH] feat(pipeline): quarantine failed jobs with evidence; classify causes; stage timings The error path destroyed all evidence: rmtree on failure threw away the demucs stderr, the stage, and the device, leaving "Audio processing failed" as the only artifact -- undebuggable after the fact. - Failed jobs now move to jobs/failed/ with an error.txt recording stage, device, model, classified cause, stage timings, and the demucs stderr tail. Heavy payloads (source, stems, video) are stripped first so quarantines stay KB-scale. Expired after 7 days by a new sweep that runs even on persistent-library deployments (failure evidence is diagnostics, not library content). The TTL sweep skips failed/. - New app/pipeline/errors.py: SeparationError carries the stderr tail + device out of separate(); classify_failure() maps failure text to out-of-memory / unsupported-device / disk-full / bad-input / unknown. The classified cause surfaces as Job.error_detail, shown in the studio as a muted secondary line under the generic error message. - Per-stage wall-clock timings (download/prepare, analyze, separate, post) recorded on the job, written to metadata.json, included in error.txt, and emitted as a one-line completion summary with the compute device -- performance regressions and the CPU-vs-GPU question are now answerable from logs. Closes #277 Closes #294 Closes #293 --- app/core/config.py | 4 ++ app/core/models.py | 13 +++++ app/main.py | 13 +++-- app/pipeline/collect.py | 20 +++++++ app/pipeline/errors.py | 88 +++++++++++++++++++++++++++++ app/pipeline/runner.py | 98 +++++++++++++++++++++++++++++++- app/pipeline/separate.py | 11 +++- static/css/job.css | 6 ++ static/js/job.js | 12 +++- tests/test_errors.py | 43 ++++++++++++++ tests/test_pipeline_runner.py | 69 +++++++++++++++++++++++ tests/test_sweep.py | 103 +++++++++++++++++++++++++++++++--- 12 files changed, 462 insertions(+), 18 deletions(-) create mode 100644 app/pipeline/errors.py create mode 100644 tests/test_errors.py diff --git a/app/core/config.py b/app/core/config.py index dfa95075..5150c31e 100644 --- a/app/core/config.py +++ b/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/, 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) diff --git a/app/core/models.py b/app/core/models.py index 26e4dc26..ca2e1b65 100644 --- a/app/core/models.py +++ b/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, } diff --git a/app/main.py b/app/main.py index e59432fd..325b4379 100644 --- a/app/main.py +++ b/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) diff --git a/app/pipeline/collect.py b/app/pipeline/collect.py index 6f995975..70e6757f 100644 --- a/app/pipeline/collect.py +++ b/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/) 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) diff --git a/app/pipeline/errors.py b/app/pipeline/errors.py new file mode 100644 index 00000000..48b68747 --- /dev/null +++ b/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" diff --git a/app/pipeline/runner.py b/app/pipeline/runner.py index 032eee77..b32ff91a 100644 --- a/app/pipeline/runner.py +++ b/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,6 +207,8 @@ 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") @@ -192,6 +216,65 @@ def _write_metadata(job: Job, job_dir: Path) -> None: 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/ 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: diff --git a/app/pipeline/separate.py b/app/pipeline/separate.py index a54c9371..dc059d05 100644 --- a/app/pipeline/separate.py +++ b/app/pipeline/separate.py @@ -13,6 +13,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 logger = logging.getLogger("stemdeck.pipeline") @@ -26,8 +27,10 @@ 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. + # 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 = [ sys.executable, @@ -124,9 +127,11 @@ def _watchdog() -> None: 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}" - raise RuntimeError(f"demucs failed: {last}") + # 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 RuntimeError(f"demucs output not found at {stems_root}") + raise SeparationError(f"demucs output not found at {stems_root}", device=device) return stems_root diff --git a/static/css/job.css b/static/css/job.css index 3b0fd29a..823b6f0b 100644 --- a/static/css/job.css +++ b/static/css/job.css @@ -103,6 +103,12 @@ progress::-moz-progress-bar { word-break: break-word; font-size: 13px; } +.error-detail { + margin-top: 4px; + font-size: 11px; + color: var(--muted); + font-family: var(--font-mono); +} .retry-btn { flex-shrink: 0; background: transparent; diff --git a/static/js/job.js b/static/js/job.js index e3bace13..8c9e0bef 100644 --- a/static/js/job.js +++ b/static/js/job.js @@ -66,11 +66,19 @@ function stopJobPolling() { } } -export function showError(message) { +export function showError(message, detail) { errorEl.textContent = ""; const msg = document.createElement("div"); msg.className = "error-msg"; msg.textContent = message; + if (detail) { + // Classified cause from the backend (e.g. "out-of-memory — ..."), shown + // as a muted secondary line so failures are actionable, not opaque. + const detailEl = document.createElement("div"); + detailEl.className = "error-detail"; + detailEl.textContent = detail; + msg.appendChild(detailEl); + } const retry = document.createElement("button"); retry.className = "retry-btn"; retry.type = "button"; @@ -214,7 +222,7 @@ function applyState(state) { stopJobPolling(); updateTrackStatus(state.job_id, "error"); setWaveformLoading(false); - showError(state.error || "Unknown error"); + showError(state.error || "Unknown error", state.error_detail); setSubmitProcessing(false); } else if (state.status === "cancelled") { stopJobPolling(); diff --git a/tests/test_errors.py b/tests/test_errors.py new file mode 100644 index 00000000..a6f2e61b --- /dev/null +++ b/tests/test_errors.py @@ -0,0 +1,43 @@ +"""Tests for the pipeline failure classifier and SeparationError (#294, #277).""" + +from __future__ import annotations + +import pytest + +from app.pipeline.errors import SeparationError, classify_failure + + +@pytest.mark.parametrize( + ("text", "expected"), + [ + ("RuntimeError: CUDA out of memory. Tried to allocate 2.00 GiB", "out-of-memory"), + ("RuntimeError: MPS backend out of memory (MPS allocated: 5.2 GB)", "out-of-memory"), + ("OSError: cannot allocate memory", "out-of-memory"), + ("RuntimeError: no kernel image is available for execution", "unsupported-device"), + ("AssertionError: Torch not compiled with CUDA enabled", "unsupported-device"), + ("OSError: [Errno 28] No space left on device", "disk-full"), + ("Invalid data found when processing input", "bad-input"), + ("RuntimeError: no stems produced by demucs", "bad-input"), + ("something entirely novel went wrong", "unknown"), + ("", "unknown"), + ], +) +def test_classify_failure(text: str, expected: str): + assert classify_failure(text) == expected + + +def test_classify_is_case_insensitive(): + assert classify_failure("CUDA OUT OF MEMORY") == "out-of-memory" + + +def test_separation_error_carries_evidence(): + err = SeparationError("demucs failed: boom", tail=["line1", "boom"], device="mps") + assert isinstance(err, RuntimeError) + assert err.tail == ["line1", "boom"] + assert err.device == "mps" + + +def test_separation_error_defaults(): + err = SeparationError("plain") + assert err.tail == [] + assert err.device is None diff --git a/tests/test_pipeline_runner.py b/tests/test_pipeline_runner.py index 74bf0ac0..69cb2656 100644 --- a/tests/test_pipeline_runner.py +++ b/tests/test_pipeline_runner.py @@ -144,6 +144,75 @@ def boom(*args, **kwargs): assert not (tmp_path / job.id).exists(), "job dir should be removed on local error" +@pytest.mark.asyncio +async def test_pipeline_error_quarantines_evidence(tmp_path: Path): + """#277: a failed job's dir moves to jobs/failed/ with error.txt + (device, cause, stderr tail) and the heavy audio payloads stripped.""" + from app.pipeline.errors import SeparationError + + job = Job(id="abcdefabcde6") + job_dir = tmp_path / job.id + (job_dir / "stems").mkdir(parents=True) + (job_dir / "stems" / "vocals.wav").write_bytes(b"RIFF" + b"\x00" * 64) + (job_dir / "source.wav").write_bytes(b"RIFF" + b"\x00" * 64) + source = job_dir / "source.wav" + job.stage_timings = {"download": 1.2} + + def boom(*args, **kwargs): + raise SeparationError( + "demucs failed: MPS backend out of memory", + tail=["progress 50%", "RuntimeError: MPS backend out of memory"], + device="mps", + ) + + with patch("app.pipeline.runner._run_local_blocking", side_effect=boom): + await run_local_pipeline(job, source, tmp_path) + + assert job.status == "error" + assert job.error_detail is not None + assert job.error_detail.startswith("out-of-memory") + # Original dir gone; quarantine holds error.txt but no audio payloads. + assert not job_dir.exists() + quarantined = tmp_path / "failed" / job.id + report = (quarantined / "error.txt").read_text(encoding="utf-8") + assert "device: mps" in report + assert "cause: out-of-memory" in report + assert "MPS backend out of memory" in report + assert '"download": 1.2' in report + assert not (quarantined / "source.wav").exists() + assert not (quarantined / "stems").exists() + + +@pytest.mark.asyncio +async def test_pipeline_success_logs_timing_summary(tmp_path: Path, caplog): + """#293: successful jobs emit a one-line stage-timing summary.""" + import logging + + job = Job(id="abcdefabcde5") + + def fake_stages(j, url, job_dir): + j.stage_timings = {"download": 2.0, "analyze": 1.0, "separate": 30.0, "post": 3.5} + j.compute_device = "cpu" + + with ( + patch("app.pipeline.runner._run_blocking", side_effect=fake_stages), + caplog.at_level(logging.INFO, logger="stemdeck.pipeline"), + ): + await run_pipeline(job, "https://www.youtube.com/watch?v=dQw4w9WgXcQ", tmp_path) + + assert job.status == "done" + summary = next(r.message for r in caplog.records if "done device=" in r.message) + assert "device=cpu" in summary + assert "separate=30.0s" in summary + assert "total=36.5s" in summary + # Timings + device persist into metadata.json for later diagnostics. + import json as _json + + meta = _json.loads((tmp_path / job.id / "metadata.json").read_text(encoding="utf-8")) + assert meta["compute_device"] == "cpu" + assert meta["stage_timings"]["separate"] == 30.0 + + def test_extract_video_track_from_mp4(tmp_path: Path): """#219: an mp4 with a video stream yields video.mp4 and sets has_video.""" if not _ffmpeg_available(): diff --git a/tests/test_sweep.py b/tests/test_sweep.py index ef255a33..5cfbcb7d 100644 --- a/tests/test_sweep.py +++ b/tests/test_sweep.py @@ -80,15 +80,48 @@ def test_sweep_disabled_under_persistent_library(monkeypatch): @pytest.mark.asyncio -async def test_sweep_loop_returns_immediately_under_desktop(monkeypatch): - """In desktop mode the loop returns at once instead of entering the hourly - cycle (wait_for would time out if it looped).""" - import asyncio - - from app.main import _sweep_loop +async def test_sweep_loop_desktop_skips_ttl_but_sweeps_failed(monkeypatch): + """Desktop mode skips the library TTL sweep but the failed-job quarantine + still expires (#277) -- failure evidence isn't library content.""" + from app import main as main_mod monkeypatch.setenv("STEMDECK_DESKTOP", "1") - await asyncio.wait_for(_sweep_loop(), timeout=2) + ttl_calls: list = [] + failed_calls: list = [] + monkeypatch.setattr(main_mod, "sweep_old_jobs", ttl_calls.append) + monkeypatch.setattr(main_mod, "sweep_failed_jobs", failed_calls.append) + + async def stop_loop(_delay): + raise RuntimeError("stop-loop") + + monkeypatch.setattr(main_mod.asyncio, "sleep", stop_loop) + with pytest.raises(RuntimeError, match="stop-loop"): + await main_mod._sweep_loop() + + assert ttl_calls == [] + assert failed_calls == [main_mod.JOBS_DIR] + + +@pytest.mark.asyncio +async def test_sweep_loop_server_runs_both_sweeps(monkeypatch): + from app import main as main_mod + + monkeypatch.delenv("STEMDECK_DESKTOP", raising=False) + monkeypatch.delenv("STEMDECK_PERSIST_LIBRARY", raising=False) + ttl_calls: list = [] + failed_calls: list = [] + monkeypatch.setattr(main_mod, "sweep_old_jobs", ttl_calls.append) + monkeypatch.setattr(main_mod, "sweep_failed_jobs", failed_calls.append) + + async def stop_loop(_delay): + raise RuntimeError("stop-loop") + + monkeypatch.setattr(main_mod.asyncio, "sleep", stop_loop) + with pytest.raises(RuntimeError, match="stop-loop"): + await main_mod._sweep_loop() + + assert ttl_calls == [main_mod.JOBS_DIR] + assert failed_calls == [main_mod.JOBS_DIR] def test_keeps_recent_terminal_job(tmp_path: Path): @@ -119,3 +152,59 @@ def test_orphan_dir_falls_back_to_mtime(tmp_path: Path): sweep_old_jobs(tmp_path) assert not d.exists() + + +# ── failed-job quarantine sweep (#277) ── + + +def test_ttl_sweep_never_touches_failed_root(tmp_path: Path): + """sweep_old_jobs must skip jobs/failed/ even when it looks ancient -- + the quarantine has its own, longer TTL.""" + failed = _mkdir(tmp_path / "failed", "abcdefabcdef") + old = time.time() - 999_999 + import os + + os.utime(tmp_path / "failed", (old, old)) + os.utime(failed, (old, old)) + + with patch("app.pipeline.collect.JOB_TTL_SECONDS", 60): + sweep_old_jobs(tmp_path) + + assert failed.is_dir() + + +def test_sweep_failed_jobs_expires_old_keeps_fresh(tmp_path: Path): + from app.pipeline.collect import sweep_failed_jobs + + old_dir = _mkdir(tmp_path / "failed", "abcdefabcde1") + fresh_dir = _mkdir(tmp_path / "failed", "abcdefabcde2") + old = time.time() - 999_999 + import os + + os.utime(old_dir, (old, old)) + + with patch("app.pipeline.collect.FAILED_TTL_SECONDS", 3600): + sweep_failed_jobs(tmp_path) + + assert not old_dir.exists() + assert fresh_dir.is_dir() + + +def test_sweep_failed_jobs_noop_without_quarantine(tmp_path: Path): + from app.pipeline.collect import sweep_failed_jobs + + sweep_failed_jobs(tmp_path) # must not raise + + +def test_restore_ignores_failed_quarantine(tmp_path: Path): + """registry.restore must not resurrect a quarantined failure as a job.""" + from app.core.registry import restore + + quarantined = tmp_path / "failed" / "abcdefabcde3" + (quarantined / "stems").mkdir(parents=True) + (quarantined / "error.txt").write_text("evidence", encoding="utf-8") + + restore(tmp_path) + + assert "abcdefabcde3" not in _jobs + assert quarantined.is_dir() # restore must not delete it either