"""이미지 Phase StepRunner 서브클래스 — Steps 11-14.

force 모드: 기존 이미지 삭제 안 함. 새 이미지 추가 생성 (is_primary 이동).
resume 모드: 미완성분만 이어서 생성.
"""

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

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()

        # 엔티티 + 씬
        links = self.db.query(EntityEpisodeLink).filter(
            EntityEpisodeLink.project_id == self.project_id,
            EntityEpisodeLink.episode_id == self.episode_id,
        ).all()
        canon_ids = [l.canon_id for l in links]
        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 _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,
        }

    def verify_completion(self):
        """active non-location/non-outlook entity 수 ==
        ImageAsset(reference, primary, entity_id ∈ scope) row + 파일 존재.

        production reference_phase1_service.py:81은 entity_type ∉
        {location, outlook}인 모든 엔티티에 ref 생성 (character + prop + 기타).
        verify는 동일 scope을 반영해야 prop ref 누락이 false-clean으로 묻히지
        않는다 (A6 IMPORTANT).

        low_freq_skip된 entity는 expected에서 제외 (character/prop 동일).
        cleanup_artifacts override 안 함 → default noop (부분 재생성 보호).
        """
        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 EntityCanon, EntityEpisodeLink, ImageAsset

        # 에피소드의 ref 대상 canon_ids (location/outlook 제외).
        active_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,
                EntityCanon.entity_type.notin_(["location", "outlook"]),
            ).all()
        ]
        skipped = load_low_freq_skip_ids(self.project_id, self.episode_id)
        expected_ids = [cid for cid in active_ids if cid not in skipped]
        expected = len(expected_ids)

        if expected == 0:
            return CompletionReport(
                is_complete=True, missing=[], severity="clean",
                metadata={"expected": 0, "found": 0, "rows": 0},
            )

        rows = self.db.query(ImageAsset).filter(
            ImageAsset.project_id == self.project_id,
            ImageAsset.episode_id == self.episode_id,
            ImageAsset.asset_type == "reference",
            ImageAsset.is_primary == 1,
            ImageAsset.entity_id.in_(expected_ids),
        ).all()
        found = 0
        for r in rows:
            p = resolve_image_path(r.file_path)
            if p and p.exists():
                found += 1

        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": len(rows)},
            )
        return CompletionReport(
            is_complete=True, missing=[], severity="clean",
            metadata={"expected": expected, "found": found, "rows": len(rows)},
        )


# ── Step 13: composite_image_gen ──

