MQTT-FIX-T01: make main MQTT client lifecycle idempotent

.gitignore

@@ -6,3 +6,4 @@ __pycache__/
 *.pyc
 data/
 review-notes/
+development-notes/

README.md

@@ -345,6 +345,11 @@ CLI 工具为受控手工验证而设（设备需接市电），仅暴露读功
 - 每次轮询成功后推 state topic；连接成功或勾选变更时重发 retained discovery config
 - `POST /api/config/mqtt/test`：试连 broker 并发布一条测试消息（可用 MQTT Explorer 验证链路）
 
+主 MQTT 客户端由应用 lifespan 持有。相同配置下重复调用 connect 不会创建新客户端；Paho
+网络循环在短暂断线后负责自动重连，并在重连成功时恢复订阅。每分钟的 HA 发布任务会幂等地
+确保客户端存在。配置变更或应用关闭时，旧客户端先从管理器解绑，再断开并停止网络循环；旧
+客户端的迟到回调不会改变新连接状态或分发消息。DSMR source 使用各自独立的客户端生命周期。
+
 **Config 页配置流程**：
 1. 在 `/config` 的「MQTT」section 填写 broker host/port/username/password，启用 `MQTT_ENABLED`
 2. 点「发送测试消息」确认 broker 链路通

app/integrations/mqtt.py

@@ -94,6 +94,12 @@ class MqttManager:
     def __init__(self) -> None:
         self._client: mqtt.Client | None = None
         self._lock = threading.RLock()
+        # Serialize ownership changes without holding the callback lock while
+        # loop_stop() joins Paho's network thread.
+        self._lifecycle_lock = threading.Lock()
+        self._client_generation = 0
+        self._client_settings: tuple[object, ...] | None = None
+        self._client_loop_started = False
         # Source replacement/removal is serialized independently from callback
         # bookkeeping.  In particular, paho's loop_stop() joins its network
         # thread, whose callback completion also needs ``_lock``.
@@ -121,6 +127,7 @@ class MqttManager:
     @property
     def is_connected(self) -> bool:
         """True if the underlying paho client is currently connected."""
+        with self._lock:
             return self._connected
 
     # ------------------------------------------------------------------
@@ -141,15 +148,23 @@ class MqttManager:
             logger.debug("MQTT not configured or not enabled — skipping connect.")
             return
 
+        with self._lifecycle_lock:
             with self._lock:
-            self._start_client(settings)
+                if self._client is not None and self._client_settings == self._settings_key(settings):
+                    thread = getattr(self._client, "_thread", None)
+                    is_alive = getattr(thread, "is_alive", None)
+                    if thread is None and not self._client_loop_started:
+                        return
+                    if thread is not None and (not callable(is_alive) or is_alive()):
+                        return
+            self._replace_client(settings)
 
     def disconnect(self) -> None:
         """Disconnect from the broker and stop the background loop thread.
 
         No-op if no client is active.
         """
-        with self._lock:
+        with self._lifecycle_lock:
             self._stop_client()
             with self._source_lifecycle_lock:
                 with self._lock:
@@ -163,9 +178,23 @@ class MqttManager:
         Call this after saving updated MQTT configuration values.
         No-op if MQTT is not enabled or broker host is not set.
         """
-        with self._lock:
+        if not _is_configured(settings):
+            # A configuration update only owns the application-wide client.
+            # DSMR sources have independent settings and lifecycle; only app
+            # shutdown via disconnect() is allowed to stop them all.
+            with self._lifecycle_lock:
                 self._stop_client()
-        self.connect(settings)
+            return
+        with self._lifecycle_lock:
+            with self._lock:
+                if self._client is not None and self._client_settings == self._settings_key(settings):
+                    thread = getattr(self._client, "_thread", None)
+                    is_alive = getattr(thread, "is_alive", None)
+                    if thread is None and not self._client_loop_started:
+                        return
+                    if thread is not None and (not callable(is_alive) or is_alive()):
+                        return
+            self._replace_client(settings)
 
     def publish(
         self,
@@ -183,8 +212,9 @@ class MqttManager:
         """
         with self._lock:
             client = self._client
+            connected = self._connected
 
-        if client is None or not self._connected:
+        if client is None or not connected:
             logger.debug("MQTT publish skipped — not connected (topic=%s).", topic)
             return False
 
@@ -212,6 +242,7 @@ class MqttManager:
         Handler exceptions are swallowed by the ``on_message`` dispatcher so
         that a buggy handler can never crash the paho loop thread.
         """
+        with self._lock:
             self._subscriptions[topic] = handler
         # Grab the client reference outside the lock to avoid holding the lock
         # while calling back into paho (which may itself acquire internal locks).
@@ -235,8 +266,8 @@ class MqttManager:
         immediately; either way the handler is removed from the registry so it
         will not be re-subscribed on the next ``_on_connect``.
         """
-        self._subscriptions.pop(topic, None)
         with self._lock:
+            self._subscriptions.pop(topic, None)
             client = self._client
             connected = self._connected
 
@@ -388,12 +419,35 @@ class MqttManager:
     # Internal helpers
     # ------------------------------------------------------------------
 
