# -*- coding: utf-8 -*- """Regression — task-2553+30 TRACK B parallel_batch_coordinator v0 generalized. Covers §5 verbatim (9 items) + NO-CRON dogfood / frozen-guard / no-mutation / schema-conformance extras. NO-CRON (9-R.1): zero cron register/remove; self- completion via result.json + .done existence (dogfooding, §12). 1 +26 MERGED, +27 PASS, +28 DONE, +29 ACCEPT in ONE batch_state 2 normal callback missing but result ready -> RESULT_READY_NO_NORMAL_CALLBACK 3 pending fallback after result ready -> non-blocking 4 fallback duplicate -> DUPLICATE_CALLBACK_IGNORED 5 4-tuple mismatch -> TRACK_MISMATCH 6 cross-track contamination -> BATCH_HOLD 7 one track HOLD does not block independent accepted track 8 closeout eligible derived from evidence (batch_state) 9 coordinator does NOT directly merge/write/cron/closeout """ import hashlib import json 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)) from anu_v3.generic_batch_coordinator import ( # noqa: E402 GenericBatchCoordinator, GenericBatchPlan, GenericTrackPlan, FrozenWriteRefused, emit_generic_batch_state, load_plan_from_fixture, ) FIXTURE = _ROOT / "memory" / "fixtures" / "task-2553.generic-batch.fixture.json" FROZEN_V1 = _ROOT / "memory" / "events" / "task-2553.parallel-batch-state.json" STATE_SCHEMA = _ROOT / "schemas" / "generic_batch_state.schema.json" GENERIC_MODULE = _ROOT / "anu_v3" / "generic_batch_coordinator.py" # +26~+29 closeout artifacts — read-only fixtures (9-R.2). no-mutation guard. CLOSEOUT_ARTIFACTS = [ _ROOT / "memory" / "events" / "task-2553.batch-closeout-decision_260517.json", _ROOT / "memory" / "events" / "task-2553.batch-closeout.result.json", _ROOT / "memory" / "reports" / "task-2553.batch-closeout-consolidated-summary_260517.md", _ROOT / "memory" / "events" / "task-2553.parallel-runtime-registry.batch-state.json", FIXTURE, ] def _coord() -> GenericBatchCoordinator: return GenericBatchCoordinator(load_plan_from_fixture(FIXTURE)) class GenericBatchCoordinatorRegression(unittest.TestCase): # 1 def test_01_four_outcomes_one_batch_state(self): st = _coord().build_state() tr = st["track_runtime_records"] outcomes = { "task-2553+26": "MERGED", "task-2553+27": "PASS", "task-2553+28": "DONE", "task-2553+29": "ACCEPT", } self.assertEqual(set(tr), set(outcomes)) cs = st["consolidated_summary"]["tracks"] for tid, want in outcomes.items(): self.assertEqual(cs[tid]["terminal_outcome"], want) # 2 def test_02_result_ready_no_normal_callback(self): recs = _coord().track_runtime_records() for tid in ("task-2553+26", "task-2553+27", "task-2553+28", "task-2553+29"): self.assertEqual( recs[tid]["classification"], "RESULT_READY_NO_NORMAL_CALLBACK", ) self.assertTrue( recs[tid]["recovery_eligible"], "RESULT_READY is a recovery target, not a failure", ) # 3 def test_03_pending_fallback_non_blocking(self): c = _coord() jp = c.join() # +26/+27/+28 fallback PENDING yet all settle independently. self.assertEqual(jp["batch_next_action"], "BATCH_ACCEPT") self.assertEqual( jp["independent_done_tracks"], ["task-2553+26", "task-2553+27", "task-2553+28", "task-2553+29"], ) self.assertEqual(jp["held_tracks"], []) # 4 def test_04_fallback_duplicate_ignored(self): c = _coord() self.assertEqual( c.classify_fallback_fire( claimed_task_id="task-2553+26", fallback_cron_id="44AE69D5", ), "DUPLICATE_CALLBACK_IGNORED", ) # 5 def test_05_four_tuple_mismatch(self): c = _coord() # fallback cron of +27 fired claiming +26 -> TRACK_MISMATCH reasons = c.validate_callback_identity( claimed_task_id="task-2553+26", event_kind="fallback", event_cron_id="CC33E68C", ) self.assertTrue(any("belongs to" in r for r in reasons)) self.assertEqual( c.classify_fallback_fire( claimed_task_id="task-2553+26", fallback_cron_id="CC33E68C", ), "TRACK_MISMATCH", ) # 6 def test_06_cross_track_contamination_batch_hold(self): plan = load_plan_from_fixture(FIXTURE) # +27 cites a +26-owned artifact -> contamination. for tp in plan.tracks: if tp.task_id == "task-2553+27": tp.cited_artifacts = [ "memory/events/task-2553+26.result.json" ] c = GenericBatchCoordinator(plan) self.assertTrue(c.contamination()) self.assertEqual(c.join()["batch_next_action"], "BATCH_HOLD") self.assertEqual(c.batch_next_action(), "BATCH_HOLD_CONTAMINATION") self.assertFalse(c.closeout_proposal()["eligible"]) # 7 def test_07_hold_track_does_not_block_accepted(self): plan = load_plan_from_fixture(FIXTURE) for tp in plan.tracks: if tp.task_id == "task-2553+27": tp.hold_for_chair = True tp.terminal_outcome = "HOLD" c = GenericBatchCoordinator(plan) jp = c.join() self.assertIn("task-2553+27", jp["held_tracks"]) for tid in ("task-2553+26", "task-2553+28", "task-2553+29"): self.assertIn(tid, jp["independent_done_tracks"]) self.assertTrue( jp["independent_done_tracks"], "accepted tracks stay independently settled", ) self.assertEqual(c.batch_next_action(), "CHAIR_DECISION_REQUIRED") # 8 def test_08_closeout_eligible_evidence_based(self): c = _coord() prop = c.closeout_proposal() self.assertTrue(prop["eligible"]) self.assertEqual(prop["derived_from"], "batch_state") self.assertFalse(prop["confirmed"], "§7 — never confirmed here") # derivation depends only on observed evidence: flip one track to # unsettled and eligibility must drop. plan = load_plan_from_fixture(FIXTURE) for tp in plan.tracks: if tp.task_id == "task-2553+28": tp.terminal_outcome = "PENDING" tp.result_present = False tp.done_present = False self.assertFalse( GenericBatchCoordinator(plan).closeout_proposal()["eligible"] ) # 9 def test_09_coordinator_no_merge_write_cron_closeout(self): c = _coord() # (a) closeout never confirmed even when eligible self.assertFalse(c.closeout_proposal()["confirmed"]) # (b) decision logic has zero filesystem side effect with tempfile.TemporaryDirectory() as d: before = set(Path(d).iterdir()) c.build_state() c.batch_next_action() c.consolidated_summary() c.authority_packets() self.assertEqual(set(Path(d).iterdir()), before) # (c) emission refuses chair durable v1 + git-tracked paths AND # never clobbers an existing untracked non-deliverable file. with self.assertRaises(FrozenWriteRefused): emit_generic_batch_state({}, FROZEN_V1) with self.assertRaises(FrozenWriteRefused): emit_generic_batch_state( {}, _ROOT / "anu_v3" / "goal_activation_controller.py" ) with tempfile.TemporaryDirectory() as d: victim = Path(d) / "unrelated_module.py" victim.write_text("ORIGINAL", encoding="utf-8") with self.assertRaises(FrozenWriteRefused): emit_generic_batch_state({}, victim) self.assertEqual( victim.read_text(encoding="utf-8"), "ORIGINAL", "existing untracked non-deliverable must NOT be clobbered", ) # (d) module performs zero cron register/remove (NO-CRON, §7) src = GENERIC_MODULE.read_text(encoding="utf-8") for tok in ("--cron", "cron-remove", "cron-register", "--cron-remove", "cokacdir"): self.assertNotIn(tok, src, f"NO-CRON violated: {tok!r}") # 10 — NO-CRON dogfooding self-completion (§12 / 9-R.1) def test_10_dogfood_self_completion(self): with tempfile.TemporaryDirectory() as d: res = Path(d) / "task-2553+30.result.json" done = Path(d) / "task-2553+30.done" self.assertFalse( GenericBatchCoordinator.self_completion_recognized(res, done) ) res.write_text("{}", encoding="utf-8") done.write_text("", encoding="utf-8") self.assertTrue( GenericBatchCoordinator.self_completion_recognized(res, done) ) # 11 — emission to a NEW untracked path succeeds; schema conforms def test_11_emit_new_path_and_schema_conform(self): import jsonschema # available in env c = _coord() st = c.build_state() schema = json.loads(STATE_SCHEMA.read_text(encoding="utf-8")) # self-contained schema (internal #/definitions only) — no remote ref jsonschema.validate(st, schema) with tempfile.TemporaryDirectory() as d: out = Path(d) / "task-2553.generic-batch-state.json" p = emit_generic_batch_state(st, out) self.assertTrue(p.is_file()) reloaded = json.loads(p.read_text(encoding="utf-8")) self.assertEqual(reloaded["schema"], "anu_v3.generic_batch_state.v0") # 12 — no-mutation of +26~+29 closeout / fixture artifacts (9-R.2) def test_12_no_mutation_of_closeout_artifacts(self): def sha(p: Path) -> str: return hashlib.sha256(p.read_bytes()).hexdigest() pre = {p: sha(p) for p in CLOSEOUT_ARTIFACTS if p.is_file()} c = _coord() c.build_state() with tempfile.TemporaryDirectory() as d: emit_generic_batch_state( c.build_state(), Path(d) / "task-2553.generic-batch-state.json" ) post = {p: sha(p) for p in CLOSEOUT_ARTIFACTS if p.is_file()} self.assertEqual(pre, post, "read-only fixture mutated (9-R.2)") # 13 — additive: programmatic plan (generalization beyond the fixture) def test_13_generic_arbitrary_batch(self): plan = GenericBatchPlan( batch_label="adhoc-2track", tracks=[ GenericTrackPlan( track_id="A", task_id="t-A", dispatch_cron_id="DA", fallback_callback_cron_id="FA", result_present=True, terminal_outcome="PASS", fallback_state="PENDING", ), GenericTrackPlan( track_id="B", task_id="t-B", dispatch_cron_id="DB", fallback_callback_cron_id="FB", result_present=False, terminal_outcome="PENDING", fallback_state="PENDING", ), ], ) c = GenericBatchCoordinator(plan) jp = c.join() self.assertIn("t-A", jp["independent_done_tracks"]) self.assertIn("t-B", jp["waiting_tracks"]) self.assertEqual(jp["batch_next_action"], "WAIT_FOR_FALLBACK") self.assertFalse(c.closeout_proposal()["eligible"]) if __name__ == "__main__": unittest.main(verbosity=2)