# -*- coding: utf-8 -*- """Regression — task-2553+46 canonical artifact-root resolver + collector lookup. Covers §5 verbatim items 10~12, 16, 19, 20 + §5/§8/9-R invariants. The normal/fallback/dead-man collector no longer judges artifact-missing on an autoset-cwd false negative: the canonical ANU workspace root (/home/jay/workspace) is hard-coded and re-checked FIRST (회장 §2.2/§3.C). 9-R.1 Layer A: artifact_root_resolver / collector_artifact_lookup are read-only — ZERO write/cron/dispatch. 9-R.2/§5.20: the runtime checkpoint stays a recovery layer, never the primary callback replacement. """ import hashlib import importlib.util import json import subprocess import sys import tempfile 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): 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 _arr = _load("_p46_arr", "anu_v3/artifact_root_resolver.py") sys.modules["anu_v3.artifact_root_resolver"] = _arr _reg = _load("_p46_reg", "anu_v3/callback_4tuple_registry.py") sys.modules["anu_v3.callback_4tuple_registry"] = _reg _cal = _load("_p46_cal", "anu_v3/collector_artifact_lookup.py") resolve_roots = _arr.resolve_roots canonical_root = _arr.canonical_root CANONICAL_ANU_WORKSPACE_ROOT = _arr.CANONICAL_ANU_WORKSPACE_ROOT classify = _cal.classify RESULT_PRESENT = _cal.RESULT_PRESENT RESULT_MISSING = _cal.RESULT_MISSING NORMAL_COLLECTOR_COMPLETED = _cal.NORMAL_COLLECTOR_COMPLETED FORBIDDEN_WHEN_CANONICAL_PRESENT = _cal.FORBIDDEN_WHEN_CANONICAL_PRESENT # §5.20 — runtime checkpoint stays a recovery layer (real-package import; no # tests/anu_v3 shadow exists, resolves to the workspace module). 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, ) from anu_v3.executor_callback_contract import ( # noqa: E402 callback_is_primary, cancel_on_success_applies_after_normal_success, checkpoint_is_recovery_layer, fallback_is_safety_path, ) NEW_MODULES = [ _ROOT / "anu_v3" / "artifact_root_resolver.py", _ROOT / "anu_v3" / "collector_artifact_lookup.py", _ROOT / "anu_v3" / "callback_4tuple_registry.py", _ROOT / "dispatch" / "cron_dispatch_guard.py", ] FROZEN_ANCHOR = _ROOT / "utils" / "anu_delegation_completion_callback.py" DURABLE_V1 = _ROOT / "memory" / "events" / "task-2553.parallel-batch-state.json" PPE = _ROOT / "anu_v3" / "policy_profile_engine.py" PBC = _ROOT / "anu_v3" / "parallel_batch_coordinator.py" GIT_HEAD_PRE = "20456b5f83fc039f2fd6f50f4b94095c29b41bfb" GIT_BRANCH_PRE = "task/task-2553p1-f1-clean-replacement" FROZEN_ANCHOR_SHA = ( "83b3e307c8207c76a3e311c408aab4951373bd317896e51687d3007907b0c3d4" ) DURABLE_V1_SHA = ( "fe705d84274e8ae367aaa88c77df763b46bdf4c936efaa1dae78458aedd2a3bc" ) PPE_SHA = "2363e291a0a43884892f5e554f115481a077322bd5caa3000fb75bf5b72bc6be" PBC_SHA = "10529421110b3d2765785b6cf911527c8f5e964b5078fcfa6190fcb86d0f2c0f" FX39 = ( _ROOT / "memory" / "fixtures" / "task-2553plus39.artifact-root-false-missing.json" ) PLUS39_RESULT = _ROOT / "memory" / "events" / "task-2553+39.result.json" def _sha(p: Path) -> str: return hashlib.sha256(p.read_bytes()).hexdigest() def _git(*a: str) -> str: return subprocess.run( ["git", "-C", str(_ROOT), *a], capture_output=True, text=True, check=True, ).stdout.strip() class ArtifactRootResolver(unittest.TestCase): @classmethod def setUpClass(cls): cls.fx39 = json.loads(FX39.read_text(encoding="utf-8")) # canonical anchor is the hard-coded ANU workspace root (§3.C) def test_00_canonical_root_fixed(self): self.assertEqual(str(canonical_root()), "/home/jay/workspace") self.assertEqual( str(CANONICAL_ANU_WORKSPACE_ROOT), "/home/jay/workspace" ) # §5.10 — +39: autoset has no memory/events, canonical result present # -> RESULT_PRESENT (or NORMAL_COLLECTOR_COMPLETED) def test_10_plus39_canonical_present(self): self.assertTrue( PLUS39_RESULT.is_file(), "precondition: canonical +39 result must exist", ) with tempfile.TemporaryDirectory() as autoset: # autoset cwd deliberately has NO memory/events r = classify( task_id="task-2553+39", chat_id="6937032012", autoset_cwd=Path(autoset), result_basename="task-2553+39.result.json", ) self.assertIn( r.verdict, self.fx39["case"]["verdict_expected_one_of"], ) for forb in self.fx39["case"]["verdict_forbidden"]: self.assertNotEqual(r.verdict, forb) self.assertEqual(r.canonical_root, "/home/jay/workspace") self.assertTrue(r.autoset_only_miss_blocked) # §5.11 — autoset missing alone must NOT yield RESULT_MISSING def test_11_autoset_only_not_missing(self): with tempfile.TemporaryDirectory() as autoset: r = classify( task_id="task-2553+39", chat_id="6937032012", autoset_cwd=Path(autoset), result_basename="task-2553+39.result.json", ) self.assertNotIn(r.verdict, FORBIDDEN_WHEN_CANONICAL_PRESENT) self.assertTrue(r.autoset_only_miss_blocked) roots = resolve_roots(autoset_cwd=Path(tempfile.gettempdir())) # canonical FIRST in search order (§3.C re-check ordering proof) self.assertEqual(roots.search_order[0], "/home/jay/workspace") # §5.12 — canonical absent + schedule missing + stale -> RESULT_MISSING def test_12_true_missing_bot_stale(self): tm = self.fx39["true_missing_case"] with tempfile.TemporaryDirectory() as autoset, \ tempfile.TemporaryDirectory() as empty_sh: r = classify( task_id=tm["task_id"], chat_id="6937032012", autoset_cwd=Path(autoset), dispatch_stale=tm["dispatch_stale"], schedule_history_dir=Path(empty_sh), ) self.assertEqual(r.verdict, RESULT_MISSING) self.assertFalse(r.schedule_history_seen) self.assertTrue(r.dispatch_stale) # §5.12b — not stale + no history -> fail-safe defer (no false-miss) def test_12b_failsafe_no_false_missing(self): with tempfile.TemporaryDirectory() as autoset, \ tempfile.TemporaryDirectory() as empty_sh: r = classify( task_id="task-2553+NONEXISTENT-YYY", chat_id="6937032012", autoset_cwd=Path(autoset), dispatch_stale=False, schedule_history_dir=Path(empty_sh), ) self.assertNotEqual(r.verdict, RESULT_MISSING) self.assertEqual(r.verdict, "PENDING_FAILSAFE") # §5.16 — unrelated-task callback must NOT be cited def test_16_unrelated_callback_not_cited(self): with tempfile.TemporaryDirectory() as autoset, \ tempfile.TemporaryDirectory() as empty_sh: r = classify( task_id="task-2553+UNRELATED", chat_id="6937032012", autoset_cwd=Path(autoset), dispatch_stale=False, schedule_history_dir=Path(empty_sh), ) # no ledger row + no canonical artifact -> never NORMAL_COLLECTOR_* self.assertNotEqual(r.verdict, NORMAL_COLLECTOR_COMPLETED) self.assertEqual(r.registry_verdict, "NO_LEDGER_RECORD") # §5.19 — callback/fallback/cancel-on-success structure preserved def test_19_structure_preserved(self): self.assertTrue(callback_is_primary()) self.assertTrue(fallback_is_safety_path()) self.assertTrue(cancel_on_success_applies_after_normal_success()) # §5.20 — runtime checkpoint remains recovery layer, not primary callback def test_20_checkpoint_recovery_not_primary(self): self.assertTrue(checkpoint_is_recovery_layer()) self.assertFalse(checkpoint_replaces_callback_primary_path()) self.assertFalse(checkpoint_discards_fallback_safety_path()) self.assertEqual(assert_checkpoint_is_recovery_not_primary(), []) # ── invariants (§5 / §8 / 9-R) ─────────────────────────────────────── 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_unmutated(self): self.assertEqual(_sha(FROZEN_ANCHOR), FROZEN_ANCHOR_SHA) self.assertEqual(_sha(DURABLE_V1), DURABLE_V1_SHA) self.assertEqual(_sha(PPE), PPE_SHA) self.assertEqual(_sha(PBC), PBC_SHA) def test_inv_new_modules_git_untracked(self): for m in NEW_MODULES: 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") def test_inv_layer_a_read_only_no_cron(self): # Real execution / write syntax only (docstrings legitimately name # cokacdir/subprocess when stating the read-only guarantee). forbidden_syntax = ( "import subprocess", "subprocess.run", "subprocess.Popen", "os.system(", "os.popen(", ".Popen(", '"--cron', "'--cron", 'open(', '.write(', '.write_text(', '.mkdir(', '"a"', "'a'", '"w"', "'w'", # resolver/lookup never write ) for m in [ _ROOT / "anu_v3" / "artifact_root_resolver.py", _ROOT / "anu_v3" / "collector_artifact_lookup.py", ]: txt = m.read_text(encoding="utf-8") for tok in forbidden_syntax: self.assertNotIn( tok, txt, f"{m.name} (Layer A read-only) must not {tok!r}", ) if __name__ == "__main__": unittest.main(verbosity=2)