+    @staticmethod
+    def _settings_key(settings: Settings) -> tuple[object, ...]:
+        return (
+            settings.mqtt_enabled,
+            settings.mqtt_broker_host,
+            settings.mqtt_broker_port,
+            settings.mqtt_username,
+            settings.mqtt_password,
+            settings.mqtt_tls_enabled,
+            settings.mqtt_client_id,
+        )
+
+    def _replace_client(self, settings: Settings) -> None:
+        self._stop_client()
+        self._start_client(settings)
+
     def _start_client(self, settings: Settings) -> None:
-        """Build a fresh paho Client, configure it, and call loop_start + connect."""
+        """Install one generation before starting Paho's reconnecting loop."""
+        with self._lock:
+            self._client_generation += 1
+            generation = self._client_generation
+        try:
             client = mqtt.Client(
                 callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
                 client_id=settings.mqtt_client_id,
             )
+        except Exception:
+            logger.exception("MQTT client initialization failed.")
+            return
 
         # Callbacks — VERSION2 on_connect signature:
         # (client, userdata, connect_flags, reason_code, properties)
@@ -404,21 +458,21 @@ class MqttManager:
             reason_code: mqtt.ReasonCode,
             _properties: mqtt.Properties | None,
         ) -> None:
+            with self._lock:
+                if self._client is not _client or self._client_generation != generation:
+                    return
                 if reason_code.is_failure:
-                logger.warning("MQTT connection refused: %s", reason_code)
                     self._connected = False
-            else:
-                logger.info("MQTT connected (reason_code=%s).", reason_code)
+                    logger.warning("MQTT connection refused: %s", reason_code)
+                    return
                 self._connected = True
-                # Re-establish all registered subscriptions after (re-)connect.
-                for sub_topic in self._subscriptions:
+                topics = tuple(self._subscriptions)
+            logger.info("MQTT connected (reason_code=%s).", reason_code)
+            for sub_topic in topics:
                 try:
                     _client.subscribe(sub_topic)
-                        logger.debug("MQTT re-subscribed to topic=%s.", sub_topic)
                 except Exception:
-                        logger.exception(
-                            "MQTT re-subscribe failed (topic=%s).", sub_topic
-                        )
+                    logger.exception("MQTT re-subscribe failed (topic=%s).", sub_topic)
 
         # VERSION2 on_disconnect signature:
         # (client, userdata, disconnect_flags, reason_code, properties)
@@ -429,6 +483,9 @@ class MqttManager:
             reason_code: mqtt.ReasonCode,
             _properties: mqtt.Properties | None,
         ) -> None:
+            with self._lock:
+                if self._client is not _client or self._client_generation != generation:
+                    return
                 self._connected = False
             if reason_code.is_failure:
                 logger.warning("MQTT disconnected unexpectedly (reason_code=%s).", reason_code)
@@ -441,6 +498,9 @@ class MqttManager:
             _userdata: object,
             message: mqtt.MQTTMessage,
         ) -> None:
