#!/usr/bin/env python3 """ validate_breakdown.py — Script Breakdown Verification Gate Hard gate for breakdown.json — continuity errors here propagate through all downstream production (storyboard → frames → video). Usage: python3 validate_breakdown.py /leviathan/visual/breakdown.json /leviathan/ python3 validate_breakdown.py /leviathan/visual/breakdown.json /leviathan/ --report python3 validate_breakdown.py /leviathan/visual/breakdown.json /leviathan/ --prompt python3 validate_breakdown.py /leviathan/visual/breakdown.json /leviathan/ --json Exit codes: 0 = clean, 1 = hard errors (FAIL), 2 = warnings (matches CONSTANTS.md) """ # ╔════════════════════════════════════════════════════════════════════╗ # ║ DEPRECATED — Superseded by Starsend equivalents (Feb 2026). ║ # ║ Kept alive for Recoil agent protocols + referencing scripts. ║ # ║ Do NOT delete until agents/breakdown_agent.md, storyboard_agent, ║ # ║ engine_checks/structural.py, and batch_threepass.py are updated. ║ # ╚════════════════════════════════════════════════════════════════════╝ import argparse import json import re import sys from pathlib import Path from dataclasses import dataclass from typing import List, Optional, Set # ── Data Structures ──────────────────────────────────────────────────────── @dataclass class ValidationResult: tier: int # 1=hard error, 2=warning, 3=info category: str message: str fix_hint: Optional[str] = None # ── Path Resolution ──────────────────────────────────────────────────────── def resolve_project_path(project_arg: str) -> Path: """Resolve project path from argument.""" script_dir = Path(__file__).resolve().parent if script_dir.name == "tools" and script_dir.parent.name == "recoil": root = script_dir.parent.parent else: root = Path.cwd() project_name = project_arg.strip("/").strip("\\") candidate = root / project_name if candidate.is_dir(): return candidate abs_path = Path(project_arg) if abs_path.is_dir(): return abs_path cwd_path = Path.cwd() / project_name if cwd_path.is_dir(): return cwd_path print(f"ERROR: Project directory not found: {project_arg}", file=sys.stderr) sys.exit(2) # ── Validators ───────────────────────────────────────────────────────────── def validate_tier1(breakdown: dict, project_path: Path) -> List[ValidationResult]: """ Tier 1: Hard Errors (exit code 1 — blocks production) """ results = [] ep_range = breakdown.get("episode_range", [0, 0]) ep_start, ep_end = ep_range[0], ep_range[1] expected_count = ep_end - ep_start + 1 # 1. All episodes processed (no gaps) actual_count = breakdown.get("episodes_processed", 0) if actual_count < expected_count: results.append(ValidationResult( tier=1, category="episode_coverage", message=f"Episode gap: expected {expected_count} episodes ({ep_start}-{ep_end}), got {actual_count}", fix_hint=f"Re-run script_breakdown.py with --episodes {ep_start}-{ep_end} to fill gaps" )) # 2. Every character in characters.md appears in breakdown chars_file = project_path / "bible" / "characters.md" if chars_file.exists(): chars_text = chars_file.read_text(encoding="utf-8") # Find character section headers char_names_in_md = set() for m in re.finditer(r'\n## ([A-Z][A-Z\s]+?)(?:\s*—|\s*$)', chars_text): name = m.group(1).strip().split()[0] if name not in ("PURPOSE", "VOICE", "CHARACTER", "VALIDATION"): char_names_in_md.add(name) breakdown_chars = set(breakdown.get("characters", {}).keys()) missing = char_names_in_md - breakdown_chars if missing: results.append(ValidationResult( tier=1, category="character_coverage", message=f"Characters in characters.md missing from breakdown: {', '.join(sorted(missing))}", fix_hint="These characters must appear in at least one episode or be added manually" )) # 3. Every character has non-empty visual description for char_key, char_data in breakdown.get("characters", {}).items(): if not char_data.get("visual_description", "").strip(): results.append(ValidationResult( tier=1, category="character_visual", message=f"Character {char_key} has no visual description", fix_hint=f"Add visual_description to {char_key} from characters.md Visual Design section" )) # 4. Every location has at least one description sample for loc_key, loc_data in breakdown.get("locations", {}).items(): samples = loc_data.get("description_samples", []) if not samples: results.append(ValidationResult( tier=1, category="location_description", message=f"Location '{loc_key}' has no description samples", fix_hint=f"Add at least one description sample from episode action blocks" )) # 5. Wardrobe phases cover full episode range per character for char_key, char_data in breakdown.get("characters", {}).items(): wardrobe = char_data.get("wardrobe", {}) if not wardrobe: continue # No wardrobe data is not a hard error by itself char_eps = char_data.get("episodes", []) if not char_eps: continue covered_eps = set() for phase_name, phase_data in wardrobe.items(): ep_span = phase_data.get("episodes", []) if len(ep_span) == 2: for ep in range(ep_span[0], ep_span[1] + 1): covered_eps.add(ep) uncovered = set(char_eps) - covered_eps if uncovered and len(wardrobe) > 0: results.append(ValidationResult( tier=1, category="wardrobe_coverage", message=f"Character {char_key}: wardrobe phases don't cover episodes {sorted(uncovered)[:10]}{'...' if len(uncovered) > 10 else ''}", fix_hint=f"Extend or add wardrobe phases to cover all episodes where {char_key} appears" )) # 6. Story day progression is monotonic timeline = breakdown.get("story_timeline", {}) ep_to_days = timeline.get("episodes_to_days", {}) if ep_to_days: prev_day = 0 for ep_str in sorted(ep_to_days.keys(), key=lambda x: int(x)): day = ep_to_days[ep_str] if isinstance(day, int) and day < prev_day: results.append(ValidationResult( tier=1, category="timeline", message=f"Timeline backwards: episode {ep_str} is day {day}, but previous episode was day {prev_day}", fix_hint="Fix story_timeline.episodes_to_days so days increase monotonically" )) if isinstance(day, int): prev_day = day return results def validate_tier2(breakdown: dict, project_path: Path) -> List[ValidationResult]: """ Tier 2: Continuity Warnings (exit code 2 — requires review) """ results = [] # 1. Character state changes match characters.md timeline for char_key, char_data in breakdown.get("characters", {}).items(): state_changes = char_data.get("state_changes", []) char_eps = char_data.get("episodes", []) for sc in state_changes: sc_ep = sc.get("episode", 0) if sc_ep > 0 and char_eps and sc_ep > max(char_eps): results.append(ValidationResult( tier=2, category="state_continuity", message=f"{char_key}: state change at ep {sc_ep} but last appearance is ep {max(char_eps)}", fix_hint=f"Either extend {char_key}'s episode range or remove the state change" )) # 2. Props owned by characters appear in those characters' episodes for prop_key, prop_data in breakdown.get("props", {}).items(): owner = prop_data.get("owner") if not owner: continue owner_data = breakdown.get("characters", {}).get(owner) if not owner_data: continue prop_eps = set(prop_data.get("episodes", [])) owner_eps = set(owner_data.get("episodes", [])) orphan_eps = prop_eps - owner_eps if orphan_eps: results.append(ValidationResult( tier=2, category="prop_ownership", message=f"Prop '{prop_data['display_name']}' (owner: {owner}) appears in episodes without owner: {sorted(orphan_eps)[:5]}", fix_hint=f"Verify if {owner} is present in those episodes or if another character uses this prop" )) # 3. Location descriptions consistent across episodes for loc_key, loc_data in breakdown.get("locations", {}).items(): samples = loc_data.get("description_samples", []) if len(samples) >= 2: # Simple check: if samples share < 20% words, flag words_sets = [set(s.lower().split()) for s in samples] if len(words_sets) >= 2: overlap = words_sets[0] & words_sets[1] union = words_sets[0] | words_sets[1] if union and len(overlap) / len(union) < 0.1: results.append(ValidationResult( tier=2, category="location_consistency", message=f"Location '{loc_key}' has very different descriptions across episodes", fix_hint="Review description samples for visual consistency" )) # 4. Story day gaps > 5 without wardrobe/state update timeline = breakdown.get("story_timeline", {}) ep_to_days = timeline.get("episodes_to_days", {}) if ep_to_days: sorted_eps = sorted(ep_to_days.keys(), key=lambda x: int(x)) for i in range(1, len(sorted_eps)): prev_ep = sorted_eps[i - 1] curr_ep = sorted_eps[i] prev_day = ep_to_days[prev_ep] curr_day = ep_to_days[curr_ep] if isinstance(prev_day, int) and isinstance(curr_day, int): gap = curr_day - prev_day if gap > 5: results.append(ValidationResult( tier=2, category="story_day_gap", message=f"Large story day gap: {gap} days between ep {prev_ep} (day {prev_day}) and ep {curr_ep} (day {curr_day})", fix_hint="Check if characters need wardrobe/state updates for this time gap" )) # 5. Shot estimates within expected range se = breakdown.get("shot_estimates", {}) shots_per_ep = se.get("shots_per_episode", 0) if shots_per_ep < 18 or shots_per_ep > 24: results.append(ValidationResult( tier=2, category="shot_estimates", message=f"Shots per episode estimate ({shots_per_ep}) outside expected range (18-24)", fix_hint="Review shot estimation logic or episode complexity" )) return results def validate_tier3(breakdown: dict, project_path: Path) -> List[ValidationResult]: """ Tier 3: Completeness Checks (informational) """ results = [] # 1. Prompt fields populated for char_key, char_data in breakdown.get("characters", {}).items(): prompts = char_data.get("prompts", {}) has_reference = bool(prompts.get("reference")) if not has_reference: results.append(ValidationResult( tier=3, category="prompts", message=f"Character {char_key}: no prompts generated yet", fix_hint=f"Run /breakdown {breakdown.get('project', '')} --prompts ref to generate" )) # 1b. Reference prompt quality checks ENGINE_FLAGS = re.compile(r'--(?:ar|s|raw|no|v|oref|ow|lens)\b') REQUIRED_TERMS = {"photorealistic"} def _check_prompt_quality(prompt_text: str, asset_label: str): """Check a single prompt string for quality issues.""" if not prompt_text or not isinstance(prompt_text, str): return if "N/A" in prompt_text: return words = prompt_text.split() word_count = len(words) if word_count < 40: results.append(ValidationResult( tier=3, category="prompt_quality", message=f"{asset_label}: reference prompt too short ({word_count} words, want 60-120)", fix_hint="Regenerate with /breakdown --prompts ref for richer detail" )) elif word_count > 150: results.append(ValidationResult( tier=3, category="prompt_quality", message=f"{asset_label}: reference prompt very long ({word_count} words, target 60-120)", fix_hint="Trim to essential visual details" )) if ENGINE_FLAGS.search(prompt_text): results.append(ValidationResult( tier=3, category="prompt_quality", message=f"{asset_label}: reference prompt contains engine-specific flags (--ar, --s, etc.)", fix_hint="Remove MJ/engine flags — reference prompts must be engine-agnostic" )) prompt_lower = prompt_text.lower() for term in REQUIRED_TERMS: if term not in prompt_lower: results.append(ValidationResult( tier=3, category="prompt_quality", message=f"{asset_label}: reference prompt missing '{term}'", fix_hint=f"Reference prompts should include '{term}' for quality" )) for char_key, char_data in breakdown.get("characters", {}).items(): prompts = char_data.get("prompts", {}) ref_data = prompts.get("reference") if not ref_data: continue if isinstance(ref_data, dict): _check_prompt_quality(ref_data.get("hero", ""), f"Character {char_key} hero") for var_name, var_angles in ref_data.get("variants", {}).items(): if isinstance(var_angles, dict): for angle, text in var_angles.items(): _check_prompt_quality(text, f"Character {char_key}/{var_name}/{angle}") elif isinstance(var_angles, str): _check_prompt_quality(var_angles, f"Character {char_key}/{var_name}") elif isinstance(ref_data, str): _check_prompt_quality(ref_data, f"Character {char_key}") for loc_key, loc_data in breakdown.get("locations", {}).items(): ref_prompt = (loc_data.get("prompts") or {}).get("reference") if ref_prompt and isinstance(ref_prompt, str): _check_prompt_quality(ref_prompt, f"Location {loc_key}") for prop_key, prop_data in breakdown.get("props", {}).items(): ref_prompt = (prop_data.get("prompts") or {}).get("reference") if ref_prompt and isinstance(ref_prompt, str): _check_prompt_quality(ref_prompt, f"Prop {prop_key}") # 2. Reference image paths valid for char_key, char_data in breakdown.get("characters", {}).items(): refs = char_data.get("reference_images", {}) for slot, path in refs.items(): if slot.startswith("_"): continue if path and not Path(path).exists() and not path.startswith("data:"): results.append(ValidationResult( tier=3, category="reference_path", message=f"Character {char_key}: reference image path not found: {path}", fix_hint="Update reference_images path or regenerate the reference" )) # 3. All props have at least one description sample for prop_key, prop_data in breakdown.get("props", {}).items(): if not prop_data.get("description_samples"): results.append(ValidationResult( tier=3, category="prop_description", message=f"Prop '{prop_data.get('display_name', prop_key)}': no description samples", fix_hint="Add description samples from episode action blocks" )) # 4. VFX elements have production method assigned for vfx_key, vfx_data in breakdown.get("vfx_elements", {}).items(): if not vfx_data.get("production_method"): results.append(ValidationResult( tier=3, category="vfx_method", message=f"VFX '{vfx_data.get('display_name', vfx_key)}': no production method assigned", fix_hint="Set production_method to 'prompt_directly', 'post_composite', or 'hybrid'" )) # 5. Audio flags present for episodes with dialogue audio_eps = set() for af in breakdown.get("audio_flags", []): audio_eps.add(af.get("episode", 0)) ep_range = breakdown.get("episode_range", [0, 0]) missing_audio = [] for ep in range(ep_range[0], ep_range[1] + 1): if ep not in audio_eps: missing_audio.append(ep) if missing_audio: results.append(ValidationResult( tier=3, category="audio_flags", message=f"{len(missing_audio)} episodes without audio flags: {missing_audio[:5]}{'...' if len(missing_audio) > 5 else ''}", fix_hint="Audio flags are generated automatically during extraction" )) return results # ── Output Formatting ────────────────────────────────────────────────────── def format_results(results: List[ValidationResult], mode: str = "report") -> str: """Format validation results for display.""" tier1 = [r for r in results if r.tier == 1] tier2 = [r for r in results if r.tier == 2] tier3 = [r for r in results if r.tier == 3] if mode == "json": return json.dumps({ "is_valid": len(tier1) == 0, "exit_code": 1 if tier1 else (2 if tier2 else 0), "errors": len(tier1), "warnings": len(tier2), "info": len(tier3), "tier1_errors": [{"category": r.category, "message": r.message} for r in tier1], "tier2_warnings": [{"category": r.category, "message": r.message} for r in tier2], "tier3_info": [{"category": r.category, "message": r.message} for r in tier3], }, indent=2) lines = [] lines.append("=" * 60) lines.append("BREAKDOWN VALIDATION REPORT") lines.append("=" * 60) if tier1: lines.append(f"\nTIER 1: HARD ERRORS ({len(tier1)}) — BLOCKS PRODUCTION") lines.append("-" * 50) for r in tier1: lines.append(f" [ERROR] [{r.category}] {r.message}") if mode == "prompt" and r.fix_hint: lines.append(f" FIX: {r.fix_hint}") else: lines.append("\nTIER 1: No hard errors") if tier2: lines.append(f"\nTIER 2: CONTINUITY WARNINGS ({len(tier2)}) — REQUIRES REVIEW") lines.append("-" * 50) for r in tier2: lines.append(f" [WARN] [{r.category}] {r.message}") if mode == "prompt" and r.fix_hint: lines.append(f" FIX: {r.fix_hint}") else: lines.append("\nTIER 2: No warnings") if tier3: lines.append(f"\nTIER 3: COMPLETENESS ({len(tier3)}) — INFORMATIONAL") lines.append("-" * 50) for r in tier3: lines.append(f" [INFO] [{r.category}] {r.message}") if mode == "prompt" and r.fix_hint: lines.append(f" FIX: {r.fix_hint}") else: lines.append("\nTIER 3: All complete") lines.append("") lines.append("-" * 60) if tier1: lines.append(f"RESULT: FAIL — {len(tier1)} hard errors must be fixed before production") lines.append(f"EXIT CODE: 1") elif tier2: lines.append(f"RESULT: WARNINGS — {len(tier2)} continuity issues need review") lines.append(f"EXIT CODE: 2") else: lines.append(f"RESULT: CLEAN — Ready for production") lines.append(f"EXIT CODE: 0") return "\n".join(lines) # ── Main ─────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser( description="Validate breakdown.json for continuity and completeness" ) parser.add_argument("breakdown_json", help="Path to breakdown.json") parser.add_argument("project", help="Project path for cross-referencing") parser.add_argument("--report", action="store_true", help="Detailed report (default)") parser.add_argument("--prompt", action="store_true", help="Include fix instructions") parser.add_argument("--json", action="store_true", help="JSON output") args = parser.parse_args() # Load breakdown breakdown_path = Path(args.breakdown_json) if not breakdown_path.exists(): print(f"ERROR: Breakdown file not found: {breakdown_path}", file=sys.stderr) sys.exit(2) try: breakdown = json.loads(breakdown_path.read_text(encoding="utf-8")) except json.JSONDecodeError as e: print(f"ERROR: Invalid JSON in {breakdown_path}: {e}", file=sys.stderr) sys.exit(2) # Resolve project path project_path = resolve_project_path(args.project) # Run all tiers results = [] results.extend(validate_tier1(breakdown, project_path)) results.extend(validate_tier2(breakdown, project_path)) results.extend(validate_tier3(breakdown, project_path)) # Output if args.json: mode = "json" elif args.prompt: mode = "prompt" else: mode = "report" print(format_results(results, mode)) # Exit code tier1 = [r for r in results if r.tier == 1] tier2 = [r for r in results if r.tier == 2] if tier1: sys.exit(1) elif tier2: sys.exit(2) else: sys.exit(0) if __name__ == "__main__": main()