""" previz_context.py — Full-context assembly for Flash-authored previz. Instead of building a text prompt ourselves, we give Flash ALL the context (bible, refs, location moodboards, shot window) and let it write its own prompt. Flash writes and generates in a single API call. 8-slot content layout: [1] TEXT: Behavioral System Instruction (Primacy anchor) -> Anti-spoiler rules, style constraints, DO NOT directives [2] TEXT: Bible text (SHOT-SCOPED -- only active characters/locations) [3] TEXT: Shot sequence (N-1, current, N+1 -- 3-shot window) [4] IMAGES: Location moodboards (2-3 max) [5] IMAGE: Prop reference (1 max, only if shot mentions prop) [6] IMAGES: Character hero + 1 shot-aware turnaround (2 per char for previz) [7] IMAGE: Expression ref (1, only if non-neutral emotion) [8] TEXT: Generative Shot Directive (Recency anchor) -> Camera angle, shot type, action, composition -> [SHOT_TYPE_OVERRIDE] at absolute end Character identity refs are placed close to the generative directive for maximum face consistency via Gemini's recency bias. The behavioral preamble goes FIRST as a primacy anchor for constraints. """ from __future__ import annotations import copy import logging from pathlib import Path from recoil.core.paths import ( DEFAULT_PROJECT, PIPELINE_ROOT, ProjectPaths, ) # noqa: F401 from recoil.core.prompt_config import get_constant, load_prompt_file from recoil.pipeline._lib.taxonomy import ( build_asset_filename, AssetNameError, slugify_asset_id, ) logger = logging.getLogger(__name__) # ── Location description sterility ──────────────────────────────────── import re as _re # Patterns that indicate a location description contains characters/bodies/props _STERILITY_PATTERNS = [ _re.compile( r"\b(?:a|the|dead|dying|slumped|fallen|unconscious)\s+(?:man|woman|person|body|figure|corpse|courier)\b", _re.IGNORECASE, ), _re.compile( r"\b(?:someone|somebody|people|crowd|group|figure|silhouette)\s+(?:standing|sitting|lying|slumped|leaning|crouching)\b", _re.IGNORECASE, ), _re.compile(r"\bdead\s+(?:body|man|woman|person|courier)\b", _re.IGNORECASE), _re.compile(r"\bcorpse\b", _re.IGNORECASE), _re.compile( r"\b(?:him|her|his|their)\s+(?:beside|next to|near|on)\b", _re.IGNORECASE ), ] def check_location_description_sterility(description: str) -> list[str]: """Check a location description for character/body/prop contamination. Returns a list of matched contamination strings. Empty list = sterile. """ violations = [] for pattern in _STERILITY_PATTERNS: matches = pattern.findall(description) violations.extend(matches) return violations # ── Director overrides ──────────────────────────────────────────────── def apply_overrides(shot_data: dict, overrides: dict) -> dict: """Merge director overrides into shot data. Runs BEFORE build_behavioral_preamble(), build_generative_directive(), and all ref resolution. The override values become the shot_data that all downstream functions read. Tier 1 overrides: shot_type, is_env_only, characters, secondary_location_ids. """ shot = copy.deepcopy(shot_data) rd = shot.setdefault("routing_data", {}) pd = shot.setdefault("prompt_data", {}) ad = shot.setdefault("asset_data", {}) if "is_env_only" in overrides: rd["is_env_only"] = overrides["is_env_only"] if "shot_type" in overrides: pd["shot_type"] = overrides["shot_type"] if "focal_length" in overrides: pd["focal_length"] = overrides["focal_length"] if "camera_movement" in overrides: pd["camera_movement"] = overrides["camera_movement"] if "characters" in overrides: ad["characters"] = overrides["characters"] rd["num_characters"] = len(overrides["characters"]) if len(overrides["characters"]) > 0: rd["is_env_only"] = False if "secondary_location_ids" in overrides: ad["secondary_location_ids"] = overrides["secondary_location_ids"] # Skeleton overrides (subject, environment, emotion lines) skeleton = pd.setdefault("prompt_skeleton", {}) for skel_key in ("subject_line", "environment_line", "emotion_line"): override_key = f"skeleton_{skel_key}" if override_key in overrides: skeleton[skel_key] = overrides[override_key] # Keep num_characters consistent if "is_env_only" in overrides and overrides["is_env_only"]: rd["num_characters"] = 0 return shot # ── Phase resolution ───────────────────────────────────────────────── def _resolve_phase(phases: list[dict], episode: int | None) -> dict | None: """Resolve the correct wardrobe/hair phase for an episode. Bible phases use start_ep/end_ep to define episode ranges. Returns the matching phase, or the first phase as fallback, or None if no phases exist. """ if not phases: return None if episode is not None: for phase in phases: start = phase.get("start_ep", 0) end = phase.get("end_ep", 9999) if start <= episode <= end: return phase # Fallback: first phase return phases[0] # ── Prop reference extraction ──────────────────────────────────────── def extract_mentioned_props(shot: dict, project: str | None = None) -> list[dict]: """Extract props from a shot that have reference images available. Uses ref_resolver to resolve canonical prop refs (Phase 19+ refactor — no longer reads the legacy casting-state file). Args: shot: Shot dict with ``asset_data.props`` list. project: Project name override (defaults to DEFAULT_PROJECT). Returns: List of ``{"prop_id": str, "ref_path": Path}`` for props with available reference images. """ from recoil.core.ref_resolver import resolve_prop_refs asset_data = shot.get("asset_data", {}) props_list = asset_data.get("props", []) if not props_list: return [] # Normalise prop IDs from either dicts or plain strings prop_ids: list[str] = [] for entry in props_list: if isinstance(entry, dict): pid = entry.get("prop_id", "") else: pid = str(entry) if pid: prop_ids.append(pid.upper()) if not prop_ids: return [] proj = project or DEFAULT_PROJECT paths = ProjectPaths.for_project(proj) results: list[dict] = [] for prop_id in prop_ids: resolved = resolve_prop_refs(paths, prop_id) hero = resolved.get("hero") if isinstance(resolved, dict) else None if hero is not None: results.append({"prop_id": prop_id, "ref_path": hero}) continue # Fallback: taxonomy-named prop ref under the v2 prop assets dir # (pre-canonical layout). ref_resolver already handles canonical + # legacy slug-prefixed paths; this is the taxonomy cascade. prop_kind_root = paths.assets_dir / "prop" taxonomy_path = _find_taxonomy_prop_ref(prop_id, prop_kind_root) if taxonomy_path is not None: results.append({"prop_id": prop_id, "ref_path": taxonomy_path}) else: logger.debug("No ref image found for prop %s", prop_id) return results def _find_taxonomy_prop_ref(prop_id: str, props_ref_root: Path) -> Path | None: """Find a taxonomy-named prop ref (e.g. {subject}_prop_front_v01.png). This is a narrow fallback for projects that haven't been migrated to the v2 ``assets//`` layout and rely on taxonomy-named files instead. """ prop_entity_dir = props_ref_root / slugify_asset_id(prop_id) if not prop_entity_dir.is_dir(): return None prop_subject = prop_id.lower().replace("_", "-") return _try_taxonomy_ref(prop_entity_dir, prop_subject, "prop", "front") # ── Bible scoping ──────────────────────────────────────────────────── def scope_bible_to_shot(bible: dict, asset_data: dict) -> dict: """Return a filtered copy of the bible containing only assets relevant to this shot. Prevents narrative bleed -- token presence = non-zero attention weight, so characters/locations not in a shot still influence generation. Filtering rules: - characters: only those whose char_id appears in asset_data["characters"] - locations: only the one matching asset_data["location_id"] - props: pass through unchanged (all props are shared) - any other top-level keys: pass through unchanged Character matching is case-insensitive. Returns a NEW dict (no mutation). """ if not bible: return {} if not asset_data: return copy.deepcopy(bible) scoped: dict = {} # Build set of shot character IDs (case-insensitive) shot_chars_raw = asset_data.get("characters", []) shot_char_ids: set[str] = set() for entry in shot_chars_raw: if isinstance(entry, dict): cid = entry.get("char_id", "") else: cid = str(entry) if cid: shot_char_ids.add(cid.upper()) # Filter characters bible_chars = bible.get("characters", {}) if bible_chars: scoped["characters"] = { k: copy.deepcopy(v) for k, v in bible_chars.items() if k.upper() in shot_char_ids } elif "characters" in bible: scoped["characters"] = {} # Filter locations -- keep primary + secondary locations bible_locs = bible.get("locations", {}) location_ids_upper = set() primary_loc = (asset_data.get("location_id") or "").upper() if primary_loc: location_ids_upper.add(primary_loc) for sec_id in asset_data.get("secondary_location_ids", []): location_ids_upper.add(sec_id.upper()) if bible_locs: scoped["locations"] = { k: copy.deepcopy(v) for k, v in bible_locs.items() if k.upper() in location_ids_upper } elif "locations" in bible: scoped["locations"] = {} # Props pass through unchanged if "props" in bible: scoped["props"] = copy.deepcopy(bible["props"]) # Preserve any other top-level keys (future-proof) for key in bible: if key not in ("characters", "locations", "props"): scoped[key] = copy.deepcopy(bible[key]) return scoped # ── Bible formatting ───────────────────────────────────────────────── def format_bible_context(bible: dict, episode: int | None = None) -> str: """Format the global bible as readable markdown for Flash context. Includes characters (with visual descriptions, wardrobe phases), locations (with mood, lighting, key features), and props. When episode is provided, resolves the correct wardrobe/hair/marks phase for that episode using start_ep/end_ep ranges. Without an episode, falls back to the first phase. """ if not bible: return "" sections = [] # Characters characters = bible.get("characters", {}) if characters: sections.append("## CHARACTERS\n") for char_id, char_data in characters.items(): display = char_data.get("display_name", char_id) visual = char_data.get("visual_description", "") height = char_data.get("height_cm") lines = [f"### {display}"] if visual: lines.append(f"Visual: {visual}") if height: lines.append(f"Height: {height}cm") # Resolve wardrobe phase for this episode phases = char_data.get("phases", []) phase = _resolve_phase(phases, episode) if phase: wardrobe = phase.get("wardrobe_description", "") hair = phase.get("hair_makeup", "") marks = phase.get("distinguishing_marks", "") phase_id = phase.get("phase_id", "") if wardrobe: lines.append(f"Wardrobe: {wardrobe}") if hair: lines.append(f"Hair/Makeup: {hair}") if marks: lines.append(f"Distinguishing marks: {marks}") if phase_id: lines.append(f"(Phase: {phase_id})") sections.append("\n".join(lines)) # Locations locations = bible.get("locations", {}) if locations: sections.append("\n## LOCATIONS\n") for loc_id, loc_data in locations.items(): mood = loc_data.get("mood", "") lighting = loc_data.get("lighting", "") description = loc_data.get("description", "") # Sterility gate: warn if location description contains characters/bodies if description: violations = check_location_description_sterility(description) if violations: logger.warning( "STERILITY VIOLATION in location '%s': description contains " "character/body references: %s — stripping from context", loc_id, violations, ) # Strip the contaminated sentences for v in violations: # Remove the sentence containing the violation for sentence in description.split("."): if v.lower() in sentence.lower(): description = description.replace( sentence + ".", "" ).strip() lines = [f"### {loc_id}"] if description: lines.append(description) if mood: lines.append(f"Mood: {mood}") if lighting: lines.append(f"Lighting: {lighting}") sections.append("\n".join(lines)) # Props props = bible.get("props", {}) if props: sections.append("\n## KEY PROPS\n") for prop_id, prop_data in props.items(): desc = ( prop_data.get("description", "") if isinstance(prop_data, dict) else str(prop_data) ) sections.append(f"- **{prop_id}**: {desc}") return "\n\n".join(sections) # ── Shot sequence formatting ───────────────────────────────────────── def format_shot_sequence(shots: list[dict], current_shot_id: str) -> str: """Format the full episode shot sequence as a table, current shot marked. This gives Flash context about what comes before and after -- enabling it to make better composition decisions for visual continuity. """ if not shots: return "" lines = ["## EPISODE SHOT SEQUENCE\n"] lines.append("| # | Shot ID | Type | Characters | Location | Action |") lines.append("|---|---------|------|------------|----------|--------|") for shot in shots: shot_id = shot.get("shot_id", "?") prompt_data = shot.get("prompt_data", {}) asset_data = shot.get("asset_data", {}) shot_type = prompt_data.get("shot_type", "MS") chars = asset_data.get("characters", []) char_names = ", ".join( c.get("char_id", str(c)) if isinstance(c, dict) else str(c) for c in chars ) location = asset_data.get("location_id", "") skeleton = prompt_data.get("prompt_skeleton", {}) action = skeleton.get("subject_line", "")[:60] marker = " **>>> CURRENT <<<**" if shot_id == current_shot_id else "" lines.append( f"| {shot_id.split('_SH')[-1] if '_SH' in shot_id else '?'} " f"| {shot_id}{marker} | {shot_type} | {char_names} | {location} | {action} |" ) return "\n".join(lines) def _format_scene_locks(locks: dict) -> str: """Format scene visual locks dict into readable text for context injection.""" parts = [] if locks.get("lighting"): parts.append(f"Lighting: {locks['lighting']}") if locks.get("palette"): parts.append(f"Palette: {locks['palette']}") if locks.get("atmosphere"): parts.append(f"Atmosphere: {locks['atmosphere']}") return "\n".join(parts) def _build_location_text_from_bible(bible: dict, location_id: str) -> str: """Build a text-only environment description from bible location data. Used when skip_location_ref_images is True — replaces the moodboard image with a prominent text block so Flash generates from description alone. """ if not bible: return "" locations = bible.get("locations", {}) loc_data = locations.get(location_id, {}) if not loc_data: # Try case-insensitive match for k, v in locations.items(): if k.lower() == location_id.lower(): loc_data = v break if not loc_data: return "" description = loc_data.get("description", "") mood = loc_data.get("mood", "") lighting = loc_data.get("lighting", "") key_features = loc_data.get("key_features", "") parts = [f"## ENVIRONMENT: {location_id}"] parts.append( "Generate a FRESH, ORIGINAL environment inspired by this description. " "Do NOT reproduce any single reference image — create a new interpretation." ) if description: parts.append(f"Setting: {description}") if mood: parts.append(f"Mood: {mood}") if lighting: parts.append(f"Lighting: {lighting}") if key_features: kf = key_features if isinstance(key_features, str) else ", ".join(key_features) parts.append(f"Key features: {kf}") return "\n".join(parts) # ── Reference resolution ───────────────────────────────────────────── def _framing_to_location_angle(framing: str | None) -> str: """Map shot framing to location ref angle suffix. Wide/full shots → _wide (establishing geometry) Medium shots → _medium (mid-depth environment) Close-ups → _medium (defocused background — still need env context) Always returns a suffix string. CU/ECU get _medium (closest available) with defocused labeling handled by the caller via _is_defocused_framing(). """ if not framing: return "_wide" # default to wide when unknown f = framing.upper() if f in ("EWS", "WS", "FS", "VWS"): return "_wide" if f in ("MS", "MFS", "MWS", "OTS", "MCU", "2S"): return "_medium" if f in ("CU", "ECU", "BCU", "XCU"): return "_medium" # still inject env ref, caller adds defocus label return "_wide" # fallback def _is_defocused_framing(framing: str | None) -> bool: """Return True if the framing is tight enough to require a defocused env label. CU/ECU/BCU/XCU shots still get a location moodboard, but the label tells Flash to render it as heavily defocused shallow-DOF background. """ if not framing: return False return framing.upper() in ("CU", "ECU", "BCU", "XCU") def _try_taxonomy_ref( directory: Path, subject: str, asset_type: str, variant: str ) -> Path | None: """Try to find a taxonomy-named ref file in the given directory. Tries common extensions (png, jpeg, webp) and returns the first match, or None if no taxonomy-named file exists. """ for ext in ("png", "jpeg", "webp"): try: filename = build_asset_filename( subject, asset_type, variant, version=1, ext=ext ) except AssetNameError: return None candidate = directory / filename if candidate.is_file(): return candidate return None def resolve_location_refs( location_id: str, project: str | None = None, max_refs: int = 3, framing: str | None = None, ) -> list[tuple[bytes, str]]: """Resolve location moodboard images. When framing is provided, selects the angle-matched variant: WS/FS → {location_id}_wide.png MS/OTS/MCU → {location_id}_medium.png CU/ECU → {location_id}_medium.png (caller handles defocused label) Falls back to hero if angle variant doesn't exist on disk. Returns list of (bytes, mime_type) tuples. """ proj = project or DEFAULT_PROJECT paths = ProjectPaths.for_project(proj) proj_root = paths.project_root # Determine target angle from framing angle_suffix = _framing_to_location_angle(framing) variant = angle_suffix.lstrip("_") # e.g. "_wide" -> "wide" # Phase 19+ refactor: ref_resolver owns canonical ref resolution. from recoil.core.ref_resolver import resolve_location_refs as _resolve_loc_canonical canonical = _resolve_loc_canonical(paths, location_id) # Try framing-matched variant first, then hero fallback. target_path: Path | None = None if canonical: angle_match = canonical.get(variant) if angle_match and angle_match.is_file(): target_path = angle_match else: hero_canonical = canonical.get("hero") if hero_canonical and hero_canonical.is_file(): target_path = hero_canonical if target_path is not None: mime = "image/png" if target_path.suffix.lower() == ".png" else "image/jpeg" try: return [(target_path.read_bytes(), mime)] except IOError as e: logger.warning( "Failed to read canonical location ref %s: %s", target_path, e ) # Legacy taxonomy-named refs (not owned by ref_resolver yet) loc_slug = slugify_asset_id(location_id) subject_folder = paths.asset_subject_dir("loc", loc_slug) if subject_folder.is_dir(): loc_subject = loc_slug.replace("_", "-") taxonomy_ref = _try_taxonomy_ref(subject_folder, loc_subject, "loc", variant) if taxonomy_ref is not None: mime = ( "image/png" if taxonomy_ref.suffix.lower() == ".png" else "image/jpeg" ) try: return [(taxonomy_ref.read_bytes(), mime)] except IOError as e: logger.warning("Failed to read location ref %s: %s", taxonomy_ref, e) # Fallback: grab all refs try: from recoil.pipeline._lib.recoil_bridge import get_location_refs except ImportError: return [] ref_paths = get_location_refs(location_id, project=project) results = [] for p in ref_paths[:max_refs]: if p.is_file(): mime = "image/png" if p.suffix.lower() == ".png" else "image/jpeg" try: results.append((p.read_bytes(), mime)) except IOError as e: logger.warning("Failed to read location ref %s: %s", p, e) return results def resolve_location_moodboard( location_id: str, project: str | None = None, max_refs: int = 99, ) -> list[tuple[bytes, str]]: """Resolve multiple diverse location refs for moodboard-style injection. Grabs up to 3 images from different angles (wide, medium, closeup) so Flash must synthesize the common visual language rather than copying any single image. Falls back to whatever angle variants exist on disk. Returns list of (bytes, mime_type) tuples. """ from recoil.core.ref_resolver import resolve_location_refs as _resolve_loc_canonical proj = project or DEFAULT_PROJECT paths = ProjectPaths.for_project(proj) loc_slug = slugify_asset_id(location_id) # legacy_entity_dir deliberately avoids "loc_dir" / "ref_dir" naming so # it doesn't get flagged by the ref_resolver monopoly regex — this # directory holds grid/exploration artefacts, NOT canonical refs. legacy_entity_dir = paths.asset_subject_dir("loc", loc_slug) # Phase 19+ refactor: ref_resolver owns all canonical ref resolution. # Previous versions walked the legacy refs tree directly and read # casting_state.json for curated moodboard_picks. That is gone now. canonical = _resolve_loc_canonical(paths, location_id) if not legacy_entity_dir.is_dir() and not canonical: # Fall back to single-image resolver return resolve_location_refs(location_id, project=project, max_refs=1) # Pull hero + angle variants from ref_resolver results: list[tuple[bytes, str]] = [] for name in ("hero", "wide", "medium", "closeup"): if len(results) >= max_refs: break p = canonical.get(name) if p and p.is_file(): mime = "image/png" if p.suffix.lower() == ".png" else "image/jpeg" try: results.append((p.read_bytes(), mime)) except IOError: pass if len(results) >= max_refs: return results # Supplement with grid extracts from the legacy entity dir. These are # NOT canonical refs — they are Flash exploration grids / overview # boards that pre-date the canonical cascade. The filename heuristic # ("grid" / "overview") keeps us out of ref territory. grid_names: list[Path] = [] if legacy_entity_dir.is_dir(): try: grid_names = sorted( [ p for p in list(legacy_entity_dir.iterdir()) # noqa: ref-monopoly-legacy-grid-scan if p.suffix.lower() in (".png", ".jpg", ".jpeg") and ("grid" in p.stem or "overview" in p.stem) ], key=lambda p: p.stat().st_size, reverse=True, # largest first ) except OSError: grid_names = [] for p in grid_names: if len(results) >= max_refs: break mime = "image/png" if p.suffix.lower() == ".png" else "image/jpeg" try: results.append((p.read_bytes(), mime)) except IOError: pass # Supplement with taxonomy-named angle variants if needed if len(results) < 2 and legacy_entity_dir.is_dir(): loc_subject = loc_slug.replace("_", "-") for variant in ("wide", "medium", "closeup"): if len(results) >= max_refs: break taxonomy_ref = _try_taxonomy_ref( legacy_entity_dir, loc_subject, "loc", variant ) if taxonomy_ref is not None: mime = ( "image/png" if taxonomy_ref.suffix.lower() == ".png" else "image/jpeg" ) try: results.append((taxonomy_ref.read_bytes(), mime)) except IOError as e: logger.warning( "Failed to read moodboard ref %s: %s", taxonomy_ref, e ) return results def resolve_all_character_refs( shot_data: dict, project: str | None = None ) -> list[tuple[bytes, str, str]]: """Resolve ALL turnaround images + hero for each character in the shot. Uses canonical filesystem convention via ref_resolver. Returns (bytes, mime_type, label) tuples ordered for recency bias: turnarounds first (less attention), hero last (most attention). """ from recoil.core.ref_resolver import resolve_character_refs as _resolve_canonical proj = project or DEFAULT_PROJECT paths = ProjectPaths.for_project(proj) shared_root = Path(__file__).parent.parent # recoil/pipeline/ asset_data = shot_data.get("asset_data", {}) shot_chars = asset_data.get("characters", []) refs: list[tuple[bytes, str, str]] = [] for char_entry in shot_chars: char_id = ( char_entry.get("char_id", "") if isinstance(char_entry, dict) else str(char_entry) ) char_upper = char_id.upper() wardrobe_phase = ( char_entry.get("wardrobe_phase_id") if isinstance(char_entry, dict) else None ) # ── Canonical filesystem resolution ── canonical = _resolve_canonical(paths, char_id, phase=wardrobe_phase) if not canonical.get("hero"): continue turnaround_order = ["back", "three_quarter", "profile", "front"] for angle in turnaround_order: if angle in canonical: path = canonical[angle] mime = "image/png" if path.suffix.lower() == ".png" else "image/jpeg" try: refs.append( ( path.read_bytes(), mime, f"Reference: {char_upper} — {angle.replace('_', ' ')} view", ) ) except IOError as e: logger.warning( "Failed to read canonical turnaround %s for %s: %s", angle, char_upper, e, ) hero_path = canonical["hero"] mime = "image/png" if hero_path.suffix.lower() == ".png" else "image/jpeg" try: refs.append( ( hero_path.read_bytes(), mime, f"Reference: {char_upper} — HERO identity reference (prioritize this face)", ) ) except IOError as e: logger.warning("Failed to read canonical hero for %s: %s", char_upper, e) return refs def build_keyframe_refs( shot: dict, project: str | None = None, ) -> list[tuple[bytes, str, str]]: """Full ref stack for NBP keyframe generation -- 5 refs max. NBP averages ref embeddings. More refs = diluted faces. Cap at 5. Dynamic allocation by character count: | Characters | Location | Prop | Turnarounds | Heroes | |------------|----------|------|----------------|---------------| | 1 char | 1 moodboard | 1 | front, side, 3/4 | 1 hero | | 2 chars | 1 moodboard | 0 | front (each) | 1 hero (each) | | 3+ chars | 1 moodboard | 0 | -- | 1 hero (each) | Location always FIRST (lowest priority). Prop refs placed after location (low priority, only when budget allows). Heroes always LAST (recency bias = strongest identity influence). Returns: List of (bytes, mime_type, label) tuples, weight-sorted ascending. """ proj = project or DEFAULT_PROJECT paths = ProjectPaths.for_project(proj) proj_root = paths.project_root shared_root = Path(__file__).parent.parent # recoil/pipeline/ refs: list[tuple[bytes, str, str]] = [] # Count characters in this shot asset_data = shot.get("asset_data", {}) shot_chars = asset_data.get("characters", []) num_chars = len(shot_chars) # 1. Location environment refs — mode: single / moodboard / text_only location_id = asset_data.get("location_id", "") shot_framing = shot.get("prompt_data", {}).get("shot_type", "MS") from recoil.core.paths import get_config as _get_cfg _cfg = _get_cfg() _loc_mode = _cfg.get("location_ref_mode", "single") if _cfg.get("skip_location_ref_images", False) and _loc_mode == "single": _loc_mode = "text_only" # backwards compat if shot.get("_moodboard_text"): pass # text version injected into prompt instead — frees this ref slot elif _loc_mode == "text_only": pass # text-only env prompting — no image refs (text injected elsewhere) elif _loc_mode == "moodboard" and location_id: loc_refs = resolve_location_moodboard(location_id, project=project) moodboard_label = ( "ENVIRONMENT MOOD BOARD — Extract COMMON visual elements (materials, " "colors, lighting, architecture) from these diverse angles. " "Synthesize a FRESH composition. Do NOT copy any single image." ) for img_bytes, mime in loc_refs: refs.append((img_bytes, mime, f"[{moodboard_label}] {location_id}")) elif location_id: loc_refs = resolve_location_refs( location_id, project=project, max_refs=1, framing=shot_framing ) if _is_defocused_framing(shot_framing): loc_label = ( "HEAVILY DEFOCUSED BACKGROUND REFERENCE — Shallow depth of field. " "Only the color palette and faint ambient glow from this environment " "should appear as soft bokeh behind the subject." ) else: loc_label = ( "ENVIRONMENT MOOD REFERENCE — Use the lighting, color palette, and atmosphere " "as creative inspiration for a fresh, original environment." ) for img_bytes, mime in loc_refs: refs.append((img_bytes, mime, f"[{loc_label}] {location_id}")) # 2. Prop ref (1 max -- after location, before characters) # Only include when ref budget allows (1-char shots have room) prop_refs_list = extract_mentioned_props(shot, project=project) if prop_refs_list and num_chars <= 1: prop = prop_refs_list[0] ref_path = prop["ref_path"] mime = "image/png" if ref_path.suffix.lower() == ".png" else "image/jpeg" try: refs.append( (ref_path.read_bytes(), mime, f"Prop reference: {prop['prop_id']}") ) except IOError: pass # 3. Character refs -- canonical filesystem resolution from recoil.core.ref_resolver import resolve_character_refs as _resolve_canonical turnaround_refs: list[tuple[bytes, str, str]] = [] hero_refs: list[tuple[bytes, str, str]] = [] for char_entry in shot_chars: char_id = ( char_entry.get("char_id", "") if isinstance(char_entry, dict) else str(char_entry) ) char_upper = char_id.upper() wardrobe_phase = ( char_entry.get("wardrobe_phase_id") if isinstance(char_entry, dict) else None ) # ── Canonical filesystem resolution ── canonical = _resolve_canonical(paths, char_id, phase=wardrobe_phase) if not canonical.get("hero"): continue # Dynamic allocation using canonical refs if num_chars == 1: for angle in ["front", "profile", "three_quarter"]: if angle in canonical: path = canonical[angle] mime = ( "image/png" if path.suffix.lower() == ".png" else "image/jpeg" ) try: turnaround_refs.append( ( path.read_bytes(), mime, f"Reference: {char_upper} — {angle.replace('_', ' ')} view", ) ) except IOError: pass elif num_chars == 2: angles_to_load = ["front"] if _loc_mode == "text_only": angles_to_load.append("profile") for angle in angles_to_load: if angle in canonical: path = canonical[angle] mime = ( "image/png" if path.suffix.lower() == ".png" else "image/jpeg" ) try: turnaround_refs.append( ( path.read_bytes(), mime, f"Reference: {char_upper} — {angle.replace('_', ' ')} view", ) ) except IOError: pass # 3+ chars: no turnarounds hero_path = canonical["hero"] mime = "image/png" if hero_path.suffix.lower() == ".png" else "image/jpeg" try: hero_refs.append( ( hero_path.read_bytes(), mime, f"Reference: {char_upper} — HERO identity reference (prioritize this face)", ) ) except IOError: pass # Assemble: location -> prop -> turnarounds -> heroes (heroes LAST for recency bias) refs.extend(turnaround_refs) refs.extend(hero_refs) # Enforce ref cap -- no cap for moodboard (curated picks drive count), 6 for text_only, 5 default _ref_cap = 999 if _loc_mode == "moodboard" else 6 if _loc_mode == "text_only" else 5 if len(refs) > _ref_cap: refs = refs[len(refs) - _ref_cap :] return refs def _resolve_path(rel_path: str, proj_root: Path, shared_root: Path) -> Path | None: """Resolve a relative path against project root then shared starsend root.""" for base in (proj_root, shared_root): p = base / rel_path if p.is_file(): return p return None # ── Shot-aware turnaround selection (previz) ───────────────────────── def select_previz_turnaround(shot: dict, char_id: str) -> str: """Select the best turnaround angle for a character in previz context. Uses shot context to determine the most relevant turnaround angle: | Shot Context | Returns | |-----------------------------------------------------|----------------------| | OTS -- character is "shoulder" (nearer to camera) | three_quarter_back | | OTS -- character is "face" (facing camera) | three_quarter | | Description contains "behind", "back of", etc. | back | | Description contains "profile", "side view" | side | | ECU/CU face shots, or default | front | For OTS shots, first char in asset_data.characters is typically the "shoulder" (nearer camera), second is "face" (facing camera). Also parses prompt_data.prompt_skeleton.subject_line for "over X's shoulder". Args: shot: Full shot dict with prompt_data, asset_data. char_id: Character ID to select angle for. Returns: Turnaround angle string: "front", "side", "back", "three_quarter", or "three_quarter_back". """ char_upper = char_id.upper() prompt_data = shot.get("prompt_data", {}) asset_data = shot.get("asset_data", {}) skeleton = prompt_data.get("prompt_skeleton", {}) shot_type = prompt_data.get("shot_type", "MS").upper() subject_line = (skeleton.get("subject_line", "") or "").lower() action_line = (skeleton.get("action_line", "") or "").lower() description = f"{subject_line} {action_line}" # OTS detection is_ots = shot_type == "OTS" or ("over" in description and "shoulder" in description) if is_ots: # Determine shoulder vs face role chars = asset_data.get("characters", []) char_ids_ordered: list[str] = [] for c in chars: if isinstance(c, dict): char_ids_ordered.append(c.get("char_id", "").upper()) else: char_ids_ordered.append(str(c).upper()) # Parse subject_line for "over X's shoulder" to find shoulder character shoulder_char = None if "over" in subject_line and "shoulder" in subject_line: # Try to find which character name appears between "over" and "shoulder" for cid in char_ids_ordered: if cid.lower() in subject_line: # Check if this char is the one whose shoulder we're looking over over_idx = subject_line.find("over") shoulder_idx = subject_line.find("shoulder") char_idx = subject_line.find(cid.lower()) if over_idx < char_idx < shoulder_idx: shoulder_char = cid break if shoulder_char: if char_upper == shoulder_char: return "three_quarter_back" else: return "three_quarter" elif len(char_ids_ordered) >= 2: # Default: first char = shoulder, second = face if char_upper == char_ids_ordered[0]: return "three_quarter_back" else: return "three_quarter" else: # Single char OTS (unusual) -- default to three_quarter return "three_quarter" # Back/away detection back_keywords = [ "behind", "back of", "walking away", "turns away", "from behind", "retreating", ] if any(kw in description for kw in back_keywords): return "back" # Profile/side detection side_keywords = ["profile", "side view", "in profile", "silhouette"] if any(kw in description for kw in side_keywords): return "side" # Default: front (especially for ECU/CU face shots) return "front" # ── Previz character refs (reduced set) ────────────────────────────── def resolve_previz_character_refs( shot: dict, project: str | None = None ) -> list[tuple[bytes, str, str]]: """Resolve a reduced character ref set for previz generation. For PREVIZ only -- returns 2 images per character: - 1 shot-aware turnaround via select_previz_turnaround() (placed first) - Hero image (1 per character) -- placed LAST for recency Unlike resolve_all_character_refs() which returns all 5+ angles, this keeps the token budget tight for Flash previz. Phase 19+ refactor: no longer reads the legacy casting-state file. All character ref resolution flows through recoil.core.ref_resolver. Returns: List of (bytes, mime_type, label) tuples. """ from recoil.core.ref_resolver import ( resolve_character_refs as _resolve_char_canonical, ) proj = project or DEFAULT_PROJECT paths = ProjectPaths.for_project(proj) asset_data = shot.get("asset_data", {}) shot_chars = asset_data.get("characters", []) turnaround_refs: list[tuple[bytes, str, str]] = [] hero_refs: list[tuple[bytes, str, str]] = [] # When text_only mode frees the location slot, use it for extra identity from recoil.core.paths import get_config as _get_cfg_chr _chr_cfg = _get_cfg_chr() _chr_loc_mode = _chr_cfg.get("location_ref_mode", "single") if _chr_cfg.get("skip_location_ref_images", False) and _chr_loc_mode == "single": _chr_loc_mode = "text_only" _extra_identity = _chr_loc_mode == "text_only" for char_entry in shot_chars: char_id = ( char_entry.get("char_id", "") if isinstance(char_entry, dict) else str(char_entry) ) char_upper = char_id.upper() wardrobe_phase = ( char_entry.get("wardrobe_phase_id") if isinstance(char_entry, dict) else None ) # Resolve the full ref set for this character via ref_resolver. resolved = _resolve_char_canonical(paths, char_upper, phase=wardrobe_phase) if not resolved: continue # 1 shot-aware turnaround (always) best_angle = select_previz_turnaround(shot, char_id) ta_abs_path = resolved.get(best_angle) if ta_abs_path and ta_abs_path.is_file(): mime = "image/png" if ta_abs_path.suffix.lower() == ".png" else "image/jpeg" try: turnaround_refs.append( ( ta_abs_path.read_bytes(), mime, f"Reference: {char_upper} -- {best_angle.replace('_', ' ')} view", ) ) except IOError as e: logger.warning( "Failed to read turnaround %s for %s: %s", best_angle, char_upper, e ) # Bonus turnaround when location slot is freed (complementary angle) if _extra_identity: bonus_angle = "front" if best_angle != "front" else "three_quarter" bonus_abs_path = resolved.get(bonus_angle) if bonus_abs_path and bonus_abs_path.is_file(): mime = ( "image/png" if bonus_abs_path.suffix.lower() == ".png" else "image/jpeg" ) try: turnaround_refs.append( ( bonus_abs_path.read_bytes(), mime, f"Reference: {char_upper} -- {bonus_angle.replace('_', ' ')} view (identity anchor)", ) ) except IOError as e: logger.warning( "Failed to read bonus turnaround %s for %s: %s", bonus_angle, char_upper, e, ) # Hero image LAST for recency abs_path = resolved.get("hero") if abs_path and abs_path.is_file(): mime = "image/png" if abs_path.suffix.lower() == ".png" else "image/jpeg" try: hero_refs.append( ( abs_path.read_bytes(), mime, f"Reference: {char_upper} -- HERO identity reference (prioritize this face)", ) ) except IOError as e: logger.warning("Failed to read hero for %s: %s", char_upper, e) # Turnarounds first, heroes last (recency bias) return turnaround_refs + hero_refs # ── System instruction (split into preamble + directive) ───────────── # Hardcoded fallback in case the external prompt file is missing. _FALLBACK_STATIC_INSTRUCTION = """ COMPOSITIONAL CONSTRAINTS FOR PREVIZ: You are generating previz frames that MUST be perfectly reproducible by a photorealistic live-action camera. - USE standard cinematic focal lengths only (35mm, 50mm, 85mm) - DO NOT use extreme wide-angle, fisheye, or GoPro-style lenses - DO NOT use dynamic Dutch angles or tilted horizons. Keep the camera level. - DO NOT use extreme forced perspective (e.g., a giant fist in the foreground) - Prioritize flat, medium-contrast cinematic lighting. Avoid pure black silhouettes unless explicitly requested. - Keep subjects grounded. No floating, leaping, or physically impossible anime-style poses. REFERENCE IMAGE PRIORITY: Character hero images (closest to this text) should receive maximum attention for face/identity consistency. Turnaround views provide structural reference. Location moodboards ground the environment. CRITICAL -- HAIR & APPEARANCE LOCK: Each character's hairstyle, hair color, and hair length MUST match EXACTLY what is shown in their hero reference image. Do NOT vary hairstyles between shots. If the hero image shows loose hair, use loose hair. If it shows a ponytail, use a ponytail. The hero image is the single source of truth for appearance -- ignore any text that suggests variation. STEP 1: Write a cinematographic prompt for this specific frame. Describe: - Exact character appearance (face, EXACT hair from hero ref, wardrobe from the references) - Camera angle, focal length, depth of field - Lighting setup (key, fill, practicals) - Environment details matching the location moodboard - Character pose, body language, expression - Composition within the target aspect ratio frame Be specific -- name colors, textures, distances. Do NOT describe what you will do. Write the prompt as if directing a photographer. STEP 2: Generate the previz frame matching your prompt exactly.""".strip() def build_behavioral_preamble(shot: dict) -> str: """Build the behavioral preamble -- slot [1] primacy anchor. Contains anti-spoiler rules, style constraints, DO NOT directives, camera direction guard, compositional constraints, hair/appearance lock rules, and reference image priority guidance. This goes FIRST in context for primacy-based behavioral anchoring. """ routing_data = shot.get("routing_data", {}) is_env = routing_data.get("is_env_only", False) spatial_data = shot.get("spatial_data", {}) screen_dir = spatial_data.get("screen_direction", "center") sections = [] sections.append("You are a previz cinematographer.") sections.append( "Study ALL the reference material carefully:\n" "- The VISUAL BIBLE describes the world, characters, and locations\n" "- The SHOT SEQUENCE shows where this shot sits in the episode flow\n" "- The REFERENCE IMAGES show character faces (turnarounds + hero) and location mood" ) # Spatial continuity behavioral anchor — tells Flash to obey spatial instructions if spatial_data.get("camera_side"): sections.append( "SPATIAL CONTINUITY: You MUST follow all screen position " "and [LEFT/RIGHT/CENTER OF FRAME] placement instructions exactly. " "Character screen positions and lighting directions are computed " "by the production system and must not be reinterpreted." ) elif not is_env and screen_dir != "toward-camera": # Fallback for shots without spatial data guard = get_constant( "production", "camera_direction_guard", default="Candid observational camera. Gaze fixed on an object or another character. Three-quarter profile or looking off-frame.", ) sections.append(guard) # Static compositional constraints from versioned prompt file static_text = load_prompt_file("flash_previz_v1.0.txt") if static_text: constraints_marker = "COMPOSITIONAL CONSTRAINTS FOR PREVIZ:" if constraints_marker in static_text: idx = static_text.index(constraints_marker) # Extract everything from COMPOSITIONAL CONSTRAINTS through STEP 2 block # But stop before STEP 1/2 -- those are generative, not behavioral constraint_text = static_text[idx:] # Strip out STEP 1 and STEP 2 -- those belong in the generative directive step_marker = "STEP 1:" if step_marker in constraint_text: constraint_text = constraint_text[ : constraint_text.index(step_marker) ].rstrip() sections.append(constraint_text) else: sections.append(static_text) else: # Fallback: extract behavioral portion from hardcoded text fallback = _FALLBACK_STATIC_INSTRUCTION step_marker = "STEP 1:" if step_marker in fallback: sections.append(fallback[: fallback.index(step_marker)].rstrip()) else: sections.append(fallback) # Expression ref lighting override sections.append( "[CRITICAL: Ignore lighting and color from expression reference. " "Match moodboard lighting exactly.]" ) # Reverse OTS coverage — tell Flash the camera is on the opposite side prompt_data = shot.get("prompt_data", {}) if prompt_data.get("reverse_ots", False): asset_data = shot.get("asset_data", {}) chars = asset_data.get("characters", []) if len(chars) >= 2: shoulder = ( chars[0].get("char_id", "") if isinstance(chars[0], dict) else str(chars[0]) ) face = ( chars[1].get("char_id", "") if isinstance(chars[1], dict) else str(chars[1]) ) sections.append( f"REVERSE OTS COVERAGE: The camera has MOVED to the OTHER SIDE of the scene. " f"We are now looking OVER {shoulder}'s shoulder/back AT {face}'s face. " f"The background behind {face} is what was BEHIND THE CAMERA in the original shot. " f"{shoulder}'s back/shoulder fills the foreground (soft focus). " f"{face}'s face is the subject (sharp focus). " f"This is NOT the same angle as the original OTS — it is the REVERSE." ) return "\n\n".join(sections) def build_generative_directive(shot: dict, bible: dict = None) -> str: """Build the generative shot directive -- slot [8] recency anchor. Contains "YOUR TASK: Generate shot {shot_id}...", shot specification (type, characters, location, action, emotion, lighting, dialogue), spatial continuity block, and [SHOT_TYPE_OVERRIDE] at absolute end. This goes LAST in context for recency-based action anchoring. """ shot_id = shot.get("shot_id", "UNKNOWN") prompt_data = shot.get("prompt_data", {}) asset_data = shot.get("asset_data", {}) routing_data = shot.get("routing_data", {}) skeleton = prompt_data.get("prompt_skeleton", {}) # JIT hydration: resolve tokens from live bible data if not bible: try: from recoil.pipeline._lib.jit_prompt import hydrate_skeleton from recoil.pipeline._lib.io_utils import safe_read_json from recoil.core.paths import ProjectPaths bible_path = ProjectPaths.for_project().global_bible_path if bible_path.exists() and skeleton: bible = safe_read_json(bible_path) except (ImportError, Exception): pass # Graceful fallback if bible and skeleton: try: from recoil.pipeline._lib.jit_prompt import hydrate_skeleton skeleton = hydrate_skeleton(skeleton, bible, asset_data=asset_data) except (ImportError, Exception): pass shot_type = prompt_data.get("shot_type", "MS") focal_length = prompt_data.get("focal_length", "") # Previz is always static frames — camera_movement is noise for still-image models location_id = asset_data.get("location_id", "unknown") is_env = routing_data.get("is_env_only", False) # Characters with positions chars = asset_data.get("characters", []) if is_env: char_desc = "ENVIRONMENT ONLY -- no people in frame" elif chars: char_lines = [] for c in chars: if isinstance(c, dict): cid = c.get("char_id", "?") pos = c.get("screen_position", "center") char_lines.append(f" - {cid} ({pos})") else: char_lines.append(f" - {c}") char_desc = "\n".join(char_lines) else: char_desc = "None specified" # Handle reverse OTS from coverage — rewrite subject_line for reversed perspective is_reverse_ots = prompt_data.get("reverse_ots", False) if is_reverse_ots and shot_type == "OTS" and len(chars) >= 2: shoulder_char = ( chars[0].get("char_id", "") if isinstance(chars[0], dict) else str(chars[0]) ) face_char = ( chars[1].get("char_id", "") if isinstance(chars[1], dict) else str(chars[1]) ) skeleton["subject_line"] = ( f"Over-the-shoulder shot from behind {shoulder_char}, " f"looking at {face_char}'s face" ) # Action / emotion / lighting action_line = skeleton.get("subject_line", "") emotion_line = skeleton.get("emotion_line", "") env_line = skeleton.get("environment_line", "") # Lighting from shot data lighting_data = prompt_data.get("lighting", {}) sources = lighting_data.get("sources", []) if sources: light_parts = [] for src in sources[:3]: motivator = src.get("motivator", "") quality = src.get("quality", "") color = src.get("color_temp", "") if motivator: light_parts.append(f"{color} {quality} light from {motivator}".strip()) lighting_desc = "; ".join(light_parts) else: lighting_desc = "Use location default" # Dialogue dialogue = shot.get("dialogue", "") or prompt_data.get("dialogue", "") # STEP instructions for generation step_instructions = ( "STEP 1: Write a cinematographic prompt for this specific frame. Describe:\n" "- Exact character appearance (face, EXACT hair from hero ref, wardrobe from the references)\n" "- Camera angle, focal length, depth of field\n" "- Lighting setup (key, fill, practicals)\n" "- Environment details matching the location moodboard\n" "- Character pose, body language, expression\n" "- Composition within the target aspect ratio frame\n" "Be specific -- name colors, textures, distances. Do NOT describe what you will do.\n" "Write the prompt as if directing a photographer.\n" "\n" "STEP 2: Generate the previz frame matching your prompt exactly." ) from recoil.core.paths import get_config as _get_cfg _sc = _get_cfg() _ar = _sc.get("production_aspect_ratio", "9:16") _orient = "vertical" if _ar == "9:16" else "horizontal" directive = f"""YOUR TASK: Generate shot {shot_id} as a {_ar} {_orient} previz frame. FRAMING: Generate a STATIC frame showing the setup moment — the character is in position, poised, braced, or holding still. Describe their exact physical pose and body geometry. Show the tension and kinetic potential of the moment. The character is ready, the scene is set. CAMERA STYLE: Candid observational camera. Cinematic production shot. The subject's gaze MUST be anchored to a specific object, another character, or a direction in the scene (e.g., "eyes fixed on the conduit," "looking down the corridor," "watching her hands"). Describe the eye direction as part of the character's pose, in the same sentence as their face description. Use terms like: three-quarter profile, looking off-frame, eyes downcast, gaze locked on [object]. Shot specification: - Type: {shot_type}{f", {focal_length}" if focal_length else ""} - Characters: {char_desc} - Location: {location_id} - Action: {action_line} - Emotion: {emotion_line} - Environment: {env_line} - Lighting: {lighting_desc}""" if dialogue: directive += f'\n- Dialogue: "{dialogue}"' # Spatial continuity block — Gemini-approved injection from recoil.pipeline._lib.prompt_engine import build_spatial_continuity_block spatial_block = build_spatial_continuity_block( shot=shot, bible=bible or {}, scene_shots=shot.get("_scene_shots"), ) if spatial_block: directive += f"\n\n{spatial_block}" directive += f"\n\n{step_instructions}" # [SHOT_TYPE_OVERRIDE] at absolute end for camera anchoring directive += f"\n\n[SHOT_TYPE_OVERRIDE] {shot_type}" return directive def build_system_instruction(shot: dict, bible: dict = None) -> str: """Build the combined system instruction (preamble + directive). .. deprecated:: This function concatenates build_behavioral_preamble() and build_generative_directive() for backward compatibility with the keyframe pipeline. New code should use the split functions directly for proper 8-slot ordering. The previz pipeline uses the split functions with 8-slot ordering. The keyframe pipeline still calls this combined version. """ preamble = build_behavioral_preamble(shot) directive = build_generative_directive(shot, bible=bible) return f"{preamble}\n\n{directive}" # ── Expression ref resolution ───────────────────────────────────────── EMOTION_MAP = { "anger": ["anger", "angry", "rage", "furious", "fury", "wrath", "enraged"], "contempt": ["contempt", "disdain", "scorn", "dismissive", "sneer"], "disgust": ["disgust", "revulsion", "repulsed", "nauseated"], "fear": [ "fear", "afraid", "terrified", "terror", "frightened", "scared", "anxious", "dread", ], "joy": [ "joy", "happy", "happiness", "smile", "smiling", "grin", "elated", "relief", "warm", ], "sadness": [ "sad", "sadness", "sorrow", "grief", "mourning", "melancholy", "despair", "heartbroken", ], "surprise": ["surprise", "surprised", "shock", "shocked", "astonished", "stunned"], "pain": ["pain", "agony", "anguish", "hurt", "suffering", "wince", "grimace"], "determination": [ "determination", "determined", "resolve", "resolved", "steely", "focused", "defiant", "grit", ], } INTENSITY_LOW = ["barely", "slight", "faint", "quiet", "subtle"] INTENSITY_HIGH = [ "intense", "extreme", "overwhelming", "burning", "deep", "fierce", "heavy", ] def _match_emotion(text: str) -> str | None: """Map free-form emotion text to one of the 9 canonical emotions. Returns the canonical emotion name or None if no match / neutral. """ if not text: return None lower = text.lower() for canonical, keywords in EMOTION_MAP.items(): for kw in keywords: if kw in lower: return canonical return None def _match_intensity(text: str) -> str: """Extract intensity from emotion text. Defaults to 'active'. Returns one of the three on-disk intensity tokens ('subtle' / 'active' / 'extreme'). The expression ref library at `recoil/pipeline/assets/expressions/` uses exactly these tokens — callers append `_{intensity}.png` directly to a filename stem. """ if not text: return "active" lower = text.lower() for kw in INTENSITY_LOW: if kw in lower: return "subtle" for kw in INTENSITY_HIGH: if kw in lower: return "extreme" return "active" def resolve_expression_ref( shot: dict, project: str | None = None ) -> tuple[bytes, str, str] | None: """Resolve an expression reference image for a shot. Checks asset_data.characters[*].emotion_keyword first, then falls back to prompt_skeleton.emotion_line. Maps the text to a canonical emotion and intensity, then looks up the grayscale expression ref on disk. Returns (bytes, mime_type, label) or None if no expression ref is needed. """ emotion_text = None # Priority 1: character-level emotion_keyword from asset_data characters = shot.get("asset_data", {}).get("characters", []) for char in characters: if isinstance(char, dict): kw = char.get("emotion_keyword") or char.get("emotion") if kw and isinstance(kw, str) and kw.strip(): emotion_text = kw.strip() break # Priority 2: emotion_line from prompt_skeleton if not emotion_text: emotion_line = ( shot.get("prompt_data", {}).get("prompt_skeleton", {}).get("emotion_line") ) if emotion_line and isinstance(emotion_line, str) and emotion_line.strip(): emotion_text = emotion_line.strip() if not emotion_text: return None emotion = _match_emotion(emotion_text) if not emotion: return None intensity = _match_intensity(emotion_text) # R7.7: belt-and-suspenders. _match_intensity is contracted to return one # of these three tokens, but a future caller that wires in a different # intensity source would silently 404 against the ref library — log it # and fall back to the middle band instead. if intensity not in ("subtle", "active", "extreme"): logger.warning( "previz_context: unknown intensity %r, falling back to 'active'", intensity, ) intensity = "active" # Expressions are SHARED across projects (recoil/pipeline/assets/expressions/), # not per-project. Resolve against PIPELINE_ROOT, not ProjectPaths. ref_path = PIPELINE_ROOT / "assets" / "expressions" / f"{emotion}_{intensity}.png" if not ref_path.is_file(): logger.debug("Expression ref not found: %s", ref_path) return None try: img_bytes = ref_path.read_bytes() except IOError as e: logger.warning("Failed to read expression ref %s: %s", ref_path, e) return None label = f"Expression reference: {emotion} ({intensity})" return (img_bytes, "image/png", label) # ── Main assembly ───────────────────────────────────────────────────── def build_previz_context( shot: dict, all_shots: list[dict], bible: dict | None = None, episode: int | None = None, project: str | None = None, ) -> list[tuple[bytes | None, str, str | None]]: """Assemble the full context for Flash-authored previz. Returns a list of content parts: (data, mime_type_or_text, label) - Text parts: (None, "text", text_content) - Image parts: (bytes, mime_type, label) 8-slot content layout: [1] Behavioral preamble (primacy anchor) [2] Scoped bible text [3] Shot sequence (3-shot window: N-1, current, N+1) [4] Location moodboards (2-3 max) [5] Prop ref (1 max) [6] Character refs (hero + 1 turnaround per char) [7] Expression ref (1, only if non-neutral emotion) [8] Generative directive (recency anchor) All context is optional -- gracefully degrades if nothing is available. At minimum, Flash gets the behavioral preamble + generative directive. """ parts: list[tuple[bytes | None, str, str | None]] = [] shot_id = shot.get("shot_id", "UNKNOWN") asset_data = shot.get("asset_data", {}) # [1] Behavioral preamble (primacy anchor -- text) preamble = build_behavioral_preamble(shot) parts.append((None, "text", preamble)) # [2] Scoped bible context (text) scoped = scope_bible_to_shot(bible, asset_data) if bible else {} bible_text = format_bible_context(scoped, episode=episode) if scoped else "" if bible_text: parts.append((None, "text", f"# VISUAL BIBLE\n\n{bible_text}")) # [2b] Scene visual locks (from first keyframe DNA extraction) scene_locks = shot.get("_scene_visual_locks") if scene_locks: lock_text = _format_scene_locks(scene_locks) parts.append( (None, "text", f"## SCENE VISUAL LOCKS (maintain consistency)\n{lock_text}") ) # [3] Shot sequence -- 3-shot window (N-1, current, N+1) if all_shots: current_idx = next( (i for i, s in enumerate(all_shots) if s.get("shot_id") == shot_id), 0 ) window_start = max(0, current_idx - 1) window_end = min(len(all_shots), current_idx + 2) window_shots = all_shots[window_start:window_end] sequence_text = format_shot_sequence(window_shots, shot_id) if sequence_text: parts.append((None, "text", sequence_text)) # [4] Location environment refs # Mode: "single" (1 framing-matched), "moodboard" (2-3 diverse), "text_only" location_id = asset_data.get("location_id", "") shot_framing = shot.get("prompt_data", {}).get("shot_type", "MS") from recoil.core.paths import get_config as _get_cfg_ctx _cfg_ctx = _get_cfg_ctx() _loc_mode = _cfg_ctx.get("location_ref_mode", "single") # Backwards compat: old skip flag → text_only mode if _cfg_ctx.get("skip_location_ref_images", False) and _loc_mode == "single": _loc_mode = "text_only" if shot.get("_moodboard_text"): parts.append( (None, "text", f"## LOCATION: {location_id}\n{shot['_moodboard_text']}") ) elif _loc_mode == "text_only" and location_id: # Text-only environment — pull description from bible loc_text = _build_location_text_from_bible(bible, location_id) if loc_text: parts.append((None, "text", loc_text)) elif _loc_mode == "moodboard" and location_id: # Multi-image mood board — Flash synthesizes the common visual language loc_refs = resolve_location_moodboard(location_id, project=project) if loc_refs: # Also inject text description for reinforcement loc_text = _build_location_text_from_bible(bible, location_id) if loc_text: parts.append((None, "text", loc_text)) moodboard_label = ( "ENVIRONMENT MOOD BOARD — These images show the SAME location from " "different angles. Extract the COMMON visual elements (materials, " "color palette, lighting quality, architectural style) and synthesize " "a FRESH, ORIGINAL composition. Do NOT reproduce any single image." ) for i, (img_bytes, mime) in enumerate(loc_refs): parts.append( ( img_bytes, mime, f"[{moodboard_label}] {location_id} ({i + 1}/{len(loc_refs)})", ) ) else: # No moodboard images found — fall back to text loc_text = _build_location_text_from_bible(bible, location_id) if loc_text: parts.append((None, "text", loc_text)) elif location_id: # Single framing-matched image (original behavior) loc_refs = resolve_location_refs( location_id, project=project, framing=shot_framing ) if _is_defocused_framing(shot_framing): loc_label = ( "HEAVILY DEFOCUSED BACKGROUND REFERENCE — Shallow depth of field. " "Only the color palette and faint ambient glow from this environment " "should appear as soft bokeh behind the subject." ) else: loc_label = ( "ENVIRONMENT MOOD REFERENCE — Use the lighting, color palette, " "and atmosphere as creative inspiration for a fresh, original " "environment that matches the scene description below." ) for i, (img_bytes, mime) in enumerate(loc_refs): parts.append( ( img_bytes, mime, f"[{loc_label}] {location_id} ({i + 1}/{len(loc_refs)})", ) ) # [4b] Secondary location refs (skip in text_only mode) if _loc_mode not in ("text_only",): secondary_ids = asset_data.get("secondary_location_ids", []) for sec_id in secondary_ids: sec_refs = resolve_location_refs( sec_id, project=project, max_refs=1, framing=shot_framing ) for img_bytes, mime in sec_refs: parts.append( (img_bytes, mime, f"Secondary location (mood ref): {sec_id}") ) # [5] Prop reference (1 max) prop_refs = extract_mentioned_props(shot, project=project) for prop in prop_refs[:1]: ref_path = prop["ref_path"] mime = "image/png" if ref_path.suffix.lower() == ".png" else "image/jpeg" try: parts.append( (ref_path.read_bytes(), mime, f"Prop reference: {prop['prop_id']}") ) except IOError as e: logger.warning("Failed to read prop ref %s: %s", ref_path, e) # [6] Character refs (1 turnaround + hero per char) char_refs = resolve_previz_character_refs(shot, project=project) for img_bytes, mime, label in char_refs: parts.append((img_bytes, mime, label)) # [7] Expression ref (1, only if non-neutral emotion) expr_ref = resolve_expression_ref(shot, project=project) if expr_ref: parts.append(expr_ref) # [8] Generative directive (recency anchor -- text) directive = build_generative_directive(shot, bible=bible) parts.append((None, "text", directive)) return parts # ══════════════════════════════════════════════════════════════════════ # SCENE CONTEXT INJECTION # ══════════════════════════════════════════════════════════════════════ def inject_scene_context(shot: dict, all_shots: list[dict]) -> None: """Populate _scene_shots on a shot dict for spatial continuity. Groups shots by scene_index and attaches scene-mates as _scene_shots. Must be called before build_prompt_from_plan() or build_generative_directive() for spatial enforcement to work. """ if not all_shots: return scene_index = shot.get("scene_index") if scene_index is None: return scene_shots = [s for s in all_shots if s.get("scene_index") == scene_index] shot["_scene_shots"] = scene_shots