+            with self._lock:
+                if self._client is not _client or self._client_generation != generation:
+                    return
                 handler = self._subscriptions.get(message.topic)
             if handler is None:
                 logger.debug(
@@ -469,35 +529,57 @@ class MqttManager:
 
         # Credentials — never log the password
         if settings.mqtt_username:
+            try:
                 client.username_pw_set(
                     username=settings.mqtt_username,
                     password=settings.mqtt_password or None,
                 )
+            except Exception:
+                logger.exception("MQTT credentials setup failed.")
+                return
 
-        # Start background loop thread *before* connect so paho can handle
-        # the connection handshake asynchronously.
-        client.loop_start()
-
+        # Publish ownership before network processing/callbacks can begin.
+        with self._lock:
+            self._client = client
+            self._client_settings = self._settings_key(settings)
+            self._connected = False
+            self._client_loop_started = False
         try:
-            client.connect(
+            client.connect_async(
                 host=settings.mqtt_broker_host,
                 port=settings.mqtt_broker_port,
                 keepalive=60,
             )
+            result = client.loop_start()
+            if isinstance(result, int) and result != mqtt.MQTT_ERR_SUCCESS:
+                raise RuntimeError(f"MQTT loop_start failed (rc={result})")
+            with self._lock:
+                if self._client is client and self._client_generation == generation:
+                    self._client_loop_started = True
         except Exception:
+            # Remove ownership before joining the loop; callbacks from this
+            # generation become inert immediately.
+            with self._lock:
+                if self._client is client and self._client_generation == generation:
+                    self._client = None
+                    self._client_settings = None
+                    self._connected = False
+                    self._client_loop_started = False
             # Log without leaking the password
             logger.exception(
                 "MQTT connect failed (host=%s, port=%s). Stopping loop.",
                 settings.mqtt_broker_host,
                 settings.mqtt_broker_port,
             )
+            try:
+                client.disconnect()
+            except Exception:
+                pass
             try:
                 client.loop_stop()
             except Exception:
                 pass
             return
-
-        self._client = client
         logger.info(
             "MQTT client started (host=%s, port=%s).",
             settings.mqtt_broker_host,
@@ -506,11 +588,17 @@ class MqttManager:
 
     def _stop_client(self) -> None:
         """Disconnect and stop the background loop.  Idempotent."""
+        with self._lock:
             client = self._client
             if client is None:
+                self._connected = False
+                self._client_settings = None
                 return
             self._client = None
+            self._client_settings = None
             self._connected = False
+            self._client_loop_started = False
+            self._client_generation += 1
         try:
             client.disconnect()
         except Exception:

app/main.py

@@ -169,9 +169,12 @@ def _run_scheduled_ha_state_publish() -> None:
     session: Session = session_local()
     try:
         runtime_settings = build_runtime_settings(session, get_settings())
-        # Reconnect if MQTT is configured (in DB) but not yet connected.
-        if mqtt_manager.is_configured(runtime_settings) and not mqtt_manager.is_connected:
-            logger.info("_run_scheduled_ha_state_publish: MQTT configured but not connected — attempting connect.")
+        # Ensure the configured MQTT client is installed; the manager avoids
+        # replacing it while Paho is already retrying a transient disconnect.
+        if mqtt_manager.is_configured(runtime_settings):
+            # connect() is idempotent for the same settings and retains a Paho
+            # client while its own loop is retrying a transient disconnect.
+            logger.info("_run_scheduled_ha_state_publish: ensuring MQTT client is active.")
             mqtt_manager.connect(runtime_settings)
         publish_discovery(session)
         publish_states(session)

docs/design/mqtt-reliability-fixes.md

@@ -0,0 +1,124 @@
+# MQTT 可靠性修复 — 2026-10-11
+
+本次范围为 MQTT-FIX-T01 → MQTT-FIX-T02，依赖顺序执行，两张卡自动继续。
+按 [AGENTS.md](../../AGENTS.md) 与 [协作契约](README.md) 实现、独立审查及返工。
+`done` 仅代表源码自动化技术验收完成；**源码修复完成后仍须人工部署与生产验证**。
+中文实现、审查及返工简报存放于已忽略的 `review-notes/`，不入库。
+
+## 证据与边界
+
+排查日期 2026-10-11，Europe/Amsterdam。以下生产观察由用户提供，未在本轮访问生产确认：
+
+1. 应用当天约 01:00 启动，故障前未自动重启、未发生 OOM。
+2. 16:50:43 起出现 `MQTT disconnected unexpectedly (reason_code=Unspecified error).`。
+3. 截至约 22:34 累计约 270 万条断线日志；22:34–22:35 一分钟内 15,283 次断线、101 次 publish rejected。
+4. 一个 uvicorn 进程、284 个总线程；并非已经确认 284 个 MQTT 客户端。
+5. 主 client ID 为 `home-automation`；配置最后更新时间为 2026-08-28，当天未观察到变更。
+6. Modbus、WarmteLink 串口及独立 DSMR MQTT source 仍有新数据，不能描述为 broker 整体停止服务。
+7. 生产镜像的 `app/main.py`、`app/integrations/mqtt.py` SHA256 与源码一致。
+8. 用户在 broker 日志中看到大量新连接及 `session taken over`。
+9. 22:37 用户授权重启应用后，观察超过一分钟未见新异常断线或发布失败，各采集源恢复；仅属临时恢复。
+10. 首次断线触发原因未知，不能将网络故障或 broker 重启写成事实。
+
+待独立核实的链路：60 秒 HA 发布任务在未连接时调用 `connect()`；它无条件创建同 ID Paho
+client/线程、覆盖引用但未停旧 client；旧回调能覆盖共享连接状态。已有离线 fake 复现只是
+排查线索，不能代替本轮回归测试。
+
+部署背景：生产日志配置为 `{"Type":"json-file","Config":{}}`；约 276 万行日志，按内容
+估算 JSON 文件约 353 MiB（非 stat 精确大小），当时约 666 GB 空闲，无即时容量危机。
+生产 Compose 在 `/home/tianyu/.local/share/homeautomation`，与源码 Compose 是两个文件。
+
+**本轮禁止**：push、tag、改写已推送历史、修改生产配置、重启/重建生产容器、连接真实
+MQTT broker（包括 `10.238.75.82:1883`）、访问真实串口、改动生产数据库或日志；禁止
+清空历史日志、删除 Docker 文件、prune、变更全局 Docker daemon。仅提供部署步骤。
+允许源码/测试/部署文档修改、本地 commit、fixup 及整理本轮尚未推送提交。保留已有无关改动。
+
+### MQTT-FIX-T01 — 修复主 MQTT 客户端生命周期
+
+- **Status**: `done`
+- **Depends**: none
+- **Context**: 防止短暂断线后定时任务创建同 ID 客户端累积，避免旧回调污染新连接。
+
+**Files**
+- `modify app/integrations/mqtt.py`
+- `modify app/main.py`
+- `modify tests/test_mqtt_client.py`
+- `modify tests/test_mqtt_subscribe.py`
+- `create tests/test_mqtt_scheduler.py`（必要的调度任务回归）
+- `modify README.md`（MQTT 生命周期说明）
+- `modify .gitignore`（编排过程报告目录保护）
+- `create docs/design/mqtt-reliability-fixes.md`（主线维护任务与 Status）
+
+**Steps**
+1. 独立核实当前生命周期、回调及调度代码，结合安装的 Paho 2.1.0 源码核对启动顺序和失败清理。
+2. 使同配置 `connect()` 幂等，区分正在连接、等待自动重连、无客户端；现有 client 能自动重连时不再创建额外 client。
+3. 初始连接失败、初始化失败、后台循环失效均须可恢复，失败对象/循环清理，不能永久跳过重试。
+4. 配置变更安全停止、清理旧 client 后替换，受管理的主 client 只有一个。
+5. connect/disconnect/message 回调核验身份或 generation，拒绝旧状态及过期消息。
+6. 处理创建、替换、关闭与回调并发；不能持有回调需要的锁调用等待网络线程退出的 `loop_stop()`。
+7. 保持独立 DSMR source、订阅恢复、MQTT 测试身份及现有 publish 行为兼容。
+8. 增加离线故障与并发回归，更新生命周期说明，运行适用闸门并保存中文实现简报。
+
+**Out of scope / 不要碰**
+- 不使用随机 ID、单纯换 ID 或降低日志级别掩盖问题。
+- 不改变 HTTP 路由/schema/docstring；如确有必要变更，须同步 OpenAPI 和前端类型。
+- 不改变旧 milestone 完成状态；不访问生产或真实设备；不处理已有 Compose 串口改动。
+
+**Acceptance criteria**
+- [x] 同配置重复 connect 及断线后多轮定时任务不累积 client 或后台循环。
+- [x] 初次连接/初始化失败后能重试，失败对象和循环被清理，循环失效也有恢复路径。
+- [x] 配置重连后旧 client 停止，旧 connect/disconnect/message 回调不能影响新状态或处理过期消息。
+- [x] 当前 client 重连成功后恢复订阅与发布。
+- [x] shutdown、reconnect 与回调交错时能完成、不死锁。
+- [x] 新测试使用 mock/fake 及合成配置，不连接生产 broker 或真实串口；覆盖行为与并发边界。
+- [x] DSMR source、测试 client ID 及现有 publish 行为兼容。
+- [x] pytest、ruff 全绿，独立 Reviewer 重跑；如适用同步契约/检查构建上下文。
+
+**Reviewer checklist**
+- 独立核查 Paho 2.1.0 生命周期顺序、线程泄漏、状态污染、死锁及失败恢复，不仅断言内部实现细节。
+- 核查调度触发路径、多轮断线自动重连、配置替换与关闭交错，以及旧消息拒绝。
+- 核查 DSMR 独立客户端及订阅/publish/test identity 回归风险。
+- 先读对应实现/返工简报，独立重跑闸门；报告 PASS 或编号返工清单，不改实现或 Git。
+
+### MQTT-FIX-T02 — 为 Docker 日志配置轮转
+
+- **Status**: `todo`
+- **Depends**: MQTT-FIX-T01
+- **Context**: 为本项目容器限定日志保留，明确源码修改与生产容器应用配置是两个操作。
+
+**Files**
+- `modify docker-compose.yml`
+- `modify docker-compose.dev.yml`（仅必要时）
+- `modify README.md`（部署说明）
+- `modify tests/test_deployment.py`（必要的静态检查）
+- `modify docs/design/mqtt-reliability-fixes.md`（主线维护 Status）
+
+**Steps**
+1. 为 app 与 migration 均配置 `logging.driver: json-file`，字符串 options `max-size: "10m"`、`max-file: "3"`。
+2. 验证开发 override 合并后保留相同合理的轮转配置，不修改全局 daemon 或其他项目。
+3. 用 `docker compose config --quiet` 或等效静态检查验证基础与合并配置；不输出含凭据的完整配置。
+4. 说明宿主机日志路径 `/var/lib/docker/containers/<完整ID>/<完整ID>-json.log`、`docker logs`、保留策略。
+5. 明确仅 `docker restart` 不应用 logging 配置，需要重建容器。
+6. 给出生产目录 `/home/tianyu/.local/share/homeautomation` 中的最小配置同步、重建、验证与回退步骤。
+7. 写清发布源码修复镜像与仅应用日志配置各自前提，以及短暂采集中断；不覆盖生产设备/volume/env 等配置。
+8. 保存中文简报，跑 pytest / ruff 和配置验证，独立 Reviewer 核查可执行性。
+
+**Out of scope / 不要碰**
+- 不自动部署，不清空或删除日志，不 prune，不触碰设备历史数据。
+- 不修改全局 Docker daemon；不把源码配置变更描述为生产已生效。
+- 不纳入用户已有的串口映射/group_add 改动。
+
+**Acceptance criteria**
+- [ ] app、migration 最终 Compose 均含指定字符串 logging options。
+- [ ] 开发 override 合并后日志配置正确。
+- [ ] Compose 经 config --quiet 或等效静态检查，不输出凭据。
+- [ ] 文档说明日志路径、docker logs、保留策略、restart 无效而须重建。
+- [ ] 文档明确生产路径、最小同步、镜像更新前提、采集中断、验证与回退；区分镜像发布和日志配置应用。
+- [ ] 无生产操作、数据/日志删除，已有无关改动保留。
+- [ ] pytest、ruff 全绿，独立 Reviewer 重跑并核查合并及部署可执行性。
+
+**Reviewer checklist**
+- 核查 logging options 类型及基础/开发配置合并；静态配置检查的范围和凭据保护。
+- 核查生产/source Compose 分离，镜像构建发布前提与 logging 重建步骤是否明确。
+- 核查只涉及 app/migration，不变更串口、volume、生产 DB、全局 daemon；说明旧日志在重建时的 Docker 生命周期影响及可回退方式。
+- 先读对应实现/返工简报，独立重跑闸门，给 PASS 或编号返工清单。

tests/test_mqtt_client.py

@@ -4,6 +4,7 @@ All paho interaction is mocked — no real broker is required.
 """
 from __future__ import annotations
 
+import threading
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -103,7 +104,7 @@ def test_connect_creates_paho_client_with_version2() -> None:
     settings = _make_settings()
 
     mock_client = MagicMock()
-    mock_client.connect.return_value = None
+    mock_client.connect_async.return_value = None
 
     with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client) as mock_cls:
         with patch("app.integrations.mqtt.mqtt.CallbackAPIVersion"):
@@ -117,7 +118,7 @@ def test_connect_creates_paho_client_with_version2() -> None:
     # loop_start must have been called
     mock_client.loop_start.assert_called_once()
     # connect must have been called with the right host/port
-    mock_client.connect.assert_called_once_with(
+    mock_client.connect_async.assert_called_once_with(
         host="broker.test", port=1883, keepalive=60
     )
 
@@ -133,6 +134,220 @@ def test_connect_uses_configured_client_id() -> None:
     assert mock_cls.call_args.kwargs["client_id"] == "home-automation-dev"
 
 
+def test_connect_same_settings_is_idempotent_even_when_disconnected() -> None:
+    manager = MqttManager()
+    settings = _make_settings()
+    client = MagicMock()
+    with patch("app.integrations.mqtt.mqtt.Client", return_value=client) as cls:
+        manager.connect(settings)
+        manager.connect(settings)
+    cls.assert_called_once()
+    client.loop_start.assert_called_once()
+    client.connect_async.assert_called_once()
+
+
+def test_connect_replaces_client_when_paho_loop_thread_has_exited() -> None:
+    manager = MqttManager()
+    settings = _make_settings()
+    old_client, recovered_client = MagicMock(), MagicMock()
+    with patch("app.integrations.mqtt.mqtt.Client", side_effect=[old_client, recovered_client]):
+        manager.connect(settings)
+        old_client._thread = None
+        manager._client_loop_started = True
+        manager.connect(settings)
+
+    old_client.disconnect.assert_called_once()
+    old_client.loop_stop.assert_called_once()
+    recovered_client.loop_start.assert_called_once()
+
+
+def test_loop_start_failure_cleans_installed_client_for_retry() -> None:
+    manager = MqttManager()
+    failed_client = MagicMock()
+    failed_client.loop_start.return_value = -1
+    recovered_client = MagicMock()
+    with patch("app.integrations.mqtt.mqtt.Client", side_effect=[failed_client, recovered_client]):
+        manager.connect(_make_settings())
+        assert manager._client is None
+        manager.connect(_make_settings())
+
+    failed_client.disconnect.assert_called_once()
+    failed_client.loop_stop.assert_called_once()
+    assert manager._client is recovered_client
+
+
+def test_concurrent_same_config_connect_creates_one_client() -> None:
+    manager = MqttManager()
+    settings = _make_settings()
+    client = MagicMock()
+    barrier = threading.Barrier(3)
+
+    def connect() -> None:
+        barrier.wait(timeout=1)
+        manager.connect(settings)
+
+    with patch("app.integrations.mqtt.mqtt.Client", return_value=client) as cls:
+        workers = [threading.Thread(target=connect) for _ in range(2)]
+        for worker in workers:
+            worker.start()
+        barrier.wait(timeout=1)
+        for worker in workers:
+            worker.join(timeout=2)
+
+    assert all(not worker.is_alive() for worker in workers)
+    cls.assert_called_once()
+    client.loop_start.assert_called_once()
+
+
+@pytest.mark.parametrize("operation", ["disconnect", "reconnect"])
+def test_teardown_can_join_loop_while_disconnect_callback_runs(operation: str) -> None:
+    """A real callback thread must finish while teardown joins the fake loop."""
+    manager = MqttManager()
+    callback_go = threading.Event()
+    callback_entered = threading.Event()
+    callback_done = threading.Event()
+    clients: list[object] = []
+
+    class LoopClient:
+        def __init__(self) -> None:
+            self._thread: threading.Thread | None = None
+            self.on_disconnect = None
+            self.loop_joined = threading.Event()
+
+        def connect_async(self, **_kwargs: object) -> None:
+            pass
+
+        def loop_start(self) -> int:
+            def network_callback() -> None:
+                callback_go.wait(timeout=2)
+                callback_entered.set()
+                try:
+                    self.on_disconnect(self, None, None, MagicMock(is_failure=True), None)
+                finally:
+                    callback_done.set()
+
+            self._thread = threading.Thread(target=network_callback, daemon=True)
+            self._thread.start()
+            return 0
+
+        def disconnect(self) -> None:
+            callback_go.set()
+
+        def loop_stop(self) -> None:
+            assert callback_entered.wait(timeout=1), "callback thread did not start"
+            assert self._thread is not None
+            self._thread.join(timeout=1)
+            assert not self._thread.is_alive(), "callback thread did not exit before loop_stop returned"
+            self.loop_joined.set()
+
+        def cleanup(self) -> None:
+            callback_go.set()
+            if self._thread is not None:
+                self._thread.join(timeout=1)
+
+    def make_client(**_kwargs: object) -> LoopClient:
+        client = LoopClient()
+        clients.append(client)
+        return client
+
+    try:
+        with patch("app.integrations.mqtt.mqtt.Client", side_effect=make_client):
+            manager.connect(_make_settings())
+            old_client = clients[0]
+            if operation == "disconnect":
+                manager.disconnect()
+                assert manager._client is None
+            else:
+                manager.reconnect(_make_settings(mqtt_client_id="replacement"))
+                assert manager._client is clients[1]
+            assert callback_done.is_set()
+            assert old_client.loop_joined.is_set()
+            assert old_client._thread is not None and not old_client._thread.is_alive()
+    finally:
+        for client in clients:
+            client.cleanup()
+
+
+@pytest.mark.parametrize("failure", ["constructor", "tls", "credentials", "connect_async", "loop_start"])
+def test_initialization_failure_is_cleaned_and_next_connect_recovers(failure: str) -> None:
+    manager = MqttManager()
+    broken, recovered = MagicMock(), MagicMock()
+    broken.connect_async.return_value = None
+    if failure == "tls":
+        broken.tls_set.side_effect = OSError("synthetic TLS failure")
+    elif failure == "credentials":
+        broken.username_pw_set.side_effect = OSError("synthetic credential failure")
+    elif failure == "connect_async":
+        broken.connect_async.side_effect = OSError("synthetic connect setup failure")
+    elif failure == "loop_start":
+        broken.loop_start.side_effect = OSError("synthetic loop failure")
+
+    settings = _make_settings(
+        mqtt_tls_enabled=failure == "tls",
+        mqtt_username="user" if failure == "credentials" else "",
+        mqtt_password="synthetic",
+    )
+    first = (
+        [RuntimeError("synthetic constructor failure"), recovered]
+        if failure == "constructor"
+        else [broken, recovered]
+    )
+    with patch("app.integrations.mqtt.mqtt.Client", side_effect=first):
+        manager.connect(settings)
+        assert manager._client is None
+        manager.connect(settings)
+
+    assert manager._client is recovered
+    recovered.loop_start.assert_called_once()
+    recovered.connect_async.assert_called_once()
+    if failure in {"connect_async", "loop_start"}:
+        broken.disconnect.assert_called_once()
+        broken.loop_stop.assert_called_once()
+    elif failure in {"tls", "credentials"}:
+        broken.loop_start.assert_not_called()
+
+
+@pytest.mark.parametrize("old_is_failure", [False, True])
+def test_old_connect_success_and_failure_callbacks_cannot_change_replacement(
+    old_is_failure: bool,
+) -> None:
+    manager = MqttManager()
+    old_client, new_client = MagicMock(), MagicMock()
+    with patch("app.integrations.mqtt.mqtt.Client", side_effect=[old_client, new_client]):
+        manager.connect(_make_settings())
+        manager.reconnect(_make_settings(mqtt_client_id="replacement"))
+
+    manager._connected = True
+    old_client.on_connect(
+        old_client, None, None, MagicMock(is_failure=old_is_failure), None
+    )
+    assert manager._client is new_client
+    assert manager.is_connected
+    new_client.subscribe.assert_not_called()
+
+
+def test_old_client_callbacks_cannot_change_replacement_state_or_dispatch() -> None:
+    manager = MqttManager()
+    settings = _make_settings()
+    old_client, new_client = MagicMock(), MagicMock()
+    received: list[bytes] = []
+    manager.subscribe("sensor/state", received.append)
+    with patch("app.integrations.mqtt.mqtt.Client", side_effect=[old_client, new_client]):
+        manager.connect(settings)
+        manager.reconnect(_make_settings(mqtt_client_id="replacement"))
+
+    manager._connected = True
+    failed = MagicMock()
+    failed.is_failure = True
+    old_client.on_disconnect(old_client, None, None, failed, None)
+    old_client.on_message(old_client, None, MagicMock(topic="sensor/state", payload=b"stale"))
+    assert manager.is_connected
+    assert received == []
+
+    new_client.on_message(new_client, None, MagicMock(topic="sensor/state", payload=b"fresh"))
+    assert received == [b"fresh"]
+
+
 def test_deployment_client_id_variants_are_distinct() -> None:
     production_ids = {
         "home-automation",
@@ -155,7 +370,7 @@ def test_connect_sets_credentials_when_username_provided() -> None:
     settings = _make_settings(mqtt_username="user", mqtt_password="s3cr3t")
 
     mock_client = MagicMock()
-    mock_client.connect.return_value = None
+    mock_client.connect_async.return_value = None
 
     with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client):
         manager.connect(settings)
@@ -170,7 +385,7 @@ def test_connect_skips_credentials_when_username_empty() -> None:
     settings = _make_settings(mqtt_username="", mqtt_password="")
 
     mock_client = MagicMock()
-    mock_client.connect.return_value = None
+    mock_client.connect_async.return_value = None
 
     with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client):
         manager.connect(settings)
@@ -183,7 +398,7 @@ def test_connect_calls_tls_set_when_tls_enabled() -> None:
     settings = _make_settings(mqtt_tls_enabled=True)
 
     mock_client = MagicMock()
-    mock_client.connect.return_value = None
+    mock_client.connect_async.return_value = None
 
     with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client):
         manager.connect(settings)
@@ -200,7 +415,7 @@ def test_connect_does_not_raise_on_connection_failure() -> None:
     settings = _make_settings()
 
     mock_client = MagicMock()
-    mock_client.connect.side_effect = OSError("Connection refused")
+    mock_client.connect_async.side_effect = OSError("Connection refused")
 
     with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client):
         # Must not raise — failures are caught and logged
@@ -221,7 +436,7 @@ def test_disconnect_stops_loop_and_clears_client() -> None:
     settings = _make_settings()
 
     mock_client = MagicMock()
-    mock_client.connect.return_value = None
+    mock_client.connect_async.return_value = None
 
     with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client):
         manager.connect(settings)
@@ -249,7 +464,7 @@ def test_publish_passes_topic_payload_retain_to_paho() -> None:
     settings = _make_settings()
 
     mock_client = MagicMock()
-    mock_client.connect.return_value = None
+    mock_client.connect_async.return_value = None
 
     with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client):
         manager.connect(settings)
@@ -306,7 +521,7 @@ def test_reconnect_stops_old_client_and_starts_new_one() -> None:
     manager._connected = True
 
     new_client = MagicMock()
-    new_client.connect.return_value = None
+    new_client.connect_async.return_value = None
 
     with patch("app.integrations.mqtt.mqtt.Client", return_value=new_client):
         manager.reconnect(settings)
@@ -317,11 +532,60 @@ def test_reconnect_stops_old_client_and_starts_new_one() -> None:
 
     # New client must have been connected
     new_client.loop_start.assert_called_once()
-    new_client.connect.assert_called_once_with(
+    new_client.connect_async.assert_called_once_with(
         host="broker.test", port=1883, keepalive=60
     )
 
 
+@pytest.mark.parametrize(
+    "settings",
+    [
+        _make_settings(mqtt_enabled=False),
+        _make_settings(mqtt_broker_host=""),
+    ],
+    ids=["disabled", "empty-host"],
+)
+def test_unconfigured_reconnect_stops_main_client_but_keeps_dsmr_source(settings) -> None:
+    """Changing main MQTT config must not interrupt independent DSMR clients."""
+    manager = MqttManager()
+    main_client = MagicMock()
+    source_client = MagicMock()
+    received: list[bytes] = []
+
+    with patch(
+        "app.integrations.mqtt.mqtt.Client",
+        side_effect=[main_client, source_client],
+    ):
+        manager.connect(_make_settings())
+        assert manager.replace_source(
+            17,
+            host="source.test",
+            port=1883,
+            username="",
+            password="",
+            tls_enabled=False,
+            subscriptions={"meter/17": received.append},
+        )
+
+    manager.reconnect(settings)
+
+    assert manager._client is None
+    main_client.disconnect.assert_called_once()
+    main_client.loop_stop.assert_called_once()
+    assert manager._source_clients[17] is source_client
+    assert manager.source_is_active(17)
+    source_client.disconnect.assert_not_called()
+    source_client.loop_stop.assert_not_called()
+
+    message = MagicMock()
+    message.topic = "meter/17"
+    message.payload = b"reading"
+    source_client.on_message(source_client, None, message)
+    assert received == [b"reading"]
+
+    manager.disconnect()
+
+
 # ---------------------------------------------------------------------------
 # POST /api/config/mqtt/test — session + CSRF guards
 # ---------------------------------------------------------------------------

tests/test_mqtt_scheduler.py

@@ -0,0 +1,68 @@
+"""Offline regression coverage for scheduler-driven MQTT recovery."""
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from app.integrations.mqtt import MqttManager
+from app.main import _run_scheduled_ha_state_publish
+
+
+def test_scheduler_reuses_manager_across_disconnects_then_subscribes_and_publishes() -> None:
+    session = MagicMock()
+    session_local = MagicMock(return_value=session)
+    settings = MagicMock(
+        mqtt_enabled=True,
+        mqtt_broker_host="synthetic-broker",
+        mqtt_broker_port=1883,
+        mqtt_username="",
+        mqtt_password="",
+        mqtt_tls_enabled=False,
+        mqtt_client_id="offline-test",
+    )
+    manager = MqttManager()
+    clients: list[MagicMock] = []
+
+    def make_client(*_args: object, **_kwargs: object) -> MagicMock:
+        client = MagicMock()
+        clients.append(client)
+        return client
+
+    manager.subscribe("homeassistant/status", lambda _payload: None)
+
+    def publish_states(_session: object) -> None:
+        manager.publish("homeassistant/state", b"state", retain=True)
+
+    with (
+        patch("app.main.get_session_local", return_value=session_local),
+        patch("app.main.build_runtime_settings", return_value=settings),
+        patch("app.main.get_settings", return_value=MagicMock()),
+        patch("app.main.mqtt_manager", manager),
+        patch("app.main.publish_discovery") as publish_discovery,
+        patch("app.main.publish_states", side_effect=publish_states),
+        patch("app.integrations.mqtt.mqtt.Client", side_effect=make_client),
+    ):
+        for _ in range(3):
+            _run_scheduled_ha_state_publish()
+            assert len(clients) == 1
+            assert clients[0].loop_start.call_count == 1
+            clients[0].on_disconnect(
+                clients[0], None, None, MagicMock(is_failure=True), None
+            )
+            assert not manager.is_connected
+
+        # A subsequent scheduler turn sees Paho's existing loop retrying; it
+        # must not replace that client. Simulate the successful CONNACK.
+        _run_scheduled_ha_state_publish()
+        assert len(clients) == 1
+        clients[0].on_connect(clients[0], None, None, MagicMock(is_failure=False), None)
+
+        assert manager.is_connected
+        clients[0].subscribe.assert_called_with("homeassistant/status")
+        _run_scheduled_ha_state_publish()
+        assert len(clients) == 1
+        clients[0].publish.assert_called_once_with(
+            "homeassistant/state", payload=b"state", qos=0, retain=True
+        )
+        assert publish_discovery.call_count == 5
+        assert session.close.call_count == 5
+        manager.disconnect()

tests/test_mqtt_subscribe.py

@@ -143,7 +143,7 @@ def test_unsubscribe_unknown_topic_is_noop() -> None:
 # ---------------------------------------------------------------------------
 
 
-def _extract_on_message(manager: MqttManager, settings) -> object:
+def _extract_on_message(manager: MqttManager, settings) -> tuple[object, MagicMock]:
     """Connect with a mocked paho client and return the on_message callback that
     was registered on the client mock."""
     mock_client = MagicMock()
@@ -153,7 +153,7 @@ def _extract_on_message(manager: MqttManager, settings) -> object:
         manager.connect(settings)
 
     # The callback was assigned as: client.on_message = _on_message
-    return mock_client.on_message
+    return mock_client.on_message, mock_client
 
 
 def test_on_message_dispatches_to_registered_handler() -> None:
@@ -168,10 +168,10 @@ def test_on_message_dispatches_to_registered_handler() -> None:
 
     manager.subscribe("dsmr/json", handler)
 
-    on_message = _extract_on_message(manager, settings)
+    on_message, client = _extract_on_message(manager, settings)
 
     msg = _make_mqtt_message("dsmr/json", b'{"id": 1}')
-    on_message(MagicMock(), None, msg)  # simulate paho calling the callback
+    on_message(client, None, msg)  # simulate paho calling the callback
 
     assert received_payloads == [b'{"id": 1}']
 
@@ -183,11 +183,11 @@ def test_on_message_ignores_unknown_topic() -> None:
     handler = MagicMock()
     manager.subscribe("other/topic", handler)
 
-    on_message = _extract_on_message(manager, settings)
+    on_message, client = _extract_on_message(manager, settings)
 
     msg = _make_mqtt_message("unknown/topic", b"data")
     # Must not raise
-    on_message(MagicMock(), None, msg)
+    on_message(client, None, msg)
 
     handler.assert_not_called()
 
@@ -201,11 +201,11 @@ def test_on_message_swallows_handler_exception() -> None:
         raise RuntimeError("handler exploded")
 
     manager.subscribe("dsmr/json", bad_handler)
-    on_message = _extract_on_message(manager, settings)
+    on_message, client = _extract_on_message(manager, settings)
 
     msg = _make_mqtt_message("dsmr/json", b"{}")
     # This must not raise
-    on_message(MagicMock(), None, msg)
+    on_message(client, None, msg)
 
 
 def test_on_message_does_not_crash_on_handler_exception_multiple_calls() -> None:
@@ -221,13 +221,13 @@ def test_on_message_does_not_crash_on_handler_exception_multiple_calls() -> None
             raise ValueError("first call fails")
 
     manager.subscribe("dsmr/json", flaky_handler)
-    on_message = _extract_on_message(manager, settings)
+    on_message, client = _extract_on_message(manager, settings)
 
     msg1 = _make_mqtt_message("dsmr/json", b"first")
     msg2 = _make_mqtt_message("dsmr/json", b"second")
 
-    on_message(MagicMock(), None, msg1)  # should not raise despite exception in handler
-    on_message(MagicMock(), None, msg2)  # handler called again
+    on_message(client, None, msg1)  # should not raise despite exception in handler
+    on_message(client, None, msg2)  # handler called again
 
     assert call_count[0] == 2