"""야외 구조물 형태 참조 — 웹 이미지 검색 → VLM 비교선택 1장 (선행 스텝). 사용자 지시 (2026-07-28): "야외 구조물(건물이나 규모가 있는 모든 복잡한 것들)은 배경 생성시에(씨드 생성시에) 모두 **검색 기반 + VLM** 으로". 설계 = docs/superpowers/specs/ 2026-07-28-search-grounded-seed-and-conti-entity-chain-design.md ## 왜 독립 선행 스텝인가 (설계 §8-2) seed 와 (나중의) 콘티 계보가 **같은 선택 자산**을 재사용해야 하고, 검색은 네트워크·비결정·비용 수명이 seed 생성과 다르다. seed 프롬프트만 바뀔 때 비싼 검색이 재실행되면 안 된다. ## 대상 집합 (v3 — 2026-07-30 사용자 확정 "전 그룹 무조건 검색") 두 번 틀렸고, 두 번 다 **"어떤 배경이 검색을 탈 자격이 있나"를 판단하려 한 것**이 원인이었다. - **v1 = `structure_plate` 바인딩 exact parity.** 프록시를 잘못 골랐다. `structure_plate` 는 구조물의 성질이 아니라 **샷의 필요성**이다(그 그룹의 선택 샷 중 인물 뒤로 구조가 보존돼야 하는 샷이 있느냐). 금월도 2고 실측 에서 씨드 25그룹 중 **10만 검색을 탔고** 편의점·파출소·주유소·도축장· 식당 데크·목조 창고·스피커탑 등 15그룹이 순수 T2I 로 남았다. - **v2 = 바인딩 ∪ LLM scope 판정.** 판정을 얹어 넓히려 했으나 ①판정 입력에 작품 고유명사(`Geumwoldo`·`Banwol`·`Suri-young` 실측)가 실려 이름으로 의미를 판정하는 경로가 열렸고 ②validator 가 **근거 없는 NO 를 통과**시켜 "애매하면 yes" 가 강제되지 않았다. 판정기 하나가 틀리면 그 배경은 또 빠진다 — v1 과 같은 실패를 다른 옷을 입고 반복하는 구조였다. **v3 = 판정을 없앤다. 대상 = 소비자(seed)와 exact parity.** ``` 검색 대상 = collect_lane2_groups(lane, all_groups=) - (place spec 결손 그룹) ``` 프록시도 판정도 없으므로 **조용히 빠지는 경로가 존재하지 않는다.** 유일한 제외는 place spec 결손(검색어를 저작할 근거 자체가 없는 경우)이고, 그것은 `data.target.no_spec_group_ids` 에 명시 기록된다. ## 사전조사가 검색보다 먼저다 (2026-08-03) 예전에는 사전조사가 **씨드 스텝 안에서, 검색이 다 끝난 뒤에** 돌았다. 그래서 조사가 정확한 답을 내도 그 답이 이미지 질의로 이어질 방법이 없었다 — 조사와 검색이 서로 모른 채 따로 돌았다. 실측(편의점 앞 앉을 자리): 조사는 한 문항에 열 가지를 뭉쳐 물어 "증거 없음"으로 끝났고, 이미지 질의 12개는 전부 "외관 출입문 간판 보도 벤치"로 뭉쳐 나가 건물 전경만 회수했다. 그 물건의 실물 근거가 파이프 어디에도 들어오지 않았고 저작이 근거 없이 형태를 지어냈다. 지금은 한 그룹에서 이 순서다: ①당연하지 않은 대상만 남긴다(보통 하나) — 명세가 붙인 이름이 아니라 **원문 인용**이 근거다 ②그 하나를 한 줄로 원어 조사한다(독립 2회 → 일치/단독 채택) ③조사 산출을 근거로 주 구조물 검색 지시문을 쓴다 + 언어 잠금을 원어로 따로 저작해 지시문 **맨 앞**에 놓는다 ④주 구조물 사진 1장을 이중 판정으로 고른다 ⑤채택된 대상마다 **따로** 질의를 써서 검색하고 1장을 고른다(부속 참조) 부속 참조는 파일·sha 로만 남기고 자산으로 묶지 않는다 — 라운드 manifest 가 예약하는 UUID 는 하나뿐이라, 여러 개로 늘리는 것은 저장 계약 변경이다. 계보 간선은 주 참조만 갖는다(없는 간선을 지어내지 않는다). ## 후보 선택 (2026-07-30 사용자 확정) 회수 후보 중 1장 선택은 **Gemini + GPT 이중 판정**이다. 두 심판이 각자 점수를 매기고 **평균이 가장 높은 후보**를 고른다. 한쪽 심판이 죽으면 살아남은 쪽 단독 판정으로 진행하되 그 사실을 감사에 남긴다(조용한 격하 금지). ## 그룹 단위 재사용 (설계 §9) 검색은 네트워크·비결정·비용 수명이 다르다. 그룹 입력과 **검색·선택 계약 바이트**가 그대로면 재검색하지 않고 이전 선택을 재사용한다(파일 존재+sha 재검증 통과 시에만). `force` 는 재사용을 우회한다. ## 실패 계약 (설계 §8-3) 필수 대상은 **fail-closed**. 순수 T2I 자동 degrade 금지 — 조용한 하강은 사용자가 지적한 "배경이 이상해"를 그대로 재생산하면서 아무도 모르게 만든다. 단 **한 그룹의 실패가 다른 그룹을 죽이지 않는다**(그룹 단위 fail-closed). """ from __future__ import annotations import json import logging from pathlib import Path from typing import Any, Dict, List, Optional from app.core.step_runner import StepRunner # ★모듈 최상단에서 가져온다. 이 파일의 관례는 함수 안 local import 지만, # 블록을 옮기며 import 를 빠뜨려 **호출 즉시 NameError** 가 나는 사고가 # 실제로 있었다(`_group_fingerprint` 의 settings). 순환이 없는 모듈은 # 최상단에 두는 편이 그 실패 모드를 아예 없앤다. from app.modules.pipeline.typology_prior import ( SEARCH_MODELS as TYPOLOGY_SEARCH_MODELS, ) from app.modules.pipeline.typology_prior import ( TYPOLOGY_PRIOR_VERSION, ) from app.modules.pipeline.form_ref_rounds import ( PROJECTED_ASSET_ID_RULE, # ★투영 자산 id 정책은 라운드 모듈이 소유 (스텝은 UUID 를 직접 만들지 않는다) BOUND_STATES, LEGACY_CONTRACT_VERSION, LEGACY_PATH_KIND, LEGACY_ROUND_ID, MISSING, NO_REF_RESULT_STATUS as _NO_REF_RESULT_STATUS, PROVENANCE_KEYS, ROUND_CP_PROJECTION_KEYS, ROUND_PATH_KIND, ROUND_STORAGE_CONTRACT_VERSION, resolve_round_without_reference, Round, RoundState, candidate_dir, finalize_round, load_registered_round, next_candidate_index, resolve_round, transition_round, typed_equal, ) logger = logging.getLogger(__name__) def _round_marks(prior: Dict[str, Any], *, cp_schema: Any) -> Dict[str, Any]: """재사용하는 이전 산출에 붙일 **저장 계약 표기**. 라운드 계약 이전에 만들어진 산출은 그룹 디렉터리에 `cand_NN` 로 평평하게 쌓여 있다. 그것을 옮기거나 덮어쓰지 않고 `legacy` 로 명시만 한다 — `r000` 을 쓰지 않는 이유는 새 계약의 정상 순번처럼 보이기 때문이다. ★legacy 인지는 **CP 스키마 버전**으로 판정한다. "round_id 가 없으면 legacy" 로 두면 **현행 CP 가 손상돼 그 필드만 사라져도 legacy 로 강등** 되고, legacy 예외(journal 대조 생략)가 검증을 통째로 건너뛴다(실측: 그 상태에서 자산 row 까지 지웠는데 유료 0·자산 0 으로 "완료"됐다). 필드 부재는 손상이지 legacy 가 아니다. """ from app.core.errors import AppError legacy = { "round_id": LEGACY_ROUND_ID, "path_kind": LEGACY_PATH_KIND, "round_contract_version": LEGACY_CONTRACT_VERSION, } # ★"부재" 와 "존재하지만 malformed" 를 합치면 손상이 legacy 로 내려간다. # 실측: `schema_version` 을 문자열 "4" 로 바꾸고 registry 를 깨뜨렸는데 # status=ok · round_id=legacy · paid=0 으로 통과했다. # ★**키 부재만** legacy 다 — `dict.get()` 은 키 부재와 값 None 을 같은 # 것으로 만들어, 명시적 null 이 다시 구 스키마로 오인됐다(실측). if cp_schema is MISSING: return legacy if isinstance(cp_schema, bool) or not isinstance(cp_schema, int): raise AppError( code="form_reference.round_marks_invalid", message=(f"CP schema_version 이 정수가 아니다: {cp_schema!r} — " f"legacy 로 강등하지 않는다")[:500], status_code=422) schema = cp_schema if schema < ROUND_CONTRACT_SCHEMA_VERSION: return legacy # 현행 스키마 CP 는 라운드 표기를 **전부** 갖고 있어야 한다. rid = str(prior.get("round_id") or "").strip() kind = str(prior.get("path_kind") or "").strip() ver = prior.get("round_contract_version") bad = [] if not rid: bad.append("round_id 없음") if kind != ROUND_PATH_KIND: bad.append(f"path_kind={kind!r}") if isinstance(ver, bool) or not isinstance(ver, int): bad.append(f"round_contract_version={ver!r}") if bad: raise AppError( code="form_reference.round_marks_invalid", message=(f"현행 스키마(v{schema}) CP 인데 라운드 표기가 온전하지 " f"않다 — legacy 로 강등하지 않는다: " + "; ".join(bad))[:500], status_code=422) return {"round_id": rid, "path_kind": kind, "round_contract_version": int(ver)} # v2 (2026-07-29): 그룹 단위 재사용 — CP shape 에 # groups[gid].group_fingerprint·reused 추가. # v3 (2026-07-30): 대상 = seed exact parity(판정 제거) + 이중 판정 선택. # CP shape 의 data.scope 가 data.target 으로 대체됐다. # v4 (2026-08-02, A4): 라운드 불변 저장. group entry 에 round_id · path_kind · # round_contract_version 이 붙고 최상위에 round_replayed_count 가 생겼다. # ★shape 이 바뀌면 config_hash 만으로 두지 않는다 — 해시 없는 옛 CP 에는 # mismatch 검사가 적용되지 않는 경로가 있다(A7 실측). # v5 (2026-08-03): 사전조사가 이미지 검색 **앞으로** 들어왔다. group entry 에 # typology(대상·문답·채택)와 fitting_refs(부속 대상별 참조 사진)가 붙는다. # ★step_manifest.py 의 "schema_version" 과 반드시 동기 — 어긋나면 신규 CP 를 # resume 이 구 스키마로 읽는다. test_step_schema_pins.py 가 교차 검증한다. SCHEMA_VERSION = 5 STEP_ID = "outdoor_structure_form_reference" # ── [v12] 시각 권위 (2026-09-05) ────────────────────────────────────── #: 이 이야기가 지어낸 대상 — 맞출 실사 정본이 없다. 사진을 사지 않는다. GENERATIVE_AUTHORITY = "generative_authority" #: 사진으로 확인되는 대상 — 실재하거나, 널리 본 작품이 모습을 정해 둔 것. _EXTERNAL_AUTHORITIES = ("observable_real", "established_visual_canon") #: 그룹 기록의 status — `ok`(사진 잠금)·`failed`(못 찾음)와 **다른 사건**이다. #: 하류는 이 값을 보고 생성으로 일관성 정본을 만든다. # ★값의 주인은 `form_ref_rounds` 다 — 라운드의 사진 없는 **종착**을 소유한 # 모듈이 그 산출 status 도 정한다. 여기서 다시 적으면 한쪽만 고쳐진다. GENERATIVE_STATUS = _NO_REF_RESULT_STATUS #: 이 판정 계약이 바뀌면 지문이 움직여야 한다 — 안 그러면 옛 CP 가 새 계약인 #: 것처럼 재사용된다. VISUAL_AUTHORITY_CONTRACT_VERSION = "1.202609050430" # 라운드 계약이 도입된 CP 스키마. **legacy 판정의 유일한 기준**이다 — # 이 값 미만(또는 스키마 부재)만 라운드 이전 산출로 본다. ROUND_CONTRACT_SCHEMA_VERSION = 4 # 판정 합산 계약의 버전 — 심판 응답을 어떻게 검증·합산하는지가 바뀌면 올린다. # config_hash 에 실려 **완료된 CP 를 의도적으로 stale** 시킨다. 이것이 없으면 # 계약을 고쳐도 이미 끝난 프로젝트는 옛 합산 결과를 그대로 쓴다(지난 wave 실측). # v1 (2026-08-01, A6): 심판 응답을 원자 단위로 검증한다 — index 가 1..N # 정확한 순열이 아니면 그 심판 전체를 제외하고, 살아남은 심판은 모든 # 후보에 동일 분모로 기여하며, valid 0명이면 fail-closed. PICK_COMBINE_CONTRACT_VERSION = "1" # ── 사전조사 · 부속 참조 (2026-08-03) ──────────────────────────────── # 사전조사가 **이미지 검색보다 먼저** 이 스텝 안에서 돈다. 예전에는 씨드 # 스텝이 검색이 다 끝난 뒤에 조사했고, 그래서 조사 결과가 질의로 이어질 # 방법이 없었다. 계약이 바뀌면 이 버전을 올려 완료 CP 를 stale 시킨다. # v1 (2026-08-03): 대상 좁히기 → 원어 조사 → 조사 결과를 근거로 질의 → # 대상별 단독 검색 → 부속 참조 선정. PRE_RESEARCH_CONTRACT_VERSION = "1" # ★★★앞쪽(중앙 reference_acquisition + 야외 보충) 산출을 **읽어서 투영**하는 계약 — 이 소비 경계의 지문. # (Codex BLOCK 2026-09-03) 소유권이 legacy 구매 → 중앙 보충으로 바뀌었는데 config hash 가 안 움직여, 옛 legacy # 구매형 completed CP 가 새 코드에서 그대로 되쓰일 자리였다. 사는 모드에서만 hash 에 결속한다(legacy hash 불변). #: 2.202609030915 — 투영이 자산 행(ImageAsset · structure_form_ref/form_ref)을 **묶는다**. ★실측 f7cc45c576c0 09:06 (한 run 야외 통합): #: 투영은 path·sha·final_id 만 싣고 legacy 라운드만 하던 자산 결속을 안 해 `form_ref_asset_id=None` → 씨드가 fail-closed 로 섰다. FRONT_PROJECTION_CONTRACT_VERSION = "2.202609030915" # 한 그룹에 붙일 부속 참조 상한. 조사가 남기는 대상이 보통 하나이므로 둘이면 # 넉넉하다 — 늘리면 참조가 서로의 배치를 끌고 들어온다. MAX_FITTING_REFS = 2 # 부속 대상 하나당 회수할 사진 수. FITTING_IMAGE_RESULTS = 6 # 부속 선정 심판. 주 참조의 이중 판정과 달리 단독이다 — 고르는 것이 # "그 물건이 잘 보이는가" 하나뿐이라 척도 합산이 성립할 축이 없다. FITTING_PICK_JUDGE = "gemini-pro" class OutdoorStructureFormReferenceStep(StepRunner): """구조물마다 실사 형태 참조 1장을 검색·선택해 영속한다.""" def _config_hash(self) -> str: import hashlib from app.core.config import settings from app.modules.pipeline.search_grounded_ref import ( MAX_PICK_CANDIDATES, PICK_HEAD_CONTRACT_VERSION, PICK_JUDGES, SAFE_DOWNLOAD_POLICY_VERSION, SEARCH_IMAGE_RESULTS, SEARCH_ORCHESTRATOR, TARGET_POLICY_VERSION, build_web_search_tool, resolve_ref_pack_version, ) payload = self._config_payload_legacy(settings, resolve_ref_pack_version, PICK_HEAD_CONTRACT_VERSION, build_web_search_tool, SEARCH_IMAGE_RESULTS, MAX_PICK_CANDIDATES, SEARCH_ORCHESTRATOR, SAFE_DOWNLOAD_POLICY_VERSION, PICK_JUDGES, TARGET_POLICY_VERSION) payload.update(self._central_ownership_binding()) return hashlib.sha256( json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8") ).hexdigest() def _central_ownership_binding(self) -> Dict[str, Any]: """★사는 모드(`buys_reference`)에서만 — 중앙 소유권 · 보충 계약 · 앞쪽 투영 계약을 hash 에 싣는다. legacy/사지 않는 모드는 빈 dict → 기존 hash 와 byte-identical (Codex BLOCK 2026-09-03).""" from app.core.grounding_mode import buys_reference, resolve_grounding_mode from app.modules.pipeline import grounding_outdoor_supplement as gos from app.modules.pipeline import reference_acquisition as ra mode = resolve_grounding_mode(getattr(self, "project_config", None) or {}) if not buys_reference(mode): return {} return {"central_ownership": { "grounding_mode": mode, "owner": ra.OWNER_FRONT, "acquisition_contract": ra.ACQUISITION_CONTRACT_VERSION, "supplement_contract": gos.SUPPLEMENT_CONTRACT_VERSION, "front_projection_contract": FRONT_PROJECTION_CONTRACT_VERSION, }} def _config_payload_legacy(self, settings, resolve_ref_pack_version, PICK_HEAD_CONTRACT_VERSION, build_web_search_tool, SEARCH_IMAGE_RESULTS, MAX_PICK_CANDIDATES, SEARCH_ORCHESTRATOR, SAFE_DOWNLOAD_POLICY_VERSION, PICK_JUDGES, TARGET_POLICY_VERSION) -> Dict[str, Any]: """옛 payload 그대로 — 키 하나도 더하거나 빼지 않는다(legacy hash 불변 시험이 잠근다).""" return { "schema_version": SCHEMA_VERSION, "ref_pack": resolve_ref_pack_version(), # ★[v12] 대상별 시각 권위 판정 계약 — 팩 바이트가 아니라 **코드**가 # 정하는 분기(생성 정본이면 검색을 안 산다)라 팩 해시에 안 잡힌다. # 안 실으면 옛 완료 CP 가 새 계약인 것처럼 재사용된다. "visual_authority_contract": VISUAL_AUTHORITY_CONTRACT_VERSION, # 팩 바이트가 아니라 **코드**가 정하는 심판 머리말 조립 계약. # ref_pack 만으로는 이 변경이 식별되지 않는다. "pick_head_contract": PICK_HEAD_CONTRACT_VERSION, # 검색 계약 자체가 산출의 실질 입력 — 바뀌면 재실행 유도 "web_tool": build_web_search_tool(SEARCH_IMAGE_RESULTS), "candidate_cap": MAX_PICK_CANDIDATES, "search_orchestrator": SEARCH_ORCHESTRATOR, "safe_download_policy": SAFE_DOWNLOAD_POLICY_VERSION, "pick_combine_contract": PICK_COMBINE_CONTRACT_VERSION, # 후보·자산의 저장 계약(라운드 불변 경로). ★`round_id` 는 넣지 # 않는다 — 입력이 아니라 산출 identity 라, 넣으면 재개가 항상 # miss 되어 매번 새로 검색한다. "round_storage_contract": ROUND_STORAGE_CONTRACT_VERSION, # ★별칭 뒤의 **물리 모델**도 판정의 실질 입력이다. 별칭은 그대로인 # 채 모델만 갈아 끼우는 것이 실제 운영에서 일어나는 변화이고, # 그게 안 실리면 완료 CP 가 옛 심판 결과를 그대로 재사용한다. "judge_physical_gemini": str( getattr(settings, "gemini_text_model", "")), "judge_physical_openai": str( getattr(settings, "openai_model", "")), "brief_model": "gpt", # 배경 검색 후보 선택 = Gemini + GPT 이중 판정 평균 (사용자 확정 # 2026-07-30). 심판 구성이 바뀌면 선택 결과가 바뀐다. "pick_judges": list(PICK_JUDGES), # v3: 대상 집합 계약 — 판정 없이 seed 대상 exact parity. # 모집단 플래그가 바뀌면 대상 자체가 바뀌므로 산출의 실질 입력. "target_policy": TARGET_POLICY_VERSION, "target_universe_all_groups": bool(getattr( settings, "outdoor_seed_all_groups_enabled", False)), "outdoor_lane_pipe_enabled": bool( getattr(settings, "outdoor_lane_pipe_enabled", False)), # 사전조사·부속 참조 계약 — 산출(질의·회수·붙는 참조)을 지배한다. "pre_research_contract": PRE_RESEARCH_CONTRACT_VERSION, "typology_prior": TYPOLOGY_PRIOR_VERSION, "typology_search_models": list(TYPOLOGY_SEARCH_MODELS), "max_fitting_refs": MAX_FITTING_REFS, "fitting_image_results": FITTING_IMAGE_RESULTS, "fitting_pick_judge": FITTING_PICK_JUDGE, } 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("%s: %s 로드 실패: %s", STEP_ID, step_id, exc) return None # ── 산출 경로 (프로젝트 루트 안 — DB 경로 CHECK 제약) ────────── def _ref_dir(self, group_id: str) -> Path: from app.core.config import settings return ( Path(settings.projects_dir) / self.project_id / "images" / self.episode_id / "structure_form_refs" / group_id ) def _bind_ref_asset(self, *, group_id: str, asset_id: str, abs_png_path: str, sha256: str, prompt_used: str) -> str: """미리 채번한 UUID 로 자산을 **idempotent insert-or-verify** 한다. ★이 자리에 있던 upsert 가 A4 의 진짜 결함이었다. 그룹당 한 row 를 찾아 `existing.file_path = rel_path` 로 **갱신**했기 때문에, 씨드의 `input_image_ids` 가 그 asset_id 를 가리키는 상태에서 재검색 한 번이 **과거 씨드의 입력 간선을 다른 이미지로 바꿨다.** 계보 기록이 조용히 거짓이 된다. 그래서 라운드마다 새 UUID 를 쓰고, 기존 row 는 **고치지 않고 검증만** 한다. ``ImageAsset`` 에는 sha 컬럼이 없다(실측 — 메타 계열은 `pipeline_metadata_json` 뿐). 그래서 row 가 가리키는 파일을 `ImagePathType` 로 resolve 해 **실제 bytes SHA** 를 journal SHA 와 비교한다. ★`annotate_generated_asset` 에 계약 검증을 맡기지 않는다 — 그 helper 는 `except Exception:` 으로 삼키고 `logger.warning` 만 하는 non-fatal 이라(annotate.py:73) 실패가 조용히 성공으로 보인다. role 주석에만 쓴다. """ from datetime import datetime, timezone from app.core.errors import AppError from app.core.file_paths import resolve_image_path from app.models.project import ImageAsset from app.services.image_capture.annotate import ( annotate_generated_asset, ) # ★입력 자체를 먼저 본다. 신규 INSERT 경로에 이 검사가 없어서, 선택 # 뒤 파일이 지워지거나 바뀌어도 **잘못된 row 와 FINALIZED 산출이 최초 # 바인딩에서 확정**됐다(검증이 기존 row 분기에만 있었다). self._asset_preflight(group_id=group_id, abs_png_path=abs_png_path, sha256=sha256) row = ( self.db.query(ImageAsset).filter_by(id=asset_id).first() ) if row is None: row = ImageAsset( id=asset_id, project_id=self.project_id, episode_id=self.episode_id, asset_type="structure_form_ref", entity_id=group_id, variant_type="form_ref", file_path=abs_png_path, prompt_used=prompt_used, status="generated", # ImageAsset.is_primary 는 Integer 컬럼이다 — Python bool 을 # 넣으면 Postgres 가 DatatypeMismatch 로 거부한다(실측). is_primary=0, # created_at 은 NOT NULL 이고 서버 기본값이 없다 — 빠뜨리면 # NotNullViolation (실측). 기존 seed upsert 와 동형. created_at=datetime.now(timezone.utc), ) self.db.add(row) # 계보 주석 — 이 자산이 어느 단계 산출인지 기록(설계 §6.3.1). annotate_generated_asset( row, pipeline_role="structure_form_reference", input_image_ids=[]) self.db.flush() return asset_id # 이미 있으면 **재사용하되 exact 검증** — 하나라도 다르면 fail-closed. mismatches = self._asset_mismatches( row=row, group_id=group_id, abs_png_path=abs_png_path, sha256=sha256) if mismatches: raise AppError( code="form_reference.asset_bind_conflict", message=( f"{group_id}: 미리 채번한 자산 {asset_id} 가 이 라운드의 " f"산출과 다르다 — 덮어쓰지 않는다 (기존 계보를 소급 변조 " f"하지 않기 위해 fail-closed). 불일치: " + "; ".join(mismatches))[:500], status_code=422, ) return asset_id def _asset_preflight(self, *, group_id: str, abs_png_path: str, sha256: str) -> None: """묶기 전에 **입력 자체**를 확인한다 — 파일이 있고 bytes 가 맞는가.""" from app.core.errors import AppError problem = "" if not sha256: problem = "sha 가 비어 있다" elif not abs_png_path or not Path(abs_png_path).is_file(): problem = f"파일이 없다: {abs_png_path!r}" else: actual = _sha_file(Path(abs_png_path)) if actual != sha256: problem = (f"bytes sha={actual[:12]}… " f"(기대 {str(sha256)[:12]}…)") if problem: raise AppError( code="form_reference.asset_bind_input_invalid", message=(f"{group_id}: 자산으로 묶을 산출이 계약과 다르다 — " f"{problem}")[:500], status_code=422, ) def _asset_mismatches(self, *, row: Any, group_id: str, abs_png_path: str, sha256: str) -> List[str]: """그 row 가 이 라운드의 산출과 같은 것을 가리키는가.""" from app.core.file_paths import resolve_image_path out: List[str] = [] for field_name, want in ( ("project_id", self.project_id), ("episode_id", self.episode_id), ("asset_type", "structure_form_ref"), ("entity_id", group_id), ("variant_type", "form_ref"), ): got = getattr(row, field_name, None) if got != want: out.append(f"{field_name}={got!r} (기대 {want!r})") # 경로는 **같은 표현으로 맞춰** 비교한다. `ImageAsset.file_path` 는 # `ImagePathType` 이라 ORM 읽기에서 상대→절대로 복원된다 # (file_paths.py:90) — 한쪽만 상대화하면 정상 자산이 영영 어긋난다. want_path = resolve_image_path(abs_png_path) got_path = resolve_image_path(getattr(row, "file_path", "") or "") if not want_path or not got_path or str(want_path) != str(got_path): out.append(f"file_path={got_path!r} (기대 {want_path!r})") elif not Path(str(got_path)).is_file(): out.append(f"file_path 가 가리키는 파일이 없다: {got_path}") else: actual = _sha_file(Path(str(got_path))) if actual != sha256: out.append(f"bytes sha={actual[:12]}… " f"(기대 {str(sha256)[:12]}…)") return out def _verify_bound_asset(self, *, group_id: str, asset_id: str, abs_png_path: str, sha256: str) -> str: """자산이 계약대로 **실재하는지 읽기만으로** 확인한다. INSERT 금지. ★journal 이 FINALIZED 라고 해서 DB row 가 있는 것은 아니다. 자산은 DB 트랜잭션에, 라운드 상태는 파일에 산다 — 두 저장소 사이에 원자성이 없다. commit 전 crash 나 **뒤 그룹 실패의 rollback** 이 앞 그룹 row 를 지워도 journal 은 완료로 남는다. 그 상태를 성공으로 내보내면 존재하지 않는 UUID 가 계보에 실린다. 여기서 INSERT 하지 않는 이유 = 그러면 "유료 0콜 재투영"이 없는 자산을 지어내는 셈이 된다. """ from app.core.errors import AppError from app.models.project import ImageAsset if not asset_id: raise AppError( code="form_reference.asset_missing", message=f"{group_id}: 기록에 자산 id 가 없다", status_code=422) row = self.db.query(ImageAsset).filter_by(id=asset_id).first() if row is None: raise AppError( code="form_reference.asset_missing", message=(f"{group_id}: 기록된 자산 {asset_id} 가 DB 에 없다 — " "라운드는 완료라고 말하는데 자산이 없다. 다시 만들지 " "않고 선다(fail-closed)"), status_code=422) mismatches = self._asset_mismatches( row=row, group_id=group_id, abs_png_path=abs_png_path, sha256=sha256) if mismatches: raise AppError( code="form_reference.asset_bind_conflict", message=(f"{group_id}: 기록된 자산 {asset_id} 가 기록된 산출과 " f"다르다. 불일치: " + "; ".join(mismatches))[:500], status_code=422) return asset_id def _round_provenance(self) -> Dict[str, str]: """라운드를 열 때 header 에 고정할 **역사 기록**. ★이 값을 나중에 "현재 값과 같은가"로 검사하면 안 된다. 무효화 판단은 **지문**의 몫이고, 지문은 대상 정책을 **의도적으로 제외**하며 팩도 이름이 아니라 검색 계약 sha 만 접는다(byte-identical 팩 승격은 같은 해시로 재사용해야 한다 — `test_search_grounded_ref_v3` 가 잠근 계약). 현재 값과 대조하면 정상적인 역사 provenance 까지 거부하게 된다. 변조 방지는 **header 결속**이 한다 — 산출의 같은 필드가 이 header 값과 달라지면 `load_round` 가 세운다. """ from app.modules.pipeline.search_grounded_ref import ( TARGET_POLICY_VERSION, resolve_ref_pack_version, ) return { "ref_pack_version": str(resolve_ref_pack_version()), "target_policy_version": str(TARGET_POLICY_VERSION), } def _verify_round_cp(self, *, group_dir: Path, gid: str, prior: Dict[str, Any], round_id: str) -> None: """라운드 계약 CP 를 재사용하기 전에 **journal 과 exact 대조**한다. ★CP 만 보고 재사용하면 journal 손상·결손과 자산 row 결손이 조용히 건너뛰어진다 — journal 이 단일 권위라는 계약이 CP 경로에서만 뚫린다. legacy(라운드 계약 이전 산출)만 예외다. 그것은 journal 이 없다. """ from app.core.errors import AppError # ★`load_round` 를 직접 부르면 index.json 을 우회한다 — registry 가 # 깨져도 manifest 만으로 통과했다(실측). 등록 여부가 먼저다. rnd = load_registered_round(group_dir, round_id) if rnd.contract_version != ROUND_STORAGE_CONTRACT_VERSION: # ★재사용은 "계약 exact 일치" 위에서만 성립한다. 타입만 보고 값을 # 안 보면 999 짜리 라운드가 그대로 재사용된다. raise AppError( code="form_reference.round_cp_mismatch", message=(f"{gid}: {round_id} 의 저장 계약이 " f"{rnd.contract_version} 이다 (현재 " f"{ROUND_STORAGE_CONTRACT_VERSION})"), status_code=422) # ★끝난 라운드는 **둘**이다 — 자산이 있는 `FINALIZED` 와 사진 없이 # 확정된 `RESOLVED_NO_REF`. 앞의 것만 인정하면 생성 정본 그룹이 # 보통 재개에서 「CP 는 완료인데 라운드는 아니다」로 선다(실측). if rnd.state not in (RoundState.FINALIZED, RoundState.RESOLVED_NO_REF): raise AppError( code="form_reference.round_cp_mismatch", message=(f"{gid}: CP 는 {round_id} 를 완료로 적었는데 라운드는 " f"{rnd.state.value} 다"), status_code=422) # ★투영 필드를 **전부** 대조한다. 3개만 보면 path_kind·계약 버전·팩· # 정책 변조가 그대로 통과한다(실측). for key in ROUND_CP_PROJECTION_KEYS: want = rnd.result.get(key) got = prior.get(key) # 값뿐 아니라 **타입 계약**까지 — `True == 1` 이라 순수 값 비교는 # bool 을 정수로 받아들인다. # ★여기서는 **중복 방어**다: CP 쪽 bool 은 `_round_marks` 가, # journal 쪽 bool 은 `_validate_result_shape` 의 `_exact_int` 가 # 먼저 막는다(위반 주입으로 확인 — 이 줄만 되돌려도 회귀가 여전히 # 잡는다). 그래도 두 방어 중 하나가 옮겨가면 여기가 마지막 선이라 # 남긴다. if not typed_equal(want, got): raise AppError( code="form_reference.round_cp_mismatch", message=(f"{gid}: CP 의 {key} 가 라운드 기록과 다르다 " f"(CP={got!r} 라운드={want!r})")[:500], status_code=422) # ★[2026-08-03] 조사·부속 참조도 journal 에 결속한다. 투영 키 목록에 # 넣는 방법은 못 쓴다 — `_validate_result_shape` 가 그 목록을 **완료 # 산출의 필수 키**로 요구해서 기존 라운드가 전부 무효가 된다. 그래서 # 이 자리에서 결정론 해시로만 대조한다. 이것이 없으면 주 참조와 # 라운드 표기가 그대로인 채 CP 의 조사 블록·부속 목록만 달라져도 # 통과하고, 씨드가 journal 과 다른 입력을 물고 그린다(조사 블록은 # 텍스트라 소비측 파일 검증에도 걸리지 않는다). want_pr = _pre_research_digest(rnd.result) got_pr = _pre_research_digest(prior) if want_pr != got_pr: raise AppError( code="form_reference.round_cp_mismatch", message=(f"{gid}: CP 의 사전조사·부속 참조가 라운드 기록과 " f"다르다 (CP={got_pr} 라운드={want_pr})"), status_code=422) # ★사진 없이 끝난 확정에는 검증할 자산이 없다 (Codex BLOCK 1). # 라운드가 권위다 — CP 가 아니라 **journal 의 종착**으로 가른다. if rnd.has_reference: self._verify_bound_asset( group_id=gid, asset_id=str(prior.get("form_ref_asset_id") or ""), abs_png_path=str(prior.get("form_ref_path") or ""), sha256=str(prior.get("form_ref_sha256") or "")) def _close_round(self, *, group_dir: Path, rnd: Round, gid: str, rec: Dict[str, Any]) -> Dict[str, Any]: """산출에 맞춰 라운드를 전이하고, 성공이면 자산을 묶어 닫는다. ★**두 저장소의 쓰기 순서가 계약이다.** 라운드 상태는 파일에, 자산은 DB 트랜잭션에 산다. 둘 사이에 원자성이 없으므로 순서를 이렇게 고정한다: ①SELECTED + 산출을 journal 에 (DB 를 만지기 전) ②자산 insert-or-verify → **commit** (DB durable) ③ASSET_BOUND (그 다음에야 "자산 있음") ④FINALIZED + 최종 산출 반대로 하면(먼저 FINALIZED, 나중 commit) commit 전 crash 나 **뒤 그룹 실패의 rollback** 이 자산을 지워도 journal 은 완료로 남아, 다음 재개가 존재하지 않는 UUID 를 성공으로 내보낸다. 실패면 라운드를 **열어 둔 채** 둔다 — 다음 재개가 같은 라운드를 이어가고, OPENED 에서 미리 채번한 UUID 덕에 자산이 중복 INSERT 되지 않는다. 반환값은 CP entry 에 얹을 표기다. """ # 성공·실패 어느 쪽이든 **어느 라운드의 산출인지**는 남긴다. 실패 # entry 에 라운드가 없으면 남은 후보가 어디 것인지 되짚을 수 없다. where = { "round_id": rnd.round_id, "path_kind": ROUND_PATH_KIND, "round_contract_version": ROUND_STORAGE_CONTRACT_VERSION, # ★라운드 header 에 박힌 **그때의** 역사값을 산출에 싣는다. 현재 # 값을 다시 읽으면 재개 시점의 값이 섞이고, header 결속도 깨진다. **{k: rnd.provenance[k] for k in PROVENANCE_KEYS if k in rnd.provenance}, } state = rnd.state if (int(rec.get("candidate_count") or 0) and state == RoundState.OPENED): transition_round(group_dir, rnd.round_id, RoundState.CANDIDATES) state = RoundState.CANDIDATES if rec.get("status") == GENERATIVE_STATUS: # ★★사진 없이 끝난 **확정**이다 (Codex BLOCK 1 2026-09-08). # 여기서 그냥 돌아가면 라운드가 열린 채 남아, 재개가 같은 그룹의 # 판정 브리핑과 사전조사를 **매번 다시 산다**. 자산은 안 묶는다 — # 가짜 자산을 만들면 재투영이 없는 파일을 검증하러 간다. resolve_round_without_reference( group_dir, rnd.round_id, result={**rec, **where}) return where if rec.get("status") != "ok": return where # ① 선택 산출을 **DB 를 만지기 전에** durable 하게. 이게 없으면 bind # 직후 crash 가 "자산은 있는데 무엇을 골랐는지 모르는" 상태가 되어 # 재개가 유료로 다시 검색해야 한다. if state in (RoundState.OPENED, RoundState.CANDIDATES): transition_round(group_dir, rnd.round_id, RoundState.SELECTED, result={**rec, **where}) state = RoundState.SELECTED # ② 자산을 묶고 **commit 으로 durable 하게** 만든다. 그룹 단위 commit # 이라 뒤 그룹의 rollback 이 이 그룹의 자산을 되돌리지 못한다. asset_id = self._bind_ref_asset( group_id=gid, asset_id=rnd.preallocated_asset_id, abs_png_path=rec.get("form_ref_path") or "", sha256=rec.get("form_ref_sha256") or "", prompt_used=rec.get("chosen_prompt_used") or "") self.db.commit() # ③ 그 다음에야 journal 이 "자산 있음"이라고 말한다. ★자산 id 를 # **같은 원자 쓰기로** 산출에 넣는다 — header 의 예약 UUID 와 # 산출이 서로 묶여야 재개가 다른 UUID 로 새 자산을 만들지 못한다. if state == RoundState.SELECTED: transition_round(group_dir, rnd.round_id, RoundState.ASSET_BOUND, result={**rec, **where, "form_ref_asset_id": asset_id}) state = RoundState.ASSET_BOUND marks = {**where, "form_ref_asset_id": asset_id} # ★journal 이 최종 산출까지 갖고 있어야, CP 만 날아간 재개가 유료 # 0콜로 끝난다(계획 §2.3 마지막 행). finalize_round(group_dir, rnd.round_id, result={**rec, **marks}) return marks # ── 그룹 재사용 지문 ──────────────────────────────────────── def _group_fingerprint( self, *, structure_desc: str, world_facts_block: str, source_text_sha: str, research_input_sha: str, ) -> str: """이 그룹의 검색·선택 산출을 지배하는 입력 전부의 해시. 여기 들어가는 것만이 **한 그룹의 결과를 바꾸는 입력**이다. 대상 집합 정책(어느 그룹이 검색을 타는가)은 그룹 결과를 바꾸지 않으므로 넣지 않는다 — 넣으면 정책 변경만으로 이미 잘 뽑아 둔 참조를 전부 다시 받게 된다. 반대로 **심판 구성은 넣는다**(선택 결과가 달라진다). """ import hashlib # ★이 import 가 없어서 함수가 **호출 즉시 NameError** 였다(19fb2e65 가 # `_config_hash` 의 물리 모델 블록을 옮겨 오며 빠뜨렸다). 그 자리는 # `_execute` 의 그룹 try 안이라 전 그룹이 조용히 "failed" 로 떨어졌다. from app.core.config import settings from app.modules.pipeline.search_grounded_ref import ( MAX_PICK_CANDIDATES, PICK_HEAD_CONTRACT_VERSION, PICK_JUDGES, SAFE_DOWNLOAD_POLICY_VERSION, SEARCH_IMAGE_RESULTS, SEARCH_ORCHESTRATOR, search_contract_sha, ) payload = { "structure_desc": structure_desc, "world_facts_block": world_facts_block, "source_text_sha": source_text_sha, "search_contract": search_contract_sha(), # ★심판 머리말 조립은 **코드**가 정하므로 위 팩 바이트 해시에 # 잡히지 않는다. 안 실으면 같은 spec 에 대해 배치 포함(구)과 # 이름만(신)의 머리말이 **같은 지문**을 갖고, 완료 CP 가 새 # 계약인 것처럼 재사용된다. "pick_head_contract": PICK_HEAD_CONTRACT_VERSION, "search_image_results": SEARCH_IMAGE_RESULTS, "candidate_cap": MAX_PICK_CANDIDATES, "orchestrator": SEARCH_ORCHESTRATOR, "safe_download_policy": SAFE_DOWNLOAD_POLICY_VERSION, "pick_combine_contract": PICK_COMBINE_CONTRACT_VERSION, # 저장 계약이 바뀌면 그 그룹의 산출이 사는 곳과 불변성이 바뀐다 — # 실질 입력이므로 재사용 지문에도 싣는다. "round_storage_contract": ROUND_STORAGE_CONTRACT_VERSION, # ★별칭 뒤의 **물리 모델**도 판정의 실질 입력이다. 별칭은 그대로인 # 채 모델만 갈아 끼우는 것이 실제 운영에서 일어나는 변화이고, # 그게 안 실리면 완료 CP 가 옛 심판 결과를 그대로 재사용한다. "judge_physical_gemini": str( getattr(settings, "gemini_text_model", "")), "judge_physical_openai": str( getattr(settings, "openai_model", "")), "brief_model": "gpt", # 심판 구성은 **선택 결과를 지배**한다 — 단일 gemini 에서 이중 # 판정으로 바뀌면 이전 선택은 재사용할 수 없다(Codex 리뷰 6b 연장). "pick_judges": list(PICK_JUDGES), # ★사전조사가 검색 앞으로 들어오면서 **조사가 읽는 것 전부**가 # 산출의 실질 입력이 됐다 — 이 그룹의 씬 원문, 명세 항목과 그 # 원문 인용, 그리고 실내·외 설명. 전체 원문 sha 만으로는 이 # 그룹의 조사 입력이 바뀐 것을 못 본다. # ★실내·외 설명을 빠뜨리면 조용히 stale 을 재사용한다: 그 둘은 # 명세·씬이 아니라 **배경 분류의 멤버 목록과 위치 서술**에서 # 나오므로(outdoor_structure_seed.derive_seed_inputs) 나머지가 # 그대로여도 혼자 바뀔 수 있고, 그러면 조사 문항·답과 부속 검색 # 입력이 달라졌는데 지문은 같다. "research_input_sha": research_input_sha, "pre_research_contract": PRE_RESEARCH_CONTRACT_VERSION, "typology_prior": TYPOLOGY_PRIOR_VERSION, "typology_search_models": list(TYPOLOGY_SEARCH_MODELS), "max_fitting_refs": MAX_FITTING_REFS, "fitting_image_results": FITTING_IMAGE_RESULTS, "fitting_pick_judge": FITTING_PICK_JUDGE, } return hashlib.sha256( json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8") ).hexdigest()[:16] def _reusable( self, prev: Dict[str, Any], fingerprint: str, ) -> bool: """이전 산출을 재검색 없이 쓸 수 있는가 — 파일·sha 까지 재검증한다. 구 CP 엔트리에는 `group_fingerprint` 가 없다(v1 shape). 그때는 **그 엔트리가 기록한 팩의 검색 계약 바이트**가 현재와 같은지로 대신 판정한다 — v3 는 v2 의 검색·선택 4파일을 바이트 그대로 복사했으므로 같고, 그래서 승격만으로 잘 뽑힌 참조를 버리지 않는다. """ from app.modules.pipeline.search_grounded_ref import ( search_contract_sha, ) if prev.get("status") == GENERATIVE_STATUS: # ★사진 없이 끝난 **확정**은 파일이 없다 (Codex BLOCK 1). # 파일을 요구하면 이 그룹은 재개마다 판정을 다시 산다. # 지문이 같을 때만 되쓴다 — 구 CP(v1 shape)에는 지문이 없고, # 그때는 이 판정 자체가 없던 시절이라 되쓸 것도 없다. return bool(prev.get("group_fingerprint")) and \ prev.get("group_fingerprint") == fingerprint if prev.get("status") != "ok": return False path_s = prev.get("form_ref_path") or "" sha = prev.get("form_ref_sha256") or "" if not path_s or not sha: return False p = Path(path_s) if not p.is_file() or _sha_file(p) != sha: return False got = prev.get("group_fingerprint") if got: return got == fingerprint prior_pack = ((prev.get("audit") or {}).get("pack_version") or "") if not prior_pack: return False try: return (search_contract_sha(prior_pack) == search_contract_sha()) except Exception: # noqa: BLE001 — 미배선 구 팩은 재사용 불가 return False # ── 사전조사 (검색보다 **먼저**) ────────────────────────────── def _pre_research( self, *, group_id: str, structure_desc: str, interior_note_en: str, exterior_note_en: str, scene_blocks: List[str], spec_items: List[Dict[str, Any]], world_facts_block: str, client: Any, ) -> Dict[str, Any]: """당연하지 않은 대상만 골라 원어로 조사한다. ★이 조사가 **이미지 검색보다 먼저** 돌아야 한다는 것이 이 배선의 전부다. 예전에는 씨드 스텝 안에서 검색이 다 끝난 뒤에 돌았고, 그래서 조사가 답을 내도 그 답이 질의로 이어질 방법이 없었다 — 조사와 검색이 서로 모른 채 따로 돌았다(2026-08-03 실측). 조사의 목적은 형태를 글로 확정하는 것이 아니라 **무엇을 사진으로 찾을지 정하는 것**까지다. 반환 = {"block", "items", "questions", "dropped_reason_ko"}. `block` 은 검색 지시문 저작과 씨드 저작이 함께 쓰고, `items` 의 채택분이 부속 대상별 이미지 검색의 근거가 된다. ★조사 실패는 그룹을 죽이지 않는다. 빈 블록으로 내려가면 이 단계가 없던 어제까지의 동작이라 손해가 없다 — 다만 조용히 넘기지 않는다. """ from app.modules.pipeline.typology_prior import ( SEARCH_MODELS, build_typology_facts_block, combine_typology_answers, extract_gap_questions, search_typology_answers, summarize, ) empty: Dict[str, Any] = {"block": "", "items": [], "questions": [], "dropped_reason_ko": ""} try: gaps = extract_gap_questions( structure_desc=structure_desc, # layout_narration_en 은 structure_desc 에 이미 합쳐져 있다. layout_narration_en="", interior_note_en=interior_note_en, exterior_note_en=exterior_note_en, scene_blocks=scene_blocks, spec_items=spec_items, # ★확정 지역·시대 — 없으면 문항이 가리킬 데 없는 말로 # 나가고 조사가 전건 폐기된다(실측). world_facts_block=world_facts_block, project_config=self.project_config, ) questions = list(gaps.get("questions") or []) if not questions: logger.info("%s: %s 사전조사 대상 없음 — 조사 생략 (%s)", STEP_ID, group_id, gaps.get("dropped_reason_ko") or "") return {**empty, "dropped_reason_ko": gaps.get( "dropped_reason_ko") or ""} brief_native = str(gaps.get("search_brief_native") or "").strip() runs = [ search_typology_answers( client, questions=questions, model=m, brief_native=brief_native) for m in SEARCH_MODELS ] by_index = [ {a["index"]: a for a in (r.get("answers") or [])} for r in runs ] items = [ combine_typology_answers( question=q, answer_a=by_index[0].get(i), answer_b=by_index[1].get(i), project_config=self.project_config, ) for i, q in enumerate(questions, start=1) ] logger.info("%s: %s 사전조사 %s", STEP_ID, group_id, summarize({"items": items})) return { "block": build_typology_facts_block(items), "items": items, "questions": questions, "dropped_reason_ko": gaps.get("dropped_reason_ko") or "", } except Exception as exc: # noqa: BLE001 — 보조 단계, 격리한다 logger.warning("%s: %s 사전조사 실패 — 빈 블록으로 진행 (%s)", STEP_ID, group_id, exc) return empty # ── 부속 대상 참조 (조사 → 질의 → 검색 → 1장) ───────────────── def _fitting_refs( self, *, group_id: str, typology: Dict[str, Any], client: Any, candidate_dir: Path, start_index: int, exclude_sha: set, ) -> List[Dict[str, Any]]: """채택된 조사 대상마다 그 하나만을 위한 사진 1장을 회수·선정한다. ★주 구조물 질의에 섞지 않는다. 섞으면 회수가 지배적인 쪽(건물 전경) 으로 쏠려 그 물건 사진이 한 장도 오지 않는다(실측 12질의 전부). ★같은 사진을 두 번 쓰지 않는다 — 주 참조와, 앞선 부속과 bytes 가 같으면 버린다. 실험에서 같은 사진이 두 번 선정된 적이 있고, 그러면 관할이 겹쳐 한 장이 두 몫을 하는 것처럼 기록된다. """ from app.modules.llm.llm_client import call_structured from app.modules.pipeline.multiroll_gemini import png_part from app.modules.pipeline.search_grounded_ref import ( build_fitting_pick_schema, build_focus_query_schema, download_candidate, load_fitting_pick_system, load_focus_query_system, search_reference_images, ) out: List[Dict[str, Any]] = [] seen = set(exclude_sha) next_i = int(start_index) for item in (typology.get("items") or []): if len(out) >= MAX_FITTING_REFS: break if not item.get("adopted"): continue target = str(item.get("target_native") or "").strip() found = str(item.get("agreed_fact_en") or "").strip() if not target or not found: continue try: plan = call_structured( "structure_form_ref_focus_query", load_focus_query_system(), f"THE THING:\n{target}\n\n" f"WHAT THE TEXT SEARCH FOUND:\n{found}", build_focus_query_schema(), project_config={ "structure_form_ref_focus_query": {"model": "gpt"}}, schema_name="structure_form_ref_focus_query", opik_metadata={"project_id": self.project_id, "episode_id": self.episode_id, "step": STEP_ID, "target": target}, ) search = search_reference_images( client, directive_native=plan.get("directive_native") or "", terms_native=[str(t) for t in (plan.get("queries_native") or [])], language_lock_native=plan.get( "language_lock_native") or "", max_results=FITTING_IMAGE_RESULTS, ) cands: List[Dict[str, Any]] = [] for img in (search.get("images") or []): if len(cands) >= FITTING_IMAGE_RESULTS: break dest = candidate_dir / f"cand_{next_i:02d}.png" if not download_candidate(img.get("image_url") or "", dest, img.get("thumbnail_url") or ""): continue next_i += 1 sha = _sha_file(dest) if sha in seen: # 이미 다른 관할이 쥔 사진이다 — 후보에서 뺀다. continue cands.append({"index": len(cands) + 1, "path": str(dest), "sha256": sha, "source_website_url": img.get( "source_website_url")}) if not cands: logger.info("%s: %s 부속 '%s' 회수 0장", STEP_ID, group_id, target) continue parts: List[Any] = [] for c in cands: parts.append({"type": "text", "text": f"Photograph {c['index']}:"}) parts.append(png_part(Path(c["path"]).read_bytes())) parts.append({"type": "text", "text": "THE THING:\n" + target}) pick = call_structured( "structure_form_ref_fitting_pick", load_fitting_pick_system(), parts, build_fitting_pick_schema(), project_config={ "structure_form_ref_fitting_pick": { "model": FITTING_PICK_JUDGE}}, schema_name="structure_form_ref_fitting_pick", opik_metadata={"project_id": self.project_id, "episode_id": self.episode_id, "step": STEP_ID, "target": target}, ) idx = int(pick.get("index") or 0) if not (1 <= idx <= len(cands)): logger.info("%s: %s 부속 '%s' 선정 없음 — %s", STEP_ID, group_id, target, pick.get("why_ko")) continue chosen = cands[idx - 1] seen.add(chosen["sha256"]) out.append({ "target_native": target, "path": chosen["path"], "sha256": chosen["sha256"], "source_website_url": chosen.get("source_website_url"), "researched_fact_en": found, "language_lock_native": plan.get("language_lock_native"), "directive_native": plan.get("directive_native"), "queries_planned": plan.get("queries_native"), "queries_sent": search.get("queries"), "candidate_count": len(cands), "why_ko": pick.get("why_ko"), }) logger.info("%s: %s 부속 참조 확보 '%s' — %s", STEP_ID, group_id, target, Path(chosen["path"]).name) except Exception as exc: # noqa: BLE001 — 부속은 그룹을 죽이지 않는다 logger.warning("%s: %s 부속 '%s' 실패 — %s", STEP_ID, group_id, target, exc) return out # ── 그룹 1개 처리 ──────────────────────────────────────────── def _run_group( self, *, group_id: str, structure_desc: str, world_facts_block: str, source_text: str, client: Any, candidate_dir: Path, start_index: int = 1, narrow: bool = False, typology: Optional[Dict[str, Any]] = None, pick_structure_desc: str = "", ) -> Dict[str, Any]: from app.core.errors import AppError from app.modules.llm.llm_client import call_structured from app.modules.pipeline.multiroll_gemini import png_part from app.modules.pipeline.search_grounded_ref import ( MAX_PICK_CANDIDATES, PICK_JUDGES, build_audit_record, build_pick_schema, build_pick_user_head, build_search_brief_schema, build_search_brief_user, combine_pick_verdicts, download_candidate, load_brief_system, load_narrow_retry_hint, load_pick_system, search_reference_images, summarize_for_log, ) typology = typology or {"block": "", "items": [], "questions": []} # 1) 원본어 검색 지시문 저작 (씬 원문은 언어 판정 전용 — 격리) # ★사전조사 산출이 여기 들어온다. 조사를 먼저 돌린 이유가 무엇을 # 사진으로 찾을지 정하는 것이므로, 그 답이 질의의 근거여야 한다. brief = call_structured( "structure_form_ref_brief", load_brief_system(), (build_search_brief_user( structure_desc=structure_desc, world_facts_block=world_facts_block, source_text=source_text, researched_facts_block=typology.get("block") or "") + ("\n\n" + load_narrow_retry_hint() if narrow else "")), build_search_brief_schema(), project_config={"structure_form_ref_brief": {"model": "gpt"}}, schema_name="structure_form_ref_brief", opik_metadata={"project_id": self.project_id, "episode_id": self.episode_id, "step": STEP_ID}, ) # ── [v12] 대상별 시각 권위 판정 — **검색을 사기 전에** 갈린다 ── # # ★★★사진이 존재할 수 없는 대상에 사진을 요구하면 스텝이 선다 # (실측 2026-09-05: 창작 미래 폐광 → 후보 8장 · VLM 선택 0 · 실패). # 판정은 **연도가 아니라 대상**으로 한다 — 미래 이야기의 평범한 숲은 # 여전히 찍을 수 있고, 오늘 이야기가 지어낸 기계는 못 찍는다. # # ★조사 채택 수(`typology.adopted`)로 대신하지 않는다 (Codex BLOCK): # 그것은 「형태 문항에 사실을 채택했나」이지 「관객이 알아볼 시각 # 정본이 있나」가 아니다. 실재 대상도 검색 실패로 0 일 수 있고, # 창작 대상도 콘크리트·침식 같은 일반 사실은 채택될 수 있다. authority = str(brief.get("visual_authority") or "").strip() if authority == GENERATIVE_AUTHORITY: # 이 이야기가 지어낸 것 — 맞출 실사 정본이 없다. **검색을 안 산다.** # 실패가 아니라 **다른 종류의 확정**이다: 일관성 정본은 하류 # `outdoor_structure_seed` 가 생성으로 만들어 잠근다. logger.info("%s: %s 시각 권위 = 생성 — 검색 생략 (%s)", STEP_ID, group_id, brief.get("authority_reason_ko") or "") return { "status": GENERATIVE_STATUS, "group_id": group_id, "visual_authority": authority, "authority_reason_ko": str( brief.get("authority_reason_ko") or ""), "search_period_native": "", "candidate_count": 0, "audit": {"target_key": group_id, "brief": brief, "search": {}, "candidates": [], "verdict": {}, "chosen": None}, } if authority not in _EXTERNAL_AUTHORITIES: # ★모르는 값을 조용히 생성으로 접지 않는다 — 그러면 실사 정본이 # 있는 대상까지 사진 없이 내려간다. raise AppError( code="structure_form_ref.authority_unresolved", message=(f"{group_id}: 시각 권위 판정이 없거나 모르는 값이다 " f"({authority!r}) — 사진을 살지 말지 정할 수 없다"), status_code=422) _terms = [t for t in (brief.get("search_terms_native") or []) if str(t).strip()] if not _terms: # 외부 정본이 있다고 판정해 놓고 질의가 없다 — 계약 위반이다. raise AppError( code="structure_form_ref.no_search_terms", message=(f"{group_id}: 시각 권위 {authority} 인데 검색어가 " "없다 — 사진을 찾을 근거가 없다"), status_code=422) # 2) 웹 이미지 검색 (사진 결과) — 언어 잠금은 지시문 맨 앞에. search = search_reference_images( client, directive_native=brief.get("search_directive_native") or "", terms_native=brief.get("search_terms_native") or [], language_lock_native=brief.get("language_lock_native") or "", ) # 3) 후보 다운로드 (안전 정책) # ★후보는 **그 라운드 디렉터리 안에만** 쓴다. 예전에는 그룹 디렉터리에 # `cand_NN` 로 순번을 고정해 재실행이 이전 라운드를 덮어썼다(344건 # overwritten 실측). candidates: List[Dict[str, Any]] = [] for img in search.get("images") or []: if len(candidates) >= MAX_PICK_CANDIDATES: break # 파일 번호는 라운드 안에서 **이어 붙인다** — 좁혀 재검색하는 2차 # 시도가 1차 후보를 덮어쓰면 그 라운드의 감사 근거가 사라진다. # 반면 `index` 는 심판에게 주는 1..N 순번이라 시도마다 1부터다 # (A6 계약: indices 가 1..N 의 정확한 순열이어야 한다). dest = candidate_dir / f"cand_{start_index + len(candidates):02d}.png" if not download_candidate(img.get("image_url") or "", dest, img.get("thumbnail_url") or ""): continue candidates.append({ "index": len(candidates) + 1, "path": str(dest), "sha256": _sha_file(dest), "caption": img.get("caption"), "image_url": img.get("image_url"), "source_website_url": img.get("source_website_url"), }) if not candidates: return {"status": "failed", "group_id": group_id, "error": "후보 사진 0장 — 검색·다운로드 실패", "candidate_count": 0, "audit": build_audit_record( target_key=group_id, brief=brief, search=search, candidates=[], verdict={}, chosen=None)} # 4) VLM 비교 선택 (여러 장 중 1장 — 개별 판정이 아니다) # ★심판에게는 **무엇이 있는지**만 준다 (prompt diet ⑧). # pick_system 이 스스로 "배치는 판단 근거가 아니다 / 주변이 아니라 # 구조물을 판단하라" 고 못박는데도 layout narration + 항목별 위치 # 서술이 콜마다 실려 있었다. 결손이면 기존 서술로 폴백한다. head_desc = pick_structure_desc.strip() if not head_desc: # place spec 계약상 `items[].name_en` 은 nonblank 필수다 # (outdoor_place_spec validator). 여기서 비었다는 것은 stale· # malformed CP 라는 뜻이므로 **조용히 넘기지 않는다.** # 다만 그룹을 죽이지는 않는다 — 참조 사진 없이 하류가 진행되는 # 쪽이 더 나쁘고, 폴백 대상은 이 스텝이 원래 쓰던 입력이다. # 대신 덜어내기가 무효가 된다는 사실을 로그로 드러낸다. logger.warning( "%s: %s structure_names_en 결손 — place spec items[].name_en " "계약 위반(stale CP 의심). 심판 머리말을 배치 서술로 폴백한다 " "— 이 그룹에서는 덜어내기가 적용되지 않는다.", STEP_ID, group_id) head_desc = structure_desc parts: List[Any] = [{"type": "text", "text": build_pick_user_head( structure_desc=head_desc, world_facts_block=world_facts_block)}] for c in candidates: parts.append({"type": "text", "text": f"CANDIDATE {c['index']}:"}) parts.append(png_part(Path(c["path"]).read_bytes())) # ★이중 판정 (2026-07-30 사용자 확정): Gemini + GPT 가 각자 후보마다 # 점수를 매기고 **평균이 가장 높은** 1장을 고른다. 한 심판이 죽어도 # 살아남은 쪽으로 진행하되 `single_judge` 로 드러낸다(조용한 격하 금지). # 둘 다 죽으면 fail-closed. per_judge: Dict[str, Any] = {} judge_errors: Dict[str, str] = {} judge_usage: Dict[str, Any] = {} for judge in PICK_JUDGES: _sink: Dict[str, Any] = {} try: per_judge[judge] = call_structured( # ★권위를 **심판까지 운반한다** (Codex BLOCK 2). 안 넘기면 # 널리 알려진 작품의 고유 구조물이 「one-off signature # design」·「사진이 아님」으로 **반드시** 탈락한다 — 그 # 분기의 정상 입력인데도. "structure_form_ref_pick", load_pick_system(visual_authority=authority), parts, build_pick_schema(MAX_PICK_CANDIDATES), project_config={ "structure_form_ref_pick": {"model": judge}}, schema_name="structure_form_ref_pick", opik_metadata={"project_id": self.project_id, "episode_id": self.episode_id, "step": STEP_ID, "judge": judge}, # ★★**이중 판정에서 fallback 은 열려 있으면 안 된다** # (2026-08-27, #92). `call_structured` 의 Tier 3 은 # `{"model": "gpt"}` 로 간다 — `gemini-pro` 심판이 # 안전 오류로 떨어지면 **GPT 가 대신 답하는데 여기 # `per_judge["gemini-pro"]` 로 담긴다.** 두 번째 심판도 # GPT 계열이면 「이중 판정」이 실은 **같은 모델 두 번** # 이고, 합산은 그것을 독립 두 의견으로 평균한다. # ★Opik 6,000 span 실측: `gpt_fallback` 은 실제로 # 일어나지만(5회) **이 스텝에서 일어난 기록은 없다.** # 장치가 열려 있는 것을 닫는 것이지 사고를 수습하는 # 것이 아니다. # ★막으면 그 심판은 그냥 잃는다 — 기존 `single_judge`· # `rejected_judges` 가 그것을 드러낸다. 이름이 틀린 # 판정보다 **없는 판정이 정직하다.** enable_fallback=False, # ★★**이 자리는 이미 이중이다** (2026-08-27 Codex # 재리뷰). `PICK_JUDGES = ("gemini-pro", "gpt")` 로 # 두 심판을 도는데, `enable_fallback` 은 Tier 2/3 만 # 닫고 Router 는 `num_retries=3` 이라 **심판마다 최대 # 4번 전송**될 수 있다. 재시도는 다 과금되는데 # `usage_sink` 는 마지막 응답만 본다. # ★grok 교체를 기다릴 이유가 없다 — 지금도 두 심판이고 # 비용 위험은 같다. 내가 「아직 이중 아님」으로 잘못 # 분류하고 그것을 시험으로 못박았다. num_retries=0, usage_sink=_sink, ) judge_usage[judge] = _sink except Exception as exc: # noqa: BLE001 judge_errors[judge] = str(exc)[:300] judge_usage[judge] = _sink logger.warning("%s: %s 선택 심판 %s 실패 — %s", STEP_ID, group_id, judge, exc) if not per_judge: # ★**이미 낸 돈을 기록에서 버리면 안 된다** (2026-08-27 Codex # BLOCK). 응답을 받은 **뒤** JSON·schema 검증에서 떨어지면 # `usage_sink` 는 이미 토큰·비용·물리 모델을 담고 있는데, # 여기서 `verdict={}` 로 바로 나가면 그 값이 통째로 사라진다. # 「심판이 다 실패했다」와 「돈이 안 나갔다」는 다른 사건이다. fail_verdict: Dict[str, Any] = { "judges_used": [], "chosen_index": 0, "judge_errors": judge_errors, } if any(judge_usage.values()): fail_verdict["judge_usage"] = { j: dict(u) for j, u in judge_usage.items() if u} return {"status": "failed", "group_id": group_id, "error": ("선택 심판 전원 실패: " + "; ".join(f"{k}={v}" for k, v in judge_errors.items()))[:500], "candidate_count": len(candidates), "audit": build_audit_record( target_key=group_id, brief=brief, search=search, candidates=candidates, verdict=fail_verdict, chosen=None)} combined = combine_pick_verdicts(per_judge, len(candidates)) if judge_errors: combined["judge_errors"] = judge_errors # ★**무엇을 부르려 했나**와 **무엇이 답했나**를 갈라 남긴다 # (2026-08-27, #92). 심판 이름은 alias 이고, 응답이 말한 물리 # 모델은 다를 수 있다 — 판이 바뀌었는지(`gemini-3.1-pro-preview` # → 다음 판) 나중에 되짚으려면 그 값이 기록에 있어야 한다. # 토큰·비용도 같이 남긴다: 심판 하나가 얼마짜리인지 모르면 # 「둘을 부른다」의 값을 못 잰다. if judge_usage: combined["judge_usage"] = { j: dict(u) for j, u in judge_usage.items() if u} for c in candidates: per_c = (combined.get("per_candidate") or {}).get( str(c["index"])) or {} c["scores"] = {j: d.get("score") for j, d in per_c.items()} c["avg_score"] = (combined.get("averages") or {}).get( str(c["index"])) # 한 심판이라도 쓸 만하다고 본 후보는 갤러리에서 '가능' 으로 본다. c["usable"] = any(d.get("usable") for d in per_c.values()) c["reason_ko"] = " / ".join( f"{j}: {d.get('reason_ko')}" for j, d in sorted(per_c.items()) if d.get("reason_ko")) # 감사 기록은 합산 결과 + 심판별 원본을 모두 남긴다. verdict = dict(combined) verdict["verdicts"] = [ {"index": c["index"], "usable": c["usable"], "score": c.get("avg_score"), "reason_ko": c.get("reason_ko")} for c in candidates ] # ★"평균 0" 과 "유효한 심판이 하나도 없음" 은 다른 사건이다 — 같은 # 문구로 뭉치면 감사에서 원인을 되짚을 수 없다. if combined["chosen_index"]: verdict["chosen_reason_ko"] = ( f"이중 판정 평균 최고 " f"(심판 {', '.join(combined['judges_used'])})") elif not combined["judges_used"]: verdict["chosen_reason_ko"] = ( "유효한 심판 없음 — 제외 사유: " + "; ".join(f"{j}: {w}" for j, w in sorted(combined["rejected_judges"].items()))) else: verdict["chosen_reason_ko"] = "쓸 만한 후보 없음(평균 0)" verdict["per_judge_raw"] = per_judge chosen = next((c for c in candidates if c["index"] == int(combined.get("chosen_index") or 0)), None) audit = build_audit_record( target_key=group_id, brief=brief, search=search, candidates=candidates, verdict=verdict, chosen=chosen) logger.info("%s: %s", STEP_ID, summarize_for_log(audit)) if not chosen: # 필수 대상 fail-closed — 순수 T2I 자동 degrade 금지 return {"status": "failed", "group_id": group_id, "error": "쓸 만한 후보 없음 (VLM 선택 0)", "candidate_count": len(candidates), "audit": audit} # 5) 부속 대상 참조 — 조사에서 채택된 대상만, 각자 따로 검색한다. # 주 참조가 정해진 **뒤에** 돈다: 같은 사진을 두 몫으로 쓰지 # 않으려면 주 참조의 bytes 를 알고 있어야 한다. fitting_refs = self._fitting_refs( group_id=group_id, typology=typology, client=client, candidate_dir=candidate_dir, start_index=start_index + len(candidates), exclude_sha={chosen["sha256"]}) # ★자산 바인딩은 여기서 하지 않는다. 라운드 상태 전이와 UUID 는 # `_execute` 가 journal 과 함께 소유한다 — 검색 함수가 DB 를 같이 # 쥐고 있으면 "어느 라운드의 자산인가"가 두 곳에서 결정된다. # ★부속 참조는 **자산으로 묶지 않는다.** 라운드 manifest 가 예약하는 # UUID 는 하나뿐이고, 그것을 여러 개로 늘리는 것은 저장 계약 변경 # 이다. 지금은 파일 경로와 sha 를 CP·journal 에 남겨 감사 가능하게 # 두고, 계보 간선은 주 참조만 갖는다(거짓 간선을 만들지 않는다). return {"status": "ok", "group_id": group_id, # ★어느 권위로 골랐는지를 산출에 남긴다 — 심판 지시문이 # 권위마다 다르므로, 이것 없이는 이 참조가 「평범한 실물」로 # 뽑힌 것인지 「작품 정본」으로 뽑힌 것인지 되짚을 수 없다. "visual_authority": authority, "authority_reason_ko": str( brief.get("authority_reason_ko") or ""), "form_ref_path": chosen["path"], "form_ref_sha256": chosen["sha256"], "chosen_prompt_used": ( brief.get("search_directive_native") or "")[:2000], "candidate_count": len(candidates), "source_website_url": chosen.get("source_website_url"), "fitting_refs": fitting_refs, "typology": typology, "audit": audit} # ── 실행 ──────────────────────────────────────────────────── def _supplement_path(self) -> Path: from app.core.config import settings from app.modules.pipeline.grounding_outdoor_supplement import SUPPLEMENT_STEP_DIR return (Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / SUPPLEMENT_STEP_DIR / "manifest.json") def _load_supplement(self) -> Optional[Dict[str, Any]]: p = self._supplement_path() return json.loads(p.read_text(encoding="utf-8")) if p.is_file() else None def _supplement_structure_forms(self, front_cp, *, target_gids, building_groups, spec_groups, locations_by_id) -> Optional[Dict[str, Any]]: """필수 그룹의 장소마다 `structure_form` 의무를 **결정적으로** 세우고, 없는 것만 **중앙 조사기**로 산다. ★같은 공장(`reference_acquisition_step.make_search/make_download/make_judge/make_writer`) · 같은 `ca.run` 경계 · 스텝 전용 장부 · 별도 CP append-only. 사람 대기 없음.""" from app.core.config import settings from app.core.steps import reference_acquisition_step as ras from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import grounding_outdoor_supplement as gos from app.modules.pipeline.grounding_chunk_journal import ChunkJournal from app.modules.pipeline.outdoor_structure_seed import derive_seed_inputs from app.core.errors import AppError targets: List[Dict[str, Any]] = [] for gid in target_gids: members = ((building_groups.get(gid) or {}).get("members")) or [] locs = sorted({str((m or {}).get("loc_id") or (m or {}).get("location_id") or "") for m in members if isinstance(m, dict) and m.get("is_indoor") is False and ((m or {}).get("loc_id") or (m or {}).get("location_id"))}) spec = (spec_groups.get(gid) or {}).get("spec") or {} try: desc = str((derive_seed_inputs(spec=spec, building_group=building_groups.get(gid) or {}, locations_by_id=locations_by_id) or {}).get("structure_desc") or "") except Exception as exc: # noqa: BLE001 — 서술 결손은 아래 검증이 사유로 세운다 logger.warning("%s: %s structure_desc 파생 실패 — %s", STEP_ID, gid, exc) desc = "" targets.append({"gid": gid, "loc_ids": locs, "loc_id": locs[0] if len(locs) == 1 else "", "structure_desc": desc}) # ★★Codex BLOCK 3 (2026-09-03): 이미 아는 구조 오류(장소 0/여럿 · 서술 결손 · 중앙에 장소 줄 없음)는 # **provider 앞에서 한 번에** 세운다 — 다른 그룹을 산 뒤 투영에서 실패하면 돈만 쓴다. problems = gos.validate_targets(front_cp, targets) if problems: raise AppError( code="outdoor_form_reference.supplement_targets_invalid", message=("structure_form 보충 대상을 세울 수 없다 — provider 앞에서 선다. " f"{len(problems)}건: " + " | ".join(problems)), status_code=422, ) existing = self._load_supplement() # ★기존 보충 CP 는 의무를 막지 않는다 — 되쓰기/재시도/신원은 ca.run 과 장부가 판단한다 (Codex BLOCK 2) ledger = gos.structure_form_obligations(front_cp, targets) if not ledger["rows"]: return existing world = self._load_prev_checkpoint("visual_world_rules") or {} text_cp = self._load_prev_checkpoint("text_cleanup") or {} source_text = str((text_cp.get("data") or {}).get("cleaned_text") or (text_cp.get("data") or {}).get("text") or "") jpath = self._supplement_path().parent / "journal_supplement.json" jpath.parent.mkdir(parents=True, exist_ok=True) journal = ChunkJournal(jpath, contract={"wiring": ca.CONTRACT_VERSION, "supplement": gos.SUPPLEMENT_CONTRACT_VERSION}) n = len(ledger["rows"]) cap = n * max(1, int(ca.resolve_rounds(None))) workdir = Path(settings.projects_dir) / self.project_id / "references" / "grounding" / self.episode_id workdir.mkdir(parents=True, exist_ok=True) # ★물리 전송 문 — 중앙 스텝과 같은 셈(`TRANSMISSIONS_PER_LOGICAL_SLOT`). PR #82 리뷰 2026-09-03. from app.core.research_call_budget import research_calls_armed, research_run_scope from app.core.steps.reference_acquisition_step import ReferenceAcquisitionStep as _RA _raw_cap = (self.project_config or {}).get("reference_transmission_cap") _phys = int(_raw_cap) if _raw_cap else int(cap) * _RA.TRANSMISSIONS_PER_LOGICAL_SLOT with research_run_scope(cap=_phys): with research_calls_armed(): got = ca.run(ledger, journal=journal, cap=cap, workdir=workdir, rel_root=Path(settings.projects_dir).parent, search=ras.make_search(), download=ras.make_download(), judge=ras.make_judge(project_id=self.project_id, episode_id=self.episode_id, step_id=STEP_ID), write_brief=ras.make_writer(world=world, source_text=source_text, project_id=self.project_id, episode_id=self.episode_id, step_id=STEP_ID), stop_check=getattr(self, "_stop_check", None)) # ★사용자 중단 뒤 남은 대상을 안 산다 merged = gos.merge_supplement(existing, got.get("rows") or [], config_hash=self._config_hash()) from app.core.checkpoint_io import atomic_write_json p = self._supplement_path() p.parent.mkdir(parents=True, exist_ok=True) atomic_write_json(p, merged) # ★중간 crash 가 manifest 를 자르지 않게 logger.info("%s: structure_form 보충 %d개 삼 → %s", STEP_ID, n, p) return merged def _bind_projected_asset(self, *, group_id: str, loc: str, abs_png_path: str, sha256: str) -> str: """투영한 참조를 legacy 라운드와 **같은 자산 행**(ImageAsset · structure_form_ref/form_ref)으로 묶는다. ★실측 f7cc45c576c0 (2026-09-03 09:06 · 한 run 야외 통합): 투영은 path·sha·final_id 만 싣고 자산을 안 묶어 `form_ref_asset_id=None` → 씨드가 「참조 자산 id 가 없으면 계보를 남길 수 없다」로 섰다. id 는 `PROJECTED_ASSET_ID_RULE` 의 uuid5 — **같은 선택(같은 sha)은 재개마다 같은 행**(insert-or-verify · 새 UUID 금지), 다른 사진이면 새 행(기존 행은 고치지 않는다 — `_bind_ref_asset` 의 A4 원칙). bytes 복사 0 · 네트워크 0. 그룹 단위 commit — 뒤 그룹의 실패가 이 행을 되돌리지 못한다(legacy `_close_round` 와 같은 순서).""" from app.modules.pipeline.form_ref_rounds import projected_asset_id aid = projected_asset_id(project_id=str(self.project_id), episode_id=str(self.episode_id), group_id=group_id, location_id=loc, sha256=sha256) got = self._bind_ref_asset(group_id=group_id, asset_id=aid, abs_png_path=abs_png_path, sha256=sha256, prompt_used="") self.db.commit() return got def _project_front_checkpoint(self, front_cp, *, target_gids, universe, plate_gids, no_spec, building_groups, supplement_cp: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """앞쪽(중앙 조사) 산출을 **기존 야외 CP 모양으로 투영**한다. ★네트워크 0회. ★재검색·재선택·재다운로드·bytes 복사 **금지** (Codex). 같은 선택 참조를 가리키는 **계보 변환**이다 — 경로·SHA 는 sidecar·probe 와 같은 helper 로 센다. 계약 (설계 §6·§7·§8 · HITL 0): - 읽는 입구는 **날것** 중앙 CP + 보충 CP 의 merge view 다. 붙는 조건은 `usable_as_reference` (= 자동 선택 `outcome == selected`) 하나 — 사람 판정은 조건이 아니다. - 그룹 → 장소는 `building_groups[gid].members[].loc_id`(실외) 로 잇는다. **정확히 하나**여야 한다. 0개·여럿은 gid 와 loc_id 목록으로 선다(이름 대조 없음). - 중앙의 `context` 사진을 구조 형태 참조로 올리지 않는다 — **`structure_form` 의무**로 산 줄만 쓴다(`obligation_kind`). 없으면 선다: 그 장소에 있는 줄의 쓰임·상태를 사유에 적는다. - 필수 그룹에 쓸 수 있는 줄이 없으면 422 — 빈 CP 나 순수 T2I 로 낮추지 않는다. """ from pathlib import Path as _Path from app.core.config import settings from app.core.errors import AppError from app.modules.pipeline import reference_acquisition as _ra from app.modules.pipeline.grounding_reference_obligations import ( OBLIGATION_STRUCTURE_FORM) from app.modules.pipeline.grounding_sidecar_writer import ( resolved_reference_path, row_content_sha256) from app.modules.pipeline.search_grounded_ref import ( TARGET_POLICY_VERSION, resolve_ref_pack_version) if not ((front_cp or {}).get("data") or {}).get("rows"): raise AppError( code="outdoor_form_reference.front_output_missing", message=("앞쪽 reference_acquisition 이 소유자인데 산출이 없다 " "— 빈 CP 나 순수 T2I 로 낮추지 않는다. 앞쪽을 먼저 " "통과시켜라"), status_code=422, ) # ★HITL 0 (2026-09-03): 날것 CP — 사람 선택을 기다리지 않는다. ★merge view 하나(base + 보충). from app.modules.pipeline.grounding_outdoor_supplement import outdoor_reference_rows rows = outdoor_reference_rows(front_cp, supplement_cp) by_loc: Dict[str, List[Dict[str, Any]]] = {} for r in rows: led = r.get("ledger_row") or {} fid = str(led.get("final_id") or r.get("final_id") or "") if fid: by_loc.setdefault(fid, []).append(r) root = _Path(settings.projects_dir).parent groups: Dict[str, Any] = {} problems: List[str] = [] for gid in target_gids: members = ((building_groups.get(gid) or {}).get("members")) or [] locs = sorted({ str((m or {}).get("loc_id") or (m or {}).get("location_id") or "") for m in members if isinstance(m, dict) and m.get("is_indoor") is False and ((m or {}).get("loc_id") or (m or {}).get("location_id"))}) if len(locs) != 1: problems.append(f"{gid}: 실외 장소가 {locs or '없음'} — 정확히 하나여야 한다") continue loc = locs[0] have = by_loc.get(loc) or [] form_rows = [r for r in have if str((r.get("ledger_row") or {}).get("obligation_kind") or "") == OBLIGATION_STRUCTURE_FORM] if not form_rows: seen = sorted({ f"{str((r.get('ledger_row') or {}).get('purpose') or (r.get('ledger_row') or {}).get('obligation_kind') or '-')}" f"={str(r.get('outcome') or r.get('status') or '-')}" f"/{str(_ra.fidelity_of(r) or '-')}" for r in have}) problems.append( f"{gid}→{loc}: `{OBLIGATION_STRUCTURE_FORM}` 의무로 산 줄이 없다 " f"(있는 줄: {seen or '없음'}) — context 사진을 구조 형태 참조로 올리지 않는다") continue if len(form_rows) != 1: problems.append(f"{gid}→{loc}: `{OBLIGATION_STRUCTURE_FORM}` 줄이 {len(form_rows)}개다") continue row = form_rows[0] if not _ra.usable_as_reference(row): problems.append( f"{gid}→{loc}: 자동 선택된 사진이 없다 — " f"outcome={row.get('outcome') or row.get('status')!r} · why={str(row.get('why') or '')[:80]!r}") continue rel = resolved_reference_path(row) sha = row_content_sha256(row, root=root) if not rel or not sha: problems.append(f"{gid}→{loc}: 고른 사진의 경로·내용 해시가 없다 (path={rel!r})") continue acq = row.get("acquisition") or {} chosen = acq.get("chosen") or {} # ★씨드는 자산 id 없이는 그리지 않는다(outdoor_structure_seed:663 fail-closed) — legacy 라운드와 **같은 행 모양**으로 묶는다. asset_id = self._bind_projected_asset( group_id=gid, loc=loc, abs_png_path=str(root / rel), sha256=sha) groups[gid] = { "status": "ok", "group_id": gid, "form_ref_path": str(root / rel), "form_ref_sha256": sha, "form_ref_asset_id": asset_id, "chosen_prompt_used": "", "candidate_count": len(acq.get("candidates") or []), "source_website_url": chosen.get("source_website_url"), "fitting_refs": [], "typology": {"block": "", "items": [], "questions": []}, "audit": {"projected_from": "reference_acquisition", "final_id": loc, "asset_binding": PROJECTED_ASSET_ID_RULE, "research_subject_id": str(row.get("research_subject_id") or ""), "chosen_by": str(acq.get("chosen_by") or "model"), "match_quality": str(acq.get("match_quality") or "")}, } if problems: raise AppError( code="outdoor_form_reference.front_reference_unusable", message=("앞쪽(중앙 조사)이 소유자인데 필수 그룹에 쓸 수 있는 구조 형태 " f"참조가 없다 — 순수 T2I 로 낮추지 않는다. {len(problems)}건: " + " | ".join(problems)), status_code=422, ) return { "applicable_count": len(target_gids), "completed_count": len(groups), "failed_count": 0, "reused_count": 0, "round_replayed_count": 0, "projected_from": "reference_acquisition", "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": { "groups": groups, "mandatory_group_ids": list(target_gids), "target": { "policy_version": TARGET_POLICY_VERSION, "pack_version": resolve_ref_pack_version(), "universe_group_ids": list(universe), "plate_bound_group_ids": list(plate_gids), "no_spec_group_ids": list(no_spec), }, }, } def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.config import settings from app.core.steps.outdoor_structure_seed_step import ( collect_lane2_groups, partition_form_reference_targets, ) from app.modules.pipeline.outdoor_structure_seed import ( derive_seed_inputs, ) 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": {}}} # ★★★**이 주행에서 참조를 사는 자리는 하나뿐이다** (GROUNDING-V2 §2-4). # 앞쪽 `reference_acquisition`(13.8) 이 이미 샀으면 여기서 또 사면 # 같은 것을 두 번 산다. 그때 이 스텝은 **읽어서 투영만** 한다. # ★이관 중에는 앞쪽 산출이 없는 주행이 있다 — 그때는 **명시적 # legacy fallback** 으로 여기가 그대로 산다. 이 fallback 이 남아 # 있는 동안은 「이관 완료」라고 쓰지 않는다. from app.core.grounding_mode import resolve_grounding_mode from app.modules.pipeline.reference_acquisition import ( OWNER_OUTDOOR, FrontCheckpointMissing, acquisition_owner) # ★`project_config` 가 없는 자리(시험 대역·부분 조립)에서는 legacy 로 # 읽는다 — 없는 것을 v2 로 짐작하면 정상 주행이 통째로 막힌다. # 실제로 그렇게 해서 야외 시험 54건이 깨졌다. _front_cp = self._load_prev_checkpoint("reference_acquisition") try: _owner = acquisition_owner( resolve_grounding_mode(getattr(self, "project_config", None) or {}), front_checkpoint_exists=bool(_front_cp)) except FrontCheckpointMissing as exc: # ★사는 모드 + 앞쪽 CP 없음 = 의존이 안 돈 것. legacy 로 사지 않고 422 로 선다 (실측 4398a55dc0bb). from app.core.errors import AppError raise AppError(code="form_reference.front_missing", message=str(exc)[:500], status_code=422) # ★앞쪽이 소유자면 **대상 집합을 센 뒤** 읽어서 투영만 한다 (2026-09-03). # 대상(gid)·부모 장소(loc_id)는 lane plan·building_groups 에서 나오므로 # 그것들을 읽기 전에는 무엇을 투영할지 모른다. 네트워크는 여전히 0. _front_owner = _owner != OWNER_OUTDOOR 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 {} # 바인딩 대상 = 기존 계약(빼지 않는다). 판정 모집단 = seed 대상 전체. plate_gids = collect_lane2_groups(lane_data, all_groups=False) universe = sorted(set(collect_lane2_groups( lane_data, all_groups=bool(getattr( settings, "outdoor_seed_all_groups_enabled", False)), )) | set(plate_gids)) if not universe: logger.info("%s: 야외 그룹 없음 — no-op", STEP_ID) return _empty() spec_cp = self._load_prev_checkpoint("outdoor_place_spec") spec_groups = (spec_cp or {}).get("data", {}).get("groups", {}) 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) } scene_cp = self._load_prev_checkpoint("scene_save") # 씬 원문 = 검색 지시문 저작에서는 `source_language` 판정 전용(검색 # 의미에 기여 금지)이고, **사전조사에서는 1차 근거**다 — 두 쓰임의 # 관할이 다르므로 각각 어디에 실리는지 아래 호출부에서 갈린다. # ★자르지 않는다 (프로젝트 절대 규칙). scene_texts: Dict[int, str] = {} scene_headings: Dict[int, str] = {} ordered_texts: List[str] = [] for seg in (scene_cp or {}).get("data", {}).get("segments", []) or []: ordered_texts.append(seg.get("text") or "") si = seg.get("scene_index") if isinstance(si, int): scene_texts[si] = seg.get("text") or "" scene_headings[si] = seg.get("heading") or "" source_text = "\n\n".join(t for t in ordered_texts if t) # 지역·시대 확정 사실 — outdoor_place_canon 과 동일 배선 from app.core.world_context import build_world_facts_block world_facts_block = build_world_facts_block( self._load_prev_checkpoint("visual_world_rules")) or "" from app.core.steps.shot_conti_light_step import ( _resolve_openai_client, ) client = _resolve_openai_client() # ── 검색 대상 = seed 대상 exact parity (v3, 2026-07-30) ────── # 사용자 확정: "야외 등 **모든 요소 배경**이 참조" / "전 그룹 무조건 # 검색". 판정 장치를 두지 않는다 — 판정은 배제할 수 있는 구조이고, # 실제로 그것 때문에 실패했다. v1 은 `structure_plate` 바인딩을 # 프록시로 썼는데 그건 구조물의 성질이 아니라 **샷의 필요성**이라 # 금월도 2고에서 25그룹 중 10만 검색을 탔다(주유소·편의점·파출소 등 # 15그룹이 순수 T2I). v2 는 LLM scope 판정을 합집합으로 얹었으나 # ①판정 입력에 작품 고유명사가 실렸고 ②validator 가 근거 없는 NO 를 # 통과시켜 default-YES 가 강제되지 않았다(Codex 리뷰 6a). # # **v3 = 소비자(seed)와 같은 집합을 쓴다.** `universe` 는 seed 와 # 동일한 `collect_lane2_groups(..., all_groups=<같은 플래그>)` 산출 # 이므로 어긋날 수가 없다. 프록시도 판정도 없으니 조용히 빠지는 # 경로 자체가 사라진다. # # ★[2026-08-01 Codex 2차 재리뷰 BLOCKING 1] 분할을 이 자리에서 직접 # 계산하면 소비자와 **두 벌**이 된다. 소비자는 제 계산이 없으니 CP 에 # 적힌 분할을 믿을 수밖에 없고, 그래서 구 CP 의 stale no_spec 이 # 참조 없는 생성을 열었다. 분할은 공유 helper 하나가 소유한다. # # spec 결손 그룹은 검색어를 저작할 근거가 없다. 임의로 넣으면 뒤에서 # structure_desc 결손으로 확정 실패한다 — 명시 기록만 남긴다(seed 가 # 같은 결손을 "outdoor_place_spec 부재" 로 다시 세운다). # ★조용히 빼는 것이 아니라 CP 에 남긴다. target_gids, no_spec = partition_form_reference_targets( universe=universe, spec_groups=spec_groups) logger.info( "%s: 대상 %d = seed 대상 전 그룹(모집단 %d) · spec 결손 제외 %d " "· 그중 structure_plate 바인딩 %d", STEP_ID, len(target_gids), len(universe), len(no_spec), len(set(plate_gids) & set(target_gids))) if no_spec: logger.warning( "%s: place spec 결손으로 검색 대상에서 빠진 그룹 %d — %s", STEP_ID, len(no_spec), no_spec[:8]) if _front_owner: # ★★보충 획득 (2026-09-03 · 설계 §8): 필수 그룹의 장소에 `structure_form` 줄이 없으면 # **중앙 조사기와 같은 경계**로 산다(5단계 · HITL 0) → 별도 CP(append-only) → merge view. _supp_cp = self._supplement_structure_forms( _front_cp, target_gids=list(target_gids), building_groups=building_groups, spec_groups=spec_groups, locations_by_id=locations_by_id) return self._project_front_checkpoint( _front_cp, target_gids=list(target_gids), universe=list(universe), plate_gids=list(plate_gids), no_spec=list(no_spec), building_groups=building_groups, supplement_cp=_supp_cp) import hashlib as _hashlib source_text_sha = _hashlib.sha256( source_text.encode("utf-8")).hexdigest()[:16] prev_cp = self.load_checkpoint() or {} # ★legacy 판정의 유일한 기준 — entry 의 필드 유무가 아니다. prev_schema = prev_cp.get("schema_version", MISSING) prev_groups = (prev_cp.get("data") or {}).get("groups") or {} groups: Dict[str, Any] = {} done = failed = reused = replayed = generative = 0 for gid in target_gids: spec = (spec_groups.get(gid) or {}).get("spec") or {} # 라운드가 정해지기 전에 나는 실패(입력 결손·지문·CP 대조)도 있다. # 그때는 붙일 라운드 표기가 없다 — 빈 값으로 시작한다. round_where: Dict[str, Any] = {} try: seed_inputs = derive_seed_inputs( spec=spec, building_group=building_groups.get(gid) or {}, locations_by_id=locations_by_id) desc = (seed_inputs or {}).get("structure_desc") or "" if not desc.strip(): raise ValueError("structure_desc 결손") # 사전조사 입력 = 이 그룹의 씬 원문(무절단) + 명세 항목. # 씬 배정은 lane plan 이 소유한다 — 소비자(seed)와 같은 출처를 # 써야 두 스텝이 다른 씬을 보고 다른 것을 조사하지 않는다. g_scene_indices = sorted( int(si) for si in ((lane_data.get("groups") or {}).get(gid) or {} ).get("scene_indices") or [] if isinstance(si, (int, float, str)) and str(si).lstrip("-").isdigit() ) scene_blocks = [ (f"[SCENE {si}] " f"{(scene_headings.get(si) or '').strip()}").rstrip() + "\n" + (scene_texts.get(si) or "") for si in g_scene_indices ] fingerprint = self._group_fingerprint( structure_desc=desc, world_facts_block=world_facts_block, source_text_sha=source_text_sha, # 조사가 읽는 것 전부 — 씬 원문 · 명세 항목(인용 포함) · # 실내·외 설명. 하나라도 바뀌면 조사가 달라진다. research_input_sha=_hashlib.sha256( json.dumps( [scene_blocks, (spec.get("items") or []), (seed_inputs or {}).get("interior_note_en") or "", (seed_inputs or {}).get("exterior_note_en") or ""], sort_keys=True, ensure_ascii=False, default=str).encode("utf-8") ).hexdigest()[:16]) group_dir = self._ref_dir(gid) prior = prev_groups.get(gid) or {} if mode != "force" and self._reusable(prior, fingerprint): # 검색은 네트워크·비결정·비용 수명이 다르다. 입력과 # 계약이 그대로면 재검색하지 않는다 (설계 §9). marks = _round_marks(prior, cp_schema=prev_schema) if marks["round_id"] != LEGACY_ROUND_ID: # ★라운드 계약 CP 는 **journal 이 권위**다. CP 만 보고 # 넘기면 journal 손상·결손과 자산 row 결손이 조용히 # 건너뛰어진다. self._verify_round_cp( group_dir=group_dir, gid=gid, prior=prior, round_id=str(marks["round_id"])) else: # ★legacy 예외는 **journal 부재**에만 적용된다. # 자산 검증까지 건너뛰면 계보가 가리키는 row 가 # 사라진 CP 를 그대로 성공으로 내보낸다. self._verify_bound_asset( group_id=gid, asset_id=str(prior.get("form_ref_asset_id") or ""), abs_png_path=str(prior.get("form_ref_path") or ""), sha256=str(prior.get("form_ref_sha256") or "")) groups[gid] = dict(prior, group_fingerprint=fingerprint, reused=True, **marks) done += 1 reused += 1 logger.info("%s: %s 재사용 (지문 일치·파일 검증 통과)", STEP_ID, gid) continue # ── 라운드 해결 (A4 §2.3 행렬) ──────────────────────── resolution = resolve_round( group_dir, input_fp=fingerprint, force=(mode == "force"), # ★역사 기록을 여는 순간 고정한다. 나중에 현재 값과 # 비교하지 않는다 — 무효화 판단은 지문의 몫이다. provenance=self._round_provenance()) rnd = resolution.round round_where = { "round_id": rnd.round_id, "path_kind": ROUND_PATH_KIND, "round_contract_version": ROUND_STORAGE_CONTRACT_VERSION, } if resolution.reused_finalized: # ★검색·판정·다운로드·asset INSERT **0회**. 이 경로가 # 없으면 `_execute` 가 끝난 뒤 `save_checkpoint` 직전에 # 끊긴 실행이 유료 전체 재실행으로 바뀐다. rec = dict(rnd.result) # ★journal 이 완료라고 해도 **자산이 실재하는지** 본다. # 두 저장소 사이에 원자성이 없어 자산만 사라질 수 있다. # ★단 사진 없이 끝난 확정에는 검증할 자산이 없다 — 거기서 # 자산을 요구하면 생성 정본 그룹이 재개마다 선다. if rnd.has_reference: self._verify_bound_asset( group_id=gid, asset_id=str(rec.get("form_ref_asset_id") or ""), abs_png_path=str(rec.get("form_ref_path") or ""), sha256=str(rec.get("form_ref_sha256") or "")) rec["group_fingerprint"] = fingerprint rec["round_replayed"] = True groups[gid] = rec done += 1 replayed += 1 logger.info( "%s: %s 완료 라운드 %s 재투영 (유료 0콜)", STEP_ID, gid, rnd.round_id) continue if rnd.state in BOUND_STATES and rnd.result: # ★선택이 이미 durable 하다 — 자산 바인딩/마감만 남았다. # 여기서 다시 검색하면 끊긴 지점마다 유료가 되풀이된다. rec = dict(rnd.result) logger.info( "%s: %s 라운드 %s 를 %s 에서 이어간다 (유료 0콜)", STEP_ID, gid, rnd.round_id, rnd.state.value) else: cand_dir = candidate_dir(group_dir, rnd.round_id) # ★이어가는 라운드에는 이미 후보가 있을 수 있다. 1번부터 # 다시 쓰면 **그 라운드의 이전 후보를 덮어쓴다**. start = next_candidate_index(group_dir, rnd.round_id) # ★조사는 그룹당 한 번이다. `_run_group` 안에 두면 좁혀 # 재검색하는 2차 시도가 같은 조사를 유료로 되풀이한다. typology = self._pre_research( group_id=gid, structure_desc=desc, interior_note_en=(seed_inputs or {}).get( "interior_note_en") or "", exterior_note_en=(seed_inputs or {}).get( "exterior_note_en") or "", scene_blocks=scene_blocks, spec_items=list(spec.get("items") or []), world_facts_block=world_facts_block, client=client) rec = self._run_group( group_id=gid, structure_desc=desc, pick_structure_desc=(seed_inputs or {}).get( "structure_names_en") or "", world_facts_block=world_facts_block, source_text=source_text, client=client, candidate_dir=cand_dir, start_index=start, typology=typology) # ★생성 확정은 **재시도 대상이 아니다** (Codex BLOCK 1 # 2026-09-08). 「사진을 못 찾았다」가 아니라 「맞출 실사 # 정본이 없다」는 확정이라, 좁혀 다시 물으면 브리핑을 한 # 번 더 사고 둘째 답이 외부 정본으로 뒤집혀 **첫 확정이 # 무효가 된다** — 그때는 실제 검색이나 그룹 실패까지 간다. if rec.get("status") not in ("ok", GENERATIVE_STATUS): # 실측(금월도 2고): 복합 서술이라 질의가 하위 요소로 # 분산돼 엉뚱한 주제를 회수한다. 판정을 푸는 게 아니라 # **주 구조물 하나로 좁혀** 1회 재검색한다. logger.info("%s: %s 1차 실패 — 주 구조물로 좁혀 재검색", STEP_ID, gid) rec2 = self._run_group( group_id=gid, structure_desc=desc, pick_structure_desc=(seed_inputs or {}).get( "structure_names_en") or "", world_facts_block=world_facts_block, source_text=source_text, client=client, candidate_dir=cand_dir, # 1차 후보 뒤에 이어 붙인다 — 덮어쓰면 그 라운드의 # 감사 근거가 사라진다. start_index=start + int( rec.get("candidate_count") or 0), narrow=True, typology=typology) rec2["first_attempt"] = { "status": rec.get("status"), "error": rec.get("error"), "audit": (rec.get("audit") or {}), } rec2["candidate_count"] = ( int(rec.get("candidate_count") or 0) + int(rec2.get("candidate_count") or 0)) rec = rec2 # 실패 entry 에도 조사 기록은 남긴다 — 조사가 무엇을 # 남겼고 무엇을 버렸는지가 실패 귀속의 근거다. rec.setdefault("typology", typology) rec["group_fingerprint"] = fingerprint rec.update(self._close_round( group_dir=group_dir, rnd=rnd, gid=gid, rec=rec)) except Exception as exc: # noqa: BLE001 logger.warning("%s: %s 실패 — %s", STEP_ID, gid, exc) # ★그룹 단위 격리 — flush 실패로 세션이 롤백 상태가 되면 # 뒤 그룹이 전부 "transaction has been rolled back" 로 연쇄 # 실패한다(실측: 첫 그룹 실패가 나머지 9개를 죽였다). try: self.db.rollback() except Exception: # noqa: BLE001 pass # ★계약 위반(AppError)은 **code 까지** 남긴다. 메시지만 남기면 # journal 손상·라운드 충돌·자산 불일치가 전부 "그룹 실패"로 # 뭉뚱그려져 원인을 되짚을 수 없다. code = getattr(exc, "code", None) # ★라운드가 이미 정해진 뒤의 실패(자산 바인딩·마감 오류)에서도 # **어느 라운드였는지**는 남긴다. 새 dict 로 갈아치우면 그 # 감사값이 사라져 남은 후보의 출처를 되짚을 수 없다. rec = {"status": "failed", "group_id": gid, "error": (f"{code}: {exc}" if code else str(exc))[:500], **round_where} if code: rec["error_code"] = str(code) groups[gid] = rec # ★[v12] `generative` 는 「사진을 못 찾았다」가 아니라 「맞출 실사 # 정본이 없어 생성이 정본이다」라는 **다른 확정**이다. 실패로 세면 # 창작 세계관 대본이 매번 여기서 선다. if rec.get("status") in ("ok", GENERATIVE_STATUS): done += 1 if rec.get("status") == GENERATIVE_STATUS: generative += 1 else: failed += 1 self.db.commit() from app.modules.pipeline.search_grounded_ref import ( TARGET_POLICY_VERSION, resolve_ref_pack_version, ) return { "applicable_count": len(target_gids), "completed_count": done, "failed_count": failed, # ★[v12] 완료 중 **생성이 정본인** 그룹 수. `completed_count` 에 # 섞여 있으면 「사진을 몇 개 샀나」를 되짚을 수 없다. "generative_count": generative, "reused_count": reused, # 완료 라운드를 유료 호출 없이 다시 투영한 그룹 수 — "재사용"과 # 다른 사건이라 따로 센다(전자는 CP 기반, 이쪽은 journal 기반). "round_replayed_count": replayed, "schema_version": SCHEMA_VERSION, "config_hash": self._config_hash(), "data": { "groups": groups, # 소비자(seed) 의 fail-closed 기준. "mandatory_group_ids": list(target_gids), # 대상 감사 — 판정이 없어도 **왜 이 집합인지**는 남긴다. # 특히 `no_spec_group_ids` 는 유일하게 빠지는 경로라 # 육안·후속 점검이 바로 볼 수 있어야 한다. "target": { "policy_version": TARGET_POLICY_VERSION, "pack_version": resolve_ref_pack_version(), "universe_group_ids": list(universe), "plate_bound_group_ids": list(plate_gids), "no_spec_group_ids": list(no_spec), }, }, } def _sha_file(p: Path) -> str: import hashlib return hashlib.sha256(p.read_bytes()).hexdigest() def _pre_research_digest(entry: Any) -> str: """사전조사·부속 참조의 결정론 해시 — journal 과 CP 를 대조하는 값. ★두 필드를 `ROUND_CP_PROJECTION_KEYS` 에 넣지 않는 이유: 그 목록은 `_validate_result_shape` 가 **완료 산출의 필수 키**로도 쓴다. 넣으면 이 필드가 없던 기존 완료 라운드가 전부 "투영 필드 없음"으로 죽는다. 두 쪽 모두 키가 없으면 같은 값이 나온다 — 이 계약 이전 산출을 거짓 불일치로 만들지 않는다. """ import hashlib src = entry if isinstance(entry, dict) else {} payload = {"typology": src.get("typology"), "fitting_refs": src.get("fitting_refs")} return hashlib.sha256( json.dumps(payload, sort_keys=True, ensure_ascii=False, default=str).encode("utf-8") ).hexdigest()[:16]