feat: future-based result routing for async concurrent events (#567)

* feat: add future-based result routing for async concurrent events (#509)

When multiple coroutines send events concurrently, only the lock holder
processes events. Previously, non-lock-holding callers returned None,
losing both results and exceptions.

Now each external event gets an asyncio.Future attached in AsyncEngine.put().
The processing loop resolves/rejects each future with the microstep result.
Callers that couldn't acquire the lock await their future instead of
returning None.

Uses contextvars.ContextVar to distinguish reentrant calls (from within
callbacks, including asyncio.gather child tasks) from concurrent external
calls — reentrant calls don't get futures to avoid deadlocks.

* test: add regression test for issue #509 example

* refactor: move reject_futures to EventQueue, fix test latency and doctest

- Move future rejection logic from AsyncEngine into EventQueue.reject_futures()
  to respect encapsulation (avoids reaching into PriorityQueue internals)
- Reduce asyncio.sleep in test_issue509 from 0.1s to 0.01s
- Convert plain python block in docs/async.md to a testable doctest

* test: cover EventQueue.reject_futures with future=None items

Author: fgmacedo

AGENTS.md

@@ -146,6 +146,15 @@ uv run mypy statemachine/ tests/
 
 ## Design principles
 
+- **Follow SOLID principles.** In particular:
+  - **Law of Demeter:** Methods should depend only on the data they need, not on the
+    objects that contain it. Pass the specific value (e.g., a `Future`) rather than the
+    parent object (e.g., `TriggerData`) — this reduces coupling and removes the need for
+    null-checks on intermediate accessors.
+  - **Single Responsibility:** Each module, class, and function should have one clear reason
+    to change.
+  - **Interface Segregation:** Depend on narrow interfaces. If a helper only needs one field
+    from a dataclass, accept that field directly.
 - **Decouple infrastructure from domain:** Modules like `signature.py` and `dispatcher.py` are
   general-purpose (signature adaptation, listener/observer pattern) and intentionally not coupled
   to the state machine domain. Prefer this separation even for modules that are only used
@@ -163,6 +172,13 @@ uv run sphinx-build docs docs/_build/html
 uv run sphinx-autobuild docs docs/_build/html --re-ignore "auto_examples/.*"
 ```
 
+### Documentation code examples
+
+All code examples in `docs/*.md` **must** be testable doctests (using ```` ```py ```` with
+`>>>` prompts), not plain ```` ```python ```` blocks. The test suite collects them via
+`--doctest-glob=*.md`. If an example cannot be expressed as a doctest (e.g., it requires
+real concurrency), write it as a unit test in `tests/` and reference it from the docs instead.
+
 ## Git workflow
 
 - Main branch: `develop`

docs/async.md

@@ -193,13 +193,74 @@ compound states, parallel states, history pseudo-states, eventless transitions,
 and `done.state` events — are fully supported in async code. The same
 `activate_initial_state()` pattern applies:
 
-```python
-async def run():
-    sm = MyStateChart()
-    await sm.activate_initial_state()
-    await sm.send("event")
+```py
+>>> async def run():
+...     sm = AsyncStateMachine()
+...     await sm.activate_initial_state()
+...     result = await sm.send("advance")
+...     return result
+
+>>> asyncio.run(run())
+42
+
 ```
 
+### Concurrent event sending
+
+```{versionadded} 3.0.0
+```
+
+When multiple coroutines send events concurrently (e.g., via `asyncio.gather`),
+each caller receives its own event's result — even though only one coroutine
+actually runs the processing loop at a time.
+
+```py
+>>> class ConcurrentSC(StateChart):
+...     s1 = State(initial=True)
+...     s2 = State()
+...     s3 = State(final=True)
+...
+...     step1 = s1.to(s2)
+...     step2 = s2.to(s3)
+...
+...     async def on_step1(self):
+...         return "result_1"
+...
+...     async def on_step2(self):
+...         return "result_2"
+
+>>> async def run_concurrent():
+...     import asyncio as _asyncio
+...     sm = ConcurrentSC()
+...     await sm.activate_initial_state()
+...     r1, r2 = await _asyncio.gather(
+...         sm.send("step1"),
+...         sm.send("step2"),
+...     )
+...     return r1, r2
+
+>>> asyncio.run(run_concurrent())
+('result_1', 'result_2')
+
+```
+
+Under the hood, the async engine attaches an `asyncio.Future` to each
+externally enqueued event. The coroutine that acquires the processing lock
+resolves each event's future as it processes the queue. Callers that couldn't
+acquire the lock simply `await` their future.
+
+```{note}
+Futures are only created for **external** events sent from outside the
+processing loop. Events triggered from within callbacks (reentrant calls)
+follow the existing run-to-completion (RTC) model — they are enqueued and
+processed within the current macrostep, and the callback receives ``None``.
+```
+
+If an exception occurs during processing (with `error_on_execution=False`),
+the exception is routed to the caller whose event caused it. Other callers
+whose events were still pending will also receive the exception, since the
+processing loop clears the queue on failure.
+
 ### Async-specific limitations
 
 - **Initial state activation**: In async code, you must `await sm.activate_initial_state()`

docs/releases/3.0.0.md

@@ -410,6 +410,19 @@ class GameCharacter(StateChart):
 See {ref}`weighted-transitions` for full documentation.
 
 
+### Async concurrent event result routing
+
+When multiple coroutines send events concurrently via `asyncio.gather`, each
+caller now receives its own event's result (or exception). Previously, only the
+first caller to acquire the processing lock would get a result — subsequent
+callers received `None` and exceptions could leak to the wrong caller.
+
+This is implemented by attaching an `asyncio.Future` to each externally
+enqueued event in the async engine. See {ref}`async` for details.
+
+Fixes [#509](https://github.com/fgmacedo/python-statemachine/issues/509).
+
+
 ## Bugfixes in 3.0.0
 
 - Fixes [#XXX](https://github.com/fgmacedo/python-statemachine/issues/XXX).

statemachine/engines/async_.py

@@ -1,4 +1,5 @@
 import asyncio
+import contextvars
 import logging
 from itertools import chain
 from time import time
@@ -21,13 +22,57 @@
 logger = logging.getLogger(__name__)
 
 
+# ContextVar to distinguish reentrant calls (from within callbacks) from
+# concurrent external calls. asyncio propagates context to child tasks
+# (e.g., those created by asyncio.gather in the callback system), so a
+# ContextVar set in the processing loop is visible in all callbacks.
+# Independent external coroutines have their own context where this is False.
+_in_processing_loop: contextvars.ContextVar[bool] = contextvars.ContextVar(
+    "_in_processing_loop", default=False
+)
+
+
 class AsyncEngine(BaseEngine):
     """Async engine with full StateChart support.
 
     Mirrors :class:`SyncEngine` algorithm but uses ``async``/``await`` for callback dispatch.
     All pure-computation helpers are inherited from :class:`BaseEngine`.
     """
 
+    def put(self, trigger_data: TriggerData, internal: bool = False, _delayed: bool = False):
+        """Override to attach an asyncio.Future for external events.
+
+        Futures are only created when:
+        - The event is external (not internal)
+        - No future is already attached
+        - There is a running asyncio loop
+        - The call is NOT from within the processing loop (reentrant calls
+          from callbacks must not get futures, as that would deadlock)
+        """
+        if not internal and trigger_data.future is None and not _in_processing_loop.get():
+            try:
+                loop = asyncio.get_running_loop()
+                trigger_data.future = loop.create_future()
+            except RuntimeError:
+                pass  # No running loop — sync caller
+        super().put(trigger_data, internal=internal, _delayed=_delayed)
+
+    @staticmethod
+    def _resolve_future(future: "asyncio.Future[object] | None", result):
+        """Resolve a future with the given result, if present and not yet done."""
+        if future is not None and not future.done():
+            future.set_result(result)
+
+    @staticmethod
+    def _reject_future(future: "asyncio.Future[object] | None", exc: Exception):
+        """Reject a future with the given exception, if present and not yet done."""
+        if future is not None and not future.done():
+            future.set_exception(exc)
+
+    def _reject_pending_futures(self, exc: Exception):
+        """Reject all unresolved futures in the external queue."""
+        self.external_queue.reject_futures(exc)
+
     # --- Callback dispatch overrides (async versions of BaseEngine methods) ---
 
     async def _get_args_kwargs(
@@ -265,16 +310,27 @@ async def activate_initial_state(self):
         """
         return await self.processing_loop()
 
-    async def processing_loop(self):  # noqa: C901
+    async def processing_loop(  # noqa: C901
+        self, caller_future: "asyncio.Future[object] | None" = None
+    ):
         """Process event triggers with the 3-phase macrostep architecture.
 
         Phase 1: Eventless transitions + internal queue until quiescence.
         Phase 2: Remaining internal events (safety net for invoke-generated events).
         Phase 3: External events.
+
+        When ``caller_future`` is provided, the caller can ``await`` it to
+        receive its own event's result — even if another coroutine holds the
+        processing lock.
         """
         if not self._processing.acquire(blocking=False):
+            # Another coroutine holds the lock and will process our event.
+            # Await the caller's future so we get our own result back.
+            if caller_future is not None:
+                return await caller_future
             return None
 
+        _ctx_token = _in_processing_loop.set(True)
         logger.debug("Processing loop started: %s", self.sm.current_state_value)
         first_result = self._sentinel
         try:
@@ -336,26 +392,53 @@ async def processing_loop(self):  # noqa: C901
                         )
                         break
 
-                    enabled_transitions = await self.select_transitions(external_event)
-                    logger.debug("Enabled transitions: %s", enabled_transitions)
-                    if enabled_transitions:
-                        try:
+                    event_future = external_event.future
+                    try:
+                        enabled_transitions = await self.select_transitions(external_event)
+                        logger.debug("Enabled transitions: %s", enabled_transitions)
+                        if enabled_transitions:
                             result = await self.microstep(
                                 list(enabled_transitions), external_event
                             )
+                            self._resolve_future(event_future, result)
                             if first_result is self._sentinel:
                                 first_result = result
-                        except Exception:
-                            self.clear()
-                            raise
-
-                    else:
-                        if not self.sm.allow_event_without_transition:
-                            raise TransitionNotAllowed(external_event.event, self.sm.configuration)
-
+                        else:
+                            if not self.sm.allow_event_without_transition:
+                                tna = TransitionNotAllowed(
+                                    external_event.event, self.sm.configuration
+                                )
+                                self._reject_future(event_future, tna)
+                                self._reject_pending_futures(tna)
+                                raise tna
+                            # Event allowed but no transition — resolve with None
+                            self._resolve_future(event_future, None)
+                    except Exception as exc:
+                        self._reject_future(event_future, exc)
+                        self._reject_pending_futures(exc)
+                        self.clear()
+                        raise
+
+        except Exception as exc:
+            if caller_future is not None:
+                # Route the exception to the caller's future if still pending.
+                # If already resolved (caller's own event succeeded before a
+                # later event failed), suppress the exception — the caller will
+                # get their successful result via ``await future`` below, and
+                # the failing event's exception was already routed to *its*
+                # caller's future by ``_reject_future(event_future, ...)``.
+                self._reject_future(caller_future, exc)
+            else:
+                raise
         finally:
+            _in_processing_loop.reset(_ctx_token)
             self._processing.release()
-        return first_result if first_result is not self._sentinel else None
+
+        result = first_result if first_result is not self._sentinel else None
+        # If the caller has a future, await it (already resolved by now).
+        if caller_future is not None:
+            return await caller_future
+        return result
 
     async def enabled_events(self, *args, **kwargs):
         sm = self.sm

statemachine/engines/base.py

@@ -59,6 +59,18 @@ def clear(self):
         with self.queue.mutex:
             self.queue.queue.clear()
 
+    def reject_futures(self, exc: Exception):
+        """Reject all unresolved futures in the queue.
+
+        Called when the processing loop exits abnormally so that coroutines
+        awaiting their futures don't hang forever.
+        """
+        with self.queue.mutex:
+            for trigger_data in self.queue.queue:
+                future = trigger_data.future
+                if future is not None and not future.done():
+                    future.set_exception(exc)
+
     def remove(self, send_id: str):
         # We use the internal `queue` to make thins faster as the mutex
         # is protecting the block below

statemachine/engines/sync.py

@@ -57,7 +57,7 @@ def activate_initial_state(self):
                 self._processing.release()
         return self.processing_loop()
 
-    def processing_loop(self):  # noqa: C901
+    def processing_loop(self, caller_future=None):  # noqa: C901
         """Process event triggers.
 
         The event is put on a queue, and only the first event will have the result collected.

statemachine/event.py

@@ -156,8 +156,8 @@ def __call__(self, *args, **kwargs) -> Any:
         # can be called as a method. But it is not meant to be called without
         # an SM instance. Such SM instance is provided by `__get__` method when
         # used as a property descriptor.
-        self.put(*args, **kwargs)
-        return self._sm._processing_loop()  # type: ignore[union-attr]
+        trigger_data = self.put(*args, **kwargs)
+        return self._sm._processing_loop(trigger_data.future)  # type: ignore[union-attr]
 
     def split(  # type: ignore[override]
         self, sep: "str | None" = None, maxsplit: int = -1

statemachine/event_data.py

@@ -36,6 +36,13 @@ class TriggerData:
     kwargs: dict = field(default_factory=dict, compare=False)
     """All keyword arguments provided on the :ref:`Event`."""
 
+    future: Any = field(default=None, compare=False, repr=False, init=False)
+    """An optional :class:`asyncio.Future` for async result routing.
+
+    When set, the processing loop will resolve this future with the microstep
+    result (or exception), allowing the caller to ``await`` it.
+    """
+
     def __post_init__(self):
         self.model = self.machine.model
         delay = self.event.delay if self.event and self.event.delay else 0

statemachine/statemachine.py

@@ -175,8 +175,8 @@ def activate_initial_state(self) -> Any:
             return result
         return run_async_from_sync(result)
 
-    def _processing_loop(self) -> Any:
-        result = self._engine.processing_loop()
+    def _processing_loop(self, caller_future: "Any | None" = None) -> Any:
+        result = self._engine.processing_loop(caller_future)
         if not isawaitable(result):
             return result
         return run_async_from_sync(result)