"""still_recipe_service — s40/s41 레시피 스틸 생성 경로 (still_recipe_mode="v1"). SceneImageService.generate_images 의 배치 경로 대신 **스토리 순서 순차** 루프로 씬 스틸을 생성한다(prev 앵커=직전 선정·수정본 `_sel` — s40 체인 계약). 각 샷: 참조 조립(still_recipe.build_still_refs: 플레이트/콘티/prev _sel/엔티티) → 프롬프트 조립(build_still_prompt: 텍스트 안전망 전체) → 공통 파이프(multiroll_select: N롤 nb2 → Gemini 단독 판정 → 선정 → 결함 검사 → i2i 수정 → `_sel`) → ImageAsset primary 영속(save_single_scene_asset) + scene_cp 완료 마킹. 중간 산출(롤 A/B/C·수정 전 원본·record)은 `/recipe/` 아래 파일 + records.json 으로 보존 — 재개(파일 단위 skip + critique 소급)와 리뷰 갤러리 소스. DB row 삭제·덮어쓰기 없음 (feedback_never_delete_images 유지). """ from __future__ import annotations import json import logging import shutil import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Sequence, Tuple # 이 파일은 지연 import 관례를 쓰지만 아래 둘은 상단에 둔다 — 순수 # 조립 함수라 순환이 없고(world_context=stdlib only, still_recipe= # prompt_loader only), 런 스코프 클로저(`_lane_place_facts`)가 소비해 # 호출마다 재 import 할 이유가 없다. from app.core.world_context import build_world_facts_block from app.modules.pipeline.still_recipe import build_place_facts_block logger = logging.getLogger(__name__) def _now() -> str: return datetime.now(timezone.utc).isoformat() def _require_mannequin_chain( lane_entries: Dict[str, Any], *, chain_ready: bool, ) -> None: """마네킹 콘티 CP ↔ 체인 플래그 정합 — lane_chain 분기 **밖**에서 무조건 호출한다. 양방향 검사다. ① chain_ready=False + 마네킹 팩 CP = 마네킹 유출(아래 첫 루프). lane_chain=True 안에서만 검사하면 플래그를 내린 뒤 v14 CP 를 재사용하는 경로가 그대로 남는다(그때 lane_chain 은 False 라 검사 자체가 실행되지 않는다). ② chain_ready=True + 구(비마네킹) 팩 CP = 그 반대 방향의 같은 구멍 (설계 §4.2(1)·BLOCKING-4 "소비자 exact revalidate"). 이 wave 전에 lane 콘티를 구운 프로젝트의 entry 에는 `lane_sketch_pack` 키가 **아예 없다**(스탬프 자체가 이 wave 산물). 운영자가 conti 를 다시 굽지 않고 still_recipe/이미지 스텝만 재개하면(파일 상단 주석이 경고하는 부분 실행) 구 v13 인물 그림에 Step1 이 "마네킹을 회색 그대로 두라"는 bg_fill_head 를, Step2 가 "모든 마네킹을 실제 인물로 바꾸라"는 stage_head_mannequin 을 건다 — 전부 조용히, 유료 렌더를 태우면서. 일반 콘티 팩을 보는 `bgfirst_conti_pack_mismatch` 는 이 wave 가 건드리지 않은 다른 키(conti_pack)라 잡지 못한다. 키 부재·값 불일치 둘 다 fail-closed. """ from app.core.errors import AppError from app.core.steps.shot_conti_light_step import ( LANE_SKETCH_PACK_VERSION, ) from app.modules.pipeline.outdoor_marker_map import ( _MANNEQUIN_SKETCH_PACKS, resolve_sketch_pack_version, ) if chain_ready: expected = resolve_sketch_pack_version(LANE_SKETCH_PACK_VERSION) for tag, entry in (lane_entries or {}).items(): # status!="ok" = 스케치를 굽지 않은 entry(structure_plate 정책· # 실패 격리) — 팩 스탬프 대상이 아니라 검사 대상도 아니다. if not isinstance(entry, dict) or entry.get("status") != "ok": continue pack = str(entry.get("lane_sketch_pack") or "") if pack != expected: raise AppError( code="still_recipe.lane_sketch_pack_mismatch", message=( f"lane 콘티({tag})가 기대 sketch 팩({expected})이 " f"아님 (lane_sketch_pack={pack or None!r}) — " "체인 ON 은 마네킹 콘티 전제이므로 구 팩 콘티를 " "새 계약으로 소비할 수 없다. shot_conti_light 를 " "먼저 재실행하세요 (fail-closed)" ), status_code=422, ) return for tag, entry in (lane_entries or {}).items(): if not isinstance(entry, dict): continue pack = str(entry.get("lane_sketch_pack") or "") selector = pack.split(".", 1)[0] if selector in _MANNEQUIN_SKETCH_PACKS: raise AppError( code="still_recipe.lane_mannequin_chain_off", message=( f"lane 콘티({tag})가 마네킹 팩 {pack} 로 생성됐는데 " "bgfirst/bgfirst_full/lane_prev 플래그가 모두 ON 이 " "아님 — 마네킹이 legacy 조립으로 유출된다 " "(fail-closed)" ), status_code=422, ) def _load_cp( projects_dir: str, project_id: str, episode_id: str, step_id: str ) -> Dict[str, Any]: cp = ( Path(projects_dir) / project_id / "checkpoints" / "episodes" / episode_id / step_id / "manifest.json" ) if cp.exists(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: # noqa: BLE001 logger.warning("still_recipe: %s 로드 실패: %s", step_id, exc) return {} def _require_cp_data( projects_dir: str, project_id: str, episode_id: str, step_id: str ) -> Dict[str, Any]: """레시피 필수 상류 CP fail-closed (Codex 1차 리뷰 BLOCKING-1). mode=v1 에서 분류/연속성/콘티 CP 누락·비완료를 {} 로 삼아 조용히 일반 인물샷으로 degrade 하면 레시피 계약이 무너진다 — 명시 에러로 차단. """ from app.core.errors import AppError cp = _load_cp(projects_dir, project_id, episode_id, step_id) status = cp.get("status") data = cp.get("data") or {} if status != "completed" or not data: raise AppError( code="still_recipe.upstream_incomplete", message=( f"still_recipe_mode=v1 필수 상류 스텝 미완료: {step_id} " f"(status={status!r}) — 해당 스텝을 먼저 실행하세요" ), status_code=422, ) return data def parse_no_conti_exempt_tags(raw: Any) -> set: """운영자 선언 무콘티 예외 태그 파싱 (2026-07-30). 쉼표 구분 샷 태그 목록을 집합으로 만든다. **빈 값·공백·None 은 빈 집합** 이고 빈 집합은 어떤 샷에도 `tag in exempt` 를 참으로 만들지 않는다 — 즉 기본값은 기존 경로와 byte-identical 인 no-op 이다. 인라인 comprehension 이 아니라 이름 있는 함수로 둔 것은 이 no-op 성질을 유닛으로 고정하기 위해서다(Codex 리뷰 #8: 신규 심볼 테스트 0건). """ return { t.strip() for t in str(raw or "").split(",") if t.strip() } def _jit_tag_snapshot( records: "_Records", tag: str, known_tags: Optional[set] = None, ) -> str: """#77-B: 이 샷의 record 묶음 직렬화 — 지출 흔적 탐지용. ★공유 record 포함(Codex 재리뷰 BLOCK): `groupbg::장소` 처럼 샷 tag 네임스페이스 밖의 공유 record 도 이 샷을 처리하는 동안 갱신될 수 있고 (그룹 배경 재생성 = 유료), 걷기는 샷 단위 직렬이라 "이 샷 전후로 움직인 공유 record" = 이 샷의 지출이다. known_tags(전체 샷 tag 집합)를 받으면, `::` 가 있는데 접두가 샷 tag 가 아닌 키 전부를 공유로 본다 — 특정 접두 열거보다 넓어 새 공유 네임스페이스도 자동 포함된다. 유료 재개 경로(소급 critique·재판정·외부 2택1 등)는 전부 record 를 갱신한다(재개 규율 — 갱신 없는 유료 호출은 재개마다 재지출되는 결함). 그래서 "record 가 안 움직였다"가 "돈이 안 나갔다"의 근거가 된다. 산출 bytes 만 보면 결함 없음으로 끝난 소급 critique 처럼 **돈은 나가고 bytes 는 그대로**인 경우를 놓친다(2026-08-09 Codex 리뷰 재현). 표기용 키(ref_mode 등)는 뺀다 — 코드 배포로 문구만 바뀌어도 전 샷이 "지출"로 읽혀 제동이 거짓 발동하는 것을 막는다. 유료 흔적(critique· fix_rejudge·plate_select·conti_ab_decision·bgfirst 기록)은 남는다. ★tag prefix 는 반드시 `==` 또는 `::` 경계 — "S1sh1" 이 "S1sh10" 을 삼키면 안 된다. """ _presentation = { "ref_mode", "share_plan", "lane_policy", "locked_char_refs_excluded", } # 바퀴마다 뒤집히는 표식 — 지출 0 인데 지출로 읽히는 거짓 신호원. # · reused: 캐시 적중 여부(::variants 에서 False→True 확인) # · *_this_run: 이번 실행에서 했는지 표식(conti_ab.outer_judged_ # this_run 이 재사용 바퀴에 True→False — Codex 재리뷰 반례). # 중첩 깊이 어디서든 걸러낸다. 지출 흔적 본체(판정 결과·프롬프트· # 지문·산출 경로)는 남는다. _transient = {"reused"} def _scrub(v: Any) -> Any: if isinstance(v, dict): return { kk: _scrub(vv) for kk, vv in v.items() if kk not in _transient and not kk.endswith("_this_run") } if isinstance(v, list): return [_scrub(x) for x in v] return v _known = known_tags or set() bundle: Dict[str, Any] = {} for k, v in records.data.items(): _shared = ( "::" in k and k.split("::", 1)[0] not in _known and k != tag ) if k == tag or k.startswith(f"{tag}::") or _shared: if k == tag and isinstance(v, dict): v = {kk: vv for kk, vv in v.items() if kk not in _presentation} bundle[k] = _scrub(v) return json.dumps(bundle, ensure_ascii=False, sort_keys=True) class StillJitRegenLimitExceeded(RuntimeError): """#77-B 폭주 제동 — 완료 샷의 JIT 재생성이 상한에 닿았다. 완료 샷이 무더기로 어긋난다는 것은 지문 체계가 통째로 움직였다는 뜻 (팩·설정 대량 변경 또는 결함) — 확인 없이 과금으로 옮기지 않는다. 상한 도달 시 recipe 디렉토리에 래치 파일을 먼저 남기고 raise 하므로, 자동 재시도(resume 루프)는 **지출 없이 즉시 같은 예외**로 끝난다 (2026-08-09 Codex BLOCK 3 — 래치 없이는 바퀴당 상한만큼 돈이 샌다). 해제 = 사람이 원인 확인 후 래치 파일 삭제 (+필요시 STILL_JIT_REGEN_ LIMIT 상향) 후 resume.""" class StillJitVerifyIncomplete(RuntimeError): """#77-B — JIT 검증 대상 샷이 실패한 채 걷기가 끝났다. 완료 샷의 재생성 실패는 옛 primary 가 남아 있어 스텝 집계(primary 개수)에 안 잡힌다 — 그대로 두면 스텝이 completed 로 닫히고 새 config_hash CP 가 봉인돼, 다음 resume 은 whole-step SKIP 으로 그 낡은 샷을 영구 동결한다(2026-08-09 Codex BLOCK 1). 그래서 걷기 끝에 이 예외로 스텝을 실패로 남긴다 — resume 재시도는 성공분을 전부 재사용(지출 0)하고 실패분만 다시 시도한다.""" class StillCineTransformIncomplete(RuntimeError): """#108 (Codex R1 BLOCK-1) — cine 변환 실패 샷을 남긴 채 걷기가 끝났다. 실패 샷은 원본 sel 로 영속돼(체인 안전) 스텝 집계에는 안 잡힌다 — 그대로 completed 로 닫으면 config_hash 가 그대로라 다음 resume 이 whole-step SKIP 하고, "변환 스테이지 ON 완주"가 실은 일부 미변환인 채 봉인된다(결과 오독). 걷기 끝에 이 예외로 스텝을 실패로 남긴다 — resume 재진입은 건강한 샷 전량 재사용(지출 0)+실패 변환만 재시도 (~$0.03/샷)라 바퀴 비용이 유계다. 영구 실패는 autodrive 상한(3회) 에서 사람에게 올라온다.""" class StillEraPreserveIncomplete(RuntimeError): """(era R2 BLOCK-1) — era 조사 미성립으로 기존 배경을 보존한 채 걷기가 끝났다. 보존 분기는 현재 계약 검증(모델·팩·group/context_sig·prompt·conti 지문 대조)을 통째로 미룬다 — 그대로 completed 로 닫으면 새 config_hash CP 가 봉인돼 다음 resume 이 whole-step SKIP 하고, 검증 안 된 옛 배경 위에 진짜 drift 까지 영구 동결된다(StillJitVerifyIncomplete 와 같은 부류). 걷기 끝에 이 예외로 스텝을 실패로 남긴다 — resume 은 성공분 전량 재사용(지출 0)하고 보존 그룹의 조사만 재시도하며, 조사 성공 방문이 현재 계약을 정상 검증한다. 조사가 계속 실패하면 ERA_RESEARCH_ENABLED 전환 여부는 사람이 결정한다.""" JIT_LATCH_FILENAME = "JIT_REGEN_LATCH.json" def _jit_write_latch(recipe_dir: Path, **info: Any) -> Path: """상한 래치 영속 — 있으면 다음 걷기가 지출 전에 멈춘다.""" import datetime as _dt path = recipe_dir / JIT_LATCH_FILENAME payload = { "why": ( "완료 샷 JIT 재생성이 상한에 닿았다 — 지문 체계가 통째로 " "움직인 사고 의심. 사람이 원인을 확인하기 전까지 이 스텝의 " "재실행은 지출 없이 즉시 실패한다." ), "how_to_clear": ( "원인 확인 후 이 파일을 삭제하고 (필요시 STILL_JIT_REGEN_" "LIMIT 상향) resume. force 금지." ), "at": _dt.datetime.now(_dt.timezone.utc).isoformat(), **info, } tmp = path.with_suffix(".tmp") tmp.write_text( json.dumps(payload, ensure_ascii=False, indent=1), encoding="utf-8") tmp.replace(path) return path class _Records: """recipe 디렉토리의 multiroll record 영속 — 재개(critique 소급) SOT.""" def __init__(self, path: Path): self._path = path self.data: Dict[str, Any] = {} if path.exists(): try: self.data = json.loads(path.read_text(encoding="utf-8")) except Exception: # noqa: BLE001 logger.warning("still_recipe: records.json 파싱 실패 — 재생성") def save(self) -> None: tmp = self._path.with_suffix(".tmp") tmp.write_text( json.dumps(self.data, ensure_ascii=False, indent=1), encoding="utf-8", ) tmp.replace(self._path) def run_still_recipe_generation( *, db: Any, project_id: str, episode_id: str, stills: List[Dict[str, Any]], stills_orm: List[Any], entity_lookup: Dict[str, Dict[str, Any]], ref_image_map: Dict[str, bytes], reference_svc: Any, scene_ref_image_map: Dict[str, bytes], scene_ref_asset_id_map: Dict[str, str], staging_map: Dict[str, Any], scene_cp: Any, persistence_svc: Any, progress: Any, project_config: Optional[Dict[str, Any]], scene_dir: Path, already_done_stills: set, target_scenes: Optional[Sequence[int]] = None, ) -> int: """레시피 v1 순차 생성. 반환 = 생성 완료 샷 수. target_scenes (2026-07-23 슬라이스 실행, Codex 설계 합의): 표적 씬 목록 — 설정 시 **실행 루프만** effective allowlist(표적 씬 스틸+ effective prev 체인 재귀 클로저)로 제한. 위쪽 컨텍스트(tag map/ groupbg_context/group_sig/canonical origin)는 전체 stills 파생 그대로 유지(BLOCKING-1 — 지문 불변). None(default)=전체 byte- identical. """ from app.core.config import settings from app.modules.pipeline.multiroll_gemini import ( GQ_CRITIQUE_PACK_VERSION, GQ_SELECT_POLICY_VERSION, make_gemini_critique_fn, make_gemini_judge_fn, make_gq_critique_fn, make_qk_critique_fn, make_nb2_gen_fn, resolve_judge_pack_version, resolve_judge_texts, STILL_FIX_REJUDGE_HEADER_PACK_VERSION, STILL_JUDGE_PACK_VERSION, ) from app.modules.pipeline.multiroll_select import ( build_critique_schema, build_judge_schema, roll_labels, run_multiroll_select, ) from app.modules.pipeline.shot_conti_light import resolve_shot_plate_map from app.modules.pipeline.shot_ref_classify import ( derive_location_by_scene, tag_of, ) from app.modules.pipeline.still_recipe import ( build_still_prompt, build_still_refs, ve_ids_for_shot_ex, ) from app.modules.pipeline.shot_conti_light import ( build_pose_clauses, ) projects_dir = settings.projects_dir roll_count = int(settings.still_recipe_roll_count) critique_enabled = bool(settings.still_recipe_critique_enabled) # fix1 (2026-07-19): staging 구도·스케일 계약(CAMERA/FRAME 절) 주입 camera_frame_on = bool(settings.still_recipe_camera_frame_enabled) # fix⑤ (2026-07-21 E2E10): staging lighting_mood → LIGHTING & MOOD 절 # +재질 사실감 — 전 샷(bgonly·lane 포함: 조명은 배치 권위와 무충돌) lighting_on = bool( getattr(settings, "still_recipe_lighting_enabled", False)) # fix④⑤ (2026-07-22 E2E11): 연기·형상 계약(naturalism+drawn_mark 절, # 팩 v10) — naturalism=인물 샷, drawn_mark=전 샷(bgonly 포함) conduct_on = bool( getattr(settings, "still_recipe_conduct_enabled", False)) # 2026-08-12 차렷/증명사진 대응: identity 참조 역할 한정 절(팩 v16 단일 # 스템) — 캐릭터 참조 실첨부 샷에만. OFF(default)=byte-identical. identity_role_on = bool( getattr(settings, "still_identity_ref_role_enabled", False)) # BGFIRST2 이식 ②③ (2026-07-20 사용자 확정): 일반 콘티 샷=2단 체인 # (Step1 GPT 재투영 빈 배경 → Step2 nb2 인물 삽입) + 2택1(체인 vs # 무콘티). OFF(default)=기존 경로 byte-identical. bgfirst_on = bool(getattr(settings, "still_bgfirst_enabled", False)) # fix③④ full (2026-07-21 사용자 확정 "콘티=무조건 배경 우선 전면화"): # 콘티 실재 전 샷 체인 — no_plate 샷=groupbg(장소 단위 공유 배경), # complex/seed-bg 샷=체인 2택1 편입. bgfirst ON 전제(아래 fail-closed). bgfirst_full_on = bool( getattr(settings, "still_bgfirst_full_enabled", False)) # 2026-07-25 사용자 확정 (케이스1 스펙 E·F): lane·prev 샷도 배경 # 재투영 체인 편입 — full 전제 (full OFF 면 체인 자체가 없다). lane_prev_chain_on = bgfirst_full_on and bool( getattr(settings, "still_lane_prev_bgfirst_enabled", False)) # 운영자 선언 무콘티 예외 (2026-07-30): 콘티가 끝내 실패한 샷을 운영자가 # 명시적으로 BGFIRST 정책 대상 밖으로 선언 → 기존(무콘티) 경로로 생성. # ""(default)=빈 집합=기존과 byte-identical. 선언되지 않은 샷의 "legacy # 무음 하강 금지" 계약은 그대로다 — 여기서 풀리는 것은 선언된 태그뿐이고, # 적용 시 WARNING 으로 남긴다(조용한 degrade 금지). _no_conti_exempt = parse_no_conti_exempt_tags( getattr(settings, "bgfirst_no_conti_exempt_tags", "")) if _no_conti_exempt: logger.warning( "still_recipe: 운영자 선언 무콘티 예외 %d샷 — BGFIRST 대상 제외 " "후 기존 경로 생성: %s", len(_no_conti_exempt), sorted(_no_conti_exempt), ) # ── 레시피 체크포인트 로드 (필수 3종 fail-closed) ──────────────── classify = _require_cp_data( projects_dir, project_id, episode_id, "shot_ref_classify" ) classify_shots: Dict[str, Any] = classify.get("shots", {}) or {} classify_scenes: Dict[str, Any] = classify.get("scenes", {}) or {} # v2: 원문 근거 LLM 저작 세계 앵커 — " — " 접두로 스틸 헤드에 주입 # (E2E 육안 실측: 앵커 부재 시 사진 속 인물 등 정체성 드리프트) _wa = (classify.get("world_anchor_en") or "").strip() world_anchor = f" — {_wa}" if _wa else "" continuity = _require_cp_data( projects_dir, project_id, episode_id, "shot_continuity" ) conti_data = _require_cp_data( projects_dir, project_id, episode_id, "shot_conti_light" ) # BGFIRST2: 체인은 원근 가이드 콘티(v2) 전제 — 구(v1) 콘티와의 조용한 # 혼용 금지(run-all 은 conti 스텝 config_hash 변화로 자동 재생성되지만 # 부분 실행은 stale 소비 가능). CP 의 conti_pack 는 스텝이 v2 일 때만 # 기록(opt-in 키). if bgfirst_full_on and not bgfirst_on: from app.core.errors import AppError raise AppError( code="still_recipe.bgfirst_full_requires_bgfirst", message=( "still_bgfirst_full_enabled=true 는 still_bgfirst_enabled" "=true 전제 — 조용한 부분 적용 금지 (fail-closed)" ), status_code=422, ) # 2026-07-25 (Codex NARROW-6): lane·prev 체인 편입도 full 전제 — # 켜 놓고 조용히 no-op 되면 운영자가 적용됐다고 오인한다. if ( bool(getattr(settings, "still_lane_prev_bgfirst_enabled", False)) and not bgfirst_full_on ): from app.core.errors import AppError raise AppError( code="still_recipe.lane_prev_chain_requires_bgfirst_full", message=( "still_lane_prev_bgfirst_enabled=true 는 still_bgfirst_" "full_enabled=true 전제 — 체인 자체가 없는 상태에서 " "조용한 no-op 금지 (fail-closed)" ), status_code=422, ) if bgfirst_on: from app.core.errors import AppError from app.modules.pipeline.shot_conti_light import ( resolve_prompt_version as conti_pack_version, ) # full=콘티 팩 v5(키샷 레이아웃 가치, E2E13 fix①⑦ — v4 소품 # 방향+자연 연기+no_plate 헤더 승계) 전제 — 구팩 콘티와의 조용한 # 혼용 금지 _expected_conti_pack = conti_pack_version( "5" if bgfirst_full_on else "2") if conti_data.get("conti_pack") != _expected_conti_pack: raise AppError( code="still_recipe.bgfirst_conti_pack_mismatch", message=( "still_bgfirst_enabled=true 인데 shot_conti_light CP 가 " f"기대 콘티 팩({_expected_conti_pack})이 아님 " f"(conti_pack={conti_data.get('conti_pack')!r}) — " "shot_conti_light 를 먼저 재실행하세요" ), status_code=422, ) # B-3 (2026-07-19): 배경 공유 계획 — 존재 시 prev/배경 결정의 상위 # 권위(부재=classify 판정 fail-safe). 스텝 flag OFF=no-op CP 라 빈 dict. share_plans: Dict[str, Any] = {} _sp_data = _load_cp( projects_dir, project_id, episode_id, "background_share_plan" ).get("data", {}) or {} if (_sp_data.get("plan") or {}).get("shot_plans"): share_plans = dict(_sp_data["plan"]["shot_plans"]) # fix③④: 장소 단위(groupbg) 묶음 = share_plan 그룹(LLM 이 '비슷한 # 배경 공유'로 판단한 샷 집합) — tag→group_key 색인 # E2E11 ③: 그룹 evidence(씬 원문 인용)도 색인 — groupbg 장소 근거 절 group_of: Dict[str, str] = {} group_evidence: Dict[str, List[str]] = {} for _g in (_sp_data.get("plan") or {}).get("share_groups") or []: _gkey = str((_g or {}).get("group_key") or "") for _t in (_g or {}).get("shot_tags") or []: group_of[str(_t)] = _gkey _quotes = [ str((_ev or {}).get("quote_ko") or "").strip() for _ev in (_g or {}).get("evidence") or [] if isinstance(_ev, dict) ] if _gkey and any(_quotes): group_evidence[_gkey] = [q for q in _quotes if q] # E2E11 ③: groupbg LOCATION DETAIL — entity_merge 실측 loc 서술/traits # (scene→primary loc name 매칭, 실패=빈 문자열 degrade) location_detail_by_name: Dict[str, str] = {} for _loc in ( _load_cp(projects_dir, project_id, episode_id, "entity_merge") .get("data", {}) or {} ).get("locations", []) or []: if not isinstance(_loc, dict): continue _lname = str(_loc.get("name") or "").strip() if not _lname: continue _ldesc = str(_loc.get("description") or "").strip() _ltraits = [ str(t).strip() for t in (_loc.get("visual_traits") or []) if str(t).strip() ] _detail = _lname if _ldesc: _detail += f": {_ldesc}" if _ltraits: _detail += " (특징: " + "; ".join(_ltraits) + ")" location_detail_by_name[_lname] = _detail # 2026-07-26 확정 흐름: 참조 0 배경의 장소·world 권위는 텍스트다. # lane ON 일 때 런당 1회 로드해 Step1 프롬프트에 싣는다. _place_spec_groups: Dict[str, Any] = ( _load_cp(projects_dir, project_id, episode_id, "outdoor_place_spec").get("data", {}) or {} ).get("groups", {}) or {} _vwr_cp = _load_cp( projects_dir, project_id, episode_id, "visual_world_rules") _lane_world_block = build_world_facts_block(_vwr_cp) def _lane_place_facts(group_key: str) -> str: """lane 샷의 장소 사실 블록 — 결손이면 fail-closed. 성공 entry 는 {spec, attempts, outdoor_loc_ids, scene_indices} 이고 `status` 키가 **없다**(실패 entry 만 error 를 갖는다, outdoor_place_spec_step.py:228-233). canon CP 는 status 를 갖는 다른 shape 라 혼동하지 말 것. 세 번째 shape 가 더 있다 — 근거 씬 0 인 그룹의 저작 스킵 {skipped, outdoor_loc_ids} (같은 파일 197-200). spec 키 자체가 없어 어차피 아래 dict 가드에 걸리지만, 운영자에게는 "dict 아님"이 아니라 "저작이 스킵됐다"로 보여야 원인(씬 매핑 결손)에 닿는다. """ entry = (_place_spec_groups or {}).get(group_key) if not isinstance(entry, dict) or entry.get("error"): raise ValueError( f"lane 배경: place spec 그룹({group_key!r}) 결손/실패 " f"({(entry or {}).get('error')!r}) — 장소 사실 없이 " "배경 생성 금지 (fail-closed)" ) if entry.get("skipped"): raise ValueError( f"lane 배경: place spec 그룹({group_key!r}) 저작 스킵 " f"({entry.get('skipped')!r}) — 근거 씬이 매핑되지 않아 " "스펙이 없음, 장소 사실 없이 배경 생성 금지 (fail-closed)" ) spec = entry.get("spec") if not isinstance(spec, dict): raise ValueError( f"lane 배경: place spec 그룹({group_key!r}) 의 spec 이 " "dict 아님 — fail-closed" ) block = build_place_facts_block(spec) if not block.strip(): raise ValueError( f"lane 배경: place spec 그룹({group_key!r}) 에서 장소 " "사실 블록이 비었음 — fail-closed" ) return block contis: Dict[str, Any] = conti_data.get("contis", {}) or {} map_conti: Dict[str, Any] = conti_data.get("map_conti", {}) or {} # Stage D: 야외 lane 샷 — 콘티=마커 스케치, 참조=[스케치(+seed)]+엔티티 # (+prev). 플레이트/photo canon/map 참조 0 계약. 프롬프트/라벨 팩 # selector = LANE_PROMPT_VERSION (모듈 SOT). from app.modules.pipeline.still_recipe import ( COMPLEX_AB_CONTRACT_VERSION, COMPLEX_PROMPT_VERSION as _COMPLEX_PACK, LANE_PROMPT_VERSION as _LANE_PACK, SEED_BG_PROMPT_VERSION as _SEED_BG_PACK, STILL_CONDUCT_PROMPT_VERSION as _CONDUCT_PACK_SEL, ) lane_conti: Dict[str, Any] = conti_data.get("lane_conti", {}) or {} # 마네킹 유출 3층(소비자 거부): 마네킹 CP ↔ 체인 플래그 정합을 # lane_chain 분기와 무관하게 여기서 무조건 검사한다(분기 안에서 하면 # 플래그를 내린 뒤 lane_chain=False 가 되어 검사 자체가 실행되지 # 않는다 — 같은 구멍이 남는다). # # chain_ready 는 플래그를 다시 풀어쓰지 않고 위에서 계산한 파생 # local 을 그대로 쓴다 — 가드의 정당성이 "lane_chain 이 True 가 될 # 수 있는 상태"와 정확히 같아야 하는데(아래 per-shot 루프의 # `lane_chain = lane_prev_chain_on and lane_used`), 같은 조건을 두 곳에 # 손으로 적으면 한쪽만 바뀌어 드리프트한다. _require_mannequin_chain( lane_conti, chain_ready=bool(bgfirst_on and lane_prev_chain_on), ) # R1: 콘티형 샷 플레이트 권위 — shot_conti_light 선행 판정 기록 소비 plate_authority: Dict[str, Any] = ( conti_data.get("plate_authority", {}) or {} ) plate_map = resolve_shot_plate_map(projects_dir, project_id, episode_id) # 맵 기반 플레이트가 성립한 샷은 스틸 참조도 맵 플레이트로 대체 from app.modules.pipeline.shot_ref_classify import parse_tag map_plate_keys: set = set() for tag, entry in map_conti.items(): p = (entry or {}).get("plate_path") if p and Path(p).exists(): si, shi = parse_tag(tag) plate_map[f"{si}_{shi}"] = Path(p) map_plate_keys.add(f"{si}_{shi}") # E2E6 피드백 ⑤: 같은 location 복수 플레이트 → 샷별 VLM 서브공간 선택 # (flag OFF=기존 배정 byte-identical). 맵 플레이트 대체 샷은 비대상. plate_select_on = bool( getattr(settings, "still_plate_select_enabled", False)) # E2E6 ⑧: 일반 콘티 샷 A/B(콘티 포함/미포함) — OFF=기존 byte-identical conti_ab_on = bool( getattr(settings, "still_conti_ab_enabled", False)) plate_cands_by_loc: Dict[str, Any] = {} plate_assign_by_key: Dict[str, str] = {} if plate_select_on: from app.modules.pipeline.plate_select import ( load_bg_groups_and_assignment, plate_candidates_by_location, ) _bg_groups, plate_assign_by_key = load_bg_groups_and_assignment( projects_dir, project_id, episode_id) plate_cands_by_loc = plate_candidates_by_location(_bg_groups) # ── 장소 서술·시간대 ──────────────────────────────────────────── scene_save = _load_cp( projects_dir, project_id, episode_id, "scene_save" ).get("data", {}) scene_headings: Dict[int, str] = {} for seg in scene_save.get("segments", []) or []: si = seg.get("scene_index") if isinstance(si, int) and seg.get("heading"): scene_headings[si] = seg["heading"] director = _load_cp( projects_dir, project_id, episode_id, "scene_director" ).get("data", {}) scene_primary: Dict[int, str] = {} for sc in director.get("scenes", []) or []: si = sc.get("scene_index") if si is not None and sc.get("primary_location"): scene_primary[int(si)] = sc["primary_location"] location_by_scene = derive_location_by_scene( db, project_id, scene_primary, scene_headings ) # ── 엔티티 참조·traits ────────────────────────────────────────── # entity_lookup: id → {id, name, entity_type, short_id?, stable_traits?} def _entity_field(e: Dict[str, Any], key: str) -> Any: return e.get(key) if isinstance(e, dict) else None traits_by_id: Dict[str, str] = {} for eid, e in entity_lookup.items(): raw = _entity_field(e, "stable_traits") if not raw: continue try: vals = json.loads(raw) if isinstance(raw, str) else raw if isinstance(vals, list): traits_by_id[eid] = ", ".join(str(v) for v in vals) except Exception: # noqa: BLE001 continue ve_by_key: Dict[Tuple[int, int], List[str]] = {} ve_outlook_by_key: Dict[Tuple[int, int], Dict[str, str]] = {} for s in stills: si, shi = s.get("scene_index"), s.get("shot_index") if si is None or shi is None: continue ids: List[str] = [] outlooks: Dict[str, str] = {} try: for v in json.loads(s.get("visible_entities_json") or "[]"): if isinstance(v, dict): eid = v.get("id") or v.get("entity_id") if isinstance(eid, str) and eid: ids.append(eid) # 의상 composite 선택 신호 (BLOCKING-2) ol = v.get("outlook_id") if isinstance(ol, str) and ol: outlooks[eid] = ol elif isinstance(v, str) and v: ids.append(v) except json.JSONDecodeError: pass ve_by_key[(int(si), int(shi))] = ids ve_outlook_by_key[(int(si), int(shi))] = outlooks shot_desc_by_id: Dict[str, str] = { s.id: (getattr(s, "shot_description", None) or "") for s in stills_orm } # ── 의상 배정 SOT: t2i_variations_json[*].outfit_assignments (Codex 2차 # 리뷰 B1) — VE 에는 outlook 이 실리지 않는 것이 production shape. # short id(C##/O##) → UUID 정규화, variation 간 상이 배정은 fail-closed. short_to_uuid: Dict[str, str] = {} for _eid, _e in entity_lookup.items(): _sid = (_e or {}).get("short_id") if _sid: short_to_uuid[_sid] = _eid def _norm_uuid(v: str) -> Optional[str]: if not isinstance(v, str) or not v: return None if v in entity_lookup: return v return short_to_uuid.get(v) from app.modules.pipeline.still_recipe import extract_outfit_assignments outfit_by_still_id: Dict[str, Dict[str, str]] = { s_orm.id: extract_outfit_assignments( getattr(s_orm, "t2i_variations_json", None), _norm_uuid ) for s_orm in stills_orm } # ── 엔진·판정 어댑터 (multiroll_gemini 공용) ──────────────────── from app.modules.pipeline.multiroll_gemini import ( judge_pack_content_hash, resolve_select_judge_model, resolve_select_judge_model_physical, ) from app.modules.pipeline.still_recipe import ( resolve_prompt_version as recipe_pack_version, ) judge_texts = resolve_judge_texts( roll_count, judge_name="judge_still", pack_version=STILL_JUDGE_PACK_VERSION) # 선정 판정 모델 — 키 유무로 alias 가 갈린다(Claude Opus / gemini-pro). # 판정 모델이 바뀌면 어느 롤이 뽑히는지가 바뀌므로 지문에 실어 산출을 # 무효화한다. alias 만으로는 물리 모델 교체를 못 잡아 둘 다 싣는다. _select_judge_alias = resolve_select_judge_model() _select_judge_physical = resolve_select_judge_model_physical() # 입력 지문 extra — 모델·팩 변경도 산출 무효화 (Codex BLOCKING-3) # 2026-08-13 Codex R1 BLOCK-4: image_model 은 **실제 backend 의 모델** # 이어야 한다 — grok2 상태에서 gemini 모델명을 접으면 grok_image_model # 만 바꾼 재실행이 기존 샷을 fresh 로 오판한다. nb2(기본)는 기존 키·값 # 그대로(완료 산출 지문 보존), 비기본 backend 만 키 추가. _still_backend = getattr(settings, "still_image_backend", "nb2") extra_fingerprint = { "image_model": ( settings.grok_image_model if _still_backend == "grok2" else settings.gemini_image_model), "judge_model": _select_judge_alias, "select_judge_model_physical": _select_judge_physical, "recipe_pack": recipe_pack_version("1"), # Codex 배치 리뷰 HIGH-3: 판정·결함 계약 팩도 스틸 산출 실질 입력. # 2026-08-07: 스틸 전용 selector — 전역 기본을 올리면 구조물 씨드· # 배경 플레이트까지 stale 되어 재실행이 번진다. "judge_pack": resolve_judge_pack_version(STILL_JUDGE_PACK_VERSION), # 2026-08-08: 버전 문자열만으로는 부족하다 — 판정 텍스트는 생성 # 프롬프트에 안 들어가서, 셀렉터 상수·서버 메모리 중 한 층만 낡아도 # 팩을 올리고도 지문이 그대로다(최종 스틸 253/311 이 재판정 없이 # 재사용된 실제 사례). 로드 대상 디렉토리의 bytes 해시를 접는다. "judge_pack_content": judge_pack_content_hash( STILL_JUDGE_PACK_VERSION), } if _still_backend != "nb2": # 비기본 backend 만 키 추가 — nb2 기존 지문 보존(키 부재=기본 관례). extra_fingerprint["image_backend"] = _still_backend # 덜어내기 정책은 발송 직전(fit_grok_prompt)에 적용돼 지문에 접히는 # 명목 프롬프트에 안 남는다 — 정책만 바꾸면 스텝 층은 다시 열려도 # (image_steps 가 접는다) 완료 샷이 "신선"으로 읽혀 재생성이 0건이다. # 기준선과 **달라진 때만** 키를 붙여 오늘의 지문은 안 움직인다. from app.modules.pipeline.still_recipe import ( GROK_SHED_POLICY_BASELINE as _shed_baseline, grok_shed_policy_signature as _shed_sig_fn, ) _shed_sig = _shed_sig_fn() if _shed_sig != _shed_baseline: extra_fingerprint["grok_shed_policy"] = _shed_sig # (Codex R2) b 변주·몸-지지 절의 per-shot 무효화 범위는 여기 전역 # 키가 아니라 실제 소비 층이 가른다: 몸-지지 절=base prompt bytes, # b 변주 절=roll_prompts(compute_input_fingerprint 가 직접 접음), # 롤 수=같은 함수의 roll_count 인자. 전역 whole-pack 키는 bg_only # 같은 비소비 샷까지 stale 시켜 제거했다. # ── G+Q 판정 체계 (2026-08-10 설계 §2.5) — ON 일 때만 스탬프 ── # OFF=무스탬프 byte-identical 이 계약이다: 이 기본값이 완료 산출(금월도 # 256샷)의 지문을 지킨다. 모델 쌍 자체는 위 select_judge_model_physical # 이 이미 갈라 준다(resolve 가 ON 이면 gemini+qwen 을 반환). gq_judge_on = bool(getattr(settings, "multiroll_gq_judge_enabled", False)) if gq_judge_on: extra_fingerprint["gq_select_policy"] = GQ_SELECT_POLICY_VERSION extra_fingerprint["gq_critique_pack"] = resolve_judge_pack_version( GQ_CRITIQUE_PACK_VERSION) extra_fingerprint["gq_critique_pack_content"] = ( judge_pack_content_hash(GQ_CRITIQUE_PACK_VERSION)) # ── QK 판정 체계 (2026-08-12 전환) — ON 일 때만 스탬프, GQ 와 동형 ── # OFF=무스탬프 byte-identical. GQ 와 동시 ON 은 resolve_select_judge_ # models 가 fail-closed 로 막는다(여기서 또 갈라 쓰지 않는다 — 판정 # 모델 구성의 권위는 그 함수 하나다). 모델 슬롯은 settings 가 SOT 라 # 슬롯 문자열 자체를 접는다(모델 교체=산출 무효화). qk_judge_on = bool(getattr(settings, "multiroll_qk_judge_enabled", False)) if qk_judge_on: from app.modules.pipeline.multiroll_gemini import ( QK_CRITIQUE_PACK_VERSION, QK_SELECT_POLICY_VERSION, ) extra_fingerprint["qk_select_policy"] = QK_SELECT_POLICY_VERSION # 물리 모델 쌍 (2026-08-12 반전: 메인=Qwen DashScope·보조=Gemini) extra_fingerprint["qk_judge_models"] = ( f"{settings.qwen_vlm_model}|{settings.gemini_text_model}") extra_fingerprint["qk_critique_pack"] = resolve_judge_pack_version( QK_CRITIQUE_PACK_VERSION) extra_fingerprint["qk_critique_pack_content"] = ( judge_pack_content_hash(QK_CRITIQUE_PACK_VERSION)) # ── G+G46 판정 체계 (2026-08-13) — ON 일 때만 스탬프, GQ/QK 동형 ── # OFF=무스탬프 byte-identical. 동시 ON 은 resolve_select_judge_models # 가 fail-closed. 판정 물리 쌍은 select_judge_model_physical 이 이미 # 갈라 주지만 QK 관례대로 슬롯 문자열도 접는다. **fix i2i 물리 모델** # (grok_image_model)은 이 체계의 산출 실질 입력이라 함께 접는다 — # 수정본 bytes 를 만드는 모델이 바뀌면 산출이 무효화돼야 한다. gg46_judge_on = bool( getattr(settings, "multiroll_gg46_judge_enabled", False)) if gg46_judge_on: from app.modules.pipeline.multiroll_gemini import ( GG46_CRITIQUE_PACK_VERSION, GG46_SELECT_POLICY_VERSION, ) extra_fingerprint["gg46_select_policy"] = GG46_SELECT_POLICY_VERSION extra_fingerprint["gg46_judge_models"] = ( f"{settings.gemini_text_model}" f"|{getattr(settings, 'grok_judge_model', '')}") extra_fingerprint["gg46_critique_pack"] = resolve_judge_pack_version( GG46_CRITIQUE_PACK_VERSION) extra_fingerprint["gg46_critique_pack_content"] = ( judge_pack_content_hash(GG46_CRITIQUE_PACK_VERSION)) extra_fingerprint["gg46_fix_image_model"] = str( getattr(settings, "grok_image_model", "")) # ── 참조 선별 (2026-08-19 사용자 지시) — ON 일 때만 스탬프, 동형 ── # ON 은 결함 검사 스키마(needs_ref_indices 칸)와 수정 호출의 실질 # 입력(붙는 참조 장수)을 함께 바꾼다 — 산출이 무효화돼야 맞다. fix_ref_gate_on = bool( getattr(settings, "still_fix_ref_gate_enabled", False)) fix_missing_texts: Dict[str, str] = {} if fix_ref_gate_on: from app.modules.pipeline.multiroll_gemini import ( FIX_MISSING_PACK_VERSION, fix_ref_contract_sha, load_fix_missing_texts, ) fix_missing_texts = load_fix_missing_texts(FIX_MISSING_PACK_VERSION) extra_fingerprint["fix_ref_gate"] = True extra_fingerprint["fix_missing_pack"] = resolve_judge_pack_version( FIX_MISSING_PACK_VERSION) extra_fingerprint["fix_missing_pack_content"] = ( judge_pack_content_hash(FIX_MISSING_PACK_VERSION)) # 선별을 실제로 정하는 문안(번호 목록 머리글 + 스키마 안 선별 기준 # 설명)은 팩이 아니라 코드에 있어 위 팩 해시가 못 덮는다. 그 문안을 # 고치면 붙는 참조와 최종 그림이 달라지는데 지문이 안 움직이면 # 반대 정책의 산출을 "같은 조건"으로 읽게 된다. extra_fingerprint["fix_ref_contract"] = fix_ref_contract_sha() # ── 시대 인지 사전 조사 (2026-08-14) — ON 일 때만 스탬프, 동형 ── # 판별은 모든 샷·엔티티·플레이트에 적용되므로(대상 목록이 사전에 # 없다 — 판별 자체가 실질 입력) 전역 스탬프가 정당하다. OFF=무스탬프 # byte-identical. era_research_on = bool(getattr(settings, "era_research_enabled", False)) if era_research_on: from app.modules.pipeline.era_research import ( ERA_RESEARCH_POLICY_VERSION, era_pack_content_hash, resolve_era_pack, ) extra_fingerprint["era_research_policy"] = ERA_RESEARCH_POLICY_VERSION extra_fingerprint["era_research_pack"] = resolve_era_pack() extra_fingerprint["era_research_pack_content"] = ( era_pack_content_hash()) # ── 표기 문안 저작 (2026-08-14 #119②) — ON 일 때만 스탬프, 동형 ── # 판별·저작이 모든 샷에 적용되므로 전역 스탬프. OFF=무스탬프 # byte-identical (S3 백지 팻말 실측 대응). # Codex BLOCK-2: subset 기능 키를 전역 per-shot 지문에 넣지 않는다 # (confined :921 선례) — 실효 산출은 signage_en/wearing/prev 절이 # prompt·refs 바이트로 접혀 갈리고, 완료 스텝 재방문은 outer # image_steps payload 스탬프가 보장한다. signage_on = bool(getattr(settings, "signage_author_enabled", False)) # ── 인물 의상 잠금·prev 계승 스코프 (2026-08-14 #119③) — ON 일 때만 # 스탬프, 동형. prev 라벨·continues 절과 PEOPLE wardrobe 병기가 전 # 샷의 조립 실질 입력이라 전역 스탬프. OFF=무스탬프 byte-identical # (S64sh4 롤 간 의상 표변·앵커 인물 복제 실측 대응). cast_lock_on = bool( getattr(settings, "still_cast_wardrobe_lock_enabled", False)) # 아웃룩 텍스트 대체 공급 재료 (#119③) — 시트·배정이 둘 다 없는 # 인물의 의상을 그 인물의 아웃룩 서술로 잠근다 (S64sh4 김창완·지훈 # 실측: 시트 0장 + 배정 빈 배열 → 롤마다 의상 발명). 배정이 있으면 # 그 아웃룩, 없고 정확히 1개면 그것 — 여럿이면 공급하지 않는다 # (어느 것인지 지어내지 않는다). OFF = 미조회·무영향. # Codex BLOCK-1: era 조사 실패의 걷기(1바퀴) 단위 sentinel — 같은 # 걷기에서 같은 대상 유료 사슬 재구매 금지, 다음 resume 걷기는 재시도. era_failed_memo: set = set() outlooks_by_char: Dict[str, List[str]] = {} outlook_desc_by_id: Dict[str, str] = {} if cast_lock_on: from app.models.project import CharacterOutlook, EntityCanon for _co in ( db.query(CharacterOutlook) .filter(CharacterOutlook.project_id == project_id) .all() ): outlooks_by_char.setdefault( _co.character_id, []).append(_co.outlook_id) _o_ids = {o for lst in outlooks_by_char.values() for o in lst} if _o_ids: for _o in ( db.query(EntityCanon) .filter(EntityCanon.id.in_(_o_ids)) .all() ): _desc = (getattr(_o, "description", None) or "").strip() if _desc: outlook_desc_by_id[_o.id] = _desc # ── confined fp 경로 (2026-08-11 설계 §— ON 일 때만 스탬프) ── # OFF=무스탬프 byte-identical 이 계약. ON 이면 confined 판별 통과 # 샷의 산출 실질 입력(팩 내용·fp 모델·정책)이 지문에 접힌다. confined_fp_on = bool( getattr(settings, "still_confined_fp_enabled", False)) # ★confined 키는 **전역 extra_fingerprint 에 넣지 않는다** (Codex # BLOCK-1): 전역에 넣으면 팩·모델이 바뀔 때 non-confined 샷까지 지문 # 불일치 → 유료 재생성으로 번진다(subset 기능이 전체 재생성을 부름). # confined 샷의 branch 지문에만 병합한다(아래 _confined_active 분기). confined_extra_fingerprint: Dict[str, Any] = {} if confined_fp_on: from app.modules.pipeline.confined_fp import ( CONFINED_FP_POLICY_VERSION, confined_fp_pack_content_hash, resolve_confined_fp_pack, ) confined_extra_fingerprint = { "confined_fp_pack": resolve_confined_fp_pack(), "confined_fp_pack_content": confined_fp_pack_content_hash(), "confined_fp_model": str( getattr(settings, "openai_image_model", "")), # 판별·readback 의 물리 모델도 산출 실질 입력 (BLOCK-2) "confined_fp_readback_model": str( getattr(settings, "gemini_text_model", "")), "confined_fp_policy": CONFINED_FP_POLICY_VERSION, } if camera_frame_on: # fix1: CAMERA/FRAME 절 스템 팩도 산출 실질 입력 (OFF=무스탬프 # byte-identical) from app.modules.pipeline.still_recipe import ( CAMERA_FRAME_PROMPT_VERSION as _CAM_PACK, ) extra_fingerprint["recipe_camera_frame_pack"] = recipe_pack_version( _CAM_PACK) if lighting_on: # fix⑤: LIGHTING & MOOD 절 스템 팩도 산출 실질 입력 (OFF=무스탬프 # byte-identical) from app.modules.pipeline.still_recipe import ( LIGHTING_MOOD_PROMPT_VERSION as _LIGHT_PACK, ) extra_fingerprint["recipe_lighting_pack"] = recipe_pack_version( _LIGHT_PACK) if conduct_on: # fix④⑤: 연기·형상 절 스템 팩도 산출 실질 입력 (OFF=무스탬프) from app.modules.pipeline.still_recipe import ( STILL_CONDUCT_PROMPT_VERSION as _CONDUCT_PACK, ) extra_fingerprint["recipe_conduct_pack"] = recipe_pack_version( _CONDUCT_PACK) if bgfirst_on: # BGFIRST2: 체인 팩·계약·Step1 엔진·판정 물리 모델 전부 산출 실질 # 입력 (ON 시만 기여 — OFF byte-identical) from app.modules.pipeline.still_recipe import ( BGFIRST_BG_IMAGE_MODEL as _BGFIRST_BG_MODEL, BGFIRST_CONTRACT_VERSION as _BGFIRST_CONTRACT, BGFIRST_PROMPT_VERSION as _BGFIRST_PACK, ) extra_fingerprint["recipe_bgfirst_pack"] = recipe_pack_version( _BGFIRST_PACK) extra_fingerprint["bgfirst_contract"] = _BGFIRST_CONTRACT extra_fingerprint["bgfirst_bg_model"] = _BGFIRST_BG_MODEL extra_fingerprint["bgfirst_judge_model_physical"] = str( _select_judge_physical) if bgfirst_full_on: # fix③④ full: 전면화 계약·전용 스템 팩(groupbg/seed 절)도 # 산출 실질 입력 (ON 시만 기여) from app.modules.pipeline.still_recipe import ( BGFIRST_FULL_CONTRACT_VERSION as _BGF_FULL_CONTRACT, BGFIRST_FULL_PROMPT_VERSION as _BGF_FULL_PACK, ) extra_fingerprint["bgfirst_full_contract"] = _BGF_FULL_CONTRACT extra_fingerprint["recipe_bgfirst_full_pack"] = ( recipe_pack_version(_BGF_FULL_PACK)) if bool(getattr(settings, "still_lane_prev_bgfirst_enabled", False)): # 2026-07-26 lane 확정 흐름: bg_fill 스템 팩 + 장소·world # 사실 계약도 lane 샷 산출의 실질 입력 (OFF=무스탬프 # byte-identical). full 전제는 위 fail-closed 가 보장한다. from app.modules.pipeline.still_recipe import ( BGFIRST_LANE_CONTRACT_VERSION as _BGF_LANE_CONTRACT, BGFIRST_LANE_PROMPT_VERSION as _BGF_LANE_PACK, CHAIN_BG_LOCATION_PROMPT_VERSION as _CHAIN_LOC_PACK, ) extra_fingerprint["bgfirst_lane_pack"] = recipe_pack_version( _BGF_LANE_PACK) extra_fingerprint["bgfirst_lane_contract"] = _BGF_LANE_CONTRACT # I-4: 체인 Step2 LOCATION 스템 selector — 샷 팩과 분리된 # 독립 축이라 lane 팩 스탬프가 덮어 준다는 보장이 없다 # (지금 같은 디렉토리로 풀리는 건 우연). 나란히 스탬프. extra_fingerprint["chain_bg_location_pack"] = ( recipe_pack_version(_CHAIN_LOC_PACK)) if lane_conti: # lane 팩(라벨+location authority)도 lane 샷 산출의 실질 입력 extra_fingerprint["recipe_lane_pack"] = recipe_pack_version( _LANE_PACK) # R3: 복잡 구조물 A/B 는 flag 무관 상시 강제 — v4 팩·계약 버전도 # 스틸 산출 실질 입력 (구 산출 무효화) extra_fingerprint["recipe_complex_pack"] = recipe_pack_version( _COMPLEX_PACK) extra_fingerprint["recipe_seed_bg_pack"] = recipe_pack_version( _SEED_BG_PACK) extra_fingerprint["complex_ab_contract"] = ( COMPLEX_AB_CONTRACT_VERSION) from app.modules.llm.gemini_image_client import GeminiImageClient from app.modules.prompt_sanitizer import PromptSanitizer # 최종 스틸 생성 클라이언트는 backend SOT 로 갈라 만든다 (Codex R1 # BLOCK-3): 여기서 항상 Gemini 를 주입하면 make_nb2_gen_fn 의 backend # 스위치(주입이 None 일 때만 동작)가 영원히 안 타서, grok2 설정인데 # 롤/fix/regen 산출·기록이 전부 Gemini 로 남는다(결과·기록 오독). if getattr(settings, "still_image_backend", "nb2") == "grok2": from app.modules.llm.grok_image_client import GrokImageClient shared_image_client = GrokImageClient() else: shared_image_client = GeminiImageClient() # E2E 실측: 시신·핏자국류 샷이 moderation 거절될 수 있음 — sanitizer # 1회 재시도 배선(실험 _soften 의 production 대응 경로) shared_sanitizer = PromptSanitizer(project_config=project_config) # ── i2i 시네마틱 변환 스테이지 (2026-08-13 #108, 사용자 확정 하이브리드 # "nb2 생성 → grok i2i 변환") — OFF(default)=byte-identical. 재료(문안· # 스템 해시·클라이언트)는 걷기 전에 한 번 준비: 문안 로드 실패는 즉시 # 크게 실패(샷 격리 대상이 아니라 배선 결함이다). 변환 클라이언트는 # backend 스위치와 무관하게 항상 grok — nb2 백엔드에서도 변환은 grok. cine_on = bool(getattr(settings, "still_cine_transform_enabled", False)) cine_prompt = "" cine_stem_hash = "" cine_pack = "" cine_client: Any = None if cine_on: from app.modules.llm.grok_image_client import ( GrokImageClient as _CineClient, ) from app.modules.pipeline.cine_transform import CINE_TRANSFORM_STEM from app.modules.pipeline.still_recipe import ( CINE_TRANSFORM_PROMPT_VERSION, build_cine_transform_prompt, recipe_stem_content_hash, resolve_prompt_version as _cine_pack_resolve, ) cine_prompt = build_cine_transform_prompt() cine_stem_hash = recipe_stem_content_hash( CINE_TRANSFORM_PROMPT_VERSION, CINE_TRANSFORM_STEM) cine_pack = _cine_pack_resolve(CINE_TRANSFORM_PROMPT_VERSION) cine_client = _CineClient() def make_shot_gen_fn(still_id: str, _branch_tag: str = ""): # per-shot still_id trace — llm_call_log 구조키 매칭(BLOCKING-2: # generation_call_id resolve)의 전제. 롤/fix 단위 multiroll_tag # 는 make_nb2_gen_fn 이 호출별 태그로 동적 기록(Codex H2) — # 여기서 고정 branch 태그를 넣지 않는다. return make_nb2_gen_fn( project_id=project_id, episode_id=episode_id, operation_type="still_recipe_roll", gemini_client=shared_image_client, sanitizer=shared_sanitizer, context_extra={"still_id": still_id}, ) # ── G+G46 fix i2i 백엔드 (2026-08-13 사용자 확정 "수정은 Grok") ── # 롤 생성은 backend SOT(위 shared_image_client — 기본 nb2) 그대로 두고 # **fix 편집만** grok(cine 변환과 동일 클라이언트 클래스·모델)으로 # 가른다. operation_type·multiroll_tag 기록 관례는 기존 fix 와 동일 — # 어느 모델이 편집했는지는 llm_call_log 의 model 이 가른다. OFF=주입 # 없음(run_branch_select 가 기존 gen 재사용) byte-identical. gg46_fix_client: Any = None if gg46_judge_on: from app.modules.llm.grok_image_client import ( GrokImageClient as _Gg46FixClient, ) gg46_fix_client = _Gg46FixClient() def make_shot_fix_gen_fn(still_id: str, _branch_tag: str = ""): return make_nb2_gen_fn( project_id=project_id, episode_id=episode_id, operation_type="still_recipe_roll", gemini_client=gg46_fix_client, sanitizer=shared_sanitizer, context_extra={"still_id": still_id}, ) judge_fn = make_gemini_judge_fn( judge_sys=judge_texts["judge_sys"], judge_schema=build_judge_schema( roll_labels(roll_count), with_physics=True), project_config=project_config, step_tag="still_recipe_judge", ) if gg46_judge_on: # G+G46 수정 흐름: Gemini+grok-4.6 양쪽 관찰 → Gemini 취합 # (CritiqueFn 계약 동일 — 하류 partition·심각도 게이트·fix 조립· # 재판정 무변경). 팩=v11(2관찰자 취합) — severity 구조 필드 필수 # (#106 게이트 전제, QK v10 관례 동형). from app.modules.pipeline.multiroll_gemini import ( make_gg46_critique_fn, ) critique_fn = make_gg46_critique_fn( critique_schema=build_critique_schema( with_severity=True, with_ref_gate=fix_ref_gate_on), project_config=project_config, step_tag="still_recipe_critique", ref_gate=fix_ref_gate_on, ) elif qk_judge_on: # QK 수정 흐름: Gemini 관찰 → Qwen 취합 (CritiqueFn 계약 동일 — # 하류 partition·fix 조립·재판정 무변경). 팩=v10 — 취합 이슈에 # severity 구조 필드 필수(#106 심각도 게이트의 전제). 구 경로 # (v7 critique·GQ v8)는 무severity 스키마 그대로 — 지문 불변. critique_fn = make_qk_critique_fn( critique_schema=build_critique_schema( with_severity=True, with_ref_gate=fix_ref_gate_on), project_config=project_config, step_tag="still_recipe_critique", ref_gate=fix_ref_gate_on, ) elif gq_judge_on: # G+Q 수정 흐름: Qwen 관찰 → Gemini 취합 (CritiqueFn 계약 동일 — # 하류 partition·fix 조립·재판정 무변경). critique_sys(v7)는 이 # 경로에서 안 쓴다 — 관찰·취합 계약은 v8 스템이 SOT. critique_fn = make_gq_critique_fn( critique_schema=build_critique_schema( with_ref_gate=fix_ref_gate_on), project_config=project_config, step_tag="still_recipe_critique", ref_gate=fix_ref_gate_on, ) else: critique_fn = make_gemini_critique_fn( critique_sys=judge_texts["critique_sys"], critique_schema=build_critique_schema( with_ref_gate=fix_ref_gate_on), project_config=project_config, step_tag="still_recipe_critique", ref_gate=fix_ref_gate_on, ) # ── fix-rejudge (E2E10 fix②): i2i 수정본 무판정 확정 → [원본 vs # 수정본] 2후보 블라인드 재판정. 전 스틸 브랜치 공용 — flag OFF= # 미전달(run_branch_select 가 kwargs 자체를 생략 → byte-identical) ── fix_rejudge_fn: Any = None if bool(getattr(settings, "multiroll_fix_rejudge_enabled", False)): from app.modules.pipeline.multiroll_gemini import ( load_fix_rejudge_header, ) _rj_texts = resolve_judge_texts( 2, judge_name="judge_still", pack_version=STILL_JUDGE_PACK_VERSION) # Codex HIGH-4: 기본 헤더('generated from this')는 수정본(repair # prompt 생성)에 거짓 — provenance 비노출 중립 헤더(팩 v3)로 판정 _rj_header = load_fix_rejudge_header( STILL_FIX_REJUDGE_HEADER_PACK_VERSION) fix_rejudge_fn = make_gemini_judge_fn( judge_sys=_rj_texts["judge_sys"], judge_schema=build_judge_schema( roll_labels(2), with_physics=True), project_config=project_config, step_tag="still_recipe_fix_rejudge", prompt_header=_rj_header, ) # 재판정 물리 모델·헤더 — 산출 실질 입력 (variants HIGH-3 관례) # 재판정도 `make_gemini_judge_fn` 경유이므로 선정 판정과 같은 모델이다. extra_fingerprint["fix_rejudge_judge_model_physical"] = str( _select_judge_physical ) extra_fingerprint["fix_rejudge_judge_header"] = _rj_header # ── GPT 구도 critique (E2E11 fix③): Gemini critique 와 합산 수정 — # flag OFF=미전달(run_branch_select kwargs 생략 → byte-identical) ── composition_critique_fn: Any = None # G+Q/QK/G+G46 ON 이면 GPT 구도 critique 는 배선하지 않는다(2026-08-10 # 합의 3 — 구도 관찰은 관찰 축에 포함. G+G46 은 관찰자가 이미 둘이라 # 셋째 관찰자를 더하지 않는다). if critique_enabled and not gq_judge_on and not qk_judge_on \ and not gg46_judge_on and bool( getattr(settings, "multiroll_gpt_composition_enabled", False) ): from app.modules.pipeline.multiroll_gemini import ( GPT_COMPOSITION_PACK_VERSION, make_gpt_composition_critique_fn, resolve_judge_pack_version as _judge_pack_resolve, ) composition_critique_fn = make_gpt_composition_critique_fn( critique_schema=build_critique_schema( with_ref_gate=fix_ref_gate_on), ref_gate=fix_ref_gate_on, project_config=project_config, step_tag="still_recipe_gpt_composition", ) extra_fingerprint["gpt_composition_pack"] = _judge_pack_resolve( GPT_COMPOSITION_PACK_VERSION) extra_fingerprint["gpt_composition_model_physical"] = str( getattr(settings, "openai_model", "") ) # ── 스틸 변형 2종×4택1 (2026-07-17 사용자 확정, flag OFF=byte-identical) variants_on = bool(getattr(settings, "still_variants_enabled", False)) sv_judge_header: Optional[str] = None judge_fn_by_count: Dict[int, Any] = {} judge_texts_by_count: Dict[int, Dict[str, str]] = {} if variants_on: from app.modules.pipeline.still_variants import ( STILL_VARIANTS_CONTRACT_VERSION, load_variant_judge_header, resolve_variants_pack_version as _sv_pack_version, ) # HIGH-6: 변형 모드 judge header — 기본 header('generated from # this')는 변형 롤에서 거짓. 지문 기여는 run 자체 지문(judge_ # prompt_header param)이 담당. sv_judge_header = load_variant_judge_header("1").strip() # 변형 저작 팩·계약 버전 — 스틸 산출 실질 입력 (ON 시만 기여) extra_fingerprint["still_variants_pack"] = _sv_pack_version("1") extra_fingerprint["still_variants_contract"] = ( STILL_VARIANTS_CONTRACT_VERSION ) # 리뷰 HIGH-3: judge_model alias 는 물리 모델 교체를 감지 못함 — # 변형 모드 판정(gemini-pro→gemini_text_model 해석)의 물리 모델을 # per-shot 지문에 병행 스탬프 (lane/conti_ab 우연 ON 의존 제거) extra_fingerprint["still_variants_judge_model_physical"] = str( _select_judge_physical ) # 후보 수별 판정·결함 계약 — A/B 4택1=4, 비A/B 변형=2 (count # word·라벨 집합이 judge_sys/critique_sys·schema 실질 입력. # 리뷰 HIGH-2: critique 도 'best pick of {count}' 문구라 legacy # roll_count 계약 재사용 금지) critique_fn_by_count: Dict[int, Any] = {} for _n in (2, 4): _jt = resolve_judge_texts( _n, judge_name="judge_still", pack_version=STILL_JUDGE_PACK_VERSION) judge_texts_by_count[_n] = _jt judge_fn_by_count[_n] = make_gemini_judge_fn( judge_sys=_jt["judge_sys"], judge_schema=build_judge_schema( roll_labels(_n), with_physics=True), project_config=project_config, step_tag="still_recipe_judge", prompt_header=sv_judge_header, ) if gg46_judge_on: # G+G46 도 count 무관(2단 계약 동형). 스키마는 본선과 같은 # with_severity=True — v11 취합 스템이 severity·observation_ # index 를 필수로 지시하므로 스키마가 그 계약과 어긋나면 # 안 된다(심각도 게이트도 본선과 동일하게 산다). from app.modules.pipeline.multiroll_gemini import ( make_gg46_critique_fn as _make_gg46_cf, ) critique_fn_by_count[_n] = _make_gg46_cf( critique_schema=build_critique_schema( with_severity=True, with_ref_gate=fix_ref_gate_on), project_config=project_config, step_tag="still_recipe_critique", ref_gate=fix_ref_gate_on, ) elif qk_judge_on: # QK 수정 흐름도 count 무관(GQ 와 같은 2단 계약) — variants # 샷이 legacy 흐름으로 빠지면 지문(qk_critique_pack)과 # 실행이 어긋난다(GQ BLOCK-2 와 동축). # ★2026-08-20: 본선 QK(:1161)와 같은 with_severity=True — # 빠져 있으면 additionalProperties:false 라 취합기가 severity # 를 못 실어 심각도 게이트(#106)가 이 갈래에서만 죽는다. # v10 팩 취합 스템이 그 칸을 요구하므로 계약과도 어긋났다. critique_fn_by_count[_n] = make_qk_critique_fn( critique_schema=build_critique_schema( with_severity=True, with_ref_gate=fix_ref_gate_on), project_config=project_config, step_tag="still_recipe_critique", ref_gate=fix_ref_gate_on, ) elif gq_judge_on: # G+Q 수정 흐름은 count 무관(한 장 검사 — 관찰·취합 스템에 # count word 가 없다) — 같은 어댑터를 count 칸에 공유해도 # 계약이 같다. Codex 리뷰 BLOCK-2: 기본 critique 만 갈랐더니 # variants 샷이 legacy 흐름을 타면서 지문(gq_critique_pack) # 과 실행이 어긋났다. critique_fn_by_count[_n] = make_gq_critique_fn( critique_schema=build_critique_schema( with_ref_gate=fix_ref_gate_on), project_config=project_config, step_tag="still_recipe_critique", ref_gate=fix_ref_gate_on, ) else: critique_fn_by_count[_n] = make_gemini_critique_fn( critique_sys=_jt["critique_sys"], critique_schema=build_critique_schema( with_ref_gate=fix_ref_gate_on), project_config=project_config, step_tag="still_recipe_critique", ref_gate=fix_ref_gate_on, ) # ── BGFIRST2 전용 어댑터 (2택1 판정·Step1 GPT 엔진) ───────────── bgfirst_judge_fn: Any = None bgfirst_critique_fn: Any = None bgfirst_judge_texts: Any = None bgfirst_judge_header: Any = None _bgfirst_gpt_client: Any = None if bgfirst_on: from app.core.openai_keys import openai_client from app.modules.pipeline.still_recipe import ( load_bgfirst_judge_header, ) # 2택1 중립 헤더 — 체인/무콘티 후보는 참조·프롬프트가 다르므로 # 기본 헤더('generated from this')가 거짓 (variants HIGH-6 연속) bgfirst_judge_header = load_bgfirst_judge_header() bgfirst_judge_texts = resolve_judge_texts( 2, judge_name="judge_still", pack_version=STILL_JUDGE_PACK_VERSION) bgfirst_judge_fn = make_gemini_judge_fn( judge_sys=bgfirst_judge_texts["judge_sys"], judge_schema=build_judge_schema( roll_labels(2), with_physics=True), project_config=project_config, step_tag="still_recipe_judge", prompt_header=bgfirst_judge_header, ) if gg46_judge_on: # BGFIRST 샷도 G+G46 수정 흐름을 타야 지문(gg46_critique_pack) # 과 실행이 일치한다 (GQ BLOCK-2 와 동축). 스키마는 본선과 # 같은 with_severity=True — v11 취합 스템이 severity· # observation_index 를 필수로 지시한다. from app.modules.pipeline.multiroll_gemini import ( make_gg46_critique_fn as _make_gg46_bgf, ) bgfirst_critique_fn = _make_gg46_bgf( critique_schema=build_critique_schema( with_severity=True, with_ref_gate=fix_ref_gate_on), project_config=project_config, step_tag="still_recipe_critique", ref_gate=fix_ref_gate_on, ) elif qk_judge_on: # BGFIRST 샷도 QK 수정 흐름을 타야 지문(qk_critique_pack)과 # 실행이 일치한다 (GQ BLOCK-2 와 동축). # ★2026-08-20: 본선 QK 와 같은 with_severity=True (위 변형 갈래 # 주석과 같은 이유 — 없으면 이 갈래만 심각도 게이트가 죽는다). bgfirst_critique_fn = make_qk_critique_fn( critique_schema=build_critique_schema( with_severity=True, with_ref_gate=fix_ref_gate_on), project_config=project_config, step_tag="still_recipe_critique", ref_gate=fix_ref_gate_on, ) elif gq_judge_on: # Codex 리뷰 BLOCK-2: BGFIRST 샷도 G+Q 수정 흐름을 타야 # 지문(gq_critique_pack)과 실행이 일치한다. bgfirst_critique_fn = make_gq_critique_fn( critique_schema=build_critique_schema( with_ref_gate=fix_ref_gate_on), project_config=project_config, step_tag="still_recipe_critique", ref_gate=fix_ref_gate_on, ) else: bgfirst_critique_fn = make_gemini_critique_fn( critique_sys=bgfirst_judge_texts["critique_sys"], critique_schema=build_critique_schema( with_ref_gate=fix_ref_gate_on), project_config=project_config, step_tag="still_recipe_critique", ref_gate=fix_ref_gate_on, ) # Step1 재투영 엔진 — gpt-image-2 (background_render_step 과 동일 # 클라이언트 패턴) _bgfirst_gpt_client = openai_client( timeout=float(settings.llm_timeout_image_gen), ) # ── 순차 생성 (스토리 순서 — prev 앵커 체인) ──────────────────── recipe_dir = scene_dir / "recipe" recipe_dir.mkdir(parents=True, exist_ok=True) records = _Records(recipe_dir / "records.json") def _run_bgfirst_bg( tag: str, still_id: str, conti_path: Path, plate_path: Optional[Path], bg_prompt: str, conti_asset_id: Optional[str], plate_asset_id_override: Optional[str] = None, seed_path: Optional[Path] = None, seed_asset_id: Optional[str] = None, authority_kind: str = "plate", # #119①: era 조사 참조 — 지문·생성 참조·record 에 함께 접힌다. # None = 기존 경로 byte-identical. era_ref_path: Optional[Path] = None, era_meta: Optional[Dict[str, Any]] = None, ) -> Tuple[Path, str]: """BGFIRST2 Step1 — 콘티 카메라 기준의 인물 0 빈 배경 (gpt-image-2). **참조 목록은 authority_kind 에 달렸다**: 위치 권위 사진을 쓰는 모드는 그 사진을 콘티 카메라로 재투영하므로 참조= [콘티, 플레이트] 순서 고정(프롬프트의 FIRST/SECOND 지칭과 동조), LANE_CONTI_ONLY 는 참조=[콘티] 1장뿐이다(아래 2026-07-26 항). 재개=records `{tag}::bgfirst_bg` 지문(프롬프트+참조 내용+엔진·팩· 계약) 일치 + **비어 있지 않은 파일**(Codex 리뷰 6) 시 재사용, mismatch=stale 아카이브 후 재생성(conti 관례). moderation= sanitizer 1회 재시도(effective_prompt 로 provenance 병록). 반환=(bg_path, bg_asset_id) — 등록(아래 _register)은 재사용 경로에서도 매번 idempotent upsert 라 file+row+fingerprint 가 함께 검증된다(Codex 리뷰 2). fix③④ full 확장: plate_path=위치 권위(플레이트/seed-bg/groupbg — plate_asset_id_override 로 UUID 직접 전달), seed_path=STRUCTURE LOOK 3번째 참조(complex 샷 — bg_prompt 에 seed 절은 호출자가 포함, input_ids 3 UUID). 신규 인자 default=기존 경로 지문·참조 byte-identical. 2026-07-26 확정 흐름: authority_kind=LANE_CONTI_ONLY 면 plate_path=None 이고 참조는 [콘티] 1장이다 — 외부 사진 없이 콘티 자체를 편집해 배경을 입힌다(장소 사실은 텍스트 권위). """ import uuid as _uuid from app.core.file_paths import to_relative_image_path from app.models.project import ImageAsset from app.modules.llm.gpt_image_primitive import call_gpt_image_bytes from app.modules.pipeline.multiroll_gemini import atomic_write_bytes from app.modules.pipeline.multiroll_select import ( compute_input_fingerprint, ) from app.modules.pipeline.shot_conti_light import ( archive_stale_output, ) from app.modules.pipeline.still_recipe import ( BGFIRST_BG_IMAGE_MODEL, BGFIRST_BG_SIZE, BGFIRST_CONTRACT_VERSION, register_bgfirst_bg_asset, ) from app.services.image_capture.annotate import ( annotate_generated_asset, ) out = recipe_dir / f"{tag}__bgfirst_bg.png" key = f"{tag}::bgfirst_bg" _fp_refs = [("conti", conti_path)] if plate_path is not None: _fp_refs.append(("plate", plate_path)) _fp_extra: Dict[str, Any] = { "model": BGFIRST_BG_IMAGE_MODEL, "size": BGFIRST_BG_SIZE, "pack": recipe_pack_version(_BGFIRST_PACK), # 리뷰 6: 체인 계약 전환도 Step1 산출 무효화 대상 "contract": BGFIRST_CONTRACT_VERSION, } if seed_path is not None: _fp_refs.append(("seed", seed_path)) if era_ref_path is not None: _fp_refs.append(("era", era_ref_path)) if authority_kind != "plate" or seed_path is not None: # full 확장 사용 시만 기여 — 기존(플레이트 2참조) 지문 불변 _fp_extra["authority"] = authority_kind fp = compute_input_fingerprint( prompt=bg_prompt, labeled_refs=_fp_refs, roll_count=1, critique_enabled=False, extra=_fp_extra, ) prev_rec = records.data.get(key) or {} effective_prompt = bg_prompt # 리뷰 6: exists 만으로 재사용 금지 — 0-byte/비파일 고착 차단 reuse = ( out.is_file() and out.stat().st_size > 0 and prev_rec.get("input_fingerprint") == fp ) if not reuse: if out.exists(): logger.warning( "still_recipe %s: bgfirst 배경 지문 불일치/무효 파일 — " "stale 아카이브 후 재생성", tag, ) archive_stale_output(out) current = bg_prompt for attempt in (1, 2): try: png = call_gpt_image_bytes( _bgfirst_gpt_client, mode="edit", prompt=current, ref_paths=( [conti_path] + ([plate_path] if plate_path is not None else []) + ([seed_path] if seed_path is not None else []) + ([era_ref_path] if era_ref_path is not None else []) ), call_kwargs={ "model": BGFIRST_BG_IMAGE_MODEL, "size": BGFIRST_BG_SIZE, "quality": "high", "n": 1, }, capture_role="still_recipe_bgfirst_bg", capture_metadata={ "operation_type": "still_recipe_bgfirst_bg", "still_id": still_id, "shot_tag": tag, }, ) if not png: raise RuntimeError( "empty image bytes (bgfirst step1)") atomic_write_bytes(out, png) effective_prompt = current break except Exception as exc: # noqa: BLE001 msg = str(exc).lower() moderated = ( "moderation" in msg or "safety" in msg or "content_policy" in msg or "rejected" in msg ) if attempt == 1 and moderated: sr = shared_sanitizer.sanitize( current, str(exc), [], attempt) sanitized = sr.get("sanitized_prompt") or "" if sanitized: logger.warning( "still_recipe %s: bgfirst 배경 moderation " "— sanitize 재시도", tag, ) current = sanitized continue raise else: effective_prompt = prev_rec.get("effective_prompt") or bg_prompt # ── intermediate ImageAsset 명시 upsert (Codex 리뷰 2 BLOCKING — # capture 는 scope 부재 no-op + UUID 경로라 lineage SOT 불가. # 재사용 경로에서도 매번 호출 = row 실재·lineage 최신 보증) ── # 재리뷰 1 (HIGH): plate UUID 미해결/조회 예외(None)=등록 전 # ValueError → 샷 fail-closed (불완전 lineage 를 성공 record 로 # 영속 금지 — warning-only fail-open 제거) from app.modules.pipeline.still_recipe import ( bgfirst_require_input_ids, ) # LANE_CONTI_ONLY(참조 0)는 플레이트 자체가 없다 — UUID 조회를 # 아예 타지 않고, 검증도 그 모드에서만 면제된다(아래 kind 전달). _plate_aid = None if plate_path is not None: _plate_aid = plate_asset_id_override or ( (map_conti.get(tag) or {}).get("asset_id") if (map_conti.get(tag) or {}).get("plate_path") == str(plate_path) else None ) or _plate_asset_id(plate_path) input_ids = bgfirst_require_input_ids( conti_asset_id=conti_asset_id, plate_asset_id=_plate_aid, plate_path=plate_path, seed_asset_id=seed_asset_id, seed_attached=seed_path is not None, authority_kind=authority_kind, ) def _find_by_rel(rel: str): return ( db.query(ImageAsset) .filter( ImageAsset.project_id == project_id, ImageAsset.file_path == rel, ) .first() ) def _new_asset(*, rel, asset_type, model, prompt): asset = ImageAsset( id=str(_uuid.uuid4()), project_id=project_id, asset_type=asset_type, entity_id=None, still_id=still_id, episode_id=episode_id, file_path=rel, prompt_used=prompt, generation_model=model, status="generated", created_at=_now(), ) db.add(asset) return asset def _annotate(asset, *, role, input_ids, meta): annotate_generated_asset( asset, pipeline_role=role, stage="still_recipe", input_image_ids=input_ids, pipeline_metadata={"shot_tag": tag, **meta}, ) bg_asset = register_bgfirst_bg_asset( bg_path=out, prompt=effective_prompt, input_ids=input_ids, rel_fn=to_relative_image_path, find_asset_by_rel=_find_by_rel, new_asset=_new_asset, annotate_fn=_annotate, ) db.flush() records.data[key] = { "input_fingerprint": fp, "prompt": bg_prompt, "effective_prompt": effective_prompt, "bg_path": str(out), "asset_id": bg_asset.id, "input_asset_ids": input_ids, # #119①: 조사가 성립했을 때만 키 — 기록 없는 유료 조사 금지. **({"era_research": { k: era_meta[k] for k in ( "subject", "queries", "picked_url", "sha256", "file") if k in era_meta}} if era_meta else {}), } records.save() return out, bg_asset.id def _run_groupbg( group_key: str, tag: str, still_id: str, conti_path: Path, conti_asset_id: Optional[str], place_text: str, time_of_day_en: str, location_detail_en: str = "", scene_evidence: Tuple[str, ...] = (), ) -> Tuple[Path, str]: """no_plate 그룹의 장소 단위 공유 배경 — 그룹당 1회 생성 (fix③④). 묶음=share_plan 그룹(LLM 의 '비슷한 배경 공유' 판단). 생성 근거= 이 그룹에서 스토리 순으로 **처음 도달한** bgfirst 샷(origin, record.origin_tag 로 고정 — 처리 루프가 스토리 순이라 결정론)의 **현재 상류 SOT** 콘티+장소. Codex 3·4차 리뷰 현행 계약: 모든 멤버가 canonical origin 기준 지문(프롬프트+origin 콘티 bytes)을 대조해 origin 입력 drift 를 감지하며(자기 콘티 지문 금지 — 멤버별 진동 차단), 재생성도 항상 canonical origin 입력으로 수행한다. share 그룹 재구성(group_sig 변경)·origin 해석 실패=fail-closed. 샷 특정 정보(SHOT TEXT/카메라)는 프롬프트에 불포함 — 샷별 카메라 는 Step1 재투영 담당. 등록=bgfirst_bg 와 동일 upsert (is_intermediate), role/asset_type=bgfirst_group_bg. """ import hashlib as _hashlib import uuid as _uuid from app.core.file_paths import to_relative_image_path from app.models.project import ImageAsset from app.modules.llm.gpt_image_primitive import call_gpt_image_bytes from app.modules.pipeline.multiroll_gemini import atomic_write_bytes from app.modules.pipeline.multiroll_select import ( compute_input_fingerprint, ) from app.modules.pipeline.shot_conti_light import ( archive_stale_output, ) from app.modules.pipeline.still_recipe import ( BGFIRST_BG_IMAGE_MODEL, BGFIRST_BG_SIZE, BGFIRST_FULL_CONTRACT_VERSION, BGFIRST_FULL_PROMPT_VERSION, build_groupbg_prompt, decide_groupbg_reuse, era_preserve_applies, groupbg_context_sig, groupbg_require_input_ids, register_bgfirst_bg_asset, resolve_groupbg_canonical_origin, validate_groupbg_conti_source, ) from app.services.image_capture.annotate import ( annotate_generated_asset, ) # 파일명 안전화 — group_key 는 LLM 저작 문자열(공백·기호 가능), # 해시 접미로 충돌 차단 (결정론) _safe = "".join( c if c.isalnum() or c in "-_" else "_" for c in group_key )[:48] _safe += "_" + _hashlib.sha256( group_key.encode("utf-8")).hexdigest()[:6] out = recipe_dir / f"groupbg_{_safe}.png" key = f"groupbg::{group_key}" prev_rec = records.data.get(key) or {} group_sig = { "key": group_key, "tags": sorted( t for t, g in group_of.items() if g == group_key ), } # Codex 3·4차 NARROW: canonical origin 입력 = **origin 의 현재 # 상류 SOT** (record.origin_tag 보존, follower 값 미사용). 지문 # 대조·프롬프트·재생성 전부 이 기준 — fresh run 과 partial resume # 의 산출 동일성 계약. group_sig 변경(그룹 재저작)·origin 해석 # 실패=ValueError fail-closed → 샷 실패 격리. canonical = resolve_groupbg_canonical_origin( prev_rec=prev_rec, tag=tag, current_inputs={ "place_text": place_text, "time_of_day_en": time_of_day_en, "conti_path": conti_path, "conti_asset_id": conti_asset_id, }, origin_lookup_fn=_groupbg_origin_now, current_group_sig=group_sig, ) origin_tag = canonical["origin_tag"] is_origin = origin_tag == tag place_text = canonical["place_text"] time_of_day_en = canonical["time_of_day_en"] conti_asset_id = canonical["conti_asset_id"] # 4차 NARROW-2: canonical 콘티 소스 검증(자기/타 origin 공통) — # 파일 실재·비어있지 않음·asset UUID 필수, 생성 전 fail-closed conti_path = validate_groupbg_conti_source( conti_path=canonical["conti_path"], conti_asset_id=conti_asset_id, origin_tag=origin_tag, ) bg_prompt = build_groupbg_prompt( place_text=place_text, time_of_day_en=time_of_day_en, world_anchor=world_anchor, # E2E11 ③: 장소 근거 강화 — loc 서술/traits + share 그룹 # 씬 원문 인용 (v11+ 조립, 빈 값=절 생략) location_detail_en=location_detail_en, scene_evidence=scene_evidence, ) # ── 시대 인지 사전 조사 (2026-08-14 사용자 확정 "무조건") ── # 1980년대 열차 실내류 실측: 생성 지식만으로는 시대 집기가 틀린다. # 서울역 재교체(regen_period_bg)에서 통한 [콘티+시대 참조] edit # 구조의 정식 편입 — 조사 성공 시 참조 bytes·역할문·meta 가 지문 # 에 접혀 재사용/재생성이 갈린다. 조사 실패=기존 경로 비차단. era_bg_meta: Optional[Dict[str, Any]] = None era_bg_path: Optional[Path] = None if era_research_on: from app.modules.pipeline.era_research import ( assess_and_research_cached as _era_cached_gb, build_ref_role as _era_role, ) # (2026-08-17 결함 수리, '마지막 임무' 실측) 직호출 판별+조사는 # 재방문마다 검색을 다시 돌려 회수 사진이 그때그때 다르고, 그 # sha 가 groupbg 지문(era_research_sha+참조 bytes)에 접혀 # **완성된 배경을 통째로 재생성**시켰다(06:04 S6sh3 국밥집 → # bgfirst 배경 → 하류 샷 연쇄 재구매). 샷별 판·confined 과 # 같은 records 사이드카 캐시로 교체 — 같은 대상은 최초 1회만 # 조사하고 이후 방문은 내용 주소화된 같은 참조 파일을 재사용해 # 지문이 결정론이 된다. 캐시 신원=대상·세계관·정책·팩 내용 # (진짜 계약 변경은 여전히 재생성 유발 — #77 유지). 판별 # "비대상"도 캐시, 조사 실패는 캐시 안 함+걷기 단위 sentinel # (완성될 때까지 다음 걷기가 재시도 — 비차단 계약 그대로). _era_gb_outcome: Dict[str, Any] = {} _era_gb = _era_cached_gb( step_tag="era_research_groupbg", subject_text="\n".join(filter(None, [ place_text, location_detail_en])), world_facts_block=world_anchor, out_dir=recipe_dir, cache_get=lambda k: ( records.data.get(k) if isinstance(records.data.get(k), dict) else None), cache_put=lambda k, v: ( records.data.__setitem__(k, v), records.save()), project_config=project_config, openai_client=_bgfirst_gpt_client, failed_memo=era_failed_memo, outcome=_era_gb_outcome, ) if _era_gb: era_bg_meta = _era_gb era_bg_path = Path(_era_gb["path"]) bg_prompt = bg_prompt + "\n\n" + _era_role( _era_gb["subject"]) _era_refs = ([("era", era_bg_path)] if era_bg_path else []) meta = { "model": BGFIRST_BG_IMAGE_MODEL, "size": BGFIRST_BG_SIZE, "pack": recipe_pack_version(BGFIRST_FULL_PROMPT_VERSION), "contract": BGFIRST_FULL_CONTRACT_VERSION, # Codex BLOCKING-1: share plan 재저작(그룹 구성 변화)이 sidecar # 자체에서도 무효화되도록 그룹 시그니처를 계약 meta 에 포함 # (canonical origin 해석의 group_sig fail-closed 와 동일 값) "group_sig": group_sig, # Codex NARROW-4: 장소 근거(그룹 안정 파생) 지문 — origin 이 # 아닌 멤버도 상류 근거 drift 를 meta 불일치로 감지·재생성 "context_sig": groupbg_context_sig( location_detail_en=location_detail_en, scene_evidence=scene_evidence, ), # 시대 조사 — 조사가 성립했을 때만 키가 생겨 지문이 갈린다 # (OFF/비대상=키 부재 byte-identical). sha 는 참조 bytes 의 # 신원 — 참조 교체=재생성. **({"era_research_sha": era_bg_meta["sha256"]} if era_bg_meta else {}), } file_ok = out.is_file() and out.stat().st_size > 0 # (Codex era R1 BLOCK-1) 조사 미성립(실패) 시 완성 배경 보존: # prev meta 에 era_research_sha 가 있는데 이번 방문 조사가 "실패" # (비대상 아님)면, meta 대조가 era 강등으로 읽혀 완성 배경을 era # 없이 유료 재생성하고 다음 걷기 성공 시 또 재생성한다(이중 지출). # 실패는 계약 변경이 아니다 — 기존 meta·지문을 그대로 신뢰하고 # 재사용한다. 진짜 계약 변경(팩·모델·group_sig 등)은 조사 성공 # 방문에서 정상 감지된다. _prev_meta = (prev_rec.get("meta") if isinstance(prev_rec.get("meta"), dict) else {}) _era_preserve = era_preserve_applies( era_on=era_research_on, era_meta=era_bg_meta, outcome=_era_gb_outcome if era_research_on else None, file_ok=file_ok, prev_meta=_prev_meta) if _era_preserve: logger.warning( "still_recipe %s: groupbg(%s) era 조사 미성립 — 기존 " "완성 배경 보존(재생성 억제)", tag, group_key) meta = dict(_prev_meta) regenerate, fp = False, prev_rec.get("input_fingerprint") # record 의 era_research 감사 블록도 이전 값 그대로 — # 빠뜨리면 record bytes 가 움직여 유령 지출로 읽히고 감사가 # 끊긴다. if isinstance(prev_rec.get("era_research"), dict): era_bg_meta = prev_rec["era_research"] # (era R2 BLOCK-1) 보존은 현재 계약 검증을 미룬 상태 — 걷기 # 끝에 스텝 미봉인으로 남겨 resume 이 조사를 재시도하게 한다. if group_key not in era_preserved_keys: era_preserved_keys.append(group_key) else: regenerate, fp = decide_groupbg_reuse( prev_rec=prev_rec, meta=meta, file_ok=file_ok, # canonical origin 기준 지문 — meta 일치 시 **모든 멤버**가 # 대조해 origin 의 현재 입력 drift 를 감지 (3차 NARROW) fingerprint_fn=lambda: compute_input_fingerprint( prompt=bg_prompt, labeled_refs=[("conti", conti_path), *_era_refs], roll_count=1, critique_enabled=False, extra=meta, ), ) # 감사용 canonical origin 입력 (복원 소스 아님 — canonical 은 # 항상 현재 상류 SOT 에서 재해석) origin_inputs = { "place_text": place_text, "time_of_day_en": time_of_day_en, "conti_asset_id": conti_asset_id, } if not regenerate: effective_prompt = prev_rec.get("effective_prompt") or bg_prompt stored_fp = prev_rec.get("input_fingerprint") stored_prompt = prev_rec.get("prompt") or bg_prompt # 4차 NARROW-2: reuse 도 lineage=**현재 canonical** 콘티 UUID # 로 정규화 — 동일 bytes 신규 UUID 교체 시 origin_inputs 와 # ImageAsset input edge 가 모순되던 감사 결손 봉합 lineage_conti_aid = conti_asset_id else: if fp is None: fp = compute_input_fingerprint( prompt=bg_prompt, labeled_refs=[("conti", conti_path), *_era_refs], roll_count=1, critique_enabled=False, extra=meta, ) if out.exists(): logger.warning( "still_recipe %s: groupbg(%s) 지문/계약 불일치 — " "stale 아카이브 후 재생성", tag, group_key, ) archive_stale_output(out) current = bg_prompt effective_prompt = bg_prompt for attempt in (1, 2): try: png = call_gpt_image_bytes( _bgfirst_gpt_client, mode="edit", prompt=current, ref_paths=[conti_path, *([era_bg_path] if era_bg_path else [])], call_kwargs={ "model": BGFIRST_BG_IMAGE_MODEL, "size": BGFIRST_BG_SIZE, "quality": "high", "n": 1, }, capture_role="still_recipe_groupbg", capture_metadata={ "operation_type": "still_recipe_groupbg", "still_id": still_id, # shot_tag=재생성 trigger 감사, origin_tag= # canonical 생성 기준 (Codex 3차 비차단 노트) "shot_tag": tag, "origin_tag": origin_tag, "group_key": group_key, }, ) if not png: raise RuntimeError("empty image bytes (groupbg)") atomic_write_bytes(out, png) effective_prompt = current break except Exception as exc: # noqa: BLE001 msg = str(exc).lower() moderated = ( "moderation" in msg or "safety" in msg or "content_policy" in msg or "rejected" in msg ) if attempt == 1 and moderated: sr = shared_sanitizer.sanitize( current, str(exc), [], attempt) sanitized = sr.get("sanitized_prompt") or "" if sanitized: logger.warning( "still_recipe %s: groupbg moderation — " "sanitize 재시도", tag, ) current = sanitized continue raise # 재생성 — canonical origin 기준 record 재작성 (origin_tag 는 # resolve_groupbg_canonical_origin 이 보존/확정) stored_fp = fp stored_prompt = bg_prompt lineage_conti_aid = conti_asset_id input_ids = groupbg_require_input_ids( conti_asset_id=lineage_conti_aid) def _find_by_rel(rel: str): return ( db.query(ImageAsset) .filter( ImageAsset.project_id == project_id, ImageAsset.file_path == rel, ) .first() ) def _new_asset(*, rel, asset_type, model, prompt): asset = ImageAsset( id=str(_uuid.uuid4()), project_id=project_id, asset_type=asset_type, entity_id=None, still_id=still_id, episode_id=episode_id, file_path=rel, prompt_used=prompt, generation_model=model, status="generated", created_at=_now(), ) db.add(asset) return asset def _annotate(asset, *, role, input_ids, meta): annotate_generated_asset( asset, pipeline_role=role, stage="still_recipe", input_image_ids=input_ids, pipeline_metadata={"group_key": group_key, **meta}, ) bg_asset = register_bgfirst_bg_asset( bg_path=out, prompt=effective_prompt, input_ids=input_ids, rel_fn=to_relative_image_path, find_asset_by_rel=_find_by_rel, new_asset=_new_asset, annotate_fn=_annotate, asset_type="bgfirst_group_bg", role="bgfirst_group_bg", chain="bgfirst_groupbg", ) db.flush() records.data[key] = { "input_fingerprint": stored_fp, "meta": meta, "prompt": stored_prompt, "effective_prompt": effective_prompt, "bg_path": str(out), "asset_id": bg_asset.id, "input_asset_ids": input_ids, "origin_tag": origin_tag, "place_text": place_text, # NARROW-1: canonical origin 입력 영속 — follower 가 drift # 재생성 시 origin 기준 프롬프트·콘티를 복원하는 SOT "origin_inputs": origin_inputs, # 시대 조사 감사 — 질의·나간 질의·선택 URL·sha (키 부재=미조사) **({"era_research": era_bg_meta} if era_bg_meta else {}), } records.save() return out, bg_asset.id # 플레이트(배경 렌더 산출) asset UUID 조회 — 상대경로 구조키, 캐시 plate_asset_cache: Dict[str, Optional[str]] = {} def _plate_asset_id(p: Path) -> Optional[str]: key = str(p) if key not in plate_asset_cache: try: from app.core.file_paths import to_relative_image_path from app.models.project import ImageAsset rel = to_relative_image_path(key) row = ( db.query(ImageAsset.id) .filter( ImageAsset.project_id == project_id, ImageAsset.file_path == rel, ) .first() ) plate_asset_cache[key] = row[0] if row else None except Exception: # noqa: BLE001 — 조회 실패=None 정규화. # legacy plate attach 경로에선 unresolved 진단(비치명), # bgfirst 경로에선 bgfirst_require_input_ids 가 None 을 # fail-closed 로 수렴시킨다 (Codex 최종 재리뷰 메모). plate_asset_cache[key] = None return plate_asset_cache[key] primary_asset_by_tag: Dict[str, str] = {} ordered = sorted( [s for s in stills if s.get("scene_index") is not None and s.get("shot_index") is not None], key=lambda s: (int(s["scene_index"]), int(s["shot_index"])), ) still_id_by_tag = { tag_of(int(s["scene_index"]), int(s["shot_index"])): s["id"] for s in ordered } # E2E11 ③ (Codex NARROW-4): groupbg 장소 근거 = **그룹 안정 파생** — # 멤버 샷들의 씬 → primary loc 상세를 그룹 단위로 합성(샷별 값 금지: # 멤버 간 지문 불일치로 공유 배경이 흔들리는 역결함 차단). 근거 # drift 는 context_sig(meta)로 origin 아닌 멤버도 감지. _si_by_tag = { tag_of(int(s["scene_index"]), int(s["shot_index"])): int(s["scene_index"]) for s in ordered } groupbg_context: Dict[str, Dict[str, Any]] = {} for _gkey in set(group_of.values()): if not _gkey: continue _scenes = sorted({ _si_by_tag[_t] for _t, _g in group_of.items() if _g == _gkey and _t in _si_by_tag }) _details = sorted({ location_detail_by_name.get(location_by_scene.get(_s, ""), "") for _s in _scenes } - {""}) groupbg_context[_gkey] = { "detail": " / ".join(_details), "evidence": tuple(group_evidence.get(_gkey, ())), } def _groupbg_origin_now(o_tag: str) -> Dict[str, Any]: """origin 샷의 **현재** 상류 SOT 입력 해석 (Codex 3차 NARROW). origin 이 already_done 으로 skip 되는 partial resume 에서도 follower 가 origin 의 현재 place/time/콘티(bytes)로 지문을 대조·재생성할 수 있게 한다. 해석 실패=ValueError fail-closed (원 origin 재실행 요구 — follower 값 대체 금지). """ o_si = _si_by_tag.get(o_tag) if o_si is None: raise ValueError( f"groupbg origin {o_tag} 스틸 미존재 — canonical 입력 " "해석 불가 (fail-closed)" ) o_entry = contis.get(o_tag) or {} o_cpath = o_entry.get("image_path") if not (o_cpath and Path(o_cpath).exists()): raise ValueError( f"groupbg origin {o_tag} 콘티 미해결 — canonical 입력 " "해석 불가 (fail-closed)" ) o_cls = classify_shots.get(o_tag) or {} return { "place_text": ( o_cls.get("place_en") or classify_scenes.get(str(o_si), {}).get("place_en") or location_by_scene.get(o_si, "") ), "time_of_day_en": ( classify_scenes.get(str(o_si), {}) .get("time_of_day_en") or "" ), "conti_path": Path(o_cpath), "conti_asset_id": o_entry.get("asset_id"), } # ── 표적 씬 슬라이스 scope (2026-07-23, Codex 설계 합의) ───────── # 위 컨텍스트(tag map/_si_by_tag/groupbg_context/group_sig/canonical # origin)는 전체 stills 파생 그대로(BLOCKING-1) — 여기서부터의 실행 # 루프만 effective allowlist 로 제한. 클로저=classify+share_plan 적용 # 후 effective prev 체인 재귀(BLOCKING-2 — bg_only 등 실행 분기보다 # 보수적 상한: 초과 포함=안전, 미포함=결손). audit=단일 SOT(HIGH-3, # 스텝 카운트·verify 가 동일 파일 소비). None=전체 byte-identical. execution_scope: Optional[set] = None if target_scenes: from app.modules.pipeline.scene_image_scope import ( build_effective_prev_map, build_scope_audit, closure_over_prev_chain, requested_still_ids, save_scope_audit, scope_audit_path, ) _requested_ids = requested_still_ids(ordered, target_scenes) _tag_by_id = {v: k for k, v in still_id_by_tag.items()} _requested_tags = [_tag_by_id[rid] for rid in _requested_ids] _prev_of = build_effective_prev_map( list(still_id_by_tag), classify_shots, share_plans) _eff_tags, _added_tags = closure_over_prev_chain( _requested_tags, _prev_of, set(still_id_by_tag)) execution_scope = {still_id_by_tag[t] for t in _eff_tags} save_scope_audit( scope_audit_path(projects_dir, project_id, episode_id), build_scope_audit( target_scenes=target_scenes, requested_ids=_requested_ids, dependency_added_ids=[ still_id_by_tag[t] for t in _added_tags], path_kind="recipe", ), ) logger.info( "still_recipe 표적 scope: scenes=%s requested=%d dep_added=%d " "effective=%d", list(target_scenes), len(_requested_ids), len(_added_tags), len(execution_scope), ) total = len(ordered) # 표적 모드 진행률 (Codex 재리뷰 NARROW-5): 분자·분모=effective∩미완료 # (전체 에피소드 index 는 표적 실행에서 큰 점프·과대 분모). scope # None=기존 표기 byte-identical. exec_total = total exec_seen = 0 if execution_scope is not None: exec_total = len([ s2 for s2 in ordered if s2["id"] in execution_scope and s2["id"] not in already_done_stills ]) done = 0 # #77-B (2026-08-09 합의): 완료 샷 JIT 검증 계수기 — 그대로 통과 / # 지출 있었으나 산출 동일 / 낡아서 재생성 / record 없어 검증 불가 # (예전 그대로 skip) / 검증 중 실패(스텝 completed 봉인 금지용). jit_fresh = 0 jit_regen_count = 0 jit_spent_same = 0 jit_no_record = 0 jit_failed_tags: List[str] = [] # #108: 이번 걷기에서 변환이 실패(원본 fallback 영속)한 샷 — 걷기 # 끝에 스텝 미봉인 raise 의 근거 (Codex R1 BLOCK-1). cine_failed_tags: List[str] = [] # 검열로 포기한 변환 — 실패와 구분해서 센다(스텝을 막지 않는다). cine_declined_tags: List[str] = [] # (era R2 BLOCK-1) era 조사 미성립으로 기존 배경을 보존한 그룹 — # 현재 계약 검증이 미뤄진 상태라 걷기 끝 미봉인 raise 의 근거. era_preserved_keys: List[str] = [] jit_snapshot_before = "" jit_regen_limit = max( 0, int(getattr(settings, "still_jit_regen_limit", 64))) # 공유 record(groupbg:: 등) 판별용 전체 샷 tag 집합 — 스냅샷이 이 # 샷 밖 공유 지출까지 본다(Codex 재리뷰 BLOCK). jit_known_tags = { tag_of(int(s2["scene_index"]), int(s2["shot_index"])) for s2 in ordered } jit_enabled = bool(getattr(settings, "still_jit_verify_enabled", True)) if jit_enabled and (recipe_dir / JIT_LATCH_FILENAME).exists(): # Codex BLOCK 3: 래치가 있으면 **지출 전에** 멈춘다 — 자동 # 재시도 루프(failed→resume 최대 3회 등)가 바퀴마다 상한만큼 # 다시 쓰는 것을 차단. 해제는 사람이 파일 삭제로. raise StillJitRegenLimitExceeded( f"JIT 재생성 상한 래치가 남아 있다: " f"{recipe_dir / JIT_LATCH_FILENAME} — 이전 걷기에서 완료 샷 " f"재생성이 상한에 닿았다. 원인 확인 후 래치 파일을 삭제하고 " f"resume (자동 재시도는 이 예외로 지출 없이 끝난다).") for i, s in enumerate(ordered): si, shi = int(s["scene_index"]), int(s["shot_index"]) tag = tag_of(si, shi) still_id = s["id"] if execution_scope is not None and still_id not in execution_scope: # 표적 밖 샷 = 실행 skip (클로저가 표적의 prev 체인을 이미 # scope 에 포함 — 표적 샷의 앵커가 여기서 잘리는 일 없음). # #77-B: JIT 게이트보다 먼저 — 표적 밖 완료 샷이 검증·제동 # 계산에 들어가면 안 된다. continue jit_verify = False if still_id in already_done_stills: # #77-B: 완료 skip 이 지문 비교보다 앞에 있어서, prev 앵커가 # 낡은 채 완성된 샷이 동결되고(2026-08-09 S95sh18·S96sh1) # 팩만 올린 재실행이 253장을 그대로 재사용했다(2026-08-07). # record 가 있는 완료 샷은 몸통에 들여보내 **기존 단계 지문 # 게이트**가 그대로 판정하게 한다 — 전부 일치하면 지출 0, # 산출 동일이면 아래 영속 직전에 아무것도 안 쓰고 넘어간다. # 방문 시점이라 prev 는 이번 걷기의 확정 bytes — 스토리 순서 # 걷기가 곧 의존 순서라 낡음의 연쇄가 한 바퀴에 다 잡힌다. if not getattr(settings, "still_jit_verify_enabled", True): # 되돌림 레버 off = 예전 그대로 완전 skip. continue if not isinstance(records.data.get(tag), dict): # record 가 아예 없는 완료 샷(구세대 산출·records.json # 손상 포함): 지문을 잴 근거가 없는데 재생성부터 하면 # 에피소드 전체 재지출로 번진다(과거 "앵커 소급 생성" # 결함의 재발 방향) — 예전대로 skip 하되 세어서 알린다. jit_no_record += 1 if jit_no_record == 1: logger.warning( "still_recipe %s: 완료 샷에 record 가 없어 검증 " "없이 skip — 같은 부류는 끝 요약에 센다", tag) continue if jit_regen_count >= jit_regen_limit: latch = _jit_write_latch( recipe_dir, tripped_at_tag=tag, fresh=jit_fresh, regen=jit_regen_count, spent_same=jit_spent_same, no_record=jit_no_record) raise StillJitRegenLimitExceeded( f"완료 샷 JIT 재생성이 상한({jit_regen_limit})에 " f"닿았다 — 다음 검증 대상 {tag} 앞에서 멈춘다. " f"완료 샷이 무더기로 어긋나는 것은 지문 체계가 통째로 " f"움직였다는 뜻이라 사람 확인이 필요하다. 래치 " f"{latch} 를 남겼다 — 원인 확인 후 삭제하고 resume " f"(그대로 {jit_fresh}·재생성 {jit_regen_count}" f"·record없음 {jit_no_record})") jit_verify = True # Codex BLOCK 2: 지출 탐지의 근거는 bytes 가 아니라 record — # 유료 재개 경로(소급 critique 등)는 전부 record 를 갱신한다. jit_snapshot_before = _jit_tag_snapshot( records, tag, jit_known_tags) cls = classify_shots.get(tag) or {} bg_only = not cls.get("person_visible", True) # 분류 팩 v4 (2026-08-07) — 구 체크포인트엔 없으므로 빈 문자열. handled_by = str(cls.get("handled_by") or "").strip() # ── lane 샷 판정 (2026-07-16 사용자 확정 재설계) — 실패/결손은 # fail-closed (조용한 degrade 금지). 레인1(map_marker)=스케치 # 경로 유지 / 복잡 구조물(structure_plate)=일반 경로 병합 + # A/B 상시(ab_select_ready) 또는 명시 bypass(bg_only/prev). lane_entry = lane_conti.get(tag) or {} lane_used = False complex_ab = False lane_policy = "" lane_sketch_path: Optional[Path] = None structure_seed_path: Optional[Path] = None structure_seed_asset_id: Optional[str] = None seed_bg_path: Optional[Path] = None seed_bg_asset_id: Optional[str] = None if lane_entry and lane_entry.get("lane") == "structure_plate": _status = lane_entry.get("status") or "" lane_policy = _status + ( f":{lane_entry.get('bypass_reason')}" if _status == "ab_select_bypass" else "" ) _bg_source = lane_entry.get("bg_source") or "plate" # N4 재리뷰: enum default-deny — 허용 밖 값은 fail-closed if _bg_source not in ("plate", "seed"): err = ( f"complex 샷 bg_source 허용 밖 값: {_bg_source!r} — " "fail-closed (shot_conti_light 재실행 필요)" ) logger.error("still_recipe %s: %s — fail-closed", tag, err) scene_cp.mark_failed(still_id, err) continue if _status in ("ab_select_ready", "ab_select_bypass"): _is_prev_bypass = ( lane_entry.get("bypass_reason") == "prev") if _bg_source == "seed" and not _is_prev_bypass: # seed-bg 승격 (Codex 조건 1·4): seed=LOCATION 단일 # 권위, 정확히 1회 부착 — STRUCTURE LOOK 별도 부착 # 금지. prev bypass 는 prev-only(미부착). _bgp = lane_entry.get("bg_path") or "" _seed_aid = lane_entry.get("seed_asset_id") if ( not _bgp or not Path(_bgp).is_file() or not _seed_aid ): err = ( f"seed-bg 결손/asset 부재: {_bgp!r} — " "LOCATION 권위 없이 진행 금지" ) logger.error( "still_recipe %s: %s — fail-closed", tag, err) scene_cp.mark_failed(still_id, err) continue # 조건 2: 이후 플레이트가 새로 생겼으면 seed→plate # 전환 재생성 필수 — stale seed-bg 진행 금지 if plate_map.get(f"{si}_{shi}") is not None: err = ( "seed-bg entry 인데 현재 플레이트 존재 — " "seed→plate 전환 재생성 필요 " "(shot_conti_light 재실행)" ) logger.error( "still_recipe %s: %s — fail-closed", tag, err) scene_cp.mark_failed(still_id, err) continue seed_bg_path = Path(_bgp) seed_bg_asset_id = _seed_aid elif _bg_source != "seed": # plate 케이스 — seed=구조물 look 앵커 (R6): A/B 양 # 브랜치·bg_only 공통 부착, prev bypass 미부착. _need_seed = ( _status == "ab_select_ready" or lane_entry.get("bypass_reason") == "bg_only" ) if _need_seed: _seed = lane_entry.get("seed_path") or "" _seed_aid = lane_entry.get("seed_asset_id") # 재리뷰 NARROW-3: 파일 실재만 (is_file) if ( not _seed or not Path(_seed).is_file() or not _seed_aid ): err = ( f"complex 샷 seed/asset 결손: {_seed!r} — " "구조물 look 앵커·lineage 없이 진행 금지" ) logger.error( "still_recipe %s: %s — fail-closed", tag, err) scene_cp.mark_failed(still_id, err) continue structure_seed_path = Path(_seed) structure_seed_asset_id = _seed_aid complex_ab = _status == "ab_select_ready" else: # failed/pending/미지 상태 = fail-closed (B 단독·일반 # 경로 조용한 하강 금지 — Codex R2) err = ( "complex 샷 lane 정책 실패/결손 — " + str(lane_entry.get("error") or f"status={_status}") ) logger.error("still_recipe %s: %s — fail-closed", tag, err) scene_cp.mark_failed(still_id, err) continue elif lane_entry: _sketch = lane_entry.get("image_path") or "" if ( lane_entry.get("status") != "ok" or not _sketch or not Path(_sketch).exists() ): if tag in _no_conti_exempt: # 운영자 선언 무콘티 예외 — lane 스케치가 끝내 실패한 # 샷(스텝 계수 연장과 짝, 2026-08-11). 스케치 없이 # 기존 무콘티 경로로 생성한다. 다른 샷의 fail-closed # 계약은 그대로. logger.warning( "still_recipe %s: lane conti 실패 — 운영자 무콘티 " "예외, 일반 경로로 생성", tag, ) else: err = ( "lane conti 실패/결손 — " + str(lane_entry.get("error") or f"sketch 부재: {_sketch}") ) logger.error( "still_recipe %s: %s — fail-closed", tag, err) scene_cp.mark_failed(still_id, err) continue else: lane_used = True lane_sketch_path = Path(_sketch) prev_tag = cls.get("prev") share_shot_plan = share_plans.get(tag) if share_shot_plan is not None: from app.modules.pipeline.still_recipe import ( apply_share_plan_prev, ) prev_tag = apply_share_plan_prev(prev_tag, share_shot_plan) # E2E9 육안 #2 (2026-07-19 사용자 확정): 계획의 prev 지휘는 bgonly # 계약("prev 무시·플레이트 강제")보다 상위 — 배경 전용 샷도 공유 # 그룹 후속이면 이전 샷 스틸이 배경 SOT (S12sh9=시체 샷 prev). share_prev_directed = ( isinstance(share_shot_plan, dict) and share_shot_plan.get("ref_plan") == "prev" ) prev_sel: Optional[Path] = None prev_primary_asset_id: Optional[str] = None if (not bg_only or share_prev_directed) and prev_tag: cand = recipe_dir / f"{prev_tag}_sel.png" if cand.exists(): prev_sel = cand # 이 run 의 레시피 선정·수정본 앵커 else: # prev 샷이 already_done 이면 기존 scene primary 가 앵커 # (DB SOT — asset UUID 도 lineage 에 직결) from app.core.file_paths import resolve_image_path from app.models.project import ImageAsset prev_sid = still_id_by_tag.get(prev_tag) row = ( db.query(ImageAsset) .filter( ImageAsset.project_id == project_id, ImageAsset.episode_id == episode_id, ImageAsset.still_id == prev_sid, ImageAsset.asset_type == "scene", ImageAsset.is_primary == 1, ) .first() ) if prev_sid else None p = resolve_image_path(row.file_path) if row else None if p is not None and p.exists(): prev_sel = p prev_primary_asset_id = row.id else: logger.warning( "still_recipe %s: prev %s 선정본·primary 부재 — " "플레이트만 참조", tag, prev_tag, ) # B-3 소비 시점 권위 (2026-07-19, E2E9 실측 S7sh5/S28sh9): 계획이 # prev 로 지휘한 complex 샷 = A/B 전제 결손이 아니라 계획된 prev # 파이프. finalize(conti 단계)는 classify prev 기준이라 계획과 # 시차가 생길 수 있다 — 계약("스틸 소비 시 계획이 상위 권위")대로 # 여기서 치유한다. seed/seed-bg 는 prev 샷 미부착 계약(R6·조건 5) # 에 맞춰 해제. prev 선정본 결손이면 바이패스하지 않는다(아래 # ab_active fail-closed 유지 — 권위 없는 조용한 하강 금지). if ( complex_ab and prev_sel is not None and share_prev_directed ): complex_ab = False lane_policy = ( (lane_policy or "ab_select_ready") + ":share_plan_prev_bypass" ) structure_seed_path = None structure_seed_asset_id = None seed_bg_path = None seed_bg_asset_id = None logger.info( "still_recipe %s: share_plan prev 지휘 — complex A/B 대신 " "prev 파이프 (anchor=%s)", tag, prev_tag, ) # E2E9 육안 #2: bgonly 샷의 계획 prev — prev 스틸이 배경 단일 # 권위(권위 이중화 금지): 플레이트/seed/seed-bg 전부 미부착, # refs=[prev]. prev 결손이면 기존 bgonly 경로 fail-safe. if bg_only and share_prev_directed and prev_sel is not None: lane_policy = ( (lane_policy + ":" if lane_policy else "") + "share_plan_prev_bgonly" ) structure_seed_path = None structure_seed_asset_id = None seed_bg_path = None seed_bg_asset_id = None logger.info( "still_recipe %s: share_plan prev 지휘(배경 전용) — " "플레이트 대신 prev 참조 (anchor=%s)", tag, prev_tag, ) # 2026-07-25 사용자 확정 (케이스1 스펙 E·F): lane 체인 편입 시 # **마커 스케치가 곧 콘티** — 배경 재투영(Step1)·인물 삽입 # (Step2)의 구도 SOT 가 된다. 편입 OFF = 기존과 동일(lane 샷은 # 플레이트·콘티 미사용, 스케치 단독 참조). # ★좁고 복잡한 구조물 내부 — 기하 권위 계약을 켠다(팩 v15). # lane 샷은 제외한다: 마커 스케치가 이미 배치 SOT 이고 그 경로는 # stage_head 대신 별도 계약을 쓴다(두 권위를 겹치지 않는다). # ★lane 판정이 **끝난 뒤**여야 한다 — 위에서 계산하면 lane_used 가 # 아직 없어 첫 샷에서 UnboundLocalError 로 스텝이 죽는다(Codex # 2차 리뷰 실측). 순서가 계약인 자리다. # ── confined fp 적용 판별 — legacy 게이트(bgfirst 판정·complex # A/B·lineage 수집)보다 **앞** (Codex BLOCK-3: 뒤에 두면 confined # 로 처리할 샷이 legacy 결손 fail-closed 에서 먼저 죽고, bgfirst_ # used=True 인 채 fp 멀티롤을 만든 record 가 bgfirst 후처리에서 # KeyError 로 실패한다). 판별 캐시는 입력 지문(apt_fingerprint — # 브리프+계약+스키마+물리 모델) 대조 — 존재 캐시는 새 설정을 옛 # 판별로 봉인한다(BLOCK-2, #77 부류). _confined_active = False if (confined_fp_on and bool(cls.get("confined_structure")) and not lane_used): from app.modules.pipeline.confined_fp import ( apt_fingerprint, judge_fp_applicability, ) # 판별 입력 = 조립 전 원천 재료 — 최종 브리프(prompt)는 이 # 지점(게이트 앞)에서 아직 조립 전이다. ★carried(인접 샷 # 유지 상태 — 누가 어느 자리에 있는가의 원천)를 반드시 포함 # (Codex 재리뷰 HIGH-3: 좌석·탑승 관계가 carried 에만 있는 # 샷이 false 로 봉인돼 D2 를 영구 우회하던 구멍). _apt_fix = (continuity.get("pose_fix") or {}).get(tag) or {} _apt_carried = str( _apt_fix.get("carried_en") or ((continuity.get("carried") or {}).get(tag) or {}) .get("carried_en") or "") _apt_brief = "\n\n".join(x for x in ( "SHOT TEXT: " + str( shot_desc_by_id.get(still_id) or s.get("still_frame_prompt") or ""), "CARRIED STATE: " + _apt_carried, "LOCATION: " + str( cls.get("place_en") or classify_scenes.get(str(si), {}).get("place_en") or location_by_scene.get(si, "")), ) if x.split(": ", 1)[-1].strip()) _apt_key = f"{tag}::confined_fp_apt" _apt_fp = apt_fingerprint(_apt_brief) _apt = records.data.get(_apt_key) if (not isinstance(_apt, dict) or "applies" not in _apt or _apt.get("input_fingerprint") != _apt_fp): _apt = judge_fp_applicability( tag, _apt_brief, project_config=project_config) _apt["input_fingerprint"] = _apt_fp records.data[_apt_key] = _apt records.save() _confined_active = bool(_apt.get("applies")) if not _confined_active: logger.info( "still_recipe %s: confined fp 판별 제외 — %s", tag, str(_apt.get("reason_ko"))[:80]) # confined fp 활성 샷은 geom_authority 계약을 겹치지 않는다 — # fp+장면 설명이 그 자리의 권위다(두 기하 권위 중복 금지). geom_authority = (bool(cls.get("confined_structure")) and not lane_used and not _confined_active) lane_chain = lane_prev_chain_on and lane_used # 2026-07-26 사용자 확정(사전 배경 플레이트 금지): lane 샷은 미리 # 구운 빈 배경을 **해석 자체를 하지 않는다**. 미리 만든 플레이트는 # 마네킹 콘티 구도와 어긋나 구조물이 겹쳐 보였고(실측), 배경은 # 콘티 자체를 i2i 로 편집해 입힌다 — 장소 사실은 텍스트 권위. # 아래 LANE_CONTI_ONLY 진입 assert 는 이 줄이 지켜지는지 감시하는 # 이중 방어이지 공급원이 아니다: 체인 ON 이면 조회를 되살리던 # 직전 판(2026-07-25)은 그 assert 를 스스로 때려 실측 lane 샷 # 14/57(9 에피소드 중 7)을 전량 죽였다. # # 예외=bg_only lane 샷 — 체인 비대상이고(bgfirst_eligible_full 이 # bg_only 를 제외), 체인 ON 에서는 스케치가 conti 슬롯으로 옮겨가 # build_still_refs 의 bg_only 조기 return 에 걸린다. 플레이트가 # 유일한 참조라 여기서 끊으면 참조 0(text-only)이 된다. plate = ( None if (lane_used and not (lane_chain and bg_only)) else plate_map.get(f"{si}_{shi}") ) conti_entry = ( lane_entry if lane_chain else ({} if lane_used else (contis.get(tag) or {})) ) conti_path = conti_entry.get("image_path") conti = ( Path(conti_path) if conti_path and Path(conti_path).exists() else None ) # R1 (Codex 설계 리뷰 BLOCKING-1): 콘티형 샷의 plate_select 는 # shot_conti_light 가 콘티 생성 **전** 판정·영속(plate_authority) — # 스틸은 그 기록을 재판정 없이 소비해 콘티·A/B 양 브랜치·스틸이 # 정확히 같은 플레이트를 쓴다. 스틸 시점 판정은 bg_only(콘티 # 없음) 샷만 잔존. ★명시 tradeoff (Codex 배치 리뷰 NARROW): # bg_only 판정은 생성 시점 수행·record 는 multiroll 완료 후 저장 — # 판정 후 crash 시 재판정(비용 1콜)될 수 있음. plate_select_rec: Optional[Dict[str, Any]] = None from app.modules.pipeline.still_recipe import plate_flow_mode if _confined_active: # confined fp 샷은 plate/seed 권위 해석 비대상 — 유료 plate # 선택(bg_late_select)·reconcile fail-closed 전부 건너뛴다 # (Codex 재리뷰 BLOCK-1: 여기서 못 막으면 confined branch # 도달 전에 안 쓸 plate 에 돈을 쓰거나 결손으로 죽는다). _flow = "none" elif seed_bg_path is not None: # seed-bg: LOCATION 권위=seed 고정 — plate 권위 해석 비대상. # 콘티가 정확히 이 seed 를 참조했는지만 검증 (조건 1). if ( conti is not None and str(conti_entry.get("plate_path")) != str(seed_bg_path) ): err = ( "콘티 배경 참조가 seed-bg 와 불일치 " f"(conti={conti_entry.get('plate_path')!r}) — " "shot_conti_light 재실행 필요" ) logger.error( "still_recipe %s: %s — fail-closed", tag, err) scene_cp.mark_failed(still_id, err) continue _flow = "none" else: _flow = plate_flow_mode( plate_select_on=plate_select_on, complex_ab=complex_ab, bg_only=bg_only, prev_used=prev_sel is not None, lane_used=lane_used, is_map_plate=f"{si}_{shi}" in map_plate_keys, plate_present=plate is not None, ) if _flow == "bg_late_select": # plate_present 는 plate_flow_mode 가 보장 — 즉석 판정 본문 if plate is not None: _assigned = plate_assign_by_key.get(f"{si}_{shi}") or "" from app.modules.pipeline.plate_select import ( location_of_bg_id, select_plate_for_shot, ) _loc = location_of_bg_id(_assigned) _cands = plate_cands_by_loc.get(_loc or "", []) if _assigned and len(_cands) > 1: from app.modules.llm.llm_client import call_structured _chosen_id, _chosen_path, plate_select_rec = ( select_plate_for_shot( shot_desc=shot_desc_by_id.get(still_id) or s.get("still_frame_prompt") or "", place_text=( # fix2 v3: 샷별 서브공간 우선 — 플레이트 # 선택도 샷의 실내외 서브공간 기준 cls.get("place_en") or classify_scenes.get(str(si), {}) .get("place_en") or location_by_scene.get(si, "") ), assigned_bg_id=_assigned, candidates=_cands, call_structured_fn=call_structured, project_config=project_config, ) ) if _chosen_id != _assigned: logger.info( "still_recipe %s: plate_select %s→%s (%s)", tag, _assigned, _chosen_id, (plate_select_rec or {}).get("reason_ko", ""), ) plate = _chosen_path elif _flow == "reconcile": # 콘티형 샷 plate reconciliation (R1 배선 완성 + 재리뷰 # HIGH-2): flag ON 은 plate mapping 이 비어 있어도 권위/콘티 # 플레이트로 해석, complex_ab 는 flag 무관 상시 — 콘티가 # 그려진 플레이트=SOT(결손·불일치=fail-closed). flag OFF # 일반 샷은 plate_flow_mode 가 "none" (legacy 유지). from app.modules.pipeline.still_recipe import ( resolve_conti_plate_authority, ) try: plate, plate_select_rec = resolve_conti_plate_authority( authority_entry=( plate_authority.get(tag) if plate_select_on else None ), conti_plate_path=conti_entry.get("plate_path"), current_plate=plate, ) except ValueError as _pa_exc: logger.error( "still_recipe %s: %s — fail-closed", tag, _pa_exc) scene_cp.mark_failed(still_id, str(_pa_exc)) continue # 배치 리뷰 BLOCKING-2/HIGH-3 + 재리뷰 NARROW-3: 복잡 구조물 공통 # 가드 — ①plate 필수·**파일 실재**(A/B·bg_only 전부 — seed-only # 조립·디렉토리/소실 파일 통과 금지) ②prev bypass 는 prev # 선정본/primary 가 유일한 구조물 권위 — 결손 시 seed 도 A/B 도 # 없는 일반 plate 경로로 조용한 하강 금지 if structure_seed_path is not None and not _confined_active and ( plate is None or not plate.is_file() ): err = ( f"복잡 구조물 샷 LOCATION 플레이트 결손/비파일({plate}) — " "seed-only 금지" ) logger.error("still_recipe %s: %s — fail-closed", tag, err) scene_cp.mark_failed(still_id, err) continue if lane_policy == "ab_select_bypass:prev" and prev_sel is None \ and not _confined_active: err = ( "complex prev bypass 인데 prev 선정본/primary 결손 — " "구조물 권위 없는 일반 경로 하강 금지" ) logger.error("still_recipe %s: %s — fail-closed", tag, err) scene_cp.mark_failed(still_id, err) continue # E2E6 피드백 ④: prev 참조 샷에서 '움직일 수 없는 인물'(pose_canon # =LLM 가동성 판정 SOT)의 별도 캐릭터 참조 제외 — 계약·조건은 # locked_pose_short_ids docstring 참조. from app.modules.pipeline.still_recipe import locked_pose_short_ids locked_pose_sids = ( locked_pose_short_ids( continuity.get("pose_canon") or [], tag, prev_tag) if prev_sel is not None else set() ) # ── VE → 캐릭터/소품 참조 (Codex BLOCKING-2: composite/state 우선 # + 실첨부 asset UUID 수집) ────────────────────────────────── # 두 번째 값 = 샷 확정 배정인가. False 면 씬 단위 추측이므로 # PEOPLE 절에 배타 조항을 붙이지 않는다(still_recipe 팩 v14). ve_ids, ve_exact = ve_ids_for_shot_ex(ve_by_key, (si, shi)) # 의상 SOT = outfit_assignments (2차 B1), VE outlook 은 보조. # 3차 M3: 손상/미해결 SOT 도 fail-closed (base 조용한 degrade 금지) outfit_map = outfit_by_still_id.get(still_id) or {} _outfit_err = ( f"outfit_assignments conflict: {outfit_map['__conflict__']}" if "__conflict__" in outfit_map else f"outfit_assignments invalid: {outfit_map['__invalid__']}" if "__invalid__" in outfit_map else None ) if _outfit_err: logger.error("still_recipe %s: %s — fail-closed", tag, _outfit_err) scene_cp.mark_failed(still_id, _outfit_err) continue outlook_by_eid = { **{ k: (_norm_uuid(v) or v) for k, v in ve_outlook_by_key.get((si, shi), {}).items() }, **outfit_map, } ve_detail = reference_svc.get_visible_entities( s.get("visible_entities_json") ) state_sids = reference_svc.detect_state_variant_sids( ve_detail, entity_lookup, scene_ref_image_map, staging_map.get(f"{si}_{shi}") if staging_map else None, ) sid_by_eid = { v.get("id"): v.get("short_id") for v in ve_detail if isinstance(v, dict) } attached_refs: List[Dict[str, Any]] = [] unresolved_refs: List[Dict[str, Any]] = [] def _attach(role: str, label: str, asset_id: Optional[str], **extra) -> None: if asset_id: attached_refs.append( {"asset_id": asset_id, "role": role, "label": label, **extra} ) else: unresolved_refs.append( {"role": role, "label": label, **extra} ) char_refs: List[Tuple[str, Any]] = [] prop_refs: List[Tuple[str, Any]] = [] char_names: List[str] = [] locked_excluded: List[str] = [] # E2E6 ④ 감사 기록용 for eid in ve_ids: e = entity_lookup.get(eid) if not e: continue name = e.get("name") or eid etype = e.get("entity_type") or "" if etype == "character": traits = traits_by_id.get(eid) char_names.append(f"{name} ({traits})" if traits else name) _ve_sid = sid_by_eid.get(eid) or e.get("short_id") or "" if _ve_sid and _ve_sid in locked_pose_sids: # 정본 자세 인물: prev 스틸=자세·외형 SOT — 별도 # 캐릭터 참조 미첨부 (PEOPLE 절 이름·traits 는 유지) locked_excluded.append(f"{name}({_ve_sid})") logger.info( "still_recipe %s: %s(%s) pose-locked by prev %s — " "캐릭터 참조 제외", tag, name, _ve_sid, prev_tag, ) continue from app.modules.pipeline.still_recipe import ( resolve_char_ref_key, ) key = resolve_char_ref_key( eid=eid, sid=sid_by_eid.get(eid) or e.get("short_id") or "", state_sids=state_sids, outlook_by_eid=outlook_by_eid, scene_ref_image_map=scene_ref_image_map, ) if key: char_refs.append((name, scene_ref_image_map[key])) _attach( "character_ref", name, scene_ref_asset_id_map.get(key), ref_key=key, ) elif cast_lock_on: # #119③: 시트(참조) 없는 인물 — 의상을 아웃룩 서술로 # 텍스트 잠금. 배정 우선, 없으면 그 인물의 유일 # 아웃룩만(여럿이면 지어내지 않는다). _aol = outlook_by_eid.get(eid) _cands = ([_aol] if _aol else outlooks_by_char.get(eid) or []) _wdesc = (outlook_desc_by_id.get(_cands[0]) if len(_cands) == 1 else None) if _wdesc: char_names[-1] = ( f"{char_names[-1]} — wearing: {_wdesc}") logger.info( "still_recipe %s: %s 시트·배정 없음 — 아웃룩 " "서술로 의상 잠금", tag, name) else: logger.info( "still_recipe %s: %s 시트 없음·아웃룩 후보 " "%d개 — 의상 공급 없음", tag, name, len(_cands)) elif etype == "prop": if eid in scene_ref_image_map: prop_refs.append((name, scene_ref_image_map[eid])) _attach( "prop_ref", name, scene_ref_asset_id_map.get(eid), ref_key=eid, ) # ── BGFIRST2 대상 판정 (일반 콘티 샷만 — 순수 로직) ────────── # plate/conti lineage 는 승자 확정 후 부착해야 하므로 여기서 먼저 # 판정한다 (체인 승=콘티+재투영 배경, 무콘티 승=플레이트 — 거짓 # edge 방지). Codex 리뷰 1 (BLOCKING): 정책 대상 여부는 콘티 산출 # 존재와 분리 — 대상인데 콘티 실패/파일·asset 결손이면 legacy # 무콘티 생성으로 조용히 하강하지 않고 샷 단위 fail-closed. from app.modules.pipeline.still_recipe import ( bgfirst_conti_defect, bgfirst_eligible, bgfirst_eligible_full, bgfirst_structural_skip, ) bgfirst_used = False # conti_present=콘티 스텝의 구조적 대상 선언(skipped_reason 3종 제외) # — 산출 존재(파일·asset)와는 여전히 분리, 결손은 아래 defect 로 # fail-closed. E2E10 결함 1호: no_plate/prev 앵커 부재 샷이 정책 # 대상으로 오분류돼 15샷 연쇄 fail-closed → 기존 경로 유지로 교정. _conti_structural_skip = bgfirst_structural_skip(conti_entry) if ( bgfirst_on and bgfirst_full_on and _conti_structural_skip and not _confined_active and (conti_entry or {}).get("skipped_reason") == "no_plate" ): # full 모드 전제=콘티 팩 v3(no_plate 도 저작) — no_plate skip # entry 는 stale CP 모순, legacy 무음 하강 금지 err = ( "BGFIRST full 인데 콘티가 no_plate skip — shot_conti_light " "CP stale (fail-closed)" ) logger.error("still_recipe %s: %s", tag, err) scene_cp.mark_failed(still_id, err) continue # 운영자 선언 무콘티 예외 (2026-07-30): 이 샷은 콘티가 끝내 실패해 # 운영자가 정책 대상 밖으로 선언했다 — bgfirst 판정 자체를 타지 않고 # 기존 무콘티 경로로 간다. 선언은 샷 태그 단위라 다른 샷의 fail-closed # 계약에는 영향이 없다. _op_no_conti = tag in _no_conti_exempt if _op_no_conti: logger.warning( "still_recipe %s: 운영자 선언 무콘티 예외 — BGFIRST 대상 " "제외, 기존 경로로 생성", tag, ) # fix④ full: 콘티 실재 전 샷 체인(complex/seed/no_plate 포함) / # 기존 v2 계약: 일반 콘티 샷만 # confined fp 활성 샷은 bgfirst 대상이 아니다 (Codex BLOCK-3 — # 여기서 못 막으면 bgfirst_used=True 인 채 fp 멀티롤을 만든 record # 가 bgfirst 후처리(record["bgfirst"] 읽기)에서 KeyError 로 죽는다). if bgfirst_on and not _op_no_conti and not _confined_active and ( bgfirst_eligible_full( conti_present=not _conti_structural_skip, bg_only=bg_only, prev_used=prev_sel is not None, lane_used=lane_used, chain_lane_prev=lane_prev_chain_on, ) if bgfirst_full_on else bgfirst_eligible( conti_present=not _conti_structural_skip, bg_only=bg_only, prev_used=prev_sel is not None, lane_used=lane_used, complex_ab=complex_ab, structure_seed_attached=structure_seed_path is not None, seed_bg_attached=seed_bg_path is not None, ) ): _conti_defect = bgfirst_conti_defect(conti_entry, conti) if _conti_defect: err = ( "BGFIRST2 대상 샷 콘티 결손 — legacy 하강 금지: " + _conti_defect ) logger.error("still_recipe %s: %s — fail-closed", tag, err) scene_cp.mark_failed(still_id, err) continue bgfirst_used = True # 플레이트/콘티/prev/lane/seed 실첨부 lineage (BLOCKING-2) # 체인 편입 lane 샷은 스케치가 conti 슬롯 — 승자 확정 후 부착 # (탈락 시 거짓 edge 방지, conti 관례와 동일) if lane_used and lane_sketch_path is not None and not lane_chain: _attach( "lane_storyboard_sketch", "STORYBOARD SKETCH", lane_entry.get("asset_id"), file_path=str(lane_sketch_path), ) # 복잡 구조물 STRUCTURE LOOK — A/B 양 브랜치 공통 참조라 lineage # 는 항상 기록 (R6, asset_id 는 위에서 fail-closed 검증됨). # E2E10 Codex HIGH-3: bgfirst 샷은 체인 승 시 seed 가 final 에 # 직접 첨부되지 않음(Step1 중간 bg 의 input edge 소유) — 승자 # 확정 후 부착으로 유예 (false direct edge 방지, plate 관례 동일). if structure_seed_path is not None and not bgfirst_used \ and not _confined_active: _attach( "structure_seed_look", "STRUCTURE LOOK", structure_seed_asset_id, file_path=str(structure_seed_path), ) # seed-bg 단일 권위 — role 을 plate/structure-look 과 구분해 # 정확히 1회 기록 (Codex 조건 5). bgfirst 샷은 승자 확정 후(HIGH-3). if seed_bg_path is not None and (bg_only or prev_sel is None) \ and not bgfirst_used and not _confined_active: _attach( "location_seed_bg", "LOCATION STRUCTURE PHOTOGRAPH", seed_bg_asset_id, file_path=str(seed_bg_path), ) # E2E6 ⑦: prev_used 샷은 플레이트 미첨부 — lineage 도 실첨부와 동기 # (build_still_refs 와 동일 조건). 2026-07-19: bgonly 도 계획 prev # 지휘 시 prev 가 배경 SOT — prev_sel 실재가 단일 판정 기준. # BGFIRST2 샷은 승자 확정 후 부착(무콘티 승만 플레이트 직접 참조). def _attach_plate() -> None: _map_entry = map_conti.get(tag) or {} _plate_aid = ( _map_entry.get("asset_id") if _map_entry.get("plate_path") == str(plate) else None ) or _plate_asset_id(plate) _attach( "location_plate", "LOCATION PHOTOGRAPH", _plate_aid, file_path=str(plate), ) if plate is not None and prev_sel is None and not bgfirst_used \ and not _confined_active: _attach_plate() # 체인 편입 prev 샷은 prev 스틸이 Step1 입력(재투영 소스)이라 # final 직접 첨부가 아니다 — 승자 확정 후 부착(plate 관례 동일) if prev_sel is not None and not bgfirst_used \ and not _confined_active: _prev_aid = ( primary_asset_by_tag.get(prev_tag or "") or prev_primary_asset_id ) if _prev_aid: _attach("prev_still", f"prev={prev_tag}", _prev_aid) else: # 구조키 post-hoc resolve (_resolve_prev_frame_asset_ids) unresolved_refs.append({ "pipeline_role": "scene_prev_frame", "role": "prev_still", "label": f"prev={prev_tag}", "source_still_id": still_id_by_tag.get(prev_tag or ""), }) # E2E6 ⑧: A/B 대상 샷은 콘티 lineage 를 승자 확정 후 부착(아래) — # 탈락 시 거짓 conti edge 방지. # 2026-07-16 사용자 확정: 복잡 구조물(ab_select_ready) 샷은 # still_conti_ab_enabled 와 무관하게 A/B **상시** — A/B 가 이 # 샷들의 파이프 자체 (opt-in 아님). ab_active = ( (conti_ab_on or complex_ab) and not lane_used and not bg_only and prev_sel is None and conti is not None # BGFIRST2 가 4택1/A-B 를 대체 (eligible=일반 콘티 샷 한정 — # complex_ab 샷은 eligibility 에서 제외라 R2 fail-closed 불변) and not bgfirst_used # confined fp 가 A/B·2택1 을 대체 (Codex BLOCK-3) and not _confined_active ) if complex_ab and not ab_active and not bgfirst_used \ and not _confined_active: # ready 판정인데 A/B 전제 결손(콘티 소실·분류 불일치 등) = # fail-closed — B 단독 조용한 진행 금지 (Codex R2). # full 모드에선 bgfirst 2택1 이 complex A/B 를 대체(fix④). err = ( "complex A/B 전제 결손 — conti=" + str(conti) + f", bg_only={bg_only}, prev={prev_tag!r}" ) logger.error("still_recipe %s: %s — fail-closed", tag, err) scene_cp.mark_failed(still_id, err) continue if not ab_active and not bg_only and prev_sel is None \ and conti is not None and not bgfirst_used \ and not _confined_active: _attach( "conti_light", "LAYOUT SKETCH", conti_entry.get("asset_id"), file_path=str(conti), ) # #119③: prev 스틸 인물 계승 스코프 — 앵커 샷과 이 샷의 인물 # 겹침을 코드가 계산한다(VE 확정 데이터, LLM 불요). 겹친 인물만 # 의상 잠금, 겹침 0 이면 계승 금지 명문 (S64sh4 실측: 앵커 인물 # 외모·의상이 다른 인물에게 복제). OFF/앵커 불명 = 빈 문자열 = # 레거시 스템 byte-identical. prev_people_rule = "" if cast_lock_on and prev_sel is not None and prev_tag: import re as _re _pm = _re.match(r"S(\d+)sh(\d+)$", str(prev_tag)) if _pm: from app.modules.pipeline.still_recipe import ( build_prev_people_rule, ) _a_ids, _a_exact = ve_ids_for_shot_ex( ve_by_key, (int(_pm.group(1)), int(_pm.group(2)))) # Codex BLOCK-4: 씬 union 추측(exact=False)을 확정 인물로 # 선언하면 없는 인물까지 shared 로 잠근다 — 양쪽 다 샷 # 확정일 때만 known, 교집합은 ID 로 내고 표시명 렌더. if _a_exact and ve_exact: def _char_ids_of(_ids: Any) -> set: return { _i for _i in _ids if (entity_lookup.get(_i) or {}).get( "entity_type") == "character" } _shared_ids = _char_ids_of(_a_ids) & _char_ids_of( ve_ids) prev_people_rule = build_prev_people_rule( sorted( (entity_lookup.get(_i) or {}).get("name") or _i for _i in _shared_ids), cast_known=True, ) fix = (continuity.get("pose_fix") or {}).get(tag) or {} carried = ( fix.get("carried_en") or ((continuity.get("carried") or {}).get(tag) or {}) .get("carried_en") or "" ) # 샷별 팩 selector — lane(레인1)=v3 / 복잡 구조물(seed 부착)=v4 / # 일반=v1 (byte-identical 보존) _pack = ( _LANE_PACK if lane_used else (_SEED_BG_PACK if seed_bg_path is not None else (_COMPLEX_PACK if structure_seed_path is not None else "1")) ) # grok 백엔드(2026-08-13 사용자 확정): xAI 8,000바이트 텍스트 상한 # 대응 — 지도 절을 컴팩트 팩(v17)으로 로드하고 시네마틱 마감 절을 # 동반한다. nb2(기본)는 guidance 빈 문자열 = 조립 byte-identical. # 2026-08-18: 이 값이 컴팩트 스템 갈림의 단일 신호이기도 하다 # (still_recipe.grok_stem_version) — camera_frame 절이 이 블록보다 # 먼저 렌더되므로 여기로 끌어올렸다. _grok_backend = ( getattr(settings, "still_image_backend", "nb2") == "grok2" ) _guidance = "" if _grok_backend: from app.modules.pipeline.still_recipe import ( STILL_COMPACT_PROMPT_VERSION as _guidance, # noqa: N813 ) # fix1 (2026-07-19): staging 구도·스케일 계약 렌더 — lane 샷 제외 # (마커 스케치=배치 SOT, build_still_prompt 상호 배타 가드와 동조) camera_frame_en = "" if camera_frame_on and not lane_used: from app.modules.pipeline.still_recipe import ( build_camera_frame_clause, ) camera_frame_en = build_camera_frame_clause( staging_map.get(f"{si}_{shi}") if staging_map else None, version_selector=_guidance, ) # fix⑤: 조명·무드 절 — camera_frame 과 달리 lane 샷도 포함(조명은 # 마커 스케치 배치 권위와 충돌하지 않음), bgonly 샷 포함(붉은 원 # 평면 낙서화 실측의 주 대상) lighting_mood_en = "" if lighting_on: from app.modules.pipeline.still_recipe import ( build_lighting_mood_clause, ) lighting_mood_en = build_lighting_mood_clause( staging_map.get(f"{si}_{shi}") if staging_map else None ) # 2026-08-12 차렷/증명사진 대응: identity 참조 역할 한정 절 — # 조건은 build_still_refs 가 **본문 캐릭터 참조(char_label)** 를 # 싣는 조건(char_refs 비어 있지 않음 AND not bg_only)과 같다 # (참조 없는 샷에 절이 나가면 거짓 문장). ★bg_only+handled_by 의 # HAND OWNER REFERENCE 는 의도된 제외 — 절의 지시 대상이 그 샷에 # 없고 손 라벨이 자체 역할 한정을 내장한다(Codex HIGH-4 명문화, # 정합은 unit 테스트가 잠근다). # #119②: 장면이 부르는 실물 표기의 원어 문안 저작 공급 — 판별+ # 저작 1콜(flash), records 사이드카 캐시(지문=입력+정책+팩). # 실패=비차단·비캐시(다음 방문 재시도), 빈 목록=캐시(재지출 0). signage_en = "" if signage_on: import hashlib as _sg_hashlib from app.modules.pipeline.signage_author import ( SIGNAGE_POLICY_VERSION as _sg_policy, author_inscriptions, resolve_signage_pack as _sg_pack_fn, signage_pack_content_hash as _sg_pack_content, ) from app.modules.pipeline.still_recipe import ( build_signage_section, ) _sg_shot = (shot_desc_by_id.get(still_id) or s.get("still_frame_prompt") or "") _sg_place = ( cls.get("place_en") or classify_scenes.get(str(si), {}).get("place_en") or location_by_scene.get(si, "") ) # Codex BLOCK-3: 팩 내용 해시까지 캐시 신원에 접는다. _sg_fp = _sg_hashlib.sha256("\n".join([ _sg_shot, _sg_place, world_anchor, _sg_policy, _sg_pack_fn(), _sg_pack_content(), ]).encode("utf-8")).hexdigest()[:16] _sg_key = f"{tag}::signage" _sg_rec = records.data.get(_sg_key) if not (isinstance(_sg_rec, dict) and _sg_rec.get("fp") == _sg_fp): try: _sg_items = author_inscriptions( step_tag="signage_author", shot_text=_sg_shot, place_text=_sg_place, world_facts_block=world_anchor, project_config=project_config, ) except Exception as _sg_exc: # noqa: BLE001 — 비차단 logger.warning( "still_recipe %s: 표기 저작 실패 — %r", tag, _sg_exc) _sg_rec = None else: _sg_rec = {"fp": _sg_fp, "inscriptions": _sg_items} records.data[_sg_key] = _sg_rec records.save() if isinstance(_sg_rec, dict): signage_en = build_signage_section( _sg_rec.get("inscriptions") or []) identity_role_en = "" # (_grok_backend·_guidance 는 camera_frame 절보다 앞서 필요해 # 이 함수 위쪽에서 이미 계산했다.) if identity_role_on and char_refs and not bg_only: from app.modules.pipeline.still_recipe import ( build_identity_ref_role_clause, ) # 2026-08-13 육안 8건 wave 로 얼굴 가림 우선 조항이 v18 에 # 들어왔고, 그때는 컴팩트 v17 오버라이드를 껐다 — v17 스템이 # 그 조항 이전 판이라 되돌리는 셈이었기 때문이다. # 2026-08-18: 되돌리는 대신 **v18 을 보고 새로 저작한** 컴팩트판 # (팩 v22)을 grok 조립에서만 읽는다 — 얼굴 가림 조항은 그대로 # 살아 있다. nb2 는 selector 가 비어 v18 그대로(바이트 동일). identity_role_en = build_identity_ref_role_clause( True, version_selector=_guidance) prompt = build_still_prompt( shot_desc=shot_desc_by_id.get(still_id) or s.get("still_frame_prompt") or "", # fix2 v3: 샷별 서브공간 place_en(실내외 특정) 최우선 → # v2 씬 place_en → 엔티티 이름/헤딩 fallback place_text=( cls.get("place_en") or classify_scenes.get(str(si), {}).get("place_en") or location_by_scene.get(si, "") ), time_of_day_en=( classify_scenes.get(str(si), {}).get("time_of_day_en") or "" ), world_anchor=world_anchor, bg_only=bg_only, prev_used=prev_sel is not None, prev_usage_en=cls.get("usage_en") or "", pose_clauses=( [] if bg_only else build_pose_clauses( continuity.get("pose_canon") or [], tag ) ), movement_en=fix.get("movement_en") or "", figures_en=fix.get("figures_en") or "", carried_en=carried, # 분류 팩 v4 — 그 순간 물건을 다루는 사람(없으면 빈 문자열). handled_by=handled_by, char_names=char_names, char_names_exact=ve_exact, # Stage D HIGH-3: lane(레인1) 샷은 사진 없는 sketch location # authority. 복잡 구조물은 structure_look 절(v4)로 플레이트· # seed 관할 명시. # # 2026-07-27 리뷰 I-3: 단 `lane_chain and bg_only` 는 예외 — # 위 plate 해석이 이 샷에만 플레이트를 되살리고(체인 비대상+ # bg_only 조기 return 이라 그게 유일한 참조), build_still_refs # 가 LOCATION PHOTOGRAPH 라벨로 붙인다. 그런데 sketch lock 은 # "장소 사진은 첨부되지 않았다"고 단언해 방금 되살린 사진을 # 부정한다(wave 이전엔 참조가 실제로 0 이라 자기정합이었다). # 조건을 plate 해석과 **같은 모양**으로 맞춘다. lane_ref_mode=( "sketch" if (lane_used and not (lane_chain and bg_only)) else "" ), structure_seed_attached=structure_seed_path is not None, seed_bg_mode=seed_bg_path is not None, camera_frame_en=camera_frame_en, lighting_mood_en=lighting_mood_en, conduct_version=( _CONDUCT_PACK_SEL if conduct_on else "" ), identity_role_en=identity_role_en, prev_people_rule=prev_people_rule, signage_en=signage_en, prompt_version=_pack, guidance_version=_guidance, ) # 2026-07-25 (케이스1 스펙 E·F): 체인 후보(A)는 **재투영 배경**을 # 참조하므로 prev/lane 전용 location 문구가 거짓이 된다 — 그 # 후보만 일반 조립(prev_used=False, lane_ref_mode="")으로 별도 # 저작한다. 무콘티 후보(B)는 아래 refs/prompt 그대로(현행 경로)라 # 2택1 이 "체인 vs 현행" 비교가 된다. prompt_chain = "" if bgfirst_used and (lane_chain or prev_sel is not None): _chain_cam = camera_frame_en if camera_frame_on and not _chain_cam: from app.modules.pipeline.still_recipe import ( build_camera_frame_clause as _bcfc, ) # 2026-08-18 (Codex BLOCK): 이 재조립은 lane 샷에서만 돈다 # — 첫 조립(:3081 `not lane_used`)이 건너뛰어 camera_frame_en # 이 비기 때문이다. 그 산출이 prompt_chain 이 되어 **grok 롤로 # 그대로 나가므로** 여기에도 컴팩트 갈림을 넘겨야 한다. # 안 넘기면 lane+체인 샷만 카메라 절이 원본(+228B)으로 남아 # 경계 샷이 8,000바이트 상한에 걸려 죽는다. _chain_cam = _bcfc( staging_map.get(f"{si}_{shi}") if staging_map else None, version_selector=_guidance, ) prompt_chain = build_still_prompt( shot_desc=shot_desc_by_id.get(still_id) or s.get("still_frame_prompt") or "", place_text=( cls.get("place_en") or classify_scenes.get(str(si), {}).get("place_en") or location_by_scene.get(si, "") ), time_of_day_en=( classify_scenes.get(str(si), {}).get("time_of_day_en") or "" ), world_anchor=world_anchor, bg_only=False, prev_used=False, prev_usage_en="", pose_clauses=build_pose_clauses( continuity.get("pose_canon") or [], tag ), movement_en=fix.get("movement_en") or "", figures_en=fix.get("figures_en") or "", carried_en=carried, char_names=char_names, char_names_exact=ve_exact, lane_ref_mode="", structure_seed_attached=False, seed_bg_mode=False, # F1 (2026-07-27 리뷰): 체인 Step2 참조는 [배경본, (콘티), # 엔티티] — LOCATION PHOTOGRAPH 는 없다. 위 3개를 끄면 # 기본 fallback 이 없는 사진을 가리키므로, 배경본을 # 장소 권위로 선언하는 스템으로 교체한다. chain_bg_mode=True, camera_frame_en=_chain_cam, lighting_mood_en=lighting_mood_en, conduct_version=(_CONDUCT_PACK_SEL if conduct_on else ""), # 체인 Step2(인물 삽입) 후보도 캐릭터 참조를 실첨부하므로 # 동일 절 동반 (조건·문구 동일 — 후보 간 계약 드리프트 0) identity_role_en=identity_role_en, prompt_version="1", guidance_version=_guidance, ) if _grok_backend: # 시네마틱 마감 절 (2026-08-13 사용자 지시 "꼭 시네마틱하게") — # 체인·무콘티 양 후보의 base 에 공통 동반. CAMERA 권위는 불변. from app.modules.pipeline.still_recipe import ( build_cinematic_finish_clause, ) _cinematic = build_cinematic_finish_clause(_guidance) prompt = prompt + "\n\n" + _cinematic if prompt_chain: prompt_chain = prompt_chain + "\n\n" + _cinematic refs = build_still_refs( bg_only=bg_only, plate=plate, conti=conti, prev_sel=prev_sel, char_refs=char_refs, prop_refs=prop_refs, # 체인 편입 시 마커 스케치는 conti 슬롯이 이미 첨부 — # lane_sketch 로 또 붙이면 같은 파일이 2회 참조된다 lane_sketch=None if lane_chain else lane_sketch_path, structure_seed=structure_seed_path, seed_bg=seed_bg_path, prompt_version=_pack, handled_by=handled_by, geom_authority=geom_authority, prev_people_rule=prev_people_rule, ) if not refs: logger.warning( "still_recipe %s: 참조 0 — text-only 생성 (플레이트·콘티·" "엔티티 전부 부재)", tag, ) if execution_scope is None: progress.update( f"레시피 스틸 생성 중 ({tag})", 2 + i, total + 2 ) else: # Codex 재재리뷰 NARROW-3: 첫 샷=2/(n+2) — 기존 표기(2+i)와 # 동일한 '생성 시작 전' ordinal. increment 는 update 이후. progress.update( f"레시피 스틸 생성 중 ({tag})", 2 + exec_seen, exec_total + 2, ) exec_seen += 1 try: # phase 단위 durable persist — 크래시 창에서 critique 영구 # 생략 결함 차단 (Codex BLOCKING-3). 실행=run_branch_select # 모듈 헬퍼 (Codex 74ebf365 B1: gen/fix 동일 callable 계약을 # 실행 경로 테스트로 잠금) from app.modules.pipeline.still_recipe import run_branch_select def _run_branch(branch_tag: str, branch_refs, rec_key: str, out_stem: Path, rc: Optional[int] = None, jf: Any = None, jt: Any = None, cf: Any = None, ef: Any = None, **variant_kw): # ef: branch 전용 extra_fingerprint 병합본 — confined 샷만 # confined 키를 접는다(전역 오염 금지, Codex BLOCK-1). return run_branch_select( branch_tag=branch_tag, branch_refs=branch_refs, rec_key=rec_key, out_stem=out_stem, prompt=prompt, records=records, make_gen_fn=lambda bt: make_shot_gen_fn(still_id, bt), # G+G46 ON 이면 fix i2i 만 grok 으로 — None(기본)이면 # run_branch_select 가 기존 gen 을 fix 에 재사용한다. make_fix_gen_fn=( (lambda bt: make_shot_fix_gen_fn(still_id, bt)) if gg46_fix_client is not None else None), judge_fn=jf if jf is not None else judge_fn, critique_fn=cf if cf is not None else critique_fn, roll_count=rc if rc is not None else roll_count, critique_enabled=critique_enabled, judge_texts=jt if jt is not None else judge_texts, extra_fingerprint=( ef if ef is not None else extra_fingerprint), run_fn=run_multiroll_select, fix_rejudge_fn=fix_rejudge_fn, composition_critique_fn=composition_critique_fn, # 참조 선별(2026-08-19) — OFF 면 run_branch_select 가 # kwargs 자체를 생략한다(byte-identical 계약 동형). fix_ref_gate=fix_ref_gate_on, fix_missing_texts=fix_missing_texts, **variant_kw, ) def _ab_roll_prompts(base_prompt: str, labels): """2롤 ab 변주 (2026-08-13 사용자 확정 — **전 백엔드·전 경로 무조건**): 첫 롤=기준, 나머지 롤=구도 변주 절 동반. a 와 b 가 같은 구도·앵글로 나오는 일이 없어야 한다는 사용자 원칙의 구조 재료 — 이전 "grok 전용·confined/2택1 제외" 게이트는 같은 날 사용자 재지시로 제거. 구현·시험 SOT= still_recipe.build_ab_roll_prompt_map.""" from app.modules.pipeline.still_recipe import ( build_ab_roll_prompt_map, ) return build_ab_roll_prompt_map( base_prompt, labels, _guidance) def _broll_variant(base_prompt: str) -> str: """둘째 후보용 프롬프트(base+구도 변주 절) — 두 후보가 별도 브랜치/라벨로 생성되는 경로(2택1·레거시 A/B)에서 b 쪽에 부착한다.""" _labs = roll_labels(2) return _ab_roll_prompts(base_prompt, _labs)[_labs[1]] winner_branch_tag = f"still_{tag}" variants_data: Optional[Dict[str, Any]] = None # ── confined fp 생성 (판별은 legacy 게이트 앞에서 확정됨) ── # base·샷 fp 캐시는 **입력 지문 대조** — 존재 캐시는 팩·모델 # 변경을 옛 산출로 봉인한다(Codex BLOCK-2). confined 지문 키는 # 이 branch 에만 병합(ef) — 전역 오염 금지(BLOCK-1). if _confined_active: import hashlib as _hl from app.modules.pipeline.confined_fp import ( FP_REF_LABEL, base_fingerprint, build_confined_gen_prompt, extract_brief_sections, produce_base_fp, produce_shot_fp, shot_fingerprint, ) _place_text = str( cls.get("place_en") or classify_scenes.get(str(si), {}).get("place_en") or location_by_scene.get(si, "") ) _space_key = _hl.sha256( _place_text.encode("utf-8")).hexdigest()[:12] _base_key = f"confinedfp::{_space_key}" _base_path = ( recipe_dir / f"confinedfp_base_{_space_key}.png") _base_fp_sig = base_fingerprint(_place_text) _base_rec = records.data.get(_base_key) # 같은 공간 샷들이 base 도면 1장을 공유 — 샷마다 다른 # 공간을 그리던 결함 제거(08-11 실측·사용자 확정). if not (_base_path.exists() and _base_path.stat().st_size > 0 and isinstance(_base_rec, dict) and _base_rec.get("input_fingerprint") == _base_fp_sig): produce_base_fp(_space_key, _place_text, _base_path) records.data[_base_key] = { "path": str(_base_path), "place_text": _place_text, "input_fingerprint": _base_fp_sig, } records.save() _sections = extract_brief_sections(prompt) _shot_fp = recipe_dir / f"{tag}_confinedfp.png" _cfp_key = f"{tag}::confined_fp" _shot_fp_sig = shot_fingerprint(_base_path, _sections) _cfp = records.data.get(_cfp_key) if not (isinstance(_cfp, dict) and _shot_fp.exists() and _shot_fp.stat().st_size > 0 and str(_cfp.get("scene_description_en") or "").strip() and _cfp.get("input_fingerprint") == _shot_fp_sig): _, _cfp = produce_shot_fp( tag, _base_path, _sections, _shot_fp, project_config=project_config) _cfp["input_fingerprint"] = _shot_fp_sig records.data[_cfp_key] = _cfp records.save() # 실첨부 lineage 와 지시의 일치 (BLOCK-4): 이 branch 의 # 실제 첨부는 [fp 도면, 엔티티]뿐 — 사전 수집된 plate/ # prev/seed/conti lineage 는 위 게이트들이 이미 막았고, # fp 는 파일 기반이라 unresolved 로 남긴다(asset 없음). _attach("confined_fp", FP_REF_LABEL, None, file_path=str(_shot_fp)) _cfp_prompt = build_confined_gen_prompt( str(_cfp["scene_description_en"]), prompt) _cfp_refs: List[Tuple[str, Any]] = [ (FP_REF_LABEL, _shot_fp), *char_refs, *prop_refs] # #119①: confined 샷의 실물 look 앵커 — 도면은 배치 # 권위일 뿐 재질·집기 진실이 없다(S18 열차 실내가 공상이 # 된 실측). era 조사 참조를 도면 다음에 동봉한다 — # roll_refs 에 실리므로 지문에 자동으로 접힌다. 실패= # 비차단(기존 참조 구성 그대로). if era_research_on: from app.modules.pipeline.era_research import ( assess_and_research_cached as _era_cached, build_ref_role as _era_role_fn, ) _era_cf = _era_cached( step_tag="era_research_confined", subject_text=str(_place_text or ""), world_facts_block=world_anchor, out_dir=recipe_dir, cache_get=lambda k: ( records.data.get(k) if isinstance(records.data.get(k), dict) else None), cache_put=lambda k, v: ( records.data.__setitem__(k, v), records.save()), project_config=project_config, openai_client=_bgfirst_gpt_client, failed_memo=era_failed_memo, ) if _era_cf: _cfp_refs.insert(1, ( _era_role_fn(_era_cf["subject"]), Path(_era_cf["path"]))) # Codex BLOCK-5: 실제 첨부한 era 이미지가 최종 # 자산 입력 채널에도 남게 — asset UUID 없는 조사 # 파일은 unresolved 로 role+file+sha 병기. _attach( "era_ref", str(_era_cf.get("subject") or ""), None, file=str(_era_cf.get("file") or ""), sha256=str(_era_cf.get("sha256") or "")) _labels_n = roll_labels(roll_count) # 표준 멀티롤(_chain_only 패턴) — critique 는 confined # 프롬프트만 대조(브리프 원문의 CAMERA 절과 이중 LOCATION # 모순 차단). 2026-08-13 사용자 확정: confined 도 예외 # 없이 b=구도 변주 — 도면·readback 사다리가 기하 위반을 # 잡는 계약은 그대로(변주 절은 장소·인물·순간 불변 명문). sel_path, record = _run_branch( f"still_{tag}", _cfp_refs, tag, recipe_dir / tag, roll_prompts=_ab_roll_prompts(_cfp_prompt, _labels_n), roll_refs={ln: _cfp_refs for ln in _labels_n}, critique_selected_prompt_only=True, ef={**extra_fingerprint, **confined_extra_fingerprint}, ) # record 가 실행을 대변하게 — confined 메타 병기 (BLOCK-4) record["confined_fp"] = { "base_key": _base_key, "apt_reason": str( (records.data.get(f"{tag}::confined_fp_apt") or {}) .get("reason_ko") or ""), "fixed": bool(_cfp.get("fixed")), "mismatches": list(_cfp.get("mismatches") or []), # #119①: 조사가 성립했을 때만 키 — 기록 없는 유료 # 조사는 없다(참조 자체는 roll_refs 지문에 접힘). **({"era_research": { k: _era_cf[k] for k in ( "subject", "queries", "picked_url", "sha256", "file") if k in _era_cf}} if era_research_on and _era_cf else {}), } records.save() # BGFIRST2 샷은 2택1(체인 vs 무콘티)이 변형 4택1 을 대체 — # 변형 저작 자체를 생략(사용자 확정 "이전 2장씩 대체"). # confined fp 샷도 동일하게 저작 생략. if variants_on and not bgfirst_used and not _confined_active: # 샷별 변형 2종 저작 — 캐시(author_fp) 재사용, 검증 실패= # 재시도 1회 후 raise → 아래 except 가 샷 단위 mark_failed # (Codex 합의 4: legacy 3롤 degrade 금지) from app.modules.pipeline.still_variants import ( author_still_variants, ) _vkey = f"{tag}::variants" variants_data, _vfp, _vreused = author_still_variants( base_prompt=prompt, shot_text=shot_desc_by_id.get(still_id) or s.get("still_frame_prompt") or "", cached_entry=records.data.get(_vkey), project_config=project_config, ) records.data[_vkey] = { "author_fp": _vfp, "author": variants_data, "reused": _vreused, } records.save() if _confined_active: # confined fp 경로가 위에서 이미 sel_path/record 를 만들었다 # — bgfirst 2택1·표준/변형 분기로 재진입하지 않는다. pass elif bgfirst_used: # BGFIRST2 (2026-07-20 사용자 확정): Step1=플레이트를 콘티 # 카메라로 재투영한 인물 0 빈 배경(gpt-image-2 — nb2 는 # 플레이트 프레이밍 고수 실측·재투영 실패) → Step2=그 # 배경+콘티(인물 배치만)+엔티티로 nb2 인물 삽입 후보(A) # vs 무콘티 기존 조립 후보(B) → VLM 2택1(블라인드·정역순 # flip, judge 공유 refs=무콘티). 기존 4택1 을 대체. from app.modules.pipeline.still_recipe import ( LANE_CONTI_ONLY, build_ab_branch_refs, build_bgfirst_bg_prompt, build_bgfirst_final_prompt, build_bgfirst_refs, build_bgfirst_seed_clause, ) # ── 위치 권위 해석 (fix③④ full): 플레이트 → seed-bg → # groupbg(장소 단위 생성·재사용 — ③ 정류장류 연속성의 # 근본 해결 지점). 비 full=기존 플레이트 필수 그대로 ── _authority_kind = "plate" _authority_path = plate _authority_aid: Optional[str] = None _chain_seed_path: Optional[Path] = None _chain_seed_aid: Optional[str] = None _groupbg_key: Optional[str] = None if bgfirst_full_on: if structure_seed_path is not None: # complex 샷 — STRUCTURE LOOK 을 Step1 3번째 참조로 _chain_seed_path = structure_seed_path _chain_seed_aid = structure_seed_asset_id if lane_chain: # 2026-07-26 사용자 확정: lane 콘티는 마커 맵 # 1장만 보고 마네킹으로 그려진다. 배경은 그 콘티 # 자체를 i2i 로 편집해 입히고 외부 사진은 쓰지 # 않는다(사전 배경 플레이트 금지 — 미리 만든 빈 # 배경과 콘티 구도가 어긋나 구조물이 겹쳐 보이던 # 실측). 장소 사실은 텍스트 권위(§4.3). # # 설계 §(2) 진입 assert — 이 판정은 plate/seed_bg # 분기보다 **앞**이어야 한다. `plate is None` 을 # 분기 **조건**으로 쓰면, plate_map 이 lane 샷에 # 플레이트를 물리는 순간 조건이 조용히 빗나가 # 권위가 "plate" 로 떨어지고 Step1 참조가 [콘티, # 외부 플레이트] 가 된다 — 이 흐름이 없애려던 구도 # 불일치 그 자체다. 공급되면 죽인다(fail-closed). # # 공급원 차단은 플레이트 해석 지점이 담당한다 # (lane 샷은 조회 자체를 하지 않는다 — 위 # `plate = None if (lane_used and not (lane_chain # and bg_only)) ...`). 여기는 그 계약이 무너졌을 # 때를 잡는 이중 방어이지, 정상 흐름에서 도달하는 # 경로가 아니다. if plate is not None or seed_bg_path is not None: raise ValueError( f"{LANE_CONTI_ONLY} 샷에 플레이트/seed_bg " f"가 공급됨 (plate={plate}, " f"seed_bg={seed_bg_path}) — 사전 배경 " "플레이트 금지 계약 위반 (fail-closed)" ) _authority_kind = LANE_CONTI_ONLY _authority_path = None _authority_aid = None elif plate is None and seed_bg_path is not None: _authority_kind = "seed_bg" _authority_path = seed_bg_path _authority_aid = seed_bg_asset_id elif ( plate is None and prev_sel is not None and lane_prev_chain_on ): # 2026-07-25 사용자 확정: prev 지휘 샷 = 직전 # 스틸이 배경 권위 — 콘티 카메라로 **재투영**해 # 구도차를 흡수한다(참조만으로는 장소가 재현되지 # 않던 실측 교정, S15sh5). _authority_kind = "prev" _authority_path = prev_sel _authority_aid = ( primary_asset_by_tag.get(prev_tag or "") or prev_primary_asset_id ) elif plate is None: _groupbg_key = group_of.get(tag) if not _groupbg_key: raise ValueError( "BGFIRST full no_plate 샷인데 share_plan " "그룹 미배정 — groupbg 묶음 불가 " "(fail-closed)" ) _authority_kind = "groupbg" _authority_path, _authority_aid = _run_groupbg( _groupbg_key, tag, still_id, conti, conti_entry.get("asset_id"), ( cls.get("place_en") or classify_scenes.get(str(si), {}) .get("place_en") or location_by_scene.get(si, "") ), ( classify_scenes.get(str(si), {}) .get("time_of_day_en") or "" ), # E2E11 ③ (NARROW-4): 장소 근거 — 그룹 안정 # 파생값(멤버 씬들의 loc 상세+그룹 evidence) (groupbg_context.get(_groupbg_key) or {}) .get("detail", ""), tuple( (groupbg_context.get(_groupbg_key) or {}) .get("evidence", ()) ), ) # LANE_CONTI_ONLY 는 위치 권위 **파일**이 존재하지 않는 # 것이 정상 계약이다(외부 사진 0) — 그 모드만 면제하고 # 나머지 권위는 기존 fail-closed 그대로. if _authority_kind != LANE_CONTI_ONLY and ( _authority_path is None or not _authority_path.is_file() ): raise ValueError( f"BGFIRST2 샷 LOCATION 권위 결손({_authority_path}," f" kind={_authority_kind}) — 재투영 불가 " "(fail-closed)" ) # Step1 카메라 계약 — fix1 flag 와 독립(체인은 콘티 v2 와 # 같은 staging 구도 계약을 항상 소비 — 정본 동조) # 2026-08-18: 여기는 gpt-image-2 배경판(BGFIRST_BG_IMAGE_MODEL) # 경로다 — xAI 8,000바이트 상한과 무관하므로 컴팩트 스템을 # 쓰면 얻는 것 없이 그릴 재료만 줄고, 프롬프트가 지문 # (compute_input_fingerprint) 에 들어가 배경판이 무효가 된다. # 그래서 camera_frame_en(=selector 적용본) 재사용을 끊고 # 항상 기본 절로 새로 만든다. nb2 는 selector 가 빈 값이라 # 같은 인자·같은 결과 = 바이트 동일. from app.modules.pipeline.still_recipe import ( build_camera_frame_clause, ) _bg_cam = build_camera_frame_clause( staging_map.get(f"{si}_{shi}") if staging_map else None ) from app.modules.pipeline.still_recipe import ( BGFIRST_FULL_PROMPT_VERSION as _BGF_FULL_PACK_SEL, BGFIRST_LANE_PROMPT_VERSION as _BGF_LANE_PACK_SEL, ) # 2026-07-26 §4.3: 참조 0 인 lane 샷은 장소 외형의 # **이미지** 권위가 없다 — place spec 절제 블록+world # 사실이 그 자리를 대신한다(둘 중 하나라도 비면 위 # 헬퍼/조립이 fail-closed). _lane_fill = _authority_kind == LANE_CONTI_ONLY bg_prompt = build_bgfirst_bg_prompt( shot_desc=shot_desc_by_id.get(still_id) or s.get("still_frame_prompt") or "", place_text=( cls.get("place_en") or classify_scenes.get(str(si), {}).get("place_en") or location_by_scene.get(si, "") ), time_of_day_en=( classify_scenes.get(str(si), {}) .get("time_of_day_en") or "" ), camera_frame_en=_bg_cam, lighting_mood_en=lighting_mood_en, place_facts_block=( _lane_place_facts(lane_entry.get("group_id") or "") if _lane_fill else "" ), world_facts_block=( _lane_world_block if _lane_fill else "" ), lane_fill=_lane_fill, # fix⑥ (E2E11): full=v10 head(무인="사람만" — 상주 # 동물·물품 유지+휴먼 스케일). 비 full=v7 byte-identical. # lane=v12 bg_fill_* (참조 0 i2i 채색). prompt_version=( _BGF_LANE_PACK_SEL if _lane_fill else _BGF_FULL_PACK_SEL if bgfirst_full_on else _BGFIRST_PACK ), ) if _chain_seed_path is not None: # complex 샷: Step1 에 STRUCTURE LOOK 관할 절 동반 bg_prompt = ( bg_prompt + "\n\n" + build_bgfirst_seed_clause() ) # #119①: 샷별 배경 판 — 이 주행이 실제로 재생성하는 배경 # 경로(실측 98/215)인데 era 조사가 groupbg(B2)에만 있었다. # B2 와 같은 구조를 캐시 래퍼로 동봉한다(같은 장소의 샷들 # 은 1회만 지출). 조사 성공 시 역할문+참조가 지문에 접혀 # 재사용/재생성이 갈리고, 실패는 비차단. _era_pl = None if era_research_on: from app.modules.pipeline.era_research import ( assess_and_research_cached as _era_cached_pl, build_ref_role as _era_role_pl, ) _era_pl = _era_cached_pl( step_tag="era_research_plate", subject_text=str( cls.get("place_en") or classify_scenes.get(str(si), {}) .get("place_en") or location_by_scene.get(si, "")), world_facts_block=world_anchor, out_dir=recipe_dir, cache_get=lambda k: ( records.data.get(k) if isinstance(records.data.get(k), dict) else None), cache_put=lambda k, v: ( records.data.__setitem__(k, v), records.save()), project_config=project_config, openai_client=_bgfirst_gpt_client, failed_memo=era_failed_memo, ) if _era_pl: bg_prompt = (bg_prompt + "\n\n" + _era_role_pl(_era_pl["subject"])) bg_path, bg_asset_id = _run_bgfirst_bg( tag, still_id, conti, _authority_path, bg_prompt, conti_entry.get("asset_id"), plate_asset_id_override=_authority_aid, seed_path=_chain_seed_path, seed_asset_id=_chain_seed_aid, authority_kind=_authority_kind, era_ref_path=( Path(_era_pl["path"]) if _era_pl else None), era_meta=_era_pl, ) refs_chain = build_bgfirst_refs( bg=bg_path, conti=None if lane_chain else conti, char_refs=char_refs, prop_refs=prop_refs, # 좁고 복잡한 실내 — 스케치가 인물 배치**와** 구조 둘 # 다의 권위. lane 은 마네킹 계약이 이미 지배한다. geom_authority=geom_authority and not lane_chain, ) # 2026-07-25 사용자 확정: lane 샷은 **체인 단독**. 무콘티 # 후보는 마커 스케치를 버리므로 카메라 위치·인물 배치 # 통제가 사라진다 — 맵→마커→콘티를 거친 이유 자체가 # 무색해지고, 판정이 그쪽을 고르면 마커 단계가 무력화된다. # 복잡 구조물(케이스2)의 2택1 은 그대로 유지. # # lane 체인은 2택1이 없다 — B 후보를 조립할 이유가 없고, # LANE_CONTI_ONLY 는 plate·seed_bg 가 모두 None 이라 # build_ab_branch_refs 가 "A/B 는 LOCATION 권위 필수" # ValueError 로 샷을 죽인다(still_recipe.py:737). 그래서 # 판정을 B 조립보다 **앞**에 둔다. _chain_only = lane_chain # 무콘티 후보(B)=기존 조립 — 위치 권위·seed 를 그대로 반영 # (groupbg 는 plate 슬롯: LOCATION PHOTOGRAPH 라벨) refs_b: Any = None if not _chain_only: if _authority_kind == "prev": # 2026-07-25: prev 지휘 샷은 무콘티 후보를 만들 수 # 없다 — LOCATION 권위 슬롯에 인물이 찍힌 prev # 스틸을 넣으면 그 인물이 복제된다 # (build_ab_branch_refs 도 prev 호출 금지 계약). # B=현행 prev 경로 그대로 두어 2택1 이 "체인 vs # 현행" 비교가 되게 한다. refs_b = refs else: _, refs_b = build_ab_branch_refs( plate=( # "canon_master"(구 lane 권위)는 이 튜플에서 # 뺀다 — 유일한 배정 지점이 LANE_CONTI_ONLY # 로 대체돼 서비스에서 다시 나올 수 없다. # (bgfirst_winner_lineage 의 allowlist 는 # 구 record 해석용으로 그대로 둔다.) _authority_path if _authority_kind in ("plate", "groupbg") else None ), conti=conti, char_refs=char_refs, prop_refs=prop_refs, structure_seed=structure_seed_path, seed_bg=( seed_bg_path if _authority_kind == "seed_bg" else None ), prompt_version=_pack, ) _labels2 = roll_labels(2) # lane 체인만 마네킹 교체 스템(v12 stage_head_mannequin) — # Step1 이 마네킹을 보존한 채 배경만 실사화하고 Step2 참조에 # 콘티가 없으므로, v7 stage_head("배경 EXACTLY 유지"+없는 # LAYOUT SKETCH 지시)를 그대로 쓰면 회색 마네킹이 최종까지 # 살아남는다. 비 lane 은 인자 기본값과 동일한 v7 그대로. _chain_prompt = build_bgfirst_final_prompt( prompt_chain or prompt, prompt_version=( _BGF_LANE_PACK_SEL if lane_chain else _BGFIRST_PACK ), mannequin=lane_chain, # ★인물 삽입이 배경 기하를 다시 그리던 자리(실물 대조: # 배경은 핸들 1개인데 삽입 뒤 이중 림). 좁고 복잡한 # 실내에서만 "배경선 무시" 계약을 기하 보존으로 바꾼다. geom_authority=geom_authority, ) if _chain_only: # 표준 멀티롤 — 같은 체인 프롬프트/참조로 N롤 생성 후 # 기본 판정이 최선을 고른다(후보 비교가 아니라 품질 # 선택). bgfirst 2택1 전용 judge/texts 는 2라벨 스키마라 # 여기 쓰지 않는다. _labels_n = roll_labels(roll_count) sel_path, record = _run_branch( f"still_{tag}", refs_chain, tag, recipe_dir / tag, # 2026-08-13: grok 백엔드는 첫 롤=기준, 나머지=구도 # 변주(ab). nb2 는 기존 같은 프롬프트 그대로. roll_prompts=_ab_roll_prompts( _chain_prompt, _labels_n), roll_refs={ln: refs_chain for ln in _labels_n}, # 2026-07-27 리뷰 I-2: 없으면 critique 프롬프트가 # f"{_chain_prompt}\n\n{prompt}" 합성으로 떨어진다 — # `prompt` 는 lane_ref_mode="sketch" 조립이라 # "첨부된 STORYBOARD SKETCH 가 배치 SOT·장소 사진 # 없음"을 말하는데, 체인에서 스케치는 **의도적으로 # 미첨부**이고 _chain_prompt 는 배경본을 장소 권위로 # 선언한다(상호 모순 LOCATION lock 2개). VLM 지적은 # unfixable 마킹이 없어 그대로 build_fix_prompt 로 # 흘러 정상 스틸을 망가뜨린다. _chain_prompt 가 이미 # base 전문을 품고 roll_prompts 도 있으므로 소실 없음. critique_selected_prompt_only=True, ) else: sel_path, record = _run_branch( f"still_{tag}", refs_b, tag, recipe_dir / tag, rc=2, jf=bgfirst_judge_fn, jt=bgfirst_judge_texts, cf=bgfirst_critique_fn, roll_prompts={ # 체인 후보는 재투영 배경을 보므로 prev/lane # 전용 location 문구가 없는 일반 조립을 쓴다 # (위에서 저작). 비편입 샷은 prompt_chain="" # → 기존 동일. 2026-08-13 사용자 확정: 2택1 # 도 예외 없이 b=구도 변주 절 동반 — 참조 # 세트만 다르면 같은 staging 위에서 a·b 구도가 # 수렴한다(nb2 23샷 실측·육안 반려가 근거). _labels2[0]: _chain_prompt, _labels2[1]: _broll_variant(prompt), }, roll_refs={ _labels2[0]: refs_chain, _labels2[1]: refs_b, }, parallel_rolls=True, judge_flip=True, # 동점=콘티 체인 우선 (tie_keeps_conti 정책 연속) flip_priority=list(_labels2), critique_selected_prompt_only=True, judge_prompt_header=bgfirst_judge_header, ) record = dict(record) _chain_won = ( True if _chain_only else record.get("selected") == _labels2[0] ) record["bgfirst"] = { "bg_path": str(bg_path), "bg_asset_id": bg_asset_id, "bg_record_key": f"{tag}::bgfirst_bg", "chain_winner": _chain_won, } if bgfirst_full_on: # fix③④ 감사 — 위치 권위 종류·groupbg 묶음 키 record["bgfirst"]["authority"] = _authority_kind if _groupbg_key: record["bgfirst"]["group_key"] = _groupbg_key record["bgfirst"]["groupbg_asset_id"] = ( _authority_aid) if _chain_seed_path is not None: record["bgfirst"]["seed_attached"] = True # 승자 기준 lineage — 체인 승=콘티+재투영 배경(명시 등록 # asset UUID — Codex 리뷰 2), 무콘티 승=플레이트 (거짓 # edge 방지, 위에서 유예) from app.modules.pipeline.still_recipe import ( bgfirst_winner_lineage, ) for _role in bgfirst_winner_lineage( chain_won=_chain_won, authority_kind=_authority_kind, structure_seed_attached=_chain_seed_path is not None, # lane 체인 final 은 Step2 참조에서 콘티를 뺐다 — # 콘티 UUID 는 중간 bgfirst_bg 의 input edge 로만 # 남는다(conti → bgfirst_bg → final) conti_attached=not lane_chain, ): if _role == "conti": _attach( "conti_light", "LAYOUT SKETCH", conti_entry.get("asset_id"), file_path=str(conti), ) elif _role == "bgfirst_bg": _attach( "bgfirst_bg", "SHOT BACKGROUND", bg_asset_id, file_path=str(bg_path), ) elif _role == "plate": _attach_plate() elif _role == "groupbg": # 무콘티 승 — groupbg 가 위치 참조 (③ 공유 계보) _attach( "bgfirst_group_bg", "LOCATION PHOTOGRAPH", _authority_aid, file_path=str(_authority_path), ) elif _role == "canon_master": # lane 무콘티 승 — 콘티가 본 장소 실사가 직접 참조 _attach( "lane_canon_master", "LOCATION PHOTOGRAPH", _authority_aid, file_path=str(_authority_path), ) elif _role == "prev": # 무콘티(=현행 prev 경로) 승 — prev 스틸이 직접 # 참조. 구조키 fallback 은 즉시 부착 경로와 동일. if _authority_aid: _attach( "prev_still", f"prev={prev_tag}", _authority_aid, ) else: unresolved_refs.append({ "pipeline_role": "scene_prev_frame", "role": "prev_still", "label": f"prev={prev_tag}", "source_still_id": still_id_by_tag.get( prev_tag or ""), }) elif _role == "seed_bg": _attach( "location_seed_bg", "LOCATION STRUCTURE PHOTOGRAPH", seed_bg_asset_id, file_path=str(seed_bg_path), ) elif _role == "structure_seed": _attach( "structure_seed_look", "STRUCTURE LOOK", structure_seed_asset_id, file_path=str(structure_seed_path), ) elif ab_active and variants_on: # 4택1 단일 호출 — 변형2×콘티유무2, 블라인드+정역순 flip, # 4롤 병렬. 기존 2단(브랜치 3롤+outer)을 대체. from app.modules.pipeline.still_recipe import ( build_ab_branch_refs, ) from app.modules.pipeline.still_variants import ( ab_label_map, assemble_still_roll_prompts, build_ab_roll_refs, winner_uses_conti, ) refs_a, refs_b = build_ab_branch_refs( plate=plate, conti=conti, char_refs=char_refs, prop_refs=prop_refs, structure_seed=structure_seed_path, seed_bg=seed_bg_path, prompt_version=_pack, ) _labels4 = roll_labels(4) # judge 공유 refs=무콘티(refs_b) — 어느 후보가 콘티 참조 # 인지 구조적으로 미노출 (블라인드). out_stem=관례 경로 # (recipe_dir/tag) — prev 앵커 경로 자동 성립. sel_path, record = _run_branch( f"still_{tag}", refs_b, tag, recipe_dir / tag, rc=4, jf=judge_fn_by_count[4], jt=judge_texts_by_count[4], cf=critique_fn_by_count[4], roll_prompts=assemble_still_roll_prompts( base_prompt=prompt, variants=variants_data["variants"], labels=_labels4, ), roll_refs=build_ab_roll_refs(refs_a, refs_b), parallel_rolls=True, judge_flip=True, # 동점=콘티 변형 우선 (tie_keeps_conti 정책 연속) flip_priority=list(_labels4), critique_selected_prompt_only=True, judge_prompt_header=sv_judge_header, ) record = dict(record) _w_conti = winner_uses_conti(record["selected"]) record["variant_map"] = ab_label_map() record["conti_winner"] = _w_conti # 승자가 콘티 사용 후보일 때만 conti lineage 부착 (거짓 # edge 방지 — 위에서 유예) if _w_conti: _attach( "conti_light", "LAYOUT SKETCH", conti_entry.get("asset_id"), file_path=str(conti), ) elif variants_on: # 비A/B 샷(bgonly/prev/일반) — 2롤 변형 병렬, 단일 판정 # (flip 은 콘티 모드 결정이 걸린 4택1 전용 — Codex 합의 1) from app.modules.pipeline.still_variants import ( assemble_still_roll_prompts, ) sel_path, record = _run_branch( f"still_{tag}", refs, tag, recipe_dir / tag, rc=2, jf=judge_fn_by_count[2], jt=judge_texts_by_count[2], cf=critique_fn_by_count[2], roll_prompts=assemble_still_roll_prompts( base_prompt=prompt, variants=variants_data["variants"], labels=roll_labels(2), ), parallel_rolls=True, critique_selected_prompt_only=True, judge_prompt_header=sv_judge_header, ) record = dict(record) elif ab_active: # E2E6 ⑧: A(콘티 포함)/B(미포함) 풀 파이프 ×2 → outer # VLM 블라인드·순서 뒤집기 2회 → 승자 (동점=콘티 사용본 A). # Codex 8e70d4c0 B1: outer 결정=durable+지문화 — 지문 일치 # resume 은 outer 0콜로 동일 winner 재사용, 판정 실패는 # 샷 실패 격리(사전 결정 뒤집힘 창 제거). from app.modules.pipeline.conti_ab import ( resolve_or_run_outer, ) from app.modules.pipeline.still_recipe import ( build_ab_branch_refs, ) # R6: A/B 두 브랜치 refs 를 공용 helper 로 조립 — 콘티 # 유무만 다르고 플레이트·seed·엔티티·팩 라벨은 동일 # (브랜치별 개별 조립의 seed/라벨 누락 드리프트 차단) refs_a, refs_b = build_ab_branch_refs( plate=plate, conti=conti, char_refs=char_refs, prop_refs=prop_refs, structure_seed=structure_seed_path, seed_bg=seed_bg_path, prompt_version=_pack, ) # 2026-08-13 #106 (Codex R1 BLOCK-2): 이 레거시 경로는 # 브랜치×기본 롤 수로 4장 이상을 만들고 변주도 없었다 — # "후보는 a/b 두 장, 서로 구도 상이" 계약으로 봉인: # 브랜치당 1롤, 무콘티(B) 쪽에 변주 절. _ab1 = roll_labels(1) sel_a, rec_a = _run_branch( f"still_{tag}_ab_conti", refs_a, f"{tag}::ab_conti", recipe_dir / f"{tag}__ab_conti", rc=1) sel_b, rec_b = _run_branch( f"still_{tag}_ab_noconti", refs_b, f"{tag}::ab_noconti", recipe_dir / f"{tag}__ab_noconti", rc=1, roll_prompts={_ab1[0]: _broll_variant(prompt)}) decision = resolve_or_run_outer( prompt=prompt, sel_a=Path(sel_a), sel_b=Path(sel_b), prior_decision=records.data.get( f"{tag}::conti_ab_decision"), judge_model_physical=settings.gemini_text_model, project_config=project_config, ) # canonical copy **이전** durable persist (B1 crash 창) records.data[f"{tag}::conti_ab_decision"] = { k: decision[k] for k in ("fingerprint", "winner", "outer") } records.save() winner = decision["winner"] winner_branch_tag = ( f"still_{tag}_ab_conti" if winner == "A" else f"still_{tag}_ab_noconti") sel_path, record = ( (sel_a, dict(rec_a)) if winner == "A" else (sel_b, dict(rec_b))) record["conti_ab"] = { "winner": winner, "outer": decision["outer"], "outer_judged_this_run": decision.get("judged", True), "sel_conti": str(sel_a), "sel_noconti": str(sel_b), } # prev 앵커 관례 경로(recipe_dir/tag_sel.png)에 승자 복사 # — 결정 record 기준 idempotent 복원(원자 쓰기) from app.modules.pipeline.multiroll_gemini import ( atomic_write_bytes, ) atomic_write_bytes( recipe_dir / f"{tag}_sel.png", Path(sel_path).read_bytes()) sel_path = recipe_dir / f"{tag}_sel.png" # 승자가 콘티 사용본일 때만 conti lineage 부착 (거짓 edge # 방지 — 위에서 유예) if winner == "A": _attach( "conti_light", "LAYOUT SKETCH", conti_entry.get("asset_id"), file_path=str(conti), ) else: # 표준(같은 프롬프트) 멀티롤의 ab 변주 — 2026-08-13 사용자 # 확정으로 전 백엔드 무조건(이전 grok 전용 게이트 제거). # critique 는 선정 롤 프롬프트만 대조(값이 base 전문을 # 포함 — 이중 전문 합성 방지, chain_only 관례 동형). _labels_d = roll_labels(roll_count) sel_path, record = _run_branch( f"still_{tag}", refs, tag, recipe_dir / tag, roll_prompts=_ab_roll_prompts(prompt, _labels_d), critique_selected_prompt_only=True, ) _seed_note = "+seed" if structure_seed_path is not None else "" _bg_note = ( "seed-bg" if seed_bg_path is not None else "플레이트") if _confined_active: # confined fp — 실제 첨부([fp 도면, 엔티티])와 동기 # (Codex BLOCK-4: ref_mode 갈래 부재로 기록이 실행을 # 대변하지 못했다). record["ref_mode"] = "confined_fp: 도면+장면설명+엔티티" elif lane_used: record["ref_mode"] = ( f"lane({lane_entry.get('lane')}): 스케치" + ("" if bg_only else ("+prev+엔티티" if prev_sel is not None else "+엔티티")) ) elif bgfirst_used: _auth_note = { "seed_bg": "seed-bg", "groupbg": "그룹 배경", }.get(record["bgfirst"].get("authority") or "plate", "플레이트") record["ref_mode"] = ( "재투영 배경+콘티+엔티티 (2택1: 체인 승)" if record["bgfirst"]["chain_winner"] else f"{_auth_note}+엔티티 (2택1: 무콘티 승)" ) elif ab_active and variants_on: _sel_lab = record.get("selected") _kind = "복잡구조물 4택1" if complex_ab else "4택1" _vm = (record.get("variant_map") or {}).get(_sel_lab) or {} _vnote = f"{_sel_lab}=변형{_vm.get('variant')}" record["ref_mode"] = ( f"{_bg_note}+콘티{_seed_note}+엔티티" f" ({_kind}: 콘티 승·{_vnote})" if record.get("conti_winner") else f"{_bg_note}{_seed_note}+엔티티" f" ({_kind}: 무콘티 승·{_vnote})" ) elif ab_active: _w = record.get("conti_ab", {}).get("winner") _kind = "복잡구조물 A/B" if complex_ab else "A/B" record["ref_mode"] = ( f"{_bg_note}+콘티{_seed_note}+엔티티 ({_kind}: 콘티 승)" if _w == "A" else f"{_bg_note}{_seed_note}+엔티티 ({_kind}: 무콘티 승)" ) else: record["ref_mode"] = ( ("prev만 (배경 전용·공유 계획)" if prev_sel is not None else f"{_bg_note}{_seed_note}만 (배경 전용)") if bg_only else ("prev+엔티티" if prev_sel is not None else "플레이트+콘티+엔티티") ) if share_shot_plan is not None: record["share_plan"] = dict(share_shot_plan) if lane_policy: # 복잡 구조물 정책 감사 기록 (R6 — bypass 사유 명시) record["lane_policy"] = lane_policy if locked_excluded: record["locked_char_refs_excluded"] = locked_excluded if plate_select_rec: record["plate_select"] = plate_select_rec # ── i2i 시네마틱 변환 스테이지 (2026-08-13 #108) — sel 확정 # 직후. applied 면 변환본(_cine)이 최종 영속 원본이 되고 원본 # _sel 은 그대로 prev 체인 앵커. 실패=원본 fallback+기록(재방문 # 재시도). ImageCallBudgetExceeded 는 안에서 전파 → 아래 샷 # 실패 격리로 떨어진다(조용한 미변환 완주 금지). final_src = Path(sel_path) cine_rec: Optional[Dict[str, Any]] = None if cine_on: from app.modules.pipeline.cine_transform import ( resolve_or_run_cine_transform, ) cine_rec = resolve_or_run_cine_transform( tag=tag, sel_path=Path(sel_path), recipe_dir=recipe_dir, records=records, client=cine_client, prompt=cine_prompt, model=settings.grok_image_model, stem_content_hash=cine_stem_hash, pack=cine_pack, context={ "project_id": project_id, "episode_id": episode_id, "operation_type": "still_cine_transform", "still_id": still_id, "multiroll_tag": f"still_{tag}_cine", "scene_index": si, "shot_index": shi, }, ) if cine_rec.get("applied") and cine_rec.get("file"): final_src = recipe_dir / str(cine_rec["file"]) elif cine_rec.get("declined"): # ★검열로 포기한 변환 (2026-08-19 사용자 결정) — 미완이 # 아니라 **확정된 결말**이다. 실패로 세면 스텝이 영영 안 # 닫히고, 재개마다 앞 단계가 다시 돌아 연쇄 재생성이 # 난다(08-19 실측: 한 바퀴에 그림 66장). 원본이 최종본. cine_declined_tags.append(tag) logger.warning( "still_recipe %s: 변환을 검열로 포기 — 원본을 " "최종본으로 확정 (사유=%s)", tag, cine_rec.get("declined_reason") or "moderation") else: # 원본 fallback 영속 — 스텝 집계에 안 잡히므로 걷기 # 끝 raise 로 봉인을 막는다 (BLOCK-1). cine_failed_tags.append(tag) records.data[tag] = record records.save() except Exception as exc: # noqa: BLE001 — 샷 단위 실패 격리 logger.exception("still_recipe %s: 생성 실패", tag) scene_cp.mark_failed(still_id, str(exc)) if jit_verify: # Codex BLOCK 1: 완료 샷의 검증 실패는 옛 primary 가 남아 # 스텝 집계에 안 잡힌다 — 걷기 끝에 모아 스텝을 실패로 # 남겨서 completed/새 hash 봉인을 막는다. jit_failed_tags.append(tag) continue if jit_verify: # #77-B 판정 (Codex BLOCK 2 반영): 두 신호를 따로 쓴다. # · bytes 동일 여부 → "새 asset 영속이 필요한가"만 정한다. # · record 묶음 변화 → "돈이 나갔는가"를 정한다 — 유료 재개 # 경로(소급 critique·재판정·외부 2택1 등)는 전부 record 를 # 갱신하므로(재개 규율), bytes 가 그대로여도 record 가 # 움직였으면 지출로 세어 상한에 넣는다. 표기용 키는 # _jit_tag_snapshot 이 걸러 거짓 지출을 막는다. from app.core.file_paths import resolve_image_path from app.models.project import ImageAsset _records_changed = ( _jit_tag_snapshot(records, tag, jit_known_tags) != jit_snapshot_before) _prior = ( db.query(ImageAsset) .filter( ImageAsset.project_id == project_id, ImageAsset.episode_id == episode_id, ImageAsset.still_id == still_id, ImageAsset.asset_type == "scene", ImageAsset.is_primary == 1, ) .first() ) _prior_path = ( resolve_image_path(_prior.file_path) if _prior else None) try: # #108: 비교 대상은 "이번에 영속할 원본"(final_src — 변환 # ON 이면 _cine, 아니면 _sel). sel 로 고정하면 변환 ON 의 # 재사용 바퀴마다 prior(변환본)≠sel(원본)이라 전 샷이 # 거짓 재생성으로 읽혀 상한 래치까지 간다. _unchanged = ( _prior_path is not None and _prior_path.exists() and final_src.exists() and _prior_path.read_bytes() == final_src.read_bytes() ) except OSError: _unchanged = False if _unchanged and not _records_changed: # 지출 흔적 0 + 산출 동일 = 진짜 재사용 — 아무것도 안 쓴다. jit_fresh += 1 # 후속 샷 prev lineage 는 기존 primary 로 직결 유지. primary_asset_by_tag[tag] = _prior.id continue # #108 (Codex R2 BLOCK-2): 이번 걷기에서 **유료 신규 변환이 # 성공**했으면 bytes 동일이어도 영속을 생략하지 않는다 — # 생략하면 primary 메타(applied=false·롤 모델·롤 호출)가 # 실제 상태(변환 성공·grok 호출·변환 팩)와 반대로 남은 채 # 스텝이 봉인된다. 실패→실패·재사용 바퀴는 기존대로 생략 # (churn 불필요 — SOT 는 records). _cine_fresh_success = bool( cine_rec and cine_rec.get("applied") and not cine_rec.get("reused")) # 2026-08-20: 이번 걷기에 **새로 확정된 포기**도 같이 영속한다 — # 포기는 미완이 아니라 결말인데, 자산이 "applied=false + 검열 # 오류"로만 남은 채 봉인되면 나중에 일시 실패와 구분할 길이 # 글자 대조뿐이다(금지). 산출 bytes 는 원본 그대로라 하류가 # 참조하는 내용은 안 바뀌고, 지문에 들어가는 값도 없다. # 다음 방문부터는 포기 재사용 갈래가 records 를 안 건드려 # (_records_changed=False) 위 jit_fresh 로 조용히 건너뛴다. _cine_fresh_decline = bool( cine_rec and cine_rec.get("declined") and not cine_rec.get("reused")) if _unchanged and not (_cine_fresh_success or _cine_fresh_decline): # 돈은 나갔는데(record 갱신) 산출은 그대로 — 새 asset 은 # 불필요하지만 지출이므로 상한 계수에 넣는다. jit_spent_same += 1 jit_regen_count += 1 logger.warning( "still_recipe %s: JIT 검증 — 지출 흔적(record 갱신) " "있으나 산출 동일, 영속 생략 (%d/%d)", tag, jit_regen_count, jit_regen_limit) primary_asset_by_tag[tag] = _prior.id continue jit_regen_count += 1 logger.warning( "still_recipe %s: JIT 검증 — 입력이 낡아 재생성 " "(%d/%d)", tag, jit_regen_count, jit_regen_limit) # ── 영속: 최종 산출(final_src — 변환 ON+성공=_cine, 그 외=_sel) # → scene ImageAsset primary ───────────────────────────────── final_path = scene_dir / f"{uuid.uuid4()}.png" final_path.parent.mkdir(parents=True, exist_ok=True) shutil.copy(final_src, final_path) # 2차 리뷰 H5: input(실첨부 전체) vs reference(entity 전용) 분리 from app.modules.pipeline.still_recipe import ( effective_prompt_used, fix_stage_won, partition_lineage_ids, ) attached_ids, entity_ref_ids = partition_lineage_ids(attached_refs) # 감사 링크 — nb2 롤 llm_call_log 구조키(still_id) 매칭 (BLOCKING-2: # save_single_scene_asset 기본 op(single_scene_image_gen)와 어긋남 교정) _cine_applied = bool(cine_rec and cine_rec.get("applied")) # (Codex GG46 R1 BLOCK-2) 수리 단계 승패의 단일 판정 — exact 호출 # 태그(winner_exact_multiroll_tag 내부)·생성 모델·prompt_used· # fix_applied·lineage 가 전부 이 값 하나로 갈린다. 수리가 최종이면 # 그 실제 직접 입력(선정 롤)은 등록 자산이 아니므로 unresolved # 채널용 구조 신원(sha256+파일명)을 만들어 둔다(cine source 관례 # 동형). _fix_won = fix_stage_won(record) _fix_src_desc: Optional[Dict[str, Any]] = None if _fix_won: import hashlib as _hashlib _sel_lab = str(record.get("selected") or "").strip().lower() _fix_src_path = recipe_dir / f"{tag}_{_sel_lab}.png" _fix_src_desc = { "role": "fix_source_roll", "file": _fix_src_path.name, "sha256": ( _hashlib.sha256( _fix_src_path.read_bytes()).hexdigest() if _fix_src_path.is_file() else None), } _gen_call_id = None _base_roll_call_id = None try: # Codex 8e70d4c0 H2(재리뷰): 승자 branch 의 **selected roll/ # fix exact** 호출 태그로 resolve — 패자 branch/다른 롤 거짓 # 링크 금지 (exact miss=None) from app.modules.pipeline.still_recipe import ( winner_exact_multiroll_tag, ) _base_roll_call_id = persistence_svc._resolve_generation_call_id( still_id, episode_id, operation_type="still_recipe_roll", multiroll_tag=winner_exact_multiroll_tag( winner_branch_tag, record), ) except Exception: # noqa: BLE001 — 감사 링크 실패 비치명 pass # (Codex safety-ladder BLOCK-2) fallback 성공 자산의 provenance 를 # 호출 SOT(llm_call_log)로 정정 — 승자 롤/fix 호출 metadata 에 # safety_ladder(비 primary 단계)가 있을 때만 값이 온다. 미발화· # 플래그 OFF 는 None = 기존 경로 byte-identical. _ladder_prov = persistence_svc.safety_ladder_call_provenance( _base_roll_call_id) if _cine_applied: # #108 (Codex R1 BLOCK-3): 최종 bytes 는 변환 호출이 만들었다 # — exact 만, miss=None. 롤 fallback 은 grok 산출을 nb2 호출에 # 잇는 거짓 링크(H2 의 exact-miss=None 계약과 정면 충돌)라 # 두지 않는다. 롤 링크는 review_notes.base_recipe 로 분리. try: _gen_call_id = persistence_svc._resolve_generation_call_id( still_id, episode_id, operation_type="still_cine_transform", multiroll_tag=f"still_{tag}_cine", ) except Exception: # noqa: BLE001 — 감사 링크 실패 비치명 pass else: _gen_call_id = _base_roll_call_id # #108 (Codex R1 BLOCK-3): 변환 적용 자산의 직접 입력은 grok 이 # 실제 본 원본 sel 1장뿐 — nb2 롤의 plate/char/prev 를 "직접 # 첨부"로 기록하면 실행과 기록이 어긋난다. 원본 sel 은 등록 # 자산이 아니므로 unresolved 채널(구조 신원=sha256+파일명)로 # 남기고, 롤 단계 실첨부·팩·호출 링크는 review_notes.base_recipe # 로 분리 보존한다. reference_image_ids(entity 구조 lineage, H5 # 의미 채널)는 유지 — 실첨부 채널이 아니라 "이 그림이 어느 # 엔티티를 그리는가"다. if _cine_applied: _direct_attached_ids: List[str] = [] _direct_attached_refs: List[Dict[str, Any]] = [] _direct_unresolved: List[Dict[str, Any]] = [{ "role": "cine_source_sel", "label": "SOURCE STILL", "file": (cine_rec or {}).get("source_file"), "sha256": (cine_rec or {}).get("source_sha256"), }] _prompt_used = cine_prompt _pack_provenance = cine_pack else: _direct_attached_ids = attached_ids _direct_attached_refs = attached_refs _direct_unresolved = unresolved_refs if _fix_won and _fix_src_desc: # (Codex GG46 R1 BLOCK-2) 최종 bytes 는 수리 i2i 산출 — # 실제 직접 입력은 [선정 롤 + critique 참조]다. 선정 롤만 # 구조 신원으로 병기한다. # ★2026-08-20 정정: 여기 attached 채널은 "이 샷이 본 참조 # 전체(롤 단계 포함)"다. 참조 선별(STILL_FIX_REF_GATE_ # ENABLED)이 켜지면 수리 호출에는 그중 지적이 요구한 것만 # (0장일 수도) 가므로, 수리 호출이 실제로 받은 부분집합은 # 아래 review_notes 의 still_recipe.fix_ref_gate 가 말한다. # 번호를 자산 신원으로 옮기지 않는 이유: 선별 번호가 가리키는 # 목록과 이 채널은 순서도 신원 표기도 달라(인물·소품 항목엔 # 파일 경로가 없다) 잘못 맞추면 캔버스 계보 선이 엉뚱한 # 자산을 가리킨다 — 과다 기재보다 나쁘다. _direct_unresolved = list(unresolved_refs) + [_fix_src_desc] # (Codex safety-ladder BLOCK-2) 사다리 회복 자산은 실제 성공 # 호출이 보낸 프롬프트(연화본 포함)가 prompt_used 다 — 조립 # 원문을 남기면 실행과 기록이 어긋난다. _prompt_used = ( (_ladder_prov or {}).get("user_prompt") or effective_prompt_used(record, prompt)) _pack_provenance = recipe_pack_version(_pack) asset = persistence_svc.save_single_scene_asset( { "id": str(uuid.uuid4()), "asset_type": "scene", "entity_id": None, "still_id": still_id, "episode_id": episode_id, "file_path": str(final_path), # 리뷰 NARROW-4: 변형 모드=선정 변형 전문(시각 접근 # provenance), legacy=base 그대로. base 는 record sidecar # (prompt 키)와 아래 review_notes 가 보존. # #108: 변환 적용이면 이 자산을 만든 프롬프트는 변환 # 문안이다 — 조립 전문은 review_notes.base_recipe·records. "prompt_used": _prompt_used, # #108: 최종 bytes 의 생성 모델을 기록 — 변환 적용=grok, # 미적용=롤 백엔드 실물(grok2 스위치 포함 — 이전엔 grok2 # 런도 gemini 로 남던 오기). (Codex GG46 R1 BLOCK-2) # G+G46 에서 수리가 최종이면 그 bytes 는 grok fix i2i # 산출이다 — nb2 백엔드여도 gemini 로 남기면 오기. # (Codex safety-ladder BLOCK-2) 사다리 회복 자산은 설정 # 플래그 추정이 아니라 실제 성공 호출의 모델이 SOT — # Gemini→Grok 교차 성공을 Gemini 로(역방향은 Grok 으로) # 남기는 오기를 막는다. 변환 적용 자산은 변환 모델 유지. "generation_model": ( settings.grok_image_model if _cine_applied else ((_ladder_prov or {}).get("model_name") or ( settings.grok_image_model if (getattr(settings, "still_image_backend", "nb2") == "grok2" or (gg46_judge_on and _fix_won)) else settings.gemini_image_model))), "width": None, "height": None, "status": "generated", # 실첨부 lineage (BLOCKING-2) — input_image_ids/캔버스 # edge SOT. #108: 변환 적용 시 이 채널은 grok 의 실제 # 입력(원본 sel)만 담는다. "actual_attached_image_ids": _direct_attached_ids, "actual_attached_refs": _direct_attached_refs, "unresolved_attached_refs": _direct_unresolved, "generation_call_id": _gen_call_id, "review_notes": json.dumps( { "still_recipe": { "selected": record.get("selected"), "totals": record.get("totals"), "ref_mode": record.get("ref_mode"), # (Codex GG46 R1 BLOCK-2) 존재 판정 아님 — # 재판정에서 진 fix 는 적용된 것이 아니다. "fix_applied": _fix_won, # 2026-08-20: 수리가 최종이면 그 호출이 실제로 # 받은 참조는 이 기록이 SOT 다 — 위 # actual_attached_refs 는 이 샷이 본 참조 # **전체**(롤 단계 포함)이고, 참조 선별이 켜지면 # 수리 호출에는 그중 일부만(0장일 수도) 간다. **({"fix_ref_gate": record["fix_ref_gate"]} if (_fix_won and record.get("fix_ref_gate")) else {}), "issues": (record.get("critique") or {}).get( "issues" ), }, # #108: 변환 provenance — OFF 면 키 부재 # byte-identical. 실패 fallback(원본 영속)도 # applied=False+error 로 자산에 남는다(오독 방지). **({"cine_transform": { "applied": _cine_applied, # 2026-08-20: 포기(검열 거부 누적)는 **결말**이라 # 다시 시도되지 않는다 — 이 표식이 없으면 # applied=False+error 만 남아 일시 실패와 구분이 # 안 되고, 자산은 봉인 뒤 고칠 창이 없다. **({"declined": True, "declined_reason": (cine_rec or {}).get( "declined_reason")} if (cine_rec or {}).get("declined") else {}), "pack": (cine_rec or {}).get("pack"), "fingerprint": (cine_rec or {}).get( "fingerprint"), "error": (cine_rec or {}).get("error"), "source_file": (cine_rec or {}).get( "source_file"), "source_sha256": (cine_rec or {}).get( "source_sha256"), }} if cine_on else {}), # (Codex safety-ladder BLOCK-2) 사다리 발화 근거 # — 어느 단계에서 어떤 모델이 최종 성공했는지. # 미발화=키 부재 byte-identical. **({"safety_ladder": { "stage": _ladder_prov["stage"], "generation_model": _ladder_prov["model_name"], "generation_call_id": _base_roll_call_id, }} if (_ladder_prov and not _cine_applied) else {}), # #108 (BLOCK-3): 롤 단계 provenance 분리 보존 — # 변환 적용 자산의 직접 채널에서 뺀 조립 팩· # 실첨부·롤 호출 링크는 여기가 담는다. **({"base_recipe": { "prompt_file_version": recipe_pack_version( _pack), "actual_attached_image_ids": attached_ids, "actual_attached_refs": attached_refs, "unresolved_attached_refs": unresolved_refs, "generation_call_id": _base_roll_call_id, # (Codex GG46 R1 BLOCK-2) 수리가 최종이면 # 변환 전 단계(base)의 실제 직접 입력에도 # 선정 롤이 있었다 — 구조 신원으로 병기. **({"fix_source": _fix_src_desc} if (_fix_won and _fix_src_desc) else {}), # (Codex safety-ladder BLOCK-2) 변환 밑의 # 롤이 사다리 회복분이면 여기 병기 — 변환 # 자산의 직접 provenance(cine)는 안 바꾼다. **({"safety_ladder": { "stage": _ladder_prov["stage"], "generation_model": _ladder_prov["model_name"], }} if _ladder_prov else {}), }} if _cine_applied else {}), }, ensure_ascii=False, ), "created_at": _now(), }, { # 배치 리뷰 NARROW-7 명시 계약: prompt_type/code_version # = **알고리즘 계보**(still_recipe_mode="v1" 레시피 # 파이프) 고정 — 팩 selector 가 아니다. 팩 provenance 는 # 아래 prompt_file_version 이 담당. "prompt_type": "still_recipe_v1", "code_version": "still_recipe_v1", # R3: 샷별 실제 사용 팩 (v1 하드코딩이 lane v3/complex # v4 샷의 provenance 를 오기하던 결함 교정) # #108: 변환 적용 자산은 변환 문안 팩 — 조립 팩은 # review_notes.base_recipe 로 분리(BLOCK-3). "prompt_file_version": _pack_provenance, # H5: entity 구조 lineage 만 — plate/conti/prev 는 # input_image_ids(actual_attached) 채널 전용 "reference_image_ids": json.dumps(entity_ref_ids), }, ) primary_asset_by_tag[tag] = asset.id # 후속 prev lineage 직결 scene_cp.mark_completed( still_id, { "asset_ids": [asset.id], "primary_id": asset.id, "primary_path": str(final_path), "recipe_tag": tag, }, ) done += 1 logger.info( "still_recipe %s: 완료 (selected=%s, fix=%s)", tag, record.get("selected"), bool(record.get("fix_prompt")), ) if jit_fresh or jit_regen_count or jit_no_record or jit_failed_tags: # #77-B 요약 — record없음>0 은 검증이 닿지 못한 완료 샷이 있다는 # 뜻(records.json 손상이면 전 샷이 여기 몰린다). 조용히 지나가지 # 않도록 한 줄로 남긴다. logger.info( "still_recipe JIT 검증 요약: 그대로 %d · 재생성 %d (그중 " "산출동일 %d) · record없음(검증불가 skip) %d · 실패 %d", jit_fresh, jit_regen_count, jit_spent_same, jit_no_record, len(jit_failed_tags)) if jit_regen_limit > 0 and jit_regen_count >= jit_regen_limit: # Codex BLOCK 3 경계: 정확히 상한으로 걷기가 끝나도 completed 로 # 봉인하지 않는다 — 상한 도달 = 사람 확인 요구가 합의된 의미다. latch = _jit_write_latch( recipe_dir, tripped_at_tag="(걷기 끝)", fresh=jit_fresh, regen=jit_regen_count, spent_same=jit_spent_same, no_record=jit_no_record, failed=list(jit_failed_tags)) raise StillJitRegenLimitExceeded( f"완료 샷 JIT 재생성이 상한({jit_regen_limit})에 닿은 채 " f"걷기가 끝났다 — completed 로 닫으면 확인 없이 지나간다. " f"래치 {latch} 를 남겼다 — 원인 확인 후 삭제하고 resume " f"(그대로 {jit_fresh}·재생성 {jit_regen_count}·산출동일 " f"{jit_spent_same}·record없음 {jit_no_record})") if jit_failed_tags: # Codex BLOCK 1: 완료 샷의 검증 실패는 옛 primary 가 남아 스텝 # 집계(primary 수·verify)에 안 잡힌다 — 여기서 스텝을 실패로 # 남기지 않으면 새 config_hash CP 가 completed 로 봉인돼 다음 # resume 이 whole-step SKIP, 낡은 샷이 영구 동결된다. 성공분은 # 샷 단위로 이미 영속됐으므로 resume 재시도는 실패분만 다시 본다. raise StillJitVerifyIncomplete( f"JIT 검증 대상 {len(jit_failed_tags)}샷이 실패했다: " f"{', '.join(jit_failed_tags[:8])}" f"{' …' if len(jit_failed_tags) > 8 else ''} — 옛 primary 가 " f"실패를 가리므로 스텝을 완료로 닫지 않는다. resume 이 " f"실패분만 다시 시도한다 (그대로 {jit_fresh}·재생성 " f"{jit_regen_count}·record없음 {jit_no_record})") if cine_declined_tags: # 포기는 미완이 아니라 확정된 결말이라 스텝을 막지 않는다 — # 다만 "변환 ON 완주"가 실은 몇 장 미변환이라는 사실은 남긴다. logger.warning( "cine 변환 검열 포기 %d샷(원본이 최종본): %s%s", len(cine_declined_tags), ", ".join(cine_declined_tags[:8]), " …" if len(cine_declined_tags) > 8 else "") if cine_failed_tags: # #108 (Codex R1 BLOCK-1): 실패 샷은 원본 fallback 으로 영속돼 # 집계에 안 잡힌다 — completed 봉인이면 resume whole-step SKIP # 으로 미변환이 영구 동결된다. 재시도 바퀴 비용 = 실패 변환만 # (~$0.03/샷, 건강 샷 전량 재사용 지출 0). raise StillCineTransformIncomplete( f"cine 변환 실패 {len(cine_failed_tags)}샷(원본 fallback 영속): " f"{', '.join(cine_failed_tags[:8])}" f"{' …' if len(cine_failed_tags) > 8 else ''} — 스텝을 완료로 " f"닫지 않는다. resume 재진입이 실패 변환만 다시 시도한다. " f"영구 실패로 판단되면 STILL_CINE_TRANSFORM_ENABLED=false " f"전환 여부는 사람이 결정한다.") if era_preserved_keys: # (era R2 BLOCK-1) 보존 그룹은 현재 계약 검증(모델·팩·context· # conti 지문 대조)이 조사 성공 방문까지 미뤄진 상태 — completed # 봉인이면 다음 resume 이 whole-step SKIP 해 검증 안 된 배경과 # 그 밑의 진짜 drift 가 영구 동결된다. raise StillEraPreserveIncomplete( f"era 조사 미성립으로 기존 배경을 보존한 그룹 " f"{len(era_preserved_keys)}건: " f"{', '.join(era_preserved_keys[:8])}" f"{' …' if len(era_preserved_keys) > 8 else ''} — 현재 계약 " f"검증이 미뤄졌으므로 스텝을 완료로 닫지 않는다. resume 이 " f"보존 그룹의 조사만 재시도한다(성공분 재사용 지출 0). 조사가 " f"계속 실패하면 ERA_RESEARCH_ENABLED 전환 여부는 사람이 " f"결정한다.") progress.update( "레시피 스틸 생성 완료", (total if execution_scope is None else exec_total) + 2, (total if execution_scope is None else exec_total) + 2, ) return done