fix(imports): detect truncated downloads (real audio shorter than container)

The actual HiFi/Monochrome bug isn't silence padding — it's a TRUNCATED file:
the container claims the full length (e.g. 3:08) but only ~30s of audio
decodes. silencedetect finds nothing (there's no silent audio, just missing
audio) and ffmpeg's time= even reports 0 with no error, so the duration and
quality guards all pass.

Detect it by decoding and comparing the real audio length (astats sample
count / sample rate) against the container duration: reject when the real
audio covers < 85% of the claimed length. detect_broken_audio() runs this
truncation check first, then the silence-ratio check. Wire it into the guard
that runs at the integrity/length verification point.

Verified on the real file: 'only ~30s actually decodes of a 188s file (16%)';
a normal 180s file is not flagged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

core/imports/pipeline.py

@@ -35,7 +35,7 @@ from core.imports.context import (
 from core.imports.file_integrity import check_audio_integrity, expected_duration_for_check, resolve_duration_tolerance
 from core.imports.filename import extract_track_number_from_filename
 from core.imports.guards import check_flac_bit_depth, check_quality_target, move_to_quarantine
-from core.imports.silence import detect_mostly_silent
+from core.imports.silence import detect_broken_audio
 from core.imports.quarantine import (
     approve_quarantine_entry,
     entry_id_from_quarantined_filename,
@@ -163,7 +163,7 @@ def import_rejection_reason(context: dict) -> str | None:
     if context.get('_bitdepth_rejected'):
         return "rejected by quality filter"
     if context.get('_silence_rejected'):
-        return "rejected by silence guard (file is mostly silence)"
+        return "rejected by audio guard (incomplete or silent audio)"
     if context.get('_race_guard_failed'):
         return "source file disappeared before import completed"
     return None
@@ -356,29 +356,30 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
                 f"drift={integrity.checks.get('length_drift_s', 'n/a')})"
             )
 
-        # Silence guard — runs right where the length is verified, BEFORE the
+        # Audio-completeness guard — runs right where the length is verified,
-        # AcoustID/quality gates, so a mostly-silent file (correct container
+        # BEFORE the AcoustID/quality gates, so a truncated file (container
-        # duration but only ~30s real audio, e.g. HiFi/Monochrome HLS padding)
+        # claims the full length but only ~30s actually decodes, e.g. HiFi/
-        # is caught regardless of its quality verdict and gets the right reason.
+        # Monochrome HLS assembly) or a mostly-silent file is caught regardless
-        # Same quarantine + next-candidate retry pattern as the integrity check.
+        # of its quality verdict and gets the right reason. Same quarantine +
-        _skip_silence = _should_skip_quarantine_check(context, 'silence')
+        # next-candidate retry pattern as the integrity check.
-        silence_reason = None if _skip_silence else detect_mostly_silent(file_path)
+        _skip_audio = _should_skip_quarantine_check(context, 'silence')
-        if silence_reason:
+        audio_reason = None if _skip_audio else detect_broken_audio(file_path)
-            logger.error(f"[Silence] Rejected {_basename}: {silence_reason}")
+        if audio_reason:
+            logger.error(f"[AudioGuard] Rejected {_basename}: {audio_reason}")
             context['_silence_rejected'] = True
             try:
                 quarantine_path = move_to_quarantine(
-                    file_path, context, silence_reason, automation_engine,
+                    file_path, context, audio_reason, automation_engine,
                     trigger='silence',
                 )
                 _mark_task_quarantined(context, quarantine_path)
-                logger.warning("File quarantined — mostly silent: %s", quarantine_path)
+                logger.warning("File quarantined — incomplete/silent audio: %s", quarantine_path)
             except Exception as quarantine_error:
                 logger.error(f"Quarantine failed ({quarantine_error}), deleting file: {file_path}")
                 try:
                     os.remove(file_path)
                 except Exception as del_error:
-                    logger.debug("delete silent file fallback: %s", del_error)
+                    logger.debug("delete broken file fallback: %s", del_error)
 
             with matched_context_lock:
                 matched_downloads_context.pop(context_key, None)
@@ -389,15 +390,15 @@ def post_process_matched_download(context_key, context, file_path, runtime, meta
             # the integrity / AcoustID / quality failures.
             if _requeue_quarantined_task_for_retry(task_id, batch_id, 'silence'):
                 logger.info(
-                    "Mostly-silent file for task %s — retrying next-best candidate: %s",
+                    "Incomplete/silent audio for task %s — retrying next-best candidate: %s",
-                    task_id, silence_reason,
+                    task_id, audio_reason,
                 )
                 return
             if task_id:
                 with tasks_lock:
                     if task_id in download_tasks:
                         download_tasks[task_id]['status'] = 'failed'
-                        download_tasks[task_id]['error_message'] = f"Silence guard: {silence_reason}"
+                        download_tasks[task_id]['error_message'] = f"Audio guard: {audio_reason}"
             if task_id and batch_id:
                 _notify_download_completed(batch_id, task_id, success=False)
             return

core/imports/silence.py

@@ -1,16 +1,19 @@
-"""Silence guard — detect files whose container duration looks right but whose
+"""Audio-completeness guard — detect files whose container duration looks
-audio is mostly silence.
+right but whose REAL audio is far shorter, or mostly silence.
 
-Motivating bug: HiFi/Monochrome HLS assembly can yield a file with the full
+Motivating bug: HiFi/Monochrome HLS assembly can yield a file whose container
-track duration in its container while only the first ~30s carry real audio and
+claims the full track length (e.g. 3:08) while only ~30s of audio actually
-the rest is silence. The duration-agreement and quality guards both pass (the
+decodes — the rest is missing. The duration-agreement and quality guards both
-container says 3:30 and the format/bitrate are fine), so nothing catches it —
+pass (mutagen reads the container's 3:08 and the format/bitrate are fine), so
-until you listen. This guard runs ffmpeg ``silencedetect`` over the real audio
+nothing catches it until you listen. ffmpeg's ``time=`` even reports 0 with no
-and flags a file whose silent fraction exceeds a threshold.
+error on such a file, so the robust signal is to DECODE the audio and compare
+the real duration (sample count / sample rate, via ``astats``) against the
+container duration. A separate ``silencedetect`` pass also flags genuine
+silence-padding.
 
-The parser (``silence_ratio_from_output``) is pure and unit-tested; the ffmpeg
+The parsers here are pure and unit-tested; the ffmpeg invocations are
-invocation is integration glue that fails open (returns None) when ffmpeg or
+integration glue that fails open (returns None) when ffmpeg or mutagen can't
-mutagen can't run, so it never blocks a legitimate import on tooling problems.
+run, so a tooling problem never blocks a legitimate import.
 """
 
 from __future__ import annotations
@@ -23,6 +26,14 @@ from utils.logging_config import get_logger
 
 logger = get_logger("imports.silence")
 
+# Real decoded audio must cover at least this fraction of the container
+# duration. A legit file decodes to ~100% (encoder padding aside); a truncated
+# file decodes to a small fraction (the Blossom file: 30s of a 188s container
+# = 16%). 0.85 leaves generous headroom against false positives.
+DEFAULT_MIN_DURATION_RATIO = 0.85
+
+_SAMPLES_RE = re.compile(r"Number of samples:\s*([0-9]+)")
+
 # Defaults: treat audio below -50 dB lasting >= 2s as silence, and reject when
 # more than half the track is silent. A normal song — even with quiet intros/
 # outros — sits far below 0.5; a 30s-real + padded-silence file sits near 0.85.
@@ -127,3 +138,91 @@ def detect_mostly_silent(
 
     stderr = proc.stderr.decode("utf-8", errors="replace") if proc.stderr else ""
     return is_mostly_silent_reason(stderr, total_duration_s, threshold=threshold)
+
+
+# ── Truncation: real decoded duration vs container duration ────────────────
+
+def measured_duration_from_astats(astats_stderr: str, sample_rate: int) -> Optional[float]:
+    """Real decoded audio duration in seconds from ffmpeg ``astats`` output.
+
+    ``astats`` reports the per-channel ``Number of samples``; dividing by the
+    sample rate gives the true decoded length. Returns None when the sample
+    count or sample rate is unavailable.
+    """
+    if not sample_rate or sample_rate <= 0:
+        return None
+    m = _SAMPLES_RE.search(astats_stderr or "")
+    if not m:
+        return None
+    return int(m.group(1)) / float(sample_rate)
+
+
+def incomplete_audio_reason(
+    measured_s: Optional[float],
+    container_s: Optional[float],
+    *,
+    min_ratio: float = DEFAULT_MIN_DURATION_RATIO,
+) -> Optional[str]:
+    """Return a rejection reason when the real decoded duration falls short of
+    the container duration (a truncated file whose metadata over-states length).
+    """
+    if not measured_s or not container_s or container_s <= 0:
+        return None
+    if measured_s >= container_s * min_ratio:
+        return None
+    pct = round(measured_s / container_s * 100)
+    return (
+        f"Incomplete audio: only ~{round(measured_s)}s actually decodes of a "
+        f"{round(container_s)}s file ({pct}%) — truncated/broken download "
+        f"(container duration over-states the real audio)"
+    )
+
+
+def _measured_audio_duration_s(file_path: str, sample_rate: int) -> Optional[float]:
+    try:
+        proc = subprocess.run(
+            ["ffmpeg", "-hide_banner", "-nostats", "-i", file_path,
+             "-af", "astats=metadata=1", "-f", "null", "-"],
+            stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, timeout=120,
+        )
+    except (subprocess.SubprocessError, OSError) as exc:
+        logger.debug("astats run failed for %s: %s", file_path, exc)
+        return None
+    stderr = proc.stderr.decode("utf-8", errors="replace") if proc.stderr else ""
+    return measured_duration_from_astats(stderr, sample_rate)
+
+
+def detect_incomplete_audio(
+    file_path: str,
+    *,
+    min_ratio: float = DEFAULT_MIN_DURATION_RATIO,
+) -> Optional[str]:
+    """Decode the file and reject when the real audio is far shorter than the
+    container claims. Fails open when ffmpeg/mutagen are unavailable.
+    """
+    if not _ffmpeg_available():
+        return None
+    try:
+        from mutagen import File as MutagenFile
+        audio = MutagenFile(file_path)
+        if not (audio and audio.info):
+            return None
+        container_s = float(getattr(audio.info, "length", 0) or 0)
+        sample_rate = int(getattr(audio.info, "sample_rate", 0) or 0)
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.debug("container probe failed for %s: %s", file_path, exc)
+        return None
+
+    measured_s = _measured_audio_duration_s(file_path, sample_rate)
+    return incomplete_audio_reason(measured_s, container_s, min_ratio=min_ratio)
+
+
+def detect_broken_audio(file_path: str) -> Optional[str]:
+    """Combined post-download audio guard: reject a file that is truncated
+    (real audio far shorter than the container) or mostly silence. Returns the
+    first failure reason, or None when the audio looks complete.
+    """
+    reason = detect_incomplete_audio(file_path)
+    if reason:
+        return reason
+    return detect_mostly_silent(file_path)

tests/imports/test_silence_guard.py

@@ -1,11 +1,17 @@
-"""Silence guard — catches files whose container duration is correct but whose
+"""Audio-completeness guard — catches files whose container duration is
-audio is mostly silence (e.g. HiFi/Monochrome 30s-preview padded out to the
+correct but whose real audio is far shorter (HiFi/Monochrome truncated files:
-full track length). Pure parser is tested here; the ffmpeg call is integration.
+container claims 3:08 but only ~30s actually decodes) or mostly silence. Pure
+parsers are tested here; the ffmpeg call is integration.
 """
 
 import pytest
 
-from core.imports.silence import silence_ratio_from_output, is_mostly_silent_reason
+from core.imports.silence import (
+    silence_ratio_from_output,
+    is_mostly_silent_reason,
+    measured_duration_from_astats,
+    incomplete_audio_reason,
+)
 
 
 _ONE_LONG_TAIL = """
@@ -57,3 +63,42 @@ def test_reason_when_mostly_silent():
 
 def test_no_reason_for_normal_song():
     assert is_mostly_silent_reason(_TWO_GAPS, total_duration_s=210.0, threshold=0.5) is None
+
+
+# ── truncation: real decoded duration vs container duration ────────────────
+
+_ASTATS_TRUNCATED = """
+[Parsed_astats_0 @ 0x55] Number of samples: 1318912
+[Parsed_astats_0 @ 0x55] Number of NaNs: 0
+"""
+
+_ASTATS_FULL = "[Parsed_astats_0 @ 0x55] Number of samples: 9261000\n"
+
+
+def test_measured_duration_from_samples():
+    # 1318912 samples / 44100 Hz ≈ 29.9s (the real Blossom file)
+    assert measured_duration_from_astats(_ASTATS_TRUNCATED, 44100) == pytest.approx(29.9, abs=0.1)
+
+
+def test_measured_duration_none_without_samples():
+    assert measured_duration_from_astats("no stats here", 44100) is None
+
+
+def test_measured_duration_none_without_sample_rate():
+    assert measured_duration_from_astats(_ASTATS_TRUNCATED, 0) is None
+
+
+def test_incomplete_reason_for_truncated_file():
+    # 30s of real audio in a 188s container → truncated.
+    reason = incomplete_audio_reason(29.9, 188.4, min_ratio=0.85)
+    assert reason is not None
+    assert "30s" in reason and "188s" in reason
+
+
+def test_no_incomplete_reason_for_full_file():
+    assert incomplete_audio_reason(187.5, 188.4, min_ratio=0.85) is None
+
+
+def test_no_incomplete_reason_when_unmeasurable():
+    assert incomplete_audio_reason(None, 188.4, min_ratio=0.85) is None
+    assert incomplete_audio_reason(30.0, 0, min_ratio=0.85) is None

webui/static/downloads.js

@@ -3711,7 +3711,7 @@ function processModalStatusUpdate(playlistId, data) {
                         // Distinguish quarantine outcomes from generic
                         // failures — the file is recoverable, not lost.
                         const _em = (task.error_message || '').toLowerCase();
-                        if (_em.includes('integrity check failed') || _em.includes('bit depth filter') || _em.includes('verification failed') || _em.includes('quality filter') || _em.includes('silence guard') || _em.includes('quarantin')) {
+                        if (_em.includes('integrity check failed') || _em.includes('bit depth filter') || _em.includes('verification failed') || _em.includes('quality filter') || _em.includes('audio guard') || _em.includes('silence guard') || _em.includes('quarantin')) {
                             isQuarantinedTask = true;
                             statusText = '🛡️ Quarantined';
                         } else {