feat: propagate constructor kwargs to initial state callbacks

docs/invoke.md

@@ -302,7 +302,37 @@ True
 
 ```
 
-For initial states (entered automatically, not via an event), `kwargs` is empty.
+For initial states, any extra keyword arguments passed to the `StateChart` constructor
+are forwarded as event data. This makes self-contained machines that start processing
+immediately especially useful:
+
+```py
+>>> config_file = Path(tempfile.mktemp(suffix=".json"))
+>>> _ = config_file.write_text('{"theme": "dark"}')
+
+>>> class AppLoader(StateChart):
+...     loading = State(initial=True)
+...     ready = State(final=True)
+...     done_invoke_loading = loading.to(ready)
+...
+...     def on_invoke_loading(self, config_path=None, **kwargs):
+...         """config_path comes from the constructor: AppLoader(config_path=...)."""
+...         return json.loads(Path(config_path).read_text())
+...
+...     def on_enter_ready(self, data=None, **kwargs):
+...         self.config = data
+
+>>> sm = AppLoader(config_path=str(config_file))
+>>> time.sleep(0.2)
+
+>>> "ready" in sm.configuration_values
+True
+>>> sm.config
+{'theme': 'dark'}
+
+>>> config_file.unlink()
+
+```
 
 ## Error handling
 

docs/releases/3.0.0.md

@@ -65,6 +65,24 @@ and wait for all results:
 
 ```
 
+Constructor keyword arguments are forwarded to initial state callbacks, so self-contained
+machines can receive context at creation time:
+
+```py
+>>> class Greeter(StateChart):
+...     idle = State(initial=True)
+...     done = State(final=True)
+...     idle.to(done)
+...
+...     def on_enter_idle(self, name=None, **kwargs):
+...         self.greeting = f"Hello, {name}!"
+
+>>> sm = Greeter(name="Alice")
+>>> sm.greeting
+'Hello, Alice!'
+
+```
+
 See {ref}`invoke` for full documentation.
 
 ### Compound states

statemachine/engines/async_.py

@@ -309,12 +309,15 @@ async def _run_microstep(self, enabled_transitions, trigger_data):  # pragma: no
         except Exception as e:
             self._handle_error(e, trigger_data)
 
-    async def activate_initial_state(self):
+    async def activate_initial_state(self, **kwargs):
         """Activate the initial state.
 
         In async code, the user must call this method explicitly (or it will be lazily
         activated on the first event). There's no built-in way to call async code from
         ``StateMachine.__init__``.
+
+        Any ``**kwargs`` are forwarded to initial state entry callbacks via dependency
+        injection, just like event kwargs on ``send()``.
         """
         return await self.processing_loop()
 

statemachine/engines/base.py

@@ -181,11 +181,11 @@ def _send_error_execution(self, error: Exception, trigger_data: TriggerData):
             return
         self.sm.send(_ERROR_EXECUTION, error=error, internal=True)
 
-    def start(self):
+    def start(self, **kwargs):
         if self.sm.current_state_value is not None:
             return
 
-        BoundEvent("__initial__", _sm=self.sm).put()
+        BoundEvent("__initial__", _sm=self.sm).put(**kwargs)
 
     def _initial_transitions(self, trigger_data):
         empty_state = State()

statemachine/engines/sync.py

@@ -31,13 +31,13 @@ def _run_microstep(self, enabled_transitions, trigger_data):
         except Exception as e:  # pragma: no cover
             self._handle_error(e, trigger_data)
 
-    def start(self):
+    def start(self, **kwargs):
         if self.sm.current_state_value is not None:
             return
 
-        self.activate_initial_state()
+        self.activate_initial_state(**kwargs)
 
-    def activate_initial_state(self):
+    def activate_initial_state(self, **kwargs):
         """
         Activate the initial state.
 
@@ -48,7 +48,9 @@ def activate_initial_state(self):
         may depend on async code from the StateMachine.__init__ method.
         """
         if self.sm.current_state_value is None:
-            trigger_data = BoundEvent("__initial__", _sm=self.sm).build_trigger(machine=self.sm)
+            trigger_data = BoundEvent("__initial__", _sm=self.sm).build_trigger(
+                machine=self.sm, **kwargs
+            )
             transitions = self._initial_transitions(trigger_data)
             self._processing.acquire(blocking=False)
             try:

statemachine/statemachine.py

@@ -165,7 +165,7 @@ def __init__(
         # for async code, the user should manually call `await sm.activate_initial_state()`
         # after state machine creation.
         self._engine = self._get_engine()
-        self._engine.start()
+        self._engine.start(**kwargs)
 
     def _get_engine(self):
         if self._callbacks.has_async_callbacks:

tests/test_statemachine.py

@@ -729,3 +729,81 @@ class SM(StateChart):
             warnings.simplefilter("ignore", DeprecationWarning)
             with pytest.raises(exceptions.InvalidStateValue):
                 _ = sm.current_state
+
+
+class TestInitKwargsPropagation:
+    """Constructor kwargs are forwarded to initial state entry callbacks."""
+
+    async def test_kwargs_available_in_on_enter_initial(self, sm_runner):
+        class SM(StateChart):
+            idle = State(initial=True)
+            done = State(final=True)
+            go = idle.to(done)
+
+            def on_enter_idle(self, greeting=None, **kwargs):
+                self.greeting = greeting
+
+        sm = await sm_runner.start(SM, greeting="hello")
+        assert sm.greeting == "hello"
+
+    async def test_kwargs_flow_through_eventless_transitions(self, sm_runner):
+        class Pipeline(StateChart):
+            start = State(initial=True)
+            processing = State()
+            done = State(final=True)
+
+            start.to(processing)
+            processing.to(done)
+
+            def on_enter_start(self, task_id=None, **kwargs):
+                self.task_id = task_id
+
+        sm = await sm_runner.start(Pipeline, task_id="abc-123")
+        assert sm.task_id == "abc-123"
+        assert "done" in sm.configuration_values
+
+    async def test_no_kwargs_still_works(self, sm_runner):
+        class SM(StateChart):
+            idle = State(initial=True)
+            done = State(final=True)
+            go = idle.to(done)
+
+            def on_enter_idle(self, **kwargs):
+                self.entered = True
+
+        sm = await sm_runner.start(SM)
+        assert sm.entered is True
+
+    async def test_multiple_kwargs(self, sm_runner):
+        class SM(StateChart):
+            idle = State(initial=True)
+            done = State(final=True)
+            go = idle.to(done)
+
+            def on_enter_idle(self, host=None, port=None, **kwargs):
+                self.host = host
+                self.port = port
+
+        sm = await sm_runner.start(SM, host="localhost", port=5432)
+        assert sm.host == "localhost"
+        assert sm.port == 5432
+
+    async def test_kwargs_in_invoke_handler(self, sm_runner):
+        """Init kwargs flow to invoke handlers via dependency injection."""
+
+        class SM(StateChart):
+            loading = State(initial=True)
+            ready = State(final=True)
+            done_invoke_loading = loading.to(ready)
+
+            def on_invoke_loading(self, url=None, **kwargs):
+                return f"fetched:{url}"
+
+            def on_enter_ready(self, data=None, **kwargs):
+                self.result = data
+
+        sm = await sm_runner.start(SM, url="https://example.com")
+        await sm_runner.sleep(0.2)
+        await sm_runner.processing_loop(sm)
+        assert "ready" in sm.configuration_values
+        assert sm.result == "fetched:https://example.com"