Orchestrator

Orchestrator coordinates the machine, state store, guards, and actions for stateful event processing (load state, process, persist, run actions, schedule delayed transitions).

Orchestrator

Orchestrator(machine: StateMachine, state_store: StateStore | AsyncStateStore, guards: GuardRegistry | None = None, actions: ActionRegistry | None = None, *, context_validator: ContextValidatorFn | None = None, executor: _ActionExecutor | None = None, scheduler: SchedulerAdapter | None = None, use_initial_state_when_missing: bool = True, invoke_adapter: InvokeAdapter | None = None, hooks: list[TransitionHook] | None = None, audit_store: AuditStore | None = None, event_store: EventStore | None = None)

Runs the full FSM sandwich loop: load -> process -> persist -> execute.

Parameters:

Name	Type	Description	Default
machine	StateMachine	The StateMachine definition.	required
state_store	StateStore | AsyncStateStore	Sync or async state store.	required
guards	GuardRegistry | None	Guard registry (bound to machine automatically).	None
actions	ActionRegistry | None	Action registry.	None
context_validator	ContextValidatorFn | None	Optional callable for pre-transition context validation.	None
executor	_ActionExecutor | None	Optional _ActionExecutor (created automatically if omitted).	None
scheduler	SchedulerAdapter | None	Optional scheduler for delayed transitions.	None
use_initial_state_when_missing	bool	Use machine's initial state for new entities.	True
invoke_adapter	InvokeAdapter | None	Optional adapter to start/stop invoked services on state enter/exit.	None
hooks	list[TransitionHook] | None	Optional list of TransitionHook implementations for lifecycle callbacks.	None
audit_store	AuditStore | None	Optional AuditStore for transition audit logging.	None

Context validation options (choose one or combine):

1. ``context_validator`` (recommended for external validation, e.g. PyCharter):
   Callable with signature ``(entity_id, state, trigger, context) -> (ok, errors)``.
   Runs BEFORE guards. Use for schema or contract validation.

2. ``machine.meta["validate_context"]``: Config-driven, set in YAML/dict.
   Validates that required keys exist in context. Lightweight, no external deps.

3. Guards: Use for business-rule validation that depends on state + context.
   Runs AFTER context_validator. Best for domain logic (e.g., "is_cancellable").

Parameters:

