feat: add enabled_events() method to check guard conditions (#520)

`allowed_events` returns events reachable from the current state but does
not evaluate `cond`/`unless` guards. The new `enabled_events()` method
evaluates conditions and returns only events that can actually fire.

It accepts `*args`/`**kwargs` forwarded to condition callbacks, works with
both sync and async engines, and treats condition exceptions as enabled
(permissive behavior).

Closes #520

Author: fgmacedo

docs/guards.md

@@ -47,7 +47,7 @@ To control the evaluation order, declare transitions in the desired order:
 ```python
 # Declare in the order you want them checked:
 first = state_a.to(state_b, cond="check1")   # Checked FIRST
-second = state_a.to(state_c, cond="check2")  # Checked SECOND  
+second = state_a.to(state_c, cond="check2")  # Checked SECOND
 third = state_a.to(state_d, cond="check3")   # Checked THIRD
 
 my_event = first | second | third  # Order matches declaration
@@ -159,6 +159,79 @@ So, a condition `s1.to(s2, cond=lambda: [])` will evaluate as `False`, as an emp
 **falsy** value.
 ```
 
+### Checking enabled events
+
+The {ref}`StateMachine.allowed_events` property returns events reachable from the current state,
+but it does **not** evaluate `cond`/`unless` guards. To check which events actually have their
+conditions satisfied, use {ref}`StateMachine.enabled_events`.
+
+```{testsetup}
+
+>>> from statemachine import StateMachine, State
+
+```
+
+```py
+>>> class ApprovalMachine(StateMachine):
+...     pending = State(initial=True)
+...     approved = State(final=True)
+...     rejected = State(final=True)
+...
+...     approve = pending.to(approved, cond="is_manager")
+...     reject = pending.to(rejected)
+...
+...     is_manager = False
+
+>>> sm = ApprovalMachine()
+
+>>> [e.id for e in sm.allowed_events]
+['approve', 'reject']
+
+>>> [e.id for e in sm.enabled_events()]
+['reject']
+
+>>> sm.is_manager = True
+
+>>> [e.id for e in sm.enabled_events()]
+['approve', 'reject']
+
+```
+
+`enabled_events` is a method (not a property) because conditions may depend on runtime
+arguments. Any `*args`/`**kwargs` passed to `enabled_events()` are forwarded to the
+condition callbacks, just like when triggering an event:
+
+```py
+>>> class TaskMachine(StateMachine):
+...     idle = State(initial=True)
+...     running = State(final=True)
+...
+...     start = idle.to(running, cond="has_enough_resources")
+...
+...     def has_enough_resources(self, cpu=0):
+...         return cpu >= 4
+
+>>> sm = TaskMachine()
+
+>>> sm.enabled_events()
+[]
+
+>>> [e.id for e in sm.enabled_events(cpu=8)]
+['start']
+
+```
+
+```{tip}
+This is useful for UI scenarios where you want to show or hide buttons based on whether
+an event's conditions are currently satisfied.
+```
+
+```{note}
+An event is considered **enabled** if at least one of its transitions from the current state
+has all conditions satisfied. If a condition raises an exception, the event is treated as
+enabled (permissive behavior).
+```
+
 ## Validators
 
 

statemachine/engines/async_.py

@@ -97,6 +97,34 @@ async def _trigger(self, trigger_data: TriggerData):
 
         return result if executed else None
 
+    async def enabled_events(self, *args, **kwargs):
+        sm = self.sm
+        enabled = {}
+        for transition in sm.current_state.transitions:
+            for event in transition.events:
+                if event in enabled:
+                    continue
+                extended_kwargs = kwargs.copy()
+                extended_kwargs.update(
+                    {
+                        "machine": sm,
+                        "model": sm.model,
+                        "event": getattr(sm, event),
+                        "source": transition.source,
+                        "target": transition.target,
+                        "state": sm.current_state,
+                        "transition": transition,
+                    }
+                )
+                try:
+                    if await sm._callbacks.async_all(
+                        transition.cond.key, *args, **extended_kwargs
+                    ):
+                        enabled[event] = getattr(sm, event)
+                except Exception:
+                    enabled[event] = getattr(sm, event)
+        return list(enabled.values())
+
     async def _activate(self, trigger_data: TriggerData, transition: "Transition"):
         event_data = EventData(trigger_data=trigger_data, transition=transition)
         args, kwargs = event_data.args, event_data.extended_kwargs

statemachine/engines/sync.py

@@ -99,6 +99,32 @@ def _trigger(self, trigger_data: TriggerData):
 
         return result if executed else None
 
+    def enabled_events(self, *args, **kwargs):
+        sm = self.sm
+        enabled = {}
+        for transition in sm.current_state.transitions:
+            for event in transition.events:
+                if event in enabled:
+                    continue
+                extended_kwargs = kwargs.copy()
+                extended_kwargs.update(
+                    {
+                        "machine": sm,
+                        "model": sm.model,
+                        "event": getattr(sm, event),
+                        "source": transition.source,
+                        "target": transition.target,
+                        "state": sm.current_state,
+                        "transition": transition,
+                    }
+                )
+                try:
+                    if sm._callbacks.all(transition.cond.key, *args, **extended_kwargs):
+                        enabled[event] = getattr(sm, event)
+                except Exception:
+                    enabled[event] = getattr(sm, event)
+        return list(enabled.values())
+
     def _activate(self, trigger_data: TriggerData, transition: "Transition"):
         event_data = EventData(trigger_data=trigger_data, transition=transition)
         args, kwargs = event_data.args, event_data.extended_kwargs

statemachine/statemachine.py

@@ -295,6 +295,24 @@ def allowed_events(self) -> "List[Event]":
         """List of the current allowed events."""
         return [getattr(self, event) for event in self.current_state.transitions.unique_events]
 
+    def enabled_events(self, *args, **kwargs):
+        """List of the current enabled events, considering guard conditions.
+
+        An event is **enabled** if at least one of its transitions from the current
+        state has all ``cond``/``unless`` guards satisfied.
+
+        Args:
+            *args: Positional arguments forwarded to condition callbacks.
+            **kwargs: Keyword arguments forwarded to condition callbacks.
+
+        Returns:
+            A list of enabled :ref:`Event` instances.
+        """
+        result = self._engine.enabled_events(*args, **kwargs)
+        if not isawaitable(result):
+            return result
+        return run_async_from_sync(result)
+
     def _put_nonblocking(self, trigger_data: TriggerData):
         """Put the trigger on the queue without blocking the caller."""
         self._engine.put(trigger_data)

tests/test_async.py

@@ -199,3 +199,81 @@ async def test_async_state_should_be_initialized(async_order_control_machine):
 
     await sm.activate_initial_state()
     assert sm.current_state == sm.waiting_for_payment
+
+
+class TestAsyncEnabledEvents:
+    async def test_passing_async_condition(self):
+        class MyMachine(StateMachine):
+            s0 = State(initial=True)
+            s1 = State(final=True)
+
+            go = s0.to(s1, cond="is_ready")
+
+            async def is_ready(self):
+                return True
+
+        sm = MyMachine()
+        await sm.activate_initial_state()
+        assert [e.id for e in await sm.enabled_events()] == ["go"]
+
+    async def test_failing_async_condition(self):
+        class MyMachine(StateMachine):
+            s0 = State(initial=True)
+            s1 = State(final=True)
+
+            go = s0.to(s1, cond="is_ready")
+
+            async def is_ready(self):
+                return False
+
+        sm = MyMachine()
+        await sm.activate_initial_state()
+        assert await sm.enabled_events() == []
+
+    async def test_kwargs_forwarded_to_async_conditions(self):
+        class MyMachine(StateMachine):
+            s0 = State(initial=True)
+            s1 = State(final=True)
+
+            go = s0.to(s1, cond="check_value")
+
+            async def check_value(self, value=0):
+                return value > 10
+
+        sm = MyMachine()
+        await sm.activate_initial_state()
+        assert await sm.enabled_events() == []
+        assert [e.id for e in await sm.enabled_events(value=20)] == ["go"]
+
+    async def test_async_condition_exception_treated_as_enabled(self):
+        class MyMachine(StateMachine):
+            s0 = State(initial=True)
+            s1 = State(final=True)
+
+            go = s0.to(s1, cond="bad_cond")
+
+            async def bad_cond(self):
+                raise RuntimeError("boom")
+
+        sm = MyMachine()
+        await sm.activate_initial_state()
+        assert [e.id for e in await sm.enabled_events()] == ["go"]
+
+    async def test_mixed_enabled_and_disabled_async(self):
+        class MyMachine(StateMachine):
+            s0 = State(initial=True)
+            s1 = State()
+            s2 = State(final=True)
+
+            go = s0.to(s1, cond="cond_true")
+            stop = s0.to(s2, cond="cond_false")
+
+            async def cond_true(self):
+                return True
+
+            async def cond_false(self):
+                return False
+
+        sm = MyMachine()
+        await sm.activate_initial_state()
+        assert [e.id for e in await sm.enabled_events()] == ["go"]

tests/test_statemachine.py

@@ -503,3 +503,134 @@ def __bool__(self):
 
     machine.produce()
     assert model.state == "producing"
+
+
+class TestEnabledEvents:
+    def test_no_conditions_same_as_allowed_events(self, campaign_machine):
+        """Without conditions, enabled_events should match allowed_events."""
+        sm = campaign_machine()
+        assert [e.id for e in sm.enabled_events()] == [e.id for e in sm.allowed_events]
+
+    def test_passing_condition_returns_event(self):
+        class MyMachine(StateMachine):
+            s0 = State(initial=True)
+            s1 = State(final=True)
+
+            go = s0.to(s1, cond="is_ready")
+
+            def is_ready(self):
+                return True
+
+        sm = MyMachine()
+        assert [e.id for e in sm.enabled_events()] == ["go"]
+
+    def test_failing_condition_excludes_event(self):
+        class MyMachine(StateMachine):
+            s0 = State(initial=True)
+            s1 = State(final=True)
+
+            go = s0.to(s1, cond="is_ready")
+
+            def is_ready(self):
+                return False
+
+        sm = MyMachine()
+        assert sm.enabled_events() == []
+
+    def test_multiple_transitions_one_passes(self):
+        """Same event with multiple transitions: included if at least one passes."""
+
+        class MyMachine(StateMachine):
+            s0 = State(initial=True)
+            s1 = State()
+            s2 = State(final=True)
+
+            go = s0.to(s1, cond="cond_false") | s0.to(s2, cond="cond_true")
+
+            def cond_false(self):
+                return False
+
+            def cond_true(self):
+                return True
+
+        sm = MyMachine()
+        assert [e.id for e in sm.enabled_events()] == ["go"]
+
+    def test_final_state_returns_empty(self, campaign_machine):
+        sm = campaign_machine()
+        sm.produce()
+        sm.deliver()
+        assert sm.enabled_events() == []
+
+    def test_kwargs_forwarded_to_conditions(self):
+        class MyMachine(StateMachine):
+            s0 = State(initial=True)
+            s1 = State(final=True)
+
+            go = s0.to(s1, cond="check_value")
+
+            def check_value(self, value=0):
+                return value > 10
+
+        sm = MyMachine()
+        assert sm.enabled_events() == []
+        assert [e.id for e in sm.enabled_events(value=20)] == ["go"]
+
+    def test_condition_exception_treated_as_enabled(self):
+        """If a condition raises, the event is treated as enabled (permissive)."""
+
+        class MyMachine(StateMachine):
+            s0 = State(initial=True)
+            s1 = State(final=True)
+
+            go = s0.to(s1, cond="bad_cond")
+
+            def bad_cond(self):
+                raise RuntimeError("boom")
+
+        sm = MyMachine()
+        assert [e.id for e in sm.enabled_events()] == ["go"]
+
+    def test_mixed_enabled_and_disabled(self):
+        class MyMachine(StateMachine):
+            s0 = State(initial=True)
+            s1 = State()
+            s2 = State(final=True)
+
+            go = s0.to(s1, cond="cond_true")
+            stop = s0.to(s2, cond="cond_false")
+
+            def cond_true(self):
+                return True
+
+            def cond_false(self):
+                return False
+
+        sm = MyMachine()
+        assert [e.id for e in sm.enabled_events()] == ["go"]
+
+    def test_unless_condition(self):
+        class MyMachine(StateMachine):
+            s0 = State(initial=True)
+            s1 = State(final=True)
+
+            go = s0.to(s1, unless="is_blocked")
+
+            def is_blocked(self):
+                return True
+
+        sm = MyMachine()
+        assert sm.enabled_events() == []
+
+    def test_unless_condition_passes(self):
+        class MyMachine(StateMachine):
+            s0 = State(initial=True)
+            s1 = State(final=True)
+
+            go = s0.to(s1, unless="is_blocked")
+
+            def is_blocked(self):
+                return False
+
+        sm = MyMachine()
+        assert [e.id for e in sm.enabled_events()] == ["go"]