Plugins & Observability

Observe every event, transition, action, guard, and service with pluggable hooks.

Plugins are observers that watch machine execution without modifying behavior. They hook into the interpreter’s lifecycle to add cross-cutting concerns like logging, metrics, debugging, auditing, and performance monitoring.

🔌 What are Plugins?

flowchart LR
    I["⚙️ interpreter"] -- on_transition --> L["🪵 LoggingInspector"]
    I -- on_action_execute --> M["📈 MetricsPlugin"]
    I -- on_guard_error / on_action_error --> A["🚨 AlertPlugin"]
    L -. "observe only — never mutate" .-> I

A plugin implements the Observer pattern: it subscribes to lifecycle events in the interpreter and reacts to them. Plugins never alter the machine’s state or transitions — they only observe. This makes them safe to add or remove without affecting business logic.

Every plugin extends the PluginBase class and overrides the hook methods it cares about. You register plugins with the interpreter using the .use() method.

from xstate_statemachine import SyncInterpreter, PluginBase

class MyPlugin(PluginBase):
    def on_transition(self, interpreter, from_states, to_states, transition):
        print("A transition just happened!")

machine = create_machine(config)
interp = SyncInterpreter(machine)
interp.use(MyPlugin())  # Register the plugin
interp.start()

🪵 Built-in: LoggingInspector

The library ships with LoggingInspector, a ready-to-use plugin that emits detailed, emoji-prefixed log messages for every significant machine event. It is invaluable for debugging complex state machines.

🔒 Redaction (0.9.0). LoggingInspector redacts sensitive keys before logging context or event payloads: any key containing password, secret, token, api_key, authorization, credential, card, cvv, … (see DEFAULT_REDACT_KEYS) is written as "***", recursively. Extend the list with LoggingInspector(redact_keys=(*DEFAULT_REDACT_KEYS, "ssn")), or opt out explicitly with redact_keys=(). Pass log_context=False to skip the per-transition context dump on large machines. The redact() helper is exported for your own plugins.

Setup and Usage

import logging
from xstate_statemachine import (
    create_machine,
    SyncInterpreter,
    MachineLogic,
    LoggingInspector,
)

# Configure Python logging to see the output
logging.basicConfig(level=logging.INFO)

config = {
    "id": "demo",
    "initial": "idle",
    "context": {
        "count": 0,
    },
    "states": {
        "idle": {
            "on": {
                "START": {
                    "target": "running",
                    "actions": "increment",
                },
            },
        },
        "running": {
            "on": {
                "STOP": "idle",
                "FINISH": "done",
            },
        },
        "done": {
            "type": "final",
        },
    },
}

logic = MachineLogic(
    actions={
        "increment": lambda i, ctx, e, a: ctx.update(
            {
                "count": ctx["count"] + 1,
            }
        ),
    }
)

machine = create_machine(config, logic=logic)
interp = SyncInterpreter(machine)
interp.use(LoggingInspector())  # Attach the inspector
interp.start()

interp.send("START")
interp.send("FINISH")
interp.stop()

Output Format

All LoggingInspector messages are prefixed with 🕵️ [INSPECT] for easy filtering:

🕵️ [INSPECT] Event Received: START
🕵️ [INSPECT] Executing Action: increment
🕵️ [INSPECT] Transition: ['demo.idle'] -> ['demo.running'] on Event 'START'
🕵️ [INSPECT] New Context: {'count': 1}
🕵️ [INSPECT] Event Received: FINISH
🕵️ [INSPECT] Transition: ['demo.running'] -> ['demo.done'] on Event 'FINISH'
🕵️ [INSPECT] New Context: {'count': 1}

Service-related messages use different prefixes:

🚀 [INSPECT] Service 'fetchData' (ID: loading) starting...
✅ [INSPECT] Service 'fetchData' (ID: loading) completed. Result: {...}
❌ [INSPECT] Service 'fetchData' (ID: loading) failed. Error: ConnectionError

Guard evaluation messages include pass/fail indicators:

🕵️ [INSPECT] Guard 'isAuthenticated' evaluated for event 'ACCESS' -> ✅ Passed
🕵️ [INSPECT] Guard 'hasPermission' evaluated for event 'DELETE' -> ❌ Failed

Tip: Set logging.basicConfig(level=logging.INFO) at the top of your script to see LoggingInspector output. Without this, Python’s default logging level (WARNING) will suppress the messages.

🧩 Custom Plugins with PluginBase

Create custom plugins by subclassing PluginBase and overriding any hooks you need:

Shortcut: xsm gt machine.json --with-plugin generates a PluginBase subclass that overrides exactly the hooks that chart can fire, each logging one JSON line — see the companion templates.

from xstate_statemachine import PluginBase

class MetricsPlugin(PluginBase):
    """Tracks transition counts and event frequencies."""

    def __init__(self):
        self.transition_count = 0
        self.event_counts = {}

    def on_event_received(self, interpreter, event):
        self.event_counts[event.type] = self.event_counts.get(event.type, 0) + 1

    def on_transition(self, interpreter, from_states, to_states, transition):
        self.transition_count += 1
        from_ids = sorted(s.id for s in from_states if s.is_atomic or s.is_final)
        to_ids = sorted(s.id for s in to_states if s.is_atomic or s.is_final)
        if from_ids != to_ids:
            print(f"Transition #{self.transition_count}: {from_ids} -> {to_ids}")

    def report(self):
        print(f"Total transitions: {self.transition_count}")
        print(f"Event frequencies: {self.event_counts}")

