Files
aios/active-runtime/app/aios_control_plane_runtime_status.py
T

49 lines
1.9 KiB
Python

"""Separate read, write, and release plane readiness evaluation."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Mapping
@dataclass(frozen=True, slots=True)
class RuntimeStatus:
read_plane: Mapping[str, bool]
write_plane: Mapping[str, bool]
release_plane: Mapping[str, bool]
required_facts: Mapping[str, frozenset[str]] | None = None
def _readiness_reason(self) -> str:
if self.required_facts is None:
return "REQUIRED_FACT_INVENTORY_MISSING"
expected_groups = {"READ_PLANE", "WRITE_PLANE", "RELEASE_PLANE"}
if set(self.required_facts) != expected_groups:
return "REQUIRED_FACT_INVENTORY_INVALID"
planes = {
"READ_PLANE": self.read_plane,
"WRITE_PLANE": self.write_plane,
"RELEASE_PLANE": self.release_plane,
}
for group, facts in self.required_facts.items():
if not isinstance(facts, frozenset) or not facts or any(not isinstance(name, str) or not name for name in facts):
return "REQUIRED_FACT_INVENTORY_INVALID"
plane = planes[group]
if set(plane) != set(facts):
return "REQUIRED_FACT_INVENTORY_MISMATCH"
if any(type(value) is not bool or value is not True for value in plane.values()):
return "REQUIRED_FACT_NOT_TRUE"
return "READY"
@property
def release_control_plane_ready(self) -> bool:
return self._readiness_reason() == "READY"
def as_dict(self) -> dict[str, object]:
return {
"READ_PLANE": dict(self.read_plane),
"WRITE_PLANE": dict(self.write_plane),
"RELEASE_PLANE": dict(self.release_plane),
"release_control_plane_ready": self.release_control_plane_ready,
"readiness_reason": self._readiness_reason(),
}