feat(ffmpeg): add measured podcast audio analysis (#479)

ffmpeg/README.md

@@ -61,7 +61,7 @@ This skill provides a repeatable intake-to-acceptance workflow. It separates tec
 | `scripts/media-intake` | Read-only input inventory with bounded `ffprobe` metadata |
 | `scripts/extract-review-frames` | Bounded timestamp frame extraction for human or vision review |
 | `scripts/render-edl` | Validate single- or multi-source EDLs and emit non-executing concat-filter or concat-demuxer plans |
-| `scripts/audio-inspect` | Read-only audio metadata inspection with bounded probing |
+| `scripts/audio-inspect` | Bounded silence, loudness, peak/clipping, transcript-candidate, and podcast-plan evidence |
 | `scripts/media-verify` | Compare input/output probe documents against basic criteria |
 | `scripts/editorial-workflow-example` | Generate synthetic audio/video and exercise the complete workflow with durable evidence |
 | `scripts/generate-media-fixtures` | Generate a bounded sanitized fixture battery and versioned evidence manifest |
@@ -109,6 +109,16 @@ scripts/generate-media-fixtures /tmp/ffmpeg-fixtures --json
 
 Generated media remains task-local; commit the generator and test assertions, not the binary outputs.
 
+Produce measured audio evidence and a reviewable plan without changing the source:
+
+```sh
+scripts/audio-inspect input.wav \
+  --measure-silence --silence-threshold=-45dB --silence-duration 0.7 \
+  --measure-loudness --measure-clipping --json
+```
+
+Optional timed transcript JSON must disclose alignment quality. Detector intervals and transcript ranges remain candidates until listening review approves an edit.
+
 ## Triggers
 
 Load this skill for:

ffmpeg/SKILL.md

@@ -99,6 +99,8 @@ Run `scripts/generate-media-fixtures` when tests need deterministic non-personal
 
 Use `scripts/render-edl` for a non-executing single- or multi-source plan. Default to decoded concat-filter assembly; select concat-demuxer stream copy only with matching probe-derived signatures and verified packet/keyframe boundaries. Unsupported transitions must remain explicit errors.
 
+Use `scripts/audio-inspect` for bounded silence, EBU R128, peak/clipping, and transcript-alignment evidence. Request each measurement explicitly, preserve unavailable filters as `UNAVAILABLE`, and treat every interval or transcript range as a listening-review candidate. Its optional report output refuses overwrite.
+
 Copy a template into the task workspace and replace its placeholder/example values. Do not put private paths, media, transcripts, or review evidence in the public skill repository.
 
 ## Non-Negotiable Checks

ffmpeg/evals/evals.json

@@ -131,6 +131,19 @@
         "distinguishes concat filter, concat demuxer, and concat protocol without executing"
       ],
       "case_set": "release"
+    },
+    {
+      "id": "podcast-audio-evidence-plan",
+      "prompt": "Analyze this supplied podcast WAV for long pauses, loudness, true peak, clipping, and transcript-aligned edit candidates. Save a plan, but do not alter the recording or make editorial cuts.",
+      "expected_output": "Probe the source, inventory required filters, run explicitly bounded silence, EBU R128, and peak measurements with recorded build/commands/thresholds, validate transcript timing quality, and create a no-overwrite plan whose ranges retain reasons, evidence, confidence, handles, fades, output contract, and needs-listening-review status.",
+      "assertions": [
+        "records exact measurement commands, build, window, thresholds, and limitations",
+        "reports missing filters or measurements as unavailable rather than inventing values",
+        "keeps silence, clipping, and transcript-derived ranges as candidates pending listening review",
+        "preserves the raw recording and refuses report overwrite",
+        "records source ranges, actions, reasons, evidence, confidence, handles, fades, and output contract"
+      ],
+      "case_set": "release"
     }
   ]
 }

ffmpeg/references/audio-and-podcast-editing.md

@@ -27,6 +27,10 @@ Relevant filters include:
 
 Record the exact interval, filter options, channel mode, FFmpeg build, and unfiltered source. Noise floors, breaths, room tone, music, cross-talk, and codec artifacts can invalidate a generic threshold.
 