Usage:

metrics = MetricsPlugin()
interp = SyncInterpreter(machine)
interp.use(metrics)
interp.start()

interp.send("START")
interp.send("PROCESS")
interp.send("FINISH")
interp.stop()

metrics.report()
# Total transitions: 3
# Event frequencies: {'START': 1, 'PROCESS': 1, 'FINISH': 1}

🪝 Plugin Hooks Reference

Every hook receives the interpreter instance as its first argument, giving plugins full read access to the machine’s current state, context, and configuration.

LoggingInspector implements on_event_received, on_event_processed, on_transition, on_action_execute, on_guard_evaluated, on_service_start/on_service_done/on_service_error, on_transition_failed, on_guard_error, on_unhandled_event, on_error, and on_done. It does not implement on_interpreter_start, on_interpreter_stop, on_action_error, or on_event_dropped — action failures and dropped events pass through silently unless you add your own plugin for them.

⚠️ Hooks are synchronous callbacks. An async def override is never awaited — the engine dispatches hooks from inside a transition and cannot suspend there. Since 0.9.0 such a hook is closed explicitly and reported through on_plugin_error / interpreter.last_plugin_error as a TypeError, instead of vanishing with only a Python RuntimeWarning. If a hook needs async work, keep it def and schedule a task inside it.

Plugin failures of any kind — including a hook raising asyncio.CancelledError — are contained: the interpreter keeps running, and the failure is visible to other plugins via on_plugin_error.

Hook Details

on_interpreter_start(interpreter)

Fires once when start() is called. Use it for setup tasks like opening database connections, starting timers, or initializing counters.

def on_interpreter_start(self, interpreter):
    self.start_time = time.time()
    print(f"Machine '{interpreter.id}' started")

on_interpreter_stop(interpreter)

Fires once when stop() is called. Use it for teardown tasks like flushing buffers, closing connections, or generating reports.

def on_interpreter_stop(self, interpreter):
    elapsed = time.time() - self.start_time
    print(f"Machine '{interpreter.id}' ran for {elapsed:.2f}s")

on_event_received(interpreter, event)

Fires immediately after an event enters the processing queue. The event object has a .type (string) and .payload (dict or None).

def on_event_received(self, interpreter, event):
    print(f"Event received: {event.type}")
    if hasattr(event, "payload") and event.payload:
        print(f"  Payload: {event.payload}")

on_before_send(interpreter, event) -> Receipt | None [0.11.0]

The one hook that can stop a caller’s event before the machine sees it. It fires from send(), send_events() and send_threadsafe() after the admission checks (strict, event_schemas) and before the event is queued. Return None to let it through; return a Receipt to short-circuit — the event is never queued, on_event_received and on_event_processed do not fire for it, and the caller receives your receipt (a send(wait=True) resolves to it immediately on both engines). The first plugin that returns a receipt wins.

This is how the idempotency inbox answers a redelivered webhook with the original outcome, and how a rate limiter or maintenance-mode switch says “not now” without touching state:

from xstate_statemachine import (
    PluginBase,
    Receipt,
    SyncInterpreter,
    create_machine,
    stub_logic,
)


class MaintenanceMode(PluginBase):
    """Refuse every user event while enabled; the machine stays untouched."""

    enabled = False

    def on_before_send(self, interpreter, event):
        if not self.enabled:
            return None  # proceed normally
        return Receipt(
            frozenset(interpreter.current_state_ids),
            changed=False,
            error=None,
            deferred=False,
            denied=True,  # "a rule refused it", not "does not apply"
        )


cfg = {
    "id": "m",
    "initial": "a",
    "states": {
        "a": {
            "on": {
                "GO": "b",
            },
        },
        "b": {},
    },
}
gate = MaintenanceMode()
interp = (
    SyncInterpreter(create_machine(cfg, logic=stub_logic(cfg)))
    .use(gate)
    .start()
)

gate.enabled = True
receipt = interp.send("GO", wait=True)
assert receipt.denied and interp.current_state_ids == {"m.a"}  # never queued

gate.enabled = False
assert interp.send("GO", wait=True).changed  # normal path

Fail-open. Plugin hooks are error-contained: if your interceptor raises, the failure is reported through on_plugin_error and the event is admitted. A blocker must return a receipt, not raise. The same rule applies to a wrong return type: anything other than a Receipt or None is reported as a TypeError through on_plugin_error and the event is admitted. Hooks contain Exception and asyncio.CancelledError; KeyboardInterrupt and SystemExit raised from a hook reach the caller of send(). Engine-minted events (after, done.invoke, error.platform) never pass through this hook — nobody “sends” them.

Interceptor rules under load (battle-tested on both engines with 8 producer threads, 1 000 concurrent gather(send()) calls and a seeded 2 000-operation stress run — tests/test_battle_304_concurrency.py):

