# -*- coding: utf-8 -*- """anu_v3.parallel_runtime_registry — code/file-level runtime registry. task-2553+29 NO-CRON variant (9-R.1). The registry auto-registers a track at dispatch, then reconciles dispatch / result / normal-callback / fallback / collector / closeout state purely by reading files — it registers/removes ZERO crons (§7, 9-R.1). Self-completion of +29 itself is recognised by the existence of result.json + .done (dogfooding, §12). Standalone (9-R.3): zero import/mutation of anu_v3.parallel_batch_coordinator; the frozen anu_v3.batch_join_policy / parallel_batch_state durable v1 files are read-only inputs only. batch_state authority == the NEW writable file memory/events/task-2553.parallel-runtime-registry.batch-state.json. """ from __future__ import annotations import json from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional from anu_v3.callback_4tuple_index import Callback4TupleIndex, Tuple4 from anu_v3.result_ready_recovery import ( RuntimeObservation, classify_runtime, classify_fallback_fire, is_recovery_eligible, recovery_note, can_recover_now, ) @dataclass class TaskRuntimeRecord: """구현목표 1·2 — the record written at dispatch time and reconciled.""" task_id: str executor: str dispatch_cron_id: str fallback_callback_cron_id: str expected_artifacts: List[str] = field(default_factory=list) dispatch_status: str = "ok" expected_normal_collector_cron_id: Optional[str] = None normal_collector_registered: bool = False normal_collector_executed: bool = False by_design_no_normal_collector: bool = False result_present: bool = False done_present: bool = False fallback_state: str = "PENDING" fallback_fire_kst: Optional[str] = None terminal_outcome: str = "UNKNOWN" hold_for_chair: bool = False # derived classification: str = "" recovery_eligible: bool = False recovery_note_text: str = "" track_mismatch_reasons: List[str] = field(default_factory=list) def to_json(self) -> Dict[str, object]: return { "task_id": self.task_id, "executor": self.executor, "dispatch_cron_id": self.dispatch_cron_id, "dispatch_status": self.dispatch_status, "expected_normal_collector_cron_id": self.expected_normal_collector_cron_id, "fallback_callback_cron_id": self.fallback_callback_cron_id, "expected_artifacts": list(self.expected_artifacts), "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, "result_present": self.result_present, "done_present": self.done_present, "fallback_state": self.fallback_state, "fallback_fire_kst": self.fallback_fire_kst, "classification": self.classification, "recovery_eligible": self.recovery_eligible, "recovery_note": self.recovery_note_text, "terminal_outcome": self.terminal_outcome, "hold_for_chair": self.hold_for_chair, "track_mismatch_reasons": list(self.track_mismatch_reasons), } class ParallelRuntimeRegistry: """In-memory authority registry; emits the authority batch-state file.""" def __init__(self) -> None: self._records: Dict[str, TaskRuntimeRecord] = {} self._index = Callback4TupleIndex() # ---- 구현목표 1·2: dispatch-time registration ------------------------- def register_dispatch(self, rec: TaskRuntimeRecord) -> None: self._records[rec.task_id] = rec self._index.register( Tuple4( task_id=rec.task_id, dispatch_cron_id=rec.dispatch_cron_id, normal_collector_cron_id=rec.expected_normal_collector_cron_id, fallback_callback_cron_id=rec.fallback_callback_cron_id, ) ) def get(self, task_id: str) -> Optional[TaskRuntimeRecord]: return self._records.get(task_id) # ---- observation helpers --------------------------------------------- def _obs(self, rec: TaskRuntimeRecord) -> RuntimeObservation: return RuntimeObservation( dispatch_ok=(rec.dispatch_status == "ok"), result_present=rec.result_present, done_present=rec.done_present, normal_collector_executed=rec.normal_collector_executed, by_design_no_normal_collector=rec.by_design_no_normal_collector, fallback_state=rec.fallback_state, ) # ---- 구현목표 3·4·6·7·8: reconcile one track -------------------------- def reconcile_track(self, task_id: str) -> TaskRuntimeRecord: rec = self._records[task_id] # 구현목표 8 / regression 7-10 — 4-tuple guard runs first. reasons = self._index.validate_tuple(task_id) rec.track_mismatch_reasons = list(reasons) if reasons: rec.classification = "TRACK_MISMATCH" rec.recovery_eligible = False return rec obs = self._obs(rec) rec.classification = classify_runtime(obs) rec.recovery_eligible = is_recovery_eligible(rec.classification) rec.recovery_note_text = recovery_note( rec.classification, rec.by_design_no_normal_collector ) return rec def reconcile_all(self) -> Dict[str, TaskRuntimeRecord]: for tid in list(self._records): self.reconcile_track(tid) return dict(self._records) # ---- 구현목표 7: fallback fire classification ------------------------ def classify_fallback( self, *, claimed_task_id: str, fallback_cron_id: str, fallback_task_id: Optional[str] = None, ) -> str: mismatch = self._index.classify_event( claimed_task_id=claimed_task_id, event_kind="fallback", event_task_id=fallback_task_id, event_cron_id=fallback_cron_id, ) rec = self._records.get(claimed_task_id) if rec is None: return "TRACK_MISMATCH" return classify_fallback_fire(self._obs(rec), track_mismatch=bool(mismatch)) # ---- 구현목표 6: immediate collector recovery feasibility ------------ def collector_recovery_possible(self, task_id: str) -> bool: rec = self._records[task_id] return can_recover_now(self._obs(rec)) def missing_normal_collectors(self) -> List[str]: out: List[str] = [] for tid, rec in self._records.items(): if rec.by_design_no_normal_collector: continue if not rec.normal_collector_executed: out.append(tid) return sorted(out) # ---- §12 / 9-R.1: dogfooding self-completion (NO-CRON) --------------- @staticmethod def self_completion_recognized( result_path: Path, done_path: Path ) -> bool: """+29 recognises its own completion by result.json + .done existence. No cron callback is registered or removed (§7, 9-R.1). The registry reconciles its own terminal state the same way it reconciles tracks. """ return result_path.is_file() and done_path.is_file() # ---- serialisation ---------------------------------------------------- def records_json(self) -> Dict[str, Dict[str, object]]: return {tid: rec.to_json() for tid, rec in self._records.items()} @staticmethod def load_fixture(path: Path) -> "ParallelRuntimeRegistry": data = json.loads(Path(path).read_text(encoding="utf-8")) reg = ParallelRuntimeRegistry() for t in data.get("tracks", []): reg.register_dispatch( TaskRuntimeRecord( task_id=t["task_id"], executor=t.get("executor", ""), dispatch_cron_id=t["dispatch_cron_id"], fallback_callback_cron_id=t["fallback_callback_cron_id"], expected_artifacts=list(t.get("expected_artifacts", [])), dispatch_status=t.get("dispatch_status", "ok"), expected_normal_collector_cron_id=t.get( "expected_normal_collector_cron_id" ), normal_collector_registered=t.get( "normal_collector_registered", False ), normal_collector_executed=t.get( "normal_collector_executed", False ), by_design_no_normal_collector=t.get( "by_design_no_normal_collector", False ), result_present=t.get("result_present", False), done_present=t.get("done_present", False), fallback_state=t.get("fallback_state", "PENDING"), fallback_fire_kst=t.get("fallback_fire_kst"), terminal_outcome=t.get("terminal_outcome", "UNKNOWN"), hold_for_chair=t.get("hold_for_chair", False), ) ) return reg