+`scripts/audio-inspect` implements a bounded evidence path. Select `--measure-silence`, `--measure-loudness`, and/or `--measure-clipping`; each requested measurement records the command, analysis window, filter, thresholds, and local build. A missing filter produces `UNAVAILABLE`, not a fabricated value. Silence intervals and peak-based clipping flags are candidates with a listening-review gate.
+
+Pass `--transcript` only a timed JSON document with a non-empty `quality` object and `segments` containing `start`, `end`, and `text`. The report retains timing and a text digest rather than reproducing transcript content. A supplied `proposed_action` remains `needs_listening_review`. Output codec/rate/layout and loudness limits form a declared contract; handles and fades are plan fields, not executed edits. `--report-output` creates a new file exclusively and refuses overwrite.
+
 For loudness normalization, a measured first pass followed by a parameterized second pass is more reviewable than assuming one-pass behavior meets a delivery policy. Verify the rendered output again; a filter’s reported target is not acceptance evidence by itself.
 
 ## Editing and processing

ffmpeg/references/synthetic-media-fixtures.md

@@ -18,7 +18,7 @@ The command refuses a non-empty directory and bounds every FFmpeg/FFprobe comman
 | Irregular frame selection and offset/drift candidate | Container rates and stream durations are not a complete account of cadence or sync |
 | Compatible pair and concat output | Concat-demuxer success for the exact recorded streams and build |
 | Incompatible concat candidate | Dimensions, cadence, and sample-rate differences cause pre-concat rejection |
-| Audio analysis and faded WAVs | Declared silence/over-range regions plus a mechanical fade output for bounded measurement tests |
+| Audio analysis, speech-like, and faded WAVs | Declared silence/over-range regions, a frequency-modulated voiced-like source, and a mechanical fade output for bounded measurement tests |
 | SRT and subtitle-stream MKV | Subtitle source and explicit stream preservation; burn-in is exercised when the local filter exists and otherwise recorded unavailable |
 | Three boundary PNGs | Samples around one timestamp with an explicit no-whole-video-claim boundary |
 

ffmpeg/scripts/audio-inspect

@@ -1,42 +1,447 @@
 #!/usr/bin/env python3
-"""Inspect audio metadata with bounded, read-only ffprobe."""
+"""Produce bounded audio evidence and a reviewable podcast edit plan."""
+
+from __future__ import annotations
+
 import argparse
+import hashlib
 import json
+import math
+import os
+import re
 import shutil
 import subprocess
 import sys
