MQTT for IoT skill for Codex

Done well means: the topic tree encodes identity so ACLs can restrict each device to its own branch, QoS and retain flags are chosen per message type rather than globally, every connection uses TLS with per-device credentials, and clients survive broker restarts without losing or duplicating commands that matter.

Inspect first

1. Read AGENTS.md, the broker config (mosquitto.conf, EMQX/HiveMQ/AWS IoT policy) and every place the code calls publish and subscribe.
2. Inventory: list each topic, publisher, subscribers, QoS, retain, payload format and rate. Most bugs show up as an inconsistent row.
3. Capture real traffic before redesigning (needs network to the broker):

mosquitto_sub -h broker.local -p 8883 --cafile ca.crt \
  -u ops -P "$MQTT_PASS" -t 'site/#' -v -W 600 > observed.txt
python3 scripts/topic_check.py match -f 'site/+/telemetry/#' -f 'site/+/cmd/#' \
  --topics-file observed.txt

scripts/topic_check.py validates names and filters against the MQTT matching rules, flags lint issues (leading slash, mixed case, deep trees, # at root) and lists which observed topics each filter covers and which no filter covers. It exits 1 on any invalid input. -v prints the topic first on each line, which is what --topics-file expects; -W 600 stops the capture after ten minutes.

Topic design

{org}/{site}/{device_id}/telemetry/{metric}    device -> cloud, QoS 0, not retained
{org}/{site}/{device_id}/state                 device -> cloud, QoS 1, retained
{org}/{site}/{device_id}/cmd/{name}            cloud -> device, QoS 1, not retained
{org}/{site}/{device_id}/cmd/{name}/result     device -> cloud, QoS 1
{org}/{site}/{device_id}/status                LWT: "offline" retained, "online" on connect

• Lowercase, no leading slash, no spaces, IDs that match the client ID or certificate CN.
• Put the device ID at a fixed level so one ACL pattern covers every device.
• Do not encode values or timestamps in topic levels; they explode subscription matching and broker memory. Payload carries data.
• Version the payload ("v": 2 field or a content type property in MQTT 5), not the topic, unless the change is breaking for all consumers.
• Wildcard filters starting with + or # never match topics beginning with $ ($SYS, $aws), by specification.

QoS and retain decisions

Message	QoS	Retain	Why
High-rate telemetry	0	No	Next sample replaces a lost one; QoS 1 costs a PUBACK per message
Device state / config shadow	1	Yes	New subscribers need the last value immediately
Commands (open valve, set point)	1 + idempotency key	No	Retained commands replay on every reconnect
Billing or audit events	1 + dedupe on consumer	No	QoS 2 is rarely worth its 4-packet handshake end to end
Online/offline	1	Yes (via LWT)	Dashboards need presence without waiting

QoS 1 means at least once: consumers must tolerate duplicates. Delivered QoS is the minimum of publish and subscribe QoS. Clear a retained message by publishing an empty retained payload: mosquitto_pub -t 'site/a/dev1/state' -r -n. Confirm with the user before clearing retained topics on a production broker; dashboards and devices may depend on them.

Sessions and client IDs

• Client IDs must be unique. A second connection with the same ID disconnects the first, which then reconnects and kicks the second: the classic flapping loop. Derive it from the device serial, never a constant in firmware.
• Persistent sessions (MQTT 3.1.1 clean_session=false, MQTT 5 clean_start=false with a session_expiry_interval) queue QoS 1 messages while a device is offline. Set an expiry in MQTT 5; infinite sessions for dead devices fill broker memory.
• MQTT 5 message expiry on commands stops a device from executing a stale command hours later.
• Keepalive 30-60 s for mains devices; longer for battery devices on cellular, balanced against how quickly the LWT should fire (about 1.5 times keepalive).
• Shared subscriptions ($share/workers/site/+/telemetry/#) load-balance backend consumers; without them every consumer instance gets every message.

Broker security (Mosquitto 2.x)

# mosquitto.conf
per_listener_settings false
allow_anonymous false

listener 8883
cafile   /etc/mosquitto/certs/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile  /etc/mosquitto/certs/server.key
require_certificate true
use_identity_as_username true        # certificate CN becomes the username

acl_file /etc/mosquitto/acl

# /etc/mosquitto/acl
# %c = client id, %u = username (here the certificate CN)
pattern write acme/+/%u/telemetry/#
pattern readwrite acme/+/%u/state
pattern write acme/+/%u/status
pattern read  acme/+/%u/cmd/#
pattern write acme/+/%u/cmd/+/result

user backend
topic read acme/#
topic write acme/+/+/cmd/#

• Mosquitto 2.0+ refuses remote connections without an explicit listener and auth; do not “fix” that with allow_anonymous true on a reachable port.
• Username/password devices: mosquitto_passwd -c /etc/mosquitto/passwd dev-0001 (prompts for the password; -c overwrites the file, so drop it when adding users) and password_file instead of certificates.
• Plain 1883 only on localhost or an isolated network. Port 8883 with TLS everywhere else.
• Per-device credentials so one leaked device can be revoked alone. Keep private keys and passwords out of the repo; see secrets-management.
• Validate payloads on the backend; a device that passes the ACL can still send garbage.

Client: paho-mqtt 2.x (Python)

import json, ssl, uuid
import paho.mqtt.client as mqtt

DEVICE = "dev-0001"

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print("connect failed:", reason_code)
        return
    client.publish(f"acme/s1/{DEVICE}/status", "online", qos=1, retain=True)
    client.subscribe(f"acme/s1/{DEVICE}/cmd/#", qos=1)       # resubscribe on every connect

def on_message(client, userdata, msg):
    cmd = json.loads(msg.payload)
    if seen_before(cmd["id"]):                                # QoS 1 duplicates happen
        return
    handle(msg.topic, cmd)

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=DEVICE,
                     protocol=mqtt.MQTTv5)
client.tls_set(ca_certs="ca.crt", certfile=f"{DEVICE}.crt", keyfile=f"{DEVICE}.key",
               tls_version=ssl.PROTOCOL_TLS_CLIENT)
client.will_set(f"acme/s1/{DEVICE}/status", "offline", qos=1, retain=True)
client.reconnect_delay_set(min_delay=1, max_delay=120)
client.on_connect, client.on_message = on_connect, on_message
client.connect("broker.example.com", 8883, keepalive=60)
client.loop_forever(retry_first_connection=True)

paho-mqtt 2.0 made the callback API version argument mandatory; 1.x examples with on_connect(client, userdata, flags, rc) fail. Subscribe inside on_connect, never only once at start, or a reconnect without a persistent session leaves the client deaf. On microcontrollers the same rules apply with ESP-MQTT, Zephyr’s MQTT client or umqtt.

Gotchas

• Large payloads: brokers cap message size (Mosquitto message_size_limit, AWS IoT 128 KB). Send files over HTTPS and publish a reference.
• Ordering is guaranteed only per topic per QoS level from one publisher.
• Bridges and wildcard re-publishing create loops; set bridge topic directions explicitly.
• A device that publishes QoS 1 faster than the link drains fills its in-flight window and stalls; rate-limit or drop to QoS 0 for telemetry.

Verification

• topic_check.py passes on every topic and filter in code and ACLs
• With device A’s credentials, publishing to device B’s topics is refused (test it)
• Broker restart: clients reconnect, resubscribe and presence returns to “online”
• Kill a device’s network: LWT “offline” appears within about 1.5 times keepalive
• Duplicate a command message: the device executes it once

Report

**Broker:** Mosquitto 2.0.20, TLS 8883 with client certificates, ACL by certificate CN
**Topic tree:** acme/{site}/{device}/{telemetry|state|status|cmd}; 9 topics, 3 filters
**Changes:** commands no longer retained; telemetry QoS 1 -> 0; LWT added; session expiry 1 h
**Tests:** cross-device publish denied; broker restart recovered in 4 s; LWT fired at 82 s