# -*- coding: utf-8 -*- """anu_v3.self_collector_guard — executor self-collector / self-adjudication / self-dispatch 실 runtime 구조적 차단 (task-2553+49 AUTHORITATIVE §3/§5.A). 회장 §3 필수목표: "executor self-callback 차단 · executor self-collector 차단 · executor self-adjudication 차단 · executor self-dispatch 차단" 을 실제 runtime path 에 결선한다. narrow +49 ``dispatch.callback_owner_enforcer`` 의 ``assert_not_self_adjudication`` / ``assert_not_self_dispatch`` 를 runtime guard 로 결선하고, collector 회수단계에서 self-session collector-result 를 무효화하는 ``guard_self_collector_session`` 을 추가한다. dispatch entrypoint 및 collector 회수 경로가 **반드시 경유**해야 하는 fail-closed 게이트. reuse, not re-implement: self-adjudication / self-dispatch 규칙은 비-frozen ``dispatch.callback_owner_enforcer`` 단일 진실원을 호출. Layer A / NO-CRON: 순수 검증. ZERO cron / dispatch / subprocess / cokacdir. """ from __future__ import annotations from dataclasses import dataclass, field from typing import List, Optional from dispatch.callback_owner_enforcer import ( FAIL, PASS, assert_not_self_adjudication, assert_not_self_dispatch, ) GUARD_SCHEMA = "anu_v3.self_collector_guard.v1" SELF_COLLECTOR_FORBIDDEN = "SELF_COLLECTOR_FORBIDDEN" EXECUTOR_SELF_ADJUDICATION_FORBIDDEN = "EXECUTOR_SELF_ADJUDICATION_FORBIDDEN" EXECUTOR_SELF_DISPATCH_FORBIDDEN = "EXECUTOR_SELF_DISPATCH_FORBIDDEN" # §6 spec uses EXECUTOR_SELF_DISPATCH_FORBIDDEN; the narrow enforcer emitted # SELF_DISPATCH_FORBIDDEN. Carry both so callers/regression can match the # §6-verbatim token without mutating the byte-0-adjacent narrow module. SELF_DISPATCH_FORBIDDEN_ALIASES = frozenset( {EXECUTOR_SELF_DISPATCH_FORBIDDEN, "SELF_DISPATCH_FORBIDDEN"} ) @dataclass class SelfCollectorGuardResult: schema: str verdict: str # PASS | FAIL classification: Optional[str] is_executor_self_session: bool blocked_action: Optional[str] reasons: List[str] = field(default_factory=list) @property def ok(self) -> bool: return self.verdict == PASS def to_json(self) -> dict: return { "schema": self.schema, "verdict": self.verdict, "classification": self.classification, "is_executor_self_session": self.is_executor_self_session, "blocked_action": self.blocked_action, "reasons": list(self.reasons), } def guard_self_collector_session( *, executor_key: str, collector_key: Optional[str] = None, actor_key: Optional[str] = None, collector_role: str = "ANU", is_executor_self_session: Optional[bool] = None, ) -> SelfCollectorGuardResult: """collector 회수단계 fail-closed: executor self-session 이 자기 completion collector 를 *소유/운영* 하면 무효 (§3 / §5.A). self iff: actor_key == executor_key, OR collector_key == executor_key, OR explicit self-session flag (escalate-only — never relaxes the key-derived signal, narrow enforcer 동형). """ key_self = bool(actor_key) and actor_key == executor_key collector_self = bool(collector_key) and collector_key == executor_key explicit = bool(is_executor_self_session) if ( is_executor_self_session is not None ) else False self_session = key_self or collector_self or explicit if self_session or collector_role != "ANU": cls = ( SELF_COLLECTOR_FORBIDDEN if self_session else "CALLBACK_COLLECTOR_NOT_ANU" ) reason = ( "executor self-session owns/operates its own completion " "collector — structurally forbidden; the collector MUST be an " "independent ANU session (§3 / §5.A)." if self_session else f"collector_role={collector_role!r} != 'ANU' — collector " "must run as an independent ANU session (§5.A)." ) return SelfCollectorGuardResult( schema=GUARD_SCHEMA, verdict=FAIL, classification=cls, is_executor_self_session=self_session, blocked_action="self_collector", reasons=[reason], ) return SelfCollectorGuardResult( schema=GUARD_SCHEMA, verdict=PASS, classification=None, is_executor_self_session=False, blocked_action=None, reasons=["collector is an independent ANU session (no self-chain)."], ) def guard_self_adjudication( *, executor_key: str, actor_key: Optional[str] = None, is_codex_audit: bool = False, is_adjudication: bool = False, is_executor_self_session: Optional[bool] = None, ) -> SelfCollectorGuardResult: """executor self-session Codex audit / ANU-Codex adjudication 차단 (§3 / §5.A / regression 11). Delegates to the narrow enforcer rule.""" r = assert_not_self_adjudication( executor_key=executor_key, actor_key=actor_key, is_codex_audit=is_codex_audit, is_adjudication=is_adjudication, is_executor_self_session=is_executor_self_session, ) return SelfCollectorGuardResult( schema=GUARD_SCHEMA, verdict=r.verdict, classification=( EXECUTOR_SELF_ADJUDICATION_FORBIDDEN if r.verdict == FAIL else None ), is_executor_self_session=r.is_executor_self_session, blocked_action="self_adjudication" if r.verdict == FAIL else None, reasons=list(r.reasons), ) def guard_self_dispatch( *, executor_key: str, actor_key: Optional[str] = None, is_followup_dispatch: bool = False, is_executor_self_session: Optional[bool] = None, ) -> SelfCollectorGuardResult: """executor self-session 후속 dispatch / delegation 차단 (§3 / §5.A / regression 12). Delegates to the narrow enforcer rule.""" r = assert_not_self_dispatch( executor_key=executor_key, actor_key=actor_key, is_followup_dispatch=is_followup_dispatch, is_executor_self_session=is_executor_self_session, ) return SelfCollectorGuardResult( schema=GUARD_SCHEMA, verdict=r.verdict, classification=( EXECUTOR_SELF_DISPATCH_FORBIDDEN if r.verdict == FAIL else None ), is_executor_self_session=r.is_executor_self_session, blocked_action="self_dispatch" if r.verdict == FAIL else None, reasons=list(r.reasons), ) __all__ = [ "GUARD_SCHEMA", "PASS", "FAIL", "SELF_COLLECTOR_FORBIDDEN", "EXECUTOR_SELF_ADJUDICATION_FORBIDDEN", "EXECUTOR_SELF_DISPATCH_FORBIDDEN", "SELF_DISPATCH_FORBIDDEN_ALIASES", "SelfCollectorGuardResult", "guard_self_collector_session", "guard_self_adjudication", "guard_self_dispatch", ]