fix(movie): preserve rejected narration and prior takes

Prevent failed narration scenes from entering the cache manifest, while leaving their generated WAV files available as failure evidence. Add filesystem-backed regressions covering repeated rejected chat synthesis, accepted-scene reuse, and strict ASR rejection of cached audio.

Refuse nonempty recording directories both before CLI session side effects and at the direct film boundary. The regression preserves existing numbered frames and a sentinel byte-for-byte across run, key, watch, and direct film refusal.

Make subtitle capability tests independent of the host FFmpeg installation, skip the real pixel test before probing unavailable tools, retain strict skipped-capability rejection, and remove only the unused websockets runner dependency.

skills/proving-it-works-with-a-movie/examples/film-terminal.py

@@ -137,10 +137,16 @@ def key_params(key):
     raise SystemExit(f"unknown key {key!r}: use one character or one of {', '.join(KEYS)}")
 
 
+def require_empty_take(out):
+    if out.exists() and any(out.iterdir()):
+        raise SystemExit(f"{out} is not empty: use a new take directory")
+
+
 def film(out, seconds, hold, capture, finished, clock=time.monotonic, sleep=time.sleep):
     """Write PNG frames on the FPS grid until `finished()` plus `hold` seconds,
     or `seconds` in all. A slow capture repeats the previous frame, so the
     directory plays back at exactly FPS. Returns the frame count."""
+    require_empty_take(out)
     out.mkdir(parents=True, exist_ok=True)
     start, index, last, stop = clock(), -1, None, None
     while True:
@@ -516,6 +522,8 @@ def main():
     args = parser.parse_args()
     if args.verb == "watch" and not args.record:
         parser.error("watch needs --record")
+    if getattr(args, "record", None):
+        require_empty_take(args.record)
     return {"serve": serve, "run": run, "key": key, "watch": watch, "close": close}[args.verb](args)
 
 

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

@@ -281,6 +281,8 @@ def main():
             break
         else:
             failures.append(sid)
+        if sid in failures:
+            continue
         manifest.append({"id": sid, "text": text, "wav": wav.name,
                          "duration": duration(wav)})
 

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

@@ -1,7 +1,7 @@
 #!/usr/bin/env -S uv run --script
 # /// script
 # requires-python = ">=3.10"
-# dependencies = ["pyyaml", "pillow", "websockets", "websocket-client==1.9.0"]
+# dependencies = ["pyyaml", "pillow", "websocket-client==1.9.0"]
 # ///
 """Run the proving-it-works-with-a-movie regression suites portably."""
 

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

@@ -87,6 +87,52 @@ class NarrationDriftRegression(unittest.TestCase):
                 self.assertNotEqual(module.main(), 0)
             self.assertIn("clip: required verification unavailable", stderr.getvalue())
             self.assertIn("FAILED verbatim delivery: ['clip']", stderr.getvalue())
+            self.assertEqual(
+                json.loads((output / "manifest.json").read_text(encoding="utf-8")),
+                [],
+            )
+            self.assertEqual((output / "clip.wav").read_bytes(), b"cached audio fixture")
+
+    def test_rejected_chat_audio_is_never_cached_but_accepted_audio_is(self):
+        import json
+        import sys
+        module = fixtures.load_script("narrate")
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            scenes = root / "scenes.yaml"
+            scenes.write_text(json.dumps({"scenes": [
+                {"id": "accepted", "narration": "Read this sentence exactly."},
+                {"id": "rejected", "narration": "Keep this evidence out of the manifest."},
+            ]}), encoding="utf-8")
+            output = root / "voice"
+            calls = []
+
+            def synthesize(key, text, wav, voice):
+                calls.append(text)
+                wav.write_bytes(f"render {len(calls)}".encode())
+                if text.startswith("Keep"):
+                    return "Unrelated invented preamble with entirely different words here."
+                return text
+
+            argv = ["narrate", str(scenes), str(output), "--engine", "openai-chat",
+                    "--verify", "off"]
+            rejected_renders = []
+            with patch.object(sys, "argv", argv), \
+                 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):
+                for _ in range(2):
+                    with redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
+                        self.assertEqual(module.main(), 1)
+                    manifest = json.loads(
+                        (output / "manifest.json").read_text(encoding="utf-8")
+                    )
+                    self.assertEqual([entry["id"] for entry in manifest], ["accepted"])
+                    rejected_renders.append((output / "rejected.wav").read_bytes())
+
+            self.assertEqual(calls.count("Read this sentence exactly."), 1)
+            self.assertEqual(calls.count("Keep this evidence out of the manifest."), 4)
+            self.assertNotEqual(rejected_renders[0], rejected_renders[1])
 
 class TranscriptionProtocolRegression(unittest.TestCase):
     def test_owned_json_is_used_instead_of_library_stdout(self):

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

