fix(review): preserve recovered takes and stale-video guards

app/services/director_pipeline.py

@@ -2192,8 +2192,10 @@ def _backfill_clip_video_attempts(state: dict, state_dir: str) -> dict:
             selected = ""
         clip["selected_video_filename"] = selected or None
         if selected:
+            # A Studio selection is the playback authority, but it does not
+            # refresh inputs. Image reruns keep video_stale so Rejoin/export
+            # cannot assemble a take that no longer matches the start frame.
             clip["video_filename"] = selected
-            clip["video_stale"] = False
         clip["video_attempts"] = sorted(
             attempts_by_clip[index].values(),
             key=lambda item: (float(item.get("created_at") or 0), item["filename"]),
@@ -4356,6 +4358,19 @@ def _rejoin_clips_impl(out_dir: str, pid: str) -> dict:
     state = _ensure_h3_segment_state(state)
     clips = state.get("clips", [])
     video_files = []
+    # Image reruns keep video_stale on the clip even when a Studio selection or
+    # H3 segment list still points at playable files. Gate Rejoin before the
+    # H3 branch, which otherwise treats those files as current.
+    stale_clip_numbers = [
+        str(index + 1)
+        for index, clip in enumerate(clips)
+        if clip.get("video_stale")
+    ]
+    if stale_clip_numbers:
+        raise ValueError(
+            "Regenerate stale video clip(s) "
+            f"{', '.join(stale_clip_numbers)} before rejoining."
+        )
     legacy_h3_segments = (
         _is_sequential_h3_model(state.get("video_model"))
         and any(clip.get("h3_segments") for clip in clips)
@@ -4390,17 +4405,6 @@ def _rejoin_clips_impl(out_dir: str, pid: str) -> dict:
         if stale:
             raise ValueError("Regenerate stale H3 continuations before rejoining the final video")
     else:
-        stale_clip_numbers = [
-            str(index + 1)
-            for index, clip in enumerate(clips)
-            if clip.get("video_stale")
-        ]
-        if stale_clip_numbers:
-            raise ValueError(
-                "Regenerate stale video clip(s) "
-                f"{', '.join(stale_clip_numbers)} before rejoining."
-            )
-
         if shot_images_required(_saved_pipeline_shot_image_policy(state)):
             invalid_start_numbers = _invalid_saved_media_numbers(
                 [clip.get("start_image_filename") for clip in clips],

app/services/director_review.py

@@ -1,11 +1,10 @@
 """Persist review decisions onto the existing Director pipeline, atomically."""
 from pathlib import Path
-import json
 import time
 
 from services.director_pipeline import (
-    _exclusive_pipeline_operation, _find_pipeline_file, _pipeline_file_lock,
-    _write_pipeline_json_unlocked,
+    _exclusive_pipeline_operation, _find_pipeline_file, _load_pipeline_state_locked,
+    _pipeline_file_lock, _write_pipeline_json_unlocked, hydrate_queue_clips,
 )
 
 
@@ -69,7 +68,11 @@ def save_review(workspace: str, pid: str, commands: list) -> dict:
     if not path or Path(path).resolve().parent != Path(workspace).resolve():
         raise ValueError("Production not found in this workspace")
     with _pipeline_file_lock:
-        state = json.loads(Path(path).read_text(encoding="utf-8"))
+        # Same projection GET uses: sidecar/output_files histories are visible
+        # on the desk, so persist must accept and return those takes.
+        state = _load_pipeline_state_locked(workspace, pid)
+        if not state or str(state.get("pipeline_id") or "") != pid:
+            raise ValueError("Production not found in this workspace")
         _apply_review(state, commands, Path(workspace), pid)
         _write_pipeline_json_unlocked(path, state)
-    return state
+    return hydrate_queue_clips(state)

tests/test_director_cancellation.py

@@ -1153,6 +1153,46 @@ def test_rejoin_rejects_stale_video_instead_of_omitting_clip(self):
 
         concatenate.assert_not_called()
 
+    def test_rejoin_rejects_stale_video_even_when_a_take_is_selected(self):
+        pid = "pipe-stale-selected-rejoin"
+        record = self._add_pipeline(pid, "completed")
+        record["clip_plans"] = [
+            {"image_prompt": "one", "video_prompt": "one"},
+            {"image_prompt": "two", "video_prompt": "two"},
+        ]
+        record["_clip_video_files"] = ["one.mp4", "two.mp4"]
+        for filename in record["_clip_video_files"]:
+            self._write_media(filename, b"video")
+        self.assertTrue(pipeline._save_pipeline_state(pid))
+
+        def mark_selected_stale(state):
+            clip = state["clips"][0]
+            clip["selected_video_filename"] = clip["video_filename"]
+            clip["video_stale"] = True
+
+        pipeline._update_saved_pipeline(self.temp_dir.name, pid, mark_selected_stale)
+        loaded = pipeline.load_pipeline_state(self.temp_dir.name, pid)
+        self.assertTrue(loaded["clips"][0]["video_stale"])
+        self.assertEqual(loaded["clips"][0]["selected_video_filename"], "one.mp4")
+
+        def keep_notes(state):
+            state["clips"][0]["review_notes"] = "keep stale"
+
+        pipeline._update_saved_pipeline(self.temp_dir.name, pid, keep_notes)
+        raw_path = pipeline._find_pipeline_file(self.temp_dir.name, pid)
+        with open(raw_path, encoding="utf-8") as handle:
+            saved = json.load(handle)
+        self.assertTrue(saved["clips"][0]["video_stale"])
+        self.assertEqual(saved["clips"][0]["review_notes"], "keep stale")
+
+        concatenate = Mock(return_value=True)
+        pipeline._wgp.concatenate_multi_clip_videos = concatenate
+        with self.assertRaisesRegex(
+            ValueError, "stale video clip.*1.*before rejoining",
+        ):
+            pipeline.rejoin_clips(self.temp_dir.name, pid)
+        concatenate.assert_not_called()
+
     def test_rejoin_rejects_clip_whose_start_image_is_missing(self):
         pid = "pipe-missing-rejoin-start"
         record = self._add_pipeline(pid, "completed")

tests/test_director_h3_workflow_edits.py

@@ -2,6 +2,8 @@
 from pathlib import Path
 from unittest.mock import patch
 
+import pytest
+
 from app.services import director_pipeline
 
 
@@ -346,3 +348,59 @@ def concatenate_multi_clip_videos(paths, destination, _audio, **_kwargs):
         director_pipeline.rejoin_clips(str(tmp_path), "h3-selection")
 
     assert joined == ["shot0_studio.mp4", "shot1.mp4"]
+
+
+@pytest.mark.parametrize("video_model", ["minimax_h3", "minimax_h3_legacy"])
+@pytest.mark.parametrize("selected", [None, "shot0_studio.mp4"])
+def test_h3_rejoin_rejects_stale_clip_even_when_segments_are_playable(tmp_path: Path, video_model, selected):
+    filenames = ("shot0_a.mp4", "shot0_b.mp4", "shot1.mp4")
+    for filename in filenames:
+        (tmp_path / filename).write_bytes(b"video")
+    if selected:
+        (tmp_path / selected).write_bytes(b"selected video")
+    _write_pipeline(tmp_path, {
+        "pipeline_id": "h3-stale-rejoin",
+        "created_at": 10.0,
+        "status": "completed",
+        "pipeline_type": "short_film_story",
+        "video_model": video_model,
+        "clips": [
+            {
+                "index": 0,
+                "video_filename": "shot0_b.mp4",
+                "selected_video_filename": selected,
+                "video_stale": True,
+                "video_prompt": "Whole shot zero",
+                "h3_segments": [
+                    {"index": 0, "filename": "shot0_a.mp4", "stale": False},
+                    {"index": 1, "filename": "shot0_b.mp4", "stale": False},
+                ],
+            },
+            {
+                "index": 1,
+                "video_filename": "shot1.mp4",
+                "video_prompt": "Whole shot one",
+                "h3_segments": [
+                    {"index": 0, "filename": "shot1.mp4", "stale": False},
+                ],
+            },
+        ],
+        "output_files": list(filenames),
+        "workspace": "default",
+    })
+    checkpoint = Path(director_pipeline._find_pipeline_file(str(tmp_path), "h3-stale-rejoin"))
+    before = checkpoint.read_bytes()
+    joined = []
+
+    class FakeWgp:
+        @staticmethod
+        def concatenate_multi_clip_videos(paths, destination, _audio, **_kwargs):
+            joined.extend(Path(path).name for path in paths)
+            Path(destination).write_bytes(b"joined")
+            return True
+
+    with patch.object(director_pipeline, "_wgp", FakeWgp()):
+        with pytest.raises(ValueError, match="stale video clip.*1.*before rejoining"):
+            director_pipeline.rejoin_clips(str(tmp_path), "h3-stale-rejoin")
+    assert joined == []
+    assert checkpoint.read_bytes() == before

tests/test_director_review.py

@@ -38,8 +38,11 @@ def test_review_persists_exact_take_tag_and_notes_and_preserves_other_shots(tmp_
     saved = json.loads(path.read_text())
     assert saved['clips'][0]['selected_video_filename'] == 'old.mp4'
     assert saved['clips'][0]['review_notes'] == '  literal\nnotes  '
-    assert saved['clips'][0]['video_attempts'] == state['clips'][0]['video_attempts']
-    assert saved['clips'][1] == state['clips'][1]
+    assert {item['filename'] for item in saved['clips'][0]['video_attempts']} == {'old.mp4', 'new.mp4'}
+    for original in state['clips'][0]['video_attempts']:
+        actual = next(item for item in saved['clips'][0]['video_attempts'] if item['filename'] == original['filename'])
+        assert {key: actual[key] for key in original} == original
+    assert {key: saved['clips'][1][key] for key in state['clips'][1]} == state['clips'][1]
 
 
 def test_review_rejects_invalid_batch_without_partial_save(tmp_path):
@@ -75,3 +78,94 @@ def test_switching_an_approved_take_updates_h3_selection_without_approving_it(tm
     assert saved['clips'][0]['h3_segments'][0]['filename'] == 'old.mp4'
     assert saved['clips'][0]['h3_segments'][0]['stale'] is False
     assert 'old.mp4' in saved['output_files']
+
+
+def _sidecar_history(tmp_path):
+    state = {"pipeline_id": "review-test", "status": "completed", "clips": [
+        {"index": 0, "video_filename": "new.mp4", "video_prompt": "literal prompt", "tag": None},
+    ]}
+    path = tmp_path / f"{pipeline._PIPELINE_FILE_PREFIX}review-test.json"
+    path.write_text(json.dumps(state))
+    (tmp_path / "new.mp4").write_bytes(b"current take")
+    (tmp_path / "old.mp4").write_bytes(b"recovered take")
+    (tmp_path / "old.mp4.meta.json").write_text(json.dumps({
+        "output_filename": "old.mp4",
+        "director_pipeline_id": "review-test",
+        "director_clip_index": 0,
+        "created_at": 1,
+        "params": {"_director_clip_index": 0, "prompt": "older take"},
+    }))
+    return path
+
+
+def test_review_can_select_a_sidecar_take_missing_from_the_checkpoint(tmp_path):
+    path = _sidecar_history(tmp_path)
+    saved = save_review(str(tmp_path), "review-test", [
+        {"type": "select_take", "pipelineId": "review-test", "clipIndex": 0, "filename": "old.mp4"},
+        {"type": "tag_clip", "pipelineId": "review-test", "clipIndex": 0, "tag": "good"},
+    ])
+    names = {item["filename"] for item in saved["clips"][0]["video_attempts"]}
+    assert names == {"old.mp4", "new.mp4"}
+    assert saved["clips"][0]["selected_video_filename"] == "old.mp4"
+    assert saved["clips"][0]["tag"] == "good"
+    disk = json.loads(path.read_text())
+    assert disk["clips"][0]["selected_video_filename"] == "old.mp4"
+    assert {item["filename"] for item in disk["clips"][0]["video_attempts"]} == names
+
+
+def test_review_notes_keep_recovered_takes_in_the_saved_pipeline(tmp_path):
+    _sidecar_history(tmp_path)
+    saved = save_review(str(tmp_path), "review-test", [
+        {"type": "note_clip", "pipelineId": "review-test", "clipIndex": 0, "notes": "keep history"},
+    ])
+    names = {item["filename"] for item in saved["clips"][0]["video_attempts"]}
+    assert names == {"old.mp4", "new.mp4"}
+    assert saved["clips"][0]["review_notes"] == "keep history"
+    assert saved["clips"][0]["video_filename"] == "new.mp4"
+
+
+def test_review_cannot_select_a_sidecar_from_another_production(tmp_path):
+    path = _sidecar_history(tmp_path)
+    sidecar = tmp_path / 'old.mp4.meta.json'
+    metadata = json.loads(sidecar.read_text())
+    metadata['director_pipeline_id'] = 'another-production'
+    sidecar.write_text(json.dumps(metadata))
+    before = path.read_bytes()
+    with pytest.raises(ValueError, match='existing take'):
+        save_review(str(tmp_path), 'review-test', [
+            {'type': 'select_take', 'pipelineId': 'review-test', 'clipIndex': 0, 'filename': 'old.mp4'},
+        ])
+    assert path.read_bytes() == before
+
+
+def test_invalid_review_does_not_persist_hydrated_history_or_partial_notes(tmp_path):
+    path = _sidecar_history(tmp_path)
+    before = path.read_bytes()
+    with pytest.raises(ValueError, match='Invalid review decision'):
+        save_review(str(tmp_path), 'review-test', [
+            {'type': 'note_clip', 'pipelineId': 'review-test', 'clipIndex': 0, 'notes': 'not committed'},
+            {'type': 'tag_clip', 'pipelineId': 'review-test', 'clipIndex': 0, 'tag': 'invalid'},
+        ])
+    assert path.read_bytes() == before
+
+
+def test_review_notes_keep_a_stale_selected_take_stale(tmp_path):
+    path, state = fixture(tmp_path)
+    state['clips'][0].update(
+        selected_video_filename='old.mp4',
+        video_filename='old.mp4',
+        video_stale=True,
+        tag='good',
+    )
+    path.write_text(json.dumps(state))
+    loaded = pipeline.load_pipeline_state(str(tmp_path), 'review-test')
+    assert loaded['clips'][0]['video_stale'] is True
+    assert loaded['clips'][0]['selected_video_filename'] == 'old.mp4'
+    save_review(str(tmp_path), 'review-test', [
+        {'type': 'note_clip', 'pipelineId': 'review-test', 'clipIndex': 0, 'notes': 'keep stale'},
+    ])
+    saved = json.loads(path.read_text())
+    assert saved['clips'][0]['video_stale'] is True
+    assert saved['clips'][0]['selected_video_filename'] == 'old.mp4'
+    assert saved['clips'][0]['review_notes'] == 'keep stale'
+    assert saved['clips'][0]['tag'] == 'good'