fix: widen type annotations for compound/parallel state transitions (#580)

* fix: widen type annotations for compound/parallel state transitions

NestedStateFactory.__new__ returns a State instance at runtime, but mypy
sees class Foo(State.Compound) as type[Foo]. Accept NestedStateFactory in
the union type of _ToState, _FromState, and the metaclass stubs so that
compound/parallel states work as transition targets without type: ignore.

Also includes: ai_shell example, invoke/error.execution engine support,
event id helpers, and related test coverage.

Author: fgmacedo

AGENTS.md

@@ -180,21 +180,28 @@ uv run mypy statemachine/ tests/
 
 - **Formatter/Linter:** ruff (line length 99, target Python 3.9)
 - **Rules:** pycodestyle, pyflakes, isort, pyupgrade, flake8-comprehensions, flake8-bugbear, flake8-pytest-style
-- **Imports:** single-line, sorted by isort
+- **Imports:** single-line, sorted by isort. **Always prefer top-level imports** — only use
+  lazy (in-function) imports when strictly necessary to break circular dependencies
 - **Docstrings:** Google convention
 - **Naming:** PascalCase for classes, snake_case for functions/methods, UPPER_SNAKE_CASE for constants
 - **Type hints:** used throughout; `TYPE_CHECKING` for circular imports
 - Pre-commit hooks enforce ruff + mypy + pytest
 
 ## Design principles
 
-- **Follow SOLID principles.** In particular:
+- **Use GRASP/SOLID patterns to guide decisions.** When refactoring or designing, explicitly
+  apply patterns like Information Expert, Single Responsibility, and Law of Demeter to decide
+  where logic belongs — don't just pick a convenient location.
+  - **Information Expert (GRASP):** Place logic in the module/class that already has the
+    knowledge it needs. If a method computes a result, it should signal or return it rather
+    than forcing another method to recompute the same thing.
   - **Law of Demeter:** Methods should depend only on the data they need, not on the
     objects that contain it. Pass the specific value (e.g., a `Future`) rather than the
     parent object (e.g., `TriggerData`) — this reduces coupling and removes the need for
     null-checks on intermediate accessors.
   - **Single Responsibility:** Each module, class, and function should have one clear reason
-    to change.
+    to change. Functions and types belong in the module that owns their domain (e.g.,
+    event-name helpers belong in `event.py`, not in `factory.py`).
   - **Interface Segregation:** Depend on narrow interfaces. If a helper only needs one field
     from a dataclass, accept that field directly.
 - **Decouple infrastructure from domain:** Modules like `signature.py` and `dispatcher.py` are

statemachine/engines/async_.py

@@ -402,6 +402,7 @@ async def processing_loop(  # noqa: C901
 
                 # Spawn invoke handlers for states entered during this macrostep.
                 await self._invoke_manager.spawn_pending_async()
+                self._check_root_final_state()
 
                 # Phase 2: remaining internal events
                 while not self.internal_queue.is_empty():  # pragma: no cover

statemachine/engines/base.py

@@ -99,6 +99,7 @@ def __init__(self, sm: "StateChart"):
         self._macrostep_count: int = 0
         self._microstep_count: int = 0
         self._log_id = f"[{type(sm).__name__}]"
+        self._root_parallel_final_pending: "State | None" = None
 
     def empty(self):  # pragma: no cover
         return self.external_queue.is_empty()
@@ -614,6 +615,8 @@ def _handle_final_state(self, target: State, on_entry_result: list):
                     BoundEvent(f"done.state.{grandparent.id}", _sm=self.sm, internal=True).put(
                         *donedata_args, **donedata_kwargs
                     )
+                    if grandparent.parent is None:
+                        self._root_parallel_final_pending = grandparent
 
     def _enter_states(  # noqa: C901
         self,
@@ -908,6 +911,29 @@ def add_ancestor_states_to_enter(
                             default_history_content,
                         )
 
+    def _check_root_final_state(self):
+        """SCXML spec: terminate when the root configuration is final.
+
+        For top-level parallel states, the machine terminates when all child
+        regions have reached their final states — equivalent to the SCXML
+        algorithm's ``isInFinalState(scxml_element)`` check.
+
+        Uses a flag set by ``_handle_final_state`` (Information Expert) to
+        avoid re-scanning top-level states on every macrostep.  The flag is
+        deferred because ``done.state`` events queued by ``_handle_final_state``
+        may trigger transitions that exit the parallel, so we verify the
+        parallel is still in the configuration before terminating.
+        """
+        state = self._root_parallel_final_pending
+        if state is None:
+            return
+        self._root_parallel_final_pending = None
+        # A done.state transition may have exited the parallel; verify it's
+        # still in the configuration before terminating.
+        if state in self.sm.configuration and self.is_in_final_state(state):
+            self._invoke_manager.cancel_all()
+            self.running = False
+
     def is_in_final_state(self, state: State) -> bool:
         if state.is_compound:
             return any(s.final and s in self.sm.configuration for s in state.states)

statemachine/engines/sync.py

@@ -118,6 +118,7 @@ def processing_loop(self, caller_future=None):  # noqa: C901
 
                 # Spawn invoke handlers for states entered during this macrostep.
                 self._invoke_manager.spawn_pending_sync()
+                self._check_root_final_state()
 
                 # Process remaining internal events before external events.
                 # Note: the macrostep loop above already drains the internal queue,

statemachine/event.py

@@ -16,6 +16,24 @@
     from .transition_list import TransitionList
 
 
+def _expand_event_id(key: str) -> str:
+    """Apply naming conventions for special event prefixes.
+
+    Converts underscore-based Python attribute names to their dot-separated
+    event equivalents. Returns a space-separated string so ``Events.add()``
+    registers both forms.
+    """
+    if key.startswith("done_invoke_"):
+        suffix = key[len("done_invoke_") :]
+        return f"{key} done.invoke.{suffix}"
+    if key.startswith("done_state_"):
+        suffix = key[len("done_state_") :]
+        return f"{key} done.state.{suffix}"
+    if key.startswith("error_"):
+        return f"{key} {key.replace('_', '.')}"
+    return key
+
+
 _event_data_kwargs = {
     "event_data",
     "machine",

statemachine/factory.py

@@ -9,6 +9,7 @@
 from .callbacks import CallbackPriority
 from .callbacks import CallbackSpecList
 from .event import Event
+from .event import _expand_event_id
 from .exceptions import InvalidDefinition
 from .graph import disconnected_states
 from .graph import iterate_states
@@ -271,29 +272,13 @@ def add_from_attributes(cls, attrs):  # noqa: C901
             if isinstance(value, State):
                 cls.add_state(key, value)
             elif isinstance(value, (Transition, TransitionList)):
-                event_id = key
-                if key.startswith("error_"):
-                    event_id = f"{key} {key.replace('_', '.')}"
-                elif key.startswith("done_invoke_"):
-                    suffix = key[len("done_invoke_") :]
-                    event_id = f"{key} done.invoke.{suffix}"
-                elif key.startswith("done_state_"):
-                    suffix = key[len("done_state_") :]
-                    event_id = f"{key} done.state.{suffix}"
+                event_id = _expand_event_id(key)
                 cls.add_event(event=Event(transitions=value, id=event_id, name=key))
             elif isinstance(value, (Event,)):
                 if value._has_real_id:
                     event_id = value.id
-                elif key.startswith("error_"):
-                    event_id = f"{key} {key.replace('_', '.')}"
-                elif key.startswith("done_invoke_"):
-                    suffix = key[len("done_invoke_") :]
-                    event_id = f"{key} done.invoke.{suffix}"
-                elif key.startswith("done_state_"):
-                    suffix = key[len("done_state_") :]
-                    event_id = f"{key} done.state.{suffix}"
                 else:
-                    event_id = key
+                    event_id = _expand_event_id(key)
                 new_event = Event(
                     transitions=value._transitions,
                     id=event_id,

statemachine/state.py

@@ -4,11 +4,13 @@
 from typing import Dict
 from typing import Generator
 from typing import List
+from typing import cast
 from weakref import ref
 
 from .callbacks import CallbackGroup
 from .callbacks import CallbackPriority
 from .callbacks import CallbackSpecList
+from .event import _expand_event_id
 from .exceptions import InvalidDefinition
 from .exceptions import StateMachineError
 from .i18n import _
@@ -32,8 +34,10 @@ def __call__(self, *states: "State", **kwargs):
 
 
 class _ToState(_TransitionBuilder):
-    def __call__(self, *states: "State | None", **kwargs):
-        transitions = TransitionList(Transition(self._state, state, **kwargs) for state in states)
+    def __call__(self, *states: "State | NestedStateFactory | None", **kwargs):
+        transitions = TransitionList(
+            Transition(self._state, cast("State | None", state), **kwargs) for state in states
+        )
         self._state.transitions.add_transitions(transitions)
         return transitions
 
@@ -43,11 +47,12 @@ def any(self, **kwargs):
         """Create transitions from all non-final states (reversed)."""
         return self.__call__(AnyState(), **kwargs)
 
-    def __call__(self, *states: "State", **kwargs):
+    def __call__(self, *states: "State | NestedStateFactory", **kwargs):
         transitions = TransitionList()
         for origin in states:
-            transition = Transition(origin, self._state, **kwargs)
-            origin.transitions.add_transitions(transition)
+            state = cast(State, origin)
+            transition = Transition(state, self._state, **kwargs)
+            state.transitions.add_transitions(transition)
             transitions.add_transitions(transition)
         return transitions
 
@@ -78,7 +83,7 @@ def __new__(  # type: ignore [misc]
                 value._set_id(key)
                 states.append(value)
             elif isinstance(value, TransitionList):
-                value.add_event(key)
+                value.add_event(_expand_event_id(key))
             elif callable(value):
                 callbacks[key] = value
 
@@ -87,15 +92,17 @@ def __new__(  # type: ignore [misc]
         )
 
     @classmethod
-    def to(cls, *args: "State", **kwargs) -> "_ToState":  # pragma: no cover
+    def to(cls, *args: "State | NestedStateFactory", **kwargs) -> "_ToState":  # pragma: no cover
         """Create transitions to the given target states.
         .. note: This method is only a type hint for mypy.
             The actual implementation belongs to the :ref:`State` class.
         """
         return _ToState(State())
 
     @classmethod
-    def from_(cls, *args: "State", **kwargs) -> "_FromState":  # pragma: no cover
+    def from_(  # pragma: no cover
+        cls, *args: "State | NestedStateFactory", **kwargs
+    ) -> "_FromState":
         """Create transitions from the given target states (reversed).
         .. note: This method is only a type hint for mypy.
             The actual implementation belongs to the :ref:`State` class.