Refactor MQTT (#24010)

* refactor mqtt so that Frigate owns the transport lifecycle instead of delegating it to paho

* release the shutdown barrier on worker crash and replay retained publishes the broker never acked

* collapse in-flight retained values by topic and release the shutdown barrier from a finally

* replay the outage buffer before the publish queue so newer values are not reverted

frigate/app.py

@@ -335,6 +335,7 @@ class FrigateApp:
             self.ptz_metrics,
             comms,
         )
+        self.dispatcher.start_communicators()
 
     def init_profile_manager(self) -> None:
         self.profile_manager = ProfileManager(

frigate/comms/base_communicator.py

@@ -1,11 +1,27 @@
 from abc import ABC, abstractmethod
 from collections.abc import Callable
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from frigate.comms.dispatcher import Dispatcher
 
 
 class Communicator(ABC):
     """pub/sub model via specific protocol."""
 
+    def attach_dispatcher(self, dispatcher: "Dispatcher") -> None:
+        """Receive the owning dispatcher.
+
+        Transports that need more than the receiver callback (the command topic
+        surface, the snapshot API) take it here rather than reaching through the
+        bound receiver.
+        """
+        return None
+
+    def start(self) -> None:
+        """Start background I/O after receiver wiring is complete."""
+        return None
+
     @abstractmethod
     def publish(self, topic: str, payload: Any, retain: bool = False) -> None:
         """Send data via specific protocol."""

frigate/comms/dispatcher.py

@@ -49,6 +49,11 @@ from frigate.util.services import restart_frigate
 
 logger = logging.getLogger(__name__)
 
+# <camera>/<command>/<sub_command>/set, one segment longer than the rest
+SUB_COMMAND_TOPICS = frozenset({"motion_mask", "object_mask", "zone"})
+
+BARE_COMMAND_TOPICS = frozenset({"onConnect", "restart"})
+
 
 class Dispatcher:
     """Handle communication between Frigate and communicators."""
@@ -103,12 +108,114 @@ class Dispatcher:
         }
         self.profile_manager: ProfileManager | None = None
 
-        for comm in self.comms:
-            comm.subscribe(self._receive)
-
         self.web_push_client = next(
             (comm for comm in communicators if isinstance(comm, WebPushClient)), None
         )
+
+        for comm in self.comms:
+            comm.subscribe(self._receive)
+            comm.attach_dispatcher(self)
+
+    def start_communicators(self) -> None:
+        """Start communicators after dispatcher wiring is fully initialized."""
+        for comm in self.comms:
+            comm.start()
+
+    def is_command_topic(self, topic: str) -> bool:
+        """Whether a prefix-stripped topic maps to a command handler.
+
+        Transports that fan a whole topic tree in must filter on this:
+        _receive() republishes anything it does not recognize, so forwarding
+        unfiltered would echo Frigate's own publishes back.
+        """
+        parts = topic.split("/")
+
+        if topic in BARE_COMMAND_TOPICS:
+            return True
+
+        if len(parts) == 2 and parts[1] == "ptz":
+            return True
+
+        if len(parts) == 2 and parts[1] == "set":
+            return parts[0] in self._global_settings_handlers
+
+        if len(parts) == 3 and parts[2] == "set":
+            return (
+                parts[1] in self._camera_settings_handlers
+                and parts[1] not in SUB_COMMAND_TOPICS
+            )
+
+        if len(parts) == 3 and parts[2] == "suspend":
+            return parts[1] == "notifications"
+
+        if len(parts) == 4 and parts[3] == "set":
+            return parts[1] in SUB_COMMAND_TOPICS
+
+        return False
+
+    def _build_camera_activity_snapshot(self) -> tuple[dict[str, Any], dict[str, Any]]:
+        """Build the current runtime activity snapshot for reconnect consumers."""
+        camera_status = {
+            camera: status
+            for camera, status in self.camera_activity.last_camera_activity.copy().items()
+            if camera in self.config.cameras
+        }
+        audio_detections = self.audio_activity.current_audio_detections.copy()
+        cameras_with_status = camera_status.keys()
+
+        for camera in self.config.cameras.keys():
+            if camera not in cameras_with_status:
+                camera_status[camera] = {}
+
+            camera_status[camera]["config"] = {
+                "detect": self.config.cameras[camera].detect.enabled,
+                "enabled": self.config.cameras[camera].enabled,
+                "snapshots": self.config.cameras[camera].snapshots.enabled,
+                "record": self.config.cameras[camera].record.enabled,
+                "audio": self.config.cameras[camera].audio.enabled,
+                "audio_transcription": self.config.cameras[
+                    camera
+                ].audio_transcription.live_enabled,
+                "notifications": self.config.cameras[camera].notifications.enabled,
+                "notifications_suspended": int(
+                    self.web_push_client.suspended_cameras.get(camera, 0)
+                )
+                if self.web_push_client
+                and camera in self.web_push_client.suspended_cameras
+                else 0,
+                "autotracking": self.config.cameras[camera].onvif.autotracking.enabled,
+                "alerts": self.config.cameras[camera].review.alerts.enabled,
+                "detections": self.config.cameras[camera].review.detections.enabled,
+                "object_descriptions": self.config.cameras[
+                    camera
+                ].objects.genai.enabled,
+                "review_descriptions": self.config.cameras[camera].review.genai.enabled,
+            }
+
+        return camera_status, audio_detections
+
+    def publish_runtime_snapshot(
+        self,
+        publisher: Callable[[str, Any, bool], None] | None = None,
+    ) -> None:
+        """Publish the runtime snapshot for newly connected listeners."""
+        publish = publisher or self.publish
+        camera_status, audio_detections = self._build_camera_activity_snapshot()
+
+        publish("camera_activity", json.dumps(camera_status), False)
+        publish("model_state", json.dumps(self.model_state.copy()), False)
+        publish(
+            "embeddings_reindex_progress",
+            json.dumps(self.embeddings_reindex.copy()),
+            False,
+        )
+        publish("birdseye_layout", json.dumps(self.birdseye_layout.copy()), False)
+        publish("audio_detections", json.dumps(audio_detections), False)
+        publish(
+            "profile/state",
+            self.config.active_profile or "none",
+            True,
+        )
         if self.web_push_client is not None:
             self.web_push_client.set_suspension_broadcaster(self.publish)
 
@@ -127,17 +234,11 @@ class Dispatcher:
 
             try:
                 if command_type == "set":
-                    # Commands that require a sub-command (mask/zone name)
-                    sub_command_required = {
-                        "motion_mask",
-                        "object_mask",
-                        "zone",
-                    }
                     if sub_command:
                         self._camera_settings_handlers[command](
                             camera_name, sub_command, payload
                         )
