diff --git a/app/services/director_pipeline.py b/app/services/director_pipeline.py index bfb6e209..8e89594a 100644 --- a/app/services/director_pipeline.py +++ b/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], diff --git a/app/services/director_review.py b/app/services/director_review.py index 0bfd2f65..bffe1f38 100644 --- a/app/services/director_review.py +++ b/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) diff --git a/tests/test_director_cancellation.py b/tests/test_director_cancellation.py index 65fef171..088cda30 100644 --- a/tests/test_director_cancellation.py +++ b/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") diff --git a/tests/test_director_h3_workflow_edits.py b/tests/test_director_h3_workflow_edits.py index 525e5834..430eaebf 100644 --- a/tests/test_director_h3_workflow_edits.py +++ b/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 diff --git a/tests/test_director_review.py b/tests/test_director_review.py index 3b08f2c1..426b361a 100644 --- a/tests/test_director_review.py +++ b/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'