"""T2I 검수 StepRunner — entity_t2i + scene_detail 검증 후 치환.""" import logging from typing import Any, Dict, List, Optional, Tuple from app.core.errors import AppError from app.core.step_runner import StepRunner from app.core.steps._owned_helpers import refresh_t2i_prompt_hash from app.core.steps.detail_steps import _DetailStepMixin logger = logging.getLogger(__name__) class T2iReviewStep(_DetailStepMixin, StepRunner): """entity_t2i와 scene_detail의 T2I 프롬프트를 검수하고 치환 적용.""" def _config_hash(self) -> str: """검수 **범위**를 지문에 접는다 (2026-08-27, Codex BLOCK-3). v1 에서 씬 갈래를 건너뛰면 산출 모양이 달라진다(`scene_fixes` 가 0 이고 `scene_review_status` 가 붙는다). 그것을 안 접으면 **옛 CP 가 새 계약인 것처럼** current 로 읽혀 조용히 skip 되고, 나중에 「검사했는데 깨끗하다」로 오독된다. ★OFF/legacy 에서는 종전과 같은 값이 나오게 둔다 — 그 경로는 범위가 안 바뀌었다. """ import hashlib import json as _json from app.core.config import settings from app.core.step_runner import compute_config_hash base = compute_config_hash(self.project_config) mode = getattr(settings, "still_recipe_mode", "off") skip = bool(getattr(settings, "t2i_review_skip_scene_on_v1", True)) if mode != "v1" or not skip: return base # legacy 는 byte-identical return hashlib.sha256( _json.dumps({"base": base, "review_scope": "entity_only_v1"}, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _execute(self, mode="resume") -> Dict[str, Any]: # 의존 체크포인트 로드 entity_t2i_cp = self._load_prev_checkpoint("entity_t2i") scene_detail_cp = self._load_prev_checkpoint("scene_detail") entity_merge_cp = self._load_prev_checkpoint("entity_merge") entity_detail_cp = self._load_prev_checkpoint("entity_detail") shot_extract_cp = self._load_prev_checkpoint("shot_validator") # Phase 9.2: t2i_review v2의 신규 검증 룰 (close_framing_existing_ref / # physical_inconsistency)이 shot_staging.camera_direction 비교를 요구. shot_staging_cp = self._load_prev_checkpoint("shot_staging") vwr_cp = self._load_prev_checkpoint("visual_world_rules") if not entity_t2i_cp or not scene_detail_cp: from app.core.errors import AppError raise AppError( code="step.no_input", message="entity_t2i 또는 scene_detail 결과 없음", status_code=400, ) entity_t2i_data = entity_t2i_cp.get("data", {}) scene_detail_data = scene_detail_cp.get("data", {}) entity_merge_data = entity_merge_cp.get("data", {}) if entity_merge_cp else {} entity_detail_data = entity_detail_cp.get("data", {}) if entity_detail_cp else {} shot_extract_data = shot_extract_cp.get("data", {}) if shot_extract_cp else {} shot_staging_data = shot_staging_cp.get("data", {}) if shot_staging_cp else {} vwr_data = vwr_cp.get("data", {}) if vwr_cp else {} from app.modules.pipeline.t2i_review import run_t2i_review # Area #6 v1 (2026-05-18+): t2i_review = diagnostic-only, mutation=0. # `_apply_*_fixes` 는 W1 에서 폐기됐고 run_t2i_review 는 entity_applied=0 # / scene_applied=0 / scene_applied_indices=[] 만 반환. 본 pre-capture # 호출은 legacy defensive guard 로 보존 — 미래 mutation 재도입 시 자동 # 보호 (Patch A baseline, V2 patch P2 / spec V5 S1 carry). pre_card_state = _capture_card_hash_state(scene_detail_data) review_result = run_t2i_review( entity_t2i_data=entity_t2i_data, scene_detail_data=scene_detail_data, entity_merge_data=entity_merge_data, entity_detail_data=entity_detail_data, shot_extract_data=shot_extract_data, shot_staging_data=shot_staging_data, vwr_data=vwr_data, opik_metadata=self.build_opik_metadata(extra_tags=["t2i_review"]), ) # Area #6 v1 (2026-05-18+): scene_detail_data 는 변경 0 # (mutation=0 invariant). pre_card_state 는 legacy defensive baseline # 이며 W1 에서 entity_applied/scene_applied > 0 branch 는 unreached. # 아래 if-branch 2 개는 legacy defensive guard — entity_applied > 0 # / scene_applied > 0 시 checkpoint overwrite 의무 (mutation 재도입 # 시 자동 보호, Patch B 보존). v1 에서는 mutation=0 invariant 로 모두 # unreached. if review_result["entity_applied"] > 0: self._save_checkpoint_data("entity_t2i", entity_t2i_data) logger.info("t2i_review: entity_t2i checkpoint updated (%d fixes)", review_result["entity_applied"]) if review_result["scene_applied"] > 0: # AC-A1, A2, A3: sentinel refresh + card hash assert (Block A hotfix) applied_indices = review_result["scene_applied_indices"] try: refreshed = _refresh_scene_detail_sentinels( scene_detail_data, applied_indices ) _assert_card_hash_unchanged( scene_detail_data, applied_indices, pre_card_state ) except (AppError, KeyError, IndexError) as exc: # AC-A2: refresh / assert 실패 → t2i_review failed (silent # corruption 차단, feedback_no_silent_fallback.md). if isinstance(exc, AppError): raise raise AppError( code="t2i_review.sentinel_refresh_failed", message=f"sentinel refresh 실패: {exc}", ) from exc self._save_checkpoint_data("scene_detail", scene_detail_data) logger.info( "t2i_review: scene_detail checkpoint updated (%d fixes, %d sentinels refreshed)", review_result["scene_applied"], refreshed, ) # Phase 9.2: cascade 정책 = 1-pass (step_runner.py:388-398). t2i_review가 # entity_t2i/scene_detail을 수정해도 downstream 자동 invalidate 안 함. 이미 # 완료된 shot_dependency_t2i/scene_image_pipeline은 review 이전 prompt 기반. # 운영자가 review 결과를 production에 반영하려면 downstream 수동 force 권장. if review_result["entity_applied"] > 0 or review_result["scene_applied"] > 0: logger.warning( "t2i_review: %d fixes applied to entity_t2i + %d to scene_detail. " "downstream(shot_dependency_t2i / scene_image_pipeline)이 이미 완료 상태면 " "수정 결과 반영을 위해 force 재실행이 필요합니다 (cascade=1-pass policy).", review_result["entity_applied"], review_result["scene_applied"], ) return { "review_summary": { "entity_detected": review_result["entity_fixes"], "scene_detected": review_result["scene_fixes"], "entity_applied": review_result["entity_applied"], # canary: always 0 "scene_applied": review_result["scene_applied"], # canary: always 0 "regeneration_required": { "entity": review_result["entity_fixes"], "scene": review_result["scene_fixes"], }, # ★2026-08-27: 모듈이 돌려준 건너뜀 신원을 **버리지 않는다.** # 이 칸이 없으면 체크포인트에 `scene_detected=0` 만 남아 # 「검사했는데 깨끗하다」로 오독된다 — 이 판이 막겠다고 한 # 바로 그것이다 (Codex BLOCK-2). "scene_review_status": review_result["scene_review_status"], }, "diagnostics": review_result["diagnostics"], "scene_applied_indices": review_result["scene_applied_indices"], # canary: always [] # ★2026-08-27 (Codex 재리뷰 BLOCK): **저장값과 재개 기대값이 # 같아야 한다.** `_config_hash` 를 새로 만들었는데 반환 최상위에 # 안 실으면 `save_checkpoint` 가 `project_config` fallback 을 # 저장한다 — 저장 `99914b93…` vs 다음 재개 기대 `0fe556f8…` 라 # **첫 완료 뒤 재개가 항상 contract drift BLOCK** 이다. # `shot_dependency_t2i_step.py:472` 가 이미 같은 계약을 쓴다. "config_hash": self._config_hash(), } def _save_checkpoint_data(self, step_id: str, data: Dict): """다른 step의 체크포인트 데이터를 덮어쓰기 (아카이브 + 원자적 쓰기). B1 patch (Block A closure stabilization, Codex BLOCKING #1): manifest missing / unreadable 시 모두 fail-fast. 옛 동작 (logger.warning + return) 은 mutator 가 성공처럼 끝나지만 cp 미저장 — silent fallback 정책 위반 (feedback_no_silent_fallback.md). caller (StepRunner) 가 step 실패로 surface — silent corruption 차단. """ import shutil from datetime import datetime, timezone from pathlib import Path from app.core.checkpoint_io import atomic_write_json, read_json_safe from app.core.config import settings cp_dir = Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id manifest = cp_dir / "manifest.json" if not manifest.exists(): # B1 patch: silent skip 금지 — mutator 성공 오판 차단. raise AppError( code="t2i_review.checkpoint_save_failed", message=( f"t2i_review: target {step_id} checkpoint manifest missing — " f"silent skip 금지 (mutator 성공 오판 차단). path={manifest}" ), ) # 아카이브 (기존 버전 보존) — fail-fast 전에 archive 생성하여 pre-mutation # state 디스크 보존. archive copy 자체 실패 (disk full / permission) 도 # checkpoint_save_failed 로 surface (Codex IMPORTANT #2: contract symmetry — # OSError 가 raw 로 leak 되면 caller 의 try 가 못 잡음). archive_name = f"manifest_{datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')}_pre_review.json" try: shutil.copy2(str(manifest), str(cp_dir / archive_name)) except OSError as exc: raise AppError( code="t2i_review.checkpoint_save_failed", message=( f"t2i_review: target {step_id} archive copy failed — " f"silent skip 금지. {type(exc).__name__}: {exc}" ), ) from exc # 원자적 쓰기 (tmp + rename) cp = read_json_safe(manifest) if cp is None: # B1 patch: silent skip 금지 — archive 는 위에서 보존됨. raise AppError( code="t2i_review.checkpoint_save_failed", message=( f"t2i_review: target {step_id} manifest unreadable — " f"silent skip 금지. archive 보존: {archive_name}" ), ) cp["data"] = data atomic_write_json(manifest, cp) # ───────────────────────────────────────────────────────────────────────────── # Module-level helpers (Block A T6 — V2 patch P2 / spec V5 S1). # t2i_review 가 mutator → mutation 후 victim invariant (owned_validation # sentinel) 갱신 + card payload 변경 차단 assert. # ───────────────────────────────────────────────────────────────────────────── def _refresh_scene_detail_sentinels( scene_detail_data: Dict[str, Any], applied_indices: List[Tuple[int, int]], ) -> int: """t2i_review 가 수정한 (scene_list_idx, variation_idx) 항목의 owned_validation sentinel 의 t2i_prompt_hash 만 갱신. AC-A1, A2 (spec §3.4): applied_indices 외 항목은 건드리지 않음. sentinel 누락 / shape 위반 시 raise → caller 가 t2i_review failed 처리. """ refreshed = 0 scenes = scene_detail_data["scenes"] for s_list_idx, v_idx in applied_indices: scene = scenes[s_list_idx] variation = scene["t2i_variations"][v_idx] sentinel = variation.get("owned_validation") if not sentinel: raise AppError( code="t2i_review.owned_validation_missing", message=( f"scenes[{s_list_idx}].t2i_variations[{v_idx}]." "owned_validation 없음 — schema 위반" ), ) if refresh_t2i_prompt_hash( sentinel, variation["t2i_prompt"], where=f"t2i_review_refresh@s{s_list_idx}_v{v_idx}", ): refreshed += 1 return refreshed def _capture_card_hash_state( scene_detail_data: Dict[str, Any], ) -> Dict[Tuple[int, int], Optional[str]]: """mutation 직전 모든 (scene_idx, var_idx) 의 render_prompt_card_hash 전체 capture. V2 patch P2 (Codex BLOCKING #2 + spec §3.5 V5 patch S1): run_t2i_review() 호출 **전** capture 의무 — 호출 후엔 이미 mutation 끝남. applied_indices 는 mutation 후에야 알 수 있으므로 **전체** scenes/variations 매핑. under-coverage 위험 0 — applied_indices 는 항상 전체의 부분집합. """ state: Dict[Tuple[int, int], Optional[str]] = {} for s_idx, scene in enumerate(scene_detail_data.get("scenes", [])): for v_idx, v in enumerate(scene.get("t2i_variations", [])): state[(s_idx, v_idx)] = v.get("render_prompt_card_hash") return state def _assert_card_hash_unchanged( scene_detail_data: Dict[str, Any], applied_indices: List[Tuple[int, int]], pre_state: Dict[Tuple[int, int], Optional[str]], ) -> None: """t2i_review 는 card payload 변경 금지 — pre/post hash 일치 검증 (AC-A3). V2 patch P2: pre_state 는 caller 가 _capture_card_hash_state( scene_detail_data) 로 mutation 전 만든 dict — 전체 매핑. 본 helper 는 applied_indices 만 비교. """ for s_idx, v_idx in applied_indices: v = scene_detail_data["scenes"][s_idx]["t2i_variations"][v_idx] post_hash = v.get("render_prompt_card_hash") if pre_state.get((s_idx, v_idx)) != post_hash: raise AppError( code="t2i_review.card_hash_unexpected_change", message=( f"t2i_review 가 card hash 를 변경함 — scope 외 " f"mutation: scenes[{s_idx}].t2i_variations[{v_idx}]" ), )