axiomata

Automata that obey their axioms

A type-safe state machine library that enforces invariants and requirements through a declarative API. Perfect for modeling complex workflows with strict business rules, distributed systems, and out-of-order event handling.

Features

• Invariants: Declare conditions that must hold for state transitions to succeed
• Requirements: Define preconditions that actions must satisfy before execution
• Automatic Event Queuing: Handle out-of-order events with configurable tolerance windows
• Type Safety: Full TypeScript support with comprehensive type inference
• Hierarchical States: Support for nested states and parallel regions
• Functional Design: Pure functions and immutable state updates

Installation

npm install axiomata

Core Concepts

State Machines

An axiomata state machine extends XState with formal verification of business rules through:

1. Actions: Pure functions that transform context based on events
2. Requirements: Validators that gate action execution
3. Invariants: Conditions that must hold for state transitions
4. States: Hierarchical state definitions with entry actions and transitions

Actions

Actions are pure functions that return partial context updates. They can optionally declare requirements that must be satisfied before execution.

type AxiomAction<Context, Event> = (input: { context: Context; event: Event }) => Partial<Context>;

Requirements

Requirements are predicates that validate context and event properties. They return either a boolean or an AxiomVerdict for fine-grained control.

type RequirementValidator<Context, Event> = (input: {
  context: Context;
  event: Event;
}) => boolean | AxiomVerdict;

Invariants

Invariants are conditions that must hold for a transition to succeed. When an invariant fails, the transition is rejected and the machine remains in its current state.

Basic Usage

Defining a State Machine

import {
  createAxiomStateMachine,
  type AxiomMachineState,
  type AxiomMachineContext,
} from 'axiomata';

// Define your types
type ActionKey = 'enterIdle' | 'clockIn' | 'takeBreak' | 'resumeWork' | 'clockOut';
type EventType = 'CLOCK_IN' | 'TAKE_BREAK' | 'RESUME_WORK' | 'CLOCK_OUT';
type RequirementKey = 'stateIsIdle' | 'stateIsActive' | 'stateIsWorking';

// Define your context
type MyContext = AxiomMachineContext & {
  state: { type: 'idle' } | { type: 'active'; phase: 'working' | 'onBreak' };
};

type MyEvent = {
  type: EventType;
  occurredAt: Date;
};

// Create the machine
const machine = createAxiomStateMachine<
  'initial',
  EventType,
  ActionKey,
  never, // No global invariants
  RequirementKey,
  MyContext,
  MyEvent
>({
  id: 'timekeeping',
  initialState: 'idle',
  actions: {
    enterIdle: {
      run: ({ context }) => ({ state: { type: 'idle' } }),
    },
    clockIn: {
      requires: ['stateIsIdle'], // Action gated by requirement
      run: ({ context, event }) => ({
        state: { type: 'active', phase: 'working', startedAt: event.occurredAt },
      }),
    },
    takeBreak: {
      requires: ['stateIsActive', 'stateIsWorking'],
      run: ({ context, event }) => ({
        state: { ...context.state, phase: 'onBreak' },
      }),
    },
  },
  requirements: {
    stateIsIdle: ({ context }) => context.state.type === 'idle',
    stateIsActive: ({ context }) => context.state.type === 'active',
    stateIsWorking: ({ context }) =>
      context.state.type === 'active' && context.state.phase === 'working',
  },
  states: {
    idle: {
      entry: 'enterIdle',
      on: {
        CLOCK_IN: { target: 'active.working', action: 'clockIn' },
        TAKE_BREAK: {
          target: 'idle',
          invariant: [{ requires: 'stateIsWorking', message: 'Cannot take break while idle' }],
        },
      },
    },
    active: {
      initial: 'working',
      states: {
        working: {
          on: {
            TAKE_BREAK: { target: 'active.onBreak', action: 'takeBreak' },
            CLOCK_OUT: { target: 'idle', action: 'clockOut' },
          },
        },
        onBreak: {
          on: {
            RESUME_WORK: { target: 'active.working', action: 'resumeWork' },
            CLOCK_OUT: { target: 'idle', action: 'clockOut' },
          },
        },
      },
    },
  },
});

Using the Machine

// Get initial state
const { initialState } = machine;

// Transition synchronously
const nextState = machine.transition(initialState, {
  type: 'CLOCK_IN',
  occurredAt: new Date(),
});

// Apply transition and get result tuple
const [newState, event] = machine.apply(initialState, {
  type: 'CLOCK_IN',
  occurredAt: new Date(),
});

// Process multiple events
const finalState = machine.reduce(initialState, [
  { type: 'CLOCK_IN', occurredAt: new Date() },
  { type: 'TAKE_BREAK', occurredAt: new Date() },
  { type: 'RESUME_WORK', occurredAt: new Date() },
]);

Advanced Features

Queue for Out-of-Order Events

The queue handles events that arrive out of order, which is common in distributed systems. It maintains per-user queues and retries events within a tolerance window.

const machine = createAxiomStateMachine({
  id: 'rollout',
  queue: {
    shouldRetryAll: true,
    toleranceMs: 1_000, // Wait up to 1 second for earlier events
    buildExpiredError: (event, lastErrorMessage) =>
      new Error(`Event expired: ${event.type}. ${lastErrorMessage}`),
  },
  // ... rest of machine config
});

