# -*- coding: utf-8 -*- """anu_v3.coordinator_profile_binding — batch coordinator ← profile decision output 소비 seam. task-2553+39 TRACK B (회장 GO 병렬 4트랙 Track B, 9-R.1 우선). 목표 (회장 verbatim §1): batch coordinator 가 policy profile engine 의 decision output (gate / HOLD / allowed / forbidden / completion-packet / evidence schema)을 직접 소비하여 track 판단·통합에 활용한다. 9-R.1 (HIGH 해소, 본문 우선): "직접 소비" 의 결선 산물 = **이 신규 additive 모듈 자체가 coordinator-consumable binding seam(파일레벨 contract)**. +37 entrypoint· +29/+30 별도모듈 선례 동일 패턴 — 신규 binding 모듈이 곧 소비 연결점이며 frozen ``parallel_batch_coordinator.py`` / +29 registry / +30 generic coordinator 의 in-place 편집·import 결합 불요. "직접 소비" 달성 = binding 모듈 존재 + 파일레벨 contract + regression 으로 coordinator → engine output 소비 경로 입증. 기존 coordinator 가 이 binding 을 실 호출하는 in-place adoption 은 별도 운영단계. 설계 invariant (§3 / §5): - 신규 별도 모듈 (additive). engine·+29·+30·frozen coordinator 원본 무수정. **import 결합 0** — anu_v3 import 전무, 순수 stdlib only. - engine decision output 은 **파일레벨 contract** 로만 소비: ``anu_v3.policy_profile_engine.decision.v1`` JSON 형태(파일 or dict)를 read-only 입력으로 받는다 (engine 모듈 import·호출·mutation 0). - coordinator = 판단보조 소비만. closeout / merge **자동확정 0** — binding output 은 어떤 입력에서도 ``closeout_authority`` / ``merge_authority`` / ``auto_confirm`` 를 절대 True 로 내지 않는다 (hard-pinned False, fail-closed). - engine 부재 / schema mismatch → fail-closed 안전 (자동확정 0, coordinator 가 settle 못 하도록 HOLD/UNAVAILABLE 신호). - fallback safety path / callback mandatory rule / runtime checkpoint recovery layer 약화·격상·대체 0 (이 모듈은 그 경로를 만지지 않는다). - 모든 decision-logic 메서드 = derive / propose / read only. 파일 I/O 는 decision 경계 밖의 명시 hard-guarded helper 1개(`emit_binding`)에 한정. - NO-SCHEDULER: 스케줄러/외부 dispatch tooling 호출 0 (decision 경계 밖 I/O 는 hard-guarded helper 1개 한정, 그 외 부작용 전무). """ 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, Optional BINDING_MODULE: Final[str] = "anu_v3.coordinator_profile_binding" BINDING_VERSION: Final[str] = "task-2553+39.B.v1" BINDING_SCHEMA: Final[str] = "anu_v3.coordinator_profile_binding.v1" # 소비 대상 = C1 engine 의 파일레벨 contract (decision output schema marker). # 이 marker 가 아니면 fail-closed (engine 부재/mismatch 안전). ACCEPTED_DECISION_SCHEMA: Final[str] = "anu_v3.policy_profile_engine.decision.v1" # engine RESOLVED status 토큰 (engine 모듈 import 없이 contract 상수로 미러). _DECISION_RESOLVED: Final[str] = "RESOLVED" _DECISION_HOLD: Final[str] = "HOLD_FOR_CHAIR" # coordinator 가 binding 소비 후 취할 수 있는 판단보조 신호 (확정 권한 아님). CONSUME_OK: Final[str] = "CONSUME_OK" CONSUME_HOLD_ENGINE: Final[str] = "ENGINE_DECISION_HOLD" CONSUME_HOLD_RUNTIME: Final[str] = "RUNTIME_HOLD_OBSERVED" CONSUME_UNAVAILABLE: Final[str] = "DECISION_UNAVAILABLE" # durable v1 chair 파일 — byte-0 immutable, 절대 emission 대상 아님 (§5). _FROZEN_DURABLE_V1: Final[str] = "task-2553.parallel-batch-state.json" _DELIVERABLE_SUFFIX: Final[str] = ".coordinator-profile-binding.json" _DELIVERABLE_MARKER: Final[str] = f'"binding_schema": "{BINDING_SCHEMA}"' class CoordinatorProfileBindingError(ValueError): """binding 소비 실패. ``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): """chair durable v1 / git-tracked / unrelated untracked 파일 write 거부. binding authority 는 별도 NEW untracked deliverable 이어야 한다 (§4/§5).""" # --------------------------------------------------------------------------- # 1. profile decision output 로더 (파일레벨 contract, read-only 소비) # --------------------------------------------------------------------------- def load_profile_decision( source: "str | Path | Mapping[str, Any]", ) -> Dict[str, Any]: """C1 engine 의 decision output(``decision.v1``)을 read-only 소비. source = 파일 경로 또는 이미 로드된 dict (engine 모듈 import·호출 0 — 파일레벨 contract 만). schema marker mismatch / 파일 부재 / 파싱 실패 = fail-closed (CoordinatorProfileBindingError, 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 CoordinatorProfileBindingError( "decision_source_unreadable", f"engine decision output 부재/읽기 실패 {p}: {e}", ) from e try: obj = json.loads(raw) except json.JSONDecodeError as e: raise CoordinatorProfileBindingError( "decision_source_unparsable", f"engine decision output JSON 파싱 실패 {p}: {e}", ) from e if not isinstance(obj, dict): raise CoordinatorProfileBindingError( "decision_not_object", f"decision output 최상위 object 아님: {type(obj).__name__}", ) schema = obj.get("schema") if schema != ACCEPTED_DECISION_SCHEMA: raise CoordinatorProfileBindingError( "decision_schema_mismatch", f"engine decision schema mismatch: expected " f"{ACCEPTED_DECISION_SCHEMA!r}, got {schema!r} " f"(engine 부재/버전 mismatch — fail-closed, 자동확정 0)", ) return obj # --------------------------------------------------------------------------- # 2. coordinator-consumable binding (모든 메서드 = read-only derive, 9-R.1) # --------------------------------------------------------------------------- @dataclass class CoordinatorProfileBinding: """engine decision output 을 batch coordinator 가 track 판단·통합에 그대로 소비할 수 있는 평면 contract 로 변환하는 read-only seam. 어떤 입력에서도 closeout/merge 를 자동 확정하지 않는다 — coordinator 는 판단보조 소비만 (확정 권한은 chair/ANU out-of-band).""" decision: Dict[str, Any] # -- engine decision output 의 의미 보존 read-only 추출 -------------- def goal_id(self) -> str: return str(self.decision.get("goal_id", "")) def goal_type(self) -> str: return str(self.decision.get("goal_type", "")) def profile_id(self) -> str: return str(self.decision.get("profile_id", "")) def engine_status(self) -> str: return str(self.decision.get("status", "")) def gate_conditions(self) -> List[Dict[str, Any]]: out: List[Dict[str, Any]] = [] for g in self.decision.get("gate", []) or []: if isinstance(g, Mapping): out.append({"name": str(g.get("name", "")), "expected": g.get("expected")}) return out def hold_trigger_conditions(self) -> List[str]: return [str(h) for h in (self.decision.get("hold_trigger_conditions", []) or [])] def allowed_actions(self) -> List[str]: return [str(a) for a in (self.decision.get("allowed_actions", []) or [])] def forbidden_actions(self) -> List[str]: return [str(a) for a in (self.decision.get("forbidden_actions", []) or [])] def completion_packet_meta_ref(self) -> Optional[str]: cps = self.decision.get("completion_packet_schema") or {} if isinstance(cps, Mapping): ref = cps.get("meta_schema_ref") return str(ref) if ref is not None else None return None def evidence_meta_ref(self) -> Optional[str]: evs = self.decision.get("evidence_schema") or {} if isinstance(evs, Mapping): ref = evs.get("meta_schema_ref") return str(ref) if ref is not None else None return None # -- HOLD 평가기 (engine HOLD 의미 미러: 정의-시점 enablement vs # 런타임 발생 분리). ANY enabled trigger 가 런타임 관측 => HOLD. -- def evaluate_hold( self, runtime_signals: Optional[Mapping[str, Any]] = None ) -> Dict[str, Any]: triggers = self.hold_trigger_conditions() sig = runtime_signals or {} fired = [c for c in triggers if sig.get(c)] engine_hold = self.engine_status() == _DECISION_HOLD return { "engine_status_hold": engine_hold, "enabled_hold_triggers": triggers, "runtime_fired_triggers": fired, "hold": bool(engine_hold or fired), "semantics": ( "engine HOLD 의미 미러 — enabled trigger 는 정의-시점 " "enablement, ANY trigger 런타임 관측 시 HOLD(action 0). " "binding 은 HOLD 판정을 전파만 하며 자동확정 0." ), } # -- coordinator 가 track 판단에 소비할 평면 view ------------------ 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. coordinator 는 이 view 를 판단보조로만 쓰고 settle/merge 권한은 갖지 않는다.""" return { "goal_id": self.goal_id(), "goal_type": self.goal_type(), "profile_id": self.profile_id(), "engine_status": self.engine_status(), "gate_condition_names": [g["name"] for g in self.gate_conditions()], "gate_semantics": "AND — ALL conditions must hold for PASS", "hold_trigger_conditions": self.hold_trigger_conditions(), "allowed_actions": self.allowed_actions(), "forbidden_actions": self.forbidden_actions(), "completion_packet_meta_ref": self.completion_packet_meta_ref(), "evidence_meta_ref": self.evidence_meta_ref(), # ── 자동확정 0 (hard-pinned, fail-closed §5) ── "coordinator_role": "decision_consumer_only", "closeout_authority": False, "merge_authority": False, "auto_confirm": False, } # -- coordinator 판단보조 신호 (확정 권한 아님) ------------------- def consumption_decision( self, runtime_signals: Optional[Mapping[str, Any]] = None ) -> Dict[str, Any]: """engine decision output 소비 결과를 coordinator 판단보조 신호로 derive. 어떤 경로에서도 auto_confirm/merge/closeout 확정 0. - engine status HOLD_FOR_CHAIR → ENGINE_DECISION_HOLD (안전) - 런타임 HOLD trigger 관측 → RUNTIME_HOLD_OBSERVED (안전) - 그 외 RESOLVED → CONSUME_OK (판단보조 소비 가능, 그래도 closeout/merge 자동확정은 절대 0) - status 미상 → DECISION_UNAVAILABLE (fail-closed) """ hold = self.evaluate_hold(runtime_signals) status = self.engine_status() if hold["engine_status_hold"]: signal = CONSUME_HOLD_ENGINE reason = "engine decision status = HOLD_FOR_CHAIR — coordinator 소비 보류" elif hold["runtime_fired_triggers"]: signal = CONSUME_HOLD_RUNTIME reason = ( f"runtime HOLD trigger 관측: {hold['runtime_fired_triggers']} " f"— coordinator 소비 보류" ) elif status == _DECISION_RESOLVED: signal = CONSUME_OK reason = ( "engine RESOLVED — coordinator 가 gate/allowed/forbidden 을 " "판단보조로 소비 가능 (closeout/merge 자동확정은 여전히 0)" ) else: signal = CONSUME_UNAVAILABLE reason = ( f"engine decision status 미상({status!r}) — fail-closed, " f"coordinator settle 불가" ) return { "signal": signal, "reason": reason, "authority": "judgment_assist_only", "auto_confirm": False, # hard-pinned "closeout_authority": False, # hard-pinned "merge_authority": False, # hard-pinned "hold": hold, } # -- 전체 binding contract (read-only derive) --------------------- def build_binding( self, runtime_signals: Optional[Mapping[str, Any]] = None ) -> Dict[str, Any]: return { "binding_schema": BINDING_SCHEMA, "binding_module": BINDING_MODULE, "binding_version": BINDING_VERSION, "consumed_decision_schema": ACCEPTED_DECISION_SCHEMA, "decision_logic": ( "read-only: derive/propose/read only; zero " "execute/confirm/write/merge/cron/closeout (9-R.1)." ), "source_engine": self.decision.get("engine"), "source_engine_version": self.decision.get("engine_version"), "track_consumption_view": self.track_consumption_view(), "consumption_decision": self.consumption_decision(runtime_signals), "auto_confirm_invariant": ( "coordinator=판단보조 소비만; closeout/merge 자동확정 0 " "(hard-pinned, 모든 입력에서 불변)." ), } # --------------------------------------------------------------------------- # 3. callable entrypoint — coordinator 소비 연결점 (9-R.1 seam) # --------------------------------------------------------------------------- def consume_for_coordinator( source: "str | Path | Mapping[str, Any]", *, runtime_signals: Optional[Mapping[str, Any]] = None, ) -> Dict[str, Any]: """batch coordinator → profile decision output 소비의 단일 callable entrypoint. engine decision output(파일레벨 contract)을 read-only 로 소비해 coordinator-consumable binding contract 를 반환한다. engine 부재/mismatch 는 fail-closed safe binding 으로 변환(예외를 coordinator 가 settle 신호로 오인하지 않도록 DECISION_UNAVAILABLE). """ try: decision = load_profile_decision(source) except CoordinatorProfileBindingError as e: return { "binding_schema": BINDING_SCHEMA, "binding_module": BINDING_MODULE, "binding_version": BINDING_VERSION, "consumed_decision_schema": ACCEPTED_DECISION_SCHEMA, "decision_logic": "read-only fail-closed (engine 부재/mismatch).", "source_engine": None, "source_engine_version": None, "track_consumption_view": { "coordinator_role": "decision_consumer_only", "closeout_authority": False, "merge_authority": False, "auto_confirm": False, "engine_status": "UNAVAILABLE", }, "consumption_decision": { "signal": CONSUME_UNAVAILABLE, "reason": f"[{e.code}] {e.message}", "authority": "judgment_assist_only", "auto_confirm": False, "closeout_authority": False, "merge_authority": False, "hold": {"hold": False, "enabled_hold_triggers": [], "runtime_fired_triggers": [], "engine_status_hold": False}, }, "auto_confirm_invariant": ( "coordinator=판단보조 소비만; closeout/merge 자동확정 0 " "(fail-closed 경로에서도 불변)." ), "fail_closed": True, "error_code": e.code, } return CoordinatorProfileBinding(decision).build_binding(runtime_signals) # --------------------------------------------------------------------------- # fixture 로더 (read/parse/reference only; source mutation 0) # --------------------------------------------------------------------------- @dataclass class BindingFixture: decision: Dict[str, Any] runtime_signals: Dict[str, Any] = field(default_factory=dict) def load_binding_fixture(fixture_path: "str | Path") -> BindingFixture: data = json.loads(Path(fixture_path).read_text(encoding="utf-8")) return BindingFixture( decision=dict(data["decision"]), runtime_signals=dict(data.get("runtime_signals", {})), ) # --------------------------------------------------------------------------- # 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_binding(binding: Mapping[str, Any], out_path: "str | Path") -> Path: """NEW untracked binding authority 파일 write. decision component 아님 — runner/ANU 가 수행하는 명시 I/O (9-R.1 경계 밖). Hard write-guard (§4/§5 — +30 가드 미러, 무관 파일 절대 clobber 0): * chair durable v1 name -> REFUSE (byte-0 immutable) * 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) if out.name == _FROZEN_DURABLE_V1: raise FrozenWriteRefused( "refusing to write chair durable v1 (§5 byte-0); binding " "authority MUST be a separate NEW untracked deliverable" ) 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 seam's own " f"binding deliverable may be written)" ) out.parent.mkdir(parents=True, exist_ok=True) out.write_text( json.dumps(binding, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) return out __all__ = [ "BINDING_MODULE", "BINDING_VERSION", "BINDING_SCHEMA", "ACCEPTED_DECISION_SCHEMA", "CONSUME_OK", "CONSUME_HOLD_ENGINE", "CONSUME_HOLD_RUNTIME", "CONSUME_UNAVAILABLE", "CoordinatorProfileBindingError", "FrozenWriteRefused", "CoordinatorProfileBinding", "BindingFixture", "load_profile_decision", "load_binding_fixture", "consume_for_coordinator", "emit_binding", ]