A Broker Between Every Device

# labs/mqtt_bridge.py
def topic_matches(subscription: str, topic: str) -> bool:
    """True when a broker would deliver `topic` to a subscriber of `subscription`."""
    levels = subscription.split("/")
    parts = topic.split("/")
    if parts[0].startswith("$") and levels[0] in ("+", "#"):
        return False
    for i, level in enumerate(levels):
        if level == "#":
            return True
        if i >= len(parts):
            return False
        if level != "+" and level != parts[i]:
            return False
    return len(levels) == len(parts)

CASES = [
    ("home/+/+/event", "home/kitchen/motion/event"),
    ("home/+/+/event", "home/front/doorbell/event"),
    ("home/+/+/event", "home/kitchen/motion/state"),
    ("home/+/+/event", "home/kitchen/counter/motion/event"),
    ("home/kitchen/#", "home/kitchen/motion/event"),
    ("home/kitchen/#", "home/kitchen"),
    ("home/#", "home/hall/lights/set"),
    ("#", "$SYS/broker/uptime"),
]

if __name__ == "__main__":
    for subscription, topic in CASES:
        verdict = "deliver" if topic_matches(subscription, topic) else "drop   "
        print(f"{verdict}  {subscription:<16} <- {topic}")

$ uv run python -m labs.mqtt_bridge
deliver  home/+/+/event   <- home/kitchen/motion/event
deliver  home/+/+/event   <- home/front/doorbell/event
drop     home/+/+/event   <- home/kitchen/motion/state
drop     home/+/+/event   <- home/kitchen/counter/motion/event
deliver  home/kitchen/#   <- home/kitchen/motion/event
deliver  home/kitchen/#   <- home/kitchen
deliver  home/#           <- home/hall/lights/set
drop     #                <- $SYS/broker/uptime

No broker was involved in that run, and that is the point of writing the function at all: your naming scheme becomes something you can test in one second, before any hardware exists to be renamed. The fourth case is the one people get wrong from memory. A plus sign is not a wildcard for "anything below here"; it fills exactly one level, so a sensor published under an extra level of nesting goes silently undelivered no matter how reasonable its name looks.

Two smaller rules hide in the last three cases. A hash matches the parent level too, so home/kitchen/# also delivers a message published on home/kitchen itself. And a leading wildcard skips the $SYS tree, where brokers publish their own statistics; without that rule a client subscribed to # for debugging would be flooded with the broker reporting on itself. Stage 4 reuses this same function on the receiving side, so her dispatch rows and her subscriptions are decided by one implementation of one rule.

import paho.mqtt.client as mqtt

BROKER_HOST = "127.0.0.1"
BROKER_PORT = 1883
CLIENT_ID = "glados-bridge"
STATUS_TOPIC = "glados/status"
SUBSCRIPTIONS = [("home/+/+/event", 1), ("home/+/+/state", 1)]

def on_connect(client, userdata, flags, reason_code, properties) -> None:
    if reason_code != 0:
        print(f"[MQTT] refused: {reason_code}")
        return
    print(f"[MQTT] connected to {BROKER_HOST}:{BROKER_PORT} "
          f"(session_present={flags.session_present})")
    client.subscribe(SUBSCRIPTIONS)
    client.publish(STATUS_TOPIC, "online", qos=1, retain=True)
    for topic, qos in SUBSCRIPTIONS:
        print(f"[MQTT] subscribed {topic} qos {qos}")

def on_message(client, userdata, msg) -> None:
    payload = msg.payload.decode("utf-8", errors="replace")
    print(f"[MQTT] {msg.topic} = {payload!r} (qos {msg.qos}, retain {int(msg.retain)})")

def build_client() -> mqtt.Client:
    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2,
                         client_id=CLIENT_ID, clean_session=False)
    client.will_set(STATUS_TOPIC, "offline", qos=1, retain=True)
    client.on_connect = on_connect
    client.on_message = on_message
    return client

if __name__ == "__main__":
    client = build_client()
    client.connect(BROKER_HOST, BROKER_PORT, keepalive=60)
    client.loop_forever()

$ uv run python -m labs.mqtt_bridge   # terminal 1, then two publishes from terminal 2
[MQTT] connected to 127.0.0.1:1883 (session_present=False)
[MQTT] subscribed home/+/+/event qos 1
[MQTT] subscribed home/+/+/state qos 1
[MQTT] home/kitchen/motion/event = 'detected' (qos 1, retain 0)
[MQTT] home/office/window/state = 'open' (qos 0, retain 0)

$ mosquitto_pub -t 'home/kitchen/motion/event' -m detected -q 1   # terminal 2
$ mosquitto_pub -t 'home/office/window/state' -m open -r

