feat: invoke manager hooks, deferred imports promotion, and session tracking

- Promote `asyncio`, `iterate_states`, and `InvokeManager` imports to
  top-level (no circular dependency for these).
- Replace class-level attr hacks with `InvokeSession` dataclass and
  `_start` constructor parameter for child StateMachines.
- Add engine hook methods on InvokeManager (on_state_entered,
  on_state_exiting, spawn_pending_sync/async, handle_external_event).
- Wire engines to delegate invoke lifecycle to InvokeManager hooks.

Author: fgmacedo

statemachine/engines/async_.py

@@ -173,12 +173,8 @@ 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 self._invoke is not None:
+                self._invoke.on_state_exiting(info.state)
 
             if info.state is not None:  # pragma: no branch
                 await self.sm._callbacks.async_call(
@@ -250,8 +246,8 @@ async def _enter_states(  # noqa: C901
                 )
 
             # Track states with invocations for post-macrostep spawning
-            if getattr(target, "invocations", None):
-                self.states_to_invoke.add(target)
+            if self._invoke is not None:
+                self._invoke.on_state_entered(target)
 
             # Handle final states
             if target.final:
@@ -383,13 +379,8 @@ async def processing_loop(  # noqa: C901
                         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()
+                if self._invoke is not None:
+                    await self._invoke.spawn_pending_async(internal_event)
 
                 # Phase 2: remaining internal events
                 while not self.internal_queue.is_empty():  # pragma: no cover
@@ -412,23 +403,13 @@ async def processing_loop(  # noqa: C901
                         # transitions can be processed while we wait.
                         break
 
-                    # Forward delayed cross-session events to their target
-                    if external_event.forward_target:
-                        self._forward_to_target(external_event)
-                        continue
+                    # Handle invoke-related event processing (forward, finalize, autoforward)
+                    if self._invoke is not None:
+                        if self._invoke.handle_external_event(external_event):
+                            continue
 
                     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,13 +20,13 @@
 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
 from ..transition import Transition
 
 if TYPE_CHECKING:
+    from ..invoke import InvokeManager
     from ..statemachine import StateChart
 
 logger = logging.getLogger(__name__)
@@ -95,8 +95,7 @@ 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()
+        self._invoke: "InvokeManager | None" = None  # set by StateChart if needed
 
     def empty(self):  # pragma: no cover
         return self.external_queue.is_empty()
@@ -138,33 +137,6 @@ def cancel_event(self, send_id: str):
         """Cancel the event with the given send_id."""
         self.external_queue.remove(send_id)
 
-    def _forward_to_target(self, trigger_data: TriggerData):
-        """Forward an event to a cross-session target instead of processing it.
-
-        Called by the processing loop when ``trigger_data.forward_target`` is set.
-        This supports delayed cross-session sends: the event sits on the child's
-        queue with a delay, and when it expires the processing loop forwards it
-        to the named target.
-        """
-        target = trigger_data.forward_target
-        event_name = str(trigger_data.event)
-        if target in ("#_parent", "parent"):
-            parent_sm = getattr(self.sm, "_parent_sm", None)
-            if parent_sm is not None:
-                child_invokeid = getattr(self.sm, "_invokeid", None)
-                parent_sm.send(
-                    event_name,
-                    *trigger_data.args,
-                    invokeid=child_invokeid,
-                    **trigger_data.kwargs,
-                )
-            else:
-                self.sm.send("error.communication", internal=True)
-        elif target == "#_child":
-            self.invoke_manager.send_to_child(event_name, **trigger_data.kwargs)
-        else:
-            logger.warning("Unknown forward_target: %s", target)
-
     def _on_error_handler(self) -> "Callable[[Exception], None] | None":
         """Return a per-block error handler, or ``None``.
 
@@ -528,12 +500,8 @@ 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)
+            if self._invoke is not None:
+                self._invoke.on_state_exiting(info.state)
 
             # Execute `onexit` handlers — same per-block error isolation as onentry.
             if info.state is not None:  # pragma: no branch
@@ -602,16 +570,16 @@ def _add_state_to_configuration(self, target: State):
         if not self.sm.atomic_configuration_update:
             self.sm.configuration |= {target}
 
-    def _terminate_machine(self):
-        """SCXML termination: exit all active states in exit order, firing onexit handlers.
+    def _terminate_child_session(self):
+        """Exit all active states (firing onexit handlers) and stop the engine.
 
-        Per SCXML spec, when a top-level final state is reached, the machine
-        terminates. All active states have their ``onexit`` handlers fired
-        (in reverse document order), but the final configuration is preserved
-        so that observers can see which final state was reached.
+        Called when a child session (one with ``_invoke_session``) reaches a
+        top-level final state. Per SCXML spec, all active states have their
+        ``onexit`` handlers fired (in reverse document order), but the final
+        configuration is preserved so that observers can see which final state
+        was reached.
         """
         on_error = self._on_error_handler()
-        # Sort active states by document_order (reverse) for exit order
         active_states = sorted(
             self.sm.configuration,
             key=lambda s: s.document_order,
@@ -621,24 +589,24 @@ def _terminate_machine(self):
         event_data = EventData(trigger_data=trigger_data, transition=None)
         args, kwargs = event_data.args, event_data.extended_kwargs
         for state in active_states:
-            if getattr(state, "invocations", None):
-                self.invoke_manager.cancel_for_state(state)
+            if self._invoke is not None:
+                self._invoke.on_state_exiting(state)
             self.sm._callbacks.call(state.exit.key, *args, on_error=on_error, **kwargs)
-        # Note: we intentionally do NOT clear configuration — the final state
-        # must remain visible for assertions and done.invoke reporting.
-        self.invoke_manager.cancel_all()
+        if self._invoke is not None:
+            self._invoke.on_terminate()
         self.running = False
 
     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:
-            if getattr(self.sm, "_parent_sm", None) is not None:
+            if getattr(self.sm, "_invoke_session", None) is not None:
                 # Child session: fire onexit on all active states before terminating
                 # so that #_parent sends in <onexit> of final states are delivered
-                self._terminate_machine()
+                self._terminate_child_session()
             else:
                 self.running = False
-                self.invoke_manager.cancel_all()
+                if self._invoke is not None:
+                    self._invoke.on_terminate()
         else:
             parent = target.parent
             grandparent = parent.parent
@@ -729,8 +697,8 @@ def _enter_states(  # noqa: C901
                 )
 
             # Track states with invocations for post-macrostep spawning
-            if getattr(target, "invocations", None):
-                self.states_to_invoke.add(target)
+            if self._invoke is not None:
+                self._invoke.on_state_entered(target)
 
             # Handle final states
             if target.final:

statemachine/engines/sync.py

@@ -108,13 +108,8 @@ def processing_loop(self, caller_future=None):  # noqa: C901
                         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:
-                        self.invoke_manager.spawn_sync(state, config, internal_event)
-                self.states_to_invoke.clear()
+                if self._invoke is not None:
+                    self._invoke.spawn_pending_sync(internal_event)
 
                 # Process remaining internal events before external events.
                 # Note: the macrostep loop above already drains the internal queue,
@@ -139,23 +134,13 @@ def processing_loop(self, caller_future=None):  # noqa: C901
                         # transitions can be processed while we wait.
                         break
 
-                    # Forward delayed cross-session events to their target
-                    if external_event.forward_target:
-                        self._forward_to_target(external_event)
-                        continue
+                    # Handle invoke-related event processing (forward, finalize, autoforward)
+                    if self._invoke is not None:
+                        if self._invoke.handle_external_event(external_event):
+                            continue
 
                     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
-                                )
-
                     enabled_transitions = self.select_transitions(external_event)
                     logger.debug("Enabled transitions: %s", enabled_transitions)
                     if enabled_transitions:

statemachine/event.py

@@ -98,20 +98,20 @@ def is_same_event(self, *_args, event: "str | None" = None, **_kwargs) -> bool:
         if self == event:
             return True
         if event is not None:
-            event_str = str(event)
-            self_dot = str(self).replace("_", ".")
-            event_dot = event_str.replace("_", ".")
-
-            # Exact match with dot/underscore normalization
-            if self_dot == event_dot:
-                return True
-
-            # SCXML prefix matching: descriptor "done.invoke.active" matches
-            # actual event "done.invoke.active.abc123"
-            if event_dot.startswith(self_dot + "."):
-                return True
+            return self._is_prefix_match(str(event))
         return False
 
+    def _is_prefix_match(self, event_str: str) -> bool:
+        """SCXML prefix matching with dot/underscore normalization.
+
+        ``'done.invoke.x'`` matches ``'done.invoke.x.uuid'``.
+        """
+        self_dot = str(self).replace("_", ".")
+        event_dot = event_str.replace("_", ".")
+        if self_dot == event_dot:
+            return True
+        return event_dot.startswith(self_dot + ".")
+
     def _add_callback(self, callback, grouper: CallbackGroup, is_event=False, **kwargs):
         if self._transitions is None:
             raise InvalidDefinition(
@@ -141,7 +141,6 @@ def put(
         *args,
         send_id: "str | None" = None,
         invokeid: "str | None" = None,
-        forward_target: "str | None" = None,
         **kwargs,
     ):
         # The `__call__` is declared here to help IDEs knowing that an `Event`
@@ -154,7 +153,6 @@ def put(
             machine=self._sm,
             send_id=send_id,
             invokeid=invokeid,
-            forward_target=forward_target,
             **kwargs,
         )
         self._sm._put_nonblocking(trigger_data, internal=self.internal)
@@ -166,7 +164,6 @@ def build_trigger(
         machine: "StateChart",
         send_id: "str | None" = None,
         invokeid: "str | None" = None,
-        forward_target: "str | None" = None,
         **kwargs,
     ):
         if machine is None:
@@ -178,7 +175,6 @@ def build_trigger(
             event=self,
             send_id=send_id,
             invokeid=invokeid,
-            forward_target=forward_target,
             args=args,
             kwargs=kwargs,
         )

statemachine/event_data.py

@@ -30,16 +30,6 @@ class TriggerData:
     execution_time: float = field(default=0.0)
     """The time at which the :ref:`Event` should run."""
 
-    forward_target: "str | None" = field(compare=False, default=None)
-    """When set, the processing loop forwards this event to the named target
-    (e.g. ``"#_parent"``, ``"#_child"``) instead of processing it normally.
-
-    Used for delayed cross-session sends: the event sits on the child's queue
-    with a delay, and when the delay expires it is forwarded to the target.
-    If the child terminates first, the event is never processed — automatic
-    cancellation per SCXML spec.
-    """
-
     model: Any = field(init=False, compare=False)
     """A reference to the underlying model that holds the current :ref:`State`."""