// Create a queue instance
const queue = machine.createQueue();

// Apply transitions asynchronously with automatic ordering
// The stateId parameter identifies whose state is being tracked (user, order, shift, etc.)
const [newSnapshot, processedEvent] = await queue.applyTransition({
  stateId: 'user-123',
  snapshot: currentSnapshot,
  event: { type: 'ADVANCE_STEP', occurredAt: new Date() },
});

// Configure tolerance at runtime
queue.configureTolerance(2_000);

// Reset all queues
queue.reset();

Queue Behavior

• Events are grouped by state identifier to maintain independent queues per entity
• Events with occurredAt timestamps are ordered chronologically
• If an event arrives too early, it waits for preceding events within the tolerance window
• After the tolerance expires, events that cause violations are rejected
• Retryable errors (based on shouldRetry) cause the queue to wait and retry

Snapshot Refreshing

When events are queued and waiting, the snapshot stored in the queue can become stale. To handle this, you can configure the queue to refresh snapshots from your data source:

const machine = createAxiomStateMachine({
  id: 'timekeeping',
  queue: {
    shouldRetryAll: true,
    toleranceMs: 1_000,
    buildExpiredError: (event, lastErrorMessage) =>
      new Error(`Event expired: ${event.type}. ${lastErrorMessage}`),
    // Optional: Control refresh frequency (default: always refresh)
    snapshotRefreshIntervalMs: 500,
    // Function to fetch fresh data from your database
    // The identifier parameter matches what you pass to applyTransition
    refreshSnapshot: async (identifier, staleSnapshot) => {
      const freshData = await fetchFromDatabase(identifier);
      return buildSnapshot(freshData);
    },
  },
  // ... rest of machine config
});

How it works:

• If refreshSnapshot is provided without snapshotRefreshIntervalMs (or set to 0 or negative), the snapshot is refreshed every time a queued event is processed
• If snapshotRefreshIntervalMs is set to a positive number, the snapshot is only refreshed when that interval has elapsed
• Snapshot refreshing is only disabled if you don't provide a refreshSnapshot function
• The fresh snapshot is used for processing the current and subsequent events
• The identifier passed to refreshSnapshot is the same value you provide to applyTransition (e.g., a user ID, order ID, or any entity identifier)

Use cases:

• Systems where entity state can be modified by other processes while events are queued
• Long-running queues where database state may change while events wait
• Systems requiring strong consistency guarantees despite out-of-order event processing
• Use snapshotRefreshIntervalMs to reduce database load when perfect freshness isn't required

Invariants and Requirements

Action Requirements

Requirements gate action execution. If any requirement fails, the action doesn't run and the transition is rejected.

actions: {
  promoteRollout: {
    requires: ['stateIsActive', 'hasNoOpenIncident'],
    run: ({ context, event }) => ({
      state: { type: 'idle' },
      lastRollout: buildCompletedRollout(context, event),
    }),
  },
}

Transition Invariants

Invariants validate that a transition is legal. They can reference requirements or be standalone validators.

states: {
  idle: {
    on: {
      START_ROLLOUT: { target: 'rollout.ramping', action: 'startRollout' },
      ADVANCE_STEP: {
        target: 'idle',
        invariant: [{ requires: 'stateIsActive', message: 'No rollout is currently active.' }],
      },
    },
  },
  rollout: {
    initial: 'ramping',
    states: {
      ramping: {
        on: {
          ADVANCE_STEP: {
            target: 'rollout.ramping',
            action: 'advanceStep',
            invariant: [
              { requires: 'stateIsActive', message: 'No rollout is currently active.' },
              { requires: 'hasNextStep', message: 'Rollout already at final step.' },
            ],
          },
        },
      },
    },
  },
}

Multiple Transitions per Event

You can define multiple possible transitions for the same event, each with different invariants. The first matching transition is taken.

on: {
  RESUME: [
    {
      target: 'rollout.ramping',
      action: 'resumeRollout',
      invariant: [
        { requires: ['stateIsPaused', 'resumePhaseIsRamping'], message: 'Not paused in ramping' },
      ],
    },
    {
      target: 'rollout.monitoring',
      action: 'resumeRollout',
      invariant: [
        { requires: ['stateIsPaused', 'resumePhaseIsMonitoring'], message: 'Not paused in monitoring' },
      ],
    },
  ],
}

Requirement Definitions

Requirements can be defined as simple predicates or with advanced patterns:

requirements: {
  // Simple predicate
  stateIsIdle: ({ context }) => context.state.type === 'idle',

  // Multiple requirements combined
  stateIsRamping: ({ context }) =>
    context.state.type === 'active' && context.state.phase === 'ramping',

  // Complex validation with verdict
  hasValidExposure: ({ context }) => {
    if (context.state.type !== 'active') return false;
    if (context.state.exposurePct < 0 || context.state.exposurePct > 100) {
      return {
        kind: 'violate',
        code: 'InvalidExposure',
        detail: `Exposure ${context.state.exposurePct} out of range`,
      };
    }
    return true;
  },
}

