"""★참조 획득 — **엔티티마다 참고 사진을 찾아 온다.** GROUNDING-V2 §2-4. 사용자 설계(2026-08-30): 이미지 모델이 만들기 어려운 것은 한 번만 나와도 엔티티에 등록하고, 그 엔티티는 **검색 → 이미지 검색**을 탄다. 사전 검색은 판정이 아니라 **「무엇을 찾을지」 자료를 모으는 것**이다. VLM 은 **「우리가 찾던 종류가 맞나」만** 본다. 없으면 좁혀서 한 번 더. ★★**새로 만드는 것이 아니다.** 검색·안전 다운로드·선택 primitive 는 `search_grounded_ref` 에 이미 있고 지금은 **야외 구조물(21.915)** 만 쓴다. 이 모듈은 그 **호출 소유권을 앞으로 옮긴다** — primitive 를 복사하지 않는다. 호출자는 딱 하나여야 한다. 앞뒤 두 곳이 각각 사면 같은 것을 두 번 산다. """ from __future__ import annotations import hashlib import logging from typing import Any, Dict, Optional logger = logging.getLogger(__name__) ACQUISITION_CONTRACT_VERSION = 1 STEP_NAME = "reference_acquisition" def acquisition_contract_sha(*, coarse_version: Optional[str] = None, ref_version: Optional[str] = None, rounds: int) -> str: """★★**참조 신원을 지배하는 것 전부**의 해시. 한 곳에서만 만든다. 왜 한 곳인가 — 앞쪽 스텝과 `image_steps` 가 **각각 산식을 적으면** 한쪽만 고쳐진다. 그러면 팩을 바꿔도 바깥 이미지 스텝이 stale 이 안 돼 **옛 참조가 영구 봉인**된다(`image_steps` 주석에 그 사고가 적혀 있다). 접는 것: - **거친 종류 선택 팩** — 지문·스키마·좁힘 문안. 무엇을 고르는지를 지배한다 - **검색·다운로드 계약** — `search_grounded_ref` 의 brief/질의 저작 팩 - **라운드 계약** — 라운드 수·라운드당 후보 상한. 바뀌면 다른 실험이다 - **계약 버전** — 이 모듈이 무엇을 뜻하는지 ★안 접는 것: **어느 엔티티가 이 길을 타는가**(`generation_difficulty`). 그것은 대상 집합을 정할 뿐 **한 대상의 참조 산출에 기여하지 않는다** — 접으면 다른 대상의 난이도가 바뀌었다고 이미 뽑아 둔 참조가 무효가 된다. """ from app.modules.pipeline import coarse_type_pick as ctp from app.modules.pipeline import search_grounded_ref as sgr from app.modules.prompt_loader import resolve_effective h = hashlib.sha256() h.update(str(ACQUISITION_CONTRACT_VERSION).encode()) h.update(b"\x00") cv = coarse_version or ctp.PROMPT_PACK_VERSION for stem in (ctp.SYSTEM_STEM, ctp.SCHEMA_STEM, ctp.NARROW_STEM): # ★★`load_prompt` 는 **schema 를 못 읽는다.** 그것으로 스키마를 부르면 # `FileNotFoundError` 가 나고 위 `except` 가 **빈 값으로 흡수**해서 # 스키마가 **조용히 안 접혔다** — 스키마를 바꿔도 참조가 재사용된다. # ★Codex 가 요구한 「stem 마다 한 바이트」 control 이 이걸 잡았다. # 「없는 팩과 견주기」로는 셋이 다 빈 값이 되어 **못 가른다**. # → `coarse_type_pick.load_pack` 이 쓰는 **같은 함수**로 읽는다. kind = "schema" if stem == ctp.SCHEMA_STEM else "prompt" try: got = resolve_effective(ctp.PROMPT_MODULE, stem, kind=kind, version=cv) h.update(str((got or {}).get("content") or "").encode("utf-8")) except FileNotFoundError: # ★없는 stem 만 빈 값으로 흡수한다. 다른 고장(loader 손상·권한)은 # 올린다 — 전부 삼키면 서로 다른 고장이 같은 해시가 된다. h.update(b"") h.update(b"\x00") # ★검색·다운로드 쪽은 기존 계약을 **그대로 접는다.** 다시 세지 않는다. h.update(sgr.search_contract_sha( ref_version or sgr.REF_PACK_VERSION).encode()) h.update(b"\x00") # ★라운드 계약도 산출을 지배한다 — 4장에서 고른 것과 12장에서 고른 것은 # 같은 참조가 아니고, **한 번 찾은 것과 좁혀 두 번 찾은 것도** 다르다. # # ★★`rounds` 를 **호출부가 준다.** 상수를 그대로 접었더니 키는 2라운드 # 계약을 접었는데 **실제 late 경로는 1라운드**여서 어긋났다(Codex). # 키는 **그 호출이 실제로 하는 것**을 접어야 한다 — 안 그러면 「2라운드로 # 찾은 참조」와 「1라운드로 찾은 참조」가 같은 키를 갖는다. # 이관이 끝나면 중앙 하나만 남고 값도 하나가 된다. # ★★**필수**다. 기본값을 두면 「이 호출이 실제로 몇 번 찾는가」와 키가 # 갈린다. ★그리고 `int()` 변환도 **안 한다** — `"1"` 과 `True` 가 # 1 로 통과한다(Codex BLOCK-1). 실제로 몇 번 찾았는지는 **정수**로만 # 말할 수 있다. if type(rounds) is not int or not 1 <= rounds <= ctp.MAX_ROUNDS: raise ValueError( f"rounds 는 1..{ctp.MAX_ROUNDS} 의 **정수**여야 한다 " f"(받은 값 {rounds!r})") n = rounds h.update(f"{n}:{ctp.PER_ROUND_CAP}:" f"{n * ctp.PER_ROUND_CAP}".encode()) return h.hexdigest()[:16] #: 최종 상태 — `coarse_type_pick.decide_next_round` 의 `next` 와 짝이다. STATUS_SELECTED = "selected" #: ★두 라운드 다 정상으로 돌았는데 종류가 맞는 것이 없다. **terminal** 이다. STATUS_NO_MATCH = "no_match_after_retry" #: ★provider·다운로드·시간·예산 때문에 **다 못 봤다.** 「없다」가 아니다. STATUS_RETRYABLE = "retryable" #: 참조 없이 내려가도 되는가 — **그렇다** (사용자 확정 2026-08-31). #: #: ★★「궁극적 목적은 자동화이니 HITL을 무조건 필요한 요소로 하면 안 된다. #: 차후 UI에서 수동 수정하게 고칠 예정이니 지금은 무조건 HITL이 없어야 한다.」 #: #: 앞 계약은 `False` 였고 「없으면 그냥 만들자 override 를 두지 않는다」였다. #: 그러면 참조를 못 구한 대상에서 **주행이 사람을 기다린다** — 그것이 HITL 이다. #: #: 이제는 못 구해도 **멈추지 않는다**. 다만 — #: #: ①`reference_unavailable` 을 **durable 하게 남긴다** #: ②**잘못된 부류의 사진을 억지로 쓰지 않는다** #: ③검색이 모은 **글 근거는 생성에 계속 쓴다** #: #: 참조가 있었나 없었나는 기록에 남으므로, 나중에 UI 에서 사람이 고칠 수 있다 #: — 그것은 **생성 뒤 선택**이지 실행 선행조건이 아니다. ALLOW_MISSING_REFERENCE = True #: 참조를 못 구했다 — **자동으로 그대로 간다**. STATUS_UNAVAILABLE = "reference_unavailable" #: ★★raw 상태의 **닫힌 목록**. 새 값을 더하면 그 값이 지나는 술어를 전부 #: 훑어야 한다 — 앞에 새 enum 값이 옛 술어를 그대로 통과한 적이 있다. #: ★여기 한 벌만 적는다. 부르는 쪽이 각자 목록을 만들면 한쪽만 고쳐진다. KNOWN_RAW_STATUSES: tuple = (STATUS_SELECTED, STATUS_NO_MATCH, STATUS_RETRYABLE, STATUS_UNAVAILABLE) #: **다 보고 끝난** 상태들. ★`retryable` 은 여기 없다 — 그것은 「못 봤다」다. #: ★★★이 표가 없어서 「없다」로 끝난 줄을 **재개마다 다시 판정**했다 #: (실측 2026-09-02: `rejudge_pending` 이 5 로 안 줄었다). 같은 것을 두 번 #: 사지 않으려면 「다 보고 없었다」가 **끝난 것**이어야 한다. TERMINAL_RAW_STATUSES: tuple = (STATUS_SELECTED, STATUS_NO_MATCH) #: ★★★**고증 축은 coarse 판정과 다른 축이다** (Codex BLOCK 2026-09-02). #: coarse 심판은 「무엇인가 · 보이는가」만 본다 — 시대·나라·정확성은 **안 #: 묻는다**(사용자 확정 08-31). 그러니 `selected` 는 「그 종류의 사진을 #: 하나 골랐다」일 뿐 **「그 시대·그 지역 것이 맞다」가 아니다**. #: 이 둘을 한 칸으로 두면 `selected` 5장이 sidecar 로 그대로 승격된다. FIDELITY_UNVERIFIED = "unverified" FIDELITY_VERIFIED = "verified" FIDELITY_REJECTED = "rejected" KNOWN_FIDELITY: tuple = (FIDELITY_UNVERIFIED, FIDELITY_VERIFIED, FIDELITY_REJECTED) #: 고증 축의 계약. ★coarse 계약과 **따로** 올린다. FIDELITY_CONTRACT_VERSION = "1.202609020700" def fidelity_of(row: Dict[str, Any]) -> str: """이 줄의 **고증 축**. ★없으면 `unverified` — 기본이 「아직 아니다」다. ★모르는 값은 `unverified` 로 접는다. 「모르는 것」이 통과가 되면 안 된다. """ got = str(((row or {}).get("grounding_fidelity") or {}).get("state") or "") return got if got in KNOWN_FIDELITY else FIDELITY_UNVERIFIED def usable_as_reference(row: Dict[str, Any]) -> bool: """하류가 **참조로 써도 되나**. ★HITL 0 — 자동 선택(`outcome == selected`) **하나**면 된다. (아래 옛 문단은 2026-09-02 의 「두 축」 계약 기록이다 — 2026-09-03 사용자 재지시로 뒤집혔다.) ★★★`selected` 만 보면 안 된다 (Codex BLOCK 2026-09-02): 그것은 coarse 심판이 「그 종류의 사진」이라고 한 것일 뿐이다. 고증이 확인되지 않은 것을 참조로 붙이면, 사람이 못 본 사진이 그림의 근거가 된다. ★확인이 안 됐다고 **파이프라인을 막지 않는다** — 참조 **없이** 내려간다 (`reference_unavailable` 과 같은 대접). HITL 0 이 이 판의 계약이다. ★어느 칸을 보나: 소비자가 보는 것은 **접은 값**(`outcome`)이다. 그것이 없는 줄이면 raw `status` 를 접어서 본다 — 두 벌로 적지 않는다. (실측 2026-09-02: sidecar 투영 줄은 `outcome` 만 들고 있어 `status` 만 보던 판이 **verified 인 줄까지 안 붙였다**.) """ got = str((row or {}).get("outcome") or "") if not got: got = acquisition_outcome(str((row or {}).get("status") or "")) # ★★★HITL 0 — 최상위 불변식 (사용자 2026-09-03 재지시 · Codex 정정으로 옛 요구 폐기): # 참조는 **자동 선택 한 장만으로 붙는다**. 사람 판정/선택 행이 0건이어도 붙어야 # 하고, `verified` 부재가 attachment 0 · blocked · partial 어느 것으로도 작동하면 # 실패다. 앞 판은 `verified` 를 요구해 검토 75건 뒤에도 usable 1/18 이었다 — # 사람을 파이프라인 안의 문으로 세운 것. 사람 판정은 canary·A/B·사후 평가 도구 # (`grounding_fidelity_review.apply_reviews`)에서만 값을 바꾼다 — production # 소비자는 이 함수 하나로 묻고, 이 함수는 판정 칸을 **읽지 않는다**. return got == STATUS_SELECTED def is_terminal(status: str) -> bool: """그 줄이 **끝났나**. ★「없다」와 「못 봤다」를 가른다.""" return str(status) in TERMINAL_RAW_STATUSES def downstream_blocked(status: str) -> bool: """이 대상을 하류가 **막아야 하는가** — ★**아니다**. ★사람 대기를 만들지 않는다 (사용자 확정 2026-08-31). 못 구한 것은 `reference_unavailable` 로 남기고 **참조 없이 자동으로** 내려간다. ★★이 함수는 **늘 False** 다. 남겨 두는 까닭은 호출부가 이미 이것을 묻고 있고, 지우면 그 자리들이 각자 판단하게 되기 때문이다. 「막을지」를 묻는 자리는 여기 하나여야 나중에 정책이 또 바뀌어도 한 곳만 고친다. """ return False def acquisition_outcome(status: str) -> str: """하류에 넘길 **처지**. ★사람이 볼 상태를 production 에 안 만든다. selected 참조를 쓴다 reference_unavailable 참조 없이 간다 (못 구했다 · 다 못 봤다) ★`no_match_after_retry` 와 `retryable` 을 **둘 다** unavailable 로 접는다. 둘의 차이는 **기록에 그대로 남지만**, 하류는 어느 쪽이든 참조 없이 간다. """ return (STATUS_SELECTED if status == STATUS_SELECTED else STATUS_UNAVAILABLE) # ───────────────────────────────────────────────────────────────────── # 소유권 — **한 주행에서 참조를 사는 자리는 하나뿐이다** # ───────────────────────────────────────────────────────────────────── OWNER_FRONT = "reference_acquisition" # 13.8 — 앞쪽 공용 producer OWNER_OUTDOOR = "outdoor_structure_form_reference" # 21.915 — 야외 legacy class FrontCheckpointMissing(RuntimeError): """사는 모드인데 중앙 CP 가 없다 — 의존이 안 돈 것이지 legacy 주행이 아니다.""" def acquisition_owner(mode: str, *, front_checkpoint_exists: bool) -> str: """이 주행에서 **누가 참조를 사는가**. ★★앞뒤가 각각 사면 **같은 것을 두 번 산다.** 그래서 소유자를 한 곳으로 못박고, 소유자가 아닌 자리는 **네트워크를 0회** 쓴다. 이관 중에는 갈래가 둘이다 (Codex): - v2 이고 **앞쪽 산출이 있으면** → 앞쪽이 소유자. 야외 스텝은 그것을 **읽어서 기존 CP 모양으로 투영만** 한다 - 그 밖에는 야외 스텝이 그대로 산다 (**명시적 legacy fallback**). ★이 fallback 이 남아 있는 동안은 「이관 완료」라고 쓰지 않는다 ★`front_checkpoint_exists` 를 **명시로 받는다.** 「v2 면 앞쪽이 있겠지」로 유추하면, 앞쪽이 아직 안 돈 주행에서 야외가 조용히 아무것도 못 사고 순수 T2I 로 내려간다. """ # ★★manifest 술어(`_if_grounding_reference`)와 **같은 함수**를 본다 (2026-09-03). # 앞 판은 `buys_v2_research`(= v2 만)를 봐서 `v2_chunk` 에선 중앙 CP 가 있어도 # 소유자가 outdoor 로 나와 같은 장소를 **다시 샀다**. from app.core.grounding_mode import buys_reference if buys_reference(mode): if not front_checkpoint_exists: # ★★★실측 canary 4398a55dc0bb (2026-09-03): 닫힘에 중앙 스텝이 없어 앞쪽 CP 가 없었고, 여기가 # OWNER_OUTDOOR 를 돌려줘 야외 스텝이 **legacy 로 5라운드를 샀다**(보충 경로 0회). 사는 모드에서 # 앞쪽이 없는 것은 「legacy 주행」이 아니라 **의존이 안 돈 것**이다 — 사지 않고 선다. raise FrontCheckpointMissing( f"grounding_mode={mode!r} 는 중앙(reference_acquisition)이 참조를 사는 모드인데 앞쪽 CP 가 없다 — " "의존 스텝이 안 돌았다. 야외 스텝은 legacy 로 사지 않는다(이중 구매 금지)") return OWNER_FRONT return OWNER_OUTDOOR def assert_not_buying(owner: str, where: str) -> None: """소유자가 아닌 자리에서 사려 하면 **선다**. ★조용히 지나가면 이중 구매가 되고, 그건 「몇 번 샀나」를 통째로 거짓말로 만든다. 늦게 죽는 것보다 여기서 드러나는 편이 낫다. """ if owner != where: from app.core.errors import AppError raise AppError( code="reference_acquisition.not_the_owner", message=(f"이 주행의 참조 소유자는 {owner!r} 인데 {where!r} 가 " "사려 한다 — 앞뒤가 각각 사면 같은 것을 두 번 산다"), status_code=409, )