Snapshots & State Restoration

Save and restore machine state — persistence, crash recovery, and testing.

Snapshots let you capture and restore a machine’s state at any point in time. This implements the Memento pattern, enabling persistence, crash recovery, workflow checkpointing, and targeted testing.

📸 What are Snapshots?

flowchart LR
    R1["▶️ running<br/><small>state · context · inbox</small>"] -- "get_snapshot()" --> J["💾 JSON<br/><small>version · machine_hash · state_ids · context</small>"]
    J -- "from_snapshot()" --> R2["▶️ running again<br/><small>same machine, later — or another process</small>"]
    J -. "SnapshotDriftError if the machine changed" .-> X["⛔ refused"]

A snapshot is a JSON string that captures the runtime state of an interpreter inside a versioned envelope (layout v4 since 0.11.0 — see Snapshot Envelope for every key). The essentials:

• status — the interpreter’s lifecycle status ("running", "stopped", etc.)
• context — the full context dictionary (all mutable data)
• state_ids / configuration — the active leaves and the full active configuration
• in-flight work — pending_events (with their lane and engine provenance), deferred events, armed scheduled_sends, history, child actors
• what went wrong — error, and the runaway-chain record chain_trips / last_chain_error

You create a snapshot with get_snapshot() and restore from one with from_snapshot().

📸 get_snapshot() — Capturing State

The get_snapshot() method returns a JSON string representing the interpreter’s current state. Call it on any running interpreter at any time.

from xstate_statemachine import create_machine, SyncInterpreter

config = {
    "id": "workflow",
    "initial": "step1",
    "context": {
        "progress": 0,
        "data": {},
    },
    "states": {
        "step1": {
            "on": {
                "NEXT": {
                    "target": "step2",
                    "actions": "updateProgress",
                },
            },
        },
        "step2": {
            "on": {
                "NEXT": {
                    "target": "step3",
                    "actions": "updateProgress",
                },
            },
        },
        "step3": {
            "on": {
                "NEXT": {
                    "target": "step4",
                    "actions": "updateProgress",
                },
            },
        },
        "step4": {
            "type": "final",
        },
    },
}

from xstate_statemachine import MachineLogic

logic = MachineLogic(
    actions={
        "updateProgress": lambda i, ctx, e, a: ctx.update(
            {
                "progress": ctx["progress"] + 25,
            }
        ),
    }
)

machine = create_machine(config, logic=logic)
interp = SyncInterpreter(machine).start()

# Advance to step2
interp.send("NEXT")
interp.context["data"]["step1_result"] = "validated"

# Capture a snapshot
snapshot_json = interp.get_snapshot()
print(snapshot_json)

Output:

{
  "status": "running",
  "context": {
    "progress": 25,
    "data": {
      "step1_result": "validated"
    }
  },
  "state_ids": ["workflow.step2"]
}

The snapshot is a standard JSON string — you can store it anywhere: files, databases, message queues, or environment variables.

interp.stop()

♻️ from_snapshot() — Restoring State

The from_snapshot() class method creates a new interpreter instance pre-configured with the saved state. You provide the snapshot JSON string and the same machine definition that was used to create the original interpreter.

The restored context is deep-copied and merged over the machine’s default context — defaults fill in any keys missing from the snapshot, while persisted values win on conflict — and the snapshot also carries any events that were deferred (onUnhandled: "defer") at save time.

import json

# Re-create the same machine definition
machine = create_machine(config, logic=logic)

# Restore from the snapshot
restored = SyncInterpreter.from_snapshot(snapshot_json, machine)

print(restored.current_state_ids)  # {'workflow.step2'}
print(restored.context["progress"])  # 25
print(restored.context["data"])  # {'step1_result': 'validated'}
print(restored.status)  # 'running'

# Continue from where we left off
restored.send("NEXT")  # step2 -> step3
print(restored.current_state_ids)  # {'workflow.step3'}
print(restored.context["progress"])  # 50

restored.send("NEXT")  # step3 -> step4 (final)
restored.stop()

Note: from_snapshot() is a class method on both SyncInterpreter and Interpreter. Call it on the class you want to create:

restored_sync = SyncInterpreter.from_snapshot(snapshot_json, machine)
restored_async = Interpreter.from_snapshot(snapshot_json, machine)

⚠️ Important Limitations

Warning: from_snapshot() performs a static restoration. Be aware of these constraints:

