# -*- coding: utf-8 -*- """utils.bot_settings_policy_loader — callback policy loader for bot lifecycle. task-2640 Track C (회장 verbatim unfork #5) — bot_settings 정책 위치 + 정합 검증 helper. dev bot 후보 위치를 제안하고 누락 시 안전 기본값을 적용한다. 기본값 (회장 verbatim · spec §4.1): - require_anu_callback_on_finish: True - self_collector_forbidden: True - sendfile_only_forbidden: True - not_registered_forbidden: True - anu_key_single_source: "c119085addb0f8b7" - fail_closed_on_violation: True 후보 위치 (우선순위, dev bot 제안): 1. /home/jay/workspace/config/callback_policy.json (workspace config 영역) 2. /home/jay/.cokacdir/bot_settings.json 의 ``callback_policy`` 키 3. /home/jay/workspace/memory/bot_settings_callback_policy.json (memory 영역) 본 loader 는 위 후보를 순차적으로 시도하여 첫 번째 존재 파일을 채택한다. 모두 부재 시 안전 기본값을 사용 (fail-closed 보장). 본 모듈은 Layer A / NO-CRON: subprocess / cokacdir / merge / cron 0 호출. 순수 데이터 로드 + 정책 검사만 수행. """ from __future__ import annotations import json import os from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional, Sequence POLICY_LOADER_SCHEMA = "utils.bot_settings_policy_loader.v1" # spec §4.1 verbatim defaults — safe fallback values (fail-closed). DEFAULT_POLICY: Dict[str, Any] = { "require_anu_callback_on_finish": True, "self_collector_forbidden": True, "sendfile_only_forbidden": True, "not_registered_forbidden": True, "anu_key_single_source": "c119085addb0f8b7", "fail_closed_on_violation": True, } # Verdict classifications. BOT_FINALIZE_POLICY_VIOLATION = "BOT_FINALIZE_POLICY_VIOLATION" SELF_COLLECTOR_FORBIDDEN_FINALIZE = "SELF_COLLECTOR_FORBIDDEN_FINALIZE" SENDFILE_ONLY_FORBIDDEN_FINALIZE = "SENDFILE_ONLY_FORBIDDEN_FINALIZE" NOT_REGISTERED_FORBIDDEN_FINALIZE = "NOT_REGISTERED_FORBIDDEN_FINALIZE" ANU_KEY_MISMATCH_FINALIZE = "ANU_KEY_MISMATCH_FINALIZE" PASS = "PASS" FAIL = "FAIL" # Candidate settings file paths (dev bot 후보 제안). def _candidate_paths(workspace_root: Optional[Path] = None) -> List[Path]: ws = workspace_root or Path( os.environ.get("WORKSPACE_ROOT", str(Path.home() / "workspace")) ) home = Path(os.environ.get("HOME", str(Path.home()))) return [ ws / "config" / "callback_policy.json", home / ".cokacdir" / "bot_settings.json", ws / "memory" / "bot_settings_callback_policy.json", ] @dataclass class PolicyLoadResult: schema: str source: str # path or "default-safe" policy: Dict[str, Any] fell_back_to_default: bool reasons: List[str] = field(default_factory=list) def to_json(self) -> dict: return { "schema": self.schema, "source": self.source, "policy": dict(self.policy), "fell_back_to_default": self.fell_back_to_default, "reasons": list(self.reasons), } def load_callback_policy( *, workspace_root: Optional[Path] = None, candidate_paths: Optional[Sequence[Path]] = None, ) -> PolicyLoadResult: """callback policy 를 후보 경로에서 순차 로드한다. 부재 시 ``DEFAULT_POLICY`` 를 안전 기본값으로 적용한다 (fail-closed — require_anu_callback_on_finish=True 등 강한 정책이 기본). Returns: PolicyLoadResult — 채택 source 경로/`"default-safe"`, 최종 policy dict, fall-back 여부, 사유. """ paths = ( list(candidate_paths) if candidate_paths is not None else _candidate_paths(workspace_root) ) reasons: List[str] = [] for p in paths: try: if not p.exists(): reasons.append(f"candidate not present: {p}") continue raw = p.read_text(encoding="utf-8") obj = json.loads(raw) except (OSError, json.JSONDecodeError) as e: reasons.append(f"candidate {p} unreadable / invalid JSON: {e}") continue # accept either top-level dict or {"callback_policy": {..}} if isinstance(obj, dict) and "callback_policy" in obj and isinstance( obj["callback_policy"], dict ): extracted = obj["callback_policy"] elif isinstance(obj, dict): extracted = obj else: reasons.append( f"candidate {p} has non-dict root (skipped)" ) continue # merge over defaults so any missing key still applies safe default. merged: Dict[str, Any] = dict(DEFAULT_POLICY) merged.update( {k: v for k, v in extracted.items() if k in DEFAULT_POLICY} ) reasons.append( f"policy loaded from {p} (overlay over DEFAULT_POLICY; missing " "keys filled with safe defaults)." ) return PolicyLoadResult( schema=POLICY_LOADER_SCHEMA, source=str(p), policy=merged, fell_back_to_default=False, reasons=reasons, ) reasons.append( "no candidate present — using DEFAULT_POLICY (safe defaults: " "require_anu_callback_on_finish=True, fail_closed_on_violation=True)." ) return PolicyLoadResult( schema=POLICY_LOADER_SCHEMA, source="default-safe", policy=dict(DEFAULT_POLICY), fell_back_to_default=True, reasons=reasons, ) @dataclass class PolicyCheckResult: schema: str verdict: str # PASS | FAIL classifications: List[str] policy: Dict[str, Any] finalize_state: Dict[str, Any] reasons: List[str] = field(default_factory=list) @property def ok(self) -> bool: return self.verdict == PASS @property def primary_classification(self) -> Optional[str]: return self.classifications[0] if self.classifications else None def to_json(self) -> dict: return { "schema": self.schema, "verdict": self.verdict, "classifications": list(self.classifications), "primary_classification": self.primary_classification, "policy": dict(self.policy), "finalize_state": dict(self.finalize_state), "reasons": list(self.reasons), } def check_finalize_policy( finalize_state: Dict[str, Any], *, policy: Optional[Dict[str, Any]] = None, ) -> PolicyCheckResult: """봇 lifecycle finalize 진입점에서 정책 필드 검사 (fail-closed). ``finalize_state`` 는 봇이 finalize 직전에 수집한 callback 상태: - callback_registered: bool — ANU normal callback cron 등록 여부 - schedule_id: Optional[str] — 등록된 cron schedule id (non-null 필요) - sendfile_only: bool — sendfile 만 수행하고 cron 0 (NOT_REGISTERED 변종) - self_collector_attempted: bool — executor self-key 로 등록 시도 - anu_key: str — 등록에 사용된 collector key - executor_key: str — executor self key (anu_key 와 달라야 함) Returns: PolicyCheckResult — verdict + classifications + reasons. """ pol = dict(policy) if policy is not None else dict(DEFAULT_POLICY) cls: List[str] = [] reasons: List[str] = [] callback_registered = bool(finalize_state.get("callback_registered")) schedule_id = finalize_state.get("schedule_id") sendfile_only = bool(finalize_state.get("sendfile_only")) self_collector_attempted = bool( finalize_state.get("self_collector_attempted") ) anu_key_used = finalize_state.get("anu_key") or "" executor_key = finalize_state.get("executor_key") or "" if pol.get("require_anu_callback_on_finish", True): if not callback_registered or not schedule_id: cls.append(BOT_FINALIZE_POLICY_VIOLATION) reasons.append( "require_anu_callback_on_finish=True but callback NOT " "registered (or schedule_id null) — finalize fail-closed." ) if pol.get("self_collector_forbidden", True) and self_collector_attempted: cls.append(SELF_COLLECTOR_FORBIDDEN_FINALIZE) reasons.append( "self_collector_forbidden=True but executor self-key collector " "attempted — SELF_COLLECTOR 변종 차단 (회장 §2/§10)." ) if pol.get("self_collector_forbidden", True) and executor_key and anu_key_used and executor_key == anu_key_used: cls.append(SELF_COLLECTOR_FORBIDDEN_FINALIZE) reasons.append( f"executor_key == anu_key ({executor_key!r}) at finalize — " "SELF_COLLECTOR_FORBIDDEN." ) if pol.get("sendfile_only_forbidden", True) and sendfile_only: cls.append(SENDFILE_ONLY_FORBIDDEN_FINALIZE) reasons.append( "sendfile_only_forbidden=True but sendfile-only path observed " "(cron 등록 0 + sendfile 만) — NOT_REGISTERED 변종." ) if pol.get("not_registered_forbidden", True) and not callback_registered: cls.append(NOT_REGISTERED_FORBIDDEN_FINALIZE) reasons.append( "not_registered_forbidden=True but callback NOT registered — " "NOT_REGISTERED 변종 차단." ) expected_anu_key = pol.get("anu_key_single_source") if expected_anu_key and anu_key_used and anu_key_used != expected_anu_key: cls.append(ANU_KEY_MISMATCH_FINALIZE) reasons.append( f"anu_key_used {anu_key_used!r} != single source " f"{expected_anu_key!r} — ANU key 단일 출처 위반." ) seen: set = set() ordered: List[str] = [] for c in cls: if c not in seen: seen.add(c) ordered.append(c) verdict = PASS if not ordered else FAIL if verdict == PASS: reasons.append( "finalize policy PASS — callback registered with schedule_id, " "no self_collector / sendfile_only / not_registered violation, " "anu_key 단일 출처 일치." ) return PolicyCheckResult( schema=POLICY_LOADER_SCHEMA, verdict=verdict, classifications=ordered, policy=pol, finalize_state=dict(finalize_state), reasons=reasons, ) __all__ = [ "POLICY_LOADER_SCHEMA", "DEFAULT_POLICY", "BOT_FINALIZE_POLICY_VIOLATION", "SELF_COLLECTOR_FORBIDDEN_FINALIZE", "SENDFILE_ONLY_FORBIDDEN_FINALIZE", "NOT_REGISTERED_FORBIDDEN_FINALIZE", "ANU_KEY_MISMATCH_FINALIZE", "PASS", "FAIL", "PolicyLoadResult", "load_callback_policy", "PolicyCheckResult", "check_finalize_policy", ]