@@ -28,7 +28,7 @@ class SubtitlePathRegression(unittest.TestCase):
                 return True
 
             stdout, stderr = io.StringIO(), io.StringIO()
-            with patch.object(sys, "argv", ["burn-subtitles", str(movie), str(subs), str(output)]), patch.object(module, "has_libass", return_value=True), patch.object(module, "run", side_effect=fake_run), redirect_stdout(stdout), redirect_stderr(stderr):
+            with patch.object(sys, "argv", ["burn-subtitles", str(movie), str(subs), str(output)]), patch.object(module.shutil, "which", return_value="ffmpeg"), patch.object(module, "has_libass", return_value=True), patch.object(module, "run", side_effect=fake_run), redirect_stdout(stdout), redirect_stderr(stderr):
                 self.assertEqual(module.main(), 0)
             self.assertIn("burned into the picture", stdout.getvalue())
             command, cwd = calls[0]
@@ -46,7 +46,7 @@ class SubtitlePathRegression(unittest.TestCase):
             movie.write_bytes(b"movie")
             subs.write_text("1\n00:00:00,000 --> 00:00:01,000\ncaption\n", encoding="utf-8")
             stdout, stderr = io.StringIO(), io.StringIO()
-            with patch.object(sys, "argv", ["burn-subtitles", str(movie), str(subs), str(output)]), patch.object(module, "has_libass", return_value=True), patch.object(module, "run", return_value=False), redirect_stdout(stdout), redirect_stderr(stderr):
+            with patch.object(sys, "argv", ["burn-subtitles", str(movie), str(subs), str(output)]), patch.object(module.shutil, "which", return_value="ffmpeg"), patch.object(module, "has_libass", return_value=True), patch.object(module, "run", return_value=False), redirect_stdout(stdout), redirect_stderr(stderr):
                 self.assertEqual(module.main(), 1)
             self.assertIn("burn failed", stderr.getvalue())
 
@@ -69,6 +69,9 @@ class SubtitleIntegrationRegression(unittest.TestCase):
 
     def test_hard_subtitles_are_pixels_in_nested_special_path(self):
         import subprocess
+        missing = fixtures.missing_executables("uv", "ffmpeg")
+        if missing:
+            self.skipTest(f"required executable(s) not on PATH: {', '.join(missing)}")
         module = fixtures.load_script("burn-subtitles")
         if not module.has_libass():
             self.skipTest("libass FFmpeg is required for hard subtitle pixels")
@@ -93,7 +96,7 @@ class SubtitleIntegrationRegression(unittest.TestCase):
                 movie, subs = root / "in.mp4", root / "in.srt"
                 movie.touch(); subs.touch()
                 stdout, stderr = io.StringIO(), io.StringIO()
