"""ImageRunner — modality `image_t2i`. Thin wrapper over StepRunner.execute_keyframe. Constructed with the StepRunner the caller already owns; .run(payload) translates the payload dict into execute_keyframe kwargs, calls it, and adapts the StepResult into a RunResult. Payload schema: { "shot_id": str (required), "prompt": str (required), "model": str (required, e.g. "nbp", "gemini-3-pro-image-preview"), "scene_ref_path": Optional[Path | str], "pose_ref_path": Optional[Path | str], "identity_refs": Optional[list[Path | str]], "expression_refs": Optional[list[Path | str]], "aspect_ratio": Optional[str] (REQUIRED — no default; omit only when dispatch() injects from Project), "gates": Optional[list[GateFunction]], "max_gate_retries": Optional[int] (default 3), "inputs_snapshot": Optional[dict], } CP-4 surface; CP-5 may move payload validation upstream. """ from __future__ import annotations import math import time from pathlib import Path from typing import Any, Optional from recoil.pipeline.core.registry import ( MODALITY_IMAGE_T2I, RunResult, register_runner, # noqa: F401 # re-exported for Phase 6 callers ) from recoil.pipeline.core.runners._shared import ( _failure_metadata_production as _failure_metadata, _require_aspect_ratio, ) class ImageRunner: """ModalityRunner for `image_t2i` — keyframe / single-image generation. Stateless after construction. Construction takes the same StepRunner the rest of the system already builds (per current bootstrap). The runner does NOT own its own StepRunner; the caller passes one in. """ modality: str = MODALITY_IMAGE_T2I def __init__(self, step_runner) -> None: self._step_runner = step_runner def run(self, payload: dict) -> RunResult: """Run a keyframe generation via the wrapped StepRunner. Returns a RunResult. On exception, captures the message in RunResult.error and sets success=False — does NOT re-raise. """ shot_id = payload.get("shot_id") prompt = payload.get("prompt") model = payload.get("model") if not shot_id or not prompt or not model: return RunResult( id=f"{shot_id or 'unknown'}_image_t2i_{int(time.time())}", modality=self.modality, success=False, error=( f"ImageRunner payload missing required keys: " f"shot_id={shot_id!r}, prompt={'...' if prompt else None!r}, " f"model={model!r}" ), metadata=_failure_metadata(), ) def _to_path(v: Any) -> Optional[Path]: if v is None: return None return v if isinstance(v, Path) else Path(v) def _to_path_list(v: Any) -> Optional[list[Path]]: if v is None: return None return [p if isinstance(p, Path) else Path(p) for p in v] try: step_result = self._step_runner.execute_keyframe( shot_id=shot_id, prompt=prompt, model=model, scene_ref_path=_to_path(payload.get("scene_ref_path")), pose_ref_path=_to_path(payload.get("pose_ref_path")), identity_refs=_to_path_list(payload.get("identity_refs")), expression_refs=_to_path_list(payload.get("expression_refs")), # gpt-image-2 (and other fal-routed multi-ref image models) # accept a flat `reference_images` list distinct from the # legacy Google-shaped {identity/expression/scene/pose}_refs. # Forwarded here so execute_keyframe can populate # UnifiedVideoPayload.reference_images for the fal adapter. # Wired 2026-05-25 to fix silent ref drop on gpt-image-2 i2i. reference_images=_to_path_list(payload.get("reference_images")), # gpt-image-2 also reads `quality` from the payload via the # fal adapter's _resolve_gpt_image_2_quality; forward it # through the dispatch chain instead of dropping it. quality=payload.get("quality"), # gpt-image-2 size_override (2K/4K). Adapter validates the # string and falls back to the aspect-ratio preset with a # WARNING on out-of-spec values, so callers can pass freely. size_override=payload.get("size_override"), aspect_ratio=_require_aspect_ratio( payload, payload.get("shot_id", "unknown") ), gates=payload.get("gates"), max_gate_retries=payload.get("max_gate_retries", 3), inputs_snapshot=payload.get("inputs_snapshot"), ) except Exception as e: # noqa: BLE001 return RunResult( id=f"{shot_id}_image_t2i_{int(time.time())}", modality=self.modality, success=False, error=f"{type(e).__name__}: {e}", metadata=_failure_metadata(), ) return _step_result_to_run_result(step_result, shot_id, self.modality) # Metadata key contract: see RUN_RESULT_METADATA_KEYS in pipeline.core.registry def _step_result_to_run_result(step_result, shot_id: str, modality: str) -> RunResult: """Adapt a StepResult into a RunResult. StepResult fields (recoil/execution/step_types.py:81-94, verified 2026-04-27): shot_id, success, final_state, output_path, cost_usd, error, take_index, gate_verdict (singular), model, pipeline Use the boolean `success` field directly — DO NOT infer from `final_state` string matching. StepResult already carries the authoritative success bit. Hardened post-CP-4 (CP-5 Phase 2): - Bug 1.1: output_path coerced to str if non-None (Path serializability). - Bug 1.2: cost_usd NaN sanitized to 0.0 so receipts can sum costs. - Bug 6.3: success=True with output_path=None is coerced to success=False with an explanatory error message. StepRunner reporting success but producing no artifact is a contract violation; callers expecting a file from a successful RunResult should never get None. """ raw_path = getattr(step_result, "output_path", None) output_path: Optional[str] = str(raw_path) if raw_path is not None else None raw_cost = getattr(step_result, "cost_usd", None) if raw_cost is None: cost_usd: float = 0.0 elif isinstance(raw_cost, float) and math.isnan(raw_cost): cost_usd = 0.0 else: cost_usd = raw_cost success = bool(getattr(step_result, "success", False)) error = getattr(step_result, "error", None) if success and output_path is None: # StepRunner reported success but produced no artifact. Coerce to # failure so downstream code never sees a successful RunResult with # output_path=None. success = False if error is None: error = "StepRunner reported success but no output_path produced." metadata: dict[str, Any] = { "final_state": getattr(step_result, "final_state", None), "cost_usd": cost_usd, "gate_verdict": getattr(step_result, "gate_verdict", None), "take_index": getattr(step_result, "take_index", None), "take_id": getattr(step_result, "take_id", None), "model": getattr(step_result, "model", None), "pipeline": getattr(step_result, "pipeline", None), } return RunResult( id=f"{shot_id}_{modality}_{int(time.time())}", modality=modality, output_path=output_path, output_url=None, metadata=metadata, success=success, error=error, ) def _make_image_runner(step_runner=None): """Factory used by Phase 6 module-load registration when no step_runner is available at registration time. Phase 6 design: the registry stores a factory that defers StepRunner construction until first use. Callers that already own a StepRunner can pass it via `register_runner(MODALITY_IMAGE_T2I, ImageRunner(my_step_runner))` instead of the factory pattern. """ if step_runner is None: # Defer constructing a default StepRunner — it requires ExecutionStore + # ProjectPaths, which are project-context-dependent. Phase 6 leaves # the runner unregistered until a caller provides a step_runner. raise RuntimeError( "ImageRunner factory was invoked without a step_runner. " "Pass step_runner explicitly OR register an ImageRunner " "instance via pipeline.core.registry.register_runner()." ) return ImageRunner(step_runner)