Helper Utilities

Building Snapshots from Timelines

For testing or replay scenarios, build machine snapshots from event sequences:

import { getSnapshot } from 'axiomata';

const events = [
  { type: 'START_ROLLOUT', occurredAt: new Date('2025-01-01T08:00:00Z') },
  { type: 'ADVANCE_STEP', occurredAt: new Date('2025-01-01T09:00:00Z') },
  { type: 'PROMOTE', occurredAt: new Date('2025-01-01T18:00:00Z') },
];

const snapshot = events.reduce(
  (state, event) => getSnapshot(machine.apply(state, event)),
  machine.initialState,
);

Error Handling

import { AxiomTransitionRejectedError } from 'axiomata';

try {
  const nextState = machine.transition(currentState, event);
} catch (error) {
  if (error instanceof AxiomTransitionRejectedError) {
    console.error('Transition rejected:', error.message);
    console.error('Machine ID:', error.machineId);
    console.error('Event type:', error.eventType);
    console.error('Reject code:', error.code);
  }
}

Type Parameters

State machines require several type parameters:

createAxiomStateMachine<
  ContextKey, // Union of context initializer keys
  EventType, // Union of event type strings
  ActionKey, // Union of action keys
  InvariantKey, // Union of global invariant keys (use 'never' if none)
  RequirementKey, // Union of requirement keys
  Context, // Context type (extends AxiomMachineContext)
  Event // Event type (must have a 'type' property)
>;

Best Practices

Naming Conventions

Use descriptive names for better error messages:

• Actions: Imperative verbs (startRollout, pauseRollout, recordIncident)
• Requirements: Boolean predicates (stateIsActive, hasNextStep, hasNoOpenIncident)
• Events: Past tense or commands (START_ROLLOUT, PAUSE, GUARDRAIL_TRIP)

Error Messages

Provide clear, actionable error messages in invariants:

invariant: [
  { requires: 'stateIsActive', message: 'No rollout is currently active.' },
  { requires: 'hasNextStep', message: 'Rollout already at final step.' },
];

Immutable Updates

Always return new objects from actions; never mutate context:

// Good
run: ({ context, event }) => ({
  state: { ...context.state, phase: 'paused' },
  updatedAt: event.occurredAt,
});

// Bad - mutates context
run: ({ context, event }) => {
  context.state.phase = 'paused'; // Don't do this!
  return context;
};

Requirement Composition

Keep requirements atomic and compose them:

requirements: {
  stateIsActive: ({ context }) => context.state.type === 'active',
  stateIsRamping: ({ context }) =>
    context.state.type === 'active' && context.state.phase === 'ramping',
  hasNextStep: ({ context }) =>
    context.state.type === 'active' && context.state.currentStepIndex < LAST_INDEX,
},
actions: {
  advanceStep: {
    requires: ['stateIsActive', 'stateIsRamping', 'hasNextStep'],
    run: ({ context }) => ({ /* ... */ }),
  },
}

Testing

Test both valid transitions and invariant violations:

describe('state machine', () => {
  it('allows valid transitions', () => {
    const nextState = machine.transition(initialState, {
      type: 'START_ROLLOUT',
      occurredAt: new Date(),
    });
    expect(nextState.context.state.type).toBe('active');
  });

  it('rejects invalid transitions', () => {
    expect(() =>
      machine.transition(initialState, {
        type: 'ADVANCE_STEP',
        occurredAt: new Date(),
      }),
    ).toThrow('No rollout is currently active.');
  });
});

Real-World Example

See the included example machines for comprehensive patterns:

• Progressive Rollout Machine (progressiveRolloutStateMachine.ts): Models a multi-stage feature rollout with incident tracking, pause/resume, and monitoring phases

The example demonstrate:

• Complex hierarchical states
• Multiple invariants per transition
• Queue configuration for distributed events
• Helper functions for building snapshots from timelines
• Comprehensive test coverage

API Reference

createAxiomStateMachine(config)

Creates a state machine with invariants and requirements.

Config Properties:

• id: string - Unique identifier for the machine
• initialState: string - Name of the initial state
• actions: Record<ActionKey, Action | ActionDescriptor> - Action definitions
• states: Record<string, StateNode> - State hierarchy
• requirements?: Record<RequirementKey, Validator> - Requirement validators
• queue?: QueueConfig - Optional queue configuration

Returns:

• machine - XState machine instance
• initialState - Initial state snapshot
• transition(state, event) - Synchronous transition function
• apply(state, event) - Returns [newState, event] tuple
• reduce(state, events) - Process multiple events
• createQueue() - Create a queue instance for async operations

Queue Configuration

queue: {
  shouldRetryAll: boolean; // Retry all events after ordering
  toleranceMs: number; // Wait window for out-of-order events
  buildExpiredError: (event, lastErrorMessage) => Error;
}

Verdict Types

type AxiomVerdictPass = { kind: 'pass' };
type AxiomVerdictReject = { kind: 'reject'; code: string; detail?: string };
type AxiomVerdictViolate = { kind: 'violate'; code: string; detail?: string };

License

MIT