# -*- coding: utf-8 -*- """dispatch.spawn_callback_contract_validator — pre-spawn contract validator. task-2640 Track A 신규 helper (회장 verbatim unfork #1/#2). 봇 spawn 직전 호출되어 callback contract 가 ANU-owned 정합 상태인지 fail-closed 검증한다. SELF_COLLECTOR / SENDFILE_ONLY / NOT_REGISTERED 변종을 dispatch 진입 시점에 차단하여 prompt 텍스트만으로 callback 정책 전달되는 구조적 결함을 봉쇄. 검증 5축 (task md ANCHOR-3 verbatim): 1. executor_key 비공란 + ≠ anu_key (self-key 차단) 2. anu_key 가 DEFAULT_ANU_KEYS 에 포함 (independent ANU 정합) 3. prompt_text 에 anu_key 텍스트 포함 (코드 contract 와 prompt 일치) 4. prompt_text 에 collector_role=ANU doctrine 포함 5. prompt_text 에 SELF_COLLECTOR / SENDFILE_ONLY / NOT_REGISTERED 차단 verbatim block 포함 이 모듈은 Layer A / NO-CRON: subprocess / cokacdir / merge / cron register 0. 순수 데이터 검증만 수행하고 verdict 를 반환한다. 실패 시 호출자가 spawn 진입 자체를 차단해야 한다 (fail-closed). """ from __future__ import annotations from dataclasses import dataclass, field from typing import List, Optional, Sequence from dispatch.callback_owner_enforcer import ( ANU_KEY_2553, COLLECTOR_ROLE_ANU, DEFAULT_ANU_KEYS, FAIL, HOLD, PASS, SELF_COLLECTOR_FORBIDDEN, is_anu_key, ) VALIDATOR_SCHEMA = "dispatch.spawn_callback_contract_validator.v1" NO_OP_SPAWN_CONTRACT_FAILED = "NO_OP_SPAWN_CONTRACT_FAILED" PROMPT_DOCTRINE_MISSING = "PROMPT_DOCTRINE_MISSING" PROMPT_ANU_KEY_MISSING = "PROMPT_ANU_KEY_MISSING" PROMPT_COLLECTOR_ROLE_MISSING = "PROMPT_COLLECTOR_ROLE_MISSING" EXECUTOR_KEY_MISSING = "EXECUTOR_KEY_MISSING" ANU_KEY_NOT_INDEPENDENT = "ANU_KEY_NOT_INDEPENDENT" TASK_ID_MISSING = "TASK_ID_MISSING" REQUIRED_DOCTRINE_TOKENS: tuple = ( "SELF_COLLECTOR", "SENDFILE_ONLY", "NOT_REGISTERED", ) @dataclass class ContractValidationResult: schema: str verdict: str # PASS | FAIL | HOLD_FOR_CHAIR classifications: List[str] task_id: str executor_key: str anu_key: str prompt_has_anu_key: bool prompt_has_collector_role: bool prompt_has_required_doctrine: bool 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, "task_id": self.task_id, "executor_key": self.executor_key, "anu_key": self.anu_key, "prompt_has_anu_key": self.prompt_has_anu_key, "prompt_has_collector_role": self.prompt_has_collector_role, "prompt_has_required_doctrine": ( self.prompt_has_required_doctrine ), "reasons": list(self.reasons), } def _prompt_has_all_doctrine_tokens(prompt_text: str) -> bool: if not prompt_text: return False return all(t in prompt_text for t in REQUIRED_DOCTRINE_TOKENS) def validate_spawn_callback_contract( *, task_id: str, executor_key: str, anu_key: str = ANU_KEY_2553, prompt_text: str, anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), anu_keys_resolvable: bool = True, ) -> ContractValidationResult: """봇 spawn 직전 호출되는 fail-closed contract validator. 검증 실패 시 verdict=FAIL · 호출자(dispatch.core.dispatch_to_bot)는 subprocess(cokacdir --cron) 발사를 차단해야 함. Args: task_id: spawn 대상 task id (non-empty 필수) executor_key: spawn 대상 봇의 self key (ANU key 와 달라야 함) anu_key: independent ANU collector key (DEFAULT_ANU_KEYS 포함 필수) prompt_text: build_prompt 산출 prompt 본문 anu_keys: 허용 ANU key 집합 (override 가능) anu_keys_resolvable: ANU key 집합 resolution 가능 여부 (HOLD 경로) Returns: ContractValidationResult — verdict 와 classification list, reasons. """ cls: List[str] = [] reasons: List[str] = [] if not anu_keys_resolvable or anu_keys is None or not anu_keys: return ContractValidationResult( schema=VALIDATOR_SCHEMA, verdict=HOLD, classifications=[], task_id=task_id or "", executor_key=executor_key or "", anu_key=anu_key or "", prompt_has_anu_key=False, prompt_has_collector_role=False, prompt_has_required_doctrine=False, reasons=[ "ANU key set unresolvable — spawn contract validation cannot " "fail-closed enforce owner pin in this environment " "(HOLD_FOR_CHAIR). NO silent pass." ], ) if not task_id: cls.append(TASK_ID_MISSING) reasons.append("task_id empty — spawn contract requires non-empty id.") if not executor_key: cls.append(EXECUTOR_KEY_MISSING) reasons.append( "executor_key empty — spawn contract cannot verify owner pin." ) if executor_key and anu_key and executor_key == anu_key: cls.append(SELF_COLLECTOR_FORBIDDEN) reasons.append( f"executor_key == anu_key ({executor_key!r}) — SELF_COLLECTOR_" "FORBIDDEN; the executor bot must NEVER be its own callback " "collector (회장 §2/§10)." ) if not is_anu_key(anu_key, anu_keys): cls.append(ANU_KEY_NOT_INDEPENDENT) reasons.append( f"anu_key {anu_key!r} not in configured DEFAULT_ANU_KEYS — " "the collector key must be an independent ANU key (회장 §10)." ) prompt_has_anu_key = bool(prompt_text) and bool(anu_key) and anu_key in prompt_text if not prompt_has_anu_key: cls.append(PROMPT_ANU_KEY_MISSING) reasons.append( "prompt body does NOT contain the ANU key text — code contract " "(spawn validator) and prompt text are out-of-sync; SELF_COLLECTOR " "변종 재발 차단을 위해 텍스트+코드 이중 단언 필요." ) prompt_has_collector_role = bool(prompt_text) and ( COLLECTOR_ROLE_ANU in prompt_text or "collector_role=ANU" in prompt_text ) if not prompt_has_collector_role: cls.append(PROMPT_COLLECTOR_ROLE_MISSING) reasons.append( "prompt body does NOT mention collector_role=ANU — collector " "role doctrine must be explicit at prompt-level." ) prompt_has_required_doctrine = _prompt_has_all_doctrine_tokens(prompt_text) if not prompt_has_required_doctrine: cls.append(PROMPT_DOCTRINE_MISSING) reasons.append( "prompt body missing one or more required doctrine tokens " f"{REQUIRED_DOCTRINE_TOKENS!r} — verbatim doctrine block 자동 " "주입 누락 (task md ANCHOR-4)." ) # de-dup classifications, preserve order 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( "spawn callback contract PASS — executor_key != anu_key, " "anu_key is an independent ANU key, prompt contains anu_key + " "collector_role=ANU + SELF_COLLECTOR/SENDFILE_ONLY/NOT_REGISTERED " "doctrine tokens. dispatch subprocess gated through." ) else: # primary 분류는 SELF_COLLECTOR_FORBIDDEN > 그 외 순으로. if SELF_COLLECTOR_FORBIDDEN in ordered and ordered[0] != SELF_COLLECTOR_FORBIDDEN: ordered.remove(SELF_COLLECTOR_FORBIDDEN) ordered.insert(0, SELF_COLLECTOR_FORBIDDEN) return ContractValidationResult( schema=VALIDATOR_SCHEMA, verdict=verdict, classifications=ordered, task_id=task_id or "", executor_key=executor_key or "", anu_key=anu_key or "", prompt_has_anu_key=prompt_has_anu_key, prompt_has_collector_role=prompt_has_collector_role, prompt_has_required_doctrine=prompt_has_required_doctrine, reasons=reasons, ) __all__ = [ "VALIDATOR_SCHEMA", "NO_OP_SPAWN_CONTRACT_FAILED", "PROMPT_DOCTRINE_MISSING", "PROMPT_ANU_KEY_MISSING", "PROMPT_COLLECTOR_ROLE_MISSING", "EXECUTOR_KEY_MISSING", "ANU_KEY_NOT_INDEPENDENT", "TASK_ID_MISSING", "REQUIRED_DOCTRINE_TOKENS", "ContractValidationResult", "validate_spawn_callback_contract", ]