"""활성화 **안전문** — 하나라도 빠지면 조립 자체가 선다. ★새 SOT 를 안 만든다. ## 왜 여기인가 Codex (2026-09-01) — 「`grounding_mode` 가 아니라 별도 계약 모듈 + `step_catalog._build()` 의 조립 직후가 맞습니다. `step_catalog` 가 `STEP_MANIFEST` 와 `STEP_CLASSES` 가 **처음 실제로 합쳐지는** composition root 라 순환 없이 **실제 조립물**을 검사할 수 있습니다.」 ★★**값을 import 해 새 SOT 를 만들지 않는다.** 조립된 catalog 와 실제 `GROUNDING_MODES`/`PLANNED_MODES`/`REFERENCE_SUPPORTED_OWNERS` 를 **인자로** 받는다. 그래야 이 모듈이 「또 하나의 목록」이 되지 않는다. ## 무엇을 보나 켠 판이면 (accepted) ①받는 집합에 있으면 **planned 에서 빠졌나** ②새 producer 와 중앙 조사기의 runner 가 catalog 에 **실제로 결속**됐나 ③차례·직접 의존이 판별·계획·아웃룩 **뒤**이고 씬 상세 **앞**인가 ④옛 유료 갈래(A0·plan·research)가 그 판에서 **미적용**이고 중앙 참조 조사만 적용인가 ⑤`REFERENCE_SUPPORTED_OWNERS` 가 중앙 장부가 낼 **다섯 갈래를 다** 받나 아직 안 켠 판이면 (planned) ⑥그 새 경로가 **어떤 받는 mode 에서도 닿으면 안 된다** ★이미 있는 inert 함수 자체는 금지 대상이 아니다 — **닿는 것**이 문제다 ## 왜 한꺼번에인가 하나만 먼저 켜면 **반쪽 판**이 된다 — 옛 producer 는 켜져 있고 새 producer 는 manifest 에 없어서, 옛 유료 사슬만 돌고 새 산출은 안 난다. 2026-08-31 에 실제로 그랬다. """ from __future__ import annotations from typing import Any, Dict, Mapping, Sequence, Set ACTIVATION_CONTRACT_VERSION = "1.202609013200" #: 새 갈래를 이루는 스텝들. ★이름은 manifest 것을 그대로 쓴다. CHUNK_PRODUCER_STEP = "grounding_chunk" SCREEN_STEP = "grounding_screen" CENTRAL_ACQUISITION_STEP = "reference_acquisition" PLAN_STEP = "grounding_plan" OUTLOOK_LAST_STEP = "outlook_phase3" SCENE_DETAIL_STEP = "scene_detail" #: 켜면 **꺼져야 하는** 옛 유료 갈래. ★`if_grounding_v2` 를 쓰는 것들. LEGACY_PAID_STEPS = ("grounding_a0", "grounding_plan", "grounding_research") #: ★★옛 **추출** 여섯. C(c) 는 같은 판독에서 엔티티 줄을 함께 내므로 이것들이 #: 같이 돌면 **같은 것을 두 번 산다** (Codex 재현 · 2026-09-01). #: ★앞 판 안전문은 위 셋만 봐서 이 중복 구매를 **초록으로 통과시켰다**. LEGACY_EXTRACT_STEPS = ("entity_all_character", "entity_all_location", "entity_all_prop", "entity_extract_character", "entity_extract_location", "entity_extract_prop") #: 그것들이 써야 하는 applicability 이름. NOT_CHUNK_APPLICABILITY = "if_not_chunk_producer" #: 새 갈래가 켜졌을 때 판별이 쓰는 applicability 이름. APPLICABILITY_DISABLED = "disabled" #: 옛 유료 갈래 셋이 쓰는 applicability 이름. ★이 이름이 `buys_v2_research` #: 를 묻고, 그 술어가 `mode == v2` **정확 비교**라 `v2_chunk` 에서 자동으로 #: 꺼진다. 그래서 정적으로 `disabled` 로 바꿀 필요가 **없다** — `v2` 가 #: 여전히 받는 값이라 그렇게 하면 **v2 주행이 깨진다**. LEGACY_APPLICABILITY = "if_grounding_v2" class ActivationContractError(RuntimeError): """활성화가 **반쪽**이다. ★조립 자체가 선다 — provider 앞이다.""" def _entry(catalog: Mapping[str, Any], sid: str): got = catalog.get(sid) if got is None: raise ActivationContractError( f"`{sid}` 가 catalog 에 없다 — 활성화에 필요한 스텝이 빠졌다") return got def _order(catalog: Mapping[str, Any], sid: str) -> float: return float(getattr(_entry(catalog, sid), "order")) def assert_activation(catalog: Mapping[str, Any], *, accepted_modes: Set[str], planned_modes: Set[str], chunk_mode: str, reference_owners: Sequence[str], ledger_owners: Sequence[str], buys_legacy_research: Any = None) -> None: """활성화가 **온전한가**. ★하나라도 빠지면 여기서 선다. Args: catalog: `step_catalog._build()` 가 **방금 만든** 것. accepted_modes: 실제 `GROUNDING_MODES`. planned_modes: 실제 `PLANNED_MODES`. chunk_mode: `GROUNDING_MODE_V2_CHUNK`. reference_owners: 실제 `REFERENCE_SUPPORTED_OWNERS`. ledger_owners: 중앙 장부가 낼 갈래 — `MATERIALIZABLE_OWNER_TYPES`. buys_legacy_research: `buys_v2_research`. ★켠 판에서 이것이 `chunk_mode` 에 **False** 를 내야 옛 유료 갈래가 꺼진다. 안 켠 판에서는 그 값을 물으면 거절당하므로 **안 부른다**. """ if chunk_mode in accepted_modes and chunk_mode in planned_modes: raise ActivationContractError( f"{chunk_mode!r} 가 받는 집합과 계획 집합에 **둘 다** 있다 — " "켠 것인지 안 켠 것인지 모른다") if chunk_mode in accepted_modes: _assert_switched_on(catalog, chunk_mode=chunk_mode, reference_owners=reference_owners, ledger_owners=ledger_owners, buys_legacy_research=buys_legacy_research) else: _assert_still_unreachable(catalog) def _assert_switched_on(catalog: Mapping[str, Any], *, chunk_mode: str, reference_owners: Sequence[str], ledger_owners: Sequence[str], buys_legacy_research: Any = None) -> None: # ②runner 가 **실제로 결속**됐나 — manifest 에만 있고 클래스가 없으면 # 그 스텝은 못 돈다 for sid in (CHUNK_PRODUCER_STEP, SCREEN_STEP, CENTRAL_ACQUISITION_STEP): if getattr(_entry(catalog, sid), "runner_cls", None) is None: raise ActivationContractError( f"`{sid}` 에 runner 가 안 붙었다 — manifest 에만 있으면 못 돈다") if str(getattr(_entry(catalog, sid), "applicability", "")) == \ APPLICABILITY_DISABLED: raise ActivationContractError( f"`{sid}` 가 아직 `{APPLICABILITY_DISABLED}` 다 — 안 돈다") # ③차례와 **직접 의존** — 아웃룩 뒤, 씬 상세 앞 central = _entry(catalog, CENTRAL_ACQUISITION_STEP) for before in (SCREEN_STEP, PLAN_STEP, OUTLOOK_LAST_STEP): if _order(catalog, before) >= _order(catalog, CENTRAL_ACQUISITION_STEP): raise ActivationContractError( f"`{CENTRAL_ACQUISITION_STEP}` 가 `{before}` 보다 앞이다 — " "그 산출 없이 조사하게 된다") if OUTLOOK_LAST_STEP not in set(getattr(central, "depends_on", ())): raise ActivationContractError( f"`{CENTRAL_ACQUISITION_STEP}` 가 `{OUTLOOK_LAST_STEP}` 에 " "직접 의존하지 않는다 — 차례만 맞고 재개에서 어긋난다") if _order(catalog, CENTRAL_ACQUISITION_STEP) >= _order(catalog, SCENE_DETAIL_STEP): raise ActivationContractError( f"`{CENTRAL_ACQUISITION_STEP}` 가 `{SCENE_DETAIL_STEP}` 보다 " "뒤다 — 참조 없이 씬이 만들어진다") # ④옛 유료 갈래는 **그 판에서** 꺼져야 한다 # ★정적으로 `disabled` 로 바꾸면 안 된다 — `v2` 가 여전히 받는 값이라 # 그렇게 하면 v2 주행이 깨진다. 대신 **술어가 그 판에서 False 를 내는지** # 본다. 그러려면 ⓐ그 스텝들이 그 술어를 쓰고 ⓑ술어가 False 여야 한다. wrong = [sid for sid in LEGACY_PAID_STEPS if sid in catalog and str(getattr(catalog[sid], "applicability", "")) not in (LEGACY_APPLICABILITY, APPLICABILITY_DISABLED)] if wrong: raise ActivationContractError( f"옛 유료 갈래 {wrong} 가 `{LEGACY_APPLICABILITY}` 를 안 쓴다 — " "그 판에서 꺼지는지 알 수 없다") if buys_legacy_research is not None and buys_legacy_research(chunk_mode): raise ActivationContractError( f"`{chunk_mode}` 에서 옛 유료 조사 술어가 아직 참이다 — " "옛 갈래와 새 갈래가 **같이 돌아 두 번 산다**") # ★★씬이 **중앙 조사 CP 에 의존**해야 한다 — 차례만 맞고 의존이 없으면 # 조사가 다시 돌아도 씬이 옛 참조를 그대로 쓴다. if CENTRAL_ACQUISITION_STEP not in set( getattr(_entry(catalog, SCENE_DETAIL_STEP), "depends_on", ())): raise ActivationContractError( f"`{SCENE_DETAIL_STEP}` 가 `{CENTRAL_ACQUISITION_STEP}` 에 직접 " "의존하지 않는다 — 조사가 다시 돌아도 씬이 옛 참조를 쓴다") # ★★옛 **추출** 여섯도 그 판에서 꺼져야 한다 — 안 그러면 두 번 산다 loud = [sid for sid in LEGACY_EXTRACT_STEPS if sid in catalog and str(getattr(catalog[sid], "applicability", "")) not in (NOT_CHUNK_APPLICABILITY, APPLICABILITY_DISABLED)] if loud: raise ActivationContractError( f"옛 추출 {loud} 가 `{NOT_CHUNK_APPLICABILITY}` 를 안 쓴다 — " "C(c) 판에서 같이 돌아 **같은 것을 두 번 산다**") # ⑤다섯 갈래를 **다** 받나 missing = [o for o in ledger_owners if o not in set(reference_owners)] if missing: raise ActivationContractError( f"중앙 장부가 낼 갈래 {missing} 를 `REFERENCE_SUPPORTED_OWNERS` 가 " "안 받는다 — 그 갈래는 참조를 못 만든다") # ⑥집행자가 **선언만 있고 구현이 없으면** 선다 (Codex 2026-09-01). # ★「누락을 안다」는 완료가 아니다 — 값을 하나 더 적어 놓고 아무도 # 그것을 집행 안 하면 그 갈래의 참조는 조용히 사라진다. _assert_every_gate_is_implemented() #: 집행자를 **실제로 구현하는** 모듈들. ★각 모듈이 자기 값을 선언한다 — #: 여기 목록을 손으로 다시 적으면 한쪽만 고쳐진다. GATE_IMPLEMENTORS = ( ("app.core.steps.episode_reference_policy_step", "ENFORCES_GATE"), ("app.modules.pipeline.grounding_bundle_projection", "ENFORCES_GATE"), ("app.modules.pipeline.grounding_sidecar_writer", "ENFORCES_GATE"), ) def implemented_gates() -> Set[str]: """지금 **실제로 집행되는** 값들. ★공개 끝점.""" import importlib got = set() for mod, attr in GATE_IMPLEMENTORS: m = importlib.import_module(mod) val = getattr(m, attr, None) if val: got.add(str(val)) return got def _assert_every_gate_is_implemented() -> None: from app.modules.pipeline.grounding_entity_contract import ( ENFORCEMENT_GATES, REFERENCE_ENFORCEMENT_BY_OWNER) have = implemented_gates() orphan = [g for g in ENFORCEMENT_GATES if g not in have] if orphan: raise ActivationContractError( f"집행자 {orphan} 를 선언해 놓고 **구현한 모듈이 없다** — " f"그 갈래(" f"{[o for o, g in REFERENCE_ENFORCEMENT_BY_OWNER.items() if g in orphan]}" f")의 참조는 아무 데도 안 붙는다") unknown = [g for g in sorted(have) if g not in ENFORCEMENT_GATES] if unknown: raise ActivationContractError( f"모듈이 모르는 집행자 {unknown} 를 선언한다 — " f"{ENFORCEMENT_GATES}") def _assert_still_unreachable(catalog: Mapping[str, Any]) -> None: """아직 안 켰다 — 새 경로가 **어떤 받는 mode 에서도 닿으면 안 된다**. ★이미 있는 inert 함수 자체는 금지 대상이 아니다. **catalog 에 붙어 돌 수 있는 것**이 문제다. """ for sid in (CHUNK_PRODUCER_STEP,): if sid in catalog: raise ActivationContractError( f"`{sid}` 가 아직 안 켠 판인데 catalog 에 있다 — " "새 producer 만 켜진 **반쪽 판**이다") scr = catalog.get(SCREEN_STEP) if scr is not None and str(getattr(scr, "applicability", "")) != \ APPLICABILITY_DISABLED: raise ActivationContractError( f"`{SCREEN_STEP}` 가 아직 안 켠 판인데 `" f"{APPLICABILITY_DISABLED}` 가 아니다")