• Where it runs. on_before_send runs on the sending thread for send() / send_events(), and on the owning thread (sync) or the event loop (async) for events arriving via send_threadsafe(). It blocks that engine for as long as it runs, but send_threadsafe() producers are never blocked by it — the mailbox lock is not held while hooks run, so a parked interceptor cannot stall other threads’ enqueues.
• Re-entrancy. An on_before_send that calls send() on its own interpreter is allowed: the nested event is fully processed before the original is queued. An interceptor that unconditionally re-sends the same event recurses until RecursionError, which is contained like any other exception (reported via on_plugin_error, original event admitted) — add your own guard.
• Coherent receipts. A receipt’s state_ids is always a complete configuration — never a half-finished one — even when another thread is mid-transition, because both engines build it on the owning thread from an atomic frozenset.
• Mailbox growth. A producer that calls send_threadsafe() in a tight loop can outpace the owner’s drain indefinitely; the mailbox is unbounded by design. Bound your producers (batch, back-pressure, or a semaphore) — the engine will not do it for you.

on_event_processed(interpreter, event, receipt) [0.11.0]

Fires exactly once per event that entered the machine — user events and engine-minted ones alike — after its macrostep has settled, or after it was denied, unhandled, deferred or dropped. receipt is the same Receipt a send(wait=True) caller gets, so this is the hook that knows the outcome: on_transition fires per transition and carries no event, and a guard-denied or unhandled event fires no transition at all. Audit logs, state-coverage collectors, tracing spans and the idempotency inbox’s “mark” all attach here.

from xstate_statemachine import (
    PluginBase,
    SyncInterpreter,
    create_machine,
    stub_logic,
)


class Outcomes(PluginBase):
    def __init__(self):
        self.seen = []

    def on_event_processed(self, interpreter, event, receipt):
        kind = (
            "error"
            if receipt.error
            else (
                "deferred"
                if receipt.deferred
                else (
                    "denied"
                    if receipt.denied
                    else "changed" if receipt.changed else "no-op"
                )
            )
        )
        self.seen.append((event.type, kind))


cfg = {
    "id": "o",
    "initial": "a",
    "states": {
        "a": {
            "on": {
                "GO": "b",
                "LOCKED": {
                    "target": "b",
                    "guard": "never",
                },
            },
        },
        "b": {},
    },
}
out = Outcomes()
interp = (
    SyncInterpreter(create_machine(cfg, logic=stub_logic(cfg, guards=False)))
    .use(out)
    .start()
)
interp.send("LOCKED")  # guard says no
interp.send("NOPE")  # nothing handles it
interp.send("GO")  # transitions
assert out.seen == [("LOCKED", "denied"), ("NOPE", "no-op"), ("GO", "changed")]

Both engines build the receipt from a per-event before-image, so a send() on the sync engine that also drains a due timer reports two outcomes — the user’s event and the after event — each with its own flags (the timer’s outcome is reported first, since it was due before the user event ran). The per-event bookkeeping only runs when some attached plugin overrides this hook.

Two things worth knowing before you build on it:

• Sending from the hook. A send() made from on_event_processed on SyncInterpreter runs at once, inside the hook dispatch, so plugins registered after yours see the follow-up event’s on_event_processed before the event that caused it. Register observers that care about order before plugins that send. On the async engine the follow-up is queued and reported in order.
• Timer events have no payload. Events the engine mints for after timers carry type but no payload attribute; read it defensively with getattr(event, "payload", None) if your hook handles both kinds.
• What a failing guard or action looks like. A guard that raises is treated as False, so the receipt arrives with denied=True and error=None. An action that raises mid-list delivers one receipt with error set, after the remaining actions in that list were skipped — the hook still fires exactly once.
• Receipt has six fields. It is a NamedTuple and duplicate was appended in 0.11.0; unpacking it into five names raises ValueError: too many values to unpack. Use attribute access (receipt.changed, receipt.duplicate) so the next field addition cannot break you.
• Your receipt is yours. When on_before_send returns a Receipt, the engine hands it to the caller as-is — state_ids is not coerced or validated. Build it with frozenset(interpreter.current_state_ids) as in the example above.

on_transition(interpreter, from_states, to_states, transition)

Fires after a transition completes. from_states and to_states are sets of StateNode objects. The transition object contains .event, .target_str, .guard, and .actions.

def on_transition(self, interpreter, from_states, to_states, transition):
    from_ids = {s.id for s in from_states if s.is_atomic or s.is_final}
    to_ids = {s.id for s in to_states if s.is_atomic or s.is_final}
    if from_ids != to_ids:
        print(f"State changed: {from_ids} -> {to_ids}")

on_action_execute(interpreter, action)

Fires right before each action is executed. The action is an ActionDefinition with a .type (name) and optional .params.

def on_action_execute(self, interpreter, action):
    print(f"Executing: {action.type}")

on_action_error(interpreter, action, error)

Fires when a user-supplied action raises, before the machine’s action_error_policy ("continue" / "rollback" / "fail") is applied — so it fires under every policy, regardless of how the machine recovers. Use it to route action failures to Sentry, a metrics counter, or a dead-letter queue. See Actions — When an Action Raises.

def on_action_error(self, interpreter, action, error):
    print(f"Action '{action.type}' raised: {error!r}")

on_guard_evaluated(interpreter, guard_name, event, result)

Fires after a guard function returns. result is the boolean return value.

def on_guard_evaluated(self, interpreter, guard_name, event, result):
    status = "PASS" if result else "FAIL"
    print(f"Guard '{guard_name}' for '{event.type}': {status}")

on_service_start(interpreter, invocation)

Fires when an invoked service is about to run. The invocation is an InvokeDefinition with .src (service name) and .id.

