Fix inserted narration and movie audio subtitle checks

Address the final three review findings on PR #2214, as approved by Drew. Count both sides of every non-equal transcript span so a short insertion or expanded replacement cannot evade the drift gate on a longer script. Preserve the existing length and run thresholds.

Honor --no-expect-audio for encoded silent tracks. Base subtitle requirements on detected audible speech so opting out of expected audio does not suppress captions for speech that is present.

Extract the first embedded subtitle stream as SRT when no sidecar is present and apply the same cue-end check to either source. Empty cues fail even for short narration, and extraction or malformed timing errors are reported as failures. Preserve the silent end-card allowance.

Validation: the new tests first reproduced ten failing cases across narration insertion, silent-track opt-out, and embedded subtitle handling. All 35 focused narration, checker-policy, and subtitle-text tests now pass. External media commands, audio and picture sampling, contact-sheet creation, synthesis, and ASR were mocked; no actual media inspection or live ASR was performed. Drew retains final video acceptance.

skills/proving-it-works-with-a-movie/scripts/check-movie

@@ -121,6 +121,19 @@ def contact_sheet(paths, out_path, count=12):
     return [paths.index(p) for p in picks]
 
 
+def subtitle_end(text):
+    """Return the last SRT cue's end time, or None when there are no cues."""
+    ends = []
+    for line in text.splitlines():
+        if "-->" in line:
+            end = line.split("-->")[1].strip().split()[0]
+            hh, mm, rest = end.split(":")
+            ss, _, ms = rest.partition(",")
+            ends.append(int(hh) * 3600 + int(mm) * 60 + int(ss)
+                        + int(ms or 0) / 1000)
+    return max(ends, default=None)
+
+
 def main():
     for stream in (sys.stdout, sys.stderr):
         if hasattr(stream, "reconfigure"):
@@ -180,40 +193,48 @@ def main():
         failures.append(f"duration is {duration:.2f}s - that is not a movie")
     if args.expect_audio and not as_:
         failures.append("expected narration but there is no audio stream")
-    if levels and not talking:
+    if args.expect_audio and levels and not talking:
         failures.append("the audio track is silent end to end")
 
     # a narrated movie with no subtitles fails for everyone watching it muted
-    if as_ and args.expect_subs:
+    if talking and args.expect_subs:
         srt = args.subs or args.movie.with_suffix(".srt")
         embedded = any(s["codec_type"] == "subtitle" for s in info["streams"])
+        subtitle_text, source = None, srt.name
         if srt.exists():
-            last = 0.0
-            for line in srt.read_text(encoding="utf-8-sig", errors="replace").splitlines():
-                if "-->" in line:
-                    end = line.split("-->")[1].strip().split()[0]
-                    hh, mm, rest = end.split(":")
-                    ss, _, ms = rest.partition(",")
-                    last = max(last, int(hh) * 3600 + int(mm) * 60 + int(ss)
-                               + int(ms or 0) / 1000)
-            # compare against where the narration ends, not the runtime: a
-            # silent end card is normal and must not read as missing subtitles
-            speech_end = float(last_talk + 1) if last_talk is not None else duration
-            print(f"subtitles   {srt.name}, last cue ends at {last:.1f}s "
-                  f"(narration ends {speech_end:.0f}s)")
-            if last < speech_end - 3.0:
-                failures.append(
-                    f"subtitles stop at {last:.0f}s but the narration runs to "
-                    f"{speech_end:.0f}s - {speech_end - last:.0f}s of speech "
-                    f"has no subtitles")
+            subtitle_text = srt.read_text(encoding="utf-8-sig", errors="replace")
         elif embedded:
-            print("subtitles   embedded subtitle stream present")
+            extracted = subprocess.run(
+                ["ffmpeg", "-nostdin", "-v", "error", "-i", str(args.movie),
+                 "-map", "0:s:0", "-f", "srt", "-"],
+                capture_output=True, text=True, encoding="utf-8", errors="replace")
+            if extracted.returncode != 0:
+                die(f"embedded subtitle extraction failed: {extracted.stderr.strip()[:200]}")
+            subtitle_text, source = extracted.stdout, "embedded"
         else:
             failures.append(
                 f"narrated, but no subtitles: expected {srt.name} beside the "
                 f"movie (or an embedded track). Run make-subtitles and burn "
                 f"them in; pass --no-expect-subtitles only for a movie nobody "
                 f"will ever watch muted.")
