State Machine

State Machine - Finite State Machines for Complex Flows

Description

Enforces deterministic, minimal state machines for any multi-step flow in NodeJS-Starter-V1. Codifies the project's existing patterns (TaskStatus, ExecutionStatus, NodeStatus) and provides a reusable framework for defining states, transition maps, guard conditions, and side effects across Python and TypeScript.

When to Apply

Positive Triggers

• Designing multi-step workflows with distinct status phases
• Adding new Enum status fields to Pydantic models
• Implementing UI components with loading/error/success states
• Reviewing state transitions for completeness and determinism
• Adding retry, escalation, or verification loops
• User mentions: "state machine", "status", "workflow state", "transitions", "FSM"

Negative Triggers

• Defining simple boolean flags (is_active, is_published)
• Implementing form validation (use data-validation instead)
• Designing animation state transitions (use scientific-luxury + Framer Motion)
• Classifying error types without state transitions (use error-taxonomy instead)

Core Directives

The Three Laws of State Machines

1. Deterministic: Every (state, event) pair produces exactly one next state
2. Minimal: No unreachable states, no duplicate transitions
3. Complete: Every state has defined exit transitions or is a terminal state

State Definition Convention

Use str, Enum for all status fields. String enums serialise cleanly to JSON and are readable in logs.

from enum import Enum

class OrderStatus(str, Enum):
    """Status of an order — each value is a distinct FSM state."""

    DRAFT = "draft"
    SUBMITTED = "submitted"
    PROCESSING = "processing"
    COMPLETED = "completed"
    CANCELLED = "cancelled"

// Frontend mirror — use const assertion, not enum
const ORDER_STATUS = {
  DRAFT: 'draft',
  SUBMITTED: 'submitted',
  PROCESSING: 'processing',
  COMPLETED: 'completed',
  CANCELLED: 'cancelled',
} as const;

type OrderStatus = (typeof ORDER_STATUS)[keyof typeof ORDER_STATUS];

Existing Project State Machines

TaskStatus (Orchestrator)

Location: apps/backend/src/agents/orchestrator.py

10 states governing the task lifecycle with verification loop:

PENDING → IN_PROGRESS → AWAITING_VERIFICATION
                              ↓
                    VERIFICATION_IN_PROGRESS
                         ↙          ↘
           VERIFICATION_PASSED    VERIFICATION_FAILED
                    ↓                    ↓
               COMPLETED          (retry → IN_PROGRESS)
                                  (max retries → ESCALATED_TO_HUMAN)

Cross-cutting:
  Any non-terminal → BLOCKED (dependency unmet)
  Any non-terminal → FAILED (unrecoverable error)

Terminal states: COMPLETED, FAILED, ESCALATED_TO_HUMAN

ExecutionStatus (Workflow Engine)

Location: apps/backend/src/workflow/models.py

4 states for workflow execution lifecycle:

PENDING → RUNNING → COMPLETED
                  → FAILED

Terminal states: COMPLETED, FAILED

NodeStatus (Workflow Nodes)

Location: apps/backend/src/workflow/state.py

5 states for individual node execution:

PENDING → RUNNING → COMPLETED
                  → FAILED
       → SKIPPED (conditional branch not taken)

Terminal states: COMPLETED, FAILED, SKIPPED

Design Patterns

Pattern 1: Transition Map

Define allowed transitions as a dictionary. Reject any transition not in the map.

from typing import TypeVar

S = TypeVar("S", bound=Enum)

