# -*- coding: utf-8 -*- """Regression — task-2553+32 EXECUTOR COMPLETION CALLBACK MANDATORY RULE 복원. Covers §6 verbatim (1~15) + invariants (git HEAD/branch equal, frozen anchor / durable v1 / +31 checkpoint original sha equal, NO-CRON token scan, additive no-mutation). The executor normal completion callback is restored as a MANDATORY lifecycle signal in executable code (§1/§4). 9-R.1: zero cron register/remove anywhere; cancel-on-success path only asserted preserved (execute 0). 9-R.2: finalization-authority escalation forbidden while executor self-task lifecycle outputs are REQUIRED. """ import hashlib import importlib.util import json import subprocess import sys import unittest from pathlib import Path _ROOT = Path(__file__).resolve().parents[2] if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) def _load(modname: str, relpath: str): """Hermetic file-path import. pytest places tests/ on sys.path and a shadow package tests/dispatch/ exists, so a bare `import dispatch.*` can resolve to the wrong package depending on collection order. Loading the real workspace files by explicit path is collision-proof and touches no shared config (§8). """ spec = importlib.util.spec_from_file_location(modname, _ROOT / relpath) mod = importlib.util.module_from_spec(spec) sys.modules[modname] = mod spec.loader.exec_module(mod) return mod _ecc = _load( "_p32_executor_completion_contract", "dispatch/executor_completion_contract.py", ) # Pre-seed the canonical dotted name so spec_template_validator's internal # `from dispatch.executor_completion_contract import ...` resolves to the # real workspace module regardless of the tests/dispatch shadow package. sys.modules["dispatch.executor_completion_contract"] = _ecc Callback4Tuple = _ecc.Callback4Tuple ExecutorCloseoutEvidence = _ecc.ExecutorCloseoutEvidence NO_CRON_CORRECTED_DEFINITION = _ecc.NO_CRON_CORRECTED_DEFINITION RESULT_READY_NO_NORMAL_CALLBACK = _ecc.RESULT_READY_NO_NORMAL_CALLBACK NORMAL_COLLECTOR_COMPLETED = _ecc.NORMAL_COLLECTOR_COMPLETED classify_completion = _ecc.classify_completion closeout_is_valid = _ecc.closeout_is_valid is_accepted_normal_lifecycle = _ecc.is_accepted_normal_lifecycle is_executor_completion_callback_a_cron_violation = ( _ecc.is_executor_completion_callback_a_cron_violation ) is_recovery_state = _ecc.is_recovery_state tuple_is_valid = _ecc.tuple_is_valid validate_4tuple = _ecc.validate_4tuple validate_closeout_evidence = _ecc.validate_closeout_evidence _stv = _load( "_p32_spec_template_validator", "dispatch/spec_template_validator.py", ) FAIL = _stv.FAIL HOLD = _stv.HOLD PASS = _stv.PASS inject_completion_callback_clause = _stv.inject_completion_callback_clause spec_has_normal_callback_clause = _stv.spec_has_normal_callback_clause validate_spec = _stv.validate_spec from anu_v3.executor_callback_contract import ( # noqa: E402 callback_is_primary, cancel_on_success_applies_after_normal_success, checkpoint_is_recovery_layer, closeout_authority_violation, CloseoutAuthorityCheck, doctrine_verdict, executor_completion_callback_is_cron_add_violation, fallback_is_safety_path, plus32_executor_removes_any_cron, registry_is_primary_completion_mechanism, ) from anu_v3.runtime_reconcile_checkpoint_recovery_layer import ( # noqa: E402 assert_checkpoint_is_recovery_not_primary, checkpoint_discards_fallback_safety_path, checkpoint_replaces_callback_primary_path, ) FIXTURE = ( _ROOT / "memory" / "fixtures" / "task-2553plus31.no-normal-callback-regression.json" ) NEW_MODULES = [ _ROOT / "dispatch" / "executor_completion_contract.py", _ROOT / "dispatch" / "spec_template_validator.py", _ROOT / "anu_v3" / "executor_callback_contract.py", _ROOT / "anu_v3" / "runtime_reconcile_checkpoint_recovery_layer.py", ] PLUS31_CKPT = _ROOT / "anu_v3" / "runtime_reconcile_checkpoint.py" FROZEN_ANCHOR = _ROOT / "utils" / "anu_delegation_completion_callback.py" DURABLE_V1 = _ROOT / "memory" / "events" / "task-2553.parallel-batch-state.json" # Baselines captured at +32 start (task md §8 invariant anchors). GIT_HEAD_PRE = "20456b5f83fc039f2fd6f50f4b94095c29b41bfb" GIT_BRANCH_PRE = "task/task-2553p1-f1-clean-replacement" FROZEN_ANCHOR_SHA = ( "83b3e307c8207c76a3e311c408aab4951373bd317896e51687d3007907b0c3d4" ) DURABLE_V1_SHA = ( "fe705d84274e8ae367aaa88c77df763b46bdf4c936efaa1dae78458aedd2a3bc" ) PLUS31_CKPT_SHA = ( "6018730f60bccd3b165b5b83b1f18a3281c71d4e06f3657a1ae6cbe29d1ef853" ) def _sha(p: Path) -> str: return hashlib.sha256(p.read_bytes()).hexdigest() def _git(*args: str) -> str: return subprocess.run( ["git", "-C", str(_ROOT), *args], capture_output=True, text=True, check=True, ).stdout.strip() class ExecutorCompletionCallbackMandatory(unittest.TestCase): @classmethod def setUpClass(cls): cls.fx = json.loads(FIXTURE.read_text(encoding="utf-8")) # §6.1 — executor task spec WITH normal completion callback -> PASS def test_01_spec_with_callback_pass(self): spec = ( "executor 는 작업 완료 직후 ANU 에 normal completion callback " "cron 을 반드시 발사해야 한다 (mandatory)." ) self.assertTrue(spec_has_normal_callback_clause(spec)) self.assertEqual(validate_spec(spec).verdict, PASS) # §6.2 — executor task spec MISSING normal callback -> FAIL def test_02_spec_missing_callback_fail(self): spec = "executor 는 작업을 수행하고 result.json 과 .done 을 남긴다." self.assertFalse(spec_has_normal_callback_clause(spec)) r = validate_spec(spec) self.assertEqual(r.verdict, FAIL) self.assertTrue(any("MISSING" in x for x in r.reasons)) # §6.3 — NO-CRON registry task does NOT strip the mandatory callback def test_03_no_cron_does_not_strip_mandatory(self): self.assertIn("MANDATORY", NO_CRON_CORRECTED_DEFINITION) self.assertIn("registry/checkpoint", NO_CRON_CORRECTED_DEFINITION) # A NO-CRON-labelled spec still must carry the executor callback. spec = ( "NO-CRON variant. 그래도 executor 는 완료 직후 normal completion " "callback 을 반드시 발사한다 (mandatory)." ) self.assertEqual(validate_spec(spec).verdict, PASS) # §6.4 — registry/checkpoint itself keeps cron add/remove == 0 def test_04_registry_checkpoint_zero_cron(self): # §6.4 — modules perform ZERO cron *execution*. The scan targets # actual execution vectors, not explanatory prose (the words # "cron"/"cron-remove" legitimately appear in docstrings stating # the NO-CRON guarantee). self.assertFalse(registry_is_primary_completion_mechanism()) self.assertFalse(plus32_executor_removes_any_cron()) exec_vectors = ( "--cron", # cokacdir cron CLI flag "--cron-remove", "cokacdir", # the scheduling binary "subprocess", # any process spawn "os.system", "Popen", "os.popen", ) for m in NEW_MODULES: txt = m.read_text(encoding="utf-8") for tok in exec_vectors: self.assertNotIn( tok, txt, f"{m.name} must not execute cron/process ({tok!r})", ) # §6.5 — executor completion callback NOT classified as cron-add violation def test_05_callback_not_cron_violation(self): self.assertFalse(is_executor_completion_callback_a_cron_violation()) self.assertFalse(executor_completion_callback_is_cron_add_violation()) # §6.6 — fallback callback preserved as safety path def test_06_fallback_safety_path(self): self.assertTrue(fallback_is_safety_path()) self.assertFalse(checkpoint_discards_fallback_safety_path()) # §6.7 — fallback cancel-on-success path preserved after normal success def test_07_cancel_on_success_preserved(self): self.assertTrue(cancel_on_success_applies_after_normal_success()) # +32 asserts only; executes zero cron remove (9-R.1). self.assertFalse(plus32_executor_removes_any_cron()) # §6.8 — result.json + .done but no normal callback -> # RESULT_READY_NO_NORMAL_CALLBACK def test_08_result_ready_no_normal_callback(self): c = classify_completion( dispatch_ok=True, result_present=True, done_present=True, normal_callback_registered=False, ) self.assertEqual(c, RESULT_READY_NO_NORMAL_CALLBACK) # §6.9 — that state is RECOVERY, not a normal lifecycle complete def test_09_recovery_not_complete(self): self.assertTrue(is_recovery_state(RESULT_READY_NO_NORMAL_CALLBACK)) self.assertFalse( is_accepted_normal_lifecycle(RESULT_READY_NO_NORMAL_CALLBACK) ) self.assertTrue( is_accepted_normal_lifecycle(NORMAL_COLLECTOR_COMPLETED) ) # §6.10 — the +31 past-error case reproduced by the fixture def test_10_plus31_error_reproduced(self): ec = self.fx["error_case"] spec = ec["spec_excerpt"] self.assertFalse(spec_has_normal_callback_clause(spec)) self.assertEqual(validate_spec(spec).verdict, FAIL) obs = ec["observed_runtime"] self.assertEqual( classify_completion( dispatch_ok=obs["dispatch_ok"], result_present=obs["result_present"], done_present=obs["done_present"], normal_callback_registered=obs["normal_callback_registered"], ), ec["classification_expected"], ) t = ec["callback_4tuple"] self.assertFalse( tuple_is_valid(Callback4Tuple( t["task_id"], t["dispatch_cron_id"], t["normal_collector_cron_id"], t["fallback_callback_cron_id"], )) ) # §6.11 — generated dispatch prompt auto-includes the callback clause def test_11_prompt_auto_includes_clause(self): bare = "task-XXXX 를 수행하라." self.assertFalse(spec_has_normal_callback_clause(bare)) injected = inject_completion_callback_clause(bare) self.assertTrue(spec_has_normal_callback_clause(injected)) # idempotent self.assertEqual( inject_completion_callback_clause(injected), injected ) # §6.12 — 4-tuple missing normal_collector_cron_id -> invalid def test_12_4tuple_requires_normal_collector(self): bad = Callback4Tuple("t", "D", None, "F") reasons = validate_4tuple(bad) self.assertTrue( any("normal_collector_cron_id missing" in r for r in reasons) ) good = Callback4Tuple("t", "D", "N", "F") self.assertTrue(tuple_is_valid(good)) # §6.13 — checkpoint claiming to replace callback primary path -> FAIL def test_13_checkpoint_not_primary(self): self.assertTrue(callback_is_primary()) self.assertTrue(checkpoint_is_recovery_layer()) self.assertFalse(checkpoint_replaces_callback_primary_path()) self.assertEqual(assert_checkpoint_is_recovery_not_primary(), []) # §6.14 — checkpoint discarding fallback safety path -> FAIL def test_14_checkpoint_keeps_fallback(self): self.assertFalse(checkpoint_discards_fallback_safety_path()) # §6.15 — write/merge/closeout/cron-remove authority -> FAIL; # 9-R.2: executor self-task lifecycle outputs REQUIRED def test_15_finalization_authority_forbidden(self): bad = closeout_authority_violation(CloseoutAuthorityCheck( finalization_authority_escalated=True, executor_self_lifecycle_outputs_present=True, )) self.assertTrue(any("finalization" in r for r in bad)) ok = closeout_authority_violation(CloseoutAuthorityCheck( finalization_authority_escalated=False, executor_self_lifecycle_outputs_present=True, )) self.assertEqual(ok, []) # closeout WITHOUT normal-callback evidence == FAIL (§4.4/§4.6) ev = ExecutorCloseoutEvidence( result_json_present=True, done_marker_present=True, report_present=True, normal_callback_registration_evidence=False, normal_collector_cron_id=None, ) self.assertFalse(closeout_is_valid(ev)) self.assertTrue( any("normal callback registration evidence missing" in r for r in validate_closeout_evidence(ev)) ) # ── invariants (§8 / 9-R) ──────────────────────────────────────────── def test_inv_spec_location_unknown_holds(self): r = validate_spec("", spec_location_known=False) self.assertEqual(r.verdict, HOLD) def test_inv_git_head_branch_equal(self): self.assertEqual(_git("rev-parse", "HEAD"), GIT_HEAD_PRE) self.assertEqual( _git("rev-parse", "--abbrev-ref", "HEAD"), GIT_BRANCH_PRE ) def test_inv_frozen_anchor_durable_ckpt_unmutated(self): self.assertEqual(_sha(FROZEN_ANCHOR), FROZEN_ANCHOR_SHA) self.assertEqual(_sha(DURABLE_V1), DURABLE_V1_SHA) # +31 checkpoint original byte-equal (additive sidecar, zero mutation) self.assertEqual(_sha(PLUS31_CKPT), PLUS31_CKPT_SHA) def test_inv_new_modules_git_untracked(self): for m in NEW_MODULES + [FIXTURE]: rc = subprocess.run( ["git", "-C", str(_ROOT), "ls-files", "--error-unmatch", str(m.relative_to(_ROOT))], capture_output=True, text=True, ).returncode self.assertNotEqual( rc, 0, f"{m.name} must be git-untracked (§8 invariant)" ) def test_inv_doctrine_verdict(self): v = doctrine_verdict() self.assertIn("MANDATORY", v) self.assertIn("NO-CRON", v) if __name__ == "__main__": unittest.main(verbosity=2)