# -*- coding: utf-8 -*- """anu_v3.authoritative_verdict_selector — self-chain verdict 영구 비권위 + independent ANU verdict 만 authoritative 선택 (task-2553+49 AUTHORITATIVE §5.D). 회장 §2.5/§2.6/§5.D verbatim: * self-chain verdict 는 authoritative verdict 가 될 수 없다. * independent ANU verification verdict 가 있으면 그것이 authoritative verdict. * executor self-session 산출 collector-result / Codex audit / adjudication 은 authoritative 로 선택 불가 -> ``SELF_CHAIN_QUARANTINED``. * dead-man ``DUPLICATE_CALLBACK_IGNORED`` 판정은 *self-chain collector 정당성* 근거가 아니라 *authoritative verdict 존재 여부* 근거로만 유효. * self-chain verdict 만 존재 -> ``AUTHORITATIVE_VERDICT_PENDING`` (HOLD 후보). * independent ANU verdict 없이 self-chain 만으로 PASS 확정 -> FAIL (``AUTHORITATIVE_PASS`` 절대 부여 금지). 이 모듈은 **runtime 결선** 대상이다. dispatch/cron_dispatch_guard 의 ``select_runtime_authoritative_verdict`` 결선부 및 collector 회수단계에서 실제 verdict 선택 시 호출된다 — narrow +49 enforcer 가 빠뜨린 §5.D 권위 선택 경로를 코드로 강제. Layer A / NO-CRON: 순수 분류·선택. ZERO cron / dispatch / subprocess / cokacdir. verdict 레코드를 *선택* 만 한다 (회수·발사·등록 0). """ from __future__ import annotations from dataclasses import dataclass, field from typing import List, Optional, Sequence SELECTOR_SCHEMA = "anu_v3.authoritative_verdict_selection.v1" # ── verdict selection classifications (§6) ─────────────────────────────────── SELF_CHAIN_QUARANTINED = "SELF_CHAIN_QUARANTINED" AUTHORITATIVE_VERDICT_PENDING = "AUTHORITATIVE_VERDICT_PENDING" AUTHORITATIVE_PASS = "AUTHORITATIVE_PASS" EXECUTOR_SELF_ADJUDICATION_FORBIDDEN = "EXECUTOR_SELF_ADJUDICATION_FORBIDDEN" PASS = "PASS" FAIL = "FAIL" HOLD = "HOLD_FOR_CHAIR" COLLECTOR_ROLE_ANU = "ANU" # Verdict origin kinds. ORIGIN_INDEPENDENT_ANU = "independent_anu" ORIGIN_SELF_CHAIN = "self_chain" # Verdict producer kinds (all equally quarantined when self-chain). KIND_COLLECTOR_RESULT = "collector_result" KIND_CODEX_AUDIT = "codex_audit" KIND_ADJUDICATION = "adjudication" KIND_DEADMAN = "deadman_duplicate_callback" @dataclass(frozen=True) class VerdictRecord: """A single verdict candidate presented to the authoritative selector. ``origin`` is *derived* — never trusted from a free-text claim. The selector recomputes it from owner identity (executor_key vs collector_key, collector_role, anu_keys) so a self-session that *claims* independence is still quarantined (회장 §5.D / prompt-text 무권위). """ kind: str # collector_result | codex_audit | adjudication | deadman_* verdict: str # PASS | FAIL | HOLD_FOR_CHAIR | DUPLICATE_CALLBACK_IGNORED task_id: str executor_key: str collector_key: str collector_role: str session_is_executor_self: bool = False claimed_origin: str = "" # untrusted hint only — recomputed detail: str = "" def derived_origin(self, anu_keys: Sequence[str]) -> str: """Recompute origin from owner identity (claim is ignored). Independent-ANU iff: collector_key is a configured ANU key AND collector_key != executor_key AND collector_role == 'ANU' AND the session is NOT the executor self-session. Anything else = self_chain. """ keyset = set(k for k in anu_keys if k) independent = ( bool(self.collector_key) and self.collector_key in keyset and self.collector_key != self.executor_key and self.collector_role == COLLECTOR_ROLE_ANU and not self.session_is_executor_self ) return ORIGIN_INDEPENDENT_ANU if independent else ORIGIN_SELF_CHAIN @dataclass class AuthoritativeSelectionResult: schema: str verdict: str # PASS | FAIL | HOLD_FOR_CHAIR classification: str # SELF_CHAIN_QUARANTINED | _PENDING | AUTHORITATIVE_PASS task_id: str authoritative_verdict: Optional[str] authoritative_source_kind: Optional[str] quarantined_count: int independent_anu_count: int deadman_duplicate_seen: bool deadman_valid: bool 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, "task_id": self.task_id, "authoritative_verdict": self.authoritative_verdict, "authoritative_source_kind": self.authoritative_source_kind, "quarantined_count": self.quarantined_count, "independent_anu_count": self.independent_anu_count, "deadman_duplicate_seen": self.deadman_duplicate_seen, "deadman_valid": self.deadman_valid, "reasons": list(self.reasons), } def select_authoritative_verdict( records: Sequence[VerdictRecord], *, task_id: str, anu_keys: Sequence[str], anu_keys_resolvable: bool = True, ) -> AuthoritativeSelectionResult: """회장 §5.D 권위 verdict 선택 — fail-closed. Verdict logic: * ANU key set unresolvable -> HOLD_FOR_CHAIR (§11; the only conditional escalation) * every record is self-chain -> SELF_CHAIN_QUARANTINED, classification PENDING, verdict FAIL (independent ANU verdict 없이 self-chain 만으로 PASS 확정 금지 — §5.D / regression 13) * >=1 independent-ANU verdict present -> that verdict is AUTHORITATIVE (regression 14/17). Its PASS yields AUTHORITATIVE_PASS; its FAIL/HOLD propagates. * dead-man DUPLICATE_CALLBACK_IGNORED -> only counts as a valid disposition when an authoritative (independent -ANU) verdict also exists (regression 15); never on its own legitimises a self-chain collector. self-session collector-result / Codex audit / adjudication are ALL quarantined regardless of producer kind (회장 §5.D — executor self-session 산출은 무권위). """ if not anu_keys_resolvable or not list(anu_keys): return AuthoritativeSelectionResult( schema=SELECTOR_SCHEMA, verdict=HOLD, classification=AUTHORITATIVE_VERDICT_PENDING, task_id=task_id, authoritative_verdict=None, authoritative_source_kind=None, quarantined_count=0, independent_anu_count=0, deadman_duplicate_seen=False, deadman_valid=False, reasons=[ "ANU key set unresolvable — independent-ANU origin cannot be " "code-derived; authoritative selection cannot be automated " "(§11 HOLD_FOR_CHAIR; the only conditional escalation)." ], ) reasons: List[str] = [] independent: List[VerdictRecord] = [] quarantined: List[VerdictRecord] = [] deadman_seen = False for rec in records: if rec.kind == KIND_DEADMAN or ( rec.verdict == "DUPLICATE_CALLBACK_IGNORED" ): deadman_seen = True origin = rec.derived_origin(anu_keys) if origin == ORIGIN_INDEPENDENT_ANU and rec.kind != KIND_DEADMAN: independent.append(rec) else: quarantined.append(rec) if rec.session_is_executor_self and rec.kind in ( KIND_CODEX_AUDIT, KIND_ADJUDICATION, ): reasons.append( f"executor self-session {rec.kind} quarantined — " "self-adjudication/self-Codex is never authoritative " "(§5.D / EXECUTOR_SELF_ADJUDICATION_FORBIDDEN)." ) else: reasons.append( f"{rec.kind} (verdict={rec.verdict!r}) is self-chain " "origin -> QUARANTINED, permanently non-authoritative " "(§5.D)." ) independent_anu_count = len(independent) quarantined_count = len(quarantined) if independent_anu_count == 0: # self-chain only (or only dead-man). §5.D: independent ANU verdict # 없이 self-chain 만으로 PASS 확정 -> FAIL. Dead-man duplicate alone # does NOT legitimise a self-chain collector (regression 15). deadman_valid = False if deadman_seen: reasons.append( "dead-man DUPLICATE_CALLBACK_IGNORED present but NO " "independent-ANU authoritative verdict exists — the dead-man " "disposition is judged on authoritative-verdict existence, " "NOT on self-chain collector legitimacy; it is NOT valid " "here (§5.D / regression 15)." ) reasons.append( "only self-chain verdict(s) exist — authoritative verdict is " "PENDING; confirming PASS from a self-chain verdict alone is " "FORBIDDEN (회장 §5.D / regression 13). AUTHORITATIVE_PASS NOT " "granted." ) return AuthoritativeSelectionResult( schema=SELECTOR_SCHEMA, verdict=FAIL, classification=AUTHORITATIVE_VERDICT_PENDING, task_id=task_id, authoritative_verdict=None, authoritative_source_kind=None, quarantined_count=quarantined_count, independent_anu_count=0, deadman_duplicate_seen=deadman_seen, deadman_valid=deadman_valid, reasons=reasons, ) # An independent-ANU verdict exists -> it is authoritative (§5.D / # regression 14/17). Strictest independent verdict wins: FAIL > HOLD > # PASS so a self-session cannot drown out an independent FAIL. def _rank(v: str) -> int: return {FAIL: 3, HOLD: 2, PASS: 1}.get(v, 0) chosen = max(independent, key=lambda r: _rank(r.verdict)) deadman_valid = deadman_seen # valid only because authoritative exists if deadman_seen: reasons.append( "dead-man DUPLICATE_CALLBACK_IGNORED is valid here ONLY because " "an independent-ANU authoritative verdict also exists (§5.D / " "regression 15)." ) reasons.append( f"independent-ANU verdict selected as authoritative: " f"kind={chosen.kind!r} verdict={chosen.verdict!r} " f"(collector_key={chosen.collector_key!r}, role=ANU, != executor) " "(회장 §2.6/§5.D / regression 14)." ) if chosen.verdict == PASS: classification = AUTHORITATIVE_PASS verdict = PASS elif chosen.verdict == HOLD: classification = AUTHORITATIVE_VERDICT_PENDING verdict = HOLD else: # FAIL or unknown -> fail-closed classification = AUTHORITATIVE_VERDICT_PENDING verdict = FAIL return AuthoritativeSelectionResult( schema=SELECTOR_SCHEMA, verdict=verdict, classification=classification, task_id=task_id, authoritative_verdict=chosen.verdict, authoritative_source_kind=chosen.kind, quarantined_count=quarantined_count, independent_anu_count=independent_anu_count, deadman_duplicate_seen=deadman_seen, deadman_valid=deadman_valid, reasons=reasons, ) __all__ = [ "SELECTOR_SCHEMA", "SELF_CHAIN_QUARANTINED", "AUTHORITATIVE_VERDICT_PENDING", "AUTHORITATIVE_PASS", "EXECUTOR_SELF_ADJUDICATION_FORBIDDEN", "PASS", "FAIL", "HOLD", "COLLECTOR_ROLE_ANU", "ORIGIN_INDEPENDENT_ANU", "ORIGIN_SELF_CHAIN", "KIND_COLLECTOR_RESULT", "KIND_CODEX_AUDIT", "KIND_ADJUDICATION", "KIND_DEADMAN", "VerdictRecord", "AuthoritativeSelectionResult", "select_authoritative_verdict", ]