File size: 2,481 Bytes
82f262a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | """
RuntimeFSM - governing state machine for the MORPH-AI pipeline.
Pure logic, no ML dependencies. Enforces legal phase transitions so the
model can only be reached from valid states.
"""
from typing import Dict, List, Set, Tuple
class IllegalTransitionError(Exception):
"""Raised when a state transition is not in the whitelist."""
class RuntimeFSM:
TRANSITIONS: Dict[str, Set[str]] = {
"IDLE": {"INTAKE"},
"INTAKE": {"GUARD_IN"},
"GUARD_IN": {"VISION", "AUDIO", "VIDEO", "DOCUMENT", "SEARCH_GATE", "ROUTED", "GEN", "RESPOND"},
"VISION": {"SEARCH_GATE"},
"AUDIO": {"SEARCH_GATE"},
"VIDEO": {"SEARCH_GATE"},
"DOCUMENT": {"SEARCH_GATE"},
"SEARCH_GATE": {"SEARCH", "ROUTED", "GEN"},
"SEARCH": {"FACT_EXTRACT"},
"FACT_EXTRACT": {"ROUTED"},
"ROUTED": {"GEN"},
"GEN": {"REFINE", "VERIFY", "TOOL_USE"},
"TOOL_USE": {"VERIFY"},
"REFINE": {"VERIFY"},
"VERIFY": {"GEN", "GUARD_OUT", "RESPOND"},
"GUARD_OUT": {"RESPOND"},
"RESPOND": {"IDLE"},
"FAULT": {"IDLE", "FAULT"},
}
def __init__(self, start: str = "IDLE"):
self._state = start
self._journal: List[Tuple[str, str]] = []
@property
def state(self) -> str:
return self._state
def can(self, next_state: str) -> bool:
return next_state in self.TRANSITIONS.get(self._state, set())
def transition(self, next_state: str) -> str:
if next_state not in self.TRANSITIONS.get(self._state, set()):
raise IllegalTransitionError(
f"Illegal transition: {self._state} -> {next_state}"
)
self._journal.append((self._state, next_state))
self._state = next_state
return self._state
def fault(self) -> str:
"""Global trap state - reachable from anywhere."""
self._journal.append((self._state, "FAULT"))
self._state = "FAULT"
return self._state
def recover(self) -> str:
if self._state == "FAULT":
return self.transition("IDLE")
raise IllegalTransitionError(f"Cannot recover from {self._state}")
def reset(self) -> str:
self._journal.append((self._state, "IDLE"))
self._state = "IDLE"
return self._state
def journal(self) -> List[Tuple[str, str]]:
return list(self._journal)
def __repr__(self) -> str:
return f"RuntimeFSM(state={self._state})" |