def on_service_start(self, interpreter, invocation):
    print(f"Service '{invocation.src}' starting...")

on_service_done(interpreter, invocation, result)

Fires when a service completes successfully. result is whatever the service returned.

def on_service_done(self, interpreter, invocation, result):
    print(f"Service '{invocation.src}' completed: {result}")

on_service_error(interpreter, invocation, error)

Fires when a service raises an exception. error is the Exception object.

def on_service_error(self, interpreter, invocation, error):
    print(f"Service '{invocation.src}' failed: {error}")

on_transition_failed(interpreter, transition, failed_actions)

Fires under every actionErrorPolicy whenever one or more actions raised — in the transition’s own action list or in the entry / exit list of any state the transition crossed. failed_actions is a list of (ActionDefinition, exception) pairs in execution order. Under "continue" it is followed by on_transition (the transition committed); under "rollback" and "fail" it is not. See Actions — When an Action Raises.

def on_transition_failed(self, interpreter, transition, failed_actions):
    for action, error in failed_actions:
        print(f"Action '{action.type}' failed: {error!r}")

on_guard_error(interpreter, guard_name, event, error)

Fires when a guard raises instead of returning, before the raise is substituted with a result per guardErrorPolicy. See Guards — Error Handling in Guards.

def on_guard_error(self, interpreter, guard_name, event, error):
    print(f"Guard '{guard_name}' raised on '{event.type}': {error!r}")

on_unhandled_event(interpreter, event, active_state_ids, disposition)

