# -*- coding: utf-8 -*- """anu_v3.profile_adoption_planner — STEP 2 profile engine operational adoption 준비 planner (task-2553+42). 회장 3단계 지시 Step 2 (회장 GO, 코드/파일 자동화 — 문서-only 아님): +38 seam(`anu_v3.dispatch_profile_selection`) · +39 binding seam (`anu_v3.coordinator_profile_binding`) · C1 engine (`anu_v3.policy_profile_engine`) 를 **read-only 소비**하여, profile engine 을 실 운영 dispatch selection / batch coordinator 경로에 연결할 때 touch 될 live touchpoint 를 기계가독 enumeration 하고, adoption plan (expected_files allowlist 후보 · conflict set · risk tier(LOW/MED/HIGH)) + read-only adoption dry-run(실 in-place adoption 0) 을 산출한다. 9-R.1 (본문 우선 — 2-layer 분리): - Layer A (deliverable module 경계): 본 모듈·dry-run 은 callback/collector **소스 경로를 읽거나 수정하지 않으며**, adoption 을 **실제 적용하지 않는다**(plan + simulate only). seam/engine/coordinator/frozen anchor **byte-0**. module side-effect 0 (순수 — decision 경계 밖 I/O 는 명시 hard-guarded helper 1개 `emit_adoption_plan` 한정). - Layer B (executor lifecycle): §8 normal completion callback 은 executor 프로세스 종료 신호로 본 모듈의 책임 밖이다 (본 모듈은 cron 을 등록/ 제거/조작하지 않는다). 설계 invariant (§3 / §5 / 9-R.1): - 신규 별도 additive 모듈. engine·+38·+39·frozen `parallel_batch_coordinator.py`·callback frozen anchor·durable v1· +22~+41 원본 무수정. - +38/+39/C1 은 **import-only read-only 소비** (정본 module attribute introspection 만 — engine resolve_policy 를 live profile dir 에 대해 실행하지 않는다; mutation 0, 파일 write 0, network 0, git 0, PR 0, merge 0, cron 0). - adoption 은 **계획·시뮬레이션만**. 실 in-place 연결 실행 0 — 실 채택은 본 task 후 별도 회장 GO. - closeout/merge/auto-confirm 자동확정 0 (hard-pinned False). - 모든 decision-logic 메서드 = derive / propose / read only. 파일 I/O 는 decision 경계 밖의 명시 hard-guarded helper 1개(`emit_adoption_plan`) 한정. - 외부 의존성 0 (stdlib only — offline 100%). """ from __future__ import annotations import json import subprocess from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, Final, List, Mapping import anu_v3.coordinator_profile_binding as _cpb import anu_v3.dispatch_profile_selection as _dps import anu_v3.policy_profile_engine as _ppe PLANNER_MODULE: Final[str] = "anu_v3.profile_adoption_planner" PLANNER_VERSION: Final[str] = "task-2553+42.Step2.v1" PLAN_SCHEMA: Final[str] = "anu_v3.profile_adoption_planner.adoption_plan.v1" DRY_RUN_SCHEMA: Final[str] = "anu_v3.profile_adoption_planner.dry_run.v1" # 본 planner 는 plan + simulate only — 실 adoption / write / merge / cron / # PR 실행 0 (회장 §5 / 9-R.1 Layer A). ADOPTION_LIFECYCLE_EFFECT: Final[str] = "none" DRY_RUN_APPLIED_COUNT: Final[int] = 0 # 실 in-place adoption 0 (불변). # risk tier (회장 §3.1). RISK_LOW: Final[str] = "LOW" RISK_MED: Final[str] = "MED" RISK_HIGH: Final[str] = "HIGH" _RISK_ORDER: Final[Dict[str, int]] = {RISK_LOW: 0, RISK_MED: 1, RISK_HIGH: 2} # §4 expected_files allowlist (이 외 write 0 — plan 산출물이 합법적으로 # write 될 수 있는 경로 후보. planner 자체는 write 하지 않는다). EXPECTED_FILES_ALLOWLIST: Final[tuple[str, ...]] = ( "anu_v3/profile_adoption_planner.py", "schemas/profile_adoption_plan.schema.json", "tests/regression/test_profile_adoption_planner_2553plus42.py", "memory/fixtures/task-2553+42.*", "memory/events/task-2553+42.decision.json", "memory/events/task-2553+42.result.json", "memory/events/task-2553+42.adoption-plan.json", "memory/reports/task-2553+42.md", ) # frozen / byte-0 anchor — 실 adoption 이 절대 in-place touch 하면 안 되는 # 경로 (conflict set 의 기반). 회장 §5 verbatim + frozen anchor. FROZEN_ANCHORS: Final[tuple[str, ...]] = ( "anu_v3/policy_profile_engine.py", # C1 engine, byte-0 "anu_v3/dispatch_profile_selection.py", # +38 seam, byte-0 "anu_v3/coordinator_profile_binding.py", # +39 binding seam, byte-0 "anu_v3/parallel_batch_coordinator.py", # frozen coordinator "utils/anu_delegation_completion_callback.py", # 83b3e307… frozen "task-2553.parallel-batch-state.json", # durable v1, byte-0 ) # callback/collector 소스 경로 — 본 planner 가 절대 읽거나 수정하지 않는 # (Layer A 무접촉) 경로. adoption 도 이 경로를 touch 하면 안 된다. CALLBACK_COLLECTOR_PATHS: Final[tuple[str, ...]] = ( "utils/anu_delegation_completion_callback.py", "anu_v3/executor_callback_contract.py", ) # durable v1 chair 파일 — emit 절대 금지 (byte-0). _FROZEN_DURABLE_V1: Final[str] = "task-2553.parallel-batch-state.json" _DELIVERABLE_SUFFIX: Final[str] = ".adoption-plan.json" _DELIVERABLE_MARKER: Final[str] = f'"plan_schema": "{PLAN_SCHEMA}"' class ProfileAdoptionPlannerError(ValueError): """planner 산출 실패. ``self.code`` = 실패 사유 코드 (fail-closed).""" def __init__(self, code: str, message: str) -> None: super().__init__(f"[{code}] {message}") self.code = code self.message = message class FrozenWriteRefused(RuntimeError): """frozen / git-tracked / 무관 untracked 파일 write 거부. plan deliverable 은 별도 NEW untracked 경로여야 한다 (§4/§5).""" # --------------------------------------------------------------------------- # 1. seam / engine read-only introspection (import-only, mutation 0) # --------------------------------------------------------------------------- def introspect_seams() -> Dict[str, Any]: """+38 seam · +39 binding seam · C1 engine 의 정본 contract 를 **read-only attribute introspection** 으로 소비한다. engine resolve_policy 를 live profile dir 에 대해 실행하지 않는다 — module attribute 만 읽는다 (mutation 0, side-effect 0). 반환은 순수 dict (engine/seam 객체 노출 0). """ return { "engine": { "module": _ppe.ENGINE_MODULE, "version": _ppe.ENGINE_VERSION, "decision_schema": "anu_v3.policy_profile_engine.decision.v1", "primary_api": ["parse_goal_request", "resolve_policy"], "side_effect_contract": ( "pure — GitHub API 0, token 0, 파일 write 0 " "(docstring invariant)" ), }, "dispatch_seam": { "module": _dps.SEAM_MODULE, "version": _dps.SEAM_VERSION, "selection_schema": _dps.SELECTION_SCHEMA_ID, "entrypoints": [ "select_profile_for_dispatch", "run_dispatch_profile_selection", ], "lifecycle_effect": _dps.DISPATCH_LIFECYCLE_EFFECT, "fail_closed_status": [ _dps.STATUS_SELECTED, _dps.STATUS_REFUSED, _dps.STATUS_HOLD, ], }, "coordinator_binding_seam": { "module": _cpb.BINDING_MODULE, "version": _cpb.BINDING_VERSION, "binding_schema": _cpb.BINDING_SCHEMA, "accepted_decision_schema": _cpb.ACCEPTED_DECISION_SCHEMA, "entrypoints": ["consume_for_coordinator", "load_profile_decision"], "auto_confirm_invariant": ( "closeout/merge/auto_confirm hard-pinned False " "(coordinator=판단보조 소비만)" ), }, } # --------------------------------------------------------------------------- # 2. adoption touchpoint enumeration (기계가독, read-only derive) # --------------------------------------------------------------------------- @dataclass class AdoptionTouchpoint: """실사용 연결 시 touch 될 live 경로 1건의 기계가독 기술자.""" touchpoint_id: str track: str seam_module: str seam_entrypoint: str consumes: str live_target_candidates: List[str] adoption_kind: str in_place_edit_required: bool touches_frozen_anchor: bool touches_callback_collector: bool risk_tier: str rationale: str def to_dict(self) -> Dict[str, Any]: return { "touchpoint_id": self.touchpoint_id, "track": self.track, "seam_module": self.seam_module, "seam_entrypoint": self.seam_entrypoint, "consumes": self.consumes, "live_target_candidates": list(self.live_target_candidates), "adoption_kind": self.adoption_kind, "in_place_edit_required": self.in_place_edit_required, "touches_frozen_anchor": self.touches_frozen_anchor, "touches_callback_collector": self.touches_callback_collector, "risk_tier": self.risk_tier, "rationale": self.rationale, } def _is_frozen(path: str) -> bool: return path in FROZEN_ANCHORS def _is_callback_collector(path: str) -> bool: return path in CALLBACK_COLLECTOR_PATHS def enumerate_touchpoints(introspection: Mapping[str, Any]) -> List[AdoptionTouchpoint]: """seam introspection → 실사용 연결 touchpoint enumeration. 실사용 grep 결과(현재 어떤 live dispatch/coordinator 도 seam 을 호출하지 않음 — 실 adoption 미수행)를 전제로, adoption 시 touch 될 live 경로를 derive 한다. live target 후보는 **문자열로만 기술**(읽기·수정 0). """ ds = introspection["dispatch_seam"] cb = introspection["coordinator_binding_seam"] eng = introspection["engine"] tps: List[AdoptionTouchpoint] = [] # ── TA: Track A — live dispatch 경로 → dispatch selection seam 결선 ── ta_targets = [ "dispatch/core.py", "dispatch/prompt.py", "anu_v3/active_dispatch_scanner.py", ] tps.append( AdoptionTouchpoint( touchpoint_id="TA.dispatch_selection_wire", track="A", seam_module=ds["module"], seam_entrypoint="run_dispatch_profile_selection", consumes=f"{eng['module']} (parse_goal_request, resolve_policy)", live_target_candidates=ta_targets, adoption_kind=( "live dispatch 경로가 seam file-level contract 를 호출하도록 " "in-place wire (seam byte-0 — live 측만 추가 호출)" ), in_place_edit_required=True, touches_frozen_anchor=any(_is_frozen(p) for p in ta_targets), touches_callback_collector=any( _is_callback_collector(p) for p in ta_targets ), risk_tier=RISK_MED, rationale=( "seam DISPATCH_LIFECYCLE_EFFECT=='none' (선택만, 실행 0) — " "live 측 tracked dispatch 코드에 1 call-site 추가 필요 " "(가역, frozen 무접촉). 실 적용은 별도 회장 GO." ), ) ) # ── TB: Track B — batch coordinator → binding seam decision 소비 ── tb_frozen = "anu_v3/parallel_batch_coordinator.py" tb_safe = "anu_v3/generic_batch_coordinator.py" tps.append( AdoptionTouchpoint( touchpoint_id="TB.coordinator_binding_consume", track="B", seam_module=cb["module"], seam_entrypoint="consume_for_coordinator", consumes=( f"{cb['accepted_decision_schema']} (engine decision.v1 " "file-level contract)" ), live_target_candidates=[tb_frozen, tb_safe], adoption_kind=( "batch coordinator 가 engine decision.v1 을 binding seam " "통해 read-only 소비 (closeout/merge 자동확정 0 불변)" ), in_place_edit_required=True, touches_frozen_anchor=_is_frozen(tb_frozen), touches_callback_collector=False, risk_tier=RISK_HIGH, rationale=( f"후보 {tb_frozen} 는 frozen anchor(byte-0) — 그 경로 in-place " f"adoption 은 §5/§6 위반(CONFLICT). 안전 route={tb_safe} " "(non-frozen, MED)로 한정해야 함. 실 적용은 별도 회장 GO." ), ) ) # ── TE: engine decision output 생산 (additive, in-place edit 불요) ── tps.append( AdoptionTouchpoint( touchpoint_id="TE.engine_decision_emit", track="C1", seam_module=eng["module"], seam_entrypoint="resolve_policy → PolicyResolution.to_decision_dict", consumes="(none — engine 은 decision.v1 생산측)", live_target_candidates=[], adoption_kind=( "engine 은 이미 decision.v1 contract 를 노출 — adoption 시 " "engine in-place edit 불요 (byte-0 유지)" ), in_place_edit_required=False, touches_frozen_anchor=False, touches_callback_collector=False, risk_tier=RISK_LOW, rationale=( "engine 정본 API 가 이미 완비 — additive contract, " "adoption 시 engine 무변 (LOW)." ), ) ) return tps # --------------------------------------------------------------------------- # 3. conflict set + risk tier (read-only derive, fail-closed) # --------------------------------------------------------------------------- def classify_conflicts( touchpoints: List[AdoptionTouchpoint], ) -> Dict[str, Any]: """frozen anchor / callback-collector 와 충돌하는 adoption 후보 식별. conflict = frozen anchor 를 in-place touch 하거나 callback/collector 경로를 건드리는 live target 후보 (그 route 는 §5/§6 위반 — 안전 route 로 한정 필요). fail-closed: 충돌 후보를 누락 없이 등재. """ frozen_conflicts: List[Dict[str, Any]] = [] callback_conflicts: List[Dict[str, Any]] = [] for tp in touchpoints: for cand in tp.live_target_candidates: if _is_frozen(cand): frozen_conflicts.append( { "touchpoint_id": tp.touchpoint_id, "candidate": cand, "reason": "frozen anchor byte-0 — in-place adoption 금지", "required_mitigation": ( "non-frozen 안전 route 로 한정 (frozen 경로 배제)" ), } ) if _is_callback_collector(cand): callback_conflicts.append( { "touchpoint_id": tp.touchpoint_id, "candidate": cand, "reason": "callback/collector 경로 — §5 무접촉", "required_mitigation": "해당 route 전면 배제", } ) # conflict_count = 충돌 route 의 **distinct (touchpoint, candidate)** # 개수 — frozen ∩ callback 인 candidate 가 양쪽에 등재돼도 1회만 계수 # (Codex (f) double-count fix). distinct_routes = { (c["touchpoint_id"], c["candidate"]) for c in (*frozen_conflicts, *callback_conflicts) } return { "frozen_anchor_conflicts": frozen_conflicts, "callback_collector_conflicts": callback_conflicts, "frozen_anchors": list(FROZEN_ANCHORS), "callback_collector_paths": list(CALLBACK_COLLECTOR_PATHS), "conflict_count": len(distinct_routes), } def overall_risk_tier(touchpoints: List[AdoptionTouchpoint]) -> str: """touchpoint 중 최고 risk tier (fail-closed — 최악 우선).""" if not touchpoints: return RISK_LOW return max( (tp.risk_tier for tp in touchpoints), key=lambda r: _RISK_ORDER.get(r, _RISK_ORDER[RISK_HIGH]), ) # --------------------------------------------------------------------------- # 4. adoption plan (read-only derive) # --------------------------------------------------------------------------- def build_adoption_plan(introspection: Mapping[str, Any] | None = None) -> Dict[str, Any]: """expected_files allowlist 후보 · conflict set · risk tier 를 담은 기계가독 adoption plan. 실 adoption 0 (plan only).""" intro = dict(introspection) if introspection is not None else introspect_seams() tps = enumerate_touchpoints(intro) conflicts = classify_conflicts(tps) overall = overall_risk_tier(tps) return { "plan_schema": PLAN_SCHEMA, "planner_module": PLANNER_MODULE, "planner_version": PLANNER_VERSION, "task": "task-2553+42", "step": "Step 2 — profile engine operational adoption 준비", "adoption_lifecycle_effect": ADOPTION_LIFECYCLE_EFFECT, "real_in_place_adoption_count": 0, "seam_introspection": intro, "touchpoints": [tp.to_dict() for tp in tps], "expected_files_allowlist": list(EXPECTED_FILES_ALLOWLIST), "conflict": conflicts, "risk": { "overall_tier": overall, "per_touchpoint": { tp.touchpoint_id: tp.risk_tier for tp in tps }, "tiers": [RISK_LOW, RISK_MED, RISK_HIGH], }, "byte0_anchors": list(FROZEN_ANCHORS), "callback_collector_untouched": True, "auto_confirm_invariant": ( "closeout/merge/auto_confirm 자동확정 0 (hard-pinned False, " "모든 입력 불변)" ), "real_adoption_gate": ( "실 in-place adoption 은 본 task 후 별도 회장 GO — 본 산출은 " "plan + dry-run only" ), } # --------------------------------------------------------------------------- # 5. read-only adoption dry-run (시뮬레이션만, 실 적용 0) # --------------------------------------------------------------------------- def dry_run_adoption(plan: Mapping[str, Any] | None = None) -> Dict[str, Any]: """실 in-place 연결을 **시뮬레이션만** — 예상 diff·conflict·rollback 지점을 산출하되 적용 0 (write/merge/cron/PR 0, callback/collector 무접촉). """ p = dict(plan) if plan is not None else build_adoption_plan() simulated_diffs: List[Dict[str, Any]] = [] rollback_points: List[Dict[str, Any]] = [] blocked: List[Dict[str, Any]] = [] blocked_callback: List[Dict[str, Any]] = [] for tp in p["touchpoints"]: if not tp["in_place_edit_required"]: simulated_diffs.append( { "touchpoint_id": tp["touchpoint_id"], "change": "NONE (additive contract 이미 완비)", "applied": False, } ) continue safe_targets = [ c for c in tp["live_target_candidates"] if c not in FROZEN_ANCHORS and c not in CALLBACK_COLLECTOR_PATHS ] frozen_targets = [ c for c in tp["live_target_candidates"] if c in FROZEN_ANCHORS ] callback_targets = [ c for c in tp["live_target_candidates"] if c in CALLBACK_COLLECTOR_PATHS ] for ft in frozen_targets: blocked.append( { "touchpoint_id": tp["touchpoint_id"], "candidate": ft, "decision": "BLOCKED (frozen byte-0 — dry-run 거부)", } ) # callback/collector route 는 명시 BLOCK — simulated diff 로 표현 # 하지 않는다 (Codex (f) callback misrepresentation fix). 본 모듈은 # 그 경로를 읽지도 시뮬레이트하지도 않는다 (Layer A 무접촉 불변). for ct in callback_targets: blocked_callback.append( { "touchpoint_id": tp["touchpoint_id"], "candidate": ct, "decision": "BLOCKED (callback/collector — §5 무접촉, dry-run 거부)", } ) if not safe_targets: # safe route 부재 → simulated diff 0 (오표현 금지). 차단 사실만 # 기록하고 chair routing 으로 escalate (Codex (f) fix). simulated_diffs.append( { "touchpoint_id": tp["touchpoint_id"], "change": ( "NO_SAFE_ROUTE — 모든 후보가 frozen/callback-collector " "(chair routing 필요, 시뮬레이트 0)" ), "expected_hunk": "(none — safe route 부재)", "applied": False, } ) continue simulated_diffs.append( { "touchpoint_id": tp["touchpoint_id"], "change": ( f"+1 call-site → {tp['seam_entrypoint']} " f"(safe route 후보: {safe_targets})" ), "expected_hunk": ( "live 측 단일 호출 추가 (seam byte-0; live 코드만 가역 추가)" ), "applied": False, } ) rollback_points.append( { "touchpoint_id": tp["touchpoint_id"], "rollback": ( "추가된 single call-site 제거 (git revert of live edit) — " "seam/engine/coordinator 무변이므로 rollback 은 live 1-hunk 한정" ), } ) return { "dry_run_schema": DRY_RUN_SCHEMA, "planner_module": PLANNER_MODULE, "planner_version": PLANNER_VERSION, "task": "task-2553+42", "mode": "READ_ONLY_SIMULATION", "applied_count": DRY_RUN_APPLIED_COUNT, "writes": 0, "merges": 0, "cron_ops": 0, "pr_ops": 0, "callback_collector_touched": False, "frozen_anchor_touched": False, "simulated_diffs": simulated_diffs, "blocked_frozen_routes": blocked, "blocked_callback_routes": blocked_callback, "rollback_points": rollback_points, "side_effect": "none (simulation only — 실 in-place adoption 0)", } # --------------------------------------------------------------------------- # 6. single callable entrypoint (read-only derive) # --------------------------------------------------------------------------- def run_adoption_planner() -> Dict[str, Any]: """plan + dry-run 을 단일 호출로 산출 (순수 — write/merge/cron/PR 0).""" plan = build_adoption_plan() dry = dry_run_adoption(plan) return { "schema": "anu_v3.profile_adoption_planner.bundle.v1", "task": "task-2553+42", "adoption_plan": plan, "dry_run": dry, "real_in_place_adoption": False, "callback_collector_untouched": True, } # --------------------------------------------------------------------------- # fixture 로더 (read/parse/reference only; source mutation 0) # --------------------------------------------------------------------------- @dataclass class AdoptionPlannerFixture: cases: List[Dict[str, Any]] = field(default_factory=list) def load_planner_fixture(fixture_path: "str | Path") -> AdoptionPlannerFixture: data = json.loads(Path(fixture_path).read_text(encoding="utf-8")) return AdoptionPlannerFixture(cases=list(data.get("cases", []))) # --------------------------------------------------------------------------- # explicit hard-guarded emission — OUTSIDE the decision boundary (9-R.1) # --------------------------------------------------------------------------- def _is_git_tracked(p: Path) -> bool: try: top = subprocess.run( ["git", "-C", str(p.parent), "rev-parse", "--show-toplevel"], capture_output=True, text=True, ) if top.returncode != 0: return False repo = top.stdout.strip() rc = subprocess.run( ["git", "-C", repo, "ls-files", "--error-unmatch", str(p.resolve())], capture_output=True, text=True, ) return rc.returncode == 0 except Exception: return False def emit_adoption_plan(plan: Mapping[str, Any], out_path: "str | Path") -> Path: """NEW untracked adoption-plan 파일 write. decision component 아님 — runner/ANU 가 수행하는 명시 I/O (9-R.1 경계 밖, Layer A 의 module side-effect 아님). Hard write-guard (§4/§5 — +39 가드 미러, 무관 파일 절대 clobber 0): * chair durable v1 name -> REFUSE (byte-0 immutable) * frozen anchor basename -> REFUSE (byte-0) * git-tracked path -> REFUSE (tracked HEAD invariant) * untracked & non-existent -> ALLOW (NEW deliverable) * untracked & existing sanctioned -> ALLOW (idempotent re-emit) * untracked & existing non-deliver. -> REFUSE (무관 파일 보호) """ out = Path(out_path) frozen_names = {Path(a).name for a in FROZEN_ANCHORS} if out.name == _FROZEN_DURABLE_V1 or out.name in frozen_names: raise FrozenWriteRefused( f"refusing to write frozen/byte-0 anchor {out} (§5)" ) # callback/collector 경로는 read 도전 이전에 refuse (Codex (c) HIGH fix — # Layer A 무접촉: 본 helper 가 callback/collector source 를 절대 읽지 # 않도록 read_text 도달 전 차단). basename + 정규화 경로 모두 검사. _cc_names = {Path(c).name for c in CALLBACK_COLLECTOR_PATHS} try: _rel = str(out.resolve()).replace("\\", "/") except OSError: _rel = str(out).replace("\\", "/") if out.name in _cc_names or any( _rel.endswith(c) for c in CALLBACK_COLLECTOR_PATHS ): raise FrozenWriteRefused( f"refusing to write/read callback·collector path {out} " "(§5 무접촉, 9-R.1 Layer A — read 도달 전 차단)" ) if _is_git_tracked(out): raise FrozenWriteRefused( f"refusing to write git-tracked path {out} (§5 tracked HEAD " "invariant — only NEW untracked deliverables)" ) if out.exists(): sanctioned = out.name.endswith(_DELIVERABLE_SUFFIX) if not sanctioned: try: sanctioned = _DELIVERABLE_MARKER in out.read_text( encoding="utf-8" ) except (OSError, UnicodeDecodeError): sanctioned = False if not sanctioned: raise FrozenWriteRefused( f"refusing to overwrite existing untracked non-deliverable " f"{out} (only a NEW untracked path or this planner's own " f"adoption-plan deliverable may be written)" ) out.parent.mkdir(parents=True, exist_ok=True) out.write_text( json.dumps(plan, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) return out __all__ = [ "PLANNER_MODULE", "PLANNER_VERSION", "PLAN_SCHEMA", "DRY_RUN_SCHEMA", "ADOPTION_LIFECYCLE_EFFECT", "DRY_RUN_APPLIED_COUNT", "RISK_LOW", "RISK_MED", "RISK_HIGH", "EXPECTED_FILES_ALLOWLIST", "FROZEN_ANCHORS", "CALLBACK_COLLECTOR_PATHS", "ProfileAdoptionPlannerError", "FrozenWriteRefused", "AdoptionTouchpoint", "AdoptionPlannerFixture", "introspect_seams", "enumerate_touchpoints", "classify_conflicts", "overall_risk_tier", "build_adoption_plan", "dry_run_adoption", "run_adoption_planner", "load_planner_fixture", "emit_adoption_plan", ]