+        if subtitle_text is not None:
+            try:
+                last = subtitle_end(subtitle_text)
+            except (ValueError, IndexError):
+                die(f"invalid subtitle timing in {source}")
+            # compare against where the narration ends, not the runtime: a
+            # silent end card is normal and must not read as missing subtitles
+            speech_end = float(last_talk + 1)
+            if last is None:
+                failures.append(f"{source}: subtitles contain no cues")
+            else:
+                print(f"subtitles   {source}, last cue ends at {last:.1f}s "
+                      f"(narration ends {speech_end:.0f}s)")
+                if last < speech_end - 3.0:
+                    failures.append(
+                        f"subtitles stop at {last:.0f}s but the narration runs to "
+                        f"{speech_end:.0f}s - {speech_end - last:.0f}s of speech "
+                        f"has no subtitles")
     if not changes:
         failures.append("the picture never reaches a new state - this is a still, "
                         "not a movie")

skills/proving-it-works-with-a-movie/scripts/narrate

@@ -109,12 +109,12 @@ def structural_drift(text, heard):
     scores as more similar than two mispronounced ones. What is detectable,
     and what actually matters, is missing or invented CONTENT: a sentence
     the voice skipped, or a preamble it invented. Returns
-    (length_delta_fraction, longest_run_of_missing_or_changed_words).
+    (length_delta_fraction, longest_run_of_missing_added_or_changed_words).
     """
     want, got = norm(text), norm(heard)
     delta = abs(len(got) - len(want)) / max(1, len(want))
     ops = difflib.SequenceMatcher(a=want, b=got).get_opcodes()
-    worst = max((i2 - i1 for tag, i1, i2, _, _ in ops if tag in ("delete", "replace")),
+    worst = max((max(i2 - i1, j2 - j1) for tag, i1, i2, j1, j2 in ops if tag != "equal"),
                 default=0)
     return delta, worst
 

tests/proving-it-works-with-a-movie/test_checker.py

@@ -1,12 +1,111 @@
+import io
+import json
+import subprocess
+import sys
 import tempfile
+from contextlib import redirect_stdout
 import unittest
 from pathlib import Path
+from unittest.mock import patch
 
 from PIL import Image
 
 import fixtures
 
 
+class CheckerPolicyRegression(unittest.TestCase):
+    def check(self, expected_exit, *options, audio=True, levels=None,
+              embedded=None, sidecar=None, extraction_exit=0):
+        module = fixtures.load_script("check-movie")
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            movie = root / "movie.mp4"
+            movie.write_bytes(b"metadata fixture only")
+            if sidecar is not None:
+                movie.with_suffix(".srt").write_text(sidecar, encoding="utf-8-sig")
+            streams = [{"index": 0, "codec_type": "video", "codec_name": "h264",
+                        "width": 640, "height": 360}]
+            if audio:
+                streams.append({"index": 1, "codec_type": "audio", "codec_name": "aac"})
+            if embedded is not None:
+                streams.append({"index": 2, "codec_type": "subtitle", "codec_name": "mov_text"})
+            levels = levels if levels is not None else [-20.0] * 20
+            info = {"format": {"duration": str(len(levels))}, "streams": streams}
+
+            def media_command(cmd, **kwargs):
+                if cmd[0] == "ffprobe":
+                    return subprocess.CompletedProcess(cmd, 0, json.dumps(info), "")
+                if cmd[0] == "ffmpeg" and embedded is not None:
+                    self.assertEqual(cmd[cmd.index("-i") + 1], str(movie))
+                    self.assertEqual(cmd[cmd.index("-map") + 1], "0:s:0")
+                    self.assertEqual(cmd[cmd.index("-f") + 1], "srt")
+                    return subprocess.CompletedProcess(cmd, extraction_exit, embedded,
+                                                       "subtitle decode failed" if extraction_exit else "")
+                raise AssertionError(f"unexpected media command: {cmd}")
+
+            output = root / "report"
+            argv = ["check-movie", str(movie), "--out", str(output), "--json", *options]
+            stdout = io.StringIO()
+            with patch.object(sys, "argv", argv), \
+                 patch.object(module.shutil, "which", return_value="test-tool"), \
+                 patch.object(module.subprocess, "run", side_effect=media_command), \
+                 patch.object(module, "sample_picture", return_value=([], [0.1] * (len(levels) - 1))), \
+                 patch.object(module, "sample_sound", return_value=levels if audio else []), \
+                 patch.object(module, "contact_sheet", return_value=[]), \
+                 redirect_stdout(stdout):
+                try:
+                    code = module.main()
+                except SystemExit as error:
+                    code = error.code
+            self.assertEqual(code, expected_exit, stdout.getvalue())
+            report = output / "check.json"
+            failures = json.loads(report.read_text(encoding="utf-8"))["failures"] if report.exists() else []
+            return failures, stdout.getvalue()
+
+    def test_silent_encoded_track_passes_when_audio_is_not_expected(self):
+        failures, _ = self.check(0, "--no-expect-audio", levels=[-120.0] * 20)
+        self.assertEqual(failures, [])
+
+    def test_absent_audio_passes_when_audio_is_not_expected(self):
+        failures, _ = self.check(0, "--no-expect-audio", audio=False)
+        self.assertEqual(failures, [])
+
+    def test_silent_encoded_track_fails_when_audio_is_expected(self):
+        failures, _ = self.check(1, levels=[-120.0] * 20)
+        self.assertTrue(any("silent" in failure for failure in failures))
+
+    def test_audio_opt_out_still_requires_captions_for_audible_speech(self):
+        failures, _ = self.check(1, "--no-expect-audio")
+        self.assertTrue(any("no subtitles" in failure for failure in failures))
+
+    def test_subtitle_opt_out_allows_audible_speech_without_captions(self):
+        failures, _ = self.check(0, "--no-expect-audio", "--no-expect-subtitles")
+        self.assertEqual(failures, [])
+
+    def test_empty_embedded_track_fails_even_for_short_narration(self):
+        for seconds in (2, 20):
+            with self.subTest(seconds=seconds):
+                failures, _ = self.check(1, embedded="", levels=[-20.0] * seconds)
+                self.assertTrue(any("subtitle" in failure for failure in failures))
+
+    def test_sidecar_and_embedded_cues_must_reach_the_end_of_speech(self):
+        for source in ("sidecar", "embedded"):
+            for end, expected_exit in (("06,000", 1), ("10,000", 0)):
+                with self.subTest(source=source, end=end):
+                    subtitles = f"1\n00:00:00,000 --> 00:00:{end}\nUnicode λ caption\n"
+                    failures, _ = self.check(expected_exit, levels=[-20.0] * 10 + [-120.0] * 10,
+                                             **{source: subtitles})
+                    self.assertEqual(bool(failures), bool(expected_exit))
+
+    def test_embedded_extraction_failure_is_not_accepted(self):
+        _, diagnostics = self.check(2, embedded="", extraction_exit=1)
+        self.assertIn("subtitle", diagnostics)
+
+    def test_malformed_embedded_cue_is_a_reported_failure(self):
+        _, diagnostics = self.check(2, embedded="1\n00:00:00,000 --> invalid\ncaption\n")
+        self.assertIn("subtitle", diagnostics)
+
+
 class CheckerRegression(unittest.TestCase):
     @classmethod
     def setUpClass(cls) -> None:

tests/proving-it-works-with-a-movie/test_narration.py

@@ -17,7 +17,7 @@ SCRIPT = (
 
 
 class NarrationDriftRegression(unittest.TestCase):
-    def drift(self, expected_exit: int, heard: str) -> None:
+    def drift(self, expected_exit: int, heard: str, script: str = SCRIPT) -> None:
         missing = fixtures.missing_executables("uv")
         if missing:
             self.skipTest(
@@ -27,7 +27,7 @@ class NarrationDriftRegression(unittest.TestCase):
             work = Path(directory)
             script_path = work / "script.txt"
             heard_path = work / "heard.txt"
-            script_path.write_text(SCRIPT, encoding="utf-8")
+            script_path.write_text(script, encoding="utf-8")
             heard_path.write_text(heard, encoding="utf-8")
             result = fixtures.run_tool(
                 "narrate",
@@ -61,6 +61,22 @@ class NarrationDriftRegression(unittest.TestCase):
     def test_empty_clip_fails(self):
         self.drift(1, "you")
 
+    def test_inserted_runs_fail_even_when_total_length_is_close(self):
+        words = [f"word{i}" for i in range(50)]
+        for position in (0, 25, 50):
+            with self.subTest(position=position):
+                heard = words[:position] + "Before we begin please listen".split() + words[position:]
+                self.drift(1, " ".join(heard), " ".join(words))
+
+    def test_expanded_replacement_counts_the_added_words(self):
+        words = [f"word{i}" for i in range(50)]
+        heard = words[:25] + "Before we begin please listen".split() + words[26:]
+        self.drift(1, " ".join(heard), " ".join(words))
+
+    def test_short_insertions_keep_the_existing_tolerance(self):
+        words = [f"word{i}" for i in range(50)]
+        self.drift(0, "Please listen closely " + " ".join(words), " ".join(words))
+
     def test_cached_audio_requires_requested_verification(self):
         import json
         import sys