test: add thread safety stress tests and document thread safety

Add TestThreadSafety with stress tests exercising real contention —
multiple threads sending events to the same SM simultaneously via
barriers. Tests verify no lost events, state consistency, correct
callback counts, and safe concurrent reads.

Document thread safety guarantees in docs/processing_model.md (linking
to atomic_configuration_update for transient None behavior) and
AGENTS.md, noting the PriorityQueue-based event queue must remain
thread-safe.

Author: fgmacedo

AGENTS.md

@@ -77,6 +77,16 @@ current event.
 - `on_error_execution()` works via naming convention but **only** when a transition for
   `error.execution` is declared — it is NOT a generic callback.
 
+### Thread safety
+
+- The sync engine is **thread-safe**: multiple threads can send events to the same SM instance
+  concurrently. The processing loop uses a `threading.Lock` so at most one thread executes
+  transitions at a time. Event queues use `PriorityQueue` (stdlib, thread-safe).
+- **Do not replace `PriorityQueue`** with non-thread-safe alternatives (e.g., `collections.deque`,
+  plain `list`) — this would break concurrent access guarantees.
+- Stress tests in `tests/test_threading.py::TestThreadSafety` exercise real contention with
+  barriers and multiple sender threads. Any change to queue or locking internals must pass these.
+
 ### Invoke (`<invoke>`)
 
 - `invoke.py` — `InvokeManager` on the engine manages the lifecycle: `mark_for_invoke()`,

docs/processing_model.md

@@ -315,3 +315,50 @@ The machine starts, enters `trying` (attempt 1), and the eventless
 self-transition keeps firing as long as `can_retry()` returns `True`. Once
 the limit is reached, the second eventless transition fires — all within a
 single macrostep triggered by initialization.
