| """ |
| 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})" |
