115 lines
3.7 KiB
Python
115 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from zoneinfo import ZoneInfo
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PLAN_PATH = ROOT / "implementation" / "WORK_PLAN.json"
|
|
STATE_PATH = ROOT / "BUILD_STATE.json"
|
|
|
|
ALLOWED_PACKET_STATUSES = {
|
|
"pending",
|
|
"in_progress",
|
|
"completed",
|
|
"blocked",
|
|
"failed",
|
|
}
|
|
|
|
|
|
class PackError(RuntimeError):
|
|
"""Raised when the design-pack state is structurally invalid."""
|
|
|
|
|
|
def load_json(path: Path) -> Any:
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except FileNotFoundError as exc:
|
|
raise PackError(f"missing file: {path.relative_to(ROOT)}") from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise PackError(
|
|
f"invalid JSON in {path.relative_to(ROOT)}: "
|
|
f"line {exc.lineno}, column {exc.colno}: {exc.msg}"
|
|
) from exc
|
|
|
|
|
|
def load_plan() -> dict[str, Any]:
|
|
value = load_json(PLAN_PATH)
|
|
if not isinstance(value, dict) or not isinstance(value.get("packets"), list):
|
|
raise PackError("implementation/WORK_PLAN.json must contain a packets array")
|
|
return value
|
|
|
|
|
|
def load_state() -> dict[str, Any]:
|
|
value = load_json(STATE_PATH)
|
|
if not isinstance(value, dict) or not isinstance(value.get("packets"), dict):
|
|
raise PackError("BUILD_STATE.json must contain a packets object")
|
|
return value
|
|
|
|
|
|
def packet_map(plan: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
|
result: dict[str, dict[str, Any]] = {}
|
|
for packet in plan["packets"]:
|
|
packet_id = packet.get("id") if isinstance(packet, dict) else None
|
|
if not isinstance(packet_id, str) or not packet_id:
|
|
raise PackError("every work-plan packet requires a non-empty id")
|
|
if packet_id in result:
|
|
raise PackError(f"duplicate work packet id: {packet_id}")
|
|
result[packet_id] = packet
|
|
return result
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PacketSelection:
|
|
packet: dict[str, Any] | None
|
|
reason: str
|
|
|
|
|
|
def select_next_packet(plan: dict[str, Any], state: dict[str, Any]) -> PacketSelection:
|
|
packets = packet_map(plan)
|
|
statuses = state.get("packets", {})
|
|
|
|
unknown = set(statuses) - set(packets)
|
|
missing = set(packets) - set(statuses)
|
|
if unknown or missing:
|
|
raise PackError(
|
|
f"work-plan/state mismatch; unknown={sorted(unknown)}, missing={sorted(missing)}"
|
|
)
|
|
|
|
for packet in plan["packets"]:
|
|
packet_id = packet["id"]
|
|
if statuses[packet_id] == "in_progress":
|
|
return PacketSelection(packet, "resume in-progress packet")
|
|
|
|
for packet in plan["packets"]:
|
|
packet_id = packet["id"]
|
|
if statuses[packet_id] not in {"pending", "failed"}:
|
|
continue
|
|
dependencies = packet.get("deps", [])
|
|
if all(statuses.get(dep) == "completed" for dep in dependencies):
|
|
reason = (
|
|
"retry failed packet"
|
|
if statuses[packet_id] == "failed"
|
|
else "first ready pending packet"
|
|
)
|
|
return PacketSelection(packet, reason)
|
|
|
|
blocked = [pid for pid, status in statuses.items() if status == "blocked"]
|
|
unfinished = [pid for pid, status in statuses.items() if status != "completed"]
|
|
if not unfinished:
|
|
return PacketSelection(None, "all packets completed")
|
|
if blocked:
|
|
return PacketSelection(None, f"no ready packet; blocked packets: {', '.join(blocked)}")
|
|
return PacketSelection(None, "no packet is ready; dependency or state inconsistency")
|
|
|
|
|
|
def brussels_now() -> str:
|
|
return datetime.now(ZoneInfo("Europe/Brussels")).isoformat(timespec="seconds")
|
|
|
|
|
|
def relative(path: Path) -> str:
|
|
return path.resolve().relative_to(ROOT.resolve()).as_posix()
|