"""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 from pathlib import Path from typing import ( Any, Callable, Dict, List, Mapping, Optional, Sequence, Tuple, ) logger = logging.getLogger(__name__) # 판정 스키마 라벨 상한 = 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그룹은 편집 가능한 결함이 하나도 없었다. 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." ) def build_regen_prompt( issues: Sequence[Dict[str, str]], base_prompt: str, regen_head: str = "", regen_tail: str = "", ) -> str: """원 브리프 + 구조 결함 목록 → 재생성 프롬프트(참조 없음). base_prompt 는 선정된 롤이 실제로 쓴 프롬프트다. 같은 브리프를 다시 주되 무엇이 틀렸는지를 덧붙인다 — 프롬프트를 바꿔 버리면 무엇이 효과를 냈는지 알 수 없고, 브리프 준수 여부도 판정할 수 없다. """ faults = "STRUCTURAL FAULTS OF THE REJECTED ATTEMPT:\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 _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 = "", regeneration_issue_policy: str = REGEN_ISSUE_POLICY_LEGACY_EDIT, ) -> Path: """선정 원본 결함 검사 → 결함 시 i2i 수정 → _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 fixable: List[Dict[str, Any]] = [] deferred: List[Dict[str, Any]] = [] unfixable: List[Dict[str, Any]] = [] for issue in issues: if issue.get("unfixable"): unfixable.append(issue) elif deferring and issue.get("needs_regeneration"): 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 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) if regen_issues and regen_gen_fn is not None and regen_prompt.strip(): regen = build_regen_prompt( regen_issues, regen_prompt, regen_head, regen_tail) record["regen_prompt"] = regen record["regen_issue_count"] = len(regen_issues) record["repair_mode"] = "regenerate" logger.info( "multiroll_select[%s]: 구조 결함 %d건 — i2i 편집 대신 재생성", tag, len(regen_issues)) # ★원본 산출만 빼고 원래 참조는 그대로 넘긴다. 원본을 참조로 # 물리면 같은 매스가 재현되지만, 참조를 통째로 비우면 형태 권위 # (구조 스케치 등)까지 함께 사라진다 — 재생성은 "이 브리프와 이 # 참조로 다시 찍는 것"이지 "맨손으로 다시 그리는 것"이 아니다. 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" # ★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} 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 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, 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 = "", regeneration_issue_policy: str = REGEN_ISSUE_POLICY_LEGACY_EDIT, ) -> 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) # ★첫 유료 호출(롤 생성) 전에 재생성 계약을 검증한다 — 결손을 기존 # 편집으로 하강시키지 않고 막는다. 브리프는 판정 뒤에야 라벨이 정해지므로 # **모든 라벨에 대해 실제 조립 함수를 돌려** 비지 않는지 본다. 원천만 # 훑으면 조립 규칙이 바뀔 때 검사와 실물이 어긋난다. 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 {} 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) _fp_extra = extra_fingerprint 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} 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 = {} # ── 재개 판정 (지문 일치 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"): return sel, record # 완결 — 전부 skip elif not critique_enabled: record["critique_skipped"] = True # 재개 경로에도 최종 판정을 남긴다 (Codex 3차 리뷰) — 빠지면 # **크래시 여부에 따라 같은 이미지가** 재촬영 대상/비대상으로 # 다르게 읽힌다. _set_needs_reshoot(record) _persist() return sel, record else: # sel 존재 + critique 미실행 → 검사·수정만 소급 (s40 _resume) 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, regeneration_issue_policy=regeneration_issue_policy, ) _set_needs_reshoot(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 관례 동일) record["roll_refs"] = { lab: [ { "label": str(l), "path": ( str(p) if isinstance(p, (str, Path)) else f"" ), } for l, p in roll_refs[lab] ] for lab in sorted(roll_refs) } _persist() # ── N롤 생성 (지문 일치 시 롤 파일 존재 = skip) ── labels = roll_labels(roll_count) 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 missing = [ lab for lab in labels if not _roll_path(out_stem, lab).exists() ] 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.services.image_capture.context import ( bind_current_generation_context, ) bound = bind_current_budget(bind_current_generation_context(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 } 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), ) cand_paths: List[Path] = [_roll_path(out_stem, lab) for lab in labels] # ── 판정·선정 ── 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"] # (2026-08-13 Codex GG46 R1 BLOCK-1) combined 합산 근거도 durable — # 정규화·adjusted·위반 합집합·양쪽 승자가 없으면 combined 선정을 # 기록만으로 되짚을 수 없다. combined 경로에서만 존재하는 키라 # agree/priority/단독 판정 record 는 키 부재 그대로다. if judge_result.get("dual"): record["dual"] = judge_result["dual"] record.update( { "totals": totals, "selected": selected, "ranking": ranking, "verdicts": verdicts, "refs": [ { "label": lab, # bytes 참조(엔티티)는 내용 대신 표식만 — record 가 # 수십 MB 로 오염되던 E2E 실측 결함 fix "path": ( str(p) if isinstance(p, (str, Path)) else f"" ), } for lab, p in labeled_refs ], } ) # 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) # ── 결함 검사·수정 (게이트) ── if not critique_enabled: record["critique_skipped"] = True # 수정을 돌리지 않으므로 최종 산출 = 초기 선정본이다. _set_needs_reshoot(record) _persist() return sel, record 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, regeneration_issue_policy=regeneration_issue_policy, ) _set_needs_reshoot(record) _persist() # phase 2: 결함 처리 완결 return sel, record