"""cine 산출 검증 검사 — 최종 픽셀을 재는 유일한 자리. ## 무엇을 재는가 (2026-08-26 계약 교체) **변환본 한 장이 사진으로 성립하는가**만 본다. 손가락이 여섯 개, 사람이 공중에 떠 있음, 차 문이 여덟 개 — 카메라 앞에 있을 수 없는 것만 잡는다. 성립하면 **무조건** 변환본을 최종본으로 쓴다. 성립하지 않으면 변환본을 버리고 앞 단계가 고른 원본이 최종본이 된다. ### 왜 바꿨나 종전 계약은 「변환본이 **원본의** 불변축(장소·인물·사물·순간·광원)을 지켰나」였다. 그런데 이 단계는 시네마틱 변환이다 — 빛을 바꾸고 카메라를 옮기는 것이 **시켜서 하는 일**이다. 그것을 기각 근거로 삼으면 단계 자체가 무의미해진다. 2026-08-26 실측(7샷)에서 반려 3건의 사유가 전부 그런 것이었다: 바닥의 깨진 그릇이 사라짐 · 형광등 둘 중 하나가 꺼짐 · 카메라가 가까워짐. 장당 $0.25 를 쓰고 버린 셈이다(8장 사서 3장 채택, 62% 버림). ★그리고 사유의 상당 부분이 **거짓**이었다. 판정기가 「벽이 물결치듯 휘고 여러 번 반복된다」고 했는데, 같은 변환본을 원본 크기로 줄여 다시 물으니 그 말이 **통째로 사라졌다**(A/B, 3샷×2조건×2회차). 내용은 한 비트도 안 바뀌었으니 원인은 그림이 아니라 **크기**다 — 변환본은 5456×3045, 원본은 1376×768 인데 축소 없이 그대로 보내고 있었다. ## 계약 - 재는 것은 **범주 넷**뿐이다(`DEFECTS`). 무엇이 보이는지는 VLM 이 말한다 — 코드에 「손가락은 다섯 개」 같은 수를 두지 않는다. - **두 번 묻고 둘 다 파탄이라 할 때만 기각한다.** 한 번만 물으면 모델의 흔들림이 그대로 기각이 되고, 이 검사은 「의심스러우면 통과」가 원칙이다. - **판정 전에 원본 크기로 줄인다.** 위 A/B 가 그 필요를 실측으로 보였다. 이제 `duplication` 이 기각 범주라 안 줄이면 그 거짓이 바로 기각이 된다. - 기각은 **실패가 아니라 확정된 결말**이다 — 원본이 최종본이 되고 스텝은 닫힌다. 실패로 세면 스텝이 영영 안 닫혀 재개마다 앞 단계가 다시 돌고 연쇄 재생성이 난다(2026-08-19 실측: 한 바퀴에 그림 66장). - 판정 자체가 못 돌면 **기각하지 않는다** — 판정기 사정으로 이미 산 그림을 버리면 돈만 나가고 결과가 나빠진다. - 제공자와 무관하다. 이 검사은 픽셀만 본다 — reve 2.1 을 다른 것으로 갈아 끼워도 그대로 쓴다. """ from __future__ import annotations import io import logging from typing import Any, Dict, List, Optional, Tuple from app.modules.prompt_loader import load_prompt, pack_dir_content_hash logger = logging.getLogger(__name__) _MODULE = "cine_verify" PACK_VERSION_MAP = { "1": "1.202608251555", # 옛 계약(원본 대조) — 되돌릴 때를 위해 남긴다 "2": "2.202608260400", # 지금 계약(온전성) } CINE_VERIFY_PACK_VERSION = "2" # 판정 모델 — **두 모델이 각각 한 번씩** (2026-08-27 사용자 지시, # 「gpt sol vlm 은 성능이 안좋아 … 무조건 gemini 3.1 pro 와 grok 최신 모델 # 둘을 사용해야해」). # # ★종전에는 `gpt` **한 모델에게 같은 그림을 두 번** 물었다. 회차가 둘인 # 이유는 「둘 다 파탄이라 해야 기각」이라는 합의 강도였는데, 같은 모델을 # 두 번 부르면 그 둘이 독립이 아니다 — 한 모델의 버릇이 두 표로 센다. # **모델을 갈면 같은 콜 수로 더 독립적인 두 의견**을 얻는다(Codex 판정: # 2모델×2회=4콜은 근거 없이 중복이다). # # ★순서는 고정한다 — 기록의 안정성 때문이고, 합산은 순서를 안 본다. CINE_VERIFY_MODELS: Tuple[str, ...] = ("gemini-pro", "grok") # 집계 의미 버전 — **코드가 정하는 부분**이라 팩 해시에 안 잡힌다. # 무엇을 기각으로 세는지의 **뜻**이 바뀌면 올린다. CINE_VERIFY_CONTRACT_VERSION = "cine_verify_v2_plausibility_both_agree" # ── 범주 = 데이터 계약 ──────────────────────────────────────────────── # 넷 다 기각 근거다. 「예쁜가·잘 찍었나」는 묻지 않는다 — 그 물음은 못 # 믿는다는 실측이 있다(Qwen 판정 실험: 관찰·셈은 좋고 `matches_brief` 는 # 못 믿는다). 코드에 사물 이름이나 개수를 두지 않는다. DEFECTS: Tuple[Tuple[str, str], ...] = ( ("anatomy", "사람이나 동물의 몸이 성립하는가 — 손가락·팔·다리의 수, 관절이 꺾인 " "방향, 얼굴과 머리의 형태가 사람의 몸으로 가능한가"), ("object_integrity", "사물이 성립하는가 — 문·바퀴·다리·손잡이 같은 부분의 수가 그 물건으로 " "가능한가, 물건이 녹거나 도중에 끊겨 있지 않은가"), ("support", "무게가 받쳐지는가 — 사람이나 사물이 아무것도 없는 공중에 떠 있거나, " "닿을 수 없는 곳에 서 있지 않은가"), ("duplication", "같은 것이 까닭 없이 되풀이되는가 — 한 사람이 화면에 두 번 나오거나 " "배경의 같은 조각이 복제되어 이어 붙어 있지 않은가"), ) GATED: Tuple[str, ...] = tuple(name for name, _ in DEFECTS) # 회차 — 같은 그림을 **따로 두 번** 묻는다. 원본 대조가 아니므로 좌우를 # 바꿀 자리가 없고, 남은 것은 「독립으로 두 번 물어 둘 다 그렇다고 할 때만 # 기각」이라는 규칙뿐이다. MAX_ROUNDS = 2 _warned_rounds: set = set() # 판정기에 보낼 때의 상한 — 원본 크기를 못 읽었을 때만 쓴다. _FALLBACK_MAX_SIDE = 1536 def normalize_rounds(rounds: Any) -> int: """설정값을 실제로 돌 수 있는 회차로 — **갈림의 단일 지점**. ★(2026-08-25 Codex BLOCK) 종전에는 하한만 걸어 `max(1, rounds)` 였다. 그러면 3 을 넣었을 때 호출은 2번 나가는데 `len(seen) == requested` 가 영원히 거짓이라 **돈은 쓰고 검사은 한 번도 기각하지 못한다.** 조용한 무력화라 로그에도 안 남았다. 상한을 넘긴 값은 잘라 쓰되 **잘랐다는 것을 남긴다** — 말없이 줄이면 운영자는 자기가 설정한 대로 도는 줄 안다. """ try: raw = int(rounds) except (TypeError, ValueError): raw = MAX_ROUNDS out = max(1, min(MAX_ROUNDS, raw)) if raw != out and raw not in _warned_rounds: _warned_rounds.add(raw) logger.warning( "cine_verify: 회차 설정 %r 은 돌 수 없다 — %d 회가 상한이라 " "%d 로 잘라 쓴다", rounds, MAX_ROUNDS, out) return out def resolve_verify_pack(selector: str = CINE_VERIFY_PACK_VERSION) -> str: try: return PACK_VERSION_MAP[selector] except KeyError: raise ValueError( f"unknown cine_verify pack selector {selector!r} " f"(known: {sorted(PACK_VERSION_MAP)})") def resolve_verify_model_physical(alias: Optional[str] = None) -> str: """alias 뒤의 **물리** 모델. alias 만 신원에 넣으면 판 교체가 판정 캐시를 못 뚫는다. 인자를 안 주면 **짝 전체**를 `,` 로 이어 돌려준다 — 계약 지문이 「누가 판정하나」를 통째로 담아야 하기 때문이다. """ from app.core.config import settings table = {"gpt": settings.openai_model, "gemini-pro": settings.gemini_text_model, "gemini-flash": settings.gemini_flash_model, "grok": getattr(settings, "grok_judge_model", "grok")} if alias is not None: return table.get(alias, alias) return ",".join(table.get(a, a) for a in CINE_VERIFY_MODELS) def verify_models(rounds: Any = None) -> Tuple[str, ...]: """판정할 모델들 — **언제나 짝 전체**. ★★**회차로 짝을 줄이면 안 된다** (2026-08-27 Codex BLOCK, 내 앞 구현을 물린다). 처음엔 `CINE_VERIFY_MODELS[:normalize_rounds(rounds)]` 로 썼다. 그러면 공개 설정 `STILL_CINE_VERIFY_ROUNDS=1` 하나로 · grok 이 **조용히** 빠지고 (사용자 지시 「무조건 둘」 위반) · Gemini 한 번의 `broken` 만으로 최종본을 기각한다 (「둘 다 동의할 때만 기각」 위반) 두 계약이 한꺼번에 깨진다. **더 나쁜 것은 내가 그 동작을 정답으로 못박는 시험을 썼다는 것이다** — `test_lowering_rounds_drops_the_second_model_visibly`. ★그래서 `rounds` 는 이 계약에서 **아무 뜻이 없다.** 검사를 끄는 것은 `still_cine_verify_enabled` 플래그 하나가 한다. 인자를 남겨 둔 것은 호출부 호환 때문이고, 값은 무시한다. """ del rounds # 이 계약에서는 회차가 모델 수를 못 바꾼다 return tuple(CINE_VERIFY_MODELS) def verify_contract_sha(rounds: int = MAX_ROUNDS) -> str: """검사 계약 지문 — 판정을 정하는 것 **전부**. ★(2026-08-25 Codex BLOCK) 종전에는 범주 **이름**만 접었다. 그런데 모델이 보는 것은 질문 문장이라, 문장을 고쳐도 sha 가 안 움직여 옛 판정이 「지금 계약」으로 통했다. 나가는 문장 그대로를 접는다. 팩 파일이 아니라 **코드가 소유한 부분**(집계 규칙)은 팩 해시에 안 잡히므로 계약 버전으로 따로 못박는다. ★**회차는 더 이상 안 접는다** (2026-08-27, #92 이중 모델). 옛 계약에서는 2→1 이 「둘 다 동의할 때만 기각」을 「한 번 보고 기각」으로 바꿔 다른 판정이었다. 지금은 `verify_models()` 가 언제나 짝 전체를 돌려주므로 회차가 아무것도 안 바꾸고, 접어 두면 **쓰지도 않는 설정을 바꿀 때 전량 재판정**이 난다. 대신 **짝의 크기**를 접는다 — 짝이 셋이 되면 합의 강도가 실제로 달라지기 때문이다. """ import hashlib basis = "\n".join([ CINE_VERIFY_CONTRACT_VERSION, pack_dir_content_hash(_MODULE, resolve_verify_pack()), # ★**짝 전체**를 접는다 (2026-08-27). 한 모델만 접으면 짝의 나머지를 # 갈아도 지문이 안 움직여 옛 판정이 「지금 계약」으로 통한다. ",".join(verify_models()), resolve_verify_model_physical(), build_defect_user_head(), # 나가는 문장 그대로 ",".join(GATED), # ★회차가 아니라 **짝의 크기**를 접는다 (2026-08-27). 회차는 이 # 계약에서 아무 뜻이 없는데 접으면, 쓰지도 않는 설정을 바꿀 때 # 전량 재판정이 난다. str(len(CINE_VERIFY_MODELS)), ]) return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:16] def build_defect_schema() -> Dict[str, Any]: return { "type": "object", "properties": { "checks": { "type": "array", "minItems": len(DEFECTS), "maxItems": len(DEFECTS), "items": { "type": "object", "properties": { "category": {"type": "string", "enum": [c for c, _ in DEFECTS]}, "sound": {"type": "boolean"}, "what_is_impossible_ko": {"type": "string"}, }, "required": ["category", "sound", "what_is_impossible_ko"], "additionalProperties": False, }, }, }, "required": ["checks"], "additionalProperties": False, } def build_defect_user_head() -> str: """범주 목록은 **코드가 소유하는 데이터 계약**이라 팩 본문과 나눠 붙인다.""" return "CATEGORIES:\n" + "\n".join(f"- {c}: {q}" for c, q in DEFECTS) def shrink_for_judging( result_png: bytes, source_png: Optional[bytes] = None ) -> Tuple[bytes, Tuple[int, int]]: """판정기에 보낼 크기로 줄인다 — **원본 크기까지만**. 왜 필요한가: 변환본은 원본의 4배(5456×3045 vs 1376×768)로 나오는데 종전에는 축소 없이 보냈다. 그 크기에서 판정기는 있지도 않은 「반복· 휘어짐」을 보고했고, 같은 그림을 원본 크기로 줄이자 그 말이 통째로 사라졌다(2026-08-26 A/B 실측). `duplication` 이 기각 범주가 된 지금은 안 줄이면 그 거짓이 곧바로 기각이 된다. ★**줄이기만 한다. 비율은 안 건드린다.** 목표 칸에 맞춰 `resize(target)` 을 그냥 부르면 세로가 긴 그림이 가로가 긴 원본을 만났을 때 **폭을 몇 배로 늘리고 비율을 깨뜨린다**(예: 결과 100×1000, 원본 500×500 → 500×500). 늘어난 자국과 찌그러진 사람은 그대로 「파탄」으로 읽히므로, 판정을 도우려던 것이 판정을 망친다. 그래서 **한 배율**(`min(1, 가로비, 세로비)`)로만 줄인다. Returns: (보낼 PNG bytes, 그 크기) """ from PIL import Image im = Image.open(io.BytesIO(result_png)) target: Optional[Tuple[int, int]] = None if source_png: try: src = Image.open(io.BytesIO(source_png)) target = src.size except Exception as exc: # noqa: BLE001 — 원본을 못 읽어도 판정은 돈다 logger.debug("cine_verify: 원본 크기를 못 읽었다 — %r", exc) if target is None: target = (_FALLBACK_MAX_SIDE, _FALLBACK_MAX_SIDE) # 한 배율로만 줄인다. 1.0 이면 이미 충분히 작다는 뜻이라 손대지 않는다. ratio = min(1.0, target[0] / im.width, target[1] / im.height) if ratio >= 1.0: return result_png, im.size 새크기 = (max(1, round(im.width * ratio)), max(1, round(im.height * ratio))) out = io.BytesIO() im.resize(새크기, Image.LANCZOS).save(out, format="PNG") return out.getvalue(), 새크기 def _ask_all( judged_png: bytes, *, step_tag: str, models: Tuple[str, ...], project_config: Optional[Dict[str, Any]], opik_metadata: Optional[Dict[str, Any]], ): """두 모델에게 각각 한 번 — **계약은 `dual_vlm` 한 벌만 쓴다.** ★★종전에는 여기서 직접 `call_structured` 를 돌았다. 그래서 「두 모델을 각각 한 번, fallback 봉인, Router 재시도 봉인, 모델별 usage 기록」이라는 **같은 계약이 두 벌** 있었고, 둘째 벌에 봉인이 빠져 Codex 가 **두 번** BLOCK 했다(`enable_fallback` 은 넣고 `num_retries` 를 빠뜨림). 한 벌로 합친다 — 그 자리에 봉인을 넣는 것을 잊을 방법이 없어진다. Returns: `dual_vlm.DualResult`. 합의는 여기(호출부)가 한다. """ from app.modules.llm.dual_vlm import ask_both from app.modules.pipeline.multiroll_gemini import png_part sys_txt = load_prompt(_MODULE, "plausibility_system", version=resolve_verify_pack()).strip() parts: List[Dict[str, Any]] = [ {"type": "text", "text": build_defect_user_head()}, {"type": "text", "text": "PHOTOGRAPH:"}, png_part(judged_png), ] # ★`project_config` 는 안 넘긴다. 종전에는 호출부 설정을 합쳐 놓고 # `{tag: {"model": model}}` 로 **항상 덮었다** — 이 스텝의 모델을 # 호출부가 바꿀 수 없었다는 뜻이다. `ask_both` 가 같은 일을 하므로 # 나가는 것은 그대로고, 못 하던 것을 새로 못 하게 만들지도 않는다. del project_config return ask_both( step_tag, sys_txt, parts, build_defect_schema(), aliases=models, schema_name=step_tag, opik_metadata=opik_metadata, # ★두 심판을 **동시에** 부른다 (2026-08-29 사용자 지시 "2, 4, 8 # 병렬로"). 둘은 같은 그림을 각자 보고 각자 답할 뿐 서로를 안 # 본다 — 합치는 것은 이 함수 밖 코드다. # `ask_both` 기본은 False 로 두었다: 참조 검증·비교도 같은 함수를 # 쓰는데 손대지 않은 자리의 동시성까지 바꾸면 안 된다. # 결과 순서는 `aliases` 순이라 기록이 안 흔들리고, 입력·모델이 # 같으므로 **지문은 안 움직인다**. parallel=True, ) def checks_by_category(payload: Optional[Dict[str, Any]]) -> Dict[str, Any]: """모델 응답 → 범주별 판정. **판단은 안 한다** — 모양만 바꾼다.""" if not isinstance(payload, dict): return {} return {str(d.get("category")): d for d in (payload.get("checks") or []) if isinstance(d, dict)} def verify_cine_result( *, result_png: bytes, source_png: Optional[bytes] = None, step_tag: str = "still_cine_verify", rounds: int = 2, project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """변환본이 사진으로 성립하는가 — 따로 두 번 묻는다. Args: result_png: 판정 대상. **이것 한 장만 본다.** source_png: 판정에 안 쓴다 — 얼마나 줄일지의 기준으로만 쓴다. 반환: {"ok": bool, "broken": [범주], "split": [범주], "findings": {범주: {"sound":.., "note":.., "agreed":..}}, "rounds": n, "requested": n, "complete": bool, "judged_size": [w, h], "contract": sha, "error": str|None} ★`ok=False` 는 **기각**이지 실패가 아니다 — 호출측은 원본을 최종본으로 확정하고 스텝을 닫는다. `error` 가 채워진 경우(판정 자체가 못 돌았다)는 **기각하지 않는다**(`ok=True`). """ # ★**회차가 아니라 모델 수다** (2026-08-27 Codex BLOCK). 공개 설정 # `STILL_CINE_VERIFY_ROUNDS=1` 로 grok 이 조용히 빠지고 Gemini 한 번의 # 파탄만으로 최종본이 기각되던 것을 막는다. 검사를 끄는 것은 # `still_cine_verify_enabled` 플래그 하나가 한다. models = verify_models() requested = len(models) del rounds out: Dict[str, Any] = {"ok": True, "broken": [], "split": [], "findings": {}, "rounds": 0, "requested": requested, "complete": False, "judged_size": None, "contract": verify_contract_sha(), "error": None} try: judged_png, judged_size = shrink_for_judging(result_png, source_png) out["judged_size"] = list(judged_size) except Exception as exc: # noqa: BLE001 — 축소 실패는 기각 근거 아님 # ★「원본 크기로」가 아니다 — 줄이지 못했으니 **변환본 크기 그대로** # 나간다. 그 크기에서 판정기가 없는 반복을 볼 수 있다는 것이 이 축소를 # 넣은 이유이므로, 로그가 그 위험을 감추면 안 된다. logger.warning( "cine_verify[%s]: 축소 실패 — 변환본 크기 그대로 묻는다 " "(그 크기에서 없는 반복을 볼 수 있다) %r", step_tag, exc) judged_png = result_png out["models"] = list(models) # ★한 모델이 죽어도 **나머지는 묻는다** — 안 그러면 「못 물어봤다」와 # 「둘 다 나쁘다」가 구분이 안 된다. 기각 권한은 `complete` 가 정한다. # 그 계약은 `dual_vlm.ask_both` 가 소유한다. dual = _ask_all(judged_png, step_tag=step_tag, models=models, project_config=project_config, opik_metadata=opik_metadata) out["model_usage"] = {c.alias: dict(c.usage) for c in dual.calls} for c in dual.calls: if not c.ok: logger.warning("cine_verify[%s]: %s 판정 실패 — %s", step_tag, c.alias, c.error) out["error"] = f"{c.alias}: {c.error}"[:300] out["provenance"] = dual.provenance() seen = [checks_by_category(p) for p in dual.payloads] out["rounds"] = len(seen) # ★요청한 회차가 **전부** 성공했을 때만 기각 근거를 셀 수 있다 # (2026-08-25 Codex BLOCK-1). 관찰 자체는 기록에 남긴다 — 못 쓰는 것은 # 기각 권한뿐이다. out["complete"] = len(seen) == requested if not seen: return out for category, _q in DEFECTS: rows = [r for r in (s.get(category) for s in seen) if isinstance(r, dict)] if not rows: continue sounds = [bool(r.get("sound")) for r in rows] note = next((str(r.get("what_is_impossible_ko") or "") for r in rows if not r.get("sound")), "") out["findings"][category] = {"sound": all(sounds), "note": note, "agreed": len(set(sounds)) == 1} # 한 회차라도 그 범주의 칸이 비면 「둘 다 동의」를 말할 수 없다. if not (out["complete"] and len(rows) == requested): if not all(sounds): out["split"].append(category) continue if all(not s for s in sounds): out["broken"].append(category) # 두 회차가 **둘 다** 파탄이라 했다 elif any(not s for s in sounds): out["split"].append(category) # 엇갈렸다 — 기각하지 않는다 out["ok"] = not out["broken"] return out