+from pathlib import Path
+from typing import Any
 
 
-def main():
+class AudioError(Exception):
+    def __init__(self, status: str, message: str, exit_code: int = 2) -> None:
+        super().__init__(message)
+        self.status = status
+        self.message = message
+        self.exit_code = exit_code
+
+
+def resolve_tool(name: str) -> str:
+    expanded = os.path.expanduser(name)
+    if os.path.sep in expanded:
+        candidate = Path(expanded)
+        if candidate.is_file() and os.access(candidate, os.X_OK):
+            return str(candidate.resolve())
+    else:
+        resolved = shutil.which(name)
+        if resolved:
+            return resolved
+    raise AudioError("missing_tool", f"executable not found: {name}", 3)
+
+
+def run(argv: list[str], timeout: float) -> subprocess.CompletedProcess[str]:
+    try:
+        return subprocess.run(
+            argv,
+            stdin=subprocess.DEVNULL,
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            check=False,
+        )
+    except subprocess.TimeoutExpired as exc:
+        raise AudioError("timeout", f"command exceeded {timeout:g} seconds", 4) from exc
+    except OSError as exc:
+        raise AudioError("tool_start_failed", str(exc), 3) from exc
+
+
+def require_success(result: subprocess.CompletedProcess[str], status: str) -> None:
+    if result.returncode:
+        detail = result.stderr.strip()[-2000:] or result.stdout.strip()[-2000:]
+        raise AudioError(status, detail or f"command exited {result.returncode}", 1)
+
+
+def finite(value: Any) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)
+
+
+def ffmpeg_window(args: argparse.Namespace) -> list[str]:
+    values: list[str] = []
+    if args.analysis_start:
+        values.extend(["-ss", f"{args.analysis_start:g}"])
+    values.extend(["-t", f"{args.analysis_duration:g}"])
+    return values
+
+
+def inventory_filters(ffmpeg: str, timeout: float) -> set[str]:
+    result = run([ffmpeg, "-hide_banner", "-filters"], timeout)
+    require_success(result, "filter_inventory_failed")
+    names: set[str] = set()
+    for line in result.stdout.splitlines():
+        parts = line.split()
+        if len(parts) >= 2 and set(parts[0]).issubset(set(".TSCAVN|->")):
+            names.add(parts[1])
+    return names
+
+
+def parse_probe(stdout: str) -> dict[str, Any]:
+    try:
+        document = json.loads(stdout)
+    except json.JSONDecodeError as exc:
+        raise AudioError("invalid_json", f"invalid ffprobe JSON: {exc}", 1) from exc
+    if not isinstance(document, dict):
+        raise AudioError("invalid_json", "ffprobe JSON root must be an object", 1)
+    streams = document.get("streams", [])
+    if not isinstance(streams, list) or not any(
+        isinstance(stream, dict) and stream.get("codec_type") == "audio" for stream in streams
+    ):
+        raise AudioError("audio_stream_missing", "probe did not report an audio stream", 1)
+    return document
+
+
+def probe_duration(document: dict[str, Any]) -> float | None:
+    raw = document.get("format", {}).get("duration")
+    try:
+        return float(raw)
+    except (TypeError, ValueError):
+        return None
+
+
+def parse_silence(log: str, analysis_end: float | None) -> list[dict[str, Any]]:
+    starts = [float(value) for value in re.findall(r"silence_start:\s*([-+0-9.eE]+)", log)]
+    ends = [
+        (float(end), float(duration))
+        for end, duration in re.findall(
+            r"silence_end:\s*([-+0-9.eE]+)\s*\|\s*silence_duration:\s*([-+0-9.eE]+)",
+            log,
+        )
+    ]
+    intervals: list[dict[str, Any]] = []
+    for index, start in enumerate(starts):
+        if index < len(ends):
+            end, measured_duration = ends[index]
+            intervals.append({"start": start, "end": end, "duration": measured_duration})
+        else:
+            intervals.append(
+                {
+                    "start": start,
+                    "end": analysis_end,
+                    "duration": None if analysis_end is None else max(0.0, analysis_end - start),
+                    "open_ended": True,
+                }
+            )
+    return intervals
+
+
+def last_measure(pattern: str, log: str) -> float | None:
+    values = re.findall(pattern, log, flags=re.MULTILINE)
+    if not values:
+        return None
+    value = values[-1]
+    if value.lower() in {"-inf", "inf", "+inf"}:
+        return None
+    return float(value)
+
+
+def measurement(
+    ffmpeg: str,
+    source: str,
+    filter_name: str,
+    filter_expression: str,
+    filters: set[str],
+    args: argparse.Namespace,
+) -> tuple[dict[str, Any], str | None]:
+    if filter_name not in filters:
+        return {
+            "status": "UNAVAILABLE",
+            "filter": filter_name,
+            "reason": f"local FFmpeg filter inventory does not contain {filter_name}",
+        }, None
+    command = [
+        ffmpeg,
+        "-nostats",
+        "-v",
+        "info",
+        *ffmpeg_window(args),
+        "-i",
+        source,
+        "-map",
+        "0:a:0",
+        "-af",
+        filter_expression,
+        "-f",
+        "null",
+        "-",
+    ]
+    result = run(command, args.timeout)
+    require_success(result, f"{filter_name}_failed")
+    return {
+        "status": "MEASURED",
+        "filter": filter_name,
+        "command": command,
+        "analysis_window": {
+            "start_seconds": args.analysis_start,
+            "duration_seconds": args.analysis_duration,
+        },
+    }, result.stderr
+
+
+def load_transcript(path: str | None, source_duration: float | None) -> dict[str, Any]:
+    if path is None:
+        return {"status": "NOT_PROVIDED", "candidates": []}
+    try:
+        document = json.loads(Path(path).read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise AudioError("invalid_transcript", f"could not load transcript JSON: {exc}") from exc
+    if not isinstance(document, dict) or not isinstance(document.get("segments"), list):
+        raise AudioError("invalid_transcript", "transcript must contain a segments array")
+    quality = document.get("quality")
+    if not isinstance(quality, dict) or not quality:
+        raise AudioError("invalid_transcript", "transcript must disclose timing/alignment quality")
+    candidates: list[dict[str, Any]] = []
+    for index, segment in enumerate(document["segments"]):
+        if not isinstance(segment, dict):
+            raise AudioError("invalid_transcript", f"segment {index} must be an object")
+        start, end, text = segment.get("start"), segment.get("end"), segment.get("text")
+        if (
+            not finite(start)
+            or not finite(end)
+            or start < 0
+            or end <= start
+            or not isinstance(text, str)
+        ):
+            raise AudioError("invalid_transcript", f"segment {index} has invalid timing or text")
+        if source_duration is not None and end > source_duration:
+            raise AudioError("invalid_transcript", f"segment {index} exceeds source duration")
+        candidates.append(
+            {
+                "id": segment.get("id", f"transcript-{index + 1}"),
+                "source_range": {"in": start, "out": end},
+                "action": segment.get("proposed_action", "review"),
+                "reason": segment.get("reason", "transcript navigation candidate"),
+                "evidence": {
+                    "type": "transcript",
+                    "segment_index": index,
+                    "text_sha256": hashlib.sha256(text.encode()).hexdigest(),
+                    "timing_quality": quality,
+                },
+                "confidence": segment.get("confidence"),
+                "review_status": "needs_listening_review",
+            }
+        )
+    return {"status": "CANDIDATES_ONLY", "quality": quality, "candidates": candidates}
+
+
+def write_exclusive(path: str, value: object) -> None:
+    target = Path(path).expanduser()
+    try:
+        with target.open("x", encoding="utf-8") as handle:
+            json.dump(value, handle, indent=2, sort_keys=True)
+            handle.write("\n")
+    except FileExistsError as exc:
+        raise AudioError("output_exists", f"refusing to overwrite report: {target}") from exc
+    except OSError as exc:
+        raise AudioError("report_write_failed", str(exc)) from exc
+
+
+def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("input")
+    parser.add_argument("--ffmpeg", default="ffmpeg")
     parser.add_argument("--ffprobe", default="ffprobe")
-    parser.add_argument("--timeout", type=float, default=15.0)
-    parser.add_argument("--silence", action="store_true")
+    parser.add_argument("--timeout", type=float, default=30.0)
+    parser.add_argument("--analysis-start", type=float, default=0.0)
+    parser.add_argument("--analysis-duration", type=float, default=300.0)
+    parser.add_argument("--silence", action="store_true", help="alias for --measure-silence")
+    parser.add_argument("--measure-silence", action="store_true")
+    parser.add_argument("--measure-loudness", action="store_true")
+    parser.add_argument("--measure-clipping", action="store_true")
+    parser.add_argument("--silence-threshold", default="-40dB")
+    parser.add_argument("--silence-duration", type=float, default=0.5)
+    parser.add_argument("--clipping-threshold-db", type=float, default=-0.1)
+    parser.add_argument("--transcript", help="timed transcript JSON with disclosed quality")
+    parser.add_argument("--handles", type=float, default=0.05)
+    parser.add_argument("--fade-duration", type=float, default=0.01)
+    parser.add_argument("--target-lufs", type=float)
+    parser.add_argument("--true-peak-limit", type=float)
+    parser.add_argument("--output-codec")
+    parser.add_argument("--output-sample-rate", type=int)
+    parser.add_argument("--output-channel-layout")
+    parser.add_argument("--report-output", help="write report to a new path; overwrite is refused")
     parser.add_argument("--json", action="store_true")
-    args = parser.parse_args()
-    tool = shutil.which(args.ffprobe)
-    if not tool:
-        print(json.dumps({"ok": False, "status": "missing_tool", "error": "ffprobe not found"}))
-        return 3
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = build_parser().parse_args(argv)
     try:
-        result = subprocess.run([tool, "-v", "error", "-show_format", "-show_streams", "-of", "json", args.input], capture_output=True, text=True, timeout=args.timeout)
-    except subprocess.TimeoutExpired:
-        print(json.dumps({"ok": False, "status": "timeout", "error": "ffprobe timed out"}))
-        return 4
-    if result.returncode:
-        print(json.dumps({"ok": False, "status": "probe_failed", "error": result.stderr.strip() or "ffprobe failed"}))
-        return 1
-    try:
-        document = json.loads(result.stdout)
-    except json.JSONDecodeError:
-        print(json.dumps({"ok": False, "status": "invalid_json", "error": "invalid ffprobe JSON"}))
-        return 1
-    payload = {"ok": True, "status": "ok", "input": args.input, "probe": document}
-    if args.silence:
-        payload["silence"] = {"status": "candidate_only", "note": "silence intervals require review; no editorial cut was made"}
-    print(json.dumps(payload, indent=2, sort_keys=True))
-    return 0
+        if not 0 < args.timeout <= 300:
+            raise AudioError("invalid_limit", "--timeout must be greater than 0 and at most 300")
+        if not args.analysis_start >= 0 or not 0 < args.analysis_duration <= 3600:
+            raise AudioError("invalid_limit", "analysis window must be within 0-3600 seconds")
+        if not 0 < args.silence_duration <= args.analysis_duration:
+            raise AudioError(
+                "invalid_threshold", "silence duration must fit inside the analysis window"
+            )
+        if not -20 <= args.clipping_threshold_db <= 0:
+            raise AudioError(
+                "invalid_threshold", "clipping threshold must be between -20 and 0 dBFS"
+            )
+        if args.handles < 0 or args.fade_duration < 0:
+            raise AudioError("invalid_plan", "handles and fade duration must be non-negative")
+
+        ffprobe = resolve_tool(args.ffprobe)
+        source = str(Path(args.input).expanduser())
+        probe_command = [
+            ffprobe,
+            "-v",
+            "error",
+            "-show_format",
+            "-show_streams",
+            "-of",
+            "json",
+            source,
+        ]
+        probe_result = run(probe_command, args.timeout)
+        require_success(probe_result, "probe_failed")
+        probe = parse_probe(probe_result.stdout)
+        source_duration = probe_duration(probe)
+
+        requested = (
+            args.silence or args.measure_silence or args.measure_loudness or args.measure_clipping
+        )
+        ffmpeg: str | None = None
+        filters: set[str] = set()
+        ffmpeg_version: str | None = None
+        if requested:
+            ffmpeg = resolve_tool(args.ffmpeg)
+            version_result = run([ffmpeg, "-version"], args.timeout)
+            require_success(version_result, "version_failed")
+            ffmpeg_version = version_result.stdout.splitlines()[0]
+            filters = inventory_filters(ffmpeg, args.timeout)
+
+        evidence: dict[str, Any] = {}
+        if args.silence or args.measure_silence:
+            result, log = measurement(
+                ffmpeg or "ffmpeg",
+                source,
+                "silencedetect",
+                f"silencedetect=noise={args.silence_threshold}:d={args.silence_duration:g}",
+                filters,
+                args,
+            )
+            result.update(
+                {
+                    "evidence_class": "threshold_candidate",
+                    "threshold": args.silence_threshold,
+                    "minimum_duration_seconds": args.silence_duration,
+                    "editorial_status": "needs_listening_review",
+                }
+            )
+            if log is not None:
+                analysis_end = args.analysis_start + min(
+                    args.analysis_duration,
+                    source_duration or args.analysis_duration,
+                )
+                result["intervals"] = parse_silence(log, analysis_end)
+            evidence["silence"] = result
+
+        if args.measure_loudness:
+            result, log = measurement(
+                ffmpeg or "ffmpeg",
+                source,
+                "ebur128",
+                "ebur128=peak=true",
+                filters,
+                args,
+            )
+            if log is not None:
+                result["integrated_lufs"] = last_measure(r"^\s*I:\s+(-?inf|[-+0-9.]+)\s+LUFS", log)
+                result["true_peak_dbfs"] = last_measure(
+                    r"^\s*Peak:\s+(-?inf|[-+0-9.]+)\s+dBFS", log
+                )
+                result["target_lufs"] = args.target_lufs
+                result["true_peak_limit_dbfs"] = args.true_peak_limit
+            evidence["loudness"] = result
+
+        if args.measure_clipping:
+            result, log = measurement(
+                ffmpeg or "ffmpeg",
+                source,
+                "astats",
+                "astats=metadata=1:reset=0",
+                filters,
+                args,
+            )
+            if log is not None:
+                peak = last_measure(r"Peak level dB:\s+(-?inf|[-+0-9.]+)", log)
+                result.update(
+                    {
+                        "peak_level_dbfs": peak,
+                        "candidate_threshold_dbfs": args.clipping_threshold_db,
+                        "clipping_candidate": peak is not None
+                        and peak >= args.clipping_threshold_db,
+                        "editorial_status": "needs_listening_review",
+                    }
+                )
+            evidence["clipping"] = result
+
+        transcript = load_transcript(args.transcript, source_duration)
+        plan = {
+            "schema_version": 1,
+            "source": source,
+            "source_preservation": "original_untouched",
+            "overwrite_policy": "refuse",
+            "output_contract": {
+                "target_lufs": args.target_lufs,
+                "true_peak_limit_dbfs": args.true_peak_limit,
+                "codec": args.output_codec,
+                "sample_rate": args.output_sample_rate,
+                "channel_layout": args.output_channel_layout,
+            },
+            "candidates": [
+                {
+                    **candidate,
+                    "handles_seconds": args.handles,
+                    "fade_seconds": args.fade_duration,
+                }
+                for candidate in transcript["candidates"]
+            ],
+            "assembly_options": {
+                "voice_music_separation": "UNPLANNED",
+                "ducking": "UNPLANNED",
+                "intro_outro": "UNPLANNED",
+                "chapters": "UNPLANNED",
+                "metadata": "UNPLANNED",
+            },
+            "approval_gate": "no candidate becomes an edit without listening review",
+        }
+        report = {
+            "ok": True,
+            "status": "MEASURED_WITH_BOUNDARIES" if evidence else "PROBED",
+            "input": source,
+            "tool_context": {"ffprobe_command": probe_command, "ffmpeg_version": ffmpeg_version},
+            "probe": probe,
+            "analysis": evidence,
+            "transcript": transcript,
+            "podcast_edit_plan": plan,
+            "unverified": [
+                "speaker identity",
+                "transcript semantic accuracy",
+                "editorial suitability of silence or clipping candidates",
+                "listening quality",
+                "downstream compatibility",
+            ],
+        }
+        if args.report_output:
+            write_exclusive(args.report_output, report)
+        print(
+            json.dumps(report, sort_keys=True, separators=(",", ":"))
+            if args.json
+            else json.dumps(report, indent=2, sort_keys=True)
+        )
+        return 0
+    except AudioError as exc:
+        print(json.dumps({"ok": False, "status": exc.status, "error": exc.message}, sort_keys=True))
+        return exc.exit_code
 
 
 if __name__ == "__main__":

ffmpeg/scripts/generate-media-fixtures

@@ -321,8 +321,7 @@ def main(argv: list[str] | None = None) -> int:
 
         concat_list = workspace / "compatible.ffconcat"
         concat_list.write_text(
-            "ffconcat version 1.0\n"
-            + "".join(f"file '{path.name}'\n" for path in compatible),
+            "ffconcat version 1.0\n" + "".join(f"file '{path.name}'\n" for path in compatible),
             encoding="utf-8",
         )
         fixtures.append(
@@ -472,6 +471,29 @@ def main(argv: list[str] | None = None) -> int:
             args.timeout,
         )
 
+        speech_like = workspace / "speech-like-audio.wav"
+        speech_like_command = generate(
+            speech_like,
+            [
+                "-f",
+                "lavfi",
+                "-i",
+                "aevalsrc=if(between(t\\,1\\,1.7)\\,0\\,0.35*sin(2*PI*(180+40*sin(2*PI*3*t))*t)):s=48000:d=3",
+                "-c:a",
+                "pcm_s16le",
+            ],
+        )
+        add_media_fixture(
+            fixtures,
+            ffprobe,
+            speech_like,
+            "synthetic-speech-like-analysis-source",
+            speech_like_command,
+            ["frequency-modulated voiced-like tone", "declared 0.7 second silence interval"],
+            ["synthetic modulation is not human speech and cannot validate intelligibility"],
+            args.timeout,
+        )
+
         subtitle_text = workspace / "captions.srt"
         subtitle_text.write_text(
             "1\n00:00:00,200 --> 00:00:00,800\nSynthetic caption\n",

ffmpeg/scripts/test_media_workflows.py

@@ -308,10 +308,210 @@ def test_audio_inspect_reports_missing_ffprobe(tmp_path: Path) -> None:
     assert json.loads(result.stdout) == {
         "ok": False,
         "status": "missing_tool",
-        "error": "ffprobe not found",
+        "error": f"executable not found: {missing_ffprobe}",
     }
 
 
+def test_audio_inspect_rejects_malformed_probe_output(tmp_path: Path) -> None:
+    fake_probe = tmp_path / "ffprobe"
+    fake_probe.write_text("#!/bin/sh\nprintf 'not-json\\n'\n")
+    fake_probe.chmod(0o755)
+
+    result = run_script("audio-inspect", "input.wav", "--ffprobe", str(fake_probe), "--json")
+
+    assert result.returncode == 1
+    assert json.loads(result.stdout)["status"] == "invalid_json"
+
+
+def test_audio_inspect_reports_missing_measurement_filter(tmp_path: Path) -> None:
+    fake_probe = tmp_path / "ffprobe"
+    fake_probe.write_text(
+        '#!/bin/sh\nprintf \'%s\\n\' \'{"streams":[{"codec_type":"audio"}],"format":{"duration":"3"}}\'\n'
+    )
+    fake_probe.chmod(0o755)
+    fake_ffmpeg = tmp_path / "ffmpeg"
+    fake_ffmpeg.write_text(
+        "#!/bin/sh\ncase \"$*\" in *-version*) printf 'ffmpeg version fake\\n' ;; *-filters*) printf 'Filters:\\n' ;; *) exit 99 ;; esac\n"
+    )
+    fake_ffmpeg.chmod(0o755)
+
+    result = run_script(
+        "audio-inspect",
+        "input.wav",
+        "--ffprobe",
+        str(fake_probe),
+        "--ffmpeg",
+        str(fake_ffmpeg),
+        "--measure-silence",
+        "--json",
+    )
+
+    assert result.returncode == 0, result.stdout
+    silence = json.loads(result.stdout)["analysis"]["silence"]
+    assert silence["status"] == "UNAVAILABLE"
+    assert silence["filter"] == "silencedetect"
+    assert "intervals" not in silence
+
+
+def test_audio_inspect_measures_synthetic_candidates_and_builds_plan(tmp_path: Path) -> None:
+    if shutil.which("ffmpeg") is None or shutil.which("ffprobe") is None:
+        import pytest
+
+        pytest.skip("ffmpeg and ffprobe are required for measured audio evidence")
+
+    fixture_workspace = tmp_path / "fixtures"
+    fixture_result = run_script("generate-media-fixtures", str(fixture_workspace), "--json")
+    assert fixture_result.returncode == 0, fixture_result.stdout + fixture_result.stderr
+    transcript = write_json(
+        tmp_path / "transcript.json",
+        {
+            "quality": {
+                "method": "synthetic fixture timing",
+                "alignment": "declared, not speech-recognized",
+            },
+            "segments": [
+                {
+                    "id": "segment-1",
+                    "start": 0.2,
+                    "end": 0.8,
+                    "text": "synthetic phrase placeholder",
+                    "proposed_action": "keep",
+                    "reason": "exercise transcript alignment",
+                    "confidence": 1.0,
+                }
+            ],
+        },
+    )
+    report_path = tmp_path / "audio-report.json"
+    arguments = (
+        str(fixture_workspace / "speech-like-audio.wav"),
+        "--measure-silence",
+        "--measure-loudness",
+        "--measure-clipping",
+        "--silence-threshold",
+        "-50dB",
+        "--silence-duration",
+        "0.5",
+        "--transcript",
+        str(transcript),
+        "--target-lufs",
+        "-16",
+        "--true-peak-limit",
+        "-1",
+        "--output-codec",
+        "pcm_s16le",
+        "--output-sample-rate",
+        "48000",
+        "--output-channel-layout",
+        "mono",
+        "--report-output",
+        str(report_path),
+        "--json",
+    )
+
+    result = run_script("audio-inspect", *arguments)
+
+    assert result.returncode == 0, result.stdout + result.stderr
+    report = json.loads(result.stdout)
+    assert report == json.loads(report_path.read_text())
+    assert report["analysis"]["silence"]["status"] == "MEASURED"
+    assert any(
+        interval["duration"] >= 0.5 for interval in report["analysis"]["silence"]["intervals"]
+    )
+    assert report["analysis"]["loudness"]["integrated_lufs"] is not None
+    assert report["analysis"]["loudness"]["true_peak_dbfs"] is not None
+    assert report["analysis"]["clipping"]["peak_level_dbfs"] is not None
+    candidate = report["podcast_edit_plan"]["candidates"][0]
+    assert candidate["source_range"] == {"in": 0.2, "out": 0.8}
+    assert candidate["review_status"] == "needs_listening_review"
+    assert candidate["handles_seconds"] == 0.05
+    assert candidate["fade_seconds"] == 0.01
+    assert "text_sha256" in candidate["evidence"]
+    assert report["podcast_edit_plan"]["overwrite_policy"] == "refuse"
+    assert "listening quality" in report["unverified"]
+
+    repeated_arguments = (*arguments[:-3], "--json")
+    repeated = run_script("audio-inspect", *repeated_arguments)
+    assert repeated.returncode == 0, repeated.stdout + repeated.stderr
+    repeated_report = json.loads(repeated.stdout)
+    assert repeated_report == report
+
+    overwrite = run_script("audio-inspect", *arguments)
+    assert overwrite.returncode == 2
+    assert json.loads(overwrite.stdout)["status"] == "output_exists"
+
+    clipping_result = run_script(
+        "audio-inspect",
+        str(fixture_workspace / "audio-analysis.wav"),
+        "--measure-clipping",
+        "--json",
+    )
+    assert clipping_result.returncode == 0, clipping_result.stdout + clipping_result.stderr
+    assert json.loads(clipping_result.stdout)["analysis"]["clipping"]["clipping_candidate"] is True
+
+    treated = tmp_path / "treated.wav"
+    render = subprocess.run(
+        [
+            "ffmpeg",
+            "-v",
+            "error",
+            "-n",
+            "-i",
+            str(fixture_workspace / "speech-like-audio.wav"),
+            "-af",
+            "afade=t=in:d=0.05,afade=t=out:st=2.95:d=0.05",
+            "-c:a",
+            "pcm_s16le",
+            str(treated),
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert render.returncode == 0, render.stderr
+
+    probe_paths: list[Path] = []
+    for name, media_path in (
+        ("source", fixture_workspace / "speech-like-audio.wav"),
+        ("treated", treated),
+    ):
+        probe_result = subprocess.run(
+            [
+                "ffprobe",
+                "-v",
+                "error",
+                "-show_format",
+                "-show_streams",
+                "-of",
+                "json",
+                str(media_path),
+            ],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        assert probe_result.returncode == 0, probe_result.stderr
+        probe_paths.append(
+            write_json(tmp_path / f"{name}-probe.json", json.loads(probe_result.stdout))
+        )
+    verify = run_script("media-verify", *(str(path) for path in probe_paths))
+    assert verify.returncode == 0, verify.stdout + verify.stderr
+    assert json.loads(verify.stdout)["status"] == "pass"
+
+
+def test_audio_inspect_rejects_invalid_thresholds(tmp_path: Path) -> None:
+    result = run_script(
+        "audio-inspect",
+        str(tmp_path / "missing.wav"),
+        "--silence-duration",
+        "0",
+        "--json",
+    )
+
+    assert result.returncode == 2
+    assert json.loads(result.stdout)["status"] == "invalid_threshold"
+
+
 def test_media_verify_passes_matching_probe_files(tmp_path: Path) -> None:
     input_probe = write_json(tmp_path / "input-probe.json", probe("video", "audio"))
     output_probe = write_json(tmp_path / "output-probe.json", probe("video", "audio"))
@@ -437,6 +637,7 @@ def test_generate_media_fixtures_covers_real_boundaries(tmp_path: Path) -> None:
         "audio-offset-and-duration-drift-candidate",
         "audio-silence-and-peak-candidates",
         "audio-fade-output",
+        "synthetic-speech-like-analysis-source",
         "subtitle-source-text",
         "subtitle-stream-survival",
         "bounded-boundary-frame",

ffmpeg/templates/podcast-edit-plan.md

@@ -5,20 +5,25 @@
 - Episode/source ID:
 - Intended listener and destination:
 - Output format/loudness requirements:
+- Codec, sample rate, and channel layout:
+- True-peak limit and measurement method:
 - Original preservation path:
+- Overwrite policy: refuse
 
 ## Evidence
 
 - Transcript and timing quality:
 - Audio probe:
-- Waveform/silence candidate method:
+- Silence threshold/minimum duration and exact command:
+- Loudness/true-peak method and exact command:
+- Peak/clipping candidate method and threshold:
 - Listening review segments:
 
 ## Decisions
 
-| Range | Action | Reason | Evidence | Confidence | Review |
-|---|---|---|---|---|---|
-| | keep/remove/shorten/treat | | | | |
+| Source range | Action | Reason | Evidence | Confidence | Handles/fades | Review status |
+|---|---|---|---|---|---|---|
+| | keep/remove/shorten/treat | | | | | needs listening review |
 
 ## Safety
 
@@ -27,3 +32,4 @@
 - Handles and fades:
 - Clipping/noise policy:
 - Missing measurement or playback evidence:
+- Voice/music separation, ducking, intro/outro, chapters, and metadata status: