""" blocking_pass.py — Stage 2.5: Physical Blocking Pass. Generates physical blocking descriptions for every shot in an episode. Processes per-scene: reads the scene's source text + bible + Stage 2 plan, produces a Scene Blocking Document (SBD), then decomposes into per-shot freeze-frame blocking. Single-call architecture: one Gemini Pro call per scene outputs both the SBD and per-shot blocking in a single structured JSON response. Designed for easy split into two calls if quality requires it. Usage: from orchestrator.blocking_pass import BlockingPass bp = BlockingPass(project="tartarus", project_root=Path("../projects/tartarus")) result = bp.run(episode_num=1) # result = bp.run(episode_num=1, scene_indices=[3, 5]) # re-run specific scenes """ from __future__ import annotations import hashlib import json import logging import os import sys import time from datetime import datetime, timezone from pathlib import Path from typing import Optional _PROJECT_ROOT = Path(__file__).parent.parent if str(_PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(_PROJECT_ROOT)) from recoil.core.model_profiles import get_model from recoil.pipeline._lib.render_schema import ( BlockingMetadata, CharacterBlocking, CoverageAnchor, EpisodePlan, GazeSequenceEntry, GlobalBible, HandState, PropLedgerEntry, PropTransition, SceneBlockingDocument, ShotRecord, ) logger = logging.getLogger(__name__) BLOCKING_MODEL = get_model("pro", "text") # --------------------------------------------------------------------------- # System Prompt # --------------------------------------------------------------------------- BLOCKING_SYSTEM_PROMPT = """\ You are a Script Supervisor on a film set. Your job is to generate physical \ blocking notes for every shot in a scene. You will receive: 1. A scene's worth of shots (grouped by scene_index) with source_text, \ shot_type, characters, props, dialogue, and spatial data. 2. Relevant Global Bible entries (character builds, props with states). You MUST output a JSON object with TWO sections: A) scene_blocking — a continuous prose choreography of the scene's physical \ action, plus structured prop tracking and gaze sequences. B) shot_blocking — a per-shot array of freeze-frame blocking records. BLOCKING RULES: 1. BODY POSITION, NOT APPEARANCE. Never mention hair color, clothing style, \ skin texture, or facial features. Those come from the character bible later. \ You describe WHERE the body is and WHAT it is doing. 2. LEFT/RIGHT SPECIFICITY. Always specify which hand holds which prop, which \ foot bears weight, which direction the head turns. 3. PROP TRACKING. Every prop in the scene must be accounted for in every shot. \ If a character picks up a prop in shot 3, they must still have it in shot 4 \ unless the narrative specifies they set it down. 4. GAZE TRACKING. Every character is always looking at something specific. \ Name the target (character name, prop, location landmark, "floor", "distance"). 5. INFER NATURAL POSITIONS. If the source text says "she checks the console," \ infer: standing, facing console, one or both hands on console surface, gaze at \ console. Write the full body position even when the source text only implies it. 6. CONTINUITY. A character's position at the END of one shot must match their \ position at the START of the next shot in the same scene, unless the source \ text describes movement between them. 7. SHOT-TYPE FILTERING. Adjust blocking detail to match what the camera sees: - WS/LS: Full body. Stance, all limb positions, spatial relationship \ between characters. - FS/MS: Torso up. Hand positions, arm positions, prop grip. Omit feet. - MCU/CU: Shoulders and head. Gaze direction is critical. Mention hands \ only if near face/head or holding a prominent prop. - ECU: Face or detail only. Gaze direction. No body position. If INSERT \ of a prop, describe prop state and hand contact. - ENV shots (is_env_only=true): No character blocking. Set characters to \ empty array. The compiled_subject_line MUST still describe the spatial \ arrangement of objects and set pieces visible in frame — treat it as camera \ blocking. Example: "heavy brushed steel cryo-pod suspended by thick anchor \ cables, hanging over vast dark maintenance shaft, bottomless drop below." \ Be vivid and action-oriented, describing physical relationships and states. 8. OFF-SCREEN CHARACTERS. Characters with visibility=off_screen should NOT \ receive physical blocking. Only generate blocking for in_frame characters. 9. CAMERA SIDE COMPLIANCE. If spatial_data.camera_side is "A" and the scene \ has two characters, describe the blocking from camera side A's perspective. \ Character screen positions must match spatial_data.screen_direction. 10. GAZE DETERMINES 180-LINE. If character A looks right at character B, and \ the camera is on side A, character A should be facing screen-right. If any \ shot's spatial_data contradicts this, set axis_violation=true. SUBJECT_LINE GRAMMAR: [Character] [stance] at [landmark], [dominant hand action], \ [secondary hand action], [head/gaze toward target] Multi-character: separate with semicolons. Example: "Torch crouched at corridor junction, right hand gripping salvage \ hook handle, left hand bracing against bulkhead, head turned back over right \ shoulder toward darkness" COVERAGE ANCHORS: Identify 1-3 moments in the scene that are significant for multi-camera \ coverage: establishing frames where all characters are visible, key dramatic \ beats where camera cuts are likely, reveal moments.""" # --------------------------------------------------------------------------- # Blocking Pass Output Schema (for Gemini response_schema) # --------------------------------------------------------------------------- def _build_blocking_output_schema() -> dict: """Build the JSON schema for the blocking pass output. This combines the SBD + per-shot blocking in a single structured output so Gemini generates both in one call (native CoT). """ return { "type": "object", "properties": { "scene_blocking": { "type": "object", "description": "Scene-level continuous blocking — write this FIRST as Chain-of-Thought before per-shot decomposition.", "properties": { "blocking_narrative": { "type": "string", "description": "Continuous prose choreography. Script supervisor blocking notes in present tense.", }, "starting_positions": { "type": "string", "description": "Where each character is positioned when the scene begins.", }, "prop_ledger": { "type": "array", "items": { "type": "object", "properties": { "prop_id": {"type": "string"}, "initial_holder": {"type": ["string", "null"]}, "transitions": { "type": "array", "items": { "type": "object", "properties": { "at_shot_approx": {"type": "string"}, "state": {"type": "string"}, }, "required": ["at_shot_approx", "state"], }, }, }, "required": ["prop_id"], }, }, "gaze_sequence": { "type": "array", "items": { "type": "object", "properties": { "character": {"type": "string"}, "sequence": { "type": "array", "items": {"type": "string"}, }, }, "required": ["character", "sequence"], }, }, "coverage_anchors": { "type": "array", "items": { "type": "object", "properties": { "moment": {"type": "string"}, "description": {"type": "string"}, "approximate_shot": {"type": "string"}, "coverage_note": {"type": "string"}, }, "required": ["moment", "description"], }, }, }, "required": ["blocking_narrative", "starting_positions", "prop_ledger", "gaze_sequence"], }, "shot_blocking": { "type": "array", "description": "Per-shot freeze-frame blocking, derived from the scene_blocking above.", "items": { "type": "object", "properties": { "shot_id": {"type": "string"}, "is_env_shot": {"type": "boolean"}, "compiled_subject_line": { "type": "string", "description": "The final string that replaces prompt_skeleton.subject_line.", }, "characters": { "type": "array", "items": { "type": "object", "properties": { "character_id": {"type": "string"}, "stance": {"type": "string"}, "torso_facing": {"type": "string"}, "head_facing": {"type": "string"}, "gaze_target": {"type": "string"}, "dominant_hand": { "type": "object", "properties": { "hand": {"type": "string"}, "action": {"type": "string"}, "target": {"type": ["string", "null"]}, "prop_id": {"type": ["string", "null"]}, }, "required": ["hand", "action"], }, "secondary_hand": { "type": "object", "properties": { "hand": {"type": "string"}, "action": {"type": "string"}, "target": {"type": ["string", "null"]}, "prop_id": {"type": ["string", "null"]}, }, "required": ["hand", "action"], }, "weight_bearing": {"type": ["string", "null"]}, }, "required": [ "character_id", "stance", "torso_facing", "head_facing", "gaze_target", "dominant_hand", "secondary_hand", ], }, }, "prop_states": { "type": "object", "description": "prop_id → state string for this shot", }, "axis_violation": {"type": "boolean"}, }, "required": ["shot_id", "is_env_shot", "compiled_subject_line", "characters", "prop_states"], }, }, }, "required": ["scene_blocking", "shot_blocking"], } # --------------------------------------------------------------------------- # BlockingPass # --------------------------------------------------------------------------- class BlockingPass: """Stage 2.5: Physical Blocking Pass. Reads an episode plan + bible, groups shots by scene_index, generates blocking per scene, writes back to the plan JSON. """ def __init__( self, project: str, project_root: Path | None = None, dry_run: bool = False, model: str | None = None, ): self.project = project self.dry_run = dry_run self.model = model or BLOCKING_MODEL self._client = None from recoil.core.paths import ProjectPaths state_root = ProjectPaths.for_project(project).visual_state_dir self.plans_dir = state_root / "plans" self.bible_path = state_root / "global_bible.json" self.blocking_dir = state_root / "blocking" @property def client(self): """Lazy-init Gemini client.""" if self._client is None: from google import genai api_key = os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY") if not api_key: raise RuntimeError("GOOGLE_API_KEY or GEMINI_API_KEY not set") self._client = genai.Client(api_key=api_key) return self._client def _call_gemini( self, system_prompt: str, user_prompt: str, response_schema: dict | None = None, temperature: float = 0.3, max_tokens: int = 65536, ) -> str | None: """Call Gemini with optional response_schema enforcement.""" from google.genai import types as genai_types config = genai_types.GenerateContentConfig( temperature=temperature, max_output_tokens=max_tokens, system_instruction=system_prompt, ) if response_schema: config.response_mime_type = "application/json" config.response_json_schema = response_schema char_count = len(system_prompt) + len(user_prompt) logger.info(f"Blocking pass: calling {self.model} ({char_count:,} chars input)") if self.dry_run: logger.info(f"[DRY RUN] Would call {self.model} ({char_count:,} chars)") return None t0 = time.time() response = self.client.models.generate_content( model=self.model, contents=user_prompt, config=config, ) elapsed = time.time() - t0 text = response.text if hasattr(response, "text") else str(response) logger.info(f"Blocking pass response: {len(text):,} chars in {elapsed:.1f}s") return text def _load_plan(self, episode_num: int) -> dict: """Load an episode plan as a raw dict (for mutation).""" path = self.plans_dir / f"ep_{episode_num:03d}_plan.json" if not path.exists(): raise FileNotFoundError(f"Plan not found: {path}") return json.loads(path.read_text()) def _save_plan(self, episode_num: int, plan_dict: dict) -> None: """Save the mutated plan dict back to disk.""" path = self.plans_dir / f"ep_{episode_num:03d}_plan.json" path.write_text(json.dumps(plan_dict, indent=2, default=str)) logger.info(f"Saved updated plan: {path}") def _load_bible(self) -> GlobalBible: """Load the global bible.""" if not self.bible_path.exists(): raise FileNotFoundError(f"Bible not found: {self.bible_path}") return GlobalBible.model_validate_json(self.bible_path.read_text()) def _group_shots_by_scene(self, shots: list[dict]) -> dict[int, list[dict]]: """Group shot dicts by scene_index.""" scenes: dict[int, list[dict]] = {} for shot in shots: si = shot.get("scene_index", 1) scenes.setdefault(si, []).append(shot) return scenes def _build_scene_context( self, scene_shots: list[dict], bible: GlobalBible, episode_num: int, ) -> str: """Build the user prompt for one scene's blocking pass call.""" parts = [] # Scene metadata scene_index = scene_shots[0].get("scene_index", 1) parts.append(f"# SCENE {scene_index} — {len(scene_shots)} shots\n") # Shots for shot in scene_shots: shot_id = shot.get("shot_id", "?") source_text = shot.get("source_text", "") shot_type = shot.get("prompt_data", {}).get("shot_type", "MS") is_env = shot.get("routing_data", {}).get("is_env_only", False) camera_side = shot.get("spatial_data", {}).get("camera_side", "A") screen_dir = shot.get("spatial_data", {}).get("screen_direction", "center") # Characters in shot (only in_frame — skip off_screen) chars = shot.get("asset_data", {}).get("characters", []) char_ids = [ c.get("char_id", c) if isinstance(c, dict) else c for c in chars if not isinstance(c, dict) or c.get("visibility", "in_frame") == "in_frame" ] # Props in shot props = shot.get("asset_data", {}).get("props", []) prop_ids = [p.get("prop_id", p) if isinstance(p, dict) else p for p in props] prop_interaction = shot.get("asset_data", {}).get("prop_interaction", "none") # Dialogue dialogue = shot.get("audio_data", {}).get("dialogue", []) dialogue_text = "" if dialogue: lines = [] for d in dialogue: if isinstance(d, dict): lines.append(f"{d.get('character', '?')}: \"{d.get('text', '')}\"") dialogue_text = " | ".join(lines) parts.append( f"[{shot_id}] shot_type={shot_type}, is_env={is_env}, " f"camera_side={camera_side}, screen_direction={screen_dir}\n" f" characters: {char_ids}\n" f" props: {prop_ids} (interaction: {prop_interaction})\n" f" dialogue: {dialogue_text or 'none'}\n" f" source_text: {source_text}\n" ) # Bible entries for characters in this scene all_char_ids = set() all_prop_ids = set() for shot in scene_shots: chars = shot.get("asset_data", {}).get("characters", []) for c in chars: cid = c.get("char_id", c) if isinstance(c, dict) else c all_char_ids.add(cid) props = shot.get("asset_data", {}).get("props", []) for p in props: pid = p.get("prop_id", p) if isinstance(p, dict) else p all_prop_ids.add(pid) parts.append("\n# CHARACTER BIBLE ENTRIES\n") for char_id in sorted(all_char_ids): char = bible.characters.get(char_id) if char: phase = char.phase_for_episode(episode_num) parts.append( f"- {char_id} ({char.display_name}): " f"build={char.visual_description[:100]}... " ) if phase: # Include associated props from wardrobe parts.append(f" wardrobe hints: {phase.wardrobe_description[:300]}...") parts.append("\n# PROP BIBLE ENTRIES\n") for prop_id in sorted(all_prop_ids): prop = bible.props.get(prop_id) if prop: parts.append( f"- {prop_id}: {prop.description} " f"(associated: {prop.associated_characters})" ) return "\n".join(parts) def _parse_blocking_response( self, raw: str, scene_index: int, scene_shots: list[dict], ) -> tuple[SceneBlockingDocument, list[dict]]: """Parse the Gemini response into SBD + per-shot blocking updates.""" data = json.loads(raw) sb = data.get("scene_blocking", {}) sbd = SceneBlockingDocument( scene_index=scene_index, characters_present=[], # filled below scene_duration_shots=len(scene_shots), blocking_narrative=sb.get("blocking_narrative", ""), starting_positions=sb.get("starting_positions", ""), prop_ledger=[ PropLedgerEntry( prop_id=p.get("prop_id", ""), initial_holder=p.get("initial_holder"), transitions=[ PropTransition(**t) for t in p.get("transitions", []) ], ) for p in sb.get("prop_ledger", []) ], gaze_sequence=[ GazeSequenceEntry(**g) for g in sb.get("gaze_sequence", []) ], coverage_anchors=[ CoverageAnchor(**ca) for ca in sb.get("coverage_anchors", []) ], ) # Collect characters from scene shots char_ids = set() for shot in scene_shots: for c in shot.get("asset_data", {}).get("characters", []): cid = c.get("char_id", c) if isinstance(c, dict) else c char_ids.add(cid) sbd.characters_present = sorted(char_ids) # Location from first shot first_loc = scene_shots[0].get("asset_data", {}).get("location_id", "") sbd.location_id = first_loc # Parse per-shot blocking shot_updates = [] sbd_hash = hashlib.sha256(sbd.blocking_narrative.encode()).hexdigest()[:16] now = datetime.now(timezone.utc).isoformat() for sb_shot in data.get("shot_blocking", []): shot_id = sb_shot.get("shot_id", "") is_env = sb_shot.get("is_env_shot", False) characters = [] for c in sb_shot.get("characters", []): dh = c.get("dominant_hand", {}) sh = c.get("secondary_hand", {}) characters.append(CharacterBlocking( character_id=c.get("character_id", ""), stance=c.get("stance", "standing"), torso_facing=c.get("torso_facing", "camera"), head_facing=c.get("head_facing", "camera"), gaze_target=c.get("gaze_target", "camera"), dominant_hand=HandState( hand=dh.get("hand", "right"), action=dh.get("action", "hanging_at_side"), target=dh.get("target"), prop_id=dh.get("prop_id"), ), secondary_hand=HandState( hand=sh.get("hand", "left"), action=sh.get("action", "hanging_at_side"), target=sh.get("target"), prop_id=sh.get("prop_id"), ), weight_bearing=c.get("weight_bearing"), )) blocking_meta = BlockingMetadata( characters=characters, prop_states=sb_shot.get("prop_states", {}), scene_blocking_hash=sbd_hash, axis_violation=sb_shot.get("axis_violation", False), blocking_pass_model=self.model, blocking_pass_timestamp=now, ) shot_updates.append({ "shot_id": shot_id, "is_env_shot": is_env, "compiled_subject_line": sb_shot.get("compiled_subject_line", ""), "blocking_metadata": blocking_meta.model_dump(mode="json"), }) return sbd, shot_updates def _save_sbd(self, episode_num: int, sbd: SceneBlockingDocument) -> Path: """Save an SBD as a sidecar JSON file.""" ep_dir = self.blocking_dir / f"ep_{episode_num:03d}" ep_dir.mkdir(parents=True, exist_ok=True) path = ep_dir / f"scene_{sbd.scene_index:03d}_sbd.json" path.write_text(sbd.model_dump_json(indent=2)) logger.info(f"Saved SBD: {path}") return path def _save_manifest( self, episode_num: int, sbds: list[SceneBlockingDocument], violations: list[str], ) -> None: """Save the blocking manifest index.""" ep_dir = self.blocking_dir / f"ep_{episode_num:03d}" ep_dir.mkdir(parents=True, exist_ok=True) manifest = { "episode_id": f"EP{episode_num:03d}", "blocking_pass_version": "1.0", "generated_at": datetime.now(timezone.utc).isoformat(), "model": self.model, "scenes": [ { "scene_index": sbd.scene_index, "sbd_file": f"scene_{sbd.scene_index:03d}_sbd.json", "sbd_hash": hashlib.sha256( sbd.blocking_narrative.encode() ).hexdigest()[:16], "shot_count": sbd.scene_duration_shots, "characters": sbd.characters_present, } for sbd in sbds ], "violations": violations, } path = ep_dir / "blocking_manifest.json" path.write_text(json.dumps(manifest, indent=2)) logger.info(f"Saved manifest: {path}") def run( self, episode_num: int, scene_indices: list[int] | None = None, ) -> dict: """Run the blocking pass for an episode. Args: episode_num: Episode number to process. scene_indices: Optional list of specific scene indices to (re-)process. If None, processes all scenes. Returns: Dict with stats: scenes_processed, shots_blocked, violations. """ episode_id = f"EP{episode_num:03d}" logger.info(f"=== Stage 2.5: Blocking Pass {episode_id} ===") plan_dict = self._load_plan(episode_num) bible = self._load_bible() shots = plan_dict.get("shots", []) scenes = self._group_shots_by_scene(shots) # Filter to requested scenes if scene_indices: scenes = {si: s for si, s in scenes.items() if si in scene_indices} sbds = [] all_shot_updates = [] total_blocked = 0 for scene_index in sorted(scenes.keys()): scene_shots = scenes[scene_index] logger.info( f"Processing scene {scene_index} " f"({len(scene_shots)} shots)" ) # Build context user_prompt = self._build_scene_context( scene_shots, bible, episode_num ) # Call Gemini schema = _build_blocking_output_schema() raw = self._call_gemini( system_prompt=BLOCKING_SYSTEM_PROMPT, user_prompt=user_prompt, response_schema=schema, ) if raw is None: logger.info(f"[DRY RUN] Scene {scene_index} skipped") continue # Parse response sbd, shot_updates = self._parse_blocking_response( raw, scene_index, scene_shots ) sbds.append(sbd) all_shot_updates.extend(shot_updates) # Save SBD sidecar self._save_sbd(episode_num, sbd) total_blocked += len(shot_updates) # Apply shot updates to plan dict shot_map = {s.get("shot_id"): s for s in shots} for update in all_shot_updates: shot = shot_map.get(update["shot_id"]) if shot is None: logger.warning( f"Shot {update['shot_id']} not found in plan — skipping" ) continue # Overwrite subject_line if update["compiled_subject_line"]: shot.setdefault("prompt_data", {}).setdefault( "prompt_skeleton", {} )["subject_line"] = update["compiled_subject_line"] # Set blocking_metadata shot["blocking_metadata"] = update["blocking_metadata"] # Run validators violations = [] try: from recoil.pipeline._lib.blocking_validators import run_all_validators violations = run_all_validators(plan_dict, bible) except ImportError: logger.warning("blocking_validators not found — skipping validation") # Check shot_id match rate — error if too few matched if all_shot_updates: matched = sum(1 for u in all_shot_updates if u["shot_id"] in shot_map) if matched < len(all_shot_updates) * 0.5: logger.error( f"Only {matched}/{len(all_shot_updates)} shot_ids matched plan — " f"possible Gemini shot_id format mismatch" ) violations.append( f"[HARD] Shot ID match rate too low: {matched}/{len(all_shot_updates)}" ) # Gate on hard violations — do NOT save corrupted plan hard_violations = [v for v in violations if v.startswith("[HARD]")] if hard_violations: logger.error( f"Blocking pass has {len(hard_violations)} hard violations — " f"plan NOT saved. Fix violations and re-run." ) for hv in hard_violations: logger.error(f" {hv}") else: # Save updated plan only if no hard violations self._save_plan(episode_num, plan_dict) # Save manifest self._save_manifest(episode_num, sbds, violations) result = { "episode_id": episode_id, "scenes_processed": len(sbds), "shots_blocked": total_blocked, "violations": violations, "violation_count": len(violations), "blocked": bool(hard_violations), } logger.info( f"Blocking pass complete: {result['scenes_processed']} scenes, " f"{result['shots_blocked']} shots blocked, " f"{result['violation_count']} violations" ) return result