M8-R02: report DSMR MQTT source health

app/integrations/mqtt.py

@@ -248,6 +248,7 @@ class MqttManager:
         tls_enabled: bool,
         subscriptions: dict[str, Callable[[bytes], None]],
         base_client_id: str = "home-automation",
+        state_handler: Callable[[str], None] | None = None,
     ) -> bool:
         """Replace one source-owned client and its handlers.
 
@@ -259,6 +260,7 @@ class MqttManager:
         with self._source_lifecycle_lock:
             self._stop_source_client(source_id)
             if not host:
+                self._report_source_state(state_handler, "error", source_id)
                 return False
             self._next_source_generation += 1
             generation = self._next_source_generation
@@ -279,14 +281,18 @@ class MqttManager:
                     if not self._is_current_source_client(source_id, generation, connected_client):
                         return
                     if reason_code.is_failure:
+                        self._source_connected.discard(source_id)
                         logger.warning("DSMR MQTT connection refused for source_id=%s", source_id)
-                        return
+                        state = "error"
-                    self._source_connected.add(source_id)
+                    else:
-                    for topic in captured_subscriptions:
+                        self._source_connected.add(source_id)
-                        try:
+                        state = "online"
-                            connected_client.subscribe(topic)
+                        for topic in captured_subscriptions:
-                        except Exception:
+                            try:
-                            logger.exception("DSMR MQTT re-subscribe failed for source_id=%s", source_id)
+                                connected_client.subscribe(topic)
+                            except Exception:
+                                logger.exception("DSMR MQTT re-subscribe failed for source_id=%s", source_id)
+                self._report_source_state(state_handler, state, source_id)
 
             def _on_disconnect(
                 disconnected_client: mqtt.Client,
@@ -299,6 +305,7 @@ class MqttManager:
                     if not self._is_current_source_client(source_id, generation, disconnected_client):
                         return
                     self._source_connected.discard(source_id)
+                self._report_source_state(state_handler, "error", source_id)
 
             def _on_message(
                 message_client: mqtt.Client,
@@ -334,6 +341,7 @@ class MqttManager:
                     client.tls_set()
                 except Exception:
                     logger.exception("DSMR MQTT TLS setup failed for source_id=%s", source_id)
+                    self._report_source_state(state_handler, "error", source_id)
                     return False
             if username:
                 client.username_pw_set(username=username, password=password or None)
@@ -345,11 +353,13 @@ class MqttManager:
                 self._source_subscriptions[source_id] = captured_subscriptions
                 self._source_generations[source_id] = generation
                 self._source_states[source_id] = _SourceClientState(client, generation)
+            self._report_source_state(state_handler, "connecting", source_id)
             client.loop_start()
             try:
                 client.connect(host=host, port=port, keepalive=60)
             except Exception:
                 logger.exception("DSMR MQTT connect failed (source_id=%s, host=%s)", source_id, host)
+                self._report_source_state(state_handler, "error", source_id)
                 self._stop_source_client(source_id)
                 return False
             return True
@@ -541,6 +551,18 @@ class MqttManager:
             and self._source_clients.get(source_id) is client
         )
 
+    @staticmethod
+    def _report_source_state(
+        state_handler: Callable[[str], None] | None, state: str, source_id: int
+    ) -> None:
+        """Invoke an optional health callback without exposing connection credentials."""
+        if state_handler is None:
+            return
+        try:
+            state_handler(state)
+        except Exception:
+            logger.exception("DSMR MQTT source state update failed for source_id=%s", source_id)
+
 
 # ---------------------------------------------------------------------------
 # Module-level singleton — shared across lifespan and route handlers

app/services/dsmr_ingest.py

@@ -185,9 +185,11 @@ def apply_dsmr_subscription(settings: "Settings | None" = None) -> None:
         # Every source owns a distinct MQTT client, so equal topics from
         # different sources/brokers are dispatchable.  A telegram and tariff
         # topic on the *same* client would overwrite one handler, however.
+        rejected_source_ids: set[int] = set()
         for source_id, snapshot in list(desired.items()):
             if snapshot.tariff_topic and snapshot.topic == snapshot.tariff_topic:
                 logger.error("DSMR source_id=%s rejected: telegram/tariff topic collision", source_id)
+                rejected_source_ids.add(source_id)
                 desired.pop(source_id)
 
         with _subscription_lock:
@@ -203,6 +205,19 @@ def apply_dsmr_subscription(settings: "Settings | None" = None) -> None:
                 _subscription_tokens.pop(source_id, None)
                 set_current_tariff(source_id, None)
 
+        # A disabled source has no installed MQTT owner.  Persist that fact
+        # after invalidating its callback token, so a retained callback cannot
+        # revive an earlier online state while teardown is in progress.
+        for source_id in stale_source_ids:
+            _mark_disabled_source_inactive(source_id)
+
+        # A topic collision is a configuration error for an enabled source,
+        # not a disabled-state transition.  It must therefore replace any
+        # earlier online state even when this process started without a
+        # matching runtime subscription to tear down.
+        for source_id in rejected_source_ids:
+            _mark_rejected_source_error(source_id)
+
         for source_id in stale_source_ids:
             mqtt_manager.remove_source(source_id)
 
@@ -250,6 +265,9 @@ def apply_dsmr_subscription(settings: "Settings | None" = None) -> None:
                 tls_enabled=snapshot.tls_enabled,
                 subscriptions=handlers,
                 base_client_id=base_client_id,
+                state_handler=lambda state, captured=snapshot, captured_token=token: (
+                    handle_captured_source_state(captured, captured_token, state)
+                ),
             )
             if not applied:
                 with _subscription_lock:
@@ -293,6 +311,72 @@ def handle_captured_tariff_message(
         handle_tariff_message(payload_bytes, snapshot.source_id)
 
 
+def handle_captured_source_state(snapshot: DsmrSourceSnapshot, token: object, state: str) -> None:
+    """Persist one active generation's connection health in a short DB session."""
+    with _subscription_lock:
+        if _subscription_tokens.get(snapshot.source_id) is not token:
+            return
+        session_local = get_session_local()
+        session = session_local()
+        try:
+            source = session.get(MeterSource, snapshot.source_id)
+            if source is None or not source.enabled or source.kind != "dsmr_mqtt":
+                return
+            source.status = state
+            source.last_error = "MQTT connection failed." if state == "error" else None
+            source.updated_at = datetime.now(timezone.utc)
+            session.commit()
+        except Exception:
+            session.rollback()
+            logger.exception("DSMR source health update failed for source_id=%s", snapshot.source_id)
+        finally:
+            session.close()
+
+
+def _mark_disabled_source_inactive(source_id: int) -> None:
+    """Clear an obsolete online health state for a disabled DSMR source.
+
+    The caller has already invalidated the source's generation token.  This
+    helper deliberately opens its own short session so reconcile never shares
+    a callback-thread transaction.  Deleted sources simply have no row left
+    to update.
+    """
+    session_local = get_session_local()
+    session = session_local()
+    try:
+        source = session.get(MeterSource, source_id)
+        if source is None or source.enabled or source.kind != "dsmr_mqtt":
+            return
+        source.status = "unknown"
+        source.last_error = None
+        source.updated_at = datetime.now(timezone.utc)
+        session.commit()
+    except Exception:
+        session.rollback()
+        logger.exception("DSMR disabled source health update failed for source_id=%s", source_id)
+    finally:
+        session.close()
+
+
+def _mark_rejected_source_error(source_id: int) -> None:
+    """Persist a non-sensitive error for an enabled source rejected by reconcile."""
+    session_local = get_session_local()
+    session = session_local()
+    try:
+        source = session.get(MeterSource, source_id)
+        if source is None or not source.enabled or source.kind != "dsmr_mqtt":
+            return
+        source.status = "error"
+        source.last_error = "DSMR source configuration invalid."
+        source.updated_at = datetime.now(timezone.utc)
+        session.commit()
+    except Exception:
+        session.rollback()
+        logger.exception("DSMR rejected source health update failed for source_id=%s", source_id)
+    finally:
+        session.close()
+
+
 def _handle_message_inner(payload_bytes: bytes, snapshot: DsmrSourceSnapshot) -> None:
     try:
         data = json.loads(payload_bytes)
