"""정책 owner(character·prop)의 **사람이 확인한** 조사 사진 → canonical ref 생성의 실제 입력. ★★★실측 (Codex BLOCK 2026-09-02): 중앙 정책(`episode_reference_policy`)은 verified C/P 를 `required_refs` 의 **ID 로만** 올렸다. 그런데 `ReferencePhase1Service` 의 `extra_references` 는 의존 엔티티(변형/기본형)의 ref map 만 봤다 — 중앙 조사에서 받은 사진 bytes 를 읽는 길이 **없었다**. 그래서 P01 의 사람이 확인한 사진은 「P01 참조가 필요하다」는 boolean 만 만들고, 실제 P01 canonical ref 는 그 사진 **없이** 다시 생성됐다. 조사를 했는데 결과가 그림에 안 닿는 결함이다. 이 모듈은 그 사진을 canonical ref 생성의 **labeled extra reference** 로 잇는다. 결속 규칙: - 키는 typed id 하나 — 중앙 CP 의 `ledger_row.final_id` == `EntityCanon.short_id`. 이름·부분문자열 결속 없음. - 갈래는 계약이 정한다 — `canonical_ref_owner_types()` (= 정책 집행 갈래). 배경 sidecar 갈래(location·location_part)는 여기 **안 든다** — 그 갈래의 집행자는 sidecar 하나다. - 붙여도 되는가는 `reference_acquisition.usable_as_reference` **한 곳**이 판단한다 (selected ∧ 사람이 verified). unverified·rejected·unavailable → 0장. - 사진 bytes·sha 는 sidecar·probe 와 **같은 helper** 로 읽는다 (`grounding_sidecar_writer.row_content_sha256` · `resolved_reference_path`). - 이미지 호출 수를 늘리지 않는다 — 같은 한 번의 생성 호출에 입력만 더 실린다. 사진이 없으면 기존 C/P 생성 동작은 한 바이트도 안 바뀐다. """ from __future__ import annotations import logging from pathlib import Path from typing import Any, Dict, List, Optional logger = logging.getLogger(__name__) #: 생성 호출에 실리는 참조 라벨 — **역할** 라벨이다(이름·장소·시대 없음). 기존 #: `"Reference: …"` · `"Base form — KEEP THIS FACE: …"` 와 같은 자리에 선다. GROUNDING_REFERENCE_LABEL = ( "Real-world grounding photo of this subject (human-verified) — " "match its actual form, materials and proportions:") #: 자산에 남기는 출처 칸 — `ImageAsset.pipeline_metadata_json["grounding_inputs"]`. GROUNDING_INPUTS_KEY = "grounding_inputs" #: 자산·참조 체크포인트 행 **둘 다**에 남기는 입력 지문 칸. GROUNDING_INPUT_DIGEST_KEY = "grounding_input_digest" #: 지문 계약판 — 지문에 접는 좌표(final_id · acquisition_identity · content_sha256)가 #: 바뀌면 올린다. 옛 지문과 새 지문이 같은 값을 낼 수 없게. GROUNDING_INPUT_DIGEST_CONTRACT = "1.202609021800" CENTRAL_STEP_ID = "reference_acquisition" class GroundingInputError(RuntimeError): """중앙 CP 의 verified 줄이 canonical ref 입력이 되기에 모자란다.""" def grounding_input_digest(refs: Any) -> str: """한 subject 의 **입력 지문** — 결정적. ★「입력 없음」도 **명시 값**이다. ★★★Codex BLOCK (2026-09-02, 재개 계약): verified 사진이 생성 입력이 됐는데 resume 은 「primary 자산 파일이 있다」만 보고 already_done 으로 넣었다. 그러면 P01 을 나중에 verified 로 바꾸거나 사진 SHA 가 바뀌어도 옛 canonical ref 가 영구히 primary 다. 입력의 좌표를 정렬해 지문으로 접고, 자산과 CP 행 둘 다에 남기며, resume 은 둘이 지금 값과 같을 때만 되쓴다. """ import hashlib import json rows = sorted((str(g.get("final_id") or ""), str(g.get("acquisition_identity") or ""), str(g.get("content_sha256") or "")) for g in (refs or ())) payload = {"contract": GROUNDING_INPUT_DIGEST_CONTRACT, "inputs": [list(r) for r in rows]} return hashlib.sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()[:16] #: 입력이 하나도 없을 때의 지문 — 옛 자산(칸 없음)과 **다르게** 읽힌다(아래 `is_fresh`). EMPTY_INPUT_DIGEST = grounding_input_digest([]) def is_fresh(stored: Optional[str], current: str) -> bool: """저장된 지문이 지금 입력과 같은가. ★옛 자산(지문 칸이 없다)은 **지금 입력이 없을 때만** 신선하다 — 이 기능 전에 구운 legacy/v2 프로젝트의 참조 수백 장을 resume 마다 다시 굽지 않는다. 지금 입력이 있으면(사람이 확인한 사진) 옛 자산은 stale 이다 — 그 사진 없이 구워진 것이므로. 저장된 지문이 있으면 **정확히 같아야** 한다: grounded 자산이 rejected 로 바뀌면 지금 지문은 EMPTY 라 달라져 다시 굽는다. """ if not stored: return current == EMPTY_INPUT_DIGEST return str(stored) == str(current) def stored_digest_of_asset(asset: Any) -> Optional[str]: """`ImageAsset.pipeline_metadata_json` 에 남긴 지문. 없으면 None.""" import json raw = getattr(asset, "pipeline_metadata_json", None) if not raw: return None try: parsed = json.loads(raw) except (ValueError, TypeError): return None if not isinstance(parsed, dict): return None got = parsed.get(GROUNDING_INPUT_DIGEST_KEY) return str(got) if got else None def project_grounding_mode(db: Any, project_id: str) -> str: """이 프로젝트의 grounding 모드 — production 의 같은 두 함수로.""" from app.core.grounding_mode import resolve_grounding_mode from app.services.step_execution_service import _load_project_config return resolve_grounding_mode(_load_project_config(db, str(project_id)) or {}) def central_cp_is_required(grounding_mode: str) -> bool: """중앙 조사가 **반드시 돈** 판인가 — `v2_chunk` 만. legacy/v2 는 CP 없음 → 옛 길.""" from app.core.grounding_mode import uses_chunk_producer return bool(uses_chunk_producer(str(grounding_mode))) class _CheckpointReader: """`central_cp_with_reviews(runner)` 가 읽는 **네 속성만** 가진 읽기 전용 모양. ★스텝이 아닌 자리(orchestrator — RefImageGenStep 과 API 가 **둘 다** 여기로 온다)에서 production 과 같은 reader 를 쓰기 위한 어댑터다. 판정을 얹는 법은 `grounding_fidelity_review.central_cp_with_reviews` 한 곳이 갖고 있다 — 여기서 CP 를 따로 읽고 판정을 따로 얹으면 두 벌이 된다. """ def __init__(self, db: Any, project_id: str, episode_id: str) -> None: self.db = db self.project_id = str(project_id) self.episode_id = str(episode_id) def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]: import json from app.core.config import settings p = (Path(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / step_id / "manifest.json") if not p.is_file(): return None return json.loads(p.read_text(encoding="utf-8")) def verified_policy_references(db: Any, *, project_id: str, episode_id: str, root: Optional[Path] = None, grounding_mode: Optional[str] = None, fail_closed: Optional[bool] = None, ) -> Dict[str, List[Dict[str, Any]]]: """final_id → 그 subject 의 **사람이 확인한** 사진들(bytes 포함). ★`v2_chunk` 판(`fail_closed`)에서 중앙 CP 가 없거나 깨졌으면 **provider 앞에서 선다** — public `/generate-reference-images` 가 중앙 조사 전에 눌리면 고증 없는 canonical ref 가 만들어지고 resume 이 그것을 영구 재사용한다 (Codex BLOCK 2026-09-02 E). legacy/v2 는 CP 없음 → `{}` 옛 길 그대로. ★판정 기록을 **못 읽으면** 선다(`ReviewsUnreadable`) — 「없다」가 아니다. """ from app.modules.pipeline import grounding_sidecar_writer as sw from app.modules.pipeline import reference_acquisition as ra from app.modules.pipeline.grounding_entity_contract import ( canonical_ref_owner_types, ) if fail_closed is None: mode = grounding_mode or project_grounding_mode(db, project_id) fail_closed = central_cp_is_required(mode) reader = _CheckpointReader(db, project_id, episode_id) # ★HITL 0 (2026-09-03): 날것 CP — 사람 판정 표는 production 의존성이 아니다 cp = reader._load_prev_checkpoint("reference_acquisition") if fail_closed: sw.assert_central_checkpoint(cp) # ★부재·손상이면 여기서 선다 if cp is None: return {} owners = set(canonical_ref_owner_types()) base = Path(root) if root is not None else sw.default_reference_root() out: Dict[str, List[Dict[str, Any]]] = {} for row in ((cp.get("data") or {}).get("rows") or ()): ledger = row.get("ledger_row") or {} if str(ledger.get("owner_type") or "") not in owners: continue # ★sidecar·아웃룩 갈래 — 집행자가 다르다 if not ra.usable_as_reference(row): continue # ★outcome selected 만 (HITL 0) — 한 곳의 판단 final_id = str(ledger.get("final_id") or "") if not final_id: raise GroundingInputError( f"{row.get('research_subject_id')!r} 는 verified 인데 `final_id` 가 " "없다 — 어느 canonical ref 에 실을지 모르는 채 안 싣는다") rel = sw.resolved_reference_path(row) sha = sw.row_content_sha256(row, root=base) path = base / rel if rel else None if not rel or not sha or path is None or not path.is_file(): raise GroundingInputError( f"{final_id} 는 verified 인데 사진 파일이 없다 (path={rel!r}) — " "판정은 그 bytes 위에 선 것이라 다른 것을 대신 싣지 않는다") out.setdefault(final_id, []).append({ "final_id": final_id, "identity": str(row.get("research_subject_id") or ""), "acquisition_identity": str(row.get("identity") or ""), "content_sha256": sha, "path": str(path), "bytes": path.read_bytes(), }) if out: logger.info("grounding canonical inputs: %s", {k: [g["content_sha256"][:12] for g in v] for k, v in out.items()}) return out def provenance_of(refs: List[Dict[str, Any]]) -> List[Dict[str, str]]: """자산에 남길 출처 — bytes 를 뺀 좌표만.""" return [{k: str(g.get(k) or "") for k in ("final_id", "identity", "acquisition_identity", "content_sha256")} for g in refs] def digests_by_final_id(grounding_references: Dict[str, List[Dict[str, Any]]] ) -> Dict[str, str]: """final_id → 입력 지문. 없는 subject 는 부르는 쪽이 `EMPTY_INPUT_DIGEST` 로 읽는다.""" return {fid: grounding_input_digest(refs) for fid, refs in (grounding_references or {}).items()} def canonical_grounding_digest(db: Any, *, project_id: str, episode_id: str, grounding_mode: str) -> str: """이 에피소드 canonical ref 입력 **전체**의 지문 — `RefImageGenStep._config_hash` 가 `v2_chunk` 에서 접는다(Codex D). ★여기서는 CP 부재에 서지 않는다 — 지문은 늘 계산돼야 하고(부재는 `"central_cp": "absent"` 로 접힌다), 서는 문은 생성 입구(`verified_policy_references(fail_closed=True)`)다.""" import hashlib import json reader = _CheckpointReader(db, project_id, episode_id) cp = reader._load_prev_checkpoint("reference_acquisition") # ★HITL 0 — 날것 CP if cp is None: payload: Dict[str, Any] = {"contract": GROUNDING_INPUT_DIGEST_CONTRACT, "central_cp": "absent"} else: refs = verified_policy_references(db, project_id=project_id, episode_id=episode_id, grounding_mode=grounding_mode, fail_closed=False) payload = {"contract": GROUNDING_INPUT_DIGEST_CONTRACT, "by_final_id": dict(sorted(digests_by_final_id(refs).items()))} return hashlib.sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()[:16]