# -*- coding: utf-8 -*- """anu_v3.proposal_authorization_gate — classify a +54 runtime-event-loop proposal candidate and decide the single bounded action the enactor is permitted to take (task-2553+55 §2.2~2.7 / §8). 회장 §2 verbatim 2~7: 2. proposal type 구분: consolidated_summary · closeout_candidate · next_dispatch_candidate · hold_packet. 3. 권한 있는 action만 enact. 4. 권한 없는 action은 proposal 상태로 유지. 5. closeout은 additive artifact로만 enact. 6. next dispatch는 ANU key + owner/key guard + callback mandatory contract 통과 시에만 candidate->actual dispatch 승격. 7. merge/PR/credential/branch write 기본 금지. This gate is the *authority boundary*. It NEVER enacts; it only classifies and authorizes. Every decision routes through the +49 single sources of truth (no rule re-implementation): * ``dispatch.callback_owner_enforcer.is_anu_key`` — ANU-key ownership. * ``anu_v3.callback_owner_validator.validate_callback_owner_runtime`` — callback owner/key/role fail-closed contract (§2.6). * ``anu_v3.self_collector_guard.guard_self_collector_session`` / ``anu_v3.authoritative_verdict_selector.select_authoritative_verdict`` — self-chain / non-ANU collector permanent quarantine (§8). * ``anu_v3.executor_callback_contract`` — the callback-mandatory doctrine (a next-dispatch candidate with no mandatory ANU callback is NOT promotable; §2.6). Track boundary (회장 §3 / §8): * consolidated_summary / closeout_candidate -> ``ENACT_ADDITIVE`` (Track A: additive artifact ONLY — zero merge/PR/branch/write to any existing file). * next_dispatch_candidate -> ``DISPATCH_READY`` *iff* the §2.6 triple passes — but Track B is **verify-only**: dispatch-ready is NOT execution; the actual candidate->dispatch promotion requires an explicit chair grant (``allow_actual_dispatch``) which this task never sets. Otherwise ``PROPOSAL_ONLY``. * hold_packet (self-chain / non-ANU / mismatch) -> ``HOLD_ROUTED`` (Track C: routing/safety-gate verification only). * any merge / PR / credential / OWNER_PAT / branch-write target -> ``BLOCKED`` (stays a proposal; 회장 §6/§7). Layer A / NO-CRON: pure classification. ZERO cron / dispatch / subprocess / cokacdir / merge / PR / branch / credential. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Dict, List, Optional, Sequence from anu_v3.authoritative_verdict_selector import ( # pyright: ignore[reportMissingImports] # noqa: E501 AUTHORITATIVE_PASS, COLLECTOR_ROLE_ANU, VerdictRecord, select_authoritative_verdict, ) from anu_v3.callback_owner_validator import ( # pyright: ignore[reportMissingImports] # noqa: E501 is_anu_key, validate_callback_owner_runtime, ) from anu_v3.self_collector_guard import ( # pyright: ignore[reportMissingImports] # noqa: E501 guard_self_collector_session, ) GATE_SCHEMA = "anu_v3.proposal_authorization_gate.v1" # ── §2.2 proposal types ────────────────────────────────────────────────────── PROPOSAL_CONSOLIDATED_SUMMARY = "consolidated_summary" PROPOSAL_CLOSEOUT_CANDIDATE = "closeout_candidate" PROPOSAL_NEXT_DISPATCH_CANDIDATE = "next_dispatch_candidate" PROPOSAL_HOLD_PACKET = "hold_packet" PROPOSAL_UNKNOWN = "unknown_proposal" # ── bounded actions (the ONLY actions the enactor may take) ────────────────── ACTION_ENACT_ADDITIVE = "ENACT_ADDITIVE" # Track A — additive only ACTION_DISPATCH_READY = "DISPATCH_READY" # Track B — verify only ACTION_PROPOSAL_ONLY = "PROPOSAL_ONLY" # unauthorized -> stays proposal ACTION_BLOCKED = "BLOCKED" # merge/PR/cred/branch -> §6 ACTION_HOLD_ROUTED = "HOLD_ROUTED" # Track C — routing only # Structured forbidden-target tokens (§2.7/§6). Matched ONLY against the # *structured* next_phase action/kind/target fields + explicit boolean # requirement flags — NEVER against free-form note/disclaimer text (whose # wording legitimately contains "merge"/"closeout" in a negation). FORBIDDEN_ACTION_TOKENS = frozenset( { "merge", "pr", "pull_request", "pullrequest", "credential", "owner_pat", "pat", "branch_write", "branch-write", "main_write", "git_push", "push", "secret", } ) FORBIDDEN_REQUIREMENT_FLAGS = ( "requires_credential", "requires_branch_write", "requires_merge", "requires_pr", "requires_owner_pat", "requires_main_write", ) @dataclass class AuthorizationDecision: schema: str proposal_type: str action: str authorized: bool enact_as_additive: bool dispatch_ready: bool blocked: bool source: str source_id: str owner_validation: Optional[Dict[str, object]] selector: Optional[Dict[str, object]] self_guard: Optional[Dict[str, object]] callback_contract_ok: bool reasons: List[str] = field(default_factory=list) def to_json(self) -> Dict[str, object]: return { "schema": self.schema, "proposal_type": self.proposal_type, "action": self.action, "authorized": self.authorized, "enact_as_additive": self.enact_as_additive, "dispatch_ready": self.dispatch_ready, "blocked": self.blocked, "source": self.source, "source_id": self.source_id, "owner_validation": self.owner_validation, "selector": self.selector, "self_guard": self.self_guard, "callback_contract_ok": self.callback_contract_ok, "reasons": list(self.reasons), } # ── classification ─────────────────────────────────────────────────────────── def classify_proposal(candidate: Dict[str, object]) -> str: """Map a +54 candidate dict onto one of the §2.2 proposal types. * a consolidated-summary candidate schema / status -> consolidated_summary * a quarantined self-chain / non-ANU / mismatch packet -> hold_packet * a dispatch candidate whose declared next_phase is a *closeout* -> closeout_candidate (Track A target) * any other dispatch candidate -> next_dispatch_candidate """ schema = str(candidate.get("schema") or "") status = str(candidate.get("status") or "") if "consolidated_summary" in schema or status == ( "CONSOLIDATED_SUMMARY_CANDIDATE_READY" ): return PROPOSAL_CONSOLIDATED_SUMMARY if candidate.get("event_kind") or candidate.get("hold_packet"): return PROPOSAL_HOLD_PACKET if "dispatch_candidate" in schema or candidate.get("next_phase"): np = candidate.get("next_phase") if isinstance(np, dict): blob = " ".join( str(np.get(k) or "") for k in ("phase", "kind", "intent", "action") ).lower() if "closeout" in blob or "consolidated closeout" in blob: return PROPOSAL_CLOSEOUT_CANDIDATE return PROPOSAL_NEXT_DISPATCH_CANDIDATE return PROPOSAL_UNKNOWN def _forbidden_target(candidate: Dict[str, object]) -> Optional[str]: """Detect a structurally forbidden enact target (§2.7/§6). Only structured fields are inspected (never note text).""" np = candidate.get("next_phase") np = np if isinstance(np, dict) else {} for flag in FORBIDDEN_REQUIREMENT_FLAGS: if bool(candidate.get(flag)) or bool(np.get(flag)): return flag fa = candidate.get("forbidden_actions") if isinstance(fa, (list, tuple)) and fa: return f"forbidden_actions:{','.join(str(x) for x in fa)}" for key in ("kind", "action", "target", "intent", "operation"): val = str(np.get(key) or "").lower() tokens = set(val.replace("/", "_").replace("-", "_").split("_")) hit = tokens & FORBIDDEN_ACTION_TOKENS if hit: return f"next_phase.{key}={val!r} -> {sorted(hit)}" return None def _hold_packet_decision( candidate: Dict[str, object], ) -> AuthorizationDecision: return AuthorizationDecision( schema=GATE_SCHEMA, proposal_type=PROPOSAL_HOLD_PACKET, action=ACTION_HOLD_ROUTED, authorized=False, enact_as_additive=False, dispatch_ready=False, blocked=False, source=str(candidate.get("schema_kind") or candidate.get("source") or "quarantine"), # noqa: E501 source_id=str( candidate.get("task_id") or candidate.get("source_id") or candidate.get("event_id") or "" ), owner_validation=None, selector=None, self_guard=None, callback_contract_ok=False, reasons=[ "hold_packet (self-chain / non-ANU / registry-mismatch) — Track " "C routing/safety-gate verification ONLY; never enacted, never " "dispatched, never auto-resolved (회장 §3.C/§8). Routed to the " "independent-ANU HOLD lane." ], ) def authorize_proposal( candidate: Dict[str, object], *, anu_keys: Sequence[str], executor_self_key: str = "", allow_actual_dispatch: bool = False, ) -> AuthorizationDecision: """Decide the single bounded action for one proposal candidate. ``allow_actual_dispatch`` is the *explicit chair grant* for an actual candidate->dispatch promotion. task-2553+55 NEVER sets it (Track B is verify-only); even when a dispatch candidate passes the full §2.6 triple, the action is DISPATCH_READY (proposal stays a proposal until a future chair-granted session). """ ptype = classify_proposal(candidate) source = str(candidate.get("source") or candidate.get("schema") or "") source_id = str( candidate.get("source_id") or candidate.get("batch_id") or candidate.get("task_id") or "" ) if ptype == PROPOSAL_HOLD_PACKET: return _hold_packet_decision(candidate) # §2.7/§6 — a forbidden target is BLOCKED before anything else. forb = _forbidden_target(candidate) if forb is not None: return AuthorizationDecision( schema=GATE_SCHEMA, proposal_type=ptype, action=ACTION_BLOCKED, authorized=False, enact_as_additive=False, dispatch_ready=False, blocked=True, source=source, source_id=source_id, owner_validation=None, selector=None, self_guard=None, callback_contract_ok=False, reasons=[ f"forbidden enact target detected ({forb}) — merge / PR / " "credential / OWNER_PAT / branch-write is BLOCKED by " "default; the proposal is retained as a proposal and never " "enacted (회장 §2.7/§6/§7)." ], ) if ptype in ( PROPOSAL_CONSOLIDATED_SUMMARY, PROPOSAL_CLOSEOUT_CANDIDATE, ): all_settled = bool(candidate.get("all_settled")) all_ap = bool(candidate.get("all_authoritative_pass")) gate_ev = candidate.get("gate_evidence") if isinstance(gate_ev, dict): all_settled = all_settled or bool(gate_ev.get("all_settled")) all_ap = all_ap or bool(gate_ev.get("all_authoritative_pass")) if all_settled and all_ap: return AuthorizationDecision( schema=GATE_SCHEMA, proposal_type=ptype, action=ACTION_ENACT_ADDITIVE, authorized=True, enact_as_additive=True, dispatch_ready=False, blocked=False, source=source, source_id=source_id, owner_validation=None, selector=None, self_guard=None, callback_contract_ok=True, reasons=[ "all-settled & all-AUTHORITATIVE_PASS consolidated/" "closeout proposal — AUTHORIZED for an ADDITIVE artifact " "enact ONLY (Track A, 회장 §2.5/§3.A/§8: zero merge/PR/" "branch, zero modification of any existing artifact)." ], ) return AuthorizationDecision( schema=GATE_SCHEMA, proposal_type=ptype, action=ACTION_PROPOSAL_ONLY, authorized=False, enact_as_additive=False, dispatch_ready=False, blocked=False, source=source, source_id=source_id, owner_validation=None, selector=None, self_guard=None, callback_contract_ok=False, reasons=[ "consolidated/closeout proposal is NOT all-settled & " "all-AUTHORITATIVE_PASS — unauthorized for enact; retained " "as a proposal (회장 §2.4)." ], ) # next_dispatch_candidate — §2.6 triple: ANU key + owner/key guard + # callback mandatory contract. Pure verification (Track B). np = candidate.get("next_phase") np = np if isinstance(np, dict) else {} cc = candidate.get("callback_contract") cc = cc if isinstance(cc, dict) else {} owner_key = str( cc.get("collector_owner_key") or cc.get("collector_key") or np.get("owner_key") or "" ) raw_anu = candidate.get("anu_keys") declared_anu_keys = ( [str(k) for k in raw_anu] if isinstance(raw_anu, (list, tuple)) and raw_anu else list(anu_keys) ) anu_owner_ok = is_anu_key(owner_key, declared_anu_keys) if owner_key else False # noqa: E501 executor_key = str(cc.get("executor_key") or "") collector_key = str(cc.get("collector_key") or owner_key) collector_role = str(cc.get("collector_role") or COLLECTOR_ROLE_ANU) # The executor self key MUST never own/operate the next-phase callback # (회장 §6/§9 — +49 정본). An explicit self-key match short-circuits to # a self-chain rejection before the triple is even scored. self_key_taint = bool(executor_self_key) and executor_self_key in ( owner_key, collector_key, ) guard = guard_self_collector_session( executor_key=( executor_self_key if self_key_taint else (executor_key or "unknown_executor") ), collector_key=collector_key or None, collector_role=collector_role, is_executor_self_session=True if self_key_taint else None, ) sel = select_authoritative_verdict( [ VerdictRecord( kind="collector_result", verdict="PASS", task_id=source_id, executor_key=executor_key or "unknown_executor", collector_key=collector_key or owner_key, collector_role=collector_role, session_is_executor_self=bool(collector_key) and collector_key == executor_key, claimed_origin="independent_anu", detail="next_dispatch_candidate_gate", ) ], task_id=source_id, anu_keys=declared_anu_keys, ) contract_present = bool( cc.get("normal_collector_cron_id") and cc.get("callback_mandatory") is True ) ov = None if owner_key and executor_key: ov = validate_callback_owner_runtime( task_id=source_id or "next-dispatch-candidate", executor_key=executor_key, collector_key=collector_key or owner_key, collector_owner_key=owner_key, collector_role=collector_role, normal_collector_cron_id=str( cc.get("normal_collector_cron_id") or "" ) or None, fallback_callback_cron_id=str( cc.get("fallback_callback_cron_id") or "" ) or None, dispatch_cron_id=str(cc.get("dispatch_cron_id") or "DG-next"), chat_id=str(cc.get("chat_id") or ""), anu_keys=declared_anu_keys, no_fallback=bool(cc.get("no_fallback")), ).to_json() owner_ok = bool(ov and ov.get("registration_allowed")) guard_ok = bool(guard.ok) sel_ok = ( sel.classification == AUTHORITATIVE_PASS and sel.independent_anu_count >= 1 ) triple_ok = ( anu_owner_ok and owner_ok and guard_ok and sel_ok and contract_present ) if not triple_ok: return AuthorizationDecision( schema=GATE_SCHEMA, proposal_type=PROPOSAL_NEXT_DISPATCH_CANDIDATE, action=ACTION_PROPOSAL_ONLY, authorized=False, enact_as_additive=False, dispatch_ready=False, blocked=False, source=source, source_id=source_id, owner_validation=ov, selector=sel.to_json(), self_guard=guard.to_json(), callback_contract_ok=contract_present, reasons=[ "next_dispatch_candidate did NOT pass the §2.6 triple " f"(anu_key_owner={anu_owner_ok}, owner_validation=" f"{owner_ok}, self_guard={guard_ok}, authoritative_selector" f"={sel_ok}, callback_mandatory_contract={contract_present})" " — UNAUTHORIZED; retained as a proposal, zero dispatch " "(회장 §2.4/§2.6/§8). self-chain / non-ANU collector is " "permanently rejected via the +49 selector/guard." ], ) # §2.6 triple PASSED. Track B is verify-only: dispatch-ready, NOT # executed — the actual candidate->dispatch promotion is gated on an # explicit chair grant which task-2553+55 never sets (§3.B/§8/§10). return AuthorizationDecision( schema=GATE_SCHEMA, proposal_type=PROPOSAL_NEXT_DISPATCH_CANDIDATE, action=ACTION_DISPATCH_READY, authorized=True, enact_as_additive=False, dispatch_ready=True, blocked=False, source=source, source_id=source_id, owner_validation=ov, selector=sel.to_json(), self_guard=guard.to_json(), callback_contract_ok=True, reasons=[ "next_dispatch_candidate PASSED the §2.6 triple (ANU-key owner " "+ +49 owner/key guard + callback-mandatory contract) — " "DISPATCH-READY. Track B is verify-only: NO actual dispatch is " "executed in this task; the candidate->dispatch promotion " "requires an explicit chair grant " f"(allow_actual_dispatch={allow_actual_dispatch}); not set here " "(회장 §3.B/§8/§10 — 자동 실 dispatch 승격 0)." ], ) __all__ = [ "GATE_SCHEMA", "PROPOSAL_CONSOLIDATED_SUMMARY", "PROPOSAL_CLOSEOUT_CANDIDATE", "PROPOSAL_NEXT_DISPATCH_CANDIDATE", "PROPOSAL_HOLD_PACKET", "PROPOSAL_UNKNOWN", "ACTION_ENACT_ADDITIVE", "ACTION_DISPATCH_READY", "ACTION_PROPOSAL_ONLY", "ACTION_BLOCKED", "ACTION_HOLD_ROUTED", "FORBIDDEN_ACTION_TOKENS", "AuthorizationDecision", "classify_proposal", "authorize_proposal", ]