# -*- coding: utf-8 -*- """anu_v3.profile_coordinator_input_adapter — §3.5: profile decision → batch coordinator 입력 어댑터 (task-2553+52 / Track 3). 목표 (회장 §3.5): default profile resolver(`anu_v3.default_profile_resolver`)의 decision output 을 batch coordinator 가 track 판단·통합 입력으로 그대로 소비할 수 있는 평면 표현으로 변환한다. 설계 invariant (§5 / §6 — +39 coordinator_profile_binding 가드 미러): - 신규 별도 additive 모듈. resolver·engine·+38·+39·frozen `parallel_batch_coordinator.py`·durable v1 무수정. **import 결합 0** — anu_v3 import 전무, 순수 stdlib only (resolver decision = 파일레벨 contract 로만 소비). - coordinator = 판단보조 입력 소비만. closeout / merge **자동확정 0** — 어떤 입력에서도 ``closeout_authority`` / ``merge_authority`` / ``auto_confirm`` 를 절대 True 로 내지 않는다 (hard-pinned False, fail-closed §6 'coordinator가 closeout/merge 자동확정' 금지). - resolver 부재 / mismatch / status != RESOLVED → fail-closed 안전 (coordinator 가 settle 못 하도록 HOLD/UNAVAILABLE 신호). - callback mandatory rule / self-collector guard / fallback safety path 약화·격상·대체 0 (이 모듈은 그 경로를 만지지 않는다). - 모든 메서드 = 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_coordinator_input_adapter" ADAPTER_VERSION: Final[str] = "task-2553+52.Track3.v1" ADAPTER_SCHEMA: Final[str] = "anu_v3.profile_coordinator_input_adapter.v1" ACCEPTED_DECISION_SCHEMA: Final[str] = "anu_v3.default_profile_resolver.decision.v1" _RESOLVED: Final[str] = "RESOLVED" CONSUME_OK: Final[str] = "COORD_INPUT_OK" CONSUME_HOLD: Final[str] = "COORD_RESOLVER_HOLD" CONSUME_CONFLICT: Final[str] = "COORD_RESOLVER_CONFLICT" CONSUME_REFUSED: Final[str] = "COORD_RESOLVER_REFUSED" CONSUME_UNAVAILABLE: Final[str] = "COORD_DECISION_UNAVAILABLE" class ProfileCoordinatorInputAdapterError(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 소비. resolver 모듈 import·호출 0 — 파일레벨 contract 만. schema marker mismatch / 부재 / 파싱 실패 = fail-closed (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 ProfileCoordinatorInputAdapterError( "decision_source_unreadable", f"resolver decision 부재/읽기 실패 {p}: {e}", ) from e try: obj = json.loads(raw) except json.JSONDecodeError as e: raise ProfileCoordinatorInputAdapterError( "decision_source_unparsable", f"resolver decision JSON 파싱 실패 {p}: {e}", ) from e if not isinstance(obj, dict): raise ProfileCoordinatorInputAdapterError( "decision_not_object", f"resolver decision 최상위 object 아님: {type(obj).__name__}", ) if obj.get("schema") != ACCEPTED_DECISION_SCHEMA: raise ProfileCoordinatorInputAdapterError( "decision_schema_mismatch", f"resolver decision schema mismatch: expected " f"{ACCEPTED_DECISION_SCHEMA!r}, got {obj.get('schema')!r} " f"(resolver 부재/버전 mismatch — fail-closed, 자동확정 0)", ) return obj @dataclass class ProfileCoordinatorInputAdapter: """resolver decision → batch coordinator 판단보조 입력 (read-only). 어떤 입력에서도 closeout/merge 를 자동 확정하지 않는다 — coordinator 는 판단보조 소비만 (확정 권한은 chair/ANU out-of-band).""" 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 _signal(self) -> tuple[str, str]: st = self.status() if st == _RESOLVED: return CONSUME_OK, ( "resolver RESOLVED — coordinator 가 track gate/allowed/" "forbidden 을 판단보조 입력으로 소비 가능 (closeout/merge " "자동확정은 여전히 0)" ) if st == "HOLD_FOR_CHAIR": return CONSUME_HOLD, "resolver HOLD_FOR_CHAIR — coordinator 소비 보류" if st == "PROFILE_CONFLICT": return CONSUME_CONFLICT, ( "resolver PROFILE_CONFLICT — fail-closed, coordinator settle 불가" ) if st == "DEFAULT_RESOLUTION_REFUSED": return CONSUME_REFUSED, ( "resolver DEFAULT_RESOLUTION_REFUSED — fail-closed, " "coordinator settle 불가" ) return CONSUME_UNAVAILABLE, ( f"resolver status 미상({st!r}) — fail-closed, coordinator settle 불가" ) def track_consumption_view(self) -> Dict[str, Any]: """batch coordinator 가 track 별 gate/hold/allowed/forbidden/packet/ evidence 를 그대로 소비할 평면 입력. **자동확정 0 invariant**: closeout_authority / merge_authority / auto_confirm 는 입력과 무관하게 hard-pinned False.""" signal, reason = self._signal() return { "adapter_schema": ADAPTER_SCHEMA, "adapter_module": ADAPTER_MODULE, "adapter_version": ADAPTER_VERSION, "consumed_decision_schema": ACCEPTED_DECISION_SCHEMA, "signal": signal, "reason": reason, "consumable": signal == CONSUME_OK, "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": self.status(), "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"), "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), # ── 자동확정 0 (hard-pinned, fail-closed §6) ── "coordinator_role": "decision_consumer_only", "closeout_authority": False, "merge_authority": False, "auto_confirm": False, "auto_confirm_invariant": ( "coordinator=판단보조 입력 소비만; closeout/merge 자동확정 0 " "(hard-pinned, 모든 입력에서 불변)." ), } def adapt_for_coordinator_input( source: "str | Path | Mapping[str, Any]", ) -> Dict[str, Any]: """default resolver → batch coordinator 입력 단일 callable entrypoint. resolver 부재/mismatch 는 fail-closed safe 입력으로 변환 (예외를 coordinator 가 settle 신호로 오인하지 않도록 COORD_DECISION_UNAVAILABLE).""" try: decision = load_resolver_decision(source) except ProfileCoordinatorInputAdapterError as e: return { "adapter_schema": ADAPTER_SCHEMA, "adapter_module": ADAPTER_MODULE, "adapter_version": ADAPTER_VERSION, "consumed_decision_schema": ACCEPTED_DECISION_SCHEMA, "signal": CONSUME_UNAVAILABLE, "reason": f"[{e.code}] {e.message}", "consumable": False, "default_path": False, "resolver_status": "UNAVAILABLE", "coordinator_role": "decision_consumer_only", "closeout_authority": False, "merge_authority": False, "auto_confirm": False, "fail_closed": True, "error_code": e.code, "auto_confirm_invariant": ( "fail-closed 경로에서도 closeout/merge 자동확정 0 불변." ), } return ProfileCoordinatorInputAdapter(decision).track_consumption_view() __all__ = [ "ADAPTER_MODULE", "ADAPTER_VERSION", "ADAPTER_SCHEMA", "ACCEPTED_DECISION_SCHEMA", "CONSUME_OK", "CONSUME_HOLD", "CONSUME_CONFLICT", "CONSUME_REFUSED", "CONSUME_UNAVAILABLE", "ProfileCoordinatorInputAdapterError", "ProfileCoordinatorInputAdapter", "load_resolver_decision", "adapt_for_coordinator_input", ]