""" assembler.py — ShotAssembler: compile prompt text + reference images into API payloads. Takes a PromptPackage (prompt text, weight-sorted references, model config) and compiles it into the final google.genai Parts list for generation calls. Key ordering principle (Gemini recency bias): The LAST image Part before the text prompt gets the MOST model attention. References are sorted by weight ascending so identity refs (weight 8-10) sit closest to the prompt text, while scene refs (weight 1-2) are furthest away. Order: Scene (1-2) -> Pose (4-5) -> Expression (6-7) -> Prop (7-8) -> Identity (8-10) -> Prompt text Provenance: ported from starsend/lib/assembler.py (Execution Layer Port, Build 2, Phase 2). Import paths rewritten from legacy `lib.*` to `core.*` / `execution.*`. """ import json import logging from dataclasses import dataclass, field from pathlib import Path from typing import Optional from recoil.execution.asset_manager import ReferenceImage from recoil.core import model_profiles from recoil.core.paths import CONFIG_PATH # Optional import — google-genai may not be installed in all environments try: from google.genai import types as genai_types _HAS_GENAI = True except ImportError: genai_types = None # type: ignore[assignment] _HAS_GENAI = False logger = logging.getLogger(__name__) DEFAULT_NEGATIVE_PROMPT = "morphing, flicker, facial distortion, extra fingers, text overlay, watermark" # ====================================================================== # Reference Allocation (ADR-M01) # ====================================================================== def allocate_references( pipeline: str, # 'still', 'i2v', 't2v', 'multi_shot' model: str, # model ID num_chars: int, has_props: bool, is_env: bool, ) -> dict: """Returns ref slot allocation per type. ADR-M01 priority: Keyframe > Identity > Prop > Expression > Scene.""" allocation = {"identity": 0, "prop": 0, "expression": 0, "scene": 0, "keyframe": 0} if "veo" in model: if num_chars > 0 or has_props: raise ValueError("Veo 3.1 cannot handle characters or props. Reroute.") allocation["scene"] = 3 return allocation if "kling" in model: allocation["keyframe"] = 1 if pipeline == "i2v" else 0 allocation["identity"] = min(num_chars, 4) allocation["prop"] = 2 if has_props else 0 allocation["scene"] = 1 if is_env else 0 return allocation if "seeddance" in model: if pipeline == "r2v": slots_left = 9 allocation["identity"] = min(num_chars * 2, 4) # max 4 identity refs slots_left -= allocation["identity"] allocation["scene"] = 1 # always 1 location ref slots_left -= 1 if has_props and slots_left > 0: allocation["prop"] = min(1, slots_left) return allocation slots_left = 9 if pipeline == "multi_shot": allocation["scene"] = 1 # w=0.25, mandatory for background consistency slots_left -= 1 else: allocation["keyframe"] = 1 # w=0.70 slots_left -= 1 allocation["identity"] = min(num_chars * 2, slots_left) slots_left -= allocation["identity"] if has_props and slots_left > 0: allocation["prop"] = min(2, slots_left) return allocation if "seedream" in model: # Seedream edit endpoint: up to 10 refs as URLs # Fewer is better — labeled roles pattern needs clear Figure assignment slots_left = 10 allocation["identity"] = min(num_chars * 2, 6) slots_left -= allocation["identity"] if has_props and slots_left > 0: allocation["prop"] = min(2, slots_left) slots_left -= allocation["prop"] if not is_env and slots_left > 0: allocation["expression"] = min(num_chars, slots_left) slots_left -= allocation["expression"] if slots_left > 0: allocation["scene"] = 1 return allocation if "wan" in model: if pipeline in ("i2v", "between"): # Wan I2V / In Between: frame IS the reference. Keyframe slot for start frame. allocation["keyframe"] = 1 return allocation if pipeline == "r2v": # Wan R2V: identity refs only (frontal + 3/4 per character) # NO scene refs — prompt carries environment. Max 10 total. allocation["identity"] = min(num_chars * 2, 8) slots_left = 10 - allocation["identity"] if has_props and slots_left >= 1: allocation["prop"] = min(1, slots_left) return allocation # Wan T2V fallback — no refs needed return allocation # NBP stills (gemini): 11 slots, recency bias slots_left = 11 allocation["identity"] = min(num_chars * 3, 5) slots_left -= allocation["identity"] if has_props: allocation["prop"] = min(3, slots_left) slots_left -= allocation["prop"] allocation["expression"] = min(num_chars, slots_left) slots_left -= allocation["expression"] if slots_left > 0: allocation["scene"] = 1 return allocation # ====================================================================== # API Payload dataclasses — ready-to-send, model-specific payloads # ====================================================================== @dataclass class GenaiPayload: """Ready-to-send payload for Google genai API.""" parts: list # genai Parts list config: dict # GenerateContentConfig kwargs model: str modality: str = "image" @dataclass class KlingPayload: """Ready-to-send payload for Kling REST API.""" prompt: str model: str duration_s: int start_frame_bytes: Optional[bytes] = None end_frame_bytes: Optional[bytes] = None aspect_ratio: str = "9:16" negative_prompt: str = DEFAULT_NEGATIVE_PROMPT camera_control: Optional[dict] = None # {"pan": 0.5, "tilt": 0.0, "roll": 0.0, "zoom": 0.0} (ADR-R10) @dataclass class SeedDancePayload: """Ready-to-send payload for SeedDance via fal.ai.""" prompt: str model: str duration_s: int reference_images: list[bytes] = field(default_factory=list) reference_videos: list[str] = field(default_factory=list) reference_weights: list[float] = field(default_factory=list) # Per-ref weights (0.0-1.0) aspect_ratio: str = "9:16" audio_prompt: Optional[str] = None @dataclass class SeedreamPayload: """Ready-to-send payload for Seedream image generation via fal.ai. References are ReferenceImage objects — the client handles uploading to fal storage and building Figure N role labels in the prompt. """ prompt: str model: str reference_images: list = field(default_factory=list) # list[ReferenceImage] image_size: str = "portrait_16_9" aspect_ratio: str = "9:16" num_images: int = 1 enable_safety_checker: bool = False # ByteDance filters too aggressive for microdrama content @dataclass class KeyframeRefBundle: """Model-agnostic reference bundle for keyframe generation. StepRunner collects refs into this structure. Each client\'s generate_keyframe() method transforms it into a model-specific payload. """ prompt: str model: str aspect_ratio: str = "9:16" scene_ref: Optional[Path] = None pose_ref: Optional[Path] = None identity_refs: list = field(default_factory=list) # list[Path] expression_refs: list = field(default_factory=list) # list[Path] inputs_snapshot: Optional[dict] = None def compile_keyframe_parts(bundle: KeyframeRefBundle) -> list: """Build the genai parts list from a KeyframeRefBundle. Extracted from GoogleGenaiClient.generate_keyframe (api_client.py:554) in CP-2 Phase 6. Behavior must be byte-identical at the parts-list level — keyframe quality depends on this exact translation. Ordering follows the Gemini recency bias rule: Scene (lowest weight) first, then Pose, Expression, Identity (highest weight = closest to prompt), and finally the prompt text last. """ if not _HAS_GENAI: raise RuntimeError( "google-genai SDK not installed. pip install google-genai" ) genai_parts: list = [] # Scene ref (lowest attention weight) if bundle.scene_ref: if bundle.scene_ref.exists(): genai_parts.append(genai_types.Part.from_bytes( data=bundle.scene_ref.read_bytes(), mime_type="image/png")) genai_parts.append(genai_types.Part.from_text( text="[ENVIRONMENT MOOD REFERENCE \u2014 Use the lighting direction, " "color palette, and atmospheric mood from this image as creative " "inspiration for a fresh, original environment that captures the " "same feeling and matches the scene description below.]")) else: logger.warning( "FALLBACK_FIRED keyframe_ref_missing kind=%s path=%s", "scene", bundle.scene_ref, ) # Pose ref if bundle.pose_ref: if bundle.pose_ref.exists(): genai_parts.append(genai_types.Part.from_bytes( data=bundle.pose_ref.read_bytes(), mime_type="image/png")) genai_parts.append(genai_types.Part.from_text( text="[POSE REFERENCE \u2014 body positioning]")) else: logger.warning( "FALLBACK_FIRED keyframe_ref_missing kind=%s path=%s", "pose", bundle.pose_ref, ) # Expression refs for ref in (bundle.expression_refs or []): if ref.exists(): genai_parts.append(genai_types.Part.from_bytes( data=ref.read_bytes(), mime_type="image/png")) genai_parts.append(genai_types.Part.from_text( text="[EXPRESSION REFERENCE \u2014 emotion/facial expression]")) else: logger.warning( "FALLBACK_FIRED keyframe_ref_missing kind=%s path=%s", "expression", ref, ) # Identity refs (highest weight = closest to prompt) for ref in (bundle.identity_refs or []): if not ref.exists(): raise ValueError(f"Missing identity reference for keyframe: {ref}") suffix = ref.suffix.lower() mime = "image/jpeg" if suffix in (".jpg", ".jpeg") else "image/png" genai_parts.append(genai_types.Part.from_bytes( data=ref.read_bytes(), mime_type=mime)) genai_parts.append(genai_types.Part.from_text( text=f"[CHARACTER IDENTITY REFERENCE \u2014 {ref.stem}]")) # Prompt text LAST (maximum model attention due to recency bias) genai_parts.append(genai_types.Part.from_text(text=bundle.prompt)) return genai_parts @dataclass class MultiShotPayload: """Ready-to-send payload for Kling multi-prompt generation. Represents a batch of 2-6 shots generated in a single API call. Each shot has an index, prompt, and duration. Total duration <= 15s. Format confirmed by API probe: each entry needs {index, prompt, duration}. """ shots: list[dict] = field(default_factory=list) # [{index: int, prompt: str, duration: int}] model: str = "kling-v3" aspect_ratio: str = "9:16" mode: str = "standard" start_frame_bytes: Optional[bytes] = None shot_ids: list[str] = field(default_factory=list) # For tracking back to plan negative_prompt: str = DEFAULT_NEGATIVE_PROMPT elements_payload: Optional[dict] = None # fal.ai Elements: {"elements": [{frontal_image_url, reference_image_urls}]} cfg_scale: Optional[float] = None # Generation guidance scale (0.0-1.0) motion_presets: list[dict] = field(default_factory=list) # Per-shot [{movement, intensity}] @property def total_duration(self) -> int: return sum(s.get("duration", 3) for s in self.shots) def validate(self) -> list[str]: """Pre-flight validation. Returns list of error strings (empty = valid).""" errors = [] if len(self.shots) < 2: errors.append(f"Need >= 2 shots, got {len(self.shots)}") if len(self.shots) > 6: errors.append(f"Max 6 shots, got {len(self.shots)}") if self.total_duration > 15: errors.append(f"Total duration {self.total_duration}s > 15s max") if self.motion_presets and len(self.motion_presets) != len(self.shots): errors.append(f"motion_presets length ({len(self.motion_presets)}) != shots length ({len(self.shots)})") for i, shot in enumerate(self.shots): dur = shot.get("duration", 0) if dur < 3: errors.append(f"Shot {i+1} duration {dur}s < 3s minimum") return errors @dataclass class WanI2VPayload: """Ready-to-send payload for Wan 2.7 Image-to-Video via fal.ai. Supports three modes: - First-frame-only I2V: image_url set, end_image_url None - First+Last frame "In Between": both image_url and end_image_url set - Video continuation: video_url set (extends a prior clip) """ prompt: str image_url: str # First frame (fal storage URL or data URI) end_image_url: Optional[str] = None # Last frame for "In Between" pattern duration: int = 5 # 2-15 seconds resolution: str = "720p" # "720p" or "1080p" negative_prompt: str = DEFAULT_NEGATIVE_PROMPT enable_prompt_expansion: bool = False # ALWAYS False — pipeline controls prompts seed: Optional[int] = None audio_url: Optional[str] = None # Pre-rendered audio (WAV/MP3) video_url: Optional[str] = None # Video continuation from prior clip enable_safety_checker: bool = True def validate(self) -> list[str]: """Pre-flight validation. Returns list of error strings (empty = valid).""" errors = [] if not self.prompt: errors.append("prompt is required") if not self.image_url: errors.append("image_url (first frame) is required") if self.duration < 2 or self.duration > 15: errors.append(f"duration must be 2-15s, got {self.duration}") if self.resolution not in ("720p", "1080p"): errors.append(f"resolution must be 720p or 1080p, got {self.resolution}") if len(self.prompt) > 5000: errors.append(f"prompt exceeds 5000 char limit ({len(self.prompt)} chars)") if self.negative_prompt and len(self.negative_prompt) > 500: errors.append(f"negative_prompt exceeds 500 char limit ({len(self.negative_prompt)} chars)") return errors @dataclass class WanR2VPayload: """Ready-to-send payload for Wan 2.7 Reference-to-Video via fal.ai. Uses character reference images/videos for identity preservation. multi_shots enables intelligent multi-shot segmentation where the model decides camera changes internally. """ prompt: str reference_image_urls: list[str] = field(default_factory=list) reference_video_urls: list[str] = field(default_factory=list) multi_shots: bool = False duration: int = 5 # 2-10 seconds (shorter max than I2V) resolution: str = "720p" # "720p" or "1080p" aspect_ratio: str = "9:16" negative_prompt: str = "" seed: Optional[int] = None enable_safety_checker: bool = True def validate(self) -> list[str]: """Pre-flight validation. Returns list of error strings (empty = valid).""" errors = [] if not self.prompt: errors.append("prompt is required") if self.duration < 2 or self.duration > 10: errors.append(f"R2V duration must be 2-10s, got {self.duration}") if self.resolution not in ("720p", "1080p"): errors.append(f"resolution must be 720p or 1080p, got {self.resolution}") if self.aspect_ratio not in ("16:9", "9:16", "1:1", "4:3", "3:4"): errors.append(f"unsupported aspect_ratio: {self.aspect_ratio}") if not self.reference_image_urls and not self.reference_video_urls: errors.append("at least one reference_image_url or reference_video_url required") return errors # ====================================================================== # PromptPackage — everything needed for a single API call # ====================================================================== @dataclass class PromptPackage: """Complete generation package — everything needed to make one API call. References should already be weight-sorted ascending (lowest attention first, identity refs last) before being passed here. The assembler preserves this ordering when building the Parts list. Supports both image and video generation. For video, set modality="video" and provide duration_s and optional start_frame_path/end_frame_path. """ prompt_text: str references: list[ReferenceImage] # Weight-sorted (ascending) model: str # Model ID (e.g. "gemini-3-pro-image-preview") aspect_ratio: str # e.g. "9:16", "1:1" image_size: str # e.g. "4K", "1K" shot_id: Optional[str] = None shot_name: Optional[str] = None is_env: bool = False # True for ENV-only renders (no characters) grid_type: Optional[str] = None # GridType value if this is a grid call # From pipeline.py (consolidated): num_candidates: int = 1 # How many candidates to generate directives: list[str] = field(default_factory=list) # Behavioral directives # Video fields: modality: str = "image" # "image" or "video" duration_s: int = 0 # Video duration in seconds start_frame_path: Optional[str] = None # For I2V (Kling start_frame) end_frame_path: Optional[str] = None # For I2V (Kling end_frame) audio_prompt: Optional[str] = None # Audio/SFX direction camera_movement: Optional[str] = None # Track/crane/dolly description # PromptPackage router fields (ADR-R05): core_semantics: Optional[dict] = None # Unified semantic dict from compile_core_semantics compiled_prompts: Optional[dict] = None # All model-specific prompts from compile_all_prompts def estimated_cost(self) -> float: """Look up estimated cost from model_profiles.""" try: if self.modality == "video" and self.duration_s > 0: cost_per_sec = model_profiles.get_cost(self.model) return cost_per_sec * self.duration_s return model_profiles.get_cost(self.model) except KeyError: logger.warning("Unknown model '%s' — cannot estimate cost", self.model) return 0.0 def describe(self) -> str: """Human-readable summary of this package.""" lines = [] # Header shot_label = f"Shot {self.shot_id}" if self.shot_id is not None else "Shot ?" if self.shot_name: shot_label += f" ({self.shot_name})" lines.append(shot_label) # Model and format try: profile = model_profiles.get_profile(self.model) display = profile.get("display_name", self.model) except KeyError: display = self.model lines.append(f" Model: {display} [{self.model}]") lines.append(f" Modality: {self.modality}") lines.append(f" Aspect: {self.aspect_ratio}") lines.append(f" Size: {self.image_size}") # Video fields if self.modality == "video": lines.append(f" Duration: {self.duration_s}s") if self.start_frame_path: lines.append(f" Start frame: {self.start_frame_path}") if self.end_frame_path: lines.append(f" End frame: {self.end_frame_path}") if self.audio_prompt: lines.append(f" Audio: {self.audio_prompt}") if self.camera_movement: lines.append(f" Camera: {self.camera_movement}") # Grid type if self.grid_type: lines.append(f" Grid: {self.grid_type}") # Candidates if self.num_candidates > 1: lines.append(f" Candidates: {self.num_candidates}") # ENV flag if self.is_env: lines.append(" Type: ENV (environment only, no characters)") # Directives if self.directives: lines.append(" Directives:") for d in self.directives: lines.append(f" - {d}") # Cost cost = self.estimated_cost() lines.append(f" Cost: ${cost:.3f}") # References lines.append(f" Refs: {len(self.references)}") for ref in self.references: mirror_tag = " [M]" if ref.is_mirrored else "" exists_tag = "" if ref.path.exists() else " [MISSING]" lines.append( f" w={ref.weight:2d} {ref.label} " f"({ref.path.name}{mirror_tag}{exists_tag})" ) # Prompt (truncated for readability) prompt_preview = self.prompt_text[:200] if len(self.prompt_text) > 200: prompt_preview += "..." lines.append(f" Prompt: {prompt_preview}") return "\n".join(lines) # ====================================================================== # ShotAssembler — compile PromptPackage into API-ready payloads # ====================================================================== class ShotAssembler: """Compiles PromptPackage into google.genai API-ready Parts lists. Handles two API patterns: - genai_inline: Gemini models that accept interleaved image+text Parts - upload_bundle: Models (Kling etc.) that need files + prompt.txt on disk The genai_inline pattern orders Parts by weight ascending so that the recency bias puts identity refs (highest weight) closest to the prompt text, giving them maximum model attention. """ def to_genai_parts(self, package: PromptPackage) -> list: """Convert to google.genai Parts list. Order: refs sorted by weight ascending (lowest attention first), each ref image followed by its text label, then prompt text LAST. Returns list that can be passed directly to client.models.generate_content(contents=parts). Skips references whose image file does not exist on disk (with a warning), so the call can still proceed with available refs. Raises: RuntimeError: If google-genai is not installed. """ if not _HAS_GENAI: raise RuntimeError( "google-genai is not installed. " "Install with: pip install google-genai" ) parts: list = [] # References in weight order (ascending = lowest attention first) sorted_refs = sorted(package.references, key=lambda r: r.weight) for ref in sorted_refs: # Skip missing files gracefully if not ref.path.exists(): logger.warning( "Reference image not found, skipping: %s (label: %s)", ref.path, ref.label, ) continue # Load image bytes (handles mirroring in-memory) try: image_bytes = ref.load_bytes() except Exception as e: logger.warning( "Failed to load reference image %s: %s — skipping", ref.path.name, e, ) continue # Image Part parts.append( genai_types.Part.from_bytes( data=image_bytes, mime_type=ref.mime_type, ) ) # Label Part — helps Gemini distinguish reference types parts.append(genai_types.Part.from_text(text=ref.label)) # Behavioral directives right before the main prompt for directive in package.directives: parts.append(genai_types.Part.from_text(text=directive)) # ENV critical directive if package.is_env: parts.append(genai_types.Part.from_text( text="CRITICAL DIRECTIVE: Absolutely no humans, figures, or anatomy." )) # Prompt text is ALWAYS last — model pays most attention to the text # and the image Parts immediately preceding it parts.append(genai_types.Part.from_text(text=package.prompt_text)) return parts def to_genai_config(self, package: PromptPackage) -> dict: """Build the GenerateContentConfig kwargs for this package. Returns a dict that can be unpacked into types.GenerateContentConfig(**config) or passed as the config= parameter to client.models.generate_content(). Example usage: config = assembler.to_genai_config(package) response = client.models.generate_content( model=package.model, contents=parts, config=types.GenerateContentConfig(**config), ) """ config: dict = { "response_modalities": ["IMAGE", "TEXT"], } # Image config sub-object (SDK uses image_config / ImageConfig) image_cfg: dict = {} # Aspect ratio if package.aspect_ratio: image_cfg["aspect_ratio"] = package.aspect_ratio # Image size (Gemini supports "512px", "1K", "2K", "4K") if package.image_size: image_cfg["image_size"] = package.image_size if image_cfg: config["image_config"] = image_cfg return config def to_genai_image_payload(self, package: PromptPackage) -> GenaiPayload: """Build a complete GenaiPayload for image generation. Combines to_genai_parts() + to_genai_config() into a single ready-to-send payload object. """ return GenaiPayload( parts=self.to_genai_parts(package), config=self.to_genai_config(package), model=package.model, modality="image", ) def to_genai_video_payload(self, package: PromptPackage) -> GenaiPayload: """Build a GenaiPayload for Veo video generation. Sets response_modalities to VIDEO and includes duration config. Uses correct Veo config keys (video_generation_config, duration_seconds). Adds safety settings (BLOCK_ONLY_HIGH for all 4 categories). For pure T2V (ENV + complex camera), strips image Parts to avoid edge-smearing artifact on pullbacks/tracks. """ # Determine if this is a pure T2V call (ENV + complex camera) _COMPLEX_CAMERA = {"pan", "pull_back", "tracking", "crane", "track", "push_in"} is_pure_t2v = ( package.is_env and package.camera_movement and any(move in package.camera_movement.lower() for move in _COMPLEX_CAMERA) ) if is_pure_t2v: # Pure T2V: text Parts only — no image refs # Veo reference image forces I2V mode, destroying complex camera if not _HAS_GENAI: raise RuntimeError( "google-genai is not installed. " "Install with: pip install google-genai" ) parts = [] for directive in package.directives: parts.append(genai_types.Part.from_text(text=directive)) if package.is_env: parts.append(genai_types.Part.from_text( text="CRITICAL DIRECTIVE: Absolutely no humans, figures, or anatomy." )) parts.append(genai_types.Part.from_text(text=package.prompt_text)) else: parts = self.to_genai_parts(package) config = { "response_modalities": ["VIDEO"], "video_generation_config": { "duration_seconds": package.duration_s or 5, "aspect_ratio": package.aspect_ratio or "9:16", }, "safety_settings": [ {"category": "HARM_CATEGORY_DANGEROUS_CONTENT", "threshold": "BLOCK_ONLY_HIGH"}, {"category": "HARM_CATEGORY_HARASSMENT", "threshold": "BLOCK_ONLY_HIGH"}, {"category": "HARM_CATEGORY_HATE_SPEECH", "threshold": "BLOCK_ONLY_HIGH"}, {"category": "HARM_CATEGORY_SEXUALLY_EXPLICIT", "threshold": "BLOCK_ONLY_HIGH"}, ], } return GenaiPayload( parts=parts, config=config, model=package.model, modality="video", ) def to_kling_payload(self, package: PromptPackage) -> KlingPayload: """Build a KlingPayload from a PromptPackage. Extracts prompt, duration, optional start/end frame bytes, and camera_control from camera_movement strings (ADR-R10). Strips camera movement text from the prompt string. """ start_bytes = None if package.start_frame_path: start_path = Path(package.start_frame_path) if start_path.exists(): start_bytes = start_path.read_bytes() end_bytes = None if package.end_frame_path: end_path = Path(package.end_frame_path) if end_path.exists(): end_bytes = end_path.read_bytes() # Kling only accepts 5 or 10 seconds. Round up — human editors trim. raw_duration = package.duration_s or 5 kling_duration = 10 if raw_duration > 5 else 5 # Map camera_movement strings to structured camera_control floats camera_control = None prompt_text = package.prompt_text if package.camera_movement: camera_control = _map_camera_movement(package.camera_movement) # Strip camera movement text from prompt to avoid double-encoding prompt_text = _strip_camera_from_prompt(prompt_text, package.camera_movement) return KlingPayload( prompt=prompt_text, model=package.model, duration_s=kling_duration, start_frame_bytes=start_bytes, end_frame_bytes=end_bytes, aspect_ratio=package.aspect_ratio or "9:16", camera_control=camera_control, ) def to_seeddance_payload(self, package: PromptPackage) -> SeedDancePayload: """Build a SeedDancePayload from a PromptPackage. Loads reference image bytes for inline submission to fal.ai. Maps integer weights to SeedDance float weights (ADR-M04): Identity (w=8-10) -> 0.85, Prop (w=7-8) -> 0.50, Scene (w=1-2) -> 0.25, Keyframe -> 0.70 @Image1 ordering: primary character identity ref is enforced at index 0 of reference_images. SeedDance gives @Image1 40-50% of model attention (PROMPT_BIBLE best_practices). If identity refs are not already first, this method reorders and logs a warning. """ # Sort by weight ascending (standard ordering: scene first, identity last) sorted_refs = sorted(package.references, key=lambda r: r.weight) # Enforce @Image1 = primary identity ref for SeedDance sorted_refs = _enforce_seeddance_identity_first(sorted_refs) ref_bytes = [] ref_weights = [] for ref in sorted_refs: if ref.path.exists(): try: ref_bytes.append(ref.load_bytes()) ref_weights.append(_ref_to_upload_priority(ref)) except Exception as e: logger.warning("Failed to load ref %s for SeedDance: %s", ref.path.name, e) return SeedDancePayload( prompt=package.prompt_text, model=package.model, duration_s=package.duration_s or 5, reference_images=ref_bytes, reference_weights=ref_weights, aspect_ratio=package.aspect_ratio or "9:16", audio_prompt=package.audio_prompt, ) def to_bundle_manifest(self, package: PromptPackage) -> dict: """For upload_bundle models (Kling etc.) — package as files + prompt.txt. Returns a manifest dict describing what to write to disk: { "prompt_file": str, # prompt.txt content "reference_files": [ # list of dicts { "source_path": str, "label": str, "weight": int, "is_mirrored": bool, }, ... ], "model": str, "aspect_ratio": str, "shot_id": "int | None", "shot_name": "str | None", } The caller is responsible for writing these to a bundle directory and performing any mirroring (via load_bytes()) before upload. """ ref_entries = [] sorted_refs = sorted(package.references, key=lambda r: r.weight) for ref in sorted_refs: if not ref.path.exists(): logger.warning( "Bundle ref not found, skipping: %s", ref.path ) continue ref_entries.append({ "source_path": str(ref.path), "label": ref.label, "weight": ref.weight, "is_mirrored": ref.is_mirrored, }) return { "prompt_file": package.prompt_text, "reference_files": ref_entries, "model": package.model, "aspect_ratio": package.aspect_ratio, "shot_id": package.shot_id, "shot_name": package.shot_name, } # ------------------------------------------------------------------ # Diagnostics # ------------------------------------------------------------------ def describe_parts(self, package: PromptPackage) -> str: """Describe what the Parts list would contain without building it. Useful for zero-cost previews — does not require google-genai. """ lines = [] lines.append("Parts assembly order (top = least attention, bottom = most):") lines.append("-" * 60) sorted_refs = sorted(package.references, key=lambda r: r.weight) part_index = 0 for ref in sorted_refs: exists = ref.path.exists() status = "OK" if exists else "MISSING" mirror = " [mirrored]" if ref.is_mirrored else "" size_str = "" if exists: size_kb = ref.path.stat().st_size / 1024 size_str = f" ({size_kb:.0f} KB)" lines.append( f" [{part_index}] IMAGE w={ref.weight:2d} " f"{ref.path.name}{size_str}{mirror} [{status}]" ) part_index += 1 lines.append( f" [{part_index}] TEXT {ref.label}" ) part_index += 1 # Final prompt text prompt_preview = package.prompt_text[:120] if len(package.prompt_text) > 120: prompt_preview += "..." lines.append(f" [{part_index}] TEXT [PROMPT] {prompt_preview}") lines.append("-" * 60) lines.append( f"Total Parts: {part_index + 1} " f"({len(sorted_refs)} image + {len(sorted_refs)} label + 1 prompt)" ) return "\n".join(lines) # ====================================================================== # SeedDance Upload Priority (ADR-M04) # ====================================================================== # Maps ref_type to SeedDance float weight / upload priority (0.0-1.0). # Higher priority = uploaded first = gets @Image with lower index. _SEEDDANCE_UPLOAD_PRIORITY = { "identity": 0.85, # @Image1-2 (character hero + angles) "keyframe": 0.70, # @Image3 (if used as visual anchor) "prop": 0.50, # @Image4-5 "expression": 0.40, # @Image6 "scene": 0.25, # @Image7-8 (location/style) } def _ref_to_upload_priority(ref: 'ReferenceImage') -> float: """Map a ReferenceImage to a SeedDance upload priority float using ref_type.""" return _SEEDDANCE_UPLOAD_PRIORITY.get(ref.ref_type, 0.50) def _enforce_seeddance_identity_first(refs: list) -> list: """Ensure primary character identity ref is at index 0 for SeedDance. SeedDance gives @Image1 (index 0) 40-50% of model attention. The primary character identity ref MUST occupy this slot. If identity refs are not already first, reorder and log a warning. Only affects SeedDance payloads — other models use their own ordering. Args: refs: List of ReferenceImage objects (may be in any order). Returns: Reordered list with first identity ref at index 0. """ if not refs: return refs # Find the first identity ref first_identity_idx = None for i, ref in enumerate(refs): if ref.ref_type == "identity": first_identity_idx = i break if first_identity_idx is None: # No identity refs — nothing to enforce return refs if first_identity_idx == 0: # Already in correct position return refs # Reorder: move identity ref to index 0 reordered = list(refs) identity_ref = reordered.pop(first_identity_idx) reordered.insert(0, identity_ref) logger.warning( "SeedDance @Image1 reorder: moved identity ref '%s' from index %d to 0 " "(was behind %s ref '%s')", identity_ref.label, first_identity_idx, refs[0].ref_type, refs[0].label, ) return reordered def resolve_seedance_r2v_refs( prompt_text: str, ref_list: list, # list of ref objects with .path, .ref_type ) -> tuple[str, list[str]]: """Resolve @Image{ref_type_N} placeholders to @ImageN based on upload order. 1. Sort refs by _SEEDDANCE_UPLOAD_PRIORITY (highest first) 2. Assign @Image1 through @ImageN based on sorted position 3. Replace placeholder tokens in prompt_text 4. Return (resolved_prompt, ordered_file_paths) """ from collections import Counter # Sort by priority descending; within same type, preserve original order # (stable sort on priority alone preserves insertion order for ties) sorted_refs = sorted( ref_list, key=lambda r: _SEEDDANCE_UPLOAD_PRIORITY.get(r.ref_type, 0.50), reverse=True, ) # Track per-type counters for placeholder replacement type_counter: Counter = Counter() # Map (ref_type, occurrence_index) -> @ImageN placeholder_map: dict[str, str] = {} ordered_paths: list[str] = [] for image_index, ref in enumerate(sorted_refs, start=1): type_counter[ref.ref_type] += 1 occurrence = type_counter[ref.ref_type] # e.g. @Image{identity_1} -> @Image1, @Image{identity_2} -> @Image2 old_token = f"@Image{{{ref.ref_type}_{occurrence}}}" new_token = f"@Image{image_index}" placeholder_map[old_token] = new_token ordered_paths.append(str(ref.path)) # Apply replacements to prompt text resolved_prompt = prompt_text for old_token, new_token in placeholder_map.items(): resolved_prompt = resolved_prompt.replace(old_token, new_token) return resolved_prompt, ordered_paths # ====================================================================== # Camera Movement Mapping (ADR-R10) # ====================================================================== # Maps camera_movement strings to Kling camera_control float vectors. # Values are normalized 0.0-1.0 where 0.5 = neutral. _CAMERA_MOVEMENT_MAP = { "pan": {"pan": 0.7, "tilt": 0.5, "roll": 0.5, "zoom": 0.5}, "pan left": {"pan": 0.3, "tilt": 0.5, "roll": 0.5, "zoom": 0.5}, "pan right": {"pan": 0.7, "tilt": 0.5, "roll": 0.5, "zoom": 0.5}, "tilt": {"pan": 0.5, "tilt": 0.7, "roll": 0.5, "zoom": 0.5}, "tilt up": {"pan": 0.5, "tilt": 0.3, "roll": 0.5, "zoom": 0.5}, "tilt down": {"pan": 0.5, "tilt": 0.7, "roll": 0.5, "zoom": 0.5}, "push_in": {"pan": 0.5, "tilt": 0.5, "roll": 0.5, "zoom": 0.7}, "dolly in": {"pan": 0.5, "tilt": 0.5, "roll": 0.5, "zoom": 0.7}, "pull_back": {"pan": 0.5, "tilt": 0.5, "roll": 0.5, "zoom": 0.3}, "dolly out": {"pan": 0.5, "tilt": 0.5, "roll": 0.5, "zoom": 0.3}, "tracking": {"pan": 0.65, "tilt": 0.5, "roll": 0.5, "zoom": 0.5}, "crane": {"pan": 0.5, "tilt": 0.3, "roll": 0.5, "zoom": 0.55}, "crane up": {"pan": 0.5, "tilt": 0.3, "roll": 0.5, "zoom": 0.45}, "crane down": {"pan": 0.5, "tilt": 0.7, "roll": 0.5, "zoom": 0.55}, "handheld": {"pan": 0.52, "tilt": 0.52, "roll": 0.52, "zoom": 0.5}, "steadicam": {"pan": 0.55, "tilt": 0.5, "roll": 0.5, "zoom": 0.5}, "dolly": {"pan": 0.6, "tilt": 0.5, "roll": 0.5, "zoom": 0.5}, } # Words to strip from prompt text when camera_control is used _CAMERA_STRIP_PATTERNS = [ "panning", "tilting", "push-in", "pull-back", "tracking shot", "crane movement", "crane up", "crane down", "handheld", "Steadicam", "steadicam", "dolly movement", "dolly in", "dolly out", "camera pans", "camera tilts", "camera tracks", "camera pushes in", "camera pulls back", "camera cranes", ] def _map_camera_movement(movement: str) -> Optional[dict]: """Map a camera_movement string to structured Kling camera_control floats.""" if not movement or movement == "static": return None key = movement.lower().strip() # Direct match if key in _CAMERA_MOVEMENT_MAP: return dict(_CAMERA_MOVEMENT_MAP[key]) # Fuzzy match — check if any key is contained in the movement string for map_key, control in _CAMERA_MOVEMENT_MAP.items(): if map_key in key: return dict(control) # Default: slight pan for unknown movements return {"pan": 0.55, "tilt": 0.5, "roll": 0.5, "zoom": 0.5} def _strip_camera_from_prompt(prompt: str, movement: str) -> str: """Strip camera movement text from prompt when using camera_control. Avoids double-encoding (structured camera_control + text description). """ import re result = prompt for pattern in _CAMERA_STRIP_PATTERNS: result = result.replace(pattern, "") # Also strip the raw movement name if movement: movement_words = { "pan": "panning", "tilt": "tilting", "push_in": "push-in", "pull_back": "pull-back", "tracking": "tracking", "crane": "crane", "handheld": "handheld", "steadicam": "Steadicam", "dolly": "dolly", } for key, word in movement_words.items(): if key in movement.lower(): result = result.replace(f", {word}", "") result = result.replace(word, "") # Clean up artifacts result = re.sub(r",\s*,", ",", result) result = re.sub(r"\s{2,}", " ", result) result = re.sub(r",\s*\.", ".", result) return result.strip() __all__ = [ # Public symbols (Phase D — MF-3 + DEBT-9). # Payload dataclasses. "GenaiPayload", "KeyframeRefBundle", "KlingPayload", "MultiShotPayload", "PromptPackage", "SeedDancePayload", "SeedreamPayload", "WanI2VPayload", "WanR2VPayload", # Re-exported from recoil.execution.asset_manager (proxy callers depend on it). "ReferenceImage", # Assembler class. "ShotAssembler", # Module-level helpers. "allocate_references", "compile_keyframe_parts", "resolve_seedance_r2v_refs", ] # ====================================================================== # CLI demo — EP001 Shot 2 preview # ====================================================================== if __name__ == "__main__": import argparse import sys logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") parser = argparse.ArgumentParser( description="Preview assembled prompt packages (zero cost — no API calls)" ) parser.add_argument("--project", required=True, help="Project name (e.g. starsend-test, tartarus)") parser.add_argument("--episode", required=True, help="Episode ID (e.g. EP001, 1)") parser.add_argument("--shot", default=None, help="Shot ID (e.g. EP001_SH02). Omit for all-shots summary.") args = parser.parse_args() from recoil.core.paths import ensure_pipeline_importable ensure_pipeline_importable() from recoil.pipeline._lib.recoil_bridge import ( get_character_refs, get_location_refs, ) from recoil.execution.asset_manager import AssetManager from recoil.core.paths import ProjectPaths # Normalize episode ID ep_str = args.episode.upper() if not ep_str.startswith("EP"): ep_str = f"EP{int(ep_str):03d}" ep_num = ep_str.replace("EP", "") # Load shot plan plan_path = ProjectPaths.for_project(args.project).plans_dir / f"ep_{ep_num}_plan.json" if not plan_path.exists(): print(f"ERROR: Shot plan not found at {plan_path}") sys.exit(1) plan = json.loads(plan_path.read_text(encoding="utf-8")) all_shots = plan.get("shots", []) # Load bible for context bible_path = ProjectPaths.for_project(args.project).global_bible_path bible = json.loads(bible_path.read_text(encoding="utf-8")) if bible_path.exists() else {} # Filter to specific shot or show all if args.shot: target_shots = [s for s in all_shots if s.get("shot_id") == args.shot.upper()] if not target_shots: print(f"ERROR: Shot {args.shot} not found in {ep_str} plan") print(f"Available: {[s.get('shot_id') for s in all_shots]}") sys.exit(1) else: target_shots = all_shots print("=" * 60) print(f"ShotAssembler Preview — {args.project} {ep_str}") print(f"Shots: {len(target_shots)} of {len(all_shots)}") print("=" * 60) mgr = AssetManager() assembler = ShotAssembler() config_path = CONFIG_PATH pipeline_config = json.loads(config_path.read_text()) if config_path.exists() else {} default_model = pipeline_config.get("default_model", "gemini-3-pro-image-preview") for shot in target_shots: sid = shot.get("shot_id", "?") pd = shot.get("prompt_data") or {} ad = shot.get("asset_data") or {} cp = shot.get("compiled_prompts") or {} rd = shot.get("routing_data") or {} print() print(f"--- {sid} ---") print(f" Type: {pd.get('shot_type', '?')} | Focal: {pd.get('focal_length', '?')} | ENV: {rd.get('is_env_only', False)}") skeleton = pd.get("prompt_skeleton") or {} print(f" Subject: {skeleton.get('subject_line', '?')[:80]}") print(f" Emotion: {skeleton.get('emotion_line', '?')[:60]}") # Characters (plan format: asset_data.characters[].char_id) char_ids = [c.get("char_id", "") for c in ad.get("characters", []) if c.get("char_id")] print(f" Characters: {char_ids or '(none)'}") # Identity refs all_identity_refs: list[ReferenceImage] = [] max_refs_per_char = 2 if len(char_ids) <= 2 else 1 for char_id in char_ids: char_ref_paths = get_character_refs(char_id.lower(), project=args.project) identity_refs = mgr.get_identity_refs( char_id.upper(), char_ref_paths, max_refs=max_refs_per_char, ) all_identity_refs.extend(identity_refs) print(f" {char_id}: {len(identity_refs)} identity refs") # Location ref — shot-aware or legacy location_key = ad.get("location_id", "") scene_ref = None location_view_id = shot.get("location_view_id") if location_view_id is None and "location_view_id" in shot: print(" Location: (skipped — ECU, no location ref)") elif location_view_id: slug = location_key.lower() for prefix in ("int. ", "ext. ", "int/ext. "): if slug.startswith(prefix): slug = slug[len(prefix):] break for char in " -./": slug = slug.replace(char, "_") while "__" in slug: slug = slug.replace("__", "_") slug = slug.strip("_") view_path = ProjectPaths.for_project(args.project).asset_subject_dir("loc", slug) / location_view_id if view_path.exists(): scene_ref = mgr.get_scene_ref(view_path) print(f" Location: {view_path.name} (shot-aware, {view_path.stat().st_size // 1024}KB)") else: print(f" Location: MISSING — {location_view_id}") location_ref_paths = get_location_refs(location_key, project=args.project) if location_ref_paths: scene_ref = mgr.get_scene_ref(location_ref_paths[0]) print(f" Location: {scene_ref.path.name} (legacy fallback)") else: location_ref_paths = get_location_refs(location_key, project=args.project) if location_ref_paths: scene_ref = mgr.get_scene_ref(location_ref_paths[0]) print(f" Location: {scene_ref.path.name} (legacy)") else: print(" Location: (none)") # Expression ref — only for shots with characters (faces). # Attaching a face ref to an object-only shot risks bleeding # facial features into the generated image. emotion = skeleton.get("emotion_line", "") expression_ref = None if emotion and char_ids: expression_ref = mgr.get_expression_ref(emotion) if expression_ref: print(f" Expression: {expression_ref.path.name} (\"{emotion[:40]}\")") else: print(f" Expression: (no match for \"{emotion[:40]}\")") # Full ref stack all_refs = mgr.build_shot_refs( character_refs=all_identity_refs, scene_ref=scene_ref, expression_ref=expression_ref, ) print(f" Ref stack: {len(all_refs)} refs") # Show full detail for single-shot mode if args.shot: print() print(mgr.describe_shot_refs(all_refs)) # Prompt prompt_text = cp.get("keyframe_nbp") or cp.get("previs_flash", "(no compiled prompt)") package = PromptPackage( prompt_text=prompt_text, references=all_refs, model=default_model, aspect_ratio="9:16", image_size="4K", shot_id=sid, ) print("=" * 60) print("PROMPT PACKAGE") print("=" * 60) print(package.describe()) print() print("=" * 60) print("PARTS ASSEMBLY PREVIEW") print("=" * 60) print(assembler.describe_parts(package)) print() print("=" * 60) print("GENAI CONFIG") print("=" * 60) config = assembler.to_genai_config(package) print(json.dumps(config, indent=2)) # Summary if not args.shot: print() print("=" * 60) print("SUMMARY") print("=" * 60) total_refs = 0 for shot in target_shots: ad = shot.get("asset_data") or {} char_count = len([c for c in ad.get("characters", []) if c.get("char_id")]) has_loc = shot.get("location_view_id") is not None or "location_view_id" not in shot ref_est = char_count * 2 + (1 if has_loc else 0) + 1 # chars + loc + expression total_refs += ref_est print(f" Shots: {len(target_shots)}") print(f" Est. total refs: ~{total_refs}") print(f" Est. cost (NBP): ${len(target_shots) * 0.134:.2f}") print(f" Est. cost (Flash previz): ${len(target_shots) * 0.039:.2f}") print() stats = mgr.cache_stats() print(f"AssetManager cache: {stats['entries']} entries, {stats['total_bytes'] / 1024:.0f} KB")