fix(movie): retain incomplete cleanup and command evidence

Address both independent Task 3 review findings at 42486dbb. When an owned leader exits before tree cleanup, the existing parentage helper cannot confirm orphan descendant cleanup. Treat that state as incomplete, preserve PID metadata, return failure, and never invoke the helper on the exited leader's numeric PID. Continue releasing the other owned resources without adding descendant tracking.

Keep a recording capture failure truthful while preserving available completed command evidence: outcome remains failed, exit status remains 1, and the capture error is retained alongside ok, native exit_code, and cwd. A completed successful command does not make a failed recording successful, and no successful take manifest is published.

TDD regressions reproduced an exited leader incorrectly returning 0 and capture failures dropping completed native status for exits 0 and 7. All 14 covering lifecycle and observation tests pass using fake process handles, fake clocks, mocked capture/log boundaries, and intercepted media writes; git diff --check passes. No real media generation or inspection, process/browser launches, SessionTests, FilmGridTests, or full media suites were executed. Detailed RED/GREEN evidence is appended to .superpowers/sdd/2026-09-11-movie-committee-repairs/task-3-report.md.

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

@@ -437,8 +437,12 @@ def serve(args):
         for process in processes:
             try:
                 # Only the owner acts on handles it acquired, never stored PIDs.
-                if process.poll() is None:
-                    kill_process_tree(process.pid)
+                if process.poll() is not None:
+                    print(f"serve cleanup: child {process.pid} exited before tree cleanup; "
+                          "descendant cleanup cannot be confirmed", file=sys.stderr)
+                    cleaned = False
+                    continue
+                kill_process_tree(process.pid)
                 process.wait(timeout=5)
             except (OSError, subprocess.TimeoutExpired) as error:
                 print(f"serve cleanup: {error}", file=sys.stderr)
@@ -511,17 +515,22 @@ def observe(args, cdp, n0):
             while poll() is None and time.monotonic() < deadline:
                 time.sleep(0.05)
         prompt = poll()
+        result = {"outcome": "completed" if prompt else "running"}
     except Exception as error:
-        print(json.dumps({"outcome": "failed", "error": str(error)}))
-        return 1
-    result = {"outcome": "completed" if prompt else "running"}
+        result = {"outcome": "failed", "error": str(error)}
+        try:
+            prompt = latest()
+        except OSError:
+            prompt = None
     if prompt:
         result.update(ok=prompt["ok"], exit_code=prompt["exit_code"], cwd=prompt["cwd"])
-    if args.record:
+    if args.record and result["outcome"] != "failed":
         result["frames"] = frames
         result["scene"] = {"kind": "frames", "src": str(args.record.resolve()), "rate": FPS}
         write_json(args.record / "take.json", result)
     print(json.dumps(result))
+    if result["outcome"] == "failed":
+        return 1
     return 2 if not prompt else 0 if prompt["ok"] else 1
 
 

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

@@ -82,8 +82,11 @@ def serving(failure=None, stop_at=None, relative=False):
 
         def kill(pid):
             for child in children:
-                if child.pid == pid and failure != "child wait":
-                    child.returncode = -9
+                if child.pid == pid:
+                    if child.poll() is not None:
+                        raise AssertionError("cannot use an exited leader's PID")
+                    if failure != "child wait":
+                        child.returncode = -9
 
         def write_json(path, value):
             if failure == "session metadata" and path.name == "session.json":
@@ -108,6 +111,8 @@ def serving(failure=None, stop_at=None, relative=False):
                 clock.sleep(timeout)
                 if (directory / "ready.json").exists():
                     stop("ready")
+                    if failure == "exited leader":
+                        children[0].returncode = 0
                     if failure == "later connection":
                         raise ConnectionError("injected later disconnect")
                 else:
@@ -247,6 +252,17 @@ class ServeLifecycleTests(unittest.TestCase):
                 self.assertFalse(session.get("closed", False))
                 self.assertFalse((rig.directory / "ready.json").exists())
 
+    def test_exited_leader_cannot_confirm_descendant_cleanup(self):
+        with serving("exited leader", stop_at="ready") as rig:
+            self.assertEqual(rig.module.serve(rig.args), 1)
+            session = rig.module.read_json(rig.directory / "session.json")
+            self.assertEqual(session["pids"], [1100, 1101])
+            self.assertFalse(session.get("closed", False))
+            self.assertEqual(rig.children[0].returncode, 0)
+            self.assertEqual(rig.children[1].returncode, -9)
+            self.assertTrue(all(h.closed for h in rig.handles))
+            self.assertFalse((rig.directory / "ready.json").exists())
+
 
 class CloseContractTests(unittest.TestCase):
     def test_close_waits_for_owner_cleanup_and_repeated_close_never_kills(self):
@@ -405,6 +421,37 @@ class ObservationTests(unittest.TestCase):
             self.fail(f"stdout JSON was not portable: {error}")
         self.assertEqual((code, result["cwd"]), (1, "C:/René/λ"))
 
+    def test_capture_disconnect_during_hold_preserves_completed_command_status(self):
+        for ok, exit_code in ((True, 0), (False, 7)):
+            with self.subTest(ok=ok), tempfile.TemporaryDirectory() as temp:
+                module, clock, writes = recorder(), Clock(), []
+                marker = f"\x1b]0;MOVIE;2;{int(ok)};{exit_code};C:/René/λ\x07".encode()
+                args = SimpleNamespace(session=Path(temp), record=Path(temp) / "take",
+                                       seconds=10, hold=0.6)
+                real_film = module.film
+
+                def film(*args):
+                    return real_film(*args, clock=clock.monotonic, sleep=clock.sleep)
+
+                def capture(cdp):
+                    if clock.now >= 0.2:
+                        raise ConnectionError("capture connection closed")
+                    return b"capture-token"
+
+                with patch.object(module, "film", film), \
+                     patch.object(module, "screenshot", capture), \
+                     patch.object(module, "tail", lambda path: marker), \
+                     patch.object(Path, "write_bytes", lambda path, token: writes.append(token)), \
+                     patch.object(module, "write_json", side_effect=AssertionError("failed take publication")), \
+                     contextlib.redirect_stdout(io.StringIO()) as out:
+                    code = module.observe(args, SimpleNamespace(), 1)
+                self.assertEqual(code, 1)
+                self.assertEqual(json.loads(out.getvalue()), {
+                    "outcome": "failed", "error": "capture connection closed",
+                    "ok": ok, "exit_code": exit_code, "cwd": "C:/René/λ",
+                })
+                self.assertEqual(writes, [b"capture-token"])
+
 
 class VisibleTextTests(unittest.TestCase):
     def test_visible_prompt_survives_bel_and_st_title_markers(self):