""" ref_image_critic.py — Reference Image Validator. CriticLoop subclass that validates reference images before use as Kling elements. Catches extra limbs, wrong appendages, missing props, and background contamination. max_attempts=1: ref images need regeneration, not patching. """ 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 # Map check names to FailureMode for Dimension tagging _CHECK_FAILURE_MODES = { "LIMB_COUNT": FailureMode.ANATOMY_LIMB_MISCOUNT, "EXTRA_APPENDAGES": FailureMode.ANATOMY_LIMB_MISCOUNT, "LEGS_ON_GROUND": FailureMode.ANATOMY_LIMB_MISCOUNT, "BACKGROUND_CLEAN": FailureMode.BACKGROUND_CONTAMINATION, "PROP_HELD": FailureMode.UNKNOWN, } logger = logging.getLogger(__name__) class RefImageCritic(CriticLoop): """Validates reference images before use as Kling elements. Dimensions: LIMB_COUNT (HARD) — correct number of limbs for character type EXTRA_APPENDAGES (HARD) — no phantom/duplicate limbs PROP_HELD (HARD, if applicable) — character holding expected prop BACKGROUND_CLEAN (SOFT) — background is solid neutral color Args: character_type: "human", "quadruped", or "vehicle". expected_props: List of props the character should be holding. legs_on_ground: Expected number of legs on ground. total_limbs: Expected total limb count. experience_pool_dir: Path for JSONL logging. shot_id: Shot identifier for logging. """ def __init__( self, character_type: str = "human", expected_props: Optional[list[str]] = None, legs_on_ground: Optional[int] = None, total_limbs: Optional[int] = None, experience_pool_dir: Optional[Path] = None, shot_id: str = "", target_model: str = "", ): super().__init__( name="ref_image", max_attempts=1, experience_pool_dir=experience_pool_dir, shot_id=shot_id, ) self.character_type = character_type self.expected_props = expected_props or [] self.target_model = target_model # Set defaults based on character type if character_type == "human": self.legs_on_ground = legs_on_ground if legs_on_ground is not None else 2 self.total_limbs = total_limbs if total_limbs is not None else 4 elif character_type == "quadruped": self.legs_on_ground = legs_on_ground if legs_on_ground is not None else 4 self.total_limbs = total_limbs if total_limbs is not None else 4 else: # vehicle or other self.legs_on_ground = legs_on_ground if legs_on_ground is not None else 0 self.total_limbs = total_limbs if total_limbs is not None else 0 def _build_checks(self) -> list[dict]: """Build the check list for this character type.""" checks = [] if self.total_limbs > 0: checks.append({ "name": "LIMB_COUNT", "question": ( f"How many total limbs (arms + legs) does this character have? " f"Expected: {self.total_limbs}. Answer with just the number." ), "expected": str(self.total_limbs), "severity": "hard", }) checks.append({ "name": "EXTRA_APPENDAGES", "question": ( "Are there any extra, phantom, or duplicate limbs, fingers, " "or appendages? Answer 'no' if the anatomy looks normal, " "'yes' if there are extras." ), "expected": "no", "severity": "hard", }) if self.character_type in ("human", "quadruped") and self.legs_on_ground is not None: checks.append({ "name": "LEGS_ON_GROUND", "question": f"How many legs are touching the ground? Answer with just a number.", "expected": str(self.legs_on_ground), "severity": "hard", }) for prop in self.expected_props: checks.append({ "name": "PROP_HELD", "question": ( f"Is the character holding or carrying a {prop}? " f"Answer 'yes' or 'no'." ), "expected": "yes", "severity": "hard", }) checks.append({ "name": "BACKGROUND_CLEAN", "question": ( "Is the background a solid neutral color (white, gray, or similar) " "with no scene elements, objects, or environment contamination? " "Answer 'yes' for clean background, 'no' for contaminated." ), "expected": "yes", "severity": "soft", }) return checks def evaluate(self, artifact: Any, context: dict) -> list[Dimension]: """Evaluate a reference image against all dimensions. Args: artifact: Path to the reference image (str or Path). context: Optional context dict (unused). Returns: List of Dimension results. """ image_path = str(artifact) checks = self._build_checks() # Per-model strictness: "relaxed" downgrades HARD anatomy checks to SOFT anatomy_strictness = "standard" if self.target_model: try: from recoil.core.model_profiles import get_critic_override anatomy_strictness = get_critic_override( self.target_model, "anatomy", "strictness", "standard" ) except Exception as e: logger.debug("Could not get critic override for %s: %s", self.target_model, e) context_desc = ( f"Reference image of a {self.character_type} character for " f"film production. Should be on a clean neutral background." ) result = validate_image(image_path, checks, context_desc) dims = [] if result.get("error"): # Vision API failure — let CriticLoop catch and route to ERROR raise RuntimeError(f"RefImageCritic 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 anatomy tuning: "relaxed" downgrades HARD to SOFT if anatomy_strictness == "relaxed" and severity == Severity.HARD: if check_result["name"] in ("LIMB_COUNT", "EXTRA_APPENDAGES"): severity = Severity.SOFT message = "" if not check_result["passed"]: message = ( f"Expected '{check_result['expected']}', " f"got '{check_result['answer']}'" ) fm = None if check_result["passed"] else _CHECK_FAILURE_MODES.get(check_result["name"], FailureMode.UNKNOWN) dims.append(Dimension( name=check_result["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 ref_image_critic( image=None, reference=None, structured_output: bool = False, image_path: str | None = None, character_type: str = "human", expected_props: list | None = None, **kwargs, ): """Structured-output adapter for the reference image critic. Runs the real RefImageCritic class on the provided image_path and translates the Dimension list to a FailureMode-typed dict. Required for structured_output=True: image_path (or image/reference). """ if not structured_output: raise NotImplementedError( "Use RefImageCritic class directly for non-structured calls." ) # Resolve image path from various aliases img = image_path or image or reference if img is None: raise ValueError( "ref_image_critic(structured_output=True) requires " "image_path (or image/reference)." ) critic = RefImageCritic( character_type=character_type, expected_props=expected_props or [], shot_id=kwargs.get("shot_id", ""), ) _, result = critic.run(str(img)) # Translate Dimension list → FailureMode (first hard failure wins) failure_mode = FailureMode.NONE.value for dim in result.hard_failures: if dim.name in ("LIMB_COUNT", "EXTRA_APPENDAGES", "LEGS_ON_GROUND"): failure_mode = FailureMode.ANATOMY_LIMB_MISCOUNT.value break elif dim.name == "BACKGROUND_CLEAN": failure_mode = FailureMode.BACKGROUND_CONTAMINATION.value break elif dim.name == "PROP_HELD": 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 ], }