fix(movie): retain narration evidence through verification

Address Task 1 review round 1. Treat successful empty ASR output as speech failure rather than an unavailable verifier, reject empty chat claims within the existing bounded retry loop, and extend unsupported segmentation detection to supplementary CJK ideographs.

Withdraw cached acceptance before every revalidation so strict failures and interrupts cannot leave stale publication. Preserve nested manifest WAV paths on cache reacceptance, and measure a unique candidate before promotion so duration failures retain their evidence across reruns. Add structured movie geometry coverage plus both silent and source-audio mapping tests. All tests use uv --no-project with mocked synthesis, ASR, probes, and encoding; no media operation was run.

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

@@ -79,6 +79,10 @@ def has_audio_stream(path):
     return any(stream.get("codec_type") == "audio" for stream in streams)
 
 
+def movie_geometry(width, height, inner_height):
+    return {"scale": (width, inner_height), "pad": (width, height)}
+
+
 def make_card(scene, png, w, h, browser):
     if not browser:
         die("a `card` scene needs a browser (Chrome/Chromium) to render text; "
@@ -141,13 +145,15 @@ def main():
             src = base / sc["src"]
             target = dur(src)
             inner_h = int(sc.get("height", int(H * 0.82)))
+            geometry = movie_geometry(W, H, inner_h)
             source_audio = has_audio_stream(src)
             ain = ([] if source_audio
                    else ["-f", "lavfi", "-i", "anullsrc=r=44100:cl=stereo"])
             audio_map = "0:a:0" if source_audio else "1:a:0"
             run(["ffmpeg", "-nostdin", "-y", "-v", "error", "-i", str(src), *ain,
-                 "-vf", f"scale={W}:{inner_h}:force_original_aspect_ratio=decrease,"
-                        f"pad={W}:{H}:(ow-iw)/2:(oh-ih)/2:"
+                 "-vf", f"scale={geometry['scale'][0]}:{geometry['scale'][1]}:"
+                        f"force_original_aspect_ratio=decrease,"
+                        f"pad={geometry['pad'][0]}:{geometry['pad'][1]}:(ow-iw)/2:(oh-ih)/2:"
                         f"color=#101014,setsar=1",
                  "-af", f"volume={sc.get('gain_db', 0)}dB,apad",
                  "-r", str(FPS), "-t", f"{target:.3f}",

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

@@ -65,6 +65,9 @@ SEGMENTATION_SCRIPTS = (
     (0x3040, 0x30FF),  # Hiragana and Katakana
     (0x3400, 0x4DBF),  # CJK Extension A
     (0x4E00, 0x9FFF),  # CJK Unified Ideographs
+    (0xF900, 0xFAFF),  # CJK Compatibility Ideographs
+    (0x20000, 0x323AF),  # CJK Unified Ideograph extensions B through I
+    (0x2F800, 0x2FA1F),  # CJK Compatibility Ideographs Supplement
     (0x0E00, 0x0E7F),  # Thai
 )
 
@@ -119,7 +122,7 @@ def transcribe_local(wav, model="base.en"):
                 return None
             data = json.loads(result_path.read_text(encoding="utf-8"))
             text = data.get("text") if isinstance(data, dict) else None
-            return text.strip() if isinstance(text, str) and text.strip() else None
+            return text.strip() if isinstance(text, str) else None
     except (OSError, ValueError, subprocess.SubprocessError) as error:
         print(f"local ASR unavailable: {error}", file=sys.stderr)
         return None
@@ -322,8 +325,7 @@ def main():
             withdraw(sid)
             failures.append(sid)
             continue
-        if not cached:
-            withdraw(sid)
+        withdraw(sid)
         accepted = False
         for attempt in ((1,) if cached else (1, 2)):
             claimed = None
@@ -344,6 +346,9 @@ def main():
 
             # Preserve the engine transcript gate and the ASR drift thresholds.
             if claimed is not None:
+                if not norm(claimed):
+                    print(f"{sid}: chat transcript contains no speech", file=sys.stderr)
+                    continue
                 drift_result = structural_drift(text, claimed)
                 if drift_result is None:
                     print(f"{sid}: chat transcript comparison unavailable", file=sys.stderr)
@@ -381,15 +386,16 @@ def main():
                               f"change, worst run {worst})")
             if not verify:
                 print(f"{sid}: ok")
-            if not cached:
-                candidate.replace(args.outdir / f"{sid}.wav")
-                candidate = args.outdir / f"{sid}.wav"
             try:
                 clip_duration = duration(candidate)
             except Exception as error:  # unaccepted bytes remain as evidence
                 print(f"{sid}: duration failed ({error})", file=sys.stderr)
                 break
-            accept({"id": sid, "text": text, "wav": candidate.name,
+            if not cached:
+                candidate.replace(args.outdir / f"{sid}.wav")
+                candidate = args.outdir / f"{sid}.wav"
+            wav_name = previous["wav"] if cached else candidate.name
+            accept({"id": sid, "text": text, "wav": wav_name,
                     "duration": clip_duration, "synthesis": synthesis})
             accepted = True
             break

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

@@ -99,6 +99,7 @@ class NarrationDriftRegression(unittest.TestCase):
                     "--engine", "piper", "--verify", "on"]
             stdout, stderr = io.StringIO(), io.StringIO()
             with patch.object(sys, "argv", argv), \
+                 patch.object(module.shutil, "which", return_value="ffprobe"), \
                  patch.object(module, "openai_key", return_value=None), \
                  patch.object(module, "duration", return_value=1.0), \
                  patch.object(module, "say_piper", side_effect=AssertionError("expected cached clip")), \
@@ -138,6 +139,7 @@ class NarrationDriftRegression(unittest.TestCase):
                     "--verify", "off"]
             rejected_renders = []
             with patch.object(sys, "argv", argv), \
+                 patch.object(module.shutil, "which", return_value="ffprobe"), \
                  patch.object(module, "openai_key", return_value="test-key"), \
                  patch.object(module, "say_openai_chat", side_effect=synthesize), \
                  patch.object(module, "duration", return_value=1.0):
@@ -189,7 +191,10 @@ class TranscriptionProtocolRegression(unittest.TestCase):
                 diagnostics = io.StringIO()
                 with patch.object(module.subprocess, "run", side_effect=child), \
                      redirect_stderr(diagnostics):
-                    self.assertIsNone(module.transcribe_local(Path("clip.wav")))
+                    self.assertEqual(
+                        module.transcribe_local(Path("clip.wav")),
+                        "" if payload == '{"text": ""}' and code == 0 else None,
+                    )
                 self.assertIn("local ASR", diagnostics.getvalue())
 
     def test_fresh_and_off_then_on_clips_require_asr(self):
@@ -205,7 +210,7 @@ class TranscriptionProtocolRegression(unittest.TestCase):
                 def synthesize(text, wav, voice):
                     wav.write_bytes(b"branch policy fixture")
                 argv = ["narrate", str(scenes), str(output), "--engine", "piper", "--verify"]
-                with patch.object(module, "openai_key", return_value=None), patch.object(module, "say_piper", side_effect=synthesize), patch.object(module, "duration", return_value=1.0), patch.object(module, "transcribe_local", return_value=None):
+                with patch.object(module.shutil, "which", return_value="ffprobe"), patch.object(module, "openai_key", return_value=None), patch.object(module, "say_piper", side_effect=synthesize), patch.object(module, "duration", return_value=1.0), patch.object(module, "transcribe_local", return_value=None):
                     if cached:
                         with patch.object(sys, "argv", [*argv, "off"]), \
                              redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
@@ -229,6 +234,10 @@ class NarrationCacheRegression(unittest.TestCase):
         self.output = self.root / "voice"
         self.renders = []
 
+        which = patch.object(self.module.shutil, "which", return_value="ffprobe")
+        which.start()
+        self.addCleanup(which.stop)
+
         def synthesize(*args):
             text, wav, voice = args[-3:]
             self.renders.append((text, voice))

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

@@ -20,17 +20,20 @@ class NarrationPublicationContract(unittest.TestCase):
         self.scenes = self.root / "scenes.yaml"
         self.output = self.root / "narration"
 
-    def run_narrate(self, scenes, synthesize, *, verify="off", expected=0):
+    def run_narrate(self, scenes, synthesize, *, verify="off", expected=0, engine="piper",
+                    extra_options=(), transcript=None):
         self.scenes.write_text(json.dumps({"scenes": scenes}), encoding="utf-8")
-        argv = ["narrate", str(self.scenes), str(self.output), "--engine", "piper",
-                "--verify", verify]
+        argv = ["narrate", str(self.scenes), str(self.output), "--engine", engine,
+                "--verify", verify, *extra_options]
         stderr = io.StringIO()
         with patch.object(sys, "argv", argv), \
              patch.object(self.module.shutil, "which", return_value="ffprobe"), \
-             patch.object(self.module, "openai_key", return_value=None), \
+             patch.object(self.module, "openai_key", return_value="key" if engine.startswith("openai") else None), \
              patch.object(self.module, "say_piper", side_effect=synthesize), \
+             patch.object(self.module, "say_openai_chat",
+                          side_effect=lambda key, text, wav, voice: synthesize(text, wav, voice)), \
              patch.object(self.module, "duration", return_value=1.25), \
-             patch.object(self.module, "transcribe_local", return_value=None), \
+             patch.object(self.module, "transcribe_local", return_value=transcript), \
              redirect_stdout(io.StringIO()), redirect_stderr(stderr):
             self.assertEqual(self.module.main(), expected, stderr.getvalue())
         manifest = self.output / "manifest.json"
@@ -72,8 +75,12 @@ class NarrationPublicationContract(unittest.TestCase):
 
         scenes = [{"id": "reject", "narration": "Reject this"},
                   {"id": "raise", "narration": "Raise here"}]
-        first = self.run_narrate(scenes, synthesize, expected=1)
-        second = self.run_narrate(scenes, synthesize, expected=1)
+        def accepted(text, wav, voice):
+            wav.write_bytes(b"accepted")
+
+        self.run_narrate(scenes, accepted)
+        first = self.run_narrate(scenes, synthesize, expected=1, extra_options=("--force",))
+        second = self.run_narrate(scenes, synthesize, expected=1, extra_options=("--force",))
 
         self.assertEqual(first, [])
         self.assertEqual(second, [])
@@ -103,6 +110,27 @@ class NarrationPublicationContract(unittest.TestCase):
         manifest = json.loads((self.output / "manifest.json").read_text())
         self.assertEqual([entry["id"] for entry in manifest], ["accepted"])
 
+    def test_duration_failures_keep_distinct_attempt_bytes_across_reruns(self):
+        scenes = [{"id": "clip", "narration": "Measure this clip"}]
+
+        def synthesize(text, wav, voice):
+            wav.write_bytes(f"attempt {len(list(self.output.glob('.clip.attempt-*.wav'))) + 1}".encode())
+
+        for expected_attempts in (1, 2):
+            self.scenes.write_text(json.dumps({"scenes": scenes}), encoding="utf-8")
+            argv = ["narrate", str(self.scenes), str(self.output), "--engine", "piper", "--verify", "off"]
+            with patch.object(sys, "argv", argv), \
+                 patch.object(self.module.shutil, "which", return_value="ffprobe"), \
+                 patch.object(self.module, "openai_key", return_value=None), \
+                 patch.object(self.module, "say_piper", side_effect=synthesize), \
+                 patch.object(self.module, "duration", side_effect=RuntimeError("ffprobe failed")), \
+                 redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
+                self.assertEqual(self.module.main(), 1)
+            attempts = sorted(self.output.glob(".clip.attempt-*.wav"))
+            self.assertEqual(len(attempts), expected_attempts)
+            self.assertEqual(len({path.read_bytes() for path in attempts}), expected_attempts)
+            self.assertFalse((self.output / "clip.wav").exists())
+
     def test_two_accepted_scenes_are_published_together(self):
         def synthesize(text, wav, voice):
             wav.write_bytes(text.encode())
@@ -145,6 +173,19 @@ class NarrationPublicationContract(unittest.TestCase):
             with self.assertRaises(SystemExit):
                 self.module.main()
 
+    def test_fresh_unsupported_chat_transcript_is_rejected_before_synthesis(self):
+        called = []
+
+        def synthesize(*args):
+            called.append(args)
+
+        manifest = self.run_narrate(
+            [{"id": "clip", "narration": "\U00020000\U00020001"}], synthesize,
+            engine="openai-chat", expected=1,
+        )
+        self.assertEqual(manifest, [])
+        self.assertEqual(called, [])
+
     def test_cached_unsupported_chat_transcript_withdraws_acceptance_without_asr(self):
         self.output.mkdir()
         (self.output / "clip.wav").write_bytes(b"cached bytes")
@@ -157,17 +198,85 @@ class NarrationPublicationContract(unittest.TestCase):
         self.scenes.write_text(json.dumps({"scenes": [
             {"id": "clip", "narration": "\u77ed\u6587"}
         ]}), encoding="utf-8")
-        argv = ["narrate", str(self.scenes), str(self.output), "--engine", "openai-chat",
-                "--verify", "off"]
+        for mode in ("off", "auto"):
+            with self.subTest(mode=mode):
+                (self.output / "manifest.json").write_text(json.dumps([{
+                    "id": "clip", "text": "\u77ed\u6587", "wav": "clip.wav", "duration": 1.0,
+                    "synthesis": synthesis,
+                }]), encoding="utf-8")
+                argv = ["narrate", str(self.scenes), str(self.output), "--engine", "openai-chat",
+                        "--verify", mode]
+                with patch.object(sys, "argv", argv), \
+                     patch.object(self.module.shutil, "which", return_value="ffprobe"), \
+                     patch.object(self.module, "openai_key", return_value="key"), \
+                     patch.object(self.module, "say_openai_chat", side_effect=AssertionError("cached")), \
+                     patch.object(self.module, "transcribe_local", side_effect=AssertionError("ASR")), \
+                     redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
+                    self.assertEqual(self.module.main(), 1)
+                self.assertEqual(json.loads((self.output / "manifest.json").read_text()), [])
+
+    def test_empty_chat_or_asr_speech_is_rejected(self):
+        def empty_chat_synthesis(text, wav, voice):
+            wav.write_bytes(b"audio")
+            return ""
+
+        empty_chat = self.run_narrate(
+            [{"id": "clip", "narration": "Two words"}],
+            empty_chat_synthesis, engine="openai-chat", expected=1,
+        )
+        self.assertEqual(empty_chat, [])
+
+        def synthesize(text, wav, voice):
+            wav.write_bytes(b"audio")
+
+        self.scenes.write_text(json.dumps({"scenes": [{"id": "clip", "narration": "Two words"}]}),
+                              encoding="utf-8")
+        argv = ["narrate", str(self.scenes), str(self.output), "--engine", "piper", "--verify", "auto"]
         with patch.object(sys, "argv", argv), \
              patch.object(self.module.shutil, "which", return_value="ffprobe"), \
-             patch.object(self.module, "openai_key", return_value="key"), \
-             patch.object(self.module, "say_openai_chat", side_effect=AssertionError("cached")), \
-             patch.object(self.module, "transcribe_local", side_effect=AssertionError("ASR")), \
+             patch.object(self.module, "openai_key", return_value=None), \
+             patch.object(self.module, "say_piper", side_effect=synthesize), \
+             patch.object(self.module, "transcribe_local", return_value=""), \
+             patch.object(self.module, "duration", return_value=1.0), \
              redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
             self.assertEqual(self.module.main(), 1)
         self.assertEqual(json.loads((self.output / "manifest.json").read_text()), [])
 
+    def test_cached_nested_wav_keeps_its_manifest_path_after_reverification(self):
+        nested = self.output / "takes" / "clip.wav"
+        nested.parent.mkdir(parents=True)
+        nested.write_bytes(b"accepted")
+        synthesis = {"engine": "piper", "voice": self.module.PIPER_VOICE,
+                     "model": self.module.PIPER_VOICE}
+        (self.output / "manifest.json").write_text(json.dumps([{
+            "id": "clip", "text": "Nested clip", "wav": "takes/clip.wav", "duration": 1.0,
+            "synthesis": synthesis,
+        }]), encoding="utf-8")
+        manifest = self.run_narrate(
+            [{"id": "clip", "narration": "Nested clip"}],
+            lambda *args: (_ for _ in ()).throw(AssertionError("cached")), verify="on",
+            transcript="Nested clip",
+        )
+        self.assertEqual(manifest[0]["wav"], "takes/clip.wav")
+
+    def test_cached_strict_verification_withdraws_before_interrupt(self):
+        def synthesize(text, wav, voice):
+            wav.write_bytes(b"accepted")
+
+        self.run_narrate([{"id": "clip", "narration": "Interrupt safely"}], synthesize)
+        self.scenes.write_text(json.dumps({"scenes": [{"id": "clip", "narration": "Interrupt safely"}]}),
+                              encoding="utf-8")
+        argv = ["narrate", str(self.scenes), str(self.output), "--engine", "piper", "--verify", "on"]
+        with patch.object(sys, "argv", argv), \
+             patch.object(self.module.shutil, "which", return_value="ffprobe"), \
+             patch.object(self.module, "openai_key", return_value=None), \
+             patch.object(self.module, "say_piper", side_effect=AssertionError("cached")), \
+             patch.object(self.module, "transcribe_local", side_effect=KeyboardInterrupt), \
+             redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
+            with self.assertRaises(KeyboardInterrupt):
+                self.module.main()
+        self.assertEqual(json.loads((self.output / "manifest.json").read_text()), [])
+
 
 class NarrationComparisonContract(unittest.TestCase):
     def setUp(self):
@@ -178,6 +287,7 @@ class NarrationComparisonContract(unittest.TestCase):
             ("你好世界", "こんにちは世界"),
             ("短文", "短文"),
             ("mixed 日本語 words", "mixed 日本語 words"),
+            ("\U00020000\U00020001", "\U00020002\U00020003"),
             ("*** !!!", "*** !!!"),
         ):
             with self.subTest(script=script):
@@ -303,6 +413,28 @@ class AssemblyNarrationContract(unittest.TestCase):
         offsets = json.loads((self.root / "work" / "offsets.json").read_text())
         self.assertEqual(offsets, {})
 
+    def test_movie_with_audio_maps_its_source_audio(self):
+        source = self.root / "source-with-audio.mp4"
+        source.write_bytes(b"source")
+        status, calls = self.assemble(
+            [{"id": "movie", "kind": "movie", "src": source.name}],
+            run_side_effect=lambda command: subprocess.CompletedProcess(
+                command, 0,
+                json.dumps({"streams": [{"codec_type": "video"}, {"codec_type": "audio"}]}), ""
+            ) if "-show_streams" in command else subprocess.CompletedProcess(command, 0, "1.0", ""),
+        )
+        self.assertEqual(status, 0)
+        movie_encode = next(call for call in calls if call and call[0] == "ffmpeg")
+        self.assertNotIn("anullsrc=r=44100:cl=stereo", movie_encode)
+        self.assertEqual(movie_encode[movie_encode.index("-map") + 1], "0:v:0")
+        self.assertEqual(movie_encode[movie_encode.index("-map", movie_encode.index("-map") + 1) + 1], "0:a:0")
+
+    def test_movie_geometry_fits_width_and_requested_inner_height_before_padding(self):
+        self.assertEqual(self.module.movie_geometry(1920, 1080, 800), {
+            "scale": (1920, 800),
+            "pad": (1920, 1080),
+        })
+
 
 class PercentPathContract(unittest.TestCase):
     def test_sequence_pattern_escapes_only_directory_percents(self):