"""앞쪽 **판별**(screen) — 「이 대상이 참조 사진을 사야 하는가」 한 자리. ## 무엇이 바뀌나 옛 축(`generation_difficulty`)은 분류기가 **또 한 번 유료로** 내던 판단이었다. 같은 물음을 `era_research.assess_subjects` 가 **판별 + 원어 질의 저작**을 한 번에 하므로, 그쪽을 SOT 로 삼고 이 모듈은 그것을 **대상마다 돌려 장부로 남긴다.** 사는 것은 여기서 **하지 않는다** — 획득은 중앙 19.x 한 곳이다. ## 이 모듈이 지키는 것 1. **다섯 owner 전부** 장부에 남는다. `prop`·`character`·`location` 은 붙을 행이 있고, `location_part`·`outlook` 은 producer(§2-6.5a·아웃룩 단계)가 아직 없다 — 그 사실을 **`deferred_producer` 로 적지**, 이름이 닮았다고 base 갈래에 올리지 않는다. 2. **판별과 결속은 다른 축**이다. 그래서 칸이 둘이다 — `screen`(사야 하나)과 `binding`(어디에 붙나). 한 칸에 접으면 「사야 하는데 붙을 데가 없다」를 적을 자리가 없어진다. 3. 신원은 **stable subject id + controlled owner + canonical payload sha** 셋이다. 하나라도 없으면 fallback 으로 내려간다 — 「**잘못 합치는 것이 중복 조사보다 나쁘다**」. 4. 이름·부분문자열로 뜻을 정하지 않는다. 이 모듈에는 문자열 대조가 **없다**. """ from __future__ import annotations import copy import hashlib import logging from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence from app.modules.pipeline import era_research as _era from app.modules.pipeline import grounding_entity_contract as _ec from app.modules.pipeline.grounding_carry import (DISP_CARRIED, DISP_ENTITY_ONLY, DISP_CONTESTED, DISP_DEFERRED, DISP_PROMOTED, DISP_UNRESOLVED, FACET_OWNERS) logger = logging.getLogger(__name__) #: 판별 규칙이 바뀌면 올린다 — 지문에 접혀 resume 이 옛 CP 를 안 건너뛴다. SCREEN_CONTRACT_VERSION = 1 #: ★`screen` 칸 — **사야 하는가.** `era_research` 의 envelope 를 그대로 옮긴다. SCREEN_OBLIGATION = "obligation" # 판별이 대상을 냈다 — 참조를 사야 한다 SCREEN_NOT_TARGET = "not_target" # 정당한 era 없음 (**빈 목록도 캐시된다**) SCREEN_UNRESOLVED = "unresolved" # 판별이 못 섰다 — 「비대상」으로 안 내린다 SCREEN_CAPPED = "capped" # 이 걷기의 상한에 걸렸다 — 미확정이다 SCREENS = (SCREEN_OBLIGATION, SCREEN_NOT_TARGET, SCREEN_UNRESOLVED, SCREEN_CAPPED) #: ★`binding` 칸 — **어디에 붙는가.** 판별과 **직교**한다. BIND_ENTITY_ROW = "entity_row" # 이미 있는 행에 붙었다 BIND_PROMOTED_ROW = "promoted_row" # 행이 없어 앞쪽에서 만든다 BIND_DEFERRED_PRODUCER = "deferred_producer" # facet — producer 가 뒤에 있다 BIND_UNBOUND = "unbound" # 결속을 못 정했다 BINDINGS = (BIND_ENTITY_ROW, BIND_PROMOTED_ROW, BIND_DEFERRED_PRODUCER, BIND_UNBOUND) #: ★한 걷기가 살 수 있는 **판별 호출 수**의 계약 기본값. 시나리오와 무관한 #: 좌표다. 운영자가 `project_config['grounding_screen_max_assess']` 로 올린다. DEFAULT_MAX_ASSESS = 64 #: 판별 envelope → 장부 칸. **한 곳에서만** 옮긴다. _STATUS_TO_SCREEN = { _era.PLAN_OK: SCREEN_OBLIGATION, _era.PLAN_NO_SUBJECT: SCREEN_NOT_TARGET, _era.PLAN_FAILED: SCREEN_UNRESOLVED, } #: 결속 장부(`grounding_carry`)의 disposition → 이 모듈의 `binding` 칸. #: ★`contested`·`unresolved` 는 **여기 없다** — `BIND_UNBOUND` 로 떨어진다. _DISP_TO_BINDING = { DISP_CARRIED: BIND_ENTITY_ROW, # ★A0 후보가 없는 기존 엔티티 행도 **entity_row** 다 — 붙을 데가 있다. DISP_ENTITY_ONLY: BIND_ENTITY_ROW, DISP_PROMOTED: BIND_PROMOTED_ROW, DISP_DEFERRED: BIND_DEFERRED_PRODUCER, } #: ★**사지 않는** disposition. 「후보는 있는데 어느 것인지 못 정했다」이므로 #: 판별해도 그 결과를 어디에 붙일지 모른다 — 사면 그냥 버리는 돈이다. NO_BUY_DISPOSITIONS = frozenset({DISP_CONTESTED, DISP_UNRESOLVED}) STEP_TAG = "grounding_screen" def build_population(subjects: Sequence[Dict[str, Any]], candidates: Sequence[Dict[str, Any]], dispositions: Dict[str, str]) -> List[Dict[str, Any]]: """★★**판별 모집단** — 다섯 갈래가 **전부** 들어온다. 왜 필요한가 — `grounding_carry.build_subjects` 는 두 번째 pass 에서 `DISP_PROMOTED` 만 승격한다. `location_part`·`outlook` 은 `DISP_DEFERRED` 라 **subject 목록에 아예 안 들어온다**. 그것만 넣고 「다섯 owner 전부 판별한다」고 쓰면 거짓이다 (Codex 직접 재현: subjects=0). 그래서 subject 목록에 **후보 장부**를 이어 붙이고 `research_subject_id` 로 겹치는 것을 지운다. 순서는 subject 먼저 — 그쪽이 결속 provenance (`short_id` 등)를 갖고 있다. ★facet 은 붙을 행이 없지만 **판별은 산다.** 그 결과가 §2-6.5a producer 가 만들 때 그대로 쓰인다 — 그때 다시 사면 같은 것을 두 번 사는 것이다. ★★**조용히 버리지 않는다** (Codex BLOCK-3). 앞 판은 빈 id 와 겹친 id 를 `continue` 로 넘겨서, 뒤에 오는 검사가 볼 때는 **이미 사라진 뒤**였다. 같은 벌 안의 중복 → **선다**. 어느 쪽이 진짜인지 못 정한다. 빈 id → **선다**. 붙일 데를 모른다. 두 벌에 같은 id → owner 와 payload 가 **같을 때만** 접는다. 다르면 신원 충돌이라 **선다**. """ _assert_no_blank_or_dup(subjects, "subject") _assert_no_blank_or_dup(candidates, "candidate") out: List[Dict[str, Any]] = [] seen: Dict[str, Dict[str, Any]] = {} for item in list(subjects or ()) + list(candidates or ()): rsid = str(item.get("research_subject_id") or "").strip() prev = seen.get(rsid) if prev is None: seen[rsid] = item out.append(item) continue # ★같은 id 인데 **다른 것**이면 접으면 안 된다 — 앞 것을 택하는 순간 # 뒤 후보가 소리 없이 사라진다. if (str(prev.get("owner_type") or "") != str(item.get("owner_type") or "") or canonical_payload_sha(prev) != canonical_payload_sha(item)): raise AssertionError( f"같은 `research_subject_id` 가 다른 대상을 가리킨다 ({rsid}) — " f"owner={prev.get('owner_type')!r}/{item.get('owner_type')!r}") return out def _assert_no_blank_or_dup(items: Sequence[Dict[str, Any]], label: str) -> None: """한 벌 안에서 **자리별로** 본다. ★집합·합계로 보면 둘 다 빌 때 통과한다.""" ids = [str((x or {}).get("research_subject_id") or "").strip() for x in items or ()] blank = [i for i, v in enumerate(ids) if not v] if blank: raise AssertionError( f"{label} 에 `research_subject_id` 가 빈 것이 {len(blank)}개 있다 " f"(자리 {blank[:5]}) — 판별 결과를 어디에 붙일지 못 정한다") dup = sorted({v for v in ids if ids.count(v) > 1}) if dup: raise AssertionError( f"{label} 안에 같은 `research_subject_id` 가 {len(dup)}개 겹친다 " f"{dup[:5]} — 어느 쪽이 진짜인지 못 정한다") def resolve_cap(max_assess: Any) -> int: """판별 호출 상한. ★**정확히 양의 int** 만 받는다 — fail-closed. `int(x)` 로 강제하면 `True`→1 · `"2"`→2 · `0` · `-1` 이 전부 통과한다. 상한은 **승인 범위**라 조용히 넓어지거나 좁아지면 안 된다. """ if max_assess is None: return DEFAULT_MAX_ASSESS # ★`bool` 은 `int` 의 하위형이다 — 먼저 막는다. if isinstance(max_assess, bool) or type(max_assess) is not int: raise ValueError( f"판별 상한은 정확히 int 여야 한다 (받은 것: {max_assess!r})") if max_assess <= 0: raise ValueError(f"판별 상한은 1 이상이어야 한다 (받은 것: {max_assess})") return max_assess def _flush(on_row, row, plan) -> None: """★한 줄이 날 때마다 **바로** 넘긴다. ★★**저장 실패를 삼키지 않는다** (Codex BLOCK-2). 삼키고 다음 대상을 사러 가면 「호출마다 durable」이 거짓이 된다 — 디스크가 찬 채로 스무 번 더 사고, 그 스무 번이 아무 데도 안 남는다. **다음 구매 전에 선다.** """ if on_row is None: return on_row(row, plan) def source_evidence_of(subject: Dict[str, Any]) -> Dict[str, Any]: """그 대상의 **원문 증거**를 compact 하게 원형 보존한다. ★★★왜 필요한가 (Codex · 09-01) — `outlook` 은 `outlook_phase3`(order 19.2) 뒤에야 실물이 생기는데 이 판별은 13.68 이다. 그래서 앞단은 **의무와 증거만** 남기고, 실제 참조는 중앙 획득 한 곳이 산다. 그런데 앞 판의 줄에는 `research_subject_id` 와 판별 결과뿐이라 **뒤에서 결속할 증거가 없었다**. ★**없는 것을 만들지 않는다.** span 도 occurrence 도 있을 때만 싣는다 — 지어낸 좌표는 증거가 아니라 거짓이다. ★subjects 를 통째로 복제하지 않는다. 정본 행 **하나에** 담는다. """ got: Dict[str, Any] = { "surface_form": subject.get("surface_form") or "", "source_anchor": subject.get("source_anchor") or "", "source_quote": subject.get("source_quote") or "", "quote_source": subject.get("quote_source") or "", # ★payload 지문 — 같은 대상인지 뒤에서 **구조로** 견준다 "payload_sha": canonical_payload_sha(subject), } for k in ("occurrences", "source_spans", "spans"): v = subject.get(k) if v: # ★없으면 칸도 안 만든다 got[k] = copy.deepcopy(v) pv = subject.get("provenance") or {} if pv: got["provenance"] = copy.deepcopy(pv) return got def canonical_payload_sha(subject: Dict[str, Any]) -> str: """정본 payload sha. ★**구조화 칸만** 접는다 — 이름은 신원이 아니다. 접는 것: owner · 원문 anchor · 표면형 · **원문 인용**. 원문 인용이 신원에 들어가는 이유는, 같은 표면형이라도 다른 문장에서 온 것은 다른 대상일 수 있어서다. """ parts = [str(subject.get(k) or "") for k in ("owner_type", "source_anchor", "surface_form", "source_quote")] return hashlib.sha256("\x1e".join(parts).encode("utf-8")).hexdigest()[:16] def subject_text_of(subject: Dict[str, Any]) -> str: """판별에 넣을 텍스트. ★슬롯을 이어 붙일 뿐 — 낱말을 넣지 않는다.""" lines = [] surface = str(subject.get("surface_form") or "").strip() if surface: lines.append(surface) quote = str(subject.get("source_quote") or "").strip() if quote: lines.append(quote) return "\n".join(lines) def binding_of(subject_id: str, dispositions: Dict[str, str]) -> str: """그 대상이 어디에 붙는가. ★**결속 장부가 정한다** — 이름이 아니라.""" return _DISP_TO_BINDING.get(dispositions.get(subject_id or "") or "", BIND_UNBOUND) def project_from_producer(subjects, *, dispositions): """C(c) 판 — producer 판정으로 **결정적 투영**. ★모델을 안 부른다. ★★★같은 뜻을 **다시 판단하지 않는다** (사용자 확정 · Codex BLOCK 2026-09-01). C(c) producer 는 한 판독에서 두 축(`hard_to_generate` · `viewers_would_notice`)을 이미 냈다. 그것을 다시 VLM 에게 물으면 — ①같은 것을 두 번 산다 ②두 판정이 갈리면 어느 쪽이 정본인지 아무도 모른다 의무는 **같은 행에서 둘 다 참일 때만** 선다 — `grounding_chunk` 의 등록 규칙과 **같은 문장**이다. Returns: `screen_subjects` 와 **같은 모양**. `assess_bought` 는 0 이다. """ rows, counts = [], {k: 0 for k in SCREENS} for subj in subjects or (): rsid = str(subj.get("research_subject_id") or "").strip() owner = str(subj.get("owner_type") or "").strip() payload = subj.get(_ec.PRODUCER_PAYLOAD) or {} hard = payload.get("hard_to_generate") notice = payload.get("viewers_would_notice") if hard is None or notice is None: # ★★producer 가 **안 낸 것**을 「아니다」로 접지 않는다 screen = SCREEN_UNRESOLVED why = "producer 가 두 축을 안 냈다" elif hard is True and notice is True: screen, why = SCREEN_OBLIGATION, "" else: screen, why = SCREEN_NOT_TARGET, "두 축이 같은 행에서 둘 다 참이 아니다" row = { "research_subject_id": rsid, "owner_type": owner, "binding": binding_of(rsid, dispositions), "is_facet": owner in FACET_OWNERS, "source_evidence": source_evidence_of(subj), **({_ec.PRODUCER_PAYLOAD: copy.deepcopy(payload)} if payload else {}), **({_ec.FACET_BINDING: copy.deepcopy(subj[_ec.FACET_BINDING])} if subj.get(_ec.FACET_BINDING) else {}), **({_ec.HOST_CONTEXT: copy.deepcopy(subj[_ec.HOST_CONTEXT])} if subj.get(_ec.HOST_CONTEXT) else {}), "screen": screen, # ★한 번도 안 샀다 — 그래서 캐시 열쇠도 지문도 없다 "from_cache": None, "assess_key": None, } if why: row["reason"] = why counts[screen] = counts.get(screen, 0) + 1 rows.append(row) return {"contract_version": SCREEN_CONTRACT_VERSION, "rows": rows, "counts": counts, "plans": {}, "assess_bought": 0, "assess_reused": 0, "projected_from": "grounding_chunk"} def screen_subjects( subjects: Sequence[Dict[str, Any]], *, dispositions: Dict[str, str], world_facts_block: str, cache_get: Callable[[str], Any], cache_put: Callable[[str, Any], Any], project_config: Optional[Dict[str, Any]] = None, opik_metadata: Optional[Dict[str, Any]] = None, failed_memo: Optional[set] = None, max_assess: Optional[int] = None, assess_fn: Optional[Callable[..., Dict[str, Any]]] = None, on_row: Optional[Callable[[Dict[str, Any], Optional[Dict[str, Any]]], Any]] = None, ) -> Dict[str, Any]: """대상마다 판별을 돌려 **장부**를 만든다. ★참조는 **사지 않는다.** Args: subjects: `grounding_carry.build_subjects` 가 낸 subject 들. dispositions: `research_subject_id` → `grounding_carry` disposition. **결속은 그쪽이 SOT** 다 — 여기서 다시 짝짓지 않는다. max_assess: 이 걷기가 **살 수 있는** 판별 호출 수. 캐시 적중은 안 센다. 넘으면 남은 대상은 `SCREEN_CAPPED` 다 — ★「비대상」으로 **안 내린다**. on_row: 한 줄이 날 때마다 부른다. ★**산 것을 판정보다 먼저 저장**하는 자리다 — 중간에 끊겨도 앞서 산 것을 다시 사지 않는다. Returns: `rows`(대상별 한 줄) · `plans`(id → 판별 계획) · `obligations`(id 목록) · `counts` · `by_owner` · `assess_calls`. """ call = assess_fn or _era.assess_plan_cached cap = resolve_cap(max_assess) # ★★**사기 전에** 모집단을 본다. 다 사고 나서 서면 그 돈이 버려진다. _assert_population(subjects) rows: List[Dict[str, Any]] = [] plans: Dict[str, Any] = {} #: ★**산 것**과 **재사용한 것**을 가른다. 캐시 적중까지 세면 「유료 호출 #: 상한」이라는 말이 거짓이 되고, 상한이 엉뚱하게 일찍 닫힌다. bought = 0 reused = 0 for subj in subjects or (): rsid = str(subj.get("research_subject_id") or "").strip() owner = str(subj.get("owner_type") or "").strip() text = subject_text_of(subj) bind = binding_of(rsid, dispositions) row: Dict[str, Any] = { "research_subject_id": rsid, "owner_type": owner, "binding": bind, # ★**facet 인지**를 칸으로 남긴다. 뒤에서 owner 이름을 다시 # 판단하면 같은 규칙이 두 곳이 된다. "is_facet": owner in FACET_OWNERS, # ★★뒤(19.2 이후 중앙 획득)가 결속할 **원문 증거**. 여기 한 줄에 # 담는다 — `facet_debt` 는 이 rows 를 거른 view 지 딴 SOT 가 아니다. "source_evidence": source_evidence_of(subj), # ★★producer 가 **유료로 낸 구조화 판정**. 원문 증거와 **따로** # 둔다 — 하나는 원고가 말한 것이고 하나는 모델이 판단한 것이다. # ★없으면 칸도 안 만든다. **({_ec.PRODUCER_PAYLOAD: copy.deepcopy(subj[_ec.PRODUCER_PAYLOAD])} if subj.get(_ec.PRODUCER_PAYLOAD) else {}), # ★facet 좌표 — 뒤(19.2 이후 결속)가 **누구 것인지**를 이것으로 안다 **({_ec.FACET_BINDING: copy.deepcopy(subj[_ec.FACET_BINDING])} if subj.get(_ec.FACET_BINDING) else {}), # ★맥락 칸 — 의무 계획이 맥락 조사 재료를 여기서 얻는다 **({_ec.HOST_CONTEXT: copy.deepcopy(subj[_ec.HOST_CONTEXT])} if subj.get(_ec.HOST_CONTEXT) else {}), } if dispositions.get(rsid) in NO_BUY_DISPOSITIONS: # ★「불렸는데 못 정했다」 — 판별해도 붙일 데를 모른다. row["screen"] = SCREEN_UNRESOLVED row["reason"] = f"carry_{dispositions.get(rsid)}" rows.append(row) _flush(on_row, row, None) continue if not text: # ★근거가 없으면 **판별을 사지 않는다.** 원문 문장도 표면형도 # 없는 것에 물으면 모델이 상상으로 답한다. row["screen"] = SCREEN_UNRESOLVED row["reason"] = "no_subject_text" rows.append(row) _flush(on_row, row, None) continue # ★★**이미 있는 것은 상한을 안 먹는다** (Codex). 상한에 먼저 걸어 # 버리면 앞판이 사 둔 것까지 `capped` 가 되어 **공짜로 쓸 수 있는 # 것을 못 쓴다**. 「살 것인가」를 **사기 전에** 캐시로 가른다. # ★키는 `era_research` 가 낸다 — 여기서 조립하면 두 곳이 갈리고, # 갈리면 「공짜인 줄 알고 불렀다가 실제로 샀다」가 된다. cached = _era.assess_cache_key( subject_text=text, world_facts_block=world_facts_block, canonical_scope_id=rsid, canonical_scope_role=owner, canonical_scope_sha=canonical_payload_sha(subj)) pre = cache_get(cached) will_buy = not (isinstance(pre, dict) and "subjects" in pre) if will_buy and bought >= cap: row["screen"] = SCREEN_CAPPED row["reason"] = f"max_assess={cap}" rows.append(row) _flush(on_row, row, None) continue # ★★**호출 전** 상태로 센다. 끝난 뒤에 캐시를 다시 읽으면 판별이 # 방금 써 넣은 값을 「적중」으로 세게 되고, 그러면 `bought` 가 영영 # 0 이라 **상한이 아무것도 안 막는다**. # 실측: 상한 2 로 걸고 5개를 넣었더니 5개를 다 샀다. probe = {"read": False, "hit": not will_buy} def _watch(key, _p=probe, _get=cache_get): v = _get(key) if str(key).startswith("era_assess::"): _p["read"] = True return v outcome: Dict[str, Any] = {} env = call( step_tag=STEP_TAG, subject_text=text, world_facts_block=world_facts_block, cache_get=_watch, cache_put=cache_put, project_config=project_config, opik_metadata=opik_metadata, failed_memo=failed_memo, outcome=outcome, # ★정본 신원 셋. 셋이 다 있어야 `era_research` 가 정본으로 잡는다. canonical_scope_id=rsid, canonical_scope_role=owner, canonical_scope_sha=canonical_payload_sha(subj), ) or {} row["assess_key"] = cached status = str(env.get("status") or "") # ★★셈은 **돈 쪽으로 보수적**이다. # # 호출 전에 이미 있었다 → 되쓴 것 # 읽었는데 없었다 → 산 것 # 읽지도 않았는데 실패였다 → 걷기 memo · 사기 전 거절 — 안 샀다 # 읽지도 않았는데 값이 왔다 → **산 것으로 센다** # # 마지막 갈래를 「안 샀다」로 두면 캐시를 안 보는 구현이 끼었을 때 # 상한이 통째로 무력해진다. 늦게 닫히는 것보다 일찍 닫히는 것이 낫다. if probe["hit"]: reused += 1 row["from_cache"] = True elif probe["read"] or status != _era.PLAN_FAILED: bought += 1 row["from_cache"] = False else: row["from_cache"] = None row["screen"] = _STATUS_TO_SCREEN.get(status, SCREEN_UNRESOLVED) if env.get("reason"): row["reason"] = str(env["reason"]) if row["screen"] == SCREEN_OBLIGATION and env.get("plan"): plans[rsid] = env["plan"] row["assess_sha"] = str((env["plan"] or {}).get("assess_sha") or "") row["subject_count"] = len((env["plan"] or {}).get("subjects") or []) rows.append(row) # ★★**산 것을 판정보다 먼저 남긴다.** N번째에서 끊기면 앞의 N-1 유료 # 결과가 통째로 사라진다 — 실제로 겪은 부류다. _flush(on_row, row, plans.get(rsid)) counts = {s: sum(1 for r in rows if r["screen"] == s) for s in SCREENS} by_owner: Dict[str, Dict[str, int]] = {} for r in rows: by_owner.setdefault(r["owner_type"], {}) by_owner[r["owner_type"]][r["screen"]] = ( by_owner[r["owner_type"]].get(r["screen"], 0) + 1) _assert_ledger_covers(subjects, rows) return { "contract_version": SCREEN_CONTRACT_VERSION, "rows": rows, "plans": plans, "obligations": [r["research_subject_id"] for r in rows if r["screen"] == SCREEN_OBLIGATION], "counts": counts, "by_owner": by_owner, # ★셋을 따로 남긴다 — 합쳐 두면 「상한이 무엇을 막았나」를 못 읽는다. "assess_bought": bought, "assess_reused": reused, "assess_calls": bought + reused, "max_assess": cap, } def _assert_population(subjects: Sequence[Dict[str, Any]]) -> None: """★**사기 전에** 신원을 본다. 빈 id 는 붙일 데를 모르는 것이다. 자리별로 본다 — 집합·합계로 보면 양쪽이 다 비었을 때 통과한다. """ blank = [i for i, s in enumerate(subjects or ()) if not str(s.get("research_subject_id") or "").strip()] if blank: raise AssertionError( f"대상에 `research_subject_id` 가 빈 것이 {len(blank)}개 있다 " f"(자리 {blank[:5]}) — 판별 결과를 어디에 붙일지 못 정한다") def _assert_ledger_covers(subjects: Sequence[Dict[str, Any]], rows: Sequence[Dict[str, Any]]) -> None: """★**id 집합으로** 잰다 — 개수만 세면 항진식이다. 대상마다 행을 하나씩 넣고 그 행 수를 다시 세면 합은 **언제나** 맞는다. 빈 id 도 중복 id 도 그렇게는 안 잡힌다 (`grounding_carry` 에서 겪었다). """ want = [str(s.get("research_subject_id") or "").strip() for s in subjects or ()] got = [str(r.get("research_subject_id") or "").strip() for r in rows] if want != got: raise AssertionError( f"판별 장부가 대상과 안 맞는다 — 대상 {len(want)}개, 장부 {len(got)}개") dup = sorted({v for v in want if v and want.count(v) > 1}) if dup: raise AssertionError( f"같은 `research_subject_id` 가 {len(dup)}개 겹친다 {dup[:5]} — " "장부가 어느 대상인지 못 적는다") def obligation_ids(screen_data: Optional[Dict[str, Any]]) -> set: """판별 CP → **참조를 사야 하는 id 들.** ★읽는 자리를 **한 곳**으로 둔다. `entity_filter` 보호·앞쪽 등록·중앙 획득이 전부 이것을 부른다. 세 곳이 각자 `data["obligations"]` 를 파면 한 곳만 고쳐진다. """ ids = {str(x or "").strip() for x in ((screen_data or {}).get("obligations") or [])} ids.discard("") return ids def unresolved_ids(screen_data: Optional[Dict[str, Any]]) -> set: """판별이 **못 선** id 들. ★「비대상」과 다르다 — 하류가 막혀야 한다.""" out = {str(r.get("research_subject_id") or "").strip() for r in ((screen_data or {}).get("rows") or []) if r.get("screen") in (SCREEN_UNRESOLVED, SCREEN_CAPPED)} out.discard("") return out def promotable_obligations(screen_data: Optional[Dict[str, Any]]) -> set: """★**사야 하는데 붙을 행이 없는 base 대상**의 id. 앞쪽 등록(`materialize_missing_entities`)이 만들 것이 정확히 이것이다. 두 칸(`screen`·`binding`)을 **여기서 한 번** 합친다 — 호출부가 각자 합치면 한쪽만 고쳐진다. ★facet 은 `BIND_DEFERRED_PRODUCER` 라 자연히 빠진다. owner 이름을 다시 보지 않는다. """ out = {str(r.get("research_subject_id") or "").strip() for r in ((screen_data or {}).get("rows") or []) if r.get("screen") == SCREEN_OBLIGATION and r.get("binding") == BIND_PROMOTED_ROW} out.discard("") return out def facet_obligations(screen_data: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]: """★**사야 하는데 붙을 데가 없는 것들.** §2-6.5a producer 가 받을 몫이다. 이 목록이 비어 있지 않은 채로 「다섯 갈래 완료」를 말하면 거짓이다. """ return [r for r in ((screen_data or {}).get("rows") or []) if r.get("screen") == SCREEN_OBLIGATION and r.get("binding") == BIND_DEFERRED_PRODUCER]