refactor: extract Configuration class to encapsulate state representation

Move the dual scalar/OrderedSet representation, caching, validation,
and incremental mutation logic from StateChart and BaseEngine into a
dedicated Configuration class (Information Expert pattern).

- StateChart properties (configuration, current_state_value, etc.) now
  delegate to self._config
- Engine add/remove methods reduced to one-line _config.add/discard calls
- Remove vestigial list handling in current_state (was a v3 dev artifact)
- Cache uses identity check against raw value to detect external bypasses

Author: fgmacedo

statemachine/configuration.py

@@ -0,0 +1,172 @@
+from typing import TYPE_CHECKING
+from typing import Any
+from typing import Dict
+from typing import MutableSet
+from weakref import ref
+
+from .exceptions import InvalidStateValue
+from .i18n import _
+from .orderedset import OrderedSet
+
+_SENTINEL = object()
+
+if TYPE_CHECKING:
+    from .state import State
+    from .statemachine import StateChart
+
+
+class Configuration:
+    """Encapsulates the dual representation of the active state configuration.
+
+    Internally, ``current_state_value`` is either a scalar (single active state)
+    or an ``OrderedSet`` (parallel regions).  This class hides that detail behind
+    a uniform interface for reading, mutating, and caching the resolved
+    ``OrderedSet[State]``.
+    """
+
+    __slots__ = (
+        "_machine_ref",
+        "_model",
+        "_state_field",
+        "_states_map",
+        "_for_instance",
+        "_cached",
+        "_cached_value",
+    )
+
+    def __init__(
+        self,
+        machine: "StateChart",
+        model: Any,
+        state_field: str,
+        states_map: "Dict[Any, State]",
+        for_instance_cache: "Dict[State, State]",
+    ):
+        self._machine_ref: "ref[StateChart]" = ref(machine)
+        self._model = model
+        self._state_field = state_field
+        self._states_map = states_map
+        self._for_instance = for_instance_cache
+        self._cached: "OrderedSet[State] | None" = None
+        self._cached_value: Any = _SENTINEL
+
+    # -- Raw value (persisted on the model) ------------------------------------
+
+    @property
+    def value(self) -> Any:
+        """The raw state value stored on the model (scalar or ``OrderedSet``)."""
+        return getattr(self._model, self._state_field, None)
+
+    @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)
+
+    @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])
+
+    # -- 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:
+            return self._cached
+        if csv is None:
+            return OrderedSet()
+
+        machine = self._machine_ref()
+        assert machine is not None
+
+        if not isinstance(csv, MutableSet):
+            result = OrderedSet(
+                [self._states_map[csv].for_instance(machine=machine, cache=self._for_instance)]
+            )
+        else:
+            result = OrderedSet(
+                [
+                    self._states_map[v].for_instance(machine=machine, cache=self._for_instance)
+                    for v in csv
+                ]
+            )
+
+        self._cached = result
+        self._cached_value = csv
+        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)
+
+    # -- 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):
+            csv.add(state.value)
+            self._invalidate()
+        else:
+            self.value = OrderedSet([csv, state.value])
+
+    def discard(self, state: "State"):
+        """Remove *state* from the configuration, normalizing back to scalar."""
+        csv = self.value
+        if isinstance(csv, MutableSet):
+            csv.discard(state.value)
+            self._invalidate()
+            if len(csv) == 1:
+                self.value = next(iter(csv))
+            elif len(csv) == 0:
+                self.value = None
+        elif csv == state.value:
+            self.value = None
+
+    # -- Deprecated v2 compat --------------------------------------------------
+
+    @property
+    def current_state(self) -> "State | OrderedSet[State]":
+        """Resolve the current state with validation.
+
+        Unlike ``states`` (which returns an empty set for ``None``), this
+        raises ``InvalidStateValue`` when the value is ``None`` or not
+        found in ``states_map`` — matching the v2 ``current_state`` contract.
+        """
+        csv = self.value
+        if csv is None:
+            raise InvalidStateValue(
+                csv,
+                _(
+                    "There's no current state set. In async code, "
+                    "did you activate the initial state? "
+                    "(e.g., `await sm.activate_initial_state()`)"
+                ),
+            )
+        try:
+            config = self.states
+            if len(config) == 1:
+                return next(iter(config))
+            return config
+        except KeyError as err:
+            raise InvalidStateValue(csv) from err
+
+    # -- Internal --------------------------------------------------------------
+
+    def _invalidate(self):
+        self._cached = None
+        self._cached_value = _SENTINEL

statemachine/engines/base.py

