Finishing the transport seam would open the SDK to hosts that own their MQTT connection

[cayossarian]

Today the SDK can be embedded only in an application willing to let it own a paho client and a
network thread. That's a fine default, but it's currently the only option, and it rules out hosts
that would otherwise be natural consumers:

• Any asyncio application. Device.stop() blocks up to ~3s on the caller's thread
  (publish_and_flush at :1449, then mqttc.stop(timeout=2.0) at :1455). On an event loop
  that's a stalled loop.
• Home Assistant specifically. HA detects and logs both blocking calls and slow callbacks; a
  component that stalls the loop for seconds isn't shippable there. Its MQTT integration is also
  single_config_entry, so a second SDK-owned connection isn't available even in principle — the
  connection has to come from the host.
• Test harnesses. With a transport seam, a consumer can drive the SDK from a fake client and test
  eBus behaviour with no broker and no paho mocking at all.

None of that needs a new concurrency model. It needs the seam you've already started to be finishable
from the outside. The cleanliness is a side effect; the point is who can use it.

You already have two halves of this

Seam	State
Inbound dispatch — Property.async_loop (homie.py:464, :914)	Exists. /set callbacks go via asyncio.ensure_future(..., loop=self.async_loop) when supplied, else are called directly. But it lives on Property — not plumbed from Device or Node, so a tree with 200 properties sets it 200 times. Also declared Optional[asyncio.SelectorEventLoop] = False, a bool default against a non-bool annotation, which py.typed now surfaces downstream.
Outbound transport — Controller.mqttc (0.13.0)	Exists, consumer-side only. Device has no equivalent, so a producer in a host that already owns a connection must cede the socket.
Housekeeping — ha/bridge.py:132, :185	Not abstracted. A raw threading.Timer for reconcile, inside the HA-facing module.

Plus property.py's Lock/RLock, which exist to be safe against paho's callback thread — a cost
that only earns its keep while the SDK owns the threading model.

The concrete gap

homie.py:2152    mqttc: Optional[MqttClient] = None     # Controller — has it
homie.py:1219    mqtt_cfg: Optional[dict] = None        # Device — no injection path
homie.py:1234    self.mqttc = None

Device.__init__:1225 already rejects parent= with mqtt_cfg= ("children share the root's MQTT
connection"), so a root device is the single place a connection enters a tree — the natural insertion
point.

Not only a parameter. Controller honours "an injected client is not started or stopped by the
SDK" because both calls are guarded (start at :2238 inside the from_config(...) branch, stop
at :2831 behind if self._owns_client). Device has no such guard:

homie.py:1401    root.mqttc.start()            # unguarded
homie.py:1455    mqttc.stop(timeout=2.0)       # unguarded
homie.py:678     mqttc.start()                 # unguarded (property path)

And one design question I'd rather you answered than I assumed: with a caller-owned client, the
caller implements publish_and_flush — so should an injected client be documented as permitted to
implement it non-blocking (returning False for "not confirmed"), or should stop() grow a way to
skip the flush for callers that can't block?

A range of scopes — your call

1. Patch — mqttc= on Device plus the ownership flag. Strictly additive: mqttc=None default,
   so every existing caller takes today's path unchanged, and the new branch is unreachable without
   injection. No MQTT behaviour changes, nothing existing moves. This alone unlocks the async-host
   cases above.
2. Consolidate — one transport-and-dispatch seam per tree: async_loop promoted off Property
   to the root, mqttc alongside it, threading.Timer behind the same seam. Larger, and it would
   touch existing signatures.
3. Sans-I/O — the SDK as a protocol state machine emitting commands, all I/O performed by the
   caller. The pure form, and a rewrite of homie.py that breaks every consumer. Alpha is the only
   time it's cheap, but it's your cost.

I'm not arguing for 3 — mentioning it so the range is visible and 1 isn't chosen by default merely
because it's what someone turned up asking for.

Offer

Since this isn't a PR-up-front situation: happy to prototype whichever direction interests you and
put it up as a branch or draft PR to look at, rather than something you're asked to merge. Equally
happy to leave it with you, or to hear the current shape is deliberate and the adapting belongs on
our side — that's a useful answer too.

