"""i2i 시네마틱 변환 스테이지 (2026-08-13 #108) — sel 확정 후 grok 2.0 변환. 사용자 확정 구조(하이브리드): 최종 스틸 = nb2 생성·판정·수정(sel 확정) → grok 2.0 i2i "시네마틱 재구성" → 변환본이 scene ImageAsset primary 로 영속. 원본 `_sel.png` 은 그대로 보존되어 prev 체인 앵커(기하 SOT)로 남는다 — 변환은 최종 룩 패스일 뿐 체인 참조 재료가 아니다. 계약: · 문안 = still_recipe 팩 단일 스템 `cine_transform` (CINE_TRANSFORM_PROMPT_VERSION) — 짧은 범용 지시 하나, 원본 조립 프롬프트·시나리오 고유명사 불사용. 화풍 교체 = 팩 버전 교체. · 지문 = 계약 버전 + 문안 스템 내용 해시 + 모델 + 원본 sel bytes — sel 재생성·문안 개정·모델 교체 어느 것이든 변환을 다시 이끈다. · 재사용 = 지문 일치 + applied + 산출 파일 실재 → 유료 호출 0, records 무변경 (#77-B `_jit_tag_snapshot` 지출 탐지가 거짓 지출을 읽지 않는 전제 — 저장 record 를 그대로 돌려준다). · 실패 = 원본 fallback + 실패 기록(records) — 샷은 원본으로 영속돼 체인(prev)을 막지 않는다. ★재시도는 JIT 가 아니라 **스텝 미봉인**이 이끈다(Codex R1 BLOCK-1): 완료+config 일치 스텝은 whole-step SKIP 이라 "다음 방문"이 오지 않는다 — 걷기 끝에 서비스가 StillCineTransformIncomplete 를 raise 해 스텝을 완료로 닫지 않고, resume 재진입(전 샷 재사용 지출 0)이 실패 변환만 다시 산다. 예외: ImageCallBudgetExceeded 는 삼키지 않는다 — 예산 브레이크를 fallback 으로 눌러 버리면 "변환된 최종본" 오독의 조용한 미변환 완주가 된다(전파 → 샷 실패 격리 → 스텝 실패 → 운영자 확인). · **검열 포기**(2026-08-19 사용자 결정) — 위 되풀이가 끝나지 않는 경우가 실측으로 나왔다. 같은 원본이 xAI 검열에 나흘간 일곱 번 전부 막혔고(S49sh10·S73sh6), 막는 것이 지시문이 아니라 **그림 내용**이라 다시 보낼수록 같은 자리에서 막힌다. 게다가 **거부당해도 요금이 나간다**(거부 응답에 `cost_in_usd_ticks` 가 붙는다). 그래서 같은 지문에서 검열 거부가 `still_cine_moderation_give_up_after` 회 쌓이면 그 변환을 포기로 기록하고(`declined`) 원본을 최종본으로 확정한다 — 다음 방문은 유료 호출 없이 그 기록을 돌려주고, 서비스는 그 샷을 미완으로 세지 않아 스텝이 닫힌다. 원본 sel 이 다시 만들어지면 지문이 달라져 저절로 한 번 더 시도한다(포기는 이 입력에 대한 것이지 이 샷에 대한 영구 선고가 아니다). · 쓰기 순서 = 산출 파일 durable(원자 쓰기) 후 records 기록 — 크래시 창에서는 기록 없는 파일이 남고, 다음 방문이 재변환한다(재지출 ~$0.03 1회 허용이 기록 없는 자산 신뢰보다 싸다). 경계(알려진 것): · prev 앵커 fallback(still_recipe_service — recipe `_sel.png` 부재 시 DB primary 사용)은 변환 ON 완주본에서 변환본을 앵커로 집는다. 변환은 장소·인물·순간 보존 계약이라 기하 권위는 유지되지만 룩이 섞인다 — 정상 걷기(스토리 순서, sel 실재)에서는 닿지 않는 경로. """ from __future__ import annotations import hashlib import logging from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, Optional from app.core.image_call_budget import ImageCallBudgetExceeded from app.modules.llm.image_moderation import is_moderation_error logger = logging.getLogger(__name__) # 포기 기준 기본값 — 설정이 없거나 못 읽으면 이 값. 0 이하면 포기하지 # 않는다(종전 동작). CINE_MODERATION_GIVE_UP_AFTER_DEFAULT = 2 def _is_terminal_provider_error(exc: BaseException) -> bool: """제공자가 **결말을 지었다**고 말한 실패인가 (재시도·다시 가져오기 불가). 접수 기반 제공자에서만 뜻이 있다. 다시 가져오기할 산출이 아예 없다는 뜻이므로 접수 표식을 걷어야 다음 방문이 새로 살 수 있다 — 안 걷으면 없는 작업을 영원히 가져오려 든다. """ try: from app.modules.llm.reve_image_client import ReveTerminalError except Exception: # noqa: BLE001 — client 가 없는 판이면 해당 없음 return False if not isinstance(exc, ReveTerminalError): return False # ★접수 **기록**만 실패한 것은 결말이 아니다 (2026-08-26 Codex PR#4 재리뷰). # 작업은 fal 에 살아 있고 번호도 안다. 이것을 결말로 세면 아래 마무리가 # `if not _terminal_fail` 에서 접수 신원 복사를 통째로 건너뛰어, # **번호 없는 실패 기록만 남고 다음 걷기가 새로 보낸다**($0.25 이중). if getattr(exc, "error_type", "") == "submit_record_failed": return False return True def _give_up_after() -> int: from app.core.config import settings raw = getattr( settings, "still_cine_moderation_give_up_after", CINE_MODERATION_GIVE_UP_AFTER_DEFAULT) try: return int(raw) except (TypeError, ValueError): return CINE_MODERATION_GIVE_UP_AFTER_DEFAULT # 문안 스템 이름 — still_recipe 팩 안 단일 스템 (팩 v19 도입). CINE_TRANSFORM_STEM = "cine_transform" # ★재료 절 스템 — **이것도 나가는 문안의 일부다** (2026-08-27, 과제 #94). # 재료를 넘기는 판은 두 스템을 이어 붙여 보낸다 # (`build_cine_transform_prompt`: base + 이 절). 종전에는 앞엣것만 # 지문에 접혀, 이 절의 문안을 고치면 나가는 글은 바뀌는데 지문이 안 # 움직여 **완주한 샷이 옛 그림 그대로 통과**했다. CINE_STAGE_DIRECTION_STEM = "cine_stage_direction" # 참조 라벨 — 파일럿(grok_cine_batch) 실측 계약 그대로. CINE_SOURCE_LABEL = "SOURCE STILL" # 결말을 못 본 옛 요청을 몇 건까지 들고 있나 — **최근 것만 남는 기록**이다. # 이보다 오래된 것은 밀려난다(덧붙이기 전용 대장이 아니다). _UNRESOLVED_LOG_LIMIT = 20 # 변환 계약 버전 — 참조 구성·산출 규약(1장 변환·_cine 파일)이 바뀌면 bump. CINE_CONTRACT_VERSION = "cine_v1" def _verify_on() -> bool: from app.core.config import settings return bool(getattr(settings, "still_cine_verify_enabled", False)) def _verify_rounds() -> int: """설정 회차 — **이제 판정을 안 바꾼다** (2026-08-27, #92 이중 모델). ★옛 뜻: 「같은 모델에게 몇 번 물을까」. 지금 계약은 「두 모델에게 각 한 번씩」이고 `cine_verify.verify_models()` 가 언제나 짝 전체를 돌려준다 — 이 값은 모델 수를 못 바꾼다. ★그런데 함수는 남겨 둔다: 호출부가 `rounds=` 로 넘기고 있고, 인자를 지우면 「무엇을 무시하는지」가 코드에서 사라진다. 검사를 끄는 것은 `still_cine_verify_enabled` 플래그 하나다. """ from app.core.config import settings from app.modules.pipeline.cine_verify import normalize_rounds return normalize_rounds(getattr(settings, "still_cine_verify_rounds", 2)) def _verify_is_current(prior_verify: Any) -> bool: """저장된 판정이 **지금 계약의 것**인가. 판정 문안·**모델 짝**·물리 모델명·잡는 축이 바뀌면 옛 판정은 다른 질문의 답이다. 이 갈래는 그림을 다시 사지 않고 판정만 다시 산다 — 그래서 계약 sha 를 변환 지문(`cine_fingerprint`)이 아니라 여기서만 본다. ★**회차는 이제 계약이 아니다** (2026-08-27). 짝 전체가 언제나 판정하고 `verify_contract_sha` 도 회차 대신 짝 크기를 접는다 — 설정을 2→1 로 바꿔도 다시 판정하지 않는다(나가는 물음이 한 글자도 안 바뀐다). """ if not isinstance(prior_verify, dict): return False if not _verify_on(): return False # 검사가 꺼졌으면 「지금 계약의 판정」이 아니다 from app.modules.pipeline.cine_verify import verify_contract_sha # 회차도 신원이다 — 2→1 은 「둘 다 동의할 때만 기각」을 「한 번 보고 # 기각」으로 바꾸므로 다른 판정이다(Codex BLOCK-3). return prior_verify.get("contract") == verify_contract_sha( _verify_rounds()) def _apply_verify( rec: Dict[str, Any], *, tag: str, source_png: bytes, result_png: bytes, project_config: Optional[Dict[str, Any]] = None, context: Optional[Dict[str, Any]] = None, ) -> None: """산출을 재고 기각 여부를 rec 에 접는다. 2026-08-26 계약 교체 — **변환본이 사진으로 성립하는가**만 본다. 손가락 여섯 개·공중에 뜬 사람·문 여덟 개짜리 차 같은 파탄만 잡고, 성립하면 **무조건** 변환본을 최종본으로 쓴다. 빛이 바뀌고 카메라가 옮겨 간 것은 이 단계가 시켜서 하는 일이라 기각 근거가 아니다. 자세한 근거는 `cine_verify` 머리말. ★기각은 **실패가 아니라 확정된 결말**이다 — `applied=False` + `rejected=True` 로 두어 호출측이 원본을 최종본으로 확정하고 스텝을 닫게 한다. 실패로 세면 스텝이 영영 안 닫혀 재개마다 앞 단계가 다시 돌고 연쇄 재생성이 난다(검열 포기가 같은 이유로 이 모양이다). ★판정 자체가 못 돌았으면(판정기 오류) **기각하지 않는다** — 이미 산 그림을 판정기 사정으로 버리면 돈만 나가고 결과가 나빠진다. """ if not _verify_on(): return from app.modules.pipeline.cine_verify import verify_cine_result meta = dict(context or {}) v = verify_cine_result( result_png=result_png, source_png=source_png, rounds=_verify_rounds(), project_config=project_config, opik_metadata={k: meta[k] for k in ("project_id", "episode_id", "still_id", "scene_index", "shot_index") if k in meta}, ) rec["verify"] = v if v.get("ok"): return broken = v.get("broken") or [] rec["applied"] = False rec["rejected"] = True rec["rejected_reason"] = "implausible:" + ",".join(broken) logger.warning( "cine_transform %s: 변환본이 사진으로 성립하지 않는다(%s) — 원본을 " "최종본으로 확정한다. %s", tag, ", ".join(broken), "; ".join(f"{c}={(v.get('findings') or {}).get(c, {}).get('note', '')}" for c in broken)[:300]) # ── 사람의 거절 (2026-09-19) ───────────────────────────────────────── #: 샷 단위 기록 — `records["{tag}::cine_human"]`. 값은 아래 helper 가 쓴다. CINE_HUMAN_SUFFIX = "::cine_human" CINE_HUMAN_KEEP_ORIGINAL = "keep_original" def cine_human_key(tag: str) -> str: return f"{tag}{CINE_HUMAN_SUFFIX}" def human_keep_original(records: Any, tag: str) -> Optional[Dict[str, Any]]: """사람이 이 샷의 변환본을 거절했나 — 그 기록, 없으면 None.""" rec = records.data.get(cine_human_key(tag)) if isinstance(rec, dict) and rec.get("decision") == CINE_HUMAN_KEEP_ORIGINAL: return rec return None def record_human_keep_original( records: Any, tag: str, *, reason: str, rejected_cine: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """사람이 변환본을 거절했다 — 이 샷은 **변환 전 원본**을 최종본으로 쓴다. 실측(컨트리로드 2판, 사용자 육안): 변환이 로봇 얼굴을 사람 얼굴로, 파란 홀로그램을 불투명한 실물로 바꾸고, 사람을 둘로 늘리고, 팔을 셋으로 만들었다 — 변환 전 원본은 다섯 샷 모두 멀쩡했다. ★`::cine` 기록에 rejected 를 손으로 심지 않는다. 그 기각은 검사가 꺼져 있거나 판정 계약이 바뀌면 **복권된다**(`resolve_or_run_cine_transform` 의 rejected 갈래) — 판정기의 기각과 사람의 거절은 다른 것이라 섞지 않는다. ★**샷 단위**다. 샷을 다시 만들어 원본이 바뀌어도 유지된다 — 사람이 본 것은 「이 샷에서 변환이 망가뜨린다」이고, 새 변환이 괜찮다는 보장이 없다. 되돌리려면 이 키를 지운다. """ rec: Dict[str, Any] = { "decision": CINE_HUMAN_KEEP_ORIGINAL, "reason": reason, "at": datetime.now(timezone.utc).isoformat(), # 무엇을 보고 거절했나 — 거절한 변환본의 신원 "rejected_cine": { k: (rejected_cine or {}).get(k) for k in ("file", "staged_sha256", "source_sha256", "fingerprint", "provider", "model")}, } records.data[cine_human_key(tag)] = rec records.save() return rec def cine_record_key(tag: str, slot: str = "") -> str: """변환 기록의 **키**. 슬롯이 있으면 뒤에 붙인다. ★한 자리에 둔다 — 이 키는 `records` 에 쓰는 곳과 **발송 맥락을 씌우는 소유자 자리** 두 곳이 쓴다. 두 번 적으면 한쪽만 고쳐져 「본체 생성」과 「뒤의 변환」이 같은 이름으로 섞인다 (2026-09-20 Codex BLOCK 2). ★장부 쪽에서 규칙을 **다시 짐작하지 않는다** — 여기서 받아 쓴다. """ return f"{tag}::cine" + (f"::{slot}" if slot else "") def cine_output_path(recipe_dir: Path, tag: str, slot: str = "") -> Path: """변환본 관례 경로 — `_sel.png` 과 나란한 `{tag}_cine.png`. ★``slot`` — 제공자 자리. 빈 문자열이면 **주 제공자**이고 종전 경로 그대로다 (byte-identical). 검열 대체처럼 **다른 제공자**가 같은 샷을 변환할 때만 `{tag}_cine_{slot}.png` 로 갈라 둔다. ★[2026-09-09 Codex BLOCK 1] 안 가르면 대체 성공이 주 제공자의 거절 기록을 **덮는다.** 그러면 같은 샷을 다시 방문할 때 주부터 또 호출하고, 대체 지문은 주와 안 맞아 재사용도 못 해 **둘 다 다시 산다.** 제공자별 거절 누적 (`moderation_refusals`)도 사라져 「둘 다 거절 후 확정」이 안 쌓인다. """ suffix = f"_{slot}" if slot else "" return recipe_dir / f"{tag}_cine{suffix}.png" def cine_fingerprint( *, stem_content_hash: str, model: str, sel_bytes: bytes, ) -> str: """변환 1건의 재사용 지문 v1 — 문안(exact-stem)·모델·원본 bytes. ★옛 판이다. provider·endpoint 칸이 없어 **모델 문자열만** 적었는데, 그 시절 변환은 전부 Grok 이었으므로 v1 record 는 「그 Grok 산출」을 뜻한다. 지우지 않는 이유는 완주본을 다시 사지 않기 위해서다 (dual-read — `_fingerprint_matches` 참조). """ sel_sha = hashlib.sha256(sel_bytes).hexdigest() return hashlib.sha256( f"{CINE_CONTRACT_VERSION}|{stem_content_hash}|{model}|{sel_sha}" .encode("utf-8") ).hexdigest() # 지문 판 (2026-08-25 provider 교환 도입) — 새로 쓰는 record 는 전부 이 판. CINE_FINGERPRINT_VERSION = 2 def _stop_gate(where: str = "") -> None: """지금 이 결과를 **확정해도 되는지** 확인한다. 스텝 경계에서 걸어 둔 손잡이를 그대로 부른다. 표가 안 걸린 직접 호출 에서는 아무 일도 안 한다(no-op). ★부르는 자리가 둘이다 (2026-08-26 Codex 2차 재리뷰 BLOCK-2). ⓐ 제공자 결과를 받은 직후 — 그림은 파일로 남기고 확정만 멈춘다. ⓑ **판정이 끝난 뒤 기록을 쓰기 직전** — 판정 VLM 이 도는 동안 주인이 바뀔 수 있고, 그러면 옛 worker 가 뒤늦게 `applied`/`rejected` 를 남의 주행 자리에 적는다. ⓐ 만으로는 그 창이 안 닫힌다. """ from app.core.image_call_budget import get_current_stop_check _g = get_current_stop_check() if _g is None: return try: _g() except Exception: logger.warning("cine_transform: 중단 확인에 걸렸다 — %s", where or "안전 지점") raise def _verify_and_gate( rec: Dict[str, Any], *, tag: str, source_png: bytes, result_png: bytes, project_config: Any, context: Dict[str, Any], where: str, run_verify: bool = True, ) -> None: """판정하고 → **아직 내가 주인인지 보고** 나온다. 기록은 호출부가 한다. ★두 동작을 한 함수로 묶는 이유 (2026-08-26 Codex 3차 재리뷰 BLOCK-1): 판정을 부르는 자리가 다섯 곳인데 확인을 손으로 하나씩 붙였더니 **두 곳을 빠뜨렸다**(이미 쓰던 그림을 새 계약으로 다시 보는 갈래, 버렸던 그림을 되살려 다시 보는 갈래). 판정 VLM 은 오래 도는데 그동안 주인이 바뀌면 옛 worker 가 남의 자리에 결과를 적는다. 그래서 `_apply_verify` 를 직접 부르지 말고 **반드시 이 함수를 지나게** 한다. 빠뜨릴 자리를 없애는 것이 기억에 기대는 것보다 낫다. """ if run_verify: _apply_verify(rec, tag=tag, source_png=source_png, result_png=result_png, project_config=project_config, context=context) _stop_gate(where) def cine_fingerprint_v2( *, prompt: str, stem_content_hash: str, identity: Dict[str, str], sel_bytes: bytes, ) -> str: """변환 1건의 재사용 지문 v2 — 문안·스템·제공자 신원·원본. v1 에 두 가지가 더해진다. ⓐ `provider|endpoint`. 같은 문안·같은 원본이라도 **다른 제공자가 만든 그림은 다른 산출**이다. 신원을 안 넣으면 제공자를 바꿔도 옛 산출이 「지금 계약의 것」으로 통한다. ⓑ **실제 나가는 문안**. 연출 재료가 들어오면서 문안이 **샷마다 달라졌기** 때문이다 — 스템만 보면 카메라 지시가 바뀌어도 지문이 안 움직여, 다른 지시로 만든 그림이 「같은 조건」으로 읽힌다. ★스템 해시는 **빼지 않는다**. 문안 sha 로 갈음하면 「팩 교체 = 전 샷 재변환」이 호출부가 스템에서 문안을 만든다는 전제에 얹히게 된다. 그 전제가 언젠가 깨지면 화풍을 바꿔도 옛 그림이 그대로 남는다 — v1 이 막던 실패라 그대로 지킨다(둘을 같이 접는 값은 0 이다). ## 원본은 **바이트 그대로** 접는다 — 의도한 비용 정책이다 `sel_bytes` 는 PNG 파일의 날바이트다. 그림이 눈에 똑같아 보여도 메타데이터가 다르거나 다시 인코딩했으면 지문이 움직이고, 그러면 변환 요금이 한 번 더 나간다 (2026-08-26 Codex 2차 재리뷰 지적). 그래도 바이트로 접는 이유는, 「같은 그림인가」를 값싸고 **틀리지 않게** 판정할 방법이 이것뿐이기 때문이다. 픽셀만 비교하면 원본이 실제로 조금 달라진 경우를 같은 것으로 읽어 옛 변환본을 최종본으로 통과시킨다 — 잘못 아끼는 쪽의 손해가 잘못 쓰는 쪽보다 크다. ★대신 상류가 **같은 그림을 쓸데없이 다시 쓰지 않는 것**이 전제다. 원본 sel 이 내용은 같은데 바이트만 달라지는 일이 실제로 관찰되면, 그것은 여기서 막을 일이 아니라 상류에서 고칠 일이다. """ sel_sha = hashlib.sha256(sel_bytes).hexdigest() prompt_sha = hashlib.sha256((prompt or "").encode("utf-8")).hexdigest() basis = "|".join(( CINE_CONTRACT_VERSION, f"fp{CINE_FINGERPRINT_VERSION}", stem_content_hash, prompt_sha, identity.get("provider", ""), identity.get("endpoint", ""), identity.get("model", ""), sel_sha, )) return hashlib.sha256(basis.encode("utf-8")).hexdigest() def _fingerprint_matches( prior: Any, *, fp_v2: str, fp_v1: str, identity: Dict[str, str], ) -> bool: """저장된 record 가 **지금 조건의 것**인가 — 두 판을 나란히 읽는다. provider 칸을 그냥 지문에 더하면 **완주 판의 변환이 전량 재실행**된다 (v1 record 의 지문이 하나도 안 맞으므로). 그래서 판 번호를 보고 갈린다: v2 record 그대로 v2 지문과 댄다 v1 record 지금 선택이 **정확히 옛 Grok** 일 때만 지금 것으로 읽는다 ★legacy 갈래에서 records 를 **1비트도 쓰지 않는다** — 판 번호를 붙여 주려고 기록을 건드리면 재사용 바퀴가 지출로 세어진다(#77-B). ★소급 동등 처리 금지: 제공자 교체는 산출 계약이 실제로 다르다. 조금이라도 다르면 재생성이 맞다. """ if not isinstance(prior, dict): return False if prior.get("fingerprint_version") == CINE_FINGERPRINT_VERSION: return prior.get("fingerprint") == fp_v2 from app.modules.pipeline.cine_provider import is_legacy_identity return (is_legacy_identity(identity) and prior.get("fingerprint") == fp_v1) def resolve_or_run_cine_transform( *, tag: str, sel_path: Path, recipe_dir: Path, records: Any, client: Any, prompt: str, model: str, stem_content_hash: str, pack: str, context: Dict[str, Any], project_config: Optional[Dict[str, Any]] = None, identity: Optional[Dict[str, str]] = None, slot: str = "", ) -> Dict[str, Any]: """재사용 우선 변환 — 반환 record 로 최종 영속 원본이 갈린다. ★``slot`` — 제공자 자리(기본 "" = 주 제공자, 종전과 byte-identical). 검열 대체처럼 다른 제공자가 **같은 샷**을 변환할 때 이것을 주면 저장 키와 파일이 갈려, 주의 거절 기록과 대체의 결과가 **각각 보존**된다. 반환: records 저장본과 동일 내용 + transient `reused`(스냅샷 제외 키 — _jit_tag_snapshot 이 걸러낸다). applied=True 면 호출자가 `file` 을 최종 영속 원본으로 쓰고, False 면 원본 sel 로 fallback. `identity` = 제공자 신원(provider·endpoint·model). 생략하면 지금 설정 에서 푼다. `model` 은 **옛 지문(v1) 계산과 record 표시용**이라 호출부는 신원과 같은 모델 문자열을 넘겨야 한다 — 어긋나면 legacy 갈래가 엉뚱한 지문을 대게 된다. """ sel_bytes = sel_path.read_bytes() if identity is None: from app.modules.pipeline.cine_provider import cine_provider_identity _ident: Dict[str, str] = cine_provider_identity() else: _ident = identity fp_v1 = cine_fingerprint( stem_content_hash=stem_content_hash, model=model, sel_bytes=sel_bytes) fp = cine_fingerprint_v2( prompt=prompt, stem_content_hash=stem_content_hash, identity=_ident, sel_bytes=sel_bytes) def _is_current(p: Any) -> bool: return _fingerprint_matches( p, fp_v2=fp, fp_v1=fp_v1, identity=_ident) source_sha = hashlib.sha256(sel_bytes).hexdigest() rec_key = cine_record_key(tag, slot) out_path = cine_output_path(recipe_dir, tag, slot) prior = records.data.get(rec_key) if ( isinstance(prior, dict) and _is_current(prior) and prior.get("applied") is True and out_path.exists() ): if not _verify_on() or _verify_is_current(prior.get("verify")): # 재사용 — 저장 record 그대로(값 재계산·시간 갱신 금지: 재사용 # 바퀴에 record 가 1비트라도 움직이면 #77-B 가 지출로 센다). return {**prior, "reused": True} # 검사를 **나중에 켰거나 판정 계약이 바뀐** 경우 — 그림은 다시 사지 # 않고 판정만 산다. 실제 지출(VLM 1~2콜)이라 record 가 움직이는 것이 # 맞다. ★판정기를 고쳤다고 유료 이미지를 다시 사면 안 된다. rec_v = {k: v for k, v in prior.items()} _verify_and_gate(rec_v, tag=tag, source_png=sel_bytes, result_png=out_path.read_bytes(), project_config=project_config, context=context, where="다시 본 판정을 확정하기 직전") records.data[rec_key] = rec_v records.save() return {**rec_v, "reused": True} if ( isinstance(prior, dict) and _is_current(prior) and prior.get("rejected") is True and out_path.exists() ): # 기각한 변환 — 같은 입력에 **그림을 다시 사지 않는다**. 다만 # (2026-08-25 Codex BLOCK-2) 그대로 돌려주기만 하면 기각이 영구가 # 된다: 검사를 꺼도 이미 산 그림이 안 쓰이고, 판정 계약을 고쳐도 # 다시 볼 길이 없다. 검사 상태와 판정 신원을 보고 갈린다. if _verify_is_current(prior.get("verify")): logger.info( "cine_transform %s: 사진으로 성립하지 않아 버린 변환 — " "원본 유지, 호출 없음 (%s)", tag, prior.get("rejected_reason") or "") return {**prior, "reused": True} # 검사를 껐거나 판정 계약이 바뀌었다 — **복권 가능 상태로 되돌린 # 뒤** 다시 본다. 되돌리지 않으면 새 판정이 통과해도 기각 표식이 # 남아 그림이 영영 안 쓰인다. rec_r = {k: v for k, v in prior.items() if k not in ("rejected", "rejected_reason", "verify")} rec_r["applied"] = True if not _verify_on(): logger.info( "cine_transform %s: 검사가 꺼져 있어 버렸던 것을 되살린다 — " "이미 요금이 나간 변환본을 최종본으로 쓴다 (호출 없음)", tag) _verify_and_gate(rec_r, tag=tag, source_png=sel_bytes, result_png=out_path.read_bytes(), project_config=project_config, context=context, where="되살린 변환을 확정하기 직전", run_verify=_verify_on()) records.data[rec_key] = rec_r records.save() return {**rec_r, "reused": True} if ( isinstance(prior, dict) and _is_current(prior) and prior.get("staged") is True and prior.get("applied") is not True and out_path.exists() ): # ★요금은 이미 나갔고 그림도 디스크에 있는데 **확정을 못 하고 끝난** # 경우다(정지·락 상실·크래시). 제공자를 안 가리고 다시 가져오는 자리 — # 접수 hook 이 있는 reve 든 동기 왕복이든 여기서 걸린다. # # ★파일이 그때 그 파일인지 **바이트로 확인한다.** 지문만 맞으면 # 되는 게 아니다 — 같은 자리에 다른 그림이 덮여 있으면 요금이 # 나간 적 없는 그림을 확정해 버린다. staged_bytes = out_path.read_bytes() staged_sha = str(prior.get("staged_sha256") or "") if staged_sha and hashlib.sha256(staged_bytes).hexdigest() == staged_sha: # `staged`·`pending` 은 둘 다 「아직 확정 못 함」 표식이다 — # 확정하면서 같이 걷는다. 접수 번호·조회 주소는 남겨 둔다(감사용). rec_s = {k: v for k, v in prior.items() if k not in ("staged", "pending")} rec_s["applied"] = True rec_s["recovered"] = True _verify_and_gate(rec_s, tag=tag, source_png=sel_bytes, result_png=staged_bytes, project_config=project_config, context=context, where="남겨 둔 변환을 확정하기 직전", run_verify=_verify_on()) records.data[rec_key] = rec_s records.save() logger.info( "cine_transform %s: 확정을 못 하고 끝난 변환을 되찾았다 — " "다시 요청하지 않았다(요금 없음)", tag) return {**rec_s, "reused": False} # ★해시가 다르다고 곧바로 새로 요청하지 않는다 (2026-08-26 Codex # 2차 재리뷰 BLOCK-1). 접수 번호가 남아 있으면 **무료로 다시 가져올 # 수 있다** — 아래 접수 다시 가져오기 갈래가 이어서 잡는다. 그 갈래는 # `pending` 을 보므로 여기서 표식을 지우면 안 된다. logger.warning( "cine_transform %s: 남겨 둔 변환 파일이 기록의 해시와 다르다 — " "확정하지 않는다 (접수 번호=%s)", tag, prior.get("request_id") or "없음") if ( isinstance(prior, dict) and _is_current(prior) and prior.get("declined") is True ): # 포기한 변환 — 같은 입력에 다시 돈을 쓰지 않는다. 재사용과 같은 # 규칙으로 저장 record 를 그대로 돌려준다(거짓 지출 방지). logger.info( "cine_transform %s: 검열로 포기한 변환 — 원본 유지, 호출 없음", tag) return {**prior, "reused": True} if ( isinstance(prior, dict) and _is_current(prior) and prior.get("submission_unknown") is True ): # ★접수됐는지 **모르는** 채로 끝난 변환. 자동으로 다시 보내지 않는다 — # fal 이 받아서 요금이 나갔을 수 있고, 다시 보내면 같은 이미지에 # 요금이 두 번 나간다. 사람이 fal 대시보드에서 접수 여부를 보고 # 정한다: 안 됐으면 이 칸을 지우고 다시 걷고, 됐으면 그 번호를 # `request_id` + `pending` 으로 넣으면 결과 조회 경로가 무료로 가져온다. logger.warning( "cine_transform %s: 접수 여부를 모르는 변환이 남아 있다 — 자동으로 " "다시 보내지 않는다(요금 이중 방지). 사람이 확인해야 한다 (%s)", tag, prior.get("submission_unknown_reason") or "") return {**prior, "reused": True} # ── 접수됐는데 결말을 못 본 작업 — **다시 사지 않고 다시 가져온다** ── # 접수 기반 제공자(fal queue)는 submit 1회가 유료 작업 1건이다. 크래시나 # 응답 유실로 결말을 못 본 것을 그냥 새로 보내면 **같은 작업을 두 번 # 산다**($0.25). 동기 왕복($0.03)에서 감수하던 1회 재지출과 값이 다르다. if ( isinstance(prior, dict) and prior.get("pending") is True and prior.get("request_id") and _is_current(prior) and hasattr(client, "fetch_submitted") ): rid = str(prior.get("request_id") or "") rec_p = {k: v for k, v in prior.items()} try: import inspect client.set_context(**context) # 접수 때 받아 둔 조회 주소를 넘긴다 — 이 프로세스가 접수한 것이 # 아니면 인스턴스에 아무것도 없어 조립에 기대게 되고, 그 조립을 # 틀려 405 를 받은 적이 있다. 그 주소를 안 받는 제공자면 뺀다. _fetch_kw: Dict[str, Any] = { "prompt": prompt, "labeled_references": [(CINE_SOURCE_LABEL, sel_bytes)], } _params = inspect.signature(client.fetch_submitted).parameters if "status_url" in _params: _fetch_kw["status_url"] = str(prior.get("status_url") or "") _fetch_kw["response_url"] = str( prior.get("response_url") or "") png, elapsed_ms = client.fetch_submitted(rid, **_fetch_kw) except Exception as exc: # noqa: BLE001 rec_p["error"] = f"{type(exc).__name__}: {exc}"[:500] if _is_terminal_provider_error(exc): # 그 작업은 결말이 났고 결과가 없다 — 접수 표식을 걷어야 # 다음 방문이 새로 살 수 있다. 안 걷으면 없는 작업을 영원히 # 가져오려 든다. rec_p.pop("pending", None) rec_p.pop("request_id", None) logger.warning( "cine_transform %s: 접수한 변환이 결과 없이 끝났다 — " "접수 표식을 걷는다 (%s)", tag, rec_p["error"]) else: logger.warning( "cine_transform %s: 접수한 변환(%s) 가져오기 실패 — 원본 " "fallback, 다음 방문에 다시 가져온다 (%s)", tag, rid, rec_p["error"]) # ★조회가 길게 돌다 실패했다면 그 사이 주인이 바뀌었을 수 있다. # 여기서 그냥 쓰면 새 주인의 기록을 덮고 접수 표식까지 걷어 # 버린다. 이미 남아 있는 pending 기록은 그대로 두는 편이 안전하다. _stop_gate("가져오기 실패를 적기 직전") records.data[rec_key] = rec_p records.save() return {**rec_p, "reused": False} from app.modules.pipeline.multiroll_gemini import atomic_write_bytes atomic_write_bytes(out_path, png) # ★새로 요청한 경로와 **같은 확인**을 여기도 둔다. 접수한 작업을 # 다시 가져오는 동안 조회가 길게 돌 수 있고, 그 사이 정지가 걸리거나 락을 # 놓칠 수 있다. 그림은 위에서 파일로 남겼으니 잃지 않고, **확정만** # 멈춘다. _stop_gate("접수한 변환을 되찾은 직후") rec_p.pop("pending", None) rec_p.pop("staged", None) rec_p["applied"] = True rec_p["file"] = out_path.name rec_p["staged_sha256"] = hashlib.sha256(png).hexdigest() rec_p["latency_ms"] = int(elapsed_ms) rec_p["recovered"] = True rec_p.pop("error", None) _verify_and_gate(rec_p, tag=tag, source_png=sel_bytes, result_png=png, project_config=project_config, context=context, where="되찾은 변환을 확정하기 직전") records.data[rec_key] = rec_p records.save() logger.info( "cine_transform %s: 접수한 변환(%s)을 다시 가져왔다 — 다시 사지 " "않았다", tag, rid) return {**rec_p, "reused": False} # ── 결말을 못 본 옛 요청의 신원을 잃지 않는다 ───────────────────── # 여기까지 흘러왔다 = **이제 새로 요청한다**. 그런데 옛 기록이 「접수 # 여부 모름」이나 「접수는 됐는데 결말 못 봄」이면 아래에서 같은 칸을 # 덮어써 그 번호와 시각이 사라진다. 사람이 fal 대시보드에서 요금이 # 나갔는지 대조할 실마리가 그것뿐인데 없어진다 (2026-08-26 Codex # 재리뷰 BLOCK-3). # # ★지문이 바뀌었다고 새 요청 자체를 막지는 않는다. 원본이 다시 # 만들어졌으면 그건 **다른 그림**이고, 옛 결과는 어차피 못 쓴다. # 막으면 사람이 손댈 때까지 그 샷이 영영 안 나온다. 잃는 것만 막는다. if isinstance(prior, dict) and ( prior.get("pending") is True or prior.get("submission_unknown") is True ): # ★이름 그대로 **최근 것만 남는 기록**이다 — 덧붙이기 전용이 아니다 # (2026-08-26 Codex 2차 재리뷰). 최근 _UNRESOLVED_LOG_LIMIT 건만 # 남고 그보다 오래된 것은 밀려난다. 오래 보관해야 하는 감사 대장이 # 필요해지면 여기가 아니라 별도 저장소를 둔다. _orphan_key = f"{rec_key}::unresolved_log" _orphans = records.data.get(_orphan_key) if not isinstance(_orphans, list): _orphans = [] _orphans.append({ # 옮겨 적은 시각 — **요청한 시각이 아니다.** 대조에 쓸 시각은 # 아래 `attempted_at`(시도 시작)·`submitted_at`(접수 확정)이다. "archived_at": datetime.now(timezone.utc).isoformat(), "attempted_at": prior.get("attempted_at", ""), "submitted_at": prior.get("submitted_at", ""), "reason": ("submission_unknown" if prior.get("submission_unknown") is True else "pending"), "request_id": prior.get("request_id", ""), "provider": prior.get("provider", ""), "endpoint": prior.get("endpoint", ""), "status_url": prior.get("status_url", ""), "response_url": prior.get("response_url", ""), "fingerprint": prior.get("fingerprint", ""), "detail": prior.get("submission_unknown_reason", "") or prior.get("error", ""), }) records.data[_orphan_key] = _orphans[-_UNRESOLVED_LOG_LIMIT:] records.save() logger.warning( "cine_transform %s: 결말을 못 본 옛 요청(%s · 번호=%s · 시도=%s)을 " "따로 남기고 새로 요청한다 — fal 대시보드에서 요금 여부를 대조할 것", tag, _orphans[-1]["reason"], _orphans[-1]["request_id"] or "없음", _orphans[-1]["submitted_at"] or _orphans[-1]["attempted_at"] or "?") # 같은 지문에서 검열 거부가 몇 번 쌓였는가 — 지문이 다르면 새 입력 # 이므로 0 부터다. prior_refusals = 0 if isinstance(prior, dict) and _is_current(prior): _pr = prior.get("moderation_refusals") if isinstance(_pr, int) and not isinstance(_pr, bool) and _pr > 0: prior_refusals = _pr rec: Dict[str, Any] = { "applied": False, # 이 시도를 **언제 시작했나**. 접수 번호를 못 받은 채 끝난 경우 # (`submission_unknown`) fal 대시보드와 맞출 실마리가 이것뿐이다. "attempted_at": datetime.now(timezone.utc).isoformat(), "fingerprint": fp, "fingerprint_version": CINE_FINGERPRINT_VERSION, # 제공자 신원 — 이 그림을 **누가 만들었는가**. 지문이 접는 것과 # 같은 값을 기록에도 남겨야 나중에 판을 갈라 읽을 수 있다. "provider": _ident.get("provider", ""), "endpoint": _ident.get("endpoint", ""), "model": model, # ★[2026-09-09 Codex BLOCK 2] 이 변환이 **어느 호출 태그로** 나갔는가. # 영속부가 자산의 generation_call_id·모델을 이 값으로 찾는다. 안 담으면 # 주 태그만 찾다가 대체 산출을 **주 제공자 모델로 잘못 기록**한다. "multiroll_tag": str((context or {}).get("multiroll_tag") or ""), "slot": slot, "pack": pack, # 최종 자산 lineage 용 — 변환기가 실제로 본 직접 입력(원본 sel)의 # 신원. 자산 기록이 nb2 롤 참조를 "직접 첨부"로 오기하지 않는 # 전제 재료(Codex R1 BLOCK-3). "source_file": sel_path.name, "source_sha256": source_sha, } # 접수된 작업의 신원 — 실패 갈래에서도 읽어 record 에 실어야 다시 가져온다. submitted_info: Dict[str, Any] = {} _terminal_fail = False def _note_submit(info: Dict[str, Any]) -> None: """접수 즉시 신원을 durable 하게 — **산출 파일보다 먼저**. 동기 왕복($0.03)에서는 「파일 durable 후 기록」이 맞았다: 크래시 창에서 기록 없는 파일이 남고 다음 방문이 재변환하는데, 그 1회 재지출이 기록 없는 자산을 믿는 것보다 쌌다. 접수 기반 $0.25 에는 반대다 — **접수된 작업의 신원을 먼저 남겨야** 그것을 다시 가져온다. """ # 접수 **시각**을 여기서 못박는다 — fal 대시보드와 맞출 때 실제로 # 쓰이는 값이다. 나중에 기록을 옮긴 시각으로는 대조가 안 된다. info.setdefault("submitted_at", datetime.now(timezone.utc).isoformat()) submitted_info.update(info) records.data[rec_key] = { **rec, "pending": True, "submitted_at": info.get("submitted_at", ""), "provider": info.get("provider", "") or _ident.get("provider", ""), "endpoint": info.get("endpoint", "") or _ident.get("endpoint", ""), "request_id": info.get("request_id", ""), # 접수 때 제공자가 준 조회 주소 **둘 다** 남긴다 — 프로세스가 # 죽었다 살아나면 이것이 결과를 다시 가져오는 유일한 손잡이다. "status_url": info.get("status_url", ""), "response_url": info.get("response_url", ""), } records.save() if hasattr(client, "set_submit_hook"): client.set_submit_hook(_note_submit) try: client.set_context(**context) png, elapsed_ms = client.generate_image( prompt, labeled_references=[(CINE_SOURCE_LABEL, sel_bytes)], ) from app.modules.pipeline.multiroll_gemini import atomic_write_bytes atomic_write_bytes(out_path, png) # durable 먼저, 기록은 그 뒤 # ★파일을 남겼으면 **그 사실을 곧바로 기록에도** 남긴다. # # 2026-08-26 Codex 재리뷰 BLOCK-5. 바로 아래 확인에서 멈추면 최종 # 기록을 못 쓰고 나간다. reve 는 접수 hook 이 `pending` 을 남겨 둬서 # 다음 걷기가 결과만 가져오지만, hook 이 없는 제공자(동기 왕복)는 # 그 경로가 없다 — 요금이 나간 그림이 디스크에 있는데도 기록이 없어 # 다음 걷기가 **처음부터 다시 요청한다.** # # 그래서 제공자와 무관하게 「요청이 끝났고 파일로 남겼다」를 먼저 # 적는다. 아직 최종본이 아니므로 `applied` 는 안 붙인다 — 이 표식 # 만으로는 그림이 쓰이지 않고, 다음 걷기가 판정만 돌려 확정한다. # ★접수 신원도 **같이 실어야 한다** (2026-08-26 Codex 2차 재리뷰 # BLOCK-1). 안 실으면 이 기록이 `_note_submit` 이 남긴 pending 을 # 덮어써 접수 번호·조회 주소가 사라진다. 그러면 남겨 둔 파일이 # 없어지거나 해시가 어긋났을 때 **무료로 다시 가져올 손잡이가 # 없어** 곧바로 새 요청으로 간다($0.25 가 또 나간다). rec_staged = { **rec, "staged": True, "file": out_path.name, "staged_sha256": hashlib.sha256(png).hexdigest(), "latency_ms": int(elapsed_ms), } if submitted_info.get("request_id"): rec_staged["pending"] = True rec_staged["request_id"] = str( submitted_info.get("request_id") or "") rec_staged["status_url"] = str( submitted_info.get("status_url") or "") rec_staged["response_url"] = str( submitted_info.get("response_url") or "") # ★접수 시각도 실어야 한다 (2026-08-26 Codex 3차 재리뷰). # 여기서 멈춘 뒤 다음 걷기가 이 기록으로 확정하면, 빠뜨린 시각은 # 최종본에서 **영영 비어 있게** 된다. rec_staged["submitted_at"] = str( submitted_info.get("submitted_at") or "") records.data[rec_key] = rec_staged records.save() # ★요금이 나간 **직후**에 다시 확인한다. # # 돈 쓰기 직전의 확인은 이미 `reserve_current_call` 에 있다. 그런데 # 그 뒤 이 그림을 최종본으로 확정하기까지 사이에 락을 놓칠 수 있고, # 그러면 남의 주행 자리에 내 결과를 적는다. 요금이 나간 것은 파일로 # 남겨 두고(위에서 이미 했다) **확정만 멈춘다** — 돈은 이미 # 나갔으니 결과까지 버릴 이유가 없다. # # 손잡이는 스텝 경계에서 걸어 둔 그것 그대로다. _stop_gate("변환 결과를 받은 직후") rec["applied"] = True rec["file"] = out_path.name rec["staged_sha256"] = hashlib.sha256(png).hexdigest() rec["latency_ms"] = int(elapsed_ms) if submitted_info.get("request_id"): # 확정본에도 접수 신원을 남긴다 — 나중에 이 그림이 어느 요청에서 # 나왔는지 대조할 수 있어야 한다. `pending` 은 안 붙이므로 다시 가져오기 # 갈래를 다시 타지 않는다. rec["request_id"] = str(submitted_info.get("request_id") or "") rec["status_url"] = str(submitted_info.get("status_url") or "") rec["response_url"] = str( submitted_info.get("response_url") or "") rec["submitted_at"] = str( submitted_info.get("submitted_at") or "") _verify_and_gate(rec, tag=tag, source_png=sel_bytes, result_png=png, project_config=project_config, context=context, where="새 변환을 확정하기 직전") except ImageCallBudgetExceeded: raise except Exception as exc: # noqa: BLE001 — 원본 fallback + 기록 # 멈추라는 말과 락을 놓친 것은 **변환 실패가 아니다.** 여기서 삼키면 # 「이 샷은 변환이 안 됐다」로 적히고 다음 샷으로 넘어가 계속 돈을 # 쓴다. 이미 산 그림은 위에서 파일로 남겼으니 잃지 않는다. from app.core.run_control import is_abort # ★코드 목록은 **한 곳**(`ABORT_CODES`)이다 — # 여기 다시 적으면 한쪽만 고쳐진다 if is_abort(exc): logger.warning("cine_transform %s 중단 — %s", tag, getattr(exc, "message", str(exc))) raise rec["error"] = f"{type(exc).__name__}: {exc}"[:500] # ★접수됐는지 모르는 채로 끝났다 — **다음 걷기가 자동으로 다시 보내면 # 안 된다.** fal 이 받아서 요금이 나갔을 수 있고, 다시 보내면 같은 # 이미지에 요금이 두 번 나간다. 표식을 남겨 사람이 정하게 한다. # ★클래스 **이름**이 아니라 타입으로 판정한다 (2026-08-26 자체 리뷰). # 종전에는 `type(exc).__name__ == "ReveSubmissionUnknown"` 이었는데, # Grok·Gemini 가 던지는 공용 `ImageSubmissionUnknown` 은 이름이 달라 # **못 알아봤다.** 그리고 grok 이 기본 변환 제공자다 # (`config.still_cine_provider`) — 즉 이 판이 막으려던 재요청 차단이 # **기본 경로에서 열려 있었다.** 지금은 reve 쪽이 공용 예외의 # 하위라 `isinstance` 하나가 둘 다 잡는다. from app.modules.llm.image_send_state import ImageSubmissionUnknown if isinstance(exc, ImageSubmissionUnknown): rec["submission_unknown"] = True rec["submission_unknown_reason"] = str( getattr(exc, "cause", "") or type(exc).__name__)[:200] logger.error( "cine_transform %s: 접수 여부 불명 — 자동 재요청을 멈춘다" "(요금 이중 방지). fal 대시보드에서 접수 여부를 확인하고, " "안 됐으면 records 의 submission_unknown 을 지운 뒤 다시 " "걷는다. 됐으면 그 번호를 request_id + pending 으로 넣으면 " "추가 요금 없이 결과만 가져온다. (%s)", tag, rec["error"]) _terminal_fail = _is_terminal_provider_error(exc) # ★검열 거부만 센다 — 서버가 그날만 까다로웠던 것(5xx·타임아웃)은 # 다시 보내면 되는 일이라 포기 근거가 아니다. if is_moderation_error(exc): refusals = prior_refusals + 1 rec["moderation_refusals"] = refusals limit = _give_up_after() if limit > 0 and refusals >= limit: rec["declined"] = True rec["declined_reason"] = "moderation" logger.warning( "cine_transform %s: 검열 거부 %d회 — 이 원본의 변환을 " "포기하고 원본을 최종본으로 확정한다 (거부에도 요금이 " "나가므로 다시 보내지 않는다)", tag, refusals, ) else: logger.warning( "cine_transform %s: 검열 거부 %d회 (포기 기준 %d) — " "원본 sel fallback", tag, refusals, limit, ) else: if prior_refusals: # 검열 이력은 남긴다 — 셈은 검열 거부에만 늘어난다. rec["moderation_refusals"] = prior_refusals logger.warning( "cine_transform %s: 변환 실패 — 원본 sel fallback (%s)", tag, rec["error"], ) # ★접수는 됐는데 결말을 못 본 경우 다시 가져오기 신원을 record 에 남긴다. # 이 줄이 없으면 아래 저장이 `_note_submit` 이 쓴 pending 을 덮어써 # **이미 산 작업을 영영 못 찾는다**($0.25 를 두 번 쓴다). terminal 실패 # (검열·결과 없음)는 다시 가져오기할 산출이 아예 없으므로 남기지 않는다. if (not rec.get("applied") and not _terminal_fail and submitted_info.get("request_id")): rec["pending"] = True rec["request_id"] = str(submitted_info.get("request_id") or "") # ★`status_url` 도 같이 복사한다 (2026-08-26 Codex 2차 재리뷰 # BLOCK-1). 종전에는 `response_url` 만 옮겨서, 접수 hook 이 애써 # 남긴 조회 주소가 **여기서 다시 사라졌다.** 그러면 다음 걷기가 # 주소를 직접 조립하게 되고, 그 조립을 틀려 405 를 받은 적이 있다. rec["status_url"] = str(submitted_info.get("status_url") or "") rec["response_url"] = str(submitted_info.get("response_url") or "") rec["submitted_at"] = str(submitted_info.get("submitted_at") or "") records.data[rec_key] = {k: v for k, v in rec.items()} records.save() return {**rec, "reused": False}