Fail timing checks when ffprobe cannot read a duration

src/docgen/validate.py

@@ -24,6 +24,38 @@
     from docgen.config import Config
 
 
+def _unusable_duration(duration: object) -> bool:
+    if duration is None or isinstance(duration, bool) or not isinstance(duration, (int, float)):
+        return True
+    value = float(duration)
+    return not math.isfinite(value) or value <= 0
+
+
+def _ffprobe_launch_error(path: Path, exc: BaseException) -> str:
+    if isinstance(exc, subprocess.TimeoutExpired):
+        return f"ffprobe timed out on {path.name}"
+    return "ffprobe not found in PATH"
+
+
+def _positive_ffprobe_duration(out: subprocess.CompletedProcess[str]) -> float:
+    if out.returncode != 0:
+        extra = (out.stderr or "").strip()
+        msg = f"ffprobe failed (exit {out.returncode})"
+        if extra:
+            msg = f"{msg}: {extra[:200]}"
+        raise RuntimeError(msg)
+    raw = (out.stdout or "").strip()
+    try:
+        duration = float(raw)
+    except (TypeError, ValueError) as exc:
+        raise RuntimeError(f"ffprobe duration is not a number: {raw[:80]!r}") from exc
+    if not math.isfinite(duration) or duration <= 0:
+        raise RuntimeError(
+            f"ffprobe duration is not a finite positive number: {raw[:80]!r}"
+        )
+    return duration
+
+
 @dataclass
 class CheckResult:
     name: str
@@ -833,16 +865,9 @@ def _check_timing_sync(self, seg_id: str) -> CheckResult:
                 )
             return CheckResult("timing_sync", True, ["Empty timing entry (skipped, non-manim)"])
 
-        audio_dur = self._probe_media_duration(audio)
-        if audio_dur is None:
-            return CheckResult(
-                "timing_sync",
-                False,
-                [
-                    f"cannot probe audio duration for {audio.name} — "
-                    "ffprobe failed; timing_sync cannot compare the mp3 to timing.json"
-                ],
-            )
+        audio_dur, probe_fail = self._duration_or_fail(audio, "timing_sync")
+        if probe_fail is not None:
+            return probe_fail
 
         max_tail = float(ts_cfg.get("max_tail_gap_sec", 3.0))
         max_overrun = float(ts_cfg.get("max_end_overrun_sec", 1.0))
@@ -947,16 +972,9 @@ def _check_story_end(self, seg_id: str) -> CheckResult:
 
         audio = self._find_audio(seg_id)
         if audio and not _is_lfs_pointer(audio):
-            audio_end = self._probe_media_duration(audio)
-            if audio_end is None or audio_end <= 0:
-                return CheckResult(
-                    "story_end",
-                    False,
-                    [
-                        f"cannot probe audio duration for {audio.name} — "
-                        "story_end cannot compare last paced reveal to the mp3"
-                    ],
-                )
+            audio_end, probe_fail = self._duration_or_fail(audio, "story_end")
+            if probe_fail is not None:
+                return probe_fail
             end_t = audio_end
         else:
             # No local mp3 (or LFS pointer): compare against transcript end only.
@@ -1073,22 +1091,48 @@ def _timing_last_end(block: dict[str, Any]) -> float | None:
                     return max(ends)
         return None
 
+    def _duration_or_fail(self, path: Path, check_name: str) -> tuple[float, None] | tuple[None, CheckResult]:
+        """Duration for *check_name*, or a failed check when the probe cannot be used.
+
+        A non-zero ffprobe exit, timeout, missing binary, or unusable duration
+        fails the check that asked (``passed=False``). ``None`` is not a skip.
+        """
+        try:
+            duration = self._probe_media_duration(path)
+        except RuntimeError as exc:
+            return None, CheckResult(
+                check_name,
+                False,
+                [f"cannot probe audio duration for {path.name} — {exc}"],
+            )
+        if _unusable_duration(duration):
+            return None, CheckResult(
+                check_name,
+                False,
+                [
+                    f"cannot probe audio duration for {path.name} — "
+                    f"ffprobe duration is not a finite positive number: {duration!r}"
+                ],
+            )
+        return float(duration), None
+
     @staticmethod