-                    elif command in sub_command_required:
+                    elif command in SUB_COMMAND_TOPICS:
                         logger.error(
                             "Command %s requires a sub-command (mask/zone name)",
                             command,
@@ -271,67 +372,11 @@ class Dispatcher:
         def handle_birdseye_layout() -> None:
             self.publish("birdseye_layout", json.dumps(self.birdseye_layout.copy()))
 
-        def handle_on_connect() -> None:
-            camera_status = {
-                camera: status
-                for camera, status in self.camera_activity.last_camera_activity.copy().items()
-                if camera in self.config.cameras
-            }
-            audio_detections = self.audio_activity.current_audio_detections.copy()
-            cameras_with_status = camera_status.keys()
-
-            for camera in self.config.cameras.keys():
-                if camera not in cameras_with_status:
-                    camera_status[camera] = {}
-
-                camera_status[camera]["config"] = {
-                    "detect": self.config.cameras[camera].detect.enabled,
-                    "enabled": self.config.cameras[camera].enabled,
-                    "snapshots": self.config.cameras[camera].snapshots.enabled,
-                    "record": self.config.cameras[camera].record.enabled,
-                    "audio": self.config.cameras[camera].audio.enabled,
-                    "audio_transcription": self.config.cameras[
-                        camera
-                    ].audio_transcription.live_enabled,
-                    "notifications": self.config.cameras[camera].notifications.enabled,
-                    "notifications_suspended": int(
-                        self.web_push_client.suspended_cameras.get(camera, 0)
-                    )
-                    if self.web_push_client
-                    and camera in self.web_push_client.suspended_cameras
-                    else 0,
-                    "autotracking": self.config.cameras[
-                        camera
-                    ].onvif.autotracking.enabled,
-                    "alerts": self.config.cameras[camera].review.alerts.enabled,
-                    "detections": self.config.cameras[camera].review.detections.enabled,
-                    "object_descriptions": self.config.cameras[
-                        camera
-                    ].objects.genai.enabled,
-                    "review_descriptions": self.config.cameras[
-                        camera
-                    ].review.genai.enabled,
-                }
-
-            self.publish("camera_activity", json.dumps(camera_status))
-            self.publish("model_state", json.dumps(self.model_state.copy()))
-            self.publish(
-                "embeddings_reindex_progress",
-                json.dumps(self.embeddings_reindex.copy()),
-            )
-            self.publish("birdseye_layout", json.dumps(self.birdseye_layout.copy()))
-            self.publish("audio_detections", json.dumps(audio_detections))
-            self.publish(
-                "profile/state",
-                self.config.active_profile or "none",
-                retain=True,
-            )
-
         def handle_notification_test() -> None:
             self.publish("notification_test", "Test notification")
 
         # Dictionary mapping topic to handlers
-        topic_handlers = {
+        topic_handlers: dict[str, Callable[[], Any]] = {
             INSERT_MANY_RECORDINGS: handle_insert_many_recordings,
             REQUEST_REGION_GRID: handle_request_region_grid,
             INSERT_PREVIEW: handle_insert_preview,
@@ -354,7 +399,7 @@ class Dispatcher:
             "jobState": handle_job_state,
             "audioTranscriptionState": handle_audio_transcription_state,
             "birdseyeLayout": handle_birdseye_layout,
-            "onConnect": handle_on_connect,
+            "onConnect": self.publish_runtime_snapshot,
         }
 
         if topic.endswith("set") or topic.endswith("ptz") or topic.endswith("suspend"):

frigate/comms/inter_process.py

@@ -18,10 +18,13 @@ SOCKET_REP_REQ = "ipc:///tmp/cache/comms"
 
 class InterProcessCommunicator(Communicator):
     def __init__(self) -> None:
+        # bound eagerly so subprocesses starting before start_communicators()
+        # can still connect; their requests queue in zmq until the reader runs
         self.context = zmq.Context()
         self.socket = self.context.socket(zmq.REP)
         self.socket.bind(SOCKET_REP_REQ)
         self.stop_event: MpEvent = mp.Event()
+        self.reader_thread: threading.Thread | None = None
 
     def publish(self, topic: str, payload: Any, retain: bool = False) -> None:
         """There is no communication back to the processes."""
@@ -29,6 +32,8 @@ class InterProcessCommunicator(Communicator):
 
     def subscribe(self, receiver: Callable) -> None:
         self._dispatcher = receiver
+
+    def start(self) -> None:
         self.reader_thread = threading.Thread(target=self.read)
         self.reader_thread.start()
 
@@ -61,7 +66,8 @@ class InterProcessCommunicator(Communicator):
 
     def stop(self) -> None:
         self.stop_event.set()
-        self.reader_thread.join()
+        if self.reader_thread is not None:
+            self.reader_thread.join()
         self.socket.close(linger=0)
         self.context.destroy(linger=0)
 

frigate/comms/mqtt.py

@@ -1,7 +1,12 @@
+from __future__ import annotations
+
 import logging
+import queue
 import threading
+import time
 from collections.abc import Callable
-from typing import Any
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
 
 import paho.mqtt.client as mqtt
 from paho.mqtt.enums import CallbackAPIVersion
@@ -9,8 +14,25 @@ from paho.mqtt.enums import CallbackAPIVersion
 from frigate.comms.base_communicator import Communicator
 from frigate.config import FrigateConfig, birdseye_modes_to_mqtt_payload
 
+if TYPE_CHECKING:
+    from frigate.comms.dispatcher import Dispatcher
+
 logger = logging.getLogger(__name__)
 
+MQTT_LOOP_TIMEOUT = 1.0
+MQTT_RECONNECT_INTERVAL = 10.0
+MQTT_SHUTDOWN_FLUSH_TIMEOUT = 5.0
+MQTT_ON_CONNECT_RATE_LIMIT = 1.0
+MQTT_PUBLISH_WAIT_INTERVAL = 0.1
+
+
+@dataclass(slots=True)
+class QueuedPublish:
+    topic: str
+    payload: Any
+    retain: bool
+    done: threading.Event | None = None
+
 
 class MqttClient(Communicator):
     """Frigate wrapper for mqtt client."""
@@ -19,28 +41,80 @@ class MqttClient(Communicator):
         self.config = config
         self.mqtt_config = config.mqtt
         self.connected = False
+        self.client: mqtt.Client | None = None
+        self._dispatcher: Callable[[str, Any], Any] | None = None
+        self._command_router: Dispatcher | None = None
+        self._worker: threading.Thread | None = None
+        self._stop_event = threading.Event()
+        self._publish_queue: queue.Queue[QueuedPublish] = queue.Queue()
+        self._callback_queue: queue.Queue[tuple[Any, ...]] = queue.Queue()
+        self._retained_lock = threading.Lock()
+        self._pending_retained: dict[str, tuple[Any, bool]] = {}
+        self._inflight_retained: dict[int, tuple[str, Any]] = {}
+        self._subscription_mid: int | None = None
+        self._subscription_ready = False
+        self._next_connect_time = 0.0
+        self._last_on_connect_dispatch = 0.0
 
     def subscribe(self, receiver: Callable) -> None:
         """Wrapper for allowing dispatcher to subscribe."""
         self._dispatcher = receiver
-        self._start()
+
+    def attach_dispatcher(self, dispatcher: Dispatcher) -> None:
+        """Take Dispatcher's command surface and snapshot API."""
+        self._command_router = dispatcher
+
+    def start(self) -> None:
+        """Start the MQTT worker after all receiver wiring is complete."""
+
+        if self._worker and self._worker.is_alive():
+            return
+
+        self._stop_event.clear()
+        self._start_worker()
 
     def publish(self, topic: str, payload: Any, retain: bool = False) -> None:
         """Wrapper for publishing when client is in valid state."""
+        full_topic = f"{self.mqtt_config.topic_prefix}/{topic}"
+
         if not self.connected:
-            logger.debug(f"Unable to publish to {topic}: client is not connected")
+            if retain:
+                self._queue_retained(full_topic, payload, retain)
+            else:
+                logger.debug("Unable to publish to %s: client is not connected", topic)
             return
 
-        self.client.publish(
+        self._publish_queue.put(QueuedPublish(full_topic, payload, retain))
-            f"{self.mqtt_config.topic_prefix}/{topic}",
-            payload,
-            qos=self.config.mqtt.qos,
-            retain=retain,
-        )
 
     def stop(self) -> None:
-        self.publish("available", "stopped", retain=True)
+        if self._worker is None:
-        self.client.disconnect()
+            return
+
+        if self.connected and self._subscription_ready:
+            publish_done = threading.Event()
+            self._publish_queue.put(
+                QueuedPublish(
+                    f"{self.mqtt_config.topic_prefix}/available",
+                    "stopped",
+                    True,
+                    publish_done,
+                )
+            )
+            publish_done.wait(MQTT_SHUTDOWN_FLUSH_TIMEOUT)
+
+        self._stop_event.set()
+
+        if self.client is not None:
+            try:
+                self.client.disconnect()
+            except Exception:
+                logger.debug("MQTT disconnect raised during shutdown", exc_info=True)
+
+        if self._worker.is_alive():
+            self._worker.join(MQTT_SHUTDOWN_FLUSH_TIMEOUT + MQTT_LOOP_TIMEOUT)
+
+        self._cleanup_client()
+        self._worker = None
 
     def _notifications_enabled_in_config(self) -> bool:
         """Whether notifications are configured globally or on any camera.
@@ -54,17 +128,17 @@ class MqttClient(Communicator):
             for cam in self.config.cameras.values()
         )
 
-    def _set_initial_topics(self) -> None:
+    def _publish_retained_state(self) -> None:
-        """Set initial state topics."""
+        """Publish retained MQTT state after a successful subscribe."""
         for camera_name, camera in self.config.cameras.items():
             self.publish(
                 f"{camera_name}/enabled/state",
-                "ON" if camera.enabled_in_config else "OFF",
+                "ON" if camera.enabled else "OFF",
                 retain=True,
             )
             self.publish(
                 f"{camera_name}/recordings/state",
-                "ON" if camera.record.enabled_in_config else "OFF",
+                "ON" if camera.record.enabled else "OFF",
                 retain=True,
             )
             self.publish(
@@ -74,7 +148,7 @@ class MqttClient(Communicator):
             )
             self.publish(
                 f"{camera_name}/audio/state",
-                "ON" if camera.audio.enabled_in_config else "OFF",
+                "ON" if camera.audio.enabled else "OFF",
                 retain=True,
             )
             self.publish(
@@ -89,7 +163,7 @@ class MqttClient(Communicator):
             )
             self.publish(
                 f"{camera_name}/motion/state",
-                "ON",
+                "ON" if camera.motion.enabled else "OFF",
                 retain=True,
             )
             self.publish(
@@ -99,7 +173,7 @@ class MqttClient(Communicator):
             )
             self.publish(
                 f"{camera_name}/ptz_autotracker/state",
-                "ON" if camera.onvif.autotracking.enabled_in_config else "OFF",
+                "ON" if camera.onvif.autotracking.enabled else "OFF",
                 retain=True,
             )
             self.publish(
@@ -133,22 +207,22 @@ class MqttClient(Communicator):
             )
             self.publish(
                 f"{camera_name}/review_alerts/state",
-                "ON" if camera.review.alerts.enabled_in_config else "OFF",
+                "ON" if camera.review.alerts.enabled else "OFF",
                 retain=True,
             )
             self.publish(
                 f"{camera_name}/review_detections/state",
-                "ON" if camera.review.detections.enabled_in_config else "OFF",
+                "ON" if camera.review.detections.enabled else "OFF",
                 retain=True,
             )
             self.publish(
                 f"{camera_name}/object_descriptions/state",
-                "ON" if camera.objects.genai.enabled_in_config else "OFF",
+                "ON" if camera.objects.genai.enabled else "OFF",
                 retain=True,
             )
             self.publish(
                 f"{camera_name}/review_descriptions/state",
-                "ON" if camera.review.genai.enabled_in_config else "OFF",
+                "ON" if camera.review.genai.enabled else "OFF",
                 retain=True,
             )
 
@@ -189,13 +263,521 @@ class MqttClient(Communicator):
         )
         self.publish("available", "online", retain=True)
 
-    def on_mqtt_command(
+    def _create_client(self) -> mqtt.Client:
-        self, client: mqtt.Client, userdata: Any, message: mqtt.MQTTMessage
+        """Build a fresh paho client for a single connect attempt."""
-    ) -> None:
+        client = mqtt.Client(
-        self._dispatcher(
+            callback_api_version=CallbackAPIVersion.VERSION2,
-            message.topic.replace(f"{self.mqtt_config.topic_prefix}/", "", 1),
+            client_id=self.mqtt_config.client_id,
-            message.payload.decode(),
+            reconnect_on_failure=False,
         )
+        client.on_connect = self._on_connect
+        client.on_disconnect = self._on_disconnect
+        client.on_message = self._on_message
+        client.on_subscribe = self._on_subscribe
+        client.on_publish = self._on_publish
+        client.will_set(
+            self.mqtt_config.topic_prefix + "/available",
+            payload="offline",
+            qos=1,
+            retain=True,
+        )
+
+        if self.mqtt_config.tls_ca_certs is not None:
+            if (
+                self.mqtt_config.tls_client_cert is not None
+                and self.mqtt_config.tls_client_key is not None
+            ):
+                client.tls_set(
+                    self.mqtt_config.tls_ca_certs,
+                    self.mqtt_config.tls_client_cert,
+                    self.mqtt_config.tls_client_key,
+                )
+            else:
+                client.tls_set(self.mqtt_config.tls_ca_certs)
+
+        if self.mqtt_config.tls_insecure is not None:
+            client.tls_insecure_set(self.mqtt_config.tls_insecure)
+
+        if self.mqtt_config.user is not None:
+            client.username_pw_set(
+                self.mqtt_config.user,
+                password=self.mqtt_config.password,
+            )
+
+        return client
+
+    def _start_worker(self) -> None:
+        self._worker = threading.Thread(
+            target=self._worker_main, name="mqtt", daemon=True
+        )
+        self._worker.start()
+        logger.info("MQTT worker started")
+
+    def _worker_main(self) -> None:
+        """Run the worker loop.
+
+        An unexpected crash disables MQTT for this session rather than taking
+        Frigate down with it, so it has to announce itself: without the offline
+        publish, consumers keep the last retained values and see a healthy
+        Frigate that has simply stopped updating.
+        """
+        try:
+            self._mqtt_loop_worker()
+        except Exception:
+            if not self._stop_event.is_set():
+                logger.exception("MQTT worker crashed, disabling MQTT for this session")
+                self._stop_event.set()
+                self._subscription_ready = False
+                self._publish_offline_availability()
+                self.connected = False
+        finally:
+            # nothing drains the queue once the loop is gone, so release any
+            # waiter here or stop() blocks for the full flush timeout
+            self._requeue_disconnected_publishes()
+            self._cleanup_client()
+
+    def _publish_offline_availability(self) -> None:
+        """Announce that MQTT is going away after a worker crash.
+
+        _cleanup_client() disconnects cleanly, which tells the broker to
+        suppress the will, so the retained topic would otherwise stay "online".
+        """
+        if self.client is None:
+            return
+
+        try:
+            message_info = self.client.publish(
+                f"{self.mqtt_config.topic_prefix}/available",
+                "offline",
+                qos=self.config.mqtt.qos,
+                retain=True,
+            )
+
+            # pumped here rather than through _wait_for_publish() so the drain
+            # that may have just crashed is not re-entered
+            deadline = time.monotonic() + MQTT_SHUTDOWN_FLUSH_TIMEOUT
+            while not message_info.is_published() and time.monotonic() < deadline:
+                if (
+                    self.client.loop(timeout=MQTT_PUBLISH_WAIT_INTERVAL)
+                    != mqtt.MQTT_ERR_SUCCESS
+                ):
+                    break
+        except Exception:
+            logger.warning(
+                "MQTT is dormant and the broker could not be told Frigate is offline",
+                exc_info=True,
+            )
+
+    def _mqtt_loop_worker(self) -> None:
+        # The worker owns all socket I/O so reconnect, subscribe, and publish
+        # ordering stays serialized in one place.
+        while not self._stop_event.is_set():
+            if self.client is None:
+                wait_time = self._next_connect_time - time.monotonic()
+                if wait_time > 0:
+                    self._stop_event.wait(min(wait_time, MQTT_LOOP_TIMEOUT))
+                    continue
+
+                if not self._connect_client():
+                    self._next_connect_time = time.monotonic() + MQTT_RECONNECT_INTERVAL
+                    continue
+
+            assert self.client is not None
+            try:
+                result = self.client.loop(timeout=MQTT_LOOP_TIMEOUT)
+            except (OSError, mqtt.WebsocketConnectionError) as err:
+                logger.warning("MQTT loop error: %s", err)
+                self._schedule_reconnect()
+                continue
+
+            self._drain_callback_queue()
+            self._drain_publish_queue()
+
+            if self._stop_event.is_set():
+                break
+
+            if result != mqtt.MQTT_ERR_SUCCESS and self.client is not None:
+                logger.error("MQTT loop returned error code: %s", result)
+                self._schedule_reconnect()
+
+    def _connect_client(self) -> bool:
+        """Create and connect a new client instance owned by the worker thread."""
+        try:
+            self.client = self._create_client()
+            self.client.connect(self.mqtt_config.host, self.mqtt_config.port, 60)
+        except Exception as err:
+            logger.error("Unable to connect to MQTT server: %s", err)
+            self._cleanup_client()
+            return False
+
+        return True
+
+    def _cleanup_client(self) -> None:
+        """Drop session-specific state and release the current paho client."""
+        self.connected = False
+        self._subscription_ready = False
+        self._subscription_mid = None
+        self._requeue_inflight_retained()
+
+        client = self.client
+        self.client = None
+
+        if client is None:
+            return
+
+        try:
+            client.disconnect()
+        except Exception:
+            logger.debug("MQTT client cleanup raised disconnect error", exc_info=True)
+
+    def _schedule_reconnect(self) -> None:
+        """Tear down the current session and arm the next reconnect attempt."""
+        if self._stop_event.is_set():
+            return
+
+        self.connected = False
+        self._subscription_ready = False
+        self._subscription_mid = None
+        self._requeue_disconnected_publishes()
+        self._next_connect_time = time.monotonic() + MQTT_RECONNECT_INTERVAL
+        logger.info("MQTT reconnect scheduled in %.1fs", MQTT_RECONNECT_INTERVAL)
+        self._cleanup_client()
+
+    def _requeue_inflight_retained(self) -> None:
+        """Rebuffer retained publishes paho took but the broker never acked.
+
+        Dropping the client drops paho's outbound queue with it, and the session
+        is clean, so the broker will not resume delivery on the new one.
+        """
+        with self._retained_lock:
+            # mids are insertion ordered, so collapsing by topic keeps the
+            # newest value when several updates to one topic were in flight
+            latest = {
+                topic: payload for topic, payload in self._inflight_retained.values()
+            }
+            self._inflight_retained.clear()
+
+        for topic, payload in latest.items():
+            self._queue_retained(topic, payload, True, overwrite=False)
+
+    def _buffer_undelivered(
+        self, queued_publish: QueuedPublish, overwrite: bool = True
+    ) -> None:
+        """Handle a publish that never reached the broker.
+
+        Releasing the waiter matters on every path: stop() blocks on it, so a
+        broker error would otherwise stall shutdown for the full flush timeout.
+        """
+        if queued_publish.retain:
+            self._queue_retained(
+                queued_publish.topic,
+                queued_publish.payload,
+                queued_publish.retain,
+                overwrite=overwrite,
+            )
+
+        if queued_publish.done is not None:
+            queued_publish.done.set()
+
+    def _requeue_disconnected_publishes(self) -> None:
+        while True:
+            try:
+                queued_publish = self._publish_queue.get_nowait()
+            except queue.Empty:
+                break
+
+            self._buffer_undelivered(queued_publish)
+
+    def _drain_callback_queue(self) -> None:
+        # Paho callbacks only enqueue transport events; state transitions run
+        # here on the worker thread.
+        while True:
+            try:
+                event = self._callback_queue.get_nowait()
+            except queue.Empty:
+                break
+
+            event_type = event[0]
+
+            if event_type == "connect":
+                self._handle_connect_event(event[1])
+            elif event_type == "connect_failure":
+                self._handle_connect_failure(event[1])
+            elif event_type == "disconnect":
+                self._handle_disconnect_event(event[1])
+            elif event_type == "subscribed":
+                self._handle_subscribe_event(event[1], event[2])
+            elif event_type == "message":
+                self._handle_inbound_message(event[1], event[2])
+            elif event_type == "published":
+                self._handle_publish_event(event[1])
+
+    def _drain_publish_queue(self) -> None:
+        """Publish queued work only after the session is fully subscribed.
+
+        Oldest first: the outage buffer replays before the queue, so a topic
+        that changed since the reconnect ends up on its newest value rather
+        than being reverted by the replay.
+        """
+        if self.connected and not self._subscription_ready:
+            return
+
+        self._flush_pending_retained()
+
+        while True:
+            try:
+                queued_publish = self._publish_queue.get_nowait()
+            except queue.Empty:
+                break
+
+            if not self.connected:
+                self._buffer_undelivered(queued_publish)
+                continue
+
+            self._publish_direct(queued_publish)
+
+    def _flush_pending_retained(self) -> None:
+        """Replay the latest retained state once the broker session is ready."""
+        if not self.connected or not self._subscription_ready:
+            return
+
+        with self._retained_lock:
+            pending = list(self._pending_retained.items())
+            self._pending_retained.clear()
+
+        for topic, (payload, retain) in pending:
+            self._publish_direct(QueuedPublish(topic, payload, retain))
+
+    def _publish_direct(self, queued_publish: QueuedPublish) -> None:
+        """Publish a queued message from the worker thread's serialized context.
+
+        The waiter is released however this exits. The message is already off
+        the queue by now, so nothing else can recover it for a stop() that is
+        blocked waiting on it.
+        """
+        try:
+            if self.client is None:
+                # never attempted, so anything already buffered for this topic
+                # was written later and has to survive
+                self._buffer_undelivered(queued_publish, overwrite=False)
+                return
+
+            try:
+                message_info = self.client.publish(
+                    queued_publish.topic,
+                    queued_publish.payload,
+                    qos=self.config.mqtt.qos,
+                    retain=queued_publish.retain,
+                )
+            except (OSError, mqtt.WebsocketConnectionError) as err:
+                logger.warning(
+                    "MQTT publish failed for %s: %s", queued_publish.topic, err
+                )
+                # a newer buffered value for this topic wins over the failed one
+                self._buffer_undelivered(queued_publish, overwrite=False)
+                self._schedule_reconnect()
+                return
+
+            if message_info.rc != mqtt.MQTT_ERR_SUCCESS:
+                logger.error(
+                    "Unable to publish to %s: mqtt error %s",
+                    queued_publish.topic,
+                    message_info.rc,
+                )
+                self._buffer_undelivered(queued_publish, overwrite=False)
+                self._schedule_reconnect()
+                return
+
+            # a successful rc only means paho accepted the message; above qos 0
+            # it is not durable until the broker acks, so keep a copy for replay
+            if queued_publish.retain and not message_info.is_published():
+                with self._retained_lock:
+                    self._inflight_retained[message_info.mid] = (
+                        queued_publish.topic,
+                        queued_publish.payload,
+                    )
+
+            if queued_publish.done is not None:
+                self._wait_for_publish(message_info)
+        finally:
+            if queued_publish.done is not None:
+                queued_publish.done.set()
+
+    def _handle_publish_event(self, mid: int) -> None:
+        """Drop the replay copy once the broker has acknowledged the message."""
+        with self._retained_lock:
+            self._inflight_retained.pop(mid, None)
+
+    def _wait_for_publish(self, message_info: mqtt.MQTTMessageInfo) -> None:
+        """Pump the loop until a shutdown-critical publish is acknowledged."""
+        deadline = time.monotonic() + MQTT_SHUTDOWN_FLUSH_TIMEOUT
+
+        while not message_info.is_published() and time.monotonic() < deadline:
+            if self.client is None:
+                return
+
+            try:
+                result = self.client.loop(timeout=MQTT_PUBLISH_WAIT_INTERVAL)
+            except (OSError, mqtt.WebsocketConnectionError) as err:
+                logger.warning("MQTT publish wait failed: %s", err)
+                self._schedule_reconnect()
+                return
+
+            self._drain_callback_queue()
+
+            if result != mqtt.MQTT_ERR_SUCCESS:
+                logger.error(
+                    "MQTT loop returned error code while waiting for publish: %s",
+                    result,
+                )
+                self._schedule_reconnect()
+                return
+
+    def _queue_retained(
+        self,
+        topic: str,
+        payload: Any,
+        retain: bool,
+        overwrite: bool = True,
+    ) -> None:
+        """Store the last retained value per topic for replay after reconnect."""
+        with self._retained_lock:
+            if overwrite or topic not in self._pending_retained:
+                self._pending_retained[topic] = (payload, retain)
+
+    def _handle_connect_event(self, reason_code: mqtt.ReasonCode) -> None:  # type: ignore[name-defined]
+        """Begin a new session by subscribing before any replay is published."""
+        if self.client is None:
+            return
+
+        self.connected = True
+        self._subscription_ready = False
+        self._subscription_mid = None
+        logger.debug("MQTT connected")
+
+        try:
+            result, mid = self.client.subscribe(
+                f"{self.mqtt_config.topic_prefix}/#",
+                qos=self.config.mqtt.qos,
+            )
+        except (OSError, mqtt.WebsocketConnectionError) as err:
+            logger.warning("MQTT subscribe failed: %s", err)
+            self._schedule_reconnect()
+            return
+
+        if result != mqtt.MQTT_ERR_SUCCESS:
+            logger.error(
+                "Unable to subscribe to MQTT command tree: mqtt error %s", result
+            )
+            self._schedule_reconnect()
+            return
+
+        self._subscription_mid = mid
+
+    def _handle_connect_failure(self, reason_code: mqtt.ReasonCode) -> None:  # type: ignore[name-defined]
+        """Record a failed connect attempt and transition into reconnect state."""
+        self.connected = False
+        logger.error(
+            "Unable to connect to MQTT server: %s", self._reason_code_name(reason_code)
+        )
+        self._schedule_reconnect()
+
+    def _handle_disconnect_event(self, reason_code: mqtt.ReasonCode) -> None:  # type: ignore[name-defined]
+        """Handle broker disconnects idempotently from the worker thread."""
+        if not self.connected:
+            return
+
+        self.connected = False
+        self._subscription_ready = False
+        self._subscription_mid = None
+
+        if self._stop_event.is_set():
+            logger.debug("MQTT disconnected")
+            self._cleanup_client()
+            return
+
+        logger.error("MQTT disconnected: %s", self._reason_code_name(reason_code))
+        self._schedule_reconnect()
+
+    def _handle_subscribe_event(
+        self,
+        mid: int,
+        reason_codes: list[mqtt.ReasonCode],  # type: ignore[name-defined]
+    ) -> None:
+        """Mark the session ready after SUBACK, then replay retained/runtime state."""
+        if mid != self._subscription_mid:
+            return
+
+        if any(
+            getattr(reason_code, "is_failure", False) for reason_code in reason_codes
+        ):
+            logger.error("MQTT subscription was rejected by the broker")
+            self._schedule_reconnect()
+            return
+
+        self._subscription_ready = True
+        self._subscription_mid = None
+
+        # a bug in replay should cost a snapshot, not the MQTT session
+        try:
+            self._publish_retained_state()
+
+            if self._command_router is not None:
+                self._command_router.publish_runtime_snapshot(self.publish)
+        except Exception:
+            logger.exception("Error replaying MQTT state after subscribe")
+
+    def _handle_inbound_message(self, topic: str, payload: str) -> None:
+        """Forward supported command topics into Dispatcher semantics."""
+        if self._dispatcher is None:
+            return
+
+        if not self._is_supported_command_topic(topic):
+            return
+
+        if topic == "onConnect":
+            now = time.monotonic()
+            if now - self._last_on_connect_dispatch < MQTT_ON_CONNECT_RATE_LIMIT:
+                logger.debug("Skipping MQTT onConnect replay request due to rate limit")
+                return
+            self._last_on_connect_dispatch = now
+
+        # a raise here used to end the network thread and take MQTT down
+        try:
+            self._dispatcher(topic, payload)
+        except Exception:
+            logger.exception("Error handling MQTT command topic %s", topic)
+
+    def _is_supported_command_topic(self, topic: str) -> bool:
+        """Filter the wildcard subscription down to Dispatcher's command surface.
+
+        Load-bearing rather than an optimization: the broker echoes Frigate's own
+        publishes back through frigate/#, and Dispatcher republishes topics it
+        does not recognize, so forwarding unfiltered would loop.
+        """
+        if self._command_router is None:
+            return False
+
+        # mirrors the gate on the state topic in _publish_retained_state()
+        if topic == "notifications/set" and not self._notifications_enabled_in_config():
+            return False
+
+        return self._command_router.is_command_topic(topic)
+
+    def _strip_topic_prefix(self, topic: str) -> str:
+        return topic.replace(f"{self.mqtt_config.topic_prefix}/", "", 1)
+
+    def _is_success_reason_code(self, reason_code: mqtt.ReasonCode) -> bool:  # type: ignore[name-defined]
+        if hasattr(reason_code, "is_failure"):
+            return not bool(reason_code.is_failure)
+
+        return bool(reason_code == 0)
+
+    def _reason_code_name(self, reason_code: mqtt.ReasonCode) -> str:  # type: ignore[name-defined]
+        if hasattr(reason_code, "getName"):
+            return str(reason_code.getName())
+
+        return str(reason_code)
 
     def _on_connect(
         self,
@@ -205,29 +787,11 @@ class MqttClient(Communicator):
         reason_code: mqtt.ReasonCode,  # type: ignore[name-defined]
         properties: Any,
     ) -> None:
-        """Mqtt connection callback."""
+        """Handle broker connect notifications from paho."""
-        threading.current_thread().name = "mqtt"
+        if self._is_success_reason_code(reason_code):
-        if reason_code != 0:
+            self._callback_queue.put(("connect", reason_code))
-            if reason_code == "Server unavailable":
+        else:
-                logger.error(
+            self._callback_queue.put(("connect_failure", reason_code))
-                    "Unable to connect to MQTT server: MQTT Server unavailable"
-                )
-            elif reason_code == "Bad user name or password":
-                logger.error(
-                    "Unable to connect to MQTT server: MQTT Bad username or password"
-                )
-            elif reason_code == "Not authorized":
-                logger.error("Unable to connect to MQTT server: MQTT Not authorized")
-            else:
-                logger.error(
-                    "Unable to connect to MQTT server: Connection refused. Error code: %s",
-                    reason_code.getName(),
-                )
-
-        self.connected = True
-        logger.debug("MQTT connected")
-        client.subscribe(f"{self.mqtt_config.topic_prefix}/#", qos=self.config.mqtt.qos)
-        self._set_initial_topics()
 
     def _on_disconnect(
         self,
@@ -237,126 +801,63 @@ class MqttClient(Communicator):
         reason_code: mqtt.ReasonCode,  # type: ignore[name-defined]
         properties: Any,
     ) -> None:
-        """Mqtt disconnection callback."""
+        """Handle broker disconnect notifications from paho."""
-        self.connected = False
+        self._callback_queue.put(("disconnect", reason_code))
-        logger.error("MQTT disconnected")
 
-    def _start(self) -> None:
+    def _on_subscribe(
-        """Start mqtt client."""
+        self,
-        self.client = mqtt.Client(
+        client: mqtt.Client,
-            callback_api_version=CallbackAPIVersion.VERSION2,
+        userdata: Any,
-            client_id=self.mqtt_config.client_id,
+        mid: int,
-        )
+        reason_codes: list[mqtt.ReasonCode],  # type: ignore[name-defined]
-        self.client.on_connect = self._on_connect
+        properties: Any,
-        self.client.on_disconnect = self._on_disconnect
+    ) -> None:
-        self.client.will_set(
+        """Handle subscribe acknowledgements from paho."""
-            self.mqtt_config.topic_prefix + "/available",
+        self._callback_queue.put(("subscribed", mid, reason_codes))
-            payload="offline",
-            qos=1,
-            retain=True,
-        )
 
-        # register callbacks
+    def _on_publish(
-        callback_types = [
+        self,
-            "enabled",
+        client: mqtt.Client,
-            "recordings",
+        userdata: Any,
-            "snapshots",
+        mid: int,
-            "detect",
+        reason_code: mqtt.ReasonCode,  # type: ignore[name-defined]
-            "audio",
+        properties: Any,
-            "audio_transcription",
+    ) -> None:
-            "motion",
+        """Handle publish acknowledgements from paho.
-            "improve_contrast",
-            "ptz_autotracker",
-            "motion_threshold",
-            "motion_contour_area",
-            "birdseye",
-            "birdseye_modes",
-            "review_alerts",
-            "review_detections",
-            "object_descriptions",
-            "review_descriptions",
-            "notifications",
-        ]
 
-        for name in self.config.cameras.keys():
+        Only tracked retained messages need an event. At the default qos 0
-            for callback in callback_types:
+        nothing is tracked, so this stays off the hot publish path.
-                self.client.message_callback_add(
+        """
-                    f"{self.mqtt_config.topic_prefix}/{name}/{callback}/set",
+        with self._retained_lock:
-                    self.on_mqtt_command,
+            if mid not in self._inflight_retained:
-                )
+                return
 
-            # notifications suspend doesn't follow the /set topic pattern
+        self._callback_queue.put(("published", mid))
-            self.client.message_callback_add(
-                f"{self.mqtt_config.topic_prefix}/{name}/notifications/suspend",
-                self.on_mqtt_command,
-            )
 
-            if self.config.cameras[name].onvif.host:
+    def _on_message(
-                self.client.message_callback_add(
+        self,
-                    f"{self.mqtt_config.topic_prefix}/{name}/ptz",
+        client: mqtt.Client,
-                    self.on_mqtt_command,
+        userdata: Any,
-                )
+        message: mqtt.MQTTMessage,
+    ) -> None:
+        """Queue inbound MQTT messages for processing in the worker loop."""
+        topic = self._strip_topic_prefix(message.topic)
 
-            for mask_name in self.config.cameras[name].motion.mask.keys():
+        # Ignore everything outside Frigate's command surface before decoding or
-                self.client.message_callback_add(
+        # dispatching into the rest of the app.
-                    f"{self.mqtt_config.topic_prefix}/{name}/motion_mask/{mask_name}/set",
+        if not self._is_supported_command_topic(topic):
-                    self.on_mqtt_command,
-                )
-
-            for mask_name in self.config.cameras[name].objects.mask.keys():
-                self.client.message_callback_add(
-                    f"{self.mqtt_config.topic_prefix}/{name}/object_mask/{mask_name}/set",
-                    self.on_mqtt_command,
-                )
-
-            for zone_name in self.config.cameras[name].zones.keys():
-                self.client.message_callback_add(
-                    f"{self.mqtt_config.topic_prefix}/{name}/zone/{zone_name}/set",
-                    self.on_mqtt_command,
-                )
-
-        if self._notifications_enabled_in_config():
-            self.client.message_callback_add(
-                f"{self.mqtt_config.topic_prefix}/notifications/set",
-                self.on_mqtt_command,
-            )
-
-        self.client.message_callback_add(
-            f"{self.mqtt_config.topic_prefix}/profile/set",
-            self.on_mqtt_command,
-        )
-
-        self.client.message_callback_add(
-            f"{self.mqtt_config.topic_prefix}/onConnect", self.on_mqtt_command
-        )
-
-        self.client.message_callback_add(
-            f"{self.mqtt_config.topic_prefix}/restart", self.on_mqtt_command
-        )
-
-        if self.mqtt_config.tls_ca_certs is not None:
-            if (
-                self.mqtt_config.tls_client_cert is not None
-                and self.mqtt_config.tls_client_key is not None
-            ):
-                self.client.tls_set(
-                    self.mqtt_config.tls_ca_certs,
-                    self.mqtt_config.tls_client_cert,
-                    self.mqtt_config.tls_client_key,
-                )
-            else:
-                self.client.tls_set(self.mqtt_config.tls_ca_certs)
-        if self.mqtt_config.tls_insecure is not None:
-            self.client.tls_insecure_set(self.mqtt_config.tls_insecure)
-        if self.mqtt_config.user is not None:
-            self.client.username_pw_set(
-                self.mqtt_config.user, password=self.mqtt_config.password
-            )
-        try:
-            # https://stackoverflow.com/a/55390477
-            # with connect_async, retries are handled automatically
-            self.client.connect_async(self.mqtt_config.host, self.mqtt_config.port, 60)
-            self.client.loop_start()
-        except Exception as e:
-            logger.error(f"Unable to connect to MQTT server: {e}")
             return
+
+        try:
+            payload = message.payload.decode()
+        except UnicodeDecodeError:
+            logger.debug("Ignoring non-UTF-8 MQTT payload for topic %s", topic)
+            return
+
+        self._callback_queue.put(
+            (
+                "message",
+                topic,
+                payload,
+            )
+        )

frigate/comms/webpush.py

@@ -63,14 +63,8 @@ class WebPushClient(Communicator):
         self.last_notification_time: float = 0
         self.user_cameras: dict[str, set[str]] = {}
         self.notification_queue: queue.Queue[PushNotification] = queue.Queue()
-        self.notification_thread = threading.Thread(
+        self.notification_thread: threading.Thread | None = None
-            target=self._process_notifications, daemon=True
+        self.suspension_thread: threading.Thread | None = None
-        )
-        self.notification_thread.start()
-        self.suspension_thread = threading.Thread(
-            target=self._process_suspensions, daemon=True
-        )
-        self.suspension_thread.start()
 
         if not self.config.notifications.email:
             logger.warning("Email must be provided for push notifications to be sent.")
@@ -99,6 +93,16 @@ class WebPushClient(Communicator):
         """Wrapper for allowing dispatcher to subscribe."""
         pass
 
+    def start(self) -> None:
+        self.notification_thread = threading.Thread(
+            target=self._process_notifications, daemon=True
+        )
+        self.notification_thread.start()
+        self.suspension_thread = threading.Thread(
+            target=self._process_suspensions, daemon=True
+        )
+        self.suspension_thread.start()
+
     def check_registrations(self) -> None:
         # check for valid claim or create new one
         now = datetime.datetime.now().timestamp()
@@ -608,4 +612,5 @@ class WebPushClient(Communicator):
 
     def stop(self) -> None:
         logger.info("Closing notification queue")
-        self.notification_thread.join()
+        if self.notification_thread is not None:
+            self.notification_thread.join()

frigate/comms/ws.py

@@ -466,7 +466,6 @@ class WebSocketClient(Communicator):
 
     def subscribe(self, receiver: Callable) -> None:
         self._dispatcher = receiver
-        self.start()
 
     def start(self) -> None:
         """Start the websocket client."""

frigate/test/test_mqtt_lifecycle.py

@@ -0,0 +1,907 @@
+import os
+import threading
+import unittest
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import paho.mqtt.client as mqtt
+
+from frigate.comms.dispatcher import Dispatcher
+from frigate.comms.mqtt import MqttClient, QueuedPublish
+from frigate.config import FrigateConfig
+from frigate.const import MODEL_CACHE_DIR
+
+
+class RuntimeSnapshotReceiver:
+    def __init__(self) -> None:
+        self.messages: list[tuple[str, str]] = []
+
+    def _receive(self, topic: str, payload: str) -> None:
+        self.messages.append((topic, payload))
+
+
+class FakeMessage:
+    def __init__(self, topic: str, payload: bytes) -> None:
+        self.topic = topic
+        self.payload = payload
+
+
+class FakeCommunicator:
+    def __init__(self) -> None:
+        self.receiver = None
+        self.dispatcher = None
+        self.started = False
+
+    def subscribe(self, receiver) -> None:
+        self.receiver = receiver
+
+    def attach_dispatcher(self, dispatcher) -> None:
+        self.dispatcher = dispatcher
+
+    def start(self) -> None:
+        self.started = True
+
+    def publish(self, topic: str, payload: Any, retain: bool = False) -> None:
+        return None
+
+    def stop(self) -> None:
+        return None
+
+
+class FakeSnapshotCommunicator(FakeCommunicator):
+    def start(self) -> None:
+        self.started = True
+        assert self.dispatcher is not None
+        self.dispatcher.publish_runtime_snapshot(lambda *_args, **_kwargs: None)
+
+
+def build_config() -> FrigateConfig:
+    config = {
+        "mqtt": {
+            "host": "mqtt",
+            "client_id": "frigate-test",
+            "topic_prefix": "frigate",
+        },
+        "notifications": {"enabled": True},
+        "cameras": {
+            "front": {
+                "ffmpeg": {
+                    "inputs": [
+                        {
+                            "path": "rtsp://10.0.0.1:554/video",
+                            "roles": ["detect", "audio"],
+                        }
+                    ]
+                },
+                "detect": {
+                    "height": 1080,
+                    "width": 1920,
+                    "fps": 5,
+                },
+                "audio": {"enabled": True},
+                "notifications": {"enabled": False},
+                "onvif": {"host": "10.0.0.5"},
+                "motion": {
+                    "mask": {
+                        "motion_mask_1": {
+                            "coordinates": "0,0,1,0,1,1,0,1",
+                        }
+                    }
+                },
+                "objects": {
+                    "track": ["person"],
+                    "mask": {
+                        "object_mask_1": {
+                            "coordinates": "0,0,1,0,1,1,0,1",
+                        }
+                    },
+                },
+                "zones": {
+                    "driveway": {
+                        "coordinates": "0.1,0.1,0.9,0.1,0.9,0.9,0.1,0.9",
+                        "objects": ["person"],
+                    }
+                },
+            }
+        },
+    }
+    return FrigateConfig(**config)
+
+
+def build_dispatcher(config: FrigateConfig, communicators: list[Any]) -> Dispatcher:
+    """Build a real Dispatcher with only the activity managers stubbed out."""
+    with (
+        patch("frigate.comms.dispatcher.CameraActivityManager") as mock_camera_activity,
+        patch("frigate.comms.dispatcher.AudioActivityManager") as mock_audio_activity,
+    ):
+        mock_camera_activity.return_value.last_camera_activity = {}
+        mock_audio_activity.return_value.current_audio_detections = {}
+
+        return Dispatcher(config, MagicMock(), MagicMock(), {}, communicators)
+
+
+class TestMqttClientLifecycle(unittest.TestCase):
+    def setUp(self) -> None:
+        if not os.path.exists(MODEL_CACHE_DIR) and not os.path.islink(MODEL_CACHE_DIR):
+            os.makedirs(MODEL_CACHE_DIR)
+
+        self.config = build_config()
+        self.client = MqttClient(self.config)
+        self.receiver = RuntimeSnapshotReceiver()
+        self.client.attach_dispatcher(build_dispatcher(self.config, []))
+
+    def test_subscribe_stores_receiver_without_starting_worker(self) -> None:
+        client = MqttClient(self.config)
+
+        with patch.object(client, "_start_worker") as mock_start_worker:
+            client.subscribe(self.receiver._receive)
+
+        self.assertIsNotNone(client._dispatcher)
+        self.assertIs(client._dispatcher.__self__, self.receiver)
+        self.assertIs(client._dispatcher.__func__, RuntimeSnapshotReceiver._receive)
+        mock_start_worker.assert_not_called()
+
+    def test_attach_dispatcher_supplies_command_surface(self) -> None:
+        client = MqttClient(self.config)
+
+        self.assertFalse(client._is_supported_command_topic("front/detect/set"))
+
+        dispatcher = build_dispatcher(self.config, [])
+        client.attach_dispatcher(dispatcher)
+
+        self.assertIs(client._command_router, dispatcher)
+        self.assertTrue(client._is_supported_command_topic("front/detect/set"))
+
+    def test_start_starts_worker_after_receiver_registration(self) -> None:
+        self.client.subscribe(self.receiver._receive)
+
+        with patch.object(self.client, "_start_worker") as mock_start_worker:
+            self.client.start()
+
+        mock_start_worker.assert_called_once()
+
+    def test_dispatcher_initializes_state_before_starting_communicators(self) -> None:
+        fake_comm = FakeSnapshotCommunicator()
+        dispatcher = build_dispatcher(self.config, [fake_comm])
+
+        self.assertIsNone(dispatcher.web_push_client)
+        self.assertIs(fake_comm.receiver.__self__, dispatcher)
+        self.assertIs(fake_comm.dispatcher, dispatcher)
+        self.assertFalse(fake_comm.started)
+
+        dispatcher.start_communicators()
+
+        self.assertTrue(fake_comm.started)
+
+    def test_publish_drops_ephemeral_and_coalesces_retained_while_disconnected(
+        self,
+    ) -> None:
+        self.client.publish("front/events", "payload")
+        self.assertTrue(self.client._publish_queue.empty())
+
+        self.client.publish("profile/state", "armed", retain=True)
+        self.client.publish("profile/state", "disarmed", retain=True)
+
+        self.assertEqual(
+            self.client._pending_retained["frigate/profile/state"],
+            ("disarmed", True),
+        )
+
+    def test_outage_buffer_replays_before_newer_queued_value(self) -> None:
+        """A topic that changed since the reconnect must not be reverted by the
+        replay of the value buffered during the outage."""
+        mock_client = MagicMock()
+        mock_client.publish.return_value = MagicMock(
+            rc=mqtt.MQTT_ERR_SUCCESS, mid=1, **{"is_published.return_value": True}
+        )
+        self.client._pending_retained = {"frigate/front/detect/state": ("OFF", True)}
+        self.client.client = mock_client
+        self.client.connected = True
+        self.client._subscription_ready = True
+        self.client._publish_queue.put(
+            QueuedPublish("frigate/front/detect/state", "ON", True)
+        )
+
+        self.client._drain_publish_queue()
+
+        published = [
+            (call.args[0], call.args[1]) for call in mock_client.publish.call_args_list
+        ]
+        self.assertEqual(
+            published,
+            [
+                ("frigate/front/detect/state", "OFF"),
+                ("frigate/front/detect/state", "ON"),
+            ],
+        )
+
+    def test_flush_failure_does_not_clobber_newer_queued_value(self) -> None:
+        """If the session dies mid-replay, the newer queued value that gets
+        rebuffered has to win over the older entries still being replayed."""
+        self.client._pending_retained = {
+            "frigate/a/detect/state": ("OLD", True),
+            "frigate/front/detect/state": ("OFF", True),
+        }
+        self.client.connected = True
+        self.client._subscription_ready = True
+        self.client._publish_queue.put(
+            QueuedPublish("frigate/front/detect/state", "ON", True)
+        )
+
+        mock_client = MagicMock()
+        mock_client.publish.side_effect = BrokenPipeError("broken pipe")
+        self.client.client = mock_client
+
+        self.client._drain_publish_queue()
+
+        self.assertEqual(
+            self.client._pending_retained["frigate/front/detect/state"], ("ON", True)
+        )
+
+    def test_publish_buffers_messages_until_subscription_ready(self) -> None:
+        self.client.connected = True
+
+        self.client.publish("front/events", "payload")
+        self.client.publish("profile/state", "armed", retain=True)
+
+        self.assertEqual(self.client._publish_queue.qsize(), 2)
+        self.assertEqual(self.client._pending_retained, {})
+
+    @patch("frigate.comms.mqtt.mqtt.Client")
+    def test_connect_client_initializes_manual_loop_client(
+        self, mock_client_cls
+    ) -> None:
+        mock_client = MagicMock()
+        mock_client_cls.return_value = mock_client
+
+        connected = self.client._connect_client()
+
+        self.assertTrue(connected)
+        mock_client_cls.assert_called_once_with(
+            callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
+            client_id="frigate-test",
+            reconnect_on_failure=False,
+        )
+        self.assertIs(mock_client.on_connect.__self__, self.client)
+        self.assertIs(mock_client.on_connect.__func__, MqttClient._on_connect)
+        self.assertIs(mock_client.on_disconnect.__self__, self.client)
+        self.assertIs(mock_client.on_disconnect.__func__, MqttClient._on_disconnect)
+        self.assertIs(mock_client.on_message.__self__, self.client)
+        self.assertIs(mock_client.on_message.__func__, MqttClient._on_message)
+        self.assertIs(mock_client.on_subscribe.__self__, self.client)
+        self.assertIs(mock_client.on_subscribe.__func__, MqttClient._on_subscribe)
+        mock_client.connect.assert_called_once_with("mqtt", 1883, 60)
+
+    def test_handle_connect_failure_leaves_client_disconnected(self) -> None:
+        self.client.client = MagicMock()
+        reason_code = MagicMock()
+        reason_code.getName.return_value = "Not authorized"
+
+        with patch.object(
+            self.client, "_schedule_reconnect"
+        ) as mock_schedule_reconnect:
+            self.client._handle_connect_failure(reason_code)
+
+        self.assertFalse(self.client.connected)
+        mock_schedule_reconnect.assert_called_once()
+
+    def test_handle_connect_event_subscribes_wildcard_topic(self) -> None:
+        self.client.client = MagicMock()
+        self.client.client.subscribe.return_value = (mqtt.MQTT_ERR_SUCCESS, 42)
+
+        self.client._handle_connect_event(MagicMock())
+
+        self.assertTrue(self.client.connected)
+        self.assertEqual(self.client._subscription_mid, 42)
+        self.client.client.subscribe.assert_called_once_with("frigate/#", qos=0)
+
+    def test_handle_connect_event_reconnects_on_recoverable_subscribe_error(
+        self,
+    ) -> None:
+        self.client.client = MagicMock()
+        self.client.client.subscribe.side_effect = BrokenPipeError("broken pipe")
+
+        with patch.object(
+            self.client, "_schedule_reconnect"
+        ) as mock_schedule_reconnect:
+            self.client._handle_connect_event(MagicMock())
+
+        mock_schedule_reconnect.assert_called_once()
+
+    def test_handle_subscribe_event_publishes_snapshots_after_matching_suback(
+        self,
+    ) -> None:
+        self.client.connected = True
+        self.client._subscription_mid = 7
+
+        with (
+            patch.object(self.client, "_publish_retained_state") as mock_retained,
+            patch.object(
+                self.client._command_router, "publish_runtime_snapshot"
+            ) as mock_snapshot,
+        ):
+            self.client._handle_subscribe_event(7, [MagicMock(is_failure=False)])
+
+        self.assertTrue(self.client._subscription_ready)
+        mock_retained.assert_called_once()
+        mock_snapshot.assert_called_once_with(self.client.publish)
+
+    def test_handle_subscribe_event_ignores_other_suback_mid(self) -> None:
+        self.client.connected = True
+        self.client._subscription_mid = 8
+
+        with (
+            patch.object(self.client, "_publish_retained_state") as mock_retained,
+            patch.object(
+                self.client._command_router, "publish_runtime_snapshot"
+            ) as mock_snapshot,
+        ):
+            self.client._handle_subscribe_event(9, [MagicMock(is_failure=False)])
+
+        self.assertFalse(self.client._subscription_ready)
+        mock_retained.assert_not_called()
+        mock_snapshot.assert_not_called()
+
+    def test_on_message_strips_prefix_and_dispatches_supported_topic(self) -> None:
+        dispatcher = MagicMock()
+        self.client._dispatcher = dispatcher
+
+        self.client._on_message(
+            MagicMock(),
+            None,
+            FakeMessage("frigate/front/detect/set", b"ON"),
+        )
+        self.client._drain_callback_queue()
+
+        dispatcher.assert_called_once_with("front/detect/set", "ON")
+
+    def test_on_message_ignores_unsupported_state_topic(self) -> None:
+        dispatcher = MagicMock()
+        self.client._dispatcher = dispatcher
+
+        self.client._on_message(
+            MagicMock(),
+            None,
+            FakeMessage("frigate/front/detect/state", b"ON"),
+        )
+        self.client._drain_callback_queue()
+
+        dispatcher.assert_not_called()
+
+    def test_on_message_ignores_non_utf8_payloads(self) -> None:
+        self.client._on_message(
+            MagicMock(),
+            None,
+            FakeMessage("frigate/onConnect", b"\xff"),
+        )
+
+        self.assertTrue(self.client._callback_queue.empty())
+
+    def test_on_connect_requests_are_rate_limited(self) -> None:
+        dispatcher = MagicMock()
+        self.client._dispatcher = dispatcher
+
+        with patch("frigate.comms.mqtt.time.monotonic", side_effect=[100.0, 100.1]):
+            self.client._handle_inbound_message("onConnect", "")
+            self.client._handle_inbound_message("onConnect", "")
+
+        dispatcher.assert_called_once_with("onConnect", "")
+
+    def test_supported_command_topics_preserve_command_surface(self) -> None:
+        for topic in (
+            "front/detect/set",
+            "front/audio_transcription/set",
+            "front/notifications/set",
+            "front/notifications/suspend",
+            "front/zone/driveway/set",
+            "front/motion_mask/motion_mask_1/set",
+            "front/ptz",
+            "notifications/set",
+            "profile/set",
+            "onConnect",
+            "restart",
+        ):
+            with self.subTest(topic=topic):
+                self.assertTrue(self.client._is_supported_command_topic(topic))
+
+        for topic in (
+            # Frigate's own publishes echoing back through the wildcard
+            "front/detect/state",
+            "available",
+            "front/notifications/suspended",
+            "front/zone/set",
+            "front/nonsense/set",
+            "nonsense/set",
+        ):
+            with self.subTest(topic=topic):
+                self.assertFalse(self.client._is_supported_command_topic(topic))
+
+    def test_command_surface_tracks_dispatcher_handlers(self) -> None:
+        """The allowlist is derived, so a new handler is routable for free."""
+        router = self.client._command_router
+        router._camera_settings_handlers["brand_new_toggle"] = MagicMock()
+
+        self.assertTrue(
+            self.client._is_supported_command_topic("front/brand_new_toggle/set")
+        )
+
+    def test_global_notifications_set_follows_config_gate(self) -> None:
+        """The command gate must match the publish gate in
+        _publish_retained_state: with notifications unconfigured there is no
+        state topic, so the command must not flip runtime state either."""
+        self.assertTrue(self.client._notifications_enabled_in_config())
+        self.assertTrue(self.client._is_supported_command_topic("notifications/set"))
+
+        with patch.object(
+            self.client, "_notifications_enabled_in_config", return_value=False
+        ):
+            self.assertFalse(
+                self.client._is_supported_command_topic("notifications/set")
+            )
+            # per-camera topics stay routable regardless of the global gate
+            self.assertTrue(
+                self.client._is_supported_command_topic("front/notifications/set")
+            )
+            self.assertTrue(
+                self.client._is_supported_command_topic("front/notifications/suspend")
+            )
+
+    def test_publish_direct_waits_for_flush_barrier(self) -> None:
+        mock_client = MagicMock()
+        mock_client.loop.return_value = mqtt.MQTT_ERR_SUCCESS
+        self.client.client = mock_client
+        message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS, mid=1)
+        # inflight tracking checks once, then _wait_for_publish polls
+        message_info.is_published.side_effect = [False, False, True]
+        mock_client.publish.return_value = message_info
+        barrier = MagicMock()
+
+        self.client._publish_direct(
+            QueuedPublish("frigate/available", "stopped", True, barrier)
+        )
+
+        mock_client.loop.assert_called_once()
+        barrier.set.assert_called_once()
+
+    def test_shutdown_barrier_releases_when_publish_raises(self) -> None:
+        """stop() waits on this barrier, so a broker error must not stall
+        shutdown for the full flush timeout."""
+        self.client.client = MagicMock()
+        self.client.client.publish.side_effect = BrokenPipeError("broken pipe")
+        barrier = threading.Event()
+
+        with patch.object(self.client, "_schedule_reconnect"):
+            self.client._publish_direct(
+                QueuedPublish("frigate/available", "stopped", True, barrier)
+            )
+
+        self.assertTrue(barrier.is_set())
+        self.assertEqual(
+            self.client._pending_retained["frigate/available"], ("stopped", True)
+        )
+
+    def test_shutdown_barrier_releases_on_publish_error_code(self) -> None:
+        self.client.client = MagicMock()
+        self.client.client.publish.return_value = MagicMock(rc=mqtt.MQTT_ERR_NO_CONN)
+        barrier = threading.Event()
+
+        with patch.object(self.client, "_schedule_reconnect"):
+            self.client._publish_direct(
+                QueuedPublish("frigate/available", "stopped", True, barrier)
+            )
+
+        self.assertTrue(barrier.is_set())
+
+    def test_shutdown_barrier_releases_when_requeued_while_disconnected(self) -> None:
+        barrier = threading.Event()
+        self.client._publish_queue.put(
+            QueuedPublish("frigate/available", "stopped", True, barrier)
+        )
+
+        self.client._requeue_disconnected_publishes()
+
+        self.assertTrue(barrier.is_set())
+        self.assertEqual(
+            self.client._pending_retained["frigate/available"], ("stopped", True)
+        )
+
+    def test_shutdown_barrier_releases_when_worker_crashes(self) -> None:
+        """stop() can queue the final publish just as the worker dies, and
+        nothing drains the queue after that."""
+        barrier = threading.Event()
+        self.client._publish_queue.put(
+            QueuedPublish("frigate/available", "stopped", True, barrier)
+        )
+
+        with patch.object(
+            self.client,
+            "_mqtt_loop_worker",
+            side_effect=RuntimeError("unexpected bug"),
+        ):
+            self.client._worker_main()
+
+        self.assertTrue(barrier.is_set())
+
+    def test_shutdown_barrier_releases_when_publish_raises_unexpectedly(self) -> None:
+        """The message is off the queue by the time this runs, so crash cleanup
+        cannot recover it and only _publish_direct can release the waiter."""
+        self.client.client = MagicMock()
+        self.client.client.publish.side_effect = RuntimeError("unexpected bug")
+        barrier = threading.Event()
+
+        with self.assertRaises(RuntimeError):
+            self.client._publish_direct(
+                QueuedPublish("frigate/available", "stopped", True, barrier)
+            )
+
+        self.assertTrue(barrier.is_set())
+
+    def test_newest_inflight_retained_value_wins(self) -> None:
+        """Several updates to one topic can be unacked at once above qos 0, and
+        the newest is the one subscribers should end up with."""
+        mock_client = MagicMock()
+        self.client.client = mock_client
+        self.client.connected = True
+
+        for mid, payload in ((1, "ON"), (2, "OFF")):
+            message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS, mid=mid)
+            message_info.is_published.return_value = False
+            mock_client.publish.return_value = message_info
+            self.client._publish_direct(
+                QueuedPublish("frigate/front/detect/state", payload, True)
+            )
+
+        self.client._requeue_inflight_retained()
+
+        self.assertEqual(
+            self.client._pending_retained["frigate/front/detect/state"], ("OFF", True)
+        )
+
+    def test_inflight_retained_does_not_clobber_queued_value(self) -> None:
+        """Anything still queued was written later than anything in flight."""
+        self.client._pending_retained = {"frigate/front/detect/state": ("OFF", True)}
+        self.client._inflight_retained = {1: ("frigate/front/detect/state", "ON")}
+
+        self.client._requeue_inflight_retained()
+
+        self.assertEqual(
+            self.client._pending_retained["frigate/front/detect/state"], ("OFF", True)
+        )
+
+    def test_unacked_retained_publish_survives_reconnect(self) -> None:
+        """Above qos 0 a successful rc only means paho queued the message, and
+        dropping the client drops its outbound queue with it."""
+        mock_client = MagicMock()
+        message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS, mid=12)
+        message_info.is_published.return_value = False
+        mock_client.publish.return_value = message_info
+        self.client.client = mock_client
+        self.client.connected = True
+
+        self.client._publish_direct(
+            QueuedPublish("frigate/profile/state", "armed", True)
+        )
+        self.assertEqual(
+            self.client._inflight_retained[12], ("frigate/profile/state", "armed")
+        )
+
+        self.client._cleanup_client()
+
+        self.assertEqual(self.client._inflight_retained, {})
+        self.assertEqual(
+            self.client._pending_retained["frigate/profile/state"], ("armed", True)
+        )
+
+    def test_acked_retained_publish_is_not_replayed(self) -> None:
+        mock_client = MagicMock()
+        message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS, mid=12)
+        message_info.is_published.return_value = False
+        mock_client.publish.return_value = message_info
+        self.client.client = mock_client
+        self.client.connected = True
+
+        self.client._publish_direct(
+            QueuedPublish("frigate/profile/state", "armed", True)
+        )
+        self.client._on_publish(mock_client, None, 12, MagicMock(), None)
+        self.client._drain_callback_queue()
+
+        self.assertEqual(self.client._inflight_retained, {})
+
+        self.client._cleanup_client()
+
+        self.assertEqual(self.client._pending_retained, {})
+
+    def test_already_published_retained_is_not_tracked(self) -> None:
+        """At the default qos 0 paho reports the message as published inline,
+        so there is nothing to wait on."""
+        mock_client = MagicMock()
+        message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS, mid=12)
+        message_info.is_published.return_value = True
+        mock_client.publish.return_value = message_info
+        self.client.client = mock_client
+        self.client.connected = True
+
+        self.client._publish_direct(
+            QueuedPublish("frigate/profile/state", "armed", True)
+        )
+
+        self.assertEqual(self.client._inflight_retained, {})
+
+    def test_wait_for_publish_survives_disconnect_during_wait(self) -> None:
+        mock_client = MagicMock()
+        self.client.client = mock_client
+        self.client.connected = True
+        message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS)
+        message_info.is_published.side_effect = [False, False]
+
+        loop_calls = [0]
+
+        def loop_side_effect(timeout: float) -> int:
+            if loop_calls[0] == 0:
+                self.client._callback_queue.put(("disconnect", 1))
+            loop_calls[0] += 1
+            return mqtt.MQTT_ERR_SUCCESS
+
+        mock_client.loop.side_effect = loop_side_effect
+
+        self.client._wait_for_publish(message_info)
+
+        self.assertEqual(loop_calls[0], 1)
+        self.assertIsNone(self.client.client)
+
+    def test_publish_direct_reconnects_on_recoverable_publish_error(self) -> None:
+        self.client.client = MagicMock()
+        self.client.client.publish.side_effect = BrokenPipeError("broken pipe")
+
+        with patch.object(
+            self.client, "_schedule_reconnect"
+        ) as mock_schedule_reconnect:
+            self.client._publish_direct(
+                QueuedPublish("frigate/profile/state", "armed", True)
+            )
+
+        self.assertEqual(
+            self.client._pending_retained["frigate/profile/state"],
+            ("armed", True),
+        )
+        mock_schedule_reconnect.assert_called_once()
+
+    def test_mqtt_loop_worker_reconnects_on_recoverable_loop_error(self) -> None:
+        self.client.client = MagicMock()
+        self.client.client.loop.side_effect = OSError("socket closed")
+
+        def stop_after_reconnect() -> None:
+            self.client._stop_event.set()
+
+        with patch.object(
+            self.client,
+            "_schedule_reconnect",
+            side_effect=stop_after_reconnect,
+        ) as mock_schedule_reconnect:
+            self.client._mqtt_loop_worker()
+
+        mock_schedule_reconnect.assert_called_once()
+
+    def test_worker_main_goes_dormant_on_unexpected_exception(self) -> None:
+        with patch.object(
+            self.client,
+            "_mqtt_loop_worker",
+            side_effect=RuntimeError("unexpected bug"),
+        ):
+            self.client._worker_main()
+
+        self.assertTrue(self.client._stop_event.is_set())
+        self.assertFalse(self.client.connected)
+        self.assertFalse(self.client._subscription_ready)
+        self.assertIsNone(self.client.client)
+
+    def test_worker_crash_announces_offline_before_disconnecting(self) -> None:
+        """The clean disconnect in cleanup suppresses the will, so a dormant
+        MQTT session must say so itself or consumers keep the stale values."""
+        mock_client = MagicMock()
+        mock_client.publish.return_value = MagicMock(
+            rc=mqtt.MQTT_ERR_SUCCESS, **{"is_published.return_value": True}
+        )
+        self.client.client = mock_client
+        self.client.connected = True
+
+        with patch.object(
+            self.client,
+            "_mqtt_loop_worker",
+            side_effect=RuntimeError("unexpected bug"),
+        ):
+            self.client._worker_main()
+
+        mock_client.publish.assert_called_once_with(
+            "frigate/available",
+            "offline",
+            qos=self.config.mqtt.qos,
+            retain=True,
+        )
+        # the announcement has to land before the socket is torn down
+        self.assertLess(
+            mock_client.method_calls.index(
+                next(c for c in mock_client.method_calls if c[0] == "publish")
+            ),
+            mock_client.method_calls.index(
+                next(c for c in mock_client.method_calls if c[0] == "disconnect")
+            ),
+        )
+
+    def test_worker_crash_stays_dormant_when_offline_publish_fails(self) -> None:
+        self.client.client = MagicMock()
+        self.client.client.publish.side_effect = BrokenPipeError("broken pipe")
+
+        with patch.object(
+            self.client,
+            "_mqtt_loop_worker",
+            side_effect=RuntimeError("unexpected bug"),
+        ):
+            self.client._worker_main()
+
+        self.assertTrue(self.client._stop_event.is_set())
+        self.assertIsNone(self.client.client)
+
+    def test_command_handler_exception_does_not_kill_worker(self) -> None:
+        """A raise in a dispatcher handler used to end the network thread and
+        take MQTT down until the next Frigate restart."""
+        self.client._dispatcher = MagicMock(side_effect=RuntimeError("handler bug"))
+        self.client._callback_queue.put(("message", "front/detect/set", "ON"))
+        self.client._callback_queue.put(("message", "front/motion/set", "ON"))
+
+        with self.assertLogs("frigate.comms.mqtt", level="ERROR"):
+            self.client._drain_callback_queue()
+
+        self.assertEqual(self.client._dispatcher.call_count, 2)
+
+    def test_snapshot_replay_exception_does_not_kill_worker(self) -> None:
+        self.client.connected = True
+        self.client._subscription_mid = 3
+
+        with (
+            patch.object(
+                self.client,
+                "_publish_retained_state",
+                side_effect=RuntimeError("replay bug"),
+            ),
+            self.assertLogs("frigate.comms.mqtt", level="ERROR"),
+        ):
+            self.client._handle_subscribe_event(3, [MagicMock(is_failure=False)])
+
+        self.assertTrue(self.client._subscription_ready)
+
+    def test_schedule_reconnect_drops_stale_ephemeral_and_preserves_retained(
+        self,
+    ) -> None:
+        mock_client = MagicMock()
+        message_info = MagicMock(rc=mqtt.MQTT_ERR_SUCCESS)
+        mock_client.publish.return_value = message_info
+        self.client.client = mock_client
+        self.client.connected = True
+        self.client._subscription_ready = True
+
+        self.client.publish("front/events", "ephemeral")
+        self.client.publish("profile/state", "armed", retain=True)
+
+        self.client._schedule_reconnect()
+
+        self.assertTrue(self.client._publish_queue.empty())
+        self.assertEqual(
+            self.client._pending_retained["frigate/profile/state"],
+            ("armed", True),
+        )
+
+        self.client.client = mock_client
+        self.client.connected = True
+        self.client._subscription_ready = True
+        self.client._drain_publish_queue()
+
+        mock_client.publish.assert_called_once_with(
+            "frigate/profile/state",
+            "armed",
+            qos=self.config.mqtt.qos,
+            retain=True,
+        )
+
+    def test_stop_disconnects_client_and_joins_worker(self) -> None:
+        worker = MagicMock()
+        worker.is_alive.return_value = True
+        self.client._worker = worker
+        mock_client = MagicMock()
+        self.client.client = mock_client
+
+        self.client.stop()
+
+        self.assertTrue(self.client._stop_event.is_set())
+        self.assertGreaterEqual(mock_client.disconnect.call_count, 1)
+        worker.join.assert_called_once()
+
+    def test_stop_skips_stopped_publish_until_subscription_ready(self) -> None:
+        worker = MagicMock()
+        worker.is_alive.return_value = True
+        self.client._worker = worker
+        self.client.client = MagicMock()
+        self.client.connected = True
+        self.client._subscription_ready = False
+
+        with patch.object(self.client._publish_queue, "put") as mock_queue_put:
+            self.client.stop()
+
+        mock_queue_put.assert_not_called()
+
+    def test_duplicate_disconnect_callback_is_safe(self) -> None:
+        self.client.connected = True
+
+        with patch.object(
+            self.client, "_schedule_reconnect"
+        ) as mock_schedule_reconnect:
+            self.client._handle_disconnect_event(MagicMock())
+            self.client._handle_disconnect_event(MagicMock())
+
+        mock_schedule_reconnect.assert_called_once()
+
+    def test_publish_retained_state_emits_expected_topic_families(self) -> None:
+        published_topics: list[tuple[str, Any, bool]] = []
+
+        with patch.object(
+            self.client,
+            "publish",
+            side_effect=lambda topic, payload, retain=False: published_topics.append(
+                (topic, payload, retain)
+            ),
+        ):
+            self.client._publish_retained_state()
+
+        topics = {topic for topic, _, _ in published_topics}
+        self.assertIn("front/enabled/state", topics)
+        self.assertIn("front/motion_mask/motion_mask_1/state", topics)
+        self.assertIn("front/object_mask/object_mask_1/state", topics)
+        self.assertIn("front/zone/driveway/state", topics)
+        self.assertIn("notifications/state", topics)
+        self.assertIn("profile/state", topics)
+        self.assertIn("available", topics)
+
+    def test_publish_retained_state_uses_runtime_values(self) -> None:
+        camera = self.config.cameras["front"]
+        camera.enabled_in_config = True
+        camera.enabled = False
+        camera.record.enabled_in_config = True
+        camera.record.enabled = False
+        camera.audio.enabled_in_config = True
+        camera.audio.enabled = False
+        camera.motion.enabled = False
+        camera.onvif.autotracking.enabled_in_config = True
+        camera.onvif.autotracking.enabled = False
+        camera.review.alerts.enabled_in_config = True
+        camera.review.alerts.enabled = False
+        camera.review.detections.enabled_in_config = True
+        camera.review.detections.enabled = False
+        camera.objects.genai.enabled_in_config = True
+        camera.objects.genai.enabled = False
+        camera.review.genai.enabled_in_config = True
+        camera.review.genai.enabled = False
+        self.config.notifications.enabled_in_config = True
+        self.config.notifications.enabled = False
+
+        published_states: dict[str, Any] = {}
+
+        with patch.object(
+            self.client,
+            "publish",
+            side_effect=lambda topic, payload, retain=False: (
+                published_states.__setitem__(topic, payload)
+            ),
+        ):
+            self.client._publish_retained_state()
+
+        self.assertEqual(published_states["front/enabled/state"], "OFF")
+        self.assertEqual(published_states["front/recordings/state"], "OFF")
+        self.assertEqual(published_states["front/audio/state"], "OFF")
+        self.assertEqual(published_states["front/motion/state"], "OFF")
+        self.assertEqual(published_states["front/ptz_autotracker/state"], "OFF")
+        self.assertEqual(published_states["front/review_alerts/state"], "OFF")
+        self.assertEqual(published_states["front/review_detections/state"], "OFF")
+        self.assertEqual(published_states["front/object_descriptions/state"], "OFF")
+        self.assertEqual(published_states["front/review_descriptions/state"], "OFF")
+        self.assertEqual(published_states["notifications/state"], "OFF")

frigate/test/test_mqtt_topic_registration.py

@@ -1,100 +0,0 @@
-"""Tests for MQTT command topic callback registration."""
-
-import unittest
-from unittest.mock import MagicMock, patch
-
-from frigate.comms.mqtt import MqttClient
-
-
-def _make_camera_mock(
-    *,
-    enabled: bool = True,
-    notifications_enabled_in_config: bool = False,
-) -> MagicMock:
-    """Build a camera config mock with the fields _start() reads."""
-    camera = MagicMock()
-    camera.enabled = enabled
-    camera.notifications.enabled_in_config = notifications_enabled_in_config
-    camera.onvif.host = None
-    camera.motion.mask = {}
-    camera.objects.mask = {}
-    camera.zones = {}
-    return camera
-
-
-def _registered_topics(
-    cameras: dict[str, MagicMock],
-    *,
-    global_notifications_enabled_in_config: bool = False,
-) -> set[str]:
-    """Start an MqttClient against a mocked paho client and collect the
-    topics registered via message_callback_add."""
-    config = MagicMock()
-    config.cameras = cameras
-    config.notifications.enabled_in_config = global_notifications_enabled_in_config
-    config.mqtt.topic_prefix = "frigate"
-    config.mqtt.client_id = "frigate"
-    config.mqtt.user = None
-    config.mqtt.tls_ca_certs = None
-    config.mqtt.tls_insecure = None
-
-    with patch("frigate.comms.mqtt.mqtt.Client") as client_cls:
-        mqtt_client = MqttClient(config)
-        mqtt_client.subscribe(MagicMock())
-
-    paho_client = client_cls.return_value
-    return {call.args[0] for call in paho_client.message_callback_add.call_args_list}
-
-
-class TestMqttTopicRegistration(unittest.TestCase):
-    def test_camera_notification_topics_registered(self):
-        """Per-camera notification set/suspend must be registered so paho
-        routes them to the dispatcher (unregistered topics drop silently)."""
-        topics = _registered_topics(
-            {"front_door": _make_camera_mock(notifications_enabled_in_config=True)}
-        )
-
-        self.assertIn("frigate/front_door/notifications/set", topics)
-        self.assertIn("frigate/front_door/notifications/suspend", topics)
-
-    def test_global_set_registered_with_camera_only_notifications(self):
-        """The global topic must work when notifications are enabled only at
-        the camera level, matching the WebPushClient gating in app.py."""
-        topics = _registered_topics(
-            {"front_door": _make_camera_mock(notifications_enabled_in_config=True)},
-            global_notifications_enabled_in_config=False,
-        )
-
-        self.assertIn("frigate/notifications/set", topics)
-
-    def test_global_set_registered_with_global_notifications(self):
-        topics = _registered_topics(
-            {"front_door": _make_camera_mock()},
-            global_notifications_enabled_in_config=True,
-        )
-
-        self.assertIn("frigate/notifications/set", topics)
-
-    def test_global_set_not_registered_when_notifications_unconfigured(self):
-        topics = _registered_topics(
-            {"front_door": _make_camera_mock()},
-            global_notifications_enabled_in_config=False,
-        )
-
-        self.assertNotIn("frigate/notifications/set", topics)
-
-    def test_disabled_camera_does_not_enable_global_set(self):
-        topics = _registered_topics(
-            {
-                "front_door": _make_camera_mock(
-                    enabled=False, notifications_enabled_in_config=True
-                )
-            },
-            global_notifications_enabled_in_config=False,
-        )
-
-        self.assertNotIn("frigate/notifications/set", topics)
-
-
-if __name__ == "__main__":
-    unittest.main()

frigate/test/test_profiles.py

@@ -738,6 +738,53 @@ class TestProfileManager(unittest.TestCase):
         assert "objects" in api_base
         assert api_base["objects"]["track"] == ["person"]
 
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_update_config_reapplies_active_profile(self, mock_persist):
+        """Replacing the config preserves the active profile overrides."""
+        self.manager.activate_profile("armed")
+
+        new_config = FrigateConfig(**self.config_data)
+        self.manager.update_config(new_config)
+
+        assert self.manager.config.active_profile == "armed"
+        assert self.manager.config.cameras["front"].notifications.enabled is True
+        assert self.manager.config.cameras["front"].objects.track == [
+            "person",
+            "car",
+            "package",
+        ]
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_update_config_deactivates_removed_profile(self, mock_persist):
+        """Replacing the config clears the active profile if it no longer exists."""
+        self.manager.activate_profile("armed")
+
+        new_config_data = json.loads(json.dumps(self.config_data))
+        del new_config_data["profiles"]["armed"]
+        del new_config_data["cameras"]["front"]["profiles"]["armed"]
+        del new_config_data["cameras"]["back"]["profiles"]["armed"]
+
+        new_config = FrigateConfig(**new_config_data)
+        self.manager.update_config(new_config)
+
+        assert self.manager.config.active_profile is None
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_enabled_state_is_published_via_dispatcher_when_profile_changes(
+        self, mock_persist
+    ):
+        """Enabled state updates are published through the dispatcher when present."""
+        self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away")
+        self.config.cameras["front"].profiles["away"] = CameraProfileConfig(
+            enabled=False
+        )
+        dispatcher = MagicMock()
+        self.manager = ProfileManager(self.config, self.mock_updater, dispatcher)
+
+        self.manager.activate_profile("away")
+
+        dispatcher.publish.assert_any_call("front/enabled/state", "OFF", retain=True)
+
     def test_base_configs_for_api_are_json_serializable(self):
         """API base configs are JSON-serializable (mode='json')."""
         import json