"""Phase state machine for RPI workflow."""
from enum import Enum
from typing import Self
from pydantic import BaseModel
class Phase(str, Enum):
"""Workflow phases."""
RESEARCH = "research"
PLAN = "plan"
IMPLEMENT = "implement"
# Valid transitions: forward-only
_VALID_TRANSITIONS: dict[Phase, set[Phase]] = {
Phase.RESEARCH: {Phase.PLAN},
Phase.PLAN: {Phase.IMPLEMENT},
Phase.IMPLEMENT: set(), # Terminal phase
}
class PhaseState(BaseModel):
"""Tracks current phase and enforces transition guards."""
session_id: str
current_phase: Phase = Phase.RESEARCH
def can_transition_to(self, target: Phase) -> bool:
"""Check if transition to target phase is valid."""
return target in _VALID_TRANSITIONS.get(self.current_phase, set())
def transition_to(self, target: Phase) -> Self:
"""Transition to target phase. Raises ValueError if invalid."""
if not self.can_transition_to(target):
raise ValueError(
f"Invalid transition: {self.current_phase.value} -> {target.value}"
)
return self.model_copy(update={"current_phase": target})
@property
def is_research(self) -> bool:
return self.current_phase == Phase.RESEARCH
@property
def is_plan(self) -> bool:
return self.current_phase == Phase.PLAN
@property
def is_implement(self) -> bool:
return self.current_phase == Phase.IMPLEMENT