# -*- coding: utf-8 -*- """anu_v3.runtime_reconcile_checkpoint — ANU runtime reconcile checkpoint. task-2553+31 ANU_RUNTIME_RECONCILE_CHECKPOINT orchestrator (§1/§2/§11). Solves the operationalization gap: tools existed (+29 registry / +30 generic coordinator) but did not run automatically — ANU only discovered completion when the chair asked. This module is the read-only checkpoint that ANU can auto-wire at an active-dispatch turn boundary OR call before responding (§11 결선) so a finished NO-CRON task with no normal callback is detected WITHOUT a chair question (§1/§15 dogfooding). DESIGN BOUNDARY (§10, regression 12/15): * read-only detection only — ZERO write / cron / merge / dispatch / closeout * batch_state output == ONE additive versioned PROPOSAL artifact (9-R.1) * next_action == recommendation only * callback primary/safety/cancel-on-success paths preserved, NOT replaced (§2) """ from __future__ import annotations import hashlib import json from dataclasses import dataclass from pathlib import Path from typing import Dict, List, Optional from anu_v3.active_dispatch_scanner import ActiveDispatchScanner, DispatchRecord from anu_v3.task_artifact_detector import TaskArtifactDetector from anu_v3.runtime_next_action_resolver import ( resolve, TERMINAL, NONTERMINAL, ALL_CLASSIFICATIONS, NO_CRON_TASK_DONE, RESULT_READY_NO_NORMAL_CALLBACK, NORMAL_COLLECTOR_COMPLETED, DUPLICATE_CALLBACK_IGNORED, TRACK_MISMATCH, RUNNING, WAIT_FOR_RESULT, FALLBACK_PENDING, STALE_OR_BOT_STUCK_CANDIDATE, ) from anu_v3.runtime_batch_state_updater import ( build_proposal, emit_runtime_batch_state, FrozenWriteRefused, ) RESULT_SCHEMA = "anu_v3.runtime_reconcile_checkpoint.result.v1" # Default stale threshold: a dispatched track with no result after this many # seconds past its fallback fire time is a STALE_OR_BOT_STUCK_CANDIDATE. DEFAULT_STALE_SECONDS = 3600 @dataclass(frozen=True) class RuntimeTaskObservation: """Normalized read-only observation for one track (구현목표 1-8).""" task_id: str dispatch_ok: bool result_present: bool done_present: bool normal_collector_registered: bool normal_collector_executed: bool by_design_no_normal_collector: bool fallback_state: str # NONE | PENDING | FIRED | CANCELLED terminal_outcome: str = "" stale: bool = False track_mismatch: bool = False track_mismatch_reasons: Optional[List[str]] = None def to_json(self) -> Dict[str, object]: return { "task_id": self.task_id, "dispatch_ok": self.dispatch_ok, "result_present": self.result_present, "done_present": self.done_present, "normal_collector_registered": self.normal_collector_registered, "normal_collector_executed": self.normal_collector_executed, "by_design_no_normal_collector": self.by_design_no_normal_collector, "fallback_state": self.fallback_state, "terminal_outcome": self.terminal_outcome, "stale": self.stale, "track_mismatch": self.track_mismatch, "track_mismatch_reasons": list(self.track_mismatch_reasons or []), } def classify_observation(obs: RuntimeTaskObservation) -> str: """Deterministic taxonomy classifier (9-R.2/9-R.4; §5/§6 정합). Precedence: 1 TRACK_MISMATCH (4-tuple integrity first; reg 8-11) 2 NORMAL_COLLECTOR_COMPLETED (primary path; reg 5, fixture 4) 3 has result: a fallback FIRED -> DUPLICATE_CALLBACK_IGNORED (reg 7) b NO-CRON done & no fallback -> NO_CRON_TASK_DONE (reg 1, fixture 1) c otherwise -> RESULT_READY_NO_NORMAL_CALLBACK (reg 2/6, fixture 2/3) 4 no result: a stale / dispatch !ok -> STALE_OR_BOT_STUCK_CANDIDATE (reg 4) b fallback PENDING -> FALLBACK_PENDING c else -> RUNNING (reg 3) """ if obs.track_mismatch: return TRACK_MISMATCH if obs.normal_collector_executed: return NORMAL_COLLECTOR_COMPLETED has_result = obs.result_present or obs.done_present if has_result: if obs.fallback_state == "FIRED": return DUPLICATE_CALLBACK_IGNORED if ( obs.done_present and obs.by_design_no_normal_collector and obs.fallback_state == "NONE" and not obs.normal_collector_registered ): return NO_CRON_TASK_DONE return RESULT_READY_NO_NORMAL_CALLBACK if obs.stale or not obs.dispatch_ok: return STALE_OR_BOT_STUCK_CANDIDATE if obs.fallback_state == "PENDING": return FALLBACK_PENDING return RUNNING def _four_tuple_mismatch( track: dict, dispatch: Optional[DispatchRecord] ) -> List[str]: """Detect 4-tuple contamination (구현목표 13; reg 8-11). Honours an explicit ``track_mismatch`` flag/reasons in the fixture, then cross-checks task_id / dispatch_cron_id / collector / fallback ownership. """ reasons: List[str] = [] if track.get("track_mismatch"): reasons.extend(track.get("track_mismatch_reasons") or ["explicit track_mismatch flag"]) owner = track["task_id"] binding = track.get("callback_binding", {}) or {} if binding: if binding.get("task_id", owner) != owner: reasons.append("task_id mismatch") if dispatch and binding.get("dispatch_cron_id") and ( binding["dispatch_cron_id"] != dispatch.dispatch_cron_id ): reasons.append("dispatch_cron_id mismatch") nc_owner = binding.get("normal_collector_owner_task") if nc_owner and nc_owner != owner: reasons.append("normal collector belongs to different task") fb_owner = binding.get("fallback_owner_task") if fb_owner and fb_owner != owner: reasons.append("fallback belongs to different task") # de-dup, preserve order seen: set = set() return [r for r in reasons if not (r in seen or seen.add(r))] class RuntimeReconcileCheckpoint: """Read-only checkpoint orchestrator. Auto-wireable entrypoint (§11).""" def __init__( self, repo_root: Path, *, stale_seconds: int = DEFAULT_STALE_SECONDS, ) -> None: self.repo_root = Path(repo_root) self.stale_seconds = stale_seconds self.detector = TaskArtifactDetector(self.repo_root) self.scanner = ActiveDispatchScanner(self.repo_root) # ---- observation building ------------------------------------------- def _observe_track( self, track: dict, dispatch: Optional[DispatchRecord] ) -> RuntimeTaskObservation: """Fold fixture truth + live read-only artifact detection. Fixture values are authoritative for deterministic regression; when a fixture omits result/done presence the live detector fills it (still read-only). Existing +26/+27/+30 artifacts are read/reference only. """ tid = track["task_id"] live = self.detector.detect(tid) result_present = bool(track.get("result_present", live.result_present)) done_present = bool(track.get("done_present", live.done_present)) mismatch_reasons = _four_tuple_mismatch(track, dispatch) return RuntimeTaskObservation( task_id=tid, dispatch_ok=bool(track.get( "dispatch_ok", dispatch.dispatch_ok if dispatch else str(track.get("dispatch_status", "ok")).lower() in ("ok", "fired", "success"))), result_present=result_present, done_present=done_present, normal_collector_registered=bool( track.get("normal_collector_registered", False)), normal_collector_executed=bool( track.get("normal_collector_executed", False)), by_design_no_normal_collector=bool( track.get("by_design_no_normal_collector", False)), fallback_state=str(track.get("fallback_state", "NONE")).upper(), terminal_outcome=str(track.get("terminal_outcome", "")), stale=bool(track.get("stale", False)), track_mismatch=bool(mismatch_reasons), track_mismatch_reasons=mismatch_reasons, ) # ---- core read-only run --------------------------------------------- def run( self, fixture_path: Path, *, generated_ts_kst: str = "", prior_version: int = 0, emit: bool = False, ) -> Dict[str, object]: """Read-only reconcile. Returns the full result document. emit=False (default) writes NOTHING. emit=True writes ONLY the allowlisted additive PROPOSAL artifact via the hard-guarded emitter (9-R.1); any other path raises FrozenWriteRefused. """ fixture_path = Path(fixture_path) raw = json.loads(fixture_path.read_text(encoding="utf-8")) tracks = raw.get("tracks", raw) if isinstance(raw, dict) else raw dispatches = { d.task_id: d for d in ActiveDispatchScanner.from_fixture(fixture_path) } track_records: Dict[str, dict] = {} for track in tracks: tid = track["task_id"] obs = self._observe_track(track, dispatches.get(tid)) classification = classify_observation(obs) na = resolve(classification) rec = obs.to_json() rec["classification"] = classification rec["terminal"] = classification in TERMINAL rec["recovery_eligible"] = ( classification == RESULT_READY_NO_NORMAL_CALLBACK ) rec["next_action"] = na.to_json() rec["expected"] = track.get("expected") # fixture self-check track_records[tid] = rec frozen_v1 = ( self.repo_root / "memory" / "events" / "task-2553.parallel-batch-state.json" ) frozen_v1_sha = ( hashlib.sha256(frozen_v1.read_bytes()).hexdigest() if frozen_v1.is_file() else "" ) proposal = build_proposal( track_records, source_fixture=str(fixture_path), source_fixture_sha256=hashlib.sha256( fixture_path.read_bytes()).hexdigest(), frozen_v1_ref=str(frozen_v1), frozen_v1_sha256=frozen_v1_sha, generated_ts_kst=generated_ts_kst, prior_version=prior_version, ) consolidated = self._consolidated_summary(track_records, proposal) result: Dict[str, object] = { "schema": RESULT_SCHEMA, "task_id": "task-2553+31", "checkpoint": "ANU_RUNTIME_RECONCILE_CHECKPOINT", "mode": "READ_ONLY", "generated_ts_kst": generated_ts_kst, "source_fixture": str(fixture_path), "track_records": track_records, "batch_state_proposal": proposal, "consolidated_summary": consolidated, "zero_side_effect_proof": { "write": 0, "cron": 0, "merge": 0, "dispatch": 0, "closeout_confirm": 0, "note": "checkpoint is read-only detection; emit (opt-in) " "is the single hard-guarded additive PROPOSAL only.", }, } if emit: out = ( self.repo_root / "memory" / "events" / "task-2553.runtime-reconcile-checkpoint.batch-state.json" ) sha = emit_runtime_batch_state(proposal, out, self.repo_root) result["batch_state_proposal_emitted"] = { "path": str(out), "sha256": sha, "kind": "ADDITIVE_VERSIONED_PROPOSAL (9-R.1)", } return result # ---- consolidated summary (구현목표 15, §13) ------------------------ def _consolidated_summary( self, track_records: Dict[str, dict], proposal: Dict[str, object] ) -> Dict[str, object]: by_class: Dict[str, List[str]] = {} for tid, rec in track_records.items(): by_class.setdefault(rec["classification"], []).append(tid) no_cron_done = by_class.get(NO_CRON_TASK_DONE, []) result_ready = by_class.get(RESULT_READY_NO_NORMAL_CALLBACK, []) return { "tracks_total": len(track_records), "by_classification": by_class, "no_cron_task_done": no_cron_done, "result_ready_no_normal_callback": result_ready, "normal_collector_completed": by_class.get(NORMAL_COLLECTOR_COMPLETED, []), "duplicate_callback_ignored": by_class.get(DUPLICATE_CALLBACK_IGNORED, []), "track_mismatch": by_class.get(TRACK_MISMATCH, []), "batch_next_action": proposal["batch_next_action"], "closeout_proposal": proposal["closeout_proposal"], "callback_paths_status": proposal["callback_paths_status"], "fixture_self_check_pass": all( rec.get("expected") in (None, rec["classification"]) for rec in track_records.values() ), } # ---- §11 auto-wireable entrypoint ----------------------------------- def checkpoint_entrypoint( self, fixture_path: Path, *, generated_ts_kst: str = "", ) -> Dict[str, object]: """ANU pre-response entrypoint (§11). Designed to be called at an active-dispatch turn boundary OR by ANU before answering, so a finished NO-CRON task with no normal callback is detected WITHOUT a chair question. Read-only (emit=False). Returns a compact decision packet ANU can act on. """ res = self.run(fixture_path, generated_ts_kst=generated_ts_kst, emit=False) cs = res["consolidated_summary"] return { "auto_wireable": True, "wiring": "active-dispatch turn boundary OR ANU pre-response call", "chair_question_required": False, "no_cron_completion_detected": bool(cs["no_cron_task_done"]), "result_ready_no_normal_callback": cs["result_ready_no_normal_callback"], "batch_next_action": cs["batch_next_action"], "decision_packet": res, } # ---- §15 dogfooding self-detection ---------------------------------- def detect_self_completion(self) -> Dict[str, object]: """+31 self-completion detector (ultimate dogfooding, §15). The checkpoint detects its OWN completion by reading its result.json + .done existence — no cron, no chair question. This is the very mechanism §15 mandates: ANU calls this entrypoint manually to recognize +31 done without asking the chair. """ obs = self.detector.detect("task-2553+31") done_p = ( self.repo_root / "memory" / "events" / "task-2553+31.done" ) result_p = ( self.repo_root / "memory" / "events" / "task-2553.runtime-reconcile-checkpoint.result.json" ) result_present = result_p.is_file() done_present = done_p.is_file() self_obs = RuntimeTaskObservation( task_id="task-2553+31", dispatch_ok=True, result_present=result_present, done_present=done_present, normal_collector_registered=False, normal_collector_executed=False, by_design_no_normal_collector=True, fallback_state="NONE", terminal_outcome="DONE", ) cls = classify_observation(self_obs) return { "task_id": "task-2553+31", "result_json_present": result_present, "done_present": done_present, "classification": cls, "self_completion_detected": cls == NO_CRON_TASK_DONE, "chair_question_required": False, "mechanism": "read-only result.json + .done existence " "(NO-CRON dogfooding, §15)", }