"""참고 사진 획득 — GROUNDING-V2 §2-4. `entity_filter`(13.7) 뒤 · 13.8. ★★**참조를 사는 경계는 하나다** — `grounding_central_acquisition.run` 과 이 모듈의 공장 (`make_search/make_download/make_judge/make_writer`). 이 스텝(19.3)이 중앙 의무를 사고, 야외 구조물(21.915)은 **같은 경계·같은 공장**으로 `structure_form` 보충 의무만 산다 (2026-09-03 · 설계 §8). 검색·다운로드·VLM 선택 코드를 다른 자리에 두지 않는다 — 앞뒤가 각각 제 손으로 사면 같은 것을 두 번 산다. primitive 는 `search_grounded_ref` 에 이미 있다. **복사하지 않고 부른다.** 대상은 `generation_difficulty` 가 `hard`/`uncertain` 인 것뿐이다. ★`route` 로 고르면 안 된다 — 실측에서 아홉 축이 **전부** `research` 였다. """ from __future__ import annotations import logging from typing import Any, Dict, List, Optional from app.core.step_runner import StepRunner from app.core.steps.entity_steps import _EntityStepMixin logger = logging.getLogger(__name__) STEP_ID = "reference_acquisition" #: ★중앙 획득이 도는 라운드 수 — 좁혀 한 번 더. late(1라운드)와 다르다. #: 실제 loop 는 step 5 에서 넣는다. CENTRAL_ROUNDS = 2 class OwnerCoverageRefused(RuntimeError): """이 판의 대상 갈래가 **기대와 다르다**. ★provider 0 으로 선다.""" def assert_owner_coverage(obligations, required) -> Dict[str, Any]: """장부가 **그 갈래들을 다 갖고 있나**. ★공개 끝점. ★★이름·부분문자열로 안 본다 — 장부가 적은 `owner_type` 만 본다 (Codex 2026-09-02). Raises: OwnerCoverageRefused: 빠진 갈래가 있다. 조사를 **시작도 안 한다**. """ from app.modules.pipeline import grounding_central_inputs as ci got = set(ci.owners_reaching_the_door(obligations)) want = {str(x) for x in required} missing = sorted(want - got) if missing: raise OwnerCoverageRefused( f"이 판의 조사 대상에 갈래 {missing} 가 없다 — 있는 것 " f"{sorted(got)}. 기대한 것과 다른 것을 재게 되므로 provider 를 " f"안 부른다(inconclusive)") return {"required": sorted(want), "present": sorted(got)} # ── 획득 공장 — ★production 한 곳 (2026-09-03): 중앙 스텝(19.3)과 야외 보충(21.915)이 **같은 함수**를 부른다. # 스텝 메서드는 얇은 wrapper 다. 같은 규칙이 두 곳에 있으면 한쪽만 고쳐진다. #: 사는 lane 의 provider client 는 **SDK 재시도 0** — 물리 전송은 `FailoverOpenAIClient._invoke` 의 키 슬롯 loop 와 #: `llm_client._completion` 의 슬롯 loop **직전에서만** 예약되므로, 그 안의 SDK(max_retries 기본 2)·Router(num_retries 기본 3) #: 재시도는 안 세어진다. 0 으로 고정해야 예약 수 = 전송 수 (Codex PR #82 재리뷰 C). LANE_CLIENT_KWARGS = {"max_retries": 0} LANE_LLM_NUM_RETRIES = 0 def make_search(): from app.core.openai_keys import openai_client from app.modules.pipeline.era_research import MAX_CANDIDATES from app.modules.pipeline.search_grounded_ref import search_reference_images client = openai_client(**LANE_CLIENT_KWARGS) def _fn(**kw): kw.setdefault("max_results", MAX_CANDIDATES) return search_reference_images(client, **kw) return _fn def make_download(): from app.modules.pipeline.search_grounded_ref import download_candidate return download_candidate def world_facts_block_of(world) -> str: """창작자 확정 세계 사실 블록 — `build_grounding_world_facts` (다섯 갈래 공용 · 실패 안 삼킴).""" from app.core.world_context import build_grounding_world_facts return str(build_grounding_world_facts(world) or "") def make_writer(*, world, source_text: str, project_id: str, episode_id: str, step_id: str): """검색 지시문 **조사 저작기** (사용자 5단계 ①·②·④) — `grounding_target_research.make_writer`.""" from app.modules.pipeline import grounding_search_brief as gsb from app.modules.pipeline import grounding_target_research as gtr from app.core.openai_keys import openai_client co = gsb.coordinates_of(world) return gtr.make_writer( world_facts=world_facts_block_of(world), source_text=source_text, era=co["era"], region=co["region"], client=openai_client(**LANE_CLIENT_KWARGS), # ★SDK 재시도 0 (전송 수 = 예약 수) opik_metadata={"project_id": project_id, "episode_id": episode_id, "step": step_id}) def make_judge(*, project_id: str, episode_id: str, step_id: str): """거친 종류·가시성 + CRITERIA 일치·닮은 정도 **1심**. ★시대·국가·품질은 안 묻는다.""" from pathlib import Path as _P from app.modules.llm.llm_client import call_structured from app.modules.pipeline import coarse_type_pick as ctp from app.modules.pipeline.era_research import PICK_MODEL from app.modules.pipeline.multiroll_gemini import png_part pack = ctp.load_pack() sys_text = pack["stems"][ctp.SYSTEM_STEM]["content"].strip() def _fn(candidates, criteria: str = ""): parts = [{"type": "text", "text": ("THE KIND OF THING the photograph must show:\n" + str((candidates[0] or {}).get("kind_name") or ""))}] if criteria: parts.append({"type": "text", "text": "CRITERIA (what the photograph should show):\n" + criteria}) from app.core.config import settings as _settings for g in candidates: parts.append({"type": "text", "text": f"PHOTOGRAPH {g['index']}:"}) pth = _P(g["path"]) if not pth.is_absolute(): pth = _P(_settings.projects_dir).parent / pth parts.append(png_part(pth)) tag = "grounding_coarse_pick" raw = call_structured( tag, sys_text, parts, pack["stems"][ctp.SCHEMA_STEM]["content"], project_config={tag: {"model": PICK_MODEL}}, schema_name=tag, opik_metadata={"project_id": project_id, "episode_id": episode_id, "step": step_id}, num_retries=LANE_LLM_NUM_RETRIES) # ★Router 재시도 0 — 전송 수 = 예약 수 (Codex PR #82 C) return {PICK_MODEL: raw} return _fn #: ★검색·받기(outbound)를 **사는** 스텝의 계약 — 이 모듈의 공장(make_search/make_download)을 부르는 스텝 전부. #: canary 의 검색·받기 문은 「중앙이 끝났나」가 아니라 **이 목록 중 남은 것이 있나**로 연다 (Codex 2026-09-03 07:20 · 실측 f7cc45c576c0 #: 5판 dry: 중앙 completed 라 문 0 인데 야외 보충이 남아 있었다). 새 소비자가 공장을 import 하면서 여기 안 적으면 AST 시험이 선다. from app.modules.pipeline.grounding_outbound_consumers import OUTBOUND_CONSUMER_STEPS # noqa: E402 — 가벼운 모듈에서 다시 내보낸다 REPLAY_JOURNAL_NAME = "journal_replay.json" class ReferenceAcquisitionStep(_EntityStepMixin, StepRunner): """★어려운 대상마다 참고 사진 하나를 구해 둔다. 한 대상의 흐름: ``` 검색 지시문 저작 ← 「무엇을 찾을지」 자료 모으기 (판정 아님) ↓ 이미지 검색 (라운드당 `era_research.MAX_CANDIDATES` 장 = 4) ↓ 안전 다운로드 ↓ VLM 거친 종류 선택 ← 「자동차인지 화폐인지」만 ↓ 없으면 질의를 좁혀 **한 번 더** (총 2라운드) ↓ 그래도 없으면 no_match_after_retry — **terminal**. ★하류를 **막지 않는다** ``` ★★종결 상태가 하류를 막나 안 막나는 **여기서 정하지 않는다** — `reference_acquisition.acquisition_outcome` / `downstream_blocked` 한 곳이 정한다. 못 구하면 `reference_unavailable` 로 적고 참조 **없이** 내려간다 (사용자 확정 2026-08-31: 「HITL 을 무조건 필요한 요소로 하면 안 된다」). 이 줄은 HITL 을 없앨 때 **안 고쳐진 자리**였다 (Codex 2026-09-01). """ def _targets(self) -> List[Dict[str, Any]]: """이 주행이 참조를 살 대상. ★**최종 엔티티 집합** 위에서 고른다. ★두 자리를 **모두** 만족해야 한다: ① `grounding_plan` 이 `generation_difficulty` 로 연 것 ② `entity_filter` 를 **살아남은** 행 ①만 보면 필터가 지운 것까지 사고, ②만 보면 흔한 것까지 산다. """ from app.modules.pipeline.grounding_planner import reference_acquisition_ids plan = (self._load_prev_checkpoint("grounding_plan") or {}).get("data") or {} decided = plan.get("decided") or [] want = set(reference_acquisition_ids(decided)) if not want: return [] filt = (self._load_prev_checkpoint("entity_filter") or {}).get("data") or {} rows = filt.get("filtered_entities") or {} # ★표면형이 아니라 **subject id** 로 잇는다. 이름으로 이으면 추출이 # 이름을 줄인 순간 못 찾는다. by_subject = {str(d.get("research_subject_id") or ""): d for d in decided} out: List[Dict[str, Any]] = [] for key, items in (rows.items() if isinstance(rows, dict) else []): for e in (items or []): sid = str(e.get("research_subject_id") or "") # ★`materialize_missing_entities` 가 만든 행은 sid 를 갖는다. # 추출이 건진 행은 안 가지므로 표면형으로 한 번 더 본다. if not sid: sid = self._sid_for_surface(e.get("name"), by_subject) if sid and sid in want: out.append({**e, "entity_key": key, "research_subject_id": sid, "decided": by_subject.get(sid) or {}}) return out @staticmethod def _sid_for_surface(name, by_subject) -> str: """이름으로 subject 를 찾는다. ★**완전성 확인**이지 뜻 판단이 아니다. `grounding_overlay.missing_candidates` 와 **같은 규칙**을 쓴다 — 「고무줄로 묶인 회수권 뭉치」가 「회수권 뭉치」로 줄어도 잡는다. """ from app.modules.pipeline.grounding_overlay import _norm want = _norm(str(name or "")) if not want: return "" for sid, d in by_subject.items(): form = _norm(str(d.get("_surface_form") or d.get("surface_form") or "")) if form and (form in want or want in form): return sid return "" # ───────────────────────────────────────────────────────────── # 중앙 갈래 — **의무를 조사 앞에서** 세우고 한 곳에서 조사한다 # ★2026-09-01 켜졌다: `v2_chunk` 가 `GROUNDING_MODES` 로 옮겨졌고 # `_execute` 가 `uses_chunk_producer` 로 이 갈래를 부른다. 앞 판의 # 「아직 못 켠다 · inert」 주석은 사실이 아니게 되어 지운다. # ───────────────────────────────────────────────────────────── def central_obligations(self) -> Dict[str, Any]: """이 주행의 **의무 장부**. ★CP 셋을 읽어 합치기만 한다.""" from app.modules.pipeline import grounding_central_inputs as ci return ci.build_obligations( self._load_prev_checkpoint("grounding_screen"), self._load_prev_checkpoint("grounding_plan"), self._load_prev_checkpoint("outlook_phase3")) def central_result(self, obligations: Dict[str, Any], *, journal, cap: int, workdir, search, download, judge, write_brief=None, rel_root=None, stop_check=None, rounds: Optional[int] = None) -> Dict[str, Any]: """의무 장부 → **조사 결과**. ★조사기는 하나뿐이다. ★부르는 것을 **주입받는다** — 이 스텝이 provider 를 직접 안 연다. """ from app.modules.pipeline import grounding_central_acquisition as ca return ca.run(obligations, journal=journal, cap=cap, workdir=workdir, rel_root=rel_root, search=search, download=download, judge=judge, write_brief=write_brief, stop_check=stop_check, rounds=rounds) def central_rejudge(self, obligations: Dict[str, Any], result: Dict[str, Any], *, journal, cap: int, root=None, stop_check=None) -> Dict[str, Any]: """이미 **받아 둔** 후보만 다시 판정한다. ★검색 0 · 받기 0. ★★★왜 (실측 2026-09-02 유료 canary ①): 검색 10회·받기 38장을 다 사고 12대상 **전부**가 `판정 실패: [Errno 2]` 로 떨어졌다. 산 것을 버리고 다시 사지 않으려면 받아 둔 것으로 **판정만** 다시 하는 자리가 있어야 한다. ★이 메서드는 **조립만** 한다 — `search`·`download` 를 안 만들고, 중앙의 `rejudge_rows` 에게 넘기지도 않는다(그 함수는 받지도 않는다). """ from app.core.config import settings from pathlib import Path as _P from app.modules.pipeline import grounding_central_acquisition as ca # ★★★두 lane 의 회계를 **먼저** 바로잡는다 (Codex BLOCK 2026-09-02). # 앞 판이 재판정을 구매 장부에 적어, 구매 lane 의 `bought()` 가 12 가 # 아니라 16 으로 읽혔다. 줄은 안 지우고 **덧붙여 정정**한다. lanes = ca.reconcile_lanes(self._journal(), journal) got = ca.rejudge_rows( result.get("rows") or (), journal=journal, root=_P(root or _P(settings.projects_dir).parent), judge=self._judge(), cap=cap, stop_check=stop_check) merged = {**result, "rows": got["rows"], "replay": {**{k: v for k, v in got.items() if k != "rows"}, "lanes": lanes}} return self.central_wrap(obligations, merged) @staticmethod def central_wrap(obligations: Dict[str, Any], result: Dict[str, Any]) -> Dict[str, Any]: """조사 결과 → **이 스텝의 CP 모양**. ★들어온 줄이 전부 남는다. ★못 구한 것도 `rows` 에 남는다 — 나중 UI 가 「무엇이 없었나」를 읽는다. """ from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import grounding_central_inputs as ci from app.modules.pipeline import grounding_reference_obligations as ro from app.modules.pipeline.reference_acquisition import ( ACQUISITION_CONTRACT_VERSION, STATUS_SELECTED) cov = ca.ledger_coverage(obligations, result) if not cov["ok"]: raise RuntimeError(f"의무가 산출에서 사라졌다 — {cov}") left = ca.unfinished_rows(result) if left: raise RuntimeError( f"사람을 기다리는 줄이 {len(left)}개다 — 이 판의 정책은 HITL 0") rows = result.get("rows") or [] # ★★★「애초에 참조가 필요 없었다」를 「못 구했다」로 세지 않는다 # (Codex 재현 · 09-01). 앞 판은 `outcome != selected` 로 세서 # `not_applicable`(결과가 **없는** 줄)까지 `unresolved` 에 넣었다 — # 갈라 놓은 구분을 이 자리에서 다시 뭉갠 것이다. # ★판단은 `acquisition_outcome_of` 한 곳이 한다. needed = [r for r in rows if ca.acquisition_outcome_of(r) is not None] # ★★★**산 것을 못 본 판은 끝난 판이 아니다** (실측 2026-09-02). # 유료 판이 검색 10회·받기 38장을 다 사고 12대상 전부 판정에서 # 죽었는데, 이 자리가 `failed_count: 0` 으로 **completed** 를 닫았다. # 그래서 재개가 `SKIP` 으로 지나가고 — 받아 둔 사진을 다시 볼 길이 # 없었다. 「없다」와 「못 봤다」를 여기서도 갈라야 한다: # `no_match_after_retry` 는 **다 보고 없었다**(끝난 것), # `retryable` 은 **못 봤다**(안 끝난 것)다. pending = ReferenceAcquisitionStep.pending_rows(result) # ★후보 있는 줄 + 후보 없는 retryable — 둘 다 미완료 # ★`completed_count` 는 **1 로 둔다** — `step_runner` 는 # `failed==0 → completed · completed>0 → partial · 아니면 failed` # 로 접는다(`step_runner.py:1717`). 0 으로 두면 `failed` 가 되어 # 하류가 통째로 막힌다. 우리가 원하는 것은 **`partial`** 이다: # 재개가 `RERUN_SELF` 로 다시 돌되 **cleanup 은 안 한다** — # 그래야 받아 둔 사진 38장과 장부가 안 지워진다. return { "completed_count": 1, "applicable_count": 1, # ★retryable 은 reference_unavailable 로 보여도 raw terminal 이 아니다 — completed 로 봉인하지 않는다 "failed_count": int(pending["total"]), "data": { # ★왜 안 끝났는지를 **산출에** 남긴다 — 수만 남기면 다음 판이 # 무엇을 다시 봐야 하는지 모른다 "rejudge_pending": int(pending["rejudge"]), "research_retry_pending": int(pending["research_retry"]), "pending_total": int(pending["total"]), "contract_version": ACQUISITION_CONTRACT_VERSION, "obligation_contract": ro.OBLIGATION_CONTRACT_VERSION, "inputs_contract": ci.CENTRAL_INPUTS_CONTRACT_VERSION, "wiring_contract": ca.CONTRACT_VERSION, "owners_present": list( ci.owners_reaching_the_door(obligations)), "rows": rows, "purchases": result.get("purchases") or {}, "dispositions": result.get("dispositions") or {}, # ★참조가 **필요했던** 줄만 대상이다 "target_count": len(needed), "selected_count": sum( 1 for r in needed if ca.acquisition_outcome_of(r) == STATUS_SELECTED), # ★옛 소비자가 읽는 칸 — **필요했는데** 못 구한 것만 "unresolved": [r["research_subject_id"] for r in needed if ca.acquisition_outcome_of(r) != STATUS_SELECTED], # ★애초에 살 것이 아니었던 줄 — 감사용으로 **따로** 센다 "not_applicable_count": len(rows) - len(needed), "ledger_rows": len(rows), }, } # ───────────────────────────────────────────────────────────── # provider 배선 — ★부품은 **전부 기존 것**이다. 여기서 새로 안 짓는다. # ───────────────────────────────────────────────────────────── def verify_completion(self): """★★★**산 것을 못 본 판은 끝난 판이 아니다** (실측 2026-09-02). 유료 판이 검색 10회·받기 38장을 다 사고 12대상 전부 판정에서 죽었는데 `completed` 로 닫혔다. 그러면 재개가 **`SKIP`** 으로 지나가고 — 받아 둔 사진을 다시 볼 길이 없다. 이 자리가 「아직 못 본 것이 있다」를 말해야 재개가 `RERUN_SELF` 로 다시 돈다. ★`origin='artifact_missing'` 이라 **cleanup 이 안 일어난다** — `force` 로 가면 `cleanup_artifacts`+`clear_checkpoint` 가 받아 둔 사진과 장부를 지운다. 산 것을 지우면 안 된다. """ from app.core.integrity_report import CompletionReport # ★★끝날 때의 검증은 **방금 만든 결과**를 본다 (실측 2026-09-03 새벽): runner 는 exit 검증을 # `save_checkpoint` **앞**에서 부르므로, 여기서 CP 를 되읽으면 **앞 판의** CP(미판정 1)를 보고 # 이번 결과(미판정 0)를 partial 로 접는다 — 구매 0 인 attempt 를 한 번 더 돌려야 닫혔다. # 재개 때(결과 없음)는 durable CP 가 진실이다. last = getattr(self, "_last_execute_result", None) if isinstance(last, dict) and isinstance(last.get("data"), dict): data = last["data"] else: cp = self._load_prev_checkpoint(STEP_ID) data = (cp or {}).get("data") or {} pending = self.pending_rows(data) if not pending["total"]: return CompletionReport(is_complete=True, missing=[], severity="clean", metadata={}, origin="clean") missing = [] if pending["rejudge"]: missing.append(f"판정 못 한 대상 {pending['rejudge']}개 — 받아 둔 사진이 있는데 아무것도 못 골랐다") if pending["research_retry"]: missing.append(f"다시 사야 할 대상 {pending['research_retry']}개 — 후보 없이 끝났다(정지·검색 실패)") # ★자동 재시도 빚이다 — 사람을 기다리지 않는다. 재개(RERUN_SELF)가 장부를 되쓰며 그 대상만 다시 산다. return CompletionReport( is_complete=False, missing=missing, severity="partial", metadata={"rejudge_pending": pending["rejudge"], "research_retry_pending": pending["research_retry"], "pending_total": pending["total"], "★means": ("재개가 자동으로 처리한다 — 후보 있는 줄은 판정만, 후보 없는 줄은 그 대상만 " "다시 산다. 이미 고른 줄은 장부 되쓰기(provider 0)")}, origin="artifact_missing") def _workdir(self): """받은 사진을 둘 자리. ★**서빙 가능한 곳**이어야 사람이 본다.""" from pathlib import Path as _P from app.core.config import settings d = (_P(settings.projects_dir) / self.project_id / "references" / "grounding" / self.episode_id) d.mkdir(parents=True, exist_ok=True) return d def _search(self): return make_search() @staticmethod def _download(): return make_download() def _world(self): return self._load_prev_checkpoint("visual_world_rules") or {} def _write_brief(self): """검색 지시문 저작기. ★**production 이 이것을 넘긴다** — 공장은 모듈 함수 `make_writer` 하나.""" return make_writer(world=self._world(), source_text=self._source_sample(), project_id=self.project_id, episode_id=self.episode_id, step_id=STEP_ID) def _world_facts_block(self, world) -> str: return world_facts_block_of(world) def _source_sample(self) -> str: """언어 판정용 원문 표본. ★`search_grounded_ref` 와 같은 상수로 자른다. ★이 자리만 발췌한다 — 사용자 2026-08-03 조건부 예외(정보 하나를 얻는 용도). 나머지 어디서도 원문을 안 자른다.""" cp = self._load_prev_checkpoint("text_cleanup") or {} data = cp.get("data") or {} return str(data.get("cleaned_text") or data.get("text") or "") def _judge(self): """거친 종류·가시성 **1심** — 공장은 모듈 함수 `make_judge` 하나.""" return make_judge(project_id=self.project_id, episode_id=self.episode_id, step_id=STEP_ID) #: 재판정 장부 이름 — 구매 장부와 **다른 장부**다 (`_journal` docstring). ★Codex BLOCK 2026-09-03: #: 공장 추출 때 이 상수를 잃어 재판정 경로가 AttributeError 로 죽을 자리였다. REPLAY_JOURNAL = REPLAY_JOURNAL_NAME def _journal(self, name: str = "journal.json"): """이 스텝의 장부. ★구매와 재판정은 **다른 장부**다. ★★★왜 갈랐나 (실측 2026-09-02): 재판정을 구매 장부에 적었더니 `reserve` 가 **이미 산 12건**을 제 상한에서 빼서, 12대상 중 **4개만** 되보고 멈췄다. 상한은 「이 판이 몇 번 부르나」인데 앞 판의 구매가 그것을 먹은 것이다. 문이 다르면 장부도 다르다. ★원 구매 줄은 **한 글자도 안 건드린다** (Codex 계약 2026-09-02). """ from app.core.config import settings from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline.grounding_chunk_journal import ChunkJournal from pathlib import Path as _P path = (_P(settings.projects_dir) / self.project_id / "checkpoints" / "episodes" / self.episode_id / STEP_ID / str(name)) path.parent.mkdir(parents=True, exist_ok=True) return ChunkJournal(path, contract={"wiring": ca.CONTRACT_VERSION}) def _cap(self, obligations) -> int: """이 주행의 **논리 조사 상한** = 대상 수 × pass 수. ★★pass 마다 구매 신원이 다르다(1차 전원 → 2차 못 구한 것만). 앞 판은 상한이 대상 수 N 이라 1차가 N 을 다 쓰고 **2차가 문 앞에서 전부 막혔다** — 시험은 `buy_or_reuse` 를 대역으로 바꿔 못 봤다 (Codex BLOCK 1, 2026-09-02). `grounding_reference_cap` 은 **대상 수** 단위로 받고 여기서 같이 곱한다.""" from app.modules.pipeline import grounding_acquisition_ledger as gl from app.modules.pipeline import grounding_central_acquisition as ca n = len(gl.acquisition_targets(obligations)) raw = (self.project_config or {}).get("grounding_reference_cap") per_target = int(raw) if isinstance(raw, int) and not isinstance( raw, bool) and 0 <= raw <= n else n return per_target * max(1, int(ca.resolve_rounds(None))) #: 논리 자리 하나가 부를 수 있는 **물리 전송**의 상한. SDK·Router 재시도는 0 으로 잠갔으므로(`LANE_CLIENT_KWARGS` · #: `LANE_LLM_NUM_RETRIES`) 전송은 예약 자리에서만 난다: 조사 1 + 검색 ≤2 + 판정 ≤2 × fallback 3 tier = 9 논리, #: 키 슬롯 failover 가 각각을 물리 둘로 만들 수 있어 ×2 = 18. `project_config["reference_transmission_cap"]` 이 이긴다. #: ★PR #82 리뷰 2026-09-03: 이 lane 은 `research_run_scope` 를 안 열어 물리 전송을 세는 자리가 **한 곳도 없었다**. #: ★Codex 재리뷰(C): 「논리×10」은 SDK·Router 재시도를 못 봤다 — 재시도를 0 으로 잠근 뒤의 셈이 이것이다. TRANSMISSIONS_PER_LOGICAL_SLOT = 18 #: 논리 상한을 모를 때(대상 0 · 시험 대역)의 물리 상한 — `grounding_research` 의 기본값과 같은 수 DEFAULT_TRANSMISSION_CAP = 60 def _transmission_cap(self, logical_cap) -> int: raw = (self.project_config or {}).get("reference_transmission_cap") if raw: return int(raw) try: n = int(logical_cap or 0) except (TypeError, ValueError): n = 0 return n * self.TRANSMISSIONS_PER_LOGICAL_SLOT if n > 0 else self.DEFAULT_TRANSMISSION_CAP def _central(self) -> Dict[str, Any]: """중앙 갈래 — 의무를 세우고 **한 곳에서** 조사한다. ★물리 전송 문(`research_run_scope`) 안에서 판다 (PR #82 리뷰).""" from app.core.config import settings from app.core.research_call_budget import research_calls_armed, research_run_scope ob = self.central_obligations() # ★★★**첫 outbound 전에** 이 판이 무엇을 재는지 확인한다 # (Codex 2026-09-02). 대상 갈래가 다르면 canary 가 **다른 것을** # 재는 것이라 그대로 두면 초록이 거짓말이 된다. # ★이름을 추론하지 않는다 — 장부의 `owner_type` 만 본다. want = (self.project_config or {}).get("required_owner_types") if want: assert_owner_coverage(ob, want) # ★물리 전송 문 — 이 블록 안의 조사·검색·판정 전송만 센다(worker 도 `bind_current_research_budget` 로 물려받는다). with research_run_scope(cap=self._transmission_cap(self._cap(ob))): with research_calls_armed(): res = self.central_result( ob, journal=self._journal(), cap=self._cap(ob), workdir=self._workdir(), rel_root=__import__("pathlib").Path(settings.projects_dir).parent, search=self._search(), download=self._download(), judge=self._judge(), write_brief=self._write_brief(), stop_check=getattr(self, "_stop_check", None)) # ★★★앞 판이 **사고 나서 판정에서 죽은** 줄이 있으면, 다시 사지 않고 # 받아 둔 사진으로 **판정만** 다시 한다 (Codex GO 2026-09-02). # ★검색·받기는 이 갈래에 안 넘어간다 — `central_rejudge` 가 받지도 # 않는다. 되볼 것이 없으면 아무 일도 안 한다(VLM 0). if self.rows_to_rejudge(res): return self.central_rejudge( ob, res, journal=self._journal(self.REPLAY_JOURNAL), cap=self._cap(ob), stop_check=getattr(self, "_stop_check", None)) return self.central_wrap(ob, res) @staticmethod def pending_rows(result: Dict[str, Any]) -> Dict[str, Any]: """**자동 재시도 빚** — 아직 raw terminal 이 아닌 줄을 둘로 갈라 센다 (Codex BLOCK 2026-09-03 05:30): · `rejudge`: 받아 둔 후보가 있는데 아직 못 본 줄 → 재개가 **사지 않고 판정만** 다시 한다 · `research_retry`: 후보가 하나도 없는 retryable(정지·검색 실패) → 재개가 **그 대상만** 다시 산다 둘 다 `total` 에 들어가 completed 봉인을 막는다. ★사람 대기가 아니다 — 재개가 자동으로 처리한다. ★실측 4398a55dc0bb: 정지 요청에 맞은 9줄이 후보 0 이라 `rows_to_rejudge` 에 안 세어져 completed 로 봉인됐고 plain resume 이 SKIP 으로 지나갈 자리였다. terminal(`selected` · `no_match_after_retry`)·이미 고른 줄·안 산 줄은 안 센다.""" from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import reference_acquisition as ra rejudge: List[str] = [] research_retry: List[str] = [] for r in (result.get("rows") or ()): acq = r.get("acquisition") or {} if r.get("disposition") != ca.DISP_ACQUIRED or acq.get("chosen"): continue if ra.is_terminal(str(r.get("status") or acq.get("status") or "")): continue sid = str(r.get("research_subject_id") or "") if any((rd.get("downloaded_candidates") or ()) for rd in (acq.get("rounds") or ())): rejudge.append(sid) else: research_retry.append(sid) return {"rejudge": len(rejudge), "research_retry": len(research_retry), "total": len(rejudge) + len(research_retry), "rejudge_subjects": rejudge, "research_retry_subjects": research_retry} @staticmethod def rows_to_rejudge(result: Dict[str, Any]) -> int: """되볼 줄이 **몇 개**인가. ★고르는 규칙을 한 곳에만 적는다. 받아 둔 후보가 있는데 **아직 못 본** 줄이다. 이미 고른 줄·애초에 안 산 줄·**다 보고 없었던 줄**은 세지 않는다 — 그래야 「되볼 것이 없으면 VLM 0」이 성립한다. ★★★`no_match_after_retry` 를 빼는 것이 중요하다 (실측 2026-09-02): 「고른 것이 없다」로만 세면 **다 보고 없었던 줄**까지 매 재개마다 다시 판정한다 — 같은 것을 계속 사는 것이고, 스텝이 영원히 `partial` 이다. """ from app.modules.pipeline import grounding_central_acquisition as ca from app.modules.pipeline import reference_acquisition as ra # ★규칙은 `pending_rows` 한 곳 — 여기서는 「후보 있는 줄」만 돌려준다(재판정기가 쓴다) return int(ReferenceAcquisitionStep.pending_rows(result)["rejudge"]) def _execute(self, mode="resume") -> Dict[str, Any]: from app.core.grounding_mode import (resolve_grounding_mode, uses_chunk_producer) from app.modules.pipeline.reference_acquisition import ( ACQUISITION_CONTRACT_VERSION, acquisition_contract_sha) # ★★★C(c) 판이면 **중앙 한 곳**이 조사한다. 옛 갈래와 **병존하지 # 않는다** — 술어가 배타적이라 한 판에서 둘 다 돌 수 없다. if uses_chunk_producer(resolve_grounding_mode(self.project_config)): return self._central() targets = self._targets() # ★★중앙 획득은 **2라운드 계약**이다 — 좁혀 한 번 더 찾는다. # 다만 실제 loop 는 아직 없다(step 5). 그래서 지금 이 스텝은 대상이 # 있으면 `NotImplementedError` 로 선다 — 키만 2라운드로 내고 실행은 # 1라운드인 상태를 **만들지 않는다**. contract = acquisition_contract_sha(rounds=CENTRAL_ROUNDS) if not targets: # ★「살 것이 없다」와 「못 샀다」는 다르다. 대상 0 은 정상이다. return {"completed_count": 1, "applicable_count": 1, "failed_count": 0, "data": {"contract_version": ACQUISITION_CONTRACT_VERSION, "acquisition_contract": contract, "targets": [], "records": [], "target_count": 0, "selected_count": 0, "unresolved": []}} # ★★여기서부터가 유료다. 승인 없이 안 돈다 — 지금은 대상만 세고 선다. raise NotImplementedError( "참조 획득 orchestration 은 아직 안 배선했다 — 대상 " f"{len(targets)}개. 유료 승인과 acceptance 가 먼저다")