"""사람이 **그 사진이 그 시대·그 지역 것인지** 정한 기록. ★DB 가 정본. coarse 심판은 「무엇인가 · 보이는가」만 본다 — 시대·나라·정확성은 **안 묻는다** (사용자 확정 2026-08-31). 그러니 `selected` 는 「그 종류의 사진을 골랐다」일 뿐이고, 그 사진이 맞는지는 **사람만** 정한다. ★★★계약 (Codex 2026-09-02) — · 정본은 **PostgreSQL 의 append-only 표**다. `artifact/` 의 HTML·JSON 은 지워도 되는 **화면 산출**이지 결정이 아니다 · 결정은 **입력 신원**에 결속된다 — 사진 SHA · acquisition identity · 요구 좌표(era/region). 하나라도 바뀌면 옛 결정은 **안 따라간다** · 정정은 `UPDATE` 가 아니라 `supersedes_id` 를 든 **새 행** · 서로 모순되는 미정리 최신 행이 둘이면 **미확인 + 감사 결함**으로 fail-safe 한다 — HITL 때문에 파이프라인을 막지는 않는다 · CP 는 **투영만** 한다. 사람이 CP 를 직접 고치는 길은 만들지 않는다 """ from __future__ import annotations import hashlib import json from typing import Any, Dict, List, Optional, Sequence #: 이 화면·이 판정 계약의 판. ★화면이 바뀌면 올린다. REVIEW_CONTRACT_VERSION = "1.202609020900" VERDICT_VERIFIED = "verified" VERDICT_REJECTED = "rejected" KNOWN_VERDICTS = (VERDICT_VERIFIED, VERDICT_REJECTED) #: 입력 신원에 접히는 칸들. ★여기 없는 것은 결정을 바꾸지 않는다. INPUT_FIELDS = ("project_id", "episode_id", "research_subject_id", "final_id", "purpose", "acquisition_identity", "image_sha256", "era", "region", "source_url", "query_provenance_digest") class ReviewInputBroken(ValueError): """판정 입력이 모자라다. ★모르는 채로 결정을 안 받는다.""" class ReviewConflict(RuntimeError): """서로 모순되는 최신 결정이 둘이다. ★사람이 정리해야 한다.""" def canonical_input(payload: Dict[str, Any]) -> str: """판정 입력의 **정본 bytes**. ★hash 는 이 문자열에서만 나온다. ★없는 좌표는 **빈 값**이다 — 만들지 않는다. 그래서 「시대 선언이 없는 현대 원고」와 「시대가 있는데 안 실은 판」이 **다른 신원**이 된다. Raises: ReviewInputBroken: 신원 칸이 비었다 — 무엇을 본 결정인지 모른다. """ got = {k: str(payload.get(k) or "") for k in INPUT_FIELDS} for must in ("project_id", "episode_id", "research_subject_id", "acquisition_identity", "image_sha256"): if not got[must]: raise ReviewInputBroken( f"{must} 가 비었다 — 무엇을 본 결정인지 모르는 채로 안 적는다") return json.dumps(got, ensure_ascii=False, sort_keys=True, separators=(",", ":")) def input_hash(payload: Dict[str, Any]) -> str: """그 입력의 신원. ★사진·구매판·좌표 중 하나만 바뀌어도 달라진다.""" return hashlib.sha256( canonical_input(payload).encode("utf-8")).hexdigest()[:32] def provenance_digest(rounds: Sequence[Dict[str, Any]]) -> str: """사람이 본 **질의 출처 표시**의 지문. ★화면이 보여 준 것과 결속한다. 우리가 준 질의와 provider 가 더한 질의를 **갈라 센 수**를 접는다 — 화면이 다른 것을 보여 주게 바뀌면 옛 결정이 안 따라간다. """ rows = [] for rd in rounds or (): prov = (rd or {}).get("query_provenance") or {} rows.append({"round": int((rd or {}).get("round_no") or 0), "provided": int(prov.get("provided") or 0), "expanded": int(prov.get("expanded") or 0)}) return hashlib.sha256( json.dumps(rows, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()[:16] def effective_verdict(rows: Sequence[Dict[str, Any]], *, want_hash: str ) -> Dict[str, Any]: """이 입력의 **유효 결정 하나**. 없거나 어긋나면 미확인이다. ★`supersedes_id` 를 **실제로 해석한다** — 대신된 행은 죽는다. ★살아 있는 최신 행이 **둘 이상 서로 다르면** 미확인 + 감사 결함이다. 「둘 중 아무거나」로 고르면 사람이 안 정한 것을 정한 것으로 만든다. Returns: `{"verdict", "row", "fault"}` — `verdict` 는 `verified`/`rejected`/ `None`(미확인). `fault` 는 왜 못 정했는지. """ mine = [r for r in (rows or ()) if str(r.get("review_input_hash") or "") == str(want_hash)] if not mine: return {"verdict": None, "row": None, "fault": "이 입력에 대한 결정이 없다"} dead = {str(r.get("supersedes_id")) for r in mine if r.get("supersedes_id")} live = [r for r in mine if str(r.get("id")) not in dead] if not live: return {"verdict": None, "row": None, "fault": "살아 있는 결정이 없다 — 전부 대신되었다"} kinds = {str(r.get("verdict") or "") for r in live} if len(live) > 1 and len(kinds) > 1: return {"verdict": None, "row": None, "fault": (f"서로 모순되는 최신 결정이 {len(live)}개다 " f"({sorted(kinds)}) — 사람이 정리해야 한다")} got = sorted(live, key=lambda r: str(r.get("reviewed_at") or ""))[-1] kind = str(got.get("verdict") or "") if kind not in KNOWN_VERDICTS: return {"verdict": None, "row": got, "fault": f"모르는 판정 {kind!r} — 미확인으로 접는다"} return {"verdict": kind, "row": got, "fault": ""} def fidelity_state(rows: Sequence[Dict[str, Any]], *, want_hash: str ) -> Dict[str, Any]: """DB 결정 → CP 에 적을 **고증 축**. ★투영이지 판단이 아니다.""" from app.modules.pipeline import reference_acquisition as ra got = effective_verdict(rows, want_hash=want_hash) kind = got["verdict"] state = (ra.FIDELITY_VERIFIED if kind == VERDICT_VERIFIED else ra.FIDELITY_REJECTED if kind == VERDICT_REJECTED else ra.FIDELITY_UNVERIFIED) out: Dict[str, Any] = { "state": state, "contract": ra.FIDELITY_CONTRACT_VERSION, "review_contract": REVIEW_CONTRACT_VERSION, "review_input_hash": str(want_hash), } if got["row"] is not None: out["review_id"] = str(got["row"].get("id") or "") out["reviewed_at"] = str(got["row"].get("reviewed_at") or "") out["reviewer_actor"] = str(got["row"].get("reviewer_actor") or "") if got["fault"]: out["fault"] = got["fault"] return out def assert_observed_matches(payload: Dict[str, Any], observed: Optional[Dict[str, Any]]) -> None: """사람이 **실제로 본 것**이 이 입력과 같나. ★다르면 안 받는다. 화면이 옛 사진을 캐시해 보여 주고 서버가 새 사진에 결정을 붙이면, 사람이 안 본 것을 승인한 것이 된다. Raises: ReviewInputBroken: 본 것과 붙일 것이 다르다. """ seen = dict(observed or {}) for axis in ("image_sha256", "era", "region"): want = str(payload.get(axis) or "") got = str(seen.get(axis) or "") if got != want: raise ReviewInputBroken( f"사람이 본 {axis} 가 {got!r} 인데 붙일 것은 {want!r} 다 — " f"안 본 것을 승인할 수 없다") def review_inputs_for(cp: Optional[Dict[str, Any]], *, project_id: str, episode_id: str, coordinates: Dict[str, str], sha_of, strict: bool = True) -> List[Dict[str, Any]]: """이 CP 에서 **사람이 볼 것들**과 그 입력 신원. ★한 곳이 만든다. 화면·서버 끝점·CP 투영이 **같은 함수**를 쓴다. 세 곳이 각각 조립하면 지문이 갈리고, 그러면 사람이 본 것과 붙는 것이 달라진다. Args: coordinates: **런타임 구조화 칸**(`visual_world_rules.era`/`.region`) 에서 온 선언 좌표. ★여기서 짐작하지 않는다 — 저작기가 쓴 것과 **같은 SOT** 여야 사람이 본 것과 붙는 것이 같아진다. 없으면 빈 값이고 **만들지 않는다**. sha_of: `(row) -> str` — 그 줄의 **사진 내용 지문**. 파일을 여는 일은 부르는 쪽이 한다(이 모듈은 디스크를 안 본다). Raises: ReviewInputBroken: 기록에 적힌 요구 축과 준 좌표가 어긋난다. Args (이어서): strict: 참이면 신원을 못 만드는 줄에서 **선다**. 거짓이면 그 줄을 `fault` 와 함께 돌려준다. ★★★왜 가르나 (2026-09-02): 사진 파일 하나가 없다고 **파이프라인 전체**를 죽이면, 멀쩡한 다른 컷까지 못 그린다. 그 줄은 어차피 **안 붙는다**(미확인) — 위험한 것은 「확인된 것이 조용히 안 붙는 것」과 「안 본 것이 붙는 것」이고 둘 다 안 일어난다. 반대로 **DB·CP 를 통째로 못 읽는 것**은 확인된 것이 전부 사라지는 일이라 거기서는 선다. Returns: `[{row, candidate, payload, hash, chosen_path, model_pick, rounds}]` — **행 × 후보**. 받아 둔 사진이 있는 줄은 모델이 골랐든 아니든 후보마다 한 줄씩이다(Codex BLOCK 3). `strict=False` 에서 신원을 못 만든 줄은 `{row, fault}` 만 갖는다. """ out: List[Dict[str, Any]] = [] for r in ((cp or {}).get("data") or {}).get("rows") or (): acq = r.get("acquisition") or {} cands = candidates_of(r) if not cands: continue # ★받아 둔 사진이 없다 — 사람이 볼 것이 없다 led = r.get("ledger_row") or {} rounds = list(acq.get("rounds") or ()) coords = {k: str(v or "").strip() for k, v in (coordinates or {}).items()} try: _assert_axes_agree(rounds, coords) except ReviewInputBroken as exc: if strict: raise out.append({"row": r, "fault": str(exc)}) continue for cand in cands: # ★후보 하나를 「고른 사진」자리에 앉힌 행 그림자 — sha 계산은 production 의 # `row_content_sha256(row)` 그대로다(같은 함수 · 같은 칸). view = {**r, "acquisition": {**acq, "chosen": cand, "chosen_path": cand.get("path")}} payload = { "project_id": str(project_id), "episode_id": str(episode_id), "research_subject_id": str(r.get("research_subject_id") or ""), "final_id": str(led.get("final_id") or ""), "purpose": str(led.get("purpose") or ""), "acquisition_identity": str(r.get("identity") or ""), "image_sha256": str(sha_of(view) or ""), "era": coords.get("era", ""), "region": coords.get("region", ""), "source_url": str(cand.get("url") or ""), "query_provenance_digest": provenance_digest(rounds), } try: got = input_hash(payload) except ReviewInputBroken as exc: if strict: raise out.append({"row": r, "fault": str(exc), "candidate": cand}) continue out.append({"row": r, "candidate": cand, "payload": payload, "hash": got, "chosen_path": str(cand.get("path") or ""), "model_pick": bool(cand.get("model_pick")), "rounds": rounds}) return out def candidates_of(row: Dict[str, Any]) -> List[Dict[str, Any]]: """한 줄에서 **사람이 볼 사진 전부** — 받아 둔 후보(라운드 순 · 경로로 중복 제거) 에 모델이 고른 것을 더한다. ★★Codex BLOCK 3 (2026-09-02 밤): 앞 판은 모델(VLM)이 고른 한 장만 사람 앞에 갔다 — `unsure` 하나면 후보 전부가 사람 눈에 안 닿았다(O01 넉 장이 그렇게 사라졌다). 모델 선택은 **표시**(`model_pick` · `model_eligible`)일 뿐 문이 아니다. """ acq = row.get("acquisition") or {} chosen = acq.get("chosen") or {} chosen_path = str(chosen.get("path") or acq.get("chosen_path") or "") out: List[Dict[str, Any]] = [] seen: set = set() for rd in (acq.get("rounds") or ()): eligible = {int(x) for x in (rd.get("eligible") or ()) if str(x).isdigit()} for c in (rd.get("downloaded_candidates") or ()): path = str((c or {}).get("path") or "") if not path or path in seen: continue seen.add(path) idx = (c or {}).get("index") out.append({**c, "path": path, "round_no": rd.get("round_no"), "model_pick": path == chosen_path, "model_eligible": (int(idx) in eligible) if isinstance(idx, int) else None}) if chosen_path and chosen_path not in seen: out.append({**chosen, "path": chosen_path, "round_no": None, "model_pick": True, "model_eligible": True}) return out def _assert_axes_agree(rounds: Sequence[Dict[str, Any]], coords: Dict[str, str]) -> None: """그때 **요구됐던 축**과 지금 준 좌표가 같나. ★다르면 안 적는다. 나간 기록은 어느 축이 의무였는지를 안다(`coordinates_carried.declared`). 지금 세계 규칙이 바뀌었다면 그때 본 것과 다른 것을 승인하게 된다. ★옛 CP 는 그 칸이 없다 — 그때는 검사할 것이 없으니 지나간다. """ for rd in reversed(list(rounds or ())): got = (rd or {}).get("coordinates_carried") or {} decl = got.get("declared") if not isinstance(decl, (list, tuple)): continue want = {str(x) for x in decl} now = {k for k, v in (coords or {}).items() if str(v or "").strip()} if want != now: raise ReviewInputBroken( f"그때 요구된 축은 {sorted(want)} 인데 지금 준 좌표는 " f"{sorted(now)} 다 — 그때 본 것과 다른 것을 승인할 수 없다") return def apply_reviews(cp: Dict[str, Any], reviews: Sequence[Dict[str, Any]], *, project_id: str, episode_id: str, coordinates: Dict[str, str], sha_of, selections: Sequence[Dict[str, Any]] = ()) -> Dict[str, Any]: """DB 의 유효 결정을 CP 에 **투영한다**. ★판단이 아니라 옮기기다. 결정 없음·불일치·깨짐은 전부 `unverified` 다. `rejected` 도 안 붙는다. ★★후보별 판정 + 대상당 선택 하나 (Codex BLOCK 3 · 재리뷰 2026-09-02 밤): - 판정은 후보 한 장마다 「그 시대·그 곳 것인가」다 — **여러 장이 verified 여도 옳다**. - 어느 장을 쓸지는 `selections`(대상당 살아 있는 선택 하나)가 정한다. 선택된 후보가 verified 면 그것이 `acquisition.chosen`(`chosen_by: human` · 모델 선택은 `model_pick` 보존) · `outcome/status = selected` — 모델이 `no_match` 라 했어도 참조가 된다. - 선택이 없고 verified 가 **딱 하나**면 그것을 쓴다(고를 것이 하나뿐이다). - 선택이 없고 verified 가 여럿이면 미확인 + 「한 장 고르세요」— 기록은 다 유효하다. - 선택된 후보가 verified 가 아니면(거절·미판정) 미확인 + 사유. 확인이 없고 모델 선택이 `rejected` 면 `rejected`. 나머지는 미확인. """ import copy from app.modules.pipeline import reference_acquisition as ra got = copy.deepcopy(cp or {}) # ★한 줄이 깨졌다고 판 전체를 죽이지 않는다 — 그 줄만 미확인 + 까닭. # ★DB·CP 를 통째로 못 읽는 것은 **부르는 쪽**이 이미 세웠다. by_row: Dict[int, List[Dict[str, Any]]] = {} for x in review_inputs_for(got, project_id=project_id, episode_id=episode_id, coordinates=coordinates, sha_of=sha_of, strict=False): by_row.setdefault(id(x["row"]), []).append(x) for r in ((got.get("data") or {}).get("rows") or ()): inputs = by_row.get(id(r)) if not inputs: continue faults = [x for x in inputs if x.get("fault")] if faults and len(faults) == len(inputs): r["grounding_fidelity"] = { "state": ra.FIDELITY_UNVERIFIED, "contract": ra.FIDELITY_CONTRACT_VERSION, "review_contract": REVIEW_CONTRACT_VERSION, "fault": faults[0]["fault"]} continue live = [x for x in inputs if not x.get("fault")] verdicts = {x["hash"]: effective_verdict(reviews, want_hash=x["hash"]) for x in live} verified = [x for x in live if verdicts[x["hash"]]["verdict"] == VERDICT_VERIFIED] pick = next((x for x in live if x.get("model_pick")), None) sid = str(r.get("research_subject_id") or "") sel = effective_selection(selections, subject_id=sid) chosen_input = None sel_fault = "" if sel is None and len(live_selections(selections, subject_id=sid)) > 1: sel_fault = "이 대상에 살아 있는 선택이 둘 이상이다 — 다시 고르면 하나가 된다" if sel is not None: chosen_input = next((x for x in live if x["hash"] == str(sel.get("review_input_hash") or "")), None) if chosen_input is None: sel_fault = "고른 사진이 지금 후보에 없다 — 사진이나 좌표가 달라졌다" elif chosen_input not in verified: kind = verdicts[chosen_input["hash"]]["verdict"] sel_fault = ("고른 사진이 rejected 다 — 다른 장을 고르거나 판정을 바꿔야 한다" if kind == VERDICT_REJECTED else "고른 사진이 아직 verified 가 아니다") chosen_input = None elif len(verified) == 1: chosen_input = verified[0] elif len(verified) > 1: # ★★HITL 0 (사용자 2026-09-03): 「맞다」가 여럿이고 고른 장이 없으면 **다시 묻지 # 않는다** — 모델이 고른 장이 그 안에 있으면 그것, 아니면 후보 차례의 첫 장. # 앞 판은 「한 장 고르세요」로 미확인에 세워 두었다(실측: 75건 뒤 usable 1/18). chosen_input = pick if (pick is not None and pick in verified) else verified[0] if chosen_input is not None: one = chosen_input acq = dict(r.get("acquisition") or {}) cand = dict(one["candidate"]) was = acq.get("chosen") human_changed = not cand.get("model_pick") acq["chosen"] = {k: v for k, v in cand.items() if k not in ("model_pick", "model_eligible", "round_no")} acq["chosen_path"] = cand.get("path") acq["status"] = ra.STATUS_SELECTED acq["outcome"] = ra.STATUS_SELECTED acq["chosen_by"] = "human" if human_changed else "model+human" if human_changed: acq["model_pick"] = was # ★모델이 골랐던 것은 지우지 않는다 r["acquisition"] = acq r["outcome"] = ra.STATUS_SELECTED r["grounding_fidelity"] = fidelity_state(reviews, want_hash=one["hash"]) if sel is not None: r["grounding_fidelity"]["selection_id"] = str(sel.get("id") or "") if sel_fault: r["grounding_fidelity"]["fault"] = sel_fault # ★기록의 어긋남은 남긴다 — 막지는 않는다 continue if sel_fault: r["grounding_fidelity"] = { "state": ra.FIDELITY_UNVERIFIED, "contract": ra.FIDELITY_CONTRACT_VERSION, "review_contract": REVIEW_CONTRACT_VERSION, "selection_id": str((sel or {}).get("id") or ""), "fault": sel_fault} continue if pick is not None: # ★모델 선택이 있고 사람이 그것을 거절했거나 아직 안 봤다 — 그 신원의 상태 그대로 r["grounding_fidelity"] = fidelity_state(reviews, want_hash=pick["hash"]) if sel_fault: r["grounding_fidelity"]["fault"] = sel_fault if verdicts.get(pick["hash"], {}).get("verdict") == VERDICT_REJECTED: # ★★도구의 override (HITL 0 · 2026-09-03): 사람이 모델 선택을 **거절**했고 다른 # 맞다가 없다 — production 은 판정 칸을 안 읽으므로 `outcome` 자체를 옮긴다. # canary·A/B 도구가 이 산출을 먹일 때만 반영된다. 기본 주행은 이 함수를 안 부른다. acq = dict(r.get("acquisition") or {}) acq.update({"status": ra.STATUS_NO_MATCH, "outcome": ra.STATUS_UNAVAILABLE, "rejected_by_human": True}) r["acquisition"] = acq r["status"] = ra.STATUS_NO_MATCH r["outcome"] = ra.STATUS_UNAVAILABLE r["why"] = "사람이 모델 선택을 거절했고 다른 확인된 후보가 없다 (도구 override)" continue r["grounding_fidelity"] = { "state": ra.FIDELITY_UNVERIFIED, "contract": ra.FIDELITY_CONTRACT_VERSION, "review_contract": REVIEW_CONTRACT_VERSION, "fault": f"받아 둔 후보 {len(live)}장을 사람이 아직 안 골랐다"} return got class ReviewsUnreadable(RuntimeError): """판정 기록을 못 읽었다. ★「없다」가 **아니다** — 모르는 것이다.""" def read_selections(db: Any, *, project_id: str, episode_id: str ) -> List[Dict[str, Any]]: """이 에피소드의 **쓸 사진 선택** 전부. ★못 읽으면 선다 — 「선택이 없다」가 아니다.""" from app.models.project import GroundingReferenceSelection as S try: rows = (db.query(S) .filter_by(project_id=str(project_id), episode_id=str(episode_id)).all()) except Exception as exc: # noqa: BLE001 raise ReviewsUnreadable( f"사진 선택 기록을 못 읽었다 ({exc}) — 「선택이 없다」가 아니라 **모른다**는 " f"뜻이다. `alembic upgrade head` 를 먼저 보라") from exc return [{"id": r.id, "research_subject_id": r.research_subject_id, "review_input_hash": r.review_input_hash, "selected_at": r.selected_at, "selected_by": r.selected_by, "supersedes_id": r.supersedes_id} for r in rows] def effective_selection(selections: Sequence[Dict[str, Any]], *, subject_id: str ) -> Optional[Dict[str, Any]]: """그 대상의 **살아 있는 선택 하나**. `supersedes_id`(`;` 로 여럿)를 해석한 뒤 살아 있는 것이 **정확히 하나**일 때만 낸다 — 둘 이상이면 `None`(고르지 않는다 · 다음 선택이 둘 다 대신한다).""" live = live_selections(selections, subject_id=subject_id) if len(live) != 1: # ★둘 이상 살아 있으면 **고르지 않는다** — 「아무거나」는 사람이 안 정한 것을 정한 것이다 # (Codex NON-BLOCK 2026-09-02 밤). 서버는 같은 대상의 선택을 행 잠금으로 직렬화하므로 # 보통 하나다; 둘이면 다음 선택이 둘 다 대신한다. return None return live[0] def live_selections(selections: Sequence[Dict[str, Any]], *, subject_id: str) -> List[Dict[str, Any]]: """그 대상의 살아 있는 선택들 — `supersedes_id`(여러 개면 `;` 로 이어진다)를 해석한 뒤.""" mine = [s for s in (selections or ()) if str(s.get("research_subject_id") or "") == str(subject_id)] dead: set = set() for s in mine: for x in str(s.get("supersedes_id") or "").split(";"): if x.strip(): dead.add(x.strip()) return sorted([s for s in mine if str(s.get("id")) not in dead], key=lambda s: str(s.get("selected_at") or "")) def read_reviews(db: Any, *, project_id: str, episode_id: str ) -> List[Dict[str, Any]]: """이 에피소드의 판정 전부. ★못 읽으면 **선다**. ★★「못 읽었다」를 「판정이 없다」로 접으면, 사람이 맞다고 한 사진이 조용히 안 붙는다 — 그건 기록 손실이지 판정 결과가 아니다. """ from app.models.project import GroundingReferenceFidelityReview as M try: rows = (db.query(M) .filter_by(project_id=str(project_id), episode_id=str(episode_id)).all()) except Exception as exc: # noqa: BLE001 raise ReviewsUnreadable( f"고증 판정 기록을 못 읽었다 ({exc}) — 「판정이 없다」가 아니라 " f"**모른다**는 뜻이다. `alembic upgrade head` 를 먼저 보라") from exc return [{"id": r.id, "review_input_hash": r.review_input_hash, "verdict": r.verdict, "reviewed_at": r.reviewed_at, "reviewer_actor": r.reviewer_actor, "supersedes_id": r.supersedes_id} for r in rows] def central_cp_with_reviews(runner: Any, *, required: bool = True ) -> Optional[Dict[str, Any]]: """**소비자가 읽는** 중앙 체크포인트. ★DB 의 사람 판정을 얹어서 준다. ★★★왜 여기 (Codex 재리뷰 2026-09-02): `apply_reviews` 를 만들어 놓고 **production 호출자가 0곳**이었다. 그러면 사람이 「맞다」로 정해도 scene_detail·sidecar·정책은 여전히 `unverified` 를 보고 아무것도 안 붙인다 — 「판정 후 3장 부착」은 내 도구 안에서만 참이었다. 소비자가 **하나같이 이 함수로** CP 를 읽는다. 두 곳이 각각 읽으면 한쪽만 판정을 반영한다. Raises: ReviewsUnreadable · ReviewInputBroken: 기록을 모르는 채로 안 붙인다. """ from pathlib import Path from app.core.config import settings # ★CP 부재·손상을 **여기서 말하지 않는다** — `acquisition_projection` # (`required=True`)과 `assert_central_checkpoint` 가 이미 그 문이다. # 여기서 또 세우면 그 문들의 말이 가려지고, 「무엇이 없었나」가 바뀐다. cp = runner._load_prev_checkpoint("reference_acquisition") if cp is None: return None world = runner._load_prev_checkpoint("visual_world_rules") or {} data = (world.get("data") or {}) coords = {"era": str(data.get("era") or "").strip(), "region": str(data.get("region") or "").strip()} root = Path(settings.projects_dir).parent seen: Dict[str, str] = {} def _sha(row): # ★사이드카·probe 와 **같은 helper** — 세 곳이 같은 bytes 를 센다 # (실측 2026-09-02: production 사이드카가 CP 에 없는 sha 칸을 읽어 실패) from app.modules.pipeline.grounding_sidecar_writer import ( _chosen_rel_path, row_content_sha256) rel = _chosen_rel_path(row) if rel not in seen: seen[rel] = row_content_sha256(row, root=root) return seen[rel] reviews = read_reviews(runner.db, project_id=runner.project_id, episode_id=runner.episode_id) selections = read_selections(runner.db, project_id=runner.project_id, episode_id=runner.episode_id) return apply_reviews(cp, reviews, selections=selections, project_id=str(runner.project_id), episode_id=str(runner.episode_id), coordinates=coords, sha_of=_sha) def reviews_digest(reviews: Sequence[Dict[str, Any]], selections: Sequence[Dict[str, Any]] = ()) -> str: """**유효 결정 묶음**의 지문. ★사람이 마음을 바꾸면 이 값이 달라진다. ★★★왜 (Codex 조건 4 · 2026-09-02): 이미 끝난 정책·`scene_detail` 이 **옛 결과를 되쓴다**. 사람이 「맞다」를 「아니다」로 바꿨는데 체크포인트가 그대로면, 안 붙어야 할 사진이 계속 붙는다. ★**구매 신원에는 안 넣는다** — 판정이 바뀌었다고 사진을 다시 사지 않는다. 소비자의 **처리 지문**에만 접는다. ★대신된 줄(`supersedes_id`)은 빼고, 남은 것을 `(입력신원, 판정)` 으로 정렬해 접는다 — 적힌 순서·시각이 달라도 같은 결론이면 같은 지문이다. """ dead = {str(r.get("supersedes_id")) for r in (reviews or ()) if r.get("supersedes_id")} live = sorted( (str(r.get("review_input_hash") or ""), str(r.get("verdict") or "")) for r in (reviews or ()) if str(r.get("id")) not in dead) # ★쓸 사진 선택도 접는다 — 선택을 바꾸면 붙는 사진이 바뀐다 subjects = {str(s.get("research_subject_id") or "") for s in (selections or ())} picks = sorted((sid, str((effective_selection(selections, subject_id=sid) or {}).get("review_input_hash") or "")) for sid in subjects) return hashlib.sha256( json.dumps([live, picks], sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()[:16] def reviews_digest_for(runner: Any) -> str: """그 주행이 보는 판정 묶음의 지문. ★소비자 지문에 접는다.""" return reviews_digest(read_reviews(runner.db, project_id=runner.project_id, episode_id=runner.episode_id), read_selections(runner.db, project_id=runner.project_id, episode_id=runner.episode_id))