From ab4e7f015686cf5e1637d67ea40db552d45a8269 Mon Sep 17 00:00:00 2001
From: Fernando Macedo <fgmacedo@gmail.com>
Date: Tue, 13 Oct 2026 20:06:31 -0300
Subject: [PATCH 1/3] fix: answer async callers after their macrostep and hand
 the loop over on cancellation

On the async engine a caller is now answered once the macrostep of its
event ends, so an exception raised by the internal events or eventless
transitions that follow it reaches that caller. When nobody awaits the
failing event, the coroutine running the loop gets the exception, as on
the sync engine.

Cancelling the coroutine running the processing loop no longer leaves
the callers of queued events waiting: one of them takes the loop over.

Fixes #689
Fixes #690

Signed-off-by: Fernando Macedo <fgmacedo@gmail.com>
---
 docs/async.md                  |  13 +-
 docs/releases/3.2.2.md         |  71 ++++++++-
 statemachine/engines/async_.py | 137 +++++++++++------
 tests/test_async_futures.py    | 258 +++++++++++++++++++++++++++++++--
 tests/test_cancel_event.py     |   2 +-
 5 files changed, 418 insertions(+), 63 deletions(-)

diff --git a/docs/async.md b/docs/async.md
index fbc18992..cc1f22c0 100644
--- a/docs/async.md
+++ b/docs/async.md
@@ -165,9 +165,16 @@ model: they are enqueued and processed by the running loop.
 If an exception occurs during processing (with `catch_errors_as_events=False`),
 the exception is routed to the caller whose event raised it. Other callers
 whose events were still pending will also receive the exception, since the
-processing loop clears the queue on failure. An exception raised by the
-internal events or eventless transitions that follow an event goes only to
-the callers still pending: the caller of that event already has its result.
+processing loop clears the queue on failure. A caller is answered once the
+macrostep of its event ends, so an exception raised by the internal events or
+eventless transitions that follow its event reaches it too. When no coroutine
+awaits that event, because a callback sent it or its caller was cancelled, the
+exception goes to the coroutine running the processing loop.
+
+If the coroutine running the processing loop is cancelled, only that coroutine
+stops: a coroutine awaiting a queued event takes the loop over, and returns
+once the queue is empty. The caller of the event being processed when the
+cancellation hit gets `CancelledError`, since its callbacks were interrupted.
 
 
 (syncengine)=
diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md
index 64c40671..1f2de676 100644
--- a/docs/releases/3.2.2.md
+++ b/docs/releases/3.2.2.md
@@ -74,6 +74,9 @@ fail after upgrading:
   transition. Internal events raised before a callback fails, and events sent before a failed
   lazy initial activation on the async engine, are dropped too, instead of running at the next
   event. See {ref}`release-3.2.2-macrostep`.
+- On the async engine, with `catch_errors_as_events=False`, `await send()` raises an exception
+  that escapes the internal events or eventless transitions following its event. It used to
+  return the event's result, and the exception was lost. See {ref}`release-3.2.2-async-loop`.
 - An attribute named with the `error_`, `done_state_` or `done_invoke_` prefix now registers
   two events, so `events`, `allowed_events` and `enabled_events()` list both ids. See
   {ref}`release-3.2.2-prefixed-events`.