@@ -494,7 +494,7 @@ def _prepare_exit_states(
     def _remove_state_from_configuration(self, state: State):
         """Remove a state from the configuration if not using atomic updates."""
         if not self.sm.atomic_configuration_update:
-            self.sm.configuration -= {state}
+            self.sm._config.discard(state)
 
     def _exit_states(
         self, enabled_transitions: List[Transition], trigger_data: TriggerData
@@ -576,7 +576,7 @@ def _prepare_entry_states(
     def _add_state_to_configuration(self, target: State):
         """Add a state to the configuration if not using atomic updates."""
         if not self.sm.atomic_configuration_update:
-            self.sm.configuration |= {target}
+            self.sm._config.add(target)
 
     def __del__(self):
         try:

statemachine/state.py

@@ -298,7 +298,7 @@ def __str__(self):
     def __get__(self, machine, owner):
         if machine is None:
             return self
-        return self.for_instance(machine=machine, cache=machine._states_for_instance)
+        return self.for_instance(machine=machine, cache=machine._config._for_instance)
 
     def __set__(self, instance, value):
         raise StateMachineError(

statemachine/statemachine.py

@@ -16,6 +16,7 @@
 from .callbacks import CallbacksRegistry
 from .callbacks import SpecListGrouper
 from .callbacks import SpecReference
+from .configuration import Configuration
 from .dispatcher import Listener
 from .dispatcher import Listeners
 from .engines.async_ import AsyncEngine
@@ -150,7 +151,13 @@ def __init__(
             [start_value] if start_value is not None else list(self.start_configuration_values)
         )
         self._callbacks = CallbacksRegistry()
-        self._states_for_instance: Dict[State, State] = {}
+        self._config = Configuration(
+            machine=self,
+            model=self.model,
+            state_field=self.state_field,
+            states_map=self.states_map,
+            for_instance_cache={},
+        )
         self._listeners: Dict[int, Any] = {}
         """Listeners that provides attributes to be used as callbacks."""
 
@@ -215,15 +222,21 @@ def __repr__(self):
     def __getstate__(self):
         state = self.__dict__.copy()
         del state["_callbacks"]
-        del state["_states_for_instance"]
+        del state["_config"]
         del state["_engine"]
         return state
 
     def __setstate__(self, state: Dict[str, Any]) -> None:
         listeners = state.pop("_listeners")
         self.__dict__.update(state)  # type: ignore[attr-defined]
         self._callbacks = CallbacksRegistry()
-        self._states_for_instance = {}
+        self._config = Configuration(
+            machine=self,
+            model=self.model,
+            state_field=self.state_field,
+            states_map=self.states_map,
+            for_instance_cache={},
+        )
         self._listeners = {}
 
         # _listeners already contained both class-level and runtime listeners
@@ -335,44 +348,16 @@ def _graph(self):
     def configuration_values(self) -> OrderedSet[Any]:
         """The state configuration values is the set of currently active states's values
         (or ids if no custom value is defined)."""
-        if isinstance(self.current_state_value, OrderedSet):
-            return self.current_state_value
-        return OrderedSet([self.current_state_value])
+        return self._config.values
 
     @property
     def configuration(self) -> OrderedSet["State"]:
         """The set of currently active states."""
-        if self.current_state_value is None:
-            return OrderedSet()
-
-        if not isinstance(self.current_state_value, MutableSet):
-            return OrderedSet(
-                [
-                    self.states_map[self.current_state_value].for_instance(
-                        machine=self,
-                        cache=self._states_for_instance,
-                    )
-                ]
-            )
-
-        return OrderedSet(
-            [
-                self.states_map[value].for_instance(
-                    machine=self,
-                    cache=self._states_for_instance,
-                )
-                for value in self.current_state_value
-            ]
-        )
+        return self._config.states
 
     @configuration.setter
     def configuration(self, new_configuration: OrderedSet["State"]):
-        if len(new_configuration) == 0:
-            self.current_state_value = None
-        elif len(new_configuration) == 1:
-            self.current_state_value = new_configuration.pop().value
-        else:
-            self.current_state_value = OrderedSet(s.value for s in new_configuration)
+        self._config.states = new_configuration
 
     @property
     def current_state_value(self):
@@ -381,17 +366,11 @@ def current_state_value(self):
         This is a low level API, that can be used to assign any valid state value
         completely bypassing all the hooks and validations.
         """
-        return getattr(self.model, self.state_field, None)
+        return self._config.value
 
     @current_state_value.setter
     def current_state_value(self, value):
-        if (
-            value is not None
-            and not isinstance(value, MutableSet)
-            and value not in self.states_map
-        ):
-            raise InvalidStateValue(value)
-        setattr(self.model, self.state_field, value)
+        self._config.value = value
 
     @property
     def current_state(self) -> "State | MutableSet[State]":
@@ -405,36 +384,7 @@ def current_state(self) -> "State | MutableSet[State]":
             DeprecationWarning,
             stacklevel=2,
         )
-        current_value = self.current_state_value
-
-        try:
-            if isinstance(current_value, list):
-                return OrderedSet(
-                    [
-                        self.states_map[value].for_instance(
-                            machine=self,
-                            cache=self._states_for_instance,
-                        )
-                        for value in current_value
-                    ]
-                )
-
-            state: State = self.states_map[current_value].for_instance(
-                machine=self,
-                cache=self._states_for_instance,
-            )
-            return state
-        except KeyError as err:
-            if self.current_state_value is None:
-                raise InvalidStateValue(
-                    self.current_state_value,
-                    _(
-                        "There's no current state set. In async code, "
-                        "did you activate the initial state? "
-                        "(e.g., `await sm.activate_initial_state()`)"
-                    ),
-                ) from err
-            raise InvalidStateValue(self.current_state_value) from err
+        return self._config.current_state
 
     @current_state.setter
     def current_state(self, value):  # pragma: no cover