feat: integrate invoke spawn/cancel into engines and event routing

- Add InvokeManager and states_to_invoke to BaseEngine
- Track states with invocations in _enter_states(), cancel in _exit_states()
- Spawn child sessions in sync (daemon thread) and async (asyncio task) engines
- Implement finalize and autoforward for invoke events
- Add invokeid field to TriggerData for cross-session event tracking
- Implement #_parent, #_child, and #_<invokeid> send targets
- Convert InvokeDefinition to InvokeConfig in SCXML processor
- Add State(invoke=...) parameter for Python API

Author: fgmacedo

statemachine/engines/async_.py

@@ -173,6 +173,13 @@ async def _exit_states(  # type: ignore[override]
         for info in ordered_states:
             args, kwargs = await self._get_args_kwargs(info.transition, trigger_data)
 
+            # Cancel invocations for states being exited
+            if info.state is not None and getattr(info.state, "invocations", None):
+                self.invoke_manager.cancel_for_state(info.state)
+
+            # Remove from states_to_invoke if we exit before invoking
+            self.states_to_invoke.discard(info.state)
+
             if info.state is not None:  # pragma: no branch
                 await self.sm._callbacks.async_call(
                     info.state.exit.key, *args, on_error=on_error, **kwargs
@@ -358,6 +365,15 @@ async def processing_loop(  # noqa: C901
                         took_events = True
                         await self._run_microstep(enabled_transitions, internal_event)
 
+                # Spawn invocations for states entered during this macrostep
+                for state in sorted(
+                    self.states_to_invoke,
+                    key=lambda s: s.document_order,
+                ):
+                    for config in state.invocations:
+                        await self.invoke_manager.spawn_async(state, config, internal_event)
+                self.states_to_invoke.clear()
+
                 # Phase 2: remaining internal events
                 while not self.internal_queue.is_empty():  # pragma: no cover
                     internal_event = self.internal_queue.pop()
@@ -381,6 +397,16 @@ async def processing_loop(  # noqa: C901
 
                     logger.debug("External event: %s", external_event.event)
 
+                    # Handle invoke finalize and autoforward
+                    for state in self.sm.configuration:
+                        for inv in self.invoke_manager.active_for_state(state):
+                            if external_event.invokeid and inv.invokeid == external_event.invokeid:
+                                self.invoke_manager.apply_finalize(inv, external_event)
+                            if inv.config.autoforward and external_event.event:
+                                self.invoke_manager.forward_event(
+                                    inv, str(external_event.event), external_event
+                                )
+
                     # Handle lazy initial state activation.
                     # Break out of phase 3 so the outer loop restarts from phase 1
                     # (eventless/internal), ensuring internal events queued during

statemachine/engines/base.py

@@ -20,6 +20,7 @@
 from ..event_data import TriggerData
 from ..exceptions import InvalidDefinition
 from ..exceptions import TransitionNotAllowed
+from ..invoke import InvokeManager
 from ..orderedset import OrderedSet
 from ..state import HistoryState
 from ..state import State
@@ -94,6 +95,8 @@ def __init__(self, sm: "StateChart"):
         self.running = True
         self._processing = Lock()
         self._cache: Dict = {}  # Cache for _get_args_kwargs results
+        self.invoke_manager = InvokeManager(self)
+        self.states_to_invoke: "OrderedSet[State]" = OrderedSet()
 
     def empty(self):  # pragma: no cover
         return self.external_queue.is_empty()
@@ -485,6 +488,13 @@ def _exit_states(
         for info in ordered_states:
             args, kwargs = self._get_args_kwargs(info.transition, trigger_data)
 
+            # Cancel invocations for states being exited
+            if info.state is not None and getattr(info.state, "invocations", None):
+                self.invoke_manager.cancel_for_state(info.state)
+
+            # Remove from states_to_invoke if we exit before invoking
+            self.states_to_invoke.discard(info.state)
+
             # Execute `onexit` handlers — same per-block error isolation as onentry.
             if info.state is not None:  # pragma: no branch
                 self.sm._callbacks.call(info.state.exit.key, *args, on_error=on_error, **kwargs)
@@ -556,6 +566,7 @@ def _handle_final_state(self, target: State, on_entry_result: list):
         """Handle final state entry: queue done events. No direct callback dispatch."""
         if target.parent is None:
             self.running = False
+            self.invoke_manager.cancel_all()
         else:
             parent = target.parent
             grandparent = parent.parent
@@ -645,6 +656,10 @@ def _enter_states(  # noqa: C901
                     new_configuration=new_configuration,
                 )
 
+            # Track states with invocations for post-macrostep spawning
+            if getattr(target, "invocations", None):
+                self.states_to_invoke.add(target)
+
             # Handle final states
             if target.final:
                 self._handle_final_state(target, on_entry_result)

statemachine/engines/sync.py

@@ -107,11 +107,14 @@ def processing_loop(self, caller_future=None):  # noqa: C901
                         took_events = True
                         self._run_microstep(enabled_transitions, internal_event)
 
-                # TODO: Invoke platform-specific logic
-                # for state in sorted(self.states_to_invoke, key=self.entry_order):
-                #     for inv in sorted(state.invoke, key=self.document_order):
-                #         self.invoke(inv)
-                # self.states_to_invoke.clear()
+                # Spawn invocations for states entered during this macrostep
+                for state in sorted(
+                    self.states_to_invoke,
+                    key=lambda s: s.document_order,
+                ):
+                    for config in state.invocations:
+                        self.invoke_manager.spawn_sync(state, config, internal_event)
+                self.states_to_invoke.clear()
 
                 # Process remaining internal events before external events.
                 # Note: the macrostep loop above already drains the internal queue,
@@ -137,18 +140,16 @@ def processing_loop(self, caller_future=None):  # noqa: C901
                         break
 
                     logger.debug("External event: %s", external_event.event)
-                    # # TODO: Handle cancel event
-                    # if self.is_cancel_event(external_event):
-                    #     self.running = False
-                    #     return
-
-                    # TODO: Invoke states
-                    # for state in self.configuration:
-                    #     for inv in state.invoke:
-                    #         if inv.invokeid == external_event.invokeid:
-                    #             self.apply_finalize(inv, external_event)
-                    #         if inv.autoforward:
-                    #             self.send(inv.id, external_event)
+
+                    # Handle invoke finalize and autoforward
+                    for state in self.sm.configuration:
+                        for inv in self.invoke_manager.active_for_state(state):
+                            if external_event.invokeid and inv.invokeid == external_event.invokeid:
+                                self.invoke_manager.apply_finalize(inv, external_event)
+                            if inv.config.autoforward and external_event.event:
+                                self.invoke_manager.forward_event(
+                                    inv, str(external_event.event), external_event
+                                )
 
                     enabled_transitions = self.select_transitions(external_event)
                     logger.debug("Enabled transitions: %s", enabled_transitions)

statemachine/event.py

@@ -121,17 +121,26 @@ def __get__(self, instance, owner):
             return self
         return BoundEvent(id=self.id, name=self.name, delay=self.delay, _sm=instance)
 
-    def put(self, *args, send_id: "str | None" = None, **kwargs):
+    def put(self, *args, send_id: "str | None" = None, invokeid: "str | None" = None, **kwargs):
         # The `__call__` is declared here to help IDEs knowing that an `Event`
         # can be called as a method. But it is not meant to be called without
         # an SM instance. Such SM instance is provided by `__get__` method when
         # used as a property descriptor.
         assert self._sm is not None
-        trigger_data = self.build_trigger(*args, machine=self._sm, send_id=send_id, **kwargs)
+        trigger_data = self.build_trigger(
+            *args, machine=self._sm, send_id=send_id, invokeid=invokeid, **kwargs
+        )
         self._sm._put_nonblocking(trigger_data, internal=self.internal)
         return trigger_data
 
-    def build_trigger(self, *args, machine: "StateChart", send_id: "str | None" = None, **kwargs):
+    def build_trigger(
+        self,
+        *args,
+        machine: "StateChart",
+        send_id: "str | None" = None,
+        invokeid: "str | None" = None,
+        **kwargs,
+    ):
         if machine is None:
             raise RuntimeError(_("Event {} cannot be called without a SM instance").format(self))
 
@@ -140,6 +149,7 @@ def build_trigger(self, *args, machine: "StateChart", send_id: "str | None" = No
             machine=machine,
             event=self,
             send_id=send_id,
+            invokeid=invokeid,
             args=args,
             kwargs=kwargs,
         )

statemachine/event_data.py

@@ -24,6 +24,9 @@ class TriggerData:
     Allow revoking a delayed :ref:`TriggerData` instance.
     """
 
+    invokeid: "str | None" = field(compare=False, default=None)
+    """The invoke id of the child session that generated this event, if any."""
+
     execution_time: float = field(default=0.0)
     """The time at which the :ref:`Event` should run."""
 

statemachine/invoke.py

@@ -168,7 +168,11 @@ def _create_child(
         invocation: Invocation,
         trigger_data: Any,
     ) -> "StateChart | None":
-        """Create and return a child StateChart instance."""
+        """Create and return a child StateChart instance.
+
+        Sets ``_parent_sm`` and ``_invokeid`` on the child class before
+        instantiation so that ``#_parent`` sends work during initial entry.
+        """
         from .io.scxml.processor import SCXMLProcessor
 
         bridge = ParentBridge(self.sm, invokeid, invocation)
@@ -177,10 +181,21 @@ def _create_child(
         child_sm: "StateChart | None" = None
 
         if child_class is not None:
-            child_sm = child_class(listeners=[bridge])
+            # For user-provided classes, set parent refs on the class temporarily
+            child_class._parent_sm = self.sm  # type: ignore[attr-defined]
+            child_class._invokeid = invokeid  # type: ignore[attr-defined]
+            try:
+                child_sm = child_class(listeners=[bridge])
+            finally:
+                # Clean up class-level attrs (they're now on the instance)
+                del child_class._parent_sm  # type: ignore[attr-defined]
+                del child_class._invokeid  # type: ignore[attr-defined]
         elif config.content:
             processor = SCXMLProcessor()
             processor.parse_scxml(f"invoke_{invokeid}", config.content)
+            child_cls = next(iter(processor.scs.values()))
+            child_cls._parent_sm = self.sm  # type: ignore[attr-defined]
+            child_cls._invokeid = invokeid  # type: ignore[attr-defined]
             child_sm = processor.start(listeners=[bridge])
         elif config.src:
             from pathlib import Path
@@ -191,13 +206,16 @@ def _create_child(
                 path = Path(parsed.path) if parsed.scheme == "file" else Path(config.src)
                 processor = SCXMLProcessor()
                 processor.parse_scxml_file(path)
+                child_cls = next(iter(processor.scs.values()))
+                child_cls._parent_sm = self.sm  # type: ignore[attr-defined]
+                child_cls._invokeid = invokeid  # type: ignore[attr-defined]
                 child_sm = processor.start(listeners=[bridge])
 
         if child_sm is None:
             logger.warning("Could not create child for invoke %s", invokeid)
             return None
 
-        # Set parent references on child
+        # Ensure instance-level parent references are set
         child_sm._parent_sm = self.sm  # type: ignore[attr-defined]
         child_sm._invokeid = invokeid  # type: ignore[attr-defined]
 

statemachine/io/__init__.py

@@ -46,6 +46,7 @@ class BaseStateKwargs(TypedDict, total=False):
     enter: "str | ActionProtocol | Sequence[str] | Sequence[ActionProtocol]"
     exit: "str | ActionProtocol | Sequence[str] | Sequence[ActionProtocol]"
     donedata: "ActionProtocol | None"
+    invoke: "Any"
 
 
 class StateKwargs(BaseStateKwargs, total=False):

statemachine/io/scxml/actions.py

@@ -112,6 +112,7 @@ class EventDataWrapper:
     def __init__(self, event_data):
         self.event_data = event_data
         self.sendid = event_data.trigger_data.send_id
+        self.invokeid = event_data.trigger_data.invokeid or ""
         if event_data.trigger_data.event is None or event_data.trigger_data.event.internal:
             if "error.execution" == event_data.trigger_data.event:
                 self.type = "platform"
@@ -364,9 +365,60 @@ def raise_action(*args, **kwargs):
     return raise_action
 
 
+def _send_to_parent(machine: "StateChart", event: str, content: Any, params_values: dict):
+    """Route an event to the parent session via #_parent."""
+    parent_sm = getattr(machine, "_parent_sm", None)
+    if parent_sm is not None:
+        child_invokeid = getattr(machine, "_invokeid", None)
+        parent_sm.send(event, *content, invokeid=child_invokeid, **params_values)
+    else:
+        machine.send("error.communication", internal=True)
+
+
+def _send_to_child(machine: "StateChart", event: str, params_values: dict):
+    """Route an event to the first active child via #_child."""
+    if hasattr(machine, "_engine") and hasattr(machine._engine, "invoke_manager"):
+        machine._engine.invoke_manager.send_to_child(event, **params_values)
+    else:
+        machine.send("error.communication", internal=True)
+
+
+def _send_to_invokeid(
+    machine: "StateChart", target: str, event: str, action: SendAction, **kwargs
+):
+    """Route an event to a specific child by #_<invokeid>."""
+    invokeid = target[2:]
+    if hasattr(machine, "_engine") and hasattr(machine._engine, "invoke_manager"):
+        params_values = {}
+        for param in action.params:
+            if param.expr:
+                params_values[param.name] = _eval(param.expr, **kwargs)
+        machine._engine.invoke_manager.send_to_invokeid(invokeid, event, **params_values)
+    elif target.startswith("#_scxml_"):
+        machine.send("error.communication", internal=True)
+    else:
+        raise ValueError(f"Invalid target: {target}")
+
+
+def _eval_send_params(action: SendAction, **kwargs) -> dict:
+    """Evaluate namelist and <param> into a dict of keyword arguments."""
+    machine: "StateChart" = kwargs["machine"]
+    names = []
+    for name in (action.namelist or "").strip().split():
+        if not hasattr(machine.model, name):
+            raise NameError(f"Namelist variable '{name}' not found on model")
+        names.append(Param(name=name, expr=name))
+    params_values = {}
+    for param in chain(names, action.params):
+        if param.expr is None:
+            continue
+        params_values[param.name] = _eval(param.expr, **kwargs)
+    return params_values
+
+
 def create_send_action_callable(action: SendAction) -> Callable:  # noqa: C901
     content: Any = ()
-    _valid_targets = (None, "#_internal", "internal", "#_parent", "parent")
+    _valid_targets = (None, "#_internal", "internal", "#_parent", "parent", "#_child")
     if action.content:
         try:
             content = (eval(action.content, {}, {}),)
@@ -379,22 +431,35 @@ def send_action(*args, **kwargs):
         target = action.target if action.target else None
 
         if action.type and action.type != "http://www.w3.org/TR/scxml/#SCXMLEventProcessor":
-            # Per SCXML spec 6.2.3, unsupported type raises error.execution
             raise ValueError(
                 f"Unsupported send type: {action.type}. "
                 "Only 'http://www.w3.org/TR/scxml/#SCXMLEventProcessor' is supported"
             )
+
         if target not in _valid_targets:
             if target and target.startswith("#_scxml_"):
-                # Valid SCXML session reference but undispatchable → error.communication
+                # Valid SCXML session reference but undispatchable -> error.communication
                 machine.send("error.communication", internal=True)
+                return
+            elif target and target.startswith("#_"):
+                # Handle #_invokeid target (send to specific child by invoke id)
+                _send_to_invokeid(machine, target, event, action, **kwargs)
+                return
             else:
-                # Invalid target expression → error.execution (raised as exception)
                 raise ValueError(f"Invalid target: {target}. Must be one of {_valid_targets}")
             return
 
-        internal = target in ("#_internal", "internal")
+        params_values = _eval_send_params(action, **kwargs)
 
+        # Handle cross-session targets
+        if target in ("#_parent", "parent"):
+            _send_to_parent(machine, event, content, params_values)
+            return
+        if target == "#_child":
+            _send_to_child(machine, event, params_values)
+            return
+
+        internal = target in ("#_internal", "internal")
         send_id = None
         if action.id:
             send_id = action.id
@@ -403,20 +468,6 @@ def send_action(*args, **kwargs):
             setattr(machine.model, action.idlocation, send_id)
 
         delay = ParseTime.parse_delay(action.delay, action.delayexpr, **kwargs)
-
-        # Per SCXML spec, if namelist evaluation causes an error (e.g., variable not found),
-        # the send MUST NOT be dispatched and error.execution is raised.
-        names = []
-        for name in (action.namelist or "").strip().split():
-            if not hasattr(machine.model, name):
-                raise NameError(f"Namelist variable '{name}' not found on model")
-            names.append(Param(name=name, expr=name))
-        params_values = {}
-        for param in chain(names, action.params):
-            if param.expr is None:
-                continue
-            params_values[param.name] = _eval(param.expr, **kwargs)
-
         Event(id=event, name=event, delay=delay, internal=internal, _sm=machine).put(
             *content,
             send_id=send_id,