-                with patch.object(sys, "argv", ["burn-subtitles", str(movie), str(subs), str(root / "out.mp4")]), patch.object(module, "has_libass", return_value=libass), patch.object(module, "run", side_effect=[False, True] if libass else [True]), redirect_stdout(stdout), redirect_stderr(stderr):
+                with patch.object(sys, "argv", ["burn-subtitles", str(movie), str(subs), str(root / "out.mp4")]), patch.object(module.shutil, "which", return_value="ffmpeg"), patch.object(module, "has_libass", return_value=libass), patch.object(module, "run", side_effect=[False, True] if libass else [True]), redirect_stdout(stdout), redirect_stderr(stderr):
                     self.assertEqual(module.main(), 0)
                 self.assertEqual("no libass" in stdout.getvalue(), not libass)
                 self.assertEqual("burn failed" in stderr.getvalue(), libass)

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

@@ -11,6 +11,7 @@ import tempfile
 import time
 import unittest
 from pathlib import Path
+from unittest.mock import patch
 
 import fixtures
 
@@ -78,6 +79,26 @@ class PromptTests(unittest.TestCase):
 
 
 class FilmGridTests(unittest.TestCase):
+    def test_filming_refuses_a_nonempty_take_without_changing_its_contents(self):
+        module = recorder()
+        with tempfile.TemporaryDirectory() as directory:
+            out = Path(directory) / "take"
+            out.mkdir()
+            (out / "f00000.png").write_bytes(b"old frame")
+            (out / "notes.txt").write_bytes(b"sentinel evidence")
+            before = {path.name: path.read_bytes() for path in out.iterdir()}
+            captures = []
+
+            with self.assertRaisesRegex(SystemExit, "not empty.*new take directory"):
+                module.film(out, seconds=1, hold=0,
+                            capture=lambda: captures.append(True) or b"new frame",
+                            finished=lambda: True)
+
+            self.assertEqual(captures, [])
+            self.assertEqual(
+                {path.name: path.read_bytes() for path in out.iterdir()}, before
+            )
+
     def test_a_slow_capture_repeats_the_previous_frame_and_filming_holds_after_the_prompt(self):
         module = recorder()
         clock = {"now": 0.0}
@@ -120,6 +141,44 @@ class ServeArgumentTests(unittest.TestCase):
             self.assertIn("--cwd is not a directory", result.stderr)
             self.assertFalse((Path(directory) / "session" / "session.json").exists())
 
+    def test_recording_verbs_refuse_a_nonempty_take_before_session_side_effects(self):
+        module = recorder()
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            session = root / "session"
+            session.mkdir()
+            (session / "ready.json").write_text("{}", encoding="utf-8")
+            (session / "session.json").write_text(
+                json.dumps({"debug_port": 1}), encoding="utf-8"
+            )
+            (session / "terminal.log").write_bytes(
+                b"\x1b]0;MOVIE;1;1;0;/tmp\x07"
+            )
+            for verb, positional in (("run", ["echo hello"]),
+                                     ("key", ["Enter"]),
+                                     ("watch", [])):
+                with self.subTest(verb=verb):
+                    take = root / verb
+                    take.mkdir()
+                    (take / "f00000.png").write_bytes(b"old frame")
+                    (take / "sentinel.txt").write_bytes(b"keep me")
+                    before = {path.name: path.read_bytes() for path in take.iterdir()}
+                    argv = ["film-terminal", verb, str(session), *positional,
+                            "--record", str(take)]
+                    with patch.object(sys, "argv", argv), \
+                         patch.object(module, "connect",
+                                      side_effect=AssertionError("connected")), \
+                         patch.object(module, "type_text",
+                                      side_effect=AssertionError("typed")), \
+                         patch.object(module, "press",
+                                      side_effect=AssertionError("pressed")), \
+                         self.assertRaisesRegex(SystemExit,
+                                                "not empty.*new take directory"):
+                        module.main()
+                    self.assertEqual(
+                        {path.name: path.read_bytes() for path in take.iterdir()}, before
+                    )
+
 
 @unittest.skipUnless(TTYD and BROWSER, "ttyd and a Chrome-family browser are required")
 class SessionTests(unittest.TestCase):