[cayossarian]

Line references verified against origin/main @ 24e1a3c (0.14.0), 2026-08-02.

---

Where this stands after 0.14.0

Since this was opened, 0.14.0 landed mqtt_cfg=None — a Device tree that opens no socket at
all. That came out of the #9 defect rather than from this thread, but it removes one of the
motivating cases listed here: building the Homie model with no broker, for test harnesses or
for deriving topics and $description offline, now works without any seam.

The half that remains is the opposite case: an embedder that already has a live MQTT
connection and wants the SDK to publish through it. Transport-free construction does not help
there — the caller has a client, it just has no way to hand it over on the producer side.

That case is not niche. Any application whose MQTT lifecycle is owned elsewhere hits it — an
asyncio service with a single shared connection, a plugin inside a framework that manages
broker auth centrally, or a test harness that needs to assert on what was published rather
than merely build the tree. The common shape is: the connection exists and is not ours to
open or close.

The gap, precisely

Controller takes a client (homie.py:2179). Device does not — Device.__init__
(:1231-1306) takes id, name, type, parent, nodes, extensions, description_extras, mqtt_cfg, qos.

The asymmetry is not only the parameter. Controller enforces "an injected client is neither
started nor stopped by the SDK" structurally:

2255    if self.mqttc:        # _connect_broker
2256        return            # injected → the start() at :2265 is unreachable
2857    if self._owns_client: # mqttc is None, fixed at construction
2858        self.mqttc.stop()

Device has no equivalent, because it has never had a foreign client to protect: unguarded
start at :691 (Property) and :1428 (Device), unguarded stop at :1482. So the change
is the parameter plus an ownership flag threaded to those three sites — mirroring
Controller. Worth naming explicitly, since it is the difference between the feature working
and it silently tearing down a caller's connection.

What the contract would be

Draft PR #12 adds MqttTransport for the Controller injection point: publish,
subscribe, unsubscribe, and nothing else, because those are the only members reachable on
an injected client there.

A Device-side injection point needs a wider contract, but a specific one — still not
start/stop. Every remaining member is Device-path:

Member	Where
is_connected	:1445, :1473
is_running	:690, :744, :821, :1427
publish_and_flush	:1476 — sole occurrence in the SDK

So the natural shape is a second protocol deriving from MqttTransport with those three
added. Flagging it here so #12 does not read as a partial answer to this: it is the
Controller half of the same contract, deliberately not pre-judging the Device half.

One wart in that shape, and it is the same objection we raise against a flat eight-member
protocol: deriving would hand Device-side implementors unsubscribe, which nothing on the
Device path calls (Property.set_subscribe at :957 subscribes; nothing unsubscribes). One
inherited member is a far smaller tax than five, so our inclination is to wear it — but a
publish + subscribe base with unsubscribe on the Controller side is equally
straightforward if you prefer it. Either way it is your call; raising it now only because #12
is still in draft and the name is easy to change until it is not.

The question from the original post, still open

Restating because it gates the shape rather than following from it: may an injected client
implement publish_and_flush non-blocking, returning False for "not confirmed"?

Current behaviour already leans permissive — the sole call site comments that it "is bounded
and returns False (never blocks/raises) when the broker is unreachable" (:1470-1472), and
:1479 logs the result with nothing branching on it.

Worth noting #12 cannot answer this even by accident: publish_and_flush is not in
MqttTransport at all, because nothing on the Controller path calls it. The first written
contract for that member will be whatever a Device-side protocol says, which is exactly the
decision we are asking you for rather than one we want to make by shipping a docstring.

The alternative, if you would rather not commit to it: stop() grows a way to skip the flush
for callers that cannot block. Device.stop(*, flush_timeout=1.0, stop_timeout=2.0)
(:1447) is bounded at roughly three seconds, which is fine on a thread and not fine on an
event loop.

Two smaller things in the same area

Neither is worth its own thread, and both are visible from the same code:

• Property.async_loop is the only inbound dispatch seam, and it lives on Property
  (:464, :505, :927-928). A tree with two hundred properties sets it two hundred times.
  If a transport seam lands on the root, this is the natural companion — one place per tree
  rather than per property.
