"""이미지 Phase StepRunner 서브클래스 — Steps 11-14. force 모드: 기존 이미지 삭제 안 함. 새 이미지 추가 생성 (is_primary 이동). resume 모드: 미완성분만 이어서 생성. """ import hashlib import json import logging from typing import Any, Dict, Optional from app.core.name_matcher import build_name_index, lookup_name from app.core.step_runner import StepRunner from app.core.subject_state import get_visual_descriptor, is_immobilized_state from app.services.image_capture.annotate import annotate_generated_asset from app.core.entity_identity import PRESENCE_SHELVED as _PRESENCE_SHELVED logger = logging.getLogger(__name__) class _ImageStepMixin: """이미지 단계 공통.""" def _setup_image_tracer_context(self): """ImageTracer에 Opik context 전달 — thread pool worker에서도 유지.""" try: from app.modules.llm.image_tracer import get_image_tracer get_image_tracer().set_context(self.build_opik_metadata()) except Exception as exc: logger.debug("image tracer context setup failed: %s — 본 처리 계속", exc) def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict]: from pathlib import Path from app.core.config import settings cp = Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" if cp.exists(): return json.loads(cp.read_text(encoding="utf-8")) return None def _get_system_actor_id(self) -> str: """시스템 작업용 actor_id — admin 사용자 ID 반환.""" from app.models.catalog import UserAccount user = self.db.query(UserAccount).filter(UserAccount.role == "admin").first() if not user: user = self.db.query(UserAccount).first() if not user: raise RuntimeError("No user accounts exist; cannot determine system actor_id") return user.id def _sync_analysis_to_db(self): """StepRunner 분석 체크포인트 → DB 동기화. 이미지 서비스는 episode.status='analyzed' + EntityCanon 존재를 요구. StepRunner는 체크포인트에만 저장하므로 이미지 단계 전에 DB에 반영. Phase 2.1 Service 오케스트레이터에 위임 (API private 역의존 해소). W4 P3-2 (Codex High 반영): 이 경로는 image step **실행 전** 선행 sync 이므로 step_id를 전달하지 않는다. post-step 신호가 아님. 실패 시 별도 분석 step의 sync_status에 영향을 주면 잘못된 해석을 유발. """ from app.services.checkpoint_sync import orchestrate_full_sync orchestrate_full_sync(self.project_id, self.episode_id, self.db) # ── Step 11: world_guide ── class WorldGuideStep(_ImageStepMixin, StepRunner): def _execute(self, mode="resume") -> Dict[str, Any]: self._setup_image_tracer_context() from app.models.project import Episode from sqlalchemy.orm import undefer from app.modules.world_guide_generator import WorldGuideGenerator from app.models.project import WorldGuide, EntityCanon, EntityEpisodeLink, SceneStill import hashlib, uuid from datetime import datetime, timezone ep = self.db.query(Episode).options(undefer(Episode.fulltext)).filter(Episode.id == self.episode_id).first() # 엔티티 + 씬 # ★보류(shelved) 제외 — 정본 하나 (2026-09-04). from app.core.entity_identity import active_episode_canon_ids canon_ids = active_episode_canon_ids( self.db, self.project_id, self.episode_id) entities_orm = self.db.query(EntityCanon).filter(EntityCanon.id.in_(canon_ids)).all() if canon_ids else [] entities = [{"id": e.id, "name": e.name, "entity_type": e.entity_type, "description": e.description or ""} for e in entities_orm] # shot-more: is_selected=True만 world_guide에 포함 stills_orm = self.db.query(SceneStill).filter( SceneStill.project_id == self.project_id, SceneStill.episode_id == self.episode_id, SceneStill.is_selected == True, # noqa: E712 SceneStill.still_index >= 0, SceneStill.status != "stale", ).all() stills = [{"still_frame_prompt": s.still_frame_prompt or ""} for s in stills_orm] ft = ep.fulltext or "" wg_hash = hashlib.md5(f"{ft[:500]}:{len(entities)}:{len(stills)}".encode()).hexdigest() # resume: 기존 WorldGuide 재사용 if mode == "resume": existing = self.db.query(WorldGuide).filter( WorldGuide.project_id == self.project_id, WorldGuide.episode_id == self.episode_id, ).order_by(WorldGuide.created_at.desc()).first() if existing and existing.source_hash == wg_hash: return {"completed_count": 1, "applicable_count": 1, "failed_count": 0, "data": json.loads(existing.guide_json)} wg_gen = WorldGuideGenerator() world_guide = wg_gen.generate( fulltext=ft, language=ep.language or "ko", source_file=ep.source_filename or "episode", entities=entities, stills=stills, project_config=self.project_config, ) now = datetime.now(timezone.utc).isoformat() self.db.add(WorldGuide( id=str(uuid.uuid4()), project_id=self.project_id, episode_id=self.episode_id, guide_json=json.dumps(world_guide, ensure_ascii=False), source_hash=wg_hash, created_at=now, )) self.db.commit() return {"completed_count": 1, "applicable_count": 1, "failed_count": 0, "data": world_guide} # ── Step 12: ref_image_gen ── class RefImageGenStep(_ImageStepMixin, StepRunner): def _config_hash(self) -> str: """`v2_chunk` 에서만 canonical ref 입력 전체의 지문을 접는다 (Codex D 2026-09-02). 사람이 사진을 verified/rejected 로 바꾸면 이 값이 달라져 스텝이 SKIP 으로 빠지지 않는다 — 실제로 다시 굽는 것은 orchestrator 의 지문 대조가 정한다 (바뀐 entity 만). legacy/v2 는 base 지문 **그대로**(옛 CP 보존). """ from app.core.grounding_mode import resolve_grounding_mode from app.core.step_runner import compute_config_hash from app.modules.pipeline.grounding_canonical_ref_inputs import ( canonical_grounding_digest, central_cp_is_required, ) base = compute_config_hash(self.project_config) mode = resolve_grounding_mode(self.project_config) if not central_cp_is_required(mode): return base import hashlib as _hl digest = canonical_grounding_digest( self.db, project_id=self.project_id, episode_id=self.episode_id, grounding_mode=mode) return _hl.sha256( f"{base}|grounding_inputs={digest}".encode("utf-8") ).hexdigest()[:len(base)] def _execute(self, mode="resume") -> Dict[str, Any]: self._setup_image_tracer_context() from app.services.reference_image_service import ReferenceImageService # StepRunner 분석 결과를 DB에 동기화 (이미지 생성 게이트 통과용) self._sync_analysis_to_db() if mode == "force": # 기존 이미지 삭제 안 함 — is_primary만 해제하여 새 이미지가 primary 됨. # Phase 3.1 Codex 리뷰 반영: composite(outfit) primary는 CompositeImageGenStep # 책임이므로 여기서 건드리지 않음. prompt_used의 '[outfit:' 마커로 제외. from sqlalchemy import text self.db.execute(text( "UPDATE image_asset SET is_primary = 0 " "WHERE project_id = :pid AND episode_id = :eid " "AND asset_type = 'reference' AND is_primary = 1 " "AND COALESCE(prompt_used, '') NOT LIKE '[outfit:%' " "AND COALESCE(prompt_used, '') NOT LIKE '[composite:%'" ), {"pid": self.project_id, "eid": self.episode_id}) self.db.commit() # 체크포인트 삭제 from pathlib import Path from app.core.config import settings cp_path = Path(settings.projects_dir) / self.project_id / "checkpoints" / "images" / self.episode_id / "reference_checkpoint.json" if cp_path.exists(): cp_path.unlink() svc = ReferenceImageService(db=self.db, project_id=self.project_id, actor_id=self._get_system_actor_id()) # Phase 3b.4: generate_base_references public API로 전환 — RefImageGenStep의 # 책임(Phase 1 base reference만) 명시화. # Phase 3.1 경계 재정의: 이전에는 composite까지 생성 후 _mark_composite_done으로 # 상태 붕괴를 유발했음. result = svc.generate_base_references( episode_id=self.episode_id, mode="resume", ) generated = result.get("reference_count", 0) or result.get("generated", 0) skipped = result.get("skipped", 0) failed = result.get("failed", 0) total = result.get("total", generated + skipped + failed) return { "completed_count": generated + skipped, "applicable_count": total, "failed_count": failed, "data": result, # ★저장/비교 대칭 — 비교 문은 `_config_hash` 를 보므로 저장도 그 값 "config_hash": self._config_hash(), } def verify_completion(self): """이 에피소드의 ref 대상이 **정본 참조를 갖고 있는가**. 「**대상은 episode, 자산은 canon/project**」 — 게이트와 같은 물음이다. 대상은 이 에피소드에 연결된 `entity_type ∉ {location, outlook}` 엔티티 (생산자 `reference_phase1_service.py:92` 와 같은 형태 — prop 누락이 false-clean 으로 묻히지 않는다), 저빈도 스킵은 뺀다. 자산은 `project + entity` 의 primary 참조 + **파일 실재**. ★`ImageAsset.episode_id` 로 묻지 않는다. 참조의 primary 는 프로젝트 안에서 canon 당 하나이고(새 참조가 그 범위로 옛 primary 를 내린다), 에피소드로 물으면 **재사용된 canon 이 거짓 미완료**가 된다. cleanup_artifacts override 안 함 → default noop (부분 재생성 보호). """ from app.core.integrity_report import CompletionReport from app.core.pipeline_gate import ( entities_with_reference, reference_asset_row_count, reference_targets, ) # ★게이트(`pipeline_gate.check_scene_images_ready`)와 **같은 정본**을 쓴다. # 종전에는 자산을 `ImageAsset.episode_id == self.episode_id` 로 물어서, # 앞 에피소드에서 만든 참조를 다시 쓰는 canon 을 **거짓 미완료**로 찍었다 # (2026-08-29 실측: 3화 expected 4 · found 2 인데 게이트는 4/4 통과이고 # 스틸컷 6장이 다 나왔다). 참조의 primary 는 canon/project 당 하나다. # 거짓 미완료는 `partial` 로 굳고 매 resume 마다 RERUN_SELF 를 부른다. expected_entities = reference_targets( self.db, self.project_id, self.episode_id) expected_ids = [e.id for e in expected_entities] expected = len(expected_ids) if expected == 0: return CompletionReport( is_complete=True, missing=[], severity="clean", metadata={"expected": 0, "found": 0, "rows": 0}, ) found = len(entities_with_reference( self.db, self.project_id, expected_ids)) # 행 수는 **진단**이다 — 「행이 없다」와 「파일이 사라졌다」를 가른다. rows = reference_asset_row_count(self.db, self.project_id, expected_ids) if found < expected: return CompletionReport( is_complete=False, missing=[ f"{expected - found} character ref images missing " f"(expected={expected}, found={found})" ], severity="missing" if found == 0 else "partial", metadata={"expected": expected, "found": found, "rows": rows}, ) return CompletionReport( is_complete=True, missing=[], severity="clean", metadata={"expected": expected, "found": found, "rows": rows}, ) # ── Step 13: composite_image_gen ── def _drift_ack_decision(step_id: str, mismatch_reason: str): """`STEP_CONFIG_DRIFT_ACK` 에 이 스텝이 적혀 있으면 **비파괴 재실행**. ## 왜 공용 함수인가 이 규칙은 `SceneImagePipelineStep` 안에만 있었다. 합성·상태변형에 지문을 넣자(`a58417b2`) 그 둘이 stale 이 됐는데 승인 문이 없어 「force 로 재실행하라」만 나왔다 — force 는 **멀쩡한 자산까지 전부** 다시 굽는다(합성 24장 중 고칠 것은 4장뿐이었다). 같은 규칙을 세 곳에 베껴 두면 한 곳만 고쳐진다. 그래서 한 곳에 둔다. ★`RERUN_SELF` 는 cleanup/invalidate 없이 `_execute(resume)` 를 다시 부르는 것이라 **완료 자산이 안 지워진다**. 서비스가 파일이 있는 것은 건너뛰므로, 다시 만들 것만 다시 만든다. """ from app.core.config import settings as _settings from app.core.step_runner import ResumeAction, ResumeDecision ack = { x.strip() for x in str(getattr(_settings, "step_config_drift_ack", "")).split(",") if x.strip() } if step_id not in ack: return None logger.warning( "Step %s: config drift 를 env 승인으로 통과 — 비파괴 resume 재실행 (%s)", step_id, mismatch_reason) return ResumeDecision( action=ResumeAction.RERUN_SELF, reason=("config drift — step_config_drift_ack 명시 승인, " f"비파괴 resume 재실행: {mismatch_reason}"), origin="config_drift_ack", ) def _fold_prompt_packs(base: str, module: str, stems) -> str: """`base` 지문에 팩 **버전**을 접는다 (2026-09-19). ## 왜 필요한가 합성·상태변형 단계에는 `_config_hash` 가 **아예 없었다**. 둘 다 빈 dict 지문(`99914b932bd37a50`)을 쓰고 있어서, `ref_image_prompts` 팩을 고쳐도 스텝이 무효화되지 않았다 — 고친 문안이 **나가는 프롬프트까지 못 내려간다.** `#94`(cine_stage_direction)·`1-E`(prev_usage_clause)와 같은 부류다. 실측으로 그 자리를 밟았다: 옷 참조의 마네킹이 같이 그려지는 결함을 팩 v7 로 고쳤는데, 지문이 안 움직여 보통 resume 에서는 옛 그림이 그대로 남았을 것이다. ★버전 문자열만 접는다. 바이트를 접으면 주석 한 줄만 고쳐도 유료 단계가 통째로 다시 돈다 — 발행된 팩은 덮어쓰지 않는 것이 이 저장소 규칙이라 버전이 곧 내용이다. """ import hashlib as _hl from app.modules.prompt_loader import resolve_effective parts = [] for stem in stems: try: parts.append( f"{module}/{stem}=" f"{resolve_effective(module, stem, kind='prompt')['version']}") except Exception: # noqa: BLE001 # ★못 읽은 것을 「없다」로 읽지 않는다 — 표식을 남겨 지문이 # 조용히 옛 값과 같아지는 일을 막는다. parts.append(f"{module}/{stem}=") return _hl.sha256( ("|".join([base] + sorted(parts))).encode("utf-8") ).hexdigest()[:len(base)] class CompositeImageGenStep(_ImageStepMixin, StepRunner): def _evaluate_contract_drift(self, mismatch_reason: str): """합성 팩 수리로 지문이 움직였을 때, env 승인이 있으면 **비파괴** 재실행. 없으면 기존대로 BLOCK — 승인 없는 drift 는 막는다.""" _acked = _drift_ack_decision(self.step_id, mismatch_reason) if _acked is not None: return _acked return super()._evaluate_contract_drift(mismatch_reason) def _config_hash(self) -> str: """합성이 쓰는 **팩 버전**을 접는다 — 없으면 팩 수리가 안 내려간다.""" from app.core.step_runner import compute_config_hash return _fold_prompt_packs( compute_config_hash(self.project_config), "ref_image_prompts", ("character_outlook_ref", "character_composite_ref", "character_composite_derive_ref")) def _execute(self, mode="resume") -> Dict[str, Any]: self._setup_image_tracer_context() from app.services.reference_image_service import ReferenceImageService from app.models.project import CharacterOutlook, ImageAsset, EntityEpisodeLink, EntityCanon from sqlalchemy import text from pathlib import Path import re # W20F10 R1: composite step 단독 force/resume 또는 RefImageGenStep 을 거치지 않은 # 실행 경로에서도 ReferencePipelineOrchestrator.run 의 episode.status guard 가 # stale projection 으로 막히지 않도록, RefImageGenStep 과 동일하게 DB 조회 전에 sync. self._sync_analysis_to_db() # 에피소드에 연결된 캐릭터 ID # ★보류(shelved) 제외 — 정본 하나 (2026-09-04). from app.core.entity_identity import active_episode_canon_ids _active = active_episode_canon_ids(self.db, self.project_id, self.episode_id) char_ids = set() for _cid in _active: ec = self.db.query(EntityCanon).filter( EntityCanon.id == _cid, EntityCanon.entity_type == "character" ).first() if ec: char_ids.add(ec.id) # 에피소드 스코프 조합 (O00 Null Outlook 제외) _o00_ids = { e.id for e in self.db.query(EntityCanon).filter( EntityCanon.project_id == self.project_id, EntityCanon.short_id == "O00", EntityCanon.entity_type == "outlook", ).all() } _all_combos = self.db.query(CharacterOutlook).filter( CharacterOutlook.project_id == self.project_id, CharacterOutlook.character_id.in_(char_ids), ).all() if char_ids else [] # ref_image_gen에서 저빈도 스킵된 캐릭터의 combo도 applicable에서 제외. # 그렇지 않으면 ref_image_gen이 스킵 → composite도 스킵 → but combo_total에는 # 포함돼 failed_count 부풀려짐 (실측 Breakout Sample partial 1/2 원인). from app.core.low_freq_skip import load_low_freq_skip_ids _low_freq_skip_ids: set = load_low_freq_skip_ids(self.project_id, self.episode_id) combo_total = sum( 1 for c in _all_combos if c.outlook_id not in _o00_ids and c.character_id not in _low_freq_skip_ids ) if mode == "force": # 기존 합성 이미지의 DB 레코드를 삭제하지 않고, # 서비스의 skip 조건(file exists)을 우회하기 위해 # 기존 합성 이미지 prompt_used에서 outlook_id를 변경하여 매칭 안 되게 함 # → 서비스가 "없는 것"으로 판단하고 새로 생성 composites = self.db.query(ImageAsset).filter( ImageAsset.project_id == self.project_id, ImageAsset.asset_type == "reference", ImageAsset.prompt_used.like("%composite:%"), ImageAsset.entity_id.in_(char_ids) if char_ids else False, ).all() for ca in composites: # prompt_used에 _old 접미사 추가 → 새 생성 시 매칭 안 됨 if ca.prompt_used and "_old_" not in ca.prompt_used: ca.prompt_used = ca.prompt_used.replace("composite:", "composite_old:") self.db.commit() logger.info("Composite force: marked %d existing composites as old", len(composites)) # 합성 생성 (누락분만) — Phase 3b.4: generate_composites public API 사용. # resume 모드로 Phase 1(base ref)은 이미 완료된 entity만 skip되므로 # 실질적으로 Phase 2 (outfit 단독) + Phase 3 (composite 전신) 만 실행. svc = ReferenceImageService(db=self.db, project_id=self.project_id, actor_id=self._get_system_actor_id()) svc.generate_composites(episode_id=self.episode_id, mode="resume") # 결과 카운트 — 에피소드 스코프 캐릭터의 합성만 # composite 형식: [composite:{char_id}:{outlook_id}] composite_assets = self.db.query(ImageAsset).filter( ImageAsset.project_id == self.project_id, ImageAsset.asset_type == "reference", ImageAsset.prompt_used.like("%composite:%"), ImageAsset.entity_id.in_(char_ids) if char_ids else False, ).all() if char_ids else [] # 분모(combo_total)와 동일한 필터를 분자(existing_keys)에도 적용: # O00 Null Outlook 및 저빈도 스킵 캐릭터의 기존 composite 레코드는 집계 제외. # (과거 생성본이 남아있어도 현재 기준 "해당 없음"으로 처리해 failed_count 오판 방지) existing_keys = set() for ca in composite_assets: m = re.search(r'composite:([a-f0-9-]+):([a-f0-9-]+)', ca.prompt_used or "") if not m: continue char_id, outlook_id = m.group(1), m.group(2) if outlook_id in _o00_ids: continue if char_id in _low_freq_skip_ids: continue existing_keys.add(f"{char_id}:{outlook_id}") # ★★**착용자 0인 아웃룩을 산출에 남긴다** (2026-09-19, Codex NON-BLOCK). # 검사기는 이것을 기대에서 빼는데, 그 수가 `CompletionReport.metadata` # 에만 있으면 **완료 체크포인트에는 한 글자도 안 남는다** — 나중에 # 「그때 옷이 몇 벌 비어 있었나」를 물을 자리가 없다. # 이 수는 「무배정과 저빈도 제외가 다르다」만 말한다. 무배정이 옳다는 # 증명은 아니다 — 그건 사람이 본다. _no_wearer = [ o.id for o in self.db.query(EntityCanon).join( EntityEpisodeLink, EntityEpisodeLink.canon_id == EntityCanon.id, ).filter( EntityEpisodeLink.project_id == self.project_id, EntityEpisodeLink.episode_id == self.episode_id, EntityCanon.entity_type == "outlook", EntityCanon.short_id != "O00", ).all() if o.id not in {c.outlook_id for c in _all_combos} ] return { "completed_count": len(existing_keys), "applicable_count": combo_total, "failed_count": max(0, combo_total - len(existing_keys)), "data": { "composite_done": len(existing_keys), "combo_total": combo_total, "outlook_no_wearer_count": len(_no_wearer), "outlook_no_wearer_ids": sorted(_no_wearer), }, } def verify_completion(self): """Composite step의 두 산출물 검증: 1. char×outlook composite (prompt_used~'composite:char:outlook') — Phase 1 2. outlook standalone (asset_type='reference', entity_type='outlook', O00 제외) — Phase 2 O00 + low_freq skip 제외. cleanup_artifacts override 안 함 → default noop (부분 재생성 보호). """ import re as _re from app.core.file_paths import resolve_image_path from app.core.integrity_report import CompletionReport from app.core.low_freq_skip import load_low_freq_skip_ids from app.models.project import ( CharacterOutlook, EntityCanon, EntityEpisodeLink, ImageAsset, ) # === 공통 — 에피소드의 character + outlook ids === char_ids = { r[0] for r in self.db.query(EntityEpisodeLink.canon_id).join( EntityCanon, EntityCanon.id == EntityEpisodeLink.canon_id ).filter( EntityEpisodeLink.project_id == self.project_id, EntityEpisodeLink.episode_id == self.episode_id, # ★보류(shelved) 제외 (2026-09-04). EntityEpisodeLink.presence_status != _PRESENCE_SHELVED, EntityCanon.entity_type == "character", ).all() } outlook_ids = [ r[0] for r in self.db.query(EntityEpisodeLink.canon_id).join( EntityCanon, EntityCanon.id == EntityEpisodeLink.canon_id ).filter( EntityEpisodeLink.project_id == self.project_id, EntityEpisodeLink.episode_id == self.episode_id, # ★보류(shelved) 제외 (2026-09-04). EntityEpisodeLink.presence_status != _PRESENCE_SHELVED, EntityCanon.entity_type == "outlook", EntityCanon.short_id != "O00", ).all() ] # O00 ids 제외 o00_ids = { e.id for e in self.db.query(EntityCanon).filter( EntityCanon.project_id == self.project_id, EntityCanon.short_id == "O00", EntityCanon.entity_type == "outlook", ).all() } skipped = load_low_freq_skip_ids(self.project_id, self.episode_id) missing_messages: list[str] = [] metadata: Dict[str, Any] = {} # === Phase 1: composite pair (char × outlook) === # ★이 화의 배정만 (Codex BLOCK 2026-09-04). from app.core.entity_identity import episode_outlook_rows all_combos = [ c for c in episode_outlook_rows(self.db, self.project_id, self.episode_id) if c.character_id in char_ids ] if char_ids else [] expected_pairs = [ (c.character_id, c.outlook_id) for c in all_combos if c.outlook_id not in o00_ids and c.character_id not in skipped ] pair_expected = len(expected_pairs) pair_found = 0 if pair_expected > 0: composite_rows = self.db.query(ImageAsset).filter( ImageAsset.project_id == self.project_id, ImageAsset.asset_type == "reference", ImageAsset.prompt_used.like("%composite:%"), ).all() existing_pairs = set() for r in composite_rows: m = _re.search(r'composite:([a-f0-9-]+):([a-f0-9-]+)', r.prompt_used or "") if not m: continue p = resolve_image_path(r.file_path) if p and p.exists(): existing_pairs.add((m.group(1), m.group(2))) pair_found = sum(1 for pair in expected_pairs if pair in existing_pairs) if pair_found < pair_expected: missing_messages.append( f"{pair_expected - pair_found} composites missing " f"(expected={pair_expected}, found={pair_found})" ) metadata["composite_expected"] = pair_expected metadata["composite_found"] = pair_found # === Phase 2: outlook standalone === # A7 fix: outlook은 character/prop UUID set과 직접 비교하면 type mismatch # (outlook UUID ∉ character UUID set → 항상 false → filter no-op). # production reference_phase2_service.py:122는 wearer character_id가 # low-freq면 outlook generation을 skip한다 (per-row). verify는 이를 # outlook 단위로 aggregate: 모든 wearer가 low-freq일 때만 skip. # ★★wearer 0명인 outlook 은 **낼 곳이 없다** (2026-09-19). # 만드는 쪽(reference_phase2_service)은 **착용자 쌍을 돌면서만** # 만든다. 쌍이 0이면 그 옷은 영원히 안 만들어지므로, 기대에 넣으면 # 「계약은 있는데 내는 곳이 없다」가 되어 멀쩡한 주행이 영구 partial 이 # 된다. 실측(컨트리로드 2판): O30 연구원가운·O31 청소작업복·O40 미군복. # ★조용히 넘기지는 않는다 — 몇 개가 그랬는지 metadata 로 남긴다. # ★이 셋은 사라진 게 아니다. 아웃룩 목록 CP 에는 주인이 적혀 있고 # (O30→C34·O31→C35·O40→C73) 그 주인들이 어느 샷에도 안 나와서 # 씬 배정이 없을 뿐이다. 즉 추출 누락이 아니라 **쓸 자리가 없는 옷**. # ★이 화의 배정만 — 다른 화의 착용자가 이 화의 skip 판단에 섞이면 # 안 된다 (Codex BLOCK 2026-09-04). wearer_links = [ c for c in episode_outlook_rows(self.db, self.project_id, self.episode_id) if c.outlook_id in outlook_ids ] if outlook_ids else [] wearers_by_outlook: Dict[str, set] = {} for link in wearer_links: wearers_by_outlook.setdefault(link.outlook_id, set()).add(link.character_id) def _outlook_kept(oid: str) -> bool: wearers = wearers_by_outlook.get(oid) if not wearers: return False # 착용자 0 = producer 0 — 기대에 넣지 않는다 return not wearers.issubset(skipped) outlook_expected_ids = [oid for oid in outlook_ids if _outlook_kept(oid)] outlook_expected = len(outlook_expected_ids) # ★넘긴 것을 **세어서 남긴다.** 이 수가 0 이 아니면 아웃룩 배정이 # 어딘가 비었다는 뜻이니, 통과했다고 안 본 게 아니라 본 것이다. metadata["outlook_no_wearer_skipped"] = sum( 1 for oid in outlook_ids if not wearers_by_outlook.get(oid)) outlook_found = 0 if outlook_expected > 0: rows = self.db.query(ImageAsset).filter( ImageAsset.project_id == self.project_id, ImageAsset.asset_type == "reference", ImageAsset.entity_id.in_(outlook_expected_ids), ).all() for r in rows: p = resolve_image_path(r.file_path) if p and p.exists(): outlook_found += 1 if outlook_found < outlook_expected: missing_messages.append( f"{outlook_expected - outlook_found} outlook standalones missing " f"(expected={outlook_expected}, found={outlook_found})" ) metadata["outlook_expected"] = outlook_expected metadata["outlook_found"] = outlook_found # === 종합 판정 === total_expected = pair_expected + outlook_expected total_found = pair_found + outlook_found if total_expected == 0: return CompletionReport( is_complete=True, missing=[], severity="clean", metadata=metadata, ) if total_found < total_expected: return CompletionReport( is_complete=False, missing=missing_messages, severity="missing" if total_found == 0 else "partial", metadata=metadata, ) return CompletionReport( is_complete=True, missing=[], severity="clean", metadata=metadata, ) # ── Step 14: scene_image_pipeline (compound) ── class SceneImagePipelineStep(_ImageStepMixin, StepRunner): def validate_mode(self, mode: str) -> None: """표적 씬 슬라이스+force 병용 금지 (Codex 재리뷰 BLOCKING-1). run() 최선두(claim/cleanup_artifacts/invalidate_downstream/ clear_checkpoint 이전) preflight — _execute 내부 가드만으로는 public run(force) 의 파괴 경로가 먼저 지나간다. force 는 에피소드 전체 primary 해제+checkpoint 삭제+recipe 아카이브라 표적 모드와 병용 시 비표적 자산까지 무효화 (부분 force=별도 범위). #77-B 상한 래치(Codex 3차 리뷰): 같은 이유로 래치 검사도 여기 — _execute 첫 문장 검사만으로는 public run(force) 이 그 전에 invalidate_downstream·clear_checkpoint 로 기록을 변형한다. 래치가 있으면 mode 무관, 어떤 변형·지출보다 먼저 멈춘다. (_execute 초입·recipe 걷기 초입 검사는 다른 진입로 방어로 유지) """ from app.core.config import settings as _s77 if bool(getattr(_s77, "still_jit_verify_enabled", True)): from pathlib import Path as _P77 from app.services.still_recipe_service import ( JIT_LATCH_FILENAME as _JIT_LATCH, StillJitRegenLimitExceeded as _JitLatchErr, ) _latch = ( _P77(_s77.projects_dir) / self.project_id / "images" / self.episode_id / "scene" / "recipe" / _JIT_LATCH ) if _latch.exists(): raise _JitLatchErr( f"JIT 재생성 상한 래치가 남아 있다: {_latch} — 사람 " f"확인 전에는 mode 와 무관하게 이 스텝을 실행하지 " f"않는다(선행 유료 구간·downstream 무효화·checkpoint " f"삭제 전부 차단). 원인 확인 후 래치 파일을 삭제하고 " f"resume.") if mode != "force": return from app.core.config import settings from app.modules.pipeline.scene_image_scope import ( parse_target_scenes, ) if parse_target_scenes( getattr(settings, "scene_image_target_scenes", "")): from app.core.errors import AppError raise AppError( code="scene_image.target_force_forbidden", message=( "scene_image_target_scenes 설정과 mode=force 병용 금지 " "— force 는 에피소드 전체 무효화라 비표적 자산을 " "파괴한다 (fail-closed)" ), status_code=422, ) def _evaluate_contract_drift(self, mismatch_reason: str): """표적 목록 변경만의 drift = 비파괴 RERUN_SELF (Codex 재리뷰 BLOCKING-2). 표적은 config_hash 에 접혀 있어(아래 _config_hash) 목록 변경 시 cp_mismatch → 기본 정책은 BLOCK+force 요구인데 표적+force 는 금지 — A scope 완주 후 B scope 로 바꾸면 영구 BLOCK 이던 교착 해소. CP 에 영속된 target_scope_base_hash(표적 무관 base)가 현재 base 와 같으면 '표적 목록만 바뀐 drift' 로 분류 — RERUN_SELF (origin=contract_drift) 는 cleanup/invalidate/clear 없이 _execute(resume) 만 수행하므로 비표적 자산 무손실. base 자체가 다르면(팩·모델 등 실질 변경) 기본 BLOCK 유지. """ if "config_hash mismatch" in mismatch_reason: from app.core.step_runner import ResumeAction, ResumeDecision cp = self.load_checkpoint() stored_base = (cp or {}).get("target_scope_base_hash") if not stored_base: # 무표적 CP 전이 fallback (Codex 재재리뷰 HIGH-2): 무표적 # CP 는 표적 필드를 영속하지 않으므로(shape 불변 계약) # 저장된 config_hash 자체가 표적 fold 없는 값 — base 와 # 직접 대조. save_checkpoint override(3차 HIGH)가 신규 # CP 에 step-local hash 를 저장해 이 대조가 실제 작동. # 이 override 이전의 구 CP(md5 fallback 저장분)는 recipe # ON base(sha256)와 불일치 → 기본 BLOCK 유지(안전: # false-RERUN 없음, 오탐 방향은 항상 BLOCK). stored_base = (cp or {}).get("config_hash") if stored_base and stored_base == self._config_hash_base(): return ResumeDecision( action=ResumeAction.RERUN_SELF, reason=( "target-scope drift only (base hash 동일) — " f"비파괴 resume 재실행: {mismatch_reason}" ), origin="contract_drift", ) # #77-A (2026-08-09 합의): base 자체가 움직인 drift 도, 사용자가 # env 로 이 스텝을 **명시 승인**했을 때만 비파괴 resume 재실행. # RERUN_SELF 는 cleanup/invalidate/clear 없이 _execute(resume) # 라 완료 자산 무손실이고(runner :1141), origin 이 runner 의 # 어느 force 격상 갈래에도 안 걸려 mode=resume 그대로 떨어진다. # 걷는 동안은 샷별 JIT 지문 검증(#77-B)이 낡은 샷만 재생성. # 승인 없는 config drift 는 기존대로 BLOCK(사용자 의도 확인). _acked = _drift_ack_decision(self.step_id, mismatch_reason) if _acked is not None: return _acked return super()._evaluate_contract_drift(mismatch_reason) def _config_hash(self) -> str: """표적 씬 슬라이스 스탬프 (Codex NARROW-6): 표적 설정은 산출 집합의 실질 입력 — recipe OFF(legacy) 경로 포함 전 경로에서 표적 변경=stale. 빈 설정=_config_hash_base byte-identical. scope 계약 버전(SCOPE_CONTRACT_VERSION)도 병행 스탬프 (Codex 재리뷰 HIGH-3). """ from app.core.config import settings from app.modules.pipeline.scene_image_scope import ( SCOPE_CONTRACT_VERSION, parse_target_scenes, ) base = self._config_hash_base() targets = parse_target_scenes( getattr(settings, "scene_image_target_scenes", "")) if not targets: return base import hashlib as _hl return _hl.sha256( ( f"{base}|scene_targets={','.join(map(str, targets))}" f"|scope_contract={SCOPE_CONTRACT_VERSION}" ).encode("utf-8") ).hexdigest()[:len(base)] def _config_hash_base(self) -> str: """still_recipe(2026-07-13) opt-in stamping — mode OFF 는 base hash byte-identical(legacy cp 보존), "v1" 일 때만 레시피 키를 섞는다. 표적 목록은 여기 접지 않는다 — target-scope drift 판별의 비교 기준으로 CP(target_scope_base_hash)에 영속된다.""" from app.core.config import settings from app.core.step_runner import compute_config_hash base = compute_config_hash(self.project_config) # ★zoom 크롭이 켜진 판에서 **bbox 를 내는 모델**을 접는다 # (2026-08-27 Codex BLOCK 2회차). # # ★★**조기 return 앞이어야 한다.** 첫 판은 이 아래 recipe 구간에 # 넣었는데, `still_recipe_mode == "off"` 면 여기서 먼저 돌아가 # **정작 `generate_continuity_crop_png` 가 도는 legacy batch # 경로에서는 지문에 한 번도 안 들어갔다.** 내 시험이 recipe # 모드를 안 고정해서 활성 기본값만 재고 반대편을 놓쳤다. # # ★OFF 에서는 칸을 안 만든다 — byte-identical. if getattr(settings, "zoom_continuity_anchor_enabled", False): import hashlib as _zh import json as _zj from app.services.zoom_continuity_render_service import ( VISION_MODEL_DEFAULT as _zoom_bbox_model, ) base = _zh.sha256(_zj.dumps( {"base": base, "zoom_bbox_model": _zoom_bbox_model}, sort_keys=True).encode("utf-8")).hexdigest()[:16] if getattr(settings, "still_recipe_mode", "off") == "off": return base import hashlib import json as _json from app.modules.pipeline.multiroll_gemini import ( STILL_JUDGE_PACK_VERSION as _still_judge_sel, judge_pack_content_hash as _judge_pack_content, resolve_judge_pack_version as _judge_pack_resolved, ) from app.modules.pipeline.shot_continuity_author import ( CARRIED_PERSON_PROJECTION_VERSION as _carried_person_proj, ) from app.modules.pipeline.still_recipe import ( TEXT_ONLY_LOCATION_PROMPT_VERSION as _TEXT_ONLY_LOC_SEL, recipe_pack_content_hash as _recipe_pack_content, resolve_prompt_version as _recipe_pack_resolved, ) payload = { "base": base, "still_recipe_mode": settings.still_recipe_mode, # ★CARRIED 인물 문장을 정본 표기에 결속하는 계약 (#104, # 2026-08-29). 프롬프트 **바이트가 바뀐다** — 접지 않으면 # outer 가 clean-skip 해 샷별 JIT 까지 못 내려간다. # ★조기 return **뒤**가 맞다: 이 값을 쓰는 두 still 호출부 # (`still_recipe_service:2982·3584`)가 **둘 다** # `run_still_recipe_generation` 안, 즉 레시피 경로다. # legacy(`still_recipe_mode=="off"`)는 이 값을 안 쓴다 — # 거기서 접으면 안 닿는 것을 stale 로 만든다. "carried_person_projection": _carried_person_proj, "still_recipe_roll_count": settings.still_recipe_roll_count, "still_recipe_critique_enabled": settings.still_recipe_critique_enabled, # Codex 1차 리뷰 HIGH-5: 팩·모델·맵 분기도 출력 실질 입력 # #77-A (2026-08-09): 하드코딩 버전 문자열 → 실사용 selector # 해석값 + 팩 '내용' 해시. 문자열 스탬프는 셀렉터 상수·해석 # 코드·서버 메모리 중 어느 층이 낡아도 안 움직여, 팩만 올린 # 재실행이 whole-step clean skip 되던 구멍(최종 스틸 253/311 # 재사용 사고의 스텝 층). 여기가 움직이면 resume 은 승인 env # (step_config_drift_ack) 아래 비파괴 재실행으로 걷고, 샷별 # JIT 검증(#77-B)이 낡은 샷만 다시 만든다. "still_recipe_pack": _recipe_pack_resolved("1"), "still_recipe_pack_content": _recipe_pack_content("1"), # E2E6 ⑥: 하드코딩 스탬프 → 실사용 selector 해석값으로 동기. # 2026-08-07: 스틸이 실제로 로드하는 selector 와 같아야 한다 — # 전역 기본을 읽으면 스틸 팩만 올렸을 때 이 hash 가 안 움직여 # whole-step clean skip 이 먼저 일어난다. "multiroll_judge_pack": _judge_pack_resolved(_still_judge_sel), "multiroll_judge_pack_content": _judge_pack_content( _still_judge_sel), "nb2_model": settings.gemini_image_model, "judge_model": "gemini-pro", "outdoor_map_conti_enabled": bool( getattr(settings, "outdoor_map_conti_enabled", False) ), } # ── G+Q 판정 체계 (2026-08-10 설계 §2.5) — ON 일 때만 스탬프. # OFF=키 부재 byte-identical: 이 배포·재기동만으로 완료 스텝의 # config_hash 가 움직이지 않는 것이 계약이다. ON 이면 여기가 # 움직여 step_config_drift(#77-A) → ack 승인 아래 비파괴 재진입 # → 샷별 JIT 검증이 갈라낸다. if bool(getattr(settings, "multiroll_gq_judge_enabled", False)): from app.modules.pipeline.multiroll_gemini import ( GQ_CRITIQUE_PACK_VERSION as _gq_pack_sel, GQ_SELECT_POLICY_VERSION as _gq_policy, ) payload["multiroll_gq_judge_enabled"] = True payload["gq_select_policy"] = _gq_policy payload["gq_critique_pack"] = _judge_pack_resolved(_gq_pack_sel) payload["gq_critique_pack_content"] = _judge_pack_content( _gq_pack_sel) # ── QK 판정 체계 (2026-08-12 전환) — ON 일 때만 스탬프, GQ 관례 # 동형. 이게 없으면 플래그를 올려도 완료 스텝이 clean SKIP 되어 # 전환이 조용히 실행되지 않는다 (Codex 리뷰 지적 — 돈 범주: 판정 # 없이 옛 산출이 새 체계 산출로 오독될 창). if bool(getattr(settings, "multiroll_qk_judge_enabled", False)): from app.modules.pipeline.multiroll_gemini import ( QK_CRITIQUE_PACK_VERSION as _qk_pack_sel, QK_SELECT_POLICY_VERSION as _qk_policy, ) payload["multiroll_qk_judge_enabled"] = True payload["qk_select_policy"] = _qk_policy # 물리 모델 쌍 (2026-08-12 반전: 메인=Qwen DashScope·보조=Gemini) payload["qk_judge_models"] = ( f"{settings.qwen_vlm_model}|{settings.gemini_text_model}") payload["qk_critique_pack"] = _judge_pack_resolved(_qk_pack_sel) payload["qk_critique_pack_content"] = _judge_pack_content( _qk_pack_sel) # ── G+G46 판정 체계 (2026-08-13) — ON 일 때만 스탬프, GQ/QK 관례 # 동형. 이게 없으면 플래그를 올려도 완료 스텝이 clean SKIP 되어 # 전환이 조용히 실행되지 않는다. fix i2i 물리 모델(grok)도 산출 # 실질 입력이라 함께 접는다. if bool(getattr(settings, "multiroll_gg46_judge_enabled", False)): from app.modules.pipeline.multiroll_gemini import ( GG46_CRITIQUE_PACK_VERSION as _gg46_pack_sel, GG46_SELECT_POLICY_VERSION as _gg46_policy, ) payload["multiroll_gg46_judge_enabled"] = True # 초기 선정(②)에 쓰이는 것 — repair 여부와 무관하게 늘 접는다. payload["gg46_select_policy"] = _gg46_policy # ★제시 순서 정책 (2026-08-29) — Gemini 정순 · Grok 역순 2콜. # 샷 지문(`still_recipe_service`)과 **양쪽에** 접는다. 한쪽만 # 넣으면 비대칭이 되어 다른 이유로 스텝이 열릴 때만 걸린다 — # 이번 판 fix-ref 에서 그 실수를 이미 한 번 했다(Codex BLOCK). from app.modules.pipeline.multiroll_gemini import ( CROSS_MODEL_ORDER_POLICY_VERSION as _order_policy, ) payload["select_order_policy"] = _order_policy # ★쌍을 **박지 않는다** (2026-08-29 Codex BLOCK-3). 종전에는 # `gemini|grok` 을 문자열로 지어 넣어서, 둘째 심판이 바뀌어도 # 이 값이 그대로였다 — outer hash 가 안 움직이니 완주 스텝은 # SKIP 되고 **새 판정이 내려가지도 않는다.** resolver 가 실제로 # 돌려주는 물리 쌍을 그대로 받는다. from app.modules.pipeline.multiroll_gemini import ( resolve_select_judge_model_physical as _sel_phys, ) payload["select_judge_models_physical"] = _sel_phys() # ★repair(④~⑥) 전용 — **켜져 있을 때만** (Codex 설계 리뷰). # 안 도는 단계의 팩·모델을 바꾼 것만으로 선정 롤이 stale 이 되면 # 무관한 재생성이 난다. master 기본이 OFF 로 바뀌어 그 갈래가 # 흔해졌다. 샷 지문 쪽도 같이 갈랐다 # (`still_recipe_service.py` 의 `if critique_enabled:`). if bool(getattr(settings, "still_recipe_critique_enabled", False)): payload["gg46_critique_pack"] = _judge_pack_resolved( _gg46_pack_sel) payload["gg46_critique_pack_content"] = _judge_pack_content( _gg46_pack_sel) payload["gg46_fix_image_model"] = str( getattr(settings, "grok_image_model", "")) # ── 참조 선별 (2026-08-19) — ON 일 때만 스탬프, 동형. # 결함 검사 스키마와 수정 호출의 실질 입력(붙는 참조)이 바뀌므로 # 스텝 층 drift 가 걸려야 한다. 여기 없으면 완료 스텝이 바깥에서 # 통째로 건너뛰어져 샷 지문 대조까지 내려가지 않고, 선별 없이 만든 # 산출을 "선별 적용됨"으로 읽게 된다(Codex 지적, 수용). # ★그래서 이 값을 켜면 완료된 에피소드는 재개 시 다시 만들어진다 — # 끝난 판을 다시 열지 않는 것으로 지킨다(gg46·era 관례 동형). # ★`and critique_enabled` — 참조 선별은 **수정 단계 전용**이다 # (2026-08-29, Codex 설계 리뷰). master 가 꺼지면 수정 자체가 안 # 돌아 이 계약이 산출에 닿지 않는다. 그런데도 접으면 선별 문안만 # 고쳐도 **안 도는 단계 때문에 선정 롤이 stale** 이 된다. if (bool(getattr(settings, "still_fix_ref_gate_enabled", False)) and bool(getattr(settings, "still_recipe_critique_enabled", False))): from app.modules.pipeline.multiroll_gemini import ( FIX_MISSING_PACK_VERSION as _fix_missing_sel, fix_ref_contract_sha as _fix_ref_contract, ) payload["still_fix_ref_gate_enabled"] = True payload["fix_missing_pack"] = _judge_pack_resolved( _fix_missing_sel) payload["fix_missing_pack_content"] = _judge_pack_content( _fix_missing_sel) # 선별 지시문(번호 목록 머리글 + 스키마 안 선별 기준 설명)은 # 팩 밖 코드 문자열이라 위 팩 해시가 못 덮는다 — 샷 지문과 같은 # 값을 스텝 층에도 접는다(안팎 관례 동형). payload["fix_ref_contract"] = _fix_ref_contract() # ── 수리 수단 (2026-08-29 사용자 지시) — master ON 일 때만, 동형. # # 편집본과 재생성본은 같은 브리프·같은 참조라도 **다른 그림**이다. # 수단을 바꾼 재주행이 옛 산출을 그대로 안고 가면 안 된다. # # ★`and critique_enabled` — 위 fix-ref 와 같은 이유다. master 가 # 꺼지면 수리 자체가 안 돌아 이 값이 산출에 안 닿는데, 그런데도 # 접으면 **안 도는 단계의 설정을 바꾼 것만으로 선정 롤이 stale** # 이 된다(PR #43 에서 같은 자리를 Codex 가 BLOCK 했다). # ★`edit` 은 한 글자도 안 싣는다 — 종전 산출과 byte-identical. if (str(getattr(settings, "still_repair_method", "edit") or "edit") != "edit" and bool(getattr(settings, "still_recipe_critique_enabled", False))): from app.modules.pipeline.multiroll_gemini import ( STILL_REGEN_PACK_VERSION as _regen_sel, resolve_still_regen_texts as _regen_texts, ) from app.modules.pipeline.multiroll_select import ( REPAIR_METHOD_POLICY_VERSION as _repair_policy, ) payload["still_repair_method"] = str( settings.still_repair_method) payload["still_repair_policy"] = _repair_policy payload["still_regen_pack"] = _regen_sel # ★팩 selector 만 접으면 같은 판 안에서 글을 고쳤을 때 그림이 # 안 따라 움직인다 — **실제 바이트**를 접는다(샷 지문 동형). payload["still_regen_pack_content"] = hashlib.sha256( "\n".join(_regen_texts()[k] for k in sorted( _regen_texts())).encode("utf-8")).hexdigest()[:16] # ── 시대 인지 사전 조사 (2026-08-14) — ON 일 때만 스탬프, 동형. # 판별 정책·팩이 산출 실질 입력이라 스텝 층 drift 가 걸려야 # 완료 스텝에도 조사가 실린다. OFF=키 부재 byte-identical. if bool(getattr(settings, "era_research_enabled", False)): from app.modules.pipeline.era_research import ( ERA_RESEARCH_POLICY_VERSION as _era_policy, era_pack_content_hash as _era_pack_content, resolve_era_pack as _era_pack_resolved, ) payload["era_research_enabled"] = True payload["era_research_policy"] = _era_policy payload["era_research_pack"] = _era_pack_resolved() payload["era_research_pack_content"] = _era_pack_content() # (2026-08-25 Codex BLOCK-3) 판별·선택 **모델**도 산출 실질 # 입력이다 — 팩만 찍으면 모델만 바꾼 운영에서 완료 스텝이 # whole-step SKIP 돼 새 모델이 한 샷에도 안 실린다. 실측: # 같은 팩·같은 입력에 gpt 는 17/17 을 「비대상」으로 냈다 # (모델 교체가 팩 교체만큼 산출을 바꾼다). # ★alias 만 찍으면 **설정의 실제 모델만 바뀌었을 때** 완료 # 스텝이 그대로 SKIP 된다 — `gemini-pro` 는 모델 이름이 아니라 # `gemini_text_model` 로 매핑되는 alias 다. 물리 이름을 같이 # 찍는다(select_judge_model_physical 관례 동형). from app.modules.pipeline.era_research import ( ASSESS_MODEL as _era_assess_model, PICK_MODEL as _era_pick_model, resolve_model_physical as _era_phys, ) payload["era_assess_model"] = _era_assess_model payload["era_pick_model"] = _era_pick_model payload["era_assess_model_physical"] = _era_phys( _era_assess_model) payload["era_pick_model_physical"] = _era_phys(_era_pick_model) # (era R2 BLOCK-2) 검색·선택 팩 계약도 스텝 층에 접는다 — # 없으면 picker 팩만 바뀐 운영에서 완료 스텝이 outer SKIP 돼 # recipe 몸통의 참조 신원(r_sha) 재계산 자체가 안 돌고 옛 # 참조·배경이 영구 봉인된다. ★기존 era ON 완료 에피소드는 # 이 키 추가로 1회 whole-step 재진입(이행 창 — era_ref 1회 # 미스와 동축). from app.modules.pipeline.search_grounded_ref import ( search_contract_sha as _search_contract, ) payload["era_search_contract"] = _search_contract() # ★★**앞쪽 참조 획득 계약도 같이 접는다** (GROUNDING-V2 §2-4). # 참조를 사는 자리가 앞으로 옮겨졌으므로, 그 계약이 바뀌면 # 바깥 이미지 스텝도 stale 이 돼야 한다. 안 그러면 위 주석에 # 적힌 사고가 그대로 재현된다 — 「팩만 바뀐 운영에서 완료 스텝이 # outer SKIP 돼 참조 신원 재계산이 안 돌고 **옛 참조·배경이 # 영구 봉인**된다」. # ★산식을 여기 다시 적지 않는다. **앞쪽 모듈이 export 하는 한 # callable** 을 부른다 — 두 곳에 적으면 한쪽만 고쳐진다. from app.modules.pipeline.reference_acquisition import ( acquisition_contract_sha as _acq_contract, ) # ★★어느 라운드 계약을 접나 — **중앙 획득의 것**이다. # 바깥 이미지 스텝이 봐야 하는 것은 「참조를 만드는 자리」의 # 계약이고, 이관이 끝나면 그것은 중앙 하나다. late 는 그때 # 없어진다. 지금 late(1) 을 접으면 중앙이 서는 순간 바깥이 # **또** stale 되어 두 번 재생성한다. from app.core.steps.reference_acquisition_step import ( CENTRAL_ROUNDS as _acq_rounds, ) payload["reference_acquisition_contract"] = _acq_contract( rounds=_acq_rounds) # ── 표기 문안 저작 (2026-08-14 #119②) — ON 일 때만 스탬프, 동형. if bool(getattr(settings, "signage_author_enabled", False)): from app.modules.pipeline.signage_author import ( AUTHOR_MODEL as _sg_model, SIGNAGE_POLICY_VERSION as _sg_policy, resolve_signage_pack as _sg_pack_resolved, signage_pack_content_hash as _sg_pack_content, ) from app.modules.pipeline.still_recipe import ( SIGNAGE_SECTION_PROMPT_VERSION as _SIGNAGE_SECTION_SEL, recipe_stem_content_hash as _sg_section_stem, ) payload["signage_author_enabled"] = True payload["signage_policy"] = _sg_policy payload["signage_pack"] = _sg_pack_resolved() payload["signage_pack_content"] = _sg_pack_content() # ★**저작 모델도 신원이다** (2026-08-27 Codex BLOCK ④). # 같은 입력·같은 팩이라도 모델이 바뀌면 다른 문안이 나온다. # 별칭만으로는 그 뒤의 물리 모델 교체를 못 잡으므로 둘 다. payload["signage_author_model"] = _sg_model payload["signage_author_model_resolved"] = str( getattr(settings, "gemini_flash_model", "") or "") # ★`signage_section` 스템은 **저작이 켜졌을 때** 나간다 — # 종전에는 그 내용 해시가 무관한 의상 잠금 플래그 아래에만 # 있어, 그 플래그가 꺼지면 문안을 고쳐도 지문이 안 움직였다. payload["signage_section_stem"] = _sg_section_stem( _SIGNAGE_SECTION_SEL, "signage_section") # ── 인물 의상 잠금·prev 계승 스코프 (2026-08-14 #119③) — ON 일 # 때만 스탬프, 동형. if bool(getattr( settings, "still_cast_wardrobe_lock_enabled", False)): from app.modules.prompt_loader import ( pack_dir_content_hash as _pdc, ) # (era R2 부수 발견) PACK_VERSION_MAP 는 존재한 적 없는 이름 # (0c735d7c 오타) — cast ON 이면 이 메서드 전체가 ImportError # 로 죽고, step_runner 가 경고만 남기고 base hash 로 조용히 # 강등해 **모든 ON-플래그 스탬프가 무효**가 된다(현 주행은 # 해당 경로 미실행이라 잠복). 실심볼로 교정. from app.modules.pipeline.still_recipe import ( PROMPT_VERSION_MAP as _sr_map, PREV_CAST_SCOPE_PROMPT_VERSION as _cast_sel, ) payload["still_cast_wardrobe_lock_enabled"] = True payload["cast_wardrobe_lock_policy"] = "cast_wardrobe_lock_v1" # Codex BLOCK-3 꼬리: v21 독립 스템(prev/continues/people_rule # ·signage_section) 내용 drift 도 완료 스텝 재진입을 걸어야 — # selector+내용 해시를 outer 에 접는다. payload["cast_scope_pack"] = _sr_map[_cast_sel] payload["cast_scope_pack_content"] = _pdc( "still_recipe", _sr_map[_cast_sel]) # ★`PREVIOUS STILL USAGE` 절 스템 (2026-08-27, 감사 1-E). # # ★★**이미 발행된 팩에 스템만 더하면 selector 가 안 움직인다** # (Codex BLOCK). 팩 33 은 1-H 에서 이미 나갔고, 거기에 # `prev_usage_clause` 를 더했다 — 팩 버전도 팩 디렉토리 해시도 # 그대로라 **완주한 scene_image 가 whole-step clean skip** 하고, # 옛 CP 를 지키려고 넣은 관할 한정이 나가는 프롬프트까지 못 내려간다. # # ★조건 없이 접는다 — 이 절은 prev 가 있는 샷이면 **언제나** 나가고 # (`build_still_prompt`), 그 갈래를 가르는 플래그가 없다. # ★팩 버전이 아니라 **스템 바이트**를 접는다: 같은 팩 안에서 문안을 # 고치는 것이 흔한 일이다(이 판에서만 네 번 그랬다). from app.modules.pipeline.still_recipe import ( PREV_USAGE_CLAUSE_PROMPT_VERSION as _pu_sel, recipe_stem_content_hash as _pu_stem_content, ) payload["recipe_prev_usage_pack"] = _recipe_pack_resolved(_pu_sel) payload["recipe_prev_usage_stem_content"] = _pu_stem_content( _pu_sel, "prev_usage_clause") # ── 최종 스틸 생성 엔진 (2026-08-13 grok 전환) — 비기본값일 때만 # 스탬프, 동형. 백엔드는 조립(컴팩트 v17+시네마틱+ab 변주)과 생성 # 모델을 함께 바꾸므로 스텝 층 drift 가 걸려야 샷별 JIT 검증까지 # 내려간다. nb2(기본)=키 부재 byte-identical. _img_backend = getattr(settings, "still_image_backend", "nb2") # ★★**엔진 이름만 접으면 모자란다** (2026-09-20 Codex). 백엔드마다 # 실제 물리 모델이 다른 자리에 있다 — grok 은 `grok_image_model`, # gpt25 는 `openai_image_model`. 이름만 접고 모델을 빼면 모델을 # 갈아도 지문이 안 움직여 **완료 샷이 옛 모델 그림 그대로** 남는다. if _img_backend == "gpt25": payload["still_image_backend"] = _img_backend payload["openai_image_model"] = getattr( settings, "openai_image_model", "") # ★크기도 산출 실질 입력이다 — 16:9 가 `1536x864` 인지 # `1536x1024`(3:2)인지에 따라 나오는 그림의 비율이 갈린다. from app.modules.llm.gpt_image_client import ( SIZE_BY_ASPECT as _gpt_sizes, ) payload["gpt_image_size_16_9"] = _gpt_sizes.get("16:9", "") if _img_backend == "grok2": from app.modules.pipeline.still_recipe import ( STILL_COMPACT_PROMPT_VERSION as _compact_sel, STILL_GROK_STEM_VERSION as _grok_stem_sel, ) payload["still_image_backend"] = settings.still_image_backend payload["grok_image_model"] = settings.grok_image_model payload["still_compact_pack"] = _recipe_pack_resolved( _compact_sel) payload["still_compact_pack_content"] = _recipe_pack_content( _compact_sel) # 2026-08-18 컴팩트 스템 팩(v22) — grok 조립에서만 읽히는 절 # 다섯(identity·body&support·표기 정책·b롤 변주·CAMERA&FRAME)의 # 산출 실질 입력이라 같은 조건에서 함께 접는다. 문안이 바뀌면 # 완료 샷도 JIT 검증까지 내려가야 한다. nb2=키 부재 동일. payload["still_grok_stem_pack"] = _recipe_pack_resolved( _grok_stem_sel) payload["still_grok_stem_pack_content"] = _recipe_pack_content( _grok_stem_sel) # 2026-08-18 상한 덜어내기 정책 — 상한을 넘는 롤은 일반 규칙 # 절을 덜어 보낸다(fit_grok_prompt). 지문에 접는 것은 조립 # 프롬프트뿐이라 **정책만 바뀌면 지문이 안 움직여** 낡은 그림이 # 그대로 남는다. 상한값과 덜어내기 순서를 함께 접어 그 구멍을 # 막는다(팩 내용은 위 두 키가 이미 덮는다). # ★목표(soft)와 거절 상한(hard) **둘 다** 접는다 — 둘이 함께 # 발송 여부를 정하므로 하나만 접으면 같은 명목 프롬프트가 # "발송 vs 예외"로 갈려도 완료 스텝이 안 낡는다. from app.modules.llm.grok_image_client import ( GROK_TEXT_BYTE_HARD_LIMIT as _grok_hard_limit, GROK_TEXT_BYTE_LIMIT as _grok_limit, ) from app.modules.pipeline.still_recipe import ( GROK_SHED_ORDER as _grok_shed, ) payload["still_grok_text_limit"] = _grok_limit payload["still_grok_text_hard_limit"] = _grok_hard_limit payload["still_grok_shed_order"] = list(_grok_shed) # ── 읽을 글자 정책 (2026-08-27, 감사 1-A ⑤) ───────────────────── # 승인 목록 유무로 스템이 갈리고, 배경 두 스템도 같은 팩에 산다. # 이 팩·스템 내용이 지문에 안 접히면 문안을 고쳐도 완주한 # 에피소드가 clean skip 한다. **selector 와 내용을 독립으로** 접는다 # — 다른 키(컴팩트 팩 등)가 우연히 덮는 것에 기대지 않는다. from app.modules.pipeline.still_recipe import ( TEXT_POLICY_GROK_PROMPT_VERSION as _tp_grok_sel, TEXT_POLICY_PROMPT_VERSION as _tp_sel, recipe_pack_content_hash as _tp_pack_content, resolve_prompt_version as _tp_resolved, ) payload["text_policy_pack"] = _tp_resolved(_tp_sel) payload["text_policy_pack_content"] = _tp_pack_content(_tp_sel) if getattr(settings, "still_image_backend", "nb2") == "grok2": payload["text_policy_grok_pack"] = _tp_resolved(_tp_grok_sel) payload["text_policy_grok_pack_content"] = _tp_pack_content( _tp_grok_sel) # ── 이긴 후보 실격 게이트 (2026-09-20) ────────────────────────── # ★레버만 켜면 **바깥 스텝이 같은 해시로 SKIP** 해서 샷별 # fingerprint 까지 내려가지도 않는다 — 바로 아래 카메라 절 주석이 # 적어 둔 그 결함이다(Codex BLOCK). 켜진 판에서만 접는다 — # OFF 는 기존 해시 그대로. if getattr(settings, "still_winner_gate_enabled", False): from app.modules.pipeline.multiroll_select import ( GATE_POLICY_VERSION as _gate_policy, ) payload["winner_gate_policy"] = _gate_policy # ── 비인간 인물 명시 절 (2026-09-20 사용자 지시) ──────────────── # 이 절은 **몸=신원 인물이 실린 샷에만** 붙지만, 그 샷에서는 조립 # 문안을 바꾼다. 팩 내용이 지문에 안 접히면 문안을 고쳐도 완주한 # 에피소드가 clean skip 한다(위 표기 정책이 같은 자리에서 겪은 # 구멍 · a58417b2 합성 지문 사고와 같은 부류). selector 와 내용을 # 독립으로 접는다. from app.modules.pipeline.still_recipe import ( NONHUMAN_CAST_COMPACT_PROMPT_VERSION as _nh_compact_sel, NONHUMAN_CAST_PROMPT_VERSION as _nh_sel, ) payload["nonhuman_cast_pack"] = _tp_resolved(_nh_sel) payload["nonhuman_cast_pack_content"] = _tp_pack_content(_nh_sel) if getattr(settings, "still_image_backend", "nb2") == "grok2": payload["nonhuman_cast_compact_pack"] = _tp_resolved( _nh_compact_sel) payload["nonhuman_cast_compact_pack_content"] = _tp_pack_content( _nh_compact_sel) # ── 시네마틱 마감 절 (2026-08-27) ──────────────────────────────── # 이 절은 **언제나** roll 프롬프트에 붙는다(사용자 결정). 붙는데 # 그 문안이 지문에 안 접히면, 문안을 고쳐도 완주한 에피소드가 # clean skip 해서 새 문안이 영영 그림에 안 닿는다(바로 위 덜어내기 # 정책이 같은 자리에서 겪은 구멍). 소비 범위가 스템 하나라 그 # bytes 만 접는다. # # 접을지는 백엔드가 정한다: # nb2 — 컴팩트 팩 키가 아예 없다 → **언제나** 접는다. # grok — 위 `still_compact_pack_content` 가 그 팩 디렉토리를 # 통째로 덮는다. 두 selector 가 **같은 동안은** 키를 안 # 넣는다 — 넣으면 완주한 grok 판의 지문이 통째로 움직여 # 전량 재실행된다. # ★갈라지면 안 덮인다: 마감만 25 로 올리고 컴팩트가 17 # 이면 나가는 문안은 바뀌는데 지문은 17 팩 해시 그대로다. # **지금이 그 상태다**(마감 25 / 컴팩트 17) — 그래서 두 # 백엔드 다 접는다. from app.modules.pipeline.still_recipe import ( CINEMATIC_FINISH_PROMPT_VERSION as _fin_sel, STILL_COMPACT_PROMPT_VERSION as _fin_compact_sel, recipe_stem_content_hash as _fin_stem_content, should_attach_cinematic_finish as _fin_attaches, ) # ★조립 쪽(`still_recipe_service._grok_backend`)과 **같은 식**으로 # 쓴다. 종전에는 여기가 `!= "nb2"`, 저기가 `== "grok2"` 였다 — # 지금 같은 값이 나오는 것은 `config.py` 의 validator 가 그 둘만 # 받기 때문이지, 두 줄이 같은 뜻이어서가 아니다. 백엔드가 하나 # 늘면 조립은 붙이고 지문은 안 접는 조용한 clean skip 이 난다. _fin_grok = getattr(settings, "still_image_backend", "nb2") == "grok2" if _fin_attaches( grok_backend=_fin_grok, cine_on=bool(getattr( settings, "still_cine_transform_enabled", False)) ) and ((not _fin_grok) or _fin_sel != _fin_compact_sel): payload["still_cinematic_finish_pack"] = _recipe_pack_resolved( _fin_sel) payload["still_cinematic_finish_stem_content"] = ( _fin_stem_content(_fin_sel, "cinematic_finish")) # ── i2i 시네마틱 변환 스테이지 (2026-08-13 #108) — ON 일 때만 # 스탬프, 동형. 변환은 최종 자산 bytes 를 바꾸므로 스텝 층 drift # 가 걸려야(ack 승인 아래 비파괴 재진입) 샷별 JIT 검증까지 내려가 # 완료 샷도 변환이 돈다. 소비 범위가 스템 하나(cine_transform)라 # **그 스템 bytes 만** 접는다(b 변주 스템 관례 동형 — 디렉토리 # 전체 해시는 동거 스템 개정까지 전량 drift). OFF=키 부재 # byte-identical. if bool(getattr(settings, "still_cine_transform_enabled", False)): import app.modules.pipeline.cine_transform as _cine_mod from app.modules.pipeline.still_recipe import ( cine_pack_selector as _cine_pack_selector, recipe_stem_content_hash as _cine_stem_content, ) # 연출 재료를 넘기는 판이면 팩이 v24 다 — selector 를 상수로 박아 # 두면 팩이 바뀌어도 payload 가 안 움직여 완료 스텝이 SKIP 된다. _cine_sel = _cine_pack_selector() if bool(getattr( settings, "still_cine_stage_direction_enabled", False)): # 앞 단계에서 CAMERA 산문이 빠지고 최종 i2i 에 재료 절이 # 붙는다 — 두 층의 실질 입력이 같이 바뀐다. ON 일 때만 # 스탬프(관례 동형, OFF=키 부재 byte-identical). payload["still_cine_stage_direction_enabled"] = True from app.modules.pipeline.cine_provider import ( LEGACY_PROVIDER as _CINE_LEGACY, cine_provider_identity as _cine_identity_of, ) _cine_ident = _cine_identity_of() payload["still_cine_transform_enabled"] = True # ★모델 문자열은 **제공자가 정한다** (2026-08-25 provider 교환). # 종전처럼 grok 모델을 박아 두면 reve 로 바꿔도 payload 가 안 # 움직여 완료 스텝이 whole-step SKIP 되고 **변환이 영영 안 돈다** # (관문이 같은 자리에서 겪은 창). grok 일 때 값은 종전과 같다. payload["cine_transform_model"] = _cine_ident["model"] if _cine_ident["provider"] != _CINE_LEGACY: # 비기본 제공자일 때만 스탬프 — 기본값에서는 키 부재로 # byte-identical(관례 동형). payload["cine_transform_provider"] = _cine_ident["provider"] payload["cine_transform_endpoint"] = _cine_ident["endpoint"] payload["cine_transform_pack"] = _recipe_pack_resolved(_cine_sel) payload["cine_transform_stem_content"] = _cine_stem_content( _cine_sel, _cine_mod.CINE_TRANSFORM_STEM) # ★**재료 절 스템도 접는다** (2026-08-27, 과제 #94). # 종전에는 `cine_transform` 스템 바이트만 접었다. 그런데 재료를 # 넘기는 판은 **두 스템**을 이어 붙여 보낸다 # (`build_cine_transform_prompt`: base + `cine_stage_direction`). # 뒤엣것을 고치면 나가는 문안은 바뀌는데 지문이 안 움직여 # **완주한 샷이 옛 그림 그대로 통과**한다. # ★팩 selector 만으로는 못 잡는다 — 같은 팩 안에서 문안을 # 고치는 것이 흔한 일이다(이 판에서만 세 번 그랬다). if bool(getattr( settings, "still_cine_stage_direction_enabled", False)): payload["cine_stage_direction_stem_content"] = ( _cine_stem_content( _cine_sel, _cine_mod.CINE_STAGE_DIRECTION_STEM)) # Codex R1 BLOCK-2: 계약 버전은 inner 지문에만 접혀 있으면 # 완료 스텝에 닿지 않는다 — 참조 구성·산출 규약 bump 가 # outer clean SKIP 에 막혀 영원히 실행되지 않는 창. outer # 에도 접는다. payload["cine_transform_contract"] = ( _cine_mod.CINE_CONTRACT_VERSION) # ── 검증 관문 (2026-08-25) — ON 일 때만 스탬프, 동형. # 관문을 켜도 완료 스텝이 whole-step SKIP 되면 판정이 영영 안 # 돈다(era 가 같은 자리에서 겪은 창). 켜는 순간 1회 재진입해 # **그림은 다시 사지 않고 판정만** 사도록 안쪽에 갈래가 있다. # 계약 sha 를 같이 찍어 판정 문안·모델·잡는 축이 바뀌면 다시 # 판정한다 — 이 값들은 그림을 바꾸지 않으므로 변환 지문 # (cine_fingerprint)에는 **안 넣는다**(넣으면 판정기만 고쳐도 # 유료 이미지가 전량 재구매된다). if bool(getattr(settings, "still_cine_verify_enabled", False)): from app.modules.pipeline.cine_verify import ( verify_contract_sha as _cine_verify_sha, verify_models as _cine_v_models, ) payload["still_cine_verify_enabled"] = True # 계약 sha 가 팩·**짝 전체**·물리 모델·잡는 축을 다 접는다 — # 안쪽 재사용 갈래(`_verify_is_current`)와 같은 값을 써야 # 스텝만 재진입하고 판정은 건너뛰는 창이 안 생긴다. payload["cine_verify_contract"] = _cine_verify_sha() # ★**회차가 아니라 판정 모델을 남긴다** (2026-08-27 Codex # 비차단 지적). 회차는 이제 이 계약에서 아무 뜻이 없는데 # (`verify_models` 가 언제나 짝 전체를 돌려준다) 지문에 # 남겨 두면 **쓰지도 않는 설정을 2→1 로 바꿀 때 스텝이 # 헛되이 재진입**하고, 기록은 「회차가 계약이다」라는 옛 # 뜻을 계속 말한다. payload["cine_verify_models"] = list(_cine_v_models()) # ── 2롤 ab 변주·몸-지지 절 (2026-08-13 #106, Codex R1 BLOCK-4 + # R2): 전 백엔드 공통 계약이라 **무조건** 스탬프하되, 소비 범위가 # 스템 하나이므로 **그 스템 bytes 만** 접는다 — 디렉토리 전체 # 해시는 동거 스템(컴팩트 지도 절·identity) 개정·selector 이동까지 # nb2 전량 drift 를 일으킨다(R2 BLOCK). 샷별 재생성 범위는 base # prompt(몸-지지 절)와 roll_prompts(b 변주 — compute_input_ # fingerprint 가 직접 접음)가 가른다. # # ★b 변주 스템은 **조립과 같은 함수**로 고른다 (2026-08-28 감사 # P0-B). 종전에는 여기서 `STILL_COMPACT_PROMPT_VERSION`("17")을 # 박아 접었는데 조립은 grok 에서 v22 를 렌더했다 — 컴팩트 판 # 바이트를 고쳐도 지문이 안 움직여 완주한 샷이 clean skip 한다. # `camera_frame_stem` 이 겪은 것과 같은 결함이라 같은 수법으로 # 닫는다: 실제로 나가는 스템 **하나만** 접고, 백엔드 갈림도 같은 # 함수(`still_guidance_selector`)로 만든다. from app.modules.pipeline.still_recipe import ( SUPPORT_STILLNESS_PROMPT_VERSION as _sup_sel, broll_variation_stem as _broll_stem_of, recipe_stem_content_hash as _recipe_stem_content, still_guidance_selector as _broll_sel_of_backend, ) _broll_stem_name, _broll_pack = _broll_stem_of( _broll_sel_of_backend( getattr(settings, "still_image_backend", "nb2") == "grok2")) payload["still_roll_count"] = int(settings.still_recipe_roll_count) payload["broll_variation_stem"] = _broll_stem_name payload["broll_variation_stem_content"] = _recipe_stem_content( _broll_pack, _broll_stem_name) payload["support_stillness_stem_content"] = _recipe_stem_content( _sup_sel, "support_stillness_clause") # ── identity 참조 역할 절 (2026-08-12) — ON 일 때만 스탬프, 동형. # 절이 프롬프트를 바꾸므로 스텝 층에서도 drift 가 걸려야 샷별 JIT # 검증(#77-B)까지 내려간다. if bool(getattr(settings, "still_identity_ref_role_enabled", False)): from app.modules.pipeline.still_recipe import ( IDENTITY_ROLE_PROMPT_VERSION as _idr_sel, ) payload["still_identity_ref_role_enabled"] = True payload["identity_role_pack"] = _recipe_pack_resolved(_idr_sel) payload["identity_role_pack_content"] = _recipe_pack_content( _idr_sel) # confined fp 경로 (2026-08-11 설계) — ON 일 때만 스탬프. OFF=키 # 부재 byte-identical. ON 이면 drift → ack 승인 아래 비파괴 재진입 # → 샷별 JIT 검증이 갈라낸다 (#77 경로 — gq 와 동일 규율). if bool(getattr(settings, "still_confined_fp_enabled", False)): from app.modules.pipeline.confined_fp import ( CONFINED_FP_POLICY_VERSION as _cfp_policy, confined_fp_pack_content_hash as _cfp_content, resolve_confined_fp_pack as _cfp_resolved, ) payload["still_confined_fp_enabled"] = True payload["confined_fp_pack"] = _cfp_resolved() payload["confined_fp_pack_content"] = _cfp_content() payload["confined_fp_model"] = str( getattr(settings, "openai_image_model", "")) # 판별·readback 물리 모델도 산출 실질 입력 (Codex BLOCK-2) payload["confined_fp_readback_model"] = str( getattr(settings, "gemini_text_model", "")) payload["confined_fp_policy"] = _cfp_policy # Stage D HIGH-4: lane pipe·lane 라벨 팩도 스틸 산출 실질 입력 — # ON 시만 스탬프 (기존 조합 byte-identical). 기존 완료 프로젝트에서 # lane 활성화/팩 변경 시 clean-skip 잔존 차단. from app.core.applicability import outdoor_lane_pipe_on if outdoor_lane_pipe_on(): from app.modules.pipeline.conti_ab import ( AB_JUDGE_MODEL as _ab_judge_model, resolve_prompt_version as _ab_pack_lane, ) from app.modules.pipeline.still_recipe import ( COMPLEX_AB_CONTRACT_VERSION, COMPLEX_PROMPT_VERSION, LANE_PROMPT_VERSION, SEED_BG_PROMPT_VERSION, resolve_prompt_version as _recipe_pack, ) payload["outdoor_lane_pipe_enabled"] = True payload["still_recipe_lane_pack"] = _recipe_pack( LANE_PROMPT_VERSION ) # R3 (Codex 설계 리뷰 BLOCKING-3): 복잡 구조물 A/B 는 # still_conti_ab_enabled=False 여도 강제 — lane pipe ON 이면 # v4 팩·conti_ab 판정 팩·모델(alias+물리)·계약 버전을 반드시 # 스탬프 (기존 completed still/scene CP 무효화 의도). payload["still_recipe_complex_pack"] = _recipe_pack( COMPLEX_PROMPT_VERSION ) # seed-bg 단일 권위 팩 (2026-07-17 Codex 합의) — lane pipe ON # 이면 소비 가능 경로라 함께 스탬프 payload["still_recipe_seed_bg_pack"] = _recipe_pack( SEED_BG_PROMPT_VERSION ) payload["complex_ab_contract"] = COMPLEX_AB_CONTRACT_VERSION payload["conti_ab_judge_pack"] = _ab_pack_lane("1") payload["conti_ab_judge_model"] = _ab_judge_model payload["conti_ab_judge_model_physical"] = ( settings.gemini_text_model ) # 2026-07-16 복잡 구조물=A/B 재설계: outdoor_frame_mode 소비 제거 # (direct_seed 분기 소멸) — 스탬프도 제거 (Codex 열린쟁점④: # stale flag 가 hash 에 영향 주지 않게 consumer/hash 동시 제거). # E2E6 ⑧: 콘티 A/B(outer 판정 팩·모델)도 스틸 산출 실질 입력 — # ON 시만 스탬프 (OFF byte-identical) if getattr(settings, "still_conti_ab_enabled", False): from app.modules.pipeline.conti_ab import ( AB_JUDGE_MODEL, resolve_prompt_version as _ab_pack, ) payload["still_conti_ab_enabled"] = True payload["conti_ab_judge_pack"] = _ab_pack("1") payload["conti_ab_judge_model"] = AB_JUDGE_MODEL # Codex 8e70d4c0 NARROW-3: alias 는 물리 모델 교체를 감지 # 못함(gemini-pro→settings.gemini_text_model 해석) — 물리 # 모델 병행 스탬프 payload["conti_ab_judge_model_physical"] = ( settings.gemini_text_model ) # E2E6 ⑤: 플레이트 VLM 선택도 스틸 산출 실질 입력 — ON 시만 스탬프 # (OFF=기존 조합 byte-identical) if getattr(settings, "still_plate_select_enabled", False): from app.modules.pipeline.plate_select import ( resolve_prompt_version as _psel_pack, ) payload["still_plate_select_enabled"] = True payload["plate_select_pack"] = _psel_pack("1") payload["plate_select_model"] = "gemini-pro" # R1 (Codex 설계 리뷰 BLOCKING-1): 콘티형 샷은 스틸 시점 재판정 # 대신 shot_conti_light 선행 권위(plate_authority) 소비 — # 소비 계약 버전도 실질 입력 from app.modules.pipeline.plate_select import ( PLATE_AUTHORITY_VERSION as _pa_ver, ) payload["plate_authority_version"] = _pa_ver # 스틸 변형 2종×4택1 (2026-07-17): 변형 저작 팩·모델·계약 버전도 # 스틸 산출 실질 입력 — ON 시만 스탬프 (OFF byte-identical). # flip·라벨 매핑·critique 모드는 run_multiroll_select 자체 지문 # (HIGH-4)이 담당하고, 여기서는 계약 버전이 전환 감지를 커버. if getattr(settings, "still_variants_enabled", False): from app.modules.pipeline.still_variants import ( AUTHOR_MODEL as _sv_author_model, STILL_VARIANTS_CONTRACT_VERSION as _sv_contract, resolve_variants_pack_version as _sv_pack, ) payload["still_variants_enabled"] = True payload["still_variants_pack"] = _sv_pack("1") payload["still_variants_contract"] = _sv_contract payload["still_variants_author_model"] = _sv_author_model # alias 는 물리 모델 교체를 감지 못함 — 물리 모델 병행 스탬프 payload["still_variants_author_model_physical"] = str( getattr(settings, "openai_model", "") ) # 리뷰 HIGH-3: 변형 판정(gemini-pro alias)의 물리 모델도 스틸 # 산출 실질 입력 — lane/conti_ab 우연 ON 의존 없이 명시 스탬프 payload["still_variants_judge_model_physical"] = str( settings.gemini_text_model ) # ★2026-08-27 (감사 P0-A, Codex BLOCK-2 재리뷰): text-only location # lock 스템은 **전용 selector**(v35)로 로드된다. 위 # `still_recipe_pack` 두 줄은 팩 "1" 만 접으므로 이 스템의 문안을 # 고쳐도 outer hash 가 안 움직여 완료된 스텝이 샷 루프 **전에** # clean skip 된다 — 옛 LOCATION PHOTO 문장으로 만든 그림을 현행 # 계약 산출로 읽는다. 내부 프롬프트 지문은 루프에 들어가야만 # 계산된다. # # ★chain_bg(v12)는 **이미** 아래 bgfirst 블록이 접고 있다 — 거기 # 중복으로 넣지 않는다. 이 스템은 v1 에서만 소비되므로 그때만 # 스탬프해 OFF byte-identical 을 지킨다(선례 동형). if getattr(settings, "still_recipe_mode", "off") == "v1": payload["still_recipe_text_only_loc_pack"] = ( _recipe_pack_resolved(_TEXT_ONLY_LOC_SEL)) payload["still_recipe_text_only_loc_content"] = ( _recipe_pack_content(_TEXT_ONLY_LOC_SEL)) # BGFIRST2 이식 ②③ (2026-07-20): 체인 팩·계약·Step1 엔진이 콘티 # 샷 스틸 산출을 실질 변경 — ON 시만 스탬프 (OFF byte-identical). if getattr(settings, "still_bgfirst_enabled", False): from app.modules.pipeline.still_recipe import ( BGFIRST_BG_IMAGE_MODEL as _bgf_model, BGFIRST_CONTRACT_VERSION as _bgf_contract, BGFIRST_PROMPT_VERSION as _bgf_sel, resolve_prompt_version as _bgf_pack, ) payload["still_bgfirst_enabled"] = True payload["bgfirst_pack"] = _bgf_pack(_bgf_sel) payload["bgfirst_contract"] = _bgf_contract payload["bgfirst_bg_model"] = _bgf_model # ★2026-08-27 (감사 P0-A, Codex 재리뷰 BLOCK): 체인 Step2 # LOCATION 스템 스탬프를 **여기로 올렸다.** 종전에는 아래 # `still_lane_prev_bgfirst_enabled` 안에 있었는데, 이번 판이 # 서비스의 재조립 조건을 `lane_chain or prev_sel` 에서 # `bgfirst_used` 로 넓혀 **ordinary bgfirst 도 lane_prev 가 # 꺼진 채로 이 스템을 소비**한다. 옛 자리에 두면 그 갈래에서 # 스템 문안을 고쳐도 outer hash 가 안 움직여 완료 스텝이 # clean skip 된다. # # 진짜 OFF 경계는 `still_bgfirst_enabled=False` 다 — bgfirst # 자체가 꺼지면 이 스템을 아무도 안 읽는다(byte-identical). from app.modules.pipeline.still_recipe import ( CHAIN_BG_LOCATION_PROMPT_VERSION as _chain_loc_sel, ) # ★selector 만 옮긴다 — 종전 스탬프와 **같은 모양**이다. # 팩 내용 해시까지 더하는 것은 이번 지적(게이트가 낡았다)의 # 범위 밖이고, 별건으로 남긴다. payload["chain_bg_location_pack"] = _bgf_pack(_chain_loc_sel) payload["bgfirst_judge_model_physical"] = str( settings.gemini_text_model ) # fix③④ full (2026-07-21): 전면화 계약·전용 스템 팩(groupbg/ # seed 절)도 스틸 산출 실질 입력 — ON 시만 스탬프. if getattr(settings, "still_bgfirst_full_enabled", False): from app.modules.pipeline.still_recipe import ( BGFIRST_FULL_CONTRACT_VERSION as _bgf_full_contract, BGFIRST_FULL_PROMPT_VERSION as _bgf_full_sel, ) payload["still_bgfirst_full_enabled"] = True payload["bgfirst_full_contract"] = _bgf_full_contract payload["bgfirst_full_pack"] = _bgf_pack(_bgf_full_sel) # 2026-07-25 (케이스1 스펙 E·F): lane·prev 샷 체인 편입은 # 그 샷들의 refs·프롬프트·배경 권위를 실질 변경 — ON 시만 # 스탬프 (OFF byte-identical). if getattr( settings, "still_lane_prev_bgfirst_enabled", False ): from app.modules.pipeline.still_recipe import ( BGFIRST_LANE_CONTRACT_VERSION as _bgf_lane_contract, BGFIRST_LANE_PROMPT_VERSION as _bgf_lane_sel, CHAIN_BG_LOCATION_PROMPT_VERSION as _chain_loc_sel, ) payload["still_lane_prev_bgfirst_enabled"] = True # 2026-07-27 리뷰 지적: 불리언만 찍으면 lane 전용 팩 # (bg_fill 스템+장소·world 사실 절)이 hash 에 안 잡혀, # 이미 completed 인 프로젝트가 clean-skip 으로 팩 상향을 # 통째로 건너뛴다 — 형제 분기(bgfirst_full_*)와 동형으로 # 팩·계약을 병행 스탬프. payload["bgfirst_lane_pack"] = _bgf_pack(_bgf_lane_sel) payload["bgfirst_lane_contract"] = _bgf_lane_contract # 2026-07-27 리뷰 I-4: 체인 Step2 LOCATION 스템은 샷 팩과 # **의도적으로 분리된** 자체 selector 로 로드된다 — 지금 # 값이 우연히 bgfirst_lane_pack 과 같은 디렉토리로 # 풀려 덮이고 있을 뿐이다. 이 스템만 새 팩으로 발행하고 # selector 를 올리면 완주 프로젝트가 clean-skip 으로 # 영영 못 본다(이 프로젝트가 이미 겪은 dormant 팩 결함 # 그대로). 명시 스탬프로 분리 자체를 hash 에 싣는다. # ★2026-08-27: 이 스탬프는 아래 bgfirst 범위로 **올렸다** # (Codex 재리뷰 BLOCK). 여기 남기면 lane_prev OFF 인 # ordinary bgfirst 가 이 스템을 소비하는데도 안 접힌다. pass # 좁고 복잡한 실내의 기하 권위 스템 팩 (2026-08-07, v15) + 손에 든 # 물건 절. 켜지는 샷은 분류 판정(`confined_structure`/`handled_by`)이 # 정하므로 전역 플래그가 없다 — per-shot 지문에는 프롬프트·참조 # 라벨 변화로 자동 반영되지만, whole-step clean skip 이 먼저 # 일어나면 거기에 도달하지 못한다. 팩 자체를 항상 싣는다. from app.modules.pipeline.still_recipe import ( GEOM_AUTHORITY_PROMPT_VERSION as _geom_sel, HANDLED_OBJECT_PROMPT_VERSION as _hand_sel, recipe_stem_content_hash as _hand_stem, resolve_prompt_version as _geom_pack, ) payload["recipe_geom_authority_pack"] = _geom_pack(_geom_sel) # 손 절은 감사 1-C 에서 **기하 권위 팩과 갈랐다**(v29). 프레이밍 # 문장을 걷은 것이 완주 판에 닿으려면 스텝 신원이 바뀌어야 한다 — # per-shot 지문에는 실리지만 **whole-step clean skip 이 먼저** # 일어나면 거기에 도달하지 못한다. payload["recipe_handled_object_pack"] = _geom_pack(_hand_sel) # ★팩 디렉토리 전체가 아니라 **그 스템 한 장**의 바이트만 접는다. # 디렉토리를 접으면 무관한 스템 하나만 고쳐도 전 샷이 다시 돈다. # ★**스템 둘을 다 접는다** (2026-08-27 자체 리뷰). 프레이밍 문장을 # 따로 떼어 놓고 다른 한 장만 접으면, 그 문장을 고쳐도 스텝 # 신원이 안 움직여 완주 판이 새 문안을 못 본다 — 바로 위 주석이 # 막겠다고 한 그 구멍이다. # 인물 실체 절 갈래 (감사 1-F, 2026-08-27) — 무인 샷에서 빠지고 # 컴팩트 갈래는 따로 간다. 팩 셋 + 스템 넷의 바이트를 접는다. from app.modules.pipeline.still_recipe import ( REALIZE_COMPACT_PROMPT_VERSION as _rz_c, REALIZE_PROMPT_VERSION as _rz, ) payload["recipe_realize_pack"] = "|".join( _geom_pack(x) for x in (_rz, _rz_c)) payload["recipe_realize_content"] = "|".join( _hand_stem(sel, stem) for sel in (_rz, _rz_c) for stem in ("realize_still", "human_form")) payload["recipe_handled_object_content"] = "|".join( _hand_stem(_hand_sel, stem) for stem in ("handled_object_clause", "handled_object_framing")) # fix1 (2026-07-19): CAMERA/FRAME 절 — 스틸 프롬프트 실질 입력. # ★2026-08-07 신설. 이 축만 outer hash 에서 빠져 있었다(lighting· # conduct 는 있었다). per-shot 지문에는 실려 있지만, whole-step # clean skip 이 먼저 일어나면 per-shot 검사에 도달하지 못한다 — # 이 플래그만 바뀐 실행이 stale 산출을 그대로 재사용하게 된다. if getattr(settings, "still_recipe_camera_frame_enabled", False): from app.modules.pipeline.still_recipe import ( CAMERA_FRAME_PROMPT_VERSION as _cam_sel, resolve_prompt_version as _cam_pack, ) payload["still_recipe_camera_frame_enabled"] = True payload["recipe_camera_frame_pack"] = _cam_pack(_cam_sel) # ★2026-08-27 (감사 1-H). cine 로 연출 재료를 넘기는 판은 # `- CAMERA:` 를 뺀 **다른 스템**을 쓴다. 그 팩을 안 접으면 # 각도 권위를 걷은 문안이 나가는데 지문은 옛 팩 그대로라 # whole-step clean skip 이 옛 그림을 그대로 재사용한다. # ★조건은 조립 쪽(`still_recipe_service.py:1263`)과 같은 축이다 # — 한쪽만 바뀌면 조립은 새 문안, 지문은 옛 값이 된다. if (getattr(settings, "still_cine_transform_enabled", False) and getattr( settings, "still_cine_stage_direction_enabled", False)): from app.modules.pipeline.still_recipe import ( camera_frame_stem as _cam_stem_of, recipe_stem_content_hash as _cam_stem_content, still_guidance_selector as _cam_sel_of_backend, ) # ★**실제로 나가는 스템 하나만** 접는다 (2026-08-27 Codex # 재지적, 내 앞 수리를 물린다). # # ① 처음엔 base 스템만 접어 **grok 의 컴팩트 문안 수정이 # 누락**됐다. # ② 그래서 둘 다 접었더니 **반대편이 열렸다** — nb2 설정 # 그대로 컴팩트 바이트만 고쳐도 nb2 완주 샷이 mismatch # 로 다시 산다. 안 쓰는 스템은 접으면 안 된다. # ★「`still_image_backend` 가 이미 지문에 있으니 지출이 안 # 는다」가 내 근거였는데 **틀렸다.** 그건 두 백엔드의 지문이 # 서로 다르다는 뜻일 뿐, 안 쓰는 스템이 기여하지 않는다는 # 뜻이 아니다. # # ③ 그래서 조립과 지문이 **같은 함수**(`camera_frame_stem`) # 를 부른다. 백엔드 갈림도 같은 함수 # (`still_guidance_selector`)로 만든다 — 백엔드가 하나 # 늘어도 두 자리가 같이 움직인다. _cam_stem_name, _cam_stem_pack = _cam_stem_of( _cam_sel_of_backend( getattr(settings, "still_image_backend", "nb2") == "grok2"), omit_camera_direction=True) payload["recipe_camera_frame_no_dir_pack"] = _cam_pack( _cam_stem_pack) payload["recipe_camera_frame_no_dir_stem"] = _cam_stem_name # 팩 버전만 접으면 **같은 팩 안에서 문안을 고칠 때** 안 # 움직인다 — 스템 바이트를 직접 접는다. payload["recipe_camera_frame_no_dir_stem_content"] = ( _cam_stem_content(_cam_stem_pack, _cam_stem_name)) # ★B-2b (2026-08-28, Codex #39 BLOCK) — lane 샷 스케일 쌍. # # lane 샷은 위 `camera_frame` 절을 아예 안 받는다. 대신 팩 v41 # 두 스템이 나간다 — `framing_scale_clause`(스케일 한 줄)와 # 그 짝인 `sketch_label_lane`(스케치 라벨). 이 바이트를 여기 # 안 접으면 **완료 스텝이 통째로 skip 된다**: # `StepRunner._evaluate_resume_decision` 이 config hash 가 같고 # 파일이 있으면 SKIP 이라, 샷별 `compute_input_fingerprint` 까지 # 내려가지도 않는다. 그러면 이 수정이 옛 에피소드에서 **한 번도 # 안 돈다** — 바로 위 두 주석이 적어 둔 그 결함이다. # # ★조건은 조립 쪽과 같은 축이다: 조립은 # `camera_frame_on and lane_used` 일 때만 만들고 # (`still_recipe_service.py`), 스텝 수준에서 그 짝은 # camera_frame ON + lane pipe ON 이다. 한쪽만 켜지면 이 절이 # 안 나가므로 스탬프도 안 붙는다(종전 hash 유지). # # ★**두 스템을 같이** 접는다 — 둘 중 하나만 고쳐도 계약이 # 달라진다(절만 바뀌고 라벨이 그대로면 권위가 둘이 된다). if outdoor_lane_pipe_on(): from app.modules.pipeline.still_recipe import ( FRAMING_SCALE_PROMPT_VERSION as _fs_sel, LANE_SKETCH_LABEL_PROMPT_VERSION as _lsk_sel, recipe_stem_content_hash as _stem_content, ) payload["recipe_framing_scale_pack"] = _cam_pack(_fs_sel) payload["recipe_framing_scale_stem_content"] = ( _stem_content(_fs_sel, "framing_scale_clause")) payload["recipe_lane_sketch_label_stem_content"] = ( _stem_content(_lsk_sel, "sketch_label_lane")) # E2E10 fix⑤ (2026-07-21): LIGHTING & MOOD 절 — 스틸·Step1 배경 # 프롬프트 실질 입력. ON 시만 스탬프 (OFF byte-identical). if getattr(settings, "still_recipe_lighting_enabled", False): from app.modules.pipeline.still_recipe import ( LIGHTING_MOOD_PROMPT_VERSION as _light_sel, resolve_prompt_version as _light_pack, ) payload["still_recipe_lighting_enabled"] = True payload["recipe_lighting_pack"] = _light_pack(_light_sel) # E2E11 fix④⑤ (2026-07-22): 연기·형상 절 — 스틸 프롬프트 실질 # 입력. ON 시만 스탬프 (OFF byte-identical). if getattr(settings, "still_recipe_conduct_enabled", False): from app.modules.pipeline.still_recipe import ( STILL_CONDUCT_PROMPT_VERSION as _conduct_sel, resolve_prompt_version as _conduct_pack, ) payload["still_recipe_conduct_enabled"] = True payload["recipe_conduct_pack"] = _conduct_pack(_conduct_sel) # ★`naturalism_clause` 는 이 팩에서 **나왔다** (감사 P1-A, # 2026-08-28). 위 `recipe_conduct_pack` 은 팩 **문자열**만 # 접으므로 그 절의 바이트를 고쳐도 안 움직인다. 조립과 같은 # 함수로 실제 쓰는 팩을 고르고 그 스템 바이트를 접는다 — # `broll_variation_stem` 이 닫은 것과 같은 구멍이다. from app.modules.pipeline.still_recipe import ( naturalism_pack as _nat_pack_of, recipe_stem_content_hash as _nat_stem_content, still_guidance_selector as _nat_sel_of_backend, ) _nat_pack = _nat_pack_of( _nat_sel_of_backend( getattr(settings, "still_image_backend", "nb2") == "grok2")) payload["recipe_naturalism_pack"] = _conduct_pack(_nat_pack) payload["recipe_naturalism_content"] = _nat_stem_content( _nat_pack, "naturalism_clause") # E2E10 fix② (2026-07-21): i2i 수정본 재판정 — 최종 _sel 결정 # 정책 전환이 스틸 산출 실질 입력. ON 시만 스탬프 (OFF # byte-identical). per-shot 지문 기여는 run_multiroll_select 담당. # ★`and critique_enabled` — 재판정은 「원본 vs **수정본**」이라 # master 가 꺼지면 수정본이 없어 이 단계가 아예 안 돈다 # (2026-08-29, 배선도 `still_recipe_service.py` 에서 같이 종속시켰다). if (getattr(settings, "multiroll_fix_rejudge_enabled", False) and bool(getattr(settings, "still_recipe_critique_enabled", False))): from app.modules.pipeline.multiroll_select import ( FIX_REJUDGE_POLICY_VERSION as _frj_policy, ) payload["multiroll_fix_rejudge"] = _frj_policy # ★재판정도 `make_gemini_judge_fn` 이라 **두 심판이 다 돈다** — # Gemini 하나만 적으면 둘째가 바뀌어도 지문이 안 움직인다. # ★import 를 여기서 다시 한다 — 위 `_sel_phys` 는 GG46 블록 # 안에서만 묶여, GG46 OFF + 수정 ON 이면 NameError 가 난다. from app.modules.pipeline.multiroll_gemini import ( resolve_select_judge_model_physical as _frj_sel_phys, ) payload["fix_rejudge_judge_model_physical"] = _frj_sel_phys() # Codex HIGH-4: 중립 헤더 팩(판정 계약)도 실질 입력 from app.modules.pipeline.multiroll_gemini import ( STILL_FIX_REJUDGE_HEADER_PACK_VERSION as _frj_hdr_sel, resolve_judge_pack_version as _frj_hdr_pack, ) payload["fix_rejudge_header_pack"] = _frj_hdr_pack(_frj_hdr_sel) # E2E11 fix③ (2026-07-22): GPT 구도 critique 합류 — 수정본(최종 # 후보) 실질 입력. ON 시만 스탬프. if getattr(settings, "multiroll_gpt_composition_enabled", False): from app.modules.pipeline.multiroll_gemini import ( GPT_COMPOSITION_PACK_VERSION as _gc_sel, resolve_judge_pack_version as _gc_pack, ) from app.modules.pipeline.multiroll_select import ( GPT_COMPOSITION_POLICY_VERSION as _gc_policy, ) payload["multiroll_gpt_composition"] = _gc_policy payload["gpt_composition_pack"] = _gc_pack(_gc_sel) payload["gpt_composition_model_physical"] = str( getattr(settings, "openai_model", "") ) return hashlib.sha256( _json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _execute(self, mode="resume") -> Dict[str, Any]: # #77-B 상한 래치 (Codex 재리뷰 BLOCK 2): **모든 지출·변형보다 # 앞**에서 막는다. 안쪽(recipe 걷기 초입) 검사만으로는 ①아래 # generate_images 선행 구간(world guide 캐시 미스·T2I fallback)이 # 유료라 자동 재시도가 바퀴마다 그만큼 지출하고 ②force 가 recipe # 디렉토리를 아카이브하며 래치까지 치워 우회가 된다. 그래서 이 # 검사가 tracer·force 변형·서비스 호출 전부보다 먼저다 — "사람 # 확인 전 전면 정지"를 코드로 강제(해제=래치 파일 삭제). 안쪽 # 검사는 다른 진입로 방어로 유지. from app.core.config import settings as _s77 if bool(getattr(_s77, "still_jit_verify_enabled", True)): from pathlib import Path as _P77 from app.services.still_recipe_service import ( JIT_LATCH_FILENAME as _JIT_LATCH, StillJitRegenLimitExceeded as _JitLatchErr, ) _latch = ( _P77(_s77.projects_dir) / self.project_id / "images" / self.episode_id / "scene" / "recipe" / _JIT_LATCH ) if _latch.exists(): raise _JitLatchErr( f"JIT 재생성 상한 래치가 남아 있다: {_latch} — 사람 " f"확인 전에는 mode 와 무관하게 이 스텝을 실행하지 " f"않는다(선행 유료 구간·force 아카이브 우회 차단). " f"원인 확인 후 래치 파일을 삭제하고 resume.") self._setup_image_tracer_context() from app.services.scene_image_service import SceneImageService from app.core.pipeline_gate import check_scene_images_ready # 표적 씬 슬라이스 (2026-07-23, Codex BLOCKING-5): force 는 에피소드 # 전체 primary 해제+checkpoint 삭제+recipe 아카이브라 표적 모드와 # 병용 시 비표적 자산까지 무효화 — 명시 fail-closed (부분 force 는 # still-id 단위 재설계가 필요한 별도 범위). from app.core.config import settings as _settings from app.modules.pipeline.scene_image_scope import ( parse_target_scenes as _parse_targets, ) _target_scenes = _parse_targets( getattr(_settings, "scene_image_target_scenes", "")) if _target_scenes and mode == "force": from app.core.errors import AppError raise AppError( code="scene_image.target_force_forbidden", message=( "scene_image_target_scenes 설정과 mode=force 병용 금지 " "— force 는 에피소드 전체 무효화라 비표적 자산을 " "파괴한다 (fail-closed)" ), status_code=422, ) # 게이트 (참조+합성 100% 확인) check_scene_images_ready(self.db, self.project_id, self.episode_id) svc = SceneImageService(db=self.db, project_id=self.project_id, actor_id=self._get_system_actor_id()) # force 모드: 기존 이미지 삭제 안 함 — is_primary만 해제하고 새로 생성 if mode == "force": from app.models.project import ImageAsset from sqlalchemy import text # 기존 씬 이미지 is_primary 해제 (새 이미지가 primary 됨) self.db.execute(text( "UPDATE image_asset SET is_primary = 0 " "WHERE project_id = :pid AND episode_id = :eid AND asset_type = 'scene' AND is_primary = 1" ), {"pid": self.project_id, "eid": self.episode_id}) self.db.commit() # 체크포인트 삭제 (새로 시작) from pathlib import Path from app.core.config import settings cp_path = Path(settings.projects_dir) / self.project_id / "checkpoints" / "images" / self.episode_id / "scene_checkpoint.json" if cp_path.exists(): cp_path.unlink() # still_recipe(2026-07-13, Codex BLOCKING-3): force 시 recipe # 중간 산출(롤/_sel/records.json)을 아카이브해 새 네임스페이스로 # 시작 — 옛 _sel 이 새 primary 로 재저장되는 결함 차단. 파일은 # 이동 보존(삭제 금지 규칙). if getattr(settings, "still_recipe_mode", "off") != "off": from datetime import datetime as _dt recipe_dir = ( Path(settings.projects_dir) / self.project_id / "images" / self.episode_id / "scene" / "recipe" ) if recipe_dir.exists(): archived = recipe_dir.with_name( f"recipe_archived_{_dt.now().strftime('%Y%m%d_%H%M%S')}" ) recipe_dir.rename(archived) logger.info( "scene_image_pipeline force: recipe 산출 아카이브 → %s", archived, ) gen_mode = "resume" # 체크포인트 삭제 후 resume = 전체 재생성 else: gen_mode = "resume" svc.generate_images(episode_id=self.episode_id, mode=gen_mode) # 결과 카운트 — shot-more: is_selected=True만 이미지 생성 대상. # 표적 모드=실행이 남긴 scope audit 의 effective(requested+의존 # 클로저) 집합으로 제한 — 독자 필터 재구현 금지 (Codex HIGH-3). from app.models.project import SceneStill, ImageAsset selected_still_ids = [ r[0] for r in self.db.query(SceneStill.id).filter( SceneStill.project_id == self.project_id, SceneStill.episode_id == self.episode_id, SceneStill.is_selected == True, # noqa: E712 SceneStill.still_index >= 0, SceneStill.status != "stale", ).all() ] if _target_scenes: from app.modules.pipeline.scene_image_scope import ( check_scope_universe, load_scope_audit, scope_audit_path, ) _audit = load_scope_audit( scope_audit_path( _settings.projects_dir, self.project_id, self.episode_id), expected_scenes=_target_scenes, ) # Codex 재재리뷰 HIGH-1: 현재 selected-still 우주와 결합 검증 # (같은 씬 번호의 stale audit → 교집합 0 → false CLEAN 차단) _universe = { r[0]: r[1] for r in self.db.query( SceneStill.id, SceneStill.scene_index, ).filter( SceneStill.project_id == self.project_id, SceneStill.episode_id == self.episode_id, SceneStill.is_selected == True, # noqa: E712 SceneStill.still_index >= 0, SceneStill.status != "stale", ).all() } check_scope_universe(_audit, _universe) _effective = set(_audit["effective_ids"]) selected_still_ids = [ x for x in selected_still_ids if x in _effective] primary_count = 0 if selected_still_ids: primary_count = self.db.query(ImageAsset).filter( ImageAsset.project_id == self.project_id, ImageAsset.episode_id == self.episode_id, ImageAsset.asset_type == "scene", ImageAsset.is_primary == 1, ImageAsset.still_id.in_(selected_still_ids), ).count() # ★★**실행 계수도 게이트를 본다** (2026-09-20 Codex ㉢). # 종전에는 DB 대표 행 수 그대로라, 게이트가 절반을 붙잡아도 # 「completed=239 / failed=0」으로 닫혔다 — UI·감시·복구 판단을 # 오도한다. StepRunner 는 exit verify 가 false 면 상태만 partial # 로 바꿀 뿐 **계수를 다시 세지 않는다**(:1774-1794). # ★단순히 `primary_count - len(막힌 것)` 으로 빼면 안 된다 — # **원래 대표가 없던** 보류 샷을 두 번 빼게 된다. 그래서 verify # 와 **같은 집합 계산**을 쓴다: 완료 = (파일 있는 대표) - (보류). # ★게이트가 꺼진 판의 집계는 건드리지 않는다. result: Dict[str, Any] = self._build_result_counts( selected_still_ids, primary_count) if _target_scenes: # 표적 설정 시에만 영속 (Codex 재재리뷰 HIGH-2: 표적 미설정 # default 경로는 result shape byte-identical 유지): # target_scope_base_hash=표적 무관 base — 표적 목록만 바뀐 # drift 를 비파괴 RERUN_SELF 로 분류하는 비교 기준(BLOCKING-2). # 무표적→표적 전이는 save_checkpoint override(step-local # config_hash 저장)+_evaluate_contract_drift 의 stored # config_hash fallback 이 커버. result["target_scope_base_hash"] = self._config_hash_base() return result def save_checkpoint(self, data: Dict[str, Any]) -> None: """step-local config_hash 저장 대칭 (Codex 3차 리뷰 HIGH). StepRunner.save_checkpoint 의 fallback 은 project_config md5 인데 _check_cp_mismatch 는 이 스텝의 step-local hash(_config_hash)와 비교 — 저장/비교 비대칭으로 recipe ON 무표적 CP 의 표적 전이 fallback(stored config_hash==현재 base)이 원리적으로 작동하지 않았다. config_hash 키는 CP 에 항상 존재하므로 shape 불변 — 값만 저장/비교 대칭이 된다(무표적=base 그대로, 표적=fold 값). """ if "config_hash" not in data: data["config_hash"] = self._config_hash() super().save_checkpoint(data) def _build_result_counts(self, selected_still_ids, primary_count): """실행 계수 — 게이트에 **붙잡힌 샷은 완료가 아니다**. ★종전에는 DB 대표 행 수 그대로라, 게이트가 절반을 붙잡아도 「completed=239 / failed=0」으로 닫혔다 — UI·감시·복구 판단을 오도한다. StepRunner 는 exit verify 가 false 면 **상태만** partial 로 바꿀 뿐 계수를 다시 세지 않는다. ★`대표 수 - len(보류)` 로 빼면 **원래 대표가 없던** 보류 샷을 두 번 뻐다 — 그래서 verify 와 **같은 집합 계산**을 쓴다. ★게이트가 꺼진 판의 집계는 **그대로** 둔다(result shape 포함). """ scene_total = len(selected_still_ids) completed_count = primary_count gate_data: Dict[str, Any] = {} if selected_still_ids and self._gate_counts_apply(): done, held, _, unverifiable, unjudged = ( self._policy_completed_still_ids(selected_still_ids)) completed_count = len(done) _preview = sorted(held) gate_data = { "gate_policy_completed": completed_count, "gate_held_count": len(held), # ★**잘린 목록을 전체 목록처럼 부르지 않는다** (Codex). # 세는 근거는 `gate_held_count` 고, 아래는 미리보기다. "gate_held_still_ids_preview": _preview[:200], "gate_held_still_ids_truncated": len(_preview) > 200, "gate_unjudged_count": len(unjudged), "gate_held_reason": ("gate_state_unreadable" if unverifiable else "winner_violation_gate"), } if unverifiable: gate_data["gate_unverifiable"] = True return { "completed_count": completed_count, "applicable_count": scene_total, "failed_count": max(0, scene_total - completed_count), "data": {"scene_total": scene_total, "primary_count": primary_count, **gate_data}, } def _gate_counts_apply(self) -> bool: """실행 계수를 게이트 기준으로 셀 것인가 — **적용 경계 한 자리**. ★`gate_is_on()` 하나만 본다. 소비자마다 설정 규칙을 베껴 두면 한쪽만 고쳐진다 (2026-09-20 Codex ㅢ). """ try: from app.services.still_recipe_service import gate_is_on return gate_is_on() except Exception: # noqa: BLE001 return False def _policy_completed_still_ids(self, selected_ids): """**정책상 완료**한 still_id — 실행 계수와 verify 가 함께 쓴다. Codex ㅣ 의 집합 계산 그대로: V = selected 안에서 **실제 파일이 있는** 대표의 고유 still_id H = 같은 정책에서 **보류**된 still_id(직접 보류 + 의존 보류) 완료 = V - H ★`대표 수 - len(H)` 로 빼면 **원래 대표가 없던** 보류 샷을 두 번 뻐다. 그래서 집합으로 센다. ★DB 자산을 지우거나 대표를 내리지 않는다 — **세는 자리**에서만 뻐다. 돌려주는 것: (완료 집합, 보류 집합, 대표 행 수) """ from app.core.file_paths import resolve_image_path from app.models.project import ImageAsset rows = self.db.query(ImageAsset).filter( ImageAsset.project_id == self.project_id, ImageAsset.episode_id == self.episode_id, ImageAsset.asset_type == "scene", ImageAsset.is_primary == 1, ImageAsset.still_id.in_(list(selected_ids)), ).all() seen: set = set() for r in rows: pth = resolve_image_path(r.file_path) if pth and pth.exists() and r.still_id: seen.add(r.still_id) held, unverifiable, unjudged = self._gate_blocked_still_ids( selected_ids) if unverifiable: # ★**못 읽은 것은 합격이 아니다** (2026-09-20 Codex). 켜 놓고 # 확인을 못 했으면 어느 대표도 「정책상 완료」라고 못 한다. # 파일·대표는 그대로 두고 **세는 자리**에서만 보류한다. return set(), set(selected_ids), len(rows), True, set() return (seen - held), held, len(rows), False, unjudged def _gate_blocked_still_ids(self, selected_ids): """게이트에서 **막힌** 샷 · **검사 불가** 여부 · **판정 없는** 샷. 돌려주는 것: (보류 집합, 검사 불가 여부, 판정 없는 집합) 판정 없는 집합은 보류 집합에 **포함**된다. ★게이트가 꺼져 있으면 (빈 집합, False, 빈 집합) — 종전 그대로다. ★★게이트가 **켜진** 판에서 기록을 못 읽으면 **검사 불가**다 (2026-09-20 Codex). 종전에는 빈 집합만 돌려줘서, 파일이 남아 있는 **옛 대표가 다시 완료로 집계**됐다. ★★**판정이 아예 없는 샷도 합격이 아니다** (2026-09-20 Codex). 「명시 비대상(`not_applicable`)」과 「판정 없음」은 다르다 — gate 키가 없다는 것만으로 「bgfirst 라 비대상」이라고 알 수 없다. 켜진 판에서 판정이 없으면 **아직 안 본 것**이다. ★검사 불가·미판정을 **재구매로 풀지 않는다** — 세는 자리에서 보류로 남기고, 파일·대표는 그대로 둔다. """ from app.core.config import settings as _s from app.services.still_recipe_service import gate_is_on # ★적용 경계도 **한 자리**에서 — 소비자가 각자 판단하면 빠뜨린다. if not gate_is_on(): return set(), False, set() try: import json as _json from pathlib import Path as _Path from app.models.project import SceneStill # ★소비자 넷이 **같은 판정**을 본다 (2026-09-20 Codex BLOCK) — # cine 문·prev 앵커·verify 가 각자 다른 enum 을 읽으면 같은 # 샷이 「변환은 사고 앵커로는 못 쓰이는」 어긋남이 난다. from app.services.still_recipe_service import ( GATE_HOLDING_OUTCOMES, ) rj = (_Path(_s.projects_dir) / self.project_id / "images" / self.episode_id / "scene" / "recipe" / "records.json") if not rj.exists(): # ★기록 파일이 **아예 없다**. 셀 대상이 있는데 판정 기록이 # 없으면 그 대표들이 어느 정책에서 나왔는지 알 길이 없다 — # 검사 불가다. 셀 대상이 없으면 잴 것도 없다. if selected_ids: logger.error( "scene_image_pipeline verify: 게이트가 켜져 있는데 " "판정 기록 파일이 없다(%s) — **검사 불가**", rj) return set(), True, set() return set(), False, set() data = _json.loads(rj.read_text(encoding="utf-8")) rows = self.db.query(SceneStill).filter( SceneStill.id.in_(list(selected_ids))).all() held: set = set() unjudged: set = set() for r in rows: rec = data.get(f"S{r.scene_index}sh{r.shot_index}") g = rec.get("gate") if isinstance(rec, dict) else None outcome = (str(g.get("outcome") or "") if isinstance(g, dict) else "") if not outcome: # 판정이 없다 = 이 정책으로 **아직 안 봤다** unjudged.add(r.id) held.add(r.id) elif outcome in GATE_HOLDING_OUTCOMES: held.add(r.id) if unjudged: logger.error( "scene_image_pipeline verify: 게이트가 켜져 있는데 " "**판정이 없는 샷 %d개** — 완료로 세지 않는다", len(unjudged)) return held, False, unjudged except Exception as exc: # noqa: BLE001 logger.error( "scene_image_pipeline verify: 게이트가 켜져 있는데 상태를 " "**못 읽었다** — 검사 불가다(완료 단정 금지): %s", exc) return set(), True, set() def verify_completion(self): """Group 1 #3 (visual_pipeline_contracts_plan): consumer 측 verify_completion. scene 이미지 산출물 정합성 — selected SceneStill 수 == ImageAsset(scene, is_primary=1, file 존재). producer 측 verify_completion (RefImageGenStep / CompositeImageGenStep / chain_bg) 와 symmetric. silent fallback 차단 (silent text-only 결과나 disk 누락이 status=completed 로 통과 안 되도록). """ from app.core.integrity_report import CompletionReport from app.models.project import SceneStill selected_ids = [ r[0] for r in self.db.query(SceneStill.id).filter( SceneStill.project_id == self.project_id, SceneStill.episode_id == self.episode_id, SceneStill.is_selected == True, # noqa: E712 SceneStill.still_index >= 0, SceneStill.status != "stale", ).all() ] # 표적 씬 슬라이스: verify 도 실행 audit 의 effective(requested+ # dependency_added) 기준 — 의존 스틸 파일 결손도 partial 판정 # (Codex HIGH-3·HIGH-4). audit 부재/오류=fail-closed missing. from app.core.config import settings as _settings from app.modules.pipeline.scene_image_scope import ( check_scope_universe, load_scope_audit, parse_target_scenes as _parse_targets, scope_audit_path, ) _target_scenes = _parse_targets( getattr(_settings, "scene_image_target_scenes", "")) if _target_scenes: try: _audit = load_scope_audit( scope_audit_path( _settings.projects_dir, self.project_id, self.episode_id), expected_scenes=_target_scenes, ) # Codex 재재리뷰 HIGH-1: 현재 selected-still 우주 결합 — # 스틸 세대 교체 후 stale audit 가 교집합 0 → expected=0 # → false CLEAN 이 되는 경로 차단 (요구 동등성+⊆ 우주) _universe = { r[0]: r[1] for r in self.db.query( SceneStill.id, SceneStill.scene_index, ).filter( SceneStill.project_id == self.project_id, SceneStill.episode_id == self.episode_id, SceneStill.is_selected == True, # noqa: E712 SceneStill.still_index >= 0, SceneStill.status != "stale", ).all() } check_scope_universe(_audit, _universe) except ValueError as exc: return CompletionReport( is_complete=False, missing=[str(exc)], severity="missing", metadata={"expected": 0, "found": 0, "rows": 0}, ) _effective = set(_audit["effective_ids"]) selected_ids = [x for x in selected_ids if x in _effective] expected = len(selected_ids) if expected == 0: return CompletionReport( is_complete=True, missing=[], severity="clean", metadata={"expected": 0, "found": 0, "rows": 0}, ) # ★★게이트에서 막힌 샷은 **옛 대표 파일이 있어도 완료가 아니다** # (2026-09-20 Codex). 이 검사는 `is_primary` 파일의 존재만 보므로, # 새 생성·승격을 막아도 **재개에서 거짓 완료**가 난다. # ★파일을 지우거나 대표를 내려서 실패를 표현하지 않는다 — # **세는 자리에서** 뺀다. 독립 성공 샷은 그대로 센다. # ★실행 계수와 **같은 함수**를 쓴다 (2026-09-20 Codex ㉢) — # 같은 규칙을 두 곳에 적으면 한쪽만 고쳐진다. seen_still_ids, _blocked, _rows_n, _unverifiable, _unjudged = ( self._policy_completed_still_ids(selected_ids)) if _unverifiable: logger.error( "scene_image_pipeline verify: 게이트 상태를 **못 읽어** " "완료를 단정하지 않는다 — %d샷 전부 보류", len(_blocked)) elif _blocked: logger.warning( "scene_image_pipeline verify: 게이트에 막힌 %d샷은 옛 대표가 " "있어도 완료로 세지 않는다", len(_blocked)) found = len(seen_still_ids) if found < expected: return CompletionReport( is_complete=False, missing=[ (f"게이트 상태를 못 읽어 검사 불가 " f"(expected={expected}, found={found})") if _unverifiable else (f"{expected - found} scene image(s) missing " f"(expected={expected}, found={found})") ], severity="missing" if found == 0 else "partial", metadata={"expected": expected, "found": found, "rows": _rows_n, "gate_held": len(_blocked), "gate_unjudged": len(_unjudged), "gate_unverifiable": _unverifiable}, ) return CompletionReport( is_complete=True, missing=[], severity="clean", metadata={"expected": expected, "found": found, "rows": _rows_n, "gate_held": len(_blocked), "gate_unjudged": len(_unjudged), "gate_unverifiable": _unverifiable}, ) # ── character_state_variant (order 24.5: 죽은/다친 인물 variant 참조 이미지) ── class CharacterStateVariantStep(_ImageStepMixin, StepRunner): """죽은/다친 인물의 상태 variant 참조 이미지 생성.""" def _evaluate_contract_drift(self, mismatch_reason: str): """상태 변형 팩 수리로 지문이 움직였을 때, env 승인이 있으면 **비파괴** 재실행. 없으면 기존대로 BLOCK — 승인 없는 drift 는 막는다.""" _acked = _drift_ack_decision(self.step_id, mismatch_reason) if _acked is not None: return _acked return super()._evaluate_contract_drift(mismatch_reason) def _config_hash(self) -> str: """상태 변형 팩 버전을 접는다. ★`prompt_sanitizer` 팩은 **안 접는다.** 안전 필터에 막힐 때만 도는 갈래라, 접으면 평소에 안 쓰는 팩 때문에 유료 단계가 다시 돈다. 대신 그 팩을 고칠 때는 이 스텝을 손으로 다시 돌려야 한다 — 여기 적어 둔다. """ from app.core.step_runner import compute_config_hash return _fold_prompt_packs( compute_config_hash(self.project_config), "character_state_variant", ("system",)) def _execute(self, mode="resume") -> Dict[str, Any]: self._setup_image_tracer_context() from app.modules.llm.gemini_image_client import GeminiImageClient from app.modules.pipeline.ref_image_pipeline import generate_and_validate_reference from app.modules.prompt_loader import load_prompt from app.models.project import ImageAsset from app.core.config import settings from app.core.file_paths import resolve_image_path, to_relative_image_path from pathlib import Path import uuid as _uuid from datetime import datetime, timezone # 1) shot_staging 로드 → 죽은/다친 인물 감지 staging_cp = self._load_prev_checkpoint("shot_staging") if not staging_cp: return {"completed_count": 0, "applicable_count": 0, "failed_count": 0, "data": {"state_variants": []}} affected: Dict[str, set] = {} # char_name → {state_type, ...} # (이름, 상태) → 그 상태가 나오는 샷들 — 입력 그림(옷)을 고르는 재료 occurs: Dict[tuple, list] = {} for shot in staging_cp.get("data", staging_cp).get("shots", []): for ca in shot.get("character_angles", []): state = ca["subject_state"] # required by v13 schema, KeyError = schema violation (Gate 4 fail-fast) if is_immobilized_state(state): name = ca.get("character", "") if name: affected.setdefault(name, set()).add(state) occurs.setdefault((name, state), []).append( (shot.get("scene_index"), shot.get("shot_index"))) if not affected: logger.info("character_state_variant: no dead/injured/unconscious characters found") return {"completed_count": 0, "applicable_count": 0, "failed_count": 0, "data": {"state_variants": []}} logger.info("character_state_variant: %d characters with state variants: %s", len(affected), list(affected.keys())) # 2) DB에서 이름 → UUID + description 매핑 (체크포인트에 id 없을 수 있음) # name_matcher: 원본 이름 + 괄호·공백 정규화 이름 모두 키로 등록 (드리프트 대응). from app.models.project import EntityCanon, EntityEpisodeLink _canons = [] # ★보류(shelved) 제외 (2026-09-04). char_links = self.db.query(EntityEpisodeLink).filter( EntityEpisodeLink.project_id == self.project_id, EntityEpisodeLink.episode_id == self.episode_id, EntityEpisodeLink.presence_status != _PRESENCE_SHELVED, ).all() for link in char_links: ec = self.db.query(EntityCanon).filter( EntityCanon.id == link.canon_id, EntityCanon.entity_type == "character", ).first() if ec: _canons.append(ec) name_to_uuid: Dict[str, str] = build_name_index( _canons, key_fn=lambda c: c.name, value_fn=lambda c: c.id, ) name_to_desc: Dict[str, str] = build_name_index( _canons, key_fn=lambda c: c.name, value_fn=lambda c: c.description or "", ) # ★상태 변형의 입력 = 그 상태가 나오는 **선택 샷에서 입은 옷** # (2026-09-20 — `state_variant_source` 머리말). 종전 「가장 최근 # 합성」은 찰리의 의식 없음을 외투 차림으로 만들었다(쓰는 샷은 옷 없음). from app.core.state_variant_source import ( existing_source_differs, pick_state_variant_source, source_asset_for, ) from app.models.project import SceneStill from app.modules.pipeline.still_recipe import ( build_short_id_resolver, extract_outfit_assignments, ) _ent_rows = self.db.query(EntityCanon).filter( EntityCanon.project_id == self.project_id).all() _resolver = build_short_id_resolver({ e.id: {"id": e.id, "short_id": e.short_id or "", "name": e.name, "entity_type": e.entity_type} for e in _ent_rows}) _o00_ids = {e.id for e in _ent_rows if e.short_id == "O00" and e.entity_type == "outlook"} _outfit_by_shot: Dict[tuple, Dict[str, str]] = {} for _r in self.db.query(SceneStill).filter( SceneStill.project_id == self.project_id, SceneStill.episode_id == self.episode_id, SceneStill.is_selected == True, # noqa: E712 SceneStill.still_index >= 0, SceneStill.status != "stale").all(): _om = extract_outfit_assignments( getattr(_r, "t2i_variations_json", None), _resolver) or {} if any(str(k).startswith("__") for k in _om): continue # 충돌·손상 배정 — 세지 않는다(지어내지 않는다) _outfit_by_shot[(_r.scene_index, _r.shot_index)] = _om # 3) force 모드: 기존 state variant 무효화 if mode == "force": old_variants = self.db.query(ImageAsset).filter( ImageAsset.project_id == self.project_id, ImageAsset.episode_id == self.episode_id, ImageAsset.asset_type == "reference", ImageAsset.prompt_used.like("%state_variant:%"), ).all() for ov in old_variants: if ov.prompt_used and "_old_" not in ov.prompt_used: ov.prompt_used = ov.prompt_used.replace("state_variant:", "state_variant_old:") self.db.commit() if old_variants: logger.info("character_state_variant force: marked %d old variants", len(old_variants)) # 이미지 생성 준비 reference_dir = Path(settings.projects_dir) / self.project_id / "reference_images" reference_dir.mkdir(parents=True, exist_ok=True) gemini_client = GeminiImageClient() prompt_template = load_prompt("character_state_variant", "system") or "" generated = 0 skipped = 0 failed = 0 state_variants = [] # ★채택 못 한 것을 **산출에 남긴다** — 로그로만 두면 「없었다」와 # 「만들었는데 못 쓴다」가 구별이 안 된다. unadopted: list = [] # entity_merge에 등록되지 않은 이름은 state_variant 대상에서 사전 제외. # 예: scene_consistency/scene_director가 이름 없는 단역을 "Unnamed Korean woman" 같은 # 보통명사로 식별한 경우. 이런 대상은 참조 이미지 생성이 불가능하며, 익명 단역이므로 # 참조 이미지 부재가 결과 품질 저하를 유발하지 않는다 → failed 집계 대신 제외. _unregistered = [n for n in affected if lookup_name(name_to_uuid, n) is None] if _unregistered: # warning으로 유지 — 보통명사 단역은 정상 케이스지만, 엔티티 이름 드리프트가 # name_matcher 정규화로도 잡히지 않는 수준(완전 다른 명칭)이면 여전히 알림 가치. logger.warning( "character_state_variant: skipping %d unregistered names (not in entity_merge): %s", len(_unregistered), _unregistered, ) # ref_image_gen 의 low-frequency skip policy 와 정렬한다. low_freq 로 # base reference 생성이 skip 된 registered canon 은 정상 reference 가 # 없어 state variant 도 만들 수 없고, pipeline_gate.check_scene_images_ready # 가 이 set 을 missing-ref gate 에서 제외하므로 여기 expected/generation # 에서도 동일하게 제외해야 strict projection 의 'analyzed' 전이와 충돌하지 # 않는다. 특정 이름 하드코딩이 아니라 ref_image_gen 이 산출한 skip set # 재사용 (단역 정책 일치). from app.core.low_freq_skip import load_low_freq_skip_ids _low_freq_skipped = load_low_freq_skip_ids(self.project_id, self.episode_id) _affected_registered = { n: states for n, states in affected.items() if lookup_name(name_to_uuid, n) is not None and lookup_name(name_to_uuid, n) not in _low_freq_skipped } _low_freq_excluded = [ n for n in affected if lookup_name(name_to_uuid, n) is not None and lookup_name(name_to_uuid, n) in _low_freq_skipped ] if _low_freq_excluded: logger.info( "character_state_variant: excluding %d low-frequency-skipped " "registered character(s) (no base ref per ref_image_gen policy): %s", len(_low_freq_excluded), _low_freq_excluded, ) # affected: {name → {state1, state2, ...}} → flatten to (name, state) pairs _all_pairs = [(name, st) for name, states in _affected_registered.items() for st in states] for char_name, state_type in _all_pairs: char_uuid = lookup_name(name_to_uuid, char_name) # 위에서 등록 확인됨 # resume: 이미 생성된 variant가 있으면 스킵 existing = self.db.query(ImageAsset).filter( ImageAsset.project_id == self.project_id, ImageAsset.episode_id == self.episode_id, ImageAsset.asset_type == "reference", ImageAsset.prompt_used.like(f"%state_variant:{char_uuid}:{state_type}%"), ).first() existing_path = resolve_image_path(existing.file_path) if existing else None # 그 상태가 나오는 선택 샷에서 입은 옷 → 입력 그림 _want = pick_state_variant_source( [(_outfit_by_shot.get(k) or {}).get(char_uuid) for k in occurs.get((char_name, state_type), []) if k in _outfit_by_shot], _o00_ids) _src = source_asset_for(self.db, self.project_id, char_uuid, _want) if existing and mode == "resume" and existing_path and existing_path.exists(): if existing_source_differs(existing.input_image_ids, _src.id if _src is not None else None): # ★입력이 지금 규칙과 다르다(다른 옷에서 만들었다) — 지우지 # 않고 표식을 바꿔 내린 뒤 다시 만든다(되돌리기: 표식 복원). existing.prompt_used = (existing.prompt_used or "").replace( "state_variant:", "state_variant_oldsrc:", 1) self.db.commit() logger.info( "character_state_variant: %s (%s) — 입력이 규칙과 다르다 " "(→ %s) — 내리고 다시 만든다", char_name, state_type, _src.id) else: skipped += 1 state_variants.append({ "char_name": char_name, "char_uuid": char_uuid, "state_type": state_type, "image_path": str(existing_path), }) logger.info("character_state_variant: %s (%s) — skipped (exists)", char_name, state_type) continue ref_bytes = None source_asset_id = None # persist-all Wave 1 — state_variant lineage source if _src is not None: _src_path = resolve_image_path(_src.file_path) if _src_path and _src_path.exists(): ref_bytes = _src_path.read_bytes() source_asset_id = _src.id # 규칙이 못 고르면(동수·재료 없음) 종전대로: 가장 최근 합성, 없으면 기본 comp_asset = None if ref_bytes is not None else ( self.db.query(ImageAsset) .filter( ImageAsset.project_id == self.project_id, ImageAsset.asset_type == "reference", ImageAsset.prompt_used.like(f"%composite:{char_uuid}%"), ) .order_by(ImageAsset.created_at.desc()) .first() ) comp_path = resolve_image_path(comp_asset.file_path) if comp_asset else None if ref_bytes is not None: pass elif comp_path and comp_path.exists(): ref_bytes = comp_path.read_bytes() source_asset_id = comp_asset.id else: face_asset = ( self.db.query(ImageAsset) .filter( ImageAsset.entity_id == char_uuid, ImageAsset.asset_type == "reference", ImageAsset.is_primary == 1, ) .first() ) face_path = resolve_image_path(face_asset.file_path) if face_asset else None if face_path and face_path.exists(): ref_bytes = face_path.read_bytes() source_asset_id = face_asset.id if not ref_bytes: logger.warning("character_state_variant: no ref image for %s — skipping", char_name) failed += 1 continue # 프롬프트 구성 — state_type 은 line 661 gate (is_immobilized_state) # 통과한 immobilized 값만. get_visual_descriptor 를 이 gate 밖에서 호출 # 금지 (alive / 비enum 모두 AppError 발생, Gate 4). state_desc = get_visual_descriptor(state_type) char_desc = lookup_name(name_to_desc, char_name) or "" prompt = (prompt_template .replace("{state_type}", state_type) .replace("{character_name}", char_name) .replace("{character_description}", char_desc) .replace("{state_description}", state_desc)) extra_refs = [("Reference: this character in their normal, healthy state", ref_bytes)] try: pipe_result = generate_and_validate_reference( gemini_client=gemini_client, entity_name=f"{char_name} ({state_type})", entity_description=f"{char_name} in {state_type} state: {state_desc}", entity_type="character_state_variant", t2i_prompt=prompt, output_dir=reference_dir, extra_references=extra_refs, style_context="", # W3 observability — llm_call_log PID/EID NULL 차단. trace_meta={ "project_id": self.project_id, "episode_id": self.episode_id, "operation_type": "character_state_variant", "entity_id": char_uuid, }, # ★★안전 필터에 막혀 프롬프트를 다시 쓸 때 **이 상태를 # 지우지 말라**고 알려 준다 (2026-09-19). 안 주면 # 「다친 모습」이 「말짱한 모습」으로 바뀐 채 그 상태의 # 참조로 저장된다 — 실측으로 그렇게 됐다. state_hint=state_type, ) now = datetime.now(timezone.utc) # ★★**상태 보존을 확인 못 하면 채택하지 않는다** (2026-09-19). # # 안전 필터가 프롬프트를 거부하면 사니타이저가 글을 다시 쓴다. # 실측(앰버 10세, severely_injured): 세 판을 유료로 만들었는데 # **세 판 모두 안 다친 그림**이 나왔고 어른이 옆에 같이 그려졌다. # 그런데도 `severely_injured` 참조로 저장되고 단계는 성공으로 # 셌다 — 틀린 참조가 씬에 붙는다. # # ★`sanitization_info` 가 있다는 것은 「상태가 지워졌다」가 # 아니라 **「상태가 보존됐는지 확인 못 했다」**는 뜻이다 # (Codex 지적). 그래서 사유를 그렇게 적는다. 수정 없이도 # 상태를 잘못 그릴 수 있으므로, 이것으로 상태 정확성 검증을 # 닫았다고 말해서는 안 된다. # # ★파일은 이미 쓰였다(공통 생성기). 지우지 않고 **표식을 달리 # 해서** 재사용·씬 선택에서 빠지게 한다 — 「파일이 없다」와 # 「유효한 상태 참조로 안 쓴다」는 다른 말이다. # ★대체 제공자로 만든 것은 **미확인이 아니다** — 글을 한 # 글자도 안 바꾸고 원문 그대로 다른 모델에 보낸 것이다. # 사니타이저가 돌았을 때만 「상태 보존 미확인」이다. _unverified = bool(pipe_result.get("sanitization_info")) if pipe_result.get("fallback_provider"): _unverified = False _marker = ("state_variant_unadopted" if _unverified else "state_variant") _state_asset = ImageAsset( id=str(_uuid.uuid4()), project_id=self.project_id, asset_type="reference", entity_id=char_uuid, episode_id=self.episode_id, file_path=to_relative_image_path(pipe_result["file_path"]), prompt_used=f"[{_marker}:{char_uuid}:{state_type}] {char_name}", generation_model=pipe_result.get("generation_model", ""), status=("rejected" if _unverified else "generated"), review_notes=("상태 보존 미확인으로 비채택 — 안전 필터로 " "프롬프트가 수정됐다(상태가 지워졌다는 판정은 " "아니다)" if _unverified else None), is_primary=0, created_at=now, ) self.db.add(_state_asset) # persist-all Wave 1 — state_variant lineage: 정상상태 합성/얼굴 → 상태변형. annotate_generated_asset( _state_asset, pipeline_role="character_state_variant", stage="character_state_variant", input_image_ids=[source_asset_id] if source_asset_id else [], ) self.db.commit() if _unverified: unadopted.append({ "char_name": char_name, "char_uuid": char_uuid, "state_type": state_type, "image_path": pipe_result["file_path"], "reason": "상태 보존 미확인 — 안전 필터로 프롬프트가 수정됨", }) logger.warning( "character_state_variant: %s (%s) — 비채택 " "(상태 보존 미확인)", char_name, state_type) continue generated += 1 state_variants.append({ "char_name": char_name, "char_uuid": char_uuid, "state_type": state_type, "image_path": pipe_result["file_path"], }) logger.info("character_state_variant: %s (%s) — generated", char_name, state_type) except Exception as exc: logger.warning("character_state_variant: %s (%s) failed: %s", char_name, state_type, exc) try: self.db.rollback() except Exception: pass failed += 1 return { "completed_count": generated + skipped, "applicable_count": len(_all_pairs), # ★비채택은 **실패로 센다.** 성공으로 위장하면 「그 상태 참조가 # 있다」고 하류가 읽는다. 단계가 partial 로 떨어지는 것이 사실대로다. "failed_count": failed + len(unadopted), "data": { "state_variants": state_variants, # ★「만든 적 없다」와 「만들었는데 못 쓴다」는 다른 칸이다. "unadopted_state_variants": unadopted, "unadopted_count": len(unadopted), }, } def verify_completion(self): """shot_staging 체크포인트의 dead/severely_injured/unconscious subject_state 기준 (char_uuid, state) 쌍 수 == ImageAsset(prompt_used~'state_variant:{char}:{state}') row + 파일 존재. production _execute()와 동일한 data source(shot_staging manifest.json) 사용 — DB에 shot_staging 테이블 없음. entity_merge에 미등록된 이름은 expected에서 제외 (production이 익명 단역으로 사전 제외하는 정책 일치). cleanup_artifacts override 안 함 → default noop (부분 재생성 보호). """ import re as _re from app.core.file_paths import resolve_image_path from app.core.integrity_report import CompletionReport from app.models.project import EntityCanon, EntityEpisodeLink, ImageAsset # 1) shot_staging 체크포인트 로드 → (char_name, state) 쌍 추출 staging_cp = self._load_prev_checkpoint("shot_staging") if not staging_cp: return CompletionReport( is_complete=True, missing=[], severity="clean", metadata={"expected": 0, "found": 0}, ) affected: Dict[str, set] = {} for shot in staging_cp.get("data", staging_cp).get("shots", []): for ca in shot.get("character_angles", []): state = ca["subject_state"] # required by v13 schema, KeyError = schema violation (Gate 4 fail-fast) if is_immobilized_state(state): name = ca.get("character", "") if name: affected.setdefault(name, set()).add(state) if not affected: return CompletionReport( is_complete=True, missing=[], severity="clean", metadata={"expected": 0, "found": 0}, ) # 2) DB에서 이름 → UUID 매핑 (entity_merge 미등록 이름은 익명 단역 → 제외) # ★보류(shelved) 제외 (2026-09-04). char_links = self.db.query(EntityEpisodeLink).filter( EntityEpisodeLink.project_id == self.project_id, EntityEpisodeLink.episode_id == self.episode_id, EntityEpisodeLink.presence_status != _PRESENCE_SHELVED, ).all() canons = [] for link in char_links: ec = self.db.query(EntityCanon).filter( EntityCanon.id == link.canon_id, EntityCanon.entity_type == "character", ).first() if ec: canons.append(ec) name_to_uuid: Dict[str, str] = build_name_index( canons, key_fn=lambda c: c.name, value_fn=lambda c: c.id, ) # _execute 와 동일한 low-frequency skip policy 적용 — ref_image_gen 이 # base ref 를 skip 한 registered canon 은 expected 에서 제외 (production # _execute 도 제외하므로 verify expected 와 일치시켜 sticky partial 방지). from app.core.low_freq_skip import load_low_freq_skip_ids low_freq_skipped = load_low_freq_skip_ids(self.project_id, self.episode_id) expected_pairs: list = [] for name, states in affected.items(): char_uuid = lookup_name(name_to_uuid, name) if char_uuid is None: continue # production이 사전 제외하는 익명 단역 if char_uuid in low_freq_skipped: continue # low-freq skip → base ref 없음 → _execute 도 제외 (정책 일치) for state in states: expected_pairs.append((char_uuid, state)) expected = len(expected_pairs) if expected == 0: return CompletionReport( is_complete=True, missing=[], severity="clean", metadata={"expected": 0, "found": 0}, ) # 3) ImageAsset 조회 → state_variant prompt_used 매칭 + 파일 검증 sv_rows = self.db.query(ImageAsset).filter( ImageAsset.project_id == self.project_id, ImageAsset.asset_type == "reference", ImageAsset.prompt_used.like("%state_variant:%"), ).all() existing_keys = set() for r in sv_rows: m = _re.search(r'state_variant:([a-f0-9-]+):(\w+)', r.prompt_used or "") if not m: continue p = resolve_image_path(r.file_path) if p and p.exists(): existing_keys.add((m.group(1), m.group(2))) found = sum(1 for pair in expected_pairs if pair in existing_keys) if found < expected: return CompletionReport( is_complete=False, missing=[ f"{expected - found} state variants missing " f"(expected={expected}, found={found})" ], severity="missing" if found == 0 else "partial", metadata={"expected": expected, "found": found}, ) return CompletionReport( is_complete=True, missing=[], severity="clean", metadata={"expected": expected, "found": found}, ) # ── Step Registry ── IMAGE_STEP_CLASSES = { "world_guide": WorldGuideStep, "ref_image_gen": RefImageGenStep, "composite_image_gen": CompositeImageGenStep, "character_state_variant": CharacterStateVariantStep, "scene_image_pipeline": SceneImagePipelineStep, }