# -*- coding: utf-8 -*- """task-2553+49 AUTHORITATIVE — authoritative verdict selector real-entrypoint regression (§8 reg 13~18, 24, 27 + §5.D fixtures). self-chain verdict 영구 비권위 · independent ANU verdict 만 authoritative · dead-man DUPLICATE_CALLBACK_IGNORED 는 authoritative verdict 존재 시에만 유효 · independent ANU 없이 self-chain 만으로 PASS 확정 = FAIL. 회장 §5 중요: 실 selector 를 ``dispatch.core.select_runtime_authoritative_ verdict`` (REAL dispatch entrypoint 재노출) 로 직접 호출. mock-only 금지. """ from __future__ import annotations import importlib.util import json import subprocess import sys import unittest from pathlib import Path _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 _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") _sel = _load("anu_v3.authoritative_verdict_selector", "anu_v3/authoritative_verdict_selector.py") _load("anu_v3.self_collector_guard", "anu_v3/self_collector_guard.py") _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") # real selector entrypoint (dispatch.core re-exports this verbatim). select_runtime_authoritative_verdict = ( _grd.select_runtime_authoritative_verdict ) VerdictRecord = _sel.VerdictRecord select_authoritative_verdict = _sel.select_authoritative_verdict ANU_KEY = "c119085addb0f8b7" EXEC_KEY = "0b94683120a691cf" # dev3 self key (the +47 violator) KEYS = (ANU_KEY,) FX_SELF = ( _ROOT / "memory" / "fixtures" / "task-2553plus47.self-chain-callback-violation.json" ) FX_ANU = ( _ROOT / "memory" / "fixtures" / "task-2553plus47.independent-anu-verdict.json" ) FX_SELF_DISPATCH = ( _ROOT / "memory" / "fixtures" / "task-2553plus48.self-dispatch-violation.json" ) def _self_rec(kind="collector_result", verdict="PASS"): return VerdictRecord( kind=kind, verdict=verdict, task_id="task-2553+47", executor_key=EXEC_KEY, collector_key=EXEC_KEY, collector_role="ANU", session_is_executor_self=True, ) def _anu_rec(kind="collector_result", verdict="PASS"): return VerdictRecord( kind=kind, verdict=verdict, task_id="task-2553+47", executor_key=EXEC_KEY, collector_key=ANU_KEY, collector_role="ANU", session_is_executor_self=False, ) class SelectorRegression(unittest.TestCase): def test_13_self_chain_only_pending_not_pass(self): r = select_runtime_authoritative_verdict( [_self_rec()], task_id="task-2553+47", anu_keys=KEYS ) self.assertEqual(r.verdict, "FAIL") self.assertEqual(r.classification, "AUTHORITATIVE_VERDICT_PENDING") self.assertIsNone(r.authoritative_verdict) self.assertEqual(r.independent_anu_count, 0) def test_14_independent_anu_selected_authoritative(self): r = select_runtime_authoritative_verdict( [_self_rec(), _anu_rec()], task_id="task-2553+47", anu_keys=KEYS ) self.assertEqual(r.verdict, "PASS") self.assertEqual(r.classification, "AUTHORITATIVE_PASS") self.assertEqual(r.independent_anu_count, 1) self.assertGreaterEqual(r.quarantined_count, 1) def test_15_deadman_duplicate_valid_only_with_authoritative(self): dm = VerdictRecord( kind="deadman_duplicate_callback", verdict="DUPLICATE_CALLBACK_IGNORED", task_id="task-2553+47", executor_key=EXEC_KEY, collector_key=ANU_KEY, collector_role="ANU", ) # dead-man + self-chain only -> NOT valid (no authoritative verdict). only = select_runtime_authoritative_verdict( [_self_rec(), dm], task_id="task-2553+47", anu_keys=KEYS ) self.assertFalse(only.deadman_valid) self.assertTrue(only.deadman_duplicate_seen) self.assertEqual(only.verdict, "FAIL") # dead-man + independent ANU verdict -> valid disposition. withauth = select_runtime_authoritative_verdict( [_self_rec(), dm, _anu_rec()], task_id="task-2553+47", anu_keys=KEYS, ) self.assertTrue(withauth.deadman_valid) self.assertEqual(withauth.classification, "AUTHORITATIVE_PASS") def test_16_plus47_self_chain_fixture_quarantined(self): data = json.loads(FX_SELF.read_text(encoding="utf-8")) si = data["selector_input"] recs = [VerdictRecord(**x) for x in si["records"]] r = select_authoritative_verdict( recs, task_id=si["task_id"], anu_keys=KEYS ) exp = data["expected"] self.assertEqual(r.verdict, exp["verdict"]) self.assertEqual(r.classification, exp["classification"]) self.assertEqual(r.independent_anu_count, exp["independent_anu_count"]) self.assertEqual(r.quarantined_count, exp["quarantined_count"]) def test_17_plus47_independent_anu_fixture_authoritative_pass(self): data = json.loads(FX_ANU.read_text(encoding="utf-8")) si = data["selector_input"] recs = [VerdictRecord(**x) for x in si["records"]] r = select_authoritative_verdict( recs, task_id=si["task_id"], anu_keys=KEYS ) exp = data["expected"] self.assertEqual(r.verdict, exp["verdict"]) self.assertEqual(r.classification, exp["classification"]) self.assertEqual(r.authoritative_verdict, exp["authoritative_verdict"]) self.assertTrue(r.deadman_valid) def test_18_plus48_self_dispatch_fixture(self): # routed through the real self-collector guard entrypoint. from anu_v3.self_collector_guard import ( guard_self_collector_session, guard_self_dispatch, ) data = json.loads(FX_SELF_DISPATCH.read_text(encoding="utf-8")) c = data["cases"] forbid = guard_self_dispatch( **c["dev3_self_dispatch_plus48_forbidden"]["input"] ) self.assertEqual(forbid.verdict, "FAIL") self.assertEqual( forbid.classification, "EXECUTOR_SELF_DISPATCH_FORBIDDEN" ) ok = guard_self_dispatch( **c["anu_session_dispatch_plus48_ok"]["input"] ) self.assertEqual(ok.verdict, "PASS") sc = guard_self_collector_session( **c["dev3_self_collector_plus48_forbidden"]["input"] ) self.assertEqual(sc.verdict, "FAIL") self.assertEqual(sc.classification, "SELF_COLLECTOR_FORBIDDEN") def test_24_selector_excludes_self_chain_even_if_claimed_independent(self): # a self-session that LIES (claimed_origin=independent_anu) is still # quarantined — owner identity is authoritative, not the claim. liar = VerdictRecord( kind="collector_result", verdict="PASS", task_id="task-2553+47", executor_key=EXEC_KEY, collector_key=EXEC_KEY, collector_role="ANU", session_is_executor_self=True, claimed_origin="independent_anu", ) r = select_runtime_authoritative_verdict( [liar], task_id="task-2553+47", anu_keys=KEYS ) self.assertEqual(r.classification, "AUTHORITATIVE_VERDICT_PENDING") self.assertEqual(r.independent_anu_count, 0) def test_independent_fail_outranks_self_chain_pass(self): # independent ANU FAIL must win over a self-chain PASS (fail-closed). r = select_runtime_authoritative_verdict( [_self_rec(verdict="PASS"), _anu_rec(verdict="FAIL")], task_id="task-2553+47", anu_keys=KEYS, ) self.assertEqual(r.authoritative_verdict, "FAIL") self.assertEqual(r.verdict, "FAIL") def test_anu_keys_unresolvable_holds(self): r = select_authoritative_verdict( [_anu_rec()], task_id="t", anu_keys=(), anu_keys_resolvable=False, ) self.assertEqual(r.verdict, "HOLD_FOR_CHAIR") class CancelOnSuccessNoRegression(unittest.TestCase): """§8 reg 27 — +45/+48 cancel-on-success 무회귀 (실 suite 실행).""" def test_27_plus45_plus48_no_regression(self): for suite in ( "tests/regression/" "test_cancel_on_success_live_wiring_2553plus45.py", "tests/regression/" "test_cancel_on_success_live_e2e_2553plus48.py", ): p = subprocess.run( [sys.executable, "-m", "pytest", "-q", str(_ROOT / suite)], cwd=str(_ROOT), capture_output=True, text=True, timeout=600, ) self.assertEqual( p.returncode, 0, f"no-regression FAILED for {suite}\n{p.stdout[-2000:]}", ) if __name__ == "__main__": unittest.main()