Four details in build_client are doing real work. CallbackAPIVersion.VERSION2 is required by paho 2.x, which refuses to guess which callback signature you wrote against; the version 2 signature is the one with reason_code and properties on it. A fixed client_id plus clean_session=False asks the broker to remember this client between connections, which is what makes flags.session_present meaningful and what stage 6 leans on. And will_set registers a last will: a message the broker itself publishes if this client vanishes without saying goodbye, so anything watching glados/status learns her bridge died within a keepalive or two.

Look closely at the second publish. It went out with -r, the retain flag, and it arrived with retain 0. That is correct and it surprises everybody: the flag on a delivered message does not mean "this was published with retain", it means "the broker is replaying a stored message to you now". A live delivery to a client that was already subscribed always reads zero.

$ uv run python -m labs.mqtt_bridge   # started again; nothing happened in the house meanwhile
[MQTT] connected to 127.0.0.1:1883 (session_present=True)
[MQTT] subscribed home/+/+/event qos 1
[MQTT] subscribed home/+/+/state qos 1
[MQTT] home/office/window/state = 'open' (qos 0, retain 1)

$ mosquitto_sub -t 'home/#' -v --retained-only -W 1
home/office/window/state open

$ mosquitto_pub -t 'home/office/window/state' -r -n
$ mosquitto_sub -t 'home/#' -v --retained-only -W 1

She learned the window was open without anybody publishing anything, because the broker keeps exactly one retained message per topic and hands it to every new subscriber the moment the subscription is accepted. That is the difference between state and event stated as a storage policy: a device that publishes its state retained answers the question "what is it now?" for a subscriber that was not born yet, while an event published without the flag is gone the instant the last connected subscriber has it. It is also why the window row in the next section watches a state topic: she wants the current truth at startup, not a history of hinge movements she missed.

The third command deletes it. -n publishes a zero-length payload and -r makes it retained, and a retained empty message is the one thing a broker treats as "forget this topic", which is why the second listing prints nothing at all. There is no other way to remove one. A retained message with persistence enabled outlives the process, the reboot, and your memory of having sent it.

from glados.core import GladOSCore
from labs.automation import register_action
from labs.command_loop import execute_action

CLIENT: mqtt.Client | None = None
CORE: GladOSCore | None = None

TRIGGERS: list[dict] = [
    {"filter": "home/+/motion/event",       "payload": "detected", "action": "greet_arrival"},
    {"filter": "home/front/doorbell/event", "payload": "*",        "action": "announce_visitor"},
    {"filter": "home/+/window/state",       "payload": "open",     "action": "note_open_window"},
]

def match_triggers(topic: str, payload: str) -> list[dict]:
    return [t for t in TRIGGERS
            if topic_matches(t["filter"], topic) and t["payload"] in ("*", payload)]

def split_event(text: str) -> tuple[str, str, str]:
    """'home/kitchen/motion/event detected' -> ('kitchen', 'motion', 'detected')"""
    topic, _, payload = text.partition(" ")
    levels = topic.split("/")
    return levels[1], levels[2], payload

def publish_set(area: str, device: str, value: str) -> None:
    topic = f"home/{area}/{device}/set"
    if CLIENT is None:
        print(f"[DRY] {topic} = {value}")
        return
    CLIENT.publish(topic, value, qos=1, retain=False)
    print(f"[PUB] {topic} = {value}")

def say(line: str) -> None:
    if CORE is None:
        print(f"[SAY] {line}")
        return
    CORE.speak(line)

@register_action("greet_arrival")
def greet_arrival(text: str) -> None:
    area, _, _ = split_event(text)
    publish_set(area, "lights", "ON")

@register_action("announce_visitor")
def announce_visitor(text: str) -> None:
    say("Someone is at the front door. I have decided not to be excited about it.")

@register_action("note_open_window")
def note_open_window(text: str) -> None:
    area, device, value = split_event(text)
    say(f"The {area} {device} is {value}. The heating is, of course, still running.")

def dispatch(topic: str, payload: str) -> int:
    fired = 0
    for rule in match_triggers(topic, payload):
        if execute_action(rule["action"], f"{topic} {payload}"):
            fired += 1
    if fired == 0:
        print(f"[SKIP] {topic} = {payload}")
    return fired

if __name__ == "__main__":
    for topic, payload in [
        ("home/kitchen/motion/event", "detected"),
        ("home/front/doorbell/event", "pressed"),
        ("home/office/window/state", "open"),
        ("home/office/window/state", "closed"),
        ("home/kitchen/lights/set", "ON"),
    ]:
        dispatch(topic, payload)

$ uv run python -m labs.mqtt_bridge   # no broker, no microphone, no speaker
[DRY] home/kitchen/lights/set = ON
[SAY] Someone is at the front door. I have decided not to be excited about it.
[SAY] The office window is open. The heating is, of course, still running.
[SKIP] home/office/window/state = closed
[SKIP] home/kitchen/lights/set = ON

