"""recheck_pending_qc.py — Morning re-check for pending_qc shots. When a critic errors out during a run (vision API down, network issue), shots enter the pending_qc state and the run continues. The next morning, this script walks all pending_qc shots, re-runs the appropriate critic, and either promotes them to a terminal state or marks for review. CLI: python3 -m pipeline.tools.recheck_pending_qc --project """ import argparse import logging import sys from pathlib import Path logger = logging.getLogger(__name__) def find_pending_qc_shots(store) -> list[dict]: """Return all shots currently in pending_qc state.""" all_shots = store.get_all_shots() if hasattr(store, "get_all_shots") else [] return [s for s in all_shots if s.get("status") == "pending_qc"] def _resolve_output_path(output_path_str: str, project: str) -> Path: """Resolve a (possibly relative) output path against the project root.""" output = Path(output_path_str) if output.is_absolute(): return output # Relative — join against projects_root()/{project} from recoil.core.paths import projects_root return projects_root() / project / output def _build_start_frame_critic_with_context(shot: dict): """Construct StartFrameCritic with character context from inputs_snapshot. Without this, the critic runs with empty character_descriptions and the CHARACTER_IDENTITY checks never fire — re-introduces the silent bypass bug that Task 6 fixes (Gemini Finding 2, 2026-04-09). """ from recoil.pipeline._lib.critics.start_frame_critic import StartFrameCritic inputs_snapshot = shot.get("inputs_snapshot") or {} character_descriptions = [] expected_elements = [] if inputs_snapshot: for char in inputs_snapshot.get("characters", []): character_descriptions.append({ "name": char.get("display_name") or char.get("char_id", ""), "hair": char.get("hair", ""), "facial_hair": char.get("facial_hair", ""), "clothing": char.get("wardrobe") or char.get("clothing", ""), }) expected_elements = inputs_snapshot.get("scene_elements", []) intention_context = { "character_anchor": inputs_snapshot.get("identity_anchor", "") if inputs_snapshot else "", "scene_description": inputs_snapshot.get("scene_description", "") if inputs_snapshot else "", } return StartFrameCritic( expected_background="scene", character_descriptions=character_descriptions, expected_elements=expected_elements, shot_id=shot.get("shot_id", ""), intention_context=intention_context, ) def _run_critic_for_shot(shot: dict, project: str): """Re-run the appropriate critic for a shot. Returns CriticResult.""" from recoil.core.paths import ensure_pipeline_importable ensure_pipeline_importable() from recoil.core.critic import Outcome, CriticResult pipeline = shot.get("pipeline", "") output_path_str = shot.get("output_path") if not output_path_str: return CriticResult(critic_name="recheck", outcome=Outcome.ERROR, error="no output_path on pending_qc shot") output = _resolve_output_path(output_path_str, project) if not output.exists(): return CriticResult(critic_name="recheck", outcome=Outcome.ERROR, error=f"output file missing: {output}") if pipeline == "video": from recoil.pipeline._lib.critics.video_frame_critic import VideoFrameCritic # Plumb character context into video frame critic too inputs_snapshot = shot.get("inputs_snapshot") or {} char_type = "human" # default; could be derived from inputs_snapshot in future critic = VideoFrameCritic( character_type=char_type, expected_elements=inputs_snapshot.get("scene_elements", []), num_frames=3, ) _, result = critic.run(str(output)) return result elif pipeline == "still": critic = _build_start_frame_critic_with_context(shot) _, result = critic.run(str(output)) return result elif pipeline == "previz": # Previz doesn't have hard QC — just promote return CriticResult(critic_name="recheck", outcome=Outcome.PASS) else: return CriticResult(critic_name="recheck", outcome=Outcome.ERROR, error=f"unknown pipeline: {pipeline}") def recheck_shot(store, shot: dict, project: str = "") -> str: """Re-run critic and promote/mark as needed. Returns one of: "promoted", "needs_review", "failed", "skipped". """ shot_id = shot["shot_id"] pipeline = shot.get("pipeline", "") result = _run_critic_for_shot(shot, project=project or store.project) if result.outcome.value == "pass": # Promote to the right terminal state if pipeline == "video": store.update_shot(shot_id, status="video_complete") elif pipeline == "still": store.update_shot(shot_id, status="keyframe_generated") elif pipeline == "previz": store.update_shot(shot_id, status="previs_generated") return "promoted" if result.outcome.value == "error": # Still erroring — mark for human review, don't loop store.update_shot( shot_id, status="needs_review", error_message=f"recheck still ERROR: {result.error}", ) return "needs_review" # FAIL with real failures — mark as needs_review for human action store.update_shot( shot_id, status="needs_review", error_message=f"recheck failed: {[d.name for d in result.failed_dimensions]}", ) return "failed" def main(): parser = argparse.ArgumentParser(description="Re-check pending_qc shots after a vision API outage") parser.add_argument("--project", required=True, help="Project name") parser.add_argument("--dry-run", action="store_true", help="List but don't update") args = parser.parse_args() logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") from recoil.core.paths import ensure_pipeline_importable ensure_pipeline_importable() from recoil.execution.execution_store import ExecutionStore store = ExecutionStore(project=args.project) pending = find_pending_qc_shots(store) if not pending: print(f"No pending_qc shots for project {args.project}") return 0 print(f"Found {len(pending)} pending_qc shots — re-checking...") counts = {"promoted": 0, "needs_review": 0, "failed": 0} for shot in pending: if args.dry_run: print(f" [dry-run] would recheck {shot['shot_id']}") continue outcome = recheck_shot(store, shot, project=args.project) counts[outcome] = counts.get(outcome, 0) + 1 print(f" {shot['shot_id']}: {outcome}") print(f"\nResults: {counts}") return 0 if __name__ == "__main__": sys.exit(main())