"""Tests for run_shot -- all 7 terminal statuses + FailureMode action routing. All tests mock StepRunner and API clients. No real generation calls. """ import json from unittest.mock import MagicMock, patch # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_paths(tmp_path): """Create a minimal ProjectPaths-like object with tmp_path.""" paths = MagicMock() paths.project = "test-proj" paths.project_root = tmp_path (tmp_path / "state" / "visual").mkdir(parents=True, exist_ok=True) return paths def _make_budget_guard(limit=100.0, per_shot_cap=None): from recoil.pipeline._lib.budget_manager import BudgetGuard return BudgetGuard(limit_usd=limit, label="test", per_shot_cap_usd=per_shot_cap) def _make_step_result( shot_id="SH01", success=True, final_state="keyframe_generated", output_path="/tmp/output.jpg", cost_usd=0.15, error=None, gate_verdict=None, model="test-model", ): """Create a mock StepResult.""" result = MagicMock() result.shot_id = shot_id result.success = success result.final_state = final_state result.output_path = output_path result.cost_usd = cost_usd result.error = error result.gate_verdict = gate_verdict result.model = model return result def _make_gate_verdict(passed=True, gate_name="gate_1", reason="", details=None): """Create a mock GateVerdict.""" gv = MagicMock() gv.passed = passed gv.gate_name = gate_name gv.reason = reason gv.details = details or {} gv.cost = 0.01 return gv def _make_shot(shot_id="SH01", prompt="Test prompt", pipeline="keyframe", **extra): shot = {"shot_id": shot_id, "prompt": prompt, "pipeline": pipeline} shot.update(extra) return shot def _make_step_runner(): return MagicMock() # --------------------------------------------------------------------------- # Tests: 7 terminal statuses # --------------------------------------------------------------------------- class TestRunShotTerminalStatuses: """Each of the 7 terminal statuses must be testable.""" def test_ok_happy_path(self, tmp_path): """Clean generation -> ok.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() runner.execute_keyframe.return_value = _make_step_result(success=True) result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) assert result.status == "ok" assert result.output_path == "/tmp/output.jpg" assert result.attempts >= 1 def test_budget_exhausted(self, tmp_path): """Budget pre-check fails -> budget_exhausted.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard(limit=0.01) # Pre-exhaust the budget guard._spent = 0.01 runner = _make_step_runner() result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) assert result.status == "budget_exhausted" runner.execute_keyframe.assert_not_called() def test_budget_exhausted_success(self, tmp_path): """Budget hit, last attempt IS usable -> budget_exhausted_success.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) # Guard with tight budget: allow first attempt, block second guard = _make_budget_guard(limit=0.20) runner = _make_step_runner() # First call: gate failure (non-ACCEPT) to force retry fail_verdict = _make_gate_verdict( passed=False, gate_name="gate_1", reason="anatomy issue", details={"failure_category": "anatomy_face_merge"}, ) fail_result = _make_step_result( success=False, final_state="keyframe_mechanical_failed", output_path="/tmp/attempt1.jpg", cost_usd=0.15, gate_verdict=fail_verdict, ) runner.execute_keyframe.return_value = fail_result result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) # Should return budget_exhausted_success since we have output from attempt 1 assert result.status == "budget_exhausted_success" assert result.output_path == "/tmp/attempt1.jpg" def test_attempts_exhausted(self, tmp_path): """4 outer attempts fail -> attempts_exhausted + review queue entry.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() # All attempts: non-ACCEPT failure -> AUTO_REROLL exhausts fail_verdict = _make_gate_verdict( passed=False, gate_name="gate_1", reason="anatomy issue", details={"failure_category": "anatomy_face_merge"}, ) runner.execute_keyframe.return_value = _make_step_result( success=False, final_state="keyframe_mechanical_failed", output_path="/tmp/fail.jpg", cost_usd=0.10, gate_verdict=fail_verdict, ) result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) assert result.status == "attempts_exhausted" assert result.review_queue_id is not None # Verify review queue file was created rq_path = tmp_path / "_pipeline" / "state" / "visual" / "review_queue.jsonl" assert rq_path.exists() def test_icu_escalated(self, tmp_path): """StepRunner returns icu_escalated -> icu_escalated + review queue.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() runner.execute_keyframe.return_value = _make_step_result( success=False, final_state="icu_escalated", output_path="/tmp/icu.jpg", cost_usd=0.30, ) result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) assert result.status == "icu_escalated" assert result.review_queue_id is not None def test_crashed(self, tmp_path): """Unhandled exception outside retry loop -> crashed.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() # Cause an exception OUTSIDE the inner try/except by making ops_log # blow up at op_id generation time (before the retry loop) with patch( "recoil.pipeline._lib.run_shot.ops_log.make_op_id", side_effect=RuntimeError("uuid7 broken"), ): result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) assert result.status == "crashed" def test_needs_review_unknown_failure(self, tmp_path): """UNKNOWN FailureMode -> immediate needs_review.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() fail_verdict = _make_gate_verdict( passed=False, gate_name="gate_1", reason="Unknown issue", details={"failure_category": "unknown"}, ) runner.execute_keyframe.return_value = _make_step_result( success=False, final_state="keyframe_semantic_failed", output_path="/tmp/fail.jpg", cost_usd=0.10, gate_verdict=fail_verdict, ) result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) assert result.status == "needs_review" assert result.review_queue_id is not None assert result.failure_mode == "unknown" # --------------------------------------------------------------------------- # Tests: FailureMode action routing # --------------------------------------------------------------------------- class TestRunShotActionRouting: def test_accept_soft_identity_drift(self, tmp_path): """SOFT identity_drift -> ACCEPT -> ok with validation_notes.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() # SOFT identity drift: passed=False but only soft failures gv = _make_gate_verdict( passed=False, gate_name="identity_gate", reason="soft drift", details={"failure_category": "identity_drift", "hard_failures": []}, ) runner.execute_keyframe.return_value = _make_step_result( success=False, final_state="keyframe_semantic_failed", output_path="/tmp/soft_drift.jpg", cost_usd=0.10, gate_verdict=gv, ) result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) assert result.status == "ok" assert any("identity_drift" in note for note in result.validation_notes) def test_auto_reroll_anatomy_face_merge(self, tmp_path): """ANATOMY_FACE_MERGE -> AUTO_REROLL -> exhausts -> review queue.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() gv = _make_gate_verdict( passed=False, gate_name="gate_1", reason="face merge", details={"failure_category": "anatomy_face_merge"}, ) runner.execute_keyframe.return_value = _make_step_result( success=False, final_state="keyframe_mechanical_failed", output_path="/tmp/face.jpg", cost_usd=0.10, gate_verdict=gv, ) result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) # After 4 outer attempts of AUTO_REROLL, should exhaust assert result.status == "attempts_exhausted" assert result.attempts == 4 assert runner.execute_keyframe.call_count == 4 def test_soften_retry_content_filter(self, tmp_path): """CONTENT_FILTER_HARD_BLOCK -> SOFTEN_RETRY -> softened + retry.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() # First call: content filter error runner.execute_keyframe.side_effect = [ _make_step_result( success=False, final_state="keyframe_mechanical_failed", output_path=None, cost_usd=0.10, error="content_filter violation: blocked", ), # Second call (after softening): succeeds _make_step_result( success=True, output_path="/tmp/softened.jpg", cost_usd=0.10 ), ] with patch( "recoil.pipeline._lib.run_shot.soften_prompt", return_value="softened prompt", ): with patch( "recoil.pipeline._lib.run_shot.is_model_soften_eligible", return_value=True, ): result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) assert result.status == "ok" assert result.output_path == "/tmp/softened.jpg" def test_soften_retry_double_failure_to_review_queue(self, tmp_path): """Content filter -> soften -> second failure -> REVIEW_QUEUE.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() # All calls return content filter error runner.execute_keyframe.return_value = _make_step_result( success=False, final_state="keyframe_mechanical_failed", output_path=None, cost_usd=0.10, error="content_filter violation: blocked", ) with patch( "recoil.pipeline._lib.run_shot.soften_prompt", return_value="softened prompt", ): with patch( "recoil.pipeline._lib.run_shot.is_model_soften_eligible", return_value=True, ): result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) # After softening once and second failure, should go to review queue assert result.status == "needs_review" assert result.review_queue_id is not None # --------------------------------------------------------------------------- # Tests: pending_qc mapping + ops log # --------------------------------------------------------------------------- class TestRunShotSpecialCases: def test_pending_qc_maps_to_needs_review(self, tmp_path): """pending_qc from StepRunner maps to needs_review (NOT crashed).""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() runner.execute_keyframe.return_value = _make_step_result( success=False, final_state="pending_qc", output_path="/tmp/pending.jpg", cost_usd=0.10, ) result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) assert result.status == "needs_review" assert result.output_path == "/tmp/pending.jpg" # Should NOT be crashed (would cause regeneration on resume) assert result.status != "crashed" def test_ops_log_two_line_protocol(self, tmp_path): """Verify ops.log.jsonl has matching pending + completed lines.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() runner.execute_keyframe.return_value = _make_step_result(success=True) result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) log_path = tmp_path / "_pipeline" / "state" / "visual" / "ops.log.jsonl" assert log_path.exists() lines = log_path.read_text().strip().splitlines() assert len(lines) == 2 # pending + completed pending = json.loads(lines[0]) completed = json.loads(lines[1]) assert pending["status"] == "pending" assert completed["status"] == "completed" assert pending["id"] == completed["id"] # same op_id def test_op_id_format(self, tmp_path): """Verify op_id has 'op_' format.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() runner.execute_keyframe.return_value = _make_step_result(success=True) result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) assert result.op_id.startswith("op_") assert len(result.op_id) == 15 # "op_" + 12 hex chars def test_never_raises(self, tmp_path): """Verify D6: run_shot never raises on expected failure.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() # Make everything explode at op_id generation (outside retry loop) with patch( "recoil.pipeline._lib.run_shot.ops_log.make_op_id", side_effect=Exception("Total catastrophe"), ): # This MUST NOT raise result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) assert result.status == "crashed" def test_step_runner_exceptions_exhaust_attempts(self, tmp_path): """StepRunner raising on every call -> attempts_exhausted (not crashed).""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() runner.execute_keyframe.side_effect = RuntimeError("Infra failure") result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) # StepRunner exceptions within retry loop exhaust attempts assert result.status == "attempts_exhausted" assert result.attempts == 4 def test_per_shot_budget_cap(self, tmp_path): """Per-shot budget cap enforcement.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard(limit=100.0, per_shot_cap=0.01) runner = _make_step_runner() with patch( "recoil.pipeline._lib.run_shot._get_estimated_cost", return_value=0.15 ): result = run_shot( shot=_make_shot(), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) assert result.status == "budget_exhausted" def test_video_pipeline_calls_execute_video(self, tmp_path): """Video pipeline routes to execute_video.""" from recoil.pipeline._lib.run_shot import run_shot paths = _make_paths(tmp_path) guard = _make_budget_guard() runner = _make_step_runner() runner.execute_video.return_value = _make_step_result( success=True, final_state="video_complete", ) result = run_shot( shot=_make_shot(pipeline="video"), store=MagicMock(), paths=paths, budget_guard=guard, model="test-model", step_runner=runner, ) assert result.status == "ok" runner.execute_video.assert_called_once() runner.execute_keyframe.assert_not_called()