Fires when an event matches no transition in any active state, regardless of onUnhandled policy. disposition is "ignored" (the state declares no handler for this event), "guard_denied" (a handler is declared but every candidate’s guard returned False — 0.9.0, #153), "deferred", "errored", or "dropped" (the defer buffer was full and the oldest entry was evicted). The same distinction is on the receipt as Receipt.denied. See Interpreters — Unhandled Events.

def on_unhandled_event(self, interpreter, event, active_state_ids, disposition):
    print(f"'{event.type}' unhandled in {active_state_ids}: {disposition}")

on_chain_budget_exceeded(interpreter, error, event)

Fires once per trip when maxIterations cuts the machine’s self-generated work — a zero-delay raise cycle, a completion storm, or an always loop (0.9.0, #222). error is the RunawayChainError (.limit, .dropped, .stranded); event is the first event cut, or Event("") for a settle-budget trip.

Use this — or poll interpreter.chain_trips (a monotonic counter) / interpreter.last_chain_error (a latch you clear with clear_chain_error()) — rather than last_error, which is recomputed per processed event and is erased by the next benign event. A machine with a heartbeat guarantees that event arrives, so a supervisor polling last_error loses the race every time.

def on_chain_budget_exceeded(self, interpreter, error, event):
    metrics.increment(
        "chain_trips",
        tags={
            "machine": interpreter.id,
        },
    )
    alert(
        f"{interpreter.id} cut {error.dropped} event(s) at {error.limit}; first: {event.type}"
    )

on_event_dropped(interpreter, event, reason)

Fires when an event is discarded unprocessed. reason is one of:

The drop is also logged at WARNING. Since 0.9.0 every loss site on both engines fires this hook; before, the SyncInterpreter dropped silently in several of these cases. See Interpreters — Unhandled Events.

def on_event_dropped(self, interpreter, event, reason):
    print(f"Dropped '{event.type}': {reason}")

on_resolve_error(interpreter, error, event)

Fires when a transition’s target cannot be resolved at runtime — only reachable with create_machine(..., strict_targets=False), since the default refuses an unresolvable target at build time. error is the StateNotFoundError; the step is marked failed (last_transition_ok=False, last_error) and the machine keeps running (0.9.0, #134). The third per-transition failure category alongside on_action_error and on_guard_error.

on_plugin_error(interpreter, plugin, hook, error)

Fires on every other plugin when one plugin’s hook raised, or was an async def that could not be awaited. Plugin failures never stop the machine; they are logged at ERROR, recorded on interpreter.last_plugin_error as (plugin_class_name, hook_name, error), and reported here so a metrics exporter can count observability failures. The failing plugin never hears about itself, and a raising on_plugin_error does not recurse (0.9.0, #127).

on_invalid_event(interpreter, error, raw_event)

Fires immediately before send() (or send_threadsafe()) refuses an event: a malformed value (InvalidEventError — send(123), a dict without "type"), an undeclared type under strict (UnknownEventError), or a payload a registered schema rejected (InvalidEventPayloadError). It also fires for an event a snapshot restore refused under the same rules (#214, #227) — pass from_snapshot(..., plugins=[...]) so the plugin is attached before admission runs (#230).

on_snapshot_error(interpreter, error)

Fires immediately before a snapshot is refused: SnapshotMidStepError (taken mid-macrostep, e.g. from inside an action) or SnapshotSerializationError (a pending event carries non-JSON-native data). The exception still propagates to the caller (0.9.0, #159).

on_invocation_stranded(interpreter, state_id, invoke_id, error)

Fires when a maxIterations cut discarded the done.invoke / error.platform of an invocation whose state is still active: nothing is running for it and no completion will ever arrive, so the machine rests in a state that declares invoke — a state it was never meant to rest in. error is the RunawayChainError, whose .stranded tuple names the same ids. has_dormant_invocations / pending_invocations() answer the question on demand; this is the push notification (0.9.0, #207).

on_receipt_dropped(interpreter, event_type)

Fires when a send(wait=True) receipt issued from inside an action was finalised without ever being awaited or handed out — a plain def action wrote r = i.send("B", wait=True) and received an awaitable it could never read. The engine also emits a RuntimeWarning, but that comes from a finaliser and -W error cannot reach it; this hook and interpreter.dropped_receipts are the deterministic signal (0.9.1, #232, #244). Async engine only — the sync engine refuses the call with ReentrantWaitError.

def on_receipt_dropped(self, interpreter, event_type):
    raise AssertionError(f"dropped receipt for {event_type!r} on {interpreter.id}")

on_error(interpreter, error)

Fires when the interpreter enters the terminal "error" status. interpreter.error holds the same exception.

def on_error(self, interpreter, error):
    print(f"Machine '{interpreter.id}' stopped with error: {error!r}")

on_done(interpreter, output)

Fires when the machine reaches a top-level final state. output is the machine’s output value (may be None).

def on_done(self, interpreter, output):
    print(f"Machine '{interpreter.id}' done. Output: {output!r}")

🧬 Multiple Plugins

You can attach multiple plugins to a single interpreter. They execute in registration order:

from xstate_statemachine import SyncInterpreter, LoggingInspector

metrics = MetricsPlugin()
audit = AuditLogPlugin()
inspector = LoggingInspector()

interp = SyncInterpreter(machine)
interp.use(inspector)   # Fires first
interp.use(metrics)     # Fires second
interp.use(audit)       # Fires third
interp.start()

Note: The .use() method returns the interpreter instance, so you can chain calls:

interp = SyncInterpreter(machine).use(LoggingInspector()).use(MetricsPlugin())

🌐 Global Plugins: Every Interpreter in the Process

.use() attaches a plugin to one interpreter. Some plugins want all of them — a tracing exporter, a coverage collector, a test fixture that asserts no machine ever errored — including interpreters you never construct yourself: children the engine spawns, machines a web adapter restores from a snapshot inside a request handler. For those, register once, process-wide:

from xstate_statemachine import (
    SyncInterpreter,
    create_machine,
    PluginBase,
    register_global,
    unregister_global,
    global_plugins,
)


class CountStarts(PluginBase):
    def __init__(self):
        self.count = 0

    def on_interpreter_start(self, interpreter):
        self.count += 1


counter = CountStarts()
register_global(counter)  # from now on...

cfg = {
    "id": "m",
    "initial": "a",
    "states": {
        "a": {},
    },
}
SyncInterpreter(create_machine(cfg)).start().stop()
SyncInterpreter(create_machine(cfg)).start().stop()
assert counter.count == 2
assert global_plugins() == [counter]

unregister_global(counter)  # ...until here
SyncInterpreter(create_machine(cfg)).start().stop()
assert counter.count == 2  # not attached to the later one

Rules of the registry:

• Opt-in only. The library never registers anything on import; the default process has an empty registry and a constructor pays one list read.
• Constructed after, not before. A global plugin is attached to interpreters built after register_global — both engines, from_snapshot, and spawned children. Interpreters that already exist are not touched (their lifecycle hooks have already started firing).
• Same containment as .use(). A raising global plugin is reported through on_plugin_error and never breaks the machine.
• Thread-safe. Registration and removal take a lock; global_plugins() returns a copy. Registering the same object twice is a no-op. Battle-tested with 100 threads registering while 100 threads construct interpreters: every interpreter’s plugin set is a snapshot of the registry at its construction time.
• De-duplicated against .use() by identity (0.11.0, #273 battle). The same plugin instance registered globally and attached with .use() (or .use()-d twice) receives every hook once. Two distinct instances of the same class are two observers – except for discovered plugins: attach_discovered() skips an entry point whose plugin class is already attached to that interpreter (so instrument_all(discovered=True) run twice is a no-op, and a plugin you configured by hand is never shadowed by its entry point). allow=[] loads nothing (closed), allow=None loads everything declared; a bare string is refused (TypeError) rather than matched letter by letter.
• unregister_global() of a plugin that was never registered returns False and does nothing.
• Leave it as you found it. A pytest fixture should unregister_global (or xstate_statemachine.plugins.clear_global_plugins()) in teardown.

The [testing] plugin’s pytest --xsm-coverage is built on exactly this seam: it registers one xstate_statemachine.coverage.CoverageCollector at session start and unregisters it at session end — see State & transition coverage.

🔎 Third-party plugins: discovery

Other packages can ship plugins, stores and brokers without a core change by declaring an entry point. The library finds them only when you ask: discovery is never implicit (security baseline X0.14). Importing xstate_statemachine loads nothing, no interpreter looks for plugins by itself, and nothing is attached until you call attach_discovered (or .use() what discover() returned).

from xstate_statemachine import SyncInterpreter, create_machine
from xstate_statemachine.plugins import attach_discovered, discover

for p in discover():  # [] when nothing is installed
    print(p.name, p.distribution, p.version, p.hooks)

machine = create_machine(
    {
        "id": "m",
        "initial": "a",
        "states": {
            "a": {},
        },
    }
)
interp = SyncInterpreter(machine)
attached = attach_discovered(interp, allow=["acme-audit"])  # opt in by name
interp.start().stop()

• allow= takes entry-point names or distribution names. An entry point that is not allowed is never imported.
• A broken plugin does not take you down. A loader that raises is logged at WARNING with its traceback and skipped; it is recorded in xstate_statemachine.plugin_discovery.last_skipped (a list of SkippedPlugin(name, distribution, version, group, error), reset by every discover() call). In attach_discovered, a constructor that raises, or an entry point whose object is not a PluginBase, is skipped the same way and never .use()d. Pass strict=True to re-raise instead (a TypeError for a non-PluginBase), which is what a CI check that “every plugin loads” wants.
• XSM_DISABLE_PLUGIN_DISCOVERY=1 turns discovery off for the whole process: discover() returns [] and logs that once. Use it in locked-down deployments where the environment may contain packages you did not vet.
• DiscoveredPlugin is (name, distribution, version, obj, hooks, group). hooks lists the PluginBase hooks the class overrides.
• Python 3.9. importlib.metadata.entry_points(group=...) is 3.10+; on 3.9 the library selects the group from the dict the old API returns. The behaviour is identical.
• [observability]: instrument_all(interp, discovered=True, allow=[...]) is built on attach_discovered; with no interpreter it attaches the discovered plugins to the global collector instead.

from xstate_statemachine import SyncInterpreter, create_machine
from xstate_statemachine.contrib.observability import instrument_all

machine = create_machine(
    {
        "id": "m",
        "initial": "a",
        "states": {
            "a": {},
        },
    }
)
interp = SyncInterpreter(machine)
attached = instrument_all(interp, discovered=True, allow=["acme-audit"])
interp.start().stop()

xsm plugins

xsm plugins lists everything installed in all three groups. Listing imports each entry point, since it has to load a plugin to report its hooks. That is why it is a command you run, not something the library does for you. The library itself declares its contrib brokers in xstate_statemachine.brokers, so those rows are always present. This is real output with the test fixture package (tests/fixtures/xsm_thirdparty_plugin, which ships one working plugin, one store and one plugin whose module raises on import) installed; the broken loader also logs its WARNING and traceback to stderr:

$ xsm plugins --plain
thirdparty_audit  xsm-thirdparty-plugin 1.2.3  [xstate_statemachine.plugins]
    hooks: on_interpreter_start, on_transition
thirdparty_memory  xsm-thirdparty-plugin 1.2.3  [xstate_statemachine.stores]
kafka  xstate-statemachine 0.12.0  [xstate_statemachine.brokers]
nats  xstate-statemachine 0.12.0  [xstate_statemachine.brokers]
rabbitmq  xstate-statemachine 0.12.0  [xstate_statemachine.brokers]
redis-streams  xstate-statemachine 0.12.0  [xstate_statemachine.brokers]
sqs  xstate-statemachine 0.12.0  [xstate_statemachine.brokers]
thirdparty_broken  xsm-thirdparty-plugin 1.2.3  [xstate_statemachine.plugins]  SKIPPED: RuntimeError: xsm_thirdparty_plugin.broken: import-time failure

A loaded entry point is name distribution version [group], followed by its hooks if it overrides any. An entry point whose loader raised is listed last with SKIPPED: and the exception, so “installed but broken” is never confused with “not installed”.

--json emits one object (the order of keys is fixed):

{
  "disabled": false,
  "plugins": [
    {
      "name": "thirdparty_audit",
      "distribution": "xsm-thirdparty-plugin",
      "version": "1.2.3",
      "group": "xstate_statemachine.plugins",
      "hooks": ["on_interpreter_start", "on_transition"]
    }
  ],
  "skipped": [
    {
      "name": "thirdparty_broken",
      "distribution": "xsm-thirdparty-plugin",
      "version": "1.2.3",
      "group": "xstate_statemachine.plugins",
      "error": "RuntimeError: xsm_thirdparty_plugin.broken: import-time failure"
    }
  ]
}

With XSM_DISABLE_PLUGIN_DISCOVERY=1 it prints Plugin discovery is disabled (XSM_DISABLE_PLUGIN_DISCOVERY=1). (or "disabled": true with empty lists) and imports nothing.

Exit codes. 0 whenever the listing completes, including when entries were SKIPPED. --strict turns a failing loader into exit 1: it stops at the first one and prints a single line to stderr, no traceback:

$ xsm plugins --strict
error: plugin entry point 'thirdparty_broken' from xsm-thirdparty-plugin failed to load (RuntimeError: xsm_thirdparty_plugin.broken: import-time failure); run `xsm plugins` without --strict for the full list

Use xsm plugins --strict as the CI gate for “every installed plugin imports”. It checks loading only; constructing a plugin is checked by attach_discovered(..., strict=True).

📦 Writing a third-party plugin

A minimal package needs a plugin class and one table in pyproject.toml:

acme-audit/
├── pyproject.toml
└── acme_audit/
    └── __init__.py

[project]
name = "acme-audit"
version = "1.2.3"
dependencies = ["xstate-statemachine>=1.0,<2"]

[project.entry-points."xstate_statemachine.plugins"]
acme_audit = "acme_audit:AuditPlugin"

# acme_audit/__init__.py
from xstate_statemachine import PluginBase


class AuditPlugin(PluginBase):
    def on_interpreter_start(self, interpreter):
        print("started", interpreter.id)

    def on_transition(self, interpreter, from_states, to_states, transition):
        print("->", sorted(s.id for s in to_states))

Guidelines:

• Construct with no arguments. attach_discovered calls the class (or factory) without arguments. Read configuration from environment variables or offer a factory function as the entry point, decorated with xstate_statemachine.plugin_discovery.plugin_factory – an undecorated callable is refused without being called (so an entry point like os:system can never be executed by discovery).
• Override only the hooks you need. xsm plugins reports exactly those, and every hook is optional.
• Keep import cheap and side-effect free. Your module is imported by discover() and xsm plugins. Do no I/O at import time.
• Pin a major range of xstate-statemachine. PluginBase hooks follow the deprecation policy.
• Subclass PluginBase. discover() lists any object, but attach_discovered refuses one that is not a PluginBase instance once constructed.
• Test it the way the library tests its fixture package (tests/fixtures/xsm_thirdparty_plugin): pip install -e ., then assert that discover() finds it, that xsm plugins --json lists your hooks, and that xsm plugins --strict exits 0.

pip install -e .
xsm plugins --strict

from xstate_statemachine.plugins import discover

found = [p for p in discover() if p.distribution == "acme-audit"]
# [DiscoveredPlugin(name='acme_audit', distribution='acme-audit',
#   version='1.2.3', obj=<class 'acme_audit.AuditPlugin'>,
#   hooks=('on_interpreter_start', 'on_transition'), group=...)]

Stores and brokers use the same table under their own group:

[project.entry-points."xstate_statemachine.stores"]
acme_store = "acme_audit:MemoryStore"

[project.entry-points."xstate_statemachine.brokers"]
acme_broker = "acme_audit:Broker"

🔐 Trust model. A discovered plugin runs in your process with your privileges. There is no sandbox, and it sees every event and the context. Only allow plugins from distributions you would import yourself. See SECURITY.md.

📜 Complete Example: Custom Audit Logger

import json
from datetime import datetime
from xstate_statemachine import (
    create_machine,
    SyncInterpreter,
    MachineLogic,
    PluginBase,
)


class AuditLogPlugin(PluginBase):
    """Records a detailed audit trail of all machine activity."""

    def __init__(self):
        self.log = []

    def _record(self, event_type, details):
        entry = {
            "timestamp": datetime.now().isoformat(),
            "event_type": event_type,
            **details,
        }
        self.log.append(entry)

    def on_interpreter_start(self, interpreter):
        self._record(
            "MACHINE_START",
            {
                "machine_id": interpreter.id,
            },
        )

    def on_interpreter_stop(self, interpreter):
        self._record(
            "MACHINE_STOP",
            {
                "machine_id": interpreter.id,
                "final_context": dict(interpreter.context),
            },
        )

    def on_event_received(self, interpreter, event):
        self._record(
            "EVENT_RECEIVED",
            {
                "event": event.type,
            },
        )

    def on_transition(self, interpreter, from_states, to_states, transition):
        from_ids = sorted(
            s.id for s in from_states if s.is_atomic or s.is_final
        )
        to_ids = sorted(s.id for s in to_states if s.is_atomic or s.is_final)
        if from_ids != to_ids:
            self._record(
                "TRANSITION",
                {
                    "from": from_ids,
                    "to": to_ids,
                    "event": transition.event,
                },
            )

    def on_guard_evaluated(self, interpreter, guard_name, event, result):
        self._record(
            "GUARD_EVALUATED",
            {
                "guard": guard_name,
                "event": event.type,
                "result": result,
            },
        )

    def export_json(self):
        return json.dumps(self.log, indent=2)


# Usage
config = {
    "id": "auditDemo",
    "initial": "draft",
    "context": {
        "author": "alice",
    },
    "states": {
        "draft": {
            "on": {
                "SUBMIT": "review",
            },
        },
        "review": {
            "on": {
                "APPROVE": "published",
                "REJECT": "draft",
            },
        },
        "published": {
            "type": "final",
        },
    },
}

machine = create_machine(config)
audit = AuditLogPlugin()

interp = SyncInterpreter(machine)
interp.use(audit)
interp.start()

interp.send("SUBMIT")
interp.send("APPROVE")
interp.stop()

print(audit.export_json())
# [
#   {"timestamp": "...", "event_type": "MACHINE_START", "machine_id": "auditDemo"},
#   {"timestamp": "...", "event_type": "EVENT_RECEIVED", "event": "SUBMIT"},
#   {"timestamp": "...", "event_type": "TRANSITION", "from": ["auditDemo.draft"], "to": ["auditDemo.review"], "event": "SUBMIT"},
#   {"timestamp": "...", "event_type": "EVENT_RECEIVED", "event": "APPROVE"},
#   {"timestamp": "...", "event_type": "TRANSITION", "from": ["auditDemo.review"], "to": ["auditDemo.published"], "event": "APPROVE"},
#   {"timestamp": "...", "event_type": "MACHINE_STOP", "machine_id": "auditDemo", "final_context": {"author": "alice"}}
# ]

📈 Complete Example: Performance Monitoring

import time
from xstate_statemachine import PluginBase

class PerformancePlugin(PluginBase):
    """Measures time spent in each state and total execution time."""

    def __init__(self):
        self.start_time = None
        self.state_entry_times = {}
        self.state_durations = {}
        self.action_timings = {}
        self._current_action_start = None

    def on_interpreter_start(self, interpreter):
        self.start_time = time.monotonic()

    def on_interpreter_stop(self, interpreter):
        elapsed = time.monotonic() - self.start_time
        print(f"\n--- Performance Report for '{interpreter.id}' ---")
        print(f"Total runtime: {elapsed:.4f}s")
        print(f"State durations:")
        for state_id, duration in sorted(self.state_durations.items()):
            print(f"  {state_id}: {duration:.4f}s")
        if self.action_timings:
            print(f"Action execution counts:")
            for action_name, count in sorted(self.action_timings.items()):
                print(f"  {action_name}: {count} executions")

    def on_transition(self, interpreter, from_states, to_states, transition):
        now = time.monotonic()
        # Record duration for exited states
        from_ids = {s.id for s in from_states if s.is_atomic or s.is_final}
        to_ids = {s.id for s in to_states if s.is_atomic or s.is_final}
        for state_id in from_ids - to_ids:
            if state_id in self.state_entry_times:
                duration = now - self.state_entry_times.pop(state_id)
                self.state_durations[state_id] = (
                    self.state_durations.get(state_id, 0) + duration
                )
        # Record entry time for newly entered states
        for state_id in to_ids - from_ids:
            self.state_entry_times[state_id] = now

    def on_action_execute(self, interpreter, action):
        self.action_timings[action.type] = (
            self.action_timings.get(action.type, 0) + 1
        )

Usage:

perf = PerformancePlugin()
interp = SyncInterpreter(machine)
interp.use(perf)
interp.start()

interp.send("START")
time.sleep(0.1)  # Simulate time in 'running' state
interp.send("FINISH")
interp.stop()

# --- Performance Report for 'demo' ---
# Total runtime: 0.1032s
# State durations:
#   demo.idle: 0.0001s
#   demo.running: 0.1012s
# Action execution counts:
#   increment: 1

🕰️ Complete Example: State History Tracker

from xstate_statemachine import PluginBase


class StateHistoryPlugin(PluginBase):
    """Maintains a complete history of state transitions."""

    def __init__(self, max_history=1000):
        self.history = []
        self.max_history = max_history

    def on_transition(self, interpreter, from_states, to_states, transition):
        from_ids = sorted(
            s.id for s in from_states if s.is_atomic or s.is_final
        )
        to_ids = sorted(s.id for s in to_states if s.is_atomic or s.is_final)
        if from_ids != to_ids:
            entry = {
                "from": from_ids,
                "to": to_ids,
                "event": transition.event,
                "context_snapshot": dict(interpreter.context),
            }
            self.history.append(entry)
            # Prevent unbounded memory growth
            if len(self.history) > self.max_history:
                self.history = self.history[-self.max_history :]

    @property
    def current_state(self):
        """Returns the most recent state IDs."""
        if self.history:
            return self.history[-1]["to"]
        return []

    @property
    def previous_state(self):
        """Returns the state IDs before the last transition."""
        if self.history:
            return self.history[-1]["from"]
        return []

    def get_path(self):
        """Returns the full sequence of states visited."""
        path = []
        for entry in self.history:
            if not path:
                path.append(entry["from"])
            path.append(entry["to"])
        return path

    def was_in_state(self, state_id):
        """Returns True if the machine ever visited the given state."""
        for entry in self.history:
            if state_id in entry["from"] or state_id in entry["to"]:
                return True
        return False

Usage:

config = {
    "id": "nav",
    "initial": "home",
    "states": {
        "home": {
            "on": {
                "GO_ABOUT": "about",
                "GO_CONTACT": "contact",
            },
        },
        "about": {
            "on": {
                "GO_HOME": "home",
                "GO_CONTACT": "contact",
            },
        },
        "contact": {
            "on": {
                "GO_HOME": "home",
                "GO_ABOUT": "about",
            },
        },
    },
}

machine = create_machine(config)
tracker = StateHistoryPlugin()
interp = SyncInterpreter(machine)
interp.use(tracker)
interp.start()

interp.send("GO_ABOUT")
interp.send("GO_CONTACT")
interp.send("GO_HOME")
interp.stop()

print(tracker.get_path())
# [['nav.home'], ['nav.about'], ['nav.contact'], ['nav.home']]

print(tracker.was_in_state("nav.about"))  # True
print(tracker.was_in_state("nav.settings"))  # False
print(tracker.current_state)  # ['nav.home']
print(tracker.previous_state)  # ['nav.contact']

⚡ Plugins with the Async Interpreter

Plugins work identically with both SyncInterpreter and Interpreter. The hooks themselves are always synchronous — they are called by the interpreter before/after async operations, but the hooks do not need to be async def.

import asyncio
from xstate_statemachine import create_machine, Interpreter, LoggingInspector


async def main():
    config = {
        "id": "asyncDemo",
        "initial": "idle",
        "states": {
            "idle": {
                "on": {
                    "GO": "active",
                },
            },
            "active": {
                "type": "final",
            },
        },
    }

    machine = create_machine(config)
    interp = Interpreter(machine)
    interp.use(LoggingInspector())
    await interp.start()

    await interp.send("GO")
    await asyncio.sleep(0.1)
    await interp.stop()


asyncio.run(main())

Tip: Use PluginBase[Interpreter] or PluginBase[SyncInterpreter] as the generic type when you want IDE autocompletion for a specific interpreter type:

class MyAsyncPlugin(PluginBase[Interpreter]):
    def on_interpreter_start(self, interpreter: Interpreter):
        # Full autocompletion for async Interpreter methods
        pass