homie: inbound /set async dispatch, thread-safe and promoted to the root (#15) (#23)

- Dispatch: schedule an async /set callback onto the consumer's loop with
  asyncio.run_coroutine_threadsafe (ensure_future is not safe from paho's
  network thread, where /set arrives). Branch on the callback's actual return
  (iscoroutine), not async_loop's presence, so a SYNC callback runs inline even
  when a device-level loop is set. Surface exceptions the discarded scheduling
  Future would otherwise swallow, matching the sync path's logging.
- Promotion: async_loop is now Device(async_loop=), propagated to every property
  via add_node()/Node.add_property() (mirroring _qos), children inheriting the
  root's loop. The per-Property async_loop still works and is kept when the
  device sets none.
- Fix the Optional[...] = False default to = None (bool vs loop annotation);
  widen to AbstractEventLoop.
- Fix Node's mutable default properties: dict = {} (shared-dict footgun the new
  propagation reads/writes through).

docs: CHANGELOG [Unreleased] (Added Device(async_loop=), Fixed dispatch/defaults);
README BYO note on async /set. Reviewed adversarially; 4 findings fixed. +10 tests.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: dcj

CHANGELOG.md

@@ -4,6 +4,14 @@ All notable changes to `ebus-sdk` are recorded here. Format follows [Keep a Chan
 
 ## [Unreleased]
 
+### Added
+
+- `Device(async_loop=...)`: set the consumer's asyncio event loop once on the root instead of on every settable `Property`. It propagates to every property in the tree via `add_node()` / `Node.add_property()` (like QoS) and is inherited by child devices, so inbound `/set` callbacks for a tree with N settable properties no longer configure the loop N times. The per-`Property` `async_loop` argument still works and is preserved when no device-level loop is set. ([#15](https://github.com/electrification-bus/python-sdk/issues/15))
+
+### Fixed
+
+- Inbound `/set` on a settable property with an async (coroutine) callback is now dispatched to the consumer's event loop with `asyncio.run_coroutine_threadsafe` instead of `asyncio.ensure_future`. `/set` arrives on the transport's network-loop thread, and `ensure_future` is not safe to call from a thread other than the loop's own, so the previous path could misbehave on a real async host. The dispatch now branches on whether the callback returns a coroutine (not merely on a loop being configured), so a synchronous callback keeps running inline even when a tree-wide loop is set, and an exception raised inside an async callback is logged rather than silently swallowed by the discarded scheduling future. The `async_loop` default is corrected from `False` to `None` (a bool against the event-loop annotation, which `py.typed` surfaces), and `Node`'s mutable default `properties={}` is replaced with `None` (a shared-dict footgun the new propagation reads and writes through). ([#15](https://github.com/electrification-bus/python-sdk/issues/15))
+
 ## [0.16.0] — 2026-08-02
 
 ### Added

README.md

@@ -74,6 +74,8 @@ client.connect()                               # host connects on its own loop
 
 `device.will()` returns the tree's Last Will descriptor and `device.refresh_tree()` republishes the whole tree; the `set_will` / `on_connect` / `connect` calls above are illustrative of your host's own MQTT API. Property values publish once the client is connected (the SDK gates on `is_connected()`, not on its own `start()`, which a caller-driven client never calls). `device.stop()` publishes a final retained `$state=disconnected` through the client and returns immediately, without flushing or closing it. `on_disconnect=` is inert for an injected client; register disconnect handling on your own client.
 
+For the inbound direction, if the tree has settable properties whose callbacks are async coroutines, pass `Device(async_loop=<your event loop>)`: inbound `/set` arrives on the transport's network thread, and this schedules the callback onto your loop (set once for the whole tree, not per property). A synchronous callback runs inline and needs no loop.
+
 **Two host shapes, and the will's limit.** The example above is a caller that owns a *dedicated* client and connects it itself, so it can set the will before connect. A host with a *shared* connection it does not own (Home Assistant is the archetype: one connection, up before your code loads, its will already set from the host's own config with no hook to change it) cannot set an eBus will, because MQTT allows one will per connection. There, wire `refresh_tree()` to the host's reconnect callback and let the SDK gate on `is_connected()`: the tree stays correct across every reconnect the process survives. (A consequence to be aware of: on such a shared connection the SDK cannot set the will's `$state=lost`, since the will is a CONNECT-time property the host owns.)
 
 #### Clearing a value vs. an empty-string value

src/ebus_sdk/homie.py

@@ -478,7 +478,7 @@ def __init__(
         supports_target: Optional[bool] = False,
         node: Optional[Node] = None,
         device: Optional[Device] = None,
-        async_loop: Optional[asyncio.SelectorEventLoop] = False,
+        async_loop: Optional[asyncio.AbstractEventLoop] = None,
         from_dict: Optional[dict] = None,
     ):
         if from_dict:
@@ -973,13 +973,35 @@ def _settable_callback(self, topic: str, payload: Union[bytes, bytearray]) -> No
                 # Property supports_target, publish that!
                 self.publish_target_value(payload)
             # Call the property's set_callback function
-            if self.async_loop:
-                asyncio.ensure_future(set_callback(payload), loop=self.async_loop)
-            else:
-                set_callback(payload)
+            # Run the callback: a sync callback runs inline here (on the transport's
+            # network thread, as before); an async (coroutine) callback is scheduled onto
+            # the consumer's event loop thread-safely via run_coroutine_threadsafe
+            # (ensure_future is NOT safe to call from a thread other than the loop's own).
+            # Decide on the callback's actual return, not just async_loop's presence, so a
+            # sync callback stays inline even when a device-level loop is set for the tree.
+            result = set_callback(payload)
+            if self.async_loop is not None and asyncio.iscoroutine(result):
+                future = asyncio.run_coroutine_threadsafe(result, self.async_loop)
+                # The Future is otherwise discarded, so an exception raised inside the
+                # coroutine would vanish silently (the inline path is caught below by the
+                # surrounding try/except). Surface it, matching the sync path's logging.
+                future.add_done_callback(partial(self._log_async_set_result, property_id=property_id))
         except Exception as e:
             logger.exception(f"reason=propertySetCallbackException,e={e}")
 
+    def _log_async_set_result(self, future, property_id) -> None:
+        """Done-callback for an async /set handler scheduled via run_coroutine_threadsafe.
+
+        The scheduling Future is otherwise discarded, and a discarded concurrent.futures
+        Future swallows a stored exception silently (unlike an asyncio.Task). Surface it,
+        matching the synchronous path's ``propertySetCallbackException`` logging.
+        """
+        if future.cancelled():
+            return
+        exc = future.exception()
+        if exc is not None:
+            logger.error(f"reason=propertySetAsyncCallbackException,propertyID={property_id},exc={exc!r}")
+
     def set_subscribe(self) -> None:
         """
         Subscribe to property/set topic on Homie broker
@@ -1022,7 +1044,7 @@ def __init__(
         id: Optional[str] = None,
         name: Optional[str] = None,
         type: Optional[str] = None,
-        properties: dict = {},
+        properties: Optional[dict] = None,
         device: Optional[Device] = None,
         # mqttc: Optiona[MqttClient] = None, # DCJ pretty sure we can remove this
         from_dict: Optional[dict] = None,
@@ -1047,7 +1069,7 @@ def __init__(
             else:
                 self._name = id
             self._type = type
-            self._properties = properties
+            self._properties = properties if properties is not None else {}
             self._device = device
 
     def as_dict(self) -> dict:
@@ -1112,9 +1134,11 @@ def add_property(self, property: Property) -> Property:
         """
         if not property.node():
             property.set_node(self)
-        # Propagate QoS from device if available
+        # Propagate QoS (and the async /set dispatch loop, if set) from the device.
         if self._device and hasattr(self._device, "_qos"):
             property._qos = self._device._qos
+        if self._device and getattr(self._device, "_async_loop", None) is not None:
+            property.async_loop = self._device._async_loop
         # Note set_subscribe() checks if property is settable...
         property.set_subscribe()
         # Add property to dictionary BEFORE publishing description
@@ -1331,6 +1355,7 @@ def __init__(
         mqtt_cfg: Optional[dict] = None,
         mqttc: Optional[MqttDeviceTransport] = None,
         qos: int = EBUS_HOMIE_MQTT_QOS,
+        async_loop: Optional[asyncio.AbstractEventLoop] = None,
         on_disconnect: Optional[Callable[[bool], None]] = None,
     ):
         # Root vs. child invariants — mutually exclusive. Test presence by identity
@@ -1373,6 +1398,14 @@ def __init__(
         self._owned_client: Optional[MqttClient] = None
         self._state = None
         self._qos = qos
+        # The consumer's asyncio event loop for dispatching inbound /set callbacks on
+        # settable properties. Set once per tree on the root and propagated to every
+        # property via add_node() / Node.add_property() (like _qos), rather than
+        # per-property. Children inherit the root's loop unless given their own. None
+        # means /set callbacks run synchronously on the transport's network thread.
+        self._async_loop = (
+            async_loop if async_loop is not None else (parent._async_loop if parent is not None else None)
+        )
         # Optional consumer hook: called on the ROOT's MQTT (dis)connect. Only a
         # root owns a client (children share it), so it fires on the root only.
         # Contract is transport-neutral: on_disconnect(clean: bool), never a paho
@@ -1702,9 +1735,12 @@ def add_node(self, node: Node) -> Node:
         """
         if not node.device():
             node.set_device(self)
-        # Propagate device QoS to all properties in this node
+        # Propagate device QoS (and the async /set dispatch loop, if set) to every
+        # property in this node.
         for prop in node.properties().values():
             prop._qos = self._qos
+            if self._async_loop is not None:
+                prop.async_loop = self._async_loop
         node_id = node.id()
         self._nodes.update({node_id: node})
         node.publish()

tests/test_homie_device.py

@@ -2029,6 +2029,105 @@ def test_owned_client_handle_is_set_and_cleared_for_an_sdk_built_client(self):
             assert device._owned_client is None
 
 
+class TestInboundAsyncLoop:
+    """Inbound /set async dispatch: async_loop promoted to the root, thread-safe (#15)."""
+
+    def test_async_loop_default_is_none(self):
+        # Was `Optional[...] = False` (a bool against the loop annotation); now None.
+        assert Property(id="mode").async_loop is None
+
+    def test_device_async_loop_propagates_via_add_node(self):
+        loop = MagicMock()
+        prop = Property(id="setpoint", datatype=PropertyDatatype.FLOAT)
+        node = Node(id="core", name="Core", type="sensor", properties={"setpoint": prop})
+        Device(id="dev", async_loop=loop, nodes=[node])  # add_node propagates to the node's props
+        assert prop.async_loop is loop
+
+    def test_device_async_loop_propagates_via_node_add_property(self):
+        loop = MagicMock()
+        device = Device(id="dev", async_loop=loop)  # transport-free root
+        node = Node(id="core", name="Core", type="sensor")
+        device.add_node(node)
+        prop = Property(id="setpoint", value=1.0, datatype=PropertyDatatype.FLOAT)
+        node.add_property(prop)  # node is attached -> propagates the device's loop
+        assert prop.async_loop is loop
+
+    def test_child_device_inherits_root_async_loop(self):
+        loop = MagicMock()
+        root = Device(id="root", async_loop=loop)  # transport-free root
+        child = Device(id="child", parent=root)
+        assert child._async_loop is loop  # one loop per tree, inherited by children
+        node = Node(id="core", name="Core", type="sensor")
+        child.add_node(node)
+        prop = Property(id="setpoint", value=1.0, datatype=PropertyDatatype.FLOAT)
+        node.add_property(prop)
+        assert prop.async_loop is loop
+
+    def test_async_dispatch_uses_run_coroutine_threadsafe(self):
+        """An async /set callback is scheduled onto the consumer's loop thread-safely;
+        /set arrives on the transport's network thread, so ensure_future is unsafe."""
+
+        async def cb(payload):
+            return None
+
+        loop = MagicMock()
+        prop = Property(id="mode", settable=True, set_callback=cb)
+        prop.async_loop = loop
+        with patch("asyncio.run_coroutine_threadsafe") as rct, patch("asyncio.ensure_future") as ef:
+            prop._settable_callback("ebus/5/dev/node/mode/set", b"LOAD_UP")
+        rct.assert_called_once()
+        assert rct.call_args.args[1] is loop  # scheduled onto the given loop
+        ef.assert_not_called()  # never the thread-unsafe ensure_future path
+        rct.call_args.args[0].close()  # close the un-awaited coroutine (rct is mocked)
+
+    def test_sync_dispatch_when_no_loop(self):
+        seen = []
+        prop = Property(id="mode", settable=True, set_callback=lambda v: seen.append(v))
+        assert prop.async_loop is None
+        with patch("asyncio.run_coroutine_threadsafe") as rct:
+            prop._settable_callback("ebus/5/dev/node/mode/set", b"LOAD_UP")
+        assert seen == ["LOAD_UP"]  # invoked synchronously
+        rct.assert_not_called()
+
+    def test_sync_callback_stays_inline_under_device_loop(self):
+        """A device-level loop must not force a sync callback onto the async path: the
+        dispatch branches on the callback's return, not on the loop's presence (#15)."""
+        loop = MagicMock()
+        seen = []
+        prop = Property(id="mode", settable=True, set_callback=lambda v: seen.append(v))
+        prop.async_loop = loop  # a device-level loop is propagated to every property
+        with patch("asyncio.run_coroutine_threadsafe") as rct:
+            prop._settable_callback("ebus/5/dev/node/mode/set", b"ON")
+        assert seen == ["ON"]  # ran inline, once
+        rct.assert_not_called()  # not scheduled: a sync callback returns no coroutine
+
+    def test_per_property_loop_survives_when_device_has_no_loop(self):
+        """Backward compat: a per-Property async_loop is kept when the device sets none."""
+        loop_a = MagicMock()
+        prop = Property(id="setpoint", datatype=PropertyDatatype.FLOAT, async_loop=loop_a)
+        node = Node(id="core", name="Core", type="sensor", properties={"setpoint": prop})
+        Device(id="dev", nodes=[node])  # no async_loop -> propagation is skipped
+        assert prop.async_loop is loop_a
+
+    def test_device_loop_overrides_per_property_loop(self):
+        """One loop per tree: a device-level loop wins over a pre-set per-Property loop."""
+        loop_a, loop_b = MagicMock(), MagicMock()
+        prop = Property(id="setpoint", datatype=PropertyDatatype.FLOAT, async_loop=loop_a)
+        node = Node(id="core", name="Core", type="sensor", properties={"setpoint": prop})
+        Device(id="dev", async_loop=loop_b, nodes=[node])
+        assert prop.async_loop is loop_b
+
+    def test_async_set_callback_exception_is_surfaced(self):
+        """A raising async /set handler is logged, not swallowed by the discarded Future."""
+        prop = Property(id="mode")
+        future = MagicMock()
+        future.cancelled.return_value = False
+        future.exception.return_value = ValueError("bad setpoint")
+        with patch("ebus_sdk.homie.logger") as mock_logger:
+            prop._log_async_set_result(future, "mode")
+        assert any("propertySetAsyncCallbackException" in str(c) for c in mock_logger.error.call_args_list)
+
+
 class TestDeviceWithoutTransport:
     """`mqtt_cfg=None` — the declared default — builds a device tree with no transport."""