# api/routes/generation.py """Generation pipeline endpoints — Phase 4. Ports ~18 generation endpoints from review_server.py. All generation endpoints use submit_task() from api.state and return 202 immediately. Background threads do the actual work. """ import json import os import re import time from pathlib import Path from fastapi import APIRouter, Body, Depends from fastapi.responses import JSONResponse from ..state import submit_task, gen_tracker, PROJECT_ROOT, _broadcast_task_event from ..deps import ( get_project, get_paths, get_store, to_serving_path, get_project_aspect_ratio, ) from .dailies import load_shot_data, save_log from recoil.execution.execution_store import InvalidTransitionError from recoil.pipeline.core.cost import read_cost_from_result router = APIRouter(tags=["generation"]) # Legacy model ID map — accept old IDs from clients, resolve to canonical names LEGACY_MODEL_MAP = { "kling-3.0": "kling-v3", "kling-3.0-fal": "kling-v3", "kling-v3-fal": "kling-v3", "kling-o3-fal": "kling-o3", } # ── Utility: safe frame resolution ─────────────────────────────────── def _safe_resolve_frame(rel_path, paths): """Resolve a relative frame path to an absolute path safely.""" if not rel_path: return None if ".." in str(rel_path): return None for root in (paths["project_dir"], PROJECT_ROOT): p = root / rel_path if not Path(rel_path).is_absolute() else Path(rel_path) resolved = p.resolve() if resolved.is_relative_to(root.resolve()) and resolved.exists(): return resolved return None # ═══════════════════════════════════════════════════════════════════════ # POST /api/set-previz-hero # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/set-previz-hero") def set_previz_hero( body: dict = Body(default={}), project: str = Depends(get_project), store=Depends(get_store), ): """Set a specific take as the hero for a shot.""" shot_id = body.get("shot_id") take_index = body.get("take_index") if not shot_id or take_index is None: return JSONResponse({"error": "Missing shot_id or take_index"}, status_code=400) shot = store.get_shot(shot_id) if not shot: return JSONResponse({"error": f"Shot not found: {shot_id}"}, status_code=404) takes = shot.get("takes", []) if take_index < 0 or take_index >= len(takes): return JSONResponse({"error": f"Invalid take_index: {take_index}"}, status_code=400) take = takes[take_index] file_path = take.get("file_path") if not file_path: return JSONResponse({"error": "Take has no file_path"}, status_code=400) store.update_shot( shot_id, gate_results={"hero_frame": file_path}, output_path=file_path, ) return JSONResponse({ "shot_id": shot_id, "hero_frame": file_path, "take_index": take_index, }) # ═══════════════════════════════════════════════════════════════════════ # POST /api/generate-previz # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/generate-previz") def generate_previz( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), store=Depends(get_store), ): """Generate a single previz frame for a shot (async via background thread).""" shot_id = body.get("shot_id") if not shot_id: return JSONResponse({"error": "Missing shot_id"}, status_code=400) prompt_override = body.get("prompt_override") # Previz allows concurrent generations — use unique key per request import uuid as _uuid gen_key = f"{shot_id}_previz_{_uuid.uuid4().hex[:8]}" gen_tracker.try_start(gen_key) ep_match = re.match(r"EP(\d+)_SH(\d+)([A-Za-z]?)", shot_id) if not ep_match: gen_tracker.finish(gen_key) return JSONResponse({"error": f"Invalid shot_id format: {shot_id}"}, status_code=400) ep_num = int(ep_match.group(1)) shot_num = int(ep_match.group(2)) shot_suffix = ep_match.group(3).lower() shot_label = f"{shot_num:03d}{shot_suffix}" # Upsert: auto-create shot in ExecutionStore if not present shot = store.get_shot(shot_id) if shot is None: episode_id = f"EP{ep_num:03d}" store.insert_shot({ "shot_id": shot_id, "episode_id": episode_id, "pipeline": "still", "status": "previs_pending", }) shot = store.get_shot(shot_id) # Load plan to get shot data for prompt building ep_dir = f"ep_{ep_num:03d}" plan_path = paths["plans_dir"] / f"ep_{ep_num:03d}_plan.json" shot_data = None if plan_path.exists(): try: plan = json.loads(plan_path.read_text(encoding="utf-8")) for s in plan.get("shots", []): if s.get("shot_id") == shot_id: shot_data = s break except (json.JSONDecodeError, IOError): pass # ── Apply director overrides (if any) ── use_overrides = body.get("use_overrides", False) if use_overrides and shot_data: from .overrides import _load_overrides all_ov = _load_overrides(paths["plans_dir"], ep_num) shot_overrides = all_ov.get(shot_id, {}) if shot_overrides: from recoil.pipeline._lib.previz_context import apply_overrides shot_data = apply_overrides(shot_data, shot_overrides) print(f" [OVERRIDE] {shot_id}: applied overrides: {list(shot_overrides.keys())}") # Load bible for prompt enrichment bible = None bible_path = paths["bible_path"] if bible_path.exists(): try: bible = json.loads(bible_path.read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError): pass # Load all shots for sequence context all_shots = [] if plan_path.exists(): try: plan_full = json.loads(plan_path.read_text(encoding="utf-8")) all_shots = plan_full.get("shots", []) except (json.JSONDecodeError, IOError): pass # Build context — always use full context when shot_data exists. use_full_context = (shot_data is not None) context_parts = None print(f" [DEBUG] {shot_id}: shot_data={'YES' if shot_data else 'NO'}, prompt_override={'YES' if prompt_override else 'NO'}, use_full_context={use_full_context}") # ── Moodboard-to-text: convert location image to text for 3+ char shots ── if use_full_context and shot_data: from recoil.core.paths import get_config as _get_starsend_config _ss_cfg = _get_starsend_config() if _ss_cfg.get("enable_moodboard_to_text", True): chars = shot_data.get("asset_data", {}).get("characters", []) if len(chars) >= 3: _loc_id = shot_data.get("asset_data", {}).get("location_id", "") if _loc_id: try: from recoil.pipeline._lib.previz_context import resolve_location_refs from recoil.pipeline._lib.keyframe_context import describe_moodboard _loc_refs = resolve_location_refs(_loc_id, project=project, max_refs=1) if _loc_refs: _mb_result = describe_moodboard(_loc_refs[0][0], _loc_id, _loc_refs[0][1], project=project) if _mb_result.get("success"): shot_data["_moodboard_text"] = _mb_result["text"] print(f" [MOODBOARD] {shot_id}: location '{_loc_id}' converted to text ({len(_mb_result['text'])} chars)") except Exception as e: print(f" [MOODBOARD] {shot_id}: failed — {e}") # ── Scene visual locks: load existing locks for this scene ── if _ss_cfg.get("enable_scene_visual_locks", True): _scene_index = shot_data.get("scene_index", 0) try: from recoil.pipeline._lib.keyframe_context import load_scene_locks _scene_locks = load_scene_locks(ep_num, _scene_index, project) if _scene_locks: shot_data["_scene_visual_locks"] = _scene_locks print(f" [DNA] {shot_id}: scene {_scene_index} locks loaded") except Exception as e: print(f" [DNA] {shot_id}: failed to load scene locks — {e}") # ── Scene shots: populate _scene_shots for spatial continuity ── try: from recoil.pipeline._lib.previz_context import inject_scene_context inject_scene_context(shot_data, all_shots) print(f" [SPATIAL] {shot_id}: scene context injected ({len(shot_data.get('_scene_shots', []))} shots)") except Exception as e: print(f" [SPATIAL] {shot_id}: failed to inject scene context — {e}") if use_full_context: try: from recoil.pipeline._lib.previz_context import build_previz_context context_parts = build_previz_context( shot=shot_data, all_shots=all_shots, bible=bible, episode=ep_num, project=project, ) n_img = sum(1 for d, _, _ in context_parts if d is not None) print(f" [DEBUG] {shot_id}: context built — {len(context_parts)} parts, {n_img} images") if prompt_override and context_parts: for idx in range(len(context_parts) - 1, -1, -1): data, mime, label = context_parts[idx] if data is None and mime == "text": context_parts[idx] = (None, "text", f"# GENERATE THIS FRAME\n\n{prompt_override}") print(f" [DEBUG] {shot_id}: prompt override injected into context slot {idx}") break except ImportError as e: print(f" [DEBUG] {shot_id}: ImportError building context: {e}") use_full_context = False prompt = None if not use_full_context: prompt = prompt_override if not prompt and shot_data: try: from recoil.pipeline._lib.prompt_engine import build_previs_prompt prompt = build_previs_prompt(shot_data, bible=bible) except ImportError: pass if not prompt: gen_tracker.finish(gen_key) return JSONResponse({"error": "No prompt available — provide prompt_override or ensure plan exists"}, status_code=400) # Save prompt override to generation log if provided if prompt_override: log_data = load_shot_data(ep_dir, frames_dir=paths["frames_dir"]) or {"episode": ep_num} overrides = log_data.setdefault("manual_prompt_overrides", {}) overrides[shot_id] = {"prompt": prompt_override, "override_at": time.time()} save_log(ep_dir, log_data, frames_dir=paths["frames_dir"]) # Verify tools are importable before dispatching try: from tools.generate_previs import _generate_flash_frame, PREVIS_COST except ImportError: gen_tracker.finish(gen_key) return JSONResponse({"error": "generate_previs tools not available"}, status_code=503) # Resolve character refs for legacy path (canonical filesystem resolution) legacy_refs = [] if not use_full_context and shot_data: from recoil.pipeline._lib.previz_context import resolve_all_character_refs legacy_refs = resolve_all_character_refs(shot_data, project=project) # Mark shot as generating _prior_status = (store.get_shot(shot_id) or {}).get("status", "") try: store.update_shot(shot_id, status="previs_generating") except InvalidTransitionError: print(f" [INFO] {shot_id}: status {_prior_status} can't transition to previs_generating — generating take without status change") # ── Background thread function ── # Capture closures pp = paths _previs_dir = paths["previs_dir"] _context_parts = context_parts _prompt = prompt _legacy_refs = legacy_refs _prior_status_for_bg = _prior_status def _bg_generate(): try: _current = store.get_shot(shot_id) _cur_status = _current.get("status", "") if _current else "" if _cur_status != "previs_generating" and _cur_status != _prior_status_for_bg: print(f" [SKIP] {shot_id}: status changed to {_cur_status}, aborting generation") return previs_dir = _previs_dir / ep_dir previs_dir.mkdir(parents=True, exist_ok=True) if _context_parts: gen_result = _generate_flash_frame(context_parts=_context_parts) elif _legacy_refs: gen_result = _generate_flash_frame(prompt=_prompt, ref_images=_legacy_refs) else: gen_result = _generate_flash_frame(prompt=_prompt) if not gen_result.get("success"): print(f" [WARN] Generate failed for {shot_id}: {gen_result.get('error')}") try: store.update_shot(shot_id, status="previs_mechanical_failed", error_message=gen_result.get("error", "Unknown")) except InvalidTransitionError: print(f" [WARN] Shot {shot_id}: could not transition to previs_mechanical_failed") return fresh_shot = store.get_shot(shot_id) current_takes = fresh_shot.get("takes", []) if fresh_shot else [] take_num = len(current_takes) + 1 _take_ts = int(time.time() * 1000) % 100000 take_uid = f"{take_num}_{_take_ts}" output_path = previs_dir / f"shot_{shot_label}_take{take_uid}.png" output_path.write_bytes(gen_result["image_data"]) try: from recoil.pipeline._lib.validation import Validator _v = Validator() _g1 = _v.run_gate_1_image(output_path) gate_1 = {"passed": _g1.passed, "details": _g1.details, "cost": _g1.cost} except Exception as _e: gate_1 = {"passed": True, "reason": f"Gate 1 unavailable: {_e}", "cost": 0.0} _rel_path = to_serving_path(output_path, pp) authored_prompt = gen_result.get("authored_prompt", "") display_prompt = authored_prompt or _prompt or "(flash-authored)" # ── Spatial compliance check ── spatial_compliance = None if shot_data and shot_data.get("spatial_data", {}).get("camera_side"): try: from recoil.pipeline._lib.spatial_compliance import run_spatial_compliance from recoil.pipeline._lib.prompt_engine import build_spatial_continuity_block _spatial_block = build_spatial_continuity_block( shot=shot_data, bible=bible or {}, scene_shots=shot_data.get("_scene_shots"), ) _prev_compliance = None _prev_camera_side = None if shot_data.get("_scene_shots"): _scene_shots = shot_data["_scene_shots"] _curr_idx = next( (i for i, s in enumerate(_scene_shots) if s.get("shot_id") == shot_id), -1 ) if _curr_idx > 0: _prev_shot_id = _scene_shots[_curr_idx - 1].get("shot_id") if _prev_shot_id: _prev_shot = store.get_shot(_prev_shot_id) if _prev_shot: _prev_takes = _prev_shot.get("takes") or [] for _t in reversed(_prev_takes) if isinstance(_prev_takes, list) else []: if _t.get("spatial_compliance"): _prev_compliance = _t["spatial_compliance"] break _prev_spatial = _scene_shots[_curr_idx - 1].get("spatial_data", {}) _prev_camera_side = _prev_spatial.get("camera_side") spatial_compliance = run_spatial_compliance( image_data=gen_result["image_data"], shot=shot_data, bible=bible or {}, prev_shot_compliance=_prev_compliance, prev_camera_side=_prev_camera_side, authored_prompt=authored_prompt, spatial_block=_spatial_block, ) _sev = spatial_compliance.get("severity", "PASS") _flags = [f["flag"] for f in spatial_compliance.get("flags", [])] if _sev != "PASS": print(f" [SPATIAL-CHECK] {shot_id}: {_sev} — {', '.join(_flags)}") else: print(f" [SPATIAL-CHECK] {shot_id}: PASS") except Exception as e: print(f" [SPATIAL-CHECK] {shot_id}: check failed — {e}") _total_cost = PREVIS_COST if spatial_compliance and not spatial_compliance.get("skipped"): try: from recoil.pipeline._lib.spatial_compliance import COMPLIANCE_CHECK_COST as _cc_cost _total_cost += _cc_cost except ImportError: _total_cost += 0.010 # ── Capture compiled prompt for Dailies SENT view ── _compiled_for_audit = None _compilation_bypasses = [] if shot_data: try: from recoil.pipeline._lib.prompt_engine import build_prompt_from_plan, _resolve_bypasses from recoil.pipeline._lib.recoil_bridge import load_project_config _pc = load_project_config(project=project) _bd = bible if _pc and _bd: _compiled_for_audit = build_prompt_from_plan( shot_data, _bd, _pc, episode=ep_num ) _compilation_bypasses = list(_resolve_bypasses(shot_data)) except Exception as _e: print(f" [DEBUG] {shot_id}: compiled prompt capture failed: {_e}") take_record = { "take_id": f"take_{take_uid}", "take_num": take_num, "file_path": _rel_path, "prompt": display_prompt, "authored_prompt": authored_prompt, "compiled_prompt": _compiled_for_audit, "compilation_bypasses": _compilation_bypasses, "cost": _total_cost, "gate_1": gate_1, "spatial_compliance": spatial_compliance, "generated_at": time.time(), } _new_status = "previs_generated" try: store.update_shot( shot_id, status=_new_status, output_path=_rel_path, append_take=take_record, cost_incurred=PREVIS_COST, ) except InvalidTransitionError: store.update_shot( shot_id, output_path=_rel_path, append_take=take_record, cost_incurred=PREVIS_COST, ) # Auto-select the new take fresh = store.get_shot(shot_id) if fresh: updated_takes = fresh.get("takes", []) for t in updated_takes: if t.get("take_id") == take_record["take_id"]: t["approved"] = True else: t.pop("approved", None) store.update_shot(shot_id, takes=updated_takes, gate_results={"hero_frame": _rel_path}) # Also copy to the canonical previs path import shutil as _shutil also_save = previs_dir / f"shot_{shot_label}.png" _shutil.copy2(output_path, also_save) if authored_prompt: print(f" [OK] {shot_id} take {take_num} generated (flash-authored) — ${PREVIS_COST}") print(f" Authored prompt: {authored_prompt[:120]}") else: print(f" [OK] {shot_id} take {take_num} generated — ${PREVIS_COST}") # ── Scene DNA extraction ── if shot_data: try: from recoil.core.paths import get_config as _get_cfg _cfg = _get_cfg() if _cfg.get("enable_scene_visual_locks", True): _s_idx = shot_data.get("scene_index", 0) from recoil.pipeline._lib.keyframe_context import load_scene_locks, extract_scene_visual_dna _existing = load_scene_locks(ep_num, _s_idx, project) if not _existing: _dna = extract_scene_visual_dna( gen_result["image_data"], shot_data, bible or {}, project ) if _dna.get("success"): print(f" [DNA] Scene {_s_idx} visual locks extracted") except Exception as e: print(f" [DNA] Scene DNA extraction failed: {e}") return {"shot_id": shot_id} except Exception as exc: print(f" [ERR] Background generate for {shot_id}: {exc}") try: store.update_shot(shot_id, status="previs_mechanical_failed", error_message=str(exc)) except InvalidTransitionError: print(f" [WARN] Shot {shot_id}: could not transition to previs_mechanical_failed") raise finally: gen_tracker.finish(gen_key) task_id = submit_task(shot_id, "previz", _bg_generate) return JSONResponse({"status": "generating", "shot_id": shot_id, "task_id": task_id}, status_code=202) # ═══════════════════════════════════════════════════════════════════════ # POST /api/compose-shot # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/compose-shot") def compose_shot( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), store=Depends(get_store), ): """Generate a new shot via LLM (preview only). Does NOT insert it.""" episode_id = body.get("episode_id") after_shot_id = body.get("after_shot_id") description = body.get("description", "").strip() if not episode_id: return JSONResponse({"error": "Missing episode_id"}, status_code=400) if not after_shot_id: return JSONResponse({"error": "Missing after_shot_id"}, status_code=400) if not description or len(description) < 5: return JSONResponse({"error": "Description must be at least 5 characters"}, status_code=400) # Load plan ep_num = int(episode_id.replace("EP", "")) plan_path = paths["plans_dir"] / f"ep_{ep_num:03d}_plan.json" if not plan_path.exists(): return JSONResponse({"error": f"No plan found for {episode_id}"}, status_code=404) try: plan = json.loads(plan_path.read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError) as e: return JSONResponse({"error": f"Failed to read plan: {e}"}, status_code=500) plan_shots = plan.get("shots", []) # Find anchor shots after_shot_data = None before_shot_data = None before_shot_id = None for i, s in enumerate(plan_shots): if s.get("shot_id") == after_shot_id: after_shot_data = s if i + 1 < len(plan_shots): before_shot_data = plan_shots[i + 1] before_shot_id = before_shot_data["shot_id"] break if after_shot_data is None: return JSONResponse({"error": f"Shot {after_shot_id} not found in plan"}, status_code=404) # Load bible for context and enum constraints bible = None bible_path = paths["bible_path"] if bible_path.exists(): try: bible = json.loads(bible_path.read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError): pass if bible is None: return JSONResponse({"error": "Global Bible not found — required for shot composition"}, status_code=404) # Build dynamic enum lists from bible valid_location_ids = list(bible.get("locations", {}).keys()) valid_char_ids = list(bible.get("characters", {}).keys()) valid_wardrobe_phases = {} for char_id, char_data in bible.get("characters", {}).items(): phases = [p.get("phase_id", "") for p in char_data.get("phases", []) if p.get("phase_id")] if phases: valid_wardrobe_phases[char_id] = phases # Extract relevant bible entries for anchor characters anchor_char_ids = set() for shot in [after_shot_data, before_shot_data]: if shot: for c in shot.get("asset_data", {}).get("characters", []): anchor_char_ids.add(c.get("char_id", "")) for char_id in valid_char_ids: char_name = bible.get("characters", {}).get(char_id, {}).get("display_name", "") if char_name and char_name.lower() in description.lower(): anchor_char_ids.add(char_id) bible_context = {} for cid in anchor_char_ids: if cid in bible.get("characters", {}): char = bible["characters"][cid] bible_context[cid] = { "display_name": char.get("display_name"), "visual_description": char.get("visual_description"), "phases": char.get("phases", []), } anchor_locations = set() for shot in [after_shot_data, before_shot_data]: if shot: loc = shot.get("asset_data", {}).get("location_id", "") if loc: anchor_locations.add(loc) location_context = {} for lid in anchor_locations: if lid in bible.get("locations", {}): loc = bible["locations"][lid] location_context[lid] = { "description": loc.get("description"), "atmosphere": loc.get("atmosphere"), } # Compute new shot_id from orchestrator.execution_plan import compute_insert_id new_shot_id = compute_insert_id(episode_id, after_shot_id, before_shot_id) # Build LLM prompt scene_index = after_shot_data.get("scene_index", 1) llm_prompt = f"""You are a storyboard artist for a vertical microdrama series. Generate a complete shot record for a new shot to be inserted into the sequence. DIRECTOR'S INSTRUCTION: {description} SHOT BEFORE (anchor A): {json.dumps(after_shot_data, indent=2, default=str)} {"SHOT AFTER (anchor B):" if before_shot_data else "This is the last shot in the sequence."} {json.dumps(before_shot_data, indent=2, default=str) if before_shot_data else ""} CHARACTER BIBLE (relevant entries): {json.dumps(bible_context, indent=2, default=str)} LOCATION CONTEXT: {json.dumps(location_context, indent=2, default=str)} CONSTRAINTS: - scene_index must be: {scene_index} - location_id must be one of: {json.dumps(valid_location_ids)} - char_id must be one of: {json.dumps(valid_char_ids)} - For each character, wardrobe_phase_id must match their phases in the bible - shot_type must be one of: WS, LS, FS, MS, MCU, CU, ECU, INSERT - camera_movement must be one of: static, pan, tilt, push_in, pull_back, tracking, crane, handheld, steadicam, dolly - Maintain spatial continuity with anchor shots (screen direction, camera side, 180-degree rule) - target_editorial_duration_s must be 1-30 Generate ONLY the 5 consumer data groups: routing_data, prompt_data, spatial_data, asset_data, audio_data. Also include source_text (a concise description of the shot based on the director's instruction). Do NOT generate shot_id or compiled_prompts — those are computed by the backend.""" # Call Flash for structured output try: from google import genai from google.genai import types as genai_types api_key = os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY") if not api_key: return JSONResponse({"error": "GEMINI_API_KEY not set"}, status_code=500) client = genai.Client(api_key=api_key) from recoil.core.model_profiles import get_model as _get_model compose_model = _get_model("flash", "text") config = genai_types.GenerateContentConfig( temperature=0.4, max_output_tokens=4096, response_mime_type="application/json", ) response = client.models.generate_content( model=compose_model, contents=[llm_prompt], config=config, ) raw_text = response.text if hasattr(response, "text") else str(response) except Exception as e: return JSONResponse({"error": f"LLM call failed: {e}"}, status_code=500) # Parse and validate LLM response try: generated = json.loads(raw_text) except json.JSONDecodeError: try: response2 = client.models.generate_content( model=compose_model, contents=[llm_prompt + "\n\nIMPORTANT: Output valid JSON only. No markdown, no code blocks."], config=config, ) raw_text2 = response2.text if hasattr(response2, "text") else str(response2) generated = json.loads(raw_text2) except (json.JSONDecodeError, Exception) as e2: return JSONResponse({"error": f"LLM returned invalid JSON after retry: {e2}"}, status_code=500) # Assemble full ShotRecord shot_record = { "shot_id": new_shot_id, "scene_index": scene_index, "source_text": generated.get("source_text", description), "origin": "composed", "routing_data": generated.get("routing_data", {}), "prompt_data": generated.get("prompt_data", {}), "spatial_data": generated.get("spatial_data", {}), "asset_data": generated.get("asset_data", {}), "audio_data": generated.get("audio_data", {}), } # Validate full ShotRecord schema from recoil.pipeline._lib.render_schema import ShotRecord as _ShotRecord try: _ShotRecord(**shot_record) except Exception as val_err: return JSONResponse({ "error": f"Generated shot failed schema validation: {val_err}", "shot_preview": shot_record, }, status_code=422) # Validate asset IDs against bible validation_errors = [] asset = shot_record.get("asset_data", {}) if asset.get("location_id") and asset["location_id"] not in valid_location_ids: validation_errors.append(f"Invalid location_id: {asset['location_id']}") for char in asset.get("characters", []): cid = char.get("char_id", "") if cid and cid not in valid_char_ids: validation_errors.append(f"Invalid char_id: {cid}") wpid = char.get("wardrobe_phase_id", "") if cid in valid_wardrobe_phases and wpid and wpid not in valid_wardrobe_phases[cid]: validation_errors.append(f"Invalid wardrobe_phase_id '{wpid}' for {cid}") if validation_errors: return JSONResponse({ "error": "Asset validation failed", "validation_errors": validation_errors, "shot_preview": shot_record, }, status_code=422) # Return preview WITHOUT inserting anchors = [after_shot_id] if before_shot_id: anchors.append(before_shot_id) return JSONResponse({ "ok": True, "shot": shot_record, "shot_id": new_shot_id, "anchors_used": anchors, "committed": False, }) # ═══════════════════════════════════════════════════════════════════════ # POST /api/commit-compose # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/commit-compose") def commit_compose( body: dict = Body(default={}), project: str = Depends(get_project), store=Depends(get_store), ): """Commit a previously generated composed shot.""" episode_id = body.get("episode_id") after_shot_id = body.get("after_shot_id") shot_record = body.get("shot") if not episode_id: return JSONResponse({"error": "Missing episode_id"}, status_code=400) if not after_shot_id: return JSONResponse({"error": "Missing after_shot_id"}, status_code=400) if not shot_record or not shot_record.get("shot_id"): return JSONResponse({"error": "Missing shot data"}, status_code=400) from orchestrator.execution_plan import insert_composed_shot try: insert_composed_shot( project=project, episode_id=episode_id, after_shot_id=after_shot_id, shot_record=shot_record, store=store, ) except Exception as e: return JSONResponse({"error": f"Insert failed: {e}"}, status_code=500) # Auto-compile prompts for the composed shot if missing if not shot_record.get("compiled_prompts"): try: from recoil.pipeline._lib.prompt_engine import compile_all_prompts from recoil.pipeline._lib.recoil_bridge import get_bible_path, load_project_config import json as _json bible_path = get_bible_path(project=project) bible = _json.loads(bible_path.read_text()) if bible_path.exists() else {} config = load_project_config(project=project) ep_match = __import__("re").match(r"EP(\d+)", episode_id) ep_num = int(ep_match.group(1)) if ep_match else 1 compiled = compile_all_prompts(shot_record, bible=bible, project_config=config, episode=ep_num) shot_record["compiled_prompts"] = compiled # Write back to plan from ..deps import _paths_for_project pp = _paths_for_project(project) plan_path = pp["plans_dir"] / f"ep_{ep_num:03d}_plan.json" if plan_path.exists(): plan = _json.loads(plan_path.read_text()) for s in plan.get("shots", []): if s.get("shot_id") == shot_record["shot_id"]: s["compiled_prompts"] = compiled break plan_path.write_text(_json.dumps(plan, indent=2, default=str)) except Exception as compile_err: import logging logging.getLogger(__name__).warning(f"Auto-compile failed for {shot_record.get('shot_id')}: {compile_err}") return JSONResponse({ "ok": True, "shot_id": shot_record["shot_id"], "committed": True, }) # ═══════════════════════════════════════════════════════════════════════ # POST /api/delete-shot # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/delete-shot") def delete_shot( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), store=Depends(get_store), ): """Remove a shot from plan and execution store.""" shot_id = body.get("shot_id") episode_id = body.get("episode_id") if not shot_id: return JSONResponse({"error": "Missing shot_id"}, status_code=400) if not episode_id: return JSONResponse({"error": "Missing episode_id"}, status_code=400) deleted_from_store = store.delete_shot(shot_id) deleted_from_plan = False ep_num = int(episode_id.replace("EP", "")) plan_path = paths["plans_dir"] / f"ep_{ep_num:03d}_plan.json" if plan_path.exists(): try: plan = json.loads(plan_path.read_text(encoding="utf-8")) shots = plan.get("shots", []) original_count = len(shots) shots = [s for s in shots if s.get("shot_id") != shot_id] if len(shots) < original_count: deleted_from_plan = True plan["shots"] = shots plan["total_shots"] = len(shots) plan_path.write_text(json.dumps(plan, indent=2, default=str), encoding="utf-8") except (json.JSONDecodeError, IOError) as e: print(f"Warning: Failed to update plan file for shot {shot_id} deletion: {e}") if not deleted_from_store and not deleted_from_plan: return JSONResponse({"error": f"Shot {shot_id} not found"}, status_code=404) return JSONResponse({ "ok": True, "shot_id": shot_id, "deleted_from_store": deleted_from_store, "deleted_from_plan": deleted_from_plan, }) # ═══════════════════════════════════════════════════════════════════════ # POST /api/smart-prompt # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/smart-prompt") def smart_prompt( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), store=Depends(get_store), ): """Build an NBP-optimized keyframe prompt via Flash text call.""" shot_id = body.get("shot_id") if not shot_id: return JSONResponse({"error": "Missing shot_id"}, status_code=400) director_edit = body.get("director_edit") ep_match = re.match(r"EP(\d+)_SH(\d+)", shot_id) if not ep_match: return JSONResponse({"error": f"Invalid shot_id format: {shot_id}"}, status_code=400) ep_num = int(ep_match.group(1)) plan_path = paths["plans_dir"] / f"ep_{ep_num:03d}_plan.json" if not plan_path.exists(): return JSONResponse({"error": f"Plan not found for EP{ep_num:03d}"}, status_code=404) try: plan = json.loads(plan_path.read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError) as e: return JSONResponse({"error": f"Could not load plan: {e}"}, status_code=500) all_shots = plan.get("shots", []) shot_data = next((s for s in all_shots if s.get("shot_id") == shot_id), None) if not shot_data: return JSONResponse({"error": f"Shot {shot_id} not found in plan"}, status_code=404) bible = None if paths["bible_path"].exists(): try: bible = json.loads(paths["bible_path"].read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError): pass # Find approved previz image path previz_image_path = None exec_shot = store.get_shot(shot_id) if exec_shot: takes = exec_shot.get("takes", []) approved_take = next((t for t in takes if t.get("approved")), None) if approved_take and approved_take.get("file_path"): rel = approved_take["file_path"] if rel.startswith("output/"): previz_image_path = paths["output_dir"] / rel[len("output/"):] try: from recoil.pipeline._lib.keyframe_context import build_smart_prompt except ImportError as e: return JSONResponse({"error": f"keyframe_context not available: {e}"}, status_code=503) result = build_smart_prompt( shot=shot_data, all_shots=all_shots, bible=bible or {}, episode=ep_num, project=project, previz_image_path=previz_image_path, director_edit=director_edit, ) if "error" in result: return JSONResponse({"error": result["error"]}, status_code=500) pre_compiled = shot_data.get("prompt_data", {}).get("keyframe_nbp", "") if not pre_compiled: skeleton = shot_data.get("prompt_data", {}).get("prompt_skeleton", {}) pre_compiled = ", ".join(v for v in skeleton.values() if v) resp = { "prompt": result["prompt"], "flash_reasoning": result.get("flash_reasoning", ""), "base_prompt": pre_compiled, "cost": result.get("cost", 0.001), } if result.get("critic_result"): resp["critic_result"] = result["critic_result"] return JSONResponse(resp) # ═══════════════════════════════════════════════════════════════════════ # POST /api/generate-keyframe # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/generate-keyframe") def generate_keyframe( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), store=Depends(get_store), ): """Generate keyframe via NBP (async via background thread).""" shot_id = body.get("shot_id") prompt = body.get("prompt") if not shot_id: return JSONResponse({"error": "Missing shot_id"}, status_code=400) if not prompt: return JSONResponse({"error": "Missing prompt"}, status_code=400) ep_match = re.match(r"EP(\d+)_SH(\d+)([A-Za-z]?)", shot_id) if not ep_match: return JSONResponse({"error": f"Invalid shot_id format: {shot_id}"}, status_code=400) ep_num = int(ep_match.group(1)) shot_num = int(ep_match.group(2)) shot_suffix = ep_match.group(3).lower() shot_label = f"{shot_num:03d}{shot_suffix}" if not gen_tracker.try_start(shot_id): return JSONResponse({"error": f"Generation already in progress for {shot_id}"}, status_code=409) shot = store.get_shot(shot_id) if shot is None: gen_tracker.finish(shot_id) return JSONResponse({"error": f"Shot not found: {shot_id}"}, status_code=404) force = body.get("force", False) if not force and shot["status"] not in ("previs_approved", "keyframe_generated"): print(f" [WARN] Keyframe from non-standard status '{shot['status']}' for {shot_id} — proceeding") # Load shot data for ref building plan_path = paths["plans_dir"] / f"ep_{ep_num:03d}_plan.json" shot_data = None if plan_path.exists(): try: plan = json.loads(plan_path.read_text(encoding="utf-8")) shot_data = next((s for s in plan.get("shots", []) if s.get("shot_id") == shot_id), None) except (json.JSONDecodeError, IOError): pass # Verify tools are importable try: from tools.generate_previs import _generate_nbp_frame, NBP_COST from recoil.pipeline._lib.previz_context import build_keyframe_refs except ImportError as e: gen_tracker.finish(shot_id) return JSONResponse({"error": f"Required modules not available: {e}"}, status_code=503) # Build ref stack ref_images = [] if shot_data: ref_images = build_keyframe_refs(shot_data, project=project) print(f" [DEBUG] {shot_id}: build_keyframe_refs returned {len(ref_images)} refs") for i, (_, mime, label) in enumerate(ref_images): print(f" ref[{i}]: {mime} — {label}") else: print(f" [WARN] {shot_id}: shot_data is None — plan_path={plan_path}, exists={plan_path.exists()}") # 409 guard: reject if shot is already generating current_shot = store.get_shot(shot_id) if current_shot: current_status = current_shot.get("status", "") if current_status.endswith("_generating"): gen_tracker.finish(shot_id) return JSONResponse( {"error": f"Shot {shot_id} is already generating (status: {current_status}). Wait for it to complete."}, status_code=409, ) # Mark as generating store.update_shot(shot_id, status="keyframe_generating") # ── Background generation ── pp = paths _frames_dir = paths["frames_dir"] _prompt = prompt _ref_images = ref_images ep_dir = f"ep_{ep_num:03d}" def _bg_generate_keyframe(): try: frames_dir = _frames_dir / ep_dir frames_dir.mkdir(parents=True, exist_ok=True) gen_result = _generate_nbp_frame(prompt=_prompt, ref_images=_ref_images) if not gen_result.get("success"): print(f" [WARN] Keyframe generate failed for {shot_id}: {gen_result.get('error')}") try: store.update_shot(shot_id, status="keyframe_mechanical_failed", error_message=gen_result.get("error", "Unknown")) except InvalidTransitionError: print(f" [WARN] Shot {shot_id}: could not transition to keyframe_mechanical_failed") raise RuntimeError(gen_result.get("error", "Unknown")) # Determine keyframe take number fresh_shot = store.get_shot(shot_id) current_takes = fresh_shot.get("takes", []) if fresh_shot else [] kf_takes = [t for t in current_takes if t.get("layer") == "keyframe"] kf_take_num = len(kf_takes) + 1 output_path = frames_dir / f"shot_{shot_label}_keyframe_take{kf_take_num}.png" output_path.write_bytes(gen_result["image_data"]) try: from recoil.pipeline._lib.validation import Validator _v = Validator() _g1 = _v.run_gate_1_image(output_path) gate_1 = {"passed": _g1.passed, "details": _g1.details, "cost": _g1.cost} except Exception as _e: gate_1 = {"passed": True, "reason": f"Gate 1 unavailable: {_e}", "cost": 0.0} _rel_path = to_serving_path(output_path, pp) take_record = { "take_id": f"kf_take_{kf_take_num}", "take_num": kf_take_num, "layer": "keyframe", "file_path": _rel_path, "prompt": _prompt, "compiled_prompt": _prompt, "was_user_override": bool(body.get("prompt")), "refs_count": len(_ref_images), "cost": NBP_COST, "gate_1": gate_1, "generated_at": time.time(), } store.update_shot( shot_id, status="keyframe_generated", append_take=take_record, cost_incurred=NBP_COST, ) print(f" [OK] {shot_id} keyframe take {kf_take_num} generated — ${NBP_COST}") return {"shot_id": shot_id} except Exception as e: print(f" [ERR] Background keyframe generate for {shot_id}: {e}") try: store.update_shot(shot_id, status="keyframe_mechanical_failed", error_message=str(e)) except InvalidTransitionError: print(f" [WARN] Shot {shot_id}: could not transition to keyframe_mechanical_failed") raise finally: gen_tracker.finish(shot_id) task_id = submit_task(shot_id, "keyframe", _bg_generate_keyframe) return JSONResponse({"status": "generating", "shot_id": shot_id, "task_id": task_id}, status_code=202) # ═══════════════════════════════════════════════════════════════════════ # POST /api/lock-keyframe # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/lock-keyframe") def lock_keyframe( body: dict = Body(default={}), project: str = Depends(get_project), store=Depends(get_store), ): """Lock keyframe + set anchor role.""" shot_id = body.get("shot_id") anchor_role = body.get("anchor_role", "hero_frame") frame_position = body.get("frame_position", "middle") frame_path_override = body.get("frame_path") if not shot_id: return JSONResponse({"error": "Missing shot_id"}, status_code=400) if anchor_role not in ("first_frame", "last_frame", "hero_frame", "still_only"): return JSONResponse({"error": f"Invalid anchor_role: {anchor_role}"}, status_code=400) if frame_position not in ("first", "middle", "last"): return JSONResponse({"error": f"Invalid frame_position: {frame_position}"}, status_code=400) shot = store.get_shot(shot_id) if shot is None: return JSONResponse({"error": f"Shot not found: {shot_id}"}, status_code=404) if shot["status"] == "previs_pending": return JSONResponse( {"error": f"Cannot lock from status '{shot['status']}' — generate previz first"}, status_code=400, ) takes = shot.get("takes", []) kf_takes = [t for t in takes if t.get("layer") == "keyframe"] if frame_path_override: kf_path = frame_path_override else: kf_path = kf_takes[-1].get("file_path", "") if kf_takes else "" gate_update = { "anchor_role": anchor_role, "frame_position": frame_position, } if anchor_role == "hero_frame": gate_update["hero_frame"] = kf_path elif anchor_role == "first_frame": gate_update["first_frame"] = kf_path elif anchor_role == "last_frame": gate_update["last_frame"] = kf_path try: store.update_shot(shot_id, status="keyframe_approved", gate_results=gate_update) except InvalidTransitionError as e: return JSONResponse({"error": str(e), "status": 409}, status_code=409) return JSONResponse({ "shot_id": shot_id, "status": "keyframe_approved", "anchor_role": anchor_role, "frame_position": frame_position, }) # ═══════════════════════════════════════════════════════════════════════ # POST /api/extract-frame # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/extract-frame") def extract_frame( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), store=Depends(get_store), ): """Generate the extrapolated frame (first or last) from locked keyframe. Background thread.""" shot_id = body.get("shot_id") anchor_role = body.get("anchor_role") target_frame = body.get("target_frame") prompt_override = body.get("prompt_override") reference_image_override = body.get("reference_image") if not shot_id: return JSONResponse({"error": "Missing shot_id"}, status_code=400) ep_match = re.match(r"EP(\d+)_SH(\d+)([A-Za-z]?)", shot_id) if not ep_match: return JSONResponse({"error": f"Invalid shot_id format: {shot_id}"}, status_code=400) ep_num = int(ep_match.group(1)) shot_num = int(ep_match.group(2)) shot_suffix = ep_match.group(3).lower() shot_label = f"{shot_num:03d}{shot_suffix}" shot = store.get_shot(shot_id) if shot is None: return JSONResponse({"error": f"Shot not found: {shot_id}"}, status_code=404) if shot["status"] == "previs_pending": return JSONResponse( {"error": "Cannot extract frames — generate content first"}, status_code=400, ) gate = shot.get("gate_results", {}) if not anchor_role: anchor_role = gate.get("anchor_role", "first_frame") if anchor_role not in ("first_frame", "last_frame", "hero_frame"): return JSONResponse({"error": f"Invalid anchor_role: {anchor_role}"}, status_code=400) # Find the reference image pp = paths if reference_image_override: ref_rel = reference_image_override if ".." in str(ref_rel): return JSONResponse({"error": "Invalid path"}, status_code=400) keyframe_abs = None for root in (paths["output_dir"], paths["project_dir"], PROJECT_ROOT): candidate = root / ref_rel if not Path(ref_rel).is_absolute() else Path(ref_rel) resolved = candidate.resolve() if resolved.is_relative_to(root.resolve()) and resolved.is_file(): keyframe_abs = resolved break if keyframe_abs is None: return JSONResponse({"error": f"Reference image not found: {ref_rel}"}, status_code=404) else: takes = shot.get("takes", []) kf_takes = [t for t in takes if t.get("layer") == "keyframe"] if not kf_takes: return JSONResponse({"error": "No keyframe takes found"}, status_code=400) kf_take = kf_takes[-1] kf_rel = kf_take.get("file_path", "") if kf_rel.startswith("output/"): keyframe_abs = paths["output_dir"] / kf_rel[len("output/"):] else: return JSONResponse({"error": "Keyframe path not found"}, status_code=400) if not keyframe_abs.is_file(): return JSONResponse({"error": f"Keyframe file not found: {keyframe_abs}"}, status_code=404) # Load plan + bible for extrapolation prompt plan_path = paths["plans_dir"] / f"ep_{ep_num:03d}_plan.json" shot_data = None all_shots = [] if plan_path.exists(): try: plan = json.loads(plan_path.read_text(encoding="utf-8")) all_shots = plan.get("shots", []) shot_data = next((s for s in all_shots if s.get("shot_id") == shot_id), None) except (json.JSONDecodeError, IOError): pass bible = None if paths["bible_path"].exists(): try: bible = json.loads(paths["bible_path"].read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError): pass try: from recoil.pipeline._lib.keyframe_context import build_extrapolation_prompt except ImportError as e: return JSONResponse({"error": f"Required modules not available: {e}"}, status_code=503) # Determine which frame(s) to generate if target_frame == "both": frames_to_generate = ["first_frame", "last_frame"] elif target_frame in ("first_frame", "last_frame"): frames_to_generate = [target_frame] elif anchor_role == "hero_frame": frames_to_generate = ["first_frame", "last_frame"] else: frames_to_generate = ["last_frame" if anchor_role == "first_frame" else "first_frame"] print(f" [DEBUG] {shot_id}: using edit mode (mask-free edit of hero)") # ── Background generation ── _frames_dir = paths["frames_dir"] ep_dir = f"ep_{ep_num:03d}" _anchor_role = anchor_role _frames_to_gen = frames_to_generate def _bg_extract_frames(): import traceback as _tb import sys as _sys print(f" [BG] Starting extraction for {shot_id}: {_frames_to_gen}", flush=True) try: from recoil.pipeline._lib.frame_editor import edit_hero_pose, validate_companion, build_edit_instruction, EDIT_COST frames_dir = _frames_dir / ep_dir frames_dir.mkdir(parents=True, exist_ok=True) print(f" [BG] frames_dir: {frames_dir}, anchor_role: {_anchor_role}", flush=True) consecutive_failures = 0 for gen_frame in _frames_to_gen: print(f" [BG] Generating {gen_frame} via edit mode...", flush=True) if not shot_data: print(f" [WARN] No shot data for {shot_id}, skipping {gen_frame}", flush=True) continue extrap_result = build_extrapolation_prompt( shot=shot_data, all_shots=all_shots, bible=bible or {}, anchor_role=_anchor_role, keyframe_image_path=keyframe_abs, episode=ep_num, project=project, target_frame=gen_frame, ) if "error" in extrap_result: print(f" [WARN] Pose description failed for {shot_id} {gen_frame}: {extrap_result['error']}", flush=True) consecutive_failures += 1 if consecutive_failures >= 2: print(f" [FALLBACK] {shot_id}: 2 consecutive failures — marking as single-frame-only", flush=True) store.update_shot(shot_id, gate_results={"extraction_mode": "single_frame"}) break continue pose_text = extrap_result["prompt"] if prompt_override: pose_text += f" Director note: {prompt_override}" target_type = "anticipation" if gen_frame == "first_frame" else "aftermath" edit_instruction = build_edit_instruction(target_type, pose_text) print(f" [BG] Edit instruction ({len(edit_instruction)} chars)", flush=True) max_attempts = 2 success = False companion_bytes = None for attempt in range(1, max_attempts + 1): gen_result = edit_hero_pose(keyframe_abs, edit_instruction) if not gen_result.get("success"): print(f" [WARN] Edit attempt {attempt} failed for {shot_id} {gen_frame}: {gen_result.get('error')}", flush=True) consecutive_failures += 1 if consecutive_failures >= 2: break continue hero_bytes = keyframe_abs.read_bytes() companion_bytes = gen_result["image_data"] gate_check = validate_companion(hero_bytes, companion_bytes) print(f" [GATE] {shot_id} {gen_frame} attempt {attempt}: correlation={gate_check['correlation']}, passed={gate_check['passed']}", flush=True) if gate_check["passed"]: success = True consecutive_failures = 0 break else: print(f" [GATE] Failed quality gate (correlation {gate_check['correlation']} < threshold)", flush=True) consecutive_failures += 1 if consecutive_failures >= 2: break edit_instruction = build_edit_instruction(target_type, pose_text + " Preserve every background detail exactly.") if consecutive_failures >= 2: print(f" [FALLBACK] {shot_id}: 2 consecutive failures — marking as single-frame-only", flush=True) store.update_shot(shot_id, gate_results={"extraction_mode": "single_frame"}) break if not success or companion_bytes is None: continue # Save the companion frame — versioned base_name = f"shot_{shot_label}_{gen_frame}" existing = sorted(frames_dir.glob(f"{base_name}*.png")) if existing: import re as _re max_take = 0 for ep in existing: m = _re.search(r'_take(\d+)\.png$', ep.name) if m: max_take = max(max_take, int(m.group(1))) else: max_take = max(max_take, 0) next_take = max_take + 1 else: next_take = 1 output_path = frames_dir / f"{base_name}_take{next_take}.png" output_path.write_bytes(companion_bytes) _rel_path = to_serving_path(output_path, pp) gate_update = {gen_frame: _rel_path} store.update_shot( shot_id, gate_results=gate_update, cost_incurred=EDIT_COST, ) print(f" [OK] {shot_id} {gen_frame} extracted via edit — ${EDIT_COST} (correlation: {gate_check['correlation']})", flush=True) return {"shot_id": shot_id} except Exception as e: print(f" [ERR] Background frame extraction for {shot_id}: {e}", flush=True) _tb.print_exc(file=_sys.stderr) _sys.stderr.flush() raise task_id = submit_task(shot_id, "extract", _bg_extract_frames) return JSONResponse({"status": "generating", "shot_id": shot_id, "task_id": task_id}, status_code=202) # ═══════════════════════════════════════════════════════════════════════ # POST /api/coverage-options # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/coverage-options") def coverage_options( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), store=Depends(get_store), ): """Return available coverage types for a shot plus existing coverage.""" shot_id = body.get("shot_id") if not shot_id: return JSONResponse({"error": "Missing shot_id"}, status_code=400) ep_match = re.match(r"EP(\d+)_SH(\d+)", shot_id) if not ep_match: return JSONResponse({"error": f"Invalid shot_id format: {shot_id}"}, status_code=400) ep_num = int(ep_match.group(1)) episode_id = f"EP{ep_num:03d}" plan_path = paths["plans_dir"] / f"ep_{ep_num:03d}_plan.json" if not plan_path.exists(): return JSONResponse({"error": f"Plan not found for {episode_id}"}, status_code=404) try: plan = json.loads(plan_path.read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError) as exc: return JSONResponse({"error": f"Failed to load plan: {exc}"}, status_code=500) shot_data = next((s for s in plan.get("shots", []) if s.get("shot_id") == shot_id), None) if shot_data is None: return JSONResponse({"error": f"Shot not found in plan: {shot_id}"}, status_code=404) from orchestrator.scene_planner import get_coverage_options options = get_coverage_options(shot_data) existing_coverage = [] all_shots = store.get_shots_by_episode(episode_id, include_coverage=True) existing_coverage = [ { "shot_id": s["shot_id"], "coverage_type": s.get("prompt_data", {}).get("shot_type", ""), "status": s["status"], } for s in all_shots if s.get("is_coverage") and s.get("coverage_of") == shot_id ] current_type = shot_data.get("prompt_data", {}).get("shot_type", "") return JSONResponse({ "shot_id": shot_id, "current_type": current_type, "options": options, "existing_coverage": existing_coverage, }) # ═══════════════════════════════════════════════════════════════════════ # POST /api/add-coverage # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/add-coverage") def add_coverage( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), store=Depends(get_store), ): """Add a single coverage shot for a given angle (manual director pick).""" shot_id = body.get("shot_id") coverage_type = body.get("coverage_type") if not shot_id or not coverage_type: return JSONResponse({"error": "Missing shot_id or coverage_type"}, status_code=400) ep_match = re.match(r"EP(\d+)_SH(\d+)", shot_id) if not ep_match: return JSONResponse({"error": f"Invalid shot_id format: {shot_id}"}, status_code=400) ep_num = int(ep_match.group(1)) episode_id = f"EP{ep_num:03d}" plan_path = paths["plans_dir"] / f"ep_{ep_num:03d}_plan.json" if not plan_path.exists(): return JSONResponse({"error": f"Plan not found for {episode_id}"}, status_code=404) try: plan = json.loads(plan_path.read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError) as exc: return JSONResponse({"error": f"Failed to load plan: {exc}"}, status_code=500) shot_data = next((s for s in plan.get("shots", []) if s.get("shot_id") == shot_id), None) if shot_data is None: return JSONResponse({"error": f"Shot not found in plan: {shot_id}"}, status_code=404) all_shots = store.get_shots_by_episode(episode_id, include_coverage=True) existing_cov = [s for s in all_shots if s.get("is_coverage") and s.get("coverage_of") == shot_id] coverage_num = len(existing_cov) + 1 from orchestrator.scene_planner import generate_single_coverage cov_shot = generate_single_coverage(shot_data, coverage_type, coverage_num, episode_id) cov_id = cov_shot["shot_id"] store.insert_shot({ "shot_id": cov_id, "episode_id": episode_id, "pipeline": cov_shot.get("routing_data", {}).get("pipeline", "still"), "status": "previs_pending", "is_coverage": True, "coverage_of": shot_id, }) plan_shot_ids = {s.get("shot_id") for s in plan.get("shots", [])} if cov_id not in plan_shot_ids: plan.setdefault("shots", []).append(cov_shot) try: plan_path.write_text(json.dumps(plan, indent=2, ensure_ascii=False), encoding="utf-8") except IOError as exc: print(f" [WARN] Failed to write coverage shot to plan: {exc}") # Auto-trigger previz generation in background # Note: In the original this called self._api_generate_previz which sends HTTP response. # Here we just submit the generation task directly. def _bg_previz(): try: # Re-create the generation flow inline (simplified — calls the same Flash tools) from tools.generate_previs import _generate_flash_frame, PREVIS_COST from recoil.pipeline._lib.previz_context import build_previz_context from recoil.pipeline._lib.prompt_engine import build_previs_prompt pp = paths _cov_shot_data = cov_shot bible = None if paths["bible_path"].exists(): try: bible = json.loads(paths["bible_path"].read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError): pass context_parts = None try: all_plan_shots = plan.get("shots", []) context_parts = build_previz_context( shot=_cov_shot_data, all_shots=all_plan_shots, bible=bible, episode=ep_num, project=project, ) except Exception: pass if context_parts: gen_result = _generate_flash_frame(context_parts=context_parts) else: cov_prompt = build_previs_prompt(_cov_shot_data, bible=bible) gen_result = _generate_flash_frame(prompt=cov_prompt) if gen_result.get("success"): previs_dir = paths["previs_dir"] / f"ep_{ep_num:03d}" previs_dir.mkdir(parents=True, exist_ok=True) cov_label = cov_id.replace("EP", "").replace("_SH", "_").replace("_COV_", "_cov") output_path = previs_dir / f"shot_{cov_label}_take1.png" output_path.write_bytes(gen_result["image_data"]) _rel_path = to_serving_path(output_path, pp) store.update_shot(cov_id, status="previs_generated", output_path=_rel_path, append_take={"take_id": "take_1", "take_num": 1, "file_path": _rel_path, "cost": PREVIS_COST, "generated_at": time.time()}, cost_incurred=PREVIS_COST) print(f" [OK] Coverage {cov_id} previz generated — ${PREVIS_COST}") except Exception as exc: print(f" [ERR] Background previz for coverage {cov_id}: {exc}") submit_task(cov_id, "previz", _bg_previz) return JSONResponse({ "shot_id": cov_id, "coverage_of": shot_id, "coverage_type": coverage_type, "coverage_num": coverage_num, "status": "generating", }) # ═══════════════════════════════════════════════════════════════════════ # POST /api/generate-coverage (multi-prompt I2V coverage) # Maps to _api_generate_coverage_multi in review_server.py (line 4782). # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/generate-coverage") def generate_coverage( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), store=Depends(get_store), ): """Generate coverage variants (WS/MS/CU) via multi-prompt I2V.""" shot_id = body.get("shot_id") if not shot_id: return JSONResponse({"error": "Missing shot_id"}, status_code=400) prompts = body.get("prompts", []) if not prompts or len(prompts) < 2: return JSONResponse({"error": "Need at least 2 prompts for coverage"}, status_code=400) video_model = body.get("model", "kling-v3") video_model = LEGACY_MODEL_MAP.get(video_model, video_model) shot = store.get_shot(shot_id) if shot is None: return JSONResponse({"error": f"Shot not found: {shot_id}"}, status_code=404) force = body.get("force", False) allowed_for_render = ("previs_approved", "keyframe_approved", "video_pending", "video_complete") if not force and shot["status"] not in allowed_for_render: print(f" [WARN] Coverage from non-standard status '{shot['status']}' for {shot_id} — proceeding") gate = shot.get("gate_results", {}) first_frame = gate.get("first_frame") hero_frame = gate.get("hero_frame") start_frame_path = _safe_resolve_frame(hero_frame, paths) if not start_frame_path: start_frame_path = _safe_resolve_frame(first_frame, paths) if not start_frame_path: return JSONResponse({"error": "No frame images found for coverage generation"}, status_code=400) gen_key = f"coverage_{shot_id}" if not gen_tracker.try_start(gen_key): return JSONResponse({"error": "Coverage generation already in progress for this shot"}, status_code=409) # Build multi_prompt_sequence from prompts multi_prompt_sequence = [] for i, p in enumerate(prompts): multi_prompt_sequence.append({ "index": i + 1, "prompt": p.get("text", ""), "duration": p.get("duration", 5), }) # Build batch: same shot dict repeated N times with _api_duration set batch = [] for p in prompts: shot_copy = dict(shot) shot_copy["_api_duration"] = p.get("duration", 5) batch.append(shot_copy) ep_match = re.match(r"EP(\d+)_SH(\d+)([A-Za-z]?)", shot_id) ep_num = int(ep_match.group(1)) if ep_match else 1 # ── Background generation ── _prompts = prompts _batch = batch _sequence = multi_prompt_sequence _start_path = start_frame_path _video_model = video_model def _run_coverage(): try: from recoil.execution.step_runner import StepRunner from recoil.execution.step_types import ProjectPaths paths_obj = ProjectPaths.for_episode(project, ep_num) runner = StepRunner(store=store, paths=paths_obj, episode=ep_num) results = runner.execute_multi_shot( batch=_batch, multi_prompt_sequence=_sequence, model=_video_model, start_frame=_start_path, aspect_ratio=get_project_aspect_ratio(project), ) # Tag takes with framing metadata for i, result in enumerate(results): if result.success and i < len(_prompts): framing = _prompts[i].get("framing", "") try: updated_shot = store.get_shot(shot_id) if updated_shot: takes = updated_shot.get("takes", []) if takes: take = takes[-1] take["framing"] = framing take["is_coverage"] = True store.update_shot(shot_id, takes=takes) except Exception: pass ok_count = sum(1 for r in results if r.success) total_cost = sum(r.cost_usd for r in results) print(f" [OK] Coverage for {shot_id}: {ok_count}/{len(results)} variants — ${total_cost:.3f}") return {"shot_id": shot_id, "ok_count": ok_count} except Exception as e: print(f" [ERR] Background coverage generate for {shot_id}: {e}") import traceback traceback.print_exc() try: store.update_shot( shot_id, status="video_failed", error_message=str(e), ) except Exception: pass raise finally: gen_tracker.finish(gen_key) task_id = submit_task(shot_id, "coverage", _run_coverage) return JSONResponse({ "shot_id": shot_id, "status": "coverage_submitted", "task_id": task_id, "message": f"Submitting {len(prompts)} coverage variants for {shot_id}...", "model": video_model, "prompts": len(prompts), "start_frame": str(start_frame_path), }) # ═══════════════════════════════════════════════════════════════════════ # GET /api/coverage-prompts/{shot_id} # ═══════════════════════════════════════════════════════════════════════ @router.get("/api/coverage-prompts/{shot_id}") def coverage_prompts( shot_id: str, project: str = Depends(get_project), paths: dict = Depends(get_paths), store=Depends(get_store), ): """Return enriched coverage prompts for a shot.""" shot = store.get_shot(shot_id) if shot is None: return JSONResponse({"error": f"Shot not found: {shot_id}"}, status_code=404) ep_match = re.match(r"EP(\d+)", shot_id) ep_num = int(ep_match.group(1)) if ep_match else 1 plan_path = paths["plans_dir"] / f"ep_{ep_num:03d}_plan.json" plan_shot = shot # fallback if plan_path.exists(): try: plan = json.loads(plan_path.read_text(encoding="utf-8")) found = next((s for s in plan.get("shots", []) if s.get("shot_id") == shot_id), None) if found: plan_shot = found except (json.JSONDecodeError, IOError): pass bible = {} bible_path = paths.get("bible_path") if bible_path and bible_path.exists(): try: bible = json.loads(bible_path.read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError): pass from recoil.pipeline._lib.prompt_engine import build_coverage_prompts prompts = build_coverage_prompts(plan_shot, bible=bible) return JSONResponse({"shot_id": shot_id, "prompts": prompts}) # ═══════════════════════════════════════════════════════════════════════ # POST /api/generate-sequence # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/generate-sequence") def generate_sequence( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), store=Depends(get_store), ): """Generate a multi-shot sequence across multiple shots via multi-prompt I2V.""" shot_ids = body.get("shot_ids", []) if not shot_ids or len(shot_ids) < 2: return JSONResponse({"error": "Need at least 2 shot_ids for a sequence"}, status_code=400) video_model = body.get("model", "kling-v3") video_model = LEGACY_MODEL_MAP.get(video_model, video_model) use_elements = body.get("use_elements", False) element_char_ids = body.get("element_char_ids", []) # Validate all shot_ids exist shots_data = [] for sid in shot_ids: shot = store.get_shot(sid) if shot is None: return JSONResponse({"error": f"Shot not found: {sid}"}, status_code=404) shots_data.append(shot) # Get start frame from first shot's hero_frame first_shot = shots_data[0] gate = first_shot.get("gate_results", {}) first_frame = gate.get("first_frame") hero_frame = gate.get("hero_frame") start_frame_path = _safe_resolve_frame(hero_frame, paths) if not start_frame_path: start_frame_path = _safe_resolve_frame(first_frame, paths) # Extract episode for plan loading ep_match = re.match(r"EP(\d+)", shot_ids[0]) ep_num = int(ep_match.group(1)) if ep_match else 1 # Load plan to get shot data for prompt building plan_path = paths["plans_dir"] / f"ep_{ep_num:03d}_plan.json" plan_shots = [] if plan_path.exists(): try: plan = json.loads(plan_path.read_text(encoding="utf-8")) plan_shots_map = {s.get("shot_id"): s for s in plan.get("shots", [])} for sid in shot_ids: ps = plan_shots_map.get(sid) if ps: ps["_api_duration"] = ps.get("duration", 5) plan_shots.append(ps) else: plan_shots.append({ "shot_id": sid, "_api_duration": 5, "prompt_data": {"prompt_skeleton": {"action_line": "smooth cinematic camera movement"}}, }) except (json.JSONDecodeError, IOError): pass if not plan_shots: return JSONResponse({"error": "Could not load shot data from plan"}, status_code=400) # Build elements payload first (characters + location) so we know # element count for @Element injection into prompts elements_payload = None has_location_element = False total_elements = 0 if use_elements and element_char_ids: try: from recoil.pipeline._lib.elements import ElementManager, extract_batch_location batch_location_id = extract_batch_location(plan_shots) elements_payload, has_location_element, total_elements = ( ElementManager.build_elements_with_info( element_char_ids, project, location_id=batch_location_id, ) ) except Exception as e: print(f" [WARN] Elements build failed: {e}") elements_payload = None # Extract UI prompt/duration overrides from sequence_data sequence_data = body.get("sequence_data", []) prompt_overrides = {item["shot_id"]: item["prompt"] for item in sequence_data if "shot_id" in item and "prompt" in item} duration_overrides = {item["shot_id"]: item["duration"] for item in sequence_data if "shot_id" in item and "duration" in item} # Build multi_prompt_sequence from recoil.pipeline._lib.prompt_engine import build_multi_prompt_sequence multi_prompt_sequence = build_multi_prompt_sequence( plan_shots, batch_char_ids=sorted(element_char_ids) if use_elements and element_char_ids else None, has_location_element=has_location_element, total_elements=total_elements, prompt_overrides=prompt_overrides if prompt_overrides else None, duration_overrides=duration_overrides if duration_overrides else None, ) # Duplicate check gen_key = f"sequence_{'_'.join(shot_ids[:3])}" if not gen_tracker.try_start(gen_key): return JSONResponse({"error": "Sequence generation already in progress"}, status_code=409) # ── Background generation ── _plan_shots = plan_shots _sequence = multi_prompt_sequence _start_path = start_frame_path _video_model = video_model _elements_payload = elements_payload def _run_sequence(): try: from recoil.execution.step_runner import StepRunner from recoil.execution.step_types import ProjectPaths paths_obj = ProjectPaths.for_episode(project, ep_num) runner = StepRunner(store=store, paths=paths_obj, episode=ep_num) results = runner.execute_multi_shot( batch=_plan_shots, multi_prompt_sequence=_sequence, model=_video_model, start_frame=_start_path, aspect_ratio=get_project_aspect_ratio(project), elements_payload=_elements_payload, ) ok_count = sum(1 for r in results if r.success) total_cost = sum(r.cost_usd for r in results) print(f" [OK] Sequence {shot_ids[0]}..{shot_ids[-1]}: {ok_count}/{len(results)} shots — ${total_cost:.3f}") return {"shot_ids": shot_ids, "ok_count": ok_count} except Exception as e: print(f" [ERR] Background sequence generate for {shot_ids}: {e}") import traceback traceback.print_exc() for sid in shot_ids: try: store.update_shot( sid, status="video_failed", error_message=str(e), ) except Exception: pass raise finally: gen_tracker.finish(gen_key) task_id = submit_task(shot_ids[0], "sequence", _run_sequence) return JSONResponse({ "shot_ids": shot_ids, "status": "sequence_submitted", "task_id": task_id, "message": f"Submitting {len(shot_ids)}-shot sequence...", "model": video_model, "use_elements": use_elements, "start_frame": str(start_frame_path) if start_frame_path else None, }) # ═══════════════════════════════════════════════════════════════════════ # POST /api/promote-coverage # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/promote-coverage") def promote_coverage( body: dict = Body(default={}), project: str = Depends(get_project), store=Depends(get_store), ): """Promote a coverage shot's hero frame to replace the primary shot's hero.""" coverage_shot_id = body.get("coverage_shot_id") if not coverage_shot_id: return JSONResponse({"error": "Missing coverage_shot_id"}, status_code=400) cov_shot = store.get_shot(coverage_shot_id) if cov_shot is None: return JSONResponse({"error": f"Coverage shot not found: {coverage_shot_id}"}, status_code=404) if not cov_shot.get("is_coverage"): return JSONResponse({"error": f"Shot is not a coverage shot: {coverage_shot_id}"}, status_code=400) primary_shot_id = cov_shot.get("coverage_of") if not primary_shot_id: return JSONResponse({"error": f"Coverage shot has no primary reference: {coverage_shot_id}"}, status_code=400) primary_shot = store.get_shot(primary_shot_id) if primary_shot is None: return JSONResponse({"error": f"Primary shot not found: {primary_shot_id}"}, status_code=404) cov_gate = cov_shot.get("gate_results", {}) hero_frame = cov_gate.get("hero_frame") if not hero_frame: return JSONResponse({"error": f"Coverage shot has no hero frame: {coverage_shot_id}"}, status_code=400) primary_gate = primary_shot.get("gate_results", {}) primary_gate["hero_frame"] = hero_frame store.update_shot(primary_shot_id, gate_results=primary_gate) return JSONResponse({ "ok": True, "primary_shot_id": primary_shot_id, "coverage_shot_id": coverage_shot_id, "hero_frame": hero_frame, }) # ═══════════════════════════════════════════════════════════════════════ # POST /api/assign-video-frames # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/assign-video-frames") def assign_video_frames( body: dict = Body(default={}), project: str = Depends(get_project), store=Depends(get_store), ): """Reassign which extracted frames serve as start/end for video.""" shot_id = body.get("shot_id") start_from = body.get("start_from") end_from = body.get("end_from") if not shot_id or not start_from: return JSONResponse({"error": "Missing shot_id or start_from"}, status_code=400) valid_slots = ("first_frame", "hero_frame", "last_frame") if start_from not in valid_slots: return JSONResponse({"error": f"Invalid start_from: {start_from}"}, status_code=400) if end_from and end_from not in valid_slots: return JSONResponse({"error": f"Invalid end_from: {end_from}"}, status_code=400) shot = store.get_shot(shot_id) if shot is None: return JSONResponse({"error": f"Shot not found: {shot_id}"}, status_code=404) if shot["status"] != "keyframe_approved": return JSONResponse({"error": f"Need keyframe_approved status, got '{shot['status']}'"}, status_code=400) gate = shot.get("gate_results", {}) source_path = gate.get(start_from) if not source_path: return JSONResponse({"error": f"No image at {start_from}"}, status_code=400) gate_update = dict(gate) gate_update["first_frame"] = source_path if end_from: end_path = gate.get(end_from) if not end_path: return JSONResponse({"error": f"No image at {end_from}"}, status_code=400) gate_update["last_frame"] = end_path else: gate_update.pop("last_frame", None) store.update_shot(shot_id, gate_results=gate_update) return JSONResponse({ "shot_id": shot_id, "first_frame": gate_update.get("first_frame"), "last_frame": gate_update.get("last_frame"), }) # ═══════════════════════════════════════════════════════════════════════ # POST /api/confirm-frame-pair # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/confirm-frame-pair") def confirm_frame_pair( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), store=Depends(get_store), ): """Confirm the first/last frame pair and transition to video_pending.""" shot_id = body.get("shot_id") if not shot_id: return JSONResponse({"error": "Missing shot_id"}, status_code=400) shot = store.get_shot(shot_id) if shot is None: return JSONResponse({"error": f"Shot not found: {shot_id}"}, status_code=404) force = body.get("force", False) if not force and shot["status"] not in ("keyframe_approved", "video_pending"): print(f" [WARN] Confirm pair from non-standard status '{shot['status']}' for {shot_id} — proceeding") gate = shot.get("gate_results", {}) anchor_role = gate.get("anchor_role", "first_frame") frame_position = gate.get("frame_position", "first") if anchor_role == "still_only": store.update_shot(shot_id, status="video_pending") return JSONResponse({ "shot_id": shot_id, "status": "video_pending", "anchor_role": "still_only", }) first_frame = gate.get("first_frame") hero_frame = gate.get("hero_frame") last_frame = gate.get("last_frame") takes = shot.get("takes", []) kf_takes = [t for t in takes if t.get("layer") == "keyframe"] kf_path = kf_takes[-1].get("file_path", "") if kf_takes else "" if anchor_role == "first_frame" and not first_frame: first_frame = kf_path elif anchor_role == "last_frame" and not last_frame: last_frame = kf_path elif anchor_role == "hero_frame" and not hero_frame: hero_frame = kf_path available = {} if first_frame: available["first_frame"] = first_frame if hero_frame: available["hero_frame"] = hero_frame if last_frame: available["last_frame"] = last_frame anchor_key = {"first_frame": "first_frame", "last_frame": "last_frame", "hero_frame": "hero_frame"}.get(anchor_role) if anchor_key and anchor_key not in available: return JSONResponse( {"error": f"Anchor frame ({anchor_role}) not found in gate_results"}, status_code=400, ) gate_final = { "anchor_role": anchor_role, "frame_position": frame_position, } gate_final.update(available) store.update_shot( shot_id, status="video_pending", gate_results=gate_final, ) response = { "shot_id": shot_id, "status": "video_pending", "anchor_role": anchor_role, "frame_position": frame_position, "frame_count": len(available), } response.update(available) return JSONResponse(response) # ═══════════════════════════════════════════════════════════════════════ # POST /api/generate-video # ═══════════════════════════════════════════════════════════════════════ @router.post("/api/generate-video") def generate_video( body: dict = Body(default={}), project: str = Depends(get_project), paths: dict = Depends(get_paths), store=Depends(get_store), ): """Generate video from confirmed frame pair via Kling or Veo.""" shot_id = body.get("shot_id") if not shot_id: return JSONResponse({"error": "Missing shot_id"}, status_code=400) video_model = body.get("model", "kling-v3") video_model = LEGACY_MODEL_MAP.get(video_model, video_model) user_prompt = body.get("prompt", "") shot = store.get_shot(shot_id) if shot is None: return JSONResponse({"error": f"Shot not found: {shot_id}"}, status_code=404) force = body.get("force", False) allowed_for_render = ("previs_approved", "keyframe_approved", "video_pending", "video_complete") if not force and shot["status"] not in allowed_for_render: print(f" [WARN] Video from non-standard status '{shot['status']}' for {shot_id} — proceeding") if shot["status"] in ("previs_approved", "keyframe_approved", "video_complete"): store.update_shot(shot_id, status="video_pending") gate = shot.get("gate_results", {}) first_frame = gate.get("first_frame") last_frame = gate.get("last_frame") hero_frame = gate.get("hero_frame") # Need at least one frame start_frame_path = _safe_resolve_frame(first_frame, paths) end_frame_path = _safe_resolve_frame(last_frame, paths) if not start_frame_path: start_frame_path = _safe_resolve_frame(hero_frame, paths) if not start_frame_path: return JSONResponse({"error": "No frame images found for video generation"}, status_code=400) # Check API availability for chosen model use_veo = video_model.startswith("veo") if use_veo: try: from recoil.execution.api_client import GoogleGenaiClient client = GoogleGenaiClient() if not client.is_available(): return JSONResponse( {"error": "Gemini API not configured. Set GEMINI_API_KEY environment variable."}, status_code=503, ) except Exception as e: return JSONResponse({"error": f"Veo client not available: {e}"}, status_code=503) elif video_model in ("kling-v3-direct", "kling-o3-direct"): try: from recoil.execution.api_client import KlingClient client = KlingClient() if not client.is_available(): return JSONResponse( {"error": "Kling API not configured. Set KLING_ACCESS_KEY and KLING_SECRET_KEY environment variables."}, status_code=503, ) except Exception as e: return JSONResponse({"error": f"Kling client not available: {e}"}, status_code=503) else: try: from recoil.execution.api_client import FalAiKlingClient client = FalAiKlingClient() if not client.is_available(): return JSONResponse( {"error": "fal.ai API not configured. Set FAL_KEY environment variable."}, status_code=503, ) except Exception as e: return JSONResponse({"error": f"fal.ai Kling client not available: {e}"}, status_code=503) if not gen_tracker.try_start(shot_id): return JSONResponse({"error": "Generation already in progress for this shot"}, status_code=409) store.update_shot(shot_id, status="video_submitted") # ── Background generation ── ep_match = re.match(r"EP(\d+)_SH(\d+)([A-Za-z]?)", shot_id) ep_num = int(ep_match.group(1)) if ep_match else 1 pp = paths _start_path = start_frame_path _end_path = end_frame_path _use_veo = use_veo _video_model = video_model _user_prompt = user_prompt _generate_audio = body.get("generate_audio", False) _task_id_holder = [None] def _on_fal_status(status): tid = _task_id_holder[0] if tid: _broadcast_task_event("task:status", tid, shot_id, "video", status=status) def _run_video(): try: from recoil.execution.step_runner import StepRunner from recoil.execution.step_types import ProjectPaths from recoil.pipeline.core.dispatch import dispatch from recoil.pipeline.core.dispatch_context import DispatchContext paths_obj = ProjectPaths.for_episode(project, ep_num) runner = StepRunner(store=store, paths=paths_obj, episode=ep_num) action_prompt = _user_prompt if not action_prompt: plan_path = pp["plans_dir"] / f"ep_{ep_num:03d}_plan.json" if plan_path.exists(): try: plan = json.loads(plan_path.read_text(encoding="utf-8")) shot_data = next( (s for s in plan.get("shots", []) if s.get("shot_id") == shot_id), None ) if shot_data: skeleton = shot_data.get("prompt_data", {}).get("prompt_skeleton", {}) action_prompt = skeleton.get("action_line", skeleton.get("subject_line", "")) except (json.JSONDecodeError, IOError): pass if not action_prompt: raise RuntimeError( f"no prompt resolvable for shot {shot_id} — refusing paid dispatch" ) ctx = DispatchContext( caller_id="api_routes_generation", step_runner=runner, project=project, episode=ep_num, ) receipt = dispatch( "video_i2v", { "shot_id": shot_id, "prompt": action_prompt, "model": _video_model, "start_frame": _start_path, "end_frame": _end_path, "duration": 5, "aspect_ratio": get_project_aspect_ratio(project), "generate_audio": _generate_audio, "on_status": _on_fal_status, }, context=ctx, ) result = receipt.run_result engine = "Veo" if _use_veo else "Kling" if result.success: cost_usd = read_cost_from_result(result) print(f" [OK] {shot_id} video generated via {engine} — ${cost_usd:.3f}") return {"shot_id": shot_id} else: print(f" [WARN] Video generation failed for {shot_id} via {engine}: {result.error}") raise RuntimeError(result.error or "Unknown video generation error") except Exception as e: print(f" [ERR] Background video generate for {shot_id}: {e}") import traceback traceback.print_exc() try: store.update_shot( shot_id, status="video_failed", error_message=str(e), ) except Exception: pass raise finally: gen_tracker.finish(shot_id) task_id = submit_task(shot_id, "video", _run_video) _task_id_holder[0] = task_id return JSONResponse({"status": "generating", "shot_id": shot_id, "task_id": task_id}, status_code=202) # ── Blocking Pass ───────────────────────────────────────────────────────── @router.post("/api/run-blocking-pass") def run_blocking_pass( body: dict = Body(default={}), project: str = Depends(get_project), ): """Run Stage 2.5 blocking pass for an episode (async). Request body: episode (int, required): Episode number. scene_indices (list[int], optional): Specific scenes to (re-)process. Returns 202 with task_id. Background thread runs the pass. """ episode = body.get("episode") if not episode: return JSONResponse({"error": "Missing episode"}, status_code=400) scene_indices = body.get("scene_indices") def _run_blocking(): from orchestrator.blocking_pass import BlockingPass bp = BlockingPass(project=project) return bp.run(episode_num=episode, scene_indices=scene_indices) task_id = submit_task(f"EP{episode:03d}", "blocking_pass", _run_blocking) return JSONResponse( {"status": "running", "episode": episode, "task_id": task_id}, status_code=202, )