fix: report errors raised while an invocation processes its done.invoke (#698)

Author: fgmacedo

docs/invoke.md

@@ -425,8 +425,8 @@ True
 
 ## Error handling
 
-If an invoke handler raises an exception, `error.execution` is sent to the machine's
-internal queue (when `catch_errors_as_events=True`, the default for `StateChart`). You can
+If an invoke handler raises an exception while its state is active, `error.execution` is
+sent to the machine's external queue, whatever `catch_errors_as_events` is set to. You can
 handle it with a transition for `error.execution`:
 
 ```py
@@ -451,6 +451,13 @@ True
 
 ```
 
+An error raised while the machine processes `done.invoke`, by a callback or by the
+handler's `on_finalize`, is handled like an error in any other step: it becomes `error.execution` when `catch_errors_as_events=True`,
+and propagates otherwise. When the machine is idle as the handler finishes, the handler's
+own thread or task processes `done.invoke`, and nobody awaits it: the sync engine lets the
+error propagate out of the thread, to `threading.excepthook`, and the async engine passes
+it to the event loop's exception handler.
+
 ## Multiple invokes
 
 ### Independent invokes (one event each)

statemachine/invoke.py

@@ -401,11 +401,6 @@ def _run_sync_handler(
                 result = handler.run(ctx)
             else:
                 result = callback.call(ctx=ctx, machine=ctx.machine, **ctx.kwargs)
-            if not ctx.cancelled.is_set():
-                self.sm.send(
-                    f"done.invoke.{ctx.invokeid}",
-                    data=result,
-                )
         except Exception as e:
             if not ctx.cancelled.is_set():
                 # Intentionally using the external queue (no internal=True):
@@ -415,6 +410,9 @@ def _run_sync_handler(
                 # active (the internal queue is only drained within a macrostep).
                 # This matches done.invoke, which also uses the external queue.
                 self.sm.send("error.execution", error=e)
+        else:
+            if not ctx.cancelled.is_set():
+                self.sm.send(f"done.invoke.{ctx.invokeid}", data=result)
         finally:
             invocation.terminated = True
             self._debug(
@@ -484,11 +482,6 @@ async def _run_async_handler(
                 result = await loop.run_in_executor(
                     None, lambda: callback.call(ctx=ctx, machine=ctx.machine, **ctx.kwargs)
                 )
-            if not ctx.cancelled.is_set():
-                await self.sm.send(
-                    f"done.invoke.{ctx.invokeid}",
-                    data=result,
-                )
         except asyncio.CancelledError:
             # Intentionally swallowed: the owning state was exited, so this
             # invocation was cancelled — there is nothing to propagate.
@@ -497,6 +490,9 @@ async def _run_async_handler(
             if not ctx.cancelled.is_set():
                 # External queue — see comment in _run_sync_handler.
                 await self.sm.send("error.execution", error=e)
+        else:
+            if not ctx.cancelled.is_set():
+                await self._send_done_async(ctx.invokeid, result)
         finally:
             invocation.terminated = True
             self._debug(
@@ -506,6 +502,19 @@ async def _run_async_handler(
                 ctx.cancelled.is_set(),
             )
 
+    async def _send_done_async(self, invokeid: str, result: Any):
+        """Send ``done.invoke`` and report an error raised while processing it.
+
+        The invocation's task drives the macrostep when the machine is idle, and no caller
+        awaits that task, so the error goes to the event loop's exception handler.
+        """
+        try:
+            await self.sm.send(f"done.invoke.{invokeid}", data=result)
+        except Exception as e:
+            asyncio.get_running_loop().call_exception_handler(
+                {"message": f"Error while processing done.invoke.{invokeid}", "exception": e}
+            )
+
     # --- Cancel ---
 
     def _cancel(self, invokeid: str):
@@ -589,7 +598,10 @@ def handle_external_event(self, trigger_data) -> None:
             # Note: finalize must run even after the invocation terminates, because
             # child events may still be queued when the handler thread completes.
             if is_from_child and hasattr(handler, "on_finalize"):
-                handler.on_finalize(trigger_data)
+                try:
+                    handler.on_finalize(trigger_data)
+                except Exception as e:
+                    self._engine._handle_error(e, trigger_data)
 
             # Autoforward: forward parent events to child (not events from child itself).
             # Only forward if the invocation is still running.

tests/test_invoke.py

@@ -1,15 +1,18 @@
 """Tests for the invoke callback group."""
 
+import asyncio
 import threading
 import time
 
+import pytest
 from statemachine.invoke import IInvoke
 from statemachine.invoke import InvokeContext
 from statemachine.invoke import invoke_group
 
 from statemachine import Event
 from statemachine import State
 from statemachine import StateChart
+from statemachine import StateMachine
 
 
 class TestInvokeSimpleCallable:
@@ -295,6 +298,152 @@ def on_enter_error_state(self, **kwargs):
         assert "error_state" in sm.configuration_values
 
 
+@pytest.fixture()
+def thread_errors(monkeypatch) -> "list[BaseException]":
+    errors: "list[BaseException]" = []
+    monkeypatch.setattr(threading, "excepthook", lambda args: errors.append(args.exc_value))
+    return errors
+
+
+async def finish_invocation(
+    sm, release: threading.Event, thread_errors: "list[BaseException]"
+) -> "list[BaseException]":
+    """Let the only invocation finish and return the errors its thread or task reported."""
+    (invocation,) = sm._engine._invoke_manager._active.values()
+    if invocation.task is None:
+        release.set()
+        invocation.thread.join(timeout=5)
+        return thread_errors
+    loop_errors: "list[BaseException]" = []
+    loop = asyncio.get_running_loop()
+    previous_handler = loop.get_exception_handler()
+    loop.set_exception_handler(lambda _loop, context: loop_errors.append(context["exception"]))
+    try:
+        release.set()
+        await asyncio.wait([invocation.task], timeout=5)
+    finally:
+        loop.set_exception_handler(previous_handler)
+    assert invocation.task.exception() is None
+    return loop_errors
+
+
+class TestInvokeCompletionProcessingErrors:
+    """An error raised while the invocation itself processes its ``done.invoke`` is not lost.
+
+    The invocation finishes after the machine settled, so its own thread or task processes
+    ``done.invoke`` and the error has no other caller to reach.
+    """
+
+    async def test_error_leaving_the_state_propagates(self, sm_runner, thread_errors):
+        release = threading.Event()
+
+        class SM(StateMachine):
+            s1 = State(initial=True)
+            s2 = State(final=True)
+            done_invoke_s1 = s1.to(s2, on="boom")
+
+            def on_invoke_s1(self):
+                release.wait(timeout=5)
+
+            def boom(self):
+                raise ValueError("boom")
+
+        sm = await sm_runner.start(SM)
+        errors = await finish_invocation(sm, release, thread_errors)
+
+        assert [repr(e) for e in errors] == [repr(ValueError("boom"))]
+        assert sm.configuration_values == {"s1"}
+
+    async def test_error_staying_in_the_state_propagates(self, sm_runner, thread_errors):
+        release = threading.Event()
+
+        class SM(StateMachine):
+            s1 = State(initial=True)
+            s2 = State(final=True)
+            done_invoke_s1 = s1.to.itself(internal=True, on="boom")
+            error_execution = s1.to(s2)
+
+            def on_invoke_s1(self):
+                release.wait(timeout=5)
+
+            def boom(self):
+                raise ValueError("boom")
+
+        sm = await sm_runner.start(SM)
+        errors = await finish_invocation(sm, release, thread_errors)
+
+        assert [repr(e) for e in errors] == [repr(ValueError("boom"))]
+        assert sm.configuration_values == {"s1"}
+
+    async def test_error_leaving_the_state_becomes_error_execution(self, sm_runner, thread_errors):
+        release = threading.Event()
+
+        class SM(StateChart):
+            s1 = State(initial=True)
+            s2 = State()
+            handled = State(final=True)
+            done_invoke_s1 = s1.to(s2, on="boom")
+            error_execution = s2.to(handled)
+
+            def on_invoke_s1(self):
+                release.wait(timeout=5)
+
+            def boom(self):
+                raise ValueError("boom")
+
+        sm = await sm_runner.start(SM)
+        errors = await finish_invocation(sm, release, thread_errors)
+
+        assert errors == []
+        assert sm.configuration_values == {"handled"}
+
+
+class FailingFinalize:
+    def __init__(self, release: threading.Event):
+        self.release = release
+
+    def run(self, ctx):
+        self.release.wait(timeout=5)
+
+    def on_finalize(self, trigger_data):
+        raise ValueError("finalize")
+
+
+class TestInvokeFinalizeErrors:
+    """An error raised by ``on_finalize`` is handled like an error in any other step."""
+
+    async def test_finalize_error_becomes_error_execution(self, sm_runner, thread_errors):
+        release = threading.Event()
+
+        class SM(StateChart):
+            s1 = State(initial=True, invoke=FailingFinalize(release))
+            s2 = State()
+            handled = State(final=True)
+            done_invoke_s1 = s1.to(s2)
+            error_execution = s2.to(handled)
+
+        sm = await sm_runner.start(SM)
+        errors = await finish_invocation(sm, release, thread_errors)
+
+        assert errors == []
+        assert sm.configuration_values == {"handled"}
+
+    async def test_finalize_error_propagates(self, sm_runner, thread_errors):
+        release = threading.Event()
+
+        class SM(StateMachine):
+            s1 = State(initial=True, invoke=FailingFinalize(release))
+            s2 = State(final=True)
+            done_invoke_s1 = s1.to(s2)
+            error_execution = s1.to(s2)
+
+        sm = await sm_runner.start(SM)
+        errors = await finish_invocation(sm, release, thread_errors)
+
+        assert [repr(e) for e in errors] == [repr(ValueError("finalize"))]
+        assert sm.configuration_values == {"s1"}
+
+
 class TestInvokeMultiple:
     """Multiple invokes per state — all run concurrently."""