""" start_frame_critic.py — Start Frame Validator. CriticLoop subclass that validates start frames before Kling generation. Catches blank backgrounds, wrong character identity, missing scene elements, and corrupted compositions. max_attempts=1: report only, start frames need regeneration. """ import logging from pathlib import Path from typing import Any, Optional from recoil.core.critic import CriticLoop, Dimension, Severity from recoil.core.vision_check import validate_image from . import FailureMode logger = logging.getLogger(__name__) class StartFrameCritic(CriticLoop): """Validates start frames before Kling generation. Dimensions: BACKGROUND_VALID (HARD) — no white/blank backgrounds when scene expected CHARACTER_IDENTITY (HARD) — character matches expected description SCENE_ELEMENTS (SOFT) — expected elements present COMPOSITION_VALID (HARD) — image is a proper composition, not blank/corrupted Args: expected_background: "scene" or "solid color". character_descriptions: List of {name, hair, facial_hair, clothing} dicts. expected_elements: List of scene elements to check for. experience_pool_dir: Path for JSONL logging. shot_id: Shot identifier for logging. """ def __init__( self, expected_background: str = "scene", character_descriptions: Optional[list[dict]] = None, expected_elements: Optional[list[str]] = None, experience_pool_dir: Optional[Path] = None, shot_id: str = "", intention_context: Optional[dict] = None, target_model: str = "", ): super().__init__( name="start_frame", max_attempts=1, experience_pool_dir=experience_pool_dir, shot_id=shot_id, ) self.expected_background = expected_background self.character_descriptions = character_descriptions or [] self.expected_elements = expected_elements or [] self.intention_context = intention_context self.target_model = target_model def _build_checks(self) -> list[dict]: """Build the check list for this start frame.""" checks = [] # BACKGROUND_VALID if self.expected_background == "scene": checks.append({ "name": "BACKGROUND_VALID", "question": ( "Is this image showing a proper scene/environment background " "(not a blank white, solid color, or empty background)? " "Answer 'yes' if there is a real scene, 'no' if it's blank or solid." ), "expected": "yes", "severity": "hard", }) else: checks.append({ "name": "BACKGROUND_VALID", "question": ( "Is the background a solid color (white, gray, or similar)? " "Answer 'yes' for solid color, 'no' for a scene/environment." ), "expected": "yes", "severity": "hard", }) # CHARACTER_IDENTITY — one check per character for char in self.character_descriptions: name = char.get("name", "character") desc_parts = [] if char.get("hair"): desc_parts.append(f"{char['hair']} hair") if char.get("facial_hair"): desc_parts.append(f"{char['facial_hair']}") if char.get("clothing"): desc_parts.append(f"wearing {char['clothing']}") desc_str = ", ".join(desc_parts) if desc_parts else "as described" check_name = f"CHARACTER_IDENTITY_{name.upper().replace(' ', '_')}" checks.append({ "name": check_name, "question": ( f"Is there a character matching this description: {name} with " f"{desc_str}? Answer 'yes' if the character matches, 'no' if not." ), "expected": "yes", "severity": "hard", }) # SCENE_ELEMENTS — one check per element for element in self.expected_elements: checks.append({ "name": f"SCENE_ELEMENTS_{element.upper().replace(' ', '_')}", "question": ( f"Is a {element} visible in this image? " f"Answer 'yes' or 'no'." ), "expected": "yes", "severity": "soft", }) # COMPOSITION_VALID checks.append({ "name": "COMPOSITION_VALID", "question": ( "Is this a proper, non-corrupted image with visible content " "(not entirely black, white, or glitched)? " "Answer 'yes' for a valid image, 'no' for blank/corrupted." ), "expected": "yes", "severity": "hard", }) return checks def evaluate(self, artifact: Any, context: dict) -> list[Dimension]: """Evaluate a start frame against all dimensions. Args: artifact: Path to the start frame image (str or Path). context: Optional context dict (unused). Returns: List of Dimension results. """ image_path = str(artifact) checks = self._build_checks() # Per-model identity strictness identity_strictness = "standard" if self.target_model: try: from recoil.core.model_profiles import get_critic_override identity_strictness = get_critic_override( self.target_model, "identity_drift", "strictness", "standard" ) except Exception as e: logger.debug("Could not get critic override for %s: %s", self.target_model, e) context_desc = ( f"Start frame for video generation. " f"Expected background: {self.expected_background}." ) result = validate_image(image_path, checks, context_desc, intention_context=self.intention_context) dims = [] if result.get("error"): # Vision API failure — let CriticLoop catch and route to ERROR raise RuntimeError(f"StartFrameCritic vision check failed: {result['error']}") for check_result in result.get("results", []): severity = ( Severity.HARD if check_result["severity"] == "hard" else Severity.SOFT ) # Per-model identity tuning: "relaxed" downgrades CHARACTER_IDENTITY to SOFT if identity_strictness == "relaxed" and severity == Severity.HARD: if check_result["name"].startswith("CHARACTER_IDENTITY"): severity = Severity.SOFT message = "" if not check_result["passed"]: message = ( f"Expected '{check_result['expected']}', " f"got '{check_result['answer']}'" ) # Map check name to failure mode check_name = check_result["name"] if check_result["passed"]: fm = None elif check_name == "BACKGROUND_VALID": fm = FailureMode.BACKGROUND_CONTAMINATION elif check_name.startswith("CHARACTER_IDENTITY"): fm = FailureMode.IDENTITY_DRIFT elif check_name.startswith("SCENE_ELEMENTS") or check_name == "COMPOSITION_VALID": fm = FailureMode.UNKNOWN else: fm = FailureMode.UNKNOWN dims.append(Dimension( name=check_name, severity=severity, passed=check_result["passed"], message=message, failure_mode=fm, )) return dims # ────────────────────────────────────────────────────────────────────────── # Structured-output adapter (Phase 2.5 — wired to real CriticLoop) # ────────────────────────────────────────────────────────────────────────── def start_frame_critic( image=None, reference=None, structured_output: bool = False, image_path: str | None = None, character_descriptions: list | None = None, expected_elements: list | None = None, expected_background: str = "scene", **kwargs, ): """Structured-output adapter for the start frame critic. Runs the real StartFrameCritic class and translates the result. Required for structured_output=True: image_path (or image/reference). """ if not structured_output: raise NotImplementedError( "Use StartFrameCritic class directly for non-structured calls." ) img = image_path or image or reference if img is None: raise ValueError( "start_frame_critic(structured_output=True) requires " "image_path (or image/reference)." ) critic = StartFrameCritic( expected_background=expected_background, character_descriptions=character_descriptions or [], expected_elements=expected_elements or [], shot_id=kwargs.get("shot_id", ""), intention_context=kwargs.get("intention_context"), ) _, result = critic.run(str(img)) # Translate Dimension list → FailureMode failure_mode = FailureMode.NONE.value for dim in result.hard_failures: if dim.name == "BACKGROUND_VALID": failure_mode = FailureMode.BACKGROUND_CONTAMINATION.value break elif dim.name.startswith("CHARACTER_IDENTITY"): failure_mode = FailureMode.IDENTITY_DRIFT.value break elif dim.name.startswith("SCENE_ELEMENTS") or dim.name == "COMPOSITION_VALID": failure_mode = FailureMode.UNKNOWN.value break return { "passed": result.passed, "score": 1.0 if result.passed else 0.0, "failure_mode": failure_mode, "evidence": "; ".join(d.message for d in result.failed_dimensions if d.message), "dimensions": [ {"name": d.name, "passed": d.passed, "severity": d.severity.value} for d in result.dimensions ], }