# -*- coding: utf-8 -*- """ Regression — task-2553+27 coordinator V0 closeout finalizer (§3.3 / 9-R.4 / 9-R.5). Cases: 1. normal closeout -> marker produced, 9-packet PASS, ACCEPT/CONVERGED 2. batch-state file unchanged -> sha pre == post (read-only preserve) 3. coordinator code unchanged -> sha pre == post (byte-0 anchor) 4. idempotent -> 2nd run = NO_OP, marker/decision byte-stable, no double-closeout 5. evidence absent -> HOLD_FOR_CHAIR, no marker written + write whitelist enforcement (9-R.5): non-whitelisted / tracked path refused """ import importlib.util import shutil import sys import unittest from pathlib import Path _SCRIPT = Path(__file__).resolve().parents[2] / "scripts" / "run_coordinator_v0_closeout.py" _spec = importlib.util.spec_from_file_location("rcv0c", _SCRIPT) clo = importlib.util.module_from_spec(_spec) sys.modules["rcv0c"] = clo _spec.loader.exec_module(clo) REAL_ROOT = clo.find_repo_root(_SCRIPT) def _make_fixture_repo(tmp: Path, *, drop: set[str] | None = None) -> Path: """Copy the real read-only evidence inputs into an isolated tmp repo.""" drop = drop or set() root = tmp / "repo" (root / "memory" / "events").mkdir(parents=True) (root / "memory" / "reports").mkdir(parents=True) (root / "anu_v3").mkdir(parents=True) (root / "scripts").mkdir(parents=True) for rel in list(clo.EVIDENCE.values()) + [clo.COORDINATOR_CODE]: if rel in drop: continue src = REAL_ROOT / rel if src.is_file(): dst = root / rel dst.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, dst) return root class CloseoutRegression(unittest.TestCase): def setUp(self): self._tmp = Path(__import__("tempfile").mkdtemp(prefix="p27clo_")) self.addCleanup(shutil.rmtree, self._tmp, ignore_errors=True) # ── 1. normal closeout ──────────────────────────────────────────────────── def test_1_normal_closeout(self): root = _make_fixture_repo(self._tmp) res = clo.run_closeout(root) self.assertEqual(res["verdict"], "CLOSEOUT_FINALIZED") self.assertTrue(res["ok"]) self.assertEqual(res["final_status"], "ACCEPT / CONVERGED") self.assertEqual(res["marker_branch"], "FRESH_WRITE") marker = root / clo.MARKER_REL self.assertTrue(marker.is_file()) import json m = json.loads(marker.read_text("utf-8")) self.assertEqual(m["nine_packet_check"]["result"], "PASS") self.assertEqual(len(m["nine_packet_check"]["keys"]), 9) self.assertTrue(m["high4_resolved"]) self.assertTrue(m["accept_converged"]) self.assertTrue((root / clo.DECISION_REL).is_file()) # ── 2. batch-state file unchanged ───────────────────────────────────────── def test_2_state_file_untouched(self): root = _make_fixture_repo(self._tmp) state = root / clo.EVIDENCE["batch_state"] pre = clo._sha256_file(state) res = clo.run_closeout(root) post = clo._sha256_file(state) self.assertEqual(pre, post) self.assertTrue(res["state_sha_equal"]) self.assertEqual(res["state_sha_pre"], res["state_sha_post"]) # ── 3. coordinator code unchanged ───────────────────────────────────────── def test_3_coordinator_code_untouched(self): root = _make_fixture_repo(self._tmp) coord = root / clo.COORDINATOR_CODE pre = clo._sha256_file(coord) res = clo.run_closeout(root) post = clo._sha256_file(coord) self.assertEqual(pre, post) self.assertTrue(res["coord_sha_equal"]) # ── 4. idempotent (no double-closeout, byte-stable) ─────────────────────── def test_4_idempotent_no_op(self): root = _make_fixture_repo(self._tmp) r1 = clo.run_closeout(root) self.assertEqual(r1["marker_branch"], "FRESH_WRITE") marker_b1 = (root / clo.MARKER_REL).read_bytes() decision_b1 = (root / clo.DECISION_REL).read_bytes() state_after1 = clo._sha256_file(root / clo.EVIDENCE["batch_state"]) r2 = clo.run_closeout(root) self.assertEqual(r2["verdict"], "CLOSEOUT_FINALIZED") self.assertEqual(r2["marker_branch"], "NO_OP_BYTE_STABLE") self.assertTrue(r2["no_op_already_closeout"]) self.assertEqual((root / clo.MARKER_REL).read_bytes(), marker_b1) self.assertEqual((root / clo.DECISION_REL).read_bytes(), decision_b1) self.assertEqual(clo._sha256_file(root / clo.EVIDENCE["batch_state"]), state_after1) self.assertEqual(r1["closeout_id"], r2["closeout_id"]) def test_4b_inconsistent_marker_holds(self): root = _make_fixture_repo(self._tmp) clo.run_closeout(root) (root / clo.MARKER_REL).write_text('{"tampered": true}\n', "utf-8") res = clo.run_closeout(root) self.assertEqual(res["verdict"], "HOLD_FOR_CHAIR") self.assertIn("MARKER_INCONSISTENT_WITH_DETERMINISTIC_RECOMPUTE", res["reasons"]) # ── 5. evidence absent -> HOLD, no marker ───────────────────────────────── def test_5_evidence_absent_holds(self): for missing in ("memory/events/task-2553+17.result.json", "memory/events/task-2553+19.adjudication-resolution.json"): root = _make_fixture_repo(self._tmp / missing.replace("/", "_"), drop={missing}) res = clo.run_closeout(root) self.assertEqual(res["verdict"], "HOLD_FOR_CHAIR", missing) self.assertFalse((root / clo.MARKER_REL).exists(), missing) self.assertTrue(any("MISSING" in r for r in res["reasons"]), missing) # ── 9-R.5 write whitelist enforcement ───────────────────────────────────── def test_6_write_whitelist_enforced(self): root = _make_fixture_repo(self._tmp) with self.assertRaises(PermissionError): clo._assert_writable(root, "anu_v3/parallel_batch_coordinator.py") with self.assertRaises(PermissionError): clo._assert_writable(root, "memory/events/task-2553.parallel-batch-state.json") with self.assertRaises(PermissionError): clo._assert_writable(root, "memory/events/task-2553+26.anything.json") # whitelisted untracked path is accepted self.assertTrue(str(clo._assert_writable(root, clo.MARKER_REL)).endswith(clo.MARKER_REL)) if __name__ == "__main__": unittest.main(verbosity=2)