Name	Type	Description	Default
machine	StateMachine	The StateMachine definition.	required
state_store	StateStore | AsyncStateStore	Sync or async state store for entity persistence.	required
guards	GuardRegistry | None	Optional guard registry (defaults to machine's registry).	None
actions	ActionRegistry | None	Optional action registry (defaults to machine's registry).	None
context_validator	ContextValidatorFn | None	Optional pre-transition context validator callable.	None
executor	_ActionExecutor | None	Optional _ActionExecutor (auto-created if omitted).	None
scheduler	SchedulerAdapter | None	Optional scheduler for delayed transitions.	None
use_initial_state_when_missing	bool	Use machine's initial state for new entities.	True
invoke_adapter	InvokeAdapter | None	Optional adapter for invoked services.	None
hooks	list[TransitionHook] | None	Optional list of TransitionHook implementations.	None
audit_store	AuditStore | None	Optional AuditStore for transition audit logging.	None
event_store	EventStore | None	Optional EventStore for append-only event persistence.	None

Source code in src/pystator/orchestrator.py

def __init__(
    self,
    machine: StateMachine,
    state_store: StateStore | AsyncStateStore,
    guards: GuardRegistry | None = None,
    actions: ActionRegistry | None = None,
    *,
    context_validator: ContextValidatorFn | None = None,
    executor: _ActionExecutor | None = None,
    scheduler: SchedulerAdapter | None = None,
    use_initial_state_when_missing: bool = True,
    invoke_adapter: InvokeAdapter | None = None,
    hooks: list[TransitionHook] | None = None,
    audit_store: AuditStore | None = None,
    event_store: EventStore | None = None,
) -> None:
    """Initialize the orchestrator.

    Args:
        machine: The StateMachine definition.
        state_store: Sync or async state store for entity persistence.
        guards: Optional guard registry (defaults to machine's registry).
        actions: Optional action registry (defaults to machine's registry).
        context_validator: Optional pre-transition context validator callable.
        executor: Optional _ActionExecutor (auto-created if omitted).
        scheduler: Optional scheduler for delayed transitions.
        use_initial_state_when_missing: Use machine's initial state for new entities.
        invoke_adapter: Optional adapter for invoked services.
        hooks: Optional list of TransitionHook implementations.
        audit_store: Optional AuditStore for transition audit logging.
        event_store: Optional EventStore for append-only event persistence.
    """
    self._machine = machine
    self._store = state_store

    # Bind guards if provided
    if guards is not None:
        machine.bind_guards(guards)

    # Use machine's registries if not explicitly provided
    self._guards = guards or machine.guard_registry
    self._actions = actions or machine.action_registry
    self._executor = executor or _ActionExecutor(self._actions, log_execution=True)
    self._context_validator = context_validator
    self._scheduler = scheduler
    self._use_initial = use_initial_state_when_missing
    self._invoke_adapter = invoke_adapter
    self._audit_store = audit_store
    self._event_store = event_store

    # Set up transition observer if hooks provided
    self._observer: TransitionObserver | None = None
    if hooks:
        self._observer = TransitionObserver()
        for hook in hooks:
            self._observer.add_hook(hook)

machine property

machine: StateMachine

The underlying StateMachine definition.

store property

store: StateStore | AsyncStateStore

The state store backing this orchestrator.

process_event

process_event(entity_id: str, event: str | Event, context: dict[str, Any] | None = None) -> TransitionResult

Run the sandwich loop synchronously.

Source code in src/pystator/orchestrator.py

def process_event(
    self,
    entity_id: str,
    event: str | Event,
    context: dict[str, Any] | None = None,
) -> TransitionResult:
    """Run the sandwich loop synchronously."""
    store = self._store
    if not isinstance(store, StateStore):
        raise TypeError(
            "process_event requires a sync StateStore "
            "(get_state, set_state, get_context)"
        )

    current_state = store.get_state(entity_id)
    if current_state is None:
        if self._use_initial:
            current_state = self._machine.get_initial_state().name
        else:
            raise ValueError(
                f"Entity {entity_id!r} has no state and "
                "use_initial_state_when_missing is False"
            )

    parsed = parse_stored_state(current_state, self._machine.is_parallel_state)
    if isinstance(parsed, ParallelStateConfig):
        return self._process_parallel_event(
            entity_id, parsed, event, context, store
        )

    current_state = parsed
    ctx = dict(store.get_context(entity_id))
    if isinstance(event, Event):
        ctx.update(event.payload)
    else:
        event = Event(trigger=event)
    if context:
        ctx.update(context)
    ctx["_event"] = event
    ctx["_entity_id"] = entity_id

    # Append event to event store (input, before processing)
    if self._event_store is not None:
        payload = dict(event.payload) if event.payload else None
        self._event_store.append_event(
            entity_id,
            event.trigger,
            payload=payload,
            event_id=str(event.event_id),
            source=event.source,
            machine_name=self._machine.name,
        )

    # Notify hooks: before processing
    if self._observer is not None:
        self._observer.before_process(entity_id, current_state, event.trigger, ctx)

    # Validate context if configured (inline machine validation)
    if self._machine.meta.get("validate_context"):
        valid, errors = self._machine.validate_context(ctx)
        if not valid:
            return TransitionResult.failure_result(
                source_state=current_state,
                trigger=event.trigger,
                error=FSMError(
                    "Context validation failed",
                    context={"validation_errors": errors},
                ),
                metadata={"reason": "context_validation"},
            )

    # External context validator (e.g. PyCharter data contract validation)
    if self._context_validator is not None:
        ok, val_errors = self._context_validator(
            entity_id, current_state, event.trigger, ctx
        )
        if not ok:
            return TransitionResult.failure_result(
                source_state=current_state,
                trigger=event.trigger,
                error=FSMError(
                    f"Validation failed: {'; '.join(val_errors)}",
                    context={"validation_errors": val_errors},
                ),
                metadata={"reason": "context_validation"},
            )

    # Notify hooks: transition starting
    if self._observer is not None:
        self._observer.before_transition(current_state, event.trigger, ctx)

    try:
        result = self._machine.process(current_state, event, ctx)
    except Exception as exc:
        if self._observer is not None:
            self._observer.on_error(exc, ctx)
        raise

    # Notify hooks: transition complete
    if self._observer is not None:
        self._observer.after_transition(result, ctx)

    if result.success and result.target_state is not None:
        self._machine._engine.apply_history(result)
        state_changed = result.target_state != current_state

        if state_changed:
            # Cancel pending delayed transitions (sync path)
            if self._scheduler:
                import asyncio

                try:
                    loop = asyncio.get_running_loop()
                    loop.create_task(self._scheduler.cancel_for_entity(entity_id))
                except RuntimeError:
                    asyncio.run(self._scheduler.cancel_for_entity(entity_id))

            metadata = {
                "trigger": result.trigger,
                "source_state": result.source_state,
                "target_state": result.target_state,
                **self._persist_flags(result.target_state),
            }
            if self._machine.is_parallel_state(result.target_state):
                config = self._machine.enter_parallel_state(result.target_state)
                store.set_state(entity_id, config.to_string(), metadata=metadata)
            else:
                store.set_state(entity_id, result.target_state, metadata=metadata)

        if self._invoke_adapter is not None:
            get_invoke = lambda n: self._machine.get_state(n).invoke
            run_invoke_stop(
                self._invoke_adapter,
                result.source_state,
                get_invoke,
                ctx,
            )
        if result.all_action_specs:
            ctx["new_status"] = result.target_state
            self._executor.execute(result, ctx)
        if self._invoke_adapter is not None:
            get_invoke = lambda n: self._machine.get_state(n).invoke
            run_invoke_start(
                self._invoke_adapter,
                result.target_state,
                result.source_state,
                get_invoke,
                ctx,
            )

        # Persist updated context (filter internal keys)
        persist_ctx = {k: v for k, v in ctx.items() if not k.startswith("_")}
        if hasattr(store, "set_context"):
            store.set_context(entity_id, persist_ctx)

        # Fire machine callbacks
        self._machine._fire_callbacks(result, ctx)

        # Record audit trail
        if self._audit_store is not None:
            transition_id = self._audit_store.record_transition(
                entity_id=entity_id,
                source_state=result.source_state,
                trigger=result.trigger,
                target_state=result.target_state,
                success=result.success,
            )
            if isinstance(transition_id, str):
                result.metadata["transition_id"] = transition_id

        # Post-transition chain (auto-transitions + when-routes)
        self._apply_post_transition_chain(
            entity_id, result.target_state, ctx, store
        )

    return result

async_process_event async

async_process_event(entity_id: str, event: str | Event, context: dict[str, Any] | None = None) -> TransitionResult

Run the sandwich loop asynchronously.

Source code in src/pystator/orchestrator.py

async def async_process_event(
    self,
    entity_id: str,
    event: str | Event,
    context: dict[str, Any] | None = None,
) -> TransitionResult:
    """Run the sandwich loop asynchronously."""
    store = self._store
    if not isinstance(store, AsyncStateStore):
        raise TypeError(
            "async_process_event requires an AsyncStateStore "
            "(aget_state, aset_state, aget_context)"
        )

    current_state = await store.aget_state(entity_id)
    if current_state is None:
        if self._use_initial:
            current_state = self._machine.get_initial_state().name
        else:
            raise ValueError(
                f"Entity {entity_id!r} has no state and "
                "use_initial_state_when_missing is False"
            )

    parsed = parse_stored_state(current_state, self._machine.is_parallel_state)
    if isinstance(parsed, ParallelStateConfig):
        return await self._process_parallel_event_async(
            entity_id, parsed, event, context, store
        )

    current_state = parsed
    ctx = dict(await store.aget_context(entity_id))
    if isinstance(event, Event):
        ctx.update(event.payload)
    else:
        event = Event(trigger=event)
    if context:
        ctx.update(context)
    ctx["_event"] = event
    ctx["_entity_id"] = entity_id

    # Append event to event store (input, before processing)
    if self._event_store is not None:
        payload = dict(event.payload) if event.payload else None
        self._event_store.append_event(
            entity_id,
            event.trigger,
            payload=payload,
            event_id=str(event.event_id),
            source=event.source,
            machine_name=self._machine.name,
        )

    # Notify hooks: before processing
    if self._observer is not None:
        self._observer.before_process(entity_id, current_state, event.trigger, ctx)

    if self._machine.meta.get("validate_context"):
        valid, errors = self._machine.validate_context(ctx)
        if not valid:
            return TransitionResult.failure_result(
                source_state=current_state,
                trigger=event.trigger,
                error=FSMError(
                    "Context validation failed",
                    context={"validation_errors": errors},
                ),
                metadata={"reason": "context_validation"},
            )

    # External context validator (e.g. PyCharter data contract validation)
    if self._context_validator is not None:
        if inspect.iscoroutinefunction(self._context_validator):
            ok, val_errors = await self._context_validator(
                entity_id, current_state, event.trigger, ctx
            )
        else:
            ok, val_errors = self._context_validator(
                entity_id, current_state, event.trigger, ctx
            )
        if not ok:
            return TransitionResult.failure_result(
                source_state=current_state,
                trigger=event.trigger,
                error=FSMError(
                    f"Validation failed: {'; '.join(val_errors)}",
                    context={"validation_errors": val_errors},
                ),
                metadata={"reason": "context_validation"},
            )

    # Notify hooks: transition starting
    if self._observer is not None:
        self._observer.before_transition(current_state, event.trigger, ctx)

    try:
        result = await self._machine.aprocess(current_state, event, ctx)
    except Exception as exc:
        if self._observer is not None:
            self._observer.on_error(exc, ctx)
        raise

    # Notify hooks: transition complete
    if self._observer is not None:
        self._observer.after_transition(result, ctx)

    if result.success and result.target_state is not None:
        self._machine._engine.apply_history(result)
        state_changed = result.target_state != current_state

        if state_changed:
            # Cancel pending delayed transitions
            if self._scheduler:
                await self._scheduler.cancel_for_entity(entity_id)

            metadata = {
                "trigger": result.trigger,
                "source_state": result.source_state,
                "target_state": result.target_state,
                **self._persist_flags(result.target_state),
            }
            if self._machine.is_parallel_state(result.target_state):
                config = self._machine.enter_parallel_state(result.target_state)
                await store.aset_state(
                    entity_id, config.to_string(), metadata=metadata
                )
            else:
                await store.aset_state(
                    entity_id, result.target_state, metadata=metadata
                )

            # Schedule delayed transitions from new state
            if self._scheduler:
                await self._schedule_delayed(entity_id, result.target_state, ctx)

        if self._invoke_adapter is not None:
            get_invoke = lambda n: self._machine.get_state(n).invoke
            run_invoke_stop(
                self._invoke_adapter,
                result.source_state,
                get_invoke,
                ctx,
            )
        if result.all_action_specs:
            ctx["new_status"] = result.target_state
            await self._executor.async_execute(result, ctx)
        if self._invoke_adapter is not None:
            get_invoke = lambda n: self._machine.get_state(n).invoke
            run_invoke_start(
                self._invoke_adapter,
                result.target_state,
                result.source_state,
                get_invoke,
                ctx,
            )

        # Persist updated context (filter internal keys)
        persist_ctx = {k: v for k, v in ctx.items() if not k.startswith("_")}
        if hasattr(store, "aset_context"):
            await store.aset_context(entity_id, persist_ctx)
        elif hasattr(store, "set_context"):
            store.set_context(entity_id, persist_ctx)

        # Fire machine callbacks
        await self._machine._afire_callbacks(result, ctx)

        # Record audit trail
        if self._audit_store is not None:
            transition_id = self._audit_store.record_transition(
                entity_id=entity_id,
                source_state=result.source_state,
                trigger=result.trigger,
                target_state=result.target_state,
                success=result.success,
            )
            if isinstance(transition_id, str):
                result.metadata["transition_id"] = transition_id

        # Post-transition chain (auto-transitions + when-routes)
        await self._async_apply_post_transition_chain(
            entity_id, result.target_state, ctx, store
        )

    return result

apply_data

apply_data(entity_id: str, data: dict[str, Any]) -> list[TransitionResult]

Update entity context and evaluate when-route transitions (sync).

Loads the current state and context from the store, merges data, evaluates the post-transition chain (auto + when-routes), persists any resulting state changes, and returns the list of fired results.

Requires classification.auto_route: true in the machine meta.

Source code in src/pystator/orchestrator.py

def apply_data(
    self,
    entity_id: str,
    data: dict[str, Any],
) -> list[TransitionResult]:
    """Update entity context and evaluate when-route transitions (sync).

    Loads the current state and context from the store, merges *data*,
    evaluates the post-transition chain (auto + when-routes), persists
    any resulting state changes, and returns the list of fired results.

    Requires ``classification.auto_route: true`` in the machine meta.
    """
    store = self._store
    if not isinstance(store, StateStore):
        raise TypeError("Sync apply_data requires a sync StateStore")

    current_state = store.get_state(entity_id)
    if current_state is None:
        initial = self._machine.get_initial_state()
        current_state = initial.name if initial else None
    if current_state is None:
        raise ValueError(f"No state found for entity {entity_id}")

    ctx = (
        dict(store.get_context(entity_id)) if hasattr(store, "get_context") else {}
    )
    ctx.update(data)

    results = self._machine._engine.evaluate_post_transition(current_state, ctx)
    for pr in results:
        if not (pr.success and pr.target_state):
            continue
        self._machine._engine.apply_history(pr)

        md = {
            "trigger": pr.trigger,
            "source_state": pr.source_state,
            "target_state": pr.target_state,
            **self._persist_flags(pr.target_state),
        }
        store.set_state(entity_id, pr.target_state, metadata=md)

        if pr.all_action_specs:
            ctx["new_status"] = pr.target_state
            self._executor.execute(pr, ctx)

        persist_ctx = {k: v for k, v in ctx.items() if not k.startswith("_")}
        if hasattr(store, "set_context"):
            store.set_context(entity_id, persist_ctx)

        self._machine._fire_callbacks(pr, ctx)

        if self._audit_store is not None:
            self._audit_store.record_transition(
                entity_id=entity_id,
                source_state=pr.source_state,
                trigger=pr.trigger,
                target_state=pr.target_state,
                success=pr.success,
            )

    return results

async_apply_data async

async_apply_data(entity_id: str, data: dict[str, Any]) -> list[TransitionResult]

Async version of :meth:apply_data.

Source code in src/pystator/orchestrator.py

async def async_apply_data(
    self,
    entity_id: str,
    data: dict[str, Any],
) -> list[TransitionResult]:
    """Async version of :meth:`apply_data`."""
    store = self._store
    if not isinstance(store, AsyncStateStore):
        raise TypeError("async_apply_data requires an AsyncStateStore")

    current_state = await store.aget_state(entity_id)
    if current_state is None:
        initial = self._machine.get_initial_state()
        current_state = initial.name if initial else None
    if current_state is None:
        raise ValueError(f"No state found for entity {entity_id}")

    if hasattr(store, "aget_context"):
        ctx = dict(await store.aget_context(entity_id))
    elif hasattr(store, "get_context"):
        ctx = dict(store.get_context(entity_id))
    else:
        ctx = {}
    ctx.update(data)

    results = await self._machine._engine.async_evaluate_post_transition(
        current_state, ctx
    )
    for pr in results:
        if not (pr.success and pr.target_state):
            continue
        self._machine._engine.apply_history(pr)

        md = {
            "trigger": pr.trigger,
            "source_state": pr.source_state,
            "target_state": pr.target_state,
            **self._persist_flags(pr.target_state),
        }
        await store.aset_state(entity_id, pr.target_state, metadata=md)

        if pr.all_action_specs:
            ctx["new_status"] = pr.target_state
            await self._executor.async_execute(pr, ctx)

        persist_ctx = {k: v for k, v in ctx.items() if not k.startswith("_")}
        if hasattr(store, "aset_context"):
            await store.aset_context(entity_id, persist_ctx)
        elif hasattr(store, "set_context"):
            store.set_context(entity_id, persist_ctx)

        await self._machine._afire_callbacks(pr, ctx)

        if self._audit_store is not None:
            self._audit_store.record_transition(
                entity_id=entity_id,
                source_state=pr.source_state,
                trigger=pr.trigger,
                target_state=pr.target_state,
                success=pr.success,
            )

    return results

replay

replay(entity_id: str, *, up_to_sequence: int | None = None) -> list[TransitionResult]

Replay an entity from the event store, returning results at each step.

Creates a fresh in-memory EntitySession at the machine's initial state, then replays each persisted event in sequence order.

Parameters:

Name	Type	Description	Default
entity_id	str	The entity to replay.	required
up_to_sequence	int | None	Stop after this sequence number (inclusive). If None, replays all events.	None

Returns:

Type	Description
list[TransitionResult]	List of TransitionResult objects, one per replayed event.

Raises:

Type	Description
RuntimeError	If no event store is configured.

Source code in src/pystator/orchestrator.py

def replay(
    self,
    entity_id: str,
    *,
    up_to_sequence: int | None = None,
) -> list[TransitionResult]:
    """Replay an entity from the event store, returning results at each step.

    Creates a fresh in-memory EntitySession at the machine's initial state,
    then replays each persisted event in sequence order.

    Args:
        entity_id: The entity to replay.
        up_to_sequence: Stop after this sequence number (inclusive). If
            ``None``, replays all events.

    Returns:
        List of ``TransitionResult`` objects, one per replayed event.

    Raises:
        RuntimeError: If no event store is configured.
    """
    if self._event_store is None:
        raise RuntimeError("No event store configured on this orchestrator")

    limit = up_to_sequence if up_to_sequence else 10000
    events = self._event_store.get_events(entity_id, limit=limit)

    instance = self._machine.create(entity_id=entity_id)
    results: list[TransitionResult] = []
    for evt in events:
        if up_to_sequence is not None and evt["sequence"] > up_to_sequence:
            break
        payload = evt.get("payload") or {}
        result = instance.send(evt["trigger"], **payload)
        results.append(result)
    return results

close async

close() -> None

Clean up resources (scheduler, etc.).

Source code in src/pystator/orchestrator.py

async def close(self) -> None:
    """Clean up resources (scheduler, etc.)."""
    if self._scheduler:
        await self._scheduler.close()