@@ -331,7 +415,13 @@ def _handle_message_inner(payload_bytes: bytes, snapshot: DsmrSourceSnapshot) ->
                     payload=data,
                 )
             )
-            session.commit()
+        source = session.get(MeterSource, snapshot.source_id)
+        if source is not None and source.enabled and source.kind == "dsmr_mqtt":
+            source.status = "online"
+            source.last_seen_at = datetime.now(timezone.utc)
+            source.last_error = None
+            source.updated_at = datetime.now(timezone.utc)
+        session.commit()
     except sqlalchemy.exc.IntegrityError:
         session.rollback()
     except Exception:

tests/test_dsmr_ingest.py

@@ -31,6 +31,7 @@ from sqlalchemy import create_engine, select
 from sqlalchemy.orm import Session
 
 from app.models.energy import DsmrReading
+from app.models.meter_source import MeterSource
 
 
 # ---------------------------------------------------------------------------
@@ -174,6 +175,264 @@ def test_second_00_persists_full_frame(dsmr_db):
     assert payload["phase_voltage_l2"] is None
 
 
+def test_accepted_message_marks_source_online_and_updates_last_seen(dsmr_db):
+    """A valid, accepted telegram clears stale diagnostics without changing its payload."""
+    engine, SessionLocal = dsmr_db
+    settings = _make_settings(dsmr_sample_interval_s=10)
+    with Session(engine) as session:
+        source = session.get(MeterSource, 1)
+        assert source is not None
+        source.enabled = True
+        source.status = "error"
+        source.last_error = "old error"
+        session.commit()
+
+    _call_handle_message(_SAMPLE_TELEGRAM, settings, SessionLocal)
+
+    with Session(engine) as session:
+        source = session.get(MeterSource, 1)
+        assert source is not None
+        assert source.status == "online"
+        assert source.last_error is None
+        assert source.last_seen_at is not None
+
+
+def test_source_state_callback_persists_only_current_generation(dsmr_db, monkeypatch):
+    """Connection callbacks use a short session and stale generations leave health untouched."""
+    engine, SessionLocal = dsmr_db
+    from app.services import dsmr_ingest
+
+    snapshot = _make_settings()
+    current = object()
+    monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {snapshot.source_id: current})
+    with Session(engine) as session:
+        source = session.get(MeterSource, snapshot.source_id)
+        assert source is not None
+        source.enabled = True
+        session.commit()
+    with patch.object(dsmr_ingest, "get_session_local", return_value=SessionLocal):
+        dsmr_ingest.handle_captured_source_state(snapshot, current, "connecting")
+        dsmr_ingest.handle_captured_source_state(snapshot, current, "error")
+        dsmr_ingest.handle_captured_source_state(snapshot, object(), "online")
+
+    with Session(engine) as session:
+        source = session.get(MeterSource, snapshot.source_id)
+        assert source is not None
+        assert source.status == "error"
+        assert source.last_error == "MQTT connection failed."
+
+
+def test_source_health_is_isolated_between_sources(dsmr_db, monkeypatch):
+    """Connection and message health changes only touch their own source row."""
+    engine, SessionLocal = dsmr_db
+    from app.services import dsmr_ingest
+    from app.services.dsmr_ingest import DsmrSourceSnapshot
+
+    first = _make_settings()
+    second = DsmrSourceSnapshot(2, "second/topic", "", 10)
+    first_token = object()
+    second_token = object()
+    monkeypatch.setattr(
+        dsmr_ingest, "_subscription_tokens", {first.source_id: first_token, second.source_id: second_token}
+    )
+    with Session(engine) as session:
+        first_source = session.get(MeterSource, first.source_id)
+        assert first_source is not None
+        first_source.enabled = True
+        session.add(MeterSource(
+            id=2, name="Second DSMR", kind="dsmr_mqtt", enabled=True, config={},
+            status="online", created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc),
+        ))
+        session.commit()
+
+    with patch.object(dsmr_ingest, "get_session_local", return_value=SessionLocal):
+        dsmr_ingest.handle_captured_source_state(first, first_token, "connecting")
+        dsmr_ingest.handle_captured_source_state(second, second_token, "error")
+        dsmr_ingest.handle_message(_payload(_SAMPLE_TELEGRAM), first)
+
+    with Session(engine) as session:
+        first_source = session.get(MeterSource, first.source_id)
+        second_source = session.get(MeterSource, second.source_id)
+        assert first_source is not None and second_source is not None
+        assert first_source.status == "online"
+        assert first_source.last_error is None
+        assert first_source.last_seen_at is not None
+        assert second_source.status == "error"
+        assert second_source.last_error == "MQTT connection failed."
+        assert second_source.last_seen_at is None
+
+
+def test_disable_reconcile_persists_unknown_and_rejects_retained_state_callback(dsmr_db, monkeypatch):
+    """Disable invalidates the generation before clearing a stale online health state."""
+    engine, SessionLocal = dsmr_db
+    from app.services import dsmr_ingest
+
+    class FakeMqtt:
+        def __init__(self) -> None:
+            self.removed: list[int] = []
+            self.state_handler = None
+            self.active: set[int] = set()
+
+        def replace_source(self, source_id: int, **kwargs: object) -> bool:
+            self.active.add(source_id)
+            self.state_handler = kwargs["state_handler"]
+            return True
+
+        def remove_source(self, source_id: int) -> None:
+            self.removed.append(source_id)
+            self.active.discard(source_id)
+
+        def source_is_active(self, source_id: int) -> bool:
+            return source_id in self.active
+
+    fake = FakeMqtt()
+    snapshot = _make_settings()
+    monkeypatch.setattr(dsmr_ingest, "_subscriptions", {})
+    monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {})
+    monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {})
+    monkeypatch.setattr(dsmr_ingest, "_tariffs", {})
+    monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: SessionLocal)
+    monkeypatch.setattr("app.integrations.mqtt.mqtt_manager", fake)
+    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot])
+    with Session(engine) as session:
+        source = session.get(MeterSource, snapshot.source_id)
+        assert source is not None
+        source.enabled = True
+        source.status = "online"
+        session.commit()
+    dsmr_ingest.apply_dsmr_subscription()
+    assert fake.state_handler is not None
+
+    with Session(engine) as session:
+        source = session.get(MeterSource, snapshot.source_id)
+        assert source is not None
+        source.enabled = False
+        session.commit()
+    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [])
+    dsmr_ingest.apply_dsmr_subscription()
+    fake.state_handler("online")
+
+    with Session(engine) as session:
+        source = session.get(MeterSource, snapshot.source_id)
+        assert source is not None
+        assert source.status == "unknown"
+        assert source.last_error is None
+    assert fake.removed == [snapshot.source_id]
+
+
+def test_topic_collision_persists_error_when_reconcile_removes_active_client(dsmr_db, monkeypatch):
+    """A rejected enabled source cannot retain health from its removed client."""
+    engine, SessionLocal = dsmr_db
+    from app.services import dsmr_ingest
+
+    class FakeMqtt:
+        def __init__(self) -> None:
+            self.removed: list[int] = []
+            self.active: set[int] = set()
+
+        def replace_source(self, source_id: int, **kwargs: object) -> bool:
+            self.active.add(source_id)
+            return True
+
+        def remove_source(self, source_id: int) -> None:
+            self.removed.append(source_id)
+            self.active.discard(source_id)
+
+        def source_is_active(self, source_id: int) -> bool:
+            return source_id in self.active
+
+    fake = FakeMqtt()
+    valid = _make_settings(dsmr_mqtt_topic="dsmr/telegram")
+    collision = _make_settings(dsmr_mqtt_topic="dsmr/telegram")
+    collision = dsmr_ingest.DsmrSourceSnapshot(
+        collision.source_id,
+        collision.topic,
+        collision.topic,
+        collision.sample_interval_s,
+    )
+    monkeypatch.setattr(dsmr_ingest, "_subscriptions", {})
+    monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {})
+    monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {})
+    monkeypatch.setattr(dsmr_ingest, "_tariffs", {})
+    monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: SessionLocal)
+    monkeypatch.setattr("app.integrations.mqtt.mqtt_manager", fake)
+    with Session(engine) as session:
+        source = session.get(MeterSource, valid.source_id)
+        assert source is not None
+        source.enabled = True
+        source.status = "online"
+        session.commit()
+
+    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [valid])
+    dsmr_ingest.apply_dsmr_subscription()
+    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [collision])
+    dsmr_ingest.apply_dsmr_subscription()
+
+    with Session(engine) as session:
+        source = session.get(MeterSource, valid.source_id)
+        assert source is not None
+        assert source.enabled is True
+        assert source.status == "error"
+        assert source.last_error == "DSMR source configuration invalid."
+    assert fake.removed == [valid.source_id]
+    assert fake.active == set()
+
+
+def test_topic_collision_persists_error_on_startup_without_runtime_client(dsmr_db, monkeypatch):
+    """A collision corrects stale persisted online health without an installed client."""
+    engine, SessionLocal = dsmr_db
+    from app.services import dsmr_ingest
+
+    collision = dsmr_ingest.DsmrSourceSnapshot(1, "dsmr/telegram", "dsmr/telegram", 10)
+    monkeypatch.setattr(dsmr_ingest, "_subscriptions", {})
+    monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {})
+    monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {})
+    monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: SessionLocal)
+    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [collision])
+    with Session(engine) as session:
+        source = session.get(MeterSource, collision.source_id)
+        assert source is not None
+        source.enabled = True
+        source.status = "online"
+        session.commit()
+
+    dsmr_ingest.apply_dsmr_subscription()
+
+    with Session(engine) as session:
+        source = session.get(MeterSource, collision.source_id)
+        assert source is not None
+        assert source.status == "error"
+        assert source.last_error == "DSMR source configuration invalid."
+
+
+def test_topic_collision_does_not_change_a_disabled_source(dsmr_db, monkeypatch):
+    """A concurrent disable is not overwritten by collision error handling."""
+    engine, SessionLocal = dsmr_db
+    from app.services import dsmr_ingest
+
+    collision = dsmr_ingest.DsmrSourceSnapshot(1, "dsmr/telegram", "dsmr/telegram", 10)
+    monkeypatch.setattr(dsmr_ingest, "_subscriptions", {})
+    monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {})
+    monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {})
+    monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: SessionLocal)
+    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [collision])
+    with Session(engine) as session:
+        source = session.get(MeterSource, collision.source_id)
+        assert source is not None
+        source.enabled = False
+        source.status = "unknown"
+        source.last_error = None
+        session.commit()
+
+    dsmr_ingest.apply_dsmr_subscription()
+
+    with Session(engine) as session:
+        source = session.get(MeterSource, collision.source_id)
+        assert source is not None
+        assert source.status == "unknown"
+        assert source.last_error is None
+
+
 def test_second_10_persists(dsmr_db):
     """A telegram with second=10 (another 10s boundary) must also be persisted."""
     _, SessionLocal = dsmr_db

