#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ task-2553+27 — ANU_PARALLEL_BATCH_COORDINATOR_V0 lifecycle closeout finalizer (code/file automation; NOT documentation-only). Closes out the coordinator V0 lifecycle produced by: * task-2553+17 — coordinator V0 implementation (result.json / activation-decision.json) * task-2553+19 — ACCEPT/CONVERGED via 회장 옵션 A (adjudication-resolution.json / result.json / coordinator-v0-closeout-packet.md) Behaviour (§3 / 9-R.4 / 9-R.5): * Inputs are READ-ONLY: +17/+19 artifacts + memory/events/task-2553.parallel-batch-state.json + anu_v3/parallel_batch_coordinator.py. They are never written, reverted or mutated. * Verifies the V0 lifecycle is ACCEPT/CONVERGED with the 9-item closeout packet and HIGH-4 RESOLVED (회장 옵션 A) by evidence, then emits an untracked closeout MARKER. * Idempotent: a deterministic, input-addressed marker — re-run yields byte-identical output; if the marker already exists it is preserved (no-op success, no double closeout, pre==post state sha). A pre-existing marker whose bytes diverge from the deterministic recomputation -> HOLD (partial inconsistency, 9-R.4). * Missing / inconsistent +17·+19 evidence -> HOLD (no marker written, §6). * Write surface is restricted to the exact 9-R.5 whitelist; any other path (coordinator code/state, +26, shared, git-tracked) -> hard refusal. Allowed writes (9-R.5, exactly): scripts/run_coordinator_v0_closeout.py tests/regression/test_coordinator_v0_closeout_2553plus27.py memory/events/task-2553+27.coordinator-v0-closeout.json memory/events/task-2553+27.decision.json memory/events/task-2553+27.result.json This finalizer only ever writes the latter three. """ from __future__ import annotations import hashlib import json import subprocess import sys from pathlib import Path TASK_ID = "task-2553+27" # ── 9-R.5 write whitelist (exact, repo-root-relative) ────────────────────────── WRITE_WHITELIST = frozenset({ "scripts/run_coordinator_v0_closeout.py", "tests/regression/test_coordinator_v0_closeout_2553plus27.py", "memory/events/task-2553+27.coordinator-v0-closeout.json", "memory/events/task-2553+27.decision.json", "memory/events/task-2553+27.result.json", }) # Read-only lifecycle evidence inputs (relative to repo root). EVIDENCE = { "p17_result": "memory/events/task-2553+17.result.json", "p17_activation": "memory/events/task-2553+17.activation-decision.json", "p19_adjudication": "memory/events/task-2553+19.adjudication-resolution.json", "p19_result": "memory/events/task-2553+19.result.json", "p19_packet_md": "memory/reports/task-2553+19.coordinator-v0-closeout-packet.md", "batch_state": "memory/events/task-2553.parallel-batch-state.json", } COORDINATOR_CODE = "anu_v3/parallel_batch_coordinator.py" MARKER_REL = "memory/events/task-2553+27.coordinator-v0-closeout.json" DECISION_REL = "memory/events/task-2553+27.decision.json" RESULT_REL = "memory/events/task-2553+27.result.json" # ── helpers ─────────────────────────────────────────────────────────────────── def _sha256_bytes(b: bytes) -> str: return hashlib.sha256(b).hexdigest() def _sha256_file(p: Path) -> str | None: if not p.is_file(): return None return _sha256_bytes(p.read_bytes()) def _canon(obj) -> bytes: """Deterministic JSON encoding (stable bytes across runs).""" return (json.dumps(obj, sort_keys=True, indent=2, ensure_ascii=False) + "\n").encode("utf-8") def find_repo_root(start: Path | None = None) -> Path: """Walk upward until a dir containing memory/events is found.""" cur = (start or Path(__file__).resolve()).resolve() for cand in [cur, *cur.parents]: if (cand / "memory" / "events").is_dir(): return cand raise RuntimeError("repo root (dir containing memory/events) not found") def _git_tracked(repo_root: Path, relpath: str) -> bool: try: r = subprocess.run( ["git", "ls-files", "--error-unmatch", relpath], cwd=str(repo_root), capture_output=True, text=True, ) return r.returncode == 0 except Exception: return False def _git_ref(repo_root: Path) -> dict: out = {} for k, args in ( ("head", ["rev-parse", "HEAD"]), ("branch", ["rev-parse", "--abbrev-ref", "HEAD"]), ("ref", ["symbolic-ref", "HEAD"]), ): try: r = subprocess.run(["git", *args], cwd=str(repo_root), capture_output=True, text=True) out[k] = r.stdout.strip() if r.returncode == 0 else None except Exception: out[k] = None return out def _assert_writable(repo_root: Path, relpath: str) -> Path: """Refuse any write outside the 9-R.5 whitelist or onto a git-tracked path.""" if relpath not in WRITE_WHITELIST: raise PermissionError(f"write refused — not in 9-R.5 whitelist: {relpath}") if _git_tracked(repo_root, relpath): raise PermissionError(f"write refused — git-tracked path immutable: {relpath}") return repo_root / relpath def _safe_write(repo_root: Path, relpath: str, content: bytes) -> str: """ Idempotent guarded write. Returns one of: FRESH_WRITE | NO_OP_BYTE_STABLE | INCONSISTENT Never clobbers a divergent pre-existing file (INCONSISTENT -> caller HOLDs). """ target = _assert_writable(repo_root, relpath) if target.exists(): if target.read_bytes() == content: return "NO_OP_BYTE_STABLE" return "INCONSISTENT" target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(content) return "FRESH_WRITE" # ── evidence gate ───────────────────────────────────────────────────────────── def gather_evidence(repo_root: Path) -> dict: """Read-only collection + verification of +17/+19 lifecycle evidence.""" findings: list[str] = [] shas: dict[str, str | None] = {} docs: dict[str, object] = {} for key, rel in EVIDENCE.items(): p = repo_root / rel shas[rel] = _sha256_file(p) if shas[rel] is None: findings.append(f"MISSING:{rel}") continue if rel.endswith(".json"): try: docs[key] = json.loads(p.read_text("utf-8")) except Exception as e: # noqa: BLE001 findings.append(f"UNPARSEABLE:{rel}:{e}") else: docs[key] = p.read_text("utf-8") coord = repo_root / COORDINATOR_CODE coord_sha = _sha256_file(coord) # If any input is missing/unparseable, HOLD immediately (§6 evidence absence). if findings: return { "ok": False, "verdict": "HOLD_FOR_CHAIR", "reasons": findings, "shas": shas, "coord_sha": coord_sha, } p17r: dict = docs["p17_result"] # type: ignore[assignment] p17a: dict = docs["p17_activation"] # type: ignore[assignment] adj: dict = docs["p19_adjudication"] # type: ignore[assignment] p19r: dict = docs["p19_result"] # type: ignore[assignment] md: str = docs["p19_packet_md"] # type: ignore[assignment] state: dict = docs["batch_state"] # type: ignore[assignment] # +17 — coordinator V0 implemented (DONE/PASS). if str(p17r.get("status")) != "DONE": findings.append(f"P17_STATUS_NOT_DONE:{p17r.get('status')}") if str(p17r.get("classification")) != "PASS": findings.append(f"P17_CLASS_NOT_PASS:{p17r.get('classification')}") impl = (p17r.get("implemented_files", {}) or {}).get("anu_v3_modules_new_6", []) if COORDINATOR_CODE not in impl: findings.append("P17_COORDINATOR_NOT_IN_IMPLEMENTED") if not str(p17a.get("decision", "")).startswith("ACTIVATE"): findings.append(f"P17_NOT_ACTIVATED:{p17a.get('decision')}") # +19 — chair 옵션 A adjudication: OPTION_A / 9-packet / HIGH-4 RESOLVED / ACCEPT-CONVERGED. if not str(adj.get("decision", "")).startswith("OPTION_A"): findings.append(f"P19_NOT_OPTION_A:{adj.get('decision')}") pkt = adj.get("packet_9_items", {}) or {} pkt_keys = sorted(pkt.keys()) if len(pkt_keys) != 9: findings.append(f"P19_PACKET_NOT_9:{len(pkt_keys)}") else: nums = sorted(int(k.split("_", 1)[0]) for k in pkt_keys if k.split("_", 1)[0].isdigit()) if nums != list(range(1, 10)): findings.append(f"P19_PACKET_NUMS:{nums}") item8 = str(pkt.get("8_remaining_high_critical", "")) if not ("HIGH-4" in item8 and "RESOLVED" in item8 and "unresolved HIGH/CRITICAL = 0" in item8): findings.append("P19_HIGH4_NOT_RESOLVED") item9 = str(pkt.get("9_coordinator_v0_final_status", "")) if not ("ACCEPT" in item9 and "CONVERGED" in item9): findings.append("P19_FINAL_NOT_ACCEPT_CONVERGED") # +19 result.json: pre-adjudication HOLD is expected & superseded by 회장 옵션 A; # only flag if it doesn't even acknowledge the HIGH-1..3 convergence scope. cscope = str(p19r.get("converged_scope", "")) if not ("HIGH-1" in cscope and "RESOLVED" in cscope): findings.append("P19_RESULT_SCOPE_UNEXPECTED") # +19 packet md: ACCEPT/CONVERGED + 9 numbered sections. if "ACCEPT" not in md or "CONVERGED" not in md: findings.append("P19_MD_NOT_ACCEPT_CONVERGED") md_secs = sorted({int(ln[3:].split(".", 1)[0]) for ln in md.splitlines() if ln.startswith("## ") and ln[3:].split(".", 1)[0].isdigit()}) if not set(range(1, 10)).issubset(set(md_secs)): findings.append(f"P19_MD_SECTIONS:{md_secs}") # batch state: schema + batch_id present (read-only). if str(state.get("schema")) != "anu_v3.parallel_batch_state.v1": findings.append(f"STATE_SCHEMA:{state.get('schema')}") batch_id = state.get("batch_id") if not batch_id: findings.append("STATE_NO_BATCH_ID") if findings: return { "ok": False, "verdict": "HOLD_FOR_CHAIR", "reasons": findings, "shas": shas, "coord_sha": coord_sha, } return { "ok": True, "verdict": "ACCEPT", "reasons": [], "shas": shas, "coord_sha": coord_sha, "batch_id": batch_id, "closeout_ts_kst": adj.get("ts_kst"), "authority": EVIDENCE["p19_adjudication"], "chair_decision": adj.get("decision"), "packet_9_keys": pkt_keys, "p17_task_md_sha256": p17r.get("task_md_sha256_verified"), } # ── deterministic artifact builders ─────────────────────────────────────────── def _closeout_id(ev: dict) -> str: parts = [ev["batch_id"], ev["verdict"]] for rel in sorted(ev["shas"]): parts.append(f"{rel}:{ev['shas'][rel]}") parts.append(f"{COORDINATOR_CODE}:{ev['coord_sha']}") return _sha256_bytes("|".join(parts).encode("utf-8")) def build_marker(ev: dict) -> bytes: cid = _closeout_id(ev) return _canon({ "schema": "anu_v3.task_2553p27.coordinator_v0_closeout.v1", "task_id": TASK_ID, "closeout_of": "ANU_PARALLEL_BATCH_COORDINATOR_V0 lifecycle (+17 impl / +19 옵션 A ACCEPT-CONVERGED)", "batch_id": ev["batch_id"], "final_status": "ACCEPT / CONVERGED", "authority": ev["authority"], "chair_decision": ev["chair_decision"], "closeout_ts_kst": ev["closeout_ts_kst"], "closeout_id": cid, "nine_packet_check": {"result": "PASS", "keys": ev["packet_9_keys"]}, "high4_resolved": True, "high4_resolution_path": "회장 옵션 A spec-clarification (adjudication resolution, 거짓 수렴 아님)", "accept_converged": True, "evidence_inputs_sha256": ev["shas"], "coordinator_code_immutable": { "path": COORDINATOR_CODE, "sha256": ev["coord_sha"], "mutated_by_closeout": False, }, "state_file_immutable": { "path": EVIDENCE["batch_state"], "sha256": ev["shas"][EVIDENCE["batch_state"]], "read_only_preserve": True, }, "generated_by": "scripts/run_coordinator_v0_closeout.py (code/file automation, not md-only)", "idempotent": "input-addressed; re-run byte-identical; pre-existing marker preserved (no double-closeout)", "cross_track_contamination": 0, "untracked_marker": True, }) def build_decision(ev: dict) -> bytes: """ Pure deterministic function of the read-only evidence — byte-stable across re-runs (9-R.4). Run-variant data (which idempotent branch was taken, git invariant) lives only in result.json, never in this persisted decision. state/coordinator are never written by this finalizer, so pre==post by construction; only the single immutable sha is recorded here. """ return _canon({ "schema": "anu_v3.task_2553p27.closeout_decision.v1", "task_id": TASK_ID, "decision": "CLOSEOUT_FINALIZED", "lifecycle": "ANU_PARALLEL_BATCH_COORDINATOR_V0 (+17 impl / +19 옵션 A ACCEPT-CONVERGED)", "authority": ev["authority"], "chair_decision": ev.get("chair_decision"), "evidence_refs": { "p17_result": EVIDENCE["p17_result"], "p17_activation": EVIDENCE["p17_activation"], "p19_adjudication": EVIDENCE["p19_adjudication"], "p19_result": EVIDENCE["p19_result"], "p19_packet_md": EVIDENCE["p19_packet_md"], "batch_state": EVIDENCE["batch_state"], }, "evidence_inputs_sha256": ev["shas"], "nine_packet_check": "PASS (9 items 1..9)", "high4_resolved": True, "accept_converged": True, "state_file_immutable_sha256": ev["shas"][EVIDENCE["batch_state"]], "coordinator_code_immutable_sha256": ev["coord_sha"], "read_only_preserve": "+17/+19 evidence, batch-state, coordinator code never written/reverted", "idempotent_semantics": "input-addressed; re-run byte-identical; pre-existing -> preserve (no double-closeout)", }) # ── orchestrator ────────────────────────────────────────────────────────────── def run_closeout(repo_root: Path | None = None, *, write: bool = True) -> dict: repo_root = repo_root or find_repo_root() git_pre = _git_ref(repo_root) ev = gather_evidence(repo_root) state_p = repo_root / EVIDENCE["batch_state"] coord_p = repo_root / COORDINATOR_CODE if not ev["ok"]: # §6 HOLD: write nothing to marker; surface hold packet only. git_post = _git_ref(repo_root) return { "verdict": "HOLD_FOR_CHAIR", "ok": False, "reasons": ev["reasons"], "marker_branch": "NOT_WRITTEN", "git_invariant": {"pre": git_pre, "post": git_post, "equal": git_pre == git_post}, "state_sha_pre": _sha256_file(state_p), "state_sha_post": _sha256_file(state_p), "coord_sha_pre": _sha256_file(coord_p), "coord_sha_post": _sha256_file(coord_p), } marker = build_marker(ev) if write: m_branch = _safe_write(repo_root, MARKER_REL, marker) else: existing = (repo_root / MARKER_REL).read_bytes() if (repo_root / MARKER_REL).exists() else None m_branch = ("NO_OP_BYTE_STABLE" if existing == marker else "INCONSISTENT" if existing is not None else "WOULD_WRITE") state_sha_post = _sha256_file(state_p) coord_sha_post = _sha256_file(coord_p) if m_branch == "INCONSISTENT": # 9-R.4: pre-existing marker diverges from deterministic recompute -> HOLD. git_post = _git_ref(repo_root) return { "verdict": "HOLD_FOR_CHAIR", "ok": False, "reasons": ["MARKER_INCONSISTENT_WITH_DETERMINISTIC_RECOMPUTE"], "marker_branch": m_branch, "git_invariant": {"pre": git_pre, "post": git_post, "equal": git_pre == git_post}, "state_sha_pre": ev["shas"][EVIDENCE["batch_state"]], "state_sha_post": state_sha_post, "coord_sha_pre": ev["coord_sha"], "coord_sha_post": coord_sha_post, } decision = build_decision(ev) if write: d_branch = _safe_write(repo_root, DECISION_REL, decision) if d_branch == "INCONSISTENT": git_post = _git_ref(repo_root) return { "verdict": "HOLD_FOR_CHAIR", "ok": False, "reasons": ["DECISION_INCONSISTENT_WITH_DETERMINISTIC_RECOMPUTE"], "marker_branch": m_branch, "git_invariant": {"pre": git_pre, "post": git_post, "equal": git_pre == git_post}, "state_sha_pre": ev["shas"][EVIDENCE["batch_state"]], "state_sha_post": state_sha_post, "coord_sha_pre": ev["coord_sha"], "coord_sha_post": coord_sha_post, } git_post = _git_ref(repo_root) return { "verdict": "CLOSEOUT_FINALIZED", "ok": True, "reasons": [], "batch_id": ev["batch_id"], "final_status": "ACCEPT / CONVERGED", "closeout_id": _closeout_id(ev), "marker_branch": m_branch, "marker_path": MARKER_REL, "decision_path": DECISION_REL, "no_op_already_closeout": m_branch == "NO_OP_BYTE_STABLE", "git_invariant": {"pre": git_pre, "post": git_post, "equal": git_pre == git_post}, "state_sha_pre": ev["shas"][EVIDENCE["batch_state"]], "state_sha_post": state_sha_post, "state_sha_equal": ev["shas"][EVIDENCE["batch_state"]] == state_sha_post, "coord_sha_pre": ev["coord_sha"], "coord_sha_post": coord_sha_post, "coord_sha_equal": ev["coord_sha"] == coord_sha_post, } if __name__ == "__main__": res = run_closeout() print(json.dumps(res, indent=2, ensure_ascii=False)) sys.exit(0 if res["ok"] else 2)