chimera-gfx-Public/tools/phase10ao_worker_supervisor_model.py

#!/usr/bin/env python3
# SPDX-License-Identifier: GPL-3.0-or-later
"""Host-only supervisor/worker preemption architecture model."""

from __future__ import annotations

from dataclasses import dataclass


MAX_COPY_SIZE = 128 * 1024 * 1024
MAX_TICKS = 256
MAX_EVENTS = 16

CREATE_WORKER = "CREATE_WORKER"
VERIFY_WORKER = "VERIFY_WORKER"
START_COPY = "START_COPY"
RECEIVE_RESULT = "RECEIVE_RESULT"
DEADLINE = "DEADLINE"
TERMINATE_WORKER = "TERMINATE_WORKER"
REAP_WORKER = "REAP_WORKER"
TERMINATE_CHILD = "TERMINATE_CHILD"
REAP_CHILD = "REAP_CHILD"
OK = "OK"
ERROR = "ERROR"
SUCCESS = "SUCCESS"
COPY_ERROR = "COPY_ERROR"
RESTORE_ERROR = "RESTORE_ERROR"

OPERATIONS = {
    CREATE_WORKER, VERIFY_WORKER, START_COPY, RECEIVE_RESULT, DEADLINE,
    TERMINATE_WORKER, REAP_WORKER, TERMINATE_CHILD, REAP_CHILD,
}


class SupervisorError(RuntimeError):
    """The synthetic supervisor cannot prove containment."""


@dataclass(frozen=True)
class WorkerPlan:
    worker_id: int
    child_id: int
    copy_length: int

    def __post_init__(self) -> None:
        for value in (self.worker_id, self.child_id, self.copy_length):
            if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
                raise SupervisorError("worker plan value is invalid")
        if self.worker_id == self.child_id or self.copy_length > MAX_COPY_SIZE:
            raise SupervisorError("worker plan identity or size is invalid")


@dataclass(frozen=True)
class WorkerResult:
    worker_id: int
    status: str
    copied: int
    restore_failure_bits: int

    def __post_init__(self) -> None:
        if not isinstance(self.worker_id, int) or isinstance(self.worker_id, bool) \
                or self.worker_id <= 0 or self.status not in {
                    SUCCESS, COPY_ERROR, RESTORE_ERROR}:
            raise SupervisorError("worker result identity/status is invalid")
        if not isinstance(self.copied, int) or isinstance(self.copied, bool) \
                or self.copied < 0 or not isinstance(self.restore_failure_bits, int) \
                or isinstance(self.restore_failure_bits, bool) \
                or not 0 <= self.restore_failure_bits <= 3:
            raise SupervisorError("worker result fields are invalid")


@dataclass(frozen=True)
class FakeSupervisorEvent:
    operation: str
    result: str
    worker_result: WorkerResult | None = None
    ticks: int = 1

    def __post_init__(self) -> None:
        if self.operation not in OPERATIONS or self.result not in {OK, ERROR}:
            raise SupervisorError("supervisor event is invalid")
        if (self.operation == RECEIVE_RESULT) != (self.worker_result is not None):
            raise SupervisorError("worker result binding is invalid")
        if not isinstance(self.ticks, int) or isinstance(self.ticks, bool) \
                or not 1 <= self.ticks <= MAX_TICKS:
            raise SupervisorError("supervisor event ticks are invalid")


class FakeSupervisorFacade:
    """Exact fake operations with no process, signal, clock or IPC access."""

    def __init__(self, events: tuple[FakeSupervisorEvent, ...]) -> None:
        if not isinstance(events, tuple) or not 1 <= len(events) <= MAX_EVENTS \
                or any(type(item) is not FakeSupervisorEvent for item in events):
            raise SupervisorError("supervisor script is invalid")
        self._events = list(events)
        self.ticks = 0

    def invoke(self, operation: str, terminal: bool = False) -> FakeSupervisorEvent:
        if not self._events:
            raise SupervisorError("supervisor script is exhausted")
        event = self._events.pop(0)
        if event.operation != operation:
            raise SupervisorError("supervisor event ordering differs")
        if not terminal and self.ticks + event.ticks > MAX_TICKS:
            raise SupervisorError("supervisor deadline precedes operation")
        self.ticks += event.ticks
        return event

    @property
    def remaining(self) -> int:
        return len(self._events)


@dataclass(frozen=True)
class SupervisorOutcome:
    classification: str
    success: bool
    service_alive: bool
    worker_alive: bool
    child_alive: bool
    copied: int
    automatic_restart: bool = False
    target_action_performed: bool = False
    firmware_behavior_proven: bool = False


def _terminal(facade: FakeSupervisorFacade, operation: str) -> None:
    if facade.invoke(operation, terminal=True).result != OK:
        raise SupervisorError(f"terminal operation failed: {operation}")


def run_supervisor(plan: WorkerPlan, facade: FakeSupervisorFacade,
                   deadline_first: bool = False) -> SupervisorOutcome:
    """Run one synthetic, non-restarting worker attempt."""
    if type(plan) is not WorkerPlan or type(facade) is not FakeSupervisorFacade \
            or type(deadline_first) is not bool:
        raise SupervisorError("supervisor model boundary is invalid")
    worker = started = False
    copied = 0
    failed = deadline = False

    if facade.invoke(CREATE_WORKER).result != OK:
        failed = True
    else:
        worker = True
        if facade.invoke(VERIFY_WORKER).result != OK:
            failed = True
        elif facade.invoke(START_COPY).result != OK:
            started = True
            failed = True
        else:
            started = True
            if deadline_first:
                if facade.invoke(DEADLINE).result != OK:
                    raise SupervisorError("deadline event failed")
                deadline = failed = True
            else:
                event = facade.invoke(RECEIVE_RESULT)
                result = event.worker_result
                if event.result != OK or result is None:
                    failed = True
                else:
                    copied = result.copied
                    valid = (result.worker_id == plan.worker_id and
                             result.status == SUCCESS and
                             result.copied == plan.copy_length and
                             result.restore_failure_bits == 0)
                    failed = not valid

    if worker:
        if failed:
            _terminal(facade, TERMINATE_WORKER)
        _terminal(facade, REAP_WORKER)
        worker = False
    if failed and started:
        _terminal(facade, TERMINATE_CHILD)
        _terminal(facade, REAP_CHILD)
    if facade.remaining:
        raise SupervisorError("supervisor script has unused operations")

    if not failed:
        return SupervisorOutcome("OFFLINE_WORKER_RESULT_ACCEPTED", True, True,
                                 False, True, copied)
    classification = ("OFFLINE_DEADLINE_CONTAINED" if deadline else
                      "OFFLINE_WORKER_FAILURE_CONTAINED")
    return SupervisorOutcome(classification, False, True, False,
                             not started, copied)