Stateforward SML for Python

A small, operator-composable Python state-machine runtime. It uses the Stateforward SML vocabulary while keeping Python callback, object, threading, and asyncio semantics explicit.

The distribution/project name is stateforward.sml; the public import is import sml. The repository version is 0.1.0 (sml.__version__). This README covers this Python runtime only; it makes no claim of behavioral, allocation, or performance equivalence with another implementation.

Contents

• Requirements and setup
• Quick start
• DSL reference
• Callback contract
• Async state machines
• Composite and orthogonal machines
• Bounded runtime and utilities
• Testing and benchmarks
• Scope and status

Requirements and setup

• Python 3.13–3.14 (>=3.13,<3.15)
• uv
• No runtime dependencies; pytest is a development dependency.

From the repository root:

uv sync
uv run python -c "import sml; print(sml.__version__)"

The source layout is src/sml, and package-root imports preserve the public import sml API. uv sync creates or updates a local development environment; it is not evidence of a package-index release. No package-index release is claimed here.

Quick start

Events are ordinary Python classes. * 1 marks the initial state; + starts a transition expression; / attaches an action; and == supplies a destination.

import sml

class Switch: pass
class Stop: pass

table = sml.make_transition_table(
    sml.state("Off") * 1 + sml.event[Switch] == sml.state("On"),
    sml.state("On") + sml.event[Switch] == sml.state("Off"),
    sml.state("On") + sml.event[Stop] == sml.X,
)
machine = sml.sm(table)
print(machine.state.name)       # Off
machine.process_event(Switch())
machine.process_event(Stop())
print(machine.state)             # X (terminal)

DSL reference

Need	Spelling	Notes
Ordinary state	sml.state("Ready")	Names may also be classes.
Initial state	sml.state("Ready") * 1 or sml.initial_state("Ready")	Multiple initial rows create regions.
Typed state payload	sml.state("Idle")(Payload)	A default Payload() is created on entry.
Typed event	sml.event[Go]	Dispatch Go() or Go(data).
Named event	sml.named_event["tick"]	Dispatch sml.named_event["tick"](data).
Guard	sml.event[Go][can_go]	Declaration order selects the first passing candidate.
Action/sequence	... / save; ... / sml.sequence(first, second)	Actions run in declaration order.
Internal transition	state + event / action	No destination; state lifecycle does not run.
External self transition	state + event / action == state	Exits and re-enters the state.
Terminal state	== sml.X	The region terminates; later events are unhandled.
Entry/exit lifecycle	state + sml.on_entry[...] / callback	on_entry() and on_exit() are wildcards.
Completion	state + sml.completion[...] / callback == target	Completion is stabilized automatically.
Unexpected events	state + sml.unexpected_event[sml._] / callback	Handles otherwise unknown events.
Exceptions	state + sml.exception[ValueError] / recover == target	Match an exception type or sml._.
Defer/process	event / sml.defer; event / sml.process("next")	Deferred events retry after a transition; processed events have priority.
Anonymous transition	state + sml.anonymous == target	Automatic transition after entry.
Shallow history	sml.state("History")(sml.H)	Restores the last active child/region in a composite.
Wildcards	sml._	Matches values where the trigger supports wildcards.

The factory spelling is also available: sml.transition(source, trigger, sml.guard(check), sml.action(run), target). Build definitions with sml.make_transition_table(...) and machines with sml.sm(definition).

Callback contract

Callbacks are inspected by parameter name and annotation. They can receive the current event (the default), machine/facade (machine, sm, state_machine, or facade), context (context, ctx, dependency, or dep), or typed source/source_payload and destination/destination_payload. Context is from context= or the first positional dependency to sm.

import sml
class Go: pass
class Source:
    def __init__(self, value: int = 0) -> None: self.value = value
class Destination:
    def __init__(self, value: int = 0) -> None: self.value = value

def copy_forward(source: Source, destination: Destination) -> Destination:
    destination.value = source.value + 1
    return destination

table = sml.make_transition_table(
    sml.state("A")(Source) * 1 + sml.event[Go] / copy_forward
    == sml.state("B")(Destination),
)

The source payload is current state storage; the destination is a fresh value for its declared type. Returning a destination instance installs that payload. Event, context, dependency, and callback payloads remain caller-owned and are not copied for dispatch. machine.state and machine.states return deep-copy snapshots. A callback failure restores internal machine state, but not external side effects already performed. Synchronous callbacks must not return awaitables; an awaitable during sync dispatch raises TypeError.

