# -*- coding: utf-8 -*- """dispatch.callback_owner_enforcer — executor self-collector / self-dispatch / self-adjudication 구조적 차단 + write-back role/fallback binding conflict 감사. task-2553+49 (MICRO-HARDENING, 회장 결정 — 코드/파일 자동화). Problem (봉쇄 ref: memory/events/task-2553-selfcollector-violation-containment -decision_260518.json): task-2553+47 dispatch 시 ANU 가 normal callback 을 독립 ANU key 로 발사하도록 못박지 않아, executor(dev3)가 자기 key 채널로 normal/fallback callback 을 등록하고 그 자가세션이 +47 자가 회수·자가 Codex audit·자가 adjudication·+48 자가 dispatch 를 수행했다. 분리원칙·collector 독립성·executor 신규 dispatch 0·ANU 단독 dispatch 권한 4중 위반. 근본원인 = ANU orchestration 설계 결함(ANU-key 소유 핀 누락). This standalone module encodes the missing fail-closed pins as executable code so a future executor session *cannot* own its own completion collector, self-adjudicate, or self-dispatch (회장 §1/§2): * ``enforce_callback_owner`` — normal/fallback callback owner MUST be an independent ANU key. executor_key == collector_key, collector owner == executor key, collector_role != ANU, prompt-says-ANU-but-owner-is-executor -> FAIL (§2 / regression 2~6). * ``assert_not_self_adjudication`` — executor self session running a Codex audit / ANU-Codex adjudication -> FAIL (§2 / regression 7). * ``assert_not_self_dispatch`` — executor self session dispatching a follow-up task -> FAIL (§2 / regression 8). * ``audit_writeback_binding_conflict`` — the +47 idempotency key was LOW: it keyed only on (dispatch_id, chat_id, normal_collector_cron_id) and ignored role/fallback binding. This audits role/fallback binding conflict over the +44 read-only ledger and emits ``WRITEBACK_BINDING_CONFLICT`` instead of a silent idempotent skip (§2 line 20/21). Standalone, zero-mutation (회장 §5 / byte-0 우선): imports/edits ZERO byte-0-pinned tracked module. ``anu_v3.callback_4tuple_registry`` and ``dispatch.executor_completion_contract`` are byte-0 (pinned by the existing +44/+48 regression FROZEN_SHA256 autouse invariant); they are imported read-only / lazily and NEVER mutated. The §2 4-tuple owner fields (executor_key / collector_key / collector_owner_key / collector_role) are carried by THIS module's own ``CallbackOwner4Tuple`` record + schema + decision artifact — NOT by mutating the frozen +44 legacy record (9-R / byte-0 우선 carve-out, +42/+43/+45 선례 동형). Layer A / NO-CRON (9-R.1): this module performs ZERO cron register/remove, ZERO dispatch, ZERO merge, ZERO ``cokacdir`` / ``subprocess`` exec. It only *validates* owner/key bindings and *records* a verdict. The executor's own normal completion callback (fired by an authorized session with the ANU key, §10) is a designed lifecycle signal — NOT an ad-hoc registry cron — and the firing happens outside this module; the module never execs it. """ from __future__ import annotations from dataclasses import dataclass, field from typing import List, Optional, Sequence, Tuple ENFORCER_SCHEMA = "dispatch.callback_owner_enforcement.v1" OWNER_4TUPLE_SCHEMA = "dispatch.callback_owner_4tuple.v1" # ── verdicts ───────────────────────────────────────────────────────────────── PASS = "PASS" FAIL = "FAIL" HOLD = "HOLD_FOR_CHAIR" # ── §3 classifications (회장 verbatim, 8종) ────────────────────────────────── SELF_COLLECTOR_FORBIDDEN = "SELF_COLLECTOR_FORBIDDEN" EXECUTOR_SELF_ADJUDICATION_FORBIDDEN = "EXECUTOR_SELF_ADJUDICATION_FORBIDDEN" SELF_DISPATCH_FORBIDDEN = "SELF_DISPATCH_FORBIDDEN" CALLBACK_OWNER_MISMATCH = "CALLBACK_OWNER_MISMATCH" CALLBACK_COLLECTOR_NOT_ANU = "CALLBACK_COLLECTOR_NOT_ANU" CALLBACK_4TUPLE_INVALID = "CALLBACK_4TUPLE_INVALID" DISPATCH_PATH_BYPASSED_CONTRACT = "DISPATCH_PATH_BYPASSED_CONTRACT" WRITEBACK_BINDING_CONFLICT = "WRITEBACK_BINDING_CONFLICT" ALL_CLASSIFICATIONS: Tuple[str, ...] = ( SELF_COLLECTOR_FORBIDDEN, EXECUTOR_SELF_ADJUDICATION_FORBIDDEN, SELF_DISPATCH_FORBIDDEN, CALLBACK_OWNER_MISMATCH, CALLBACK_COLLECTOR_NOT_ANU, CALLBACK_4TUPLE_INVALID, DISPATCH_PATH_BYPASSED_CONTRACT, WRITEBACK_BINDING_CONFLICT, ) # write-back binding-conflict audit verdicts. WRITEBACK_NO_CONFLICT = "WRITEBACK_NO_CONFLICT" WRITEBACK_IDEMPOTENT_SKIP = "WRITEBACK_IDEMPOTENT_SKIP" # ── collector role + key policy ────────────────────────────────────────────── # The collector MUST run under an independent ANU collector session. COLLECTOR_ROLE_ANU = "ANU" # The independent ANU key (회장 §10). The collector / normal+fallback callback # owner MUST be this (or another configured ANU key) and MUST NOT be the # executor self key. Overridable via the ``anu_keys`` argument so nothing is # hard-wired beyond the chairman-pinned default (config-driven friendly). ANU_KEY_2553 = "c119085addb0f8b7" DEFAULT_ANU_KEYS = frozenset({ANU_KEY_2553}) # The dev6 페룬 executor self key (회장 §10 — 발사 절대 금지 대상). Documented # here as the canonical counter-example; the rule is structural (owner must be # an ANU key AND must differ from the executor key), not a single denylist. EXECUTOR_SELF_KEY_EXAMPLE = "1e41a2324a3ccdd0" # Recognised (gated) dispatch entry paths. A callback owner contract presented # from an unrecognised path is a cron-direct contract bypass (§3 / # DISPATCH_PATH_BYPASSED_CONTRACT). Mirrors dispatch.cron_dispatch_guard. PATH_DISPATCH_PY = "dispatch.core.main" PATH_CRON_DIRECT = "cokacdir_cron_direct" KNOWN_ENTRY_PATHS = frozenset({PATH_DISPATCH_PY, PATH_CRON_DIRECT}) def is_anu_key(key: Optional[str], anu_keys: Sequence[str]) -> bool: """True iff ``key`` is a non-empty configured independent ANU key.""" return bool(key) and key in set(anu_keys) # ── §2 owner 4-tuple (executor_key/collector_key/collector_owner_key/role) ──── @dataclass(frozen=True) class CallbackOwner4Tuple: """The §2/§10 owner-bound callback 4-tuple. Base 4-tuple {task_id, dispatch_cron_id, normal_collector_cron_id*, fallback_callback_cron_id} + the owner binding the +44 frozen legacy record cannot carry (executor_key / collector_key / collector_owner_key / collector_role=ANU / normal_collector_cron_id / fallback_cron_id / normal_collector_cron_id_role). normal_collector_cron_id is MANDATORY. """ task_id: str dispatch_cron_id: str normal_collector_cron_id: Optional[str] fallback_callback_cron_id: Optional[str] executor_key: str collector_key: str collector_owner_key: str collector_role: str chat_id: str = "" normal_collector_cron_id_role: str = COLLECTOR_ROLE_ANU fallback_cron_id_role: str = COLLECTOR_ROLE_ANU no_fallback: bool = False def identity(self) -> dict: return { "task_id": self.task_id, "dispatch_cron_id": self.dispatch_cron_id, "normal_collector_cron_id": self.normal_collector_cron_id, "fallback_callback_cron_id": self.fallback_callback_cron_id, "executor_key": self.executor_key, "collector_key": self.collector_key, "collector_owner_key": self.collector_owner_key, "collector_role": self.collector_role, } def to_json(self) -> dict: d = dict(self.identity()) d.update( { "schema": OWNER_4TUPLE_SCHEMA, "chat_id": self.chat_id, "normal_collector_cron_id_role": ( self.normal_collector_cron_id_role ), "fallback_cron_id_role": self.fallback_cron_id_role, "no_fallback": self.no_fallback, } ) return d @dataclass class CallbackOwnerEnforcementResult: schema: str verdict: str # PASS | FAIL | HOLD_FOR_CHAIR classifications: List[str] task_id: str executor_key: str collector_key: str collector_owner_key: str collector_role: str normal_collector_cron_id: Optional[str] fallback_callback_cron_id: Optional[str] dispatch_cron_id: str normal_collector_cron_id_role: str fallback_cron_id_role: str entry_path: str prompt_claims_anu_collector: bool owner_is_independent_anu: 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, "collector_key": self.collector_key, "collector_owner_key": self.collector_owner_key, "collector_role": self.collector_role, "normal_collector_cron_id": self.normal_collector_cron_id, "fallback_callback_cron_id": self.fallback_callback_cron_id, "dispatch_cron_id": self.dispatch_cron_id, "normal_collector_cron_id_role": ( self.normal_collector_cron_id_role ), "fallback_cron_id_role": self.fallback_cron_id_role, "entry_path": self.entry_path, "prompt_claims_anu_collector": self.prompt_claims_anu_collector, "owner_is_independent_anu": self.owner_is_independent_anu, "reasons": list(self.reasons), } def enforce_callback_owner( *, task_id: str, executor_key: str, collector_key: str, collector_owner_key: Optional[str] = None, collector_role: str, normal_collector_cron_id: Optional[str], fallback_callback_cron_id: Optional[str], dispatch_cron_id: str, chat_id: str = "", prompt_claims_anu_collector: bool = False, entry_path: str = PATH_CRON_DIRECT, anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS), no_fallback: bool = False, normal_collector_cron_id_role: str = COLLECTOR_ROLE_ANU, fallback_cron_id_role: str = COLLECTOR_ROLE_ANU, anu_keys_resolvable: bool = True, ) -> CallbackOwnerEnforcementResult: """Fail-closed gate: normal/fallback callback owner MUST be an independent ANU key (회장 §2/§8/§10). Verdict logic (every violating condition is recorded — NOT silently dropped; §2 line 21): * ANU key set unresolvable in this env -> HOLD_FOR_CHAIR (§6 — the ONLY conditional escalation; un-codeable sub-control) * entry_path unrecognised -> DISPATCH_PATH_BYPASSED_ CONTRACT (cron-direct bypass) * collector_role != "ANU" -> CALLBACK_COLLECTOR_NOT_ANU (regression 5) * executor_key == collector_key -> SELF_COLLECTOR_FORBIDDEN (regression 4) * effective owner == executor key -> SELF_COLLECTOR_FORBIDDEN (regression 2/3 — normal & fallback self callback) * prompt says ANU collector but owner is the executor key / not an ANU key -> CALLBACK_OWNER_MISMATCH (regression 6) * owner not an independent ANU key -> CALLBACK_OWNER_MISMATCH * 4-tuple invalid (task_id/dispatch_cron /normal_collector_cron_id mandatory; fallback unless no_fallback) -> CALLBACK_4TUPLE_INVALID * else -> PASS (owner is an independent ANU key, regression 1). Layer A: pure validation. ZERO cron / dispatch / subprocess / cokacdir. """ owner_key = collector_owner_key or collector_key or "" reasons: List[str] = [] # §6 conditional escalation: if the ANU key set is genuinely # un-resolvable in this environment, the owner pin cannot be code- # enforced -> HOLD_FOR_CHAIR (NOT a blanket pre-declaration; 9-R.2). if not anu_keys_resolvable or not list(anu_keys): return CallbackOwnerEnforcementResult( schema=ENFORCER_SCHEMA, verdict=HOLD, classifications=[], task_id=task_id, executor_key=executor_key, collector_key=collector_key, collector_owner_key=owner_key, collector_role=collector_role, normal_collector_cron_id=normal_collector_cron_id, fallback_callback_cron_id=fallback_callback_cron_id, dispatch_cron_id=dispatch_cron_id, normal_collector_cron_id_role=normal_collector_cron_id_role, fallback_cron_id_role=fallback_cron_id_role, entry_path=entry_path, prompt_claims_anu_collector=prompt_claims_anu_collector, owner_is_independent_anu=False, reasons=[ "ANU key set unresolvable in this environment — the " "callback owner=ANU pin cannot be code-enforced here " "(§6 HOLD_FOR_CHAIR; the only conditional escalation, " "9-R.2). NO silent pass." ], ) cls: List[str] = [] if entry_path not in KNOWN_ENTRY_PATHS: cls.append(DISPATCH_PATH_BYPASSED_CONTRACT) reasons.append( f"entry_path {entry_path!r} not a recognised gated dispatch " "path — a cron-direct contract bypass (§3 / 9-R.1). The ANU " "control layer (prompt gen · 4-tuple · post-reg owner cross-" "check) must fail-closed here." ) if collector_role != COLLECTOR_ROLE_ANU: cls.append(CALLBACK_COLLECTOR_NOT_ANU) reasons.append( f"collector_role={collector_role!r} != 'ANU' — the completion " "collector MUST run as an independent ANU collector session " "(§2 / regression 5)." ) if executor_key and collector_key and executor_key == collector_key: cls.append(SELF_COLLECTOR_FORBIDDEN) reasons.append( "executor_key == collector_key — executor self-collector is " "structurally forbidden (§2 / regression 4)." ) owner_is_executor = bool(owner_key) and owner_key == executor_key if owner_is_executor: cls.append(SELF_COLLECTOR_FORBIDDEN) reasons.append( "normal/fallback callback owner key == executor self key — " "executor self-callback (normal & fallback) is structurally " "forbidden; owner MUST be an independent ANU key (§2/§10 / " "regression 2/3)." ) owner_is_anu = is_anu_key(owner_key, anu_keys) if prompt_claims_anu_collector and (owner_is_executor or not owner_is_anu): cls.append(CALLBACK_OWNER_MISMATCH) reasons.append( "prompt declares an 'ANU Result Collector' but the schedule " "owner key is the executor key / not an ANU key — invalid; " "owner identity, not prompt text, is authoritative (§2 / " "regression 6)." ) if not owner_is_anu and not owner_is_executor: cls.append(CALLBACK_OWNER_MISMATCH) reasons.append( f"callback owner key {owner_key!r} is not a configured " "independent ANU key — owner mismatch (§2/§8)." ) tuple_reasons = _validate_owner_4tuple( task_id=task_id, dispatch_cron_id=dispatch_cron_id, normal_collector_cron_id=normal_collector_cron_id, fallback_callback_cron_id=fallback_callback_cron_id, no_fallback=no_fallback, ) if tuple_reasons: cls.append(CALLBACK_4TUPLE_INVALID) reasons.extend(tuple_reasons) # de-dup classifications, preserve first-seen order (primary first). 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 owner_is_independent_anu = ( owner_is_anu and not owner_is_executor and collector_role == COLLECTOR_ROLE_ANU ) if verdict == PASS: reasons.append( "callback owner is an independent ANU key (collector_role=ANU, " "owner != executor key, 4-tuple valid) — executor self-" "collector structurally impossible from here (§2 / regression 1)." ) return CallbackOwnerEnforcementResult( schema=ENFORCER_SCHEMA, verdict=verdict, classifications=ordered, task_id=task_id, executor_key=executor_key, collector_key=collector_key, collector_owner_key=owner_key, collector_role=collector_role, normal_collector_cron_id=normal_collector_cron_id, fallback_callback_cron_id=fallback_callback_cron_id, dispatch_cron_id=dispatch_cron_id, normal_collector_cron_id_role=normal_collector_cron_id_role, fallback_cron_id_role=fallback_cron_id_role, entry_path=entry_path, prompt_claims_anu_collector=prompt_claims_anu_collector, owner_is_independent_anu=owner_is_independent_anu, reasons=reasons, ) def _validate_owner_4tuple( *, task_id: str, dispatch_cron_id: str, normal_collector_cron_id: Optional[str], fallback_callback_cron_id: Optional[str], no_fallback: bool, ) -> List[str]: """4-tuple invalidity reasons (mirrors the +32/+44 mandatory rule).""" r: List[str] = [] if not task_id: r.append("task_id empty (CALLBACK_4TUPLE_INVALID)") if not dispatch_cron_id: r.append("dispatch_cron_id empty (CALLBACK_4TUPLE_INVALID)") if not normal_collector_cron_id: r.append( "normal_collector_cron_id missing — MANDATORY independent-ANU " "normal completion callback lifecycle signal; NO-CRON does NOT " "exempt this (CALLBACK_4TUPLE_INVALID)." ) if not fallback_callback_cron_id and not no_fallback: r.append( "fallback_callback_cron_id missing and no explicit no_fallback " "contract (safety path, CALLBACK_4TUPLE_INVALID)." ) return r # ── self-adjudication / self-dispatch structural guards (§2 / reg 7,8) ─────── @dataclass class SelfActionGuardResult: schema: str verdict: str classification: Optional[str] is_executor_self_session: 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, "is_executor_self_session": self.is_executor_self_session, "reasons": list(self.reasons), } def _is_self_session( executor_key: str, actor_key: Optional[str], is_executor_self_session: Optional[bool], ) -> bool: # fail-closed (task-2553+49 Codex HIGH 해소): the key-derived # self-session signal (actor_key == executor_key) is AUTHORITATIVE and # can NEVER be downgraded by a caller-supplied hint. The explicit flag # may only ESCALATE (declare a self-session even when keys do not prove # it), never relax it — otherwise an executor-self path could pass # is_executor_self_session=False to bypass the structural self-adjud / # self-dispatch block (caller-trust → structural). key_derived = bool(actor_key) and actor_key == executor_key explicit = ( bool(is_executor_self_session) if is_executor_self_session is not None else False ) return key_derived or explicit def assert_not_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, ) -> SelfActionGuardResult: """executor self session 이 Codex audit / ANU-Codex adjudication 을 수행 하면 invalid (§2 / regression 7). Codex audit · adjudication · 후속 회수·검증 은 반드시 독립 ANU collector 세션이 수행한다. executor self session 이 그것을 하면 FAIL. """ self_session = _is_self_session( executor_key, actor_key, is_executor_self_session ) if self_session and (is_codex_audit or is_adjudication): what = [] if is_codex_audit: what.append("Codex audit") if is_adjudication: what.append("ANU-Codex adjudication") return SelfActionGuardResult( schema=ENFORCER_SCHEMA, verdict=FAIL, classification=EXECUTOR_SELF_ADJUDICATION_FORBIDDEN, is_executor_self_session=True, reasons=[ f"executor self session performing {' + '.join(what)} — " "invalid; this MUST be done by the independent ANU " "collector session, never the executor self session " "(§2 / regression 7)." ], ) return SelfActionGuardResult( schema=ENFORCER_SCHEMA, verdict=PASS, classification=None, is_executor_self_session=self_session, reasons=[ "no executor self-adjudication (Codex audit/adjudication is " "not run by the executor self session)." ], ) def assert_not_self_dispatch( *, executor_key: str, actor_key: Optional[str] = None, is_followup_dispatch: bool = False, is_executor_self_session: Optional[bool] = None, ) -> SelfActionGuardResult: """executor self session 이 후속 task dispatch 를 수행하면 invalid (§2 / regression 8). 신규 dispatch·delegation 은 독립 ANU 세션만.""" self_session = _is_self_session( executor_key, actor_key, is_executor_self_session ) if self_session and is_followup_dispatch: return SelfActionGuardResult( schema=ENFORCER_SCHEMA, verdict=FAIL, classification=SELF_DISPATCH_FORBIDDEN, is_executor_self_session=True, reasons=[ "executor self session performing a follow-up task " "dispatch/delegation — invalid; executor 자기작업중 신규 " "dispatch·delegation 0, only the independent ANU session " "may dispatch (§2/§10 / regression 8)." ], ) return SelfActionGuardResult( schema=ENFORCER_SCHEMA, verdict=PASS, classification=None, is_executor_self_session=self_session, reasons=["no executor self-dispatch."], ) # ── +47 idempotency LOW 보강: role/fallback binding conflict 감사 ──────────── @dataclass class WritebackBindingAuditResult: schema: str verdict: str # PASS (no-conflict / idempotent) | FAIL (conflict) classification: str # WRITEBACK_NO_CONFLICT | _IDEMPOTENT_SKIP | _CONFLICT task_id: str conflict: bool matched_idempotency_key: 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, "matched_idempotency_key": self.matched_idempotency_key, "conflicting_fields": list(self.conflicting_fields), "reasons": list(self.reasons), } def audit_writeback_binding_conflict( 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], ) -> WritebackBindingAuditResult: """+47 idempotency key 보강 (§2 line 20/21). The +47 ``write_back_completed`` idempotency key is ``(dispatch_id, chat_id, normal_collector_cron_id)`` ONLY — it ignores ``role`` / ``fallback_callback_cron_id``. So a SECOND write-back that shares the collector identity but carries a DIFFERENT role/fallback binding is silently treated as an idempotent skip (LOW). This audit runs over the +44 read-only ledger ``history`` (the +44 module stays byte-0; history is the value of ``registry.history_for(task_id)``): * a COMPLETED record with the SAME idempotency key AND the SAME role/fallback binding -> ``WRITEBACK_IDEMPOTENT_SKIP`` (PASS, regression 12 — a true duplicate). * a COMPLETED record with the SAME idempotency key but a DIFFERENT role / fallback binding -> ``WRITEBACK_BINDING_CONFLICT`` (FAIL, regression 11) — RECORDED, never a silent skip (§2 line 21). * no idempotency-key match -> ``WRITEBACK_NO_CONFLICT`` (PASS — a normal new append). Read-only. ZERO cron / dispatch / subprocess. The +44 registry is NOT escalated to a merge/dispatch executor (§5/§6). """ matched = False for r in history: try: status = getattr(r, "status", None) r_dispatch_id = getattr(r, "dispatch_id", None) r_chat_id = getattr(r, "chat_id", None) r_ncc = getattr(r, "normal_collector_cron_id", None) r_role = getattr(r, "role", None) r_fb = getattr(r, "fallback_callback_cron_id", None) except Exception: # pragma: no cover - fail-safe on odd record continue if status != "COMPLETED": continue # +47 idempotency key (collector identity). if ( r_dispatch_id == dispatch_id and str(r_chat_id) == str(chat_id) and r_ncc == normal_collector_cron_id ): matched = True fields: List[str] = [] if r_role != candidate_role: fields.append( f"role(record={r_role!r} vs candidate=" f"{candidate_role!r})" ) if r_fb != candidate_fallback_cron_id: fields.append( f"fallback_callback_cron_id(record={r_fb!r} vs " f"candidate={candidate_fallback_cron_id!r})" ) if fields: return WritebackBindingAuditResult( schema=ENFORCER_SCHEMA, verdict=FAIL, classification=WRITEBACK_BINDING_CONFLICT, task_id=task_id, conflict=True, matched_idempotency_key=True, conflicting_fields=fields, reasons=[ "an existing COMPLETED record shares the +47 " "idempotency key (dispatch_id, chat_id, " "normal_collector_cron_id) but has a DIFFERENT " "role/fallback binding — WRITEBACK_BINDING_CONFLICT. " "This is RECORDED, NOT a silent idempotent skip " "(§2 line 20/21 / regression 11). Conflicting: " + "; ".join(fields) ], ) if matched: return WritebackBindingAuditResult( schema=ENFORCER_SCHEMA, verdict=PASS, classification=WRITEBACK_IDEMPOTENT_SKIP, task_id=task_id, conflict=False, matched_idempotency_key=True, conflicting_fields=[], reasons=[ "COMPLETED record with identical idempotency key AND " "identical role/fallback binding — a true duplicate; " "idempotent SKIP, no duplicate append (regression 12)." ], ) return WritebackBindingAuditResult( schema=ENFORCER_SCHEMA, verdict=PASS, classification=WRITEBACK_NO_CONFLICT, task_id=task_id, conflict=False, matched_idempotency_key=False, conflicting_fields=[], reasons=[ "no COMPLETED record matches the +47 idempotency key — a " "normal new write-back (no conflict)." ], ) __all__ = [ "ENFORCER_SCHEMA", "OWNER_4TUPLE_SCHEMA", "PASS", "FAIL", "HOLD", "SELF_COLLECTOR_FORBIDDEN", "EXECUTOR_SELF_ADJUDICATION_FORBIDDEN", "SELF_DISPATCH_FORBIDDEN", "CALLBACK_OWNER_MISMATCH", "CALLBACK_COLLECTOR_NOT_ANU", "CALLBACK_4TUPLE_INVALID", "DISPATCH_PATH_BYPASSED_CONTRACT", "WRITEBACK_BINDING_CONFLICT", "WRITEBACK_NO_CONFLICT", "WRITEBACK_IDEMPOTENT_SKIP", "ALL_CLASSIFICATIONS", "COLLECTOR_ROLE_ANU", "ANU_KEY_2553", "DEFAULT_ANU_KEYS", "EXECUTOR_SELF_KEY_EXAMPLE", "PATH_DISPATCH_PY", "PATH_CRON_DIRECT", "KNOWN_ENTRY_PATHS", "is_anu_key", "CallbackOwner4Tuple", "CallbackOwnerEnforcementResult", "enforce_callback_owner", "SelfActionGuardResult", "assert_not_self_adjudication", "assert_not_self_dispatch", "WritebackBindingAuditResult", "audit_writeback_binding_conflict", ]