""" coverage_planner.py — Build editorial coverage passes from shot plans. Groups shots into scene-level coverage passes using contiguous-run grouping by (location_id, camera_side, is_env_only). Duration-aware chunking respects the 15s Kling API limit as the binding constraint. Consultation: consultations/starsend/coverage-pass-feedback-loop/SYNTHESIS.md """ import json import re from dataclasses import dataclass, field from pathlib import Path from typing import Optional from recoil.core.model_profiles import get_segment_duration_bounds from recoil.pipeline._lib.derivation_sha import shotset_hash from recoil.pipeline._lib.grouping import _coverage_shot_ids # ── Shot type ordering (widest → tightest) ── SHOT_TYPE_ORDER = { "EWS": 0, "WS": 1, "EST": 2, "LS": 3, "MS": 4, "MCU": 5, "CU": 6, "ECU": 7, "INSERT": 8, "OTS": 4, "POV": 5, "TWO-SHOT": 4, } # ── Constraints ── MAX_DURATION_S = 15 MAX_SEGMENTS = 6 MAX_CHARS_I2V = 2 MAX_CHARS_T2V = 3 # ── Shot list helpers ── _SHOT_TOKEN_FROM_ID_RE = re.compile(r"(?:EP\d+_)?SH0*(\d{1,4}[a-zA-Z]?)") def _sanitize_semantic_tag_component(s: str, fallback: str = "") -> str: """Uppercase + strip non-[A-Z0-9_] + collapse underscores. Returns `fallback` if the cleaned result is empty. """ s = s.upper() s = re.sub(r"[^A-Z0-9_]", "_", s) s = re.sub(r"_+", "_", s).strip("_") return s or fallback def shot_list_from_ids(segment_shot_ids: list[str]) -> str: """Convert ["EP001_SH33", "EP001_SH33A", "EP001_SH34"] -> "33_33a_34". Strips optional EP{N}_SH prefix and leading zeroes from the numeric portion, lowercases any alpha suffix. Used to build pass filenames whose shot_list segment is a stable, human-readable sequence of shot tokens. Wildcard sentinel IDs ("wildcard" or empty string) are silently skipped. If no real tokens remain after skipping wildcards, returns "0". Raises ValueError if any non-wildcard shot_id is malformed. """ tokens: list[str] = [] for sid in segment_shot_ids: if sid == "wildcard" or not sid: continue m = _SHOT_TOKEN_FROM_ID_RE.search(sid) if not m: raise ValueError(f"Cannot extract shot token from '{sid}'") num_part = m.group(1) digit_match = re.match(r"(\d+)([a-zA-Z]?)", num_part) if not digit_match: raise ValueError(f"Malformed shot token in '{sid}'") digits = str(int(digit_match.group(1))) # strips leading zeroes suffix = digit_match.group(2).lower() tokens.append(digits + suffix) if not tokens: return "0" return "_".join(tokens) # Phase D — MF-4: promoted from private to public. # Cross-package callers (workspace/server.py, pipeline/tools/migrate_pass_names.py, # tests/test_pass_naming.py) reached across the package boundary via the # underscore name. Promotion makes the contract explicit. The underscore alias # is preserved for one cycle so legacy in-module + intra-package call sites # continue to work; new callers should import `shot_list_from_ids`. _shot_list_from_ids = shot_list_from_ids # ── Coverage strategy config loader ── def _load_coverage_config() -> dict: """Load coverage_strategy from recoil/config/pipeline_config.json.""" config_path = ( Path(__file__).parent.parent.parent / "config" / "pipeline_config.json" ) if not config_path.exists(): return {} with open(config_path) as f: cfg = json.load(f) return cfg.get("coverage_strategy", {}) _COVERAGE_CONFIG = _load_coverage_config() _MOTION_PRESETS = _COVERAGE_CONFIG.get("motion_presets", {}) _MOTION_TIER_OVERRIDE = _COVERAGE_CONFIG.get("motion_tier_override", {}) # ── Dataclasses ── @dataclass class CoverageSegment: segment_index: int source_shot_id: str shot_type: str duration_s: ( int # clamped to per-model bounds via get_segment_duration_bounds(model) ) prompt: str transition: str = "smooth" source_text: str = "" # original script line for discoverability motion_preset: dict = field( default_factory=dict ) # {"movement": str, "intensity": float} is_wildcard: bool = False @dataclass class CoveragePass: pass_id: str episode_id: str shot_range: tuple[str, str] # (first_shot_id, last_shot_id) camera_side: str label: str focus_character: str pass_type: str # "character" or "env" location_id: str = "" segments: list[CoverageSegment] = field(default_factory=list) element_config: dict = field(default_factory=dict) generation_config: dict = field(default_factory=dict) status: str = "draft" # New fields — coverage strategy blueprint_image_path: Optional[str] = None blueprint_source: str = "" # "hero", "previz_establishing", "none" format_type: str = "B" # "A", "B", "C" takes_count: int = 1 arc_preamble: str = "" # Scene-level emotional direction (Tier 2+ only) wildcard_enabled: bool = False format_a_angle_index: int = ( -1 ) # Which angle variant this pass represents (-1 = not Format A) @property def duration_s(self) -> int: return sum(s.duration_s for s in self.segments) @property def character_count(self) -> int: return len(self.element_config.get("character_elements", [])) @classmethod def from_dict(cls, d: dict, episode_id_fallback: str = "") -> "CoveragePass": """Inverse of to_dict() — build a CoveragePass from a plain dict.""" segments = [ CoverageSegment( segment_index=s.get("segment_index", 0), source_shot_id=s.get("source_shot_id", ""), shot_type=s.get("shot_type", "MS"), duration_s=s.get("duration_s", 5), prompt=s.get("prompt", ""), transition=s.get("transition", "smooth"), source_text=s.get("source_text", ""), motion_preset=s.get("motion_preset", {}), is_wildcard=s.get("is_wildcard", False), ) for s in d.get("segments", []) ] shot_range = d.get("shot_range", ["", ""]) return cls( pass_id=d.get("pass_id", ""), episode_id=d.get("episode_id", episode_id_fallback), shot_range=(shot_range[0], shot_range[1]) if len(shot_range) >= 2 else ("", ""), camera_side=d.get("camera_side", ""), label=d.get("label", ""), focus_character=d.get("focus_character", ""), pass_type=d.get("pass_type", "character"), location_id=d.get("location_id", ""), segments=segments, element_config=d.get("element_config", {}), generation_config=d.get("generation_config", {}), status=d.get("status", "locked"), blueprint_image_path=d.get("blueprint_image_path"), blueprint_source=d.get("blueprint_source", ""), format_type=d.get("format_type", "B"), takes_count=d.get("takes_count", 1), arc_preamble=d.get("arc_preamble", ""), wildcard_enabled=d.get("wildcard_enabled", False), format_a_angle_index=d.get("format_a_angle_index", -1), ) def to_dict(self) -> dict: return { "pass_id": self.pass_id, "episode_id": self.episode_id, "shot_range": list(self.shot_range), "camera_side": self.camera_side, "label": self.label, "focus_character": self.focus_character, "pass_type": self.pass_type, "location_id": self.location_id, "duration_s": self.duration_s, "character_count": self.character_count, "shotset_hash": shotset_hash(_coverage_shot_ids(self)), "segments": [ { "segment_index": s.segment_index, "source_shot_id": s.source_shot_id, "shot_type": s.shot_type, "duration_s": s.duration_s, "prompt": s.prompt, "transition": s.transition, "source_text": s.source_text, "motion_preset": s.motion_preset, "is_wildcard": s.is_wildcard, } for s in self.segments ], "element_config": self.element_config, "generation_config": self.generation_config, "status": self.status, "blueprint_image_path": self.blueprint_image_path, "blueprint_source": self.blueprint_source, "format_type": self.format_type, "takes_count": self.takes_count, "arc_preamble": self.arc_preamble, "wildcard_enabled": self.wildcard_enabled, "format_a_angle_index": self.format_a_angle_index, } # ── Previz frame resolution ── def resolve_previz_frame(project: str, episode: int, shot_id: str) -> str | None: """Find the previz frame for a shot to use as start_frame_path. Reads v3 prep/ep_NNN/ (where generate_previs.py writes since the 2026-05-30 v3 migration). v1 output/previs/ is dead — audit F4. """ from recoil.core.paths import ProjectPaths base = ProjectPaths.for_project(project).episode_prep_dir(episode) # Keep this derivation identical to generate_previs._extract_shot_label(). match = re.search(r"SH(\d+)([A-Za-z]?)", shot_id) if match: num = int(match.group(1)) suffix = match.group(2).lower() shot_label = f"{num:03d}{suffix}" else: shot_label = "000" # EXACT match only: shot_005.png or shot_005_take.png — never the # suffixed sibling shot_005a* (EP001 has a real SH05/SH05A pair). # Prefer the HIGHEST numbered take; the base file is take 0 (the v3 # writer keeps base for the first generation and appends _takeNNN after). pattern = re.compile(rf"^shot_{re.escape(shot_label)}_take\d+\.png$") candidates = sorted( ( p for p in base.glob(f"shot_{shot_label}_take*.png") if pattern.match(p.name) ), key=lambda p: int(re.search(r"_take(\d+)", p.name).group(1)), reverse=True, ) if candidates: return str(candidates[0]) primary = base / f"shot_{shot_label}.png" return str(primary) if primary.exists() else None def resolve_blueprint( shots: list[dict], project: str, episode: int ) -> tuple[str | None, str]: """Resolve the blueprint image for a scene's coverage passes. Strategy: - 1-character scene: use that character's hero image (strongest identity lock) - 2+ character scene: use the widest previz frame (sterile environment) - Fallback: None (caller handles) Returns: (blueprint_image_path, blueprint_source) """ chars = _extract_characters(shots) if len(chars) == 1: # Single character — canonical v3 hero (audit F4; was v1 output/refs) from recoil.core.paths import ProjectPaths, RefNotFoundError char_id = chars[0].lower() try: hero = ProjectPaths.for_project(project).resolve_hero( "char", char_id, "identity" ) except RefNotFoundError: hero = None if hero is not None and Path(hero).exists(): return (str(hero), "hero") # Multi-character or no hero found — use widest previz frame widest = _widest_first_sort(shots) if widest: previz = resolve_previz_frame(project, episode, widest[0].get("shot_id", "")) if previz: return (previz, "previz_establishing") return (None, "none") def build_arc_preamble(scene_shots: list[dict], tier_map: dict[str, int]) -> str: """Build a deterministic scene-level emotional arc preamble. Returns empty string if max tier in scene is below the configured threshold. """ min_tier = _COVERAGE_CONFIG.get("arc_preamble", {}).get("min_tier", 2) tiers = [tier_map.get(s.get("shot_id", ""), 0) for s in scene_shots] if not tiers or max(tiers) < min_tier: return "" emotions = [ s.get("prompt_data", {}).get("prompt_skeleton", {}).get("emotion_line", "") for s in scene_shots ] # Detect arc shape if tiers[-1] > tiers[0]: arc = "escalating" elif tiers[-1] < tiers[0]: arc = "descending" elif max(tiers) > tiers[0] and max(tiers) > tiers[-1]: arc = "peaks then resolves" else: arc = "sustained" opening = emotions[0] if emotions[0] else "neutral" peak_idx = tiers.index(max(tiers)) peak = emotions[peak_idx] if emotions[peak_idx] else "intense" return f"[SCENE ARC: {arc}. Opens with {opening}, peaks at {peak}.]" def _resolve_motion_preset(shot_type: str, tier: int = 0) -> dict: """Look up the motion preset for a shot type, with tier intensity override.""" preset = dict( _MOTION_PRESETS.get(shot_type, {"movement": "static", "intensity": 0.5}) ) # Tier intensity override if tier >= 3: boost = _MOTION_TIER_OVERRIDE.get("climax_intensity_boost", 0.05) preset["intensity"] = min(1.0, preset["intensity"] + boost) elif tier <= 1: valley_max = _MOTION_TIER_OVERRIDE.get("valley_intensity_max", 0.55) preset["intensity"] = min(preset["intensity"], valley_max) return preset def _build_wildcard_segment( focus_character: str, model: str = "seeddance-2.0" ) -> CoverageSegment: """Build a wildcard segment. Caller is responsible for eligibility checks. Clamps duration to the active model's bounds so a too-short wildcard (e.g. configured 3s against seeddance min=4s) doesn't trigger a validator BLOCK at --lock time. """ wc_config = _COVERAGE_CONFIG.get("wildcard", {}) duration = wc_config.get("duration_s", 3) min_d, max_d = get_segment_duration_bounds(model) duration = round(max(min_d, min(max_d, duration))) char_name = focus_character if focus_character else "The scene" # Minimal prompt: subject + emotion only prompt = f"{char_name}, a moment." return CoverageSegment( segment_index=99, # Will be renumbered by caller source_shot_id="wildcard", shot_type="MS", duration_s=duration, prompt=prompt, source_text="[wildcard]", motion_preset={"movement": "static", "intensity": 0.5}, is_wildcard=True, ) def _assign_format(tier: int, pass_type: str, segment_count: int) -> list[str]: """Determine which formats (B/C) a pass should generate. Format A removed — on-demand via Claude Code instead of pre-generating. Multi-angle coverage is requested explicitly, not pre-generated. """ if pass_type == "env": return ["B"] # Format A removed — on-demand via Claude Code instead of pre-generating # Multi-angle coverage is requested explicitly, not pre-generated if tier >= 3: return ["C"] # Coverage reactions at climax only return ["B"] # ── Widest-first sort ── def _widest_first_sort(shots: list[dict]) -> list[dict]: """Pop the widest shot to position 0, leave rest in original order.""" if len(shots) <= 1: return list(shots) widest_idx = 0 widest_rank = SHOT_TYPE_ORDER.get( shots[0].get("prompt_data", {}).get("shot_type", "MS"), 5 ) for i, s in enumerate(shots[1:], 1): rank = SHOT_TYPE_ORDER.get(s.get("prompt_data", {}).get("shot_type", "MS"), 5) if rank < widest_rank: widest_rank = rank widest_idx = i result = [shots[widest_idx]] result.extend(s for i, s in enumerate(shots) if i != widest_idx) return result # ── Segment builder ── def _build_segment( shot: dict, index: int, model: str = "seeddance-2.0" ) -> CoverageSegment: """Build a CoverageSegment from a plan shot dict.""" skeleton = shot.get("prompt_data", {}).get("prompt_skeleton", {}) st = shot.get("prompt_data", {}).get("shot_type", "MS") prompt_parts = [] if skeleton.get("subject_line"): prompt_parts.append(f"{st}: {skeleton['subject_line']}") if skeleton.get("action_line"): prompt_parts.append(skeleton["action_line"]) if skeleton.get("emotion_line"): prompt_parts.append(f"[{skeleton['emotion_line']}]") prompt = " ".join(prompt_parts) if prompt_parts else shot.get("source_text", "") # Duration: routing_data is authoritative, prompt_data is fallback raw_dur = shot.get("routing_data", {}).get("target_editorial_duration_s") if raw_dur is None: raw_dur = shot.get("prompt_data", {}).get("duration_s", 5) if raw_dur is None: raw_dur = 5 min_d, max_d = get_segment_duration_bounds(model) duration = round(max(min_d, min(max_d, raw_dur))) motion = _resolve_motion_preset(st) return CoverageSegment( segment_index=index, source_shot_id=shot.get("shot_id", ""), shot_type=st, duration_s=duration, prompt=prompt, source_text=shot.get("source_text", ""), motion_preset=motion, ) # ── Character extraction ── def _extract_characters(shots: list[dict]) -> list[str]: """Return deduplicated list of character IDs across shots, ordered by frequency.""" char_counts: dict[str, int] = {} for s in shots: for c in s.get("asset_data", {}).get("characters", []): cid = c.get("char_id", "") if isinstance(c, dict) else str(c) cid = cid.upper() if cid: char_counts[cid] = char_counts.get(cid, 0) + 1 return sorted(char_counts, key=lambda c: char_counts[c], reverse=True) def _detect_focus_character(shots: list[dict]) -> str: """Return the most common character ID across a set of shots.""" chars = _extract_characters(shots) return chars[0] if chars else "" # ── Contiguous run grouping ── def _grouping_key(shot: dict) -> tuple[str, bool]: """Extract (location_id, is_env_only) from a shot. camera_side intentionally excluded — mixed-side batches are valid for multi-shot generation (W4, coverage-pass-architecture consult). """ loc = shot.get("asset_data", {}).get("location_id", "") is_env = shot.get("routing_data", {}).get("is_env_only", False) return (loc, is_env) def _build_contiguous_runs(shots: list[dict]) -> list[list[dict]]: """Group shots into contiguous runs sharing the same grouping key. Adjacent shots with the same (location, is_env) merge into one run. A change in either dimension starts a new run. """ if not shots: return [] runs: list[list[dict]] = [] current_run: list[dict] = [shots[0]] current_key = _grouping_key(shots[0]) for shot in shots[1:]: key = _grouping_key(shot) if key == current_key: current_run.append(shot) else: runs.append(current_run) current_run = [shot] current_key = key runs.append(current_run) return runs # ── Duration-aware chunking ── def _chunk_by_duration( shots: list[dict], max_duration: int = MAX_DURATION_S, max_segments: int = MAX_SEGMENTS, max_chars_i2v: int = MAX_CHARS_I2V, max_chars_t2v: int = MAX_CHARS_T2V, model: str = "seeddance-2.0", ) -> list[list[dict]]: """Split a contiguous run into chunks respecting three constraints. Priority order: 1. Duration (15s max) — the binding constraint 2. Segment count (6 max) — secondary 3. Element budget (2 chars I2V, 3 T2V) — only binding in 3+ char scenes Uses greedy packing: add shots until a constraint would be violated. """ if not shots: return [] chunks: list[list[dict]] = [] current: list[dict] = [] current_dur = 0 min_d, max_d = get_segment_duration_bounds(model) for shot in shots: # Duration: routing_data is authoritative, prompt_data is fallback raw_dur = shot.get("routing_data", {}).get("target_editorial_duration_s") if raw_dur is None: raw_dur = shot.get("prompt_data", {}).get("duration_s", 5) if raw_dur is None: raw_dur = 5 dur = round(max(min_d, min(max_d, raw_dur))) would_exceed_duration = current_dur + dur > max_duration would_exceed_segments = len(current) >= max_segments if would_exceed_duration or would_exceed_segments: if current: chunks.append(current) current = [shot] current_dur = dur else: current.append(shot) current_dur += dur if current: chunks.append(current) # Second pass: split any chunk that exceeds element budget # Use T2V limit (3 chars) as default — mode is unknown at chunk time. # Validator catches I2V overages (>2 chars) at --lock time. char_limit = max_chars_t2v final_chunks: list[list[dict]] = [] for chunk in chunks: chars = _extract_characters(chunk) if len(chars) > char_limit: sub: list[dict] = [] for shot in chunk: test_sub = sub + [shot] test_chars = _extract_characters(test_sub) if len(test_chars) > char_limit and sub: final_chunks.append(sub) sub = [shot] else: sub = test_sub if sub: final_chunks.append(sub) else: final_chunks.append(chunk) return final_chunks # ── Core function ── def build_passes( shots: list[dict], project: str, episode: int, tier_map: dict[str, int] | None = None, wildcard_override: bool | None = None, ) -> list[CoveragePass]: """Group shots into coverage passes with B/C format routing. Algorithm: 1. Build contiguous runs by (location_id, camera_side, is_env_only) 2. Resolve blueprint image per scene (1-char=hero, 2+=sterile env) 3. Build arc preamble per scene 4. Pin widest shot as opener, chunk by duration/segments/elements 5. Assign format (B/C) based on tier 6. Assign takes count based on tier 7. Optionally append wildcard segment """ tier_map = tier_map or {} takes_config = _COVERAGE_CONFIG.get("takes_per_tier", {}) runs = _build_contiguous_runs(shots) # Scene-level data (keyed by location_id) scene_blueprints: dict[str, tuple[str | None, str]] = {} scene_arcs: dict[str, str] = {} passes: list[CoveragePass] = [] pass_counter = 0 for run in runs: loc, is_env = _grouping_key(run[0]) side = run[0].get("spatial_data", {}).get("camera_side", "N") # Resolve blueprint once per location if loc not in scene_blueprints: scene_blueprints[loc] = resolve_blueprint(run, project, episode) bp_path, bp_source = scene_blueprints[loc] # Build arc preamble once per location if loc not in scene_arcs: scene_arcs[loc] = build_arc_preamble(run, tier_map) arc = scene_arcs[loc] # Pin widest shot as opener (reuse helper) run = _widest_first_sort(run) # Determine pass_type for run-level model routing pass_type = "env" if is_env else "character" # Model routing by tier (run-level, using run-wide max tier) run_tiers = [tier_map.get(s.get("shot_id", ""), 0) for s in run] run_max_tier = max(run_tiers) if run_tiers else 0 routing = _COVERAGE_CONFIG.get("model_routing", {}) if is_env: model = routing.get("env_any_tier", "seeddance-2.0") elif run_max_tier >= 3: model = routing.get("climax", "seeddance-2.0") else: model = routing.get("character_default", "seeddance-2.0") chunks = _chunk_by_duration(run, model=model) for chunk in chunks: chunk = _widest_first_sort(chunk) focus_char = _detect_focus_character(chunk) chars = _extract_characters(chunk) # Determine max tier for this chunk chunk_tiers = [tier_map.get(s.get("shot_id", ""), 0) for s in chunk] max_tier = max(chunk_tiers) if chunk_tiers else 0 # Build segments with motion presets segments = [] for i, shot in enumerate(chunk): seg = _build_segment(shot, i, model=model) seg.motion_preset = _resolve_motion_preset(seg.shot_type, max_tier) segments.append(seg) # Determine wildcard eligibility wc_config = _COVERAGE_CONFIG.get("wildcard", {}) wc_enabled = ( wildcard_override if wildcard_override is not None else wc_config.get("enabled_by_default", False) ) wc_min_tier = wc_config.get("min_tier", 3) should_wildcard = ( wc_enabled and max_tier >= wc_min_tier and pass_type != "env" ) # Optionally append wildcard as last segment if should_wildcard: wc_seg = _build_wildcard_segment(focus_char, model=model) if ( wc_seg and (sum(s.duration_s for s in segments) + wc_seg.duration_s) <= MAX_DURATION_S ): wc_seg.segment_index = len(segments) segments.append(wc_seg) # Assign format(s) formats = _assign_format(max_tier, pass_type, len(segments)) takes = takes_config.get(str(max_tier), 1) first_id = segments[0].source_shot_id if segments else "" last_id = segments[-1].source_shot_id if segments else "" start_frame = resolve_previz_frame(project, episode, first_id) has_start_frame = start_frame is not None or bp_path is not None effective_start = bp_path or start_frame cfg_scale = 0.50 if is_env else 0.55 # model resolved at run level (above _chunk_by_duration call) # Mode resolved from the model's own coverage_mode_preferences — # keeps t2v/r2v/i2v out of this file. Swap models, modes follow. from recoil.core.model_profiles import get_coverage_mode mode = get_coverage_mode(model, pass_type, has_start_frame) # Generate passes per format (B and C only) for fmt in formats: pass_counter += 1 suffix = focus_char if focus_char else "ENV" # NEW pass_id format (SYNTHESIS §1): # EP{NNN}_PASS_{CCC}_SH{shot_list}_{semantic_tag} # Where semantic_tag = {side}_{focus_or_ENV} # (format_type no longer in filename — lives on dataclass + sidecar) try: shot_list = _shot_list_from_ids( [seg.source_shot_id for seg in segments] ) except ValueError: shot_list = "UNKNOWN" clean_side = _sanitize_semantic_tag_component(side, fallback="N") clean_suffix = _sanitize_semantic_tag_component(suffix, fallback="ENV") semantic_tag = f"{clean_side}_{clean_suffix}" pass_id = ( f"EP{episode:03d}_PASS_{pass_counter:03d}" f"_SH{shot_list}_{semantic_tag}" ) passes.append( CoveragePass( pass_id=pass_id, episode_id=f"EP{episode:03d}", shot_range=(first_id, last_id), camera_side=side, label=f"{focus_char or 'ENV'} {fmt} ({loc})", focus_character=focus_char, pass_type=pass_type, location_id=loc, segments=list(segments), element_config={ "character_elements": [{"char_id": c} for c in chars], "location_id": loc, }, generation_config={ "model": model, "mode": mode, "aspect_ratio": "9:16", "cfg_scale": cfg_scale, "start_frame_path": effective_start, }, blueprint_image_path=bp_path, blueprint_source=bp_source, format_type=fmt, # still lives on dataclass takes_count=takes, arc_preamble=arc if max_tier >= 2 else "", wildcard_enabled=should_wildcard, format_a_angle_index=-1, ) ) return passes