• async_loop is declared Optional[asyncio.SelectorEventLoop] = False (:464) — a
  bool default against a non-bool annotation. Harmless at runtime, but py.typed (0.13)
  means downstream type checkers now see it.

One downstream consequence, for completeness

Not a request — recording it so the full scope is visible now rather than arriving later.

distribution-enclosure-simulator's Emitter.__init__ takes only mqtt_cfg and forwards it
to build_graph(...) → ebus_sdk.Device(mqtt_cfg=...). So a producer embedded in a host that
owns its connection would not benefit from a Device seam until Emitter gains a matching
passthrough. That is downstream work on our side, in a branch there, and needs nothing from
you here.

Offer, unchanged

Still glad to prototype whichever direction interests you and put it up as a branch or draft
PR rather than something you are asked to merge — or to hear that the current shape is
deliberate and the adapting belongs on the consumer side. That remains a useful answer.

[dcj]

Both comments and #12 read. Line refs below are current main (0.15.0). Short version: yes to the producer seam, #12 merges after one small rename, I'll answer the flush question, and there is one piece the method behind #12 structurally can't see that I want on the table before the Device protocol gets designed.

#12: rename the type, then merge

MqttTransport typed by what the SDK actually calls is the right move, and the call-site table is the right way to have argued it. One change before it merges, because it settles both the naming and the unsubscribe wart you flagged in the same stroke: make MqttTransport the shared base (publish + subscribe, the surface every injected client has), and derive the role-specific contracts from it.

• MqttTransport (base): publish, subscribe
• MqttControllerTransport(MqttTransport): adds unsubscribe — the consumer injection point this PR wires
• MqttDeviceTransport(MqttTransport): adds is_connected, is_running — lands with the Device seam, not now

So #12 becomes: rename its three-member MqttTransport to MqttControllerTransport, add the two-member MqttTransport base it derives from, export both. Two wins over the flat version. The general name goes to the general surface instead of the consumer-specific one. And because the Device contract will derive from the two-member base rather than from the three-member consumer type, it never inherits unsubscribe: the wart dissolves instead of being worn. MqttClient still satisfies all three structurally, unchanged, and your docstring line "a second protocol deriving from this one" becomes literally correct, the Device protocol deriving from the base.

The flush question, answered

May an injected client implement publish_and_flush non-blocking, returning False? Yes. The sole call site already treats False as "not confirmed" and nothing branches on it (:1507 logs flushed and moves on), so a non-blocking implementation is already safe.

But I'd rather the ownership flag carry it than a docstring. For a non-owned client, stop() should publish $state=disconnected as a plain retained message and return: no publish_and_flush, no mqttc.stop(). We flush today only because we are about to drop the socket and the process may exit; an injected client stays alive and the caller's loop delivers the retained publish. So _owns_client decides both "flush?" and "stop?", no new skip-flush parameter, mirroring what Controller.stop() already does. A side effect worth noting: this drops publish_and_flush off the injected surface entirely, the same way start/stop are owned-only. So the injected Device method surface is publish, subscribe, is_connected, is_running, nothing more.

The piece the call-site method can't see

Typing by what the SDK calls on the client is correct, but it is blind by construction to what the SDK registers and configures at connect, because those are not calls on the client. Two of them are Device correctness, not optional hooks, and both live only inside the from_config(...) call that injection short-circuits (:2013 returns before :2021-2027):

• the LWT ($state=lost, built at :2016). A producer with no last will violates the convention: subscribers never learn it died ungracefully.
• on_connect → refresh_tree (:1947), the whole-tree retained republish after a broker reconnect. Without it an injected Device that survives a broker blip never re-announces, and its retained topics may already be gone.

So a Device that satisfies MqttDeviceTransport will look complete by the call-site mapping and still ship with no will and no reconnect republish. The will especially cannot be bolted on after the fact: it is part of the CONNECT packet, and in the injected model the caller owns connect. So this is a contract, not a wider protocol.