@@ -377,9 +380,9 @@ dropped event gets that exception. Any exception escapes with `catch_errors_as_e
 as on a `StateMachine`; on a `StateChart`, one from a validator or `prepare_event` does. On
 3.2.1 the queued external events ran before those internal events and eventless transitions,
 and their callers got their own results; the exception was then raised by the sync `send()`
-that started processing, and on the async engine it reached no caller. On the async engine, as
-on 3.2.1, the coroutine whose event enabled the failing transitions gets its own event's
-result. Internal events raised by a callback before it fails are dropped too: they used to run
+that started processing, and on the async engine it reached no caller. The coroutine whose
+event enabled the failing transitions gets the exception too, see
+{ref}`release-3.2.2-async-loop`. Internal events raised by a callback before it fails are dropped too: they used to run
 at the next event. The same applies when the lazy initial activation fails on the async engine,
 where a second coroutine awaiting an event sent before it used to wait forever, and the events
 sent before it, which used to run at the next `send()`, are dropped.
@@ -389,6 +392,68 @@ sent before it, which used to run at the next `send()`, are dropped.
 [#686](https://github.com/fgmacedo/python-statemachine/issues/686),
 [#685](https://github.com/fgmacedo/python-statemachine/pull/685).
 
+(release-3.2.2-async-loop)=
+
+#### Async engine: an exception after the event, and a cancelled processing loop
+
+**An exception raised after a coroutine's event was lost.** With
+`catch_errors_as_events=False`, as on a `StateMachine`, an exception raised by the internal
+events or eventless transitions that follow an external event reached no caller on the async
+engine: `await send()` returned the event's result, and the exception was neither raised nor
+logged. The sync engine raises it from `send()`. A coroutine is now answered once the
+macrostep of its event ends, so the exception reaches it:
+
+```py
+>>> import asyncio
+
+>>> class Alarm(StateMachine):
+...     idle = State(initial=True)
+...     armed = State()
+...     ringing = State()
+...
+...     arm = idle.to(armed)
+...     trip = armed.to(ringing)
+...     reset = ringing.to(idle)
+...
+...     async def on_enter_armed(self):
+...         await self.raise_("trip")
+...
+...     async def on_trip(self):
+...         raise ValueError("sensor failure")
+
+>>> async def arm():
+...     sm = Alarm()
+...     await sm.activate_initial_state()
+...     await sm.send("arm")
+
+>>> asyncio.run(arm())
+Traceback (most recent call last):
+...
+ValueError: sensor failure
+
+```
+
+When no coroutine awaits the event whose macrostep failed, because a callback sent it or its
+caller was cancelled, the coroutine running the processing loop gets the exception, as the
+sync engine raises it from the `send()` running the loop. The callers of the events still
+queued get the exception as before. A `StateChart` is unchanged: the error becomes
+`error.execution`.
+
+**Cancelling the coroutine running the processing loop left the others waiting.** When the
+coroutine holding the processing loop was cancelled, as `asyncio.wait_for()` or a request
+timeout does, the coroutines awaiting the `send()` of events queued behind it were not
+answered until a later `send()` ran the loop again: `CancelledError` is not an `Exception`, so
+the loop's error handling did not run. The cancellation now stops only that coroutine. The
+internal events left by its macrostep are dropped, and a coroutine awaiting a queued event
+takes the processing loop over, so each gets its own result, delayed events included. Like any
+coroutine running the loop, the one taking over returns once the queue is empty. A coroutine
+awaiting the event being processed when the cancellation hit, whose callbacks were interrupted
+halfway, gets `CancelledError`. The cancelled coroutine's own event, if still queued, is
+processed, as for any coroutine that stops awaiting its `send()`.
+
+[#689](https://github.com/fgmacedo/python-statemachine/issues/689),
+[#690](https://github.com/fgmacedo/python-statemachine/issues/690).
+
 (release-3.2.2-prefixed-events)=
 
 #### Prefixed events were registered under a glued id
diff --git a/statemachine/engines/async_.py b/statemachine/engines/async_.py
index c2f9c09f..511c62d9 100644
--- a/statemachine/engines/async_.py
+++ b/statemachine/engines/async_.py
@@ -16,9 +16,38 @@
 from .base import _running_engine
 
 if TYPE_CHECKING:
+    from ..statemachine import StateChart
     from ..transition import Transition
 
 
+class _MacrostepCaller:
+    """The caller of the external event whose macrostep is running, answered once the
+    macrostep ends, so that an exception raised by its internal events or eventless
+    transitions still reaches it."""
+
+    def __init__(self):
+        self.future: "asyncio.Future[object] | None" = None
+        self.result: object = None
+
+    def answer(self):
+        AsyncEngine._resolve_future(self.future, self.result)
+        self.future = None
+        self.result = None
+
+    def reject(self, exc: Exception) -> bool:
+        """Reject the caller with ``exc``; returns whether a caller was still waiting."""
+        if self.future is None or self.future.done():
+            return False
+        self.future.set_exception(exc)
+        return True
+
+    def interrupt(self):
+        if self.future is not None:
+            self.future.cancel(
+                "The task running the processing loop was cancelled while it processed this event."
+            )
+
+
 class AsyncEngine(BaseEngine):
     """Async engine with full StateChart support.
 
@@ -26,6 +55,10 @@ class AsyncEngine(BaseEngine):
     All pure-computation helpers are inherited from :class:`BaseEngine`.
     """
 
+    def __init__(self, sm: "StateChart"):
+        super().__init__(sm)
+        self._lock_waiters: "set[asyncio.Future[object]]" = set()
+
     def put(self, trigger_data: TriggerData, internal: bool = False):
         """Override to attach an asyncio.Future to a trigger put from outside the loop.
 
@@ -50,12 +83,6 @@ def _resolve_future(future: "asyncio.Future[object] | None", result):
         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)
@@ -340,17 +367,45 @@ async def processing_loop(self, caller_future: "asyncio.Future[object] | None" =
 
         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.
+        processing lock. If that coroutine is cancelled, the caller takes the loop over.
         """
-        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
+        while not self._processing.acquire(blocking=False):
+            if caller_future is None:
+                return None
+            await self._wait_for_result_or_release(caller_future)
+            if caller_future.done():
+                return caller_future.result()
+
+        try:
+            return await self._run_processing_loop(caller_future)
+        finally:
+            self._processing.release()
+            self._wake_lock_waiters()
 
+    async def _wait_for_result_or_release(self, caller_future: "asyncio.Future[object]"):
+        """Wait until the caller's event is answered or the processing lock is released.
+
+        A caller cancelled here cancels its future, so its event, still processed, answers
+        nobody.
+        """
+        released = caller_future.get_loop().create_future()
+        self._lock_waiters.add(released)
+        try:
+            await asyncio.wait((caller_future, released), return_when=asyncio.FIRST_COMPLETED)
+        except BaseException:
+            caller_future.cancel()
+            raise
+        finally:
+            self._lock_waiters.discard(released)
+
+    def _wake_lock_waiters(self):
+        for released in self._lock_waiters:
+            self._resolve_future(released, None)
+
+    async def _run_processing_loop(self, caller_future: "asyncio.Future[object] | None"):
         self._debug("%s Processing loop started: %s", self._log_id, self.sm.current_state_value)
         first_result = self._sentinel
+        macrostep = _MacrostepCaller()
         try:
             with _processing_step(self):
                 took_events = True
@@ -360,37 +415,34 @@ async def processing_loop(self, caller_future: "asyncio.Future[object] | None" =
 
                     # Spawn invoke handlers for states entered during this macrostep.
                     await self._invoke_manager.spawn_pending_async()
+                    macrostep.answer()
                     self._check_root_final_state()
                     if not self.running:
                         break
 
                     took_external_events, first_result = await self._take_due_external_event(
-                        first_result
+                        first_result, macrostep
                     )
                     took_events = took_events or took_external_events
 
+                macrostep.answer()
                 if not self.running:
                     self._drop_external_events()
 
         except Exception as exc:
             self._reject_pending_futures(exc)
             self.clear()
-            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:
+            if not macrostep.reject(exc) and (caller_future is None or caller_future.done()):
                 raise
-        finally:
-            self._processing.release()
+        except BaseException:
+            self.internal_queue.clear()
+            macrostep.interrupt()
+            if caller_future is not None:
+                caller_future.cancel()
+            raise
 
         self._debug("%s Processing loop ended", self._log_id)
         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:
             # Resolve the future if it wasn't processed (e.g. machine terminated).
             self._resolve_future(caller_future, result)
@@ -424,11 +476,12 @@ async def _run_macrostep(self) -> bool:
                 took_events = True
                 await self.microstep(list(enabled_transitions), internal_event)
 
-    async def _take_due_external_event(self, first_result):
+    async def _take_due_external_event(self, first_result, macrostep: "_MacrostepCaller"):
         """Take the external event at the head of the queue, if it is due.
 
         One at a time, so the macrostep that follows runs the internal events and eventless
-        transitions it enabled before the next external event is taken.
+        transitions it enabled before the next external event is taken. The event's caller is
+        left in ``macrostep``, to be answered once that macrostep ends.
 
         Returns whether the queue held an event, and ``first_result`` updated with the result
         of the first event that enabled a transition.
@@ -458,35 +511,29 @@ async def _take_due_external_event(self, first_result):
             await self._enter_states(transitions, external_event, OrderedSet(), OrderedSet())
             return True, first_result
 
+        macrostep.future = external_event.future
         result = await self._take_external_event(external_event)
+        if result is not self._sentinel:
+            macrostep.result = result
         if first_result is self._sentinel:
             first_result = result
         return True, first_result
 
     async def _take_external_event(self, external_event: TriggerData):
-        """Run one external event and settle its future.
+        """Run one external event.
 
         Returns the event's microstep result, or the sentinel when it enabled no transition.
         """
         # Finalize + autoforward for active invocations
         self._invoke_manager.handle_external_event(external_event)
 
-        event_future = external_event.future
-        try:
-            enabled_transitions = await self.select_transitions(external_event)
-            self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions)
-            if enabled_transitions:
-                result = await self.microstep(list(enabled_transitions), external_event)
-                self._resolve_future(event_future, result)
-                return result
-            if not self.sm.allow_event_without_transition:
-                raise TransitionNotAllowed(external_event.event, self.sm.configuration)
-            # Event allowed but no transition — resolve with None
-            self._resolve_future(event_future, None)
-            return self._sentinel
-        except Exception as exc:
-            self._reject_future(event_future, exc)
-            raise
+        enabled_transitions = await self.select_transitions(external_event)
+        self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions)
+        if enabled_transitions:
+            return await self.microstep(list(enabled_transitions), external_event)
+        if not self.sm.allow_event_without_transition:
+            raise TransitionNotAllowed(external_event.event, self.sm.configuration)
+        return self._sentinel
 
     async def enabled_events(self, *args, **kwargs):
         sm = self.sm
diff --git a/tests/test_async_futures.py b/tests/test_async_futures.py
index e9eacf8a..45b18f5e 100644
--- a/tests/test_async_futures.py
+++ b/tests/test_async_futures.py
@@ -681,7 +681,8 @@ async def test_statemachine_callers_get_the_exception(self, failing, eventless):
         sm = gated(khazad_dum(StateMachine, failing, eventless))()
         holder, waiters = await cross_with_callers_queued(sm, *THREE_FLIGHTS)
 
-        await holder
+        with pytest.raises(ValueError, match=f"{failing} failed"):
+            await holder
         results = await asyncio.gather(*waiters, return_exceptions=True)
         assert [str(result) for result in results] == [f"{failing} failed"] * 3
         assert sm.flights == 0
@@ -706,8 +707,7 @@ async def test_statechart_callers_get_their_results(self, failing, eventless):
     async def test_machine_takes_new_events_afterwards(self):
         sm = gated(khazad_dum(StateMachine, "on_falling"))()
         holder, waiters = await cross_with_callers_queued(sm, *THREE_FLIGHTS)
-        await holder
-        await asyncio.gather(*waiters, return_exceptions=True)
+        await asyncio.gather(holder, *waiters, return_exceptions=True)
 
         assert await sm.send("flee") == 1
         assert sm.configuration_values == {"bridge"}
@@ -725,9 +725,9 @@ async def test_delayed_and_cancelled_callers_are_answered(self):
         )
         sm.cancel_event("horn")
 
-        await holder
-        results = await asyncio.gather(delayed, cancelled, flight, return_exceptions=True)
+        results = await asyncio.gather(holder, delayed, cancelled, flight, return_exceptions=True)
         assert [str(result) for result in results] == [
+            "on_falling failed",
             "on_falling failed",
             "None",
             "on_falling failed",
@@ -737,7 +737,7 @@ async def test_delayed_and_cancelled_callers_are_answered(self):
 
     @pytest.mark.timeout(5)
     @pytest.mark.asyncio()
-    async def test_failure_after_a_queued_event_answers_only_the_callers_behind_it(self):
+    async def test_failure_after_a_queued_event_answers_its_caller_and_those_behind_it(self):
         sm = gated(khazad_dum(StateMachine, "on_falling"))()
         holder, waiters = await cross_with_callers_queued(
             sm, ("flee", {}), ("retreat", {}), ("cross", {}), *THREE_FLIGHTS[:2], awake=False
@@ -745,8 +745,8 @@ async def test_failure_after_a_queued_event_answers_only_the_callers_behind_it(s
 
         assert await holder == "crossed"
         results = await asyncio.gather(*waiters, return_exceptions=True)
-        assert results[:3] == [1, None, "crossed"]
-        assert [str(result) for result in results[3:]] == ["on_falling failed"] * 2
+        assert results[:2] == [1, None]
+        assert [str(result) for result in results[2:]] == ["on_falling failed"] * 3
         assert sm.flights == 1
 
     @pytest.mark.timeout(5)
@@ -755,9 +755,8 @@ async def test_statechart_validator_failure_answers_the_callers(self):
         sm = gated(khazad_dum(StateChart, "validate_falling"))()
         holder, waiters = await cross_with_callers_queued(sm, *THREE_FLIGHTS)
 
-        await holder
-        results = await asyncio.gather(*waiters, return_exceptions=True)
-        assert [str(result) for result in results] == ["validate_falling failed"] * 3
+        results = await asyncio.gather(holder, *waiters, return_exceptions=True)
+        assert [str(result) for result in results] == ["validate_falling failed"] * 4
         assert sm.flights == 0
 
     @pytest.mark.timeout(5)
@@ -832,6 +831,243 @@ def test_sync_engine_drops_the_events_queued_behind_the_failure(self, failing, e
         assert sm.send("flee") == 1
 
 
+async def cancel_and_wait(task):
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+
+class TestMacrostepErrorReachesTheCallerOfItsEvent:
+    """An exception escaping the internal events or eventless transitions that follow an
+    external event reaches the caller of that event, on both engines. When no caller awaits
+    that event, because a callback sent it, the exception reaches the caller running the loop.
+
+    See: https://github.com/fgmacedo/python-statemachine/issues/689
+    """
+
+    @pytest.mark.timeout(5)
+    @pytest.mark.parametrize("eventless", [False, True], ids=["internal", "eventless"])
+    @pytest.mark.parametrize("failing", FAILING_CALLBACKS)
+    async def test_caller_gets_the_exception(self, sm_runner, failing, eventless):
+        sm = await sm_runner.start(khazad_dum(StateMachine, failing, eventless))
+
+        with pytest.raises(ValueError, match=f"{failing} failed"):
+            await sm_runner.send(sm, "cross")
+
+        assert await sm_runner.send(sm, "flee") == 1
+
+    @pytest.mark.timeout(5)
+    async def test_statechart_caller_gets_its_result(self, sm_runner):
+        sm = await sm_runner.start(khazad_dum(StateChart, "on_falling"))
+
+        assert await sm_runner.send(sm, "cross") == "crossed"
+
+    @pytest.mark.timeout(5)
+    async def test_event_sent_by_a_callback_fails_on_the_caller_running_the_loop(self, sm_runner):
+        sm = await sm_runner.start(khazad_dum(StateMachine, "on_falling"))
+
+        with pytest.raises(ValueError, match="on_falling failed"):
+            await sm_runner.send(sm, "cross", awake=False, queued=("retreat", "cross"))
+
+        assert sm.configuration_values == {"bridge"}
+
+    @pytest.mark.timeout(5)
+    @pytest.mark.asyncio()
+    async def test_failure_of_a_cancelled_callers_event_reaches_the_caller_running_the_loop(
+        self,
+    ):
+        sm = gated(khazad_dum(StateMachine, "on_falling"))()
+        holder, (retreat, gives_up, flee) = await cross_with_callers_queued(
+            sm, ("retreat", {}), ("cross", {}), ("flee", {}), awake=False
+        )
+        await cancel_and_wait(gives_up)
+
+        with pytest.raises(ValueError, match="on_falling failed"):
+            await holder
+        assert await retreat is None
+        with pytest.raises(ValueError, match="on_falling failed"):
+            await flee
+
+    @pytest.mark.timeout(5)
+    @pytest.mark.asyncio()
+    async def test_failure_after_the_lazy_initial_activation_reaches_every_caller(self):
+        class Lamp(StateMachine):
+            off = State(initial=True)
+            lit = State(final=True)
+
+            switch = off.to(lit)
+            flicker = Event(off.to.itself(internal=True), internal=True)
+
+            async def on_enter_off(self):
+                self.flicker.put()
+
+            def on_flicker(self):
+                raise ValueError("short circuit")
+
+        sm = Lamp()
+        results = await asyncio.gather(
+            sm.send("switch"), sm.send("switch"), return_exceptions=True
+        )
+
+        assert [str(result) for result in results] == ["short circuit", "short circuit"]
+        assert sm.configuration_values == {"off"}
+
+
+def stalled_road(base=StateMachine):
+    """``go`` and ``stall`` hold the processing lock until cancelled, ``go`` until ``gate``
+    is set first; ``step`` counts the steps taken."""
+
+    class StalledRoad(base):
+        start = State(initial=True)
+        road = State()
+
+        go = start.to(road)
+        stall = start.to.itself(internal=True) | road.to.itself(internal=True)
+        step = start.to.itself(internal=True) | road.to.itself(internal=True)
+        hop = Event(road.to.itself(internal=True), internal=True)
+
+        def __init__(self, *args, **kwargs):
+            self.gate = asyncio.Event()
+            self.stalling = asyncio.Event()
+            self.steps = 0
+            self.hops = 0
+            super().__init__(*args, **kwargs)
+
+        async def on_go(self):
+            self.hop.put()
+            await self.gate.wait()
+            return "went"
+
+        async def on_stall(self):
+            self.stalling.set()
+            await asyncio.Event().wait()
+
+        def on_step(self):
+            self.steps += 1
+            return self.steps
+
+        def on_hop(self):
+            self.hops += 1
+
+    return StalledRoad
+
+
+class TestCancelledLockHolder:
+    """Cancelling the coroutine that runs the processing loop cancels only that caller: a
+    coroutine awaiting a queued event takes the loop over, and the machine stays usable.
+
+    See: https://github.com/fgmacedo/python-statemachine/issues/690
+    """
+
+    @pytest.mark.timeout(5)
+    @pytest.mark.asyncio()
+    async def test_queued_callers_get_their_results(self):
+        sm = stalled_road()()
+        await sm.activate_initial_state()
+        holder = asyncio.ensure_future(sm.send("go"))
+        await asyncio.sleep(0)
+        waiters = await queue_behind(sm, ("step", {}), ("step", {"delay": 50}), ("step", {}))
+
+        await cancel_and_wait(holder)
+
+        assert await asyncio.gather(*waiters) == [1, 3, 2]
+        assert sm.configuration_values == {"start"}
+        assert sm.hops == 0
+        assert not sm._engine._processing.locked()
+
+    @pytest.mark.timeout(5)
+    @pytest.mark.asyncio()
+    async def test_machine_takes_new_events_afterwards(self):
+        sm = stalled_road()()
+        await sm.activate_initial_state()
+        holder = asyncio.ensure_future(sm.send("go"))
+        await asyncio.sleep(0)
+
+        await cancel_and_wait(holder)
+
+        assert not sm._engine._processing.locked()
+        assert sm._engine.internal_queue.is_empty()
+        assert await sm.send("step") == 1
+        assert sm.hops == 0
+
+    @pytest.mark.timeout(5)
+    @pytest.mark.asyncio()
+    async def test_cancelled_while_waiting_for_a_delayed_event(self):
+        sm = stalled_road()()
+        await sm.activate_initial_state()
+        holder = asyncio.ensure_future(sm.send("step", delay=100))
+        await asyncio.sleep(0)
+        (waiter,) = await queue_behind(sm, ("step", {"delay": 50}))
+
+        await cancel_and_wait(holder)
+
+        assert await waiter == 1
+        assert sm.steps == 2
+        assert sm._engine.external_queue.is_empty()
+        assert not sm._engine._processing.locked()
+
+    @pytest.mark.timeout(5)
+    @pytest.mark.asyncio()
+    async def test_caller_of_the_interrupted_event_is_cancelled(self):
+        sm = stalled_road()()
+        await sm.activate_initial_state()
+        holder = asyncio.ensure_future(sm.send("go"))
+        await asyncio.sleep(0)
+        stalled, stepper = await queue_behind(sm, ("stall", {}), ("step", {}))
+        sm.gate.set()
+        await sm.stalling.wait()
+
+        await cancel_and_wait(holder)
+
+        with pytest.raises(asyncio.CancelledError, match="was cancelled while"):
+            await stalled
+        assert await stepper == 1
+        assert sm.configuration_values == {"road"}
+
+    @pytest.mark.timeout(5)
+    @pytest.mark.asyncio()
+    async def test_cancelled_waiter_leaves_its_event_queued(self):
+        sm = stalled_road()()
+        await sm.activate_initial_state()
+        holder = asyncio.ensure_future(sm.send("go"))
+        await asyncio.sleep(0)
+        gives_up, waiter = await queue_behind(sm, ("step", {}), ("step", {}))
+
+        await cancel_and_wait(gives_up)
+        sm.gate.set()
+
+        assert await holder == "went"
+        assert await waiter == 2
+        assert sm.hops == 1
+
+    @pytest.mark.timeout(5)
+    @pytest.mark.asyncio()
+    async def test_cancelled_initial_activation_hands_the_loop_over(self):
+        class SlowStart(StateMachine):
+            booting = State(initial=True)
+            ready = State(final=True)
+
+            boot = booting.to(ready)
+
+            def __init__(self):
+                self.gate = asyncio.Event()
+                super().__init__()
+
+            async def on_enter_booting(self):
+                await self.gate.wait()
+
+        sm = SlowStart()
+        activation = asyncio.ensure_future(sm.activate_initial_state())
+        await asyncio.sleep(0)
+        (waiter,) = await queue_behind(sm, ("boot", {}))
+
+        await cancel_and_wait(activation)
+        sm.gate.set()
+
+        await waiter
+        assert sm.configuration_values == {"ready"}
+
+
 class TestEventQueueRejectFutures:
     """Unit tests for EventQueue.reject_futures."""
 
diff --git a/tests/test_cancel_event.py b/tests/test_cancel_event.py
index d7652270..3116461b 100644
--- a/tests/test_cancel_event.py
+++ b/tests/test_cancel_event.py
@@ -324,7 +324,7 @@ async def test_a_caller_awaiting_a_cancelled_event_gets_none(cancels, clock):
 
     for _ in range(cancels):
         sm.cancel_event("dawn")
-    await asyncio.sleep(0)
+    await asyncio.wait([waiting], timeout=1)
 
     assert (settled(dawn.future), settled(waiting)) == ("answered with None", "answered with None")
     clock.now += 1.0

From fd9eafb82e69f9a41284b50585f4a8d010d47a64 Mon Sep 17 00:00:00 2001
From: Fernando Macedo <fgmacedo@gmail.com>
Date: Tue, 13 Oct 2026 20:18:53 -0300
Subject: [PATCH 2/3] fix: reject the interrupted caller with StateMachineError
 and retrieve the runner's future

A CancelledError for a caller whose task was not cancelled is swallowed
by asyncio.TaskGroup, so the caller of the event interrupted by the
cancellation now gets a StateMachineError. The loop runner whose own
queued future was rejected awaits it instead of raising, so asyncio does
not log the exception as never retrieved.

Signed-off-by: Fernando Macedo <fgmacedo@gmail.com>
---
 docs/async.md                  |  6 ++--
 docs/releases/3.2.2.md         |  8 ++---
 statemachine/engines/async_.py | 15 +++++++---
 tests/test_async_futures.py    | 53 ++++++++++++++++++++--------------
 4 files changed, 50 insertions(+), 32 deletions(-)

diff --git a/docs/async.md b/docs/async.md
index cc1f22c0..c654e80d 100644
--- a/docs/async.md
+++ b/docs/async.md
@@ -152,7 +152,8 @@ processing loop at a time. The sync engine does not support this pattern.
 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 didn't
-acquire the lock simply `await` their future.
+acquire the lock `await` their future, or the release of the lock, to take
+the loop over if the coroutine holding it was cancelled.
 
 ```{note}
 Futures are created for every event sent from outside the processing loop,
