"""장부 → **중앙 획득 한 곳**. ★구매자는 여기뿐이고, **사람을 안 기다린다**. ## 무엇을 하나 **갈래 중립 장부**(`grounding_acquisition_ledger`)를 받아 — ①`inputs_from_ledger` **한 입구**로만 대상을 만든다 ②상한·신원·재개는 **기존 `ChunkJournal`** 이 본다 (새로 안 만든다) ③`reference_acquisition_rounds.acquire_one` 으로 산다 ④**모든 갈래를 durable 산출에 남긴다** — 산 것·못 산 것·안 살 것 ★★★**새 구매 구현을 만들지 않는다.** 검색·다운로드·판정은 부르는 쪽이 주고, 이 모듈은 그것을 **기존 라운드 기구**에 그대로 넘긴다. 두 벌이 되면 캐시도 장부도 갈린다. ## 상한에 닿으면 — **서지 않는다** 사용자 확정 (2026-08-31): 「궁극적 목적은 자동화이니 HITL 을 무조건 필요한 요소로 하면 안 된다.」 그래서 상한에 닿아도 — ①이미 산 것·재사용한 것은 **그대로 남는다** ②남은 대상은 `retryable`(=예산 때문에 다 못 봤다)로 적고 `acquisition_outcome` 이 `reference_unavailable` 로 접는다 ③raw 사유 `cap_reached` 를 **그 줄에 그대로** 남긴다 ④회계가 「덜 산 판」임을 **스스로 말한다** — 다 산 판인 척하지 않는다 ★앞 판은 여기서 예외를 던져 **이미 산 유료 산출까지 반환값에서 날렸다** (Codex 2026-09-01). 상한은 **더 안 사는 문**이지 주행을 멈추는 문이 아니다. ★이 모듈은 **아직 아무 스텝도 안 부른다.** manifest 배선은 D cutover 몫이다. """ from __future__ import annotations import copy import logging from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Sequence from app.modules.pipeline import grounding_acquisition_adapter as _aa logger = logging.getLogger(__name__) CONTRACT_VERSION = "1.202609012300" #: ★상한 때문에 안 산 줄의 **raw 사유**. 종결 상태와 **따로** 남긴다 — #: 접은 뒤(`reference_unavailable`)만 남기면 왜 없는지 사라진다. WHY_CAP_REACHED = "cap_reached" WHY_ABORTED = "aborted_after_fatal" # ★앞 워커의 fatal 뒤 — 이 대상은 provider 를 안 건드렸다 #: ★★**앞 판**에 답을 못 받은 호출이 있는 신원. 샀는지 모른다. #: 다시 사지도 않고 **사람을 기다리지도 않는다** — 참조 없이 간다. #: 장부에는 `uncertain` 으로 남아 운영자 화면이 나중에 정리한다. WHY_PRIOR_UNCONFIRMED = "prior_call_unconfirmed" #: ★**이번 판**에 보냈는데 답을 못 받았다. 앞 판 것과 감사에서 갈라야 한다 — #: 하나는 재개가 물려받은 빚이고 하나는 지금 난 일이다 (Codex NON-BLOCK). WHY_CURRENT_UNCONFIRMED = "current_call_unconfirmed" #: 한 줄이 이 판에서 **어떻게 됐나**. ★수는 여기서 파생한다 — 따로 세지 않는다. DISP_ACQUIRED = "acquired" # 사거나 되썼다 (골랐는지는 outcome 이) DISP_CAP_REACHED = "cap_reached" # 상한에 닿아 안 보냈다 DISP_UNCONFIRMED = "unconfirmed" # 보냈는지 모른다 — 다시 안 산다 DISP_SKIPPED = "skipped" # 저작 재료가 없어 질의를 못 만들었다 DISP_REFUSED = "refused" # 살 자격이 없는 줄이었다 DISP_AUTO_DONE = "auto_completed" # 사야 하는데 못 붙였다 — 참조 없이 간다 DISP_NOT_APPLICABLE = "not_applicable" # 애초에 살 것이 아니었다 DISPOSITIONS = (DISP_ACQUIRED, DISP_CAP_REACHED, DISP_UNCONFIRMED, DISP_SKIPPED, DISP_REFUSED, DISP_AUTO_DONE, DISP_NOT_APPLICABLE) #: ★★**참조가 필요했던** 자리들. 이것들만 `reference_unavailable` 이 뜻이 있다. #: `not_applicable` 은 애초에 살 것이 아니었으므로 「못 구했다」가 아니다 — #: 그렇게 적으면 UI·통계에서 **불필요한 대상이 검색 실패처럼 보인다** #: (Codex 2026-09-01). `grounding_acquisition_ledger.auto_completed` 의 계약이 #: 이미 그렇게 갈라 놓았는데 내가 한 상태로 뭉갰다. NEEDS_A_REFERENCE = (DISP_ACQUIRED, DISP_CAP_REACHED, DISP_UNCONFIRMED, DISP_SKIPPED, DISP_REFUSED, DISP_AUTO_DONE) class ProviderCallUncertain(Exception): """**보냈는지 아닌지 모르는** 호출. ★이것만 자동으로 접는다. ★★앞 판은 「안 접을 것」을 목록으로 적었다(blacklist). 그러면 `KeyError`· `ValueError`·`IndexError` 같은 **내 코드 결함이 provider 장애로 둔갑**한다 (Codex 2026-09-01). 방향을 뒤집는다 — **접을 것을 선언**하고 나머지는 전부 그대로 올린다. 누가 내나: 사는 쪽이 「전송했는데 답을 못 받았다」를 **알 때만** 낸다. 검색·다운로드의 흔한 실패는 `acquire_one` 이 이미 안에서 `retryable` 로 매듭지으므로 여기까지 오지 않는다. 안 오면 어떻게 되나: 예외가 위로 올라가 이 판이 선다. 장부에는 그 신원이 `uncertain` 으로 남고, **다음 재개가 사람 없이** 자동 종결한다 (`WHY_PRIOR_UNCONFIRMED`). 그래서 재전파가 HITL 을 만들지 않는다. """ #: ★★자동으로 접는 예외는 **이것뿐이다.** 늘리려면 그 종류가 「보냈는지 #: 모른다」를 뜻한다는 근거가 있어야 한다. FOLDABLE: tuple = (ProviderCallUncertain,) def wrap_if_unconfirmed(exc: BaseException, *, request_left_the_process: bool ) -> BaseException: """운반층이 **전송 여부를 아는 자리에서만** 감싼다. ★공개 끝점. Args: request_left_the_process: 요청이 **정말 나갔는지**. 운반층(HTTP·SDK)이 그것을 아는 자리에서만 참을 준다 — 소켓에 쓴 뒤 응답을 못 받았다, 읽기 시간이 지났다 같은 것. 연결조차 못 열었으면 거짓이다. Returns: 참이면 `ProviderCallUncertain`(자동 종결 대상), 거짓이면 **받은 그대로**. ★★부르는 쪽이 「무엇이든 실패했으니 불확정」이라고 감싸면 allowlist 가 다시 blacklist 가 된다 (Codex NON-BLOCK 2026-09-01). 지금 production 에서 이것을 내는 자리는 **0곳**이고, 실제 배선 때 운반층 한 자리에만 넣는다. """ if not request_left_the_process: return exc return ProviderCallUncertain( f"요청이 나갔는데 답을 못 받았다 — 샀는지 모른다 ({exc!r})") def _is_provider_failure(exc: BaseException) -> bool: """provider 가 거절했거나(503·429·연결) 답이 안 왔나(Timeout) — 예외 **종류**로만. 재판정 lane 은 이것을 한 줄의 실패로 접는다(중앙 lane 의 acquire_one 은 라운드 안에서 이미 접는다).""" names = {c.__name__ for c in type(exc).__mro__} return bool(names & {"ServiceUnavailableError", "RateLimitError", "APIConnectionError", "Timeout", "APITimeoutError", "InternalServerError", "BadGatewayError"}) def row_why(got: Dict[str, Any]) -> str: """행 위의 까닭 — `acquire_one` 결과의 `why` 가 비었으면 **마지막 라운드의 결정 까닭**을 올린다. ★실측 4398a55dc0bb: retryable 9행이 why "" 로 남아 「다 보고 없었다」와 「못 봤다」를 못 갈랐다.""" why = str(got.get("why") or "") if why: return why rounds = got.get("rounds") or () if rounds: last = rounds[-1] if isinstance(rounds[-1], dict) else {} return str(((last.get("decision") or {}).get("why")) or last.get("error") or "") return "" def must_not_swallow(exc: BaseException) -> bool: """이 예외를 **한 대상의 실패로 접으면 안 되나.** ★공개 끝점이다 — 무엇을 안 삼키는지 시험이 여기로 묻는다. ★기본값이 **안 접는다**. 중단·주인 잃음·못 읽는 게이트는 물론이고 처음 보는 오류도 그대로 올린다. """ return not isinstance(exc, FOLDABLE) def acquisition_outcome_of(row: Dict[str, Any]) -> Optional[str]: """이 줄의 **획득 결과**. ★비대상이면 `None` — 결과가 아예 없다. ★공개 끝점이다. UI·통계가 「참조를 못 구한 것」을 셀 때 이것으로 거른다 — 비대상을 `reference_unavailable` 로 세면 검색 실패처럼 보인다. """ return row.get("outcome") if row.get("disposition") in NEEDS_A_REFERENCE \ else None #: ★★하류가 읽는 **한 줄의 모양**. 부르는 쪽이 `rows` 의 dict 를 다시 #: 해석하면 같은 뜻이 두 곳에 적히고 한쪽만 고쳐진다 (Codex 2026-09-01). #: ★2026-09-02: **고증 축**이 한 칸 늘었다 — 판을 올린다. 옛 판이 #: 조용히 섞이면 소비자가 `fidelity` 없이 붙인다. ACQUISITION_PROJECTION_VERSION = "2.202609020700" #: 투영 한 줄의 칸. ★늘리면 여기와 `acquisition_projection` 이 같이 움직인다. PROJECTION_FIELDS = ("research_subject_id", "owner_type", "final_id", "status", "outcome", "disposition", "why", "why_unbought", # ★coarse 판정과 **다른 축** — 소비자가 이것도 본다 "fidelity") class ProjectionContractError(RuntimeError): """중앙 산출이 계약 밖이다. ★조용히 빈손으로 지나가지 않는다.""" def _rows_of(central_cp: Optional[Dict[str, Any]], *, required: bool): """CP → `rows`. ★`required` 면 **다섯 가지가 다 문**이다. CP 자체가 없다 · 상태가 종결이 아니다 · `data` 가 없다 · `rows` 키가 없다 · `rows` 가 목록이 아니다 """ if central_cp is None: if required: raise ProjectionContractError( "중앙 조사 체크포인트가 **없다** — 이 판은 그 스텝에 직접 " "의존한다. 없는 것을 「조사할 게 없었다」로 읽지 않는다") return [] if required: st = str(central_cp.get("status") or "") if st not in CP_TERMINAL_STATUSES: raise ProjectionContractError( f"중앙 조사 체크포인트 상태가 {st!r} 다 — " f"{CP_TERMINAL_STATUSES} 중 하나여야 한다") data = central_cp.get("data") if not isinstance(data, dict): if required: raise ProjectionContractError( "중앙 조사 체크포인트에 `data` 가 없다 — 기록이 깨졌다") return [] if "rows" not in data: if required: raise ProjectionContractError( "중앙 조사 체크포인트에 `rows` 키가 **없다** — " "빈 판이면 `rows: []` 로 적혀 있어야 한다") return [] rows = data["rows"] if not isinstance(rows, list): raise ProjectionContractError( f"`rows` 가 목록이 아니라 {type(rows).__name__} 다") return rows #: 중앙 조사 CP 가 **온전하다**고 볼 상태들. ★`partial` 은 아니다 — #: 덜 돈 판의 줄로 참조 의무를 정하면 없는 것을 있다고 읽는다. CP_TERMINAL_STATUSES = ("completed", "not_applicable") def acquisition_projection(central_cp: Optional[Dict[str, Any]], *, required: bool = False ) -> List[Dict[str, Any]]: """중앙 조사 CP → **검증된 줄들**. ★공개 끝점 — 하류는 이것만 읽는다. Args: central_cp: `reference_acquisition` 체크포인트. required: **그 스텝이 반드시 돈 판인가**. `v2_chunk` 처럼 중앙 조사가 직접 의존인 판에서는 참이어야 한다. ★★★거짓일 때만 「없으면 빈 목록」이다 — 그것이 옛 길이다. 참인데 CP 가 없거나 모양이 깨졌으면 **선다**: 그 판에서 CP 부재는 「조사할 것이 없었다」가 아니라 **기록 손실**이고, 그대로 두면 정책도 sidecar 도 동시에 사라져 **참조 없이 그림까지 간다** (Codex BLOCK 2026-09-02). ★명시적인 `rows: []` **만** 정상적인 빈 판이다. Returns: `PROJECTION_FIELDS` 만 담은 줄들. `outcome` 은 `acquisition_outcome_of` 한 곳이 정한다 — 비대상이면 ``None``. Raises: ProjectionContractError: 모르는 처분 · 모르는 raw 상태 · 비대상인데 상태가 있다 · `selected` 인데 `final_id` 가 없다 · `outcome` 이 raw 상태에서 나온 값과 **다르다** · (`required` 일 때) CP 부재·미완·`data`/`rows` 누락·목록 아님. """ from app.modules.pipeline import reference_acquisition as ra rows = _rows_of(central_cp, required=required) if not rows: return [] out: List[Dict[str, Any]] = [] for r in rows: disp = r.get("disposition") if disp not in DISPOSITIONS: raise ProjectionContractError( f"모르는 처분 {disp!r} — 아는 것 {DISPOSITIONS}") st = r.get("status") needed = disp in NEEDS_A_REFERENCE if needed and st not in ra.KNOWN_RAW_STATUSES: raise ProjectionContractError( f"모르는 raw 상태 {st!r} — 아는 것 {ra.KNOWN_RAW_STATUSES}") if not needed and st is not None: raise ProjectionContractError( f"비대상 줄에 상태 {st!r} 가 있다 — 결과가 아예 없어야 한다") outcome = acquisition_outcome_of(r) # ★★raw 상태와 적힌 처분이 **서로 모순이어도** 지나가던 자리 # (Codex 2026-09-02). 둘을 각각 믿지 않고 **한 함수로 대조**한다. if needed and outcome != ra.acquisition_outcome(st): raise ProjectionContractError( f"{r.get('research_subject_id')!r} 의 raw 상태 {st!r} 와 적힌 " f"처분 {outcome!r} 가 어긋난다 — " f"{st!r} 는 {ra.acquisition_outcome(st)!r} 여야 한다") led = r.get("ledger_row") or {} fid = str(r.get("final_id") or led.get("final_id") or "") or None if outcome == ra.STATUS_SELECTED and not fid: raise ProjectionContractError( f"{r.get('research_subject_id')!r} 가 `selected` 인데 붙일 " f"`final_id` 가 없다") out.append({ "research_subject_id": str(r.get("research_subject_id") or ""), "owner_type": str(led.get("owner_type") or ""), "final_id": fid, "status": st, "outcome": outcome, # ★★고증 축 — 없으면 `unverified` 다. 「모르는 것」이 통과가 되면 # 안 된다 (Codex BLOCK 2026-09-02). "fidelity": ra.fidelity_of(r), "disposition": str(disp), "why": str(r.get("why") or ""), "why_unbought": r.get("why_unbought"), }) return out class BadPurchaseCap(ValueError): """상한이 **정확한 0 이상 정수**가 아니다. ★아무것도 안 하고 선다. ★`int(cap)` 로 접으면 `True` 가 1 로, `"12"` 가 12 로 들어온다 — 비용 문에 그런 관용은 없다 (Codex 2026-09-01). """ class BadRoundCount(ValueError): """라운드 수가 정확한 정수가 아니다. ★계약 해시와 실제 주행이 갈린다.""" def _exact_int(value, *, what, err, low): if isinstance(value, bool) or not isinstance(value, int) or value < low: raise err(f"{what} 는 {low} 이상의 **정확한 정수**여야 한다 — " f"받은 것 {value!r} ({type(value).__name__})") return value def resolve_rounds(rounds) -> int: """이 주행이 **실제로 돌 라운드 수**. ★한 번만 정하고 두 곳에 같이 준다. 계약 해시(`acquisition_contract_sha`)와 실제 loop(`acquire_one`)가 서로 다른 수를 보면 **신원이 거짓말을 한다**. """ from app.modules.pipeline import coarse_type_pick as ctp if rounds is None: return int(ctp.MAX_ROUNDS) return _exact_int(rounds, what="rounds", err=BadRoundCount, low=1) def acquisition_is_complete(got: Any) -> bool: """한 구매의 **일이 끝났나** — 마지막 라운드가 상한·provider 로 빈손(retryable · 받은 후보 0)이면 아니다. 그런 답은 장부에 `incomplete` 로 남고 되쓰이지 않는다.""" if not isinstance(got, dict): return True rounds = list(got.get("rounds") or ()) if not rounds: return str(got.get("status") or "") != "retryable" last = rounds[-1] nxt = str(((last.get("decision") or {}).get("next")) or "") return not (nxt == "retryable" and not (last.get("downloaded_candidates") or ())) def pass_suffix(pass_no: int, resume_from: Optional[Dict[str, Any]]) -> Dict[str, Any]: """구매 신원의 pass 접미 — **한 곳**. 2차는 pass 번호와 **끝난 라운드**(질의 포함)의 지문을 접는다; 끝나지 못한 라운드(retryable · 받은 후보 0)는 안 센다(다시 돌 라운드다).""" if int(pass_no) <= 1: return {} done = [r for r in ((resume_from or {}).get("rounds") or ()) if str(((r.get("decision") or {}).get("next")) or "") != "retryable" or (r.get("downloaded_candidates") or ())] return {"pass": int(pass_no), "done_rounds": [(int(r.get("round_no") or 0), list(r.get("queries") or ())) for r in done]} def identity_of(one: Dict[str, Any], *, contract_sha: str, brief_inputs: Optional[Dict[str, Any]] = None) -> str: """이 대상의 **구매 신원**. ★같은 것을 두 번 안 사기 위한 열쇠다. Args: contract_sha: `reference_acquisition.acquisition_contract_sha(...)`. ★★**팩·검색 계약·라운드 수**를 지배하는 것 전부가 여기 있다. 앞 판은 이 모듈의 버전 문자열만 접었는데, 그러면 팩을 바꿔도 신원이 그대로라 **옛 구매가 영구히 되쓰인다**. 산식을 여기 다시 적지 않고 production 한 곳의 것을 받는다. ★재개가 이것으로 「이미 샀다」를 안다. 그래서 **질의와 좁힘 문안까지** 접는다 — 둘 중 하나만 바뀌어도 **다른 것을 사는 것**이기 때문이다. """ import hashlib import json t = dict(one.get("target") or {}) parts = { "subject_id": str(t.get("subject_id") or ""), "directive": str(t.get("directive_native") or ""), "terms": list(t.get("terms_native") or ()), "language_lock": str(t.get("language_lock_native") or ""), "narrow_hint": str(one.get("narrow_hint") or ""), "acquisition_contract": str(contract_sha), # ★★★저작 입력을 **신원에 접는다** (Codex 2026-09-02). 지역·시대· # 저작 팩이 바뀌었는데 옛 검색 결과를 되쓰면 **다른 것을 산 것**이 # 같은 것으로 읽힌다. 후처리 판만 바뀐 것은 여기 안 넣는다. "brief": dict(sorted((brief_inputs or {}).items())), "wiring": CONTRACT_VERSION, } raw = json.dumps(parts, sort_keys=True, ensure_ascii=False) return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] def _unbought(one: Dict[str, Any], why: str, code: str) -> Dict[str, Any]: """안 산 줄. ★**종결 상태 이름을 여기서 짓지 않는다.**""" from app.modules.pipeline import reference_acquisition as ra return { "subject_id": one["target"]["subject_id"], "status": ra.STATUS_RETRYABLE, # 예산 때문에 **다 못 봤다** "chosen": None, "chosen_path": "", "rounds": [], "candidates": [], "why": why, # ★사유는 **만드는 쪽이** 표시로 붙인다 — 문장을 글자로 뒤져서 # 뜻을 되짚으면 문안을 고칠 때마다 조용히 어긋난다 "why_code": code, "downstream_blocked": ra.downstream_blocked(ra.STATUS_RETRYABLE), } def _passthrough(row: Dict[str, Any], *, disposition: str, why: str) -> Dict[str, Any]: """안 산 줄을 **산 줄과 같은 모양**으로. ★원행을 그대로 안고 간다. ★★수만 남기면 나중 수동 수정 화면이 어느 줄이었는지 못 찾는다 — screen· producer payload·facet 좌표가 전부 원행에 있다 (Codex 2026-09-01). ★★★처분마다 **뜻이 다르다.** 「필요했는데 못 구했다」와 「애초에 필요 없었다」에 같은 상태를 넣으면 안 된다 — 앞 판이 넷 모두에 `STATUS_RETRYABLE` 을 넣어 비대상이 **검색 실패처럼** 보였다. """ from app.modules.pipeline import reference_acquisition as ra needed = disposition in NEEDS_A_REFERENCE # ★비대상은 **획득 결과 자체를 안 만든다** — 새 상태를 짓지도 않는다 st = ra.STATUS_RETRYABLE if needed else None return { "research_subject_id": str((row or {}).get("research_subject_id") or ""), "identity": None, # ★안 샀으니 구매 신원이 없다 "disposition": disposition, "status": st, "why_unbought": None, "why": why, "outcome": ra.acquisition_outcome(st) if needed else None, # ★어느 쪽이든 하류를 안 막는다 — 다만 이유가 다르다 "downstream_blocked": ra.downstream_blocked(st) if needed else False, "acquisition": None, "source_evidence": copy.deepcopy((row or {}).get("source_evidence") or {}), # ★★원행을 **그대로** — 여기가 복구의 유일한 근거다 "ledger_row": copy.deepcopy(row), } MAX_ACQUIRE_WORKERS = 8 # ★무제한 병렬 금지 (Codex 2026-09-03) — 설정이 더 크면 여기서 자른다 def resolve_workers(workers: Optional[int]) -> int: """대상 병렬 폭. ★설정(`grounding_acquire_workers`, 기본 4)에서 오고 1..MAX 로 자른다. 정확한 정수가 아니면 선다 — bool·문자열을 폭으로 접지 않는다.""" if workers is None: from app.core.config import settings workers = getattr(settings, "grounding_acquire_workers", 4) if isinstance(workers, bool) or not isinstance(workers, int) or workers < 1: raise BadPurchaseCap(f"대상 병렬 폭이 1 이상의 정확한 정수가 아니다 — {workers!r}") return min(int(workers), MAX_ACQUIRE_WORKERS) def run(ledger: Dict[str, Any], *, journal: Any, cap: int, workdir: Path, search: Callable[..., Dict[str, Any]], download: Callable[..., bool], judge: Callable[..., Dict[str, Any]], write_brief: Optional[Callable[..., Dict[str, Any]]] = None, rel_root: Optional[Path] = None, stop_check: Optional[Callable[[], None]] = None, rounds: Optional[int] = None, workers: Optional[int] = None) -> Dict[str, Any]: """장부 하나를 **끝까지** 처리한다. ★들어온 줄이 전부 어디로 갔는지 남는다. Args: ledger: `grounding_acquisition_ledger.merge(...)` 가 낸 **갈래 중립** 장부. ★갈래마다 잇는 법은 다르지만(엔티티 줄 · facet · phase3) 해소된 뒤의 모양은 하나다. journal: `grounding_chunk_journal.ChunkJournal`. ★상한·신원·재개를 **그것이** 본다 — 여기서 새로 세지 않는다. cap: 이 판의 **논리 구매 상한**. `buy_or_reuse` 가 잠금 안에서 센다. Returns: ``{"contract_version", "rows": [...], "purchases": {...}, "auto_completed", "not_applicable", "skipped", "refused", "ledger_rows"}`` ★`rows` 는 **대상 전부**다 — 산 것도 상한에 걸린 것도 같은 목록에 있고, 각각 `outcome` 과 raw `status`/`why` 를 같이 갖는다. 사람 검토 화면이 그것을 읽는다. """ from app.modules.pipeline import grounding_chunk_journal as cj from app.modules.pipeline import reference_acquisition as ra from app.modules.pipeline import reference_acquisition_rounds as rr # ★비용 문과 라운드 수는 **아무것도 열기 전에** 검사한다 cap_n = _exact_int(cap, what="cap", err=BadPurchaseCap, low=0) used_rounds = resolve_rounds(rounds) # ★★pass 마다 구매 신원이 다르다 — 그래서 **부르는 쪽**이 상한을 대상 수 × pass 수로 # 준다 (`reference_acquisition_step._cap`, Codex BLOCK 2026-09-02). 여기서는 받은 수를 # 그대로 센다 — 두 곳에서 곱하면 한쪽만 고쳐진다. # ★★★장부에 **지금 대상이 아닌** 주체의 자리(옛 대상 구조 — 예: 조각별 아웃룩 rs_…)가 남아 # 있으면 그만큼 상한을 더 연다 (실측 2026-09-03 새벽, attempt bb85e798: 옛 아웃룩 조각 자리 7이 # 상한 34 를 먹어 O02·L02 의 2차가 굶었다). 상한의 뜻은 「지금 대상 × pass」다. # ★★신원을 지배하는 것은 **production 한 곳**이 만든다 — 여기서 팩·라운드 # 산식을 다시 적으면 팩을 바꿔도 옛 구매가 되쓰인다. 그리고 **같은 수**를 # 해시와 실제 loop 양쪽에 준다. sha = ra.acquisition_contract_sha(rounds=used_rounds) # ★저작기가 붙어 있으면 **그 입력**도 신원의 일부다 brief_identity = getattr(write_brief, "identity_inputs", None) or {} if callable(brief_identity): brief_identity = brief_identity() # ★★입구는 **하나**다 — 여기서 장부를 다시 해석하지 않는다. plan = _aa.inputs_from_ledger(ledger) before = {"bought": int(journal.bought()), "reused": int(journal.reused())} out_rows: List[Dict[str, Any]] = [] cap_reached = False # ★옛 대상 구조의 자리를 세어 상한을 그만큼 넓힌다 — 지금 대상의 pass 가 굶지 않게 known = {str((one.get("target") or {}).get("subject_id") or "") for one in plan["targets"]} _orphan = set() for _k, _e in (getattr(journal, "entries", {}) or {}).items(): sl = str(_e.get("slot") or "") if (sl and _e.get("epoch") == getattr(journal, "epoch", None) and _k not in (getattr(journal, "transfers", {}) or {}) and _e.get("status") in ("ok", "uncertain", "reserved") and sl.rsplit(":", 1)[0] not in known): _orphan.add(sl) orphan_slots = len(_orphan) # ★★상한을 **넓히지 않는다** (PR #82 리뷰 2026-09-03 · 메모리 「코드가 스스로 승인 범위를 넓히지 않는다」). 앞 판은 # `cap_n += orphan_slots` 로 승인한 수를 장부 이력만큼 늘렸다. 대신 상한은 **지금 대상의 자리만** 센다 # (`journal.reserve(slot_universe=known)`) — 고아 자리는 세지도, 상한을 늘리지도 않는다. 수는 감사용으로 남긴다. # ★★대상 병렬 (2026-09-03 · Codex 계약): bounded pool · 1차 전원 join 뒤 2차 · 결과는 원래 index 에 · # 워커는 {ident, result} 만 돌려주고 공유 상태를 안 만진다 · 상한의 정본은 `journal.reserve`(원자) · # 예산·정지 표는 `bind_current_research_budget` 로 워커에 실어 보낸다(thread-local 이라 그냥 넘기면 못 본다). n_workers = resolve_workers(workers) _sids = [str((one.get("target") or {}).get("subject_id") or "") for one in plan["targets"]] _dup = sorted({s_ for s_ in _sids if _sids.count(s_) > 1}) if _dup: raise ValueError(f"research_subject_id 가 겹친다 — {_dup}. 같은 자리를 둘이 사면 장부가 갈린다") idents: Dict[str, str] = {} # subject_id → 마지막 pass 의 구매 신원 (main thread 만 쓴다) import threading as _th abort = _th.Event() # ★첫 fatal 뒤 — 아직 안 산 워커는 문 앞에서 돌아선다 def _acquire(one: Dict[str, Any], *, rounds: int, pass_no: int, resume_from: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """한 대상을 **한 pass** 만큼 산다. ★1차 전원 → 2차 순서 (Codex 2026-09-02): 앞 판은 대상마다 두 라운드를 다 돌아 뒤 대상(O03)이 검색 문 앞에서 굶었다.""" # ★★대상별 저작 지문도 접는다 — 갈래·부류·표기·생김새·원문 언어가 # 바뀌면 **다른 질의가 나간다**. 목록은 저작기 한 곳이 안다. _per_target = getattr(write_brief, "target_identity", None) # ★★2차 신원에는 **끝난 라운드의 지문**을 접는다 (실측 2026-09-02 밤): 앞 판의 2차가 # 상한에 막혀 빈손으로 끝난 것이 같은 신원으로 되쓰여 영원히 retryable 이었다. 끝난 # 라운드가 달라지면 다른 구매다. 자리(`slot`)는 대상×pass 하나라 상한을 안 늘린다. # ★pass 접미(pass 번호 + 끝난 라운드 지문)는 **한 곳**에서 만들어 새 신원과 옛(legacy) 신원에 # 똑같이 붙인다 (Codex BLOCK 2026-09-03: legacy 에 done_rounds 가 빠져 2차 구매를 못 이었다). suffix = pass_suffix(pass_no, resume_from) ident = identity_of( one, contract_sha=sha, brief_inputs={**dict(brief_identity), **({"target_brief": _per_target(dict(one.get("target") or {}))} if callable(_per_target) else {}), **suffix}) slot = f"{one['target']['subject_id']}:p{pass_no}" if abort.is_set(): # ★★Codex BLOCK (2026-09-03): fatal 뒤 executor 가 빈 워커에 다음 job 을 **바로** 시작한다 — # cancel 만으로는 workers-1 개가 새는다. 워커가 스스로 provider 앞에서 돌아선다. return {"ident": ident, "result": _unbought(one, "앞 대상의 fatal 뒤 — 이 판을 접는다", WHY_ABORTED)} # ★★옛 신원(저작 payload 전체 해시)으로 산 줄이 있으면 새 뼈대 신원에 **이관(alias)**한다 — # append-only 사건으로, 다시 사지 않는다 (2026-09-02 밤: 신원 계약을 바꾸며 옛 38건 보존). _legacy_t = getattr(write_brief, "legacy_target_identity", None) _legacy_b = getattr(write_brief, "legacy_identity_inputs", None) if callable(_legacy_t) and journal.get(ident) is None: try: legacy = identity_of( one, contract_sha=sha, brief_inputs={**dict(_legacy_b or {}), "target_brief": _legacy_t(dict(one.get("target") or {})), **suffix}) except Exception: # noqa: BLE001 — 옛 재료가 없으면 이관할 것도 없다 legacy = None old = journal.get(legacy) if legacy and legacy != ident else None # ★★검색 계약의 **뜻**이 바뀌었다(세부 → 뼈대). 옛 판이 **고른** 답만 잇는다 — 옛 판의 # 못 찾음(retryable/no_match)을 새 넓은 검색의 완료 답으로 접으면 새 검색이 아예 안 나간다 # (Codex BLOCK 2026-09-03). 못 찾은 옛 후보는 장부 줄에 그대로 남아 사람 검토 재료다. if isinstance(old, dict) and str(old.get("status") or "") == ra.STATUS_SELECTED: journal.alias(ident, legacy, why="구매 신원 계약이 뼈대로 바뀜(2026-09-02) — 옛 판이 고른 답을 잇는다") # ★★멈춤과 provider 실패를 **예외 종류로 못 가른다** — 둘 다 같은 # 자리에서 올라온다. 그래서 「보내기 시작했나」를 흐름으로 표시한다. sent = {"tried": False} def _send(_one=one, _sent=sent): _sent["tried"] = True return rr.acquire_one( _one["target"], workdir=Path(workdir), rel_root=rel_root, search=search, download=download, judge=judge, write_brief=write_brief, # ★저작기가 들고 있는 시대 조각을 그대로 넘긴다 — 여기서 # 다시 만들면 두 벌이 되고 한쪽만 고쳐진다 era_tokens=getattr(write_brief, "era_tokens", None), rounds=rounds, narrow_hint=_one.get("narrow_hint"), resume_from=resume_from) # ★상한의 정본은 `journal.reserve` 의 원자 문이다 — 「앞 대상이 닿았다」 flag 를 워커끼리 # 나누지 않는다(경합). 닿은 뒤의 워커는 문 앞에서 거절되고 네트워크를 안 쓴다. try: got = cj.buy_or_reuse(journal, ident, cap=cap_n, slot=slot, send=_send, stop_check=stop_check, complete=acquisition_is_complete, slot_universe=known) except cj.BudgetExceeded as exc: got = _unbought( one, f"이 판의 구매 상한 {cap_n} 에 닿았다 ({exc})", WHY_CAP_REACHED) except cj.NeedsHumanDecision as exc: # ★★「샀는지 모른다」를 **다시 사지 않는다** — 그러면 상한을 넘긴다. 그렇다고 사람을 # 기다리지도 않는다. 참조 없이 가고, 장부의 `uncertain` 은 운영자 도구가 나중에 정리한다. got = _unbought(one, f"앞 판의 이 신원을 못 매듭지었다 ({exc})", WHY_PRIOR_UNCONFIRMED) except BaseException as exc: # noqa: BLE001 if not sent["tried"] or must_not_swallow(exc): # ★멈추라는 말·주인 잃음·내 코드 결함은 **그대로 올린다** — 보내기 전에 난 것도 마찬가지. raise got = _unbought(one, f"보냈는데 답을 못 받았다 ({exc})", WHY_CURRENT_UNCONFIRMED) return {"ident": ident, "result": got} def _pool(jobs: List[tuple]) -> Dict[int, Dict[str, Any]]: """`[(index, one, kwargs), …]` 를 bounded pool 로 돈다. ★모두 join 한 뒤에만 돌아온다 — 완료 순서와 무관하게 index 로 돌려준다. 워커 예외는 그대로 올린다(삼키지 않는다).""" from concurrent.futures import ThreadPoolExecutor, as_completed, wait from app.core.research_call_budget import bind_current_research_budget _inner = bind_current_research_budget(_acquire) def bound(*a, **kw): try: return _inner(*a, **kw) except BaseException: abort.set() # ★다른 워커들이 provider 앞에서 돌아서게 raise got: Dict[int, Dict[str, Any]] = {} if not jobs: return got # ★★Codex BLOCK (2026-09-03 03:20): index 순으로 result() 하면 뒤 index 의 fatal 예외를 앞 index 가 # 끝날 때까지 못 보고, with 를 빠져도 pending future 는 취소되지 않아 17개가 계속 산다. # → 끝나는 대로 관측(as_completed) · 첫 fatal 에 **아직 시작 안 한 것은 cancel** · 이미 도는 것 # (최대 workers 개)은 join 해 장부를 닫고 · 원예외 재발생. ex = ThreadPoolExecutor(max_workers=min(n_workers, len(jobs))) futs = {ex.submit(bound, one, **kw): i for i, one, kw in jobs} fatal: Optional[BaseException] = None try: for fut in as_completed(list(futs)): i = futs[fut] try: got[i] = fut.result() except BaseException as exc: # noqa: BLE001 — 첫 fatal 만 잡아 나머지를 끊는다 fatal = exc break finally: if fatal is not None: cancelled = [f for f in futs if f.cancel()] running = [f for f in futs if not f.done()] wait(running) # 이미 도는 것만 join — 장부가 닫힌다 logger.warning("중앙 획득 병렬: fatal 뒤 대기 중 %d개 취소 · 도는 %d개 join", len(cancelled), len(running)) ex.shutdown(wait=True) if fatal is not None: raise fatal return got # pass 1 — 모든 대상이 1차를 한 번씩 먼저 받는다 first = _pool([(i, one, {"rounds": 1, "pass_no": 1}) for i, one in enumerate(plan["targets"])]) results: List[Dict[str, Any]] = [] for i, one in enumerate(plan["targets"]): idents[_sids[i]] = first[i]["ident"] results.append(first[i]["result"]) # pass 2 — 1차 **전원이 끝난 뒤에만** (공정성 · Codex 2026-09-02) · 못 구했고 실제로 샀고 안 골랐으면 if int(used_rounds) > 1: second_jobs = [] for i, one in enumerate(plan["targets"]): got = results[i] if (str(got.get("status") or "") == ra.STATUS_RETRYABLE and not got.get("why_code") and not got.get("chosen") and (got.get("rounds") or ())): second_jobs.append((i, one, {"rounds": int(used_rounds), "pass_no": 2, "resume_from": got})) second = _pool(second_jobs) for i, _one, _kw in second_jobs: got = results[i] idents[_sids[i]] = second[i]["ident"] two = second[i]["result"] if two.get("why_code"): # ★★2차가 상한·미매듭 신원에 막혔다 — **1차 결과를 덮지 않는다**. 왜 못 갔는지만 적는다. results[i] = {**got, "pass2_skipped_why_code": two.get("why_code"), "pass2_skipped_why": two.get("why_unbought") or two.get("why")} else: results[i] = two # ★워커는 flag 를 안 만진다 — 「상한에 닿았나」는 결과에서 센다 cap_reached = any(WHY_CAP_REACHED in (str(r.get("why_code") or ""), str(r.get("pass2_skipped_why_code") or "")) for r in results) for one, got in zip(plan["targets"], results): ident = idents.get(str(one["target"]["subject_id"]), "") status = str(got.get("status") or ra.STATUS_RETRYABLE) code = got.get("why_code") out_rows.append({ "research_subject_id": one["target"]["subject_id"], "identity": ident, "disposition": (DISP_CAP_REACHED if code == WHY_CAP_REACHED else DISP_UNCONFIRMED if code else DISP_ACQUIRED), # ★raw 사유를 **접기 전 상태 그대로** 남긴다 "status": status, "why_unbought": code, "why": row_why(got), # ★하류가 보는 것은 **정책 함수가 접은 것**이다 "outcome": ra.acquisition_outcome(status), "downstream_blocked": ra.downstream_blocked(status), # ★★고증 축을 **행 위로** 올린다 — 소비자는 `acquisition` 안을 # 안 뒤진다. 없으면 미확인으로 접히지만, 적어야 왜 안 붙었는지 # 검토 화면이 읽는다 (Codex BLOCK 2026-09-02). "grounding_fidelity": copy.deepcopy( got.get("grounding_fidelity")) or None, "acquisition": copy.deepcopy(got), # ★증거를 같이 둔다 — 검토 화면이 「왜 이걸 샀나」를 읽는다 "source_evidence": copy.deepcopy(one.get("source_evidence") or {}), "ledger_row": copy.deepcopy(one.get("ledger_row")), }) # ★여기까지가 **구매 갈래**다 — 아래 갈래를 넣기 전에 세어 둔다 bought_lane = len(out_rows) # ★★안 산 갈래도 **같은 목록에** 같은 모양으로 넣는다. 따로 두면 덮개 # 검사가 한쪽만 보고, 수만 남으면 원행이 사라진다. for r in plan["skipped"]: out_rows.append(_passthrough(r["ledger_row"], disposition=DISP_SKIPPED, why=str(r.get("why") or ""))) for r in plan["refused"]: out_rows.append(_passthrough(r["ledger_row"], disposition=DISP_REFUSED, why=str(r.get("why") or ""))) for r in plan["auto_completed_rows"]: out_rows.append(_passthrough( r, disposition=DISP_AUTO_DONE, why="사야 하는 줄인데 소유 인물을 못 붙였다 — 참조 없이 간다")) for r in plan["not_applicable_rows"]: out_rows.append(_passthrough( r, disposition=DISP_NOT_APPLICABLE, why="애초에 참조를 살 줄이 아니다")) by_disp = {d: sum(1 for r in out_rows if r["disposition"] == d) for d in DISPOSITIONS} return { "contract_version": CONTRACT_VERSION, "rows": out_rows, # ★수는 **행에서 파생**한다 — 따로 세면 두 수가 갈린다 "dispositions": by_disp, # ★★회계가 **덜 산 판임을 스스로 말한다** — 다 산 판인 척하지 않는다 # ★★이름이 **뜻을 그대로** 말해야 한다. 앞 판은 `attempted` 가 # 「대상 수」였는데 실제 전송 수로 읽혔다 (Codex 2026-09-01). "purchases": { "cap": cap_n, # ★옛 대상 구조의 자리 — 그만큼 상한을 넓혔다(지금 대상 × pass 가 뜻) "orphan_slots": int(orphan_slots), # ★상한이 몇 pass 를 덮는지 — 부르는 쪽이 곱해서 줬어야 한다 "cap_passes": int(used_rounds), "rounds": used_rounds, "acquisition_contract": sha, "targets_total": len(plan["targets"]), "targets_processed": bought_lane, # ★이번 판에서 **provider 경계를 지난 수** — 장부 눈금의 차이다 "dispatch_attempted_this_run": int(journal.bought()) - before["bought"], "reused_this_run": int(journal.reused()) - before["reused"], # ★이 판(epoch) **누계** — 재개해도 안 줄어든다. 답을 못 받은 # 것도 「보냈다」로 세므로 이름에 그대로 적는다 "bought_or_uncertain_epoch": int(journal.bought()), "cap_reached": cap_reached, "unbought_after_cap": by_disp[DISP_CAP_REACHED], "unconfirmed_prior_calls": sum( 1 for r in out_rows if r["why_unbought"] == WHY_PRIOR_UNCONFIRMED), "unconfirmed_current_calls": sum( 1 for r in out_rows if r["why_unbought"] == WHY_CURRENT_UNCONFIRMED), }, # ★★안 산 갈래도 **전부 남긴다** — 안 남기면 왜 안 샀는지 사라진다. # 이 수들은 위 `rows` 에서 파생한 것이고, 원행은 그 줄에 있다. "auto_completed": by_disp[DISP_AUTO_DONE], "not_applicable": by_disp[DISP_NOT_APPLICABLE], "skipped": by_disp[DISP_SKIPPED], "refused": by_disp[DISP_REFUSED], "ledger_rows": plan["ledger_rows"], } def ledger_coverage(ledger: Dict[str, Any], result: Dict[str, Any]) -> Dict[str, Any]: """★**공개 끝점** — 장부의 모든 줄이 산출에 **정확히 한 번** 있나. ★★앞 판은 자동완료·비대상을 **수로만** 냈다. 그러면 이 검사가 통과할 수가 없고, 나중 수동 수정 화면이 원행을 못 찾는다 (Codex 2026-09-01). Returns: ``{"ok", "missing", "duplicated", "extra", "ledger", "result"}`` — `ok` 가 참이 아니면 어딘가에서 줄이 사라졌거나 겹쳤다. """ from collections import Counter want = [str((r or {}).get("research_subject_id") or "") for r in ((ledger or {}).get("rows") or ())] got = Counter(str(r.get("research_subject_id") or "") for r in (result.get("rows") or ())) missing = sorted(set(want) - set(got)) dup = sorted(k for k, n in got.items() if n > 1) extra = sorted(set(got) - set(want)) return {"ok": not (missing or dup or extra), "missing": missing, "duplicated": dup, "extra": extra, "ledger": len(want), "result": sum(got.values())} def unfinished_rows(result: Dict[str, Any]) -> List[Dict[str, Any]]: """★**공개 끝점** — 사람을 기다리는 줄이 있나. 있으면 결함이다. 상한·실패·못 찾음 어느 쪽이든 `reference_unavailable` 로 접혀 하류가 참조 **없이** 내려가야 한다. 이 함수가 빈 목록이 아니면 어딘가에서 주행이 사람 입력을 기다리게 된다. """ from app.modules.pipeline import reference_acquisition as ra out = [] for r in result.get("rows") or []: disp = r.get("disposition") if disp not in DISPOSITIONS: out.append(r) # 모르는 처분 — 매듭이 아니다 continue if r.get("downstream_blocked"): out.append(r) # 하류를 막는다 = 사람을 기다린다 continue # ★★두 축을 **합치지 않는다** (Codex 2026-09-01). 참조가 필요했던 # 줄만 종결 상태를 요구하고, 비대상은 결과가 **없어야** 매듭이다. if disp in NEEDS_A_REFERENCE: if r.get("outcome") not in (ra.STATUS_SELECTED, ra.STATUS_UNAVAILABLE): out.append(r) elif r.get("outcome") is not None: out.append(r) return out #: 재판정 신원의 **앞머리** — 구매 신원과 절대 안 겹치게. REPLAY_IDENTITY_PREFIX = "replay:" def replay_identity(base_identity: str, inputs: Dict[str, Any]) -> str: """재판정 **전용** 신원. ★원 구매 신원과 한 칸도 안 겹친다. ★★★계약 (Codex 2026-09-02): 원 acquisition 신원 + **순서 있는** 후보 좌표/URL + 실제 파일 SHA + 판정·경로해석 계약을 접는다. 같은 replay 를 다시 돌리면 **VLM 0회**여야 한다. """ import hashlib import json blob = json.dumps({"base": str(base_identity), "inputs": inputs}, ensure_ascii=False, sort_keys=True, separators=(",", ":")) return (REPLAY_IDENTITY_PREFIX + hashlib.sha256(blob.encode("utf-8")).hexdigest()[:24]) def rejudge_rows(rows: Sequence[Dict[str, Any]], *, journal: Any, root: Path, judge: Callable[..., Dict[str, Any]], cap: int, stop_check: Optional[Callable[[], None]] = None ) -> Dict[str, Any]: """이미 **받아 둔** 후보만 다시 판정한다. ★검색 0 · 받기 0. ★★★왜 (실측 2026-09-02 유료 canary ①): 검색 10회·받기 38장을 다 사고 12대상 **전부**가 `판정 실패: [Errno 2]` 로 떨어졌다 — 기록용 상대 경로를 판정이 cwd 기준으로 열었다. 산 것을 버리고 다시 사지 않으려면 **받아 둔 것으로 판정만** 다시 하는 자리가 있어야 한다. ★이 함수는 **조립만** 한다 — 무엇을 어떻게 되보는지는 `reference_acquisition_rounds.rejudge_cached` 한 곳이 안다. ★`search`·`download` 는 인자로도 안 받는다. 닿을 수 없어야 한다. Args: rows: `run()` 이 낸 `rows`. ★그중 되볼 것을 **여기서 고른다**. cap: 이 판의 **논리 판정 상한**. 닿으면 자동으로 안 넓히고, 이미 끝난 것을 **보존한 채** `cap_reached` 로 끝낸다. Returns: `{"rows", "replayed", "reused", "cap_reached", "skipped"}`. `rows` 는 들어온 것과 **같은 길이**다 — 안 되본 줄도 그대로 있다. """ from app.modules.pipeline import grounding_chunk_journal as cj from app.modules.pipeline import reference_acquisition_rounds as rr cap_n = _exact_int(cap, what="cap", err=BadPurchaseCap, low=0) out: List[Dict[str, Any]] = [] replayed = reused = skipped = 0 cap_reached = False for r in rows: acq = r.get("acquisition") or {} has = any((rd.get("downloaded_candidates") or ()) for rd in (acq.get("rounds") or ())) from app.modules.pipeline import reference_acquisition as _ra if r.get("disposition") != DISP_ACQUIRED or not has \ or acq.get("chosen") \ or _ra.is_terminal(str(r.get("status") or acq.get("status") or "")): # ★안 되보는 줄도 **그대로** 목록에 남는다 — 수만 남기면 원행이 # 사라지고 덮개 검사가 한쪽만 본다 out.append(r) skipped += 1 continue if cap_reached: out.append({**r, "replay_skipped": "이 판의 판정 상한에 닿았다"}) skipped += 1 continue # ★되볼 라운드를 고르는 규칙은 `rr` 한 곳이 안다 — 여기서 다시 # 적으면 둘이 갈린다 resolved = rr.resolve_cached_candidates( rr.cached_round(acq).get("downloaded_candidates") or (), root=root) rid = replay_identity(str(r.get("identity") or ""), rr.replay_identity_inputs(acq, resolved)) was = int(journal.bought()) sent = {"tried": False} def _send_rejudge(_a=acq, _sent=sent): _sent["tried"] = True return rr.rejudge_cached(_a, root=root, judge=judge) try: got = cj.buy_or_reuse( journal, rid, cap=cap_n, stop_check=stop_check, send=_send_rejudge) except cj.BudgetExceeded: cap_reached = True out.append({**r, "replay_skipped": "이 판의 판정 상한에 닿았다"}) skipped += 1 continue except cj.NeedsHumanDecision as exc: # ★★「샀는지 모른다」를 **다시 사지도, 판을 세우지도 않는다** — 중앙 lane 과 같다 # (실측 2026-09-02 밤: 재판정 하나가 uncertain 이라 스텝이 통째로 죽었다). # 이 줄은 재판정 없이 남고, 사람이 `settle_uncertain` 으로 정한 뒤 다음 재개가 산다. out.append({**r, "replay_skipped": f"앞 판의 이 재판정을 샀는지 모른다 ({exc})", "replay_why_code": WHY_PRIOR_UNCONFIRMED}) skipped += 1 continue except BaseException as exc: # noqa: BLE001 if not sent["tried"] or (must_not_swallow(exc) and not _is_provider_failure(exc)): raise # ★provider 가 답을 못 줬다(실측 2026-09-02 밤: Gemini 503 하나에 스텝이 통째로 죽었다). # 장부는 이미 refused/uncertain 으로 적혔다. 이 줄만 재판정 없이 남기고 다음으로. out.append({**r, "replay_skipped": f"보냈는데 답을 못 받았다 ({exc})", "replay_why_code": WHY_CURRENT_UNCONFIRMED}) skipped += 1 continue # ★**장부 눈금의 차이**로 센다 — 「돌려받았다」와 「실제로 판정했다」를 # 손으로 세면 재개가 VLM 0회인지 못 말한다 if int(journal.bought()) > was: replayed += 1 else: reused += 1 out.append({ **r, "status": str(got.get("status") or r.get("status")), "outcome": got.get("outcome"), "downstream_blocked": got.get("downstream_blocked"), "grounding_fidelity": copy.deepcopy( got.get("grounding_fidelity")) or None, "acquisition": copy.deepcopy(got), "replay_identity": rid, }) return {"rows": out, "replayed": replayed, "reused": reused, "cap_reached": cap_reached, "skipped": skipped, "★means": ("받아 둔 사진만 다시 봤다 — 검색 0 · 받기 0. " "**배선 판정**이지 고증 판정이 아니다")} class LaneAccountingBroken(RuntimeError): """두 lane 의 회계가 안 맞는다. ★모르는 채로 안 이어 간다.""" def reconcile_lanes(purchase: Any, replay: Any) -> Dict[str, Any]: """구매 장부에 섞인 **재판정 줄**을 회계에서 뺀다. ★줄은 안 지운다. ★★★왜 (실측 2026-09-02): 재판정을 처음에 구매 장부에 적었다. 원 줄을 지우지 않고 재판정 장부로 **복사**했지만, 복사만으로는 회계가 안 돌아온다 — `bought()` 가 같은 epoch 의 `replay:` 4줄까지 세어 구매 lane 이 12 가 아니라 **16** 으로 읽혔다 (Codex BLOCK). ★`replay:` 라는 **뜻은 여기가 안다**. `ChunkJournal` 은 「내 몫이 아니다」만 적을 줄 알고, 무엇이 어느 lane 인지는 모른다. Raises: LaneAccountingBroken: 옮길 줄이 **옮긴 곳에 없다** — 그러면 그 판의 답이 어디에도 없게 되므로 fail-closed 다. """ moved: List[str] = [] for ident in sorted(getattr(purchase, "entries", {}) or {}): if not str(ident).startswith(REPLAY_IDENTITY_PREFIX): continue if ident in (getattr(purchase, "transfers", {}) or {}): continue # ★이미 바로잡았다 if str(ident) not in (getattr(replay, "entries", {}) or {}): raise LaneAccountingBroken( f"재판정 줄 {ident[:20]} 이 구매 장부에 있는데 재판정 장부에 " f"없다 — 회계에서 빼면 그 판의 답이 **어디에도** 없게 된다. " f"먼저 옮겨 적어야 한다") purchase.transfer_out( ident, moved_to=str(getattr(replay, "path", "재판정 장부")), why=("재판정 줄이 구매 장부에 적혔다(장부를 안 가른 탓). 줄은 " "그대로 두고 **회계에서만** 뺀다 — 구매 lane 이 그만큼 더 " "산 것으로 읽히면 상한·보고가 틀어진다")) moved.append(ident) return lane_accounting(purchase, replay, moved=moved) def lane_accounting(purchase: Any, replay: Any, *, moved: Optional[Sequence[str]] = None) -> Dict[str, Any]: """두 lane 의 **유효 회계**. ★한 줄이 두 lane 에 겹쳐 세어지지 않는다.""" p_ent = set(getattr(purchase, "entries", {}) or {}) p_out = set(getattr(purchase, "transfers", {}) or {}) r_ent = set(getattr(replay, "entries", {}) or {}) both = sorted((p_ent - p_out) & r_ent) if both: raise LaneAccountingBroken( f"{len(both)}줄이 **두 lane 에 함께** 세어진다 ({both[:3]}) — " f"이중 합산이다") return { "acquisition_effective_bought": int(purchase.bought()), "replay_effective_bought": int(replay.bought()), # ★`or` 로 쓰면 **빈 목록**이 「안 준 것」과 같아진다 — # 두 번째 판이 「또 옮겼다」로 읽힌다 "moved_to_replay": (sorted(p_out) if moved is None else list(moved)), "double_counted": 0, "★means": ("구매 lane 은 산 참조의 수, 재판정 lane 은 다시 본 수다. " "옮긴 줄은 원 장부에 **그대로 있고** 회계에서만 빠진다"), }