• Entry actions are NOT re-run — The restored interpreter is placed directly into the saved states without executing their entry actions.
• Services are NOT restarted — Any invoke services that were running when the snapshot was taken are not re-invoked by default. Inspect what’s dormant with pending_invocations(), or opt in to re-driving it with from_snapshot(..., restart_services=True) — see Restoring Dormant Invocations below.
• Timers are not resumed by default — the snapshot records each armed timer’s wall-clock deadline (deadlines), but a static restore leaves them dormant: interpreter.has_dormant_timers is True. Pass restart_timers= (defaults to the value of restart_services): True / "restart" re-arms from zero in start() (0.9.0, #128), "resume" re-arms the remaining time, and "fire_due" also fires matured timers during start() (0.11.0, #264). A timer that had already fired and was waiting in the priority lane is persisted and replays on restore (#107) — into the lane, ahead of the inbox, on both engines (#214, #233).
• The chain-trip latch crosses a restart, as a RestoredChainError (0.9.0, #226; 0.9.1, #243) — chain_trips and last_chain_error are snapshot fields. A restored machine reports the same count, and the error comes back as a RestoredChainError, which is both a RestoredError and a RunawayChainError — so isinstance(interp.last_chain_error, RunawayChainError) stays True across the restart. Its .limit / .dropped are None (JSON kept only the message); the type-independent, restart-safe guard is interp.chain_trips > 0. clear_chain_error() is still the only thing that clears it. Before #226 both reset to 0 / None on restore; in 0.9.0 the restored latch was a plain RestoredError and an isinstance guard went silently False.
• strict and event schemas apply on restore (#214, #227) — a persisted record in pending_events or scheduled_sends that send() would have refused is dropped, reported via on_invalid_event, and recorded on last_error. Pass from_snapshot(..., plugins=[...]) (#230) so the hook is attached before admission runs; otherwise the refusal is only visible on last_error.
• status is not a liveness signal after a restore — the restored object reports the persisted "running" immediately, before start() has re-invoked anything. Between from_snapshot() and start(), and after any static restore, check has_dormant_invocations / has_dormant_timers (#135).
• A snapshot must be taken from a settled interpreter — mid-macrostep (while a transition’s actions are still running) the configuration has no leaf, and get_persisted_snapshot() raises SnapshotMidStepError rather than persisting a blob that would restore as a permanently inert machine. Snapshot after send(wait=True) resolves, from an on_transition hook, or after stop(drain=True) (#102).
• A non-blocking sync child is never copied mid-step — its pump thread holds a per-child step gate for every step, and the parent’s snapshot holds the same gate across “is it settled?” and the context copy, so a step cannot begin between them. If the gate is not free within 0.5 s the parent raises SnapshotMidStepError(child=True). This guarantee covers the supported cross-thread path, send_threadsafe(); a foreign-thread send() on a SyncInterpreter is unsupported and voids it (that step runs outside the gate).
• Pending event data must be JSON-native — a Decimal / datetime in a pending DoneEvent.data used to be silently stringified; get_snapshot() now raises SnapshotSerializationError instead of handing the restored handler a str (#131).
• Malformed snapshots are refused, typed — a missing key, a non-object context, an unknown status, a running snapshot with an empty configuration (#110), a hostile value in any persisted field — version, status, history, actors, system, deferred, a non-str payload (#146) — a pending event whose type is not a string (#158), or an "error" snapshot with no recorded error (#145) all raise SnapshotCorruptError. Nothing escapes except XStateMachineError.
• Configuration legality is checked on both sides (0.9.0, #142 / #143). Legality means exactly one active leaf per region, not “some leaf exists”: a parallel machine mid-transition in one region is refused with SnapshotMidStepError even while the other region has a leaf, and a running snapshot whose configuration lost its leaf entries (["order"]) is refused on restore rather than producing a live, permanently inert machine.
• v1 (pre-kind) pending events restore as user events (0.9.0, #162). Classifying them by name laundered a user’s after.hours into an engine event that onUnhandled / strict exempted. Only the init sentinel keeps system provenance. Archives written by 0.8.0 that contain engine-shaped engine events (an escalate, an error.platform.*) in the pending queue re-persist correctly as v2 on the next save; if you replay such an archive directly, re-persist it once from a 0.9.0 interpreter.
• Restore with the clock you mean — from_snapshot(..., clock=SimulatedClock()) (0.9.0, #117); without it a restored machine ran on RealClock, which broke deterministic replay.
• Context is restored by value — The context dictionary is deserialized from JSON. Non-serializable values (functions, class instances, file handles) will be lost or converted to strings.
• Machine definition must match — The machine argument to from_snapshot() must have the same structure as the original. If state IDs have changed, restoration will fail with StateNotFoundError.

📍 from_state() — Starting in a Known State

When you only know which state something is in (a status column on a database row, say) and have no snapshot, build the interpreter straight into that state:

Interpreter.from_state(machine, state, *, context=None, clock=None,
                       plugins=None, restart_services=False,
                       restart_timers=None)   # same on SyncInterpreter

state is one id or several (for parallel regions); each is an absolute id ("order.paid.packing") or a dot path from the machine root ("paid.packing"). A compound or parallel state completes to its initial leaves, as entering it would. Entry actions are not run (exactly like a snapshot restore); call start() afterwards and it resumes instead of re-entering the initial state. context is shallow-merged over the machine’s defaults. A top-level final state restores as status == "done".

from xstate_statemachine import SyncInterpreter, create_machine

machine = create_machine(
    {
        "id": "flow",
        "initial": "draft",
        "states": {
            "draft": {
                "on": {
                    "SUBMIT": "review",
                },
            },
            "review": {
                "on": {
                    "APPROVE": "done",
                },
            },
            "done": {},
        },
    }
)

interp = SyncInterpreter.from_state(machine, "review").start()
print(sorted(interp.current_state_ids))
# ['flow.review']
interp.send("APPROVE")
print(sorted(interp.current_state_ids))
# ['flow.done']
interp.stop()

An unknown id raises StateNotFoundError; ids that do not form a legal configuration (two leaves of one non-parallel parent) raise SnapshotCorruptError. The Pythonic StateMachine(state=...) constructor uses the same path. It is implemented as a minimal snapshot handed to from_snapshot, so machine_hash verification is skipped.

🔌 Restoring Dormant Invocations

from_snapshot()’s static restore means a machine snapshotted mid-invoke comes back parked: the restored configuration says the work is in flight, but no task is actually running it. That default is deliberate — restarting a non-idempotent action (an order placement, a charge) from scratch can be worse than leaving it parked — but it is no longer silent.

pending_invocations() — what’s dormant

pending_invocations() lists every invoke in the interpreter’s active configuration that has no live service backing it. It’s empty on a running (never-restored) machine, and empty again after a restore made with restart_services=True.

status is not a liveness signal after a restore. A statically restored machine reports status == "running" — it is processing events — while every invoke it lists is dormant. A health check that trusts status alone will say “healthy” about an order that sits unfilled. Check interpreter.has_dormant_invocations (a bool, 0.9.0) or pending_invocations() instead. There is deliberately no separate "restored" status value: it would break every consumer that switches on the existing four.

Two properties the whole recovery design rests on: after a static restore entry actions do not re-run and services are not re-invoked. That is what makes it safe to reconcile each pending invocation against the outside world (the venue, the payment provider) before deciding whether to restart it.

import asyncio
from xstate_statemachine import create_machine, MachineLogic, Interpreter

OMS = {
    "id": "oms",
    "initial": "submitting",
    "context": {
        "acked": False,
    },
    "states": {
        "submitting": {
            "invoke": {
                "src": "place",
                "id": "place",
                "onDone": "live",
            },
            "on": {
                "LEAVE": "idle",
            },
        },
        "live": {
            "entry": ["ack"],
        },
        "idle": {},
    },
}


async def place(interpreter, context, event):
    await asyncio.sleep(5.0)  # simulate a slow call
    return "ok"


def ack(interpreter, context, event, action_def):
    context["acked"] = True


logic = MachineLogic(
    actions={
        "ack": ack,
    },
    services={
        "place": place,
    },
)


async def main():
    interp = await Interpreter(create_machine(OMS, logic=logic)).start()
    await asyncio.sleep(0.01)  # `place` has started, far from done
    snapshot = interp.get_snapshot()
    await interp.stop()

    restored = Interpreter.from_snapshot(
        snapshot, create_machine(OMS, logic=logic)
    )
    print(restored.pending_invocations())
    # [PendingInvocation(state_id='oms.submitting', invoke_id='place', src='place')]


asyncio.run(main())

Each entry is a PendingInvocation(state_id, invoke_id, src) namedtuple — the state that owns the invoke, the invoke’s id (explicit, or the parser’s default), and the service key. Check this list first, then decide whether to re-drive the work.

restart_services=True — re-invoking from scratch

Pass restart_services=True to from_snapshot() to have start() re-invoke every dormant invoke in the restored configuration. This runs the service again, from the beginning — it does not resume the original call — through the exact same path a normal state entry uses, so the restarted task is owner-registered and still gets cancelled if the state is exited.

restored = Interpreter.from_snapshot(
    snapshot, create_machine(OMS, logic=logic), restart_services=True
)
await restored.start()
# ... service re-runs; eventually transitions oms.submitting -> oms.live
print(restored.pending_invocations())  # []

Because the service runs again rather than resuming, use this only for idempotent work, or gate it behind a client-supplied idempotency key (an order placement, for example, should carry one). This applies identically on both Interpreter (async) and SyncInterpreter (sync).

ImplementationMissingError on a missing service

If the machine being restored into no longer registers the service a dormant invoke needs, restart_services=True raises ImplementationMissingError when start() tries to re-drive it — the same exception a live machine raises for an unregistered service:

from xstate_statemachine import ImplementationMissingError

bare = create_machine(
    OMS,
    logic=MachineLogic(
        actions={
            "ack": ack,
        }
    ),
)  # no 'place' service registered
restored = Interpreter.from_snapshot(snapshot, bare, restart_services=True)
try:
    await restored.start()
except ImplementationMissingError as exc:
    print(
        exc
    )  # "Service 'place' referenced by state 'oms.submitting' is not registered."

✉️ Snapshot Envelope

flowchart LR
    subgraph env["snapshot envelope (v4)"]
        direction TB
        H["version · machine_id · machine_hash · machine_version · taken_at"]
        B["status · state_ids · context"]
        Q["inbox · deferred · pending invocations"]
    end
    env -- "from_snapshot()" --> C{"hash matches<br/>machine?"}
    C -- yes --> R["✅ restored"]
    C -- no --> D["⛔ SnapshotDriftError"]

As of 0.8.0, get_snapshot() writes a versioned envelope around the fields described above. In addition to status, context, and state_ids, the JSON now carries:

• version — the integer payload layout version. It is bumped only when the shape of the snapshot itself changes, never on an ordinary package release, so a patch or minor upgrade of the library does not invalidate stored snapshots.
• machine_id — the id of the machine the snapshot was taken from.
• machine_hash — a 16-hex-character structural fingerprint of the machine (MachineNode.structure_hash). It is a checksum against accidental drift, not an authentication tag (#205): anyone with the config can compute it, and a party who controls the payload can write a consistent one. To authenticate a snapshot, sign the whole JSON outside the library and restore with from_snapshot(minimum_version=1, expected_machine_hash=<fingerprint you recorded>) so the payload cannot select its own level of checking. It covers states, transitions, guard/action names, invokes, and after delays — the parts of the machine that change behavior. It deliberately excludes meta, description, and key order, so editing a docstring or reordering a dict does not change the hash. Adding a guard, renaming a state, or changing a transition’s target, on the other hand, does.
• taken_at — a Unix timestamp of when the snapshot was captured.
• scheduled_sends (layout v3, #213) — every armed, unfired delayed send to the machine itself (raise / send with delay), with the delay remaining at snapshot time (remaining_ms) and its id if it has one. from_snapshot re-arms them on start() with that remaining delay, so a state whose only exit is a delayed self-raise resumes where the live machine would have been instead of parking for ever. A fired timer is in pending_events as before. The records survive a restore → re-persist hop without start() (#221): from_snapshot parks them, get_persisted_snapshot re-emits them verbatim until start() re-arms them, so a compaction or migration job that rewrites blobs without starting a machine loses no deadline. Records are checked against strict / event schemas when start() re-arms them (#227).
• chain_trips / last_chain_error (#226) — the sticky chain-trip signal (#222). Restored verbatim; the error comes back as a RestoredChainError (a RunawayChainError and a RestoredError, #243). Absent in blobs written before 0.9.0, which restore to 0 / None. A malformed value (a non-integer count, a non-string message) is SnapshotCorruptError like any other field (#241). Trust boundary (#205, #242): like every restored field these are accepted verbatim with no integrity check beyond the definition-level machine_hash — a party who can write the blob can set chain_trips to any value and last_chain_error to any message. If these fields feed a supervisor-alerting path, a replayed blob can manufacture or suppress that alert; authenticate blobs outside the library if that matters.
• pending_events — see The Inbox: pending events below. Since layout v3 (#214) each record also carries its provenance ("engine": true for a completion the engine minted) and, on the async engine, its lane ("priority" for a fired timer / completion waiting ahead of the inbox), so both round-trip; a v2 record of kind done / error / after is upcast as engine-minted, because only the engine could have written one. Restored user events pass the same strict check a send() does — a refused one is reported (on_invalid_event, last_error) and dropped, never silently accepted. Since layout v2 (0.9.0) every pending or deferred record carries a kind — event, system, done, error or after — so engine events (DoneEvent, ErrorEvent, a due after) and provenance round-trip instead of being silently dropped (#86, #87). An ErrorEvent’s exception is persisted as its repr and restored as a RestoredError. v1 snapshots restore unchanged; their engine-shaped plain events are re-derived by name at the restore boundary only.
• machine_version (layout v4, #305) — the chart’s own "version" label from the root of the config (MachineNode.version), or null when the chart declares none. machine_hash answers “did the structure change?”; this answers “which revision wrote it?” — the label a human reads in a stuck-workflow ticket and the key the snapshot migrator (#263) dispatches on. Re-labelling a chart does not change machine_hash, so old blobs still load.
• deadlines (layout v4, #305) — durable after timers anchored to the wall clock (persistence.Deadline: state_id, entry_seq, due_at_wall, delay_ms, event_type), for timers that must survive a restart hours later, possibly on another host (#264). Distinct from scheduled_sends, which stores the delay remaining and is right for a restart seconds later. entry_seq is the state-entry generation, so a deadline armed by an earlier visit to the same state is recognised as stale and dropped instead of fired. Written by get_snapshot() for every armed after whose owning state is active; persisted() resumes the remaining time and DueTimerScanner wakes the ones that elapsed while nothing was running (see Persistence & Durability).
• value — the hierarchical interpreter.value at the time of capture, included for convenience. Restore ignores it; it is derived fresh from state_ids every time.

Unversioned 0.7.x snapshots restore unchanged. A payload with no version key is treated as version 0 and accepted unconditionally — there is no breaking change for snapshots taken before 0.8.0.

Rolling deploys: a v4 blob does not load on 0.10.x

upcast() only goes forward. During a rolling deploy the old build (0.10.x, layout v3) may pick up a snapshot the new build (0.11.0, layout v4) wrote — and it refuses it with SnapshotVersionError rather than silently dropping machine_version / deadlines. Refusing loudly is deliberate (the library’s rule: silent acceptance is a bug), but it means a mixed fleet can wedge on the shared store. Before rolling out a release that bumps SNAPSHOT_VERSION:

1. Deploy readers first. Every process that calls from_snapshot must be on the new build before any writer is — new readers load both v3 and v4.
2. Or drain / pin. Stop writers, let the old readers finish, then upgrade — or pin the store to one build for the window.
3. Route on version. A SnapshotVersionError names both versions; a supervisor can park the blob for a newer worker instead of retrying on the old one.

SNAPSHOT_VERSION is bumped only when the layout changes (0.8.0 → 1, 0.9.0 → 2 and 3, 0.11.0 → 4), never on an ordinary release, so this is a rare event — but it is one to plan for.

Restore-time checks

from_snapshot() now runs two checks before restoring a versioned snapshot:

• SnapshotVersionError — raised when the snapshot’s version is newer than the library’s SNAPSHOT_VERSION. A snapshot written by a newer release cannot be read safely, so it is refused rather than partially restored.
• SnapshotDriftError — raised when the snapshot’s machine_id doesn’t match the machine being restored into, or (when hash verification is on) the machine’s structure_hash no longer matches machine_hash. This is exactly the “a guard was added, or a state was renamed since this snapshot was taken” case.

What else to expect at the boundary — each line below is pinned by a test in tests/test_battle_305_snapshots_clock_codec.py, which flips, truncates and inserts every byte of a v4 blob and mutates every top-level key on both engines:

• Corrupt bytes are always SnapshotCorruptError (or a subclass), never a bare TypeError / KeyError / OverflowError. Invalid JSON raises InvalidConfigError; a version of Infinity / NaN (which json.loads accepts) is SnapshotCorruptError.
• Unknown extra top-level keys are accepted and silently dropped on the next get_snapshot(). Only a higher version is refused — the layout is forward-tolerant, not forward-compatible.
• A chart version label mismatch (machine_version in the blob vs the root "version" key) raises MachineVersionMismatchError unless on_version_mismatch="warn" is passed or a SnapshotMigrator step is registered. The label is deliberately not part of structure_hash.
• Deadline.due_at_wall is absolute wall time. If the wall clock moves backwards between snapshot and restore, a deadline fires later than its declared delay, never earlier, and is never lost. SimulatedClock(wall_start=…) lets a test express “restarted an hour later”; wall_start must be a finite number (a datetime or string is rejected; negative, NaN and infinite values are not).

Pass verify_machine_hash=False to skip the hash check after you’ve migrated a snapshot to match a changed machine shape:

import json
from xstate_statemachine import create_machine, SyncInterpreter
from xstate_statemachine.exceptions import SnapshotDriftError

machine = create_machine(config)  # the CURRENT (changed) machine definition

try:
    restored = SyncInterpreter.from_snapshot(saved_snapshot_json, machine)
except SnapshotDriftError:
    # Migrate the payload to match the new machine shape.
    payload = json.loads(saved_snapshot_json)
    payload["context"].setdefault("new_field", None)
    migrated_json = json.dumps(payload)

    # The machine has changed on purpose -- skip the hash check.
    restored = SyncInterpreter.from_snapshot(
        migrated_json, machine, verify_machine_hash=False
    )

📥 The Inbox: pending events

The mailbox is now part of the snapshot. interpreter.pending_events exposes every event that was accepted (by send()) but not yet processed, in FIFO order; the same list is persisted under the snapshot’s pending_events key and re-enqueued, in order, on restore. Child actors’ inboxes are captured and restored recursively, so a parent’s snapshot carries its children’s queued-but-unprocessed events too.

interp.send("STEP_1")
interp.send("STEP_2")
print(interp.pending_events)  # events not yet run through a transition

drain_pending() removes and returns every pending event without processing it — useful for a shutdown path that wants to persist accepted work durably instead of losing it:

drained = interp.drain_pending()  # async: `await interp.drain_pending()`

stop(drain=True) asks the interpreter to process its inbox to empty before tearing down, so nothing the caller was told “yes” to (via send()) is silently discarded. The async Interpreter.stop() also accepts a timeout= (seconds) to bound how long it waits:

await interp.stop(drain=True, timeout=5.0)  # async
interp.stop(drain=True)                     # sync

There are two shutdown patterns, depending on what you want:

• Option A — finish queued work before stopping. Call stop(drain=True) (with a timeout= on the async engine if you need an upper bound). Every event already accepted gets processed first.
• Option B — persist with pending events still in it, then stop immediately. Call get_snapshot() first (it captures pending_events as-is), then stop() with the default drain=False. The next restore re-enqueues those events instead of losing them.

By default (drain=False), a non-empty inbox at stop() time is now logged as a warning instead of being silently discarded.

💾 Persistence: Save to File

The simplest persistence pattern writes the snapshot to a JSON file:

from pathlib import Path
from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "onboarding",
    "initial": "welcome",
    "context": {
        "user": "alice",
        "steps_completed": [],
    },
    "states": {
        "welcome": {
            "on": {
                "CONTINUE": "profile",
            },
        },
        "profile": {
            "on": {
                "CONTINUE": "preferences",
            },
        },
        "preferences": {
            "on": {
                "CONTINUE": "complete",
            },
        },
        "complete": {
            "type": "final",
        },
    },
}

machine = create_machine(config)
interp = SyncInterpreter(machine).start()

interp.send("CONTINUE")  # welcome -> profile
interp.context["steps_completed"].append("welcome")

# Save to file
snapshot = interp.get_snapshot()
Path("onboarding_state.json").write_text(snapshot)
interp.stop()

# --- Later, in a different process or after a restart ---

# Load from file
saved = Path("onboarding_state.json").read_text()
machine = create_machine(config)  # Same definition
restored = SyncInterpreter.from_snapshot(saved, machine)

print(restored.current_state_ids)  # {'onboarding.profile'}
print(restored.context["steps_completed"])  # ['welcome']

# Continue the workflow
restored.send("CONTINUE")  # profile -> preferences
restored.context["steps_completed"].append("profile")
restored.stop()

🗄️ Database Persistence: SQLite Example

For production systems, store snapshots in a database:

import sqlite3
import json
from xstate_statemachine import create_machine, SyncInterpreter


# --- Database Setup ---
def init_db(db_path="machines.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS machine_snapshots (
            machine_id TEXT PRIMARY KEY,
            snapshot TEXT NOT NULL,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.commit()
    return conn


def save_snapshot(conn, machine_id, interpreter):
    """Save the interpreter's current state to the database."""
    snapshot = interpreter.get_snapshot()
    conn.execute(
        """INSERT OR REPLACE INTO machine_snapshots
           (machine_id, snapshot, updated_at)
           VALUES (?, ?, CURRENT_TIMESTAMP)""",
        (machine_id, snapshot),
    )
    conn.commit()
    print(f"Saved snapshot for '{machine_id}'")


def load_snapshot(conn, machine_id):
    """Load a saved snapshot from the database."""
    cursor = conn.execute(
        "SELECT snapshot FROM machine_snapshots WHERE machine_id = ?",
        (machine_id,),
    )
    row = cursor.fetchone()
    return row[0] if row else None


# --- Usage ---
config = {
    "id": "order-42",
    "initial": "placed",
    "context": {
        "items": ["widget"],
        "total": 29.99,
    },
    "states": {
        "placed": {
            "on": {
                "CONFIRM": "confirmed",
            },
        },
        "confirmed": {
            "on": {
                "SHIP": "shipped",
            },
        },
        "shipped": {
            "on": {
                "DELIVER": "delivered",
            },
        },
        "delivered": {
            "type": "final",
        },
    },
}

conn = init_db()
machine = create_machine(config)
interp = SyncInterpreter(machine).start()

interp.send("CONFIRM")
save_snapshot(conn, "order-42", interp)
interp.stop()

# --- Later ---
saved_json = load_snapshot(conn, "order-42")
if saved_json:
    machine = create_machine(config)
    restored = SyncInterpreter.from_snapshot(saved_json, machine)
    print(f"Restored state: {restored.current_state_ids}")
    # {'order-42.confirmed'}
    restored.send("SHIP")
    save_snapshot(conn, "order-42", restored)
    restored.stop()

conn.close()

⚡ Async Snapshots

Snapshots work identically with the async Interpreter:

import asyncio
from xstate_statemachine import create_machine, Interpreter

config = {
    "id": "asyncWorkflow",
    "initial": "phase1",
    "context": {
        "results": [],
    },
    "states": {
        "phase1": {
            "on": {
                "ADVANCE": "phase2",
            },
        },
        "phase2": {
            "on": {
                "ADVANCE": "phase3",
            },
        },
        "phase3": {
            "type": "final",
        },
    },
}


async def main():
    machine = create_machine(config)
    interp = await Interpreter(machine).start()

    await interp.send("ADVANCE")
    await asyncio.sleep(0.1)  # Let the event process

    # Capture snapshot (synchronous method — no await needed)
    snapshot = interp.get_snapshot()
    await interp.stop()

    # Restore into an async interpreter
    machine = create_machine(config)
    restored = Interpreter.from_snapshot(snapshot, machine)
    print(restored.current_state_ids)  # {'asyncWorkflow.phase2'}
    print(restored.status)  # 'running'


asyncio.run(main())

🧪 Testing with Snapshots

Snapshots are powerful for testing because they let you jump directly to a specific state without replaying the full event sequence:

import json
from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

config = {
    "id": "checkout",
    "initial": "cart",
    "context": {
        "items": [],
        "total": 0,
        "payment_status": None,
    },
    "states": {
        "cart": {
            "on": {
                "PROCEED": "shipping",
            },
        },
        "shipping": {
            "on": {
                "CONFIRM_ADDRESS": "payment",
            },
        },
        "payment": {
            "on": {
                "PAY": {
                    "target": "processing",
                    "actions": "startPayment",
                },
            },
        },
        "processing": {
            "on": {
                "SUCCESS": "confirmed",
                "FAILURE": "payment",
            },
        },
        "confirmed": {
            "type": "final",
        },
    },
}

logic = MachineLogic(
    actions={
        "startPayment": lambda i, ctx, e, a: ctx.update(
            {
                "payment_status": "processing",
            }
        ),
    }
)


# --- Test: Jump directly to the payment state ---
def test_payment_flow():
    """Test the payment flow without navigating through cart and shipping."""
    # Create a snapshot that puts us directly in the payment state
    snapshot = json.dumps(
        {
            "status": "running",
            "context": {
                "items": [
                    {
                        "name": "Widget",
                        "price": 29.99,
                    }
                ],
                "total": 29.99,
                "payment_status": None,
            },
            "state_ids": ["checkout.payment"],
        }
    )

    machine = create_machine(config, logic=logic)
    interp = SyncInterpreter.from_snapshot(snapshot, machine)

    # Test: send PAY event
    interp.send("PAY")
    assert interp.context["payment_status"] == "processing"
    assert interp.current_state_ids == {"checkout.processing"}

    # Test: payment succeeds
    interp.send("SUCCESS")
    assert interp.current_state_ids == {"checkout.confirmed"}

    interp.stop()
    print("test_payment_flow PASSED")


def test_payment_failure_retry():
    """Test that payment failure returns to the payment state."""
    snapshot = json.dumps(
        {
            "status": "running",
            "context": {
                "items": [],
                "total": 0,
                "payment_status": None,
            },
            "state_ids": ["checkout.processing"],
        }
    )

    machine = create_machine(config, logic=logic)
    interp = SyncInterpreter.from_snapshot(snapshot, machine)

    interp.send("FAILURE")
    assert interp.current_state_ids == {"checkout.payment"}

    interp.stop()
    print("test_payment_failure_retry PASSED")


test_payment_flow()
test_payment_failure_retry()

🏁 Complete Example: Long-Running Workflow with Checkpointing

This pattern saves a snapshot after each step, enabling crash recovery:

from pathlib import Path
from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic

CHECKPOINT_FILE = Path("workflow_checkpoint.json")

config = {
    "id": "dataProcessing",
    "initial": "ingesting",
    "context": {
        "records_ingested": 0,
        "records_transformed": 0,
        "records_loaded": 0,
        "errors": [],
    },
    "states": {
        "ingesting": {
            "on": {
                "INGEST_DONE": {
                    "target": "transforming",
                    "actions": "markIngested",
                },
            },
        },
        "transforming": {
            "on": {
                "TRANSFORM_DONE": {
                    "target": "loading",
                    "actions": "markTransformed",
                },
            },
        },
        "loading": {
            "on": {
                "LOAD_DONE": {
                    "target": "complete",
                    "actions": "markLoaded",
                },
            },
        },
        "complete": {
            "type": "final",
        },
    },
}

logic = MachineLogic(
    actions={
        "markIngested": lambda i, ctx, e, a: ctx.update(
            {
                "records_ingested": e.payload.get("count", 0),
            }
        ),
        "markTransformed": lambda i, ctx, e, a: ctx.update(
            {
                "records_transformed": e.payload.get("count", 0),
            }
        ),
        "markLoaded": lambda i, ctx, e, a: ctx.update(
            {
                "records_loaded": e.payload.get("count", 0),
            }
        ),
    }
)


def checkpoint(interp):
    """Save current state to disk after each step."""
    CHECKPOINT_FILE.write_text(interp.get_snapshot())
    print(f"Checkpoint saved: {interp.current_state_ids}")


def resume_or_start():
    """Resume from checkpoint if available, otherwise start fresh."""
    machine = create_machine(config, logic=logic)

    if CHECKPOINT_FILE.exists():
        saved = CHECKPOINT_FILE.read_text()
        print("Resuming from checkpoint...")
        return SyncInterpreter.from_snapshot(saved, machine)
    else:
        print("Starting fresh...")
        return SyncInterpreter(machine).start()


# --- Main workflow ---
interp = resume_or_start()
current = interp.current_state_ids

# Only process steps that haven't been completed
if "dataProcessing.ingesting" in current:
    print("Running ingest phase...")
    interp.send("INGEST_DONE", count=1500)
    checkpoint(interp)

if "dataProcessing.transforming" in interp.current_state_ids:
    print("Running transform phase...")
    interp.send("TRANSFORM_DONE", count=1450)
    checkpoint(interp)

if "dataProcessing.loading" in interp.current_state_ids:
    print("Running load phase...")
    interp.send("LOAD_DONE", count=1450)
    checkpoint(interp)

print(f"Final context: {interp.context}")
interp.stop()

# Clean up checkpoint on success
if CHECKPOINT_FILE.exists():
    CHECKPOINT_FILE.unlink()
    print("Checkpoint cleaned up.")

🛟 Complete Example: Crash Recovery Pattern

import json
import sys
from pathlib import Path
from xstate_statemachine import create_machine, SyncInterpreter, MachineLogic


class RecoverableWorkflow:
    """A workflow wrapper that automatically persists and recovers state."""

    def __init__(self, config, logic=None, storage_path="workflow_state.json"):
        self.config = config
        self.logic = logic
        self.storage_path = Path(storage_path)
        self.interp = None

    def start(self):
        """Start or resume the workflow."""
        machine = create_machine(self.config, logic=self.logic)

        if self.storage_path.exists():
            snapshot = self.storage_path.read_text()
            self.interp = SyncInterpreter.from_snapshot(snapshot, machine)
            print(
                f"Recovered workflow at state: {self.interp.current_state_ids}"
            )
        else:
            self.interp = SyncInterpreter(machine).start()
            print(
                f"Started new workflow at state: {self.interp.current_state_ids}"
            )

        return self

    def send(self, event_type, **payload):
        """Send an event and auto-save the state."""
        self.interp.send(event_type, **payload)
        self._save()
        return self

    def _save(self):
        """Persist current state to disk."""
        self.storage_path.write_text(self.interp.get_snapshot())

    def finish(self):
        """Stop the interpreter and clean up persistence."""
        self.interp.stop()
        if self.storage_path.exists():
            self.storage_path.unlink()
        print("Workflow completed and cleaned up.")

    @property
    def state(self):
        return self.interp.current_state_ids

    @property
    def context(self):
        return self.interp.context


# Usage
config = {
    "id": "deploy",
    "initial": "building",
    "context": {
        "version": "1.2.3",
        "status": "pending",
    },
    "states": {
        "building": {
            "on": {
                "BUILD_OK": "testing",
            },
        },
        "testing": {
            "on": {
                "TESTS_PASS": "deploying",
                "TESTS_FAIL": "failed",
            },
        },
        "deploying": {
            "on": {
                "DEPLOY_OK": "done",
            },
        },
        "done": {
            "type": "final",
        },
        "failed": {
            "type": "final",
        },
    },
}

workflow = RecoverableWorkflow(config).start()
workflow.send("BUILD_OK")
print(f"State: {workflow.state}")  # {'deploy.testing'}

# If the process crashes here, the next run will resume at 'testing'

workflow.send("TESTS_PASS")
workflow.send("DEPLOY_OK")
workflow.finish()