fix: run the effect of an explicit initial transition and register it as the only initial

docs/states.md

@@ -126,6 +126,54 @@ True
 Entering a compound activates both the parent and its `initial` child. You can query
 whether a state is compound using the `is_compound` property.
 
+### Explicit initial transition
+
+To enter a compound at a nested descendant, or to run an action each time the compound
+is entered by default, declare its initial transition explicitly with
+`parent.to(target, initial=True)`. The target can be any descendant, and the `on`
+action runs after the parent's `enter` callbacks and before any child is entered:
+
+```py
+>>> class Trip(StateChart):
+...     class travel(State.Compound):
+...         class planning(State.Compound):
+...             dates = State(initial=True)
+...             route = State()
+...             next_step = dates.to(route)
+...         booked = State()
+...         book = planning.to(booked)
+...         replan = booked.to(planning)
+...     travel.to(travel.planning.route, initial=True, on="load_saved_route")
+...     home = State()
+...     cancel = travel.to(home)
+...     restart = home.to(travel)
+...
+...     def __init__(self):
+...         self.log = []
+...         super().__init__()
+...
+...     def load_saved_route(self):
+...         self.log.append("load saved route")
+...
+...     def on_enter_state(self, state):
+...         self.log.append(f"enter {state.id}")
+
+>>> sm = Trip()
+>>> sm.log
+['enter travel', 'load saved route', 'enter planning', 'enter route']
+
+>>> sm.send("cancel")
+>>> sm.log.clear()
+>>> sm.send("restart")
+>>> sm.log
+['enter travel', 'load saved route', 'enter planning', 'enter route']
+
+```
+
+An explicit initial transition replaces the default one: the first child is no longer
+picked by document order. A compound state accepts a single initial transition, so
+declaring a second one raises `InvalidDefinition`.
+
 ```{seealso}
 See {ref}`done-state-events` for completion events when a compound state's
 final child is reached.

statemachine/engines/async_.py

@@ -277,7 +277,7 @@ async def _enter_states(
             initial_transition = initial_transitions.get(target.id)
             if initial_transition is not None:
                 result += await self.sm._callbacks.async_call(
-                    initial_transition.on.key, *args, **kwargs
+                    initial_transition.on.key, *args, on_error=on_error_transition, **kwargs
                 )
 
             # Handle default history states

statemachine/engines/base.py

@@ -651,7 +651,9 @@ def _enter_states(
 
             initial_transition = initial_transitions.get(target.id)
             if initial_transition is not None:
-                result += self.sm._callbacks.call(initial_transition.on.key, *args, **kwargs)
+                result += self.sm._callbacks.call(
+                    initial_transition.on.key, *args, on_error=on_error_transition, **kwargs
+                )
 
             # Handle default history states
             default_history_transitions = [

statemachine/factory.py

@@ -153,6 +153,15 @@ def _initials_by_document_order(  # noqa: C901
             if s.initial:
                 initials.append(s)
 
+        if parent and not parent.parallel:
+            explicit_initials = [t for t in parent.transitions if t.initial]
+            if len(explicit_initials) > 1:
+                raise InvalidDefinition(
+                    _("State '{0}' has more than one initial transition.").format(parent.id)
+                )
+            if explicit_initials:
+                return
+
         if not initials and states:
             initial = states[0]
             initial._initial = True
@@ -161,11 +170,6 @@ def _initials_by_document_order(  # noqa: C901
         if not parent:
             return
 
-        # If parent already has a multi-target initial transition (e.g., from SCXML initial
-        # attribute targeting multiple parallel regions), don't create default initial transitions.
-        if any(t for t in parent.transitions if t.initial and len(t.targets) > 1):
-            return
-
         for initial in initials:
             if not any(t for t in parent.transitions if t.initial and t.target == initial):
                 parent.to(initial, initial=True)

tests/io/test_native_behavior.py

@@ -62,3 +62,36 @@ async def test_safe_expression_guards(self, sm_runner):
         sm = await sm_runner.start(load(doc, format="yaml"))
         await sm_runner.send(sm, "go")
         assert "high" in sm.configuration_values
+
+    async def test_explicit_initial_transition_to_descendant_runs_its_effect(self, sm_runner):
+        doc = """
+        datamodel:
+          - {id: trace, expr: "''"}
+        states:
+          d:
+            enter:
+              - assign: {location: trace, expr: "trace + 'enter_d;'"}
+            transitions:
+              - target: d212
+                initial: true
+                on:
+                  - assign: {location: trace, expr: "trace + 'effect;'"}
+            states:
+              d1: {}
+              d2:
+                enter:
+                  - assign: {location: trace, expr: "trace + 'enter_d2;'"}
+                states:
+                  d21:
+                    states:
+                      d211: {initial: true}
+                      d212:
+                        enter:
+                          - assign: {location: trace, expr: "trace + 'enter_d212;'"}
+        """
+        chart = load(doc, format="yaml", validate=True)
+        sm = await sm_runner.start(chart)
+
+        assert [t.target.id for t in chart.d.transitions if t.initial] == ["d212"]
+        assert set(sm.configuration_values) == {"d", "d2", "d21", "d212"}
+        assert sm.model.trace == "enter_d;effect;enter_d2;enter_d212;"

tests/test_statechart_initial_transition.py

@@ -0,0 +1,223 @@
+"""Explicit initial transitions of compound states (refs #587)."""
+
+import pytest
+from statemachine.exceptions import InvalidDefinition
+
+from statemachine import HistoryState
+from statemachine import State
+from statemachine import StateChart
+
+
+class DeepInitial(StateChart):
+    class d(State.Compound):
+        class d1(State.Compound):
+            d11 = State(initial=True)
+
+        class d2(State.Compound):
+            class d21(State.Compound):
+                d211 = State(initial=True)
+                d212 = State()
+                x = d211.to(d212)
+
+        to_d2 = d1.to(d2)
+        to_d1 = d2.to(d1)
+
+    d.to(d.d2.d21.d212, initial=True, on="init_action")
+    out = State()
+    leave = d.to(out)
+    back = out.to(d)
+
+    def __init__(self, *args, **kwargs):
+        self.log = []
+        super().__init__(*args, **kwargs)
+
+    def init_action(self):
+        self.log.append("init_action")
+
+    def on_enter_state(self, state):
+        self.log.append(f"enter:{state.id}")
+
+
+class TestExplicitInitialTransition:
+    def test_is_the_only_initial_transition_of_the_compound(self):
+        initials = [t for t in DeepInitial.d.transitions if t.initial]
+
+        assert [t.target.id for t in initials] == ["d212"]
+
+    def test_no_child_is_marked_initial_by_document_order(self):
+        assert not DeepInitial.d.d1.initial
+
+    async def test_effect_runs_between_parent_entry_and_child_entries(self, sm_runner):
+        sm = await sm_runner.start(DeepInitial)
+
+        assert set(sm.configuration_values) == {"d", "d2", "d21", "d212"}
+        assert sm.log == ["enter:d", "init_action", "enter:d2", "enter:d21", "enter:d212"]
+
+    async def test_effect_runs_again_on_reentry(self, sm_runner):
+        sm = await sm_runner.start(DeepInitial)
+        await sm_runner.send(sm, "leave")
+        sm.log.clear()
+
+        await sm_runner.send(sm, "back")
+
+        assert set(sm.configuration_values) == {"d", "d2", "d21", "d212"}
+        assert sm.log == ["enter:d", "init_action", "enter:d2", "enter:d21", "enter:d212"]
+
+    async def test_effect_receives_event_arguments(self, sm_runner):
+        received = []
+
+        class Chart(StateChart):
+            class d(State.Compound):
+                a = State()
+                b = State()
+                back = b.to(a)
+
+            out = State(initial=True)
+            enter = out.to(d)
+            d.to(d.b, initial=True, on=lambda value: received.append(value))
+
+        sm = await sm_runner.start(Chart)
+        await sm_runner.send(sm, "enter", value=42)
+
+        assert set(sm.configuration_values) == {"d", "b"}
+        assert received == [42]
+
+    def test_subclass_keeps_a_single_initial_transition(self):
+        class Child(DeepInitial):
+            pass
+
+        sm = Child()
+
+        assert [t.target.id for t in Child.d.transitions if t.initial] == ["d212"]
+        assert sm.log == ["enter:d", "init_action", "enter:d2", "enter:d21", "enter:d212"]
+
+
+class TestConflictingInitialDefinitions:
+    def test_two_explicit_initial_transitions_are_rejected(self):
+        with pytest.raises(InvalidDefinition, match="d.*more than one initial transition"):
+
+            class Chart(StateChart):
+                class d(State.Compound):
+                    a = State()
+                    b = State()
+
+                d.to(d.a, initial=True)
+                d.to(d.b, initial=True)
+
+    def test_initial_child_targeted_by_explicit_initial_transition_is_accepted(self):
+        class Chart(StateChart):
+            class d(State.Compound):
+                a = State()
+                b = State(initial=True)
+                back = b.to(a)
+
+            d.to(d.b, initial=True, on="effect")
+
+            def effect(self):
+                self.ran = True
+
+        sm = Chart()
+
+        assert [t.target.id for t in Chart.d.transitions if t.initial] == ["b"]
+        assert set(sm.configuration_values) == {"d", "b"}
+        assert sm.ran
+
+    def test_initial_child_containing_the_explicit_target_is_accepted(self):
+        class Chart(StateChart):
+            class d(State.Compound):
+                a = State()
+
+                class b(State.Compound, initial=True):
+                    b1 = State(initial=True)
+                    b2 = State()
+                    step = b2.to(b1)
+
+                back = b.to(a)
+
+            d.to(d.b.b2, initial=True)
+
+        sm = Chart()
+
+        assert set(sm.configuration_values) == {"d", "b", "b2"}
+
+
+class TestInitialTransitionEffectErrors:
+    async def test_error_in_effect_becomes_error_execution_and_entry_completes(self, sm_runner):
+        class Chart(StateChart):
+            class d(State.Compound):
+                a = State()
+                b = State()
+                back = b.to(a)
+
+            out = State(initial=True)
+            go = out.to(d)
+            d.to(d.b, initial=True, on="effect")
+
+            def effect(self):
+                raise ValueError("effect failed")
+
+        sm = await sm_runner.start(Chart)
+        await sm_runner.send(sm, "go")
+
+        assert set(sm.configuration_values) == {"d", "b"}
+
+    async def test_error_execution_handler_sees_the_effect_error(self, sm_runner):
+        class Chart(StateChart):
+            class d(State.Compound):
+                a = State()
+                b = State()
+                back = b.to(a)
+
+            out = State(initial=True)
+            go = out.to(d)
+            d.to(d.b, initial=True, on="effect")
+            error_execution = d.to(out, on="handle")
+
+            def __init__(self, *args, **kwargs):
+                self.errors = []
+                super().__init__(*args, **kwargs)
+
+            def effect(self):
+                raise ValueError("effect failed")
+
+            def handle(self, error):
+                self.errors.append(str(error))
+
+        sm = await sm_runner.start(Chart)
+        await sm_runner.send(sm, "go")
+
+        assert sm.errors == ["effect failed"]
+        assert set(sm.configuration_values) == {"out"}
+
+
+class TestInitialTransitionToHistory:
+    async def test_effect_runs_once_per_entry(self, sm_runner):
+        class Chart(StateChart):
+            class d(State.Compound):
+                h = HistoryState()
+                a = State()
+                b = State()
+                step = a.to(b)
+                h.to(a)
+
+            d.to(d.h, initial=True, on="effect")
+            out = State()
+            leave = d.to(out)
+            back = out.to(d)
+
+            def __init__(self, *args, **kwargs):
+                self.effects = 0
+                super().__init__(*args, **kwargs)
+
+            def effect(self):
+                self.effects += 1
+
+        sm = await sm_runner.start(Chart)
+        assert sm.effects == 1
+
+        await sm_runner.send(sm, "step")
+        await sm_runner.send(sm, "leave")
+        await sm_runner.send(sm, "back")
+
+        assert sm.effects == 2
+        assert set(sm.configuration_values) == {"d", "b"}