#!/usr/bin/env python3 # -*- coding: utf-8 -*- """G4.4 canary — STRICT zoom_in_detail no-new-entity detection (Task 4.2). Detects entities (`C##` / `C##O##` / `P##`) introduced in current shot's t2i_prompt that are absent in the prior shot's t2i_prompt referenced via `previous_shot_refs[i].ref_usage == "zoom_in_detail"`. Any new entity (character or prop) is a violation. Override O-7 — paired-string substring carry: zoom_in_detail body-part cross-reference is asserted via `_CONTINUITY_ID_POLICY_CROSS_REF_LITERAL` (`id_policy.body_part_focus_rule`). This canary does NOT inspect the literal itself but cross-card paired drift is enforced by unit + integration tests. Scope: shots with at least one `previous_shot_refs[i].ref_usage == "zoom_in_detail"` (R2-B4 carry). Exit criteria (analyzer, OUT-OF-BAND): candidate.metrics.zoom_in_detail_new_entity_count == 0 (STRICT, Override O-1) Usage (config mode): python scripts/canary/g4_4_zoom_in_detail_no_new_entity.py \\ --config canary_config.json \\ --cp-root data/projects//checkpoints/scene_detail \\ --prompt-version 19. \\ --role candidate \\ --output results/zoom_in_detail_candidate.json """ from __future__ import annotations import argparse import json import re import sys from datetime import datetime from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) from _g4_4_common import ( # noqa: E402 _load_scenes_or_fail, _validate_scene_preflight, build_pinning_scene_set, emit_output, load_cp_manifest, load_pinning_from_args, render_prompt_card_imports, resolve_cp_root, validate_pinning, ) _CONSTS = render_prompt_card_imports() # Override O-7 paired-string substring single source — re-export for canary # log only; semantic assertion happens in unit/integration tests. Kept here so # the literal flows through the same import boundary. _CONTINUITY_ID_POLICY_CROSS_REF_LITERAL = _CONSTS[ "_CONTINUITY_ID_POLICY_CROSS_REF_LITERAL" ] # bare C## / composite C##O## / prop P## (entity ID set extraction) _CHAR_ID_REGEX = re.compile(r"\bC\d{2}(?:O\d{2})?\b") _PROP_ID_REGEX = re.compile(r"\bP\d{2}\b") def iso8601_now() -> str: return datetime.utcnow().isoformat() + "Z" def extract_entity_set(prompt: str) -> set[str]: """t2i_prompt 안 모든 C##/C##O##/P## entity ID set (full-text scan). CLAUDE.md no-truncation carry — full prompt scan. """ chars = set(_CHAR_ID_REGEX.findall(prompt)) props = set(_PROP_ID_REGEX.findall(prompt)) return chars | props def main() -> int: parser = argparse.ArgumentParser( description=( "G4.4 canary: zoom_in_detail no-new-entity detection " "(zoom_in_detail shots only)." ) ) parser.add_argument("--config", type=str, default=None) parser.add_argument("--pid", type=str, default=None) parser.add_argument("--cp-root", type=str, default=None) parser.add_argument("--prompt-version", type=str, required=True) parser.add_argument( "--role", type=str, default="candidate", choices=["baseline", "candidate"], ) parser.add_argument("--scene-index-list", type=str, default=None) parser.add_argument("--shot-index-list-per-scene", type=str, default=None) parser.add_argument("--model-routing", type=str, default=None) parser.add_argument( "--prompt-source-mode", type=str, default=None, choices=[None, "file", "db"], ) parser.add_argument("--card-commit-hash", type=str, default=None) parser.add_argument( "--continuity-card-snapshot-hash", type=str, default=None ) parser.add_argument("--output", type=str, required=True) args = parser.parse_args() pinning = load_pinning_from_args(args) validate_pinning(pinning) cp_root = resolve_cp_root(args) cp = load_cp_manifest(cp_root) measurement_failures: list[str] = [] scenes = _load_scenes_or_fail(cp, measurement_failures) if scenes is None: out = _empty_output(args, pinning, measurement_failures) emit_output(out, args) return 1 # scene-shot lookup dict for prior shot resolution. scene_shot_lookup: dict[tuple[int, int], dict] = {} for scene in scenes: if not isinstance(scene, dict): continue si = scene.get("scene_index") shi = scene.get("_shot_index") if si is None or shi is None: continue scene_shot_lookup[(int(si), int(shi))] = scene pinning_set = build_pinning_scene_set(pinning) seen_in_cp: set[tuple[int, int]] = set() total_zoom_shots = 0 zoom_new_entity_count = 0 zoom_new_entity_per_shot: dict[str, list[str]] = {} scene_set_out: list[dict] = [] for scene in scenes: if not isinstance(scene, dict): measurement_failures.append("scene entry is not dict") continue si = scene.get("scene_index") shi = scene.get("_shot_index") if si is None or shi is None: continue si_int = int(si) shi_int = int(shi) if (si_int, shi_int) not in pinning_set: continue seen_in_cp.add((si_int, shi_int)) variations = _validate_scene_preflight( scene, si_int, shi_int, measurement_failures ) if variations is None: continue rpc = scene["render_prompt_card"] ce = rpc["continuity_elements_used"] # Trap #1 — explicit key existence + type validation. if "previous_shot_refs" not in ce: measurement_failures.append( f"s{si_int}_sh{shi_int}: continuity_elements_used." f"previous_shot_refs key missing" ) continue refs = ce["previous_shot_refs"] if not isinstance(refs, list): measurement_failures.append( f"s{si_int}_sh{shi_int}: continuity_elements_used." f"previous_shot_refs not list" ) continue zoom_refs = [ r for r in refs if isinstance(r, dict) and r.get("ref_usage") == "zoom_in_detail" ] # diagnostic — record ref_usage signature for scene_set ref_usage_signature = sorted({ r.get("ref_usage") for r in refs if isinstance(r, dict) and isinstance(r.get("ref_usage"), str) }) scene_set_out.append({ "scene_index": si_int, "shot_index": shi_int, "has_zoom_in_detail_ref": bool(zoom_refs), "ref_usage_signature": ref_usage_signature, }) if not zoom_refs: continue # scope 외 — zoom_in_detail 부재 total_zoom_shots += 1 # Aggregate prior-shot entity set across all zoom_in_detail refs. prior_entities: set[str] = set() for ref in zoom_refs: prior_si = ref.get("scene_index") prior_shi = ref.get("shot_index") if prior_si is None or prior_shi is None: measurement_failures.append( f"s{si_int}_sh{shi_int}: zoom_in_detail ref missing " f"scene_index/shot_index" ) continue try: prior_key = (int(prior_si), int(prior_shi)) except (TypeError, ValueError): measurement_failures.append( f"s{si_int}_sh{shi_int}: zoom_in_detail ref " f"scene_index/shot_index not int-coercible" ) continue prior_scene = scene_shot_lookup.get(prior_key) if prior_scene is None: measurement_failures.append( f"s{si_int}_sh{shi_int}: zoom_in_detail prior " f"s{prior_key[0]}_sh{prior_key[1]} not found in CP" ) continue # Trap #1 — explicit existence + type + non-empty (R3-B1 carry). if "t2i_variations" not in prior_scene: measurement_failures.append( f"s{prior_key[0]}_sh{prior_key[1]}: t2i_variations " f"key missing" ) continue prior_variations = prior_scene["t2i_variations"] if not isinstance(prior_variations, list): measurement_failures.append( f"s{prior_key[0]}_sh{prior_key[1]}: t2i_variations " f"not list" ) continue if not prior_variations: measurement_failures.append( f"s{prior_key[0]}_sh{prior_key[1]}: t2i_variations " f"empty" ) continue for pvar in prior_variations: if not isinstance(pvar, dict): continue pprompt = pvar.get("t2i_prompt") if isinstance(pprompt, str): prior_entities |= extract_entity_set(pprompt) # Current shot entity set current_entities: set[str] = set() for vi, var in enumerate(variations): if not isinstance(var, dict): measurement_failures.append( f"s{si_int}_sh{shi_int}_v{vi}: variation not a dict" ) continue prompt = var.get("t2i_prompt") if not isinstance(prompt, str) or not prompt.strip(): measurement_failures.append( f"s{si_int}_sh{shi_int}_v{vi}: t2i_prompt missing, " f"empty, or not str" ) continue current_entities |= extract_entity_set(prompt) new_entities = current_entities - prior_entities if new_entities: zoom_new_entity_count += len(new_entities) zoom_new_entity_per_shot[f"{si_int}_{shi_int}"] = sorted( new_entities ) # Trap #4 — pinned-tuple post-loop validation. for (si, shi) in pinning_set: if (si, shi) not in seen_in_cp: measurement_failures.append( f"pinning (s{si}_sh{shi}): not present in CP data.scenes" ) out = { "timestamp": iso8601_now(), "prompt_version": args.prompt_version, "role": args.role, "pinning": pinning, "scene_set": scene_set_out, "metrics": { "total_zoom_in_detail_shots": total_zoom_shots, "zoom_in_detail_new_entity_count": zoom_new_entity_count, "zoom_in_detail_new_entity_per_shot": zoom_new_entity_per_shot, }, "measurement_failures": measurement_failures, "exit_criteria_threshold": 0, "exit_criteria_pass": ( not measurement_failures and zoom_new_entity_count == 0 ), "measurement_scripts": { "zoom_in_detail_no_new_entity": ( "scripts/canary/g4_4_zoom_in_detail_no_new_entity.py " "(zoom_in_detail shots only — Override O-7 paired ref " f"literal: {_CONTINUITY_ID_POLICY_CROSS_REF_LITERAL})" ) }, } emit_output(out, args) print( f"[g4_4_zoom_in_detail_no_new_entity] role={args.role} " f"prompt_version={args.prompt_version}\n" f" total_zoom_in_detail_shots={total_zoom_shots}\n" f" zoom_in_detail_new_entity_count={zoom_new_entity_count}\n" f" measurement_failures: {len(measurement_failures)}\n" f"Exit criteria (STRICT): " f"candidate.metrics.zoom_in_detail_new_entity_count == 0", file=sys.stderr, ) if measurement_failures: return 1 if args.role == "candidate" and zoom_new_entity_count > 0: return 1 return 0 def _empty_output( args: argparse.Namespace, pinning: dict, measurement_failures: list, ) -> dict: return { "timestamp": iso8601_now(), "prompt_version": args.prompt_version, "role": args.role, "pinning": pinning, "scene_set": [], "metrics": { "total_zoom_in_detail_shots": 0, "zoom_in_detail_new_entity_count": 0, "zoom_in_detail_new_entity_per_shot": {}, }, "measurement_failures": measurement_failures, "exit_criteria_threshold": 0, "exit_criteria_pass": False, "measurement_scripts": { "zoom_in_detail_no_new_entity": ( "scripts/canary/g4_4_zoom_in_detail_no_new_entity.py " "(zoom_in_detail shots only)" ) }, } if __name__ == "__main__": sys.exit(main())