@@ -174,7 +175,8 @@ exception goes to the coroutine running the processing loop.
 If the coroutine running the processing loop is cancelled, only that coroutine
 stops: a coroutine awaiting a queued event takes the loop over, and returns
 once the queue is empty. The caller of the event being processed when the
-cancellation hit gets `CancelledError`, since its callbacks were interrupted.
+cancellation hit gets a `StateMachineError`, since its callbacks were
+interrupted.
 
 
 (syncengine)=
diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md
index 1f2de676..1efa0d5d 100644
--- a/docs/releases/3.2.2.md
+++ b/docs/releases/3.2.2.md
@@ -381,9 +381,9 @@ as on a `StateMachine`; on a `StateChart`, one from a validator or `prepare_even
 3.2.1 the queued external events ran before those internal events and eventless transitions,
 and their callers got their own results; the exception was then raised by the sync `send()`
 that started processing, and on the async engine it reached no caller. The coroutine whose
-event enabled the failing transitions gets the exception too, see
-{ref}`release-3.2.2-async-loop`. Internal events raised by a callback before it fails are dropped too: they used to run
-at the next event. The same applies when the lazy initial activation fails on the async engine,
+event enabled the failing transitions also gets the exception, see
+{ref}`release-3.2.2-async-loop`. Internal events raised by a callback before it fails are
+dropped too: they used to run at the next event. The same applies when the lazy initial activation fails on the async engine,
 where a second coroutine awaiting an event sent before it used to wait forever, and the events
 sent before it, which used to run at the next `send()`, are dropped.
 
@@ -448,7 +448,7 @@ internal events left by its macrostep are dropped, and a coroutine awaiting a qu
 takes the processing loop over, so each gets its own result, delayed events included. Like any
 coroutine running the loop, the one taking over returns once the queue is empty. A coroutine
 awaiting the event being processed when the cancellation hit, whose callbacks were interrupted
-halfway, gets `CancelledError`. The cancelled coroutine's own event, if still queued, is
+halfway, gets a `StateMachineError` saying so. The cancelled coroutine's own event, if still queued, is
 processed, as for any coroutine that stops awaiting its `send()`.
 
 [#689](https://github.com/fgmacedo/python-statemachine/issues/689),
diff --git a/statemachine/engines/async_.py b/statemachine/engines/async_.py
index 511c62d9..81675b03 100644
--- a/statemachine/engines/async_.py
+++ b/statemachine/engines/async_.py
@@ -7,7 +7,9 @@
 from ..event_data import EventData
 from ..event_data import TriggerData
 from ..exceptions import InvalidDefinition
+from ..exceptions import StateMachineError
 from ..exceptions import TransitionNotAllowed
+from ..i18n import _
 from ..orderedset import OrderedSet
 from ..state import State
 from .base import _ERROR_EXECUTION
@@ -42,10 +44,14 @@ def reject(self, exc: Exception) -> bool:
         return True
 
     def interrupt(self):
-        if self.future is not None:
-            self.future.cancel(
-                "The task running the processing loop was cancelled while it processed this event."
+        self.reject(
+            StateMachineError(
+                _(
+                    "The task running the processing loop was cancelled while it processed "
+                    "this event."
+                )
             )
+        )
 
 
 class AsyncEngine(BaseEngine):
@@ -430,9 +436,10 @@ async def _run_processing_loop(self, caller_future: "asyncio.Future[object] | No
                     self._drop_external_events()
 
         except Exception as exc:
+            answered = caller_future is None or caller_future.done()
             self._reject_pending_futures(exc)
             self.clear()
-            if not macrostep.reject(exc) and (caller_future is None or caller_future.done()):
+            if not macrostep.reject(exc) and answered:
                 raise
         except BaseException:
             self.internal_queue.clear()
diff --git a/tests/test_async_futures.py b/tests/test_async_futures.py
index 45b18f5e..36fafc94 100644
--- a/tests/test_async_futures.py
+++ b/tests/test_async_futures.py
@@ -12,6 +12,7 @@
 import pytest
 from statemachine.engines.base import EventQueue
 from statemachine.event_data import TriggerData
+from statemachine.exceptions import StateMachineError
 from statemachine.exceptions import TransitionNotAllowed
 
 from statemachine import Event
@@ -831,6 +832,22 @@ def test_sync_engine_drops_the_events_queued_behind_the_failure(self, failing, e
         assert sm.send("flee") == 1
 
 
+class FlickeringLamp(StateMachine):
+    """Its initial state raises an internal event whose transition fails."""
+
+    off = State(initial=True)
+    lit = State(final=True)
+
+    switch = off.to(lit)
+    flicker = Event(off.to.itself(internal=True), internal=True)
+
+    async def on_enter_off(self):
+        self.flicker.put()
+
+    def on_flicker(self):
+        raise ValueError("short circuit")
+
+
 async def cancel_and_wait(task):
     task.cancel()
     with pytest.raises(asyncio.CancelledError):
@@ -891,20 +908,7 @@ async def test_failure_of_a_cancelled_callers_event_reaches_the_caller_running_t
     @pytest.mark.timeout(5)
     @pytest.mark.asyncio()
     async def test_failure_after_the_lazy_initial_activation_reaches_every_caller(self):
-        class Lamp(StateMachine):
-            off = State(initial=True)
-            lit = State(final=True)
-
-            switch = off.to(lit)
-            flicker = Event(off.to.itself(internal=True), internal=True)
-
-            async def on_enter_off(self):
-                self.flicker.put()
-
-            def on_flicker(self):
-                raise ValueError("short circuit")
-
-        sm = Lamp()
+        sm = FlickeringLamp()
         results = await asyncio.gather(
             sm.send("switch"), sm.send("switch"), return_exceptions=True
         )
@@ -912,6 +916,16 @@ def on_flicker(self):
         assert [str(result) for result in results] == ["short circuit", "short circuit"]
         assert sm.configuration_values == {"off"}
 
+    @pytest.mark.timeout(5)
+    @pytest.mark.asyncio()
+    async def test_caller_running_the_loop_retrieves_its_rejected_future(self):
+        sm = FlickeringLamp()
+        trigger = sm.switch.put()
+
+        with pytest.raises(ValueError, match="short circuit"):
+            await sm._processing_loop(trigger.future)
+        assert not trigger.future._log_traceback
+
 
 def stalled_road(base=StateMachine):
     """``go`` and ``stall`` hold the processing lock until cancelled, ``go`` until ``gate``
@@ -1019,7 +1033,7 @@ async def test_caller_of_the_interrupted_event_is_cancelled(self):
 
         await cancel_and_wait(holder)
 
-        with pytest.raises(asyncio.CancelledError, match="was cancelled while"):
+        with pytest.raises(StateMachineError, match="was cancelled while"):
             await stalled
         assert await stepper == 1
         assert sm.configuration_values == {"road"}
@@ -1049,12 +1063,8 @@ class SlowStart(StateMachine):
 
             boot = booting.to(ready)
 
-            def __init__(self):
-                self.gate = asyncio.Event()
-                super().__init__()
-
             async def on_enter_booting(self):
-                await self.gate.wait()
+                await asyncio.Event().wait()
 
         sm = SlowStart()
         activation = asyncio.ensure_future(sm.activate_initial_state())
@@ -1062,9 +1072,8 @@ async def on_enter_booting(self):
         (waiter,) = await queue_behind(sm, ("boot", {}))
 
         await cancel_and_wait(activation)
-        sm.gate.set()
 
-        await waiter
+        assert await waiter is None
         assert sm.configuration_values == {"ready"}
 
 

From 5859fa307faf623270bab8fc427463d3038d46c8 Mon Sep 17 00:00:00 2001
From: Fernando Macedo <fgmacedo@gmail.com>
Date: Tue, 13 Oct 2026 20:28:49 -0300
Subject: [PATCH 3/3] docs: say what happens to queued events with no caller
 left to take the loop over

Signed-off-by: Fernando Macedo <fgmacedo@gmail.com>
---
 docs/async.md               |  9 +++++----
 docs/releases/3.2.2.md      | 14 ++++++++------
 tests/test_async_futures.py |  2 +-
 3 files changed, 14 insertions(+), 11 deletions(-)

diff --git a/docs/async.md b/docs/async.md
index c654e80d..07427e40 100644
--- a/docs/async.md
+++ b/docs/async.md
@@ -169,13 +169,14 @@ whose events were still pending will also receive the exception, since the
 processing loop clears the queue on failure. A caller is answered once the
 macrostep of its event ends, so an exception raised by the internal events or
 eventless transitions that follow its event reaches it too. When no coroutine
-awaits that event, because a callback sent it or its caller was cancelled, the
-exception goes to the coroutine running the processing loop.
+awaits that event, because a callback sent it or its caller was cancelled while
+awaiting it, the exception goes to the coroutine running the processing loop.
 
 If the coroutine running the processing loop is cancelled, only that coroutine
 stops: a coroutine awaiting a queued event takes the loop over, and returns
-once the queue is empty. The caller of the event being processed when the
-cancellation hit gets a `StateMachineError`, since its callbacks were
+once the queue is empty. With no coroutine waiting, the events left in the
+queue run at the next `send()`. The caller of the event being processed when
+the cancellation hit gets a `StateMachineError`, since its callbacks were
 interrupted.
 
 
diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md
index 1efa0d5d..eed34b1e 100644
--- a/docs/releases/3.2.2.md
+++ b/docs/releases/3.2.2.md
@@ -383,9 +383,10 @@ and their callers got their own results; the exception was then raised by the sy
 that started processing, and on the async engine it reached no caller. The coroutine whose
 event enabled the failing transitions also gets the exception, see
 {ref}`release-3.2.2-async-loop`. Internal events raised by a callback before it fails are
-dropped too: they used to run at the next event. The same applies when the lazy initial activation fails on the async engine,
-where a second coroutine awaiting an event sent before it used to wait forever, and the events
-sent before it, which used to run at the next `send()`, are dropped.
+dropped too: they used to run at the next event. The same applies when the lazy initial
+activation fails on the async engine, where a second coroutine awaiting an event sent before it
+used to wait forever, and the events sent before it, which used to run at the next `send()`,
+are dropped.
 
 [#668](https://github.com/fgmacedo/python-statemachine/pull/668),
 [#678](https://github.com/fgmacedo/python-statemachine/issues/678),
@@ -434,7 +435,8 @@ ValueError: sensor failure
 ```
 
 When no coroutine awaits the event whose macrostep failed, because a callback sent it or its