TASK_TRANSITIONS: dict[TaskStatus, set[TaskStatus]] = {
    TaskStatus.PENDING: {
        TaskStatus.IN_PROGRESS,
        TaskStatus.BLOCKED,
        TaskStatus.FAILED,
    },
    TaskStatus.IN_PROGRESS: {
        TaskStatus.AWAITING_VERIFICATION,
        TaskStatus.BLOCKED,
        TaskStatus.FAILED,
    },
    TaskStatus.AWAITING_VERIFICATION: {
        TaskStatus.VERIFICATION_IN_PROGRESS,
    },
    TaskStatus.VERIFICATION_IN_PROGRESS: {
        TaskStatus.VERIFICATION_PASSED,
        TaskStatus.VERIFICATION_FAILED,
    },
    TaskStatus.VERIFICATION_PASSED: {
        TaskStatus.COMPLETED,
    },
    TaskStatus.VERIFICATION_FAILED: {
        TaskStatus.IN_PROGRESS,  # Retry
        TaskStatus.ESCALATED_TO_HUMAN,  # Max retries exceeded
    },
    # Terminal states — no outgoing transitions
    TaskStatus.COMPLETED: set(),
    TaskStatus.FAILED: set(),
    TaskStatus.BLOCKED: {
        TaskStatus.PENDING,  # Unblocked
    },
    TaskStatus.ESCALATED_TO_HUMAN: set(),
}


def validate_transition(
    current: TaskStatus,
    target: TaskStatus,
    transitions: dict[TaskStatus, set[TaskStatus]] = TASK_TRANSITIONS,
) -> bool:
    """Check if a state transition is allowed."""
    allowed = transitions.get(current, set())
    return target in allowed


def transition(
    current: TaskStatus,
    target: TaskStatus,
) -> TaskStatus:
    """Execute a validated state transition."""
    if not validate_transition(current, target):
        raise ValueError(
            f"Invalid transition: {current.value} → {target.value}"
        )
    return target

Pattern 2: Guard Conditions

Guards are predicates that must be true before a transition fires.

from dataclasses import dataclass
from typing import Callable


@dataclass
class GuardedTransition:
    """A transition with a guard condition."""

    source: TaskStatus
    target: TaskStatus
    guard: Callable[[TaskState], bool]
    description: str


GUARDED_TRANSITIONS = [
    GuardedTransition(
        source=TaskStatus.VERIFICATION_FAILED,
        target=TaskStatus.IN_PROGRESS,
        guard=lambda task: task.attempts < task.max_attempts,
        description="Retry only if under max attempts",
    ),
    GuardedTransition(
        source=TaskStatus.VERIFICATION_FAILED,
        target=TaskStatus.ESCALATED_TO_HUMAN,
        guard=lambda task: task.attempts >= task.max_attempts,
        description="Escalate after max retry attempts",
    ),
    GuardedTransition(
        source=TaskStatus.BLOCKED,
        target=TaskStatus.PENDING,
        guard=lambda task: len(task.error_history) == 0,
        description="Unblock only if blocking condition resolved",
    ),
]

Pattern 3: Side Effects on Transition

Execute side effects (logging, notifications, metrics) when transitions occur.

import structlog

logger = structlog.get_logger(__name__)


async def on_transition(
    task: TaskState,
    old_status: TaskStatus,
    new_status: TaskStatus,
) -> None:
    """Execute side effects after a state transition."""

    logger.info(
        "task_transition",
        task_id=task.task_id,
        from_status=old_status.value,
        to_status=new_status.value,
    )

    # Terminal state side effects
    if new_status == TaskStatus.COMPLETED:
        task.completed_at = datetime.now().isoformat()

    if new_status == TaskStatus.ESCALATED_TO_HUMAN:
        logger.warning(
            "task_escalated",
            task_id=task.task_id,
            attempts=task.attempts,
        )

    if new_status == TaskStatus.FAILED:
        logger.error(
            "task_failed",
            task_id=task.task_id,
            error_history=task.error_history,
        )

Pattern 4: Frontend State Machine (TypeScript)

Use a reducer pattern for UI state machines:

type FetchState =
  | { status: 'idle' }
  | { status: 'loading' }
  | { status: 'success'; data: unknown }
  | { status: 'error'; error: string };

type FetchEvent =
  | { type: 'FETCH' }
  | { type: 'RESOLVE'; data: unknown }
  | { type: 'REJECT'; error: string }
  | { type: 'RESET' };

