"""Tests for execute_video start-frame and post-gen blocking.""" import pytest from unittest.mock import MagicMock, patch @pytest.fixture def step_runner(tmp_path): from recoil.execution.step_runner import StepRunner from recoil.execution.step_types import ProjectPaths store = MagicMock() store.get_shot.return_value = { "status": "video_processing", "shot_id": "EP001_SH01", } # The post-gen save path pre-computes a take number via the store; without # an int return the MagicMock propagates into `next_take_number` and raises # ("take must be >= 1"), short-circuiting to the catch-all video_failed # before the post-gen frame critic runs. store.acquire_next_take_number.return_value = 1 paths = ProjectPaths( project="test", project_root=tmp_path, frames_dir=tmp_path / "frames", video_dir=tmp_path / "video", plans_dir=tmp_path / "plans", previs_dir=tmp_path / "previs", ) paths.frames_dir.mkdir() paths.video_dir.mkdir() paths.plans_dir.mkdir() paths.previs_dir.mkdir() return StepRunner(store=store, paths=paths) def test_execute_video_blocks_on_hard_start_frame_failure(step_runner, tmp_path): """If StartFrameCritic returns Outcome.FAIL with HARD dim, do not submit to video API.""" from recoil.core.critic import Outcome, CriticResult, Dimension, Severity fake_result = CriticResult( critic_name="start_frame", outcome=Outcome.FAIL, dimensions=[ Dimension( name="CHARACTER_IDENTITY_ALICE", severity=Severity.HARD, passed=False, message="Wrong character: expected Alice with red hair, got blonde male", ) ], ) start = tmp_path / "start.png" start.write_bytes(b"fake") inputs_snapshot = { "characters": [ {"display_name": "Alice", "hair": "red", "clothing": "leather jacket"} ], } with patch( "recoil.pipeline._lib.critics.start_frame_critic.StartFrameCritic.run", return_value=("artifact", fake_result), ): with patch("recoil.execution.video_model_client.VideoModelClient") as mock_vmc_cls: mock_client = MagicMock() mock_vmc_cls.return_value = mock_client result = step_runner.execute_video( shot_id="EP001_SH01", prompt="Alice walks in", model="kling-v3", start_frame=start, inputs_snapshot=inputs_snapshot, ) # API client should NOT have been called — start frame blocked mock_client.submit.assert_not_called() assert result.success is False assert result.final_state == "video_semantic_failed" assert "Alice" in result.error or "character" in result.error.lower() def test_execute_video_routes_critic_error_to_pending_qc(step_runner, tmp_path): """If StartFrameCritic returns Outcome.ERROR (vision API down), route to pending_qc. Per JT's middle-option decision (2026-04-09): critic errors do NOT terminate the shot. Run continues, shot enters pending_qc, morning re-check script re-runs the critic when the API is back.""" from recoil.core.critic import Outcome, CriticResult fake_result = CriticResult( critic_name="start_frame", outcome=Outcome.ERROR, error="vision API timeout", ) start = tmp_path / "start.png" start.write_bytes(b"fake") with patch( "recoil.pipeline._lib.critics.start_frame_critic.StartFrameCritic.run", return_value=("artifact", fake_result), ): with patch("recoil.execution.video_model_client.VideoModelClient") as mock_vmc_cls: mock_client = MagicMock() mock_vmc_cls.return_value = mock_client result = step_runner.execute_video( shot_id="EP001_SH01", prompt="Alice walks in", model="kling-v3", start_frame=start, inputs_snapshot={"characters": []}, ) # API client should NOT have been called — critic errored mock_client.submit.assert_not_called() # Shot routed to pending_qc per JT's middle option pending_calls = [ c for c in step_runner._store.update_shot.call_args_list if c.kwargs.get("status") == "pending_qc" ] assert pending_calls, ( "Expected at least one update_shot call with status=pending_qc" ) err_msg = pending_calls[0].kwargs.get("error_message", "") assert "vision API" in err_msg or "vision_api" in err_msg.lower() # Result should also reflect pending_qc, not video_semantic_failed assert result.final_state == "pending_qc" def test_execute_video_proceeds_on_soft_failures(step_runner, tmp_path): """Soft failures alone do NOT block — PC-1 keep-bias.""" from recoil.core.critic import Outcome, CriticResult, Dimension, Severity fake_result = CriticResult( critic_name="start_frame", outcome=Outcome.PASS, # PASS because soft failures don't make it FAIL dimensions=[ Dimension( name="SCENE_ELEMENTS_DOOR", severity=Severity.SOFT, passed=False, message="No door visible", ) ], ) start = tmp_path / "start.png" start.write_bytes(b"fake") with patch( "recoil.pipeline._lib.critics.start_frame_critic.StartFrameCritic.run", return_value=("artifact", fake_result), ): with patch("recoil.execution.video_model_client.VideoModelClient") as mock_vmc_cls: mock_client = MagicMock() mock_client.submit.return_value = MagicMock(id="job1") mock_client.wait_for_job.return_value = MagicMock( success=False, error="(test stub)", cost=0.0 ) mock_vmc_cls.return_value = mock_client step_runner.execute_video( shot_id="EP001_SH01", prompt="Alice walks in", model="kling-v3", start_frame=start, inputs_snapshot={"characters": []}, ) # API client SHOULD have been called — soft failure doesn't block per PC-1 mock_client.submit.assert_called_once() def test_execute_video_post_gen_blocks_on_hard_failure(step_runner, tmp_path): """If VideoFrameCritic returns FAIL with HARD dim, mark video_semantic_failed.""" from recoil.core.critic import Outcome, CriticResult, Dimension, Severity fake_pre = CriticResult(critic_name="start_frame", outcome=Outcome.PASS) fake_post = CriticResult( critic_name="video_frame", outcome=Outcome.FAIL, dimensions=[ Dimension( name="EXTRA_LIMBS", severity=Severity.HARD, passed=False, message="Failed in frames: [2, 3, 4] — character has 3 arms", ) ], ) start = tmp_path / "start.png" start.write_bytes(b"fake") with ( patch( "recoil.pipeline._lib.critics.start_frame_critic.StartFrameCritic.run", return_value=("a", fake_pre), ), patch( "recoil.pipeline._lib.critics.video_frame_critic.VideoFrameCritic.run", return_value=("a", fake_post), ), patch("recoil.execution.video_model_client.VideoModelClient") as mock_vmc_cls, ): mock_client = MagicMock() mock_client.submit.return_value = MagicMock(id="job1") mock_client.wait_for_job.return_value = MagicMock( success=True, video_data=b"fake_video_bytes", video_url=None, cost=1.0 ) mock_vmc_cls.return_value = mock_client result = step_runner.execute_video( shot_id="EP001_SH01", prompt="Alice walks in", model="kling-v3", start_frame=start, inputs_snapshot={"characters": []}, ) assert result.success is False assert result.final_state == "video_semantic_failed" assert "EXTRA_LIMBS" in result.error or "limbs" in result.error.lower() def test_execute_video_post_gen_routes_critic_error_to_pending_qc( step_runner, tmp_path ): """If VideoFrameCritic ERRORs after generation succeeds, route to pending_qc.""" from recoil.core.critic import Outcome, CriticResult fake_pre = CriticResult(critic_name="start_frame", outcome=Outcome.PASS) fake_post = CriticResult( critic_name="video_frame", outcome=Outcome.ERROR, error="vision API timeout during frame analysis", ) start = tmp_path / "start.png" start.write_bytes(b"fake") with ( patch( "recoil.pipeline._lib.critics.start_frame_critic.StartFrameCritic.run", return_value=("a", fake_pre), ), patch( "recoil.pipeline._lib.critics.video_frame_critic.VideoFrameCritic.run", return_value=("a", fake_post), ), patch("recoil.execution.video_model_client.VideoModelClient") as mock_vmc_cls, ): mock_client = MagicMock() mock_client.submit.return_value = MagicMock(id="job1") mock_client.wait_for_job.return_value = MagicMock( success=True, video_data=b"fake", video_url=None, cost=1.0 ) mock_vmc_cls.return_value = mock_client step_runner.execute_video( shot_id="EP001_SH01", prompt="Alice walks in", model="kling-v3", start_frame=start, inputs_snapshot={"characters": []}, ) # Video was generated and saved, but the critic errored — pending_qc assert any( call.kwargs.get("status") == "pending_qc" for call in step_runner._store.update_shot.call_args_list ) def test_execute_video_gate_crash_defers_review_and_keeps_take( step_runner, tmp_path ): from recoil.execution.types import GenerationResult step_runner._validate_frames = False step_runner._store.acquire_next_take_number.return_value = 1 def crashing_gate(_video_path, _shot_data): raise RuntimeError("gate api down") with patch("recoil.execution.video_model_client.VideoModelClient") as mock_vmc_cls: mock_client = MagicMock() mock_client.submit.return_value = MagicMock(id="job1") mock_client.wait_for_job.return_value = GenerationResult( success=True, video_data=b"fake_video_bytes", video_url=None, cost=1.0, metadata={}, ) mock_vmc_cls.return_value = mock_client result = step_runner.execute_video( shot_id="EP001_SH01", prompt="Alice walks in", model="kling-v3", gates=[crashing_gate], ) assert result.success is True assert result.output_path assert result.gate_verdict.deferred is True assert result.gate_verdict.gate_name == "crashing_gate" assert result.gate_verdict.reason == "gate crashed: gate api down" step_runner._store.append_take.assert_called_once() assert any( call.kwargs.get("deferred") is True and call.kwargs.get("deferred_reason") == "gate crashed: gate api down" for call in step_runner._store.update_shot.call_args_list ) def test_execute_video_validates_grouping_before_submit(step_runner): step_runner._validate_frames = False bad_grouping = { "strategy": "coverage", "ordinal": None, "shot_ids": ["EP001_SH01"], } with patch("recoil.execution.video_model_client.VideoModelClient") as mock_vmc_cls: mock_client = MagicMock() mock_vmc_cls.return_value = mock_client with pytest.raises(ValueError, match="requires grouping identity"): step_runner.execute_video( shot_id="EP001_SH01", prompt="Alice walks in", model="kling-v3", grouping=bad_grouping, ) mock_client.submit.assert_not_called()