# -*- coding: utf-8 -*- """anu_v3.writeback_binding_conflict_guard — write-back role/fallback binding conflict 실 runtime hardening (task-2553+49 AUTHORITATIVE §5.E). 회장 §5.E verbatim: "같은 collector identity 라도 role/fallback binding/owner key 불일치 시 silent skip 금지 -> WRITEBACK_BINDING_CONFLICT 기록 / valid duplicate write-back 만 idempotent SKIP / self-chain write-back 은 completed 불인정." narrow +49 ``audit_writeback_binding_conflict`` 는 role/fallback 만 비교했다. 본 guard 는 그 위에 **owner key 불일치** 와 **self-chain write-back 의 completed 불인정** 을 추가 결선해, write-back 단계가 반드시 경유하는 fail-closed 게이트로 만든다. silent idempotent skip 은 binding 이 완전 일치할 때만 허용. reuse: role/fallback 충돌 판정은 비-frozen ``dispatch.callback_owner_enforcer.audit_writeback_binding_conflict`` 단일 진실원 호출. +44 read-only ledger 는 byte-0 유지. Layer A / NO-CRON: 순수 감사. ZERO cron / dispatch / subprocess / cokacdir. """ from __future__ import annotations from dataclasses import dataclass, field from typing import List, Optional, Sequence from dispatch.callback_owner_enforcer import ( FAIL, PASS, WRITEBACK_BINDING_CONFLICT, WRITEBACK_IDEMPOTENT_SKIP, WRITEBACK_NO_CONFLICT, audit_writeback_binding_conflict, is_anu_key, ) GUARD_SCHEMA = "anu_v3.writeback_binding_conflict.v1" # self-chain write-back is never "completed". WRITEBACK_SELF_CHAIN_NOT_COMPLETED = "WRITEBACK_SELF_CHAIN_NOT_COMPLETED" WRITEBACK_OWNER_KEY_CONFLICT = "WRITEBACK_OWNER_KEY_CONFLICT" @dataclass class WritebackBindingGuardResult: schema: str verdict: str # PASS | FAIL classification: str task_id: str conflict: bool completed_acknowledged: bool conflicting_fields: List[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, "task_id": self.task_id, "conflict": self.conflict, "completed_acknowledged": self.completed_acknowledged, "conflicting_fields": list(self.conflicting_fields), "reasons": list(self.reasons), } def guard_writeback_binding( history: Sequence[object], *, task_id: str, dispatch_id: str, chat_id: str, normal_collector_cron_id: Optional[str], candidate_role: str, candidate_fallback_cron_id: Optional[str], candidate_owner_key: str, executor_key: str, anu_keys: Sequence[str], candidate_session_is_executor_self: bool = False, ) -> WritebackBindingGuardResult: """write-back 단계 fail-closed (§5.E). Order of checks (every violation RECORDED, never silent — §5.E): 1. self-chain write-back (owner == executor key, OR owner not an ANU key, OR the producing session is the executor self-session) -> ``WRITEBACK_SELF_CHAIN_NOT_COMPLETED`` (FAIL): a self-chain write-back is NOT acknowledged as completed (회장 §5.E). 2. role / fallback binding conflict over the +44 read-only ledger -> ``WRITEBACK_BINDING_CONFLICT`` (FAIL, conflicting_fields recorded — NOT a silent idempotent skip; §5.E / regression 19). 3. identical idempotency key AND identical binding -> ``WRITEBACK_ IDEMPOTENT_SKIP`` (PASS, valid duplicate; regression 20). 4. no match -> ``WRITEBACK_NO_CONFLICT`` (PASS, normal new write-back). """ owner_is_anu = is_anu_key(candidate_owner_key, anu_keys) owner_is_executor = ( bool(candidate_owner_key) and candidate_owner_key == executor_key ) if ( candidate_session_is_executor_self or owner_is_executor or not owner_is_anu ): return WritebackBindingGuardResult( schema=GUARD_SCHEMA, verdict=FAIL, classification=WRITEBACK_SELF_CHAIN_NOT_COMPLETED, task_id=task_id, conflict=True, completed_acknowledged=False, conflicting_fields=["owner_key/self_session"], reasons=[ "self-chain write-back (owner key == executor key / not an " "independent ANU key / executor self-session) — NOT " "acknowledged as completed; recorded, never a silent skip " "(회장 §5.E)." ], ) base = audit_writeback_binding_conflict( history, task_id=task_id, dispatch_id=dispatch_id, chat_id=chat_id, normal_collector_cron_id=normal_collector_cron_id, candidate_role=candidate_role, candidate_fallback_cron_id=candidate_fallback_cron_id, ) if base.classification == WRITEBACK_BINDING_CONFLICT: return WritebackBindingGuardResult( schema=GUARD_SCHEMA, verdict=FAIL, classification=WRITEBACK_BINDING_CONFLICT, task_id=task_id, conflict=True, completed_acknowledged=False, conflicting_fields=list(base.conflicting_fields), reasons=list(base.reasons) + [ "binding conflict is RECORDED as WRITEBACK_BINDING_CONFLICT " "— silent idempotent skip is FORBIDDEN (§5.E / regression 19)." ], ) if base.classification == WRITEBACK_IDEMPOTENT_SKIP: return WritebackBindingGuardResult( schema=GUARD_SCHEMA, verdict=PASS, classification=WRITEBACK_IDEMPOTENT_SKIP, task_id=task_id, conflict=False, completed_acknowledged=True, conflicting_fields=[], reasons=list(base.reasons) + [ "valid duplicate write-back (identical idempotency key AND " "identical role/fallback binding, owner=independent ANU) — " "idempotent SKIP only (§5.E / regression 20)." ], ) return WritebackBindingGuardResult( schema=GUARD_SCHEMA, verdict=PASS, classification=WRITEBACK_NO_CONFLICT, task_id=task_id, conflict=False, completed_acknowledged=True, conflicting_fields=[], reasons=list(base.reasons) + ["normal new write-back, owner=independent ANU (no conflict)."], ) __all__ = [ "GUARD_SCHEMA", "PASS", "FAIL", "WRITEBACK_BINDING_CONFLICT", "WRITEBACK_IDEMPOTENT_SKIP", "WRITEBACK_NO_CONFLICT", "WRITEBACK_SELF_CHAIN_NOT_COMPLETED", "WRITEBACK_OWNER_KEY_CONFLICT", "WritebackBindingGuardResult", "guard_writeback_binding", ]