+
+
+(thread-safety)=
+
+## Thread safety
+
+State machines are **thread-safe** for concurrent event sending. Multiple threads
+can call `send()` or trigger events on the **same state machine instance**
+simultaneously — the engine guarantees correct behavior through its internal
+locking mechanism.
+
+### How it works
+
+The processing loop uses a non-blocking lock (`threading.Lock`). When a thread
+sends an event:
+
+1. The event is placed on the **external queue** (backed by a thread-safe
+   `PriorityQueue` from the standard library).
+2. If no other thread is currently running the processing loop, the sending
+   thread acquires the lock and processes all queued events.
+3. If another thread is already processing, the event is simply enqueued and
+   will be processed by the thread that holds the lock — no event is lost.
+
+This means that **at most one thread executes transitions at any time**, preserving
+the run-to-completion (RTC) guarantee while allowing safe concurrent access.
+
+### What is safe
+
+- **Multiple threads sending events** to the same state machine instance.
+- **Reading state** (`current_state_value`, `configuration`) from any thread
+  while events are being processed. Note that transient `None` values may be
+  observed for `current_state_value` during configuration updates when using
+  [`atomic_configuration_update`](behaviour.md#atomic_configuration_update) `= False`
+  (the default on `StateChart`, SCXML-compliant). With `atomic_configuration_update = True`
+  (the default on `StateMachine`), the configuration is updated atomically at
+  the end of the microstep, so `None` is not observed.
+- **Invoke handlers** running in background threads or thread executors
+  communicate with the parent machine via the thread-safe event queue.
+
+### What to avoid
+
+- **Do not share a state machine instance across threads with the async engine**
+  unless you ensure only one event loop drives the machine. The async engine is
+  designed for `asyncio` concurrency, not thread-based concurrency.
+- **Callbacks execute in the processing thread**, not in the thread that sent
+  the event. Design callbacks accordingly (e.g., use locks if they access
+  shared external state).

tests/test_threading.py

@@ -1,6 +1,8 @@
 import threading
 import time
+from collections import Counter
 
+import pytest
 from statemachine.state import State
 from statemachine.statemachine import StateChart
 
@@ -115,6 +117,184 @@ def __init__(self, name):
     assert c3.fsm.statuses_history == ["c3.green", "c3.green", "c3.green", "c3.yellow"]
 
 
+class TestThreadSafety:
+    """Stress tests for concurrent access to a single state machine instance.
+
+    These tests exercise real contention: multiple threads sending events to the
+    same SM simultaneously, synchronized via barriers to maximize overlap.
+    """
+
+    @pytest.fixture()
+    def cycling_machine(self):
+        class CyclingMachine(StateChart):
+            s1 = State(initial=True)
+            s2 = State()
+            s3 = State()
+            cycle = s1.to(s2) | s2.to(s3) | s3.to(s1)
+
+        return CyclingMachine()
+
+    @pytest.mark.parametrize("num_threads", [4, 8])
+    def test_concurrent_sends_no_lost_events(self, cycling_machine, num_threads):
+        """All events sent concurrently must be processed — none lost."""
+        events_per_thread = 300
+        total_events = num_threads * events_per_thread
+        barrier = threading.Barrier(num_threads)
+        errors = []
+
+        def sender():
+            try:
+                barrier.wait(timeout=5)
+                for _ in range(events_per_thread):
+                    cycling_machine.send("cycle")
+            except Exception as e:
+                errors.append(e)
+
+        threads = [threading.Thread(target=sender) for _ in range(num_threads)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=30)
+
+        assert not errors, f"Thread errors: {errors}"
+
+        # The machine cycles s1→s2→s3→s1. After N total cycle events starting
+        # from s1, the state is determined by (N % 3).
+        expected_states = {0: "s1", 1: "s2", 2: "s3"}
+        expected = expected_states[total_events % 3]
+        assert cycling_machine.current_state_value == expected
+
+    def test_concurrent_sends_state_consistency(self, cycling_machine):
+        """State must always be one of the valid states, never corrupted."""
+        valid_values = {"s1", "s2", "s3"}
+        num_threads = 6
+        events_per_thread = 500
+        barrier = threading.Barrier(num_threads + 1)  # +1 for observer
+        stop_event = threading.Event()
+        observed_values = []
+        errors = []
+
+        def sender():
+            try:
+                barrier.wait(timeout=5)
+                for _ in range(events_per_thread):
+                    cycling_machine.send("cycle")
+            except Exception as e:
+                errors.append(e)
+
+        def observer():
+            barrier.wait(timeout=5)
+            while not stop_event.is_set():
+                val = cycling_machine.current_state_value
+                observed_values.append(val)
+
+        threads = [threading.Thread(target=sender) for _ in range(num_threads)]
+        obs_thread = threading.Thread(target=observer)
+
+        for t in threads:
+            t.start()
+        obs_thread.start()
+
+        for t in threads:
+            t.join(timeout=30)
+
+        stop_event.set()
+        obs_thread.join(timeout=5)
+
+        assert not errors, f"Thread errors: {errors}"
+        # None may appear transiently during configuration updates — that's expected.
+        invalid = [v for v in observed_values if v not in valid_values and v is not None]
+        assert not invalid, f"Observed invalid state values: {set(invalid)}"
+        assert len(observed_values) > 100, "Observer didn't collect enough samples"
+
+    def test_concurrent_sends_with_callbacks(self):
+        """Callbacks must execute exactly once per transition under contention."""
+        call_log = []
+        lock = threading.Lock()
+
+        class CallbackMachine(StateChart):
+            s1 = State(initial=True)
+            s2 = State()
+            go = s1.to(s2) | s2.to(s1)
+
+            def on_enter_s2(self):
+                with lock:
+                    call_log.append("enter_s2")
+
+            def on_enter_s1(self):
+                with lock:
+                    call_log.append("enter_s1")
+
+        sm = CallbackMachine()
+        num_threads = 4
+        events_per_thread = 200
+        total_events = num_threads * events_per_thread
+        barrier = threading.Barrier(num_threads)
+        errors = []
+
+        def sender():
+            try:
+                barrier.wait(timeout=5)
+                for _ in range(events_per_thread):
+                    sm.send("go")
+            except Exception as e:
+                errors.append(e)
+
+        threads = [threading.Thread(target=sender) for _ in range(num_threads)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join(timeout=30)
+
+        assert not errors, f"Thread errors: {errors}"
+
+        # Each transition fires exactly one on_enter callback.
+        # +1 because initial activation also fires on_enter_s1.
+        counts = Counter(call_log)
+        total_callbacks = counts["enter_s1"] + counts["enter_s2"]
+        assert total_callbacks == total_events + 1
+
+    def test_concurrent_send_and_read_configuration(self, cycling_machine):
+        """Reading configuration while events are being processed must not raise."""
+        num_senders = 4
+        events_per_sender = 300
+        barrier = threading.Barrier(num_senders + 1)
+        stop_event = threading.Event()
+        errors = []
+
+        def sender():
+            try:
+                barrier.wait(timeout=5)
+                for _ in range(events_per_sender):
+                    cycling_machine.send("cycle")
+            except Exception as e:
+                errors.append(e)
+
+        def reader():
+            barrier.wait(timeout=5)
+            while not stop_event.is_set():
+                try:
+                    _ = cycling_machine.configuration
+                    _ = cycling_machine.current_state_value
+                    _ = list(cycling_machine.configuration)
+                except Exception as e:
+                    errors.append(e)
+
+        threads = [threading.Thread(target=sender) for _ in range(num_senders)]
+        reader_thread = threading.Thread(target=reader)
+
+        for t in threads:
+            t.start()
+        reader_thread.start()
+
+        for t in threads:
+            t.join(timeout=30)
+        stop_event.set()
+        reader_thread.join(timeout=5)
+
+        assert not errors, f"Thread errors: {errors}"
+
+
 async def test_regression_443_with_modifications_for_async_engine():
     """
     Test for https://github.com/fgmacedo/python-statemachine/issues/443