# -*- coding: utf-8 -*- """anu_v3.runtime_batch_state_updater — versioned/additive batch_state PROPOSAL. task-2553+31 ANU_RUNTIME_RECONCILE_CHECKPOINT leaf module (구현목표 14·15). 9-R.1 HARD BOUNDARY: "batch_state 자동 갱신" == a single NEW versioned/additive batch_state PROPOSAL artifact only: memory/events/task-2553.runtime-reconcile-checkpoint.batch-state.json The frozen durable v1 (memory/events/task-2553.parallel-batch-state.json) and ANY pre-existing batch-state-class file are read-only inputs: write/overwrite is ABSOLUTELY ZERO. The decision/derivation surface (build_proposal) is pure and writes nothing. Emission (emit_runtime_batch_state) is a SEPARATE, hard-guarded I/O function outside the decision boundary that refuses every forbidden path. """ from __future__ import annotations import hashlib import json from pathlib import Path from typing import Dict, List, Optional SCHEMA = "anu_v3.runtime_reconcile_checkpoint.batch_state.v1" BATCH_ID = "batch-task-2553-runtime-reconcile-checkpoint-2553p31" # Only this exact relative path may ever be emitted (9-R.1). ALLOWED_EMIT_RELPATH = ( "memory/events/task-2553.runtime-reconcile-checkpoint.batch-state.json" ) # Never-write list (frozen durable v1 + every other batch-state-class file). FORBIDDEN_EMIT_RELPATHS = frozenset({ "memory/events/task-2553.parallel-batch-state.json", # durable v1 "memory/events/task-2553.generic-batch-state.json", # +30 "memory/events/task-2553.parallel-runtime-registry.batch-state.json", # +29 }) class FrozenWriteRefused(RuntimeError): """Raised when emission targets a frozen / non-allowlisted path (9-R.1).""" def _sha256(path: Path) -> str: return hashlib.sha256(Path(path).read_bytes()).hexdigest() def build_proposal( track_records: Dict[str, dict], *, source_fixture: str, source_fixture_sha256: str, frozen_v1_ref: str, frozen_v1_sha256: str, generated_ts_kst: str, contamination: Optional[List[dict]] = None, prior_version: int = 0, ) -> Dict[str, object]: """Pure builder — derives the additive PROPOSAL doc. Writes NOTHING. `prior_version` lets callers chain versions additively (regression 13); the emitted doc is versioned and never overwrites durable v1. """ contamination = list(contamination or []) terminal_states = { "NO_CRON_TASK_DONE", "RESULT_READY_NO_NORMAL_CALLBACK", "NORMAL_COLLECTOR_COMPLETED", "DUPLICATE_CALLBACK_IGNORED", "TRACK_MISMATCH", } classifications = { tid: rec.get("classification") for tid, rec in track_records.items() } all_terminal = bool(track_records) and all( c in terminal_states for c in classifications.values() ) any_mismatch = any( c == "TRACK_MISMATCH" for c in classifications.values() ) blocking_any = any( rec.get("next_action", {}).get("blocking", False) for rec in track_records.values() ) if any_mismatch or contamination: batch_next_action = "RECOMMEND_HOLD_FOR_CHAIR" elif all_terminal and not blocking_any: batch_next_action = "RECOMMEND_BATCH_ACCEPT" else: batch_next_action = "RECOMMEND_CONTINUE_LOOP" # closeout proposal — derived only, NEVER confirmed (regression 16, §7). closeout_eligible = all_terminal and not any_mismatch and not contamination closeout_proposal = { "eligible": closeout_eligible, "derived_from": "runtime_reconcile_checkpoint.batch_state (read-only)", "confirmed": False, # hard-pinned — checkpoint has no closeout authority "authority_required": "CHAIR (§7/§9 closeout 확정 금지)", "reason": ( "all tracks terminal & no mismatch/contamination — closeout " "PROPOSED only; chair confirmation still required (§7)." if closeout_eligible else "not all tracks terminal or mismatch/contamination present — " "closeout NOT eligible." ), } return { "schema": SCHEMA, "artifact_kind": "ADDITIVE_VERSIONED_PROPOSAL", "batch_id": BATCH_ID, "version": prior_version + 1, "supersedes_durable_v1": False, "durable_v1_write": "ZERO (read-only provenance input, 9-R.1)", "source_fixture": source_fixture, "source_fixture_sha256": source_fixture_sha256, "frozen_batch_state_v1_ref": frozen_v1_ref, "frozen_batch_state_v1_sha256": frozen_v1_sha256, "frozen_coupling": ( "read-only provenance only; zero mutation/coupling (9-R.1)" ), "generated_ts_kst": generated_ts_kst, "tracks": track_records, "contamination": contamination, "batch_next_action": batch_next_action, "closeout_proposal": closeout_proposal, "write_authority": ( "PROPOSAL artifact only — checkpoint performs zero write to " "durable v1 / coordinator / any existing batch-state (§7/§10)" ), "callback_paths_status": { "primary_callback_path": "documented & NOT disabled (§2, reg 17)", "fallback_safety_path": "documented & NOT disabled (§2, reg 18)", "cancel_on_success": "compatible & unchanged (§2, reg 19)", }, } def emit_runtime_batch_state( doc: Dict[str, object], out_path: Path, repo_root: Path, ) -> str: """Hard-guarded emitter — the ONLY function here that touches disk. Refuses (FrozenWriteRefused) unless `out_path` resolves to exactly ``/memory/events/task-2553.runtime-reconcile-checkpoint .batch-state.json`` (9-R.1 allowlist). Never overwrites durable v1 or any other batch-state-class file. Returns the written file's sha256. """ repo_root = Path(repo_root).resolve() out_path = Path(out_path).resolve() try: rel = out_path.relative_to(repo_root).as_posix() except ValueError as exc: # outside repo root raise FrozenWriteRefused( f"emit path escapes repo_root: {out_path}" ) from exc if rel in FORBIDDEN_EMIT_RELPATHS: raise FrozenWriteRefused( f"refused: {rel} is a frozen/durable batch-state file (9-R.1)" ) if rel != ALLOWED_EMIT_RELPATH: raise FrozenWriteRefused( f"refused: {rel} not the allowlisted PROPOSAL path " f"({ALLOWED_EMIT_RELPATH}) (9-R.1)" ) out_path.parent.mkdir(parents=True, exist_ok=True) payload = json.dumps(doc, ensure_ascii=False, indent=2) out_path.write_text(payload + "\n", encoding="utf-8") return _sha256(out_path)