# -*- coding: utf-8 -*- """task-2553+49 AUTHORITATIVE — self-collector / self-adjudication / self-dispatch + write-back binding conflict runtime regression (§8 reg 11, 12, 19, 20, 28, 29). 회장 §5 중요: 실 runtime guard entrypoint 직접 호출. mock-only 금지. """ from __future__ import annotations import importlib.util import sys import unittest from dataclasses import dataclass from pathlib import Path from typing import Optional _ROOT = Path(__file__).resolve().parent.parent.parent if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) def _load(modname: str, relpath: str): if modname in sys.modules: return sys.modules[modname] spec = importlib.util.spec_from_file_location(modname, _ROOT / relpath) assert spec is not None and spec.loader is not None mod = importlib.util.module_from_spec(spec) sys.modules[modname] = mod spec.loader.exec_module(mod) return mod _ecc = _load("dispatch.executor_completion_contract", "dispatch/executor_completion_contract.py") _load("dispatch.spec_template_validator", "dispatch/spec_template_validator.py") _load("anu_v3.callback_4tuple_registry", "anu_v3/callback_4tuple_registry.py") _load("dispatch.callback_owner_enforcer", "dispatch/callback_owner_enforcer.py") _load("dispatch.normal_fallback_callback_helper", "dispatch/normal_fallback_callback_helper.py") _load("anu_v3.callback_owner_validator", "anu_v3/callback_owner_validator.py") _load("anu_v3.authoritative_verdict_selector", "anu_v3/authoritative_verdict_selector.py") _scg = _load("anu_v3.self_collector_guard", "anu_v3/self_collector_guard.py") _wbg = _load("anu_v3.writeback_binding_conflict_guard", "anu_v3/writeback_binding_conflict_guard.py") _grd = _load("dispatch.cron_dispatch_guard", "dispatch/cron_dispatch_guard.py") guard_self_adjudication = _scg.guard_self_adjudication guard_self_dispatch = _scg.guard_self_dispatch guard_writeback_binding = _wbg.guard_writeback_binding # real dispatch-path runtime composite gate (dispatch.core re-exports it). guard_dispatch_runtime_authoritative = ( _grd.guard_dispatch_runtime_authoritative ) Callback4Tuple = _ecc.Callback4Tuple ANU_KEY = "c119085addb0f8b7" EXEC_KEY = "0b94683120a691cf" KEYS = (ANU_KEY,) CHAT = "6937032012" _CLAUSE = ( "[EXECUTOR COMPLETION CALLBACK — MANDATORY] executor 는 ANU 에 normal " "completion callback cron 을 반드시 발사해야 한다." ) @dataclass class _LedgerRec: status: str dispatch_id: str chat_id: str normal_collector_cron_id: Optional[str] role: str fallback_callback_cron_id: Optional[str] class SelfActionRuntimeRegression(unittest.TestCase): def test_11_executor_self_codex_adjudication_fail(self): r = guard_self_adjudication( executor_key=EXEC_KEY, actor_key=EXEC_KEY, is_codex_audit=True, is_adjudication=True, ) self.assertEqual(r.verdict, "FAIL") self.assertEqual( r.classification, "EXECUTOR_SELF_ADJUDICATION_FORBIDDEN" ) # independent ANU actor -> PASS. ok = guard_self_adjudication( executor_key=EXEC_KEY, actor_key=ANU_KEY, is_codex_audit=True, is_adjudication=True, ) self.assertEqual(ok.verdict, "PASS") def test_12_executor_self_followup_dispatch_fail(self): r = guard_self_dispatch( executor_key=EXEC_KEY, actor_key=EXEC_KEY, is_followup_dispatch=True, ) self.assertEqual(r.verdict, "FAIL") self.assertEqual( r.classification, "EXECUTOR_SELF_DISPATCH_FORBIDDEN" ) ok = guard_self_dispatch( executor_key=EXEC_KEY, actor_key=ANU_KEY, is_followup_dispatch=True, ) self.assertEqual(ok.verdict, "PASS") def test_self_session_flag_cannot_relax_key_signal(self): # key-derived self-session is authoritative; passing # is_executor_self_session=False MUST NOT bypass the block. r = guard_self_dispatch( executor_key=EXEC_KEY, actor_key=EXEC_KEY, is_followup_dispatch=True, is_executor_self_session=False, ) self.assertEqual(r.verdict, "FAIL") class WritebackBindingRuntimeRegression(unittest.TestCase): def _hist(self, role="executor", fb="FB-ANU"): return [ _LedgerRec( status="COMPLETED", dispatch_id="DSP", chat_id=CHAT, normal_collector_cron_id="NC", role=role, fallback_callback_cron_id=fb, ) ] def test_19_role_fallback_binding_mismatch_conflict(self): r = guard_writeback_binding( self._hist(role="executor", fb="FB-ANU"), task_id="task-2553+49", dispatch_id="DSP", chat_id=CHAT, normal_collector_cron_id="NC", candidate_role="fallback", candidate_fallback_cron_id="FB-DIFFERENT", candidate_owner_key=ANU_KEY, executor_key=EXEC_KEY, anu_keys=KEYS, ) self.assertEqual(r.verdict, "FAIL") self.assertEqual(r.classification, "WRITEBACK_BINDING_CONFLICT") self.assertTrue(r.conflicting_fields) # recorded, NOT silent skip self.assertFalse(r.completed_acknowledged) def test_20_valid_duplicate_idempotent_skip(self): r = guard_writeback_binding( self._hist(role="executor", fb="FB-ANU"), task_id="task-2553+49", dispatch_id="DSP", chat_id=CHAT, normal_collector_cron_id="NC", candidate_role="executor", candidate_fallback_cron_id="FB-ANU", candidate_owner_key=ANU_KEY, executor_key=EXEC_KEY, anu_keys=KEYS, ) self.assertEqual(r.verdict, "PASS") self.assertEqual(r.classification, "WRITEBACK_IDEMPOTENT_SKIP") self.assertTrue(r.completed_acknowledged) def test_self_chain_writeback_not_completed(self): # owner == executor self key -> NOT acknowledged completed (§5.E). r = guard_writeback_binding( self._hist(), task_id="task-2553+49", dispatch_id="DSP", chat_id=CHAT, normal_collector_cron_id="NC", candidate_role="executor", candidate_fallback_cron_id="FB-ANU", candidate_owner_key=EXEC_KEY, executor_key=EXEC_KEY, anu_keys=KEYS, ) self.assertEqual(r.verdict, "FAIL") self.assertEqual( r.classification, "WRITEBACK_SELF_CHAIN_NOT_COMPLETED" ) self.assertFalse(r.completed_acknowledged) class SafetyPathInvariants(unittest.TestCase): """§8 reg 28/29 — checkpoint=recovery layer 유지(primary executor 격상 금지) · callback/fallback safety path 유지.""" def test_28_runtime_checkpoint_recovery_layer_unchanged(self): # byte-0 invariant: the recovery-layer module is NOT escalated to a # primary executor by this task (it is untouched / not imported by # the new runtime guards). rl_path = ( _ROOT / "anu_v3" / "runtime_reconcile_checkpoint_recovery_layer.py" ) self.assertTrue(rl_path.is_file()) src = rl_path.read_text(encoding="utf-8") self.assertIn("recovery", src.lower()) # the new guards must not import/escalate the checkpoint as primary. for rel in ( "anu_v3/self_collector_guard.py", "anu_v3/authoritative_verdict_selector.py", "anu_v3/callback_owner_validator.py", "anu_v3/writeback_binding_conflict_guard.py", ): txt = (_ROOT / rel).read_text(encoding="utf-8") self.assertNotIn("primary_executor", txt) self.assertNotIn("runtime_reconcile_checkpoint", txt) def test_29_callback_fallback_safety_path_preserved(self): # fallback callback remains MANDATORY unless an explicit no_fallback # contract: a missing fallback w/o no_fallback still blocks (safety # path NOT removed). from anu_v3.callback_owner_validator import ( validate_callback_owner_runtime, ) r = validate_callback_owner_runtime( task_id="task-2553+49", executor_key=EXEC_KEY, collector_key=ANU_KEY, collector_owner_key=ANU_KEY, collector_role="ANU", normal_collector_cron_id="NC", fallback_callback_cron_id=None, dispatch_cron_id="D", chat_id=CHAT, prompt_claims_anu_collector=True, anu_keys=KEYS, no_fallback=False, ) self.assertEqual(r.verdict, "FAIL") self.assertIn("CALLBACK_4TUPLE_INVALID", r.classifications) # explicit no_fallback contract -> allowed (safety path intact, not # silently removed). ok = validate_callback_owner_runtime( task_id="task-2553+49", executor_key=EXEC_KEY, collector_key=ANU_KEY, collector_owner_key=ANU_KEY, collector_role="ANU", normal_collector_cron_id="NC", fallback_callback_cron_id=None, dispatch_cron_id="D", chat_id=CHAT, prompt_claims_anu_collector=True, anu_keys=KEYS, no_fallback=True, ) self.assertEqual(ok.verdict, "PASS") def test_runtime_authoritative_composite_gate(self): # the real composite dispatch-path gate: executor self-key collector # + executor self-dispatch -> FAIL (structural, all limbs). t = Callback4Tuple( task_id="task-2553+49", dispatch_cron_id="D", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", ) base, owner, sc, sd, final = guard_dispatch_runtime_authoritative( spec_text=_CLAUSE, entry_path="cokacdir_cron_direct", tuple_=t, executor_key=EXEC_KEY, collector_key=EXEC_KEY, collector_owner_key=EXEC_KEY, collector_role="ANU", dispatch_cron_id="D", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", chat_id=CHAT, prompt_claims_anu_collector=True, anu_keys=KEYS, actor_key=EXEC_KEY, is_executor_self_session=True, is_followup_dispatch=True, ) self.assertEqual(final, "FAIL") self.assertEqual(sc.verdict, "FAIL") self.assertEqual(sd.verdict, "FAIL") # fully compliant independent-ANU path -> PASS. base2, owner2, sc2, sd2, final2 = ( guard_dispatch_runtime_authoritative( spec_text=_CLAUSE, entry_path="cokacdir_cron_direct", tuple_=t, executor_key=EXEC_KEY, collector_key=ANU_KEY, collector_owner_key=ANU_KEY, collector_role="ANU", dispatch_cron_id="D", normal_collector_cron_id="NC", fallback_callback_cron_id="FB", chat_id=CHAT, prompt_claims_anu_collector=True, anu_keys=KEYS, actor_key=ANU_KEY, is_followup_dispatch=False, ) ) self.assertEqual(final2, "PASS") if __name__ == "__main__": unittest.main()