Async state machines

Use async_sm and process_event_async for coroutine callbacks:

import asyncio, sml
class Go: pass
async def record(event: Go, context) -> None: context.append("go")
async def main() -> None:
    context: list[str] = []
    table = sml.make_transition_table(
        sml.state("A") * 1 + sml.event[Go] / record == sml.state("B"),
    )
    machine = sml.async_sm(table, context=context)
    await machine.process_event_async(Go())
    print(machine.state.name, context)  # B ['go']
asyncio.run(main())

AsyncStateMachine.process_event rejects sync dispatch. InlineScheduler runs submission immediately. FifoScheduler(capacity) is bounded FIFO single-consumer execution; capacity must be a power of two greater than one. AsyncioScheduler(capacity) is finite FIFO scheduling on the current event loop. CoroutineScheduler adapts another scheduler. An async machine has one logical owning task: serialize writes and nested submissions through that owner. Cancellation clears queued work and does not report success.

Composite and orthogonal machines

A CompositeState owns a nested table. Child dispatch is attempted before a parent transition; a terminated child can enable parent completion:

import sml
class Enter: pass
class Finish: pass
child = sml.CompositeState(
    sml.make_transition_table(sml.state("Child") * 1 + sml.event[Finish] == sml.X),
    name="Child",
)
parent = sml.make_transition_table(
    sml.state("Outside") * 1 + sml.event[Enter] == child,
    child + sml.completion[sml._] == sml.X,
)
machine = sml.sm(parent)
machine.process_event(Enter()); machine.process_event(Finish())
assert machine.is_terminated()

OrthogonalRegions broadcasts to independent machines and returns the handled count:

import sml

left = sml.sm(sml.make_transition_table(
    sml.state("L0") * 1 + sml.named_event["tick"] == sml.state("L1")))
right = sml.sm(sml.make_transition_table(
    sml.state("R0") * 1 + sml.named_event["tick"] == sml.state("R1")))
regions = sml.OrthogonalRegions([left, right])
assert regions.process_event(sml.named_event["tick"]()) == 2

Multiple initial rows also expose orthogonal regions through machine.states; machine.is_(state_a, state_b) checks all regions.

Bounded runtime and utilities

Run-to-completion work is bounded by max_work (default 1024): sml.sm(table, max_work=256, queue_capacity=64). Excessive anonymous, completion, sequence, nested, or queued work raises sml.WorkBudgetExceeded. deferred_capacity and processed_capacity independently bound internal queues.

• EventQueue(capacity) is finite FIFO storage. defer appends, process gives front priority, and overflow raises OverflowError; payloads are not copied.
• EventQueues(deferred=..., processed=..., max_work=...) drains processed events before deferred retries and returns a DispatchSummary.
• DispatchTable(machine, first_id, handlers) maps contiguous IDs to checked handlers; sml.dispatch(...) is its factory.
• SmPool(storage, factory) provides indexed/batch dispatch, including sml.with_id(index, event).
• OrthogonalRegions broadcasts one event; Hierarchical(parent, child) routes to an active child first, then parent, with HierarchicalDispatch statuses.

Testing and benchmarks

Synchronize with uv, keep examples importable from sml, and run repository tests before submitting a change. Tests use tests with src on the Python path.

uv sync
uv run pytest -q

The benchmark scripts accept simple, player-construct, and player-precreated modes:

uv run python benchmarks/benchmark_sml.py --iterations 1000 --mode simple
uv run python benchmarks/benchmark_sml.py --iterations 1000 --mode player-construct
uv run python benchmarks/benchmark_sml.py --iterations 1000 --mode player-precreated
uv run python benchmarks/benchmark_sml_async.py --iterations 1000 --mode simple
uv run python benchmarks/benchmark_sml_async.py --iterations 1000 --mode player-construct
uv run python benchmarks/benchmark_sml_async.py --iterations 1000 --mode player-precreated

Benchmark output is a local measurement of this Python/asyncio runtime, not a cross-language or package-release claim.

Scope and status

This repository documents and tests synchronous and async callbacks, bounded run-to-completion, hierarchical/composite and orthogonal routing, lifecycle/completion handling, typed payloads, and utility dispatch helpers.

It does not claim a PyPI or other package-index release, deployment system, CI matrix, persistence or serialization format, native C++/Rust equivalence, or comparable Python performance. The version and artifacts describe this checkout; package-index availability must not be inferred from 0.1.0 or local uv sync.