The shape I'd want, and it mirrors what 0.15.0 already does for the disconnect hook ("a BYO caller registers disconnect handling on its own client"):

• the SDK hands the caller the will descriptor ({topic: ebus/5/<root>/$state, payload: lost, retain, qos}) to set on their client before connecting, and
• the caller calls a public republish_tree() (i.e. refresh_tree) from their own on-connect handler.

That keeps the transport protocol to the four read/write members above (no start/stop/will_set) and preserves Homie producer correctness. That is the real design content of this thread, and it is why "add mqttc= plus a flag" is necessary but not the whole job.

The same shape, on the consumer side, is narrower than it looks

For completeness, since it is the same from_config-bypass pattern: the consumer side is mostly fine under injection. A Controller has no $state, so no LWT, and MqttClient recovers its own subscriptions on reconnect, so topic recovery does not depend on the SDK. The disconnect hook being owned-only is documented and defensible. The one thing that does ride on the bypassed on_connect: Controller._on_connect's tree-rooted bookkeeping reset (:2342), which wipes the device registry so retained state re-walks cleanly. A tree-rooted controller on an injected client would not re-walk after a reconnect. Narrow (tree-rooted + injected + reconnect), but I'll fix it in the same place we factor the producer wiring, since it is the same root cause.

The inbound seam, when it lands

Agreed that Property.async_loop is the only inbound dispatch seam and belongs on the root, one per tree, propagated the way QoS already is. One thing to fix while doing it, not just move: the dispatch is asyncio.ensure_future(..., loop=self.async_loop) (:928), and inbound /set arrives on paho's network thread. ensure_future is not safe to call from a thread other than the loop's, so this has to become run_coroutine_threadsafe for the real cross-thread case, which is exactly the async-host case this whole thread is about. And I'll take the async_loop: ... = False -> = None fix along with it. Fast-follow to the transport seam, not part of the first cut.

Two I'm not taking

• The property.py locks stay. Injection does not make us single-threaded: the injected client still runs paho on its own loop_start thread in the normal setup, so SDK callbacks still fire off the loop and the locks stay load-bearing. (And the GroupedPropertyDict RLock is observer-path reentrancy, independent of the transport thread.)
• Sans-I/O I'm setting aside, as you expected.

Offer

Yes, please prototype the producer seam, with the will/republish contract above and the _owns_client-carries-the-flush shape, so it lands mergeable rather than as a direction to argue. Equally, if you'd rather hand it back, say the word and I'll build it; you've already done the load-bearing analysis. I'll file the producer seam, the shared from_config-wiring fix, and the inbound fast-follow as issues so they're followable either way. #12 merges as soon as the rename lands (MqttTransport base + MqttControllerTransport).

[cayossarian]

Taking the rename first: MqttTransport is now the two-member base and
MqttControllerTransport derives from it, both exported, on #12 as 266e329. Thanks for the
review. Your version is better than mine and for a reason I had not seen — putting the
general name on the general surface means a Device contract derives from the base, so
unsubscribe is never inherited. The wart doesn't get worn, it stops existing. There is a
test pinning exactly that: a two-member client satisfies MqttTransport and is rejected by
MqttControllerTransport.

The flush answer I like better than the question. Having _owns_client decide both "flush?"
and "stop?" drops publish_and_flush off the injected surface the same way start/stop
are already off it, which means the Device contract is four read/write members and no
lifecycle at all. That is a smaller contract than the one I was arguing for, arrived at by
the same rule.

One thing from your consumer-side section I want to underline rather than let pass: the
tree-rooted bookkeeping reset in Controller._on_connect (:2342), and that a tree-rooted
controller on an injected client would not re-walk after a reconnect. That combination —
tree-rooted, injected, reconnect — is exactly the shape we would run: one panel, one tree, the
host's client. So of everything still open in this thread, that narrow one is the piece we
would actually consume. Good that it rides along with the from_config factoring.

Designing the producer seam before building it — one constraint you should see first

This is the constraint I hit when I sat down to design the prototype you asked for. It bears on
the shape rather than the implementation, and it is a large part of why I land where I do at
the end. Worth being explicit up front: our own path here is a consumer that emits HA
discovery, so we have no injected producer of our own. What we do have is a stake in the
outcome, described below.

