"""Regression — task-2553+17 ANU_PARALLEL_BATCH_COORDINATOR_V0 (Track2+3). Authority: task-2553+17.md §6 (chair verbatim 11) + §12 (9-R.1/.3/.5/.6/.7). All 11 chair checks + transition-table + fixture-immutability + frozen-anchor invariant must PASS. Pure stdlib + pytest; no GitHub / network / live mutation. """ from __future__ import annotations import hashlib import json import subprocess from pathlib import Path import pytest from anu_v3.batch_join_policy import ( CallbackEvent, PacketCandidate, classify_callback, final_authority_packet_selector, pending_blocks_chair_decision, ) from anu_v3.goal_loop_planner import ( GoalRequest, PolicyProfile, generate_loop_plan, ) from anu_v3.parallel_batch_coordinator import ( BatchPlan, ParallelBatchCoordinator, TrackedPathWriteRefused, TrackPlan, _is_git_tracked, ) from anu_v3.track_loop_state import ( ALL_STATES, DISPATCHED, HOLD_FOR_CHAIR, LANDING_PENDING, MERGED, MERGE_READY, PLANNED, RETRY_WITHIN_SCOPE, TERMINAL_STATES, TrackLoopState, is_legal_transition, ) REPO = Path(__file__).resolve().parents[2] FIXTURE = REPO / "memory/fixtures/task-2553.parallel-batch.fixture.json" FROZEN_ANCHORS = [ "anu_v3/pre_authorized_action_gate.py", "anu_v3/pre_authorized_contract_deriver.py", "anu_v3/pre_authorized_evidence_bundle_builder.py", "anu_v3/pre_authorized_executor_binding.py", "anu_v3/pre_authorized_activation_runner.py", "anu_v3/isolated_worktree_evidence_source.py", "anu_v3/goal_activation_controller.py", "anu_v3/branch_ref_allocator.py", "utils/anu_delegation_completion_callback.py", "utils/completion_callback_fallback_cancel.py", ] FROZEN_ANCHOR_SHA = { "anu_v3/pre_authorized_action_gate.py": "f34e2445452cd84f22a378ece0cd922d0ff952fa545cc02268923a5f9fc46532", "anu_v3/pre_authorized_contract_deriver.py": "2e1af1124cb330b4398fbdcccb0a765126443b859d46acf45a96382082259a0a", "anu_v3/pre_authorized_evidence_bundle_builder.py": "9344f9aa7bd37da9979e39222c782f8544f638b1d99f33bbdda0bff1514ea06a", "anu_v3/pre_authorized_executor_binding.py": "6a66f7b76d7ef6cb16f099b5b1fea29a4dfb7e0312e13898e4db41bf246bb834", "anu_v3/pre_authorized_activation_runner.py": "0e5a4a35db1aff434d65998d33deb302a589a1f80ad497cb3998cbc716993663", "anu_v3/isolated_worktree_evidence_source.py": "2f45801a036bb8a65d69b777b01dc1b15d451c4b4ee0ad01512bb0c8d8720e6a", "anu_v3/goal_activation_controller.py": "619baac9d366087f4f1b3532127378eae51dc03366095737f6ede5a4d0792e7b", "anu_v3/branch_ref_allocator.py": "2569c42b42d2aafbcfb4e040d4851c3097a7c828312059c3518f4d5130dcecf5", "utils/anu_delegation_completion_callback.py": "83b3e307c8207c76a3e311c408aab4951373bd317896e51687d3007907b0c3d4", "utils/completion_callback_fallback_cancel.py": "2ef661c49128ec6f1bd96bfbe6c1a227317e87ae31bf6deac025c66fd0e96d37", } def _sha(p: Path) -> str: return hashlib.sha256(p.read_bytes()).hexdigest() @pytest.fixture(scope="module") def fx() -> dict: return json.loads(FIXTURE.read_text(encoding="utf-8")) @pytest.fixture() def coord(fx) -> ParallelBatchCoordinator: tracks = [ TrackPlan( track_id=t["track_id"], task_id=t["task_id"], dispatch_cron_id=t["dispatch_cron_id"], normal_collector_cron_id=t["normal_collector_cron_id"], fallback_callback_cron_id=t["fallback_callback_cron_id"], expected_files=t.get("expected_files", []), forbidden_write_targets=t.get("forbidden_write_targets", []), depends_on=t.get("depends_on", []), own_artifacts=t.get("own_artifacts", []), cited_artifacts=t.get("cited_artifacts", []), initial_state=t.get("initial_state", "PLANNED"), retry_ceiling=t.get("retry_ceiling", 2), ) for t in fx["tracks"] ] grs = [ GoalRequest( goal_id=g["goal_id"], goal_statement=g["goal_statement"], boundary=g.get("boundary", []), policy_profile=PolicyProfile(**g["policy_profile"]), ) for g in fx.get("goal_requests", []) ] return ParallelBatchCoordinator( BatchPlan(batch_label=fx["batch_label"], tracks=tracks, goal_requests=grs) ) # -------------------------------------------------------------------------- # §6 chair verbatim 11 # -------------------------------------------------------------------------- def test_check01_A_B_contamination_zero_independent(coord): """① A↔B contamination 0, independent.""" assert coord.contamination() == [] assert coord.depmatrix.is_independent("trackA-closeout", "trackB-landing") def test_check02_A_completion_does_not_block_B(coord): """② A completion does not make B wait.""" assert not coord.depmatrix.blocks("trackA-closeout", "trackB-landing") groups = coord.depmatrix.parallelizable_groups() assert groups == [["trackA-closeout", "trackB-landing"]] def test_check03_B_does_not_modify_A_closeout(coord, fx): """③ B does not write A's closeout artifacts.""" b = next(t for t in fx["tracks"] if t["track_id"] == "trackB-landing") a_artifacts = set( next(t for t in fx["tracks"] if t["track_id"] == "trackA-closeout")[ "own_artifacts" ] ) b_writes = set(b["expected_files"]) | set(b["own_artifacts"]) assert b_writes.isdisjoint(a_artifacts) # A's closeout artifacts are in B's forbidden_write_targets assert a_artifacts & set(b["forbidden_write_targets"]) assert coord.depmatrix.forbidden_write_overlap() == [] def test_check04_tuple_mismatch_TRACK_MISMATCH(coord): """④ 4-tuple mismatch -> TRACK_MISMATCH.""" bad = CallbackEvent( track_id="trackB-landing", task_id="task-2553+15", kind="normal", identity_ok=False, normal_already_completed=False, ) assert classify_callback(bad) == "TRACK_MISMATCH" # via registry: wrong cron id is rejected ok, cls = coord.registry.validate_callback( "trackB-landing", { "task_id": "task-2553+15", "dispatch_cron_id": "WRONG", "normal_collector_cron_id": "NC-B15", "fallback_callback_cron_id": "FB-B15", }, ) assert not ok and cls == "TRACK_MISMATCH" def test_check05_normal_then_fallback_DUPLICATE_IGNORED(): """⑤ normal completed first, later fallback -> DUPLICATE_CALLBACK_IGNORED.""" ev = CallbackEvent( track_id="trackA-closeout", task_id="task-2553+13", kind="fallback", identity_ok=True, normal_already_completed=True, ) assert classify_callback(ev) == "DUPLICATE_CALLBACK_IGNORED" # state-machine level mirror st = TrackLoopState("trackA-closeout", state="NORMAL_COLLECTOR_COMPLETED") assert st.classify_callback("fallback") == "DUPLICATE_CALLBACK_IGNORED" def test_check06_pending_fallback_non_blocking(fx): """⑥ pending fallback does not block the chair decision.""" ev_def = next( e for e in fx["callback_events"] if e["name"] == "pending_fallback_non_blocking" ) ev = CallbackEvent( track_id=ev_def["track_id"], task_id=ev_def["task_id"], kind="fallback", identity_ok=True, normal_already_completed=False, authority_settled_via_normal=ev_def["authority_settled_via_normal"], batch_final_depends_on_track=ev_def["batch_final_depends_on_track"], ) assert classify_callback(ev) == "CALLBACK_PENDING" assert pending_blocks_chair_decision(ev) is False def test_check07_final_authority_packet_per_task(coord, fx): """⑦ final authority packet correctly selected per task (9-R.6).""" cands = [ PacketCandidate( task_id=c["task_id"], track_id=c["track_id"], source=c["source"], schema=c["schema"], ts=c["ts"], path=c["path"], status=c.get("status", ""), ) for c in fx["packet_candidates"] ] chosen = final_authority_packet_selector(cands) for task_id, expected_path in fx["expected_authority"].items(): assert chosen[task_id].path == expected_path # tie-break proof: +9 has normal + fallback, normal must win assert chosen["task-2553+9"].source == "normal" def test_check08_summary_chair_decision_only(coord, tmp_path): """⑧ batch summary compresses to chair final decision only.""" res = coord.run( tmp_path / "state.json", tmp_path / "summary.md" ) md = res.summary_md assert "Chair-only" in md assert "batch_next_action" in md # no raw cron secrets / no per-stage internal noise leaked assert "c119085addb0f8b7" not in md and "key " not in md.lower() # state json validates structurally state = json.loads((tmp_path / "state.json").read_text()) assert state["schema"] == "anu_v3.parallel_batch_state.v1" assert state["batch_id"].startswith("batch-") def test_check09_goal_progress_loop_per_track(coord, fx): """⑨ each track's goal-achievement loop progress is represented.""" plans = coord.fold_goal_loops() assert "trackB-landing" in plans lp = plans["trackB-landing"] assert lp.stages == [ "lint", "refine_9r", "re_lint", "impl", "audit", "adjudication", "callback", "next", ] # track loop walk advances toward the goal st = coord.track_state("trackB-landing") assert st.state == "LANDING_PENDING" coord.advance("trackB-landing", MERGE_READY) assert coord.track_state("trackB-landing").state == MERGE_READY assert coord.track_state("trackB-landing").history == [ "LANDING_PENDING", "MERGE_READY" ] def test_check10_one_track_HOLD_does_not_block_independent(): """⑩ one track HOLD does not block the independent track.""" tracks = [ TrackPlan( track_id="trackA-closeout", task_id="task-2553+13", dispatch_cron_id="DSP-A", normal_collector_cron_id="NC-A", fallback_callback_cron_id="FB-A", depends_on=[], initial_state=HOLD_FOR_CHAIR, ), TrackPlan( track_id="trackB-landing", task_id="task-2553+15", dispatch_cron_id="DSP-B", normal_collector_cron_id="NC-B", fallback_callback_cron_id="FB-B", depends_on=[], initial_state=LANDING_PENDING, ), ] c = ParallelBatchCoordinator(BatchPlan("hold-indep", tracks)) # A is HOLD; B is independent (no depends_on) and must still progress assert c.depmatrix.is_independent("trackA-closeout", "trackB-landing") c.advance("trackB-landing", MERGE_READY) c.advance("trackB-landing", MERGED) assert c.track_state("trackB-landing").state == MERGED # not blocked by A bna = c.batch_next_action() # A's HOLD surfaces for the chair, but B independently reached terminal assert bna["batch_next_action"] == "CHAIR_DECISION_REQUIRED" def test_check11_cross_track_contamination_batch_hold(fx): """⑪ cross-track contamination -> batch-level HOLD escalation.""" sc = fx["contamination_scenario"] tracks = [ TrackPlan( track_id=t["track_id"], task_id=t["track_id"], dispatch_cron_id=f"DSP-{t['track_id']}", normal_collector_cron_id=f"NC-{t['track_id']}", fallback_callback_cron_id=f"FB-{t['track_id']}", own_artifacts=t["own_artifacts"], cited_artifacts=t["cited_artifacts"], initial_state="RUNNING", ) for t in sc["tracks"] ] c = ParallelBatchCoordinator( BatchPlan(batch_label="contam", tracks=tracks) ) assert c.contamination() # non-empty assert ( c.batch_next_action()["batch_next_action"] == "BATCH_HOLD_CONTAMINATION" ) # -------------------------------------------------------------------------- # 9-R.3 transition table — legal / illegal / terminal / precedence / ceiling # -------------------------------------------------------------------------- def test_9r3_legal_path_full_merge_lifecycle(): st = TrackLoopState("t", state=PLANNED) for nxt in [ DISPATCHED, "RUNNING", "NORMAL_COLLECTOR_COMPLETED", "CODEX_AUDIT_PENDING", "ANU_ADJUDICATION_PENDING", LANDING_PENDING, MERGE_READY, MERGED, ]: st.transition(nxt) assert st.state == MERGED def test_9r3_illegal_transition_rejected(): st = TrackLoopState("t", state=PLANNED) with pytest.raises(ValueError): st.transition(MERGED) assert st.rejected and st.rejected[-1][1] == MERGED def test_9r3_terminal_has_no_successor(): for term in TERMINAL_STATES: st = TrackLoopState("t", state=term) assert st.state in ALL_STATES with pytest.raises(ValueError): st.transition(DISPATCHED) def test_9r3_precedence_hold_gt_retry_gt_microfix(): st = TrackLoopState("t", state="ANU_ADJUDICATION_PENDING") pick = st.resolve_next( ["AUTO_MICRO_FIX", RETRY_WITHIN_SCOPE, HOLD_FOR_CHAIR] ) assert pick == HOLD_FOR_CHAIR st2 = TrackLoopState("t", state="ANU_ADJUDICATION_PENDING") assert st2.resolve_next(["AUTO_MICRO_FIX", RETRY_WITHIN_SCOPE]) == ( RETRY_WITHIN_SCOPE ) def test_9r3_retry_ceiling_forces_hold(): st = TrackLoopState("t", state="ANU_ADJUDICATION_PENDING", retry_ceiling=1) st.transition(RETRY_WITHIN_SCOPE) # count -> 1 st.transition(DISPATCHED) st.transition("RUNNING") st.transition("NORMAL_COLLECTOR_COMPLETED") st.transition("CODEX_AUDIT_PENDING") st.transition("ANU_ADJUDICATION_PENDING") ok, _ = st.can_transition(RETRY_WITHIN_SCOPE) assert not ok # ceiling exhausted assert st.resolve_next([RETRY_WITHIN_SCOPE, HOLD_FOR_CHAIR]) == ( HOLD_FOR_CHAIR ) def test_9r3_table_is_13_states_and_self_consistent(): assert len(ALL_STATES) == 13 for src, dsts in __import__( "anu_v3.track_loop_state", fromlist=["TRANSITIONS"] ).TRANSITIONS.items(): for d in dsts: assert d in ALL_STATES assert is_legal_transition(src, d) # -------------------------------------------------------------------------- # 9-R.5 fixture provenance / source immutability # -------------------------------------------------------------------------- def test_9r5_source_markers_read_only(fx): for s in fx["provenance"]["sources"]: p = REPO / s["path"] assert p.exists(), f"source marker missing: {s['path']}" assert _sha(p) == s["sha256"], ( f"source marker mutated (9-R.5 violation): {s['path']}" ) # -------------------------------------------------------------------------- # 9-R.1 frozen anchor + git tracked HEAD/branch/ref invariant # -------------------------------------------------------------------------- def test_9r1_frozen_anchor_byte0(): for rel in FROZEN_ANCHORS: p = REPO / rel assert p.exists(), f"frozen anchor missing: {rel}" assert _sha(p) == FROZEN_ANCHOR_SHA[rel], ( f"frozen anchor changed (9-R.1 violation): {rel}" ) def test_9r1_git_head_branch_ref_unchanged(): def g(*args): return subprocess.run( ["git", "-C", str(REPO), *args], capture_output=True, text=True, check=True, ).stdout.strip() assert g("rev-parse", "HEAD") == "20456b5f83fc039f2fd6f50f4b94095c29b41bfb" assert g("rev-parse", "--abbrev-ref", "HEAD") == ( "task/task-2553p1-f1-clean-replacement" ) assert g("symbolic-ref", "HEAD") == ( "refs/heads/task/task-2553p1-f1-clean-replacement" ) # -------------------------------------------------------------------------- # 9-R.7 Track3 split rule — zero Track2 silent drop # -------------------------------------------------------------------------- def test_9r7_splittable_conflict_followup_no_drop(): gr = GoalRequest("g1", "goal", policy_profile=PolicyProfile("p")) plan = generate_loop_plan(gr.bind_track("trackB-landing"), ["batch_next_action"]) assert plan.status == "FOLLOW_UP_SPLIT" assert plan.split_followups == ["batch_next_action"] def test_9r7_irreducible_conflict_holds(): gr = GoalRequest("g1", "goal", policy_profile=PolicyProfile("p")) plan = generate_loop_plan( gr.bind_track("trackB-landing"), ["final_authority_packet"] ) assert plan.status == "HOLD_FOR_CHAIR" def test_9r7_no_conflict_planned(): gr = GoalRequest("g1", "goal", policy_profile=PolicyProfile("p")) plan = generate_loop_plan(gr.bind_track("trackB-landing")) assert plan.status == "PLANNED" assert plan.stages[0] == "lint" and plan.stages[-1] == "next" # -------------------------------------------------------------------------- # task-2553+19 — Codex HIGH-1..4 fix guards + end-to-end callback path # (Codex LOW: classify_callbacks -> batch_next_action was never exercised # end-to-end; that gap is closed here as the precondition for the HIGH # regression guards.) # -------------------------------------------------------------------------- def _ev(coord, track_id, kind, **kw): """CallbackEvent whose observed 4-tuple matches the registry record unless explicitly overridden (HIGH-1: observed, not reconstructed).""" rec = coord.registry.get(track_id) base = dict( track_id=track_id, task_id=rec.task_id, kind=kind, identity_ok=True, normal_already_completed=False, observed_dispatch_cron_id=rec.dispatch_cron_id, observed_normal_collector_cron_id=rec.normal_collector_cron_id, observed_fallback_callback_cron_id=rec.fallback_callback_cron_id, ) base.update(kw) return CallbackEvent(**base) def test_p19_e2e_classify_to_batch_next_action_callback_path(coord): """Codex LOW: classify_callbacks -> batch_next_action end-to-end. A clean normal on the terminal track + a non-blocking pending fallback on the in-loop track must (a) be keyed by bare track_id (HIGH-2), (b) not block the chair (HIGH-3) -> CONTINUE_LOOP.""" events = [ _ev(coord, "trackA-closeout", "normal"), _ev( coord, "trackB-landing", "fallback", authority_settled_via_normal=True, batch_final_depends_on_track=False, ), ] classes = coord.classify_callbacks(events) assert classes["trackA-closeout"] == "NORMAL_COLLECTOR_ACCEPTED" assert classes["trackB-landing"] == "CALLBACK_PENDING" bna = coord.batch_next_action(classes) by_track = {t["track_id"]: t for t in bna["tracks"]} # HIGH-2: classification actually reached the batch decision assert by_track["trackB-landing"]["callback_class"] == "CALLBACK_PENDING" # HIGH-3: non-blocking pending does not block the chair assert by_track["trackB-landing"]["blocks_chair"] is False assert bna["batch_next_action"] == "CONTINUE_LOOP" def test_p19_high1_observed_tuple_mismatch_detected(coord): """HIGH-1: a wrong *observed* cron-ID is caught even when the event self-declares identity_ok=True (no registry self-reconstruction).""" good = _ev(coord, "trackB-landing", "normal") bad = _ev( coord, "trackB-landing", "normal", observed_dispatch_cron_id="DSP-WRONG", ) assert coord.classify_callbacks([good])["trackB-landing"] == ( "NORMAL_COLLECTOR_ACCEPTED" ) assert coord.classify_callbacks([bad])["trackB-landing"] == ( "TRACK_MISMATCH" ) def test_p19_high2_classify_key_is_bare_track_id(coord): """HIGH-2: classify_callbacks keys are bare track_ids (the exact key batch_next_action looks up), never the old ``track_id:kind`` form.""" classes = coord.classify_callbacks( [_ev(coord, "trackA-closeout", "normal")] ) assert set(classes) == {"trackA-closeout"} assert "trackA-closeout:normal" not in classes bna = coord.batch_next_action(classes) a = next(t for t in bna["tracks"] if t["track_id"] == "trackA-closeout") assert a["callback_class"] == "NORMAL_COLLECTOR_ACCEPTED" # not "" def test_p19_high3_blocking_pending_blocks_else_not(coord): """HIGH-3: pending_blocks_chair_decision drives blocking, not the bare CALLBACK_PENDING class.""" non_blocking = [ _ev( coord, "trackB-landing", "fallback", authority_settled_via_normal=True, batch_final_depends_on_track=False, ) ] classes = coord.classify_callbacks(non_blocking) assert coord.batch_next_action(classes)["batch_next_action"] != ( "AWAIT_PENDING_CALLBACK" ) blocking = [ _ev( coord, "trackB-landing", "fallback", authority_settled_via_normal=False, batch_final_depends_on_track=True, ) ] classes2 = coord.classify_callbacks(blocking) assert coord.batch_next_action(classes2)["batch_next_action"] == ( "AWAIT_PENDING_CALLBACK" ) def test_p19_persisted_state_reflects_classified_callbacks(coord, tmp_path): """NEW-HIGH (cycle2): build_state()/run() must persist a batch_next_action that reflects the last classify_callbacks() pass — not a blank default that silently drops a blocking pending callback.""" blocking = [ _ev( coord, "trackB-landing", "fallback", authority_settled_via_normal=False, batch_final_depends_on_track=True, ) ] classes = coord.classify_callbacks(blocking) assert coord.batch_next_action(classes)["batch_next_action"] == ( "AWAIT_PENDING_CALLBACK" ) # persisted state must agree (no silent drop) assert coord.build_state()["batch_next_action"] == "AWAIT_PENDING_CALLBACK" res = coord.run(tmp_path / "s.json", tmp_path / "s.md") assert res.state["batch_next_action"] == "AWAIT_PENDING_CALLBACK" persisted = json.loads((tmp_path / "s.json").read_text()) assert persisted["batch_next_action"] == "AWAIT_PENDING_CALLBACK" def test_p19_high4_tracked_rejected_untracked_allowed(coord, tmp_path): """HIGH-4 / 9-R.2 full envelope: tracked -> refuse; NEW untracked -> allow; sanctioned coordinator deliverable idempotent re-write -> allow; pre-existing untracked NON-deliverable -> refuse (no clobber).""" # untracked NEW deliverable -> allowed sp = tmp_path / "state.json" mp = tmp_path / "summary.md" assert _is_git_tracked(sp) is False coord.write_state(sp) coord.write_summary(mp) assert sp.exists() and mp.exists() # idempotent overwrite-by-design of the coordinator's own deliverable # (now existing, carries the schema/summary marker) -> still allowed coord.write_state(sp) coord.write_summary(mp) # pre-existing UNTRACKED non-deliverable -> refused (never clobber) foreign = tmp_path / "unrelated.json" foreign.write_text('{"not":"a coordinator deliverable"}', encoding="utf-8") foreign_before = foreign.read_text(encoding="utf-8") with pytest.raises(TrackedPathWriteRefused): coord.write_state(foreign) with pytest.raises(TrackedPathWriteRefused): coord.write_summary(foreign) assert foreign.read_text(encoding="utf-8") == foreign_before # but a path whose NAME is the sanctioned deliverable suffix is allowed named = tmp_path / "task-2553.parallel-batch-state.json" named.write_text("{}", encoding="utf-8") coord.write_state(named) # sanctioned by name suffix -> allowed # git-tracked path -> refused, file untouched (byte-0). Uses a genuinely # git-tracked frozen anchor; the guard rejects BEFORE any write_text so # the anchor stays byte-0 (asserted via sha). tracked = REPO / "anu_v3/goal_activation_controller.py" assert _is_git_tracked(tracked) is True before = _sha(tracked) with pytest.raises(TrackedPathWriteRefused): coord.write_state(tracked) with pytest.raises(TrackedPathWriteRefused): coord.write_summary(tracked) assert _sha(tracked) == before