function fetchReducer(state: FetchState, event: FetchEvent): FetchState {
  switch (state.status) {
    case 'idle':
      if (event.type === 'FETCH') return { status: 'loading' };
      return state;

    case 'loading':
      if (event.type === 'RESOLVE') return { status: 'success', data: event.data };
      if (event.type === 'REJECT') return { status: 'error', error: event.error };
      return state;

    case 'success':
      if (event.type === 'FETCH') return { status: 'loading' };
      if (event.type === 'RESET') return { status: 'idle' };
      return state;

    case 'error':
      if (event.type === 'FETCH') return { status: 'loading' };
      if (event.type === 'RESET') return { status: 'idle' };
      return state;
  }
}

Usage with React:

function useStateMachine() {
  const [state, dispatch] = useReducer(fetchReducer, { status: 'idle' });
  return { state, dispatch };
}

Creating a New State Machine

Step 1: Enumerate States

List every distinct status. Apply the Turing Check — no redundant states:

class MyStatus(str, Enum):
    """Each state must be reachable and have at least one exit or be terminal."""

    STATE_A = "state_a"
    STATE_B = "state_b"
    STATE_C = "state_c"  # Terminal

Step 2: Define Transition Map

MY_TRANSITIONS: dict[MyStatus, set[MyStatus]] = {
    MyStatus.STATE_A: {MyStatus.STATE_B},
    MyStatus.STATE_B: {MyStatus.STATE_C},
    MyStatus.STATE_C: set(),  # Terminal
}

Step 3: Add Guards (If Needed)

Only add guards when a transition depends on runtime data.

Step 4: Add Side Effects (If Needed)

Logging, metrics, notifications on transition.

Step 5: Validate Completeness

def validate_fsm(transitions: dict[Enum, set[Enum]]) -> list[str]:
    """Validate a finite state machine for completeness."""
    issues: list[str] = []
    all_states = set(transitions.keys())
    all_targets = {t for targets in transitions.values() for t in targets}

    # Check for unreachable states (not targets and not initial)
    initial = list(transitions.keys())[0]
    unreachable = all_states - all_targets - {initial}
    for state in unreachable:
        issues.append(f"Unreachable state: {state.value}")

    # Check for undefined targets
    undefined = all_targets - all_states
    for state in undefined:
        issues.append(f"Undefined target state: {state.value}")

    # Check terminal states have no transitions
    for state, targets in transitions.items():
        if len(targets) == 0:
            continue  # Valid terminal
        # Non-terminal must have at least one reachable target
        if not targets.intersection(all_states):
            issues.append(f"Dead end: {state.value} targets undefined states")

    return issues

Anti-Patterns

Checklist for New State Machines

Design

• All states enumerated as str, Enum
• Transition map defines every (state, target) pair
• Terminal states have empty transition sets
• No unreachable states
• No undefined target states
• Guards documented for conditional transitions

Implementation

• validate_transition() called before every status change
• Side effects fire after transition, not before
• Logging includes from_status and to_status
• Frontend mirrors backend states exactly (snake_case)

Council of Logic

• Turing: Transition lookup is O(1) via dict/set
• Von Neumann: State changes are atomic — no partial transitions
• Shannon: Enum values are minimal — no redundant states

Response Format

[AGENT_ACTIVATED]: State Machine
[PHASE]: {Design | Implementation | Review}
[STATUS]: {in_progress | complete}

{state machine analysis or implementation guidance}

[NEXT_ACTION]: {what to do next}

Integration Points

Council of Logic (Turing Check)

• State machines must be deterministic and minimal
• Transition lookup must be O(1) — use dict, not if/elif chains
• No unreachable or duplicate states

Error Taxonomy

• Invalid transitions raise WORKFLOW_VALIDATION_INVALID_TRANSITION
• Terminal error states map to the appropriate error code and severity

Structured Logging

• Every state transition logs task_transition event with from_status, to_status
• Terminal states log completion or failure with context

API Contract

• Status enums are part of the API response contract
• Frontend Zod schemas must include all enum values
• Adding a new enum value is non-breaking; removing is breaking

Australian Localisation (en-AU)

• Spelling: serialise, analyse, optimise, behaviour, colour
• Date Format: ISO 8601 in transit; DD/MM/YYYY in UI display
• Timezone: AEST/AEDT — timestamps stored as UTC