tests/test_dsmr_subscription_apply.py

@@ -46,6 +46,7 @@ def fake_mqtt(monkeypatch):
     monkeypatch.setattr("app.integrations.mqtt.mqtt_manager", fake)
     monkeypatch.setattr(dsmr_ingest, "_subscriptions", {})
     monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {})
+    monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {})
     monkeypatch.setattr(dsmr_ingest, "_tariffs", {})
     return fake
 
@@ -186,3 +187,17 @@ def test_same_snapshot_aba_rejects_retained_handler(fake_mqtt, monkeypatch):
     old_handler(b"stale")  # type: ignore[operator]
     fake_mqtt.handlers[1]["same"](b"fresh")  # type: ignore[operator]
     assert received == [(b"fresh", snapshot)]
+
+
+def test_source_state_handler_is_generation_scoped(fake_mqtt, monkeypatch):
+    snapshot = _source(1, "same", broker_host="one.test")
+    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot])
+    dsmr_ingest.apply_dsmr_subscription()
+    stale = fake_mqtt.replace_calls[-1][1]["state_handler"]
+    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [])
+    dsmr_ingest.apply_dsmr_subscription()
+    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot])
+    dsmr_ingest.apply_dsmr_subscription()
+
+    monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: (_ for _ in ()).throw(AssertionError()))
+    stale("error")  # type: ignore[operator]

