"""C(c) producer — 구간당 **단일 판독** + compact evidence merge. ★아직 안 켠다. ## 무엇을 대신하나 지금은 원문 전문이 **네 번**(`grounding_a0` + `entity_extract_*` 셋), 샷 전부가 **세 번**(`entity_all_*`) 나간다. 이 스텝은 그것을 **구간 수 + 1** 로 줄인다 — 구간마다 한 번 읽고, 마지막에 한 번 합친다. ## ★아직 아무 데도 안 붙어 있다 `step_manifest` 에도 `STEP_CLASSES` 에도 없다. `uses_chunk_producer(mode)` 가 True 일 때만 도는데, 그 모드(`v2_chunk`)를 **아무도 안 쓴다.** 사람이 v3 검토 화면에서 샷 결속 의미·두 축·엔티티 품질을 보고 GO 를 준 뒤에 켠다. ## 이 파일이 **안 하는** 것 - 승인 없이 provider 를 부르지 않는다 — `uses_chunk_producer` 가 False 면 **한 푼도 안 쓰고** 돌아간다 - 구간 나누기·payload 조립·행 해석·합치기·adapter 를 **다시 적지 않는다**. 전부 이미 있는 함수를 부른다. 여기서 다시 적으면 도구가 재는 것과 주행이 굽는 것이 갈린다 - 샷 `description` 을 **자르지 않는다** """ from __future__ import annotations import logging from concurrent.futures import ThreadPoolExecutor, as_completed 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__) #: 구간 병렬 폭. ★`entity_t2i` 가 쓰는 것과 같은 관례(10)를 따른다. MAX_WORKERS = 10 #: Opik 좌표. ★신원은 **부모 trace metadata** 로 간다 — `opik_metadata` 의 #: 다른 키는 litellm 이 안 읽고, 태그는 축 whitelist 가 거른다. #: 한 주행의 **벽시계 마감**. ★상한과 별개다 — 상한은 「몇 번」, 이것은 #: 「언제까지」다. 팬아웃 안에서 이미 제출된 것들도 정지 표로 막힌다. CHUNK_DEADLINE_SECONDS = 900.0 #: ★★**한 호출**의 마감. 주행 마감만으로는 **매달린 provider 를 못 끊는다** — #: 정지 표는 *새* 전송 앞에서만 보이고, 이미 `router.completion` 안에서 #: 느린 stream 을 물고 있는 호출은 그대로 걸려 있다 (Codex 2026-08-31). #: `call_with_deadline` 이 별도 스레드로 돌려 **여기서** 끊는다. #: ★끊어도 그 daemon 스레드는 살아 있을 수 있다 — 그래서 늦게 끝난 결과가 #: **아무 데도 안 쓰이게** 해야 한다(장부는 `uncertain` 으로 남는다). CHUNK_CALL_DEADLINE_SECONDS = 300.0 #: ★★**실제 요청 계약**. `call_structured` 기본은 `enable_fallback=True` · #: `num_retries=None`(라우터 기본 재시도)이라, 안 적으면 **몇 번 나가는지 #: 모른 채** 산다. 못박고, **신원에도 접는다** — 재시도를 켠 판과 끈 판은 #: 같은 것이 아니다 (Codex 2026-08-31). REQUEST_CONTRACT: Dict[str, Any] = { "enable_fallback": False, "num_retries": 0, "temperature": 0.2, } #: 물리 전송 상한의 **기본값**. ★새 수를 지어내지 않는다 — 실제 값은 #: `project_config["research_transmission_cap"]` 이 준다(`grounding_research` #: 가 쓰는 **같은 SOT**). 여기 것은 그 키가 없을 때의 기본이다. #: #: ★앞 판은 `APPROVED_PHYSICAL_MAX = 24` 를 **내가 지어냈다** — 사용자 승인도 #: 설정 SOT 도 없는 수였다 (Codex 2026-08-31, 하드코딩 금지). DEFAULT_TRANSMISSION_CAP = 24 #: 한 논리 호출이 열 수 있는 **슬롯 배수**. 계약이 정한다 — #: `num_retries=0` · `enable_fallback=False` 라 남는 것은 **키 슬롯 loop** 뿐이고 #: `_completion` 이 `slot_count()` 만큼 돈다. #: ★재시도나 fallback 을 켜면 이 식이 **틀린다.** 그래서 계약과 같이 본다. def physical_per_logical(request_contract: Dict[str, Any], slots: int) -> int: """한 논리 호출이 여는 **물리 최대**. ★계약이 바뀌면 같이 바뀐다. ★글 호출 경로(`_completion`)에는 이미지 쪽 `reserve_current_call` 같은 **세는 자리가 없다.** 그래서 물리 수를 「막을」 수는 없고, 대신 **구조적으로 상한을 정한다** — 재시도와 fallback 을 끄면 남는 것은 키 슬롯 loop 뿐이고 그것은 `slot_count()` 로 위가 정해진다. ★이것은 **상한이지 실측이 아니다.** 실제로 몇 번 나갔는지는 Opik+provider 로그로 본다. """ tiers = 3 if request_contract.get("enable_fallback") else 1 retries = int(request_contract.get("num_retries") or 0) return max(1, slots) * tiers * (1 + retries) CHUNK_TRACE_NAME = "grounding_chunk_read" CHUNK_TAG = "op:grounding-chunk" #: 장부 줄 ↔ trace 를 잇는 키. ★감사 도구와 **같은 이름**이어야 대조가 된다. ID_META_KEY = "cc_call_identity" class GroundingChunkStep(_EntityStepMixin, StepRunner): """구간당 단일 판독 producer. ★`v2_chunk` 모드에서만 산다. ★★★`_EntityStepMixin` 을 **왜 쓰나** — `_load_prev_checkpoint` 때문이다. 이 스텝은 `scene_save`·`shot_validator`·`visual_world_rules` 체크포인트를 읽는데, 그 함수는 `StepRunner` 에 없고 스텝마다 각자 갖고 있다. 이웃인 `GroundingScreenStep` 도 같은 이유로 이 mixin 을 쓴다. ★실측 (2026-09-02 유료 canary ①): 이것이 없어서 첫 실주행이 `AttributeError: 'GroundingChunkStep' object has no attribute '_load_prev_checkpoint'` 로 죽었다. **시험은 CP 를 손으로 넣어 줘서** 이 자리를 한 번도 안 지났다 — producer 가 실제로 돈 적이 없었다. ★이 클래스가 제 `_config_hash` 를 갖고 있으므로 mixin 것에 안 가린다. """ def _config_hash(self) -> str: """★팩 **바이트**와 계약을 지문에 접는다. 안 접으면 팩·후처리 계약·adapter 를 고쳐도 resume 이 **옛 체크포인트를 그대로 건너뛴다** — 고친 것이 아무 데도 안 닿는다. 버전 문자열만 접으면 같은 버전 디렉토리의 **내용**을 고쳤을 때 못 잡는다. """ import hashlib import json as _json from app.core.step_runner import compute_config_hash from app.core.steps.grounding_steps import _pack_fingerprint from app.modules.pipeline import grounding_chunk as gc from app.modules.pipeline import grounding_chunk_adapter as ad from app.modules.pipeline import grounding_chunk_plan as cp payload = { "base": compute_config_hash(self.project_config), # ★★**사는 것을 지배하는 것**과 **해석을 지배하는 것**을 갈라 적는다. # 둘을 섞으면 후처리만 고쳐도 장부가 통째로 버려져 **유료 raw 를 # 다시 산다** (Codex BLOCK 2026-08-31). "acquisition": { **self._acquisition(), "pack_bytes": _pack_fingerprint("grounding_chunk", gc.CHUNK_PACK_VERSION), }, # ★정책은 따로 접는다 — 신원에 넣으면 정책만 바꿔도 재구매다 "admission": self._admission(), "processing": { "contract": gc.PROCESSING_CONTRACT_VERSION, "adapter": ad.ADAPTER_CONTRACT_VERSION, "chunk_plan": cp.CHUNK_PLAN_CONTRACT_VERSION, "bundle_target": cp.BUNDLE_TARGET, }, } return hashlib.sha256( _json.dumps(payload, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest()[:16] def _execute(self, mode: str = "resume") -> Dict[str, Any]: from app.core.grounding_mode import (resolve_grounding_mode, uses_chunk_producer) g_mode = resolve_grounding_mode(self.project_config) if not uses_chunk_producer(g_mode): # ★manifest 가 걸러야 정상이지만, 단독 호출로 들어와도 **한 푼도 # 안 쓰고** 선다. 「켜지 않았는데 샀다」가 없어야 한다. logger.info("grounding_chunk: mode=%s — 아무것도 안 한다", g_mode) return {"completed_count": 0, "applicable_count": 0, "failed_count": 0, "data": {"mode": g_mode, "skipped": True, "provider_calls": 0}} plan, segs, cats, world = self._plan() jr = self._journal(plan) # ★★마감을 **정지 표로** 깐다. loop 조건으로만 두면 팬아웃 안에서 # 이미 제출된 것들이 계속 나간다 (§8.5). `_completion` 이 키 전환 # 루프 **앞**에서 이 표를 보므로 네트워크 직전에 막힌다. from app.core.research_call_budget import research_run_scope # ★★**보내기 전에** 물리 상한을 본다. 넘으면 아무것도 안 산다. bound = self._assert_physical_bound(plan) # ★★`research_run_scope` 의 cap 은 **물리 전송 상한**이다 — 논리 수가 # 아니다. 예약은 `_completion` 의 슬롯 loop 안에서 **전송마다** 걸린다. with research_run_scope(cap=self._transmission_cap(), deadline_seconds=CHUNK_DEADLINE_SECONDS): return self._run(g_mode, plan, segs, cats, jr) def _run(self, g_mode, plan, segs, cats, jr): # ★앞 판에 **답을 못 받은** 호출이 있으면 자동으로 다시 안 산다. jr.assert_no_uncertain() rows, quarantined, _ = self._read_chunks(plan, segs, cats, jr) reduced = self._merge(rows, segs, jr) return self._wrap(g_mode, plan, rows, quarantined, reduced, jr) # ── 계획 ──────────────────────────────────────────────────────────── def _plan(self): """구간·원문·샷 catalog·세계 규칙. ★전부 **이미 있는 함수**로.""" from app.core.world_context import (build_grounding_world_facts, grounding_world_facts_ready) from app.modules.pipeline import grounding_chunk as gc from app.modules.pipeline import grounding_shot_catalog as sc from app.modules.pipeline.grounding_chunk_plan import ( assert_bundle_target_agrees, bundle_scenes) assert_bundle_target_agrees() scene_cp = self._load_prev_checkpoint("scene_save") or {} segments = (scene_cp.get("data") or {}).get("segments") or [] if not segments: from app.core.errors import AppError raise AppError(code="grounding_chunk.no_scenes", message="scene_save 산출이 없다", status_code=400) shot_cp = self._load_prev_checkpoint("shot_validator") or {} scenes = (shot_cp.get("data") or {}).get("scenes") or [] rules_cp = self._load_prev_checkpoint("visual_world_rules") or {} if not grounding_world_facts_ready(rules_cp): from app.core.errors import AppError raise AppError( code="grounding_chunk.no_world_facts", message=("세계 규칙이 아직 없다 — 빈 시대로 물으면 모델이 " "상상 묘사만 보고 답한다"), status_code=400) world = build_grounding_world_facts(rules_cp) segs = {f"scene-{s.get('scene_index')}": (s.get("text") or "") for s in segments} bundles = bundle_scenes( [{"idx": int(s.get("scene_index") or 0), "length": len(s.get("text") or "")} for s in segments]) plan: List[Dict[str, Any]] = [] cats: Dict[str, Any] = {} for n, b in enumerate(bundles): ids = [f"scene-{i}" for i in b] cat = sc.build_catalog(scenes, ids) cats[f"c{n}"] = cat plan.append({ "chunk_id": f"c{n}", "segment_ids": ids, "payload": gc.build_chunk_payload(ids, segs, world, shot_catalog=cat), }) return plan, segs, cats, world def _journal(self, plan) -> Any: """구간 장부. ★계약이 바뀌었으면 **이어 쓰지 않고** 새로 시작한다. 신원(`acquisition_identity`)이 팩·모델·요청 계약을 접으므로 사실은 저절로 안 맞지만, 계약 자체를 장부에 적어 두면 **왜 다시 사는지**가 기록에 남는다. """ from app.modules.pipeline.grounding_chunk_journal import ( ChunkJournal, JOURNAL_NAME) # ★★장부 계약은 **사는 것을 지배하는 것만** 담는다. 후처리를 담으면 # 해석만 고쳐도 장부가 버려져 **유료 raw 를 다시 산다**. # ★사실 신원(`acquisition_identity`)이 이미 다 접으므로, 팩·모델·요청 # 계약이 바뀌면 **저절로 안 맞고** 다시 산다. 이 계약은 **왜 안 맞는지** # 기록에 남기려고 적는다 — 지우는 근거가 아니다. jr = ChunkJournal(self._cp_dir / JOURNAL_NAME, contract={"acquisition": self._acquisition(), "chunks": len(plan)}) if jr.contract_drifted(): # ★**안 지운다.** 신원이 다르면 어차피 못 찾고, 같으면 그 raw 는 # 여전히 유효하다. 지우면 산 것을 버리는 것이다. logger.info("grounding_chunk: 획득 계약이 바뀌었다 — 신원이 다른 " "것만 다시 산다(장부는 그대로 둔다)") return jr def _acquisition(self) -> Dict[str, Any]: """**provider 응답의 뜻을 지배하는 것만**. ★신원은 이것만 접는다. ★★세 가지를 갈라야 한다 (Codex 2026-08-31) — provider 획득 payload · 모델 · **실제 전송 인자** → 바뀌면 **다른 답**이다. 다시 산다 admission 정책 마감 · 슬롯 · 예산 → 「언제·몇 번 보낼까」지 **답의 뜻이 아니다**. 여기 넣으면 정책만 바꿔도 raw 를 재구매한다 해석 후처리 · adapter · 구간 나누기 → 무료 재해석 ★`max_tokens` 는 **실제 전송 인자**라 여기 든다 — 앞 판엔 빠져 있었다. """ from app.modules.llm.llm_client import _resolve_model from app.modules.pipeline import grounding_chunk as gc return { "alias": self.step_id, "physical": _resolve_model(self.step_id, self.project_config), "pack": gc.CHUNK_PACK_VERSION, "request": dict(REQUEST_CONTRACT), } def _admission(self) -> Dict[str, Any]: """**언제·몇 번 보낼까**. ★신원에 안 접는다 — 답의 뜻이 아니다. 지문에는 **따로** 접는다. 정책이 바뀌면 「같은 조건에서 돌았다」가 거짓이 되지만, 이미 산 raw 는 **여전히 그 답**이다. """ from app.core import openai_keys slots = openai_keys.slot_count() return { "deadline_seconds": CHUNK_DEADLINE_SECONDS, "slots": slots, "physical_per_logical": physical_per_logical(REQUEST_CONTRACT, slots), "transmission_cap": self._transmission_cap(), } def _transmission_cap(self) -> int: """물리 전송 상한. ★`grounding_research` 와 **같은 설정 키**를 본다.""" return int(self.project_config.get("research_transmission_cap") or DEFAULT_TRANSMISSION_CAP) def _identity(self, payload: Dict[str, Any]) -> str: """호출 **신원** — 팩·모델·**실제 요청 계약**까지 접는다. ★production 함수 한 곳(`acquisition_identity`)이 만든다 — 감사 도구와 같은 것이라 장부·trace 대조가 성립한다. """ from app.modules.pipeline.grounding_chunk import acquisition_identity acq = self._acquisition() return acquisition_identity( payload, model_alias=acq["alias"], model_physical=acq["physical"], request_contract={"pack": acq["pack"], **acq["request"]}) def _cap(self, plan) -> int: """논리 상한 — 구간 수 + merge 1. ★계획에서 **한 번만** 센다.""" return len(plan) + 1 def _assert_physical_bound(self, plan) -> int: """물리 **상한**이 승인 범위 안인가. ★넘으면 provider 앞에서 선다. ★막는 것이 아니라 **정하는 것**이다 — 글 호출 경로에는 세는 자리가 없어서 물리 수를 그 자리에서 거절할 수 없다. 대신 재시도·fallback 을 끄고 슬롯 수로 위를 정한 뒤, 그 수가 **손으로 적은 승인 수**를 넘으면 아무것도 안 사고 선다. """ adm = self._admission() cap = adm["transmission_cap"] bound = self._cap(plan) * int(adm["physical_per_logical"]) if bound > cap: from app.core.errors import AppError raise AppError( code="grounding_chunk.physical_bound_exceeded", message=(f"물리 상한이 {bound} 인데 승인 전송 상한은 {cap} 이다 " f"(논리 {self._cap(plan)} × 슬롯·재시도 " f"{adm['physical_per_logical']}) — 사람이 " "`research_transmission_cap` 을 다시 정해야 한다. " "아무것도 안 사고 선다"), status_code=409) return bound # ── 판독 ──────────────────────────────────────────────────────────── def _read_chunks(self, plan, segs, cats, jr=None): """구간마다 **한 번** 읽는다 — 병렬. ★행 해석은 production 함수가.""" from app.modules.pipeline import grounding_chunk as gc from app.core.research_call_budget import bind_current_research_budget from app.core.run_control import is_abort rows: List[Dict[str, Any]] = [] quarantined: List[Dict[str, Any]] = [] failed: List[Any] = [] got: Dict[str, Any] = {} with ThreadPoolExecutor(max_workers=min(MAX_WORKERS, len(plan))) as ex: # ★★정지 표와 예산은 **스레드 지역**이다. 그냥 넘기면 worker 가 # 아무것도 못 봐서 마감이 그 자리에서 통째로 no-op 이 된다. send = bind_current_research_budget(self._one) futs = {ex.submit(send, c, jr, self._cap(plan)): c["chunk_id"] for c in plan} for f in as_completed(futs): cid = futs[f] try: got[cid] = f.result() except Exception as exc: # noqa: BLE001 # ★★멈추라는 말·주인 잃음·못 읽는 게이트는 **이 구간 하나의 # 실패가 아니다** (Codex BLOCK 2026-08-31). 여기서 세면 # ①나머지 worker 가 계속 돌아 돈이 나가고 ②rows 가 남아 # **merge 유료 호출까지 이어지며** ③감사에는 provider # 장애로 남는다. 즉시 위로 올린다. if is_abort(exc): logger.warning("grounding_chunk 중단 — %s", exc) raise logger.error("grounding_chunk %s 실패: %s", cid, exc) failed.append((cid, exc)) # ★결과를 **계획 순서로** 모은다 — 완료 순서로 쌓으면 같은 입력에도 # 행 순서가 달라져 지문이 안 선다. for c in plan: resp = got.get(c["chunk_id"]) if resp is None: continue out = gc.resolve_rows(resp.get("rows") or [], chunk_id=c["chunk_id"], segment_ids=c["segment_ids"], segments=segs, shot_catalog=cats.get(c["chunk_id"])) rows += out["rows"] quarantined += out["quarantined"] # ★★**한 구간이라도 못 읽었으면 산출을 안 낸다** (Codex 2026-08-31). # 등록 축이 **구간을 걸쳐** 세기 때문이다 — 구간 1과 3에 나오는 대상이 # 3이 빠지면 「1회」로 세어져 **등록이 뒤집힌다**. 그건 「모자란 것」이 # 아니라 **틀린 것**이라 partial 로 내려보내면 안 된다. if failed: from app.core.errors import AppError raise AppError( code="grounding_chunk.incomplete_read", message=(f"구간 {len(failed)}/{len(plan)} 을 못 읽었다 " f"({[c for c, _ in failed]}) — 등록 횟수는 구간을 " "걸쳐 세므로 남은 것만으로 내면 **수가 틀린다**. " "재개해서 빠진 구간만 다시 읽어야 한다"), status_code=502, ) return rows, quarantined, failed def _one(self, chunk: Dict[str, Any], jr=None, cap: int = 0 ) -> Dict[str, Any]: from app.modules.pipeline import grounding_chunk as gc p = chunk["payload"] return self._call(p, gc.CHUNK_SCHEMA_NAME, jr, cap, slot=f"chunk:{chunk.get('chunk_id')}") def _call(self, payload: Dict[str, Any], schema_name: str, jr, cap: int, *, slot: Optional[str] = None) -> Dict[str, Any]: """산다 — **장부·예산·정지를 거쳐서**. ★한 곳에서만 보낸다. ## ★신원은 `opik_metadata` 로는 **안 간다** litellm 이 `metadata["opik"]` 에서 읽는 키는 **넷뿐**이다 (`project_name`·`current_span_data`·`tags`·`thread_id`, `llm_client._build_opik_metadata` 주석). 다른 키는 조용히 버려지고, 태그로 실어도 **축 whitelist**(`is_axis_tag`)가 거른다. ★이 결함을 앞서 한 번 겪고 기억에 적었는데 **다시 냈다** (Codex 2026-08-31). 신원은 **부모 trace 의 metadata** 로 간다 — 감사 도구(`cc_runner.live_send`)가 쓰는 바로 그 자리다. """ from app.core.research_call_budget import research_calls_armed from app.modules.llm.llm_client import call_structured from app.modules.llm.opik_trace import open_trace from app.modules.pipeline.grounding_chunk_journal import buy_or_reuse from app.modules.pipeline.llm_deadline import call_with_deadline ident = self._identity(payload) def _send(): # ★부모 trace 를 열고 **그 metadata 에** 신원을 싣는다. # `open_trace` 는 기록이 꺼졌거나 실패해도 **예외 없이 None** 을 # 주므로, 이것이 본 작업을 막지 않는다. with open_trace(name=CHUNK_TRACE_NAME, tags=[CHUNK_TAG], metadata={ID_META_KEY: ident, "step": self.step_id, "schema": schema_name}, thread_id=f"{self.project_id}:{self.episode_id}", input_data={"identity": ident}): # ★**실제 요청 계약을 넘긴다.** 기본값에 맡기면 # `enable_fallback=True`·라우터 기본 재시도라 몇 번 나가는지 # 모른다. 신원에 접은 그 값과 **같은 한 벌**이다. # ★★**팔을 든다** — 이 블록 안의 전송만 조사 예산으로 센다. # 밖에서는 `reserve_current_research_call` 이 no-op 이라 # 다른 글 호출부는 한 글자도 안 바뀐다. with research_calls_armed(): # ★★한 호출 마감. 매달린 provider 를 **여기서** 끊는다. return call_with_deadline( call_structured, self.step_id, payload["system"], payload["parts"], payload["schema"], deadline_seconds=CHUNK_CALL_DEADLINE_SECONDS, project_config=self.project_config, schema_name=schema_name, opik_metadata={"tags": [CHUNK_TAG]}, temperature=REQUEST_CONTRACT["temperature"], enable_fallback=REQUEST_CONTRACT["enable_fallback"], num_retries=REQUEST_CONTRACT["num_retries"]) if jr is None: # ★시험용 경로 — 장부 없이도 돈다 return _send() return buy_or_reuse( jr, ident, cap=cap, send=_send, slot=slot, stop_check=lambda: self.checkpoint_gate( where=f"grounding_chunk {schema_name}", strict=True)) # ── 합치기 ────────────────────────────────────────────────────────── def _merge(self, rows, segs, jr=None, cap: int = 0): from app.modules.pipeline import grounding_chunk as gc from app.modules.pipeline import grounding_chunk_merge as cm if not rows: return cm.reduce_episode([], [], segments=segs) # ★구간을 다 읽는 사이에 멈추라는 말이 왔을 수 있다. merge 는 **유료**라 # 보내기 **전에** 한 번 더 본다 — `buy_or_reuse` 가 같은 순서로 한다. self.checkpoint_gate(where="grounding_chunk merge", strict=True) p = gc.build_merge_payload(rows) # ★merge 는 **한 자리** — 행 집합이 바뀌어 새 신원으로 다시 사도 상한을 안 늘린다 resp = self._call(p, gc.MERGE_SCHEMA_NAME, jr, cap or (jr.contract.get("chunks", 0) + 1 if jr else 0), slot="merge") return cm.reduce_episode(rows, resp.get("decisions") or [], segments=segs) # ── 산출 ──────────────────────────────────────────────────────────── @staticmethod def _part_of_final_ids(reduced): """`part_of` 를 **정본 ID 짝**으로. ★등록된 것만 — 없으면 안 낸다.""" reg = reduced.get("registered") or {} def _fid(lid): return str(((reg.get(str(lid)) or {}).get("final_id")) or "") out = [] for x in (reduced.get("part_of") or ()): p, w = _fid((x or {}).get("part")), _fid((x or {}).get("whole")) if p and w: out.append((p, w)) return sorted(set(out)) def _wrap(self, g_mode, plan, rows, quarantined, reduced, jr=None): """★`entity_merge` 모양과 **근거**를 함께 낸다. adapter 가 짓는다.""" from app.modules.pipeline import grounding_chunk as gc from app.modules.pipeline import grounding_chunk_adapter as ad from app.modules.pipeline import grounding_facet_binding as fb # ★★★엔티티 줄과 **후보를 함께** 낸다 (Codex BLOCK · 09-01). # 앞 판은 `to_entity_rows` 만 불러 엔티티 줄만 CP 에 적었다 — 그러면 # 판별이 A0 후보를 못 받아 **옛 유료 갈래를 다시 사거나** 빈손이 된다. # `project_rows_and_candidates` 는 **같은 reduced 행에서** 둘을 함께 # 투영하며 producer 링크를 발행한다. projected = ad.project_rows_and_candidates( reduced, project_id=self.project_id, episode_id=self.episode_id) facet = fb.bind(reduced["rows"], reduced.get("part_of") or [], reduced.get("registered") or {}) fb.assert_owner_pairs(facet["bindings"]) return { # ★★save_checkpoint 는 여기 적은 config_hash 를 보존한다. 안 적으면 base 해시가 # 저장되고 어긋남 검사는 `_config_hash()` 로 세어 **재개마다 drift→BLOCK** 이다 # (실측 2026-09-02 밤: canary 가 그 BLOCK 을 삼키고 옛 CP 로 지나갔다). "config_hash": self._config_hash(), # ★여기 오면 **전부 읽었다** — 못 읽었으면 위에서 섰다. "completed_count": len(plan), "applicable_count": len(plan), "failed_count": 0, "data": { "mode": g_mode, "chunks": len(plan), # ★논리 호출 수를 **적어 둔다** — 구간 + merge 1 # ★**장부가 센 수**를 낸다 — 계획에서 다시 세면 재개한 판에서 # 「샀다」와 「되썼다」가 섞인다. "provider_calls": (jr.bought() if jr else len(plan) + (1 if rows else 0)), "logical_cap": self._cap(plan), # ★**상한**이지 실측이 아니다. 실제 물리 수는 Opik+provider # 로그로 본다 — 여기 적힌 수를 「몇 번 나갔다」로 읽으면 안 된다. "physical_upper_bound": self._admission()[ "physical_per_logical"] * self._cap(plan), "transmission_cap": self._transmission_cap(), "complete": True, "contracts": { "pack": gc.CHUNK_PACK_VERSION, "processing": gc.PROCESSING_CONTRACT_VERSION, "adapter": ad.ADAPTER_CONTRACT_VERSION, }, **projected["rows"], "removed": [], # ★후보 — 판별이 이것으로 결속한다(이름으로 안 짝짓는다) "grounding_candidates": projected["candidates"], "grounding_facet_debt_rows": projected.get("facet_debt") or [], "grounding_skipped": projected["skipped"], "grounding_quarantined": quarantined, # ★일부를 뺀 채 남은 행 — 감사가 「어디서 무엇을 뺐나」를 본다. # (실측 2026-09-02 밤: 다른 함수의 지역 변수를 여기서 읽어 NameError 로 죽었다 — # 행에서 **파생**한다, 따로 나르지 않는다) "grounding_salvaged": [ {"local_id": x.get("local_id"), "owner_type": x.get("owner_type"), "surface_form": x.get("surface_form"), "problems": x.get("salvage_problems")} for x in rows if x.get("salvage_problems")], # ★★`part_of` 를 **정본 ID 짝**으로 낸다 — 관계 스텝이 # 이것을 호출 0 으로 투영한다. local_id 로 내면 뒤가 다시 # 짝지어야 하고, 그러면 같은 규칙이 두 곳이 된다. "grounding_part_of": [ {"part": _fid, "whole": _wid} for _fid, _wid in self._part_of_final_ids(reduced)], "grounding_facet_bindings": facet["bindings"], "grounding_facet_debt": facet["debt"], "grounding_counts": reduced.get("counts") or {}, "grounding_refused": reduced.get("refused") or {}, }, }