"""CP-7 Phase 5 — Standalone integration POC for Take/Beat keyframe→video. Proves the Take/Beat layer composes a realistic multi-take scenario end-to-end against a stubbed StepRunner. NO PRODUCTION CODE IS MODIFIED in this phase — see SPEC § Phase 5 for the rationale (mirrors CP-6 Phase 5's revised test-only POC scope; production migration deferred to a dedicated CP after CP-9). """ import sys import pathlib from types import SimpleNamespace sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent.parent.parent)) from recoil.core.paths import ensure_pipeline_importable # noqa: E402 ensure_pipeline_importable() import pytest # noqa: E402 from recoil.pipeline.core.dispatch import _reset_bootstrap_for_tests # noqa: E402 from recoil.pipeline.core.dispatch_context import DispatchContext # noqa: E402 from recoil.pipeline.core.registry import _reset_for_tests # noqa: E402 from recoil.pipeline.core.take import Beat, Scene, Take # noqa: E402 from recoil.pipeline.core.workflow import Workflow, WorkflowStep # noqa: E402 class _StubStepRunner: """Mirrors the dispatch-time interface that runners delegate to.""" def __init__(self, *, kf_succeeds=True, vid_succeeds=True): self._dispatch_path = "unknown" self.kf_succeeds = kf_succeeds self.vid_succeeds = vid_succeeds self.kf_calls = 0 self.vid_calls = 0 def execute_keyframe(self, **kwargs): self.kf_calls += 1 return SimpleNamespace( shot_id=kwargs.get("shot_id", "TEST"), success=self.kf_succeeds, final_state="keyframe_generated" if self.kf_succeeds else "keyframe_failed", output_path="/tmp/poc.png" if self.kf_succeeds else None, cost_usd=0.04, error=None if self.kf_succeeds else "kf stub fail", take_index=0, gate_verdict=None, model="nbp", pipeline="still", ) def execute_video(self, **kwargs): self.vid_calls += 1 return SimpleNamespace( shot_id=kwargs.get("shot_id", "TEST"), success=self.vid_succeeds, final_state="video_complete" if self.vid_succeeds else "video_failed", output_path="/tmp/poc.mp4" if self.vid_succeeds else None, cost_usd=0.20, error=None if self.vid_succeeds else "vid stub fail", take_index=0, gate_verdict=None, model="seeddance-2.0", pipeline="i2v", ) def _kf_video_workflow(workflow_id: str, shot_id: str = "EP001_SH02") -> Workflow: return Workflow( workflow_id=workflow_id, steps=[ WorkflowStep(step_id="keyframe", modality="image_t2i", payload={"shot_id": shot_id, "prompt": "p", "model": "nbp", "aspect_ratio": "9_16"}), WorkflowStep( step_id="video", modality="video_i2v", payload={"shot_id": shot_id, "prompt": "p", "model": "seeddance-2.0", "start_frame": "/tmp/poc.png", "duration": 5, "aspect_ratio": "9_16"}, depends_on=["keyframe"], ), ], global_provenance={"shot_id": shot_id, "scene_id": "ep001_sc02"}, ) @pytest.fixture(autouse=True) def reset(): _reset_for_tests(); _reset_bootstrap_for_tests() yield _reset_for_tests(); _reset_bootstrap_for_tests() def test_take_beat_poc_happy_path(): """Single take, succeeds. Beat picks it as primary via first_success.""" sr = _StubStepRunner() ctx = DispatchContext(caller_id="cp7_phase5_poc", step_runner=sr, project="poc_test", episode=1, receipts_log_path="DISABLED") beat = Beat(beat_id="EP001_SH02") take = beat.new_take(workflow=_kf_video_workflow("wf_take_0")) take.execute(context=ctx) assert take.status == "succeeded" assert sr.kf_calls == 1 assert sr.vid_calls == 1 chosen = beat.select_primary() assert chosen == take.take_id assert beat.primary_take is take def test_take_beat_poc_first_take_fails_second_succeeds(): """Take 0 fails (kf fails); Take 1 succeeds. Primary = take 1.""" ctx_fail = DispatchContext( caller_id="cp7_phase5_poc", step_runner=_StubStepRunner(kf_succeeds=False), receipts_log_path="DISABLED", ) ctx_ok = DispatchContext( caller_id="cp7_phase5_poc", step_runner=_StubStepRunner(kf_succeeds=True), receipts_log_path="DISABLED", ) beat = Beat(beat_id="EP001_SH02") take_a = beat.new_take(workflow=_kf_video_workflow("wf_take_0")) take_a.execute(context=ctx_fail) assert take_a.status == "failed" # Reset registry/bootstrap so the second StepRunner gets wired (CP-6 handoff §5). _reset_for_tests(); _reset_bootstrap_for_tests() take_b = beat.new_take(workflow=_kf_video_workflow("wf_take_1")) take_b.execute(context=ctx_ok) assert take_b.status == "succeeded" chosen = beat.select_primary(strategy="first_success") assert chosen == take_b.take_id # take_a failed, take_b is first success def test_take_beat_poc_partial_take_not_chosen(): """Take that ends 'partial' (kf ok, vid fails) is NOT chosen by first_success.""" sr = _StubStepRunner(vid_succeeds=False) ctx = DispatchContext(caller_id="cp7_phase5_poc", step_runner=sr, receipts_log_path="DISABLED") beat = Beat(beat_id="EP001_SH02") take = beat.new_take(workflow=_kf_video_workflow("wf_take_0")) take.execute(context=ctx) assert take.status == "partial" # kf succeeded, vid failed chosen = beat.select_primary(strategy="first_success") assert chosen is None # partial doesn't qualify def test_take_beat_poc_re_attempt_idiom(): """Re-attempt = new_take, not mutate. Stub uses fresh runners per attempt to simulate independent attempts.""" beat = Beat(beat_id="EP001_SH02") # Attempt 0 (vid fails → partial) sr_a = _StubStepRunner(vid_succeeds=False) ctx_a = DispatchContext(caller_id="cp7_phase5_poc", step_runner=sr_a, receipts_log_path="DISABLED") take_a = beat.new_take(workflow=_kf_video_workflow("wf_take_0")) take_a.execute(context=ctx_a) assert take_a.status == "partial" # Reset registry/bootstrap so the second StepRunner gets wired (CP-6 handoff §5). _reset_for_tests(); _reset_bootstrap_for_tests() # Attempt 1 (all good → succeeded) sr_b = _StubStepRunner() ctx_b = DispatchContext(caller_id="cp7_phase5_poc", step_runner=sr_b, receipts_log_path="DISABLED") take_b = beat.new_take(workflow=_kf_video_workflow("wf_take_1")) take_b.execute(context=ctx_b) assert take_b.status == "succeeded" # Beat now has 2 distinct Takes; primary is the succeeded one assert len(beat.takes) == 2 assert [t.take_index for t in beat.takes] == [0, 1] assert beat.select_primary() == take_b.take_id def test_take_beat_poc_round_trip_serialization(): """Executed Beat with multiple takes round-trips through to_dict/from_dict.""" sr = _StubStepRunner() ctx = DispatchContext(caller_id="cp7_phase5_poc", step_runner=sr, receipts_log_path="DISABLED") beat = Beat(beat_id="EP001_SH02", beat_metadata={"scene_id": "ep001_sc02"}) t0 = beat.new_take(workflow=_kf_video_workflow("wf_take_0")) t0.execute(context=ctx) t1 = beat.new_take(workflow=_kf_video_workflow("wf_take_1")) t1.execute(context=ctx) beat.select_primary() dumped = beat.to_dict() reloaded = Beat.from_dict(dumped) assert reloaded.beat_id == "EP001_SH02" assert len(reloaded.takes) == 2 assert reloaded.takes[0].status == "succeeded" assert reloaded.takes[1].status == "succeeded" assert reloaded.primary_take_id == t0.take_id # first_success picks idx 0 assert reloaded.beat_metadata == {"scene_id": "ep001_sc02"} def test_take_beat_poc_scene_groups_beats(): """Scene composes Beats. Round-trip preserves the full hierarchy.""" sr = _StubStepRunner() ctx = DispatchContext(caller_id="cp7_phase5_poc", step_runner=sr, receipts_log_path="DISABLED") scene = Scene(scene_id="ep001_sc02", scene_metadata={"location": "hallway"}) for i in range(2): beat = Beat(beat_id=f"EP001_SH0{i}") take = beat.new_take(workflow=_kf_video_workflow(f"wf_sh{i}")) take.execute(context=ctx) beat.select_primary() scene.add_beat(beat) assert len(scene.beats) == 2 assert all(b.primary_take is not None for b in scene.beats) reloaded = Scene.from_dict(scene.to_dict()) assert reloaded.scene_id == "ep001_sc02" assert len(reloaded.beats) == 2 assert reloaded.scene_metadata == {"location": "hallway"} assert all(b.primary_take_id is not None for b in reloaded.beats)