-    def _probe_media_duration(path: Path) -> float | None:
+    def _probe_media_duration(path: Path) -> float:
+        """Finite positive duration from ffprobe.
+
+        Raises ``RuntimeError`` on a non-zero exit, timeout, missing ffprobe,
+        or a duration that is not a finite positive number. Does not return
+        ``None`` for those failures (callers must fail the check).
+        """
         try:
             out = subprocess.run(
                 ["ffprobe", "-v", "error", "-show_entries", "format=duration",
                  "-of", "csv=p=0", str(path)],
                 capture_output=True, text=True, timeout=30,
             )
-        except (subprocess.TimeoutExpired, FileNotFoundError):
-            return None
-        if out.returncode != 0:
-            return None
-        try:
-            return float(out.stdout.strip())
-        except ValueError:
-            return None
+        except (subprocess.TimeoutExpired, FileNotFoundError) as exc:
+            raise RuntimeError(_ffprobe_launch_error(path, exc)) from exc
+        return _positive_ffprobe_duration(out)
 
     # ── Helpers ────────────────────────────────────────────────────────
 

tests/test_validate_timing_sync.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import subprocess
 import sys
 import types
 from pathlib import Path
@@ -212,6 +213,76 @@ def _write_scene_spec(cfg: Config, *, labels: list[str]) -> None:
     (specs / "01-x.scene.yaml").write_text(yaml.dump(raw), encoding="utf-8")
 
 
+class _FakeDurationProbe:
+    def __init__(self, returncode: int, stdout: str, stderr: str = "") -> None:
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+def _install_duration_probe(monkeypatch: pytest.MonkeyPatch, mode: str) -> None:
+    """Drive the real ``_probe_media_duration`` (do not replace the method)."""
+
+    def _run(*_args: object, **_kwargs: object) -> _FakeDurationProbe:
+        if mode == "timeout":
+            raise subprocess.TimeoutExpired(cmd=["ffprobe"], timeout=30)
+        if mode == "nonzero":
+            # Leftover stdout looks like a duration that would pass the check.
+            return _FakeDurationProbe(1, "10.4\n", stderr="Invalid data")
+        if mode == "nan":
+            return _FakeDurationProbe(0, "nan\n")
+        if mode == "inf":
+            return _FakeDurationProbe(0, "inf\n")
+        return _FakeDurationProbe(0, "not-a-duration\n")
+
+    monkeypatch.setattr(subprocess, "run", _run)
+
+
+class TestMediaDurationProbeFailClosed:
+    """ffprobe duration failures fail the check that asked; they are not a skip."""
+
+    @pytest.mark.parametrize("mode", ("nonzero", "timeout", "invalid", "nan", "inf"))
+    def test_timing_sync_fails_closed(self, cfg: Config, monkeypatch: pytest.MonkeyPatch, mode: str) -> None:
+        _write_timing(cfg, last_end=10.0)
+        _install_duration_probe(monkeypatch, mode)
+        check = Validator(cfg)._check_timing_sync("01")
+        assert check.passed is False
+        assert any("cannot probe audio duration" in detail for detail in check.details)
+        if mode == "nonzero":
+            assert any("exit 1" in detail for detail in check.details)
+        elif mode == "timeout":
+            assert any("timed out" in detail for detail in check.details)
+        else:
+            assert any("ffprobe duration" in detail for detail in check.details)
+
+    @pytest.mark.parametrize("mode", ("nonzero", "timeout", "invalid", "nan", "inf"))
+    def test_story_end_fails_closed(self, cfg: Config, monkeypatch: pytest.MonkeyPatch, mode: str) -> None:
+        words = [
+            {"word": "Alpha", "start": 2.0, "end": 2.4},
+            {"word": "Omega", "start": 80.0, "end": 80.5},
+        ]
+        (cfg.animations_dir / "timing.json").write_text(
+            json.dumps(
+                {
+                    "01-x": {
+                        "text": "Alpha Omega",
+                        "words": words,
+                        "segments": [{"start": 0.0, "end": 85.0, "text": "x"}],
+                    }
+                }
+            ),
+            encoding="utf-8",
+        )
+        _write_scene_spec(cfg, labels=["Alpha", "Omega"])
+        _install_duration_probe(monkeypatch, mode)
+        check = Validator(cfg)._check_story_end("01")
+        assert check.passed is False
+        assert any("cannot probe audio duration" in detail for detail in check.details)
+        if mode == "nonzero":
+            assert any("exit 1" in detail for detail in check.details)
+            assert not any("skipped" in detail.lower() for detail in check.details)
+
+
 class TestStoryEnd:
     def test_story_finishes_early_fails(self, cfg, monkeypatch) -> None:
         """Board done at ~10s while audio runs ~100s → story_end hard fail."""