"""OutdoorStructureSeedStep — 레인2 구조물 seed (Stage D, order 21.92). outdoor_lane_plan 에서 structure_plate 세그먼트 바인딩을 가진 그룹마다: seed = 구조물 확정 실사 — 순수 T2I nb2 멀티롤(3롤→judge→critique→fix) data.groups[] = {status, seed_png_path, seed_asset_id, seed_selected, seed_prompt} 실패 그룹 = {"status": "failed", "error": str} (그룹 단위 격리) 2026-07-16 사용자 확정(복잡 구조물=맵 제외+A/B 선택, Codex R7): siteplan (SITE PLAN 재투영) 산출 제거 — 유일 소비처(레인2 마커 맵 스케치)가 소멸. 구 siteplan 파일·ImageAsset row 는 보존만 하고 신규 생성 0. seed 입력 SOT(Codex 배선 조건 ⑥): outdoor_place_spec persistent_site 검증 서술 + excluded_transient negative contract (derive_seed_inputs). persisted lane plan 은 소비 직전 revalidate_persisted_plan 재검증(fail-closed). ImageAsset: asset_type="structure_seed", entity_id=, variant_type="seed"(UPSERT 키). """ import json import logging import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional from app.core.step_runner import StepRunner logger = logging.getLogger(__name__) # v2 (2026-07-16 R7): siteplan 산출 제거 — CP shape 에서 siteplan 키 # 소멸, 구 CP 소비 금지 # v3 (2026-07-28): 순수 T2I → 검색 선택 실사 **형태 참조 i2i**. # 필수 대상(structure_plate 바인딩)은 참조 결손 시 fail-closed. # v5 (2026-08-01): 감사 계약 shape 변경 — `seed_decision` 이 변형 전용에서 # **모든 성공 entry 의 필드**가 되고, 재생성 이슈 정책(이름·버전)과 보류된 # 결함(원문·개수)·수정 생략 사유·수정 모드가 typed 로 들어온다. config_hash # 만으로는 부족하다: 해시가 없는 옛 CP 에는 mismatch 검사가 적용되지 않는 # 경로가 있어 스키마 버전이 따로 의미를 갖는다(선례 e18a776e 의 3→4). # v6 (2026-08-03): entry 에 `seed_paths` 추가 — 어느 경로가 돌았고(직접 / # 스케치 경유) 어느 쪽이 이겼는지, 도해 파일·자산이 무엇인지. 롤 라벨의 뜻이 # 저작 변형에서 **경로**로 바뀌었으므로 옛 CP 의 `seed_selected` 를 같은 # 뜻으로 읽으면 안 된다. SCHEMA_VERSION = 6 # structure_seed 팩 — v5(2026-07-22 E2E11 fix①)=v4+HUMAN-SCALE CALIBRATION # 상시 주입(옥탑 과대 실측 — interior 결손 시 규모 계약 0 이던 창 봉합). # config_hash 에 접혀 있어 승격만으로 기존 완료 CP 자동 stale. SEED_PACK_VERSION = "5" # seed_prompt_variants 팩 — v6(2026-07-22 E2E11 ①)=v5+규모 계약 코드 조립 # (human_scale 상시 + scale_clause/INTERIOR 조건부, 롤·conformance 양쪽). # E2E11 실측: variants ON 경로에 규모 절 0(30/30) — structure_seed v5 # human_scale 은 variants OFF 전용이라 죽은 배선이던 결손 봉합. # v9 (2026-07-23 Codex v8b 재리뷰): v8 in-place 수정분 재발행(버전 불변 # 계약 복구) + room_inventory scene_index·applicability + 표시 필요 # 여부 판정·근거 + 실내 씬(INTERIOR SCENES) 배선 — 상세=seed_prompt_variants. # v10 (2026-07-23 슬라이스 E 육안): 주거 생활감 sterile 근본 대응 — # judge 타깃 active-use axis-2 실검사 + critique bare-target 문턱 강화 + # plate_rules 사실 서술 전환(전부 유형 중립) — 상세=seed_prompt_variants. # v14 (2026-08-02 사용자 지시): 표시물 통칭과 업종 목록을 프롬프트에서 # 제거 — "어떤 업종이면 간판이 필요한가"를 판단시키던 절을 없애고 활자가 # 있다면 어디서든 같은 내용이라는 한 줄로. 저작 출력도 한 칸(lettering_text). # v18 (2026-08-03): 명세가 든 **별개의 물건**의 형태·색·재질을 저작이 근거 # 없이 확정하지 않는다 — 확정해 두면 그 물건이 찍힌 참조 사진을 붙여도 # 저작이 이긴다(실측). 구조물 자신의 살에 대한 확정 규칙은 그대로다. # v19 (2026-08-03): 가장 큰 표시면의 몫은 **이 장소를 같은 종류의 다른 # 곳과 구별하는 이름**이다. 종류를 가리키는 낱말이 그 자리에 걸리면 그 # 한 가지 일을 못 한다(실측 3건). 종류 낱말이 옳은 자리는 가까이서 읽는 # 작은 실용 표시 — 규칙의 축은 낱말이 아니라 자리와 크기다. # v20 (2026-08-03): 종류 낱말을 몰아낸 자리를 **장면이 채웠다**(때+행위로 # 지은 이름). 이름은 가리키는 것이지 묘사하는 것이 아니다 — 이야기를 # 지워도 남는 것으로만 짓고, 때·사건·정서·의미에서 짓지 않는다. # v21 (2026-08-03): 금지를 쌓은 것이 원인이었다 — 작명 방식을 막아 놓고 # 실재 상호도 금지하니 남은 재료가 장면뿐이었다. 짧은 지시 하나로 줄이고 # "줄 게 없으면 지어내라"를 명시했다. SEED_VARIANTS_PACK_VERSION = "21" # ★재생성 이슈 정책 (2026-08-01) — 씨드는 재생성을 **배선하지 않는다**. # v13 팩부터 critique 가 층수·매스·footprint 처럼 국소 편집의 사정거리 밖인 # 결함을 `needs_regeneration` 으로 표시한다. 재생성 경로는 구현돼 있으나 어느 # production 호출자도 배선하지 않았고, 배선하지 않은 상태의 공용 기본 동작은 # **그 결함까지 i2i 편집 지시에 실어 보내는 것**이었다. 실측(25그룹): 8그룹이 # 그 경로를 탔고 구조 결함 11건이 편집에 실렸다 — 그중 3그룹은 편집 가능한 # 결함이 하나도 없어 고칠 수 없는 것만으로 유료 호출을 태웠다. 씨드는 재판정도 # 배선하지 않아 그 실패본이 원본을 무판정 대체한다. # # 그래서 재생성을 켜는 대신 **끈 상태의 동작을 정의**한다: 그 결함은 손대지 # 않고 무엇이 남았는지 기록만 한다(`defer`). 켜려면 생성 함수·브리프·재판정 # 계약이 함께 있어야 하며, 셋이 갖춰지지 않은 채 켜면 공용 진입점이 # fail-closed 로 막는다(`validate_regeneration_contract`). # ★이 값은 config_hash 에 실린다 — 정책이 바뀌면 완료 CP 가 stale 된다. SEED_REGEN_ISSUE_POLICY = "defer" # 형태 참조 소비 계약의 버전 — 소비 방식(어느 프롬프트에 절이 실리는지, # 어떤 계보를 남기는지, 어떤 검증을 거치는지)이 바뀌면 올린다. seed 의 # config_hash 에 실려 **완료된 CP 를 의도적으로 stale** 시킨다. # v1 (2026-08-01): 절을 전 롤에 주입 · form_ref 자산을 입력 간선으로 영속 · # producer CP partition/정책/팩 exact 검증. # v2 (2026-08-01, Codex 2차 재리뷰): partition 을 **현재 spec 기준으로 재계산해 # exact 비교**(stale no_spec 차단) · form_ref asset_id 의 # **실존·소유·경로**까지 검증(dangling UUID 차단) · # 소비한 참조를 CP 에 typed 로 영속. # ★올리지 않으면 v1 로 완료된 CP 가 그대로 재사용되어 이 # 수정이 실행에 도달하지 않는다(지난 wave 실측 교훈). FORM_REFERENCE_CONSUMER_CONTRACT_VERSION = "2" # ★두 경로 계약 (2026-08-03 사용자 확정 흐름) ───────────────────────── # 검색해 고른 사진 한 장에서 곧장 그린 것과, 그 사진을 보고 형태만 남긴 선 # 도해를 한 번 거쳐 그린 것을 **하나씩** 만들어, 참조 사진 앞에서 어느 쪽이 # 더 그 종류로 통하는지 고르게 한다. # # ★스케치 필요 여부를 **미리 판정하지 않는다.** 이전 설계는 그것을 사전에 # 정했는데, 판정이 틀리면 그대로 손해였다. 오늘 실측이 그 방향을 뒷받침한다 # — 절대 문턱은 어디에 두든 한쪽으로 쏠렸고 상대 비교만 안정적이었다. # 자연 지형처럼 선 도해가 의미 없는 대상에서는 직접 경로가 이기면 되고, # 그 승패 자체가 다음 판단의 근거가 된다. SEED_PATH_DIRECT = "A" # 사진 한 장이 형태와 표면을 다 맡는다 SEED_PATH_SKETCHED = "B" # 형태=선 도해 / 표면=사진 SEED_PATH_LABELS = [SEED_PATH_DIRECT, SEED_PATH_SKETCHED] # 경로 구성·판정 방식이 바뀌면 올린다 — config_hash 에 실려 완료 CP 를 stale. SEED_PATH_POLICY_VERSION = "path_ab_v1_ref_relative" def _regen_issue_policy_version() -> str: """공용 재생성 이슈 정책의 버전 — 정책 의미가 바뀌면 함께 올라간다.""" from app.modules.pipeline.multiroll_select import ( REGEN_ISSUE_POLICY_VERSION, ) return REGEN_ISSUE_POLICY_VERSION def _form_reference_consumer_hash_payload() -> Dict[str, Any]: """seed config_hash 에 실을 형태 참조 소비 계약 스탬프.""" import hashlib as _hl from app.modules.pipeline.search_grounded_ref import ( TARGET_POLICY_VERSION, build_fitting_ref_clause, build_form_only_clause, build_form_only_clause_sketched, build_sketch_transform_clause, resolve_ref_pack_version, ) # 두 경로가 쓰는 절 셋을 한 해시로 묶는다 — 어느 하나만 고쳐도 산출이 # 달라지므로 따로 둘 이유가 없다. 부속 관할절은 개수를 인자로 받으므로 # 대표값(1,1)으로 접는다 — 잡으려는 것은 개수가 아니라 **절 본문**이고, # 실제 개수는 그룹 지문(form_ref.fittings)이 이미 싣는다. _h = _hl.sha256() for _clause in (build_form_only_clause(), build_sketch_transform_clause(), build_form_only_clause_sketched(), build_fitting_ref_clause(head=1, count=1)): _h.update(_clause.encode("utf-8")) _h.update(b"\x00") return { "form_ref_consumer_contract": FORM_REFERENCE_CONSUMER_CONTRACT_VERSION, "form_ref_pack": resolve_ref_pack_version(), "form_ref_target_policy": TARGET_POLICY_VERSION, # 팩 버전이 같아도 절 본문이 바뀌면 생성 계약이 바뀐 것이다. "form_only_clause_sha": _h.hexdigest()[:16], } def _require(cond: bool, code: str, message: str) -> None: """계약 위반을 **항상 AppError 로** 귀결시킨다. malformed 입력이 AttributeError/TypeError 로 새면 호출측 try 가 그것을 "생성 실패"로 오해해 격리하거나, 상위에서 500 으로 뜬다. 계약 위반은 422 로 명시한다. """ if not cond: from app.core.errors import AppError raise AppError(code=code, message=message, status_code=422) def _str_set(value: Any, field: str) -> set: """목록 필드를 검증하며 집합으로 — 타입·중복·빈 문자열을 거른다.""" _require(isinstance(value, list), "structure_seed.form_reference_malformed", f"{field} 가 목록이 아니다: {type(value).__name__}") for v in value: _require(isinstance(v, str) and v.strip(), "structure_seed.form_reference_malformed", f"{field} 원소가 비어 있지 않은 문자열이 아니다: {v!r}") out = set(value) _require(len(out) == len(value), "structure_seed.form_reference_malformed", f"{field} 에 중복이 있다: {sorted(value)}") return out def build_consumed_ref_record(form_ref_fp: Optional[Dict[str, Any]] ) -> Dict[str, Any]: """CP 에 남길 **소비한 형태 참조**의 typed 기록. [2026-08-01 Codex 2차 재리뷰 BLOCKING 2] 성공 entry 에 소비한 참조의 어떤 필드도 없었다. 변형 모드가 꺼져 있으면 ``extra_fingerprint`` 조차 CP 에 안 남아, **DB 간선 하나 말고는 "어느 참조를 물고 그렸는가"를 되짚을 근거가 전혀 없다.** 간선이 거짓이면 감사 자체가 불가능해진다. 참조 없이 간 그룹도 ``asset_id=None`` 으로 **명시 기록**한다 — 키 부재로 두면 "안 물었다"와 "기록을 안 남겼다"가 구분되지 않는다. """ from app.modules.pipeline.search_grounded_ref import ( TARGET_POLICY_VERSION, resolve_ref_pack_version, ) fp = form_ref_fp or {} return { "asset_id": fp.get("asset_id"), "path": fp.get("path"), "sha256": fp.get("sha256"), # 부속 참조는 자산이 아니라 파일이다 — 간선으로는 남지 않으므로 # 무엇을 물고 그렸는지는 이 기록만이 근거다. 제외된 것도 남긴다. "fittings": list(fp.get("fittings") or []), "fittings_dropped": list(fp.get("fittings_dropped") or []), "pack_version": resolve_ref_pack_version(), "policy_version": TARGET_POLICY_VERSION, "consumer_contract": FORM_REFERENCE_CONSUMER_CONTRACT_VERSION, } def partition_form_reference_targets( *, universe: Any, spec_groups: Any, ) -> Any: """(mandatory, no_spec) — 검색 참조 대상 분할의 **단일 소유자**. [2026-08-01 Codex 2차 재리뷰 BLOCKING 1] producer 와 consumer 가 각자 분할을 갖고 있으면, 소비자는 CP 에 적힌 분할을 그대로 믿을 수밖에 없다. 그러면 구 CP 의 ``no_spec`` 에 든 그룹은 **그 뒤 spec 이 생겨 지금은 참조를 만들 수 있는 상태여도** 계속 "근거 없음"으로 남아 필수에서 빠지고, 참조 없이 조용히 순수 T2I 로 내려간다(Codex 직접 재현: completed_count=1, labeled_refs=[]). 분할 기준은 하나다 — **현재 outdoor_place_spec 에 items 가 있는가.** 검색어를 저작할 근거가 그것뿐이기 때문이다. 두 스텝이 같은 함수를 부르면 두 분할이 갈라질 수가 없다. """ mandatory: List[str] = [] no_spec: List[str] = [] for gid in sorted(set(universe or [])): entry = (spec_groups or {}).get(gid) or {} spec = entry.get("spec") if isinstance(entry, dict) else None items = spec.get("items") if isinstance(spec, dict) else None if items: mandatory.append(gid) else: no_spec.append(gid) return mandatory, no_spec #: 앞 스텝(`outdoor_structure_form_reference`)이 「생성이 정본」으로 확정한 #: 그룹의 status. **문자열을 두 곳에 적지 않는다** — 앞 스텝의 상수를 그대로 #: 가져온다. from app.core.steps.outdoor_structure_form_reference_step import ( # noqa: E402 GENERATIVE_STATUS as _FORMREF_GENERATIVE_STATUS, VISUAL_AUTHORITY_CONTRACT_VERSION as _FORMREF_AUTHORITY_CONTRACT, ) def resolve_form_reference_contract( *, formref_cp: Optional[Dict[str, Any]], target_gids: List[str], spec_groups: Dict[str, Any], ) -> set: """선행 form_reference CP 를 검증하고 **필수 그룹 집합**을 돌려준다. [2026-08-01 Codex 리뷰 BLOCKING 3 + 지적 10] 이전 구현은 mandatory = cp.get("data", {}).get("mandatory_group_ids", []) \\ or collect_lane2_groups(lane_data, all_groups=False) 였는데 이 ``or`` 폴백이 **두 상황을 구분하지 못했다** — ①선행 미실행 ②선행이 정상이고 필수 대상 0건. ②에서 폴백이 판정을 뒤집어 spec 결손으로 제외된 그룹까지 되살리고, 통과할 수 없는 게이트를 세운 뒤 "form_reference 를 먼저 통과시켜라"라는 **원인을 가리는 메시지**를 냈다. 게다가 폴백은 ``all_groups=False`` 고정인데 seed·form_reference 의 대상은 ``outdoor_seed_all_groups_enabled`` 를 따른다. 플래그가 켜지면 두 스텝이 서로 다른 우주를 보고, 구 CP 가 남아 있으면 참조 없는 그룹이 조용히 순수 T2I 로 내려간다. 새 계약: - 대상이 없으면 선행도 필요 없다(빈 집합). - CP 가 없거나 ``status != completed`` 면 **선다**. 조용한 degrade 금지. - CP 가 본 우주(``mandatory ∪ no_spec``)가 **지금 대상과 정확히 같아야** 한다. 어긋나면 구 CP 재사용이므로 선다. - 구 shape(필수 키 부재)는 **default-deny** — 관대하게 받으면 계약 이전 산출이 현재 지문으로 승격된다. """ from app.core.errors import AppError targets = set(target_gids) if not targets: return set() if not formref_cp: raise AppError( code="structure_seed.form_reference_missing", message=( "outdoor_structure_form_reference 체크포인트가 없다 — 검색 " f"선택 참조가 필요한 대상 {len(targets)}그룹이 순수 T2I 로 " "내려가지 않는다 (fail-closed). 선행 스텝을 먼저 통과시켜라"), status_code=422, ) # ★타입 검사가 먼저다 — 아래 .get() 이 malformed 입력에서 AttributeError # 로 새면 계약 위반이 "생성 실패"로 오해되거나 500 으로 뜬다. _require(isinstance(formref_cp, dict), "structure_seed.form_reference_malformed", "form_reference 체크포인트가 dict 가 아니다: " f"{type(formref_cp).__name__}") status = str(formref_cp.get("status") or "") if status != "completed": raise AppError( code="structure_seed.form_reference_incomplete", message=( "outdoor_structure_form_reference 가 완료 상태가 아니다 " f"(status={status!r}) — 부분 산출을 완료로 취급하지 않는다"), status_code=422, ) data = formref_cp.get("data") _require(isinstance(data, dict), "structure_seed.form_reference_malformed", f"form_reference data 가 dict 가 아니다: {type(data).__name__}") target_meta = data.get("target") _require(isinstance(target_meta, dict), "structure_seed.form_reference_malformed", f"form_reference target 이 dict 가 아니다: " f"{type(target_meta).__name__}") if "mandatory_group_ids" not in data: raise AppError( code="structure_seed.form_reference_stale_shape", message=( "form_reference 체크포인트에 mandatory_group_ids 가 없다 — " "계약 이전에 만들어진 산출이다. 관대하게 받으면 입력이 바뀐 " "구 산출이 현재 지문으로 승격된다. 선행을 다시 돌려라"), status_code=422, ) if "no_spec_group_ids" not in target_meta: raise AppError( code="structure_seed.form_reference_stale_shape", message=( "form_reference 체크포인트에 target.no_spec_group_ids 가 없다 " "— 그 CP 가 본 대상 우주를 복원할 수 없어 parity 를 확인할 수 " "없다(빈 목록과 키 부재는 다르다). 선행을 다시 돌려라"), status_code=422, ) # ── 정책·팩 exact match (Codex 재리뷰 BLOCKING 2) ──────────────── # 키가 있다고 계약이 같은 것은 아니다. 대상 선정 정책이나 팩이 바뀌면 # 같은 그룹 목록이어도 **다른 근거로 뽑힌 집합**이라 재사용할 수 없다. from app.modules.pipeline.search_grounded_ref import ( TARGET_POLICY_VERSION, resolve_ref_pack_version, ) for field, current in (("policy_version", TARGET_POLICY_VERSION), ("pack_version", resolve_ref_pack_version())): got = target_meta.get(field) _require( got == current, "structure_seed.form_reference_contract_drift", f"form_reference {field} 가 현재 계약과 다르다 " f"({got!r} != {current!r}) — 같은 그룹 목록이어도 다른 근거로 " "뽑힌 집합이라 재사용할 수 없다. 선행을 다시 돌려라") mandatory = _str_set(data.get("mandatory_group_ids"), "mandatory_group_ids") no_spec = _str_set(target_meta.get("no_spec_group_ids"), "target.no_spec_group_ids") # ── partition 계약 ─────────────────────────────────────────────── # union 만 보면 mandatory 와 no_spec 이 **겹쳐도** 통과한다(실측: # mandatory=["g"], no_spec=["g"] 가 accepted={'g'}). 한 그룹이 "참조 # 필수"이면서 동시에 "근거 없어 제외"일 수는 없다. overlap = mandatory & no_spec _require(not overlap, "structure_seed.form_reference_malformed", f"필수와 제외에 동시에 든 그룹이 있다: {sorted(overlap)} — " "한 그룹이 참조 필수이면서 근거 부족 제외일 수는 없다") # producer 가 남긴 우주와 소비자가 보는 대상이 같아야 한다. # ★[2026-08-01 Codex 2차 재리뷰 BLOCKING 1] 키가 없으면 분할에서 # **복원해 통과시키던** 경로를 닫는다. 복원은 producer 가 실제로 무엇을 # 봤는지 모른 채 소비자가 지어내는 것이라, 구 CP 를 조용히 승격시킨다. if "universe_group_ids" not in target_meta: raise AppError( code="structure_seed.form_reference_stale_shape", message=( "form_reference 체크포인트에 target.universe_group_ids 가 없다 " "— producer 가 본 모집단을 소비자가 복원해서는 안 된다" "(계약 이전 산출이 현재 지문으로 승격된다). 선행을 다시 " "돌려라"), status_code=422, ) universe = _str_set(target_meta.get("universe_group_ids"), "target.universe_group_ids") _require( mandatory | no_spec == universe, "structure_seed.form_reference_malformed", "producer 의 universe 와 mandatory∪no_spec 이 어긋난다 — " f"universe={sorted(universe)} " f"분할={sorted(mandatory | no_spec)}") if universe != targets: raise AppError( code="structure_seed.form_reference_target_drift", message=( "form_reference 가 본 대상 집합이 지금 seed 대상과 다르다 — " "구 체크포인트를 재사용하면 참조 없는 그룹이 조용히 순수 T2I " f"로 내려간다. 누락={sorted(targets - universe)} " f"초과={sorted(universe - targets)}"), status_code=422, ) # ── 분할을 **현재 SOT 기준으로 다시 계산해** exact 비교 ──────────── # ★[Codex 2차 재리뷰 BLOCKING 1] 여기까지의 검사는 전부 "CP 가 스스로 # 모순인가"만 본다. CP 가 내부적으로 일관되면서도 **현재 spec 과 어긋날 # 수 있다** — 그때 stale no_spec 이 참조 없는 생성을 다시 연다. expected_mandatory, expected_no_spec = partition_form_reference_targets( universe=targets, spec_groups=spec_groups) if mandatory != set(expected_mandatory) or no_spec != set( expected_no_spec): raise AppError( code="structure_seed.form_reference_partition_drift", message=( "form_reference 가 남긴 분할이 현재 outdoor_place_spec 과 " "다르다 — 구 CP 를 재사용하면 지금은 근거가 있는 그룹이 " "'근거 없음'으로 남아 참조 없이 순수 T2I 로 내려간다. " f"지금 근거가 생겼는데 제외로 남음=" f"{sorted(set(expected_mandatory) - mandatory)} " f"근거가 사라졌는데 필수로 남음=" f"{sorted(mandatory - set(expected_mandatory))}. " "선행을 다시 돌려라"), status_code=422, ) # 참조를 실제로 만든 그룹 목록도 필수 집합과 **정확히** 같아야 한다. # 부분집합(<=)만 보면 지난 라운드의 잉여 entry 가 그대로 남아 통과한다. groups = data.get("groups") _require(isinstance(groups, dict), "structure_seed.form_reference_malformed", f"form_reference groups 가 dict 가 아니다: " f"{type(groups).__name__}") _require( set(groups) == mandatory, "structure_seed.form_reference_malformed", "form_reference groups 가 필수 집합과 다르다 — " f"산출 없는 필수={sorted(mandatory - set(groups))} " f"필수 아닌 잉여 산출={sorted(set(groups) - mandatory)}") bad_shape = sorted(g for g, v in groups.items() if not isinstance(v, dict)) _require(not bad_shape, "structure_seed.form_reference_malformed", f"form_reference group entry 가 dict 가 아니다: {bad_shape}") not_ok = sorted(g for g, v in groups.items() if v.get("status") != "ok") _require( not not_ok, "structure_seed.form_reference_missing", f"필수 그룹인데 참조 산출이 ok 가 아니다: {not_ok}") return mandatory def collect_lane2_groups( lane_cp_data: Dict[str, Any], all_groups: bool = False ) -> List[str]: """seed 대상 그룹 id 목록 (결정론 정렬). 기본(False)=structure_plate 바인딩 그룹만(기존 계약 byte-identical). all_groups=True(2026-07-19 재설계 B-1): status ok 인 야외 전 그룹 — 사용자 확정 '비슷한 배경 공유 장소는 seed 기반 배경 하나' 지시의 전제(플레이트 seed 참조·공유 그룹 배경의 룩 앵커). """ out: List[str] = [] for gid in sorted((lane_cp_data or {}).get("groups", {}) or {}): entry = lane_cp_data["groups"][gid] if not isinstance(entry, dict) or entry.get("status") != "ok": continue if all_groups: out.append(gid) continue plan = entry.get("plan") or {} if any( (b or {}).get("lane") == "structure_plate" for b in plan.get("shot_bindings") or [] ): out.append(gid) return out def authority_note(entry: Optional[Dict[str, Any]]) -> Dict[str, Any]: """형태 참조 스텝이 그 그룹을 **어떻게 판정했는지** 한 줄로. ★씨드 기록에 이것이 없으면 「생성이 곧 정본이라 참조를 안 붙였다」와 「결함·결손으로 참조가 없다」가 **구별되지 않는다**. 둘 다 참조 없는 씨드로 똑같이 생겼기 때문이다 (Codex NARROW 2 2026-09-08). 가짜 자산도 가짜 간선도 만들지 않는다 — 판정만 적는다. """ e = entry or {} return { "status": str(e.get("status") or ""), "visual_authority": str(e.get("visual_authority") or ""), "authority_reason_ko": str(e.get("authority_reason_ko") or ""), "contract": _FORMREF_AUTHORITY_CONTRACT, } class OutdoorStructureSeedStep(StepRunner): """레인2 구조물 seed (Stage D) — 순수 T2I 멀티롤 (siteplan 없음).""" def _config_hash(self) -> str: import hashlib import json as _json from app.core.config import settings from app.modules.pipeline.multiroll_gemini import ( resolve_judge_pack_version, ) from app.modules.pipeline.outdoor_structure_seed import ( resolve_seed_pack_version, ) payload = { "schema_version": SCHEMA_VERSION, # 재설계 B-1: 대상 선정 모드(전 그룹 vs structure_plate 만)도 # 산출 집합의 실질 입력 — 전환=재실행 유도 "seed_all_groups": bool(getattr( settings, "outdoor_seed_all_groups_enabled", False)), "seed_pack": resolve_seed_pack_version(SEED_PACK_VERSION), # E2E6 ⑥: 판정·결함 계약 팩도 seed 선정 실질 입력 "judge_pack": resolve_judge_pack_version(), # ★그림 모델. 이름이 예전엔 특정 모델을 가리켰는데 모델을 바꾸면 # 값도 함께 바뀌어야 완료된 CP 가 stale 된다 — 그러지 않으면 # 스텝 자체가 돌지 않아 전환이 실행에 도달하지 못한다. "gen_model": str(getattr(settings, "openai_image_model", "")), "roll_count": int(settings.still_recipe_roll_count), "critique_enabled": bool(settings.still_recipe_critique_enabled), "outdoor_lane_pipe_enabled": True, # ★[2026-08-01 Codex 재리뷰 BLOCKING 1] 형태 참조 소비 계약을 # hash 축에 싣는다. 이것이 없으면 관할절·계보 배선을 바꿔도 # 완료된 seed CP 가 그대로 재사용된다 — 실증: clause 를 전혀 다른 # 문자열로 바꿔도 config_hash 가 c9bce3d8… 로 동일했다. 즉 수정이 # **실행에 도달하지 않는다.** 계약 버전·팩·대상 정책·절 본문 # 해시를 모두 실어, 이 wave 가 기존 완료 CP 를 의도적으로 # stale 시키게 한다. **_form_reference_consumer_hash_payload(), # ★[2026-08-01] 재생성 이슈 정책도 수정 단계의 실질 입력이다 — # 같은 critique 결과에서도 편집 지시와 최종 산출이 달라진다. # 싣지 않으면 완료된 CP 가 옛 편집본(고칠 수 없는 결함을 편집에 # 실어 만든 것)을 그대로 재사용해 새 계약이 실행에 도달하지 못한다. "seed_regen_issue_policy": SEED_REGEN_ISSUE_POLICY, "seed_regen_issue_policy_version": ( _regen_issue_policy_version()), } # seed 품질 2R: 변형 저작 계약(팩·저작 모델)도 seed 실질 입력 — # ON 시만 스탬프 (OFF=기존 프로젝트 hash 불변). if getattr(settings, "structure_seed_variants_enabled", False): from app.modules.pipeline.seed_prompt_variants import ( AUTHOR_MODEL, SEED_VARIANT_COUNT, resolve_variants_pack_version, ) payload["structure_seed_variants_enabled"] = True payload["seed_variants_pack"] = resolve_variants_pack_version( SEED_VARIANTS_PACK_VERSION) payload["seed_variants_author_model"] = AUTHOR_MODEL # Codex 재리뷰 HIGH-3: alias 뒤 물리 모델 교체도 저작 실질 입력 payload["seed_variants_author_physical_model"] = str( getattr(settings, "openai_model", "")) # Codex 재리뷰 HIGH-2: 변형 수=사용자 확정 3종 (roll_count 독립) payload["seed_variant_count"] = int(SEED_VARIANT_COUNT) # Codex 3차 BLOCKING-1(3): 판정 VLM 물리 모델도 실질 입력 # ('gemini-pro' alias → gemini_text_model, llm_client 매핑) payload["seed_variants_judge_physical_model"] = str( getattr(settings, "gemini_text_model", "")) # ★롤 수의 뜻이 바뀌었다(2026-08-03). 예전에는 저작 변형 수가 # 곧 롤 수였는데, 이제 롤은 **경로**다 — 직접 / 스케치 경유 둘. # 저작 변형은 그대로 여러 벌 받되 두 경로가 같은 한 벌을 쓴다 # (경로 차이만 남겨야 비교가 성립한다). payload["roll_count"] = len(SEED_PATH_LABELS) payload["seed_path_policy"] = SEED_PATH_POLICY_VERSION return hashlib.sha256( _json.dumps(payload, sort_keys=True).encode("utf-8") ).hexdigest()[:16] def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: from app.core.config import settings cp = ( Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json" ) if cp.exists(): try: return json.loads(cp.read_text(encoding="utf-8")) except Exception as exc: # noqa: BLE001 logger.warning( "outdoor_structure_seed: %s 로드 실패: %s", step_id, exc) return None def _form_ref_asset_matches(self, *, gid: str, asset_id: str, path_s: str) -> bool: """그 asset_id 가 **지금 이 에피소드의 그 그룹 참조 자산**인가. id 만 맞으면 되는 것이 아니라 project/episode/asset_type/entity_id/ variant_type 이 전부 맞고 가리키는 파일까지 CP 참조와 같아야 한다. 어느 하나라도 어긋나면 계보가 거짓을 가리키는 것이다. """ from app.core.file_paths import resolve_image_path from app.models.project import ImageAsset row = ( self.db.query(ImageAsset) .filter_by( id=asset_id, project_id=self.project_id, episode_id=self.episode_id, asset_type="structure_form_ref", entity_id=gid, variant_type="form_ref", ) .first() ) if row is None: logger.warning( "outdoor_structure_seed: %s 형태 참조 자산 %s 을 찾을 수 없다 " "— 계보가 실재하지 않는 자산을 가리킨다", gid, asset_id) return False # ★[2026-08-01] 양쪽을 **같은 표현**으로 맞춰 비교한다. 처음에는 CP # 경로를 `to_relative_image_path` 로 상대화해 `row.file_path` 와 비교 # 했는데, `ImageAsset.file_path` 는 `ImagePathType` 이라 **읽을 때 # 상대→절대로 복원**된다(file_paths.py:90). 그래서 프로젝트 루트 안에 # 있는 실제 자산은 상대 vs 절대로 영영 어긋나 **정상 참조가 전부 # form_ref_missing 으로 막혔다.** 유닛이 이를 못 잡은 이유가 더 중요 # 하다 — 픽스처가 루트 밖(/tmp)이라 상대화가 입력을 그대로 돌려줬고, # fake row 의 경로도 같은 함수로 만들어 **구현이 아니라 mock 을 검증** # 하고 있었다. want = resolve_image_path(path_s) got = resolve_image_path(getattr(row, "file_path", "") or "") if want is None or got is None or str(got) != str(want): logger.warning( "outdoor_structure_seed: %s 형태 참조 자산 경로 불일치 " "(자산=%r 참조=%r)", gid, getattr(row, "file_path", None), str(want) if want else None) return False return True def _resolve_form_ref(self, *, gid: str, entry: Dict[str, Any], mandatory: bool): """검색 선택 실사를 형태 참조로 결정한다 (소비 직전 재검증). 필수 대상(structure_plate 바인딩)인데 참조가 없거나 파일·sha 가 어긋나면 **AppError 로 그 그룹을 세운다** — 순수 T2I 로 조용히 내려가면 사용자 지시("모두 검색 기반")를 아무도 모르게 어긴다. 비필수 그룹은 참조 없이 기존 T2I 경로를 그대로 탄다. """ import hashlib from app.core.errors import AppError # ── [v12] 생성이 정본인 그룹 (2026-09-05) ────────────────────── # # ★★★「사진을 못 찾았다」와 「맞출 실사 정본이 애초에 없다」는 **다른 # 사건**이다. 앞 스텝이 대상별로 판정해 `generative` 로 확정한 그룹은 # 이 이야기가 지어낸 것이라 어떤 사진도 그것을 가리키지 못한다. # 여기서 세우면 창작 세계관 대본이 매번 막힌다(실측 2026-09-05: # 2419년 폐광 → 후보 8장 · VLM 선택 0 · 파이프라인 정지). # # ★그렇다고 일관성을 포기하는 것이 아니다 — 그 정본은 **이 스텝이 # 생성으로 만들어** 잠근다. 참조 입력만 없을 뿐 seed 자산은 그대로 # 만들어지고, 이후 모든 씬이 그 한 장을 본다. # # ★가짜 계보를 만들지 않는다 (Codex): 없는 사진의 asset_id 를 지어내 # `input_image_ids` 에 넣으면 존재하지 않는 자산을 가리키는 간선이 # 된다. 입력은 **비운 채로** 둔다. # # ★그 대신 **판정을 CP 에 적는다** (Codex NARROW 2 2026-09-08). 앞 # 판은 여기 주석이 `_authority_note` 가 남긴다고 적었는데 그런 함수도 # 호출도 없었다 — 즉 seed 쪽에는 아무 기록이 없어서, 「생성이 정본이라 # 참조를 안 붙였다」와 「결함으로 참조가 없다」가 구별되지 않았다. # 아래 `authority_note` 가 그 판정·사유·계약을 돌려주고, 호출부가 # seed entry 의 `form_reference_authority` 에 싣는다. if (entry or {}).get("status") == _FORMREF_GENERATIVE_STATUS: logger.info( "outdoor_structure_seed: %s 시각 권위 = 생성 — 실사 참조 없이 " "생성으로 정본을 만든다 (%s)", gid, str((entry or {}).get("authority_reason_ko") or "")) return [], None path_s = (entry or {}).get("form_ref_path") or "" sha = (entry or {}).get("form_ref_sha256") or "" ok = (entry or {}).get("status") == "ok" and path_s and sha if ok: p = Path(path_s) if not p.is_file(): ok = False else: actual = hashlib.sha256(p.read_bytes()).hexdigest() if actual != sha: logger.warning( "outdoor_structure_seed: %s 형태 참조 sha 불일치", gid) ok = False # ★[2026-08-01 Codex 재리뷰 BLOCKING 3] 계보를 optional 로 두지 # 않는다. 이전에는 status/path/sha 만 맞으면 asset_id 가 None 이어도 # ok 였고, 소비측 list comprehension 이 None 을 버려 input_image_ids # 가 다시 빈 목록이 됐다 — 고쳤다던 결함이 그대로 재발한다. asset_id = str((entry or {}).get("form_ref_asset_id") or "").strip() if ok and not asset_id: logger.warning( "outdoor_structure_seed: %s 형태 참조 asset_id 결손", gid) ok = False # ★[2026-08-01 Codex 2차 재리뷰 BLOCKING 2] 결손 검사가 "빈 문자열이 # 아닌가"까지였다. 존재하지 않는 UUID 를 넣어도 정상 지문으로 통과했고 # (Codex 직접 재현), `annotate_generated_asset` 은 받은 문자열을 JSON # 으로 적을 뿐 자산을 조회하지 않는다 — **가짜 간선**이 된다. # 키 존재 ≠ 계약 일치라는 같은 함정이 계보 층에서 반복됐다. if ok and not self._form_ref_asset_matches( gid=gid, asset_id=asset_id, path_s=path_s): ok = False if not ok: if mandatory: raise AppError( code="structure_seed.form_ref_missing", message=( f"{gid}: 검색 선택 형태 참조 결손/불일치 — 필수 대상은 " "순수 T2I 로 내려가지 않고, 참조 자산 id 가 없으면 " "계보를 남길 수 없다 (fail-closed). " "outdoor_structure_form_reference 를 먼저 통과시켜라"), status_code=422, ) return [], None # ── 부속 참조 (2026-08-03) ────────────────────────────────── # 앞 스텝이 조사에서 채택한 대상마다 따로 검색해 고른 사진이다. # ★주 참조와 달리 fail-closed 하지 않는다 — 부속은 있으면 더 정확해 # 지는 입력이지 그룹의 성립 조건이 아니고, 하나가 사라졌다고 그룹을 # 세우면 조사가 잘 된 그룹일수록 잘 죽는다. 대신 **조용히 빼지 # 않는다**: 사유를 로그에 남기고 소비 기록에도 남긴다. labeled: List[Any] = [("FORM REFERENCE", Path(path_s))] fittings: List[Dict[str, Any]] = [] dropped: List[str] = [] for fit in ((entry or {}).get("fitting_refs") or []): if not isinstance(fit, dict): continue fp_s = str(fit.get("path") or "") fsha = str(fit.get("sha256") or "") target = str(fit.get("target_native") or "") why = "" if not fp_s or not fsha: why = "경로/sha 결손" elif not Path(fp_s).is_file(): why = "파일 없음" elif hashlib.sha256( Path(fp_s).read_bytes()).hexdigest() != fsha: why = "bytes sha 불일치" if why: dropped.append(f"{target or fp_s}: {why}") logger.warning( "outdoor_structure_seed: %s 부속 참조 제외 (%s) — %s", gid, why, fp_s) continue labeled.append((f"FITTING REFERENCE {len(fittings) + 1}", Path(fp_s))) fittings.append({"target_native": target, "path": fp_s, "sha256": fsha}) return ( labeled, {"sha256": sha, "asset_id": asset_id, "path": path_s, # ★부속도 지문에 실린다 — 붙는 참조가 바뀌면 그림이 바뀐다. "fittings": fittings, "fittings_dropped": dropped}, ) # ── seed 품질 2R: 변형 저작 ────────────────────────────────── def _typology_facts(self, *, gid: str, formref_entry: Dict[str, Any]) -> Dict[str, Any]: """유형 사전지식 — **앞 스텝(form_reference)이 조사한 것을 읽는다.** 2026-08-03 이전에는 이 스텝이 직접 조사했다. 그런데 그 자리는 이미지 검색이 다 끝난 **뒤**라, 조사가 답을 내도 그 답이 질의로 이어질 방법이 없었다 — 조사와 검색이 서로 모른 채 따로 돌았다. 조사를 앞 스텝으로 옮기면서 이 함수는 읽기만 한다. 유료 호출이 없으므로 캐시도 필요 없다(앞 스텝의 CP·journal 이 재사용을 이미 소유한다). 반환 = {"block", "items", "questions"}. `block` 이 저작 입력에 실리는 최하위 권위 블록이고, 나머지는 감사·갤러리용 기록이다. 조사 결손은 그룹을 죽이지 않는다 — 빈 블록으로 내려가면 이 단계가 없던 때의 동작이라 손해가 없다. """ prior = (formref_entry or {}).get("typology") if not isinstance(prior, dict): logger.info("typology[%s]: 앞 스텝 조사 기록 없음 — 빈 블록", gid) return {"block": "", "items": [], "questions": []} block = str(prior.get("block") or "") logger.info("typology[%s]: 앞 스텝 조사 소비 — 대상 %d · 블록 %d자", gid, len(prior.get("items") or []), len(block)) return {"block": block, "items": list(prior.get("items") or []), "questions": list(prior.get("questions") or [])} def _make_sketch( self, *, gid: str, prompt: str, ref_path: Path, out_dir: Path, gen_fn: Any, force: bool, ) -> Optional[Path]: """형태만 남긴 선 도해 1장 — 스케치 경로의 입력. ★실패해도 그룹을 죽이지 않는다. 도해가 없으면 직접 경로만 남는데 그것이 이 단계가 없던 어제까지의 동작이라 손해가 없다. 다만 조용히 넘기지 않고 사유를 로그에 남기고, 어느 경로가 빠졌는지는 CP 에 남는다. ★파일명에 프롬프트 지문을 넣는다. 같은 이름을 덮어쓰면 브리프나 변환절이 바뀐 뒤에도 옛 도해가 남아 롤만 새로 그리게 되고, 형태 권위가 낡은 채로 전이된다. 옛 파일은 지우지 않는다 — 이력이다. ★★롤과 **다른 폴더**에 둔다. 롤 산출 정리는 `seed_<그룹>_*.png` 를 지우는데, 도해를 그 이름 아래 두면 롤을 다시 그릴 때 도해가 함께 지워진다. 실측에서 그렇게 사라진 뒤 사진 한 장만으로 그린 그림이 "스케치 경로" 산출로 판정에 올라갔다. """ import hashlib stamp = hashlib.sha256(prompt.encode("utf-8")).hexdigest()[:8] sketch_dir = out_dir / "structure_sketch" sketch_dir.mkdir(parents=True, exist_ok=True) out_path = sketch_dir / f"{gid}_{stamp}.png" if out_path.exists() and not force: return out_path try: return Path(gen_fn(f"sketch_{gid}", prompt, [("FORM REFERENCE", ref_path)], out_path)) except Exception as exc: # noqa: BLE001 — 보조 경로, 격리한다 logger.warning( "seed sketch[%s]: 실패 — 직접 경로만 진행 (%s)", gid, exc) return None def _author_seed_variants( self, *, seed_inputs: Dict[str, Any], scene_indices: List[int], scene_texts: Dict[int, str], scene_headings: Dict[int, str], world_anchor: str, count: int, cached_entry: Optional[Dict[str, Any]] = None, interior_scene_indices: Optional[List[int]] = None, typology_facts_block: str = "", ) -> tuple[Dict[str, Any], str, bool]: """Sol 저작 호출 → (저작본, author_fp, 캐시 재사용 여부). - 구조 검증 실패=1회 재시도 — 재시도 입력에 이전 위반 목록을 교정 블록으로 병기(Codex 재리뷰 NARROW-4: 동일 요청 맹복 반복 차단), 소진=예외 → 호출측 그룹 try 가 failed 격리 (fail-closed). - cached_entry(sidecar): author_fp(입력 전체 지문) 일치 + 재검증 통과 시 LLM 재호출 없이 재사용 — 크래시 후 재실행에서 비결정 저작이 지문을 바꿔 롤 전체가 무효화되는 창 제거 (NARROW-5). - interior_scene_indices (v9): 같은 building 실내 멤버 씬 — room_inventory 증거 배선. evidence-bound 검증(scene_index+ whitespace-normalized 인용 provenance)은 validator 소유 — 재시도 루프와 cache 재검증 경로 모두 커버 (Codex v8b ②③). """ import hashlib import json as _json from app.core.config import settings from app.modules.llm.llm_client import call_structured from app.modules.pipeline.seed_prompt_variants import ( AUTHOR_MODEL, build_author_schema, build_author_user_content, load_author_sys, validate_variant_output, ) interior_extra = list(interior_scene_indices or []) base_inputs = dict( scene_indices=scene_indices, scene_texts=scene_texts, scene_headings=scene_headings, structure_desc=seed_inputs["structure_desc"], interior_note_en=seed_inputs["interior_note_en"], exterior_note_en=seed_inputs["exterior_note_en"], world_anchor=world_anchor, excluded_transient_en=seed_inputs["excluded_transient_en"], count=count, interior_scene_indices=interior_extra, typology_facts_block=typology_facts_block, ) user_content = build_author_user_content(**base_inputs) author_sys = load_author_sys(count, SEED_VARIANTS_PACK_VERSION) from app.modules.pipeline.seed_prompt_variants import ( _INVENTORY_PACKS, ) inventory_on = SEED_VARIANTS_PACK_VERSION in _INVENTORY_PACKS schema = build_author_schema(count, include_inventory=inventory_on) # evidence-bound 공급 씬 집합 — validator 의 인용 provenance # 대조 대상 (공급 밖 scene_index 인용=위반, Codex v8b ③) supplied_texts = { si: scene_texts.get(si) or "" for si in {*scene_indices, *interior_extra} } if inventory_on else None physical_model = str(getattr(settings, "openai_model", "")) h = hashlib.sha256() for part in ( author_sys, user_content, _json.dumps(schema, sort_keys=True), physical_model, ): h.update(part.encode("utf-8")) h.update(b"\x00") author_fp = h.hexdigest()[:16] if ( cached_entry and cached_entry.get("author_fp") == author_fp and not validate_variant_output( cached_entry.get("author"), count, require_inventory=inventory_on, scene_texts=supplied_texts) ): return dict(cached_entry["author"]), author_fp, True pc = { **(self.project_config or {}), "seed_variant_author": {"model": AUTHOR_MODEL}, } violations: List[str] = [] for attempt in range(2): content = ( user_content if not violations else build_author_user_content( **base_inputs, prior_violations=violations) ) data = call_structured( "seed_variant_author", author_sys, content, schema, project_config=pc, schema_name="seed_variant_author", ) violations = validate_variant_output( data, count, require_inventory=inventory_on, scene_texts=supplied_texts) if not violations: return data, author_fp, False logger.warning( "seed_variant_author: 저작 검증 위반(attempt %d): %s", attempt + 1, violations[:6], ) raise ValueError( f"seed 변형 저작 검증 실패(재시도 소진): {violations[:6]}") # ── ImageAsset UPSERT (place_canon 관례) ───────────────────── def _upsert_seed_asset( self, *, group_id: str, variant_type: str, # "seed" (구 "siteplan" row 는 보존만) pipeline_role: str, abs_png_path: str, prompt_used: str, generation_model: str, input_image_ids: List[str], ) -> str: from app.core.file_paths import to_relative_image_path from app.models.project import ImageAsset from app.services.image_capture.annotate import ( annotate_generated_asset, ) rel_path = to_relative_image_path(abs_png_path) existing = ( self.db.query(ImageAsset) .filter_by( project_id=self.project_id, episode_id=self.episode_id, asset_type="structure_seed", entity_id=group_id, variant_type=variant_type, ) .first() ) if existing: existing.file_path = rel_path existing.prompt_used = prompt_used existing.status = "generated" existing.generation_model = generation_model row = existing else: row = ImageAsset( id=str(uuid.uuid4()), project_id=self.project_id, episode_id=self.episode_id, asset_type="structure_seed", entity_id=group_id, variant_type=variant_type, file_path=rel_path, prompt_used=prompt_used, generation_model=generation_model, status="generated", created_at=datetime.now(timezone.utc), ) self.db.add(row) annotate_generated_asset( row, pipeline_role=pipeline_role, input_image_ids=input_image_ids, ) return str(row.id) # ── 실행 ───────────────────────────────────────────────────── def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings from app.core.errors import AppError def _empty() -> Dict[str, Any]: return { "applicable_count": 0, "completed_count": 0, "failed_count": 0, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"groups": {}}, } if not ( getattr(settings, "outdoor_lane_pipe_enabled", False) and getattr(settings, "outdoor_lane_plan_enabled", False) ): return _empty() lane_cp = self._load_prev_checkpoint("outdoor_lane_plan") lane_data = (lane_cp or {}).get("data", {}) or {} target_gids = collect_lane2_groups( lane_data, all_groups=bool(getattr( settings, "outdoor_seed_all_groups_enabled", False)), ) if not target_gids: logger.info( "outdoor_structure_seed: structure_plate 그룹 없음 — no-op") return _empty() spec_cp = self._load_prev_checkpoint("outdoor_place_spec") spec_groups = (spec_cp or {}).get("data", {}).get("groups", {}) or {} scene_cp = self._load_prev_checkpoint("scene_save") scene_texts: Dict[int, str] = {} scene_headings: Dict[int, str] = {} for seg in (scene_cp or {}).get("data", {}).get("segments", []) or []: si = seg.get("scene_index") if isinstance(si, int): scene_texts[si] = seg.get("text") or "" scene_headings[si] = seg.get("heading") or "" bg_cp = self._load_prev_checkpoint("background_classify") building_groups = { g.get("group_id"): g for g in (bg_cp or {}).get("data", {}).get( "building_groups", []) or [] if isinstance(g, dict) } em_cp = self._load_prev_checkpoint("entity_merge") locations_by_id = { loc.get("short_id"): loc for loc in (em_cp or {}).get("data", {}).get( "locations", []) or [] if isinstance(loc, dict) } classify_cp = self._load_prev_checkpoint("shot_ref_classify") wa = ((classify_cp or {}).get("data", {}) or {}).get( "world_anchor_en") or "" world_anchor = f" — {wa.strip()}" if wa.strip() else "" # [2026-07-28] 검색 그라운딩 형태 참조 — 선행 스텝 산출 소비. # 사용자 지시: 야외 복잡 구조물은 모두 검색 기반 + VLM 선택 실사를 # 형태 참조로 물린다. **필수 대상(structure_plate 바인딩)의 참조가 # 없으면 그 그룹은 fail-closed** — 순수 T2I 자동 degrade 금지. formref_cp = self._load_prev_checkpoint( "outdoor_structure_form_reference") # 계약 검증은 한 곳이 소유한다 — 폴백·구 shape 승격 차단 포함. # ★현재 spec 을 함께 넘긴다: CP 의 분할을 그대로 믿으면 stale no_spec # 이 참조 없는 생성을 다시 연다(Codex 2차 재리뷰 BLOCKING 1). # # ★검증이 **먼저**다 (Codex 재리뷰 NARROW-2). 이전에는 `.get` 연쇄로 # groups 를 먼저 꺼냈는데, malformed CP(비-dict root/data)에서는 그 # 자리에서 AttributeError 가 나 helper 의 계약 검증에 **도달조차 못 # 했다.** 계약 위반이 "스텝 고장"으로 둔갑해 422 대신 500 이 뜬다. mandatory_gids = resolve_form_reference_contract( formref_cp=formref_cp, target_gids=target_gids, spec_groups=spec_groups) # 위 검증이 root/data/groups 의 모양을 보장한다 — 여기서 다시 방어하면 # resolver 회귀가 조용히 묻힌다. formref_groups = formref_cp["data"]["groups"] from app.modules.pipeline.gpt_image_gen import make_gpt_image_gen_fn from app.modules.pipeline.multiroll_gemini import ( make_gemini_critique_fn, make_gemini_judge_fn, make_ref_pick_judge_fn, resolve_judge_texts, ) from app.modules.pipeline.multiroll_select import ( build_critique_schema, build_judge_schema, roll_labels, run_multiroll_select, ) from app.modules.pipeline.outdoor_lane_plan import ( build_group_shot_reconstructor, revalidate_persisted_plan, ) from app.modules.pipeline.outdoor_structure_seed import ( build_structure_seed_prompt, derive_seed_inputs, ) from app.modules.prompt_sanitizer import PromptSanitizer from app.services.image_capture.context import generation_context out_dir = ( Path(settings.projects_dir) / self.project_id / "images" / "background_chain" ) out_dir.mkdir(parents=True, exist_ok=True) roll_count = int(settings.still_recipe_roll_count) judge_texts = resolve_judge_texts(roll_count) sanitizer = PromptSanitizer(project_config=self.project_config) # ★그림 모델 = gpt-image-2 (2026-08-03 사용자 결정, settings 주석에 # 맞대결 근거). 모델명은 **지문에도 실린다**(아래 gen_model) — # 그러지 않으면 기존 산출이 그대로 재사용돼 전환이 no-op 이 된다. image_model = str(getattr(settings, "openai_image_model", "")) gen_fn = make_gpt_image_gen_fn( project_id=self.project_id, episode_id=self.episode_id, operation_type="outdoor_structure_seed", sanitizer=sanitizer, ) # seed 품질 2R: 변형 저작 모드 — 판정·결함 계약을 seed 전용 팩 # v2 로 전면 교체 (Codex 재리뷰 BLOCKING-1: judge_still/critique # 는 film-still 계약이라 '동일 프롬프트 생성' 전제·샷 프레이밍· # 인물 축이 변형 seed 판정과 충돌). 변형 수=3 고정 (HIGH-2 — # still_recipe_roll_count 와 독립). variants_on = bool( getattr(settings, "structure_seed_variants_enabled", False)) critique_sys = judge_texts["critique_sys"] judge_brief_sys = judge_texts["judge_sys"] judge_header_kwargs: Dict[str, Any] = {} ref_pick_sys = "" # ★형태 참조가 붙지 않은 그룹의 롤 수. 참조가 있는 그룹은 경로 둘로 # 가고, 없는 그룹은 예전대로 변형 롤로 간다 — 참조가 없으면 두 # 경로가 성립하지 않는다(도해를 그릴 사진도, 비교할 대상도 없다). seed_roll_count = roll_count # ★그림 모델을 **변형 여부와 무관하게** 지문에 싣는다. 예전에는 # 변형 ON 경로에만 실려서, OFF 경로는 모델을 바꿔도 지문이 그대로라 # 기존 PNG 를 재사용했다 — 바꾼 적 없는 그림을 바뀐 것으로 읽게 된다. seed_extra_fp: Optional[Dict[str, Any]] = {"gen_model": image_model} force_mode = (mode == "force") if variants_on: import hashlib as _hashlib from app.modules.pipeline.multiroll_gemini import ( resolve_judge_pack_version, ) from app.modules.pipeline.search_grounded_ref import ( load_ref_pick_system, ) from app.modules.pipeline.seed_prompt_variants import ( SEED_VARIANT_COUNT, load_variant_judge_texts, resolve_variants_pack_version, ) seed_roll_count = SEED_VARIANT_COUNT variant_texts = load_variant_judge_texts( SEED_VARIANT_COUNT, SEED_VARIANTS_PACK_VERSION) judge_brief_sys = variant_texts["judge_sys"] critique_sys = variant_texts["critique_sys"] judge_header_kwargs["prompt_header"] = ( "THE SHARED CONFORMANCE BRIEF (each candidate was " "generated from a different authored phrasing of this " "brief; judge against the brief):" ) # ★참조가 붙은 그룹의 판정 계약 — 브리프 대비 절대 점수가 아니라 # **참조 사진 대비 택일**이다. 두 계약이 공존한다. ref_pick_sys = load_ref_pick_system( label_list=", ".join(SEED_PATH_LABELS)) # Codex 3차 BLOCKING-1(2): sidecar record 영속로 inner 지문이 # 재실행 경계를 넘게 됨 — 생성/판정 모델·판정 계약 전문 해시를 # 지문에 포함해 config 변경이 sidecar 재사용을 뚫지 못하게. _th = _hashlib.sha256() for part in ( judge_brief_sys, ref_pick_sys, critique_sys, judge_texts["fix_head"], judge_texts["fix_tail"], judge_texts["fix_label"], judge_header_kwargs["prompt_header"], ): _th.update(part.encode("utf-8")) _th.update(b"\x00") seed_extra_fp = { "gen_model": image_model, "judge_physical_model": str( getattr(settings, "gemini_text_model", "")), "seed_variants_pack": resolve_variants_pack_version( SEED_VARIANTS_PACK_VERSION), "judge_fix_pack": resolve_judge_pack_version(), "contract_texts_sha": _th.hexdigest()[:16], "seed_path_policy": SEED_PATH_POLICY_VERSION, } # 판정은 두 벌이다 — 그룹이 어느 쪽을 쓰는지는 형태 참조 유무가 정한다. judge_brief_fn = make_gemini_judge_fn( judge_sys=judge_brief_sys, judge_schema=build_judge_schema(roll_labels(seed_roll_count)), project_config=self.project_config, step_tag="structure_seed_judge", **judge_header_kwargs, ) # 참조 사진 앞에서 경로를 고른다 — 브리프는 판정에 주지 않는다. # 브리프 대비 절대 점수는 참조에 다 보이는 것을 빠뜨린 산출도 # 통과시켰고, 그것이 육안 반려로 돌아왔다. judge_refpick_fn = ( make_ref_pick_judge_fn( project_config=self.project_config, step_tag="structure_seed_ref_pick") if variants_on else None ) critique_fn = make_gemini_critique_fn( critique_sys=critique_sys, critique_schema=build_critique_schema(), project_config=self.project_config, step_tag="structure_seed_critique", ) # Codex 재리뷰 NARROW-5: 저작본+판정 record 를 gid별 sidecar 로 # durable persist — 크래시 후 재실행이 비결정 저작으로 지문을 # 바꿔 기존 롤을 전부 무효화하는 창 제거 + 판정 이력 감사. records_path = out_dir / "seed_variant_records.json" seed_sidecar: Dict[str, Any] = {} if variants_on: from app.modules.pipeline.shot_conti_light import ( load_sidecar_records, save_sidecar_records, ) seed_sidecar = load_sidecar_records(records_path) # Codex Stage D BLOCKING-2: 재검증 커버리지=현재 authoritative # 선택 샷 집합 (Stage A 와 동일 재구성 — plan 자기 바인딩 금지) director_cp = self._load_prev_checkpoint("scene_director") reconstruct = build_group_shot_reconstructor( staging_cp=self._load_prev_checkpoint("shot_staging"), validator_cp=self._load_prev_checkpoint("shot_validator"), selection_cp=self._load_prev_checkpoint("shot_selection"), director_cp=director_cp, ) # v9 실내 씬 배선(Codex v8b ③): scene→loc 매핑 SOT=scene_director # 확정 primary_location (visible_entities 와 동일한 코드 자동 # 구축 관례 — LLM 재판정 없음) scene_primary_loc: Dict[int, Any] = { s["scene_index"]: s.get("primary_location") for s in (director_cp or {}).get("data", {}).get( "scenes", []) or [] if isinstance(s, dict) and isinstance(s.get("scene_index"), int) } results: Dict[str, Any] = {} completed = 0 failed = 0 for gid in target_gids: lane_entry = lane_data["groups"][gid] plan = lane_entry.get("plan") or {} spec_entry = spec_groups.get(gid) or {} spec = spec_entry.get("spec") if not spec: # spec 부재 = 재구성 자체가 불가 — 그룹 실패 격리 (재검증 # 이전에 판정해 오탐 coverage 위반과 구분) results[gid] = { "status": "failed", "error": "outdoor_place_spec 부재 — seed 입력 SOT 결손", } failed += 1 continue # Codex 배선 조건(BLOCKING-2): persisted plan 소비 직전 재검증 # — 커버리지=현재 authoritative 선택 샷 집합 (fail-closed) violations = revalidate_persisted_plan( plan, scene_texts, group_shots=reconstruct(spec_entry), ) if violations: raise AppError( code="step.contract_violation.outdoor_structure_seed", message=( f"persisted lane plan 재검증 실패 (group={gid}): " + "; ".join(violations[:8]) ), status_code=422, ) try: seed_inputs = derive_seed_inputs( spec=spec, building_group=building_groups.get(gid) or {}, locations_by_id=locations_by_id, ) except ValueError as exc: # NARROW-6: persisted spec 의 transient 오염 = 그룹 실패 # 격리 (영구 seed 고착 차단) results[gid] = {"status": "failed", "error": str(exc)} failed += 1 continue scene_indices = sorted( int(si) for si in lane_entry.get("scene_indices") or [] if isinstance(si, (int, float, str)) and str(si).lstrip("-").isdigit() ) place_text = next( (scene_headings[si] for si in scene_indices if scene_headings.get(si)), "") try: author_data: Optional[Dict[str, Any]] = None roll_prompts: Optional[Dict[str, str]] = None if variants_on: # seed 품질 2R: 프롬프트 저작=LLM 위임 (규모·개구부· # 장소 서술 포함). raw 씬 헤딩(place_text)은 미주입 — # THE LOCATION 은 저작본(location_line_en)만 (진단 ③). from app.modules.pipeline.seed_prompt_variants import ( _INVENTORY_PACKS, assemble_variant_roll_prompts, build_conformance_prompt, ) from app.modules.pipeline.outdoor_structure_seed import ( collect_interior_scene_indices, ) # v9 (Codex v8b ③ 배선 결손): 같은 building 실내 # 멤버 loc 의 씬 원문을 저작 입력에 합류 — 야외 씬만 # 으로는 room_inventory 가 방 전수를 열거 못한다. # 구팩(v8 이하)=기존 입력 byte-identical. interior_sis: List[int] = [] if SEED_VARIANTS_PACK_VERSION in _INVENTORY_PACKS: interior_sis = collect_interior_scene_indices( building_group=( building_groups.get(gid) or {}), scene_primary_loc=scene_primary_loc, exclude=scene_indices, ) # 유형 사전지식 — 앞 스텝(form_reference)이 이미지 검색 # **앞에서** 조사한 것을 읽는다. 유료 호출 없음. typology = self._typology_facts( gid=gid, formref_entry=formref_groups.get(gid) or {}) author_data, author_fp, author_reused = ( self._author_seed_variants( seed_inputs=seed_inputs, scene_indices=scene_indices, scene_texts=scene_texts, scene_headings=scene_headings, world_anchor=world_anchor, count=seed_roll_count, interior_scene_indices=interior_sis, typology_facts_block=typology.get("block") or "", # Codex 3차 BLOCKING-1(1): force=명시 재생성 # 계약 — 저작 캐시도 우회 대상 cached_entry=( None if force_mode else seed_sidecar.get(gid)), ) ) if not author_reused: seed_sidecar[gid] = { "author_fp": author_fp, "author": author_data, "typology": typology, } save_sidecar_records(records_path, seed_sidecar) roll_prompts = assemble_variant_roll_prompts( data=author_data, labels=roll_labels(seed_roll_count), world_anchor=world_anchor, excluded_transient_en=( seed_inputs["excluded_transient_en"]), seed_pack_version=SEED_PACK_VERSION, variants_pack_version=SEED_VARIANTS_PACK_VERSION, # E2E11 ①: 규모 계약(코드 불변 안전절) — 저작이 # interior 증거를 떨어뜨려도 최종 프롬프트에 남는다 interior_note_en=seed_inputs["interior_note_en"], ) seed_brief = build_conformance_prompt( data=author_data, excluded_transient_en=( seed_inputs["excluded_transient_en"]), variants_pack_version=SEED_VARIANTS_PACK_VERSION, interior_note_en=seed_inputs["interior_note_en"], ) with generation_context( self.project_id, self.episode_id, stage="outdoor_structure_seed", ): if not variants_on: seed_prompt = build_structure_seed_prompt( structure_desc=seed_inputs["structure_desc"], interior_note_en=seed_inputs["interior_note_en"], exterior_note_en=seed_inputs["exterior_note_en"], world_anchor=world_anchor, place_text=place_text, excluded_transient_en=( seed_inputs["excluded_transient_en"]), prompt_version=SEED_PACK_VERSION, ) record: Dict[str, Any] = {} persist_record_fn = None if variants_on: record = dict( (seed_sidecar.get(gid) or {}).get("record") or {} ) def persist_record_fn( r: Dict[str, Any], _gid: str = gid, ) -> None: seed_sidecar.setdefault(_gid, {})["record"] = r save_sidecar_records( records_path, seed_sidecar) # [2026-07-28] 형태 참조 결정 — 필수 대상은 소비 직전 # 재검증(파일 존재+sha)까지 하고 없으면 fail-closed. form_ref_labeled, form_ref_fp = self._resolve_form_ref( gid=gid, entry=formref_groups.get(gid) or {}, mandatory=gid in mandatory_gids) _gen_prompt = seed_brief if variants_on else seed_prompt # ★지문은 **그룹마다 새로** 만든다. 예전에는 루프 변수를 # 덮어써서, 참조가 없는 그룹에 앞 그룹의 참조 지문이 그대로 # 남았다 — 앞 그룹 참조가 바뀌면 무관한 그룹이 무효화되고, # 그룹 순서만 바꿔도 지문이 달라져 재현이 안 된다. group_extra_fp: Dict[str, Any] = dict(seed_extra_fp or {}) roll_refs: Optional[Dict[str, Any]] = None sketch_path: Optional[Path] = None sketch_prompt = "" group_roll_count = seed_roll_count judge_fn = judge_brief_fn if form_ref_labeled: from app.modules.pipeline.search_grounded_ref import ( build_fitting_ref_clause, build_form_only_clause, build_form_only_clause_sketched, ) # ★부속 참조가 붙으면 관할을 나눠 말해야 한다. 여러 # 장을 무슨 몫인지 말하지 않고 붙이면 서로의 배치· # 구도가 섞여 들어온다(단일 참조에서 이미 겪은 실패). # 앞장 수를 함께 주는 이유 = 경로마다 앞이 한 장 # (사진)이거나 두 장(도해+사진)이다. _n_fit = len(form_ref_labeled) - 1 _fit_clause_direct = ( "\n\n" + build_fitting_ref_clause( head=1, count=_n_fit)) if _n_fit else "" _fit_clause_sketched = ( "\n\n" + build_fitting_ref_clause( head=2, count=_n_fit)) if _n_fit else "" _clause = build_form_only_clause() + _fit_clause_direct _gen_prompt = _gen_prompt.rstrip() + "\n\n" + _clause # ★[2026-08-01 Codex BLOCKING 2] 생성은 base prompt 를 # 보지 않는다 — multiroll 은 roll_prompts 가 있으면 # roll_prompts[label] 만 gen_fn 에 넘긴다. 절을 base 에만 # 붙이면 **웹 사진만 첨부되고 관할 계약은 어느 롤에도 # 실리지 않는다.** 카나리 A/B 실측: 주 표시면의 기관 # 표장이 A(절 없음) 2/2 없음 → B(전 롤 주입) 2/2 있음. # 절 본문이 "그 지역의 표시 관례와 서체"를 요구하므로 # "글자만 덜렁" 결함의 일부는 이 배선 결함이 만든 것이다. if variants_on and roll_prompts: # ★두 경로를 하나씩 만든다(2026-08-03 사용자 확정 # 흐름). 저작은 여러 벌 그대로 받되 **첫 한 벌**만 # 써서 두 롤에 같은 문장을 준다 — 경로 차이만 # 남아야 어느 쪽이 나은지가 판정된다. base_roll = roll_prompts[ roll_labels(seed_roll_count)[0]] from app.modules.pipeline.search_grounded_ref \ import build_sketch_transform_clause sketch_prompt = ( build_sketch_transform_clause() + "\n\n" + base_roll) sketch_path = self._make_sketch( gid=gid, prompt=sketch_prompt, ref_path=Path(form_ref_labeled[0][1]), out_dir=out_dir, gen_fn=gen_fn, force=force_mode) roll_prompts = { SEED_PATH_DIRECT: ( base_roll.rstrip() + "\n\n" + _clause), } roll_refs = { SEED_PATH_DIRECT: list(form_ref_labeled), } if sketch_path is not None: # 형태=도해 / 표면=사진 으로 관할을 쪼갠다. # 부속은 그 뒤에 붙으므로 앞장이 두 장이다. roll_prompts[SEED_PATH_SKETCHED] = ( base_roll.rstrip() + "\n\n" + build_form_only_clause_sketched() + _fit_clause_sketched) roll_refs[SEED_PATH_SKETCHED] = [ ("STRUCTURE DIAGRAM", sketch_path), *form_ref_labeled, ] group_roll_count = len(roll_prompts) judge_fn = judge_refpick_fn elif roll_prompts: roll_prompts = { lab: (txt.rstrip() + "\n\n" + _clause) for lab, txt in roll_prompts.items() } group_extra_fp["form_ref"] = form_ref_fp seed_sel, record = run_multiroll_select( tag=f"structure_seed_{gid}", prompt=_gen_prompt, # ★경로 A/B 일 때 판정에는 **주 구조물 사진 한 장만** # 보여 준다. 그 판정의 계약이 "참조 사진 대비 택일" # 이라 참조가 여럿이면 무엇에 대어 보는지가 흐려진다 # (판정문이 "the first one" 하나를 가리킨다). 생성용 # 참조는 `roll_refs` 가 롤마다 따로 쥐고 있으므로 # 부속이 빠지지 않는다. roll_refs 가 없는 경로에서는 # 이 인자가 곧 생성 참조라 전부 넘긴다. labeled_refs=(form_ref_labeled[:1] if roll_refs else form_ref_labeled), out_stem=out_dir / f"seed_{gid}", gen_fn=gen_fn, judge_fn=judge_fn, critique_fn=critique_fn, fix_gen_fn=gen_fn, roll_count=group_roll_count, roll_refs=roll_refs, # ★경로 비교일 때만 좌우를 바꿔 두 번 묻는다. 자리 # 편향이 실재해서, 두 번 다 이긴 쪽만 승자가 된다. # 우선순위가 앞선 직접 경로가 동점을 가져가므로 # 스케치 경로는 **두 번 다 이겨야** 채택된다. judge_flip=bool(roll_refs), # ★실제 롤 라벨로 준다. 상수를 그대로 주면 도해가 # 실패해 롤이 하나뿐일 때 "priority 는 labels 의 # 순열이어야 한다"에 걸려 그 그룹이 판정에서 죽는다 # — 직접 경로만으로도 살려 두려던 설계가 뒤집힌다. flip_priority=roll_labels(group_roll_count), critique_enabled=bool( settings.still_recipe_critique_enabled), fix_head=judge_texts["fix_head"], fix_tail=judge_texts["fix_tail"], fix_label=judge_texts["fix_label"], record=record, roll_prompts=roll_prompts, # [2026-08-01 HIGH 5] 절이 롤 프롬프트에 실리므로 # critique 의 공유 문안은 **절 없는 원 브리프**를 # 준다 — 그래야 생성·판정·결함 각 경로에 정확히 # 1회씩 실린다(그대로 두면 critique 에만 2회). critique_shared_prompt=( seed_brief if (variants_on and form_ref_labeled) else None), persist_record_fn=persist_record_fn, extra_fingerprint=group_extra_fp, # BLOCKING-1(1): sidecar 완료 record 가 force 를 # no-op 으로 만드는 우회 차단 (변형 모드 한정 — # 레거시는 record 비영속이라 기존 semantics 유지) force=variants_on and force_mode, # ★재생성은 배선하지 않는다 — 편집으로 고칠 수 없는 # 결함은 편집 지시에서 빼고 기록만 남긴다(모듈 상단 # SEED_REGEN_ISSUE_POLICY 주석에 실측 근거). regeneration_issue_policy=SEED_REGEN_ISSUE_POLICY, ) if variants_on: # 이후 소비(asset prompt_used/갤러리/prev 앵커)의 # seed_prompt=실제 생성 계약=선정 변형 전문. seed_prompt = roll_prompts[record["selected"]] else: # ★[2026-08-01 Codex 재리뷰 HIGH 4] variants OFF 도 # 마찬가지다. 절을 붙인 것은 `_gen_prompt` 인데 # prompt_used·CP 에는 절 없는 옛 문자열이 남아 있었다 # — 기록이 실제 생성 계약과 어긋나면 감사가 무의미하다. seed_prompt = _gen_prompt except AppError: raise except Exception as exc: # noqa: BLE001 logger.exception( "outdoor_structure_seed: group=%s 실패", gid) results[gid] = {"status": "failed", "error": str(exc)} failed += 1 continue # 형태 참조 자산 = 두 경로 공통의 입력 간선. _form_ref_asset = (form_ref_fp or {}).get("asset_id") # ★도해도 자산으로 남긴다 — 씨드가 무엇에서 형태를 받았는지가 # 간선으로 이어져야 나중에 어느 경로였는지 되짚을 수 있다. sketch_asset_id: Optional[str] = None if sketch_path is not None: sketch_asset_id = self._upsert_seed_asset( group_id=gid, variant_type="sketch", pipeline_role="structure_seed_sketch", abs_png_path=str(sketch_path), prompt_used=sketch_prompt, generation_model=image_model, input_image_ids=[ aid for aid in [_form_ref_asset] if aid], ) seed_asset_id = self._upsert_seed_asset( group_id=gid, variant_type="seed", pipeline_role="structure_seed_photo", abs_png_path=str(seed_sel), prompt_used=seed_prompt, generation_model=image_model, # [2026-08-01 Codex BLOCKING 2] 검색 선택 실사를 참조로 물고 # 그렸으면 그 자산을 입력 간선으로 남긴다 — 비워 두면 어느 # 참조에서 나온 씨드인지 추적할 수 없다. input_image_ids=[ aid for aid in [ _form_ref_asset, # ★도해는 **스케치 경로가 이겼을 때만** 이 그림의 # 입력이다. 진 경로까지 간선으로 달면 계보가 거짓이 # 된다 — 그리지도 않은 것에서 나왔다고 기록된다. (sketch_asset_id if record.get("selected") == SEED_PATH_SKETCHED else None), ] if aid ], ) entry: Dict[str, Any] = { "status": "ok", "seed_png_path": str(seed_sel), "seed_asset_id": seed_asset_id, "seed_selected": record.get("selected"), "seed_prompt": seed_prompt, # [Codex 2차 재리뷰 BLOCKING 2] 소비한 참조를 typed 로 영속. # DB 간선만으로는 감사할 수 없다 — 간선이 거짓일 수 있고, # 변형 OFF 에서는 지문조차 CP 에 남지 않는다. "form_reference_input": build_consumed_ref_record( form_ref_fp), # ★참조가 **왜** 없는지(또는 있는지)를 남긴다 — 생성 정본 # 판정과 결손을 구별하는 유일한 근거다 (Codex NARROW 2). "form_reference_authority": authority_note( formref_groups.get(gid)), # ★[2026-08-01 HIGH-1] 판정·결함·수정 이력을 **변형 여부와 # 무관하게** CP 에 남긴다. sidecar 영속은 variants ON 에서만 # 일어나므로, OFF 경로에서는 이것이 유일한 감사 근거다. # 특히 재생성 몫으로 미룬 결함(regen_deferred_*)과 수정을 # 건너뛴 사유가 없으면 "왜 원본을 그대로 뒀는지" 알 수 없다. "seed_decision": { k: record.get(k) for k in ( "input_fingerprint", "totals", "ranking", "verdicts", "critique", "critique_skipped", "fix_skipped", "fix_skip_reason", "fix_prompt", "repair_mode", "regen_deferred_issues", "regen_deferred_issue_count", # 좌우를 바꿔 두 번 물은 결과 — 두 판정이 갈렸는지가 # 여기 남는다. 없으면 어느 쪽이 자리 때문에 이겼는지 # 나중에 알 수 없다. "judge_flip", ) if k in record }, # 어느 정책으로 판단했는지 — 기록만으로 재현 가능해야 한다. # 이름만 남기면 그 뜻이 바뀐 뒤 옛 기록을 어느 계약으로 # 읽어야 할지 알 수 없으므로 버전도 함께 남긴다. "seed_regen_issue_policy": SEED_REGEN_ISSUE_POLICY, "seed_regen_issue_policy_version": ( _regen_issue_policy_version()), # ★어느 경로가 돌았고 어느 쪽이 이겼는지. 스케치가 실패해 # 직접 경로만 돈 그룹도 여기서 구분된다(labels 가 하나). "seed_paths": { "labels": sorted(roll_prompts or {}), "selected": record.get("selected"), "sketch_png_path": ( str(sketch_path) if sketch_path else None), "sketch_asset_id": sketch_asset_id, "policy_version": SEED_PATH_POLICY_VERSION, }, } if variants_on: # 저작 감사 기록 — 변형 전문·근거 인용·브리프 재추적 가능 entry["seed_variants"] = author_data entry["seed_roll_prompts"] = roll_prompts entry["seed_brief"] = seed_brief results[gid] = entry completed += 1 self.db.commit() return { "applicable_count": len(target_gids), "completed_count": completed, "failed_count": failed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": {"groups": results}, }