-caller was cancelled, the coroutine running the processing loop gets the exception, as the
+caller was cancelled while awaiting it, the coroutine running the processing loop gets the
+exception, as the
 sync engine raises it from the `send()` running the loop. The callers of the events still
 queued get the exception as before. A `StateChart` is unchanged: the error becomes
 `error.execution`.
@@ -448,8 +450,8 @@ internal events left by its macrostep are dropped, and a coroutine awaiting a qu
 takes the processing loop over, so each gets its own result, delayed events included. Like any
 coroutine running the loop, the one taking over returns once the queue is empty. A coroutine
 awaiting the event being processed when the cancellation hit, whose callbacks were interrupted
-halfway, gets a `StateMachineError` saying so. The cancelled coroutine's own event, if still queued, is
-processed, as for any coroutine that stops awaiting its `send()`.
+halfway, gets a `StateMachineError` saying so. With no coroutine waiting, the events left in
+the queue, the cancelled coroutine's own included, run at the next `send()`.
 
 [#689](https://github.com/fgmacedo/python-statemachine/issues/689),
 [#690](https://github.com/fgmacedo/python-statemachine/issues/690).
diff --git a/tests/test_async_futures.py b/tests/test_async_futures.py
index 36fafc94..cac60016 100644
--- a/tests/test_async_futures.py
+++ b/tests/test_async_futures.py
@@ -1022,7 +1022,7 @@ async def test_cancelled_while_waiting_for_a_delayed_event(self):
 
     @pytest.mark.timeout(5)
     @pytest.mark.asyncio()
-    async def test_caller_of_the_interrupted_event_is_cancelled(self):
+    async def test_caller_of_the_interrupted_event_gets_a_state_machine_error(self):
         sm = stalled_road()()
         await sm.activate_initial_state()
         holder = asyncio.ensure_future(sm.send("go"))