fix: answer callers of events dropped when the machine terminates

[fgmacedo]

Fixes #678

Why

On the async engine, 3 of 3 coroutines awaiting send() of an event queued behind one that reaches a top-level final state hang forever on main (repro below). v3.2.1 answered them with None (StateChart) or TransitionNotAllowed (StateMachine). The regression came from #668, which stops the loop at termination and leaves the dropped triggers' futures pending.

What changes

• When the machine stops, the external events still queued, delayed or not, are dropped and each awaiting caller gets what a terminated machine answers to a new event: None, or TransitionNotAllowed for its own event when allow_event_without_transition is False.
• A delayed event pending at termination is answered too. On v3.2.1 its caller hung.
• A machine stopped mid-macrostep (eventless or internal transition to a top-level final state, or a top-level parallel state completing) no longer takes one more queued external event: on main a sync StateMachine raised TransitionNotAllowed for it, and a completed root parallel state ran it and re-entered the parallel.
• Dropped triggers no longer stay in the external queue, on both engines, including events sent to a terminated StateChart.
• The 3.2.2 release note entry for fix: honor Event(internal=True) and give internal events from outside a step the external contract #668 states what awaiting callers receive.

Evidence

New tests in tests/test_async_futures.py::TestEventsDroppedAtFinalState (14, with pytest.mark.timeout(5), sync and async via sm_runner). Without the engine changes:

$ timeout 120 uv run pytest tests/test_async_futures.py -k Dropped --no-cov -q
E           Failed: Timeout (>5.0s) from pytest-timeout.   (x4)
10 failed, 11 deselected        # before the root-parallel / eventless-final tests were added

With them:

$ timeout 180 uv run pytest -n 4
TOTAL    6216      0   1956      0   100%
========== 2649 passed, 25 skipped, 44 xfailed, 9 warnings in 25.06s ===========

Repro (a finish holding the lock, two go and one go with delay=1000 queued behind it), per version:

             v3.2.1                              main                        this branch
StateChart   ['finished', None, None, HANG]      ['finished', HANG, HANG, HANG]   ['finished', None, None, None]
StateMachine ['finished', TNA, TNA, TNA]         ['finished', HANG, HANG, HANG]   ['finished', TNA, TNA, TNA]

repro script

import asyncio
from statemachine import State, StateChart, StateMachine

def build(base):
    class SM(base):
        a = State(initial=True)
        b = State()
        done = State(final=True)
        finish = a.to(done, on="slow") | b.to(done, on="slow")
        go = a.to(b) | b.to(b)
        async def slow(self):
            await asyncio.sleep(0.05)
            return "finished"
    return SM

async def concurrent(base):
    sm = build(base)()
    await sm.activate_initial_state()
    t1 = asyncio.ensure_future(sm.send("finish"))
    await asyncio.sleep(0.01)
    rest = [asyncio.ensure_future(sm.send("go")), asyncio.ensure_future(sm.send("go")),
            asyncio.ensure_future(sm.send("go", delay=1000))]
    await asyncio.wait([t1, *rest], timeout=2)
    out = []
    for t in (t1, *rest):
        if not t.done():
            out.append("HANG"); t.cancel()
        elif t.exception():
            out.append(repr(t.exception()))
        else:
            out.append(repr(t.result()))
    print(base.__name__, out)

async def main():
    for base in (StateChart, StateMachine):
        await concurrent(base)

asyncio.run(main())

Not verified

• An exception raised in the lock holder's macrostep (e.g. an eventless transition's action with catch_errors_as_events=False) still leaves other coroutines' futures pending on both main and this branch; v3.2.1 answered them with None. Out of scope here, to be tracked separately.
• Sync engine with threads: a thread that sends between the drain and the lock release leaves its trigger queued until the next send (no future, so no hang).
• A fire-and-forget async put() on a StateMachine whose event is dropped now logs "Future exception was never retrieved", as an event rejected while processed already does.
• Termination through engine.stop() (invoke cancellation) takes the same drain path but has no dedicated test.

[sonarqubecloud]

Quality Gate passed

Issues
4 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 100.00%. Comparing base (73bf57c) to head (d0be70b).

Additional details and impacted files

@@            Coverage Diff            @@
##              main      #685   +/-   ##
=========================================
  Coverage   100.00%   100.00%           
=========================================
  Files           53        53           
  Lines         5787      5808   +21     
  Branches       918       925    +7     
=========================================
+ Hits          5787      5808   +21     

Flag	Coverage Δ
unittests	100.00% <100.00%> (ø)

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.