A trigger row carries a filter, not a topic, so one row covers every room you will ever have: home/+/motion/event is written once and matches the sensor you mount in the garage next spring. The payload field is a second, cruder test, because a topic tells you which device spoke and the payload tells you what it said, and open and closed arrive on the same topic. A payload of "*" means the row does not care, which suits a doorbell: there is only one thing a doorbell has to say.

The last two lines of output are the interesting ones. closed matches the window filter but not its payload, so nothing fires. And her own command, home/kitchen/lights/set, matches no row at all, because no row ends in set. That is the feedback loop everybody builds once by accident: subscribe to home/# to catch everything, publish a command, receive your own command, act on it, publish again. The namespace makes it unrepresentable instead of catching it with a flag, and the level that does the work is the last one.

Two compromises deserve naming. Chapter 17 fixed the handler signature at one string, so the bridge packs the topic and payload into one and every handler unpacks it with split_event; a closure per handler would read better and would cost the registry its ability to be called from anywhere else. And CLIENT and CORE are module globals with printed fallbacks, in the pattern chapter 32 set for a missing voice, so the dispatch layer stays runnable with no broker and no audio hardware attached. That is why the output above reproduces on your machine.

from labs.system_config import build_default_config
from labs.wire_core import build_core

def on_message(client, userdata, msg) -> None:
    payload = msg.payload.decode("utf-8", errors="replace")
    print(f"[MQTT] {msg.topic} = {payload!r} (qos {msg.qos}, retain {int(msg.retain)})")
    dispatch(msg.topic, payload)

def main() -> None:
    global CLIENT, CORE
    CORE = build_core(build_default_config())
    CLIENT = build_client()
    CLIENT.connect(BROKER_HOST, BROKER_PORT, keepalive=60)
    try:
        CLIENT.loop_forever()
    except KeyboardInterrupt:
        CLIENT.publish(STATUS_TOPIC, "offline", qos=1, retain=True).wait_for_publish()
        CLIENT.disconnect()
        print("\n[MQTT] bridge down")

if __name__ == "__main__":
    main()

$ uv run python -m labs.mqtt_bridge   # her voice comes out of the speaker, not the terminal
Loading whisper base on cpu...
Loading F5-TTS...
[MQTT] connected to 127.0.0.1:1883 (session_present=True)
[MQTT] subscribed home/+/+/event qos 1
[MQTT] subscribed home/+/+/state qos 1
[MQTT] home/kitchen/motion/event = 'detected' (qos 1, retain 0)
[PUB] home/kitchen/lights/set = ON
[MQTT] home/front/doorbell/event = 'pressed' (qos 1, retain 0)
^C
[MQTT] bridge down

$ mosquitto_sub -t 'home/+/lights/set' -v -W 20   # terminal 3, watching what she publishes
home/kitchen/lights/set ON

There is no lamp in this house. Terminal three is standing in for one, and her code neither knows nor cares: she published an intention under a name, the broker matched it against whoever asked, and the day a real plug subscribes to that topic the only thing that changes is who receives the message. Nothing prints between the doorbell line and the ^C because the doorbell handler speaks, and speaking happens through the speaker.

Count the diff from stage 4 honestly. One added line in on_message, one main that wires the two globals, and the entire house is now connected to an automation engine written back in volume 2. Adding a smoke alarm tomorrow is one dict in TRIGGERS and one decorated function; the connection code, the subscriptions and the callbacks stay untouched, because they never knew what a smoke alarm was in the first place.

$ mosquitto_sub -i hall-lights -c -q 1 -t 'home/hall/lights/set' -v   # the plug, connecting once
^C

$ mosquitto_pub -t 'home/hall/lights/set' -m OFF -q 1   # published while the plug is down

$ mosquitto_sub -i hall-lights -c -q 1 -t 'home/hall/lights/set' -v   # the plug, coming back
home/hall/lights/set OFF

The message was held for a client that was not connected, and delivered the moment it returned. Three things had to be true at once for that: the subscriber used a stable -i client id, it asked for a session that survives disconnection with -c, and the publish used -q 1. Drop any one and the broker has nowhere to put the message. At QoS 0 it is discarded on the spot, and mosquitto_pub exits 0 without complaint, because from its point of view handing the message to the broker was the job.

A device that has never connected has no session either, so nothing can be queued against it, and a command sent to a plug you have not set up yet evaporates in total silence. That gives three possible fates for a command aimed at an absent device: lost instantly, delivered late (possibly hours late, which for "turn the lights on" is often worse than losing it), or retained and repeated forever. Pick per topic, deliberately. The way to know whether the hall lamp is on is to read home/hall/lights/state, which the lamp itself publishes retained, and never to assume that having published a set made something true.