From 6d63aedc75545119c73eb20fe54db54b54ffc80f Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Wed, 14 Oct 2026 21:51:57 -0300 Subject: [PATCH 1/2] fix: enter history states by default without running the incoming effect twice A history state with no stored value and no default transition left the machine with an empty configuration; it now enters its parent by default. Entering through history no longer runs the incoming transition's actions a second time: only the history's own default transition is default history content, and only when nothing is stored. Refs #692 Signed-off-by: Fernando Macedo --- docs/states.md | 14 +++ statemachine/engines/async_.py | 13 ++- statemachine/engines/base.py | 95 ++++++++++--------- tests/test_statechart_history.py | 156 +++++++++++++++++++++++++++++++ 4 files changed, 228 insertions(+), 50 deletions(-) diff --git a/docs/states.md b/docs/states.md index 0a01bc5e..a7fbb409 100644 --- a/docs/states.md +++ b/docs/states.md @@ -209,6 +209,20 @@ True Use `HistoryState(type="deep")` for deep history that remembers the exact leaf state in nested compounds. +On the first visit there is nothing recorded yet. A transition declared from the history +state (`h.to(b)`) is its default: its target is entered and its actions run. Without one, +the compound state is entered as usual, through its initial state. Either way, the actions +of the transition that led to the history state run exactly once. + +```py +>>> sm = WithHistory() +>>> sm.send("leave") +>>> sm.send("resume") +>>> sorted(sm.configuration_values) +['a', 'mode'] + +``` + ```{seealso} See {ref}`querying-configuration` for how to inspect which states are currently diff --git a/statemachine/engines/async_.py b/statemachine/engines/async_.py index 81675b03..ae1955b0 100644 --- a/statemachine/engines/async_.py +++ b/statemachine/engines/async_.py @@ -257,6 +257,7 @@ async def _enter_states( if self.sm.atomic_configuration_update: self.sm.configuration = new_configuration + initial_transitions = {i.state.id: i.transition for i in states_for_default_entry} for info in ordered_states: target = info.state transition = info.transition @@ -273,13 +274,11 @@ async def _enter_states( target.enter.key, *args, on_error=on_error, **kwargs ) - # Handle default initial states - if target.id in {t.state.id for t in states_for_default_entry if t.state}: - initial_transitions = [t for t in target.transitions if t.initial] - if len(initial_transitions) == 1: - result += await self.sm._callbacks.async_call( - initial_transitions[0].on.key, *args, **kwargs - ) + 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 + ) # Handle default history states default_history_transitions = [ diff --git a/statemachine/engines/base.py b/statemachine/engines/base.py index 60b5921a..c457957f 100644 --- a/statemachine/engines/base.py +++ b/statemachine/engines/base.py @@ -387,12 +387,14 @@ def get_effective_target_states(self, transition: Transition) -> OrderedSet[Stat if state.is_history: if state.id in self.sm.history_values: targets.update(self.sm.history_values[state.id]) - else: + elif state.transitions: targets.update( state for t in state.transitions for state in self.get_effective_target_states(t) ) + else: + targets.add(cast(State, state.parent)) else: targets.add(state) @@ -628,6 +630,7 @@ def _enter_states( if self.sm.atomic_configuration_update: self.sm.configuration = new_configuration + initial_transitions = {i.state.id: i.transition for i in states_for_default_entry} for info in ordered_states: target = info.state transition = info.transition @@ -646,13 +649,9 @@ def _enter_states( target.enter.key, *args, on_error=on_error, **kwargs ) - # Handle default initial states - if target.id in {t.state.id for t in states_for_default_entry if t.state}: - initial_transitions = [t for t in target.transitions if t.initial] - if len(initial_transitions) == 1: - result += self.sm._callbacks.call( - initial_transitions[0].on.key, *args, **kwargs - ) + initial_transition = initial_transitions.get(target.id) + if initial_transition is not None: + result += self.sm._callbacks.call(initial_transition.on.key, *args, **kwargs) # Handle default history states default_history_transitions = [ @@ -778,20 +777,26 @@ def _add_history_states_to_enter( states_for_default_entry, default_history_content, ): - """Add the states a history state stands for: those it recorded, or its default.""" - parent_id = state.parent and state.parent.id - default_history_content[parent_id] = [info] + """Add the states a history state stands for. + + Those it recorded; else its default transition's targets, whose content runs on entry + of the parent; else the parent's own default entry. + """ + parent = cast(State, state.parent) if state.id in self.sm.history_values: self._debug( "%s History state '%s.%s' %s restoring: '%s'", self._log_id, - state.parent, + parent, state, state.type.value, [s.id for s in self.sm.history_values[state.id]], ) - for history_state in self.sm.history_values[state.id]: - info_to_add = StateTransition(transition=info.transition, state=history_state) + infos = [ + StateTransition(transition=info.transition, state=history_state) + for history_state in self.sm.history_values[state.id] + ] + for info_to_add in infos: if state.type.is_deep: states_to_enter.add(info_to_add) else: @@ -801,46 +806,50 @@ def _add_history_states_to_enter( states_for_default_entry, default_history_content, ) - for history_state in self.sm.history_values[state.id]: - info_to_add = StateTransition(transition=info.transition, state=history_state) - self.add_ancestor_states_to_enter( - info_to_add, - state.parent, - states_to_enter, - states_for_default_entry, - default_history_content, - ) - else: - # Handle default history content + elif state.transitions: self._debug( "%s History state '%s.%s' default content: %s", self._log_id, - state.parent, + parent, state, [t.target.id for t in state.transitions if t.target], ) - - for transition in state.transitions: - target = cast(State, transition.target) - info_history = StateTransition(transition=transition, state=target) - default_history_content[parent_id].append(info_history) + infos = [ + StateTransition(transition=transition, state=cast(State, transition.target)) + for transition in state.transitions + ] + default_history_content[parent.id] = infos + for info_to_add in infos: self.add_descendant_states_to_enter( - info_history, + info_to_add, states_to_enter, states_for_default_entry, default_history_content, ) - for transition in state.transitions: - target = cast(State, transition.target) - info_history = StateTransition(transition=transition, state=target) + else: + self._debug( + "%s History state '%s.%s' has no content, default entry of '%s'", + self._log_id, + parent, + state, + parent, + ) + self.add_descendant_states_to_enter( + StateTransition(transition=info.transition, state=parent), + states_to_enter, + states_for_default_entry, + default_history_content, + ) + return - self.add_ancestor_states_to_enter( - info_history, - state.parent, - states_to_enter, - states_for_default_entry, - default_history_content, - ) + for info_to_add in infos: + self.add_ancestor_states_to_enter( + info_to_add, + parent, + states_to_enter, + states_for_default_entry, + default_history_content, + ) def _add_regions_to_enter( self, @@ -869,8 +878,8 @@ def _add_initial_states_to_enter( ): """Add the states a compound state's initial transition targets.""" state = info.state - states_for_default_entry.add(info) transition = next(t for t in state.transitions if t.initial) + states_for_default_entry.add(StateTransition(transition=transition, state=state)) # Process all targets (supports multi-target initial transitions for parallel regions) for initial_target in transition.targets: info_initial = StateTransition(transition=transition, state=initial_target) diff --git a/tests/test_statechart_history.py b/tests/test_statechart_history.py index 2e1d1923..51210cec 100644 --- a/tests/test_statechart_history.py +++ b/tests/test_statechart_history.py @@ -9,6 +9,9 @@ import pytest +from statemachine import HistoryState +from statemachine import State +from statemachine import StateChart from tests.machines.history.deep_memory_of_moria import DeepMemoryOfMoria from tests.machines.history.gollum_personality import GollumPersonality from tests.machines.history.gollum_personality_default_gollum import GollumPersonalityDefaultGollum @@ -106,3 +109,156 @@ async def test_history_with_default_transition(self, sm_runner): sm = await sm_runner.start(GollumPersonalityDefaultGollum) await sm_runner.send(sm, "enter_via_history") assert "gollum" in sm.configuration_values + + +def _build_traced_machine(kind, with_default=False): + class Traced(StateChart): + outside = State(initial=True) + + class region(State.Compound): + class nested(State.Compound): + n1 = State(initial=True) + n2 = State() + deeper = n1.to(n2) + + other = State() + h = HistoryState(type=kind) + sideways = nested.to(other) + if with_default: + h.to(other, on="default_effect") + + enter = outside.to(region, on="enter_effect") + leave = region.to(outside) + back = outside.to(region.h, on="back_effect") + + def __init__(self, *args, **kwargs): + self.trace = [] + super().__init__(*args, **kwargs) + + def enter_effect(self): + self.trace.append("enter_effect") + + def back_effect(self): + self.trace.append("back_effect") + + def default_effect(self): + self.trace.append("default_effect") + + def on_enter_state(self, target): + self.trace.append(f"enter:{target.id}") + + def on_exit_state(self, source): + self.trace.append(f"exit:{source.id}") + + return Traced + + +@pytest.mark.timeout(5) +@pytest.mark.parametrize("kind", ["shallow", "deep"]) +class TestHistoryDefaultEntry: + async def test_no_stored_value_and_no_default_enters_parent_by_default(self, sm_runner, kind): + sm = await sm_runner.start(_build_traced_machine(kind)) + sm.trace.clear() + + await sm_runner.send(sm, "back") + + assert {"region", "nested", "n1"} == set(sm.configuration_values) + assert sm.trace == [ + "exit:outside", + "back_effect", + "enter:region", + "enter:nested", + "enter:n1", + ] + + async def test_machine_keeps_running_after_default_entry_of_parent(self, sm_runner, kind): + sm = await sm_runner.start(_build_traced_machine(kind)) + await sm_runner.send(sm, "back") + await sm_runner.send(sm, "deeper") + await sm_runner.send(sm, "leave") + await sm_runner.send(sm, "back") + + expected_leaf = "n2" if kind == "deep" else "n1" + assert {"region", "nested", expected_leaf} == set(sm.configuration_values) + + async def test_stored_value_runs_incoming_effect_once(self, sm_runner, kind): + sm = await sm_runner.start(_build_traced_machine(kind)) + await sm_runner.send(sm, "enter") + await sm_runner.send(sm, "sideways") + await sm_runner.send(sm, "leave") + sm.trace.clear() + + await sm_runner.send(sm, "back") + + assert {"region", "other"} == set(sm.configuration_values) + assert sm.trace == ["exit:outside", "back_effect", "enter:region", "enter:other"] + + async def test_default_runs_incoming_effect_once_then_default_effect(self, sm_runner, kind): + sm = await sm_runner.start(_build_traced_machine(kind, with_default=True)) + sm.trace.clear() + + await sm_runner.send(sm, "back") + + assert {"region", "other"} == set(sm.configuration_values) + assert sm.trace == [ + "exit:outside", + "back_effect", + "enter:region", + "default_effect", + "enter:other", + ] + + async def test_stored_value_skips_default_effect(self, sm_runner, kind): + sm = await sm_runner.start(_build_traced_machine(kind, with_default=True)) + await sm_runner.send(sm, "enter") + await sm_runner.send(sm, "deeper") + await sm_runner.send(sm, "leave") + sm.trace.clear() + + await sm_runner.send(sm, "back") + + expected_leaf = "n2" if kind == "deep" else "n1" + assert {"region", "nested", expected_leaf} == set(sm.configuration_values) + assert sm.trace == [ + "exit:outside", + "back_effect", + "enter:region", + "enter:nested", + f"enter:{expected_leaf}", + ] + + async def test_default_entry_of_parent_enters_its_ancestors(self, sm_runner, kind): + class Nested(StateChart): + outside = State(initial=True) + + class top(State.Compound): + class region(State.Compound): + r1 = State(initial=True) + h = HistoryState(type=kind) + + t1 = State(initial=True) + + back = outside.to(top.region.h) + + sm = await sm_runner.start(Nested) + + await sm_runner.send(sm, "back") + + assert {"top", "region", "r1"} == set(sm.configuration_values) + + async def test_default_entry_of_parent_from_a_sibling(self, sm_runner, kind): + class Siblings(StateChart): + class top(State.Compound): + t1 = State(initial=True) + + class region(State.Compound): + r1 = State(initial=True) + h = HistoryState(type=kind) + + back = t1.to(region.h) + + sm = await sm_runner.start(Siblings) + + await sm_runner.send(sm, "back") + + assert {"top", "region", "r1"} == set(sm.configuration_values) From 71f1047e204aa71fbc55b9d0f6507647be380549 Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Wed, 14 Oct 2026 21:51:58 -0300 Subject: [PATCH 2/2] test: run the History PSSM cases fixed by the history default entry History_001-A, History_002-A and History_002-C pass on both engines now. History_002-B no longer runs the incoming effect twice; it still fails on the interleaving of two regions' completions, so only that blocker stays. Signed-off-by: Fernando Macedo --- tests/uml/pssm/conftest.py | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/tests/uml/pssm/conftest.py b/tests/uml/pssm/conftest.py index 90bba79f..8238605c 100644 --- a/tests/uml/pssm/conftest.py +++ b/tests/uml/pssm/conftest.py @@ -30,13 +30,7 @@ "Exit_003": "requires:exit_point: exit point effect runs before the state exit", "Exiting_004": "requires:exit_point: exit point effect runs before the state exit", "Fork_002": "requires:fork: fork branch effects interleave with entries", - "History_001-A": "bug:B2: effect into a restored history runs twice", - "History_002-A": "bug:B2: effect into a restored history runs twice", - "History_002-B": ( - "bug:B2, unclassified: effect into a restored history runs twice; completions of " - "two regions share one microstep" - ), - "History_002-C": "bug:B1: history without stored value nor default empties the config", + "History_002-B": "unclassified: completions of two regions share one microstep", "History_002-D": "conflict:K10: history restores a final state, completion loops", "Join_001": "requires:join: join effects run after all exits", "Join_002": "requires:join: join effects run after all exits",