Your framing is right — the will is part of CONNECT and in the injected model the caller owns
connect, so it has to be a contract rather than a protocol member. The thing worth putting on
the table before that contract is written: there is a class of host that cannot satisfy it,
and it is the same class this thread is about — one shared connection, owned elsewhere,
already established before our code loads. MQTT allows exactly one will per connection, so
"the caller sets our will descriptor" competes with whatever the host already put there.

Concretely, the blocking shape is a host whose MQTT layer is single-instance by design, sets
its own will from its own configuration when it builds the client, exposes no will hook to
code layered on top of it, and has the socket up before that code is even loaded. Any one of
those alone is survivable; together they close the window entirely. That is not hypothetical
— it is checkable line by line in a shipping platform (Home Assistant @ 2026.5.3, if you want
to look: mqtt/manifest.json:13, mqtt/client.py:538-544, and the absence of any will entry
in mqtt/__init__.py's __all__).

A producer injected with such a connection would satisfy the Device contract completely, look
correct by the call-site mapping, and still ship with no will — the failure you named:
subscribers never learn it died ungracefully.

The other half of your contract lands verbatim on that same shape of host, which is the
more useful half of this message. Hosts in this class generally do expose a reconnect
callback and a connectivity predicate, because their own subsystems need them — in the
example above they are async_subscribe_connection_status and is_connected, both public
(mqtt/__init__.py:605, :614). Their synchronous publish is also typically
fire-and-forget rather than blocking (mqtt/client.py:129-138), so the synchronous protocol
signature costs them nothing either. The will really is the only piece that does not bend.

The deeper constraint: the host's API caps what an injected producer can express

The will is the visible instance of something more general, and it is worth naming because it
bounds the whole producer-seam design. An injected producer can only express what the host's
API exposes — and both of Homie's liveness mechanisms live above that ceiling:

• the will is a CONNECT-time property, and the host owns connect;
• retained-message expiry is a PUBLISH-time MQTT 5 property, and hosts in this class do not
  forward properties. In the example above, the public publish is
  (topic, payload, qos, retain, encoding) and the internal one calls paho with four
  positional arguments, never passing properties (mqtt/client.py:141-148, :687-691).

Worth saying plainly because I nearly proposed expiry as the answer here: it is not available
to an injected producer, for the same reason the will is not.

There is a point in your favour buried in that. publish(topic, data, qos, retain) — the
signature you already have — is exactly what such a host offers. The transport protocol as
defined fits the constrained case precisely; it is only the two out-of-band mechanisms that
do not.

What does survive, and what it covers

The caller watches the connection and tells the SDK. Not a will substitute — when the
disconnect arrives there is nothing to publish on — but the caller is still running, and it is
the one component positioned to see the host's connection state. Notably this works through
the host's callback API rather than the wire, so the ceiling above does not apply. Given the
reconnect callback it can drive republish_tree(), keep the SDK's device state honest across
the gap, and tell a broker blip apart from a shutdown. That covers every case where the
process survives, and covers it better than a will would: a will fires lost on a
thirty-second blip, where a caller that knows better simply republishes.

Permanent death is the case with no clean answer on such a host. The host's own will fires
on the same event ours would have, so the information exists — but in the example it is
published at qos 0 and not retained (mqtt/const.py:367-372, with DEFAULT_RETAIN = False
at :309). A consumer already subscribed sees it; one that joins later sees nothing, and an
eBus $state still reading ready. Deployments can configure retention, but the producer
cannot.

Expiry is still worth having — just for a different producer. For any producer that owns
its connection (which is most of them, ours included), publishing $state with a Message
Expiry Interval means a dead producer's state evaporates instead of lying: consumers see
absent rather than a false ready, with no live actor required. You are already on MQTTv5
(ebus_mqtt_client/client.py:61), so it is a publish property rather than a protocol change.

I checked this rather than trusting the spec text, on Mosquitto 2.1.2 — two retained $state
topics, one published with message-expiry-interval 4 and one without:

T+0   ebus/5/probe/$state    ready      <- expiry 4s
      ebus/5/control/$state  ready      <- no expiry
T+6   ebus/5/control/$state  ready      <- probe topic is gone, not empty: absent

Costs: it gives $state a TTL that Homie does not specify (your call), it obliges a refresh
within the interval — a heartbeat by another name, one retained publish per device per
interval — and support is worth confirming on brokers you care about beyond Mosquitto.

Why this matters to us, given we are not the producer

Our path is a consumer that turns eBus into HA discovery, so we never publish a $state. We
are, however, the component that renders one: your _availability_from_device_state
(ha/emit.py:191-195) maps the $state topic onto HA availability, ready to online and
everything else to offline. So a producer that dies without a will leaves $state=ready
retained, and we faithfully render a dead panel as online with stale power readings presented
as live. That is the failure we are trying to avoid, one layer downstream of where it is
caused — and it is why a producer-side liveness answer is worth this much attention from a
consumer.

It is also why expiry appeals from where we sit: an evaporated $state leaves the
availability topic with no retained message, which HA holds as unavailable until one arrives.
The producer-side mechanism composes with the consumer-side rendering with nothing needed in
between, and nothing eBus-specific in the consumer.

Where a capability declaration comes in

_owns_client is already a capability contract — one bit, fixed at construction, and after
your flush change it decides three things. The will is asking for a second bit, so the shape
is familiar rather than new:

Caller declares	What the SDK does
SDK owns the client	sets the will itself, as today (:2016)
injected, I will set your will	hands over the descriptor; caller sets it pre-connect
injected, I have my own will, here it is	publishes it as a liveness pointer in $description; the consumer half subscribes and synthesises lost
injected, no will obtainable	transient covered by the wrapper; permanent death is a documented gap

Row three is the middle ground and the only genuinely new one. The host's will fires on
exactly the same event ours would have — same connection, same drop — so on a shared
connection the information is not missing, it is on the wrong topic with the wrong payload.
extensions and description_extras already merge into $description (:1537, :1539),
which is where such a pointer belongs, and declaring an extension is what that array is for.

Its costs, plainly, because there are two: a non-eBus Homie consumer still reads ready
regardless, so the pointer only helps consumers that speak eBus; and it inherits whatever
delivery guarantee the host's will has, which as above may be neither retained nor qos>0.

Worth weighing against how bad the residue actually is. The stale ready self-corrects the
moment the producer returns, so it is unbounded only when the producer never does. That is
narrower than it first sounds — but it is exactly the case you named the will for, and
$state is the one thing separating stale-values-labelled-stale from
stale-values-presented-as-live, since the values are retained too.

One option I will name so you need not wonder why it is absent: an eBus consumer could
repair the topic, publishing $state=lost on the dead producer's own topic. It closes the
gap for everyone, and I would still argue against it — it inverts the ownership model, makes
correctness depend on a consumer being present, and races when two are.

On the prototype

You offered it either way, so I will delay that effort, and I would rather clear the logging PR from #11 that I already owe you.

That is not a retreat from the analysis — it stands, but three things point the same direction. The shape is not settled: the
section above is a question about your convention, not a detail I could resolve while
implementing. We have no injected producer of our own, so I would be guessing at the ergonomics
of an API we do not use, which is how a seam ends up fitting nobody in particular. And we can
decide the will question and implement against it when ready.

One thing I should be explicit about, since we opened this thread and you would be entirely
reasonable to assume otherwise: nothing of ours is blocked on the producer seam. I think it
is the architecturally coherent completion of the injected-transport model and I would argue
for it on those grounds — but it is not gating anything we ship, so please schedule it on its
own merits rather than on ours. What this thread did gate for us has already landed:
mqtt_cfg=None taking children, and the typed injection point in #12.

That is a change since the thread opened rather than something true all along. We settled on
keeping our own transport seam in the emitter rather than handing the socket to the SDK, which
removed the arm that needed Device(mqttc=). Worth saying plainly so you are not sequencing
around a dependency that is not there.

If you would still rather I built it now, say so and I will — but please answer the will question first,
because it decides whether construction grows a parameter or not.