"""참조 획득 **라운드 loop** — 조각들을 잇는다. ★여기서 아무것도 안 부른다. ## 왜 이 파일인가 조각은 **전부 이미 있었다** (2026-08-31 실측) — 판별 grounding_planner.needs_reference_acquisition 검색 search_grounded_ref.search_reference_images 안전 다운로드 search_grounded_ref.download_candidate 후보 정리 coarse_type_pick.dedupe_candidates 거친 선택 coarse_type_pick.combine_coarse_verdicts 좁힘 문안 coarse_type_pick.load_narrow_hint 다음 라운드 coarse_type_pick.decide_next_round 종결 상태 reference_acquisition.STATUS_* **없던 것은 이 여덟을 잇는 loop 하나**다. `reference_acquisition_step` 이 대상이 있으면 `NotImplementedError` 로 서 있던 자리가 여기다. ## ★부르는 것을 **주입받는다** 검색·다운로드·판정을 인자로 받는다. 그래야 — - 이 모듈이 **provider 를 직접 안 연다** — 어디서 사는지는 호출부가 정한다 - 무료 시험이 **실물과 같은 모양**으로 돈다 (fake 가 제 모양을 지어내지 않게 실제 함수와 같은 시그니처를 쓴다) ## VLM 이 하는 일 / 안 하는 일 한다 「우리가 찾던 **종류**가 맞나」 · 「또렷이 **보이나**」 안 한다 시대 · 국가 · 제조사 · 모델 · 고증 정확성 · 좋고 나쁨 좁혀 다시 찾을 때도 그 좌표는 **질의에 남긴다** — 좁힌다는 것은 *다른 것들*을 덜어내는 것이지 대상을 못박는 좌표를 버리는 것이 아니다. ★고증 정확성과 이미지 품질은 **사람만** 판정한다 (§5·§6). ## ★후보와 판정을 **남긴다** — 사람이 볼 수 있게 URL 만 남기거나 숨은 임시 파일에 받으면 **비교 화면을 못 만든다** (Codex 2026-08-31). 라운드마다 — 받은 후보 3~5장 파일로 · 서빙 가능한 자리에 후보별 거친 판정 심판마다 무엇이라 했는지 그대로 고른 참조 어느 것을 왜 골랐는지 가 장부에 남는다. 검색 원본 URL 은 410/403 이 흔해서 **URL 만으로는 나중에 다시 못 본다** — 그래서 받은 파일이 근거다. ## 「없다」와 「못 봤다」를 가른다 selected 골랐다 no_match_after_retry 두 라운드 다 정상으로 돌았는데 종류가 맞는 것이 없다 — **terminal**. ★하류를 **막지 않는다** retryable provider·다운로드·시간·예산 때문에 **다 못 봤다** — 「없다」가 아니다. ★이것도 하류를 막지 않는다 ★★셋 중 어느 것도 **사람을 기다리지 않는다** (사용자 확정 2026-08-31: 「궁극적 목적은 자동화이니 HITL 을 무조건 필요한 요소로 하면 안 된다」). 못 구하면 `reference_unavailable` 로 적고 참조 **없이** 내려간다 — 가름은 감사 기록을 위한 것이지 문이 아니다. 정책 한 곳은 `reference_acquisition.downstream_blocked` / `acquisition_outcome` 이다. """ from __future__ import annotations import logging from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Sequence logger = logging.getLogger(__name__) ROUNDS_CONTRACT_VERSION = "1.202608312100" class SearchFn: """검색 대역이 지켜야 할 모양. ★실제 함수와 **같은 인자 이름**이다.""" def __call__(self, *, directive_native: str, terms_native: Sequence[str], language_lock_native: str, max_results: int) -> Dict[str, Any]: raise NotImplementedError def _is_abort(exc: BaseException) -> bool: """주행을 세워야 하는 예외인가 — `run_control.is_abort` 한 곳의 규칙(코드로 본다).""" from app.core.run_control import is_abort return is_abort(exc) def _rows_from(found: Dict[str, Any]) -> List[Dict[str, Any]]: """검색 산출 → 후보 행. ★칸 이름을 **검색 함수 docstring 그대로** 쓴다.""" out: List[Dict[str, Any]] = [] for im in (found.get("images") or ()): url = str(im.get("image_url") or "") if not url: continue out.append({ "url": url, "thumbnail_url": str(im.get("thumbnail_url") or ""), "source_website_url": str(im.get("source_website_url") or ""), "caption": str(im.get("caption") or ""), }) return out class WorkdirOutsideRoot(ValueError): """받는 자리가 서빙 뿌리 **밖**이다. ★조용히 넘기지 않는다.""" class CoordinatesMissing(ValueError): """저작된 질의가 **선언된 좌표**를 안 실었다. ★provider 앞에서 선다.""" def assert_coordinates_carried(directive: str, terms: Sequence[str], *, coordinates: Optional[Dict[str, str]] = None ) -> Dict[str, Any]: """나가는 것이 **선언된 좌표를 글자 그대로** 실었나. ★공개 끝점. ★★★계약 (Codex 2026-09-02) — · 런타임 **구조화** 칸(`visual_world_rules.era`/`.region`)만 본다 · **선언된 것, 비어 있지 않은 것**만 검사한다 · 지역만 있으면 **현대여도 지역은 의무**다 · 시대가 없으면 **만들지도 요구하지도 않는다** · 뜻을 짐작하지 않는다 — 장소·나라 목록도, 정규식도 없다. **선언된 문자열이 그 안에 있나**만 본다(값의 존재 확인이지 의미 판정이 아니다) 실측 (2026-09-02 유료 canary ①): 우리가 만든 검색어 50개는 좌표를 **50/50 그대로** 실었다. 잃는 것은 provider 가 스스로 더한 200개 질의 쪽이고(195개에 좌표 없음) 그건 우리가 만드는 것이 아니다. 그래서 문은 **우리 것에만** 건다. Raises: CoordinatesMissing: 선언된 좌표가 빠진 질의가 있다. """ want = {k: str(v or "").strip() for k, v in (coordinates or {}).items() if str(v or "").strip()} texts = [("지시문", str(directive or ""))] + [ (f"검색어 {i}", str(t)) for i, t in enumerate(terms or (), 1)] missing: List[Dict[str, str]] = [] for name, text in texts: if not text.strip(): continue for axis, decl in sorted(want.items()): if decl not in text: missing.append({"where": name, "axis": axis, "declared": decl, "text": text[:120]}) if missing: raise CoordinatesMissing( f"선언된 좌표가 빠진 질의가 {len(missing)}건이다 — " f"{[(m['where'], m['axis']) for m in missing[:4]]}. " f"시대 없는 질의는 앞선 시기 것이, 지역 없는 질의는 같은 언어권 " f"옆 나라 것이 온다. 뒤에서 아무도 안 본다 — provider 앞에서 선다") return {"declared": sorted(want), "checked": len(texts), "★means": ("**우리가 만든 것**만 본다. provider 가 스스로 더하는 " "질의는 우리 것이 아니라 관측값이지 문이 아니다")} def era_coverage(queries: Sequence[str], era_tokens: Sequence[str] ) -> Dict[str, Any]: """질의에 **시대 좌표가 붙었나**. ★뜻을 판단하지 않는다 — 글자만 본다. ★★사용자 실측 (2026-08-31): 찾아온 참조가 **그 시대보다 오래된 것**이었다. 질의를 보니 시대 좌표가 빠진 것이 많았다. 기록물은 주어진 시기보다 **앞선** 시기가 훨씬 많이 남아 있어서, 시대를 안 붙이면 더 오래된 것이 온다. ★심판에게 시대를 묻지 않는 것이 원칙이므로(사용자 확정), 시대는 **검색이 지켜야 한다**. 여기서 재는 것은 「지시문이 시킨 대로 시대가 붙었나」이지 「이 사진이 그 시대 것인가」가 **아니다** — 뒤의 것은 사람 몫이다. ★★★**못 재는 것을 통과로 세지 않는다** (Codex BLOCK 2026-08-31). 앞 판은 `era_tokens` 가 비면 모든 질의를 `continue` 한 뒤 `with_era=n` 을 돌려줬다. 즉 **잴 근거가 아예 없을 때 100%** 로 보고했다. 바로 위 docstring 에 「못 뽑으면 통과로 읽지 않는다」고 써 놓고 코드는 반대였다. 이제 `measured=False` 로 서고 `with_era` 는 **0**, `unknown` 에 전부 넣는다. Args: queries: 나간 질의들. era_tokens: `era_tokens_of()` 가 **구조화된 시대 칸**에서 뽑은 조각. Returns: `{"measured", "total", "with_era", "without", "outside_era", "unknown", "missing": [...], "outside": [...]}`. ★`measured=False` 면 나머지 수로 비율을 내면 **안 된다**. """ import re toks = [str(t).strip() for t in era_tokens if str(t or "").strip()] n = len(list(queries)) if not toks: # ★★잴 것이 없다. 「어긋남 0」이 아니라 **「안 쟀다」**이다. return {"measured": False, "total": n, "with_era": 0, "without": 0, "outside_era": 0, "unknown": n, "missing": [], "outside": [], "note": "시대 SOT 가 없거나 비교 불능 — 재지 못했다"} # ★★**그 십년 안의 해**도 시대 좌표다 (2026-08-31 실측). 「1960」만 찾으면 # `1962년`·`1966년`·`1969년` 처럼 **더 정확한** 질의를 「시대 없음」으로 # 세어 버린다 — 재는 도구가 더 좋은 것을 벌주는 꼴이다. # # ★뜻을 판단하지 않는다. 「같은 십년대인가」는 숫자 계산이다. decades = {t[:3] for t in toks if t.isdigit() and len(t) == 4} miss, outside = [], [] for q in queries: text = str(q or "") years = re.findall(r"\d{4}", text) if any(t in text for t in toks): pass elif any(y[:3] in decades for y in years): # ★같은 십년대의 다른 해도 시대 좌표다 pass elif years: # ★★**시대가 없는 것과 시대가 어긋난 것은 다르다** (실측 # 2026-08-31). 「1973년 …」은 시대를 적긴 적었는데 **주어진 # 십년대 밖**이다. 둘을 한 수로 세면 무엇을 고쳐야 하는지 모른다. outside.append(text) else: miss.append(text) return {"measured": True, "total": n, "with_era": n - len(miss) - len(outside), "without": len(miss), "missing": miss[:8], "outside_era": len(outside), "outside": outside[:8], "unknown": 0} def era_tokens_of(era_declaration: str) -> List[str]: """**구조화된 시대 칸**에서 비교할 조각을 뽑는다. ★뜻 추출이 아니다. ★★입력은 `visual_world_rules.era` — 창작자가 확정한 **시대 하나**다. 앞 판은 세계 사실 **전문**에 네 자리 숫자를 걸어서 값·번지·수량까지 시대라고 읽었다(Codex BLOCK 2026-08-31). 시대는 이미 제 칸이 있으므로 **그 칸을 명시로 받는다** — 넓은 글에서 짐작하지 않는다. ★연도 숫자만 집는다. 낱말을 뜻으로 고르면 그것이 의미 판정이고, 시대 이름은 언어·문화마다 다르다. ★못 뽑으면 **빈 목록**이고, `era_coverage` 는 `measured=False` 로 선다. 「없다」가 「통과」가 되지 않는다. """ import re out: List[str] = [] for m in re.finditer(r"\d{3,4}", str(era_declaration or "")): tok = m.group(0) if tok not in out: out.append(tok) return out[:6] #: 질의가 **어디서 왔나**. ★추측하지 않는다 — 우리가 준 것이 아니면 확장이다. ORIGIN_PROVIDED = "provided" ORIGIN_EXPANDED = "provider_expanded" #: 후보와 질의의 결속을 provider 가 **안 준** 경우. ★이름·내용으로 안 짐작한다. ORIGIN_UNKNOWN = "unknown" def query_provenance(sent: Sequence[str], searched: Sequence[Any], *, coordinates: Optional[Dict[str, str]] = None ) -> Dict[str, Any]: """나간 질의를 **우리 것 / provider 가 더한 것**으로 가른다. ★관측이다. ★★★문이 아니다 (Codex 2026-09-02). provider 의 질의 확장은 우리가 만드는 것이 아니므로 **막지 않고 적는다**. 사람이 §5·§6 에서 본다. 실측 (2026-09-02 유료 canary ①): 우리가 보낸 50개는 좌표를 50/50 그대로 실었고, provider 가 더한 200개 중 195개에 좌표가 없었다. 그 둘을 **합쳐 센 수**(212 중 126)는 acceptance 근거로 쓰지 않는다. Returns: `{"queries": [{text, origin}], "provided", "expanded", "expanded_without": {axis: n}}` """ ours = {str(x) for x in (sent or ())} flat: List[str] = [] for g in (searched or ()): for q in (g if isinstance(g, (list, tuple)) else [g]): flat.append(str(q)) want = {k: str(v or "").strip() for k, v in (coordinates or {}).items() if str(v or "").strip()} rows = [{"text": q, "origin": ORIGIN_PROVIDED if q in ours else ORIGIN_EXPANDED} for q in flat] exp = [r for r in rows if r["origin"] == ORIGIN_EXPANDED] return { "queries": rows, "provided": sum(1 for r in rows if r["origin"] == ORIGIN_PROVIDED), "expanded": len(exp), # ★확장 질의의 좌표 누락은 **관측값**이지 지금 gate 가 아니다 "expanded_without": {axis: sum(1 for r in exp if decl not in r["text"]) for axis, decl in sorted(want.items())}, "★means": ("우리가 준 질의와 provider 가 더한 질의를 **갈라** 센다. " "둘을 합쳐 센 수는 acceptance 근거가 아니다"), } def _rel(path: Path, root: Optional[Path]) -> str: """장부에 적을 경로. ★**상대**로 적는다 — 절대 경로는 다른 기계에서 깨진다. ★★뿌리 **밖**이면 **선다** (2026-08-31 실측). 앞 판은 조용히 원래 문자열을 돌려줬는데, 그것이 `../artifact/…` 처럼 **밖으로 새는 상대 경로**라 화면이 파일을 못 열었다 — 파일은 40장 다 있었는데 「없다」로 보였다. 「못 찾았다」를 「없다」로 읽는 부류다. 여기서 세우면 배선할 때 바로 안다. """ if root is None: return str(path) try: return str(Path(path).resolve().relative_to(Path(root).resolve())) except ValueError as exc: raise WorkdirOutsideRoot( f"받는 자리 {path} 가 서빙 뿌리 {root} 밖이다 — 그러면 화면이 " "파일을 못 연다. `workdir` 을 뿌리 안으로 두어야 한다") from exc def acquire_one( target: Dict[str, Any], *, workdir: Path, rel_root: Optional[Path] = None, search: Callable[..., Dict[str, Any]], download: Callable[..., bool], judge: Callable[[Sequence[Dict[str, Any]]], Dict[str, Any]], rounds: Optional[int] = None, per_round_cap: Optional[int] = None, narrow_hint: Optional[str] = None, write_brief: Optional[Callable[..., Dict[str, Any]]] = None, era_tokens: Optional[Sequence[str]] = None, resume_from: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """한 대상의 참조를 구한다. ★라운드는 **최대 `rounds`**(기본 2)회. Args: target: `{subject_id, directive_native, terms_native, language_lock_native}`. ★질의는 **위에서 온다** — 이 모듈이 대상의 뜻을 다시 판단하지 않는다. workdir: 받은 사진을 둘 자리. ★**서빙 가능한 곳**이어야 한다 — 숨은 임시 자리면 사람 비교 화면이 그것을 못 연다. rel_root: 장부에 적을 **상대 경로**의 기준. 보통 저장소 뿌리다. 절대 경로를 적으면 다른 기계에서 화면이 깨진다. search: `search_reference_images` 와 **같은 모양**. download: `download_candidate` 와 같은 모양. judge: 받은 사진들 → `{judge_name: verdict_payload}`. ★거친 종류와 가시성만 묻는 자리다. 이 모듈은 그 payload 를 **해석하지 않고** `combine_coarse_verdicts` 에 넘긴다. Returns: `{"subject_id", "status", "chosen", "rounds": [...], "candidates": [...]}` """ from app.modules.pipeline import coarse_type_pick as ctp from app.modules.pipeline import reference_acquisition as ra max_rounds = int(rounds or ctp.MAX_ROUNDS) cap = int(per_round_cap or ctp.PER_ROUND_CAP) sid = str(target.get("subject_id") or "") directive = str(target.get("directive_native") or "") if not directive.strip(): # ★질의가 없으면 **안 찾는다.** 빈 질의로 검색하면 아무거나 온다. return {"subject_id": sid, "status": ra.STATUS_RETRYABLE, "chosen": None, "rounds": [], "why": "질의가 비었다 — 무엇을 찾을지 모르는 채로 안 산다", "contract": ROUNDS_CONTRACT_VERSION, "candidates": []} seen_urls: List[str] = [] ledger: List[Dict[str, Any]] = [] rounds_out: List[Dict[str, Any]] = [] status = ra.STATUS_RETRYABLE chosen: Optional[Dict[str, Any]] = None match_quality = "" start_round = 1 incomplete_rounds: List[Dict[str, Any]] = [] if resume_from: # ★★1차 전원 → 2차 순서 (Codex 2026-09-02): 앞 pass 의 라운드·후보·본 URL 을 # 그대로 이어받고 **다음 라운드부터** 돈다 — 1차를 다시 사지 않는다. # ★★**끝나지 못한 라운드는 다시 돈다** (실측 2026-09-02 밤, attempt 21c47e1b→e74f4f0b): # 앞 판이 상한 문에 막혀 2라운드가 「지시문 저작 실패」로 끝났는데, 재개가 그 줄을 # 라운드로 세어 3라운드부터 시작하려다 **아무것도 안 하고** retryable 로 남았다. # `decision.next == retryable`(provider·시간·예산) 이고 받은 후보가 **없는** 라운드는 # 한 일이 없으니 세지 않는다 — 감사용으로 `incomplete_rounds` 에 남긴다. 후보를 # 받았는데 판정만 못 한 라운드는 그대로 둔다(재판정 lane 이 받아 둔 것을 판정한다). kept: List[Dict[str, Any]] = [] for r in (resume_from.get("rounds") or ()): r = dict(r) nxt = str(((r.get("decision") or {}).get("next")) or "") if nxt == ctp.NEXT_RETRYABLE and not (r.get("downloaded_candidates") or ()): incomplete_rounds.append(r) else: kept.append(r) rounds_out = kept ledger = [dict(c) for c in (resume_from.get("candidates") or ())] seen_urls = [str(c.get("url") or "") for c in ledger if c.get("url")] start_round = len(rounds_out) + 1 # ★조사 저작기(`grounding_target_research`)가 앞 라운드에 적어 둔 조사를 되쓰게 한다 — # 재개가 텍스트 조사를 다시 사지 않는다. prior = next((r.get("research") for r in reversed(kept) if isinstance(r.get("research"), dict)), None) if prior: target = {**target, "_prior_research": prior} # ★모든 라운드에서 받은 후보(경로·닮은 정도) — 마지막에 「반드시 한 장」을 고를 풀 all_got: List[Dict[str, Any]] = [] for r_ in rounds_out: for g in (r_.get("downloaded_candidates") or ()): all_got.append({**g, "round_no": r_.get("round_no"), "similarity": _similarity_of(r_.get("eligibility") or (), g.get("index"))}) for round_no in range(start_round, max_rounds + 1): coord_report: Optional[Dict[str, Any]] = None this_directive = directive this_terms = list(target.get("terms_native") or ()) this_lock = str(target.get("language_lock_native") or "") criteria: str = "" research: Optional[Dict[str, Any]] = None if write_brief is not None: # ★★**저작기가 원어로 지시문을 쓴다.** 좁힘 문안도 여기서 받는다 # — 영어 문안을 원어 지시문 뒤에 붙이면 검색이 죽는다 # (실측 2026-08-31: R1 64% vs **R2 10%**). try: brief = write_brief(target, narrow=(round_no > 1)) except Exception as exc: # noqa: BLE001 if _is_abort(exc): raise # ★멈추라는 말은 라운드 실패로 접지 않는다 rounds_out.append({"round_no": round_no, "decision": {"next": ctp.NEXT_RETRYABLE, "why": f"지시문 저작 실패: {exc}"}, "received": 0, "judged": 0, "duplicate": 0, "over_cap": 0, "downloaded": 0, "queries": [], "eligibility": [], "downloaded_candidates": [], "directive_native": ""}) status = ra.STATUS_RETRYABLE break this_directive = str(brief.get("search_directive_native") or directive) this_terms = [str(x) for x in (brief.get("search_terms_native") or this_terms)] this_lock = str(brief.get("language_lock_native") or this_lock) # ★조사 저작기가 낸 CRITERIA(어찌 생겼는가) — 심판에게 그대로 준다 criteria = str(brief.get("vlm_criteria") or "") research = brief.get("research") if isinstance(brief.get("research"), dict) else None if research: target = {**target, "_prior_research": research} # 2라운드가 되쓴다 # ★★★**보내기 전에** 좌표를 확인한다 (Codex 2026-09-02). # 저작기가 들고 있는 선언 좌표를 그대로 쓴다 — 여기서 세계 # 사실을 다시 읽으면 두 벌이 되고 한쪽만 고쳐진다. # ★대상 **하나**를 세운다 — 판 전체를 죽이면 멀쩡한 대상까지 # 못 산다. 나가지 않는 것이 문의 목적이고, 왜 안 나갔는지는 # 그 대상의 라운드에 그대로 남는다. try: coord_report = assert_coordinates_carried( this_directive, this_terms, coordinates=getattr(write_brief, "coordinates", None)) except CoordinatesMissing as exc: rounds_out.append({"round_no": round_no, "decision": {"next": ctp.NEXT_RETRYABLE, "why": str(exc)}, "received": 0, "judged": 0, "duplicate": 0, "over_cap": 0, "downloaded": 0, "queries": [], "eligibility": [], "downloaded_candidates": [], "coordinates_carried": {"ok": False}, "directive_native": this_directive, "terms_native": this_terms}) status = ra.STATUS_RETRYABLE break elif round_no > 1: hint = narrow_hint if narrow_hint is not None \ else ctp.load_narrow_hint() # ★저작기가 없을 때만 여기서 덧붙인다 — **권장하지 않는다** this_directive = f"{directive}\n\n{hint}" err = "" try: found = search(directive_native=this_directive, terms_native=this_terms, language_lock_native=this_lock, max_results=cap) except Exception as exc: # noqa: BLE001 # ★★멈추라는 말(step.cancelled 등)은 **검색 실패가 아니다** — 그대로 올린다. 실측 4398a55dc0bb # (2026-09-03 05:08): 정지 요청이 「검색 실패」로 접혀 두 라운드가 실패로 적히고 대상 9개가 # why 빈 retryable · disposition acquired 로 남았다. 규칙은 `run_control.is_abort` 한 곳. if _is_abort(exc): raise # ★**「없다」가 아니다.** 못 본 것이다. err = f"검색 실패: {exc}" found = {} rows = _rows_from(found) if not err else [] picked = ctp.dedupe_candidates(rows, seen_urls, cap=cap) ledger += picked["ledger"] seen_urls += [r["url"] for r in picked["judged"] if r.get("url")] got: List[Dict[str, Any]] = [] #: 판정이 **실제로 열 수 있는** 경로. ★기록에는 안 넣는다. # ★★★실측 (2026-09-02 유료 canary ①): 기록용 **상대** 경로를 그대로 # 판정에 넘겼고, 판정은 그것을 **프로세스 cwd 기준**으로 열어 # `[Errno 2] No such file or directory` 로 죽었다. 검색 10회·받기 # 38장을 **다 사고 나서** 12대상 전부가 그렇게 떨어졌다 # (`selected_count: 0`). 기록은 옮겨 다녀야 하니 상대로 두고, # 여는 쪽에는 절대 경로를 준다. openable: Dict[int, str] = {} if not err: for n, r in enumerate(picked["judged"], 1): dest = workdir / f"{sid or 'x'}_r{round_no}_{n:02d}.png" if download(r["url"], dest, r.get("thumbnail_url") or ""): got.append({**r, "index": n, # ★provider 가 결속을 주면 그것을, 안 주면 # **모른다**로 남긴다 — 이름·내용으로 안 짐작 "query_origin": str(r.get("query_origin") or ORIGIN_UNKNOWN), "path": _rel(dest, rel_root)}) openable[n] = str(dest) else: # ★못 받은 것도 **남긴다** — 「몇 장을 봤나」가 흔들린다 ledger.append({**r, "disposition": "download_failed"}) combined: Dict[str, Any] = {} per_judge: Dict[str, Any] = {} if got and not err: try: per_judge = _call_judge(judge, [{**g, "path": openable[g["index"]]} for g in got], criteria=criteria) combined = ctp.combine_coarse_verdicts(per_judge, len(got)) except Exception as exc: # noqa: BLE001 if _is_abort(exc): raise # ★멈추라는 말은 판정 실패로 접지 않는다 err = f"판정 실패: {exc}" elif not err and not got: # ★후보를 하나도 못 받았다 — 판정 자체가 없다 combined = {"chosen_index": 0, "candidate_count": 0, "reason": "받은 후보가 없다"} elig_table = (ctp.eligibility_table(per_judge, len(got)) if per_judge and got else []) for g in got: all_got.append({**g, "round_no": round_no, "similarity": _similarity_of(elig_table, g["index"])}) decision = ctp.decide_next_round( round_no, combined, new_candidate_count=picked["judged_count"], error=err, total_candidate_count=len(all_got)) rounds_out.append({ "round_no": round_no, "directive_native": this_directive, "terms_native": this_terms, # ★우리가 만든 것이 좌표를 실었나 — **문을 지난 기록** "coordinates_carried": coord_report, "queries": list((found or {}).get("queries") or ()), # ★우리 것과 provider 확장을 **갈라** 적는다 (관측 · 문 아님) "query_provenance": query_provenance( this_terms, (found or {}).get("queries") or (), coordinates=getattr(write_brief, "coordinates", None)), # ★지시문이 시킨 대로 **시대가 붙었나** — 사진 판정이 아니다 "era_coverage": era_coverage( [x for g in ((found or {}).get("queries") or ()) for x in (g if isinstance(g, list) else [g])], era_tokens or ()), "received": len(rows), "judged": picked["judged_count"], "duplicate": picked["duplicate_count"], "over_cap": picked["over_cap_count"], "downloaded": len(got), "judges": list(combined.get("judges") or ()), "rejected_judges": dict(combined.get("rejected_judges") or {}), "single_judge": bool(combined.get("single_judge")), "eligible": list(combined.get("eligible") or ()), "chosen_index": int(combined.get("chosen_index") or 0), "closest_index": int(combined.get("closest_index") or 0), "criteria_used": bool(criteria), **({"research": research} if research else {}), "eligibility": elig_table, # ★★**받은 후보 그 자체**를 남긴다 — 파일 · 출처 · 설명. # URL 만 남기면 410/403 때문에 나중에 못 본다. "downloaded_candidates": [ {"index": g["index"], "path": g["path"], "url": g["url"], "query_origin": g.get("query_origin") or ORIGIN_UNKNOWN, # ★★내용 지문 — 나중에 **같은 것을 다시 보는지** 가른다. # 앞 판에는 없었다. 없으면 재판정이 hash 대조를 못 한다. "sha256": _sha256(Path(openable[g["index"]])), "source_website_url": g.get("source_website_url") or "", "caption": g.get("caption") or ""} for g in got], "decision": decision, }) nxt = decision.get("next") if nxt == ctp.NEXT_SELECT: idx = int(combined.get("chosen_index") or 0) chosen = next((g for g in got if g["index"] == idx), None) status = ra.STATUS_SELECTED if chosen else ra.STATUS_RETRYABLE match_quality = "criteria" if chosen else "" break if nxt == ctp.NEXT_SELECT_CLOSEST: # ★★사용자 5단계 ⑤ — 받은 모든 라운드의 후보 중 **가장 닮은** 한 장. 비긴 것은 # 앞 라운드·앞 index(검색 상위). 안 고르는 것보다 가장 가까운 것을 고른다. pool = sorted(all_got, key=lambda g: (-(g.get("similarity") if g.get("similarity") is not None else -1.0), int(g.get("round_no") or 0), int(g["index"]))) chosen = dict(pool[0]) if pool else None status = ra.STATUS_SELECTED if chosen else ra.STATUS_NO_MATCH match_quality = "closest" if chosen else "" if chosen: chosen["forced_pick"] = True # ★판정이 없었으면(심판 실패·검색 실패) 검색 순위 첫 장이다 — 까닭을 남긴다. # 결정 자료로 정한다: 받아 둔 후보 중 판정값(similarity)이 하나도 없으면 심판이 없었다. no_judgement = all(g.get("similarity") is None for g in pool) chosen["forced_reason"] = str(decision.get("forced_reason") or ("judge_unavailable" if no_judgement else "no_criteria_match")) break if nxt == ctp.NEXT_NO_MATCH: status = ra.STATUS_NO_MATCH break if nxt == ctp.NEXT_RETRYABLE: status = ra.STATUS_RETRYABLE break # NEXT_NARROW_RETRY — 다음 라운드로 return { "subject_id": sid, "status": status, **({"match_quality": match_quality} if status == ra.STATUS_SELECTED else {}), # ★★★고증 축은 **여기서도** 명시로 적는다 (2026-09-02). 거친 종류로 # 골랐다는 것과 그 시대·그 지역 것이 맞다는 것은 다른 축이다 — # 이 판정은 앞의 것만 했다. "grounding_fidelity": { "state": ra.FIDELITY_UNVERIFIED, "contract": ra.FIDELITY_CONTRACT_VERSION, "why": ("거친 종류·가시성만 봤다. 시대·지역 적합성은 이 판정의 " "대상이 아니다 — 사람이 보거나 별도 확인기가 낸다"), }, "chosen": chosen, "rounds": rounds_out, # ★한 일이 없어 다시 돈 라운드 — 지우지 않고 여기 남긴다 **({"incomplete_rounds": incomplete_rounds} if incomplete_rounds else {}), "candidates": ledger, # ★고른 것의 **파일**까지 — 화면이 이것을 건다 "chosen_path": (chosen or {}).get("path") or "", "contract": ROUNDS_CONTRACT_VERSION, # ★하류가 막아야 하는가 — 한 곳(`reference_acquisition`)이 정한다 "downstream_blocked": ra.downstream_blocked(status), # ★★★**처분을 durable 로 적는다** (Codex BLOCK 2026-08-31). # # 앞 판은 `acquisition_outcome()` 이 **함수로만** 있고 산출에 안 # 적혔다. 그래서 「못 구한 것은 `reference_unavailable` 로 적힌다」는 # 보고가 실제 기록과 달랐다 — 얼어붙은 산출에는 raw status 뿐이었다. # # ★raw `status` 를 **지우지 않는다.** 왜 못 구했는지(두 라운드 다 # 돌았나 · 다 못 봤나)는 감사에 필요하다. `outcome` 은 그 위에 # 얹는 **하류가 보는 한 칸**이다. "outcome": ra.acquisition_outcome(status), } #: 재판정 계약. ★올리면 앞 판 재판정과 **신원이 갈린다**. REPLAY_CONTRACT_VERSION = "1.202609020600" class CandidateOutsideRoot(ValueError): """받아 둔 사진이 **허용 뿌리 밖**을 가리킨다. ★안 열고 선다.""" class CandidateMissing(FileNotFoundError): """받아 둔 사진이 **없다**. ★VLM 을 부르기 전에 선다.""" class CandidateChanged(ValueError): """받아 둔 사진의 **내용이 달라졌다**. ★같은 것을 다시 보는 게 아니다.""" class NothingToRejudge(ValueError): """되볼 후보가 **없다**. ★「없는데 봤다」가 되면 안 된다.""" def _similarity_of(elig_table: Sequence[Dict[str, Any]], index: Any) -> Optional[float]: """eligibility 표에서 그 후보의 심판 평균 similarity. 없으면 None.""" for row in elig_table or (): if int(row.get("index") or 0) == int(index or 0): vals = [j.get("similarity") for j in (row.get("by_judge") or {}).values() if isinstance(j, dict) and isinstance(j.get("similarity"), int)] return (sum(vals) / len(vals)) if vals else None return None def _call_judge(judge: Callable[..., Dict[str, Any]], candidates: List[Dict[str, Any]], *, criteria: str) -> Dict[str, Any]: """심판에 CRITERIA 를 넘긴다. ★옛 심판(인자 하나)도 그대로 돈다 — 시험·legacy 호환.""" if criteria: try: return judge(candidates, criteria=criteria) except TypeError as exc: if "criteria" not in str(exc): raise return judge(candidates) def _sha256(path: Path) -> str: import hashlib h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1 << 20), b""): h.update(chunk) return h.hexdigest() def resolve_cached_candidates(candidates: Sequence[Dict[str, Any]], *, root: Path) -> List[Dict[str, Any]]: """장부의 **상대** 경로 → 열 수 있는 절대 경로. ★VLM **전에** 다 본다. ★★★계약 (Codex 2026-09-02): 뿌리 이탈 · 파일 없음 · 내용 hash 불일치는 **VLM 호출 전에 fail-closed** 다. 돈을 쓰고 나서 「못 열었다」로 끝나는 것이 바로 앞 판에 난 일이다 — 검색 10회·받기 38장을 다 사고 12대상 전부가 `[Errno 2]` 로 떨어졌다. Returns: `[{index, path(절대), sha256, rel}]` — **장부에 쓰지 않는다**. Raises: NothingToRejudge · CandidateOutsideRoot · CandidateMissing · CandidateChanged """ base = Path(root).resolve() out: List[Dict[str, Any]] = [] for c in candidates or (): rel = str(c.get("path") or "") if not rel: raise NothingToRejudge("후보에 경로가 없다") p = Path(rel) if p.is_absolute(): raise CandidateOutsideRoot( f"장부에 **절대 경로**가 적혀 있다 ({rel}) — 기록은 상대여야 " f"다른 기계에서 열린다") full = (base / p).resolve() if base != full and base not in full.parents: raise CandidateOutsideRoot( f"후보 {rel} 이 허용 뿌리 {base} 밖을 가리킨다 — 안 연다") if not full.is_file(): raise CandidateMissing( f"받아 둔 사진이 없다: {rel} (뿌리 {base}) — VLM 을 부르기 " f"전에 선다") got = _sha256(full) was = str(c.get("sha256") or "") if was and was != got: raise CandidateChanged( f"{rel} 의 내용이 달라졌다 — 적힌 것 {was[:12]} · 지금 " f"{got[:12]}. 같은 것을 다시 보는 게 아니다") out.append({**c, "index": int(c.get("index") or (len(out) + 1)), "path": str(full), "sha256": got, "rel": rel}) if not out: raise NothingToRejudge("되볼 후보가 하나도 없다") return out def cached_round(record: Dict[str, Any], *, round_no: Optional[int] = None ) -> Dict[str, Any]: """되볼 라운드 **하나**. ★고르는 규칙을 두 곳에 안 적는다. 안 주면 후보가 있는 **마지막** 라운드다. Raises: NothingToRejudge: 받아 둔 후보가 있는 라운드가 없다. """ picks = [r for r in (record.get("rounds") or ()) if r.get("downloaded_candidates")] if round_no is not None: picks = [r for r in picks if int(r.get("round_no") or 0) == round_no] if not picks: raise NothingToRejudge( f"{record.get('subject_id')!r} 에 받아 둔 후보가 있는 라운드가 없다") return picks[-1] def replay_identity_inputs(record: Dict[str, Any], resolved: Sequence[Dict[str, Any]]) -> Dict[str, Any]: """이 재판정의 **신원 재료**. ★원 구매 신원과 **따로** 적는다. ★접는 것 — 원 acquisition 신원 · **순서 있는** 후보 좌표/URL · 실제 파일 SHA · 판정/경로해석 계약. 사진이 하나라도 바뀌거나 순서가 달라지면 **다른 판**이다 (Codex 2026-09-02). """ return { "replay_contract": REPLAY_CONTRACT_VERSION, "rounds_contract": str(record.get("contract") or ""), "subject_id": str(record.get("subject_id") or ""), "candidates": [{"index": c["index"], "rel": c["rel"], "url": str(c.get("url") or ""), "sha256": c["sha256"]} for c in resolved], } def rejudge_cached(record: Dict[str, Any], *, root: Path, judge: Callable[[Sequence[Dict[str, Any]]], Dict[str, Any]], round_no: Optional[int] = None) -> Dict[str, Any]: """이미 **받아 둔** 후보만 다시 판정한다. ★검색 0 · 받기 0. 앞 판이 `판정 실패` 로 끝났을 때, 이미 산 사진을 버리고 다시 사지 않기 위한 자리다. `search`·`download` 는 **인자로도 안 받는다** — 닿을 수 없어야 한다 (Codex 계약 2026-09-02). Args: record: 장부에 얼어붙은 `acquire_one` 산출. root: 상대 경로의 **허용 뿌리**. 보통 `settings.projects_dir`. round_no: 되볼 라운드. 안 주면 후보가 있는 **마지막** 라운드. Returns: `acquire_one` 과 **같은 모양** + `replay` 블록. 경로는 전부 상대다. """ from app.modules.pipeline import coarse_type_pick as ctp from app.modules.pipeline import reference_acquisition as ra src = cached_round(record, round_no=round_no) cands = list(src.get("downloaded_candidates") or ()) resolved = resolve_cached_candidates(cands, root=root) per_judge = judge([{k: v for k, v in c.items() if k != "rel"} for c in resolved]) combined = ctp.combine_coarse_verdicts(per_judge, len(resolved)) decision = ctp.decide_next_round( # ★**마지막** 라운드로 셈한다 — 재판정은 「한 번 더 좁혀 찾기」가 # 아니다. 여기서 못 고르면 그것이 이 후보들에 대한 답이다. int(ctp.MAX_ROUNDS), combined, new_candidate_count=0) idx = int(combined.get("chosen_index") or 0) chosen = next((c for c in cands if int(c.get("index") or 0) == idx), None) if idx else None # ★★「다 보고 없었다」는 **검색이 실제로 두 라운드 나갔을 때만** 말할 수 있다. # 받아 둔 4장만 다시 보고 새 검색 0 인데 no_match 로 접으면 「없다」가 거짓이 된다 # (Codex 2026-09-02 · O01 실측). 질의가 나간 라운드 수로 가른다. searched_rounds = sum(1 for rd in (record.get("rounds") or ()) if (rd.get("queries") or ())) terminal_ok = searched_rounds >= int(ctp.MAX_ROUNDS) status = ra.STATUS_SELECTED if chosen else ( ra.STATUS_NO_MATCH if decision.get("next") == ctp.NEXT_NO_MATCH and terminal_ok else ra.STATUS_RETRYABLE) replay = { "contract": REPLAY_CONTRACT_VERSION, "from_round": int(src.get("round_no") or 0), "searched_rounds": searched_rounds, "candidate_count": len(resolved), # ★상대 경로와 SHA 만 남긴다 — 기계 경로는 장부에 안 쓴다 "candidates": [{"index": c["index"], "path": c["rel"], "sha256": c["sha256"]} for c in resolved], "judges": list(combined.get("judges") or ()), "rejected_judges": dict(combined.get("rejected_judges") or {}), "single_judge": bool(combined.get("single_judge")), "eligible": list(combined.get("eligible") or ()), "chosen_index": idx, "eligibility": ctp.eligibility_table(per_judge, len(resolved)), "decision": decision, "★means": ("이미 받아 둔 사진만 다시 봤다 — 검색 0 · 받기 0. " "**배선 판정**이지 고증 판정이 아니다"), } return { **record, "status": status, # ★★★고증 축을 **명시로** 적는다 (Codex BLOCK 2026-09-02). # coarse 심판은 「무엇인가 · 보이는가」만 봤다 — 그 사진이 그 시대· # 그 지역 것인지는 **아무도 안 봤다**. 없으면 미확인으로 접히지만, # 적어 두면 사람 검토 화면이 왜 안 붙었는지를 바로 읽는다. "grounding_fidelity": { "state": ra.FIDELITY_UNVERIFIED, "contract": ra.FIDELITY_CONTRACT_VERSION, "why": ("거친 종류·가시성만 봤다. 시대·지역 적합성은 이 판정의 " "대상이 아니다 — 사람이 보거나 별도 확인기가 낸다"), }, "chosen": chosen, "chosen_path": (chosen or {}).get("path") or "", "downstream_blocked": ra.downstream_blocked(status), "outcome": ra.acquisition_outcome(status), "replay": replay, }