"""multiroll_select — s40 공통 생성 파이프 이식 (2026-07-13). N롤(동일 프롬프트·동일 참조) → Gemini VLM 단독 판정(0-10 + 랭킹, 동점=랭킹 우선) → 선정본 복사(`{stem}_sel.png`) → 결함 검사(CRITIQUE) → 결함 시 수정 프롬프트 저작 → i2i(선정 원본 **단독** 참조) → 수정본이 최종 `_sel`(후속 체인 앵커). 이미지 엔진·VLM 판정은 전부 callable 주입 — 이 모듈은 결정론 로직(롤 파일 재개, 선정/동점 해소, critique 소급 재개, 수정 프롬프트 조립)만 소유한다. 실험 정본: scratchpad/forest_exp/s40_full_multiref.py(_roll3/_judge_gemini/ _critfix/_select/_resume) + s39_threeroll_vlm.py:207-216(_gemini_select). 재개 계약(크래시/예산 소진 후 재실행 안전): - 롤 파일(`{stem}_{a..}.png`) 존재 → 해당 롤 skip. - `_sel` 존재 + record 에 critique 흔적 있음(critique/critique_skipped) → 전부 skip. - `_sel` 존재 + critique 미실행 + 게이트 on → 결함 검사·수정만 소급. - `_sel` 존재 + record 부재(선정 라벨 불명) → 재지출 없이 sel 유지(경고). """ from __future__ import annotations import logging import shutil import uuid from pathlib import Path from typing import ( Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple, ) logger = logging.getLogger(__name__) def build_ref_records( labeled_refs: Sequence[Tuple[str, Any]], ref_role_metadata: Optional[Sequence[Optional[Dict[str, Any]]]] = None, ) -> List[Dict[str, Any]]: """records.json 에 적을 참조 목록 — 자산 id 를 함께 나른다. ★`labeled_refs` 튜플을 넓히지 않는다. 그것은 코드 전체 243곳에 나오고 3곳에서 제자리 치환된다(`outdoor_site_layout_step.py:357,464` · `visual_continuity_anchor_step.py:490`) — 넓히면 2-튜플로 푸는 자리가 전부 깨진다. 씬 경로에 이미 있는 관례를 따른다: `ref_role_metadata[i]` 는 `labeled_refs[i]` 와 **index 로 정렬된** 병렬 목록이다 (`scene_generation_coordinator.collect_actual_attached_refs` 와 동형). 라벨로 맞추지 않는 이유 — 같은 라벨이 둘일 수 있다. ★`path` 칸은 그대로 둔다. 지금 그것을 읽는 것들이 있고, 빼면 그 자리가 조용히 빈다. """ metas = list(ref_role_metadata or []) out: List[Dict[str, Any]] = [] for i, (label, src) in enumerate(labeled_refs): meta = metas[i] if i < len(metas) and isinstance(metas[i], dict) else {} out.append({ "label": str(label), # bytes 참조(엔티티)는 내용 대신 표식만 — record 가 수십 MB 로 # 오염되던 E2E 실측 결함 fix "path": (str(src) if isinstance(src, (str, Path)) else f""), "asset_id": meta.get("asset_id"), "role": meta.get("pipeline_role"), }) return out # 판정 스키마 라벨 상한 = E (config still_recipe_roll_count validator 1~5 와 정합) ROLL_LABELS: List[str] = ["A", "B", "C", "D", "E"] # gen_fn(tag, prompt, labeled_refs, out_path) -> out_path GenFn = Callable[[str, str, Sequence[Tuple[str, Path]], Path], Path] # judge_fn(tag, prompt, labeled_refs, cand_paths, labels) -> JUDGE 스키마 dict JudgeFn = Callable[ [str, str, Sequence[Tuple[str, Path]], Sequence[Path], Sequence[str]], Dict[str, Any], ] # critique_fn(tag, prompt, labeled_refs, image_path) -> {"issues": [...]} CritiqueFn = Callable[[str, str, Sequence[Tuple[str, Path]], Path], Dict[str, Any]] def roll_labels(count: int) -> List[str]: """롤 수 → 후보 라벨 목록 (A..E, 1~5).""" if count < 1 or count > len(ROLL_LABELS): raise ValueError(f"roll count out of range: {count} (must be 1~5)") return ROLL_LABELS[:count] def build_judge_schema(labels: Sequence[str], with_readings: bool = True, with_physics: bool = False) -> Dict[str, Any]: """판정 JSON 스키마 — 실험 정본(s29 JUDGE_SCHEMA)의 라벨 가변판. ``with_readings`` (팩 v6, 2026-08-06): 후보마다 방향·복잡 구조물 내부 공간·중요 엔티티 정체를 **문장으로** 쓰게 하는 필드를 요구한다. 점수만 받으면 안 본 축을 안 본 채로 넘어간다는 것이 실측이다 — 총구가 상대를 전혀 겨누지 않은 후보가 최고점을 받았고 판정문에 조준 이야기가 한 줄도 없었다. 서술을 강제하면 그 축을 실제로 보게 된다. False 로 부르면 v5 이하와 byte 동일한 스키마다(회귀 안전판). ``with_physics`` (팩 v7, 2026-08-07): 공중에 뜬 몸·물체를 **무엇이 받치는가**를 쓰게 하는 네 번째 축. 심판 4종 대조가 근거다 — 걷는 자세를 가로로 눕혀 띄운 후보에 Opus·Sol·Qwen 셋이 최고점을 줬고 **셋 다 부유를 한 줄도 쓰지 않았다.** 같은 이미지에서 Gemini 만 "물리적으로 불가능한 연출"로 걸었다. v6 는 이것을 `hard_violations` 예시 목록 안에만 적어 두었는데, 목록에 있다는 것과 보게 만드는 것은 다르다. ★이 인자와 팩 selector 는 함께 움직인다. 팩 v7 은 PHYSICS 서술을 요구하므로 스키마에 자리가 없으면 모델이 쓸 곳을 잃는다 — 되돌릴 때도 둘을 같이 내린다. ★기본값은 **False** 다 (2026-08-07 Codex 2차 리뷰 수용). v7 은 스틸 전용이라 구조물 씨드·배경 플레이트는 v6 프롬프트를 그대로 쓰는데, 기본을 True 로 두면 그쪽 판정에 **v6 system("three axes")과 physics 필수 스키마가 모순으로** 들어간다 — 없는 축을 지어내거나 구조화 호출이 재시도로 샌다. selector 를 갈랐으면 스키마도 갈라야 한다. v7 을 쓰는 호출만 명시로 True 를 준다. """ labs = list(labels) verdict_props: Dict[str, Any] = { "label": {"type": "string", "enum": labs}, "score": {"type": "integer"}, "verdict_ko": {"type": "string"}, } required = ["label", "score", "verdict_ko"] schema: Dict[str, Any] = { "type": "object", "additionalProperties": False, "properties": { "winner": {"type": "string", "enum": labs}, "ranking": {"type": "array", "items": {"type": "string", "enum": labs}}, "verdicts": { "type": "array", "items": { "type": "object", "additionalProperties": False, "properties": verdict_props, "required": required, }, }, }, "required": ["winner", "ranking", "verdicts"], } if not with_readings: return schema schema["properties"]["readings"] = { "type": "array", "items": { "type": "object", "additionalProperties": False, "properties": { "label": {"type": "string", "enum": labs}, # 시선·총구·이동 방향이 실제로 무엇에 꽂히는가 "direction": {"type": "string"}, # 조작 장치 개수·좌석 배치·반사의 광학적 가능성 "built_space": {"type": "string"}, # 지목된 사물·인물이 그 사물·인물이 맞는가 "entities": {"type": "string"}, "hard_violations": {"type": "array", "items": {"type": "string"}}, }, "required": ["label", "direction", "built_space", "entities", "hard_violations"], }, } if with_physics: # 공중에 뜬 몸·물체를 무엇이 받치는가 — 도약의 도움닫기, 쥔 손, # 밀어낸 충격, 딛고 선 면. item = schema["properties"]["readings"]["items"] item["properties"]["physics"] = {"type": "string"} item["required"] = list(item["required"]) + ["physics"] schema["properties"]["all_candidates_fail"] = {"type": "boolean"} schema["required"] = ["winner", "ranking", "verdicts", "readings", "all_candidates_fail"] return schema def build_critique_schema( with_severity: bool = False, with_ref_gate: bool = False ) -> Dict[str, Any]: """결함 검사 JSON 스키마 — 실험 정본(s29 CRITIQUE_SCHEMA). with_severity (2026-08-13 사용자 지시 #106): 취합 이슈에 관찰자의 심각도를 **구조 필드로** 실어 보내는 계약 — v9 까지는 산문 규칙 ("critical 만 fix_en")뿐이라 취합자가 어겨도 코드가 못 막았다 (S39sh4: major 소파 지적이 fix 를 발동시켜 인물이 바뀜). True 는 QK v10 취합 전용 — 기존 경로(v7 critique·GQ v8)는 False 로 byte-identical 스키마(지문 불변). with_ref_gate (2026-08-19 사용자 지시 — 수정본 자세 변질): 지적마다 **어느 참조를 봐야 하는지** 말하는 칸. 종전에는 이 칸이 없어서 코드가 고를 방법 자체가 없었고, 그래서 편집 호출에 참조를 전부 붙였다 (`_critique_and_fix` 의 조건 없는 붙이기). 붙는 것 다수가 인물 정본 이고 그 라벨은 얼굴·머리·체형을 맞추라고 하므로 사람을 다시 그리라는 지시가 된다 — 실측 139장면 중 82장면(59%)이 개악으로 버려졌다. True 는 참조 선별 ON 경로 전용, False 는 byte-identical.""" schema: Dict[str, Any] = { "type": "object", "additionalProperties": False, "properties": { "issues": { "type": "array", "items": { "type": "object", "additionalProperties": False, "properties": { "issue_ko": {"type": "string"}, "fix_en": {"type": "string"}, # E2E13 fix⑤: 시점/카메라 이동이 필요해 i2i 국소 # 편집으로 고칠 수 없는 결함 — fix 조립에서 제외 # (판정=critique LLM, 코드는 bool 소비만) "unfixable": {"type": "boolean"}, # 2026-07-31: 건물의 전체 형태(층수·매스·footprint) # 결함은 국소 편집으로 고쳐지지 않는다 — 실측에서 # "Add a third full storey" 지시가 그대로 실패했고, # 재판정이 없던 탓에 그 실패본이 무판정 확정됐다. # true 면 i2i 편집 대신 **같은 브리프로 재생성**한다. "needs_regeneration": { "type": "boolean", "description": ( "true when the fault is in the structure's " "overall form — its storey count, massing, " "footprint or how many units it reads as " "holding — so that editing the finished " "photograph cannot repair it and the image " "must be generated again."), }, }, "required": ["issue_ko", "fix_en"], }, }, }, "required": ["issues"], } if with_severity: item = schema["properties"]["issues"]["items"] item["properties"]["severity"] = { "type": "string", "enum": ["critical", "major", "minor"]} # 관찰 배열 인덱스 왕복 (Codex R1 BLOCK-1): 취합이 severity 를 # 승격/강등하거나 관찰에 없는 이슈를 발명해도, 코드가 이 인덱스로 # 관찰 원값을 되찾아 덮어쓴다(관찰자=심각도 SOT). 미등록 인덱스는 # fail-closed 로 버려진다 — 산문 규칙이 아니라 구조가 지킨다. item["properties"]["observation_index"] = {"type": "integer"} item["required"] = list(item["required"]) + [ "severity", "observation_index"] if with_ref_gate: item = schema["properties"]["issues"]["items"] # 계약 문안은 시험(fix_ref_gate_pilot)에서 통과한 것을 그대로 옮긴다 # — 6장면 참조 23→9장, 생성 12장 전건 성공. 핵심은 "참조를 붙이는 # 것은 공짜가 아니다"를 모델에게 알리는 것이다. item["properties"]["needs_ref_indices"] = { "type": "array", "items": {"type": "integer"}, "description": ( "Numbers, from REFERENCE INDEX, of the reference " "photographs that must be seen to carry out this " "correction. Attaching a reference is not free: a " "reference the correction does not need still tells the " "editing model to look at it, and it then redraws parts " "nobody asked it to touch — people get restaged, poses " "reset, wardrobes swapped. Leave this empty when the " "correction is self-contained (erase something, move " "something, blur something, change an angle). A " "reference is needed only when the correction says to " "match, restore or copy something whose appearance is " "knowable only from that reference. Be strict; when in " "doubt, list nothing." ), } item["properties"]["adds_missing_entity"] = { "type": "boolean", "description": ( "true when the correction requires putting into the " "photograph a person or object that is NOT currently " "there at all." ), } item["properties"]["missing_entity_name"] = { "type": "string", "description": ( "What must be added, when adds_missing_entity is true; " "an empty string otherwise." ), } item["required"] = list(item["required"]) + [ "needs_ref_indices", "adds_missing_entity", "missing_entity_name"] return schema def build_gq_observe_schema() -> Dict[str, Any]: """G+Q 수정 흐름의 Qwen **관찰** 스키마 (2026-08-10). 관찰자는 "무엇이 잘못됐나"만 쓴다 — 수정문(fix_en)·수정 가능성 (unfixable/needs_regeneration)은 취합자(Gemini) 몫이라 여기 없다. severity 는 취합 참고 입력일 뿐 최종 issues 계약에는 실리지 않는다. 결함이 없으면 빈 배열 — 억지 지적은 수정 자체가 손실인 파이프라인 (critique 계약 "editing is not free")과 어긋난다. """ return { "type": "object", "additionalProperties": False, "properties": { "observations": { "type": "array", "items": { "type": "object", "additionalProperties": False, "properties": { "issue_ko": {"type": "string"}, "severity": { "type": "string", "enum": ["critical", "major", "minor"]}, }, "required": ["issue_ko", "severity"], }, }, }, "required": ["observations"], } def gemini_select(judge: Dict[str, Any]) -> Tuple[Dict[str, int], str]: """verdicts 점수 최고 후보 선정, 동점 = ranking 앞선 것 (s39 _gemini_select). verdicts 가 비면 winner 필드로 fallback (실험엔 없던 방어 — 판정 스키마가 required 라 정상 경로에선 도달하지 않는다). """ verdicts = judge.get("verdicts") or [] totals: Dict[str, int] = {v["label"]: v["score"] for v in verdicts} if not totals: winner = judge.get("winner") if not winner: raise ValueError("judge result has neither verdicts nor winner") return {}, winner best = max(totals.values()) tied = [lab for lab, t in totals.items() if t == best] rank: List[str] = judge.get("ranking") or [] selected = min(tied, key=lambda lab: rank.index(lab) if lab in rank else 99) return totals, selected # ── judge flip (still-variants 4택1, 2026-07-17 Codex BLOCKING-2) ───── # 순서 편향 방지: 정순·역순 2회 판정. 역순 호출의 display 라벨은 canonical # 라벨과 다르므로 winner 뿐 아니라 ranking·verdicts 전 라벨을 canonical 로 # 역매핑한 뒤에만 결합한다 — 미이행 시 역순 점수가 다른 후보에 합산된다. FLIP_POLICY_VERSION = 1 # 합의=승자 / 불일치=합산 최대 / 동점=priority 순 # E2E10 fix②: critique→i2i 수정본 무판정 확정 → [원본 vs 수정본] 2후보 # 블라인드 재판정(정역순 flip)으로 교체 시의 정책 버전 (지문 기여) # v2 (2026-08-07): ①동점 우선순위를 [수정본, 원본] → **[원본, 수정본]** 으로 # 반전 ②편집 호출에 critique 가 본 참조 동봉(fix_ref_label). 판정 결과의 # 의미가 달라지므로 버전을 올린다 — 기록만 보고 어느 정책으로 확정된 # 것인지 되짚을 수 있어야 한다. 지문에도 기여하므로 산출이 무효화된다. FIX_REJUDGE_POLICY_VERSION = "fix_rejudge_v2_orig_on_tie" # E2E11 fix③: GPT 구도 전담 critique 합류 — Gemini critique issues 와 # 합산해 하나의 수정 프롬프트로 i2i 하는 정책 버전 (지문 기여) GPT_COMPOSITION_POLICY_VERSION = "gpt_composition_v1_merge" # ── 재생성 이슈 정책 (2026-08-01) ───────────────────────────────────── # critique 가 `needs_regeneration` 으로 표시한 결함을 재생성 수단 없이 # 어떻게 다룰지. 호출자가 **명시로** 고른다 — 기본값을 조용히 바꾸면 이 # 공용 파이프를 쓰는 다른 소비자의 계약까지 함께 뒤집힌다. # # legacy_edit : 기존 동작. 재생성 수단이 있으면 재생성, 없으면 그 결함도 # i2i 편집 지시에 실어 보낸다. # defer : 재생성하지 않고 **손대지 않는다**. 그 결함은 편집 지시에서 # 빼고 무엇이 미처리로 남았는지 기록한다. 편집 대상이 0건이면 # 유료 호출 없이 원본을 유지한다. # # ★defer 의 근거는 실측이다. 편집의 사정거리 밖인 결함을 편집에 실으면 # 유료 호출 한 번을 버리면서 산출을 더 나쁘게 만든다 — 25그룹 실행에서 # 8그룹이 이 경로를 탔고, 그중 3그룹은 편집 가능한 결함이 하나도 없었다. # ── 수리 수단 (2026-08-29 사용자 지시) ──────────────────────────── # "지적 받은 내용을 바로 이미지를 i2i로 수정하는게 아니고 프롬프트에 # 지적 받은 내용 강화해서 다시 1장만 생성하는 형태로" # # edit : 기존 동작. 선정 원본을 i2i 로 고친다. # regenerate : 선정 롤이 **실제로 쓴 브리프**에 이번 지적을 붙여 한 장을 # 새로 그린다. 원본 산출은 참조로 안 준다(주면 같은 결함이 # 그대로 물려온다). 나온 후보는 기존 [원본 vs 수정본] # 블라인드 재판정을 그대로 탄다. # # ★기본은 `edit` 다 — 이 파이프를 공유하는 다른 소비자의 계약을 조용히 # 뒤집지 않는다. `auto`(지적 종류로 코드가 고름)는 **넣지 않았다**: # 지금 `needs_regeneration` 스키마가 층수·매스·footprint 전용이라 # 일반 스틸에서는 거의 발화하지 않아 사용자 지시와 다시 어긋난다. REPAIR_METHOD_EDIT = "edit" REPAIR_METHOD_REGENERATE = "regenerate" REPAIR_METHODS = (REPAIR_METHOD_EDIT, REPAIR_METHOD_REGENERATE) #: 지문 기여 — edit 은 기여 0(기존 소비자 byte 호환), regenerate 만 싣는다. REPAIR_METHOD_POLICY_VERSION = "repair_method_v1_regenerate" REGEN_ISSUE_POLICY_LEGACY_EDIT = "legacy_edit" REGEN_ISSUE_POLICY_DEFER = "defer" REGEN_ISSUE_POLICIES = ( REGEN_ISSUE_POLICY_LEGACY_EDIT, REGEN_ISSUE_POLICY_DEFER) # 지문 기여 — legacy 는 기여 0(기존 소비자 byte 호환), defer 만 싣는다. REGEN_ISSUE_POLICY_VERSION = "regen_issue_v1_defer" def validate_regeneration_contract( *, regeneration_issue_policy: str, regen_gen_fn: Optional[GenFn], fix_rejudge_fn: Optional[JudgeFn], brief_sources: Sequence[str] = (), ) -> None: """재생성 계약 선행 검증 — **첫 유료 호출 전에** 부른다. 재생성을 켰다면 ①원본 vs 재생성본 판정 계약과 ②다시 줄 브리프가 함께 있어야 한다. 재생성이 개선을 보장하지 않는다는 것은 실측이고(같은 개정절로 다시 그렸더니 한 그룹이 더 나빠졌다), 판정 없이 재생성본을 확정하면 그 악화가 그대로 굳는다 — v13 이 없애려던 "무판정 확정"이 방향만 바꿔 되살아난다. 그래서 기존 편집으로 **하강시키지 않고** 막는다. brief_sources: 재생성 브리프가 조립될 원천(롤 프롬프트들, 없으면 base 프롬프트). 실제 브리프는 판정으로 라벨이 뽑힌 뒤에야 정해지므로, 어느 라벨이 뽑히든 비지 않는지를 **진입 시점에** 원천으로 확인한다. 이 검사가 없으면 결손이 critique 뒤에야 드러나 롤 생성·판정·검사가 모두 유료로 나간 뒤에 막힌다(실측: gen 2 · judge 1 · critique 1). """ from app.core.errors import AppError if regeneration_issue_policy not in REGEN_ISSUE_POLICIES: raise AppError( code="regen_issue_policy_unknown", message=( f"알 수 없는 재생성 이슈 정책: {regeneration_issue_policy!r} " f"(가능: {', '.join(REGEN_ISSUE_POLICIES)})"), status_code=422) if regen_gen_fn is None: return if fix_rejudge_fn is None: raise AppError( code="regen_without_rejudge", message=( "재생성이 켜져 있는데 원본 vs 재생성본 판정 계약이 없다 — " "무판정 확정을 막기 위해 실행하지 않는다"), status_code=422) sources = [s for s in brief_sources] if not sources or not all((s or "").strip() for s in sources): raise AppError( code="regen_brief_missing", message=( "재생성이 켜져 있는데 다시 줄 브리프의 원천이 비었다 — " "롤을 그리기 전에 막는다"), status_code=422) def flip_display_to_canonical(labels: Sequence[str]) -> Dict[str, str]: """역순 제시의 display 라벨 → canonical 라벨 매핑.""" labs = list(labels) return {labs[i]: labs[len(labs) - 1 - i] for i in range(len(labs))} def _validate_judge_shape(judge: Any, labels: Sequence[str]) -> None: """판정 완전성 fail-closed — verdict N개·중복 0, ranking=permutation, winner 유효. 위반=ValueError (호출자가 샷 단위 격리). 재리뷰 NARROW-1: durable record 는 임의 JSON 손상이 가능 — root/ verdicts/ranking/각 항목의 **타입까지** 전부 ValueError 로 귀결하는 total validator 여야 재사용 경로(except ValueError)가 회복 가능하다. AttributeError/TypeError 누수=resume 영구 차단. """ labs = list(labels) if not isinstance(judge, dict): raise ValueError( f"judge result not an object: {type(judge).__name__}" ) verdicts = judge.get("verdicts") if not isinstance(verdicts, list): raise ValueError( f"judge verdicts not a list: {type(verdicts).__name__}" ) seen: List[str] = [] for v in verdicts: if not isinstance(v, dict): raise ValueError( f"judge verdict item not an object: {type(v).__name__}" ) lab = v.get("label") if not isinstance(lab, str): raise ValueError(f"judge verdict label not a string: {lab!r}") score = v.get("score") if not isinstance(score, int) or isinstance(score, bool): raise ValueError(f"judge verdict score not an int: {score!r}") seen.append(lab) if len(seen) != len(labs) or set(seen) != set(labs): raise ValueError( f"judge verdicts malformed — labels {seen} != {labs}" ) ranking = judge.get("ranking") if not isinstance(ranking, list) or not all( isinstance(r, str) for r in ranking ): raise ValueError(f"judge ranking not a string list: {ranking!r}") if sorted(ranking) != sorted(labs): raise ValueError( f"judge ranking not a permutation — {ranking} vs {labs}" ) if judge.get("winner") not in labs: raise ValueError(f"judge winner invalid — {judge.get('winner')!r}") def normalize_flip_verdict( judge_raw: Dict[str, Any], display_to_canonical: Mapping[str, str], labels: Sequence[str], ) -> Dict[str, Any]: """display 판정을 canonical 라벨로 역매핑 (winner·ranking·verdicts 전부). 입력 검증 fail-closed — malformed 판정은 결합 전에 reject. ★`all_candidates_fail` 과 `readings` 도 함께 옮긴다 (2026-08-07, Codex 리뷰 수용). 종전에는 세 키만 남기고 버렸는데, 팩은 "전 후보가 한 축을 실패하면 그렇게 선언하라, 호출측이 재촬영 여부를 그 값으로 정한다"고 적어 두었고 스키마도 필수로 요구한다. 여기서 버리면 심판이 자동차 기하 붕괴나 공중 부유를 **전 후보에서** 잡아내도 그 사실이 기록에 남지 않는다. readings(방향·공간·엔티티·물리 서술)도 같은 이유로 남긴다. """ _validate_judge_shape(judge_raw, labels) remap = dict(display_to_canonical) if set(remap) != set(labels) or set(remap.values()) != set(labels): raise ValueError(f"display→canonical 매핑 불완전: {remap}") out: Dict[str, Any] = { "winner": remap[judge_raw["winner"]], "ranking": [remap[lab] for lab in judge_raw["ranking"]], "verdicts": [ {**v, "label": remap[v["label"]]} for v in judge_raw["verdicts"] ], } if "all_candidates_fail" in judge_raw: out["all_candidates_fail"] = bool(judge_raw["all_candidates_fail"]) if judge_raw.get("readings"): out["readings"] = [ {**r, "label": remap.get(r.get("label"), r.get("label"))} for r in judge_raw["readings"] ] return out def combine_flip_verdicts( judge_fwd: Dict[str, Any], judge_rev_normalized: Dict[str, Any], labels: Sequence[str], priority: Sequence[str], ) -> Tuple[str, Dict[str, Any]]: """정순+역순(canonical 정규화 완료) 결합 — (selected, combined). 합의=그 승자 / 불일치=양회 점수 합산 최대 / 동점=priority 앞선 라벨. normalized 입력만 소비한다(raw display 판정 전달 금지). """ _validate_judge_shape(judge_fwd, labels) _validate_judge_shape(judge_rev_normalized, labels) prio = list(priority) if sorted(prio) != sorted(list(labels)): raise ValueError(f"priority 는 labels permutation 이어야: {prio}") fwd_scores = {v["label"]: v["score"] for v in judge_fwd["verdicts"]} rev_scores = { v["label"]: v["score"] for v in judge_rev_normalized["verdicts"] } totals = {lab: fwd_scores[lab] + rev_scores[lab] for lab in labels} agreement = judge_fwd["winner"] == judge_rev_normalized["winner"] if agreement: selected = judge_fwd["winner"] else: best = max(totals.values()) tied = [lab for lab in labels if totals[lab] == best] selected = min(tied, key=prio.index) ranking = sorted(labels, key=lambda lab: (-totals[lab], prio.index(lab))) combined = { "totals": totals, "ranking": ranking, "agreement": agreement, "policy": FLIP_POLICY_VERSION, } return selected, combined def collect_missing_entities( issues: Sequence[Dict[str, Any]] ) -> List[str]: """"사진에 아예 없어 새로 넣어야 하는 것" 이름 목록 (2026-08-19). 참조 선별의 짝이다 — 없는 것을 넣으라는 지적은 그 참조를 봐야 하고, 수정 지시문도 "여기 있는 것만 고치고 나머지는 그대로"에서 벗어난다는 사실을 알아야 한다(사용자 지시). 순서 보존 중복 제거. """ out: List[str] = [] for issue in issues: if not isinstance(issue, dict) or not issue.get("adds_missing_entity"): continue name = str(issue.get("missing_entity_name") or "").strip() if name and name not in out: out.append(name) return out def select_fix_refs( issues: Sequence[Dict[str, Any]], labeled_refs: Sequence[Tuple[str, Any]], ) -> Tuple[List[Tuple[str, Any]], Dict[str, Any]]: """지적이 **요구한 참조만** 고른다 (2026-08-19 사용자 지시). 반환 (고른 참조, 감사 기록). 번호는 `ref_index_part` 가 모델에게 보인 1-based 순서이며 그 순서가 계약이다. ★칸이 없는 지적이 하나라도 섞이면 **거르지 않는다.** 그 지적에게는 무엇을 봐야 하는지 물은 적이 없으므로, 여기서 빼면 참조 없이 글자로만 지시하던 2026-08-07 이전 상태로 되돌아간다(그때 실측 4건이 실패했다). 답은 "참조를 빼라"가 아니라 "요구한 것만 붙여라"다 — 물은 적 없는 지적에게는 종전대로 전부 붙이고, 그 사실을 기록에 남긴다. ★범위 밖·정수 아님은 fail-closed 로 버린다(발명한 번호로 엉뚱한 참조가 붙는 것이 안 붙는 것보다 나쁘다). 버린 값은 기록에 남는다. """ refs = list(labeled_refs) issue_list = [i for i in issues if isinstance(i, dict)] ungated = [i for i in issue_list if "needs_ref_indices" not in i] if ungated: return refs, { "mode": "all_ungated", "available": len(refs), "kept": len(refs), "ungated_issue_count": len(ungated), } # "없는 것을 새로 넣어라"인데 참조를 하나도 요구하지 않은 지적의 수. # 그때 편집 모델은 넣을 것의 생김새를 지어내야 한다 — 이름과 참조를 # 코드로 잇는 유일한 수단이 글자 대조라 금지되므로(정본 없는 일반 소품 # 이면 무조건 동봉이 오히려 틀리다) 강제하지 않고 **세기만** 한다. # 다음 주행에서 이 값이 0 이 아니면 그 샷을 눈으로 보고 문안을 손본다. missing_no_ref = sum( 1 for i in issue_list if i.get("adds_missing_entity") and not (i.get("needs_ref_indices"))) wanted: set = set() invalid: List[Any] = [] for issue in issue_list: for n in (issue.get("needs_ref_indices") or []): if isinstance(n, bool) or not isinstance(n, int) \ or not (1 <= n <= len(refs)): invalid.append(n) continue wanted.add(n) order = sorted(wanted) kept = [refs[n - 1] for n in order] rec: Dict[str, Any] = { "mode": "gated", "available": len(refs), "kept": len(kept), "requested_indices": order, } if invalid: rec["invalid_indices"] = invalid[:20] if missing_no_ref: rec["missing_without_ref"] = missing_no_ref return kept, rec def build_fix_prompt( issues: Sequence[Dict[str, str]], fix_head: str, fix_tail: str, missing_names: Sequence[str] = (), missing_head: str = "", missing_tail: str = "", ) -> str: """FIX_HEAD + CORRECTIONS 목록 + FIX_TAIL (s40 _critfix 조립). missing_* (2026-08-19): 지적이 "지금 사진에 없는 것을 새로 넣어라"고 할 때만 그 절이 CORRECTIONS 뒤에 낀다. 기본값(빈 목록·빈 문안)은 기존 조립과 byte-identical. """ corrections = "CORRECTIONS:\n" + "\n".join( f"- {i['fix_en']}" for i in issues ) parts = [fix_head, corrections] names = [n for n in missing_names if str(n).strip()] if names and missing_head.strip(): block = missing_head.strip() + "\n" + "\n".join( f"- {n}" for n in names) if missing_tail.strip(): block += "\n" + missing_tail.strip() parts.append(block) parts.append(fix_tail) return "\n\n".join(parts) # 구조 결함 재생성 — i2i 국소 편집이 손댈 수 없는 층위(2026-07-31 실측). # 편집 프롬프트와 달리 **원본을 참조로 주지 않는다**: 참조가 있으면 모델이 # 그 매스를 그대로 물려받아 같은 층수가 다시 나온다. REGEN_HEAD = ( "Generate this location photograph again from the brief below.\n\n" "An earlier attempt from this same brief was rejected for the " "structural faults listed at the end. Those faults are in the " "building's overall form — its storey count, massing or footprint — " "which no amount of retouching a finished photograph can repair. So " "this is a fresh photograph, not an edit of the previous one." ) REGEN_TAIL = ( "Build the structure so that none of those faults occurs this time, " "and keep everything else the brief requires. Where the brief states " "a storey count, count the storeys you are drawing and make them " "match it exactly — a level used for parking or storage still counts " "as a storey." ) #: 구조물 재생성의 결함 목록 머리말 — 기존 소비자(구조물 seed)의 기본값. # ★스틸은 이 말을 쓸 수 없다. 「STRUCTURAL」이 층수·매스·footprint 를 # 가리키는데 스틸의 지적은 인물·소품·시선·프레이밍이다. 그래서 인자로 # 뺐고, 스틸은 자기 팩의 머리말을 넘긴다. REGEN_FAULTS_HEADER = "STRUCTURAL FAULTS OF THE REJECTED ATTEMPT:" def build_regen_prompt( issues: Sequence[Dict[str, str]], base_prompt: str, regen_head: str = "", regen_tail: str = "", faults_header: str = "", ) -> str: """원 브리프 + 결함 목록 → 재생성 프롬프트. base_prompt 는 선정된 롤이 실제로 쓴 프롬프트다. 같은 브리프를 다시 주되 무엇이 틀렸는지를 덧붙인다 — 프롬프트를 바꿔 버리면 무엇이 효과를 냈는지 알 수 없고, 브리프 준수 여부도 판정할 수 없다. faults_header (2026-08-29): 결함 목록의 머리말. 비우면 구조물용 기본값 이라 **기존 소비자는 byte-identical** 이다. 스틸은 자기 팩의 말을 넘긴다 — 구조물 문안을 스틸에 재사용하면 지적의 대상이 뒤바뀐다. """ faults = (faults_header or REGEN_FAULTS_HEADER).rstrip() + "\n" + "\n".join( f"- {i.get('fix_en') or i.get('issue_ko') or ''}".rstrip() for i in issues ) return "\n\n".join([ (regen_head or REGEN_HEAD).strip(), base_prompt.strip(), faults, (regen_tail or REGEN_TAIL).strip(), ]) def compute_input_fingerprint( *, prompt: str, labeled_refs: Sequence[Tuple[str, Any]], roll_count: int, critique_enabled: bool, extra: Optional[Dict[str, Any]] = None, ) -> str: """생성 입력 지문 — 프롬프트+참조(라벨·내용)+롤/게이트+모델·팩(extra). Codex 1차 리뷰 BLOCKING-3: 파일 존재만으로 skip 하면 입력이 바뀌어도 옛 산출을 재사용한다 — record 의 이 지문과 대조해 mismatch 시 산출 정리. """ import hashlib import json as _json h = hashlib.sha256() h.update(prompt.encode("utf-8")) for label, src in labeled_refs: h.update(b"\x00") h.update(str(label).encode("utf-8")) data = ( Path(src).read_bytes() if isinstance(src, (str, Path)) else src ) h.update(hashlib.sha256(data).digest()) h.update(f"|{roll_count}|{critique_enabled}".encode("utf-8")) if extra: h.update(_json.dumps(extra, sort_keys=True).encode("utf-8")) return h.hexdigest()[:16] def _sha256_file(path: Path) -> str: """산출 파일의 sha — **무엇을 샀는지**의 신원. 재롤은 유료라, 「이 파일이 그 시도의 산출이 맞나」를 나중에 되짚을 수 있어야 한다. 경로만 남기면 덮어쓰기와 구분되지 않는다. """ import hashlib try: return hashlib.sha256(path.read_bytes()).hexdigest() except OSError as exc: # noqa: BLE001 logger.warning("multiroll_select: %s sha 계산 실패: %s", path, exc) return "" def gate_winner_violations( judge_result: Dict[str, Any], label: str, ) -> List[str]: """그 후보에 달린 **실격 문장 그대로** — 재롤 처방의 재료. ★판정기가 쓴 말을 **그대로 되돌려 준다**. 글자에서 뜻을 캐내 분류하지 않는다([[feedback-no-literal-substring-meaning]]) — 여기서 하는 일은 「무엇이 틀렸다고 했는지」를 다시 보여 주는 것뿐이다. """ rows = judge_result.get("readings") if not isinstance(rows, list): return [] for r in rows: if isinstance(r, dict) and r.get("label") == label: raw = r.get("hard_violations") return [str(x) for x in raw] if isinstance(raw, list) else [] return [] def _clear_outputs(out_stem: Path) -> None: """이 stem 의 롤/선정/수정/변환 산출 제거 — force·지문 mismatch·무기록 sel. ★**자기 산출 allowlist 로만 지운다** — 광역 `{stem}_*` glob 은 같은 접두를 쓰는 **타 단계 입력**까지 지웠다. 실측 두 번: · 2026-08-08 S1sh3 `__bgfirst_bg`(직전에 만든 배경판) — 밑줄 둘 네임스페이스 제외로 막았다. · 2026-08-14 S18sh5/S79sh1/S84sh2 `_confinedfp`(confined 도면) — 밑줄 하나라 그 가드 밖. 정리가 3분 전 만든 도면을 지워 직후 롤 생성이 입력 결손 FileNotFoundError 로 죽고, resume 마다 도면 재생성→정리 삭제→실패가 **결정론으로 재발**해 autodrive 표적 회수 상한 3회를 소진시켰다(1차 E2E 212/215 정지 원인). 제외 목록을 늘리는 대신 이 단계가 실제로 만드는 파일만 지운다: 롤(`_a`~`_e`)·`_sel`·`_fix`(재생성 후보도 이 이름)·`_cine`(변환 스테이지 산출 — 낡은 sel 의 파생물이라 함께 치운다, 종전 glob 도 지우던 대상이라 동작 불변). """ names = {f"{out_stem.name}_{lab.lower()}.png" for lab in roll_labels(5)} names |= {f"{out_stem.name}_sel.png", f"{out_stem.name}_fix.png", f"{out_stem.name}_cine.png"} for p in out_stem.parent.glob(f"{out_stem.name}_*.png"): if p.name not in names: continue try: p.unlink() except OSError as exc: # noqa: PERF203 logger.warning("multiroll_select: %s 제거 실패: %s", p, exc) def _roll_path(out_stem: Path, label: str) -> Path: return out_stem.parent / f"{out_stem.name}_{label.lower()}.png" def _sel_path(out_stem: Path) -> Path: return out_stem.parent / f"{out_stem.name}_sel.png" def _critique_and_fix( *, tag: str, prompt: str, labeled_refs: Sequence[Tuple[str, Path]], out_stem: Path, selected: str, critique_fn: CritiqueFn, fix_gen_fn: GenFn, fix_head: str, fix_tail: str, fix_label: str, record: Dict[str, Any], fix_ref_label: str = "", fix_ref_gate: bool = False, fix_missing_head: str = "", fix_missing_tail: str = "", fix_rejudge_fn: Optional[JudgeFn] = None, composition_critique_fn: Optional[CritiqueFn] = None, regen_gen_fn: Optional[GenFn] = None, regen_prompt: str = "", regen_head: str = "", regen_tail: str = "", regen_faults_header: str = "", regeneration_issue_policy: str = REGEN_ISSUE_POLICY_LEGACY_EDIT, repair_method: str = REPAIR_METHOD_EDIT, ) -> Path: """선정 원본 결함 검사 → 결함 시 수리(편집 또는 재생성) → _sel 확정. fix_ref_label (2026-08-07): 비어 있지 않으면 편집 호출에 **critique 가 본 참조를 그대로 동봉**한다(편집 대상이 첫 장, 참조는 이 라벨을 앞에 달고 뒤). 빈 문자열=기존 동작(원본 단독 참조) byte-identical. fix_ref_gate (2026-08-19 사용자 지시): True 면 동봉을 **지적이 요구한 참조로 좁힌다** — 판단 재료는 critique 가 구조 필드로 내놓은 `needs_ref_indices`(`build_critique_schema(with_ref_gate=True)`). 아무도 요구하지 않으면 편집 대상 원본 한 장만 간다. 없는 것을 새로 넣으라는 지적이 있으면 `fix_missing_head`/`fix_missing_tail` 절이 지시문에 끼고 그 참조도 함께 붙는다. False(기본)=byte-identical. regen_gen_fn / regen_prompt (2026-07-31): critique 가 어떤 결함을 `needs_regeneration` 으로 표시하면 그 결함은 i2i 편집으로 못 고친다 — 건물의 층수·매스·footprint 가 그렇다. 이때는 편집 대신 **원 브리프로 다시 생성**하고, 나온 후보는 기존 fix 재판정 경로를 그대로 탄다(원본 vs 재생성본 블라인드 비교). regeneration_issue_policy (2026-08-01): 재생성 수단이 없을 때 그 결함을 어떻게 다룰지 — `legacy_edit`(기본, 편집에 실어 보냄) 또는 `defer`(손대지 않고 기록만). 기본값을 바꾸지 않는 이유는 이 파이프를 공유하는 다른 소비자의 계약을 조용히 뒤집지 않기 위해서다. 상세는 모듈 상단 REGEN_ISSUE_POLICY_* 주석. composition_critique_fn (E2E11 fix③, 사용자 확정): 제공 시 GPT 구도 전담 검사(구도·배치·스케일·시선축만 — sys 계약이 그 외 봉인)를 Gemini critique 와 병행 실행, 두 issues 를 **합산한 하나의 수정 프롬프트**로 i2i. 구도 위반만 있어도 수정 진행. None(default)=기존 동작. fix_rejudge_fn (E2E10 fix②): 제공 시 수정본을 무판정 확정하지 않고 [원본, 수정본] 2후보 블라인드 재판정(정순+역순 flip — 판정자에게 어느 쪽이 수정본인지 미노출, 순서 편향은 flip 결합으로 상쇄)으로 최종 _sel 을 결정. 동점 priority=[원본, 수정본] — 2026-08-07 반전, 근거는 호출부 주석. None(default)=기존 동작. """ sel = _sel_path(out_stem) orig = _roll_path(out_stem, selected) crit = critique_fn(tag, prompt, labeled_refs, orig) record["critique"] = crit issues = list(crit.get("issues") or []) if composition_critique_fn is not None: comp = composition_critique_fn( f"{tag}_comp", prompt, labeled_refs, orig) record["composition_critique"] = comp # 합산 — Gemini(결함) 뒤에 GPT(구도) 나열, 단일 수정 프롬프트 issues += list(comp.get("issues") or []) def _atomic_place(src: Path) -> None: tmp = sel.with_name(sel.name + ".tmp") shutil.copy(src, tmp) tmp.replace(sel) if not issues: _atomic_place(orig) record["fix_skipped"] = True return sel # ── 이슈 partition — 한 번만, 우선순위 고정 ────────────────────── # ①unfixable > ②deferred(재생성 몫) > ③editable. # 두 bool 이 겹칠 때 같은 이슈가 두 갈래로 세어지면 감사 수치가 틀어진다. # # ①E2E13 fix⑤: unfixable(시점/카메라 이동류)은 i2i 국소 편집으로 고칠 # 수 없음 — fix 프롬프트에서 제외(L05B01 실측: 카메라 이동 지시가 두 # 번째 철문을 생성해 hard violation → 반전 원본이 확정되던 경로). # ②defer 정책에서 `needs_regeneration` 은 재생성이 있어야만 다룰 수 # 있는 결함이다 — 없으면 손대지 않는다(모듈 상단 주석 참조). deferring = regeneration_issue_policy == REGEN_ISSUE_POLICY_DEFER # ★재생성 모드에서는 두 배제 사유가 **둘 다 성립하지 않는다** # (2026-08-29, Codex 설계 판정 수용). # # `unfixable` 의 뜻은 「i2i 국소 편집으로 못 고침」(시점·카메라 이동류) # 이고 `needs_regeneration` 의 뜻은 「편집 대신 재생성해야 함」이다. # 둘 다 **편집을 전제로 한 배제**라, 애초에 새로 그리는 자리에서는 # 뺄 이유가 없다. 여기서 빼면 사용자가 재생성으로 고치려던 바로 그 # 결함이 지시문에서 사라진다. regenerating = repair_method == REPAIR_METHOD_REGENERATE fixable: List[Dict[str, Any]] = [] deferred: List[Dict[str, Any]] = [] unfixable: List[Dict[str, Any]] = [] for issue in issues: if issue.get("unfixable") and not regenerating: unfixable.append(issue) elif deferring and issue.get("needs_regeneration") and not regenerating: deferred.append(issue) else: fixable.append(issue) if deferred: record["regen_deferred_issue_count"] = len(deferred) record["regen_deferred_issues"] = deferred logger.info( "multiroll_select[%s]: 재생성 몫 결함 %d건 — 편집하지 않고 보류", tag, len(deferred)) if not fixable: _atomic_place(orig) record["fix_skipped"] = True # ★사유는 실제 partition 으로 정한다. 둘이 섞였는데 한쪽 이름을 # 붙이면 감사 기록이 거짓이 된다(Codex 지적, 수용). if deferred and unfixable: record["fix_skip_reason"] = "all_issues_non_editable" elif deferred: record["fix_skip_reason"] = "all_issues_need_regeneration" else: record["fix_skip_reason"] = "all_issues_unfixable" return sel # ── 심각도 게이트 (2026-08-13 사용자 지시 #106) ────────────────── # "매우 잘못된 게 없는데 계속 fix 를 한다" — critical 만 편집 대상, # major/minor 는 기록으로만 남긴다. 실측 근거=S39sh4: 관찰자가 소파 # 소실을 major 로 정확히 매겼는데 취합이 산문 규칙을 어기고 fix_en # 에 실었고, 그 fix 가 인물을 바꿨는데 재판정이 채택했다. severity # 필드가 하나도 없는 이슈 목록(구 팩 v7 critique·GQ v8 경로)은 기존 # 동작 그대로 — 게이트는 심각도가 실려 오는 계약(QK v10)에서만 산다. if any("severity" in issue for issue in fixable): below_critical = [ i for i in fixable if i.get("severity") != "critical"] fixable = [ i for i in fixable if i.get("severity") == "critical"] if below_critical: record["fix_severity_skipped_count"] = len(below_critical) record["fix_severity_skipped"] = below_critical if not fixable: _atomic_place(orig) record["fix_skipped"] = True record["fix_skip_reason"] = "no_critical_issue" return sel fixed = out_stem.parent / f"{out_stem.name}_fix.png" # ★구조 결함은 편집이 아니라 재생성으로 — 국소 편집은 매스를 못 바꾼다. regen_issues = [i for i in fixable if i.get("needs_regeneration")] if regenerating and not all( x.strip() for x in (regen_head, regen_tail, regen_faults_header)): # ★스틸 문안이 비면 `build_regen_prompt` 가 **구조물 기본값**으로 # 떨어진다 — 「STRUCTURAL FAULTS」「storey count · massing · # footprint」. 스틸 지적(인물·소품·시선·프레이밍)에 그 말이 붙으면 # 지적의 대상이 통째로 뒤바뀌고, 기록에는 스틸 재생성으로 남는다. # 조용히 떨어지지 않고 막는다. from app.core.errors import AppError raise AppError( code="repair_regenerate_wording_missing", message=( "수리 수단이 regenerate 인데 재생성 문안(head/tail/머리말)이 " "비었다 — 구조물 기본 문안으로 내려가지 않고 막는다"), status_code=422) if regenerating and regen_gen_fn is None: # ★수단을 재생성으로 골라 놓고 그릴 함수가 없으면 **편집으로 # 내려가지 않는다.** 내려가면 사용자가 바꾸라고 한 바로 그 수단이 # 조용히 옛 것으로 돌아가고 기록에는 새 것으로 남는다. from app.core.errors import AppError raise AppError( code="repair_regenerate_without_gen_fn", message=( "수리 수단이 regenerate 인데 재생성 함수가 없다 — " "편집으로 내려가지 않고 막는다"), status_code=422) if (fixable if regenerating else regen_issues) and \ regen_gen_fn is not None and not regen_prompt.strip(): # ★브리프 결손을 편집으로 하강시키지 않는다. 진입 시점에는 브리프가 # 아직 조립되지 않아 선행 검증이 잡을 수 없는 구멍이다 — 여기서 # 내려가면 편집으로 못 고치는 결함이 다시 편집 지시에 실린다. from app.core.errors import AppError raise AppError( code="regen_brief_missing", message=( "재생성이 켜져 있는데 다시 줄 브리프가 비었다 — " "편집으로 내려가지 않고 막는다"), status_code=422) # ★재생성으로 갈지 정하는 자리 (2026-08-29). # # 종전에는 `needs_regeneration` 표시가 있어야만 재생성이었다. 그 표시는 # 구조물 전용 스키마라 일반 스틸에서는 거의 안 붙는다 — 사용자가 말한 # 「지적을 프롬프트에 강화해서 다시 1장」이 그대로는 일어나지 않는다. # 그래서 **수단을 명시로 고르면** 게이트를 통과한 지적 **전체**를 한 # 번의 재생성에 싣는다. 기본(`edit`)에서는 종전과 한 글자도 다르지 않다. repair_issues = fixable if regenerating else regen_issues if repair_issues and regen_gen_fn is not None and regen_prompt.strip(): regen = build_regen_prompt( repair_issues, regen_prompt, regen_head, regen_tail, faults_header=regen_faults_header) record["regen_prompt"] = regen record["regen_issue_count"] = len(repair_issues) record["repair_mode"] = "regenerate" record["repair_method"] = repair_method logger.info( "multiroll_select[%s]: 결함 %d건 — i2i 편집 대신 재생성 (%s)", tag, len(repair_issues), "명시 수단" if regenerating else "구조 결함") # ★원본 산출만 빼고 원래 참조는 그대로 넘긴다. 원본을 참조로 # 물리면 같은 매스가 재현되지만, 참조를 통째로 비우면 형태 권위 # (구조 스케치 등)까지 함께 사라진다 — 재생성은 "이 브리프와 이 # 참조로 다시 찍는 것"이지 "맨손으로 다시 그리는 것"이 아니다. regen_gen_fn(f"{tag}_regen", regen, list(labeled_refs), fixed) else: # ── 참조 선별 (2026-08-19 사용자 지시) ─────────────────────── # 없는 것을 새로 넣으라는 지적이 있을 때만 그 절이 지시문에 낀다. missing_names = ( collect_missing_entities(fixable) if fix_ref_gate else []) fix_prompt = build_fix_prompt( fixable, fix_head, fix_tail, missing_names=missing_names, missing_head=fix_missing_head, missing_tail=fix_missing_tail, ) if missing_names: record["fix_missing_entities"] = missing_names record["repair_mode"] = "edit" record["repair_method"] = repair_method # ★critique 가 본 참조를 fix 도 본다 (2026-08-07). # # 여기서 오래 어긋나 있었다. critique 는 `labeled_refs` 전체(캐릭터 # 정본·이전 샷·배경)를 보고 지적하는데 편집은 선정 원본 한 장만 받아, # 지적 내용이 **글자로만** 전달됐다. 실측 4건: "녹색 비니를 캐릭터 # 참조와 일치시켜라" → 둘 다 비니 / "어머니를 딸로 교체" → 옷이 전부 # 달라짐 / "긴 흰머리를 묶어라" → 머리 모양이 달라짐 / "비석 글자를 # 이전 샷 형태로" → 이전 샷을 못 봐서 고치지 못함. # # 순서가 계약이다 — 편집 대상이 **첫 장**이어야 한다. 참조는 뒤에 # 붙이고 라벨로 "이건 편집 대상이 아니다"를 못 박는다(fix_ref_label). # 라벨을 갈아 끼우는 이유는 원 라벨이 생성용 문구이기 때문이다: # "이 사진에 LOCKED" 류를 편집 맥락에 그대로 넣으면 편집 대상과 # 참조 중 어느 쪽을 잠그라는 것인지 모호해진다. # # ★2026-08-19: 그런데 **요구하지도 않은 참조까지 전부** 붙었다. # 붙는 것 다수가 인물 정본이고 그 라벨은 얼굴·머리·체형을 맞추라고 # 한다 — 시계 바늘 하나 고치라는 지적에 사람을 다시 그리라는 지시가 # 함께 나가는 셈이다. 실측 139장면 중 82장면(59%)이 개악으로 # 버려졌고, 퇴짜/채택을 가른 것은 지시문 길이도 항목 수도 아닌 # **참조 장수**뿐이었다(중앙 4장 대 3장). 그래서 지적이 요구한 # 것만 붙인다. 판단은 지적을 만드는 단계가 구조 필드로 내놓는다 # (`needs_ref_indices` — 글자 대조 아님). fix_refs: List[Tuple[str, Any]] = [(fix_label, orig)] if fix_ref_label: if fix_ref_gate: gated_refs, gate_rec = select_fix_refs(fixable, labeled_refs) # 번호만 남기면 읽는 쪽이 "그 번호가 어느 목록을 가리키나"를 # 스스로 판단해야 한다 — 실제로 붙은 라벨을 함께 남긴다. gate_rec["kept_labels"] = [lab for lab, _ in gated_refs] record["fix_ref_gate"] = gate_rec if gate_rec.get("mode") == "all_ungated": logger.warning( "multiroll_select[%s]: 참조 선별 ON 인데 칸 없는 " "지적 %d건 — 이번 편집은 종전대로 전부 붙인다", tag, gate_rec.get("ungated_issue_count")) else: gated_refs = list(labeled_refs) fix_refs += [(f"{fix_ref_label} {lab}", src) for lab, src in gated_refs] record["fix_ref_count"] = len(fix_refs) fix_gen_fn(f"{tag}_fix", fix_prompt, fix_refs, fixed) record["fix_prompt"] = fix_prompt if fix_rejudge_fn is None: _atomic_place(fixed) return sel # ── 2후보 재판정 — canonical A=원본, B=수정본 ── labels2 = ["A", "B"] cand = {"A": orig, "B": fixed} # ★판정자가 순서를 스스로 소유하면 **한 번만** 부른다 (2026-08-29). # cross-model 어댑터가 Gemini 정순 + Grok 역순을 그 한 번 안에서 # 병렬로 다 한다 — 여기서 또 뒤집으면 4콜이 된다. # canonical A=원본 · B=수정본 계약과 **동점은 원본 유지** 정책은 # 그대로다(아래 `_priority` 가 그것을 정한다). if getattr(fix_rejudge_fn, "owns_order", False): judged = fix_rejudge_fn( f"{tag}_fixjudge", prompt, labeled_refs, [cand["A"], cand["B"]], labels2, ) _validate_judge_shape(judged, labels2) record["fix_rejudge"] = { "cross_model_order": judged.get("cross_model_order"), "winner": judged.get("winner"), "ranking": judged.get("ranking"), "verdicts": judged.get("verdicts"), # ★`fix_won` 은 하류·감사가 쓰는 이름이라 그대로 둔다. "fix_won": judged.get("winner") == "B", # ★원본·수정본 **둘 다** 못 쓴다는 선언 (Codex BLOCK, 2026-08-29). # # 처음엔 이 줄을 빠뜨렸다. `_set_needs_reshoot` 는 재판정 기록에 # 이 **키가 있는지**로 「재판정이 최종을 정했다」를 판별하고, # 없으면 `initial_roll_all_fail` 로 되돌아간다. 그래서 초기 롤은 # 통과했는데 재판정에서 둘 다 실패한 샷이 **재촬영 표시 없이 # 정상 산출처럼** 남았다. 값이 아니라 **키의 부재**가 갈래를 # 바꾸는 자리라 「False 는 안 쓴다」가 통하지 않는다 — 늘 쓴다. # # legacy 갈래는 정·역 **둘 다** 실패일 때만 참으로 쳤다. 여기서는 # `judged` 가 이미 두 슬롯을 합친 결과다 — agree 는 Gemini 선언, # combined 는 `n_fail == len(per_model)`, 단독 생존은 그 슬롯의 # 선언. 어느 갈래든 **최종을 정한 판정의 선언**이라 계약이 같다. "all_candidates_fail": bool(judged.get("all_candidates_fail")), } _atomic_place(cand[str(judged.get("winner") or "A")]) return sel fwd_raw = fix_rejudge_fn( f"{tag}_fixjudge", prompt, labeled_refs, [cand["A"], cand["B"]], labels2, ) fwd = normalize_flip_verdict( fwd_raw, {lab: lab for lab in labels2}, labels2) rev_raw = fix_rejudge_fn( f"{tag}_fixjudge_rev", prompt, labeled_refs, [cand["B"], cand["A"]], labels2, ) rev = normalize_flip_verdict( rev_raw, flip_display_to_canonical(labels2), labels2) # ★동점은 원본 유지 (2026-08-07 정책 반전). # # 종전 우선순위는 ["B","A"] — 동점이면 수정본이었고, 근거는 "critique 지적을 # 반영한 쪽 우선"이었다. 그런데 수정의 목적은 **지적된 것만 고치고 나머지는 # 보존**하는 것이다. 두 후보가 대등하다는 것은 지적이 해소됐다는 증거가 # 아니라 심판이 차이를 못 가렸다는 뜻이고, 그 자리에서 지시하지 않은 # 변경까지 함께 확정된다. 실측: 오프라인 재판정 223쌍에서 개악이 88장 # (39.5%)이었고 사용자가 지목한 개악 10건 중 4건은 정순·역순 판정이 # 서로 뒤집혔다 — 바로 그 대등 구간이다. winner, combined = combine_flip_verdicts(fwd, rev, labels2, ["A", "B"]) record["fix_rejudge"] = { "forward_raw": fwd_raw, "forward_normalized": fwd, "reverse_raw": rev_raw, "reverse_normalized": rev, "combined": combined, "winner": winner, "fix_won": winner == "B", # 원본·수정본 **둘 다** 못 쓴다고 본 경우 — 이 샷은 수정이 아니라 # 재촬영 대상이다. 선정 경로와 같은 이유로 typed 로만 남긴다. "all_candidates_fail": bool( fwd.get("all_candidates_fail") and rev.get("all_candidates_fail")), "policy": FIX_REJUDGE_POLICY_VERSION, } if winner == "A": logger.warning( "multiroll_select[%s]: fix-rejudge — 수정본 개악 판정, 선정 " "원본(%s) 유지", tag, selected, ) _atomic_place(cand[winner]) return sel def _compose_critique_prompt( prompt: str, roll_prompts: Optional[Mapping[str, str]], selected: str, shared_prompt: Optional[str] = None, ) -> str: """변형 롤 모드의 critique 프롬프트 = 선정 변형 전문 + 공유 브리프. critique 계약은 '프롬프트 위반만' — 생성 계약(변형 전문)과 공유 불변 사실(브리프, 진단 ② 검출 축) 둘 다 대조 대상이어야 한다. ``shared_prompt`` (2026-08-01 Codex 재리뷰 HIGH 5): 공유 쪽에 쓸 문안을 호출측이 따로 줄 수 있다. 기본값 None 이면 ``prompt`` 를 그대로 써서 기존 동작과 byte 동일하다. 필요해진 이유는 생성 계약이 롤 프롬프트에 직접 실리게 되면서(관할절) **같은 절이 롤과 base 양쪽에 들어가 critique 입력에만 2회 중복**되기 때문이다. 판정마다 같은 요구가 두 번 보이면 무게가 왜곡된다 — 각 경로에 정확히 1회만 실리게 한다. ★[2026-08-01 Codex 2차 재리뷰 NARROW-3] ``or`` 는 빈 문자열을 falsy 로 흘려 원 ``prompt`` 를 쓴다. 그런데 지문 쪽은 ``is not None`` 으로 갈라 빈 문자열도 **별개 계약**으로 기록한다 — 같은 입력이 기록에서는 다르고 실행에서는 같아진다. 기록이 실행을 대변하지 못하면 감사가 무의미하다. "공유 문안을 비운다"는 호출측의 유효한 의사이므로 그대로 존중한다. """ if roll_prompts: head = roll_prompts[selected] shared = prompt if shared_prompt is None else shared_prompt return f"{head}\n\n{shared}" if shared else head return prompt GATE_POLICY_VERSION = "winner_violation_gate_v1_admissible_first" #: 게이트 종착 — 실제 소비자가 읽는 칸이다(기록만 남기고 아무도 안 읽는 #: `needs_reshoot` 의 전철을 밟지 않는다). GATE_CLEAN = "clean" # 이긴 후보에 실격이 없다 GATE_RESELECTED = "reselected" # 허용되는 **기존 후보**로 바꿨다(무료) GATE_UNRESOLVED = "unresolved" # 허용 후보가 하나도 없다 GATE_RESOLVED = "resolved" # **재롤본**이 허용 후보로 채택됐다(유료) GATE_NOT_APPLICABLE = "not_applicable" # 이 갈래는 게이트 대상이 아니다 GATE_INCOMPLETE = "incomplete" # **판정을 못 읽었다** — 품질 문제가 아니다 #: 걷기가 소유하는 **의존 대기** 종착 — 여기서는 읽기만 한다 #: (`still_recipe_service.GATE_DEPENDENCY` 와 같은 값. 두 곳에 적는 대신 #: 값을 한 자리에서 정하고 서비스가 이것을 쓴다.) GATE_DEPENDENCY_OUTCOME = "blocked_dependency" def gate_admissible_labels( judge_result: Dict[str, Any], labels: Sequence[str], ) -> Tuple[List[str], List[str]]: """(허용 후보, **미검사** 후보) — 실격이 빈 후보만 허용한다. ## 왜 미검사를 따로 내는가 `_validate_judge_shape` 는 `readings` 를 **아예 안 본다**. 그래서 어떤 후보의 행이 통째로 없을 수 있는데, 그것은 「실격이 없다」가 아니라 **「아무도 안 봤다」**다. 둘을 같게 다루면 안 본 것을 통과로 읽는다 (2026-09-20 Codex). ★판정 팩이 `hard_violations` 를 「그 컷을 못 쓰게 만드는 실격 결함만」 이라고 규정하므로 **배열이 비었는지**만 본다 — 글자로 종류를 가르지 않는다. """ rows = judge_result.get("readings") if not isinstance(rows, list): return [], list(labels) seen: Dict[str, bool] = {} for r in rows: if not isinstance(r, dict): continue lab = str(r.get("label") or "") if lab not in labels: continue # ★**행이 있다고 「확인했다」가 아니다** (2026-09-20 Codex). # 칸이 없거나 `null` 이거나 목록이 아니면 **미검사**다 — # `(… or [])` 로 받으면 그것이 조용히 「실격 없음」이 된다. raw = r.get("hard_violations") if not isinstance(raw, list): seen[lab] = False if seen.get(lab) is False else None continue hv = [x for x in raw if str(x).strip()] prev = seen.get(lab) if prev is False: continue # 이미 불허 — 유지 ok = not hv # 같은 라벨의 다른 행이 미검사면 **빈 배열이 그것을 지우지 않는다** seen[lab] = False if not ok else (None if prev is None and lab in seen else ok) admissible = [l for l in labels if seen.get(l) is True] unchecked = [l for l in labels if seen.get(l, "미") is not False and seen.get(l) is not True] return admissible, unchecked def gate_evidence_id( judge_result: Dict[str, Any], labels: Sequence[str], input_fingerprint: str = "", ) -> Dict[str, Any]: """이 판정이 **무엇을 보고** 내려졌는지의 신원 (2026-09-20 Codex ③). 게이트 정책을 생성 지문에서 뺐으므로, 「어느 근거를 어느 정책으로 봤는가」는 `gate` 가 스스로 적어야 한다. 안 적으면 나중에 같은 `clean` 을 보고도 **어느 판정문에서 나온 허용인지** 알 수 없다. 근거 digest 는 판정문 전체를 통째로 싣지 않으려는 것이지, 판정문을 **자르는 것이 아니다** — 원문은 `readings` 에 그대로 있다. """ import hashlib as _hl import json as _js try: blob = _js.dumps(judge_result.get("readings"), ensure_ascii=False, sort_keys=True) except Exception: # noqa: BLE001 blob = repr(judge_result.get("readings")) return { "input_fingerprint": str(input_fingerprint or ""), "labels": list(labels), "readings_digest": _hl.sha256(blob.encode("utf-8")).hexdigest()[:16], } #: 게이트 재롤 정책 — 이 값이 바뀌면 「같은 정책으로 이미 한 번 샀다」가 #: 아니다. 지문과 달리 **후보를 지우지 않는다**(§15.5) — 재롤 한도만 가른다. GATE_REROLL_POLICY_VERSION = "gate_reroll_v1_one_per_shot" #: 재롤 시도의 **단계**. 「샷당 1회」를 재개까지 합산해 지키려면 발송 전후를 #: 갈라 적어야 한다 — 발송만 하고 죽은 경우에 새 1회를 주면 안 된다. REROLL_SENT = "sent" # 발송했다(결과 불명 포함) REROLL_PRODUCED = "produced" # 파일을 받았다 REROLL_JUDGED = "judged" # 재판정까지 끝났다 def reroll_attempt_matches( prior: Any, *, policy: str, fingerprint: str, ) -> bool: """이 기록의 재롤 시도가 **지금 조건의 것**인가. ★「샷당 1회」는 함수 호출당이 아니라 **같은 입력·정책의 canonical 샷당** 이다 — 재개까지 합산한다. 입력이나 정책이 바뀌면 다른 물음이므로 새 1회를 준다. """ return (isinstance(prior, dict) and prior.get("policy") == policy and prior.get("for_fingerprint") == fingerprint) def gate_reroll_ready( judge_fn: Any, regen_texts: Optional[Mapping[str, str]] = None, ) -> Tuple[bool, str]: """**이미지를 사기 전에** 재판정 배선을 확인한다 (2026-09-20 Codex). 돈을 쓰고 나서 「판정할 것이 없다」를 알면 그 돈은 버린 것이다. 두 가지를 본다: · 콜백이 실제로 있는가 · 그 판정이 **`readings` 를 남기는가** — 게이트는 그 칸으로만 판단 하므로, 안 남기는 판정기로 재롤하면 결과를 읽을 길이 없다 `readings` 보존은 호출부가 `emits_readings` 로 **선언**한다 — 판정기를 만들어 주는 쪽이 자기 스키마를 안다(`owns_order` 와 같은 관례). """ if judge_fn is None: return False, "재판정 콜백이 없다" if not getattr(judge_fn, "emits_readings", False): return False, "재판정기가 readings 를 남긴다고 선언하지 않았다" if not getattr(judge_fn, "owns_order", False): return False, "재판정기가 cross-model(owns_order) 갈래가 아니다" # ★★**스틸 재생성 문안이 없으면 안 산다** (2026-09-20 Codex BLOCK 3). # 비우고 부르면 `build_regen_prompt` 가 **구조물용 기본값**으로 # 떨어져 「이 **장소 사진**을 … 건물의 **층수**·매싱」이 나간다 — # 팔·얼굴 결함을 건물 결함으로 설명하고 사는 것이다. _t = regen_texts or {} if not (str(_t.get("regen_head") or "").strip() and str(_t.get("regen_tail") or "").strip() and str(_t.get("regen_faults_header") or "").strip()): return False, "스틸 재생성 문안이 없다(구조물 기본값으로 나간다)" return True, "" def gate_pick_within_admissible( judge_result: Dict[str, Any], labels: Sequence[str], ranking: Sequence[str], ) -> Tuple[Optional[str], List[str], List[str]]: """**허용 집합 안에서만** 순위로 고른다 — 허용이 점수보다 먼저다. 재롤 후보와 초기 후보를 **한 집합**으로 놓고 본다. 하나도 허용되지 않으면 `None` — least-bad 를 정상본으로 내보내지 않는다. """ admissible, unchecked = gate_admissible_labels(judge_result, labels) if not admissible: return None, admissible, unchecked pick = next((l for l in ranking if l in admissible), None) if pick is None: pick = next((l for l in labels if l in admissible), None) return pick, admissible, unchecked def gate_needs_reapply( record: Dict[str, Any], evidence: Dict[str, Any], ) -> bool: """이 기록에 정책을 **다시 적어야 하는가** (2026-09-20 Codex (e)). ★같은 정책·같은 근거로 이미 본 기록은 **다시 쓰지 않는다** — 캐시로 끝나는 방문마다 메타를 새로 적으면 「지출 흔적」이 매번 움직여 재생성 계수가 거짓으로 오른다(두 번째 메타 재기록 0). ★의존 대기는 **걷기가 소유**한다 — 여기서 건드리지 않는다. """ g = record.get("gate") if not isinstance(g, dict): return True if g.get("outcome") == GATE_DEPENDENCY_OUTCOME: return False # ★★**재롤이 끝난 기록은 다시 보지 않는다** (2026-09-20, 2단계). # 그 종착은 **재판정**(초기 후보 + 재롤 후보)이 냈다. 여기서 쓰는 # 근거는 `record["readings"]` = **초기 판정**이라, 다시 보면 재롤 # 결과를 옛 판정으로 덮어써 **산 그림을 버리게 된다**. if g.get("reroll"): return False return (g.get("policy") != GATE_POLICY_VERSION or g.get("evidence") != evidence) def reapply_gate_to_cached_record( *, record: Dict[str, Any], labels: Sequence[str], applicable: bool, input_fingerprint: str = "", ) -> Optional[str]: """**호출 0** 으로 캐시 기록에 정책을 적용한다 — 돌려주는 것은 새 라벨. 이미 산 후보·이미 받은 판정문만 본다. 생성도 새 VLM 호출도 **없다**. ## 왜 필요한가 (2026-09-20 Codex (e)) 정책을 생성 지문에서 뗐으므로(③), 지문이 맞는 기록은 캐시 갈래로 **판정을 안 거치고** 그대로 반환된다. 그러면 정책을 켜도 옛 기록에는 영영 판정이 안 붙고, verify 는 그 샷들을 계속 **미판정**으로 붙잡는다. ## 무엇을 안 하는가 ★**수정본이 최종인 기록은 손대지 않는다.** 저장된 `readings` 는 초기 롤을 본 판정문이라, 수정본의 판정으로 읽으면 **다른 그림의 판정**을 이 그림에 붙이게 된다. 그런 기록은 `incomplete` 로 남긴다. ★근거(`readings`)나 선택이 없으면 **미검증으로 남긴다** — 무료 소급을 명분으로 생성이나 새 판정 호출을 하지 않는다. ★라벨이 바뀌면(재선택) 호출부가 **선정 파일까지 그 후보로 맞춰야** 한다. 기록만 B 이고 파일은 A 인 채 나가면 안 된다. 돌려주는 것: 바뀐 라벨(재선택) 또는 None(바뀐 것 없음). """ if not applicable: ev_na: Dict[str, Any] = {} if gate_needs_reapply(record, ev_na): record["gate"] = {"outcome": GATE_NOT_APPLICABLE, "policy": GATE_POLICY_VERSION, "applicable": False, "evidence": ev_na, "retroactive": True} return "" # 기록만 바뀜 — 라벨은 그대로 return None readings = (record.get("readings") if isinstance(record.get("readings"), list) else None) selected = str(record.get("selected") or "") judge_result = {"readings": readings} evidence = gate_evidence_id(judge_result, labels, input_fingerprint) if not gate_needs_reapply(record, evidence): return None # ★수정본이 최종이면 이 판정문은 **그 그림의 것이 아니다**. fixed_final = bool(record.get("critique")) and not record.get( "critique_skipped") if fixed_final or not readings or not selected: record["gate"] = { "outcome": GATE_INCOMPLETE, "policy": GATE_POLICY_VERSION, "applicable": True, "evidence": evidence, "retroactive": True, "reason": ("검사·수리 이력이 있는 캐시라 보수적으로 제외한다" " — 저장된 판정문이 이 그림의 것이라는 보장이 없다" if fixed_final else "판정문이나 선택 기록이 없다"), } return "" ranking = record.get("ranking") before = selected after = apply_winner_violation_gate( record=record, judge_result=judge_result, labels=labels, selected=selected, ranking=list(ranking) if isinstance(ranking, list) else [], applicable=True, input_fingerprint=input_fingerprint) record["gate"]["retroactive"] = True if after != before: # ★★**선택 기록을 여기서 확정하지 않는다** (2026-09-20 Codex BLOCK). # # 종전에는 여기서 `record["selected"] = after` 를 먼저 해 버렸다. # 그런데 호출부가 그 후보의 **파일이 없다**는 것을 나중에 알고 # 「안 바꾼다」로 되돌리려 해도, 원래 값이 이미 지워진 뒤였다 — # 기록은 B 인데 `_sel` 은 A 로 남았다. 그러면 다음 방문에서 B 의 # 판정으로 **실격 A 가 clean 을 달고 나간다**. # # ★바꾸는 것은 **물질화가 성공한 뒤**다. 호출부가 확정한다. return after return "" def apply_winner_violation_gate( *, record: Dict[str, Any], judge_result: Dict[str, Any], labels: Sequence[str], selected: str, ranking: Sequence[str], applicable: bool, input_fingerprint: str = "", ) -> str: """선정 결과를 **허용 여부**로 한 번 더 거른다 — 최종 `selected` 를 낸다. ## 계약 (2026-09-20, Codex 설계 합의) ① 허용 후보 집합을 먼저 만든다 (실격이 빈 후보) ② 그 안에서 **기존 순위**로 고른다 ← 있으면 **무료 재선택** ③ 하나도 없으면 `unresolved` ← 정상 확정으로 안 내보낸다 재롤(유료)은 이 함수 밖이다 — 여기서는 **이미 산 후보**만 다룬다. ## 왜 필요한가 선정은 「둘 중 나은 것」을 성실히 낸다. **둘 다 못 쓸 때 멈추는 자리가 없다**(실측: 이 화 239샷 중 이긴 후보에 실격이 달린 것 81). 뿌리는 점수 가중치가 아니라 **「상대 순위」와 「최종 사용 가능」이 분리되지 않은 것**이다. ★비대상 갈래는 `not_applicable` 로 **기록**한다 — 「검사 통과」로 표시하지 않는다. ★미검사 후보는 허용으로 세지 않는다. """ if not applicable: record["gate"] = {"outcome": GATE_NOT_APPLICABLE, "policy": GATE_POLICY_VERSION, # ★비대상도 **명시**한다. 「판정 없음」과 구별이 # 안 되면 소비자가 미검증을 합격으로 읽는다. "applicable": False} return selected admissible, unchecked = gate_admissible_labels(judge_result, labels) info: Dict[str, Any] = { "policy": GATE_POLICY_VERSION, "applicable": True, # ★`clean` 이어도 **어느 정책으로 · 무엇을 보고** 허용했는지는 # 바뀐다 (2026-09-20 Codex ③). 생성 지문에서 정책을 뺀 대신 # 여기에 남긴다. "evidence": gate_evidence_id(judge_result, labels, input_fingerprint), "admissible": admissible, "unchecked": unchecked, "initial_selected": selected, } if selected in admissible: info["outcome"] = GATE_CLEAN record["gate"] = info return selected # 기존 순위를 지키며 허용 후보를 고른다 — **새로 사지 않는다** alt = next((l for l in ranking if l in admissible), None) if alt is None: alt = next((l for l in labels if l in admissible), None) if alt is not None: info["outcome"] = GATE_RESELECTED info["selected"] = alt record["gate"] = info logger.warning( "winner_violation_gate: 이긴 후보(%s)에 실격이 달렸다 — 허용되는 " "기존 후보 %s 로 바꾼다 (이미지 구매 0)", selected, alt) return alt # ★「판정이 안 됐다」를 「검사해 보니 모두 실격」으로 바꾸지 않는다 # (2026-09-20 Codex). **미검사가 하나라도 남아 있으면** 「모든 후보가 # 실격」이라고 닫을 수 없다 — 그 후보는 아직 모르는 것이다. # # A=실격·B=미검사 를 `unresolved` 로 두면 「전부 못 쓴다」가 되어 # 품질 재롤로 보내진다. 그때 필요한 것은 **판정 복구**이지 새 그림이 # 아니다. 전부 확인되고 전부 불허일 때만 품질 `unresolved` 다. if unchecked: info["outcome"] = GATE_INCOMPLETE record["gate"] = info logger.warning( "winner_violation_gate: 허용 후보가 없고 **미검사가 남았다**(%s) " "— incomplete (초기 선정 %s). 품질 unresolved 와 다르고, 필요한 " "것은 **판정 복구**이지 새 그림이 아니다", unchecked, selected) return selected info["outcome"] = GATE_UNRESOLVED record["gate"] = info logger.warning( "winner_violation_gate: **허용 후보가 없다** — 이 샷은 unresolved " "(초기 선정 %s). ★이 단계는 **기록만** 한다 — 확정을 막는 것은 " "소비자 쪽 일이다", selected) return selected def _set_needs_reshoot(record: Dict[str, Any]) -> None: """최종 `_sel` 기준 재촬영 필요 여부 — **수정이 끝난 뒤에** 계산한다. 단계별 선언을 그대로 두면 두 방향으로 오독된다(Codex 2차 리뷰): · 초기 롤이 전부 실패했어도 수정본이 정상 승리하면 재촬영은 불필요한데 초기 플래그만 보면 True 로 읽힌다. · 초기 롤은 통과했는데 [원본 vs 수정본] 재판정에서 둘 다 실패하면 초기 플래그가 없어 실패한 최종본을 놓친다. 그래서 **최종 산출을 정한 단계의 선언**을 본다. 재판정이 돌았으면 그것이 최종을 정했으므로 그 선언이 기준이고, 안 돌았으면 초기 롤 선정본이 그대로 최종이므로 초기 선언이 기준이다. 값이 참일 때만 키를 남긴다 — 기록에 False 를 흩뿌리면 없는 것과 구분되지 않는다. """ rj = record.get("fix_rejudge") if isinstance(rj, dict) and "all_candidates_fail" in rj: final_fail = bool(rj.get("all_candidates_fail")) else: final_fail = bool(record.get("initial_roll_all_fail")) if final_fail: record["needs_reshoot"] = True def run_multiroll_select( *, tag: str, prompt: str, labeled_refs: Sequence[Tuple[str, Path]], out_stem: Path, gen_fn: GenFn, judge_fn: JudgeFn, critique_fn: Optional[CritiqueFn] = None, fix_gen_fn: Optional[GenFn] = None, roll_count: int = 3, critique_enabled: bool = True, fix_head: str = "", fix_tail: str = "", fix_label: str = "", fix_ref_label: str = "", # 참조 선별 (2026-08-19) — 미사용 시 기본값이 기존 동작과 동일. fix_ref_gate: bool = False, fix_missing_head: str = "", fix_missing_tail: str = "", record: Optional[Dict[str, Any]] = None, extra_fingerprint: Optional[Dict[str, Any]] = None, persist_record_fn: Optional[Callable[[Dict[str, Any]], None]] = None, force: bool = False, roll_prompts: Optional[Mapping[str, str]] = None, # critique 의 공유 문안만 따로 준다(2026-08-01 HIGH 5). None = 기존 동작. critique_shared_prompt: Optional[str] = None, roll_refs: Optional[Mapping[str, Sequence[Tuple[str, Any]]]] = None, # 참조 신원 — `labeled_refs`/`roll_refs[lab]` 과 **index 로 정렬된** 병렬 # 목록(씬 경로의 `ref_role_metadata` 관례 동형). 튜플을 넓히지 않는 이유는 # `build_ref_records` 문서 참조. 기본 None = 기존 호출자 무영향. ref_role_metadata: Optional[Sequence[Optional[Dict[str, Any]]]] = None, roll_ref_metadata: Optional[ Mapping[str, Sequence[Optional[Dict[str, Any]]]]] = None, parallel_rolls: bool = False, judge_flip: bool = False, flip_priority: Optional[Sequence[str]] = None, critique_selected_prompt_only: bool = False, judge_prompt_header: Optional[str] = None, fix_rejudge_fn: Optional[JudgeFn] = None, composition_critique_fn: Optional[CritiqueFn] = None, regen_gen_fn: Optional[GenFn] = None, regen_head: str = "", regen_tail: str = "", regen_faults_header: str = "", regeneration_issue_policy: str = REGEN_ISSUE_POLICY_LEGACY_EDIT, repair_method: str = REPAIR_METHOD_EDIT, # ★이긴 후보의 **실격**을 걸러 내는 게이트 (2026-09-20). 호출자가 # **이 갈래가 대상인가**를 정한다 — 비대상은 `not_applicable` 로 # 기록되고 「검사 통과」로 표시되지 않는다. 기본 False = 기존 동작. winner_gate_applicable: bool = False, # ── 게이트 재롤 (2단계, 2026-09-20) ──────────────────────────── # ★`critique_enabled`(master) 와 **독립**이다 — 사용자가 시간 때문에 # 끈 관문을 켜지 않는다. 다만 master 가 켜져 있어야만 켜는 종속도 # 아니다: 재롤은 「이 그림을 못 쓴다」에 대한 답이지 수리 관문이 # 아니다. gate_reroll_enabled: bool = False, # 재판정 콜백 — **발송 전에** 있는지·`readings` 를 남기는지 본다. gate_rejudge_fn: Any = None, # 주행 단위 승인 상한 — 호출부가 소유한다(여기는 샷당 1회만 안다). # True 를 돌려주면 이번 샷의 재롤을 허용한다. gate_reroll_allow_fn: Optional[Callable[[], bool]] = None, # 재판정 모델·정책·팩 신원 — **시도 기록**에 남는다. ★생성 지문에는 # 안 넣는다(넣으면 레버만 바꿔도 초기 후보를 다시 산다, Codex BLOCK 1). gate_rejudge_identity: Optional[Mapping[str, Any]] = None, # ★재롤 전용 **스틸** 재생성 문안. 비우면 `build_regen_prompt` 가 # **구조물용 기본값**으로 떨어져 팔·얼굴 결함을 「건물 층수」로 # 설명하고 산다 (Codex BLOCK 3). 없으면 **안 산다**. gate_regen_texts: Optional[Mapping[str, str]] = None, ) -> Tuple[Path, Dict[str, Any]]: """공통 생성 파이프 1회 실행. 반환 = (최종 `_sel` 경로, record). record 는 체크포인트에 그대로 직렬화 가능한 dict: {input_fingerprint, prompt, refs, totals, selected, ranking, verdicts, critique{issues} | critique_skipped, fix_prompt | fix_skipped} 재개 계약 (Codex 1차 리뷰 BLOCKING-3 반영): - record.input_fingerprint 가 현재 입력 지문과 일치할 때만 산출 재사용. mismatch / force=True / sel 존재+record 부재(무기록 — 출처 불명) → 이 stem 산출 전부 정리 후 처음부터. - persist_record_fn 이 있으면 롤 시작 전(지문)·선정 직후·결함 처리 직후 각 phase 에서 durable 저장 — 크래시 창에서 critique 영구 생략 결함 차단. roll_prompts (seed 품질 2R, 2026-07-16): 제공 시 라벨별 상이 프롬프트로 각 1롤 생성(변형 롤). 이때 `prompt`=judge/critique 공유 브리프, critique 는 선정 변형 전문+브리프 합성, 지문·record 에 변형 전문 포함. 미제공=기존 동작·지문 byte-identical. still-variants 확장 (2026-07-17, Codex 설계 리뷰 반영): - roll_refs: 라벨별 상이 참조(A/B 4택1의 콘티 유무). gen=라벨 refs, judge=공유 labeled_refs(블라인드 — 어느 후보가 어떤 참조인지 미노출), critique=선정 라벨 refs. per-label 참조 내용은 지문 포함. - parallel_rolls: 미존재 롤 ThreadPool 병렬 생성 — budget 과 generation_context 를 worker 에 명시 전파(BLOCKING-3), 예외는 전 future 완료 후 canonical 라벨 순 첫 예외(NARROW-1). - judge_flip: 정순·역순 2회 판정(canonical 역매핑 후 결합, BLOCKING-2). 정순 직후 중간 persist(NARROW-2). - critique_selected_prompt_only: roll_prompts 가 base 전문을 포함할 때 브리프 재병합으로 base 가 중복되는 경로 차단(BLOCKING-1). - judge_prompt_header: 판정 header 는 어댑터 closure 소유 — 지문 기여용 병기(HIGH-4). 신규 kwargs 전부 default=기존 지문·record byte-identical. """ if critique_enabled and (critique_fn is None or fix_gen_fn is None): raise ValueError( "critique_enabled=True requires critique_fn and fix_gen_fn" ) if roll_prompts is not None and ( set(roll_prompts) != set(roll_labels(roll_count)) ): raise ValueError( f"roll_prompts 라벨 집합 불일치: {sorted(roll_prompts)} != " f"{roll_labels(roll_count)}" ) if roll_refs is not None and ( set(roll_refs) != set(roll_labels(roll_count)) ): raise ValueError( f"roll_refs 라벨 집합 불일치: {sorted(roll_refs)} != " f"{roll_labels(roll_count)}" ) if critique_selected_prompt_only and roll_prompts is None: raise ValueError( "critique_selected_prompt_only=True requires roll_prompts" ) def _crit_prompt(selected_label: str) -> str: if critique_selected_prompt_only: return roll_prompts[selected_label] # type: ignore[index] return _compose_critique_prompt( prompt, roll_prompts, selected_label, shared_prompt=critique_shared_prompt) # ★첫 유료 호출(롤 생성) 전에 재생성 계약을 검증한다 — 결손을 기존 # 편집으로 하강시키지 않고 막는다. 브리프는 판정 뒤에야 라벨이 정해지므로 # **모든 라벨에 대해 실제 조립 함수를 돌려** 비지 않는지 본다. 원천만 # 훑으면 조립 규칙이 바뀔 때 검사와 실물이 어긋난다. # ★★게이트 지원 범위를 **한 장도 사기 전에** 막는다 (Codex ㉣). # plain judge_flip 갈래는 정순 관찰만 싣는다 — 역순의 실격이 빠지므로 # 「실격이 빈 후보만 허용」을 절반만 보고 내리게 된다. 롤·판정을 다 # 산 뒤에 막으면 그 돈이 이미 나갔다. if winner_gate_applicable and not getattr( judge_fn, "owns_order", False): raise ValueError( "winner_violation_gate 는 이 판정 갈래를 아직 지원하지 않는다 " "— plain judge_flip 은 정순 관찰만 싣는다(역순 실격이 빠진다). " "cross-model `owns_order` 경로에서만 켠다.") validate_regeneration_contract( regeneration_issue_policy=regeneration_issue_policy, regen_gen_fn=regen_gen_fn, fix_rejudge_fn=fix_rejudge_fn, brief_sources=[ _crit_prompt(lab) for lab in roll_labels(roll_count)], ) record = dict(record) if record else {} # ★★**이번 방문의 행동 표식은 진입에서 비운다** (2026-09-20 Codex # BLOCK 3). `_this_run` 은 **이름뿐**이었다 — durable record 에 남아 # 다음 방문이 그대로 물려받는다. 그러면 한 번 재선택한 샷이 **무변경 # 방문마다** 「이번 무료 재선택」으로 집계되고, 그 방문이 변환만 다시 # 시도하면 유료 흔적까지 지워진다. # ★샷 방문 uid 를 지출 스냅샷에 새로 넣지는 않는다(거짓 지출이 된다). record.pop("gate_select_action_this_run", None) out_stem.parent.mkdir(parents=True, exist_ok=True) sel = _sel_path(out_stem) def _persist() -> None: if persist_record_fn is not None: persist_record_fn(record) def _stamp_shot_run(rec: Dict[str, Any], *, produced: bool) -> None: """이 방문의 uid 를 record 에 적고 영향 요약을 trace 에 싣는다. ★**모든 반환이 이 함수를 지나야 한다.** 한 곳에만 적으면 먼저 나가는 문(지문 일치 완결·critique 끔)에서 uid 가 빠져 세 곳이 갈린다. produced: 이번 방문이 **최종 산출의 저자인가**. False(산출 재사용)면 record·Opik 에만 남기고 `image_asset` 의 생성 계보는 **건드리지 않는다** — 그 자산을 만든 것은 이전 주행이다. ★유료 구간 진입은 여기서 안 센다 — `_mark_spend_attempt_once` 가 호출 **전에** 못박는다. 반환 직전에 세면 과금 후 실패한 시도를 놓친다(예외가 JIT 전후 비교를 통째로 건너뛴다). """ # ★`produced` 는 **언제나 현재 값**으로 덮는다. uid 가 없을 때 # (v2 OFF·scope 누락) 이 줄을 건너뛰면, record 에 남아 있던 옛 # produced=True 와 옛 uid 를 자산 metadata 가 읽어 **새 산출을 옛 # trace 의 저자로** 기록한다. rec["shot_run_produced"] = bool(produced) try: from app.modules.llm.opik_trace import ( build_influence_summary, current_shot_uid, update_trace_output) uid = current_shot_uid() if uid: rec["shot_run_uid"] = uid else: # 이번 방문에 trace 가 없으면 옛 uid 를 **지운다** — 남겨 두면 # 자산 계보가 엉뚱한 trace 를 가리킨다. rec.pop("shot_run_uid", None) summary = build_influence_summary(rec) summary["produced"] = bool(produced) update_trace_output(summary) except Exception as exc: # noqa: BLE001 — 기록은 본 작업을 안 막는다 logger.debug("_stamp_shot_run 실패 (non-fatal): %s", exc) #: 이번 방문이 **실제로 만들어 낸 라벨** — 최종 저자 판정의 한 축. # ★bool 로 두면 안 된다 (2026-09-20 Codex BLOCK). 「하나라도 만들었나」 # 와 「**최종 후보**를 만들었나」는 다르다 — U1 이 A 만 만들고 B 에서 # 죽은 뒤 U2 가 B 를 만들고 최종이 **A** 로 정해지면, bool 로는 U2 가 # 저자로 찍힌다. 성공한 라벨만 담는다. _made_here: set = set() def _this_visit_uid() -> Optional[str]: """**지금** 방문의 신원 — `record` 에 남은 옛 uid 를 읽지 않는다. ★`record["shot_run_uid"]` 는 반환 직전 `_stamp_shot_run` 에서야 쓰인다. 첫 생성이면 아직 없고(None), 캐시 방문이면 **지난 방문의** uid 다 — 그걸 「이번에 만들었다」의 근거로 쓰면 계보가 뒤집힌다 (2026-09-20 Codex BLOCK 2). """ try: from app.modules.llm.opik_trace import current_shot_uid return current_shot_uid() or None except Exception: # noqa: BLE001 return None def _final_author_is_this_visit(final_label: str) -> bool: """**최종 bytes 를 이번 방문이 만들었는가** — 라벨이 바뀐 것과 다르다. ★「선택이 달라졌다」로 저자를 정하면 두 방향으로 다 틀린다: · 옛 초기 후보 B 로 바뀐 것뿐인데 이번 방문을 저자로 찍는다 · 이번에 **실제로 산** C 인데 캐시 갈래라고 False 로 찍는다 ★출처를 모르면 **이번 uid 로 메우지 않는다**. """ rr = record.get("gate_reroll") if isinstance(rr, dict) and final_label == rr.get("label"): made = rr.get("produced_by_visit") here = _this_visit_uid() return bool(made and here and made == here) # ★**그 라벨을** 이번에 만들었는가 — 「무언가 만들었나」가 아니다. return final_label in _made_here _spend_marked = False def _mark_spend_attempt_once() -> None: """이번 방문이 유료 호출을 **하기 직전에** 한 번 못박는다. ★반환 직전이 아니라 **호출 전**이다. 과금 후 실패하면 예외가 `still_recipe_service` 의 except 로 빠져 JIT 전후 비교를 통째로 건너뛴다 — 그 시도가 기록에 안 남고 다음 resume 이 같은 자리를 다시 탄다. ★이름이 곧 계약이다: 이 값은 **「유료 호출을 할 수 있는 구간에 몇 번 들어갔나」**다. 과금 횟수가 아니다 — 콜백이 바깥으로 안 나가고 끝나는 경로도 센다(후보 하나뿐인 판정 등). 치우침은 안전한 쪽(과다 계상)이다. 방문당 한 번만 올린다. """ nonlocal _spend_marked if _spend_marked: return _spend_marked = True try: _n = int(record.get("shot_run_spend_attempt_count") or 0) except (TypeError, ValueError): _n = 0 record["shot_run_spend_attempt_count"] = _n + 1 _persist() # ★durable — 여기서 죽어도 시도가 남는다 _fp_extra = extra_fingerprint # ★수리 수단이 바뀌면 완료 샷이 **반드시 stale** 이어야 한다 # (2026-08-29, Codex 조건). 같은 브리프·같은 참조라도 편집본과 # 재생성본은 다른 그림이다 — 지문이 안 움직이면 수단만 바꾼 재주행이 # 옛 산출을 그대로 안고 간다. # # ★`edit` 은 **한 글자도 안 싣는다** — 기존 소비자·기존 산출이 # byte-identical 이어야 하고, 그래야 이 기능을 안 켠 자리에서 # 무관한 재생성이 일어나지 않는다. if repair_method != REPAIR_METHOD_EDIT: import hashlib as _hl3 _fp_extra = { **(_fp_extra or {}), "repair_method": repair_method, "repair_method_policy": REPAIR_METHOD_POLICY_VERSION, # 문안 **실제 바이트**를 접는다. 팩 selector 만 접으면 같은 # 판 안에서 글을 고쳤을 때 그림이 안 따라 움직인다. "repair_regen_pack_sha": _hl3.sha256("\n".join([ regen_head, regen_tail, regen_faults_header, ]).encode("utf-8")).hexdigest()[:16], } if critique_shared_prompt is not None: # 비기본 값만 지문에 싣는다 — None(기존 동작)은 byte 호환 유지. import hashlib as _hl2 _fp_extra = { **(_fp_extra or {}), "critique_shared_prompt_sha": _hl2.sha256( critique_shared_prompt.encode("utf-8")).hexdigest()[:16], } if roll_prompts is not None: _fp_extra = { **(_fp_extra or {}), "roll_prompts": {k: roll_prompts[k] for k in sorted(roll_prompts)}, } # still-variants 확장 지문 (HIGH-4) — 사용된 경우만 기여(미사용=불변) if roll_refs is not None: import hashlib as _hashlib def _ref_sig(refs: Sequence[Tuple[str, Any]]) -> List[List[str]]: out: List[List[str]] = [] for label, src in refs: data = ( Path(src).read_bytes() if isinstance(src, (str, Path)) else src ) out.append( [str(label), _hashlib.sha256(data).hexdigest()] ) return out _fp_extra = { **(_fp_extra or {}), "roll_refs": { lab: _ref_sig(roll_refs[lab]) for lab in sorted(roll_refs) }, } if judge_flip: _fp_extra = { **(_fp_extra or {}), "judge_flip": { "policy": FLIP_POLICY_VERSION, "priority": list(flip_priority or roll_labels(roll_count)), }, } if critique_selected_prompt_only: _fp_extra = {**(_fp_extra or {}), "critique_selected_prompt_only": True} if judge_prompt_header is not None: _fp_extra = {**(_fp_extra or {}), "judge_prompt_header": judge_prompt_header} if fix_rejudge_fn is not None: # E2E10 fix②: 재판정 정책 전환도 산출 무효화 대상 (제공 시만 기여 # — 미제공=기존 지문 byte-identical) _fp_extra = {**(_fp_extra or {}), "fix_rejudge": FIX_REJUDGE_POLICY_VERSION} if composition_critique_fn is not None: # E2E11 fix③: 구도 critique 합류도 수정본(=최종 후보) 실질 입력 _fp_extra = {**(_fp_extra or {}), "gpt_composition": GPT_COMPOSITION_POLICY_VERSION} # ★★게이트 정책은 **생성 지문에 안 접는다** (2026-09-20 Codex ③). # # 접었더니 정책을 켜는 것만으로 대상 샷 전부가 지문 불일치가 되어 # `_clear_outputs` → **롤부터 다시 사게** 됐다(이 화 실측: 표준 137샷). # 그런데 게이트가 하는 일은 **이미 산 후보 중 무엇을 쓰느냐**이지 # 후보를 다시 그리는 것이 아니다 — 같은 후보를 버릴 이유가 없다. # # ★그렇다고 정책을 안 남기는 것이 아니다. 「어느 정책으로 허용했나」는 # `record["gate"]` 가 **항상** 적는다(정책 버전 + 적용 범위 + 평가에 # 쓴 근거의 식별자). 같은 A 라도 「허용된 A」와 「unresolved 인 A」는 # 정책 상태가 다르다. # ★산출이 실제로 달라지는 것(`reselected` 로 선택 라벨·선정 파일이 # 바뀌는 것)은 그때 하류가 본다 — 후보 생성 지문을 일부러 틀려서 # 변경을 알리지 않는다. # ★스텝 단위 재평가는 바깥 `image_steps` 의 정책 해시가 맡는다 — # 「재평가해야 함」과 「롤부터 사야 함」은 다른 축이다. if regeneration_issue_policy != REGEN_ISSUE_POLICY_LEGACY_EDIT: # 2026-08-01: defer 는 수정 단계의 실질 입력이다 — 같은 critique # 결과에서도 편집 지시와 최종 산출이 달라진다. **공용 함수가 직접 # 접는다**: caller 가 extra_fingerprint 에 수동으로 중복시키는 # 방식은 새 소비자에서 빠뜨리기 쉽다(Codex 지적, 수용). # legacy 는 기여 0 — 비대상 소비자의 지문이 byte-identical 하게 남는다. _fp_extra = {**(_fp_extra or {}), "regen_issue_policy": { "policy": regeneration_issue_policy, "version": REGEN_ISSUE_POLICY_VERSION}} # (Codex R3) roll_prompts 는 위 :995 블록이 **전문 그대로** 이미 접고 # 있다 — 여기서 해시 표현으로 다시 덮으면 내용 불변인 기존 기록까지 # 표현 차이만으로 전량 stale 된다(무관 소비자 유료 재생성). b 변주 # 스템 개정 검출·미사용 키 부재·샷별 범위 전부 기존 블록으로 성립. def _crit_refs(selected_label: str) -> Sequence[Tuple[str, Any]]: if roll_refs is not None: return list(roll_refs[selected_label]) return labeled_refs fingerprint = compute_input_fingerprint( prompt=prompt, labeled_refs=labeled_refs, roll_count=roll_count, critique_enabled=critique_enabled, extra=_fp_extra, ) has_outputs = sel.exists() or any( _roll_path(out_stem, lab).exists() for lab in roll_labels(roll_count) ) stale = ( force or (record and record.get("input_fingerprint") != fingerprint) or (not record and has_outputs) # 무기록 산출 = 출처 불명 ) if stale: if not force: logger.warning( "multiroll_select[%s]: 입력 지문 불일치/무기록 산출 — " "정리 후 재생성", tag, ) _clear_outputs(out_stem) record = {} # ★여기서 정의한다 — 캐시 반환 갈래의 재롤 이음도 이 둘을 쓴다. # 종전에는 롤 생성 직전에 있어서, 캐시에서 부르면 NameError 였다. def _roll_prompt(lab: str) -> str: return roll_prompts[lab] if roll_prompts else prompt def _gen_refs(lab: str) -> Sequence[Tuple[str, Any]]: return list(roll_refs[lab]) if roll_refs is not None else labeled_refs def _materialize_reroll_pick( pick: str, rr_label: str, *, resumed: bool = False, ) -> str: """**선정본을 먼저 바꾸고, 성공한 뒤에 확정한다**. ★기록을 먼저 확정하면 복사가 실패했을 때 「기록은 C · 그림은 A」가 남고, 다음 방문은 `gate.reroll` 때문에 소급도 이음도 건너뛰어 **실격 A 를 그대로 내보낸다**(2026-09-20 Codex BLOCK 3). ★재개는 이 함수만 다시 탄다 — 이미지도 판정도 **안 산다**. resumed: **지난 방문이 예고한 결정**을 이어받는 길인가. 이 길은 이번 방문에서 아무것도 안 산다 — 그 사실을 기록에 남겨야 상한 계수가 이 방문을 유료로 세지 않는다. 같은 방문에서 사고 곧바로 확정하는 길은 False 다(그쪽은 지출 계수가 이미 올라 있어 되돌릴 것이 없다). """ nonlocal sel _before_pick = str(record.get("selected") or "") # ★이 방문이 **예고된 결정을 이어받았는가**. 이어받기(재개)는 # 이미지도 판정도 안 사는 **무료 복구**인데, 기록에서는 # `gate_reroll.pending_pick` 이 사라지는 것으로만 드러난다. # 그 한 칸을 되돌려 견줄 수 있게 **무엇을 이어받았는지** 남긴다. _resumed_pend = ( (record.get("gate_reroll") or {}).get("pending_pick") if resumed else None) src = _roll_path(out_stem, pick) if not src.exists(): # 고른 후보의 파일이 사라졌다 — 확정하지 않고 미검사로 남긴다. # ★**결정은 보존한다** (2026-09-20 Codex NON-BLOCK). 판정은 # 이미 끝났고 없는 것은 **파일**뿐이다 — `pending_pick` 을 # 지우면 파일이 돌아와도 재판정을 **다시 사게** 된다. # 「판정 미완료」와 「파일 미존재」는 다른 것이다. record["gate"]["outcome"] = GATE_INCOMPLETE record["gate"]["reroll_file_missing"] = pick record["gate"].pop("evidence", None) logger.warning( "multiroll_select[%s]: 재롤 결정(%s)의 파일이 없다 — " "결정은 남기고 확정만 미룬다", tag, pick) _persist() return str(record.get("selected") or "") _tmp = sel.with_name(sel.name + ".tmp") shutil.copy(src, _tmp) _tmp.replace(sel) # ── 여기부터가 확정이다 ── # ★★**이번 방문의 행동**을 구조로 남긴다 (2026-09-20 Codex (w)). # `_this_run` 접미는 지출 스냅샷이 걸러 낸다 — 이 기록은 # 「무엇을 했나」이지 **구매 표식이 아니다**. 영구 boolean 하나로 # 두면 「이번에 무료였나」를 다음 방문이 못 가린다. record["gate_select_action_this_run"] = { "action": "reselect_cached", "from": _before_pick, "to": pick, "to_sha": _sha256_file(src), "materialized": True, # ★이번 방문이 **발송을 했는가**는 따로 센다 — 여기 안 적는다. } if _resumed_pend is not None: record["gate_select_action_this_run"]["resumed_pending_pick"] = ( str(_resumed_pend)) record["gate_reroll"].pop("pending_pick", None) record["gate_reroll"]["adopted"] = pick record["gate"]["selected"] = pick record["gate"]["outcome"] = ( GATE_RESOLVED if pick == rr_label else GATE_RESELECTED) record["gate"]["reroll"] = ( "adopted" if pick == rr_label else "lost") record["selected"] = pick _persist() logger.warning( "multiroll_select[%s]: 재롤 뒤 %s 채택 — 종착 %s", tag, pick, record["gate"]["outcome"]) return pick def _gate_reroll_continue( selected_now: str, judge_result_now: Dict[str, Any], ranking_now: Sequence[str], ) -> str: """게이트가 **미해결**이면 후보를 하나 더 사서 다시 판정한다. ★**새 판정 직후와 캐시 반환 둘 다** 여기를 지난다 (2026-09-20 Codex BLOCK 2). 새 판정 자리에만 걸면, 주행 상한에 걸렸거나 배선이 안 됐던 샷이 다음 방문에 **영영 못 온다** — `_sel` 과 `critique_skipped` 가 저장돼 캐시 갈래로 먼저 나가기 때문이다. 돌려주는 것: 최종 라벨(바뀌었으면 호출부가 `_sel` 을 맞춘다). """ nonlocal sel if not (winner_gate_applicable and gate_reroll_enabled): return selected_now g = record.get("gate") if not isinstance(g, dict): return selected_now # ★★**미검사로 남긴 것도 다시 온다** (Codex BLOCK 5). 재판정에서 # 후보 행이 빠져 `incomplete` 로 남긴 샷은 「못 쓴다」가 아니라 # 「아직 못 봤다」다 — 같은 세 파일로 **재판정만** 다시 하면 # 살아난다(새 이미지 0). 이 길을 안 열면 복구가 막힌다. if g.get("outcome") == GATE_INCOMPLETE and ( g.get("reroll_unchecked") or g.get("reroll_file_missing")): # 미검사(재판정만 복구) · 파일 미존재(물질화만 복구) 둘 다 # **다시 온다**. 종착이 아니다. pass elif g.get("outcome") != GATE_UNRESOLVED: return selected_now rr_label = roll_labels(roll_count + 1)[-1] rr_path = _roll_path(out_stem, rr_label) labels_all = roll_labels(roll_count + 1) prior = record.get("gate_reroll") same = reroll_attempt_matches( prior, policy=GATE_REROLL_POLICY_VERSION, fingerprint=fingerprint) base_label = g.get("initial_selected") or selected_now if same and isinstance(prior, dict): # ── ① 결정만 남고 **선정본을 못 바꾼 채** 끊긴 재개 ────── # 기록에 결정을 예고해 두고 파일 교체에서 죽은 경우다. # **물질화만** 복구한다 — 이미지도 판정도 다시 안 산다. _pend = prior.get("pending_pick") if _pend and prior.get("state") == REROLL_JUDGED: return _materialize_reroll_pick( str(_pend), rr_label, resumed=True) # ── ② **확정된 실패는 다시 판정하지 않는다** (Codex BLOCK 1) ── # 전부 보고서 전부 실격이면 그것이 종착이다. 매 방문 재판정을 # 또 사면 상한도 못 막고(상한은 새 구매에서만 본다), 재답이 # 우연히 달라지면 **종착까지 뒤집힌다**. if (prior.get("state") == REROLL_JUDGED and not prior.get("unchecked") and prior.get("adopted") is None): logger.info( "multiroll_select[%s]: 재롤 결과가 **이미 확정**됐다" "(전부 관찰·전부 실격) — 다시 판정하지 않는다", tag) return selected_now stop = "" if same and not rr_path.exists(): # ★발송했는데 파일이 없다 = **결과 불명**. 자동으로 새 1회를 # 주지 않는다 — 같은 그림을 두 번 사는 길이다. stop = ("앞선 재롤 시도의 결과가 불명이다 — 새 1회를 주지 " "않는다 (사람이 확인하고 기록을 지워야 다시 산다)") elif not same: ok, why = gate_reroll_ready(gate_rejudge_fn, gate_regen_texts) if not ok: # ★**사기 전에** 막는다 — 돈을 쓰고 나서 「판정할 것이 # 없다」를 알면 그 돈은 버린 것이다. stop = f"재롤 배선이 준비되지 않았다: {why}" elif critique_enabled: # ★master ON 조합은 **아직 지원하지 않는다**. C 가 이기면 # `_crit_prompt(C)` 가 A/B 뿐인 `roll_prompts` 에서 # KeyError 로 죽는다 — **사기 전에** 막는다(Codex). stop = ("critique master 가 켜진 조합은 아직 지원하지 " "않는다 (재롤 라벨의 수정 문안 배선 없음)") elif (gate_reroll_allow_fn is not None and not gate_reroll_allow_fn()): # ★상한 거절은 **종착이 아니다** — 다음 방문에 승인 여유가 # 생기면 다시 올 수 있어야 한다. 그래서 시도 기록을 남기지 # 않는다. stop = "주행 단위 재롤 상한에 닿았다" if stop: g["reroll_skipped"] = stop logger.warning( "multiroll_select[%s]: 게이트 재롤을 하지 않는다 — %s", tag, stop) _persist() return selected_now if not same: # ★★**발송 전에** 시도를 남긴다. 여기서 죽으면 다음 방문은 # 「결과 불명」으로 읽고 새 1회를 안 준다. # ★`gate` **밖**에 쓴다 — 지출 스냅샷이 `gate` 를 빼므로 # 시도 이력을 그 안에 넣으면 지출이 사라진다. rr_prompt = build_regen_prompt( [{"issue_ko": str(v)} for v in gate_winner_violations( judge_result_now, base_label)], _roll_prompt(base_label), (gate_regen_texts or {}).get("regen_head", ""), (gate_regen_texts or {}).get("regen_tail", ""), (gate_regen_texts or {}).get("regen_faults_header", ""), ) record["gate_reroll"] = { "policy": GATE_REROLL_POLICY_VERSION, "for_fingerprint": fingerprint, "attempt_id": uuid.uuid4().hex, "state": REROLL_SENT, "label": rr_label, "from_selected": base_label, # ★**실제로 나간 전문**을 남긴다 (Codex BLOCK 4) — 안 남기면 # C 가 이겼을 때 하류가 base_prompt 를 「쓴 문안」으로 # 저장한다. "prompt": rr_prompt, "identity": dict(gate_rejudge_identity or {}), } # ★하류(`effective_prompt_used`)가 읽는 자리에도 같이 넣는다 — # 거기만 보고 자산에 적는 소비자가 있다. _rp = dict(record.get("roll_prompts") or {}) if not _rp: _rp = {lab: _roll_prompt(lab) for lab in roll_labels(roll_count)} _rp[rr_label] = rr_prompt record["roll_prompts"] = _rp _persist() _mark_spend_attempt_once() gen_fn(f"{tag}_{rr_label.lower()}", rr_prompt, _gen_refs(base_label), rr_path) record["gate_reroll"].update({ "state": REROLL_PRODUCED, "sha256": _sha256_file(rr_path), # ★이 방문이 이 파일을 만들었다 — 재개로 옛 파일을 고르면 # 저자는 **그 방문**이지 이번이 아니다. # ★**지금** 방문의 신원이다 — `record` 의 옛 uid 를 읽으면 # 첫 생성에는 None 이고 캐시 방문에는 지난 방문 것이다. "produced_by_visit": _this_visit_uid(), }) _persist() # ── 재판정: 초기 후보 + 재롤 후보를 **한 집합**으로 ── # ★재판정만 실패하면 재개는 **같은 재롤 파일을 다시 판정**한다. _mark_spend_attempt_once() rr_judge = gate_rejudge_fn( tag, prompt, labeled_refs, [_roll_path(out_stem, l) for l in labels_all], labels_all) pick, adm, unchecked = gate_pick_within_admissible( rr_judge, labels_all, rr_judge.get("ranking") or []) record["gate_reroll"].update({ "state": REROLL_JUDGED, "admissible": adm, "unchecked": unchecked, "readings": rr_judge.get("readings"), "ranking": rr_judge.get("ranking") or [], "adopted": pick, }) if pick is None and unchecked: # ★★**「못 봤다」를 「못 쓴다」로 확정하지 않는다** (Codex # BLOCK 5). 판정문에 그 후보 행이 없는 것은 미검사이지 실격이 # 아니다. `reroll` 표식을 **안 남겨서** 같은 세 파일로 재판정만 # 다시 할 길을 연다(새 이미지 0). record["gate"]["outcome"] = GATE_INCOMPLETE record["gate"]["reroll_unchecked"] = list(unchecked) # ★근거를 지운다 — 남겨 두면 소급이 「같은 근거로 이미 봤다」로 # 건너뛰어 복구가 막힌다(evidence 는 초기 판정의 것이다). record["gate"].pop("evidence", None) logger.warning( "multiroll_select[%s]: 재판정에 **안 본 후보**가 있다(%s) — " "실격으로 확정하지 않고 미검사로 남긴다", tag, unchecked) _persist() return selected_now if pick is None: # 전부 보고서 전부 실격 — 그때만 `unresolved` 종착이다. record["gate"]["reroll"] = "no_admissible" logger.warning( "multiroll_select[%s]: 재롤 뒤에도 허용 후보가 없다 — " "**unresolved** 유지 (이미지 1 + 재판정 1 지출)", tag) _persist() return selected_now # ★★**선정본을 바꾼 뒤에 확정한다** (2026-09-20 Codex BLOCK 3). # 재롤 앞에는 `_sel` = A 가 **항상 있다**. 기록을 먼저 # `resolved`·`selected=C` 로 확정하고 복사가 실패하면, 디스크는 # A 인데 기록은 「허용된 C」로 남는다 — 다음 방문은 `gate.reroll` # 때문에 소급을 건너뛰고 `resolved` 라 이음도 그냥 반환해서 # **실격 A 를 그대로 내보낸다**. `dde62916` 에서 닫은 것의 재롤판. # ★결정은 **예고**로만 남긴다 — 소비자가 읽는 종착은 아직 안 바꾼다. record["gate_reroll"]["pending_pick"] = pick _persist() return _materialize_reroll_pick(pick, rr_label) def _retro_gate() -> None: """**호출 0** 으로 캐시 기록에 정책을 적용한다 (2026-09-20 Codex (e)). 정책을 생성 지문에서 뗀 뒤로, 지문이 맞는 기록은 판정을 안 거치고 캐시로 그대로 나간다 — 그러면 정책을 켜도 옛 기록에는 영영 판정이 안 붙고 verify 가 그 샷들을 계속 **미판정**으로 붙잡는다. ★**무호출 반환이 둘**이라 두 곳 다 여기를 지난다. 한 곳만 걸면 끄기/켜기 조합에 따라 절반이 조용히 빠진다. ★재선택이 나면 **선정 파일까지** 그 후보로 맞춘다 — 기록만 B 이고 파일은 A 인 채 나가면, 중간에 끊겼다 재개해도 그 어긋남이 남는다. """ nonlocal sel changed = reapply_gate_to_cached_record( record=record, labels=roll_labels(roll_count), applicable=winner_gate_applicable, input_fingerprint=fingerprint) if changed is None: return # 이미 같은 정책·근거로 봤다 if changed: src = _roll_path(out_stem, changed) if not src.exists(): # ★후보 파일이 없으면 **선택도 `_sel` 도 그대로 둔다** — # 없는 그림을 가리키느니 미검증으로 남긴다. # ★`evidence` 를 **일부러 안 넣는다**: 파일이 복구되면 # 다시 평가해야 한다. 여기에 evidence 를 채우면 영원히 # 건너뛰어 복구할 길이 막힌다 (Codex). record["gate"] = { "outcome": GATE_INCOMPLETE, "policy": GATE_POLICY_VERSION, "applicable": True, "retroactive": True, "reason": f"재선택 후보 {changed} 의 파일이 없다", } logger.warning( "multiroll_select[%s]: 소급 게이트가 %s 로 바꾸려 했으나 " "그 후보 파일이 없다 — 선택·선정본을 그대로 두고 " "**미검증**으로 남긴다", tag, changed) else: _tmp = sel.with_name(sel.name + ".tmp") shutil.copy(src, _tmp) _tmp.replace(sel) # ★**물질화가 끝난 뒤에** 선택을 확정한다 — 기록과 파일이 # 갈리는 창을 남기지 않는다. _from = str(record.get("selected") or "") record["selected"] = changed # ★이번 방문의 **구조적 행동** — 소급 재선택은 **무료**다 # (생성도 판정도 안 샀다). `_this_run` 이라 지출 스냅샷이 # 걸러 낸다. record["gate_select_action_this_run"] = { "action": "reselect_cached", "from": _from, "to": changed, "to_sha": _sha256_file(src), "materialized": True, } logger.warning( "multiroll_select[%s]: 소급 게이트 — 허용되는 기존 " "후보 %s 로 바꿨다 (이미지 구매 0)", tag, changed) _persist() # ── 재개 판정 (지문 일치 record 한정) ── if sel.exists(): if not record.get("selected"): # 선정 persist 이전 crash 창 — sel 은 출처 불명이므로 폐기, # 지문 일치 rolls 로 판정만 재수행 (2차 리뷰 B2) logger.warning( "multiroll_select[%s]: sel 존재 + 선정 기록 없음 — sel " "폐기 후 기존 롤로 재판정", tag, ) sel.unlink() elif "critique" in record or record.get("critique_skipped"): # 정책만 적용 — 생성·판정 호출 0 _retro_gate() # ★★재롤 **이음**도 여기를 지난다 (2026-09-20 Codex BLOCK 2). # 새 판정 자리에만 걸면, 주행 상한에 걸렸거나 배선이 안 됐던 # 샷이 다음 방문에 **영영 못 온다** — `_sel`+`critique_skipped` # 가 저장돼 이 갈래로 먼저 나가기 때문이다. # ★초기 판정을 **다시 사지 않는다** — 기록에 남은 판정문을 # 그대로 쓴다. _rr_after = _gate_reroll_continue( str(record.get("selected") or ""), {"readings": record.get("readings")}, record.get("ranking") or []) # ★산출 저자는 「선택이 바뀌었나」가 아니라 **「최종 bytes 를 # 이번 방문이 만들었나」**다 (2026-09-20 Codex BLOCK 2). # 옛 후보로 바뀐 것뿐인데 저자를 가로채지 않고, 이번에 실제로 # 산 재롤인데 캐시 갈래라고 놓치지도 않는다. _stamp_shot_run( record, produced=_final_author_is_this_visit(str(_rr_after))) return sel, record elif not critique_enabled: record["critique_skipped"] = True _retro_gate() # 정책만 적용 — 생성·판정 호출 0 _rr_after2 = _gate_reroll_continue( str(record.get("selected") or ""), {"readings": record.get("readings")}, record.get("ranking") or []) # 재개 경로에도 최종 판정을 남긴다 (Codex 3차 리뷰) — 빠지면 # **크래시 여부에 따라 같은 이미지가** 재촬영 대상/비대상으로 # 다르게 읽힌다. _set_needs_reshoot(record) # ★★저자는 **최종 bytes 를 만든 방문**이다 — 「캐시면 False」라는 # 별도 규칙을 남기지 않는다 (2026-09-20 Codex). 이 갈래도 재롤 # 이음을 타므로, 여기서 C 를 새로 사서 채택하면 저자는 **이번 # 방문**이다. 무조건 False 로 두면 그 생성자를 놓친다. _stamp_shot_run( record, produced=_final_author_is_this_visit(str(_rr_after2))) _persist() return sel, record else: # sel 존재 + critique 미실행 → 검사·수정만 소급 (s40 _resume) # ★유료 구간 입구 — _critique_and_fix 안의 여섯 호출을 덮는다. # 그 함수는 모듈 수준이라 이 closure 를 못 쓴다 → 부르기 전에. _mark_spend_attempt_once() sel = _critique_and_fix( tag=tag, prompt=_crit_prompt(record["selected"]), labeled_refs=_crit_refs(record["selected"]), out_stem=out_stem, selected=record["selected"], critique_fn=critique_fn, fix_gen_fn=fix_gen_fn, # type: ignore[arg-type] fix_head=fix_head, fix_tail=fix_tail, fix_label=fix_label, record=record, fix_ref_label=fix_ref_label, fix_ref_gate=fix_ref_gate, fix_missing_head=fix_missing_head, fix_missing_tail=fix_missing_tail, fix_rejudge_fn=fix_rejudge_fn, composition_critique_fn=composition_critique_fn, regen_gen_fn=regen_gen_fn, regen_prompt=_crit_prompt(record["selected"]), regen_head=regen_head, regen_tail=regen_tail, regen_faults_header=regen_faults_header, regeneration_issue_policy=regeneration_issue_policy, repair_method=repair_method, ) _set_needs_reshoot(record) # ★소급 critique/fix 는 **유료**지만 최종 _sel 의 저자는 승패가 # 가른다 — 수리가 없었거나 졌으면 저자는 이전 방문이다. # 하나로 쓰면 계보를 가로채거나 지출을 놓친다. from app.modules.pipeline.still_recipe import fix_stage_won _stamp_shot_run(record, produced=fix_stage_won(record)) _persist() return sel, record # ── phase 0: 지문 기록 (롤 재개의 출처 증명) ── record["input_fingerprint"] = fingerprint record["prompt"] = prompt if roll_prompts is not None: record["roll_prompts"] = dict(roll_prompts) if roll_refs is not None: # bytes 참조는 표식만 — record 오염 방지 (기존 refs 관례 동일) _roll_meta = dict(roll_ref_metadata or {}) record["roll_refs"] = { lab: build_ref_records(roll_refs[lab], _roll_meta.get(lab)) for lab in sorted(roll_refs) } _persist() # ── N롤 생성 (지문 일치 시 롤 파일 존재 = skip) ── labels = roll_labels(roll_count) missing = [ lab for lab in labels if not _roll_path(out_stem, lab).exists() ] if missing: _mark_spend_attempt_once() # ★생성 전에 못박는다 if parallel_rolls and len(missing) > 1: # 병렬 worker 는 thread-local budget(W20E5 B1)과 ContextVar 기반 # generation_context(Codex BLOCKING-3) 를 상속하지 않는다 — 둘 다 # 명시 전파. 예외는 전 future 완료 후 canonical 라벨 순으로 raise # (NARROW-1: completion timing 무관 결정론), 성공 롤 파일은 남아 # resume 재사용. from concurrent.futures import ThreadPoolExecutor from app.core.image_call_budget import bind_current_budget from app.core.send_ledger import bind_current_ledger from app.modules.llm.opik_trace import bind_current_trace from app.services.image_capture.context import ( bind_current_generation_context, ) # ★**넷 다** 스레드 지역이다 — 하나라도 빠뜨리면 그 축만 조용히 # 무너진다. 발송 장부를 안 실으면 **팬아웃 안의 발송이 통째로 # 안 잡혀** 「전송 0」으로 읽힌다(2026-09-20). bound = bind_current_budget( bind_current_ledger( bind_current_generation_context(bind_current_trace(gen_fn)))) with ThreadPoolExecutor(max_workers=len(missing)) as pool: futures = { lab: pool.submit( bound, f"{tag}_{lab.lower()}", _roll_prompt(lab), _gen_refs(lab), _roll_path(out_stem, lab), ) for lab in missing } # ★**성공한 라벨만** 저자 후보로 담는다 — 예외로 죽은 라벨은 # 다음 방문이 만든다. 예외는 그 뒤에 canonical 순으로 올린다. for lab in missing: if futures[lab].exception() is None: _made_here.add(lab) for lab in missing: exc = futures[lab].exception() if exc is not None: raise exc else: for lab in missing: gen_fn( f"{tag}_{lab.lower()}", _roll_prompt(lab), _gen_refs(lab), _roll_path(out_stem, lab), ) # ★성공한 **그 라벨**만 담는다(중간에 죽으면 앞엣것만 남는다). _made_here.add(lab) cand_paths: List[Path] = [_roll_path(out_stem, lab) for lab in labels] # ── 판정·선정 ── # ★flip 분기 **앞**에서 찍는다 — judge_flip 안에 찍으면 비-flip 갈래의 # 판정 호출이 안 덮인다. _mark_spend_attempt_once() # ★판정자가 **순서를 스스로 소유**하면 여기서 또 뒤집지 않는다 # (2026-08-29 사용자 지시 "즉 두번만"). # # cross-model 어댑터는 Gemini 정순 + Grok 역순을 한 번 안에서 병렬로 # 다 한다. 그런데 아래 flip 갈래는 판정자를 **정·역 두 번** 태운다 — # 겹치면 2×2 = **다시 4콜**이 되어 지시를 어긴다. if judge_flip and getattr(judge_fn, "owns_order", False): judge_flip = False record["judge_flip_skipped"] = "judge_fn_owns_order" if judge_flip: # 정순·역순 2회 블라인드 판정 (BLOCKING-2). 판정 refs=공유 # labeled_refs — roll_refs(콘티 유무)는 미노출. # 리뷰 BLOCKING-1: 지문 일치 record 의 durable forward 는 재검증 # 후 재사용(정순 VLM 재호출 0) — stale 경로에서 record 가 이미 # 초기화되므로 여기 도달한 record 는 지문 일치가 증명된 상태. # malformed durable 은 정순부터 재판정(영구 차단 금지). # NARROW-1: 컨테이너부터 dict 정규화 — truthy 비 dict durable 이 # try 진입 전 AttributeError 로 새는 경로 차단 _prior_jf = record.get("judge_flip") if not isinstance(_prior_jf, dict): _prior_jf = {} judge_fwd_raw = _prior_jf.get("forward_raw") judge_fwd = None if judge_fwd_raw is not None: try: judge_fwd = normalize_flip_verdict( judge_fwd_raw, {lab: lab for lab in labels}, labels ) except ValueError: logger.warning( "multiroll_select[%s]: durable forward 판정 malformed" " — 정순 재판정", tag, ) judge_fwd_raw = None if judge_fwd is None: judge_fwd_raw = judge_fn( tag, prompt, labeled_refs, cand_paths, labels ) judge_fwd = normalize_flip_verdict( judge_fwd_raw, {lab: lab for lab in labels}, labels ) # 양회 raw+normalized 병록 계약 — forward 도 normalized 저장 record["judge_flip"] = { "forward_raw": judge_fwd_raw, "forward_normalized": judge_fwd, } _persist() # NARROW-2: 정순 직후 durable — crash 시 재호출 절감 judge_rev_raw = judge_fn( tag, prompt, labeled_refs, list(reversed(cand_paths)), labels ) judge_rev = normalize_flip_verdict( judge_rev_raw, flip_display_to_canonical(labels), labels ) selected, combined = combine_flip_verdicts( judge_fwd, judge_rev, labels, list(flip_priority or labels) ) record["judge_flip"].update({ "reverse_raw": judge_rev_raw, "reverse_normalized": judge_rev, "combined": combined, }) totals = dict(combined["totals"]) ranking = list(combined["ranking"]) verdicts = judge_fwd["verdicts"] # 정순 관점 기록(감사) # 정순·역순 **둘 다** 선언했을 때만 참으로 본다. 한쪽만으로 잡으면 # 신호가 무의미해진다는 것이 이 프로젝트의 실측이다(이중 판정에서 # 절반 기준으로 했다가 45%가 찍혔다 — combine_select_verdicts 주석). judge_result: Dict[str, Any] = { "all_candidates_fail": bool( judge_fwd.get("all_candidates_fail") and judge_rev.get("all_candidates_fail")), "readings": judge_fwd.get("readings"), } else: judge = judge_fn(tag, prompt, labeled_refs, cand_paths, labels) totals, selected = gemini_select(judge) ranking = judge.get("ranking") or [] verdicts = judge.get("verdicts") or [] judge_result = judge # ★심판이 "전 후보 실패"를 선언했으면 기록에 남긴다 (2026-08-07). # # 판정 팩은 "least-bad 를 좋다고 하지 말고 그렇게 선언하라, 호출측이 그 # 값으로 재촬영을 정한다"고 적어 두었고 스키마도 필수다. 그런데 소비처가 # 0이라 선언이 그대로 사라졌다 — 자동차 기하 붕괴처럼 후보 전체가 # 실패한 샷도 정상 선정본과 구분되지 않는다. 여기서 스텝을 죽이지는 # 않는다(고칠 기회는 준다). 대신 typed 로 남겨 감사·갤러리·재촬영 # 선별이 이 값을 쓸 수 있게 한다. _acf = bool(judge_result.get("all_candidates_fail")) if _acf: # ★단계 이름을 붙인다 (2026-08-07 Codex 2차 리뷰 수용). 이 값은 # **초기 롤 판정** 결과이지 최종 산출 상태가 아니다 — 뒤이어 # critique·i2i·재판정이 `_sel` 을 바꾼다. 이름 없이 top-level 에 # 두면 두 방향으로 오독된다: 초기 롤이 전부 실패했어도 수정본이 # 정상 승리하면 재촬영이 불필요한데 True 가 남고, 반대로 초기는 # 통과했는데 원본·수정본이 재판정에서 둘 다 실패하면 표시가 없다. # 최종 판단은 `needs_reshoot` 가 수정 완료 뒤에 계산한다. record["initial_roll_all_fail"] = True logger.warning( "multiroll_select[%s]: 심판이 초기 롤 전부 실패를 선언 — " "선정본(%s)은 least-bad 다", tag, selected) if judge_result.get("readings"): record["readings"] = judge_result["readings"] # G+Q 합의 경로(agree/gemini_priority/combined/single_*)·격차·양쪽 승자 # — durable 감사 계약 (Codex 리뷰 BLOCK-3: 한 심판 장애로 single_* 강등된 # 샷이 record 에서 평범한 판정과 구분되지 않았다). flip 경로는 # judge_flip.forward_raw/reverse_raw 에 판정 원본이 그대로 남아 별도 # 복사가 필요 없다. if judge_result.get("gq"): record["gq"] = judge_result["gq"] # ★cross-model 판정 기록도 durable 로 옮긴다 (2026-08-29). # 옛 `gq` 만 옮기면 새 갈래의 감사 근거가 통째로 사라진다 — 어느 모델이 # 어느 순서를 맡아 무엇을 골랐는지가 record 에 안 남는다. if judge_result.get("cross_model_order"): record["cross_model_order"] = judge_result["cross_model_order"] # (2026-08-13 Codex GG46 R1 BLOCK-1) 두 슬롯 합산 근거도 durable — # 정규화·adjusted·위반 합집합·양쪽 승자가 없으면 그 선정을 기록만으로 # 되짚을 수 없다. # ★2026-08-29 정정: 종전 주석은 「combined 경로에서만 존재」라고 적었는데 # 더는 맞지 않는다. 선정 정책 v3 부터 **두 슬롯이 다 성공하면 승자가 # 같아도 합산을 태우므로** agree 갈래에도 `dual` 이 생긴다. 키가 없는 # 것은 **한 슬롯만 살아남은 단독 경로**와 priority/단독 판정뿐이다. if judge_result.get("dual"): record["dual"] = judge_result["dual"] # ── 이긴 후보의 실격 게이트 (2026-09-20) ── # ★`record.update` **앞**이다 — 기록과 `_sel` 물질화가 같은 값을 봐야 # 한다. 여기서 바뀌는 것은 **이미 산 후보 중 무엇을 쓰느냐**이고 # 새 그림을 사지 않는다. selected = apply_winner_violation_gate( record=record, judge_result=judge_result, labels=roll_labels(roll_count), selected=selected, ranking=ranking, applicable=winner_gate_applicable, input_fingerprint=fingerprint) record.update( { "totals": totals, "selected": selected, "ranking": ranking, "verdicts": verdicts, "refs": build_ref_records(labeled_refs, ref_role_metadata), } ) # phase 1: 선정 결정을 sel 물질화 **이전에** durable persist — copy 직후 # crash 시 KeyError/출처불명 sel 창 제거 (2차 리뷰 B2). sel 은 tmp→atomic. _persist() _tmp_sel = sel.with_name(sel.name + ".tmp") shutil.copy(_roll_path(out_stem, selected), _tmp_sel) _tmp_sel.replace(sel) # ── 게이트 재롤 (2단계) — **`_sel` 을 물질화한 뒤**다 ────────────── # ★여기가 뒤인 이유: 재판정이 실패해도 `_sel` 과 선정 기록이 남아, # 재개가 캐시 갈래로 들어와 **초기 판정을 다시 사지 않고** 재롤 # 이음만 태운다 (Codex BLOCK 2b). selected = _gate_reroll_continue(selected, judge_result, ranking) # ── 결함 검사·수정 (게이트) ── if not critique_enabled: record["critique_skipped"] = True # 수정을 돌리지 않으므로 최종 산출 = 초기 선정본(또는 재롤본)이다. _set_needs_reshoot(record) # ★저자는 **최종 bytes 를 만든 방문**이다 — 재롤이 붙으면서 # 「이 방문이 롤을 돌았으니 무조건 True」가 아니게 됐다. 재개가 # 옛 재롤 파일을 채택하는 갈래가 실제로 있다(Codex BLOCK 2). _stamp_shot_run( record, produced=_final_author_is_this_visit(selected)) _persist() return sel, record _mark_spend_attempt_once() # ★유료 구간 입구 sel = _critique_and_fix( tag=tag, prompt=_crit_prompt(selected), labeled_refs=_crit_refs(selected), out_stem=out_stem, selected=selected, critique_fn=critique_fn, fix_gen_fn=fix_gen_fn, # type: ignore[arg-type] fix_head=fix_head, fix_tail=fix_tail, fix_label=fix_label, record=record, fix_ref_label=fix_ref_label, fix_ref_gate=fix_ref_gate, fix_missing_head=fix_missing_head, fix_missing_tail=fix_missing_tail, fix_rejudge_fn=fix_rejudge_fn, composition_critique_fn=composition_critique_fn, regen_gen_fn=regen_gen_fn, regen_prompt=_crit_prompt(selected), regen_head=regen_head, regen_tail=regen_tail, regen_faults_header=regen_faults_header, regeneration_issue_policy=regeneration_issue_policy, repair_method=repair_method, ) _set_needs_reshoot(record) _stamp_shot_run(record, produced=True) _persist() # phase 2: 결함 처리 완결 return sel, record