tests/test_meter_source_api.py

@@ -302,6 +302,12 @@ def test_source_channel_binding_response_contract_and_discover_capabilities(auth
         _login(client)
         source = _create_source(client, config={"username": "private-user", "password": "private-secret"})
         channel_uuid = _add_channel(engine, source["uuid"])
+        with Session(engine) as session:
+            source_model = session.query(MeterSource).filter_by(uuid=source["uuid"]).one()
+            source_model.status = "online"
+            source_model.last_seen_at = datetime.now(UTC)
+            source_model.last_error = None
+            session.commit()
         listed = client.get("/api/energy/sources")
         assert listed.status_code == 200
         assert listed.json()["total"] >= 1
@@ -332,7 +338,7 @@ def test_source_channel_binding_response_contract_and_discover_capabilities(auth
             "uuid", "label", "suggested_commodity", "unit", "device_type", "latest_value",
             "latest_at", "latest_quality", "binding_count", "bound_meter_ids", "binding_summary",
         }
-        assert channels.json()["source_status"] == "unknown"
+        assert channels.json()["source_status"] == "online"
 
         meter = client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json={
             "label": "Contract meter", "started_at": "2030-01-01T00:00:00Z", "reason": "initial",

tests/test_mqtt_subscribe.py

@@ -437,6 +437,88 @@ def test_source_sync_connack_before_connect_returns_subscribes_all_topics() -> N
     assert client.subscribed == ["telegram/topic", "tariff/topic"]
 
 
+def test_source_health_tracks_connack_disconnect_and_ignores_stale_callbacks() -> None:
+    """A source is connecting until CONNACK, and old generations cannot rewrite health."""
+    manager = MqttManager()
+    old_client = MagicMock()
+    new_client = MagicMock()
+    old_states: list[str] = []
+    new_states: list[str] = []
+    kwargs = {
+        "host": "broker.test",
+        "port": 1883,
+        "username": "",
+        "password": "",
+        "tls_enabled": False,
+        "subscriptions": {"topic": lambda _payload: None},
+    }
+    with patch("app.integrations.mqtt.mqtt.Client", side_effect=[old_client, new_client]):
+        assert manager.replace_source(1, **kwargs, state_handler=old_states.append)
+        assert old_states == ["connecting"]
+        assert manager.replace_source(1, **kwargs, state_handler=new_states.append)
+
+    accepted = MagicMock()
+    accepted.is_failure = False
+    old_client.on_connect(old_client, None, MagicMock(), accepted, None)
+    old_client.on_disconnect(old_client, None, MagicMock(), MagicMock(), None)
+    assert old_states == ["connecting"]
+
+    new_client.on_connect(new_client, None, MagicMock(), accepted, None)
+    new_client.on_disconnect(new_client, None, MagicMock(), MagicMock(), None)
+    assert new_states == ["connecting", "online", "error"]
+
+
+def test_source_health_rejected_connack_reports_error_without_disconnect() -> None:
+    """A failed CONNACK is distinct from a later disconnect callback."""
+    manager = MqttManager()
+    client = MagicMock()
+    states: list[str] = []
+    with patch("app.integrations.mqtt.mqtt.Client", return_value=client):
+        assert manager.replace_source(
+            1, host="broker.test", port=1883, username="", password="", tls_enabled=False,
+            subscriptions={"topic": lambda _payload: None}, state_handler=states.append,
+        )
+
+    accepted = MagicMock()
+    accepted.is_failure = False
+    refused = MagicMock()
+    refused.is_failure = True
+    client.on_connect(client, None, MagicMock(), accepted, None)
+    client.on_connect(client, None, MagicMock(), refused, None)
+
+    assert states == ["connecting", "online", "error"]
+    assert 1 not in manager._source_connected
+
+
+def test_source_health_disconnect_then_reconnect_connack_is_ordered_and_isolated() -> None:
+    """One source's reconnect sequence cannot change another source's health."""
+    manager = MqttManager()
+    first_client = MagicMock()
+    second_client = MagicMock()
+    first_states: list[str] = []
+    second_states: list[str] = []
+    kwargs = {
+        "host": "broker.test",
+        "port": 1883,
+        "username": "",
+        "password": "",
+        "tls_enabled": False,
+        "subscriptions": {"topic": lambda _payload: None},
+    }
+    with patch("app.integrations.mqtt.mqtt.Client", side_effect=[first_client, second_client]):
+        assert manager.replace_source(1, **kwargs, state_handler=first_states.append)
+        assert manager.replace_source(2, **kwargs, state_handler=second_states.append)
+
+    accepted = MagicMock()
+    accepted.is_failure = False
+    first_client.on_connect(first_client, None, MagicMock(), accepted, None)
+    first_client.on_disconnect(first_client, None, MagicMock(), MagicMock(), None)
+    first_client.on_connect(first_client, None, MagicMock(), accepted, None)
+
+    assert first_states == ["connecting", "online", "error", "online"]
+    assert second_states == ["connecting"]
+
+
 def test_source_teardown_with_joining_loop_stop_waits_for_callback_before_aba() -> None:
     """loop_stop may join a callback that needs the manager lock to finish."""
     manager = MqttManager()