From 867814d11d706adabd7489132cd1810b8b02d0a0 Mon Sep 17 00:00:00 2001
From: John Merrill <john@fairplay.ai>
Date: Sun, 4 Oct 2026 14:00:04 -0700
Subject: [PATCH 1/8] fix: keep events due at the same time in the order they
 were sent

Both queues are PriorityQueues ordered by TriggerData's execution_time
alone, so triggers with equal times came out in heap order rather than
the order they were put. Two events stamped alike by a coarse clock could
run out of order; twenty raised in one callback with one stamp ran as
0, 2, 6, 14, 19, 18, ...

TriggerData now numbers each trigger as it is created and compares that
number after execution_time, so triggers due together keep their order.
The new tests stamp every trigger with one time and check the order on
both engines and both queues.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/releases/3.2.2.md       |  9 ++++++
 statemachine/event_data.py   | 10 +++++++
 tests/test_event_ordering.py | 58 ++++++++++++++++++++++++++++++++++++
 3 files changed, 77 insertions(+)
 create mode 100644 tests/test_event_ordering.py

diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md
index b0a6405e..2909bc51 100644
--- a/docs/releases/3.2.2.md
+++ b/docs/releases/3.2.2.md
@@ -76,6 +76,15 @@ Reported by [@Dolecor](https://github.com/Dolecor).
 
 [#643](https://github.com/fgmacedo/python-statemachine/issues/643).
 
+### Events due at the same moment kept their order only by chance
+
+Both event queues are priority queues ordered by each trigger's due time, and triggers due at
+the same time came out in no particular order. A clock coarse enough to stamp two events alike
+could run them out of the order they were sent or raised: twenty events raised in one callback,
+all stamped with the same time, ran as `0, 2, 6, 14, 19, 18, ...`. Each trigger now also holds
+the order it was created in, and triggers due at the same moment run in that order, on both
+engines and both queues.
+
 ### `delay` and `internal` dropped from an explicit `Event`
 
 `Event(dark.to(lit), delay=50)` rebuilt the event without its `delay`, so
diff --git a/statemachine/event_data.py b/statemachine/event_data.py
index 9eebfe41..176e1f6e 100644
--- a/statemachine/event_data.py
+++ b/statemachine/event_data.py
@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 from dataclasses import field
+from itertools import count
 from time import time
 from typing import TYPE_CHECKING
 from typing import Any
@@ -10,6 +11,9 @@
     from .statemachine import StateChart
     from .transition import Transition
 
+_sequence = count()
+"""Numbers each trigger as it is created, so triggers due at the same time keep their order."""
+
 
 @dataclass(order=True)
 class TriggerData:
@@ -27,6 +31,11 @@ class TriggerData:
     execution_time: float = field(default=0.0)
     """The time at which the :ref:`Event` should run."""
 
+    sequence: int = field(default=0, init=False)
+    """The order this trigger was created in, which breaks a tie between triggers due at the
+    same time: the queues are priority queues, and without it two triggers stamped alike come
+    out in no particular order."""
+
     model: Any = field(init=False, compare=False)
     """A reference to the underlying model that holds the current :ref:`State`."""
 
@@ -47,6 +56,7 @@ def __post_init__(self):
         self.model = self.machine.model
         delay = self.event.delay if self.event and self.event.delay else 0
         self.execution_time = time() + (delay / 1000)
+        self.sequence = next(_sequence)
 
 
 @dataclass
diff --git a/tests/test_event_ordering.py b/tests/test_event_ordering.py
new file mode 100644
index 00000000..d746220e
--- /dev/null
+++ b/tests/test_event_ordering.py
@@ -0,0 +1,58 @@
+"""Events due at the same moment are processed in the order they were sent.
+
+Both queues are priority queues ordered by each trigger's due time. Triggers
+stamped with the same time, as a coarse clock stamps events sent close together,
+must still come out in the order they went in.
+
+Theme: the Riders of Rohan answering the muster, each in turn.
+"""
+
+import pytest
+from statemachine.event import BoundEvent
+
+from statemachine import State
+from statemachine import StateChart
+
+
+class Muster(StateChart):
+    camp = State(initial=True)
+
+    call = camp.to.itself(internal=True, on="call_riders")
+    ride = camp.to.itself(internal=True, on="answer")
+
+    def __init__(self, *args, **kwargs):
+        self.answered = []
+        super().__init__(*args, **kwargs)
+
+    def call_riders(self):
+        for rider in range(20):
+            BoundEvent(id="ride", name="Ride", internal=True, _sm=self).put(rider=rider)
+
+    def answer(self, rider):
+        self.answered.append(rider)
+
+
+@pytest.fixture()
+def one_moment(monkeypatch):
+    """Stamp every trigger with the same time."""
+    monkeypatch.setattr("statemachine.event_data.time", lambda: 1000.0)
+
+
+@pytest.mark.timeout(10)
+@pytest.mark.usefixtures("one_moment")
+class TestEventsDueTogether:
+    async def test_internal_events_run_in_the_order_raised(self, sm_runner):
+        sm = await sm_runner.start(Muster)
+
+        await sm_runner.send(sm, "call")
+
+        assert sm.answered == list(range(20))
+
+    async def test_external_events_run_in_the_order_sent(self, sm_runner):
+        sm = await sm_runner.start(Muster)
+        for rider in range(20):
+            BoundEvent(id="ride", name="Ride", _sm=sm).put(rider=rider)
+
+        await sm_runner.processing_loop(sm)
+
+        assert sm.answered == list(range(20))

From b1c9dbc81ecaabbeb77c3adf30a3acda62b1c8af Mon Sep 17 00:00:00 2001
From: John Merrill <john@fairplay.ai>
Date: Sun, 4 Oct 2026 14:02:27 -0700
Subject: [PATCH 2/8] fix: measure event delays on the monotonic clock

Triggers were stamped with time.time() plus their delay and held until
time.time() reached the stamp, so a step of the wall clock moved every
due time: stepped back, it held even an undelayed event until the clock
caught up; stepped forward, it ran delayed events early. Stamps and the
due checks in both engines now use time.monotonic().

The new delayed-event test drives a fake monotonic clock in all three
modules, so a stamp on one clock checked against another would hold the
event and fail. The ordering tests now fix the monotonic clock instead
of the wall clock.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/releases/3.2.2.md           |  8 ++++++++
 statemachine/engines/async_.py   |  4 ++--
 statemachine/engines/sync.py     |  4 ++--
 statemachine/event_data.py       |  7 ++++---
 tests/test_event_ordering.py     |  2 +-
 tests/test_statechart_delayed.py | 29 +++++++++++++++++++++++++++++
 6 files changed, 46 insertions(+), 8 deletions(-)

diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md
index 2909bc51..6d350a2b 100644
--- a/docs/releases/3.2.2.md
+++ b/docs/releases/3.2.2.md
@@ -85,6 +85,14 @@ all stamped with the same time, ran as `0, 2, 6, 14, 19, 18, ...`. Each trigger
 the order it was created in, and triggers due at the same moment run in that order, on both
 engines and both queues.
 
+### Delays measured on the wall clock
+
+Every trigger was stamped with `time.time()` plus its delay, and the processing loop held a
+trigger until `time.time()` reached its stamp. A step of the wall clock, from a time sync or a
+change of the system time, moved every due time with it: stepped back, it held even an event
+sent with no delay until the clock caught up; stepped forward, it ran delayed events early.
+Delays are now measured on `time.monotonic()`, which no setting of the wall clock moves.
+
 ### `delay` and `internal` dropped from an explicit `Event`
 
 `Event(dark.to(lit), delay=50)` rebuilt the event without its `delay`, so
diff --git a/statemachine/engines/async_.py b/statemachine/engines/async_.py
index 0cd6c4a8..74871f6d 100644
--- a/statemachine/engines/async_.py
+++ b/statemachine/engines/async_.py
@@ -2,7 +2,7 @@
 import contextvars
 from collections.abc import Callable
 from itertools import chain
-from time import time
+from time import monotonic
 from typing import TYPE_CHECKING
 
 from ..event_data import EventData
@@ -426,7 +426,7 @@ async def processing_loop(  # noqa: C901
                     self.clear_cache()
                     took_events = True
                     external_event = self.external_queue.pop()
-                    current_time = time()
+                    current_time = monotonic()
                     if external_event.execution_time > current_time:
                         self.put(external_event, _delayed=True)
                         await asyncio.sleep(self.sm._loop_sleep_in_ms)
diff --git a/statemachine/engines/sync.py b/statemachine/engines/sync.py
index 627b51ae..055a6a15 100644
--- a/statemachine/engines/sync.py
+++ b/statemachine/engines/sync.py
@@ -1,5 +1,5 @@
+from time import monotonic
 from time import sleep
-from time import time
 from typing import TYPE_CHECKING
 
 from statemachine.event import BoundEvent
@@ -132,7 +132,7 @@ def processing_loop(self, caller_future=None):  # noqa: C901
                     self.clear_cache()
                     took_events = True
                     external_event = self.external_queue.pop()
-                    current_time = time()
+                    current_time = monotonic()
                     if external_event.execution_time > current_time:
                         self.put(external_event, _delayed=True)
                         sleep(self.sm._loop_sleep_in_ms)
diff --git a/statemachine/event_data.py b/statemachine/event_data.py
index 176e1f6e..94d30d69 100644
--- a/statemachine/event_data.py
+++ b/statemachine/event_data.py
@@ -1,7 +1,7 @@
 from dataclasses import dataclass
 from dataclasses import field
 from itertools import count
-from time import time
+from time import monotonic
 from typing import TYPE_CHECKING
 from typing import Any
 
@@ -29,7 +29,8 @@ class TriggerData:
     """
 
     execution_time: float = field(default=0.0)
-    """The time at which the :ref:`Event` should run."""
+    """The time at which the :ref:`Event` should run, on the :func:`time.monotonic` clock, so
+    that a step of the wall clock neither holds nor hastens it."""
 
     sequence: int = field(default=0, init=False)
     """The order this trigger was created in, which breaks a tie between triggers due at the
@@ -55,7 +56,7 @@ class TriggerData:
     def __post_init__(self):
         self.model = self.machine.model
         delay = self.event.delay if self.event and self.event.delay else 0
-        self.execution_time = time() + (delay / 1000)
+        self.execution_time = monotonic() + (delay / 1000)
         self.sequence = next(_sequence)
 
 
diff --git a/tests/test_event_ordering.py b/tests/test_event_ordering.py
index d746220e..5a442c44 100644
--- a/tests/test_event_ordering.py
+++ b/tests/test_event_ordering.py
@@ -35,7 +35,7 @@ def answer(self, rider):
 @pytest.fixture()
 def one_moment(monkeypatch):
     """Stamp every trigger with the same time."""
-    monkeypatch.setattr("statemachine.event_data.time", lambda: 1000.0)
+    monkeypatch.setattr("statemachine.event_data.monotonic", lambda: 1000.0)
 
 
 @pytest.mark.timeout(10)
diff --git a/tests/test_statechart_delayed.py b/tests/test_statechart_delayed.py
index f1f7c21b..c26711eb 100644
--- a/tests/test_statechart_delayed.py
+++ b/tests/test_statechart_delayed.py
@@ -97,3 +97,32 @@ class BeaconsOfGondor(StateChart):
         await asyncio.sleep(0.1)
         await sm_runner.processing_loop(sm)
         assert "lit" in sm.configuration_values
+
+    async def test_delay_is_measured_on_the_monotonic_clock(self, sm_runner, monkeypatch):
+        """A delay comes due by ``time.monotonic``, which a wall-clock step cannot move.
+
+        The trigger is stamped and checked on the same monotonic clock: here a fake one that
+        moves only when the test moves it. A stamp on one clock checked against another would
+        hold the beacon unlit.
+        """
+
+        class BeaconsOfGondor(StateChart):
+            dark = State(initial=True)
+            lit = State(final=True)
+
+            light = dark.to(lit)
+
+        now = [1000.0]
+        for module in (
+            "statemachine.event_data",
+            "statemachine.engines.sync",
+            "statemachine.engines.async_",
+        ):
+            monkeypatch.setattr(f"{module}.monotonic", lambda: now[0])
+        sm = await sm_runner.start(BeaconsOfGondor)
+        BoundEvent(id="light", name="Light", delay=50, _sm=sm).put()
+        now[0] += 0.05
+
+        await sm_runner.processing_loop(sm)
+
+        assert "lit" in sm.configuration_values

From 080835bdd887254efe9bd55f45a65dadfa143cdf Mon Sep 17 00:00:00 2001
From: John Merrill <john@fairplay.ai>
Date: Sun, 4 Oct 2026 14:06:37 -0700
Subject: [PATCH 3/8] fix: cancel a delayed event by marking it, not removing
 it

EventQueue.remove() filtered the PriorityQueue's underlying list. A
filtered heap is not always a heap, so the triggers left could come out
of due order: of ten delayed events due from 10 to 90 ms, cancelling the
first ran the event due at 50 ms before the one due at 40 ms. The
trigger's future was dropped with it, so on the async engine a coroutine
awaiting the cancelled event's send() was never answered.

cancel_event() now marks each trigger sent with the send_id as cancelled
and answers its future with None. The triggers stay where they are, so
the heap is never rearranged from outside, and is_empty() drops a
cancelled trigger when it reaches the head, before the processing loop
could check whether it is due.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/releases/3.2.2.md       |  12 +++
 docs/statechart.md           |   2 +
 statemachine/engines/base.py |  29 +++++---
 statemachine/event_data.py   |   5 ++
 statemachine/statemachine.py |   5 +-
 tests/test_cancel_event.py   | 139 +++++++++++++++++++++++++++++++++++
 6 files changed, 181 insertions(+), 11 deletions(-)
 create mode 100644 tests/test_cancel_event.py

diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md
index 6d350a2b..9ad2d1fa 100644
--- a/docs/releases/3.2.2.md
+++ b/docs/releases/3.2.2.md
@@ -85,6 +85,18 @@ all stamped with the same time, ran as `0, 2, 6, 14, 19, 18, ...`. Each trigger
 the order it was created in, and triggers due at the same moment run in that order, on both
 engines and both queues.
 
+### Cancelling a delayed event could disorder the others and strand its caller
+
+`cancel_event()` removed a trigger by filtering the list behind the queue's heap, and a filtered
+heap is not always a heap. The triggers left could then run out of due order: of ten delayed
+events due from 10 to 90 ms, cancelling the first left the one due at 50 ms running before the
+one due at 40 ms. The trigger's future went with it, so on the async engine a coroutine
+awaiting the cancelled event's `send()` was never answered.
+
+A cancelled trigger now stays where it is, marked as cancelled, and is dropped when it reaches
+the head of the queue, without the processing loop waiting for its due time. A coroutine
+awaiting it gets `None` as soon as it is cancelled.
+
 ### Delays measured on the wall clock
 
 Every trigger was stamped with `time.time()` plus its delay, and the processing loop held a
diff --git a/docs/statechart.md b/docs/statechart.md
index 5b92d6a3..e06ed251 100644
--- a/docs/statechart.md
+++ b/docs/statechart.md
@@ -113,6 +113,8 @@ sm.send("timeout", delay=5000, send_id="my_timeout")
 sm.cancel_event("my_timeout")
 ```
 
+On the async engine, a coroutine awaiting the `send()` of a cancelled event gets `None`.
+
 ```{note}
 The delay is **blocking** in the sync engine — the processing loop sleeps until the
 delay elapses, holding the calling thread. In the async engine, delays are scheduled
diff --git a/statemachine/engines/base.py b/statemachine/engines/base.py
index 14ae301d..a7617b97 100644
--- a/statemachine/engines/base.py
+++ b/statemachine/engines/base.py
@@ -2,6 +2,7 @@
 from collections.abc import Callable
 from dataclasses import dataclass
 from dataclasses import field
+from heapq import heappop
 from itertools import chain
 from queue import PriorityQueue
 from queue import Queue
@@ -41,7 +42,13 @@ def __repr__(self):
         return f"EventQueue({self.queue.queue!r}, size={self.queue.qsize()})"
 
     def is_empty(self):
-        return self.queue.qsize() == 0
+        """Whether the queue holds no trigger that has not been cancelled. Cancelled triggers at
+        the head are dropped first, so the :meth:`pop` that follows returns one that was not."""
+        with self.queue.mutex:
+            heap = self.queue.queue
+            while heap and heap[0].cancelled:
+                heappop(heap)
+            return not heap
 
     def put(self, trigger_data: TriggerData):
         """Put the trigger on the queue without blocking the caller."""
@@ -67,15 +74,17 @@ def reject_futures(self, exc: Exception):
                 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
+    def cancel(self, send_id: str):
+        """Cancel every trigger with this ``send_id``, and answer a caller awaiting one with
+        ``None``. Each stays where it is, so the heap keeps its order, until :meth:`is_empty`
+        drops it from the head."""
         with self.queue.mutex:
-            self.queue.queue = [
-                trigger_data
-                for trigger_data in self.queue.queue
-                if trigger_data.send_id != send_id
-            ]
+            for trigger_data in self.queue.queue:
+                if trigger_data.send_id == send_id:
+                    trigger_data.cancelled = True
+                    future = trigger_data.future
+                    if future is not None and not future.done():
+                        future.set_result(None)
 
 
 _ERROR_EXECUTION = "error.execution"
@@ -130,7 +139,7 @@ def clear(self):
 
     def cancel_event(self, send_id: str):
         """Cancel the event with the given send_id."""
-        self.external_queue.remove(send_id)
+        self.external_queue.cancel(send_id)
 
     def _on_error_handler(self) -> "Callable[[Exception], None] | None":
         """Return a per-block error handler, or ``None``.
diff --git a/statemachine/event_data.py b/statemachine/event_data.py
index 94d30d69..07747721 100644
--- a/statemachine/event_data.py
+++ b/statemachine/event_data.py
@@ -53,6 +53,11 @@ class TriggerData:
     result (or exception), allowing the caller to ``await`` it.
     """
 
+    cancelled: bool = field(default=False, compare=False, init=False)
+    """Whether :meth:`~statemachine.StateChart.cancel_event` cancelled this trigger. A cancelled
+    trigger stays in its queue, so the triggers around it keep their due order, and is dropped
+    when it reaches the head."""
+
     def __post_init__(self):
         self.model = self.machine.model
         delay = self.event.delay if self.event and self.event.delay else 0
diff --git a/statemachine/statemachine.py b/statemachine/statemachine.py
index 5277c4fc..82b9c210 100644
--- a/statemachine/statemachine.py
+++ b/statemachine/statemachine.py
@@ -507,7 +507,10 @@ def raise_(
         return self.send(event, *args, delay=delay, send_id=send_id, internal=True, **kwargs)
 
     def cancel_event(self, send_id: str):
-        """Cancel all the delayed events with the given ``send_id``."""
+        """Cancel all the delayed events with the given ``send_id``.
+
+        A caller awaiting one of them gets ``None``.
+        """
         self._engine.cancel_event(send_id)
 
     @property
diff --git a/tests/test_cancel_event.py b/tests/test_cancel_event.py
new file mode 100644
index 00000000..684b8cbd
--- /dev/null
+++ b/tests/test_cancel_event.py
@@ -0,0 +1,139 @@
+"""Cancelling delayed events with ``cancel_event()``.
+
+A cancelled trigger stays where it is in its queue, so the triggers around it keep their due
+order, and is dropped when it reaches the head. A caller awaiting it is answered with ``None``.
+
+Theme: signal fires, some of them called off before they are lit.
+"""
+
+import asyncio
+
+import pytest
+from statemachine.event import BoundEvent
+
+from statemachine import State
+from statemachine import StateChart
+
+_CLOCKS = ("statemachine.event_data", "statemachine.engines.sync", "statemachine.engines.async_")
+
+
+class Signals(StateChart):
+    idle = State(initial=True)
+
+    signal = idle.to.itself(internal=True, on="answer")
+
+    def __init__(self, *args, **kwargs):
+        self.answered = []
+        super().__init__(*args, **kwargs)
+
+    def answer(self, due):
+        self.answered.append(due)
+
+
+class AsyncSignals(Signals):
+    async def answer(self, due):
+        self.answered.append(due)
+
+
+@pytest.fixture()
+def clock(monkeypatch):
+    """A monotonic clock that moves only when the test moves it."""
+    now = [1000.0]
+    for module in _CLOCKS:
+        monkeypatch.setattr(f"{module}.monotonic", lambda: now[0])
+    return now
+
+
+def signal(sm, due, send_id):
+    """Queue a signal due in ``due`` ms, without running the processing loop."""
+    BoundEvent(id="signal", name="Signal", delay=due, _sm=sm).put(send_id=send_id, due=due)
+
+
+@pytest.mark.timeout(10)
+@pytest.mark.parametrize("cancelled", ["s0", "s2", "s9"])
+async def test_cancelling_a_delayed_event_keeps_the_rest_in_due_order(sm_runner, clock, cancelled):
+    sm = await sm_runner.start(Signals)
+    delays = [10, 50, 20, 60, 70, 30, 40, 80, 90, 15]
+    for number, delay in enumerate(delays):
+        signal(sm, delay, f"s{number}")
+
+    sm.cancel_event(cancelled)
+    clock[0] += 1.0
+    await sm_runner.processing_loop(sm)
+
+    assert sm.answered == sorted(set(delays) - {delays[int(cancelled[1:])]})
+
+
+@pytest.mark.timeout(10)
+async def test_cancelling_a_send_id_cancels_every_event_sent_with_it(sm_runner, clock):
+    sm = await sm_runner.start(Signals)
+    signal(sm, 10, "dawn")
+    signal(sm, 20, "dawn")
+    signal(sm, 30, "dusk")
+    signal(sm, 40, "dawn")
+
+    sm.cancel_event("dawn")
+    clock[0] += 1.0
+    await sm_runner.processing_loop(sm)
+
+    assert sm.answered == [30]
+
+
+@pytest.mark.timeout(10)
+async def test_cancelling_an_unknown_send_id_cancels_nothing(sm_runner, clock):
+    sm = await sm_runner.start(Signals)
+    signal(sm, 10, "dawn")
+    signal(sm, 20, "dusk")
+
+    sm.cancel_event("noon")
+    clock[0] += 1.0
+    await sm_runner.processing_loop(sm)
+
+    assert sm.answered == [10, 20]
+
+
+@pytest.mark.timeout(10)
+async def test_the_loop_never_waits_for_a_cancelled_event(sm_runner, clock, monkeypatch):
+    sm = await sm_runner.start(Signals)
+    signal(sm, 10, "dawn")
+    sm.cancel_event("dawn")
+
+    def checked():
+        raise AssertionError("the loop checked whether a cancelled event was due")
+
+    for module in _CLOCKS[1:]:
+        monkeypatch.setattr(f"{module}.monotonic", checked)
+    await sm_runner.processing_loop(sm)
+
+    assert sm.answered == []
+
+
+async def awaiting(sm):
+    """Send two delayed signals, each from its own task, and return both tasks.
+
+    The first task runs the processing loop, which waits on the clock for its signal. The
+    second signal, sent with ``send_id="dawn"`` and due first, is awaited by the second task.
+    """
+    holding = asyncio.create_task(sm.send("signal", delay=20, due=20))
+    await asyncio.sleep(0)
+    waiting = asyncio.create_task(sm.send("signal", delay=10, send_id="dawn", due=10))
+    await asyncio.sleep(0)
+    return holding, waiting
+
+
+@pytest.mark.timeout(10)
+@pytest.mark.parametrize("cancels", [1, 2])
+async def test_a_caller_awaiting_a_cancelled_event_gets_none(clock, cancels):
+    sm = AsyncSignals()
+    await sm.activate_initial_state()
+    holding, waiting = await awaiting(sm)
+
+    for _ in range(cancels):
+        sm.cancel_event("dawn")
+    await asyncio.sleep(0)
+
+    assert waiting.done()
+    assert waiting.result() is None
+    clock[0] += 1.0
+    await holding
+    assert sm.answered == [20]

From 480fe7a832dddcdaec80b52440ca2e44f80728d8 Mon Sep 17 00:00:00 2001
From: John Merrill <john@fairplay.ai>
Date: Sun, 4 Oct 2026 15:49:58 -0700
Subject: [PATCH 4/8] test: cover cancelling, delays and ordering at their
 edges

Tests drawn from the change's own code rather than its description:

- cancel_event() called from a callback, on both engines, and a caller
  that stops waiting, before or while it awaits, leaving its event
  cancellable;
- a send_id matched exactly: empty, whitespace, line endings, NUL, case,
  a zero-width space, a byte-order mark, composed against decomposed
  accents, and ids up to 2048 characters, none of which cancels an event
  sent without one;
- a cancel with nothing queued cancels nothing sent later, two ids
  cancelled in turn cancel each, and a cancelled trigger is marked
  rather than removed;
- the order test cancels the first, a middle and the last event due, and
  the last one queued;
- no trigger is stamped or checked on the wall clock, the engines read
  time.monotonic, and a delay in milliseconds comes due that many
  thousandths of a second later, for none, 0, 1, 50 and -50;
- events due together keep their order for 0, 1, 2 and 20 of them.

Each was checked against mutants of the code it covers: 31, every one
failed by at least one test without waiting for a timeout.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tests/test_cancel_event.py       | 200 ++++++++++++++++++++++++++++---
 tests/test_event_ordering.py     |  17 +--
 tests/test_statechart_delayed.py |  50 +++++++-
 3 files changed, 239 insertions(+), 28 deletions(-)

diff --git a/tests/test_cancel_event.py b/tests/test_cancel_event.py
index 684b8cbd..251d1868 100644
--- a/tests/test_cancel_event.py
+++ b/tests/test_cancel_event.py
@@ -9,10 +9,13 @@
 import asyncio
 
 import pytest
+from statemachine.engines import async_
+from statemachine.engines import sync
 from statemachine.event import BoundEvent
 
 from statemachine import State
 from statemachine import StateChart
+from statemachine import event_data
 
 _CLOCKS = ("statemachine.event_data", "statemachine.engines.sync", "statemachine.engines.async_")
 
@@ -21,6 +24,7 @@ class Signals(StateChart):
     idle = State(initial=True)
 
     signal = idle.to.itself(internal=True, on="answer")
+    stand_down = idle.to.itself(internal=True, on="call_off")
 
     def __init__(self, *args, **kwargs):
         self.answered = []
@@ -29,6 +33,9 @@ def __init__(self, *args, **kwargs):
     def answer(self, due):
         self.answered.append(due)
 
+    def call_off(self, which):
+        self.cancel_event(which)
+
 
 class AsyncSignals(Signals):
     async def answer(self, due):
@@ -44,13 +51,30 @@ def clock(monkeypatch):
     return now
 
 
-def signal(sm, due, send_id):
+def signal(sm, due, send_id=None):
     """Queue a signal due in ``due`` ms, without running the processing loop."""
-    BoundEvent(id="signal", name="Signal", delay=due, _sm=sm).put(send_id=send_id, due=due)
+    return BoundEvent(id="signal", name="Signal", delay=due, _sm=sm).put(send_id=send_id, due=due)
+
+
+def test_the_clock_fixture_moves_every_clock_the_engines_read(clock):
+    modules = (event_data, sync, async_)
+    assert [module.monotonic() for module in modules] == [1000.0, 1000.0, 1000.0]
+
+    clock[0] += 0.5
+
+    assert [module.monotonic() for module in modules] == [1000.5, 1000.5, 1000.5]
 
 
 @pytest.mark.timeout(10)
-@pytest.mark.parametrize("cancelled", ["s0", "s2", "s9"])
+@pytest.mark.parametrize(
+    "cancelled",
+    [
+        pytest.param("s0", id="first-due-and-first-queued"),
+        pytest.param("s6", id="due-in-the-middle"),
+        pytest.param("s8", id="last-due"),
+        pytest.param("s9", id="last-queued"),
+    ],
+)
 async def test_cancelling_a_delayed_event_keeps_the_rest_in_due_order(sm_runner, clock, cancelled):
     sm = await sm_runner.start(Signals)
     delays = [10, 50, 20, 60, 70, 30, 40, 80, 90, 15]
@@ -93,9 +117,121 @@ async def test_cancelling_an_unknown_send_id_cancels_nothing(sm_runner, clock):
 
 
 @pytest.mark.timeout(10)
-async def test_the_loop_never_waits_for_a_cancelled_event(sm_runner, clock, monkeypatch):
+async def test_cancelling_two_send_ids_in_turn_cancels_each_and_nothing_else(sm_runner, clock):
+    sm = await sm_runner.start(Signals)
+    signal(sm, 10, "dawn")
+    signal(sm, 20, "dusk")
+    signal(sm, 30, "noon")
+
+    sm.cancel_event("dawn")
+    sm.cancel_event("dusk")
+    clock[0] += 1.0
+    await sm_runner.processing_loop(sm)
+
+    assert sm.answered == [30]
+
+
+@pytest.mark.timeout(10)
+async def test_a_cancel_with_nothing_queued_does_not_cancel_what_is_sent_later(sm_runner, clock):
+    sm = await sm_runner.start(Signals)
+
+    sm.cancel_event("dawn")
+    signal(sm, 10, "dawn")
+    clock[0] += 1.0
+    await sm_runner.processing_loop(sm)
+
+    assert sm.answered == [10]
+
+
+@pytest.mark.timeout(10)
+@pytest.mark.parametrize(
+    ("sent", "cancelled"),
+    [
+        pytest.param("dawn", "", id="empty"),
+        pytest.param("dawn", " ", id="space"),
+        pytest.param("dawn", " dawn", id="leading-space"),
+        pytest.param("dawn", "dawn ", id="trailing-space"),
+        pytest.param("dawn", "dawn\n", id="newline"),
+        pytest.param("dawn\r\n", "dawn\n", id="crlf-and-lf"),
+        pytest.param("dawn", "Dawn", id="case"),
+        pytest.param("dawn", "dawn​", id="zero-width-space"),
+        pytest.param("dawn", "dawn", id="byte-order-mark"),
+        pytest.param("da\0wn", "da", id="up-to-a-nul"),
+        pytest.param("café", "café", id="composed-and-decomposed"),
+    ],
+)
+async def test_a_send_id_cancels_only_events_sent_with_exactly_that_id(
+    sm_runner, clock, sent, cancelled
+):
+    sm = await sm_runner.start(Signals)
+    signal(sm, 10, sent)
+
+    sm.cancel_event(cancelled)
+    clock[0] += 1.0
+    await sm_runner.processing_loop(sm)
+
+    assert sm.answered == [10]
+
+
+@pytest.mark.timeout(10)
+@pytest.mark.parametrize(
+    "send_id",
+    [
+        pytest.param("", id="empty"),
+        pytest.param(" ", id="space"),
+        pytest.param("dawn\n", id="newline"),
+        pytest.param("da\0wn", id="nul"),
+        pytest.param("café", id="decomposed"),
+        pytest.param("\U0001f525", id="astral"),
+        pytest.param("\u6681", id="cjk"),
+        pytest.param("\u202edawn", id="right-to-left-override"),
+        pytest.param("x" * 255, id="255-characters"),
+        pytest.param("x" * 256, id="256-characters"),
+        pytest.param("x" * 257, id="257-characters"),
+        pytest.param("x" * 1024, id="1024-characters"),
+        pytest.param("x" * 2048, id="2048-characters"),
+    ],
+)
+async def test_any_string_is_a_send_id_that_cancels_only_its_own_events(sm_runner, clock, send_id):
+    sm = await sm_runner.start(Signals)
+    signal(sm, 10, send_id)
+    signal(sm, 20)
+
+    sm.cancel_event(send_id)
+    clock[0] += 1.0
+    await sm_runner.processing_loop(sm)
+
+    assert sm.answered == [20]
+
+
+@pytest.mark.timeout(10)
+async def test_cancelling_marks_each_trigger_sent_with_the_id(sm_runner):
+    sm = await sm_runner.start(Signals)
+    dawn = signal(sm, 10, "dawn")
+    dusk = signal(sm, 20, "dusk")
+
+    sm.cancel_event("dawn")
+
+    assert (dawn.cancelled, dusk.cancelled) == (True, False)
+
+
+@pytest.mark.timeout(10)
+async def test_an_event_can_be_cancelled_from_a_callback(sm_runner, clock):
     sm = await sm_runner.start(Signals)
+    BoundEvent(id="stand_down", name="Stand down", _sm=sm).put(which="dawn")
     signal(sm, 10, "dawn")
+    signal(sm, 20, "dusk")
+
+    clock[0] += 1.0
+    await sm_runner.processing_loop(sm)
+
+    assert sm.answered == [20]
+
+
+@pytest.mark.timeout(10)
+async def test_the_loop_never_waits_for_a_cancelled_event(sm_runner, clock, monkeypatch):
+    sm = await sm_runner.start(Signals)
+    dawn = signal(sm, 10, "dawn")
     sm.cancel_event("dawn")
 
     def checked():
@@ -105,20 +241,31 @@ def checked():
         monkeypatch.setattr(f"{module}.monotonic", checked)
     await sm_runner.processing_loop(sm)
 
-    assert sm.answered == []
+    assert (dawn.cancelled, sm.answered) == (True, [])
 
 
-async def awaiting(sm):
-    """Send two delayed signals, each from its own task, and return both tasks.
+async def awaiting(sm, started=True):
+    """Queue two delayed signals, each awaited by its own task, and return the tasks.
 
     The first task runs the processing loop, which waits on the clock for its signal. The
-    second signal, sent with ``send_id="dawn"`` and due first, is awaited by the second task.
+    second signal, sent with ``send_id="dawn"`` and due first, is awaited through its trigger's
+    future by the second task, as ``send()`` awaits it, once that task has started. The
+    trigger is returned too.
     """
     holding = asyncio.create_task(sm.send("signal", delay=20, due=20))
     await asyncio.sleep(0)
-    waiting = asyncio.create_task(sm.send("signal", delay=10, send_id="dawn", due=10))
-    await asyncio.sleep(0)
-    return holding, waiting
+    dawn = signal(sm, 10, "dawn")
+    waiting = asyncio.create_task(sm._processing_loop(dawn.future))
+    if started:
+        await asyncio.sleep(0)
+    return holding, waiting, dawn
+
+
+def settled(future):
+    """How a settled future was settled: cancelled, or answered with a value."""
+    if future.cancelled():
+        return "cancelled"
+    return f"answered with {future.result()!r}"
 
 
 @pytest.mark.timeout(10)
@@ -126,14 +273,39 @@ async def awaiting(sm):
 async def test_a_caller_awaiting_a_cancelled_event_gets_none(clock, cancels):
     sm = AsyncSignals()
     await sm.activate_initial_state()
-    holding, waiting = await awaiting(sm)
+    holding, waiting, dawn = await awaiting(sm)
 
     for _ in range(cancels):
         sm.cancel_event("dawn")
     await asyncio.sleep(0)
 
-    assert waiting.done()
-    assert waiting.result() is None
+    assert (settled(dawn.future), settled(waiting)) == ("answered with None", "answered with None")
     clock[0] += 1.0
     await holding
     assert sm.answered == [20]
+
+
+@pytest.mark.timeout(10)
+@pytest.mark.parametrize(
+    ("started", "settled_as"),
+    [
+        pytest.param(False, "answered with None", id="before-it-runs"),
+        pytest.param(True, "cancelled", id="while-it-awaits"),
+    ],
+)
+async def test_a_caller_that_stops_waiting_leaves_its_event_cancellable(
+    clock, started, settled_as
+):
+    sm = AsyncSignals()
+    await sm.activate_initial_state()
+    holding, waiting, dawn = await awaiting(sm, started)
+
+    waiting.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await waiting
+    sm.cancel_event("dawn")
+    clock[0] += 1.0
+    await holding
+
+    assert settled(dawn.future) == settled_as
+    assert sm.answered == [20]
diff --git a/tests/test_event_ordering.py b/tests/test_event_ordering.py
index 5a442c44..a8b3ce32 100644
--- a/tests/test_event_ordering.py
+++ b/tests/test_event_ordering.py
@@ -24,8 +24,8 @@ def __init__(self, *args, **kwargs):
         self.answered = []
         super().__init__(*args, **kwargs)
 
-    def call_riders(self):
-        for rider in range(20):
+    def call_riders(self, riders):
+        for rider in range(riders):
             BoundEvent(id="ride", name="Ride", internal=True, _sm=self).put(rider=rider)
 
     def answer(self, rider):
@@ -40,19 +40,20 @@ def one_moment(monkeypatch):
 
 @pytest.mark.timeout(10)
 @pytest.mark.usefixtures("one_moment")
+@pytest.mark.parametrize("riders", [0, 1, 2, 20])
 class TestEventsDueTogether:
-    async def test_internal_events_run_in_the_order_raised(self, sm_runner):
+    async def test_internal_events_run_in_the_order_raised(self, sm_runner, riders):
         sm = await sm_runner.start(Muster)
 
-        await sm_runner.send(sm, "call")
+        await sm_runner.send(sm, "call", riders=riders)
 
-        assert sm.answered == list(range(20))
+        assert sm.answered == list(range(riders))
 
-    async def test_external_events_run_in_the_order_sent(self, sm_runner):
+    async def test_external_events_run_in_the_order_sent(self, sm_runner, riders):
         sm = await sm_runner.start(Muster)
-        for rider in range(20):
+        for rider in range(riders):
             BoundEvent(id="ride", name="Ride", _sm=sm).put(rider=rider)
 
         await sm_runner.processing_loop(sm)
 
-        assert sm.answered == list(range(20))
+        assert sm.answered == list(range(riders))
diff --git a/tests/test_statechart_delayed.py b/tests/test_statechart_delayed.py
index c26711eb..2661b750 100644
--- a/tests/test_statechart_delayed.py
+++ b/tests/test_statechart_delayed.py
@@ -8,13 +8,19 @@
 """
 
 import asyncio
+import time
 
 import pytest
+from statemachine.engines import async_
+from statemachine.engines import sync
 from statemachine.event import BoundEvent
 
 from statemachine import Event
 from statemachine import State
 from statemachine import StateChart
+from statemachine import event_data
+
+_CLOCKS = ("statemachine.event_data", "statemachine.engines.sync", "statemachine.engines.async_")
 
 
 @pytest.mark.timeout(10)
@@ -103,7 +109,7 @@ async def test_delay_is_measured_on_the_monotonic_clock(self, sm_runner, monkeyp
 
         The trigger is stamped and checked on the same monotonic clock: here a fake one that
         moves only when the test moves it. A stamp on one clock checked against another would
-        hold the beacon unlit.
+        hold the beacon unlit, and the wall clock fails the test if anything reads it.
         """
 
         class BeaconsOfGondor(StateChart):
@@ -112,13 +118,14 @@ class BeaconsOfGondor(StateChart):
 
             light = dark.to(lit)
 
+        def wall_clock():
+            raise AssertionError("a trigger was stamped or checked on the wall clock")
+
+        monkeypatch.setattr("time.time", wall_clock)
         now = [1000.0]
-        for module in (
-            "statemachine.event_data",
-            "statemachine.engines.sync",
-            "statemachine.engines.async_",
-        ):
+        for module in _CLOCKS:
             monkeypatch.setattr(f"{module}.monotonic", lambda: now[0])
+            monkeypatch.setattr(f"{module}.time", wall_clock, raising=False)
         sm = await sm_runner.start(BeaconsOfGondor)
         BoundEvent(id="light", name="Light", delay=50, _sm=sm).put()
         now[0] += 0.05
@@ -126,3 +133,34 @@ class BeaconsOfGondor(StateChart):
         await sm_runner.processing_loop(sm)
 
         assert "lit" in sm.configuration_values
+
+    def test_triggers_are_stamped_and_checked_on_the_monotonic_clock(self):
+        clocks = (event_data.monotonic, sync.monotonic, async_.monotonic)
+
+        assert clocks == (time.monotonic, time.monotonic, time.monotonic)
+
+    @pytest.mark.parametrize(
+        ("delay", "due"),
+        [
+            pytest.param(None, 1000.0, id="none"),
+            pytest.param(0, 1000.0, id="zero"),
+            pytest.param(1, 1000.001, id="one-millisecond"),
+            pytest.param(50, 1000.05, id="fifty-milliseconds"),
+            pytest.param(-50, 999.95, id="negative-already-due"),
+        ],
+    )
+    def test_a_delay_in_milliseconds_comes_due_that_many_thousandths_later(
+        self, monkeypatch, delay, due
+    ):
+        class BeaconsOfGondor(StateChart):
+            dark = State(initial=True)
+            lit = State(final=True)
+
+            light = dark.to(lit)
+
+        monkeypatch.setattr("statemachine.event_data.monotonic", lambda: 1000.0)
+        sm = BeaconsOfGondor()
+
+        trigger = BoundEvent(id="light", name="Light", delay=delay, _sm=sm).put()
+
+        assert trigger.execution_time == pytest.approx(due, abs=1e-9)

From b2367c434d436dc7c67ef328bf65a58731fabb4d Mon Sep 17 00:00:00 2001
From: John Merrill <john@fairplay.ai>
Date: Sun, 4 Oct 2026 16:04:04 -0700
Subject: [PATCH 5/8] test: run the clock fixtures automatically, and escape
 the odd send_ids

SonarCloud's S9083 asks for `@pytest.fixture` without parentheses, and
this project's ruff configuration (fixture-parentheses = true) asks for
them. The two clock fixtures are now `autouse=True`, which both accept:
every test in each module runs on the substituted clock, so none reads
the real one, and a test that moves the clock asks for it by name.

The send_ids that held a zero-width space, a byte-order mark and
accents as literal characters are escapes again, so the file is ASCII.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tests/test_cancel_event.py   | 26 +++++++++++++-------------
 tests/test_event_ordering.py |  7 +++----
 2 files changed, 16 insertions(+), 17 deletions(-)

diff --git a/tests/test_cancel_event.py b/tests/test_cancel_event.py
index 251d1868..a1e1b220 100644
--- a/tests/test_cancel_event.py
+++ b/tests/test_cancel_event.py
@@ -42,9 +42,9 @@ async def answer(self, due):
         self.answered.append(due)
 
 
-@pytest.fixture()
+@pytest.fixture(autouse=True)
 def clock(monkeypatch):
-    """A monotonic clock that moves only when the test moves it."""
+    """Run every test here on a monotonic clock that moves only when the test moves it."""
     now = [1000.0]
     for module in _CLOCKS:
         monkeypatch.setattr(f"{module}.monotonic", lambda: now[0])
@@ -56,7 +56,7 @@ def signal(sm, due, send_id=None):
     return BoundEvent(id="signal", name="Signal", delay=due, _sm=sm).put(send_id=send_id, due=due)
 
 
-def test_the_clock_fixture_moves_every_clock_the_engines_read(clock):
+def test_every_engine_reads_the_test_clock(clock):
     modules = (event_data, sync, async_)
     assert [module.monotonic() for module in modules] == [1000.0, 1000.0, 1000.0]
 
@@ -75,7 +75,7 @@ def test_the_clock_fixture_moves_every_clock_the_engines_read(clock):
         pytest.param("s9", id="last-queued"),
     ],
 )
-async def test_cancelling_a_delayed_event_keeps_the_rest_in_due_order(sm_runner, clock, cancelled):
+async def test_cancelling_a_delayed_event_keeps_the_rest_in_due_order(sm_runner, cancelled, clock):
     sm = await sm_runner.start(Signals)
     delays = [10, 50, 20, 60, 70, 30, 40, 80, 90, 15]
     for number, delay in enumerate(delays):
@@ -154,14 +154,14 @@ async def test_a_cancel_with_nothing_queued_does_not_cancel_what_is_sent_later(s
         pytest.param("dawn", "dawn\n", id="newline"),
         pytest.param("dawn\r\n", "dawn\n", id="crlf-and-lf"),
         pytest.param("dawn", "Dawn", id="case"),
-        pytest.param("dawn", "dawn​", id="zero-width-space"),
-        pytest.param("dawn", "dawn", id="byte-order-mark"),
+        pytest.param("dawn", "dawn\u200b", id="zero-width-space"),
+        pytest.param("dawn", "\ufeffdawn", id="byte-order-mark"),
         pytest.param("da\0wn", "da", id="up-to-a-nul"),
-        pytest.param("café", "café", id="composed-and-decomposed"),
+        pytest.param("caf\u00e9", "cafe\u0301", id="composed-and-decomposed"),
     ],
 )
 async def test_a_send_id_cancels_only_events_sent_with_exactly_that_id(
-    sm_runner, clock, sent, cancelled
+    sm_runner, sent, cancelled, clock
 ):
     sm = await sm_runner.start(Signals)
     signal(sm, 10, sent)
@@ -181,7 +181,7 @@ async def test_a_send_id_cancels_only_events_sent_with_exactly_that_id(
         pytest.param(" ", id="space"),
         pytest.param("dawn\n", id="newline"),
         pytest.param("da\0wn", id="nul"),
-        pytest.param("café", id="decomposed"),
+        pytest.param("cafe\u0301", id="decomposed"),
         pytest.param("\U0001f525", id="astral"),
         pytest.param("\u6681", id="cjk"),
         pytest.param("\u202edawn", id="right-to-left-override"),
@@ -192,7 +192,7 @@ async def test_a_send_id_cancels_only_events_sent_with_exactly_that_id(
         pytest.param("x" * 2048, id="2048-characters"),
     ],
 )
-async def test_any_string_is_a_send_id_that_cancels_only_its_own_events(sm_runner, clock, send_id):
+async def test_any_string_is_a_send_id_that_cancels_only_its_own_events(sm_runner, send_id, clock):
     sm = await sm_runner.start(Signals)
     signal(sm, 10, send_id)
     signal(sm, 20)
@@ -229,7 +229,7 @@ async def test_an_event_can_be_cancelled_from_a_callback(sm_runner, clock):
 
 
 @pytest.mark.timeout(10)
-async def test_the_loop_never_waits_for_a_cancelled_event(sm_runner, clock, monkeypatch):
+async def test_the_loop_never_waits_for_a_cancelled_event(sm_runner, monkeypatch):
     sm = await sm_runner.start(Signals)
     dawn = signal(sm, 10, "dawn")
     sm.cancel_event("dawn")
@@ -270,7 +270,7 @@ def settled(future):
 
 @pytest.mark.timeout(10)
 @pytest.mark.parametrize("cancels", [1, 2])
-async def test_a_caller_awaiting_a_cancelled_event_gets_none(clock, cancels):
+async def test_a_caller_awaiting_a_cancelled_event_gets_none(cancels, clock):
     sm = AsyncSignals()
     await sm.activate_initial_state()
     holding, waiting, dawn = await awaiting(sm)
@@ -294,7 +294,7 @@ async def test_a_caller_awaiting_a_cancelled_event_gets_none(clock, cancels):
     ],
 )
 async def test_a_caller_that_stops_waiting_leaves_its_event_cancellable(
-    clock, started, settled_as
+    started, settled_as, clock
 ):
     sm = AsyncSignals()
     await sm.activate_initial_state()
diff --git a/tests/test_event_ordering.py b/tests/test_event_ordering.py
index a8b3ce32..f10842b8 100644
--- a/tests/test_event_ordering.py
+++ b/tests/test_event_ordering.py
@@ -32,14 +32,13 @@ def answer(self, rider):
         self.answered.append(rider)
 
 
-@pytest.fixture()
-def one_moment(monkeypatch):
-    """Stamp every trigger with the same time."""
+@pytest.fixture(autouse=True)
+def stamped_alike(monkeypatch):
+    """Stamp every trigger in these tests with the same time."""
     monkeypatch.setattr("statemachine.event_data.monotonic", lambda: 1000.0)
 
 
 @pytest.mark.timeout(10)
-@pytest.mark.usefixtures("one_moment")
 @pytest.mark.parametrize("riders", [0, 1, 2, 20])
 class TestEventsDueTogether:
     async def test_internal_events_run_in_the_order_raised(self, sm_runner, riders):

From 8e351b20536f102a8e6458d3468cbbb4c6d32fd4 Mon Sep 17 00:00:00 2001
From: John Merrill <john@fairplay.ai>
Date: Sun, 4 Oct 2026 16:37:54 -0700
Subject: [PATCH 6/8] test: queue a delayed event from a callback, and stop
 waiting for one

A delayed event queued from a callback, while the processing loop runs,
comes due on a clock that moves 10 ms with every reading, on both
engines. A coroutine that stops awaiting a delayed event, before or
while it awaits, does not stop the event, which fires in due order.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tests/test_statechart_delayed.py | 68 ++++++++++++++++++++++++++++++++
 1 file changed, 68 insertions(+)

diff --git a/tests/test_statechart_delayed.py b/tests/test_statechart_delayed.py
index 2661b750..635fa974 100644
--- a/tests/test_statechart_delayed.py
+++ b/tests/test_statechart_delayed.py
@@ -9,6 +9,7 @@
 
 import asyncio
 import time
+from itertools import count
 
 import pytest
 from statemachine.engines import async_
@@ -164,3 +165,70 @@ class BeaconsOfGondor(StateChart):
         trigger = BoundEvent(id="light", name="Light", delay=delay, _sm=sm).put()
 
         assert trigger.execution_time == pytest.approx(due, abs=1e-9)
+
+    async def test_a_delayed_event_queued_from_a_callback_fires_when_due(
+        self, sm_runner, monkeypatch
+    ):
+        """The callback runs inside the processing loop, which then waits on the event it queued.
+
+        Every reading of the clock is 10 ms later than the last, so the beacon comes due after
+        a few turns of the loop however long each takes.
+        """
+
+        class BeaconsOfGondor(StateChart):
+            dark = State(initial=True)
+            first_lit = State()
+            all_lit = State(final=True)
+
+            light_first = dark.to(first_lit, after="spread")
+            light_all = first_lit.to(all_lit)
+
+            def spread(self):
+                BoundEvent(id="light_all", name="Light all", delay=50, _sm=self).put()
+
+        readings = count()
+        for module in _CLOCKS:
+            monkeypatch.setattr(f"{module}.monotonic", lambda: 1000.0 + next(readings) / 100)
+        sm = await sm_runner.start(BeaconsOfGondor)
+
+        await sm_runner.send(sm, "light_first")
+
+        assert sm.configuration_values == {"all_lit"}
+
+    @pytest.mark.parametrize(
+        "started",
+        [pytest.param(False, id="before-it-runs"), pytest.param(True, id="while-it-awaits")],
+    )
+    async def test_a_caller_that_stops_waiting_does_not_stop_its_delayed_event(
+        self, monkeypatch, started
+    ):
+        class Beacon(StateChart):
+            dark = State(initial=True)
+
+            signal = dark.to.itself(internal=True, on="flare")
+
+            def __init__(self, *args, **kwargs):
+                self.flares = []
+                super().__init__(*args, **kwargs)
+
+            async def flare(self, which):
+                self.flares.append(which)
+
+        now = [1000.0]
+        for module in _CLOCKS:
+            monkeypatch.setattr(f"{module}.monotonic", lambda: now[0])
+        sm = Beacon()
+        await sm.activate_initial_state()
+        holding = asyncio.create_task(sm.send("signal", delay=20, which="second"))
+        await asyncio.sleep(0)
+        waiting = asyncio.create_task(sm.send("signal", delay=10, which="first"))
+        if started:
+            await asyncio.sleep(0)
+
+        waiting.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await waiting
+        now[0] += 1.0
+        await holding
+
+        assert sm.flares == ["first", "second"]

From 3a1d5cf94d8667bf8d23c81810ca46abb0980d9d Mon Sep 17 00:00:00 2001
From: John Merrill <john@fairplay.ai>
Date: Sun, 4 Oct 2026 16:38:17 -0700
Subject: [PATCH 7/8] refactor: fix what the engines' suppressions covered, and
 drop them

No `type: ignore`, `noqa`, `pragma: no cover` or `pragma: no branch` is
left in base.py, sync.py, async_.py or statemachine.py.

- BaseEngine declared select_transitions, select_eventless_transitions,
  _select_transitions, _first_transition_that_matches, microstep and
  _exit_states as synchronous, and AsyncEngine overrode them with
  coroutines, which mypy rejected. They now belong to SyncEngine, and
  BaseEngine keeps what both engines share.
- Each processing loop is split into its macrostep, the external events
  and one external event, and add_descendant_states_to_enter into its
  history, region and initial-state parts.
- Removed as unreachable: BaseEngine.empty and pop, which nothing
  called; the _run_microstep wrappers, since microstep already sends
  every error through _handle_error; the second pass over the internal
  queue, since the macrostep empties it and spawning invocations puts
  events only on the external queue; and the checks for an exited state
  being None.
- is_in_final_state takes the parallel case first, so an atomic state
  needs no branch of its own.
- StateChart: a cast for the default Model, vars(self).update, an
  assert that narrows initial_state, and _repr_svg_ through the diagram
  formatter, which renders SVG the same way.

New tests cover the two lines the pragmas hid: stop() still stops the
engine when cancelling invocations fails, and a transition into a
parallel state and a state inside one of its regions enters that state
and not the region's initial one.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 statemachine/engines/async_.py       | 248 +++++++++---------
 statemachine/engines/base.py         | 377 +++++++++++----------------
 statemachine/engines/sync.py         | 308 ++++++++++++++--------
 statemachine/statemachine.py         |  17 +-
 tests/scxml/test_parallel_regions.py |  38 +++
 tests/test_invoke.py                 |  21 ++
 6 files changed, 543 insertions(+), 466 deletions(-)
 create mode 100644 tests/scxml/test_parallel_regions.py

diff --git a/statemachine/engines/async_.py b/statemachine/engines/async_.py
index 74871f6d..e3fe65aa 100644
--- a/statemachine/engines/async_.py
+++ b/statemachine/engines/async_.py
@@ -113,7 +113,7 @@ async def _conditions_match(self, transition: "Transition", trigger_data: Trigge
             transition.cond.key, *args, on_error=on_error, **kwargs
         )
 
-    async def _first_transition_that_matches(  # type: ignore[override]
+    async def _first_transition_that_matches(
         self,
         state: State,
         trigger_data: TriggerData,
@@ -130,7 +130,7 @@ async def _first_transition_that_matches(  # type: ignore[override]
                     return transition
         return None
 
-    async def _select_transitions(  # type: ignore[override]
+    async def _select_transitions(
         self, trigger_data: TriggerData, predicate: Callable
     ) -> "OrderedSet[Transition]":
         enabled_transitions: "OrderedSet[Transition]" = OrderedSet()
@@ -147,7 +147,7 @@ async def _select_transitions(  # type: ignore[override]
     async def select_eventless_transitions(self, trigger_data: TriggerData):
         return await self._select_transitions(trigger_data, lambda t, _e: t.is_eventless)
 
-    async def select_transitions(self, trigger_data: TriggerData) -> "OrderedSet[Transition]":  # type: ignore[override]
+    async def select_transitions(self, trigger_data: TriggerData) -> "OrderedSet[Transition]":
         return await self._select_transitions(trigger_data, lambda t, e: t.match(e))
 
     async def _execute_transition_content(
@@ -172,7 +172,7 @@ async def _execute_transition_content(
 
         return result
 
-    async def _exit_states(  # type: ignore[override]
+    async def _exit_states(
         self, enabled_transitions: "list[Transition]", trigger_data: TriggerData
     ) -> "OrderedSet[State]":
         ordered_states, result = self._prepare_exit_states(enabled_transitions)
@@ -180,24 +180,22 @@ async def _exit_states(  # type: ignore[override]
 
         for info in ordered_states:
             # Cancel invocations for this state before executing exit handlers.
-            if info.state is not None:  # pragma: no branch
-                self._invoke_manager.cancel_for_state(info.state)
+            self._invoke_manager.cancel_for_state(info.state)
 
             args, kwargs = await self._get_args_kwargs(
                 info.transition, trigger_data, source=info.state
             )
 
-            if info.state is not None:  # pragma: no branch
-                self._debug("%s Exiting state: %s", self._log_id, info.state)
-                await self.sm._callbacks.async_call(
-                    info.state.exit.key, *args, on_error=on_error, **kwargs
-                )
+            self._debug("%s Exiting state: %s", self._log_id, info.state)
+            await self.sm._callbacks.async_call(
+                info.state.exit.key, *args, on_error=on_error, **kwargs
+            )
 
             self._remove_state_from_configuration(info.state)
 
         return result
 
-    async def _enter_states(  # noqa: C901
+    async def _enter_states(
         self,
         enabled_transitions: "list[Transition]",
         trigger_data: TriggerData,
@@ -329,19 +327,6 @@ async def microstep(self, transitions: "list[Transition]", trigger_data: Trigger
 
     # --- Engine loop ---
 
-    async def _run_microstep(self, enabled_transitions, trigger_data):  # pragma: no cover
-        """Run a microstep for internal/eventless transitions with error handling.
-
-        Note: microstep() handles its own errors internally, so this try/except
-        is a safety net that is not expected to be reached in normal operation.
-        """
-        try:
-            await self.microstep(list(enabled_transitions), trigger_data)
-        except InvalidDefinition:
-            raise
-        except Exception as e:
-            self._handle_error(e, trigger_data)
-
     async def activate_initial_state(self, **kwargs):
         """Activate the initial state.
 
@@ -354,14 +339,12 @@ async def activate_initial_state(self, **kwargs):
         """
         return await self.processing_loop()
 
-    async def processing_loop(  # noqa: C901
-        self, caller_future: "asyncio.Future[object] | None" = None
-    ):
-        """Process event triggers with the 3-phase macrostep architecture.
+    async def processing_loop(self, caller_future: "asyncio.Future[object] | None" = None):
+        """Process event triggers, a macrostep at a time.
 
-        Phase 1: Eventless transitions + internal queue until quiescence.
-        Phase 2: Remaining internal events (safety net for invoke-generated events).
-        Phase 3: External events.
+        Each macrostep takes eventless transitions and internal events until neither enables a
+        transition, spawns the invocations of the states it entered, and then takes the
+        external events that are due.
 
         When ``caller_future`` is provided, the caller can ``await`` it to
         receive its own event's result — even if another coroutine holds the
@@ -381,110 +364,14 @@ async def processing_loop(  # noqa: C901
             took_events = True
             while took_events and self.running:
                 self.clear_cache()
-                took_events = False
-                macrostep_done = False
-
-                # Phase 1: eventless transitions and internal events
-                while not macrostep_done:
-                    self._microstep_count = 0
-                    self._debug(
-                        "%s Macrostep %d: eventless/internal queue",
-                        self._log_id,
-                        self._macrostep_count,
-                    )
-
-                    self.clear_cache()
-                    internal_event = TriggerData(self.sm, event=None)  # null object for eventless
-                    enabled_transitions = await self.select_eventless_transitions(internal_event)
-                    if not enabled_transitions:
-                        if self.internal_queue.is_empty():
-                            macrostep_done = True
-                        else:
-                            internal_event = self.internal_queue.pop()
-                            enabled_transitions = await self.select_transitions(internal_event)
-                    if enabled_transitions:
-                        self._debug(
-                            "%s Enabled transitions: %s", self._log_id, enabled_transitions
-                        )
-                        took_events = True
-                        await self._run_microstep(enabled_transitions, internal_event)
+                took_events = await self._run_macrostep()
 
                 # Spawn invoke handlers for states entered during this macrostep.
                 await self._invoke_manager.spawn_pending_async()
                 self._check_root_final_state()
 
-                # Phase 2: remaining internal events
-                while not self.internal_queue.is_empty():  # pragma: no cover
-                    internal_event = self.internal_queue.pop()
-                    enabled_transitions = await self.select_transitions(internal_event)
-                    if enabled_transitions:
-                        await self._run_microstep(enabled_transitions, internal_event)
-
-                # Phase 3: external events
-                self._debug("%s Macrostep %d: external queue", self._log_id, self._macrostep_count)
-                while not self.external_queue.is_empty():
-                    self.clear_cache()
-                    took_events = True
-                    external_event = self.external_queue.pop()
-                    current_time = monotonic()
-                    if external_event.execution_time > current_time:
-                        self.put(external_event, _delayed=True)
-                        await asyncio.sleep(self.sm._loop_sleep_in_ms)
-                        # Break to Phase 1 so internal events and eventless
-                        # transitions can be processed while we wait.
-                        break
-
-                    self._macrostep_count += 1
-                    self._microstep_count = 0
-                    self._debug(
-                        "%s macrostep %d: event=%s",
-                        self._log_id,
-                        self._macrostep_count,
-                        external_event.event,
-                    )
-
-                    # Handle lazy initial state activation.
-                    # Break out of phase 3 so the outer loop restarts from phase 1
-                    # (eventless/internal), ensuring internal events queued during
-                    # initial entry are processed before any external events.
-                    if external_event.event == "__initial__":
-                        transitions = self._initial_transitions(external_event)
-                        await self._enter_states(
-                            transitions, external_event, OrderedSet(), OrderedSet()
-                        )
-                        break
-
-                    # 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)
-                            if first_result is self._sentinel:
-                                first_result = result
-                        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
+                took_external_events, first_result = await self._take_external_events(first_result)
+                took_events = took_events or took_external_events
 
         except Exception as exc:
             if caller_future is not None:
@@ -510,6 +397,105 @@ async def processing_loop(  # noqa: C901
             return await caller_future
         return result
 
+    async def _run_macrostep(self) -> bool:
+        """Take eventless transitions and internal events until neither enables a transition.
+
+        Returns whether any transition was taken. The internal queue is empty on return.
+        """
+        took_events = False
+        while True:
+            self._microstep_count = 0
+            self._debug(
+                "%s Macrostep %d: eventless/internal queue",
+                self._log_id,
+                self._macrostep_count,
+            )
+
+            self.clear_cache()
+            internal_event = TriggerData(self.sm, event=None)  # null object for eventless
+            enabled_transitions = await self.select_eventless_transitions(internal_event)
+            if not enabled_transitions:
+                if self.internal_queue.is_empty():
+                    return took_events
+                internal_event = self.internal_queue.pop()
+                enabled_transitions = await self.select_transitions(internal_event)
+            if enabled_transitions:
+                self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions)
+                took_events = True
+                await self.microstep(list(enabled_transitions), internal_event)
+
+    async def _take_external_events(self, first_result):
+        """Take external events in due order until the queue is empty or holds one not yet due.
+
+        Returns whether any event was taken from the queue, and ``first_result`` updated with
+        the result of the first event that enabled a transition.
+        """
+        self._debug("%s Macrostep %d: external queue", self._log_id, self._macrostep_count)
+        took_events = False
+        while not self.external_queue.is_empty():
+            self.clear_cache()
+            took_events = True
+            external_event = self.external_queue.pop()
+            current_time = monotonic()
+            if external_event.execution_time > current_time:
+                self.put(external_event, _delayed=True)
+                await asyncio.sleep(self.sm._loop_sleep_in_ms)
+                # Break to the macrostep so internal events and eventless
+                # transitions can be processed while we wait.
+                break
+
+            self._macrostep_count += 1
+            self._microstep_count = 0
+            self._debug(
+                "%s macrostep %d: event=%s",
+                self._log_id,
+                self._macrostep_count,
+                external_event.event,
+            )
+
+            # Handle lazy initial state activation.
+            # Break to the macrostep so internal events queued during
+            # initial entry are processed before any external events.
+            if external_event.event == "__initial__":
+                transitions = self._initial_transitions(external_event)
+                await self._enter_states(transitions, external_event, OrderedSet(), OrderedSet())
+                break
+
+            result = await self._take_external_event(external_event)
+            if first_result is self._sentinel:
+                first_result = result
+        return took_events, first_result
+
+    async def _take_external_event(self, external_event: TriggerData):
+        """Run one external event and settle its future.
+
+        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:
+                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)
+            return self._sentinel
+        except Exception as exc:
+            self._reject_future(event_future, exc)
+            self._reject_pending_futures(exc)
+            self.clear()
+            raise
+
     async def enabled_events(self, *args, **kwargs):
         sm = self.sm
         enabled = {}
diff --git a/statemachine/engines/base.py b/statemachine/engines/base.py
index a7617b97..796eed36 100644
--- a/statemachine/engines/base.py
+++ b/statemachine/engines/base.py
@@ -3,7 +3,6 @@
 from dataclasses import dataclass
 from dataclasses import field
 from heapq import heappop
-from itertools import chain
 from queue import PriorityQueue
 from queue import Queue
 from threading import Lock
@@ -34,6 +33,13 @@ class StateTransition:
     state: State
 
 
+#: What entering states needs, worked out before any callback runs: the states in entry order,
+#: those entered by default, the default history content, and the new configuration.
+_EntryPlan = tuple[
+    list[StateTransition], OrderedSet[StateTransition], dict[str, Any], OrderedSet[State]
+]
+
+
 class EventQueue:
     def __init__(self):
         self.queue: Queue = PriorityQueue()
@@ -106,9 +112,6 @@ def __init__(self, sm: "StateChart"):
         self._debug = logger.debug if logger.isEnabledFor(logging.DEBUG) else lambda *a, **k: None
         self._root_parallel_final_pending: "State | None" = None
 
-    def empty(self):  # pragma: no cover
-        return self.external_queue.is_empty()
-
     def clear_cache(self):
         """Clears the cache. Should be called at the start of each processing loop."""
         self._cache.clear()
@@ -131,9 +134,6 @@ def put(self, trigger_data: TriggerData, internal: bool = False, _delayed: bool
                 "internal" if internal else "external",
             )
 
-    def pop(self):  # pragma: no cover
-        return self.external_queue.pop()
-
     def clear(self):
         self.external_queue.clear()
 
@@ -333,100 +333,6 @@ def get_effective_target_states(self, transition: Transition) -> OrderedSet[Stat
 
         return targets
 
-    def select_eventless_transitions(self, trigger_data: TriggerData):
-        """
-        Select the eventless transitions that match the trigger data.
-        """
-        return self._select_transitions(trigger_data, lambda t, _e: t.is_eventless)
-
-    def select_transitions(self, trigger_data: TriggerData) -> OrderedSet[Transition]:
-        """
-        Select the transitions that match the trigger data.
-        """
-        return self._select_transitions(trigger_data, lambda t, e: t.match(e))
-
-    def _first_transition_that_matches(
-        self,
-        state: State,
-        trigger_data: TriggerData,
-        predicate: Callable,
-    ) -> "Transition | None":
-        for s in chain([state], state.ancestors()):
-            transition: Transition
-            for transition in s.transitions:
-                if (
-                    not transition.initial
-                    and predicate(transition, trigger_data.event)
-                    and self._conditions_match(transition, trigger_data)
-                ):
-                    return transition
-        return None
-
-    def _select_transitions(
-        self, trigger_data: TriggerData, predicate: Callable
-    ) -> OrderedSet[Transition]:
-        """Select the transitions that match the trigger data."""
-        enabled_transitions = OrderedSet[Transition]()
-
-        # Get atomic states, TODO: sorted by document order
-        atomic_states = (state for state in self.sm.configuration if state.is_atomic)
-
-        for state in atomic_states:
-            transition = self._first_transition_that_matches(state, trigger_data, predicate)
-            if transition is not None:
-                enabled_transitions.add(transition)
-
-        return self._filter_conflicting_transitions(enabled_transitions)
-
-    def microstep(self, transitions: list[Transition], trigger_data: TriggerData):
-        """Process a single set of transitions in a 'lock step'.
-        This includes exiting states, executing transition content, and entering states.
-        """
-        self._microstep_count += 1
-        self._debug(
-            "%s macro:%d micro:%d transitions: %s",
-            self._log_id,
-            self._macrostep_count,
-            self._microstep_count,
-            transitions,
-        )
-        previous_configuration = self.sm.configuration
-        try:
-            result = self._execute_transition_content(
-                transitions, trigger_data, lambda t: t.before.key
-            )
-
-            states_to_exit = self._exit_states(transitions, trigger_data)
-            result += self._enter_states(
-                transitions, trigger_data, states_to_exit, previous_configuration
-            )
-        except InvalidDefinition:
-            self.sm.configuration = previous_configuration
-            raise
-        except Exception as e:
-            self.sm.configuration = previous_configuration
-            self._handle_error(e, trigger_data)
-            return None
-
-        try:
-            self._execute_transition_content(
-                transitions,
-                trigger_data,
-                lambda t: t.after.key,
-                set_target_as_state=True,
-            )
-        except InvalidDefinition:
-            raise
-        except Exception as e:
-            self._handle_error(e, trigger_data)
-
-        if len(result) == 0:
-            result = None
-        elif len(result) == 1:
-            result = result[0]
-
-        return result
-
     def _get_args_kwargs(
         self,
         transition: Transition,
@@ -488,7 +394,7 @@ def _prepare_exit_states(
             state = info.state
             for history in state.history:
                 if history.type.is_deep:
-                    history_value = [s for s in self.sm.configuration if s.is_descendant(state)]  # noqa: E501
+                    history_value = [s for s in self.sm.configuration if s.is_descendant(state)]
                 else:  # shallow history
                     history_value = [s for s in self.sm.configuration if s.parent == state]
 
@@ -508,29 +414,6 @@ def _remove_state_from_configuration(self, state: State):
         if not self.sm.atomic_configuration_update:
             self.sm._config.discard(state)
 
-    def _exit_states(
-        self, enabled_transitions: list[Transition], trigger_data: TriggerData
-    ) -> OrderedSet[State]:
-        """Compute and process the states to exit for the given transitions."""
-        ordered_states, result = self._prepare_exit_states(enabled_transitions)
-        on_error = self._on_error_handler()
-
-        for info in ordered_states:
-            # Cancel invocations for this state before executing exit handlers.
-            if info.state is not None:  # pragma: no branch
-                self._invoke_manager.cancel_for_state(info.state)
-
-            args, kwargs = self._get_args_kwargs(info.transition, trigger_data, source=info.state)
-
-            # Execute `onexit` handlers — same per-block error isolation as onentry.
-            if info.state is not None:  # pragma: no branch
-                self._debug("%s Exiting state: %s", self._log_id, info.state)
-                self.sm._callbacks.call(info.state.exit.key, *args, on_error=on_error, **kwargs)
-
-            self._remove_state_from_configuration(info.state)
-
-        return result
-
     def _execute_transition_content(
         self,
         enabled_transitions: list[Transition],
@@ -558,7 +441,7 @@ def _prepare_entry_states(
         enabled_transitions: list[Transition],
         states_to_exit: OrderedSet[State],
         previous_configuration: OrderedSet[State],
-    ) -> "tuple[list[StateTransition], OrderedSet[StateTransition], dict[str, Any], OrderedSet[State]]":  # noqa: E501
+    ) -> _EntryPlan:
         """Compute entry set, ordering, and new configuration. Pure computation, no callbacks.
 
         Returns:
@@ -599,7 +482,7 @@ def stop(self):
         self.running = False
         try:
             self._invoke_manager.cancel_all()
-        except Exception:  # pragma: no cover
+        except Exception:
             self._debug("%s Error stopping engine", self._log_id, exc_info=True)
 
     def __del__(self):
@@ -642,7 +525,7 @@ def _handle_final_state(self, target: State, on_entry_result: list):
                     if grandparent.parent is None:
                         self._root_parallel_final_pending = grandparent
 
-    def _enter_states(  # noqa: C901
+    def _enter_states(
         self,
         enabled_transitions: list[Transition],
         trigger_data: TriggerData,
@@ -764,7 +647,7 @@ def compute_entry_set(
                     default_history_content,
                 )
 
-    def add_descendant_states_to_enter(  # noqa: C901
+    def add_descendant_states_to_enter(
         self,
         info: StateTransition,
         states_to_enter,
@@ -784,121 +667,164 @@ def add_descendant_states_to_enter(  # noqa: C901
         state = info.state
 
         if state and state.is_history:
-            # Handle history state
-            state = cast(HistoryState, state)
-            parent_id = state.parent and state.parent.id
-            default_history_content[parent_id] = [info]
-            if state.id in self.sm.history_values:
-                self._debug(
-                    "%s History state '%s.%s' %s restoring: '%s'",
-                    self._log_id,
-                    state.parent,
-                    state,
-                    state.type.value,
-                    [s.id for s in self.sm.history_values[state.id]],
-                )
-                for history_state in self.sm.history_values[state.id]:
-                    info_to_add = StateTransition(transition=info.transition, state=history_state)
-                    if state.type.is_deep:
-                        states_to_enter.add(info_to_add)
-                    else:
-                        self.add_descendant_states_to_enter(
-                            info_to_add,
-                            states_to_enter,
-                            states_for_default_entry,
-                            default_history_content,
-                        )
-                for history_state in self.sm.history_values[state.id]:
-                    info_to_add = StateTransition(transition=info.transition, state=history_state)
-                    self.add_ancestor_states_to_enter(
-                        info_to_add,
-                        state.parent,
-                        states_to_enter,
-                        states_for_default_entry,
-                        default_history_content,
-                    )
-            else:
-                # Handle default history content
-                self._debug(
-                    "%s History state '%s.%s' default content: %s",
-                    self._log_id,
-                    state.parent,
-                    state,
-                    [t.target.id for t in state.transitions if t.target],
-                )
-
-                for transition in state.transitions:
-                    target = cast(State, transition.target)
-                    info_history = StateTransition(transition=transition, state=target)
-                    default_history_content[parent_id].append(info_history)
-                    self.add_descendant_states_to_enter(
-                        info_history,
-                        states_to_enter,
-                        states_for_default_entry,
-                        default_history_content,
-                    )  # noqa: E501
-                for transition in state.transitions:
-                    target = cast(State, transition.target)
-                    info_history = StateTransition(transition=transition, state=target)
-
-                    self.add_ancestor_states_to_enter(
-                        info_history,
-                        state.parent,
-                        states_to_enter,
-                        states_for_default_entry,
-                        default_history_content,
-                    )  # noqa: E501
+            self._add_history_states_to_enter(
+                info,
+                cast(HistoryState, state),
+                states_to_enter,
+                states_for_default_entry,
+                default_history_content,
+            )
             return
 
         # Add the state to the entry set
-        if (
-            self.sm.enable_self_transition_entries
-            or not info.transition.internal
-            or not (
-                info.transition.is_self
-                or (
-                    info.transition.target
-                    and info.transition.target.is_descendant(info.transition.source)
-                )
-            )
-        ):
+        if self._enters_target(info.transition):
             states_to_enter.add(info)
         state = info.state
 
         if state.parallel:
-            for child_state in state.states:
-                if not any(  # pragma: no branch
-                    s.state.is_descendant(child_state) for s in states_to_enter
-                ):
-                    info_to_add = StateTransition(transition=info.transition, state=child_state)
+            self._add_regions_to_enter(
+                info, states_to_enter, states_for_default_entry, default_history_content
+            )
+        elif state.is_compound:
+            self._add_initial_states_to_enter(
+                info, states_to_enter, states_for_default_entry, default_history_content
+            )
+
+    def _enters_target(self, transition: Transition) -> bool:
+        """Whether taking ``transition`` enters its target.
+
+        An internal transition to its own source, or to a descendant of it, does not, unless
+        the machine enables self-transition entries.
+        """
+        return (
+            self.sm.enable_self_transition_entries
+            or not transition.internal
+            or not (
+                transition.is_self
+                or (transition.target and transition.target.is_descendant(transition.source))
+            )
+        )
+
+    def _add_history_states_to_enter(
+        self,
+        info: StateTransition,
+        state: HistoryState,
+        states_to_enter,
+        states_for_default_entry,
+        default_history_content,
+    ):
+        """Add the states a history state stands for: those it recorded, or its default."""
+        parent_id = state.parent and state.parent.id
+        default_history_content[parent_id] = [info]
+        if state.id in self.sm.history_values:
+            self._debug(
+                "%s History state '%s.%s' %s restoring: '%s'",
+                self._log_id,
+                state.parent,
+                state,
+                state.type.value,
+                [s.id for s in self.sm.history_values[state.id]],
+            )
+            for history_state in self.sm.history_values[state.id]:
+                info_to_add = StateTransition(transition=info.transition, state=history_state)
+                if state.type.is_deep:
+                    states_to_enter.add(info_to_add)
+                else:
                     self.add_descendant_states_to_enter(
                         info_to_add,
                         states_to_enter,
                         states_for_default_entry,
                         default_history_content,
                     )
-        elif state.is_compound:
-            states_for_default_entry.add(info)
-            transition = next(t for t in state.transitions if t.initial)
-            # Process all targets (supports multi-target initial transitions for parallel regions)
-            for initial_target in transition.targets:
-                info_initial = StateTransition(transition=transition, state=initial_target)
+            for history_state in self.sm.history_values[state.id]:
+                info_to_add = StateTransition(transition=info.transition, state=history_state)
+                self.add_ancestor_states_to_enter(
+                    info_to_add,
+                    state.parent,
+                    states_to_enter,
+                    states_for_default_entry,
+                    default_history_content,
+                )
+        else:
+            # Handle default history content
+            self._debug(
+                "%s History state '%s.%s' default content: %s",
+                self._log_id,
+                state.parent,
+                state,
+                [t.target.id for t in state.transitions if t.target],
+            )
+
+            for transition in state.transitions:
+                target = cast(State, transition.target)
+                info_history = StateTransition(transition=transition, state=target)
+                default_history_content[parent_id].append(info_history)
                 self.add_descendant_states_to_enter(
-                    info_initial,
+                    info_history,
                     states_to_enter,
                     states_for_default_entry,
                     default_history_content,
                 )
-            for initial_target in transition.targets:
-                info_initial = StateTransition(transition=transition, state=initial_target)
+            for transition in state.transitions:
+                target = cast(State, transition.target)
+                info_history = StateTransition(transition=transition, state=target)
+
                 self.add_ancestor_states_to_enter(
-                    info_initial,
-                    state,
+                    info_history,
+                    state.parent,
+                    states_to_enter,
+                    states_for_default_entry,
+                    default_history_content,
+                )
+
+    def _add_regions_to_enter(
+        self,
+        info: StateTransition,
+        states_to_enter,
+        states_for_default_entry,
+        default_history_content,
+    ):
+        """Add each region of a parallel state, unless a state inside it is already entered."""
+        for child_state in info.state.states:
+            if not any(s.state.is_descendant(child_state) for s in states_to_enter):
+                info_to_add = StateTransition(transition=info.transition, state=child_state)
+                self.add_descendant_states_to_enter(
+                    info_to_add,
                     states_to_enter,
                     states_for_default_entry,
                     default_history_content,
                 )
 
+    def _add_initial_states_to_enter(
+        self,
+        info: StateTransition,
+        states_to_enter,
+        states_for_default_entry,
+        default_history_content,
+    ):
+        """Add the states a compound state's initial transition targets."""
+        state = info.state
+        states_for_default_entry.add(info)
+        transition = next(t for t in state.transitions if t.initial)
+        # Process all targets (supports multi-target initial transitions for parallel regions)
+        for initial_target in transition.targets:
+            info_initial = StateTransition(transition=transition, state=initial_target)
+            self.add_descendant_states_to_enter(
+                info_initial,
+                states_to_enter,
+                states_for_default_entry,
+                default_history_content,
+            )
+        for initial_target in transition.targets:
+            info_initial = StateTransition(transition=transition, state=initial_target)
+            self.add_ancestor_states_to_enter(
+                info_initial,
+                state,
+                states_to_enter,
+                states_for_default_entry,
+                default_history_content,
+            )
+
     def add_ancestor_states_to_enter(
         self,
         info: StateTransition,
@@ -961,9 +887,10 @@ def _check_root_final_state(self):
             self.running = False
 
     def is_in_final_state(self, state: State) -> bool:
-        if state.is_compound:
-            return any(s.final and s in self.sm.configuration for s in state.states)
-        elif state.parallel:  # pragma: no cover — requires nested parallel-in-parallel
+        """Whether every region of a parallel state, or a compound state, rests in a final child.
+
+        An atomic state has no children, so it never does.
+        """
+        if state.parallel:
             return all(self.is_in_final_state(s) for s in state.states)
-        else:  # pragma: no cover — atomic states are never "in final state"
-            return False
+        return any(s.final and s in self.sm.configuration for s in state.states)
diff --git a/statemachine/engines/sync.py b/statemachine/engines/sync.py
index 055a6a15..e7289a9e 100644
--- a/statemachine/engines/sync.py
+++ b/statemachine/engines/sync.py
@@ -1,6 +1,7 @@
+from collections.abc import Callable
+from itertools import chain
 from time import monotonic
 from time import sleep
-from typing import TYPE_CHECKING
 
 from statemachine.event import BoundEvent
 from statemachine.orderedset import OrderedSet
@@ -8,26 +9,12 @@
 from ..event_data import TriggerData
 from ..exceptions import InvalidDefinition
 from ..exceptions import TransitionNotAllowed
+from ..state import State
+from ..transition import Transition
 from .base import BaseEngine
 
-if TYPE_CHECKING:
-    from ..transition import Transition
-
 
 class SyncEngine(BaseEngine):
-    def _run_microstep(self, enabled_transitions, trigger_data):
-        """Run a microstep for internal/eventless transitions with error handling.
-
-        Note: microstep() handles its own errors internally, so this try/except
-        is a safety net that is not expected to be reached in normal operation.
-        """
-        try:
-            self.microstep(list(enabled_transitions), trigger_data)
-        except InvalidDefinition:
-            raise
-        except Exception as e:  # pragma: no cover
-            self._handle_error(e, trigger_data)
-
     def start(self, **kwargs):
         if self.sm.current_state_value is not None:
             return
@@ -56,7 +43,7 @@ def activate_initial_state(self, **kwargs):
                 self._processing.release()
         return self.processing_loop()
 
-    def processing_loop(self, caller_future=None):  # noqa: C901
+    def processing_loop(self, caller_future=None):
         """Process event triggers.
 
         The event is put on a queue, and only the first event will have the result collected.
@@ -80,100 +67,217 @@ def processing_loop(self, caller_future=None):  # noqa: C901
             took_events = True
             while took_events and self.running:
                 self.clear_cache()
-                took_events = False
-                # Execute the triggers in the queue in FIFO order until the queue is empty
-                # while self._running and not self.external_queue.is_empty():
-                macrostep_done = False
-                enabled_transitions: "OrderedSet[Transition] | None" = None
-
-                # handles eventless transitions and internal events
-                while not macrostep_done:
-                    self._microstep_count = 0
-                    self._debug(
-                        "%s Macrostep %d: eventless/internal queue",
-                        self._log_id,
-                        self._macrostep_count,
-                    )
-
-                    self.clear_cache()
-                    internal_event = TriggerData(
-                        self.sm, event=None
-                    )  # this one is a "null object"
-                    enabled_transitions = self.select_eventless_transitions(internal_event)
-                    if not enabled_transitions:
-                        if self.internal_queue.is_empty():
-                            macrostep_done = True
-                        else:
-                            internal_event = self.internal_queue.pop()
-                            enabled_transitions = self.select_transitions(internal_event)
-                    if enabled_transitions:
-                        self._debug(
-                            "%s Enabled transitions: %s", self._log_id, enabled_transitions
-                        )
-                        took_events = True
-                        self._run_microstep(enabled_transitions, internal_event)
+                took_events = self._run_macrostep()
 
                 # Spawn invoke handlers for states entered during this macrostep.
                 self._invoke_manager.spawn_pending_sync()
                 self._check_root_final_state()
 
-                # Process remaining internal events before external events.
-                # Note: the macrostep loop above already drains the internal queue,
-                # so this is a safety net per SCXML spec for invoke-generated events.
-                while not self.internal_queue.is_empty():  # pragma: no cover
-                    internal_event = self.internal_queue.pop()
-                    enabled_transitions = self.select_transitions(internal_event)
-                    if enabled_transitions:
-                        self._run_microstep(enabled_transitions, internal_event)
-
-                # Process external events
-                self._debug("%s Macrostep %d: external queue", self._log_id, self._macrostep_count)
-                while not self.external_queue.is_empty():
-                    self.clear_cache()
-                    took_events = True
-                    external_event = self.external_queue.pop()
-                    current_time = monotonic()
-                    if external_event.execution_time > current_time:
-                        self.put(external_event, _delayed=True)
-                        sleep(self.sm._loop_sleep_in_ms)
-                        # Break to Phase 1 so internal events and eventless
-                        # transitions can be processed while we wait.
-                        break
-
-                    self._macrostep_count += 1
-                    self._microstep_count = 0
-                    self._debug(
-                        "%s macrostep %d: event=%s",
-                        self._log_id,
-                        self._macrostep_count,
-                        external_event.event,
-                    )
+                took_external_events, first_result = self._take_external_events(first_result)
+                took_events = took_events or took_external_events
+        finally:
+            self._processing.release()
+        self._debug("%s Processing loop ended", self._log_id)
+        return first_result if first_result is not self._sentinel else None
+
+    def _run_macrostep(self) -> bool:
+        """Take eventless transitions and internal events until neither enables a transition.
+
+        Returns whether any transition was taken. The internal queue is empty on return.
+        """
+        took_events = False
+        while True:
+            self._microstep_count = 0
+            self._debug(
+                "%s Macrostep %d: eventless/internal queue",
+                self._log_id,
+                self._macrostep_count,
+            )
 
-                    # Finalize + autoforward for active invocations
-                    self._invoke_manager.handle_external_event(external_event)
+            self.clear_cache()
+            internal_event = TriggerData(self.sm, event=None)  # this one is a "null object"
+            enabled_transitions = self.select_eventless_transitions(internal_event)
+            if not enabled_transitions:
+                if self.internal_queue.is_empty():
+                    return took_events
+                internal_event = self.internal_queue.pop()
+                enabled_transitions = self.select_transitions(internal_event)
+            if enabled_transitions:
+                self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions)
+                took_events = True
+                self.microstep(list(enabled_transitions), internal_event)
 
-                    enabled_transitions = self.select_transitions(external_event)
-                    self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions)
-                    if enabled_transitions:
-                        try:
-                            result = self.microstep(list(enabled_transitions), external_event)
-                            if first_result is self._sentinel:
-                                first_result = result
+    def _take_external_events(self, first_result):
+        """Take external events in due order until the queue is empty or holds one not yet due.
 
-                        except Exception:
-                            # We clear the queue as we don't have an expected behavior
-                            # and cannot keep processing
-                            self.clear()
-                            raise
+        Returns whether any event was taken from the queue, and ``first_result`` updated with
+        the result of the first event that enabled a transition.
+        """
+        self._debug("%s Macrostep %d: external queue", self._log_id, self._macrostep_count)
+        took_events = False
+        while not self.external_queue.is_empty():
+            self.clear_cache()
+            took_events = True
+            external_event = self.external_queue.pop()
+            current_time = monotonic()
+            if external_event.execution_time > current_time:
+                self.put(external_event, _delayed=True)
+                sleep(self.sm._loop_sleep_in_ms)
+                # Break to the macrostep so internal events and eventless
+                # transitions can be processed while we wait.
+                break
 
-                    else:
-                        if not self.sm.allow_event_without_transition:
-                            raise TransitionNotAllowed(external_event.event, self.sm.configuration)
+            self._macrostep_count += 1
+            self._microstep_count = 0
+            self._debug(
+                "%s macrostep %d: event=%s",
+                self._log_id,
+                self._macrostep_count,
+                external_event.event,
+            )
 
-        finally:
-            self._processing.release()
-        self._debug("%s Processing loop ended", self._log_id)
-        return first_result if first_result is not self._sentinel else None
+            result = self._take_external_event(external_event)
+            if first_result is self._sentinel:
+                first_result = result
+        return took_events, first_result
+
+    def _take_external_event(self, external_event: TriggerData):
+        """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)
+
+        enabled_transitions = self.select_transitions(external_event)
+        self._debug("%s Enabled transitions: %s", self._log_id, enabled_transitions)
+        if not enabled_transitions:
+            if not self.sm.allow_event_without_transition:
+                raise TransitionNotAllowed(external_event.event, self.sm.configuration)
+            return self._sentinel
+
+        try:
+            return self.microstep(list(enabled_transitions), external_event)
+        except Exception:
+            # We clear the queue as we don't have an expected behavior
+            # and cannot keep processing
+            self.clear()
+            raise
+
+    def select_eventless_transitions(self, trigger_data: TriggerData):
+        """
+        Select the eventless transitions that match the trigger data.
+        """
+        return self._select_transitions(trigger_data, lambda t, _e: t.is_eventless)
+
+    def select_transitions(self, trigger_data: TriggerData) -> OrderedSet[Transition]:
+        """
+        Select the transitions that match the trigger data.
+        """
+        return self._select_transitions(trigger_data, lambda t, e: t.match(e))
+
+    def _first_transition_that_matches(
+        self,
+        state: State,
+        trigger_data: TriggerData,
+        predicate: Callable,
+    ) -> "Transition | None":
+        for s in chain([state], state.ancestors()):
+            transition: Transition
+            for transition in s.transitions:
+                if (
+                    not transition.initial
+                    and predicate(transition, trigger_data.event)
+                    and self._conditions_match(transition, trigger_data)
+                ):
+                    return transition
+        return None
+
+    def _select_transitions(
+        self, trigger_data: TriggerData, predicate: Callable
+    ) -> OrderedSet[Transition]:
+        """Select the transitions that match the trigger data."""
+        enabled_transitions = OrderedSet[Transition]()
+
+        # Get atomic states, TODO: sorted by document order
+        atomic_states = (state for state in self.sm.configuration if state.is_atomic)
+
+        for state in atomic_states:
+            transition = self._first_transition_that_matches(state, trigger_data, predicate)
+            if transition is not None:
+                enabled_transitions.add(transition)
+
+        return self._filter_conflicting_transitions(enabled_transitions)
+
+    def microstep(self, transitions: list[Transition], trigger_data: TriggerData):
+        """Process a single set of transitions in a 'lock step'.
+        This includes exiting states, executing transition content, and entering states.
+        """
+        self._microstep_count += 1
+        self._debug(
+            "%s macro:%d micro:%d transitions: %s",
+            self._log_id,
+            self._macrostep_count,
+            self._microstep_count,
+            transitions,
+        )
+        previous_configuration = self.sm.configuration
+        try:
+            result = self._execute_transition_content(
+                transitions, trigger_data, lambda t: t.before.key
+            )
+
+            states_to_exit = self._exit_states(transitions, trigger_data)
+            result += self._enter_states(
+                transitions, trigger_data, states_to_exit, previous_configuration
+            )
+        except InvalidDefinition:
+            self.sm.configuration = previous_configuration
+            raise
+        except Exception as e:
+            self.sm.configuration = previous_configuration
+            self._handle_error(e, trigger_data)
+            return None
+
+        try:
+            self._execute_transition_content(
+                transitions,
+                trigger_data,
+                lambda t: t.after.key,
+                set_target_as_state=True,
+            )
+        except InvalidDefinition:
+            raise
+        except Exception as e:
+            self._handle_error(e, trigger_data)
+
+        if len(result) == 0:
+            result = None
+        elif len(result) == 1:
+            result = result[0]
+
+        return result
+
+    def _exit_states(
+        self, enabled_transitions: list[Transition], trigger_data: TriggerData
+    ) -> OrderedSet[State]:
+        """Compute and process the states to exit for the given transitions."""
+        ordered_states, result = self._prepare_exit_states(enabled_transitions)
+        on_error = self._on_error_handler()
+
+        for info in ordered_states:
+            # Cancel invocations for this state before executing exit handlers.
+            self._invoke_manager.cancel_for_state(info.state)
+
+            args, kwargs = self._get_args_kwargs(info.transition, trigger_data, source=info.state)
+
+            # Execute `onexit` handlers — same per-block error isolation as onentry.
+            self._debug("%s Exiting state: %s", self._log_id, info.state)
+            self.sm._callbacks.call(info.state.exit.key, *args, on_error=on_error, **kwargs)
+
+            self._remove_state_from_configuration(info.state)
+
+        return result
 
     def enabled_events(self, *args, **kwargs):
         sm = self.sm
diff --git a/statemachine/statemachine.py b/statemachine/statemachine.py
index 82b9c210..525c552f 100644
--- a/statemachine/statemachine.py
+++ b/statemachine/statemachine.py
@@ -5,6 +5,7 @@
 from typing import Any
 from typing import Generic
 from typing import TypeVar
+from typing import cast
 
 from statemachine.orderedset import OrderedSet
 
@@ -146,7 +147,7 @@ def __init__(
         listeners: "list[object] | None" = None,
         **kwargs: Any,
     ):
-        self.model: TModel = model if model is not None else Model()  # type: ignore[assignment]
+        self.model: TModel = model if model is not None else cast(TModel, Model())
         """The external model object that holds domain state, or an internal
         :class:`Model` instance when none is provided.  See :ref:`domain models`."""
 
@@ -259,7 +260,7 @@ def __getstate__(self):
 
     def __setstate__(self, state: dict[str, Any]) -> None:
         listeners = state.pop("_listeners")
-        self.__dict__.update(state)  # type: ignore[attr-defined]
+        vars(self).update(state)
         self._callbacks = CallbacksRegistry()
         self._config = self._build_configuration()
         self._listeners = {}
@@ -273,11 +274,11 @@ def __setstate__(self, state: dict[str, Any]) -> None:
         self._engine.start()
 
     def _get_initial_configuration(self):
-        initial_state_values = (
-            self.start_configuration_values
-            if self.start_configuration_values
-            else [self.initial_state.value]  # type: ignore[union-attr]
-        )
+        initial_state_values = self.start_configuration_values
+        if not initial_state_values:
+            # Only a machine with no states has no initial state, and none can be created.
+            assert self.initial_state is not None
+            initial_state_values = [self.initial_state.value]
         try:
             return [self.states_map[value] for value in initial_state_values]
         except KeyError as err:
@@ -362,7 +363,7 @@ def _repr_html_(self):
         return f'<div class="statemachine">{self._repr_svg_()}</div>'
 
     def _repr_svg_(self):
-        return self._graph().create_svg().decode()  # type: ignore[attr-defined]
+        return format(self, "svg")
 
     def _graph(self):
         from .contrib.diagram import DotGraphMachine
diff --git a/tests/scxml/test_parallel_regions.py b/tests/scxml/test_parallel_regions.py
new file mode 100644
index 00000000..1a212bd8
--- /dev/null
+++ b/tests/scxml/test_parallel_regions.py
@@ -0,0 +1,38 @@
+"""A transition that targets a parallel state and a state inside one of its regions.
+
+The region holding the targeted state enters that state and not its initial one; every other
+region enters its initial state.
+
+Theme: the beacons, one of them already lit.
+"""
+
+import pytest
+from statemachine.io.scxml.processor import SCXMLProcessor
+
+BEACONS_SCXML = """
+<scxml initial="dark">
+  <state id="dark">
+    <transition event="light" target="east_lit beacons"/>
+  </state>
+  <parallel id="beacons">
+    <state id="east" initial="east_dark">
+      <state id="east_dark"/>
+      <state id="east_lit"/>
+    </state>
+    <state id="west" initial="west_dark">
+      <state id="west_dark"/>
+    </state>
+  </parallel>
+</scxml>
+"""
+
+
+@pytest.mark.scxml()
+def test_a_region_with_a_targeted_state_does_not_also_enter_its_initial_state():
+    processor = SCXMLProcessor()
+    processor.parse_scxml("beacons", BEACONS_SCXML)
+    sm = processor.start()
+
+    sm.send("light")
+
+    assert set(sm.configuration_values) == {"beacons", "east", "east_lit", "west", "west_dark"}
diff --git a/tests/test_invoke.py b/tests/test_invoke.py
index e31baad2..c9dc271c 100644
--- a/tests/test_invoke.py
+++ b/tests/test_invoke.py
@@ -1294,6 +1294,27 @@ def test_stop_child_machine_exception_swallowed(self):
         _stop_child_machine(child)
 
 
+class TestEngineStopCleanup:
+    """Test BaseEngine.stop cancel_all exception handling."""
+
+    def test_stop_swallows_cancel_all_exception_and_stops(self):
+        """stop() swallows exceptions from cancel_all and still stops the engine."""
+
+        class SM(StateChart):
+            s1 = State(initial=True)
+            s2 = State(final=True)
+
+            go = s1.to(s2)
+
+        sm = SM()
+        engine = sm._engine
+        engine._invoke_manager.cancel_all = lambda: (_ for _ in ()).throw(RuntimeError("boom"))
+
+        engine.stop()
+
+        assert engine.running is False
+
+
 class TestEngineDelCleanup:
     """Test BaseEngine.__del__ cancel_all exception handling."""
 

From e62edf73de4885e8be9ea8e6c70b562e2dd22b90 Mon Sep 17 00:00:00 2001
From: John Merrill <john@fairplay.ai>
Date: Sun, 4 Oct 2026 16:49:43 -0700
Subject: [PATCH 8/8] fix: give the caller's future only to the async engine

SyncEngine.processing_loop took a caller_future it never used: only
AsyncEngine.put attaches a future to an event, so the sync engine was
always handed None. StateChart._processing_loop now passes the future
to an AsyncEngine and calls the sync loop without one, and the sync
loop no longer takes it (SonarCloud python:S1172).

The new SCXML test marks its module with `pytestmark` rather than
`@pytest.mark.scxml()`, which this project's ruff configuration
(mark-parentheses = true) and SonarCloud's python:S9083 disagree on.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 statemachine/engines/sync.py         | 2 +-
 statemachine/statemachine.py         | 6 +++++-
 tests/scxml/test_parallel_regions.py | 3 ++-
 3 files changed, 8 insertions(+), 3 deletions(-)

diff --git a/statemachine/engines/sync.py b/statemachine/engines/sync.py
index e7289a9e..3cba86a9 100644
--- a/statemachine/engines/sync.py
+++ b/statemachine/engines/sync.py
@@ -43,7 +43,7 @@ def activate_initial_state(self, **kwargs):
                 self._processing.release()
         return self.processing_loop()
 
-    def processing_loop(self, caller_future=None):
+    def processing_loop(self):
         """Process event triggers.
 
         The event is put on a queue, and only the first event will have the result collected.
diff --git a/statemachine/statemachine.py b/statemachine/statemachine.py
index 525c552f..2f88ab1e 100644
--- a/statemachine/statemachine.py
+++ b/statemachine/statemachine.py
@@ -226,7 +226,11 @@ def activate_initial_state(self) -> Any:
         return run_async_from_sync(result)
 
     def _processing_loop(self, caller_future: "Any | None" = None) -> Any:
-        result = self._engine.processing_loop(caller_future)
+        # Only the async engine gives an event a future, for its caller to await the result by.
+        if isinstance(self._engine, AsyncEngine):
+            result = self._engine.processing_loop(caller_future)
+        else:
+            result = self._engine.processing_loop()
         if not isawaitable(result):
             return result
         return run_async_from_sync(result)
diff --git a/tests/scxml/test_parallel_regions.py b/tests/scxml/test_parallel_regions.py
index 1a212bd8..6f149827 100644
--- a/tests/scxml/test_parallel_regions.py
+++ b/tests/scxml/test_parallel_regions.py
@@ -9,6 +9,8 @@
 import pytest
 from statemachine.io.scxml.processor import SCXMLProcessor
 
+pytestmark = pytest.mark.scxml
+
 BEACONS_SCXML = """
 <scxml initial="dark">
   <state id="dark">
@@ -27,7 +29,6 @@
 """
 
 
-@pytest.mark.scxml()
 def test_a_region_with_a_targeted_state_does_not_also_enter_its_initial_state():
     processor = SCXMLProcessor()
     processor.parse_scxml("beacons", BEACONS_SCXML)