Fail closed when validate stream probes ignore ffprobe exit

milestones/README.md

@@ -5,10 +5,14 @@ repositories that install `docgen` and maintain their own demo bundle. The
 library no longer ships an in-repo dogfood; consumers are the integration test
 of record.
 
-**Active:** **[whisper-prompt-caps.md](whisper-prompt-caps.md)** —
-timing-enrichment whisper count caps must not coerce bools to 1.
+**Active:** **[validate-stream-probe.md](validate-stream-probe.md)** —
+validate stream/drift checks must not trust ffprobe stdout when the
+probe exits non-zero.
 
 **Shipped:**
+- **[whisper-prompt-caps.md](whisper-prompt-caps.md)** —
+  timing-enrichment whisper count caps must not coerce bools to 1
+  (#144).
 - **[pages-ffprobe-returncode.md](pages-ffprobe-returncode.md)** —
   pages duration badges must not trust ffprobe stdout when the probe
   exits non-zero (#143).

milestones/validate-stream-probe.md

@@ -0,0 +1,36 @@
+# Milestone: validate stream/drift probes must honor ffprobe returncode
+
+**Status:** Active  
+**PR:** [#145](https://github.com/jmjava/documentation-generator/pull/145)  
+**Depends on:** `milestones/whisper-prompt-caps.md` (PR #144),
+`milestones/pages-ffprobe-returncode.md` (PR #143),
+`milestones/ffprobe-returncode.md` (PR #135)
+
+## Problem
+
+PR #135 made duration probes ignore stdout when ffprobe exits non-zero.
+Pages followed in #143. Validate **`_check_streams` / `_check_drift`**
+still `json.loads(out.stdout)` with no `returncode` check.
+
+Empty stdout already fails `json.loads` (failed check). Leftover JSON
+that lists video+audio streams can **pass** `stream_presence` /
+`av_drift` on a corrupt recording. Those checks are hard in
+`validate --pre-push` (not in the soft set).
+
+## Goal
+
+Nonzero ffprobe is a failed check. Do not trust leftover JSON.
+
+## Done when
+
+- [x] `_check_streams` / `_check_drift` fail when `returncode != 0`
+- [x] Successful probes still parse streams / drift as before
+- [x] Tests for leftover JSON on failure vs a real probe
+- [x] `ruff check src/ tests/`
+- [x] `pytest tests/` (808 passed, 1 skipped)
+- [x] `docgen benchmark` (no clock change; meets baseline)
+
+## Out of scope
+
+- Soft-check policy for layout / av_sync / freeze_ratio
+- Whisper API `start`/`end` `float(... or 0.0)` in `ai_client`

milestones/whisper-prompt-caps.md

@@ -1,6 +1,6 @@
 # Milestone: Whisper prompt caps must not coerce bools
 
-**Status:** Active  
+**Status:** Shipped  
 **PR:** [#144](https://github.com/jmjava/documentation-generator/pull/144)  
 **Depends on:** `milestones/pages-ffprobe-returncode.md` (PR #143),
 `milestones/merge-settings-numerics.md` (PR #142),

src/docgen/validate.py

@@ -658,16 +658,44 @@ def _check_layout(self, path: Path) -> CheckResult:
 
     # ── ffprobe-based checks ──────────────────────────────────────────
 
-    def _check_streams(self, path: Path) -> CheckResult:
+    @staticmethod
+    def _ffprobe_json(path: Path, *show_flags: str) -> dict[str, Any]:
+        """Run ffprobe JSON. Ignore stdout when the process exits non-zero."""
         try:
             out = subprocess.run(
-                ["ffprobe", "-v", "quiet", "-print_format", "json", "-show_streams", str(path)],
+                ["ffprobe", "-v", "quiet", "-print_format", "json", *show_flags, str(path)],
                 capture_output=True, text=True, timeout=30,
             )
-            data = json.loads(out.stdout)
+        except subprocess.TimeoutExpired as exc:
+            raise RuntimeError(f"ffprobe timed out on {path.name}") from exc
+        except FileNotFoundError as exc:
+            raise RuntimeError("ffprobe not found in PATH") from exc
+        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)
+        try:
+            payload = json.loads(out.stdout)
+        except json.JSONDecodeError as exc:
+            raise RuntimeError(f"ffprobe JSON is not valid: {exc}") from exc
+        if not isinstance(payload, dict):
+            raise RuntimeError(
+                f"ffprobe JSON root must be an object, not {type(payload).__name__}"
+            )
+        return payload
+
+    def _check_streams(self, path: Path) -> CheckResult:
+        try:
+            data = self._ffprobe_json(path, "-show_streams")
             streams = data.get("streams", [])
-            has_video = any(s.get("codec_type") == "video" for s in streams)
-            has_audio = any(s.get("codec_type") == "audio" for s in streams)
+            if not isinstance(streams, list):
+                return CheckResult(
+                    "stream_presence", False, ["ffprobe streams must be a JSON array"]
+                )
+            has_video = any(s.get("codec_type") == "video" for s in streams if isinstance(s, dict))
+            has_audio = any(s.get("codec_type") == "audio" for s in streams if isinstance(s, dict))
             issues: list[str] = []
             if not has_video:
                 issues.append("Missing video stream")
@@ -679,22 +707,27 @@ def _check_streams(self, path: Path) -> CheckResult:
 
     def _check_drift(self, path: Path, max_drift: float) -> CheckResult:
         try:
-            out = subprocess.run(
-                ["ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", "-show_streams", str(path)],
-                capture_output=True, text=True, timeout=30,
-            )
-            data = json.loads(out.stdout)
+            data = self._ffprobe_json(path, "-show_format", "-show_streams")
             durations: dict[str, float] = {}
-            for s in data.get("streams", []):
+            streams = data.get("streams", [])
+            if not isinstance(streams, list):
+                streams = []
+            for s in streams:
+                if not isinstance(s, dict):
+                    continue
                 ct = s.get("codec_type", "")
-                dur = float(s.get("duration", 0))
+                try:
+                    dur = float(s.get("duration", 0))
+                except (TypeError, ValueError):
+                    continue
                 if ct in ("video", "audio") and dur > 0:
                     durations[ct] = dur
 
             has_video_stream = any(
-                s.get("codec_type") == "video" for s in data.get("streams", [])
+                isinstance(s, dict) and s.get("codec_type") == "video" for s in streams
             )
-            fmt_dur_raw = data.get("format", {}).get("duration")
+            fmt = data.get("format")
+            fmt_dur_raw = fmt.get("duration") if isinstance(fmt, dict) else None
             if has_video_stream and "video" not in durations and fmt_dur_raw is not None:
                 try:
                     fd = float(fmt_dur_raw)

tests/test_validate.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import json
 import subprocess
 from pathlib import Path
 
@@ -617,6 +618,67 @@ def test_validate_passes_covered_beats(self, cfg_dir: Path) -> None:
         assert check["passed"], check["details"]
 
 
+# ── ffprobe JSON probes honor returncode ──────────────────────────────
+
+class _FakeProbe:
+    def __init__(self, returncode: int, stdout: str, stderr: str = "Invalid data") -> None:
+        self.returncode = returncode
+        self.stdout = stdout
+        self.stderr = stderr
+
+
+class TestFfprobeJsonReturncode:
+    def test_streams_fail_when_ffprobe_exits_nonzero(self, config, tmp_path, monkeypatch):
+        leftover = json.dumps({
+            "streams": [{"codec_type": "video"}, {"codec_type": "audio"}],
+        })
+        monkeypatch.setattr(
+            subprocess, "run", lambda *_a, **_k: _FakeProbe(1, leftover)
+        )
+        result = Validator(config)._check_streams(tmp_path / "corrupt.mp4")
+        assert result.passed is False
+        assert "ffprobe failed" in result.details[0]
+
+    def test_drift_fail_when_ffprobe_exits_nonzero(self, config, tmp_path, monkeypatch):
+        leftover = json.dumps({
+            "format": {"duration": "10.0"},
+            "streams": [
+                {"codec_type": "video", "duration": "10.0"},
+                {"codec_type": "audio", "duration": "10.0"},
+            ],
+        })
+        monkeypatch.setattr(
+            subprocess, "run", lambda *_a, **_k: _FakeProbe(1, leftover)
+        )
+        result = Validator(config)._check_drift(tmp_path / "corrupt.mp4", max_drift=2.75)
+        assert result.passed is False
+        assert "ffprobe failed" in result.details[0]
+
+    def test_streams_pass_when_ffprobe_succeeds(self, config, tmp_path, monkeypatch):
+        ok = json.dumps({
+            "streams": [{"codec_type": "video"}, {"codec_type": "audio"}],
+        })
+        monkeypatch.setattr(
+            subprocess, "run", lambda *_a, **_k: _FakeProbe(0, ok, stderr="")
+        )
+        result = Validator(config)._check_streams(tmp_path / "ok.mp4")
+        assert result.passed is True
+
+    def test_drift_pass_when_ffprobe_succeeds(self, config, tmp_path, monkeypatch):
+        ok = json.dumps({
+            "format": {"duration": "10.0"},
+            "streams": [
+                {"codec_type": "video", "duration": "10.0"},
+                {"codec_type": "audio", "duration": "10.05"},
+            ],
+        })
+        monkeypatch.setattr(
+            subprocess, "run", lambda *_a, **_k: _FakeProbe(0, ok, stderr="")
+        )
+        result = Validator(config)._check_drift(tmp_path / "ok.mp4", max_drift=2.75)
+        assert result.passed is True
+
+
 # ── Helper to create silent audio ─────────────────────────────────────
 
 def _make_silent_audio(path: Path, duration_sec: float = 10.0) -> Path: