#!/usr/bin/env python3 from __future__ import annotations import argparse import hashlib import itertools import json import shutil import sys from datetime import datetime, timezone from pathlib import Path from typing import Any, Protocol ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from packages.pet_package_schema import ALLOWED_ACTIONS from services.ai.model_router import load_image_generator from services.ai.qwen_client import QwenReviewClient from services.media_pipeline.png_transparency import ensure_png_alpha from services.pet_builder.action_atlas import FRAMES_PER_ACTION, compose_action_frame_atlas from services.pet_builder.action_semantics import evaluate_action_semantics, score_candidate_frame from services.pet_builder.photo_generation_worker import ( ACTION_FRAME_CHROMA_KEY, ACTION_FRAME_CHROMA_THRESHOLD, _build_review, _candidate_visual_detail_score, _final_contact_sheet_visual_review_passed, _frame_rejection_reason, _safe_frame_report, _safe_generation_usage, _safe_transparency_report, _slugify, _transparency_rejection_reason, _visual_review_passed, _write_install_package, _write_json, ) from services.pet_builder.qa_media import write_animation_previews, write_contact_sheet_png from services.pet_builder.size_proportion_qa import evaluate_size_proportions WALK_GENERATED_FRAME_COUNT = 2 WALK_LOOP_MODE = "two_frame_copy" class ImageGenerator(Protocol): provider: str model: str def generate_action_frame( self, *, action: str, frame_index: int, pet_name: str, notes: str, analysis: dict[str, Any], images: list[dict[str, Any]], canonical_image_bytes: bytes, previous_frame_image_bytes: bytes | None = None, candidate_index: int = 0, size: str = "1024*1024", ) -> dict[str, Any]: ... class ActionReviewer(Protocol): def review_action_frames( self, *, pet_name: str, action: str, image_bytes: bytes, image_mime: str, ) -> dict[str, Any]: ... def review_contact_sheet( self, *, pet_name: str, image_bytes: bytes, image_mime: str, ) -> dict[str, Any]: ... def main() -> int: parser = argparse.ArgumentParser(description="Repair only the walk action row in a generated pet build.") parser.add_argument("--build-dir", required=True, type=Path, help="Source build directory with qa/action-frames.") parser.add_argument( "--output-root", default=str(ROOT / "outputs" / "pet_builds"), type=Path, help="Directory where the repaired build will be written.", ) parser.add_argument("--pet-name", default="", help="Override pet display name.") parser.add_argument("--notes", default="quiet memorial desktop pet") parser.add_argument("--candidates-per-frame", type=int, default=2) parser.add_argument("--species", default="cat") parser.add_argument("--base-color", default="gray and white") parser.add_argument("--eye-color", default="copper") parser.add_argument( "--skip-visual-review", action="store_true", help="Skip Qwen visual gait review. Intended only for offline deterministic smoke tests.", ) parser.add_argument( "--seed-candidate-dir", type=Path, default=None, help="Optional directory containing existing walk--.png candidates to reuse before new generation.", ) parser.add_argument( "--derive-frame-three-from-frame-one", action="store_true", help="Deprecated compatibility flag. Walk frames 3 and 4 are now always copied from frames 1 and 2.", ) args = parser.parse_args() build_dir = repair_walk_action( build_dir=args.build_dir, output_root=args.output_root, pet_name=args.pet_name, notes=args.notes, candidates_per_frame=args.candidates_per_frame, analysis={ "species": args.species, "base_color": args.base_color, "eye_color": args.eye_color, }, action_reviewer=None if args.skip_visual_review else QwenReviewClient.from_env(), seed_candidate_dir=args.seed_candidate_dir, derive_frame_three_from_frame_one=args.derive_frame_three_from_frame_one, ) print( json.dumps( { "status": "ok", "build_dir": _relative_or_abs(build_dir), "walk_contact_sheet": _relative_or_abs(build_dir / "qa" / "action-contact-sheets" / "walk.png"), "contact_sheet": _relative_or_abs(build_dir / "qa" / "contact-sheet.png"), "validation": _relative_or_abs(build_dir / "validation.json"), "package": _relative_or_abs(build_dir / "package" / "pet.json"), }, ensure_ascii=False, indent=2, ) ) return 0 def repair_walk_action( *, build_dir: Path, output_root: Path, pet_name: str = "", notes: str = "quiet memorial desktop pet", candidates_per_frame: int = 2, analysis: dict[str, Any] | None = None, image_generator: ImageGenerator | None = None, action_reviewer: ActionReviewer | None = None, seed_candidate_dir: Path | None = None, derive_frame_three_from_frame_one: bool = False, ) -> Path: source_build = build_dir.resolve() if not source_build.exists(): raise FileNotFoundError(f"source build not found: {source_build}") source_manifest_path = source_build / "manifest.json" if not source_manifest_path.exists(): raise FileNotFoundError(f"manifest.json not found: {source_manifest_path}") source_manifest = _read_json(source_manifest_path) display_name = pet_name.strip() or str(source_manifest.get("display_name") or source_manifest.get("pet_id") or "pet") pet_id = str(source_manifest.get("pet_id") or _slugify(display_name)) output_root = output_root.resolve() output_root.mkdir(parents=True, exist_ok=True) repaired_build_id = f"{source_manifest.get('build_id', source_build.name)}-walk-repair-{_utc_stamp()}" repaired_build = output_root / repaired_build_id if repaired_build.exists(): raise FileExistsError(f"output build already exists: {repaired_build}") shutil.copytree(source_build, repaired_build) qa_dir = repaired_build / "qa" action_frame_dir = qa_dir / "action-frames" raw_candidate_dir = qa_dir / "action-frame-raw-candidates" candidate_dir = qa_dir / "action-frame-candidates" action_contact_dir = qa_dir / "action-contact-sheets" action_review_dir = qa_dir / "action-reviews" for path in (action_frame_dir, raw_candidate_dir, candidate_dir, action_contact_dir, action_review_dir): path.mkdir(parents=True, exist_ok=True) canonical_path = _canonical_reference_path(qa_dir) canonical_reference_bytes = canonical_path.read_bytes() image_generator = image_generator or load_image_generator() candidates_per_frame = max(1, min(int(candidates_per_frame), 3)) analysis = analysis or {"species": "cat", "base_color": "gray and white", "eye_color": "copper"} candidate_failures: list[dict[str, Any]] = [] frame_pools: dict[int, list[dict[str, Any]]] = {} if seed_candidate_dir: _load_seed_candidates( seed_candidate_dir=seed_candidate_dir, candidate_dir=candidate_dir, frame_pools=frame_pools, candidate_failures=candidate_failures, ) if derive_frame_three_from_frame_one: candidate_failures.append( { "action": "walk", "frame_index": 2, "candidate_index": None, "reason": "deprecated_derive_frame_three_from_frame_one_ignored", } ) for frame_index in range(WALK_GENERATED_FRAME_COUNT): frame_pools.setdefault(frame_index, []) used_candidate_indexes = { int(item["candidate_index"]) for item in frame_pools[frame_index] if isinstance(item.get("candidate_index"), int) } for candidate_index in range(candidates_per_frame): if len(frame_pools[frame_index]) >= candidates_per_frame: break if candidate_index in used_candidate_indexes: continue try: generation = image_generator.generate_action_frame( action="walk", frame_index=frame_index, pet_name=display_name, notes=notes.strip(), analysis=analysis, images=[], canonical_image_bytes=canonical_reference_bytes, previous_frame_image_bytes=None, candidate_index=candidate_index, size="1024*1024", ) except Exception as exc: candidate_failures.append( { "action": "walk", "frame_index": frame_index, "candidate_index": candidate_index, "reason": "provider_generation_failed", "provider_error": str(exc)[:260], } ) continue raw_bytes = _require_image_bytes(generation, f"walk-{frame_index}-{candidate_index}") (raw_candidate_dir / f"walk-{frame_index}-{candidate_index}.png").write_bytes(raw_bytes) candidate = _process_candidate( action="walk", frame_index=frame_index, candidate_index=candidate_index, raw_bytes=raw_bytes, candidate_path=candidate_dir / f"walk-{frame_index}-{candidate_index}.png", generation=generation, ) if candidate["ok"]: frame_pools[frame_index].append(candidate) used_candidate_indexes.add(candidate_index) else: candidate_failures.append(candidate["failure"]) if not frame_pools[frame_index]: raise ValueError(f"no acceptable walk candidate for frame {frame_index}") selected_candidates, action_semantic_qa = _select_walk_candidates(frame_pools) for frame_index, candidate in enumerate(selected_candidates): (action_frame_dir / f"walk-{frame_index}.png").write_bytes(candidate["image_bytes"]) action_frames = _read_all_action_frames(action_frame_dir) atlas = compose_action_frame_atlas(action_frames) _annotate_walk_candidates(atlas.qa, selected_candidates) action_semantic_qa = evaluate_action_semantics(atlas.qa) if action_semantic_qa.get("ok") is not True: _write_json(qa_dir / "frame-qa.json", atlas.qa) _write_json(qa_dir / "action-semantic-qa.json", action_semantic_qa) _write_json(qa_dir / "candidate-failures.json", {"failures": candidate_failures}) raise ValueError("walk repair failed action semantic QA") walk_atlas = compose_action_frame_atlas({"walk": [item["image_bytes"] for item in selected_candidates]}) visual_review = None if action_reviewer: visual_review = action_reviewer.review_action_frames( pet_name=display_name, action="walk", image_bytes=walk_atlas.image_bytes, image_mime="image/png", ) _write_json( action_review_dir / "walk.json", { "action": "walk", "attempt": 1, "ok": _visual_review_passed(visual_review, action="walk"), "semantic_qa": action_semantic_qa, "visual_review": visual_review, }, ) if not _visual_review_passed(visual_review, action="walk"): _write_json(qa_dir / "frame-qa.json", atlas.qa) _write_json(qa_dir / "action-semantic-qa.json", action_semantic_qa) _write_json(qa_dir / "candidate-failures.json", {"failures": candidate_failures}) raise ValueError("walk repair failed visual gait QA") spritesheet_bytes = atlas.image_bytes (repaired_build / "spritesheet.png").write_bytes(spritesheet_bytes) (action_contact_dir / "walk.png").write_bytes(walk_atlas.image_bytes) contact_sheet_media = write_contact_sheet_png( spritesheet_bytes=spritesheet_bytes, output_path=qa_dir / "contact-sheet.png", ) final_visual_review = None if action_reviewer and hasattr(action_reviewer, "review_contact_sheet"): final_visual_review = action_reviewer.review_contact_sheet( pet_name=display_name, image_bytes=(qa_dir / "contact-sheet.png").read_bytes(), image_mime="image/png", ) _write_json(qa_dir / "final-visual-review.json", final_visual_review) if not _final_contact_sheet_visual_review_passed(final_visual_review): _write_json(qa_dir / "frame-qa.json", atlas.qa) _write_json(qa_dir / "action-semantic-qa.json", action_semantic_qa) _write_json(qa_dir / "candidate-failures.json", {"failures": candidate_failures}) raise ValueError("walk repair failed final contact sheet visual QA") preview_media = write_animation_previews( action_frames=action_frames, output_dir=qa_dir / "previews", ) size_proportion_qa = evaluate_size_proportions(atlas.qa) _write_json(qa_dir / "size-proportion-qa.json", size_proportion_qa) checksum = hashlib.sha256(spritesheet_bytes).hexdigest() package_manifest = _write_install_package( build_dir=repaired_build, pet_id=pet_id, pet_name=display_name, spritesheet_bytes=spritesheet_bytes, ) repaired_manifest = _updated_manifest( source_manifest=source_manifest, repaired_build_id=repaired_build_id, pet_id=pet_id, display_name=display_name, checksum=checksum, image_generator=image_generator, selected_candidates=selected_candidates, atlas_qa=atlas.qa, action_semantic_qa=action_semantic_qa, size_proportion_qa=size_proportion_qa, candidates_per_frame=candidates_per_frame, candidate_failures=candidate_failures, package_manifest=package_manifest, contact_sheet_media=contact_sheet_media, preview_media=preview_media, ) validation = _build_validation(repaired_manifest) review = _build_review(frame_qa=atlas.qa, action_semantic_qa=action_semantic_qa) _write_json(repaired_build / "manifest.json", repaired_manifest) _write_json(repaired_build / "validation.json", validation) _write_json(qa_dir / "frame-qa.json", atlas.qa) _write_json(qa_dir / "action-semantic-qa.json", action_semantic_qa) _write_json(qa_dir / "candidate-failures.json", {"failures": candidate_failures}) _write_json(qa_dir / "review.json", review) _write_json( qa_dir / "walk-repair-summary.json", { "ok": True, "repaired_action": "walk", "source_build_id": source_manifest.get("build_id"), "repaired_build_id": repaired_build_id, "provider": getattr(image_generator, "provider", "unknown"), "model": getattr(image_generator, "model", "unknown"), "selected_candidate_indexes": [item["candidate_index"] for item in selected_candidates], "walk_semantic_metrics": _walk_metrics(action_semantic_qa), "size_proportion_qa": size_proportion_qa, "visual_review": visual_review, "final_visual_review": final_visual_review, "candidate_failures": len(candidate_failures), }, ) return repaired_build def _load_seed_candidates( *, seed_candidate_dir: Path, candidate_dir: Path, frame_pools: dict[int, list[dict[str, Any]]], candidate_failures: list[dict[str, Any]], ) -> None: source_dir = seed_candidate_dir.resolve() if not source_dir.exists(): raise FileNotFoundError(f"seed candidate dir not found: {source_dir}") for source_path in sorted(source_dir.glob("walk-*-*.png")): parts = source_path.stem.split("-") if len(parts) != 3 or parts[0] != "walk": continue try: frame_index = int(parts[1]) candidate_index = int(parts[2]) except ValueError: continue if frame_index < 0 or frame_index >= FRAMES_PER_ACTION: continue image_bytes = source_path.read_bytes() target_path = candidate_dir / source_path.name if source_path.resolve() != target_path.resolve(): shutil.copy2(source_path, target_path) frame_report = _frame_report_for_candidate("walk", image_bytes) rejection_reason = _frame_rejection_reason(frame_report) if rejection_reason: candidate_failures.append( { "action": "walk", "frame_index": frame_index, "candidate_index": candidate_index, "reason": rejection_reason, "frame_report": _safe_frame_report(frame_report), "source": "seed_candidate", } ) continue base_score = score_candidate_frame(action="walk", frame_report=frame_report, candidate_index=candidate_index) frame_pools.setdefault(frame_index, []).append( { "ok": True, "candidate_index": candidate_index, "image_bytes": image_bytes, "generation": { "request_id": "", "prompt_sha256": "", "usage": {}, "source": "seed_candidate", "source_path": str(source_path), }, "transparency": { "method": "existing_transparent_candidate", "source": str(source_path), }, "frame_report": frame_report, "score": base_score + _candidate_visual_detail_score(image_bytes), "base_score": base_score, } ) def _process_candidate( *, action: str, frame_index: int, candidate_index: int, raw_bytes: bytes, candidate_path: Path, generation: dict[str, Any], ) -> dict[str, Any]: image_bytes, transparency = ensure_png_alpha( raw_bytes, chroma_key=ACTION_FRAME_CHROMA_KEY, chroma_threshold=ACTION_FRAME_CHROMA_THRESHOLD, ) transparency_rejection = _transparency_rejection_reason(transparency) if transparency_rejection == "chroma_key_background_missing": image_bytes, transparency = ensure_png_alpha(raw_bytes) transparency_rejection = _transparency_rejection_reason(transparency) candidate_path.write_bytes(image_bytes) if transparency_rejection: return { "ok": False, "failure": { "action": action, "frame_index": frame_index, "candidate_index": candidate_index, "reason": transparency_rejection, "transparency": _safe_transparency_report(transparency), }, } frame_report = _frame_report_for_candidate(action, image_bytes) rejection_reason = _frame_rejection_reason(frame_report) if rejection_reason: return { "ok": False, "failure": { "action": action, "frame_index": frame_index, "candidate_index": candidate_index, "reason": rejection_reason, "frame_report": _safe_frame_report(frame_report), }, } sanitized_generation = {key: value for key, value in generation.items() if key != "image_bytes"} base_score = score_candidate_frame(action=action, frame_report=frame_report, candidate_index=candidate_index) return { "ok": True, "candidate_index": candidate_index, "image_bytes": image_bytes, "generation": sanitized_generation, "transparency": transparency, "frame_report": frame_report, "score": base_score + _candidate_visual_detail_score(image_bytes), "base_score": base_score, } def _frame_report_for_candidate(action: str, image_bytes: bytes) -> dict[str, Any]: atlas = compose_action_frame_atlas({action: [image_bytes]}) row = next((item for item in atlas.qa.get("rows", []) if item.get("action") == action), None) frames = row.get("frames", []) if isinstance(row, dict) else [] return frames[0] if frames else {"source_bbox": None, "foreground_pixels": 0} def _select_walk_candidates(frame_pools: dict[int, list[dict[str, Any]]]) -> tuple[list[dict[str, Any]], dict[str, Any]]: best_combo: tuple[dict[str, Any], ...] | None = None best_semantic_qa: dict[str, Any] | None = None best_score = float("-inf") for combo in itertools.product(*(frame_pools[index] for index in range(WALK_GENERATED_FRAME_COUNT))): selected = list(combo) selected.append(_copy_walk_candidate(selected[0], source_frame_index=0, target_frame_index=2)) selected.append(_copy_walk_candidate(selected[1], source_frame_index=1, target_frame_index=3)) frame_qa = { "ok": True, "failure_reason": None, "frames_per_action": FRAMES_PER_ACTION, "action_count": 1, "selected_frame_count": FRAMES_PER_ACTION, "rows": [ { "action": "walk", "input_frames": FRAMES_PER_ACTION, "selected_frames": FRAMES_PER_ACTION, "loop_mode": WALK_LOOP_MODE, "generated_frame_count": WALK_GENERATED_FRAME_COUNT, "frames": [dict(item["frame_report"], frame=index) for index, item in enumerate(selected)], } ], } semantic_qa = evaluate_action_semantics(frame_qa) metrics = _walk_metrics(semantic_qa) leg_motion = float(metrics.get("lower_foreground_center_motion") or 0.0) combo_score = sum(float(item["score"]) for item in combo) + leg_motion * 2.0 if semantic_qa.get("ok") is True: combo_score += 10_000.0 if combo_score > best_score: best_score = combo_score best_combo = tuple(selected) best_semantic_qa = semantic_qa if best_combo is None or best_semantic_qa is None: raise ValueError("walk candidate selection failed") return list(best_combo), best_semantic_qa def _copy_walk_candidate(source: dict[str, Any], *, source_frame_index: int, target_frame_index: int) -> dict[str, Any]: generation = dict(source.get("generation", {})) source_request_id = str(generation.get("request_id", "")) generation["request_id"] = f"derived-copy-of:{source_request_id or f'walk-{source_frame_index}'}" generation["source_mode"] = WALK_LOOP_MODE generation["derived_from_frame_index"] = source_frame_index generation["copied_to_frame_index"] = target_frame_index usage = dict(generation.get("usage", {})) if isinstance(generation.get("usage"), dict) else {} usage["derived_copy"] = True usage["external_image_count"] = 0 generation["usage"] = usage copied = dict(source) copied["generation"] = generation copied["derived_from_frame_index"] = source_frame_index copied["copied_to_frame_index"] = target_frame_index copied["derivation"] = WALK_LOOP_MODE return copied def _exact_duplicate_frame_penalty(combo: tuple[dict[str, Any], ...]) -> float: penalty = 0.0 seen: dict[bytes, int] = {} for frame_index, item in enumerate(combo): image_bytes = item.get("image_bytes") if not isinstance(image_bytes, bytes): continue previous_index = seen.get(image_bytes) if previous_index is None: seen[image_bytes] = frame_index continue penalty += 250.0 if {previous_index, frame_index} == {0, 2}: penalty += 250.0 return penalty def _read_all_action_frames(action_frame_dir: Path) -> dict[str, list[bytes]]: action_frames: dict[str, list[bytes]] = {} for action in ALLOWED_ACTIONS: frames = [] for frame_index in range(FRAMES_PER_ACTION): frame_path = action_frame_dir / f"{action}-{frame_index}.png" if not frame_path.exists(): raise FileNotFoundError(f"action frame missing: {frame_path}") frames.append(frame_path.read_bytes()) action_frames[action] = frames return action_frames def _annotate_walk_candidates(frame_qa: dict[str, Any], selected_candidates: list[dict[str, Any]]) -> None: for row in frame_qa.get("rows", []): if row.get("action") != "walk": continue row["loop_mode"] = WALK_LOOP_MODE row["generated_frame_count"] = WALK_GENERATED_FRAME_COUNT row["derived_frames"] = [ {"frame": 2, "derived_from_frame": 0}, {"frame": 3, "derived_from_frame": 1}, ] for frame in row.get("frames", []): frame_index = int(frame.get("frame", 0)) if frame_index >= len(selected_candidates): continue selected = selected_candidates[frame_index] frame["selected_candidate_index"] = int(selected["candidate_index"]) frame["candidate_score"] = float(selected["score"]) frame["base_score"] = float(selected["base_score"]) def _updated_manifest( *, source_manifest: dict[str, Any], repaired_build_id: str, pet_id: str, display_name: str, checksum: str, image_generator: ImageGenerator, selected_candidates: list[dict[str, Any]], atlas_qa: dict[str, Any], action_semantic_qa: dict[str, Any], size_proportion_qa: dict[str, Any], candidates_per_frame: int, candidate_failures: list[dict[str, Any]], package_manifest: dict[str, Any], contact_sheet_media: dict[str, Any], preview_media: dict[str, Any], ) -> dict[str, Any]: manifest = json.loads(json.dumps(source_manifest, ensure_ascii=False)) now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") manifest["build_id"] = repaired_build_id manifest["package_id"] = f"package-{_slugify(str(getattr(image_generator, 'provider', 'image')))}-{_slugify(pet_id)}-{_utc_stamp()}" manifest["pet_id"] = pet_id manifest["display_name"] = display_name manifest["checksum"] = checksum manifest["created_at"] = now manifest["user_approved"] = False manifest["spritesheet_path"] = "spritesheet.png" manifest["package_ref"] = "package/pet.json" source = manifest.setdefault("source", {}) source["image_generation_provider"] = str(getattr(image_generator, "provider", source.get("image_generation_provider", ""))) source["image_model"] = str(getattr(image_generator, "model", source.get("image_model", ""))) source["frame_qa"] = atlas_qa source["action_semantic_qa"] = action_semantic_qa source["size_proportion_qa"] = size_proportion_qa source["output_asset_ref"] = "spritesheet.png" source["output_mime"] = "image/png" source["repaired_from_build_id"] = source_manifest.get("build_id") source["repaired_action"] = "walk" action_request_ids = source.setdefault("action_request_ids", {}) action_request_ids["walk"] = [str(item["generation"].get("request_id", "")) for item in selected_candidates] action_candidate_request_ids = source.setdefault("action_candidate_request_ids", {}) action_candidate_request_ids["walk"] = [[str(item["generation"].get("request_id", ""))] for item in selected_candidates] action_prompt_sha256s = source.setdefault("action_prompt_sha256s", {}) action_prompt_sha256s["walk"] = [str(item["generation"].get("prompt_sha256", "")) for item in selected_candidates] generation_usage = source.setdefault("generation_usage", {}) actions_usage = generation_usage.setdefault("actions", {}) actions_usage["walk"] = [_safe_generation_usage(item["generation"].get("usage", {})) for item in selected_candidates] transparency = source.setdefault("transparency_processing", {}) actions_transparency = transparency.setdefault("actions", {}) if isinstance(actions_transparency, dict): actions_transparency["walk"] = [item["transparency"] for item in selected_candidates] candidate_policy = source.setdefault("candidate_policy", {}) candidate_policy["repair_mode"] = "smallest_failed_action" candidate_policy["walk_repair_candidates_per_frame"] = candidates_per_frame candidate_policy["walk_repair_selection"] = "deterministic_two_keyframe_copy_loop" candidate_policy["walk_loop_mode"] = WALK_LOOP_MODE candidate_policy["walk_generated_frames"] = WALK_GENERATED_FRAME_COUNT candidate_policy["walk_repair_candidate_failures"] = len(candidate_failures) candidate_policy.setdefault("final_semantic_repair_actions", []) if "walk" not in candidate_policy["final_semantic_repair_actions"]: candidate_policy["final_semantic_repair_actions"].append("walk") hatch = source.setdefault("hatch_pet_pipeline", {}) hatch.update( { "contact_sheet_png": "qa/contact-sheet.png", "preview_dir": "qa/previews", "package_dir": "package", "package_manifest": package_manifest, "repair_mode": "smallest_failed_action", "contact_sheet_media": contact_sheet_media, "preview_media": preview_media, } ) return manifest def _build_validation(manifest: dict[str, Any]) -> dict[str, Any]: asset_style = manifest["source"].get("asset_style") checks = { "manifest_complete": True, "spritesheet_exists": True, "checksum_sha256": len(manifest["checksum"]) == 64, "states_allowed": [state["name"] for state in manifest["states"]] == list(ALLOWED_ACTIONS), "high_fidelity_asset": asset_style in {"high_fidelity_reference_preserving"}, "image_generation_tracked": bool(manifest["source"].get("image_request_id")) and bool(manifest["source"].get("image_model")), "alpha_capable_asset": manifest["source"].get("transparency_processing", {}).get("alpha_capable") is True, "frame_qa_passed": manifest["source"].get("frame_qa", {}).get("ok") is True, "action_semantics_passed": manifest["source"].get("action_semantic_qa", {}).get("ok") is True, "size_proportion_qa_passed": manifest["source"].get("size_proportion_qa", {}).get("ok") is True, "hatch_pet_review_artifacts_ready": bool( manifest["source"].get("hatch_pet_pipeline", {}).get("deterministic_review") ) and bool(manifest["source"].get("hatch_pet_pipeline", {}).get("contact_sheet_png")) and bool(manifest["source"].get("hatch_pet_pipeline", {}).get("preview_dir")), "package_manifest_written": manifest.get("package_ref") == "package/pet.json", "user_approval_required": manifest["user_approved"] is False, "no_sensitive_payloads": True, } return {"ok": all(checks.values()), "checks": checks, "failure_reason": None} def _walk_metrics(action_semantic_qa: dict[str, Any]) -> dict[str, Any]: for row in action_semantic_qa.get("rows", []): if row.get("action") == "walk": metrics = row.get("metrics") return metrics if isinstance(metrics, dict) else {} return {} def _canonical_reference_path(qa_dir: Path) -> Path: for name in ("canonical-reference.png", "canonical.png"): path = qa_dir / name if path.exists(): return path raise FileNotFoundError(f"canonical reference not found under {qa_dir}") def _require_image_bytes(payload: dict[str, Any], label: str) -> bytes: image_bytes = payload.get("image_bytes") if not isinstance(image_bytes, bytes) or not image_bytes: raise ValueError(f"{label} generation did not return image bytes") return image_bytes def _read_json(path: Path) -> dict[str, Any]: return json.loads(path.read_text(encoding="utf-8")) def _utc_stamp() -> str: return datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") def _relative_or_abs(path: Path) -> str: resolved = path.resolve() try: return str(resolved.relative_to(ROOT)) except ValueError: return str(resolved) if __name__ == "__main__": raise SystemExit(main())