refactor: normalize Configuration internals to always use OrderedSet

Introduce two boundary functions (_read_from_model / _write_to_model)
that confine the None|scalar|OrderedSet trichotomy to the model edge.
All other methods (add, discard, states setter, values) now operate on
a uniform OrderedSet, eliminating per-method type branching.

The model still receives the same denormalized values as before
(None for empty, scalar for single state, OrderedSet for multiple).

Fixes edge case: configuration_values now returns OrderedSet() for
uninitialized config instead of OrderedSet([None]), since None is not
a valid state value.

Prepares the codebase for the persistence protocol work (#597).

Signed-off-by: Fernando Macedo <fernando.macedo@jusbrasil.com.br>

Author: fgmacedo

docs/async.md

@@ -75,7 +75,7 @@ the initial state:
 ...     print(list(sm.configuration_values))
 
 >>> asyncio.run(show_problem())
-[None]
+[]
 
 ```
 

statemachine/configuration.py

@@ -55,76 +55,53 @@ def value(self) -> Any:
 
     @value.setter
     def value(self, val: Any):
-        self._invalidate()
-        if val is not None and not isinstance(val, MutableSet) and val not in self._states_map:
-            raise InvalidStateValue(val)
-        setattr(self._model, self._state_field, val)
+        if val is None:
+            self._write_to_model(OrderedSet())
+        elif isinstance(val, MutableSet):
+            self._write_to_model(OrderedSet(val) if not isinstance(val, OrderedSet) else val)
+        else:
+            self._write_to_model(OrderedSet([val]))
 
     @property
     def values(self) -> OrderedSet[Any]:
         """The set of raw state values currently active."""
-        v = self.value
-        if isinstance(v, OrderedSet):
-            return v
-        return OrderedSet([v])
+        return self._read_from_model()
 
     # -- Resolved states -------------------------------------------------------
 
     @property
     def states(self) -> "OrderedSet[State]":
         """The set of currently active :class:`State` instances (cached)."""
-        csv = self.value
-        if self._cached is not None and self._cached_value is csv:
+        raw = self.value
+        if self._cached is not None and self._cached_value is raw:
             return self._cached
-        if csv is None:
+        if raw is None:
             return OrderedSet()
 
-        instance_states = self._instance_states
-        if not isinstance(csv, MutableSet):
-            result = OrderedSet([instance_states[self._states_map[csv].id]])
-        else:
-            result = OrderedSet([instance_states[self._states_map[v].id] for v in csv])
-
+        # Normalize inline (avoid second getattr via _read_from_model)
+        values = raw if isinstance(raw, MutableSet) else (raw,)
+        result = OrderedSet(self._instance_states[self._states_map[v].id] for v in values)
         self._cached = result
-        self._cached_value = csv
+        self._cached_value = raw
         return result
 
     @states.setter
     def states(self, new_configuration: "OrderedSet[State]"):
-        if len(new_configuration) == 0:
-            self.value = None
-        elif len(new_configuration) == 1:
-            self.value = next(iter(new_configuration)).value
-        else:
-            self.value = OrderedSet(s.value for s in new_configuration)
+        self._write_to_model(OrderedSet(s.value for s in new_configuration))
 
     # -- Incremental mutation (used by the engine) -----------------------------
 
     def add(self, state: "State"):
-        """Add *state* to the configuration, maintaining the dual representation."""
-        csv = self.value
-        if csv is None:
-            self.value = state.value
-        elif isinstance(csv, MutableSet):
-            new = OrderedSet(csv)
-            new.add(state.value)
-            self.value = new
-        else:
-            self.value = OrderedSet([csv, state.value])
+        """Add *state* to the configuration."""
+        values = self._read_from_model()
+        values.add(state.value)
+        self._write_to_model(values)
 
     def discard(self, state: "State"):
-        """Remove *state* from the configuration, normalizing back to scalar."""
-        csv = self.value
-        if isinstance(csv, MutableSet):
-            new = OrderedSet(v for v in csv if v != state.value)
-            if len(new) == 0:
-                self.value = None
-            elif len(new) == 1:
-                self.value = next(iter(new))
-            else:
-                self.value = new
-        elif csv == state.value:
-            self.value = None
+        """Remove *state* from the configuration."""
+        values = self._read_from_model()
+        values.discard(state.value)
+        self._write_to_model(values)
 
     # -- Deprecated v2 compat --------------------------------------------------
 
@@ -154,7 +131,31 @@ def current_state(self) -> "State | OrderedSet[State]":
         except KeyError as err:
             raise InvalidStateValue(csv) from err
 
-    # -- Internal --------------------------------------------------------------
+    # -- Internal: model boundary ----------------------------------------------
+
+    def _read_from_model(self) -> OrderedSet:
+        """Normalize: model value → always ``OrderedSet``."""
+        raw = self.value
+        if raw is None:
+            return OrderedSet()
+        if isinstance(raw, OrderedSet):
+            return raw
+        if isinstance(raw, MutableSet):
+            return OrderedSet(raw)
+        return OrderedSet([raw])
+
+    def _write_to_model(self, values: OrderedSet):
+        """Denormalize: ``OrderedSet`` → ``None | scalar | OrderedSet`` for model."""
+        self._invalidate()
+        if len(values) == 0:
+            raw = None
+        elif len(values) == 1:
+            raw = next(iter(values))
+        else:
+            raw = values
+        if raw is not None and not isinstance(raw, MutableSet) and raw not in self._states_map:
+            raise InvalidStateValue(raw)
+        setattr(self._model, self._state_field, raw)
 
     def _invalidate(self):
         self._cached = None

tests/test_api_contract.py

@@ -243,7 +243,7 @@ async def test_set_none_clears_configuration(sm_runner):
 
     assert model.state is None
     assert sm.current_state_value is None
-    assert sm.configuration_values == OrderedSet([None])
+    assert sm.configuration_values == OrderedSet()
     assert sm.configuration == OrderedSet()
 
 
@@ -269,10 +269,10 @@ async def test_uninitialized_then_activated(sc_class, expected_ids):
     model = Model()
     sm = sc_class(model=model, listeners=[_AsyncListener()])
 
-    # Before activation: model.state is None, configuration_values wraps it
+    # Before activation: all APIs reflect empty configuration
     assert model.state is None
     assert sm.current_state_value is None
-    assert sm.configuration_values == OrderedSet([None])
+    assert sm.configuration_values == OrderedSet()
     assert sm.configuration == OrderedSet()
 
     # After activation: full contract holds

tests/test_configuration.py

@@ -41,6 +41,42 @@ def test_set_multi_element_configuration(self):
         assert sm.current_state_value == OrderedSet([ParallelSM.s1.value, ParallelSM.s2.value])
 
 
+class TestConfigurationValueSetter:
+    def test_set_value_none_writes_none_to_model(self):
+        sm = ParallelSM()
+        assert sm.current_state_value is not None
+
+        sm.current_state_value = None
+        assert sm.current_state_value is None
+        assert sm.configuration_values == OrderedSet()
+
+    def test_set_value_plain_set_coerces_to_ordered_set(self):
+        sm = ParallelSM()
+        s1_val = ParallelSM.s1.value
+        s2_val = ParallelSM.s2.value
+
+        # Assign a plain set (MutableSet but not OrderedSet)
+        sm.current_state_value = {s1_val, s2_val}
+        # Model should store an OrderedSet (denormalized back to it)
+        assert isinstance(sm.current_state_value, OrderedSet)
+        assert sm.current_state_value == OrderedSet([s1_val, s2_val])
+
+
+class TestReadFromModelNonOrderedSet:
+    def test_read_from_model_coerces_plain_set(self):
+        """When the model stores a plain set, _read_from_model coerces it."""
+        sm = ParallelSM()
+        s1_val = ParallelSM.s1.value
+        s2_val = ParallelSM.s2.value
+
+        # Bypass the value setter to place a plain set on the model
+        setattr(sm._config._model, sm._config._state_field, {s1_val, s2_val})
+
+        values = sm._config._read_from_model()
+        assert isinstance(values, OrderedSet)
+        assert values == OrderedSet([s1_val, s2_val])
+
+
 class TestConfigurationDiscard:
     def test_discard_nonmatching_scalar(self):
         sm = ParallelSM()