"""anu_v3.default_profile_resolver — TRACK 3: policy profile engine → DEFAULT dispatch-planning 경로 연결 (task-2553+52). 회장 3-track 배치 Track 3. goal_type `policy_profile_default_dispatch_adoption`. 목표 (회장 verbatim §1): policy profile engine 을 **기본 dispatch planning 경로** 에 연결한다. 기본 경로에서 **goal_type + boundary 만으로 profile 선택** 가능해야 한다. 기존 +38 `dispatch_profile_selection` 과의 차이: +38 seam 은 chair 가 ``policy_profile.name`` 을 **명시 공급** 해야 했다. 본 Track 은 그 한 칸을 없앤다 — chair 는 default 경로에서 ``goal_type`` + ``boundary`` 만 준다. 본 resolver 가 ``goal_type → policy_profile`` mapping(`memory/policy_profiles/goal_type_profile_mapping.json`)을 단일 결정 테이블로 조회해 profile 이름을 산출하고, C1 engine 정본 API (parse_goal_request → resolve_policy)를 **read-only 소비** 하여 gate / HOLD / allowed / forbidden / evidence / completion-packet 을 자동 산출한다 (§8 회장이 풀어쓰지 않아도 ANU 가 engine+guard 로 자동산출). 본 모듈 한정 책임 (순수, 부작용 0 — 파일 write 0, network 0, git 0, GitHub API 0, 실 dispatch/PR/merge/branch 실행 0): 1. default profile resolver (§3.1) 2. goal_type → policy_profile mapping 로더·조회 (§3.2) 3. boundary → gate/HOLD/forbidden expansion 표면화 (§3.3 — engine 산출 read-only 표면화) 6. missing / unknown profile fail-closed (§3.6 — 추측·날조 0) 7. profile conflict fail-closed (§3.7) 8. selected profile evidence JSON 산출 (§3.8) 설계 invariant (§5 / §6 / §10): - 신규 별도 모듈 (strict-additive). engine·+38·+39·frozen anchor·기존 profile/mapping/anchor import-only read-only / mutation 0. - engine = `anu_v3.policy_profile_engine` 정본 API 만 호출 (parse_goal_request@187 → resolve_policy@561, PolicyEngineError, validate_against_meta, load_meta_schema). engine 내부 재구현·mutation 0. - resolver 는 profile 을 **선택만** 한다 — 실행하지 않는다. ``DISPATCH_LIFECYCLE_EFFECT == "none"``. 실 운영 in-place 채택은 별도. - missing / unknown / conflict = **fail-closed** (자동 적용 0, profile 미바인딩, 예외 dispatch 비전파). 추측·날조 0. - profile engine 이 write/merge 권한을 요구하면 본 모듈은 그것을 절대 수행하지 않고 fail-closed (§7 공통 HOLD 대상 — 회장 보고). - 외부 의존성 0 (engine 과 동일 — offline 100%). """ from __future__ import annotations from dataclasses import dataclass, field from pathlib import Path from typing import Any, Final, Mapping, Sequence from anu_v3.policy_profile_engine import ( PROFILE_JSON_DIR_DEFAULT, PROFILE_SCHEMA_DIR_DEFAULT, SCHEMA_DIR_DEFAULT, PolicyEngineError, PolicyResolution, load_meta_schema, parse_goal_request, resolve_policy, validate_against_meta, ) RESOLVER_MODULE: Final[str] = "anu_v3.default_profile_resolver" RESOLVER_VERSION: Final[str] = "task-2553+52.Track3.v1" RESOLUTION_SCHEMA_ID: Final[str] = "anu_v3.default_profile_resolver.decision.v1" EVIDENCE_SCHEMA_ID: Final[str] = "anu_v3.default_profile_resolver.selected_profile_evidence.v1" # DEFAULT dispatch-planning mapping 위치 (read-only, mutation 0). MAPPING_PATH_DEFAULT: Final[str] = "memory/policy_profiles/goal_type_profile_mapping.json" MAPPING_SCHEMA_NAME: Final[str] = "goal_type_profile_mapping.schema.json" # resolver 는 profile 을 **선택만** 한다 — 실 dispatch/PR/merge/branch 실행 0. DISPATCH_LIFECYCLE_EFFECT: Final[str] = "none" # fail-closed 4-상태 (회장 §3.6/§3.7 — 추측·날조 0). STATUS_RESOLVED: Final[str] = "RESOLVED" STATUS_REFUSED: Final[str] = "DEFAULT_RESOLUTION_REFUSED" STATUS_CONFLICT: Final[str] = "PROFILE_CONFLICT" STATUS_HOLD: Final[str] = "HOLD_FOR_CHAIR" # engine 진입 이전 정규화/조회 단계 fail-closed 코드. REFUSAL_REQUEST_NOT_MAPPING: Final[str] = "default_request_not_mapping" REFUSAL_GOAL_TYPE_MISSING: Final[str] = "goal_type_missing" REFUSAL_GOAL_TYPE_NOT_MAPPED: Final[str] = "goal_type_not_mapped" REFUSAL_MAPPING_LOAD_FAIL: Final[str] = "mapping_load_fail" REFUSAL_MAPPING_SCHEMA_FAIL: Final[str] = "mapping_schema_fail" REFUSAL_MAPPING_ENTRY_INVALID: Final[str] = "mapping_entry_invalid" CONFLICT_EXPLICIT_PROFILE: Final[str] = "explicit_profile_conflict" CONFLICT_AMBIGUOUS_MAPPING: Final[str] = "ambiguous_mapping_conflict" CONFLICT_ENGINE_ALLOW_FORBID: Final[str] = "engine_allowed_forbidden_conflict" REFUSAL_ENGINE_HOLD: Final[str] = "engine_hold_for_chair" class DefaultProfileResolverError(ValueError): """resolver 정규화/조회 실패. dispatch lifecycle 로 전파하지 않고 내부에서 fail-closed 결정(REFUSED/CONFLICT)으로 변환된다.""" def __init__(self, code: str, message: str) -> None: super().__init__(f"[{code}] {message}") self.code = code self.message = message # --------------------------------------------------------------------------- # 1. DEFAULT dispatch request — chair 는 goal_type + boundary 만 준다 # --------------------------------------------------------------------------- @dataclass class DefaultDispatchRequest: """기본 dispatch planning 경로 입력. policy_profile.name 은 chair 가 주지 않는다 — resolver 가 goal_type 으로 산출한다. ``explicit_policy_profile_name`` 은 선택적 — chair 가 굳이 명시했는데 mapping 산출과 다르면 **profile conflict fail-closed** (§3.7) 검출용. """ goal_id: str goal_statement: str goal_type: str boundary: list[str] = field(default_factory=list) explicit_policy_profile_name: str | None = None def to_goal_request(self, profile_name: str) -> dict[str, Any]: """resolver 가 산출한 profile_name 을 채워 C1 engine 정본 입력 (goal_request_2553plus33.schema.json) 형태로 변환.""" return { "goal_id": self.goal_id, "goal_statement": self.goal_statement, "goal_type": self.goal_type, "boundary": list(self.boundary), "policy_profile": {"name": profile_name}, } def build_default_request(raw: Any) -> DefaultDispatchRequest: """dispatch 가 넘긴 raw dict → DefaultDispatchRequest (fail-closed). 형태 불량은 DefaultProfileResolverError 로 raise 하되 entrypoint 가 REFUSED 안전 거부로 흡수한다 (dispatch 무파괴). """ if not isinstance(raw, Mapping): raise DefaultProfileResolverError( REFUSAL_REQUEST_NOT_MAPPING, f"default request dict 아님: {type(raw).__name__}", ) gt = raw.get("goal_type") if not gt or not isinstance(gt, str) or not gt.strip(): raise DefaultProfileResolverError( REFUSAL_GOAL_TYPE_MISSING, "default 경로 필수 입력 goal_type 누락/무효 (추측 0 — fail-closed)", ) pp = raw.get("policy_profile") or {} explicit = pp.get("name") if isinstance(pp, Mapping) else None return DefaultDispatchRequest( goal_id=str(raw.get("goal_id", "")), goal_statement=str(raw.get("goal_statement", "")), goal_type=gt.strip(), boundary=[str(b) for b in (raw.get("boundary") or [])], explicit_policy_profile_name=( explicit.strip() if isinstance(explicit, str) and explicit.strip() else None ), ) # --------------------------------------------------------------------------- # 2. goal_type → policy_profile mapping 로더·조회 (단일 결정 테이블) # --------------------------------------------------------------------------- def load_goal_type_mapping( *, mapping_path: str | Path = MAPPING_PATH_DEFAULT, schema_dir: str | Path = SCHEMA_DIR_DEFAULT, ) -> dict[str, Any]: """goal_type → policy_profile mapping 을 read-only 로딩·검증. engine 의 offline draft-07 subset validator(validate_against_meta)로 `goal_type_profile_mapping.schema.json` 검증. 파일 부재/파싱 실패/ schema 불일치 = fail-closed (DefaultProfileResolverError, code 보존). mutation 0 (mapping 파일 read-only). """ import json p = Path(mapping_path) try: raw = p.read_text(encoding="utf-8") except OSError as e: raise DefaultProfileResolverError( REFUSAL_MAPPING_LOAD_FAIL, f"mapping 부재/읽기 실패 {p}: {e}" ) from e try: obj = json.loads(raw) except json.JSONDecodeError as e: raise DefaultProfileResolverError( REFUSAL_MAPPING_LOAD_FAIL, f"mapping JSON 파싱 실패 {p}: {e}" ) from e if not isinstance(obj, dict): raise DefaultProfileResolverError( REFUSAL_MAPPING_LOAD_FAIL, f"mapping 최상위 dict 아님 {p}" ) try: meta = load_meta_schema(MAPPING_SCHEMA_NAME, schema_dir=schema_dir) except PolicyEngineError as e: raise DefaultProfileResolverError( REFUSAL_MAPPING_SCHEMA_FAIL, f"mapping meta-schema 로드 실패: {e}" ) from e errs = validate_against_meta(obj, meta) if errs: raise DefaultProfileResolverError( REFUSAL_MAPPING_SCHEMA_FAIL, "; ".join(errs) ) if not isinstance(obj.get("mappings"), dict) or not obj["mappings"]: raise DefaultProfileResolverError( REFUSAL_MAPPING_SCHEMA_FAIL, "mapping.mappings 비어있음/dict 아님" ) return obj def resolve_profile_name(mapping: Mapping[str, Any], goal_type: str) -> str: """단일 결정 테이블 조회. goal_type → policy_profile 이름 (deterministic). - goal_type 미등재 → REFUSAL_GOAL_TYPE_NOT_MAPPED (unknown, 추측 0 fail-closed §3.6) - entry 형태 불량 → REFUSAL_MAPPING_ENTRY_INVALID - entry 가 다중 후보(list) → CONFLICT_AMBIGUOUS_MAPPING (§3.7) """ entries = mapping.get("mappings") or {} if goal_type not in entries: raise DefaultProfileResolverError( REFUSAL_GOAL_TYPE_NOT_MAPPED, f"goal_type {goal_type!r} 가 default mapping 에 없음 — UNKNOWN, " f"추측·날조 0 (fail-closed §3.6)", ) entry = entries[goal_type] if isinstance(entry, (list, tuple)): raise DefaultProfileResolverError( CONFLICT_AMBIGUOUS_MAPPING, f"goal_type {goal_type!r} 다중 후보(비결정) — profile conflict " f"fail-closed (§3.7)", ) if not isinstance(entry, Mapping): raise DefaultProfileResolverError( REFUSAL_MAPPING_ENTRY_INVALID, f"goal_type {goal_type!r} entry 가 object 아님: {type(entry).__name__}", ) name = entry.get("policy_profile") if not name or not isinstance(name, str) or not name.strip(): raise DefaultProfileResolverError( REFUSAL_MAPPING_ENTRY_INVALID, f"goal_type {goal_type!r} entry 에 policy_profile(string) 누락/무효", ) return name.strip() # --------------------------------------------------------------------------- # 3 / 6 / 7 / 8. DefaultProfileResolution + decision/evidence/adapter # --------------------------------------------------------------------------- @dataclass class DefaultProfileResolution: """default 경로 산출 — dispatch planning / batch coordinator 가 그대로 소비할 profile resolution binding (fail-closed 4-상태).""" status: str profile_bound: bool auto_apply: bool goal_id: str goal_type: str resolved_profile_name: str | None profile_id: str profile_version: str mapping_source: str | None boundary_expansion: dict[str, Any] refusal_code: str | None = None refusal_reason: str | None = None dispatch_lifecycle_effect: str = DISPATCH_LIFECYCLE_EFFECT default_path: bool = True resolver: str = RESOLVER_MODULE resolver_version: str = RESOLVER_VERSION engine_decision: dict[str, Any] | None = None def to_decision_dict(self) -> dict[str, Any]: return { "schema": RESOLUTION_SCHEMA_ID, "resolver": self.resolver, "resolver_version": self.resolver_version, "status": self.status, "profile_bound": self.profile_bound, "auto_apply": self.auto_apply, "default_path": self.default_path, "dispatch_lifecycle_effect": self.dispatch_lifecycle_effect, "goal_id": self.goal_id, "goal_type": self.goal_type, "resolved_profile_name": self.resolved_profile_name, "profile_id": self.profile_id, "profile_version": self.profile_version, "mapping_source": self.mapping_source, "boundary_expansion": self.boundary_expansion, "refusal_code": self.refusal_code, "refusal_reason": self.refusal_reason, "engine_decision": self.engine_decision, } def to_selected_profile_evidence(self) -> dict[str, Any]: """§3.8 — selected profile evidence JSON. goal_type + boundary → profile 선택이 default 경로에서 engine read-only 소비로 이뤄졌음을 입증하는 평면 증거.""" return { "schema": EVIDENCE_SCHEMA_ID, "resolver": self.resolver, "resolver_version": self.resolver_version, "default_path": self.default_path, "selection_input": { "goal_type": self.goal_type, "boundary": list(self.boundary_expansion.get("explicit_boundary", [])), "chair_supplied_policy_profile_name": False, }, "selected": { "status": self.status, "resolved_profile_name": self.resolved_profile_name, "profile_id": self.profile_id, "profile_version": self.profile_version, "profile_bound": self.profile_bound, "auto_apply": self.auto_apply, }, "mapping_source": self.mapping_source, "engine_consumed_read_only": ( "anu_v3.policy_profile_engine.parse_goal_request -> resolve_policy " "(정본 API, engine import-only, mutation 0, byte-0)" ), "boundary_expansion": self.boundary_expansion, "fail_closed": self.status != STATUS_RESOLVED, "refusal_code": self.refusal_code, "refusal_reason": self.refusal_reason, } def planning_binding(self) -> dict[str, Any]: """dispatch planning adapter 소비용 평면 binding (RESOLVED 시에만 의미; 그 외 빈 contract — 자동 적용 0).""" be = self.boundary_expansion return { "status": self.status, "profile_bound": self.profile_bound, "auto_apply": self.auto_apply, "default_path": self.default_path, "goal_id": self.goal_id, "goal_type": self.goal_type, "resolved_profile_name": self.resolved_profile_name, "profile_id": self.profile_id, "gate_condition_names": list(be.get("gate_condition_names", [])), "hold_trigger_conditions": list(be.get("hold_trigger_conditions", [])), "allowed_actions": list(be.get("allowed_actions", [])), "forbidden_actions": list(be.get("forbidden_actions", [])), "completion_packet_meta_ref": be.get("completion_packet_meta_ref"), "evidence_meta_ref": be.get("evidence_meta_ref"), "dispatch_lifecycle_effect": self.dispatch_lifecycle_effect, } def _refused( code: str, reason: str, *, status: str = STATUS_REFUSED, goal_id: str = "", goal_type: str = "", resolved_profile_name: str | None = None, engine_decision: dict[str, Any] | None = None, ) -> DefaultProfileResolution: """fail-closed 안전 거부 — profile 미바인딩·자동 적용 0, dispatch 무파괴.""" return DefaultProfileResolution( status=status, profile_bound=False, auto_apply=False, goal_id=goal_id, goal_type=goal_type, resolved_profile_name=resolved_profile_name, profile_id="", profile_version="", mapping_source=None, boundary_expansion={"explicit_boundary": []}, refusal_code=code, refusal_reason=reason, engine_decision=engine_decision, ) def _boundary_expansion_from_resolution( res: PolicyResolution, request: DefaultDispatchRequest ) -> dict[str, Any]: """§3.3 — boundary → gate/HOLD/forbidden expansion 표면화. engine 이 boundary[] + profile 로 산출한 gate/HOLD/allowed/forbidden 을 그대로 표면화 (engine read-only). resolver 가 재구현하지 않는다. """ binding = res.to_coordinator_binding() return { "explicit_boundary": list(request.boundary), "gate_condition_names": list(binding["gate_condition_names"]), "hold_trigger_conditions": list(binding["hold_trigger_conditions"]), "allowed_actions": list(binding["allowed_actions"]), "forbidden_actions": list(binding["forbidden_actions"]), "completion_packet_meta_ref": binding["completion_packet_meta_ref"], "evidence_meta_ref": binding["evidence_meta_ref"], "resolved_boundary": dict(res.boundary), } def resolve_default_profile( request: DefaultDispatchRequest, *, mapping_path: str | Path = MAPPING_PATH_DEFAULT, profile_json_dir: str | Path = PROFILE_JSON_DIR_DEFAULT, profile_schema_dir: str | Path = PROFILE_SCHEMA_DIR_DEFAULT, schema_dir: str | Path = SCHEMA_DIR_DEFAULT, ) -> DefaultProfileResolution: """**DEFAULT dispatch-planning resolver entrypoint** (§3.1). goal_type + boundary 만으로 profile 을 자동 선택·로딩한다. 1. goal_type → policy_profile mapping 조회 (§3.2, 단일 결정 테이블) 2. explicit profile 명시 충돌 검출 → PROFILE_CONFLICT (§3.7) 3. C1 engine 정본 API read-only 소비: parse_goal_request → resolve_policy (engine byte-0, mutation 0) 4. boundary → gate/HOLD/forbidden expansion 표면화 (§3.3) 5. missing/unknown/conflict = fail-closed (§3.6/§3.7 — 추측·날조 0) profile 부재 / schema mismatch / engine HOLD 는 예외를 dispatch 로 전파하지 않고 fail-closed 결정으로 변환 (dispatch lifecycle 무파괴). """ # (a) mapping 로드 + goal_type 조회 (engine 진입 이전 fail-closed). try: mapping = load_goal_type_mapping( mapping_path=mapping_path, schema_dir=schema_dir ) profile_name = resolve_profile_name(mapping, request.goal_type) except DefaultProfileResolverError as e: status = STATUS_CONFLICT if e.code == CONFLICT_AMBIGUOUS_MAPPING else STATUS_REFUSED return _refused( e.code, e.message, status=status, goal_id=request.goal_id, goal_type=request.goal_type, ) mapping_source = str(mapping_path) # (b) chair 가 굳이 명시한 profile 이 mapping 산출과 다르면 conflict (§3.7). if ( request.explicit_policy_profile_name and request.explicit_policy_profile_name != profile_name ): return _refused( CONFLICT_EXPLICIT_PROFILE, f"chair 명시 profile {request.explicit_policy_profile_name!r} ≠ " f"default mapping 산출 {profile_name!r} (goal_type " f"{request.goal_type!r}) — profile conflict fail-closed (§3.7)", status=STATUS_CONFLICT, goal_id=request.goal_id, goal_type=request.goal_type, resolved_profile_name=profile_name, ) # (c) C1 engine 정본 API read-only 소비. goal_request = request.to_goal_request(profile_name) try: parse_goal_request(goal_request, schema_dir=schema_dir) res = resolve_policy( goal_request, profile_json_dir=profile_json_dir, profile_schema_dir=profile_schema_dir, schema_dir=schema_dir, ) except PolicyEngineError as e: # profile 부재(profile_load_fail)·schema mismatch 등 → fail-closed. return _refused( e.code, e.message, goal_id=request.goal_id, goal_type=request.goal_type, resolved_profile_name=profile_name, ) be = _boundary_expansion_from_resolution(res, request) decision = res.to_decision_dict() if res.status == "HOLD_FOR_CHAIR": # engine HOLD: allowed∩forbidden 충돌이면 PROFILE_CONFLICT, 그 외 # HOLD_FOR_CHAIR — 어느 쪽도 자동 적용 0 (fail-closed §3.7). reason = res.hold_reason or "engine HOLD_FOR_CHAIR" is_conflict = "충돌" in reason or "모순" in reason return DefaultProfileResolution( status=STATUS_CONFLICT if is_conflict else STATUS_HOLD, profile_bound=False, auto_apply=False, goal_id=res.goal_id, goal_type=res.goal_type, resolved_profile_name=profile_name, profile_id=res.profile_id, profile_version=res.profile_version, mapping_source=mapping_source, boundary_expansion=be, refusal_code=( CONFLICT_ENGINE_ALLOW_FORBID if is_conflict else REFUSAL_ENGINE_HOLD ), refusal_reason=reason, engine_decision=decision, ) return DefaultProfileResolution( status=STATUS_RESOLVED, profile_bound=True, auto_apply=True, goal_id=res.goal_id, goal_type=res.goal_type, resolved_profile_name=profile_name, profile_id=res.profile_id, profile_version=res.profile_version, mapping_source=mapping_source, boundary_expansion=be, engine_decision=decision, ) def run_default_profile_resolution( raw_request: Any, *, mapping_path: str | Path = MAPPING_PATH_DEFAULT, profile_json_dir: str | Path = PROFILE_JSON_DIR_DEFAULT, profile_schema_dir: str | Path = PROFILE_SCHEMA_DIR_DEFAULT, schema_dir: str | Path = SCHEMA_DIR_DEFAULT, ) -> dict[str, Any]: """파일레벨 contract entrypoint — raw dict → decision dict. 기본 dispatch planning 경로가 단일 호출로 소비할 결선점. 요청 정규화 실패도 fail-closed 안전 거부로 흡수한다 (예외 dispatch 비전파). """ try: req = build_default_request(raw_request) except DefaultProfileResolverError as e: return _refused(e.code, e.message).to_decision_dict() return resolve_default_profile( req, mapping_path=mapping_path, profile_json_dir=profile_json_dir, profile_schema_dir=profile_schema_dir, schema_dir=schema_dir, ).to_decision_dict() def run_selected_profile_evidence( raw_request: Any, *, mapping_path: str | Path = MAPPING_PATH_DEFAULT, profile_json_dir: str | Path = PROFILE_JSON_DIR_DEFAULT, profile_schema_dir: str | Path = PROFILE_SCHEMA_DIR_DEFAULT, schema_dir: str | Path = SCHEMA_DIR_DEFAULT, ) -> dict[str, Any]: """§3.8 — raw dict → selected profile evidence JSON (예외 비전파).""" try: req = build_default_request(raw_request) except DefaultProfileResolverError as e: return _refused(e.code, e.message).to_selected_profile_evidence() return resolve_default_profile( req, mapping_path=mapping_path, profile_json_dir=profile_json_dir, profile_schema_dir=profile_schema_dir, schema_dir=schema_dir, ).to_selected_profile_evidence() def default_requests_from_fixture(cases: Sequence[Mapping[str, Any]]) -> list[Any]: """fixture case[].request 목록 추출 (regression/dry-run 공용).""" return [c["request"] for c in cases] __all__ = [ "RESOLVER_MODULE", "RESOLVER_VERSION", "RESOLUTION_SCHEMA_ID", "EVIDENCE_SCHEMA_ID", "MAPPING_PATH_DEFAULT", "MAPPING_SCHEMA_NAME", "DISPATCH_LIFECYCLE_EFFECT", "STATUS_RESOLVED", "STATUS_REFUSED", "STATUS_CONFLICT", "STATUS_HOLD", "REFUSAL_REQUEST_NOT_MAPPING", "REFUSAL_GOAL_TYPE_MISSING", "REFUSAL_GOAL_TYPE_NOT_MAPPED", "REFUSAL_MAPPING_LOAD_FAIL", "REFUSAL_MAPPING_SCHEMA_FAIL", "REFUSAL_MAPPING_ENTRY_INVALID", "CONFLICT_EXPLICIT_PROFILE", "CONFLICT_AMBIGUOUS_MAPPING", "CONFLICT_ENGINE_ALLOW_FORBID", "REFUSAL_ENGINE_HOLD", "DefaultProfileResolverError", "DefaultDispatchRequest", "DefaultProfileResolution", "build_default_request", "load_goal_type_mapping", "resolve_profile_name", "resolve_default_profile", "run_default_profile_resolution", "run_selected_profile_evidence", "default_requests_from_fixture", ]