class CompositeImageGenStep(_ImageStepMixin, StepRunner):
    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
        links = self.db.query(EntityEpisodeLink).filter(
            EntityEpisodeLink.project_id == self.project_id,
            EntityEpisodeLink.episode_id == self.episode_id,
        ).all()
        char_ids = set()
        for l in links:
            ec = self.db.query(EntityCanon).filter(
                EntityCanon.id == l.canon_id, 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}")

        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},
        }

    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,
                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,
                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) ===
        all_combos = self.db.query(CharacterOutlook).filter(
            CharacterOutlook.project_id == self.project_id,
            CharacterOutlook.character_id.in_(char_ids),
        ).all() 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명인 orphan outlook은 skip 근거 없으므로 expected 유지.
        wearer_links = self.db.query(CharacterOutlook).filter(
            CharacterOutlook.project_id == self.project_id,
            CharacterOutlook.outlook_id.in_(outlook_ids),
        ).all() 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 True  # orphan: skip 근거 없음
            return not wearers.issubset(skipped)

        outlook_expected_ids = [oid for oid in outlook_ids if _outlook_kept(oid)]
        outlook_expected = len(outlook_expected_ids)

        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(사용자 의도 확인).
            from app.core.config import settings as _settings

            _ack = {
                s.strip()
                for s in str(getattr(
                    _settings, "step_config_drift_ack", "")).split(",")
                if s.strip()
            }
            if self.step_id in _ack:
                logger.warning(
                    "Step %s: config drift 를 env 승인으로 통과 — 비파괴 "
                    "resume 재실행 (%s)", self.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",
                )
        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)
        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.still_recipe import (
            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,
            "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
            payload["gg46_select_policy"] = _gg46_policy
            payload["gg46_judge_models"] = (
                f"{settings.gemini_text_model}"
                f"|{getattr(settings, 'grok_judge_model', '')}")
            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 관례 동형).
        if bool(getattr(settings, "still_fix_ref_gate_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-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()
            # (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()
        # ── 표기 문안 저작 (2026-08-14 #119②) — ON 일 때만 스탬프, 동형.
        if bool(getattr(settings, "signage_author_enabled", False)):
            from app.modules.pipeline.signage_author import (
                SIGNAGE_POLICY_VERSION as _sg_policy,
                resolve_signage_pack as _sg_pack_resolved,
                signage_pack_content_hash as _sg_pack_content,
            )

            payload["signage_author_enabled"] = True
            payload["signage_policy"] = _sg_policy
            payload["signage_pack"] = _sg_pack_resolved()
            payload["signage_pack_content"] = _sg_pack_content()
        # ── 인물 의상 잠금·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])
        # ── 최종 스틸 생성 엔진 (2026-08-13 grok 전환) — 비기본값일 때만
        # 스탬프, 동형. 백엔드는 조립(컴팩트 v17+시네마틱+ab 변주)과 생성
        # 모델을 함께 바꾸므로 스텝 층 drift 가 걸려야 샷별 JIT 검증까지
        # 내려간다. nb2(기본)=키 부재 byte-identical.
        if getattr(settings, "still_image_backend", "nb2") != "nb2":
            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)
        # ── 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_TRANSFORM_PROMPT_VERSION as _cine_sel,
                recipe_stem_content_hash as _cine_stem_content,
            )

            payload["still_cine_transform_enabled"] = True
            payload["cine_transform_model"] = settings.grok_image_model
            payload["cine_transform_pack"] = _recipe_pack_resolved(_cine_sel)
            payload["cine_transform_stem_content"] = _cine_stem_content(
                _cine_sel, _cine_mod.CINE_TRANSFORM_STEM)
            # Codex R1 BLOCK-2: 계약 버전은 inner 지문에만 접혀 있으면
            # 완료 스텝에 닿지 않는다 — 참조 구성·산출 규약 bump 가
            # outer clean SKIP 에 막혀 영원히 실행되지 않는 창. outer
            # 에도 접는다.
            payload["cine_transform_contract"] = (
                _cine_mod.CINE_CONTRACT_VERSION)
        # ── 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 가 직접 접음)가 가른다.
        from app.modules.pipeline.still_recipe import (
            STILL_COMPACT_PROMPT_VERSION as _broll_sel,
            SUPPORT_STILLNESS_PROMPT_VERSION as _sup_sel,
            recipe_stem_content_hash as _recipe_stem_content,
        )

        payload["still_roll_count"] = int(settings.still_recipe_roll_count)
        payload["broll_variation_stem_content"] = _recipe_stem_content(
            _broll_sel, "broll_composition_variation")
        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
            )
        # 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
            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 에 싣는다.
                    payload["chain_bg_location_pack"] = _bgf_pack(
                        _chain_loc_sel)
        # 좁고 복잡한 실내의 기하 권위 스템 팩 (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,
            resolve_prompt_version as _geom_pack,
        )

        payload["recipe_geom_authority_pack"] = _geom_pack(_geom_sel)
        # 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)
        # 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)
        # E2E10 fix② (2026-07-21): i2i 수정본 재판정 — 최종 _sel 결정
        # 정책 전환이 스틸 산출 실질 입력. ON 시만 스탬프 (OFF
        # byte-identical). per-shot 지문 기여는 run_multiroll_select 담당.
        if getattr(settings, "multiroll_fix_rejudge_enabled", False):
            from app.modules.pipeline.multiroll_select import (
                FIX_REJUDGE_POLICY_VERSION as _frj_policy,
            )

            payload["multiroll_fix_rejudge"] = _frj_policy
            payload["fix_rejudge_judge_model_physical"] = str(
                settings.gemini_text_model
            )
            # 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]
        scene_total = len(selected_still_ids)
        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()

        result: Dict[str, Any] = {
            "completed_count": primary_count,
            "applicable_count": scene_total,
            "failed_count": max(0, scene_total - primary_count),
            "data": {"scene_total": scene_total, "primary_count": 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 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.file_paths import resolve_image_path
        from app.core.integrity_report import CompletionReport
        from app.models.project import ImageAsset, 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},
            )

        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_(selected_ids),
        ).all()
        seen_still_ids: set = set()
        for r in rows:
            p = resolve_image_path(r.file_path)
            if p and p.exists() and r.still_id:
                seen_still_ids.add(r.still_id)
        found = len(seen_still_ids)
        if found < expected:
            return CompletionReport(
                is_complete=False,
                missing=[
                    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": len(rows)},
            )
        return CompletionReport(
            is_complete=True, missing=[], severity="clean",
            metadata={"expected": expected, "found": found, "rows": len(rows)},
        )


# ── character_state_variant (order 24.5: 죽은/다친 인물 variant 참조 이미지) ──

class CharacterStateVariantStep(_ImageStepMixin, StepRunner):
    """죽은/다친 인물의 상태 variant 참조 이미지 생성."""

    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, ...}
        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:
            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 = []
        char_links = self.db.query(EntityEpisodeLink).filter(
            EntityEpisodeLink.project_id == self.project_id,
            EntityEpisodeLink.episode_id == self.episode_id,
        ).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 "",
        )

        # 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 = []

        # 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
            if existing and mode == "resume" and existing_path and existing_path.exists():
                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

            # 기존 composite ref 로드 (없으면 face ref)
            ref_bytes = None
            source_asset_id = None  # persist-all Wave 1 — state_variant lineage source
            comp_asset = (
                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 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,
                    },
                )
                now = datetime.now(timezone.utc)
                _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"[state_variant:{char_uuid}:{state_type}] {char_name}",
                    generation_model=pipe_result.get("generation_model", ""),
                    status="generated",
                    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()
                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),
            "failed_count": failed,
            "data": {"state_variants": state_variants},
        }

    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 미등록 이름은 익명 단역 → 제외)
        char_links = self.db.query(EntityEpisodeLink).filter(
            EntityEpisodeLink.project_id == self.project_id,
            EntityEpisodeLink.episode_id == self.episode_id,
        ).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,
}