From ec8ce36274d573b4887ed9ccc78b39e7d8a9ad8c Mon Sep 17 00:00:00 2001 From: Vladimir Babin Date: Thu, 24 Sep 2026 09:53:46 +0300 Subject: [PATCH 1/9] fix: honor Event(internal=True) when the event is triggered (cherry picked from commit e633508da16b9ede57d0846d7a95780622aa0049) Signed-off-by: Fernando Macedo --- docs/processing_model.md | 2 +- docs/releases/3.2.2.md | 11 +++++++++-- statemachine/event.py | 4 +++- statemachine/statemachine.py | 1 + tests/test_events.py | 25 +++++++++++++++++++++++++ 5 files changed, 39 insertions(+), 4 deletions(-) diff --git a/docs/processing_model.md b/docs/processing_model.md index d8181316..16bceb03 100644 --- a/docs/processing_model.md +++ b/docs/processing_model.md @@ -80,7 +80,7 @@ The engine maintains two separate FIFO queues: | Queue | How to enqueue | When processed | |---|---|---| -| **Internal** | {func}`raise_() ` or `send(..., internal=True)` | Within the current macrostep | +| **Internal** | {func}`raise_() `, `send(..., internal=True)`, or an event declared with `Event(..., internal=True)` | Within the current macrostep | | **External** | {func}`send() ` | After the current macrostep ends | This distinction matters when you trigger events from inside callbacks. diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md index 47b25b89..17d5f4cc 100644 --- a/docs/releases/3.2.2.md +++ b/docs/releases/3.2.2.md @@ -117,8 +117,7 @@ Delays are now measured on `time.monotonic()`, which no setting of the wall cloc `Event(dark.to(lit), delay=50)` rebuilt the event without its `delay`, so `BeaconsOfGondor.light.delay` was `0` and the event fired immediately instead of being queued. -The same happened to `internal`. Both are now preserved, although `internal` still has no -effect at trigger time: +The same happened to `internal`. Both are now preserved: ```py >>> class BeaconsOfGondor(StateChart): @@ -132,6 +131,14 @@ effect at trigger time: ``` +### `internal` declared on an `Event` ignored at trigger time + +`Event(a.to(b), internal=True)` kept `internal` on the declaration, but neither `sm.go()` nor +`sm.send("go")` carried it to the queue, so the event always went to the external queue. Both +paths now honor the declaration and put the event on the internal queue. + +[#655](https://github.com/fgmacedo/python-statemachine/issues/655). + ### `is_active` on a state nested in a parallel region Reaching a child state through its parent (`sm.door.shut`) returned the class-level `State`, diff --git a/statemachine/event.py b/statemachine/event.py index 2c23f920..50470280 100644 --- a/statemachine/event.py +++ b/statemachine/event.py @@ -146,7 +146,9 @@ def __get__(self, instance, owner): """ if instance is None: return self - return BoundEvent(id=self.id, name=self.name, delay=self.delay, _sm=instance) + return BoundEvent( + id=self.id, name=self.name, delay=self.delay, internal=self.internal, _sm=instance + ) def put(self, *args, send_id: "str | None" = None, **kwargs): # The `__call__` is declared here to help IDEs knowing that an `Event` diff --git a/statemachine/statemachine.py b/statemachine/statemachine.py index ba7398b4..1cd3cf1c 100644 --- a/statemachine/statemachine.py +++ b/statemachine/statemachine.py @@ -488,6 +488,7 @@ def send( delay = ( delay if delay else know_event and know_event.delay or 0 ) # first the param, then the event, or 0 + internal = internal or bool(know_event and know_event.internal) event_instance = BoundEvent( id=event, name=event_name, delay=delay, internal=internal, _sm=self ) diff --git a/tests/test_events.py b/tests/test_events.py index fc00d675..2440bb29 100644 --- a/tests/test_events.py +++ b/tests/test_events.py @@ -71,6 +71,31 @@ class BeaconsOfGondor(StateChart): assert BeaconsOfGondor.light.delay == 50 assert BeaconsOfGondor().light.delay == 50 + @pytest.mark.parametrize("trigger", ["method", "send"]) + def test_declared_internal_puts_on_the_internal_queue(self, trigger): + class BeaconsOfGondor(StateChart): + dark = State(initial=True) + lit = State(final=True) + + light = Event(dark.to(lit), internal=True) + + sm = BeaconsOfGondor() + queued = [] + put = sm._engine.put + + def spy(trigger_data, internal=False, **kwargs): + queued.append(internal) + return put(trigger_data, internal=internal, **kwargs) + + sm._engine.put = spy + if trigger == "method": + sm.light() + else: + sm.send("light") + + assert queued == [True] + assert sm.lit.is_active + def test_derive_name_from_id(self): class StartMachine(StateChart): created = State(initial=True) From ad7e6ed463e2e5c530ee223f71e49a7fd0873423 Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Sun, 11 Oct 2026 22:24:15 -0300 Subject: [PATCH 2/9] fix: run a macrostep after each external event With several external events queued, both engines took them one after the other and only then ran the internal events and eventless transitions they had enabled, so an internal event raised while handling the first external event ran after the second. SCXML's main event loop, and the processing model docs, run the macrostep after every external event. The external-queue step now takes at most one due event and returns to the macrostep. Signed-off-by: Fernando Macedo --- docs/releases/3.2.2.md | 33 +++++++++++++++ statemachine/engines/async_.py | 73 +++++++++++++++++----------------- statemachine/engines/sync.py | 57 +++++++++++++------------- tests/test_event_ordering.py | 53 +++++++++++++++++++++++- 4 files changed, 151 insertions(+), 65 deletions(-) diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md index 17d5f4cc..fec9ca2b 100644 --- a/docs/releases/3.2.2.md +++ b/docs/releases/3.2.2.md @@ -174,3 +174,36 @@ Reported by [@MartinBernstorff](https://github.com/MartinBernstorff). [#651](https://github.com/fgmacedo/python-statemachine/issues/651), [#652](https://github.com/fgmacedo/python-statemachine/pull/652). + +### Each external event gets its own macrostep + +With several external events already queued, the engine took them one after the other and +only then ran the internal events and eventless transitions they had enabled. An internal +event raised while handling the first external event ran after the second, breaking the +run-to-completion order the {ref}`processing model ` describes: + +```py +>>> from statemachine.event import BoundEvent + +>>> class Beacons(StateChart): +... dark = State(initial=True) +... lit = State(final=True) +... besieged = State(final=True) +... +... signal = dark.to.itself(internal=True, on="relay") +... light = dark.to(lit) +... siege = dark.to(besieged) +... +... def relay(self): +... self.raise_("light") +... self.send("siege") + +>>> sm = Beacons() +>>> sm.send("signal") +>>> sm.lit.is_active +True + +``` + +Each external event is now followed by its macrostep before the next one is taken, on both +engines. diff --git a/statemachine/engines/async_.py b/statemachine/engines/async_.py index 9ba27f89..fba213d8 100644 --- a/statemachine/engines/async_.py +++ b/statemachine/engines/async_.py @@ -370,7 +370,9 @@ async def processing_loop(self, caller_future: "asyncio.Future[object] | None" = await self._invoke_manager.spawn_pending_async() self._check_root_final_state() - took_external_events, first_result = await self._take_external_events(first_result) + took_external_events, first_result = await self._take_due_external_event( + first_result + ) took_events = took_events or took_external_events except Exception as exc: @@ -424,45 +426,44 @@ async def _run_macrostep(self) -> bool: 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. + async def _take_due_external_event(self, first_result): + """Take the external event at the head of the queue, if it is due. + + One at a time, so the macrostep that follows runs the internal events and eventless + transitions it enabled before the next external event is taken. - Returns whether any event was taken from the queue, and ``first_result`` updated with - the result of the first event that enabled a transition. + Returns whether the queue held an event, 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_due(monotonic()) - if external_event is None: - 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, - ) + if self.external_queue.is_empty(): + return False, first_result + + self.clear_cache() + external_event = self.external_queue.pop_due(monotonic()) + if external_event is None: + await asyncio.sleep(self.sm._loop_sleep_in_ms) + return True, first_result + + 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. + if external_event.event == "__initial__": + transitions = self._initial_transitions(external_event) + await self._enter_states(transitions, external_event, OrderedSet(), OrderedSet()) + return True, first_result - # 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 + result = await self._take_external_event(external_event) + if first_result is self._sentinel: + first_result = result + return True, first_result async def _take_external_event(self, external_event: TriggerData): """Run one external event and settle its future. diff --git a/statemachine/engines/sync.py b/statemachine/engines/sync.py index e15c8d8f..7ba818e5 100644 --- a/statemachine/engines/sync.py +++ b/statemachine/engines/sync.py @@ -73,7 +73,7 @@ def processing_loop(self): self._invoke_manager.spawn_pending_sync() self._check_root_final_state() - took_external_events, first_result = self._take_external_events(first_result) + took_external_events, first_result = self._take_due_external_event(first_result) took_events = took_events or took_external_events finally: self._processing.release() @@ -107,37 +107,38 @@ def _run_macrostep(self) -> bool: took_events = True self.microstep(list(enabled_transitions), internal_event) - def _take_external_events(self, first_result): - """Take external events in due order until the queue is empty or holds one not yet due. + def _take_due_external_event(self, first_result): + """Take the external event at the head of the queue, if it is 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. + One at a time, so the macrostep that follows runs the internal events and eventless + transitions it enabled before the next external event is taken. + + Returns whether the queue held an event, 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_due(monotonic()) - if external_event is None: - 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, - ) + if self.external_queue.is_empty(): + return False, first_result + + self.clear_cache() + external_event = self.external_queue.pop_due(monotonic()) + if external_event is None: + sleep(self.sm._loop_sleep_in_ms) + return True, first_result + + self._macrostep_count += 1 + self._microstep_count = 0 + self._debug( + "%s macrostep %d: event=%s", + self._log_id, + self._macrostep_count, + external_event.event, + ) - result = self._take_external_event(external_event) - if first_result is self._sentinel: - first_result = result - return took_events, first_result + result = self._take_external_event(external_event) + if first_result is self._sentinel: + first_result = result + return True, first_result def _take_external_event(self, external_event: TriggerData): """Run one external event. diff --git a/tests/test_event_ordering.py b/tests/test_event_ordering.py index 168f994f..53b945e6 100644 --- a/tests/test_event_ordering.py +++ b/tests/test_event_ordering.py @@ -2,7 +2,8 @@ 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. +must still come out in the order they went in. And each external event gets its +own macrostep: what it enables runs before the next external event is taken. Theme: the Riders of Rohan answering the muster, each in turn. """ @@ -56,3 +57,53 @@ async def test_external_events_run_in_the_order_sent(self, sm_runner, riders): await sm_runner.processing_loop(sm) assert sm.answered == list(range(riders)) + + +class RelayedBeacons(StateChart): + dark = State(initial=True) + lit = State(final=True) + besieged = State(final=True) + + signal = dark.to.itself(internal=True, on="relay") + light = dark.to(lit) + siege = dark.to(besieged) + + def relay(self): + BoundEvent(id="light", name="Light", internal=True, _sm=self).put() + + +class WatchedBeacons(StateChart): + dark = State(initial=True) + lit = State(final=True) + besieged = State(final=True) + + signal = dark.to.itself(internal=True, on="relay") + dark.to(lit, cond="seen") + siege = dark.to(besieged) + + def __init__(self, *args, **kwargs): + self.seen = False + super().__init__(*args, **kwargs) + + def relay(self): + self.seen = True + + +@pytest.mark.timeout(10) +@pytest.mark.parametrize( + "machine", + [ + pytest.param(RelayedBeacons, id="internal-event"), + pytest.param(WatchedBeacons, id="eventless-transition"), + ], +) +async def test_what_an_external_event_enables_runs_before_the_next_external_event( + sm_runner, machine +): + sm = await sm_runner.start(machine) + BoundEvent(id="signal", name="Signal", _sm=sm).put() + BoundEvent(id="siege", name="Siege", _sm=sm).put() + + await sm_runner.processing_loop(sm) + + assert sm.configuration_values == {"lit"} From 29ed6cb59392fcfc47fb2cad71e2b5553a405620 Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Sun, 11 Oct 2026 22:27:10 -0300 Subject: [PATCH 3/9] fix: give internal triggers from outside a step the external contract Honoring Event(internal=True) at trigger time exposed how the internal queue treated a trigger that no running step had raised: - on the async engine, `await sm.go()` on a fresh machine was popped before `__initial__` entered the initial state, and silently lost; - the caller got `None` instead of the callback result, no `TransitionNotAllowed` on a StateMachine, and no future to await; - a delay was ignored, so `raise_("x", delay=50)` fired at once. The internal queue now receives an internal trigger only when the machine raises it while running a step. From outside there is no macrostep to join, so the trigger goes to the external queue, which already returns results, routes exceptions and orders `__initial__` first. The context variable that marks a running step moves from the async engine to the base, wraps both loops and the sync initial entry, and decides the queue in `BaseEngine.put`. A delay can never be internal: a step takes no time, and SCXML forbids `delay` with target `#_internal`. `Event.__new__` rejects the combination with InvalidDefinition, which covers declarations, `raise_(delay=)`, `send(delay=, internal=True)` and `` alike. Refs #655, #663. Signed-off-by: Fernando Macedo --- AGENTS.md | 8 +- statemachine/engines/async_.py | 45 +++--- statemachine/engines/base.py | 28 +++- statemachine/engines/sync.py | 25 ++-- statemachine/event.py | 9 ++ statemachine/statemachine.py | 8 +- tests/test_events.py | 43 ++---- tests/test_internal_triggers.py | 218 ++++++++++++++++++++++++++++++ tests/test_statechart_compound.py | 15 +- 9 files changed, 324 insertions(+), 75 deletions(-) create mode 100644 tests/test_internal_triggers.py diff --git a/AGENTS.md b/AGENTS.md index 49062228..ab2e4638 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -39,7 +39,13 @@ The engine follows the SCXML run-to-completion (RTC) model with two processing l ### Event queues - `send()` → **external queue** (processed after current macrostep ends). -- `raise_()` → **internal queue** (processed within the current macrostep, before external events). +- `raise_()`, `send(internal=True)`, `Event(internal=True)` → **internal queue** (processed + within the current macrostep, before external events), but only when raised while the + machine runs a step (`_in_processing_loop` in `engines/base.py`). From outside a step there + is no macrostep to join: the trigger goes to the external queue, so its caller gets the + result, the exception or `TransitionNotAllowed`. +- `delay` and `internal` together raise `InvalidDefinition`: a step takes no time. +- Each external event is followed by its macrostep before the next external event is taken. ### Error handling (`catch_errors_as_events`) diff --git a/statemachine/engines/async_.py b/statemachine/engines/async_.py index fba213d8..eec3da60 100644 --- a/statemachine/engines/async_.py +++ b/statemachine/engines/async_.py @@ -1,5 +1,4 @@ import asyncio -import contextvars from collections.abc import Callable from itertools import chain from time import monotonic @@ -13,19 +12,12 @@ from ..state import State from .base import _ERROR_EXECUTION from .base import BaseEngine +from .base import _in_processing_loop +from .base import _processing_step if TYPE_CHECKING: from ..transition import Transition -# ContextVar to distinguish reentrant calls (from within callbacks) from -# concurrent external calls. asyncio propagates context to child tasks -# (e.g., those created by asyncio.gather in the callback system), so a -# ContextVar set in the processing loop is visible in all callbacks. -# Independent external coroutines have their own context where this is False. -_in_processing_loop: contextvars.ContextVar[bool] = contextvars.ContextVar( - "_in_processing_loop", default=False -) - class AsyncEngine(BaseEngine): """Async engine with full StateChart support. @@ -35,16 +27,16 @@ class AsyncEngine(BaseEngine): """ def put(self, trigger_data: TriggerData, internal: bool = False): - """Override to attach an asyncio.Future for external events. + """Override to attach an asyncio.Future to a trigger put from outside the loop. - Futures are only created when: - - The event is external (not internal) + Such a trigger always lands on the external queue, and its caller awaits the future + for its result. Futures are only created when: - No future is already attached - There is a running asyncio loop - The call is NOT from within the processing loop (reentrant calls from callbacks must not get futures, as that would deadlock) """ - if not internal and trigger_data.future is None and not _in_processing_loop.get(): + if trigger_data.future is None and not _in_processing_loop.get(): try: loop = asyncio.get_running_loop() trigger_data.future = loop.create_future() @@ -357,23 +349,23 @@ async def processing_loop(self, caller_future: "asyncio.Future[object] | None" = return await caller_future return None - _ctx_token = _in_processing_loop.set(True) self._debug("%s Processing loop started: %s", self._log_id, self.sm.current_state_value) first_result = self._sentinel try: - took_events = True - while took_events and self.running: - self.clear_cache() - took_events = await self._run_macrostep() + with _processing_step(): + took_events = True + while took_events and self.running: + self.clear_cache() + 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() + # Spawn invoke handlers for states entered during this macrostep. + await self._invoke_manager.spawn_pending_async() + self._check_root_final_state() - took_external_events, first_result = await self._take_due_external_event( - first_result - ) - took_events = took_events or took_external_events + took_external_events, first_result = await self._take_due_external_event( + first_result + ) + took_events = took_events or took_external_events except Exception as exc: if caller_future is not None: @@ -387,7 +379,6 @@ async def processing_loop(self, caller_future: "asyncio.Future[object] | None" = else: raise finally: - _in_processing_loop.reset(_ctx_token) self._processing.release() self._debug("%s Processing loop ended", self._log_id) diff --git a/statemachine/engines/base.py b/statemachine/engines/base.py index 0fed1acb..938f937c 100644 --- a/statemachine/engines/base.py +++ b/statemachine/engines/base.py @@ -1,5 +1,8 @@ import logging from collections.abc import Callable +from collections.abc import Iterator +from contextlib import contextmanager +from contextvars import ContextVar from dataclasses import dataclass from dataclasses import field from heapq import heappop @@ -26,6 +29,23 @@ logger = logging.getLogger(__name__) +_in_processing_loop: ContextVar[bool] = ContextVar("_in_processing_loop", default=False) +"""Whether the current context is running a step of an engine. + +asyncio copies the context into the tasks a step creates, so callbacks see it set. A new +thread or an independent coroutine starts without it. +""" + + +@contextmanager +def _processing_step() -> Iterator[None]: + """Mark the current context as running a step of an engine, until the block ends.""" + token = _in_processing_loop.set(True) + try: + yield + finally: + _in_processing_loop.reset(token) + @dataclass(frozen=True, unsafe_hash=True, eq=True) class StateTransition: @@ -133,10 +153,16 @@ def clear_cache(self): self._cache.clear() def put(self, trigger_data: TriggerData, internal: bool = False): - """Put the trigger on the queue without blocking the caller.""" + """Put the trigger on the queue without blocking the caller. + + An ``internal`` trigger joins the internal queue only when the machine raises it while + running a step. Put from outside, there is no macrostep to join, so it goes to the + external queue, where its caller gets its result. + """ if not self.running and not self.sm.allow_event_without_transition: raise TransitionNotAllowed(trigger_data.event, self.sm.configuration) + internal = internal and _in_processing_loop.get() if internal: self.internal_queue.put(trigger_data) else: diff --git a/statemachine/engines/sync.py b/statemachine/engines/sync.py index 7ba818e5..2433c526 100644 --- a/statemachine/engines/sync.py +++ b/statemachine/engines/sync.py @@ -12,6 +12,7 @@ from ..state import State from ..transition import Transition from .base import BaseEngine +from .base import _processing_step class SyncEngine(BaseEngine): @@ -38,7 +39,8 @@ def activate_initial_state(self, **kwargs): transitions = self._initial_transitions(trigger_data) self._processing.acquire(blocking=False) try: - self._enter_states(transitions, trigger_data, OrderedSet(), OrderedSet()) + with _processing_step(): + self._enter_states(transitions, trigger_data, OrderedSet(), OrderedSet()) finally: self._processing.release() return self.processing_loop() @@ -64,17 +66,20 @@ def processing_loop(self): self._debug("%s Processing loop started: %s", self._log_id, self.sm.current_state_value) first_result = self._sentinel try: - took_events = True - while took_events and self.running: - self.clear_cache() - took_events = self._run_macrostep() + with _processing_step(): + took_events = True + while took_events and self.running: + self.clear_cache() + 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() + # Spawn invoke handlers for states entered during this macrostep. + self._invoke_manager.spawn_pending_sync() + self._check_root_final_state() - took_external_events, first_result = self._take_due_external_event(first_result) - took_events = took_events or took_external_events + took_external_events, first_result = self._take_due_external_event( + first_result + ) + took_events = took_events or took_external_events finally: self._processing.release() self._debug("%s Processing loop ended", self._log_id) diff --git a/statemachine/event.py b/statemachine/event.py index 50470280..6bac8f19 100644 --- a/statemachine/event.py +++ b/statemachine/event.py @@ -100,6 +100,15 @@ def __new__( _has_real_id = id is not None id = str(id) if _has_real_id else f"__event__{uuid4().hex}" + if delay and internal: + raise InvalidDefinition( + _( + "Event {id!r} cannot be both delayed and internal: the internal queue runs " + "within the current macrostep, which takes no time. Drop 'internal' to " + "deliver it later on the external queue, or drop 'delay' to run it now." + ).format(id=id) + ) + instance = super().__new__(cls, id) instance.id = id instance.delay = delay diff --git a/statemachine/statemachine.py b/statemachine/statemachine.py index 1cd3cf1c..41a23ade 100644 --- a/statemachine/statemachine.py +++ b/statemachine/statemachine.py @@ -477,6 +477,10 @@ def send( :param delay: A time delay in milliseconds to process the event. Default is 0. :param send_id: An identifier for the event, used with ``cancel_event()`` to cancel delayed events. + :param internal: Put the event on the internal queue, so it runs within the current + macrostep. Defaults to the event's declaration. Honored from a callback; sent from + outside a step, the event goes to the external queue and its result is returned. + Cannot be combined with ``delay``. :param kwargs: Additional keyword arguments to pass to the event. .. seealso:: @@ -504,7 +508,9 @@ def raise_( Events on the internal queue are processed immediately within the current macrostep, before any pending external events. This is equivalent to calling - ``send(..., internal=True)``. + ``send(..., internal=True)``: called from outside a callback, there is no macrostep + to join, so the event goes to the external queue and its result is returned. A + ``delay`` other than 0 is an :class:`~statemachine.exceptions.InvalidDefinition`. .. seealso:: diff --git a/tests/test_events.py b/tests/test_events.py index 2440bb29..7fd97c12 100644 --- a/tests/test_events.py +++ b/tests/test_events.py @@ -59,42 +59,25 @@ class StartMachine(StateChart): assert [e.name for e in StartMachine.events] == ["Start the machine"] assert StartMachine.start.name == "Start the machine" - def test_accept_delay_and_internal(self): + @pytest.mark.parametrize( + ("declared", "expected"), + [ + pytest.param({"delay": 50}, (50, False), id="delay"), + pytest.param({"internal": True}, (0, True), id="internal"), + ], + ) + def test_accept_delay_or_internal(self, declared, expected): class BeaconsOfGondor(StateChart): dark = State(initial=True) lit = State(final=True) - light = Event(dark.to(lit), delay=50, internal=True) + light = Event(dark.to(lit), **declared) (registered,) = BeaconsOfGondor.events - assert (registered.delay, registered.internal) == (50, True) - assert BeaconsOfGondor.light.delay == 50 - assert BeaconsOfGondor().light.delay == 50 - - @pytest.mark.parametrize("trigger", ["method", "send"]) - def test_declared_internal_puts_on_the_internal_queue(self, trigger): - class BeaconsOfGondor(StateChart): - dark = State(initial=True) - lit = State(final=True) - - light = Event(dark.to(lit), internal=True) - - sm = BeaconsOfGondor() - queued = [] - put = sm._engine.put - - def spy(trigger_data, internal=False, **kwargs): - queued.append(internal) - return put(trigger_data, internal=internal, **kwargs) - - sm._engine.put = spy - if trigger == "method": - sm.light() - else: - sm.send("light") - - assert queued == [True] - assert sm.lit.is_active + bound = BeaconsOfGondor().light + assert (registered.delay, registered.internal) == expected + assert (BeaconsOfGondor.light.delay, BeaconsOfGondor.light.internal) == expected + assert (bound.delay, bound.internal) == expected def test_derive_name_from_id(self): class StartMachine(StateChart): diff --git a/tests/test_internal_triggers.py b/tests/test_internal_triggers.py new file mode 100644 index 00000000..94608cc1 --- /dev/null +++ b/tests/test_internal_triggers.py @@ -0,0 +1,218 @@ +"""Which queue a trigger asked to be internal goes to, and what its caller gets back. + +The internal queue holds what the machine raises while it runs a step, so it is processed +within that macrostep, before any pending external event. A trigger put from outside a step +has no macrostep to join: it goes to the external queue, where its caller gets its result, +its exception or ``TransitionNotAllowed``. A delay can never be internal, since a step takes +no time; asking for both is an invalid definition. + +Theme: the beacons of Gondor, lit from inside the chain or called for from afar. +""" + +from inspect import isawaitable + +import pytest +from statemachine.engines.base import _in_processing_loop +from statemachine.exceptions import InvalidDefinition +from statemachine.exceptions import TransitionNotAllowed +from statemachine.io.scxml.processor import SCXMLProcessor + +from statemachine import Event +from statemachine import State +from statemachine import StateChart +from statemachine import StateMachine + +DELAYED_INTERNAL = "cannot be both delayed and internal" + + +class Beacons(StateChart): + dark = State(initial=True) + lit = State(final=True) + + light = Event(dark.to(lit, on="kindle"), internal=True) + + def kindle(self): + return "kindled" + + +async def settle(result): + return await result if isawaitable(result) else result + + +class TestFromACallback: + async def test_a_declared_internal_event_runs_before_a_pending_external_event(self, sm_runner): + class Muster(StateChart): + camp = State(initial=True) + riding = State(final=True) + besieged = State(final=True) + + summon = camp.to.itself(internal=True, on="call") + ride = Event(camp.to(riding), internal=True) + siege = camp.to(besieged) + + def call(self): + self.siege.put() + self.ride.put() + + sm = await sm_runner.start(Muster) + + await sm_runner.send(sm, "summon") + + assert sm.configuration_values == {"riding"} + + async def test_an_internal_event_raised_while_entering_the_initial_state_runs_first( + self, sm_runner + ): + class Watch(StateChart): + dark = State(initial=True) + lit = State(final=True) + fallen = State(final=True) + + light = Event(dark.to(lit), internal=True) + fall = dark.to(fallen) + + def on_enter_dark(self): + self.fall.put() + self.light.put() + + sm = await sm_runner.start(Watch) + + assert sm.configuration_values == {"lit"} + + +class TestFromOutside: + @pytest.mark.parametrize( + "trigger", + [ + pytest.param(lambda sm: sm.light(), id="method"), + pytest.param(lambda sm: sm.send("light"), id="send"), + pytest.param(lambda sm: sm.raise_("light"), id="raise"), + ], + ) + async def test_a_declared_internal_event_returns_the_callback_result(self, sm_runner, trigger): + sm = await sm_runner.start(Beacons) + + assert await settle(trigger(sm)) == "kindled" + assert sm.configuration_values == {"lit"} + + async def test_an_internal_event_with_no_transition_raises_on_a_state_machine(self, sm_runner): + class StrictBeacons(StateMachine): + dark = State(initial=True) + lit = State() + + light = dark.to(lit) + douse = Event(lit.to(dark), internal=True) + + sm = await sm_runner.start(StrictBeacons) + + with pytest.raises(TransitionNotAllowed): + await sm_runner.send(sm, "douse") + + async def test_an_exception_from_a_declared_internal_event_reaches_the_caller(self, sm_runner): + class FailingBeacons(StateChart): + catch_errors_as_events = False + + dark = State(initial=True) + lit = State(final=True) + + light = Event(dark.to(lit, on="kindle"), internal=True) + + def kindle(self): + raise RuntimeError("the fire would not catch") + + sm = await sm_runner.start(FailingBeacons) + + with pytest.raises(RuntimeError, match="the fire would not catch"): + await sm_runner.send(sm, "light") + + async def test_a_fresh_async_machine_runs_a_declared_internal_event_on_its_first_await( + self, + ): + class AsyncBeacons(Beacons): + async def kindle(self): + return "kindled" + + sm = AsyncBeacons() + + assert await sm.light() == "kindled" + assert sm.configuration_values == {"lit"} + + @pytest.mark.parametrize("outcome", ["completes", "raises"]) + async def test_the_processing_loop_leaves_the_context_as_it_found_it(self, sm_runner, outcome): + class Signal(StateChart): + catch_errors_as_events = False + + dark = State(initial=True) + lit = State(final=True) + + light = Event(dark.to(lit, on="kindle"), internal=True) + fail = dark.to.itself(internal=True, on="stumble") + + def kindle(self): + return "kindled" + + def stumble(self): + raise RuntimeError("stumbled") + + sm = await sm_runner.start(Signal) + if outcome == "raises": + with pytest.raises(RuntimeError, match="stumbled"): + await sm_runner.send(sm, "fail") + + assert _in_processing_loop.get() is False + assert await settle(sm.raise_("light")) == "kindled" + + +class TestDelayedInternal: + def test_a_declaration_cannot_be_both_delayed_and_internal(self): + with pytest.raises(InvalidDefinition, match=DELAYED_INTERNAL): + + class DelayedBeacons(StateChart): + dark = State(initial=True) + lit = State(final=True) + + light = Event(dark.to(lit), delay=50, internal=True) + + def test_a_compound_declaration_cannot_be_both_delayed_and_internal(self): + with pytest.raises(InvalidDefinition, match=DELAYED_INTERNAL): + + class DelayedBeacons(StateChart): + class chain(State.Compound): + dark = State(initial=True) + lit = State(final=True) + + light = Event(dark.to(lit), delay=50, internal=True) + + @pytest.mark.parametrize( + "trigger", + [ + pytest.param(lambda sm: sm.raise_("light", delay=50), id="raise"), + pytest.param(lambda sm: sm.send("light", delay=50, internal=True), id="send"), + pytest.param(lambda sm: sm.send("light", delay=50), id="declared-internal"), + ], + ) + async def test_a_trigger_cannot_be_both_delayed_and_internal(self, sm_runner, trigger): + sm = await sm_runner.start(Beacons) + + with pytest.raises(InvalidDefinition, match=DELAYED_INTERNAL): + await settle(trigger(sm)) + + assert sm.configuration_values == {"dark"} + + def test_an_scxml_send_to_internal_cannot_have_a_delay(self): + scxml = """ + + + + + + + + + + """ + processor = SCXMLProcessor() + processor.parse_scxml("delayed_internal", scxml) + + with pytest.raises(InvalidDefinition, match=DELAYED_INTERNAL): + processor.start() diff --git a/tests/test_statechart_compound.py b/tests/test_statechart_compound.py index de1c52f3..61dafc01 100644 --- a/tests/test_statechart_compound.py +++ b/tests/test_statechart_compound.py @@ -337,19 +337,24 @@ class shire(State.Compound): await sm_runner.send(sm, "visit_pub") assert {"shire", "green_dragon"} == set(sm.configuration_values) - def test_name_delay_and_internal_are_preserved(self): + @pytest.mark.parametrize( + ("declared", "expected"), + [ + pytest.param({"delay": 50}, (50, False), id="delay"), + pytest.param({"internal": True}, (0, True), id="internal"), + ], + ) + def test_name_delay_and_internal_are_preserved(self, declared, expected): class NamedEvent(StateChart): class shire(State.Compound): bag_end = State(initial=True) green_dragon = State(final=True) - visit_pub = Event( - bag_end.to(green_dragon), name="Visit the pub", delay=50, internal=True - ) + visit_pub = Event(bag_end.to(green_dragon), name="Visit the pub", **declared) (registered,) = NamedEvent.events assert (registered.id, registered.name) == ("visit_pub", "Visit the pub") - assert (registered.delay, registered.internal) == (50, True) + assert (registered.delay, registered.internal) == expected assert NamedEvent.visit_pub.name == "Visit the pub" def test_expanded_id_drops_the_declared_arguments(self): From 106875b5c37a8f6ce3a6575ad7cb69a08d1f1c49 Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Sun, 11 Oct 2026 22:28:10 -0300 Subject: [PATCH 4/9] docs: describe which queue an internal event goes to and why The queue table, the raise_() and delayed-event sections and the async futures note now say that an internal event joins the macrostep in progress, goes to the external queue when triggered from outside a callback, and cannot carry a delay. The 3.2.2 notes describe the fixes behind it. Signed-off-by: Fernando Macedo --- docs/async.md | 9 +++---- docs/processing_model.md | 42 +++++++++++++++++++++++++++++---- docs/releases/3.2.2.md | 51 +++++++++++++++++++++++++++++++++++++--- docs/statechart.md | 15 ++++++++++++ 4 files changed, 106 insertions(+), 11 deletions(-) diff --git a/docs/async.md b/docs/async.md index 313d03f8..9c3bb367 100644 --- a/docs/async.md +++ b/docs/async.md @@ -155,10 +155,11 @@ resolves each event's future as it processes the queue. Callers that didn't acquire the lock simply `await` their future. ```{note} -Futures are only created for **external** events sent from outside the -processing loop. Events triggered from within callbacks (via `send()` or -`raise_()`) follow the {ref}`run-to-completion ` model — they -are enqueued and processed within the current macrostep. +Futures are created for every event sent from outside the processing loop, +including one declared `internal=True`, which goes to the external queue +when there is no macrostep to join. Events triggered from within callbacks +(via `send()` or `raise_()`) follow the {ref}`run-to-completion ` +model: they are enqueued and processed by the running loop. ``` If an exception occurs during processing (with `catch_errors_as_events=False`), diff --git a/docs/processing_model.md b/docs/processing_model.md index 16bceb03..7dd99a58 100644 --- a/docs/processing_model.md +++ b/docs/processing_model.md @@ -76,17 +76,51 @@ After the macrostep completes, the engine picks the next event from the ### Event queues -The engine maintains two separate FIFO queues: +The engine maintains two separate queues, each ordered by due time. Events due at the same +moment run in the order they were sent. | Queue | How to enqueue | When processed | |---|---|---| -| **Internal** | {func}`raise_() `, `send(..., internal=True)`, or an event declared with `Event(..., internal=True)` | Within the current macrostep | -| **External** | {func}`send() ` | After the current macrostep ends | +| **Internal** | {func}`raise_() `, `send(..., internal=True)`, or an event declared with `Event(..., internal=True)`, from a callback | Within the current macrostep | +| **External** | {func}`send() `, and any event triggered from outside a callback | After the current macrostep ends | This distinction matters when you trigger events from inside callbacks. Using `raise_()` ensures the event is handled as part of the current processing cycle, while `send()` defers it to after the machine reaches -a stable configuration. +a stable configuration. An event declared `internal=True` behaves like +`raise_()` without spelling it at each call: + +```py +>>> from statemachine import Event, State, StateChart + +>>> class Muster(StateChart): +... camp = State(initial=True) +... riding = State(final=True) +... besieged = State(final=True) +... +... summon = camp.to.itself(internal=True, on="call") +... ride = Event(camp.to(riding), internal=True) +... siege = camp.to(besieged) +... +... def call(self): +... self.send("siege") # after this macrostep +... self.ride() # within this macrostep, so first + +>>> sm = Muster() +>>> sm.send("summon") +>>> sm.riding.is_active +True + +``` + +The internal queue belongs to the macrostep in progress. Triggered from +outside a callback, there is no macrostep to join: an internal event goes +to the external queue, and the call returns its result like `send()`. + +A delay cannot be internal, since a macrostep takes no time. Declaring or +sending an event with both `delay` and `internal=True`, or calling +`raise_()` with a `delay`, raises {class}`~statemachine.exceptions.InvalidDefinition`. +Use `send(..., delay=...)` to deliver an event later. ```{seealso} See {ref}`sending-events` for examples of `send()` vs `raise_()`. diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md index fec9ca2b..d4613bcc 100644 --- a/docs/releases/3.2.2.md +++ b/docs/releases/3.2.2.md @@ -117,7 +117,7 @@ Delays are now measured on `time.monotonic()`, which no setting of the wall cloc `Event(dark.to(lit), delay=50)` rebuilt the event without its `delay`, so `BeaconsOfGondor.light.delay` was `0` and the event fired immediately instead of being queued. -The same happened to `internal`. Both are now preserved: +The same happened to `internal`. Each is now preserved: ```py >>> class BeaconsOfGondor(StateChart): @@ -135,9 +135,54 @@ The same happened to `internal`. Both are now preserved: `Event(a.to(b), internal=True)` kept `internal` on the declaration, but neither `sm.go()` nor `sm.send("go")` carried it to the queue, so the event always went to the external queue. Both -paths now honor the declaration and put the event on the internal queue. +now honor the declaration: triggered from a callback, the event runs within the current +macrostep, before any pending external event. -[#655](https://github.com/fgmacedo/python-statemachine/issues/655). +Honoring it exposed how the internal queue treated a trigger that no running step had raised. +On the async engine, the first `await sm.go()` on a fresh machine was lost before the initial +state was entered. The caller got `None` instead of the callback result, and no +`TransitionNotAllowed` on a `StateMachine`. The internal queue now receives only what the +machine raises while it runs a step. Triggered from outside, an internal event goes to the +external queue and returns its result, as `send()` does: + +```py +>>> class Beacons(StateChart): +... dark = State(initial=True) +... lit = State(final=True) +... +... light = Event(dark.to(lit, on="kindle"), internal=True) +... +... def kindle(self): +... return "kindled" + +>>> Beacons().light() +'kindled' + +``` + +A delay never reached the internal queue either: `raise_("go", delay=50)` fired at once. A +macrostep takes no time, and SCXML forbids `delay` on a `` to `#_internal`, so a delayed +internal event is now an invalid definition, whether declared, sent or raised: + +```py +>>> class DelayedBeacons(StateChart): +... dark = State(initial=True) +... lit = State(final=True) +... +... light = Event(dark.to(lit), delay=50, internal=True) +Traceback (most recent call last): +... +InvalidDefinition: Event 'light' cannot be both delayed and internal: ... + +``` + +Drop `internal` to deliver the event later on the external queue, or drop `delay` to run it +within the current macrostep. + +Contributed by [@chiliec](https://github.com/chiliec). + +[#655](https://github.com/fgmacedo/python-statemachine/issues/655), +[#663](https://github.com/fgmacedo/python-statemachine/pull/663). ### `is_active` on a state nested in a parallel region diff --git a/docs/statechart.md b/docs/statechart.md index e06ed251..2f4dde39 100644 --- a/docs/statechart.md +++ b/docs/statechart.md @@ -88,6 +88,17 @@ True ``` +Called from outside a callback, `raise_()` has no macrostep to join, so it +behaves like `send()` and returns the result: + +```py +>>> sm = WithInternalEvent() +>>> sm.raise_("go") +>>> sm.c.is_active +True + +``` + Both methods accept arbitrary keyword arguments that are forwarded to all callbacks via {ref}`dependency injection `. @@ -115,6 +126,10 @@ sm.cancel_event("my_timeout") On the async engine, a coroutine awaiting the `send()` of a cancelled event gets `None`. +Only `send()` takes a delay. An internal event runs within the current macrostep, +which takes no time, so `raise_(..., delay=...)` and `send(..., delay=..., internal=True)` +raise {class}`~statemachine.exceptions.InvalidDefinition`. + ```{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 From 6f628113a63581e5df7101cb7beede6b4948e758 Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Sun, 11 Oct 2026 22:29:42 -0300 Subject: [PATCH 5/9] docs: link the 3.2.2 notes to #668 Signed-off-by: Fernando Macedo --- docs/releases/3.2.2.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md index d4613bcc..ac1ff5f3 100644 --- a/docs/releases/3.2.2.md +++ b/docs/releases/3.2.2.md @@ -182,7 +182,8 @@ within the current macrostep. Contributed by [@chiliec](https://github.com/chiliec). [#655](https://github.com/fgmacedo/python-statemachine/issues/655), -[#663](https://github.com/fgmacedo/python-statemachine/pull/663). +[#663](https://github.com/fgmacedo/python-statemachine/pull/663), +[#668](https://github.com/fgmacedo/python-statemachine/pull/668). ### `is_active` on a state nested in a parallel region @@ -252,3 +253,5 @@ True Each external event is now followed by its macrostep before the next one is taken, on both engines. + +[#668](https://github.com/fgmacedo/python-statemachine/pull/668). From 1339f03e0b49d8900759090485c10013c3bd9d0f Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Sun, 11 Oct 2026 23:06:00 -0300 Subject: [PATCH 6/9] fix: scope the running step to one machine and test it per thread and task The context variable that marks a running step held a bool, so a callback of one machine that raised an internal event on another put it on the other machine's internal queue, and the caller got None. It now holds the engine running the step, and `put` honors `internal` only for that engine. Tests pin why a context variable is the right primitive: a coroutine sending while another holds the async loop gets its own result, a thread raising while another thread runs the sync loop queues behind the events it sent before, and a callback raising on another machine gets that machine's result. Each fails against the alternative it rules out (a bool flag, or the engine's lock). Signed-off-by: Fernando Macedo --- AGENTS.md | 3 +- statemachine/engines/async_.py | 6 +-- statemachine/engines/base.py | 26 +++++----- statemachine/engines/sync.py | 4 +- tests/test_internal_triggers.py | 92 ++++++++++++++++++++++++++++++++- 5 files changed, 111 insertions(+), 20 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index ab2e4638..fc4db48a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,7 +41,8 @@ The engine follows the SCXML run-to-completion (RTC) model with two processing l - `send()` → **external queue** (processed after current macrostep ends). - `raise_()`, `send(internal=True)`, `Event(internal=True)` → **internal queue** (processed within the current macrostep, before external events), but only when raised while the - machine runs a step (`_in_processing_loop` in `engines/base.py`). From outside a step there + machine runs a step in the same thread or task (`_running_engine` in `engines/base.py`, a + `ContextVar` so it is right for threads and coroutines alike). From outside a step there is no macrostep to join: the trigger goes to the external queue, so its caller gets the result, the exception or `TransitionNotAllowed`. - `delay` and `internal` together raise `InvalidDefinition`: a step takes no time. diff --git a/statemachine/engines/async_.py b/statemachine/engines/async_.py index eec3da60..4c17a0e1 100644 --- a/statemachine/engines/async_.py +++ b/statemachine/engines/async_.py @@ -12,8 +12,8 @@ from ..state import State from .base import _ERROR_EXECUTION from .base import BaseEngine -from .base import _in_processing_loop from .base import _processing_step +from .base import _running_engine if TYPE_CHECKING: from ..transition import Transition @@ -36,7 +36,7 @@ def put(self, trigger_data: TriggerData, internal: bool = False): - The call is NOT from within the processing loop (reentrant calls from callbacks must not get futures, as that would deadlock) """ - if trigger_data.future is None and not _in_processing_loop.get(): + if trigger_data.future is None and _running_engine.get() is None: try: loop = asyncio.get_running_loop() trigger_data.future = loop.create_future() @@ -352,7 +352,7 @@ async def processing_loop(self, caller_future: "asyncio.Future[object] | None" = self._debug("%s Processing loop started: %s", self._log_id, self.sm.current_state_value) first_result = self._sentinel try: - with _processing_step(): + with _processing_step(self): took_events = True while took_events and self.running: self.clear_cache() diff --git a/statemachine/engines/base.py b/statemachine/engines/base.py index 938f937c..81cafdfa 100644 --- a/statemachine/engines/base.py +++ b/statemachine/engines/base.py @@ -29,22 +29,23 @@ logger = logging.getLogger(__name__) -_in_processing_loop: ContextVar[bool] = ContextVar("_in_processing_loop", default=False) -"""Whether the current context is running a step of an engine. +_running_engine: "ContextVar[BaseEngine | None]" = ContextVar("_running_engine", default=None) +"""The engine whose step the current thread or task is running, if any. -asyncio copies the context into the tasks a step creates, so callbacks see it set. A new -thread or an independent coroutine starts without it. +A context variable is per thread and per asyncio task: asyncio copies it into the tasks a +step creates, so callbacks see it, while another thread or an independent coroutine sending +to the same machine does not. """ @contextmanager -def _processing_step() -> Iterator[None]: - """Mark the current context as running a step of an engine, until the block ends.""" - token = _in_processing_loop.set(True) +def _processing_step(engine: "BaseEngine") -> Iterator[None]: + """Mark the current context as running a step of ``engine``, until the block ends.""" + token = _running_engine.set(engine) try: yield finally: - _in_processing_loop.reset(token) + _running_engine.reset(token) @dataclass(frozen=True, unsafe_hash=True, eq=True) @@ -155,14 +156,15 @@ def clear_cache(self): def put(self, trigger_data: TriggerData, internal: bool = False): """Put the trigger on the queue without blocking the caller. - An ``internal`` trigger joins the internal queue only when the machine raises it while - running a step. Put from outside, there is no macrostep to join, so it goes to the - external queue, where its caller gets its result. + An ``internal`` trigger joins the internal queue only when this machine raises it while + running a step. Put from outside, including from a step of another machine, there is + no macrostep to join, so it goes to the external queue, where its caller gets its + result. """ if not self.running and not self.sm.allow_event_without_transition: raise TransitionNotAllowed(trigger_data.event, self.sm.configuration) - internal = internal and _in_processing_loop.get() + internal = internal and _running_engine.get() is self if internal: self.internal_queue.put(trigger_data) else: diff --git a/statemachine/engines/sync.py b/statemachine/engines/sync.py index 2433c526..d4bcc334 100644 --- a/statemachine/engines/sync.py +++ b/statemachine/engines/sync.py @@ -39,7 +39,7 @@ def activate_initial_state(self, **kwargs): transitions = self._initial_transitions(trigger_data) self._processing.acquire(blocking=False) try: - with _processing_step(): + with _processing_step(self): self._enter_states(transitions, trigger_data, OrderedSet(), OrderedSet()) finally: self._processing.release() @@ -66,7 +66,7 @@ def processing_loop(self): self._debug("%s Processing loop started: %s", self._log_id, self.sm.current_state_value) first_result = self._sentinel try: - with _processing_step(): + with _processing_step(self): took_events = True while took_events and self.running: self.clear_cache() diff --git a/tests/test_internal_triggers.py b/tests/test_internal_triggers.py index 94608cc1..be79acf3 100644 --- a/tests/test_internal_triggers.py +++ b/tests/test_internal_triggers.py @@ -9,10 +9,12 @@ Theme: the beacons of Gondor, lit from inside the chain or called for from afar. """ +import asyncio +import threading from inspect import isawaitable import pytest -from statemachine.engines.base import _in_processing_loop +from statemachine.engines.base import _running_engine from statemachine.exceptions import InvalidDefinition from statemachine.exceptions import TransitionNotAllowed from statemachine.io.scxml.processor import SCXMLProcessor @@ -159,10 +161,96 @@ def stumble(self): with pytest.raises(RuntimeError, match="stumbled"): await sm_runner.send(sm, "fail") - assert _in_processing_loop.get() is False + assert _running_engine.get() is None assert await settle(sm.raise_("light")) == "kindled" +class TestWhoIsRunningTheStep: + """The internal queue belongs to the step of one machine, in one thread or task.""" + + @pytest.mark.timeout(10) + async def test_a_coroutine_outside_the_loop_gets_its_result_while_another_holds_it(self): + paused = asyncio.Event() + resume = asyncio.Event() + + class Relay(StateChart): + dark = State(initial=True) + waiting = State() + lit = State(final=True) + + hold = dark.to(waiting, on="pause") + light = Event(waiting.to(lit, on="kindle"), internal=True) + + async def pause(self): + paused.set() + await resume.wait() + return "held" + + def kindle(self): + return "kindled" + + async def light(): + return await sm.light() + + sm = Relay() + await sm.activate_initial_state() + holding = asyncio.create_task(sm.send("hold")) + await paused.wait() + lighting = asyncio.create_task(light()) + await asyncio.sleep(0) + resume.set() + + assert await asyncio.gather(holding, lighting) == ["held", "kindled"] + + @pytest.mark.timeout(10) + def test_a_thread_raising_while_another_runs_the_loop_queues_behind_its_sends(self): + in_step = threading.Event() + release = threading.Event() + + class Watch(StateChart): + dark = State(initial=True) + waiting = State() + lit = State(final=True) + besieged = State(final=True) + + hold = dark.to(waiting, on="pause") + light = waiting.to(lit) + siege = waiting.to(besieged) + + def pause(self): + in_step.set() + release.wait(timeout=5) + + sm = Watch() + holder = threading.Thread(target=sm.send, args=("hold",)) + holder.start() + assert in_step.wait(timeout=5) + + sm.send("siege") + sm.raise_("light") + release.set() + holder.join(timeout=5) + + assert sm.configuration_values == {"besieged"} + + async def test_a_callback_raising_on_another_machine_gets_its_result(self, sm_runner): + other = await sm_runner.start(Beacons) + + class Lighter(StateChart): + idle = State(initial=True) + done = State(final=True) + + reach = idle.to(done, on="light_other") + + def light_other(self): + return other.raise_("light") + + sm = await sm_runner.start(Lighter) + + assert await settle(await sm_runner.send(sm, "reach")) == "kindled" + assert other.configuration_values == {"lit"} + + class TestDelayedInternal: def test_a_declaration_cannot_be_both_delayed_and_internal(self): with pytest.raises(InvalidDefinition, match=DELAYED_INTERNAL): From b14eed571e79e07089e808725130033e9acc90fe Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Sun, 11 Oct 2026 23:12:38 -0300 Subject: [PATCH 7/9] fix: reject delayed internal sends at load time and let send() override internal A `` to `#_internal` with `delay` or `delayexpr`, in SCXML or in the native JSON/YAML format, loaded fine and then raised InvalidDefinition from the middle of a transition. It is now rejected when the action is built, through the same message as `Event`, and the JSON Schema rejects it too, so `validate=True`, the parser and the runtime agree. `send()` took `internal` as `internal or `, so an explicit `internal=False` could not override an event declared internal, despite the docstring. It now defaults to None and falls back to the declaration only when omitted. Also pins the cross-machine contract on a StateMachine: a callback triggering a disallowed internal event on another machine raises TransitionNotAllowed. Signed-off-by: Fernando Macedo --- statemachine/event.py | 22 +++-- statemachine/io/actions.py | 6 +- .../io/schemas/statechart.schema.json | 9 +- statemachine/statemachine.py | 5 +- tests/io/test_validation.py | 13 +++ tests/test_internal_triggers.py | 87 +++++++++++++++++-- 6 files changed, 124 insertions(+), 18 deletions(-) diff --git a/statemachine/event.py b/statemachine/event.py index 6bac8f19..4ae6e2b3 100644 --- a/statemachine/event.py +++ b/statemachine/event.py @@ -46,6 +46,20 @@ def _expand_event_id(key: str) -> str: } +def raise_delayed_internal(event_id: str): + """Reject an event that asks to be both delayed and internal. + + The internal queue runs within the current macrostep, which takes no time. + """ + raise InvalidDefinition( + _( + "Event {id!r} cannot be both delayed and internal: the internal queue runs " + "within the current macrostep, which takes no time. Drop 'internal' to " + "deliver it later on the external queue, or drop 'delay' to run it now." + ).format(id=event_id) + ) + + class Event(AddCallbacksMixin, str): """An event triggers a signal that something has happened. @@ -101,13 +115,7 @@ def __new__( id = str(id) if _has_real_id else f"__event__{uuid4().hex}" if delay and internal: - raise InvalidDefinition( - _( - "Event {id!r} cannot be both delayed and internal: the internal queue runs " - "within the current macrostep, which takes no time. Drop 'internal' to " - "deliver it later on the external queue, or drop 'delay' to run it now." - ).format(id=id) - ) + raise_delayed_internal(id) instance = super().__new__(cls, id) instance.id = id diff --git a/statemachine/io/actions.py b/statemachine/io/actions.py index cc5c17ac..11056e56 100644 --- a/statemachine/io/actions.py +++ b/statemachine/io/actions.py @@ -17,6 +17,7 @@ from ..event import BoundEvent from ..event import Event +from ..event import raise_delayed_internal from ..statemachine import StateChart from .evaluators import Evaluator from .evaluators import protected_attrs @@ -400,6 +401,9 @@ def create_send_action_callable( # noqa: C901 ) -> Callable: content: Any = () _valid_targets = (None, "#_internal", "internal", "#_parent", "parent") + internal = action.target in ("#_internal", "internal") + if internal and (action.delay or action.delayexpr): + raise_delayed_internal(action.event or action.eventexpr or "") if action.content: content = (evaluator.eval_literal(action.content),) delay_expr = ( @@ -433,8 +437,6 @@ def send_action(*args, **kwargs): # noqa: C901 _send_to_parent(action, evaluator, **kwargs) return - internal = target in ("#_internal", "internal") - send_id = None if action.id: send_id = action.id diff --git a/statemachine/io/schemas/statechart.schema.json b/statemachine/io/schemas/statechart.schema.json index b9000073..091d707b 100644 --- a/statemachine/io/schemas/statechart.schema.json +++ b/statemachine/io/schemas/statechart.schema.json @@ -117,7 +117,14 @@ "params": { "type": "array", "items": { "$ref": "#/$defs/param" } }, "content": { "type": "string" } }, - "additionalProperties": false + "additionalProperties": false, + "if": { + "required": ["target"], + "properties": { "target": { "enum": ["#_internal", "internal"] } } + }, + "then": { + "not": { "anyOf": [{ "required": ["delay"] }, { "required": ["delayexpr"] }] } + } }, "foreach": { "type": "object", diff --git a/statemachine/statemachine.py b/statemachine/statemachine.py index 41a23ade..fa805ed9 100644 --- a/statemachine/statemachine.py +++ b/statemachine/statemachine.py @@ -467,7 +467,7 @@ def send( *args, delay: float = 0, send_id: "str | None" = None, - internal: bool = False, + internal: "bool | None" = None, **kwargs, ) -> Any: """Send an :ref:`Event` to the state machine. @@ -492,7 +492,8 @@ def send( delay = ( delay if delay else know_event and know_event.delay or 0 ) # first the param, then the event, or 0 - internal = internal or bool(know_event and know_event.internal) + if internal is None: + internal = bool(know_event and know_event.internal) event_instance = BoundEvent( id=event, name=event_name, delay=delay, internal=internal, _sm=self ) diff --git a/tests/io/test_validation.py b/tests/io/test_validation.py index a7c13f83..030a78a4 100644 --- a/tests/io/test_validation.py +++ b/tests/io/test_validation.py @@ -166,6 +166,19 @@ def test_invalid_callback_ref_in_nested_action_list(self): } ) + @pytest.mark.parametrize("target", ["#_internal", "internal"]) + @pytest.mark.parametrize("delay", ["delay", "delayexpr"]) + def test_invalid_delayed_send_to_internal(self, target, delay): + send = {"event": "e", "target": target, delay: "1s"} + with pytest.raises(InvalidDefinition, match="failed schema validation"): + validate_document({"states": {"a": {"initial": True, "enter": [{"send": send}]}}}) + + @pytest.mark.parametrize( + "send", [{"event": "e", "target": "#_internal"}, {"event": "e", "delay": "1s"}] + ) + def test_valid_send_to_internal_or_delayed(self, send): + validate_document({"states": {"a": {"initial": True, "enter": [{"send": send}]}}}) + def test_invalid_invoke_extra_key(self): with pytest.raises(InvalidDefinition, match="failed schema validation"): validate_document({"states": {"a": {"initial": True, "invoke": {"bogus": 1}}}}) diff --git a/tests/test_internal_triggers.py b/tests/test_internal_triggers.py index be79acf3..f0132d6c 100644 --- a/tests/test_internal_triggers.py +++ b/tests/test_internal_triggers.py @@ -10,6 +10,7 @@ """ import asyncio +import json import threading from inspect import isawaitable @@ -17,6 +18,7 @@ from statemachine.engines.base import _running_engine from statemachine.exceptions import InvalidDefinition from statemachine.exceptions import TransitionNotAllowed +from statemachine.io.loader import load from statemachine.io.scxml.processor import SCXMLProcessor from statemachine import Event @@ -81,6 +83,26 @@ def on_enter_dark(self): assert sm.configuration_values == {"lit"} + def test_send_with_internal_false_overrides_the_declaration(self): + class Muster(StateChart): + camp = State(initial=True) + riding = State(final=True) + besieged = State(final=True) + + summon = camp.to.itself(internal=True, on="call") + ride = Event(camp.to(riding), internal=True) + siege = camp.to(besieged) + + def call(self): + self.send("siege") + self.send("ride", internal=False) + + sm = Muster() + + sm.send("summon") + + assert sm.configuration_values == {"besieged"} + class TestFromOutside: @pytest.mark.parametrize( @@ -250,6 +272,34 @@ def light_other(self): assert await settle(await sm_runner.send(sm, "reach")) == "kindled" assert other.configuration_values == {"lit"} + async def test_a_callback_triggering_a_disallowed_event_on_another_machine_raises( + self, sm_runner + ): + class StrictBeacons(StateMachine): + dark = State(initial=True) + lit = State() + + light = dark.to(lit) + douse = Event(lit.to(dark), internal=True) + + other = await sm_runner.start(StrictBeacons) + + class Douser(StateChart): + catch_errors_as_events = False + + idle = State(initial=True) + done = State(final=True) + + reach = idle.to(done, on="douse_other") + + def douse_other(self): + return other.douse() + + sm = await sm_runner.start(Douser) + + with pytest.raises(TransitionNotAllowed): + await settle(await sm_runner.send(sm, "reach")) + class TestDelayedInternal: def test_a_declaration_cannot_be_both_delayed_and_internal(self): @@ -287,20 +337,45 @@ async def test_a_trigger_cannot_be_both_delayed_and_internal(self, sm_runner, tr assert sm.configuration_values == {"dark"} - def test_an_scxml_send_to_internal_cannot_have_a_delay(self): - scxml = """ + @pytest.mark.timeout(10) + async def test_a_declared_internal_event_can_be_sent_later_as_external(self, sm_runner): + sm = await sm_runner.start(Beacons) + + assert await sm_runner.send(sm, "light", delay=10, internal=False) == "kindled" + + @pytest.mark.parametrize("delay", ['delay="50ms"', "delayexpr=\"'50ms'\""]) + @pytest.mark.parametrize("target", ["#_internal", "internal"]) + def test_an_scxml_send_to_internal_with_a_delay_fails_to_load(self, target, delay): + scxml = f""" - + """ - processor = SCXMLProcessor() - processor.parse_scxml("delayed_internal", scxml) with pytest.raises(InvalidDefinition, match=DELAYED_INTERNAL): - processor.start() + SCXMLProcessor().parse_scxml("delayed_internal", scxml) + + @pytest.mark.parametrize("validate", [True, False]) + def test_a_native_send_to_internal_with_a_delay_fails_to_load(self, validate): + document = { + "name": "DelayedInternal", + "states": { + "dark": { + "initial": True, + "enter": [ + {"send": {"event": "light", "target": "#_internal", "delay": "50ms"}} + ], + "transitions": [{"event": "light", "target": "lit"}], + }, + "lit": {"final": True}, + }, + } + + with pytest.raises(InvalidDefinition): + load(json.dumps(document), format="json", validate=validate) From 5f43965702c616efb89de01e9ef3bfbd54e0fac0 Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Sun, 11 Oct 2026 23:12:51 -0300 Subject: [PATCH 8/9] docs: show raise_() returning a result and note the per-event macrostep cost Signed-off-by: Fernando Macedo --- docs/releases/3.2.2.md | 12 +++++++----- docs/statechart.md | 15 +++++++++++---- 2 files changed, 18 insertions(+), 9 deletions(-) diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md index ac1ff5f3..8384c8b5 100644 --- a/docs/releases/3.2.2.md +++ b/docs/releases/3.2.2.md @@ -162,7 +162,8 @@ external queue and returns its result, as `send()` does: A delay never reached the internal queue either: `raise_("go", delay=50)` fired at once. A macrostep takes no time, and SCXML forbids `delay` on a `` to `#_internal`, so a delayed -internal event is now an invalid definition, whether declared, sent or raised: +internal event is now an invalid definition, whether declared, sent, raised, or loaded +from an SCXML, JSON or YAML `` to `#_internal` (also rejected by the JSON Schema): ```py >>> class DelayedBeacons(StateChart): @@ -177,7 +178,8 @@ InvalidDefinition: Event 'light' cannot be both delayed and internal: ... ``` Drop `internal` to deliver the event later on the external queue, or drop `delay` to run it -within the current macrostep. +within the current macrostep. An explicit `send(..., internal=False)` overrides the declaration, so +`send("light", delay=50, internal=False)` delivers a declared-internal event later. Contributed by [@chiliec](https://github.com/chiliec). @@ -229,8 +231,6 @@ event raised while handling the first external event ran after the second, break run-to-completion order the {ref}`processing model ` describes: ```py ->>> from statemachine.event import BoundEvent - >>> class Beacons(StateChart): ... dark = State(initial=True) ... lit = State(final=True) @@ -252,6 +252,8 @@ True ``` Each external event is now followed by its macrostep before the next one is taken, on both -engines. +engines. Eventless transitions are now checked after every external event instead of once per +batch, so a burst of queued external events costs one guard evaluation each, and an event +queued behind one that reaches a final state is no longer processed. [#668](https://github.com/fgmacedo/python-statemachine/pull/668). diff --git a/docs/statechart.md b/docs/statechart.md index 2f4dde39..3c8e3fe4 100644 --- a/docs/statechart.md +++ b/docs/statechart.md @@ -92,10 +92,17 @@ Called from outside a callback, `raise_()` has no macrostep to join, so it behaves like `send()` and returns the result: ```py ->>> sm = WithInternalEvent() ->>> sm.raise_("go") ->>> sm.c.is_active -True +>>> class Greeter(StateChart): +... idle = State(initial=True) +... done = State(final=True) +... +... greet = idle.to(done, on="say_hello") +... +... def say_hello(self): +... return "hello" + +>>> Greeter().raise_("greet") +'hello' ``` From e03da9cf799ec4df532109019e4d990f86a0af82 Mon Sep 17 00:00:00 2001 From: Fernando Macedo Date: Mon, 12 Oct 2026 06:58:23 -0300 Subject: [PATCH 9/9] docs: name the event in the 3.2.2 internal declaration example Signed-off-by: Fernando Macedo --- docs/releases/3.2.2.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/releases/3.2.2.md b/docs/releases/3.2.2.md index 8384c8b5..81e11eab 100644 --- a/docs/releases/3.2.2.md +++ b/docs/releases/3.2.2.md @@ -133,8 +133,8 @@ The same happened to `internal`. Each is now preserved: ### `internal` declared on an `Event` ignored at trigger time -`Event(a.to(b), internal=True)` kept `internal` on the declaration, but neither `sm.go()` nor -`sm.send("go")` carried it to the queue, so the event always went to the external queue. Both +`go = Event(a.to(b), internal=True)` kept `internal` on the declaration, but neither `sm.go()` +nor `sm.send("go")` carried it to the queue, so the event always went to the external queue. Both now honor the declaration: triggered from a callback, the event runs within the current macrostep, before any pending external event.