# -*- coding: utf-8 -*- """anu_v3.profile_dispatch_planning_adapter — §3.4: profile decision → DEFAULT dispatch *planning* 입력 어댑터 (task-2553+52 / Track 3). 목표 (회장 §3.4): default profile resolver(`anu_v3.default_profile_resolver`)의 decision output(`decision.v1`)을 기본 dispatch **planning** 경로가 그대로 소비할 수 있는 평면 입력으로 변환한다. 설계 invariant (§2 / §5 / §6): - 신규 별도 additive 모듈. resolver·engine·+38·+39·frozen anchor· durable v1 무수정. **import 결합 0** — anu_v3 import 전무, 순수 stdlib only (resolver decision = 파일레벨 contract 로만 소비). - planning 입력 derive 만 — 실 dispatch / PR / merge / branch 실행 0. ``DISPATCH_LIFECYCLE_EFFECT == "none"``, ``plan_only == True`` (hard-pinned). 어떤 입력에서도 write/merge/PR 자동확정 0. - resolver status != RESOLVED (REFUSED/CONFLICT/HOLD) → fail-closed: plan_admissible=False (planner 가 자동 진행 못 하도록 안전 신호). - decision schema marker mismatch / 부재 / 파싱 실패 = fail-closed (PLAN_UNAVAILABLE — 예외를 planner 가 진행 신호로 오인 불가). - 모든 메서드 = read-only derive/propose. 부작용 0 (파일 I/O 0). """ from __future__ import annotations import json from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, Final, List, Mapping ADAPTER_MODULE: Final[str] = "anu_v3.profile_dispatch_planning_adapter" ADAPTER_VERSION: Final[str] = "task-2553+52.Track3.v1" ADAPTER_SCHEMA: Final[str] = "anu_v3.profile_dispatch_planning_adapter.v1" # 소비 대상 = default resolver 의 파일레벨 contract (decision schema marker). ACCEPTED_DECISION_SCHEMA: Final[str] = "anu_v3.default_profile_resolver.decision.v1" _RESOLVED: Final[str] = "RESOLVED" # planning 입력은 **계획만** — 실 dispatch/PR/merge/branch 실행 0. DISPATCH_LIFECYCLE_EFFECT: Final[str] = "none" PLAN_OK: Final[str] = "PLAN_INPUT_OK" PLAN_FAIL_CLOSED: Final[str] = "PLAN_FAIL_CLOSED" PLAN_UNAVAILABLE: Final[str] = "PLAN_DECISION_UNAVAILABLE" class ProfileDispatchPlanningAdapterError(ValueError): """resolver decision 소비 실패. ``self.code`` = 사유 (fail-closed).""" def __init__(self, code: str, message: str) -> None: super().__init__(f"[{code}] {message}") self.code = code self.message = message def load_resolver_decision( source: "str | Path | Mapping[str, Any]", ) -> Dict[str, Any]: """default resolver decision(`decision.v1`)을 read-only 소비. source = 파일 경로 또는 이미 로드된 dict (resolver 모듈 import·호출 0 — 파일레벨 contract 만). schema marker mismatch / 부재 / 파싱 실패 = fail-closed (ProfileDispatchPlanningAdapterError, code 보존). """ if isinstance(source, Mapping): obj: Any = dict(source) else: p = Path(source) try: raw = p.read_text(encoding="utf-8") except OSError as e: raise ProfileDispatchPlanningAdapterError( "decision_source_unreadable", f"resolver decision 부재/읽기 실패 {p}: {e}", ) from e try: obj = json.loads(raw) except json.JSONDecodeError as e: raise ProfileDispatchPlanningAdapterError( "decision_source_unparsable", f"resolver decision JSON 파싱 실패 {p}: {e}", ) from e if not isinstance(obj, dict): raise ProfileDispatchPlanningAdapterError( "decision_not_object", f"resolver decision 최상위 object 아님: {type(obj).__name__}", ) if obj.get("schema") != ACCEPTED_DECISION_SCHEMA: raise ProfileDispatchPlanningAdapterError( "decision_schema_mismatch", f"resolver decision schema mismatch: expected " f"{ACCEPTED_DECISION_SCHEMA!r}, got {obj.get('schema')!r} " f"(resolver 부재/버전 mismatch — fail-closed)", ) return obj @dataclass class ProfileDispatchPlanningAdapter: """resolver decision → dispatch planning 평면 입력 (read-only derive). plan_only / DISPATCH_LIFECYCLE_EFFECT 는 입력과 무관하게 hard-pinned — 어떤 경로에서도 실 dispatch/PR/merge/branch 자동확정 0.""" decision: Dict[str, Any] def _g(self, k: str, default: Any = "") -> Any: return self.decision.get(k, default) def status(self) -> str: return str(self._g("status", "")) def _be(self) -> Dict[str, Any]: be = self.decision.get("boundary_expansion") or {} return dict(be) if isinstance(be, Mapping) else {} def _list(self, key: str) -> List[str]: return [str(x) for x in (self._be().get(key) or [])] def to_dispatch_planning_input(self) -> Dict[str, Any]: """기본 dispatch planning 경로가 그대로 소비할 평면 입력. resolver RESOLVED 일 때만 plan_admissible=True. REFUSED/CONFLICT/ HOLD = fail-closed plan_admissible=False (planner 자동 진행 차단). 실 dispatch/PR/merge/branch = plan_only hard-pinned (자동확정 0).""" st = self.status() admissible = st == _RESOLVED return { "adapter_schema": ADAPTER_SCHEMA, "adapter_module": ADAPTER_MODULE, "adapter_version": ADAPTER_VERSION, "consumed_decision_schema": ACCEPTED_DECISION_SCHEMA, "signal": PLAN_OK if admissible else PLAN_FAIL_CLOSED, "plan_admissible": admissible, "default_path": bool(self._g("default_path", False)), "goal_id": str(self._g("goal_id", "")), "goal_type": str(self._g("goal_type", "")), "resolved_profile_name": self._g("resolved_profile_name", None), "profile_id": str(self._g("profile_id", "")), "resolver_status": st, "gate_condition_names": self._list("gate_condition_names"), "gate_semantics": "AND — ALL conditions must hold for PASS", "hold_trigger_conditions": self._list("hold_trigger_conditions"), "allowed_actions": self._list("allowed_actions"), "forbidden_actions": self._list("forbidden_actions"), "explicit_boundary": self._list("explicit_boundary"), "completion_packet_meta_ref": self._be().get("completion_packet_meta_ref"), "evidence_meta_ref": self._be().get("evidence_meta_ref"), "refusal_code": self._g("refusal_code", None), "refusal_reason": self._g("refusal_reason", None), # ── 실 dispatch/PR/merge/branch 자동확정 0 (hard-pinned §2/§6) ── "dispatch_lifecycle_effect": DISPATCH_LIFECYCLE_EFFECT, "plan_only": True, "write_authority": False, "merge_authority": False, "pr_branch_authority": False, "auto_dispatch": False, "plan_only_invariant": ( "planning 입력 derive 만; 실 dispatch/PR/merge/branch 자동확정 0 " "(hard-pinned, 모든 입력에서 불변)." ), } def adapt_for_dispatch_planning( source: "str | Path | Mapping[str, Any]", ) -> Dict[str, Any]: """default resolver → dispatch planning 입력 단일 callable entrypoint. resolver 부재/mismatch 는 fail-closed safe 입력으로 변환 (예외를 planner 가 진행 신호로 오인하지 않도록 PLAN_DECISION_UNAVAILABLE).""" try: decision = load_resolver_decision(source) except ProfileDispatchPlanningAdapterError as e: return { "adapter_schema": ADAPTER_SCHEMA, "adapter_module": ADAPTER_MODULE, "adapter_version": ADAPTER_VERSION, "consumed_decision_schema": ACCEPTED_DECISION_SCHEMA, "signal": PLAN_UNAVAILABLE, "plan_admissible": False, "default_path": False, "resolver_status": "UNAVAILABLE", "refusal_code": e.code, "refusal_reason": e.message, "dispatch_lifecycle_effect": DISPATCH_LIFECYCLE_EFFECT, "plan_only": True, "write_authority": False, "merge_authority": False, "pr_branch_authority": False, "auto_dispatch": False, "fail_closed": True, "plan_only_invariant": ( "fail-closed 경로에서도 plan_only / 자동확정 0 불변." ), } return ProfileDispatchPlanningAdapter(decision).to_dispatch_planning_input() __all__ = [ "ADAPTER_MODULE", "ADAPTER_VERSION", "ADAPTER_SCHEMA", "ACCEPTED_DECISION_SCHEMA", "DISPATCH_LIFECYCLE_EFFECT", "PLAN_OK", "PLAN_FAIL_CLOSED", "PLAN_UNAVAILABLE", "ProfileDispatchPlanningAdapterError", "ProfileDispatchPlanningAdapter", "load_resolver_decision", "adapt_for_dispatch_planning", ]