Merge remote-tracking branch 'github/dev' into feat/weather-service-localization

.eslintrc.json

@@ -7,6 +7,7 @@
   "settings": {
     "html/html-extensions": [".html"]
   },
+  "root": true,
   "rules": {
     "no-undef": "warn",
     "no-unused-vars": "warn",

CHANGELOG.md

@@ -24,6 +24,47 @@ semantic versioning.
 
 ### Fixed
 
+- `path` no longer answers "No path information available in current message" on a
+  busy mesh (#255). Verifying a channel message against the RF cache only ever
+  checked the newest row, which assumes the RF log row and the decoded CHAN event
+  for one reception arrive back to back with nothing in between. On a dense mesh
+  they do not: a repeater's echo of the very same packet is routinely logged in the
+  gap. The reporter's message was heard directly (`SNR 13.25`, 0 hops) and again via
+  repeater `f0` 185 ms later (`SNR 12.0`, 1 hop); both rows carry packet hash
+  `392926C85DCB87D0`, but the check saw only the echo, disagreed on path length and
+  SNR, and left the route unresolved — so the bot withheld a path it had decoded
+  correctly. The cache is now searched for the row the payload matches instead of
+  testing just the most recent one. Rows that agree must resolve to a single packet
+  hash, so two unrelated packets that happen to agree stay a fallback and #80's
+  guarantee is unchanged: a route is still only ever attributed on evidence. SNR and
+  RSSI now come from the message's own reception too, rather than from whichever
+  packet was heard last.
+
+- MQTT brokers no longer flap in a reconnect storm (#248). Three things stacked up.
+  First, the packet-capture watchdog ran `client.reconnect()` from its own thread
+  every 30 seconds whenever `is_connected()` was false — which includes every moment
+  paho's network thread is inside its own backoff. Two threads driving one client's
+  socket produced duplicate CONNACKs, spurious `MQTT_ERR_PROTOCOL` disconnects, and a
+  fixed-interval retry that flattened paho's 1→120s backoff into a hot loop. The
+  watchdog now observes, refreshes an expiring auth token so paho's next attempt can
+  succeed, and only intervenes when the network thread is gone and nothing is
+  retrying at all. Second, the generated client ID had no per-broker component, so
+  every broker in the process connected under the same ID; two hostnames belonging to
+  one cluster (`mqtt-a` and `mqtt-b` of the same service) evicted each other's session
+  on a six-second cycle. IDs are now distinct per broker. Third, a disconnect logged a
+  bare `rc=`, which reads against the CONNACK table even though paho reports
+  `MQTT_ERR_*` there — `rc=2` is a protocol error, not "client identifier rejected" —
+  so it now names the code.
+
+- A renewed MQTT auth token is now actually put in force. MQTT presents credentials
+  once, at CONNECT, so `username_pw_set` on a live session changed nothing and the
+  connection kept running on the token it was opened with until the broker evicted it
+  at that token's `exp`. Renewal is now followed by a clean, serialized reconnect
+  (`disconnect` → `loop_stop` → `reconnect` → `loop_start`, in that order, so the
+  network thread is joined before anything else touches the socket). Set
+  `mqttN_jwt_reconnect_on_renew = false` for a broker that ignores expiry on live
+  sessions.
+
 - Startup config-lint findings now go to the log file. They were printed to stderr
   before the bot (and therefore its logger) existed, so under systemd they reached
   only the journal and never `logs/meshcore_bot.log`. The linter had correctly
@@ -133,9 +174,14 @@ semantic versioning.
   and Configuration now sit under a single **Settings** gear menu, leaving Dashboard,
   Real-time, Contacts, Mesh Graph and Logs on the bar. The current page is highlighted,
   including the gear when a settings page is open.
+- Added notes on connecting to waev.app MQTT brokers to the `packet_capture.md` file.
 
 ### Added
 
+- `mqttN_keepalive` (default 60) sets the MQTT PINGREQ interval per broker. It was
+  hardcoded at 60 before, which is long for websockets through a proxy that drops
+  idle connections.
+
 - `hops_min:N` response-template filter, alongside `pathbytes_min:N`. It clears a
   field unless the message actually travelled at least N hops, so
   `{firstlast_distance|hops_min:1|prefix_if_nonempty: | F/L Dist: }` drops the whole

config.ini.example

@@ -1937,6 +1937,8 @@ observer_name =
 #                                     # Default false. Only for self-signed brokers on a
 #                                     # trusted network — otherwise credentials are MITMable.
 # mqttN_websocket_path = /mqtt        # WebSocket path (for websockets transport)
+# mqttN_keepalive = 60                # Seconds between PINGREQs. Lower (30) for websockets
+#                                     # through a proxy that drops idle connections.
 # mqttN_username =                     # MQTT username (optional, auto-generated for auth tokens)
 # mqttN_password =                     # MQTT password (optional, auto-generated for auth tokens)
 # mqttN_use_auth_token = true/false   # Use JWT auth token instead of username/password
@@ -1944,7 +1946,9 @@ observer_name =
 # mqttN_topic_status =                # Status topic template (uses placeholders below)
 # mqttN_topic_packets =               # Packets topic template (uses placeholders below)
 # mqttN_topic_prefix =                # Legacy topic prefix (fallback if topic_status/topic_packets not set)
-# mqttN_client_id =                   # MQTT client ID (optional, auto-generated from bot name)
+# mqttN_client_id =                   # MQTT client ID (optional; one is generated per broker).
+#                                     # Two brokers must never share an ID — brokers behind one
+#                                     # cluster (mqtt-a/mqtt-b) will evict each other's session.
 # mqttN_upload_packet_types =         # Comma-separated packet types to upload (e.g. 2,4); empty = all
 # mqttN_include_decoded = true/false  # Publish the decoded object to this broker (default: include_decoded)
 # mqttN_neighbors = true/false        # Publish the zero-hop neighbours snapshot to this broker.
@@ -1983,6 +1987,9 @@ mqtt1_upload_packet_types =
 # Optional per-broker JWT (inherit globals if omitted):
 # mqtt1_jwt_ttl_seconds = 3600           # JWT exp claim: iat + this many seconds
 # mqtt1_jwt_renewal_interval = 1800      # Refresh password this often; use < ttl; 0 = no renewal loop
+# mqtt1_jwt_reconnect_on_renew = true    # Reconnect after renewing so the fresh token is in force.
+#                                        # Default true. Brokers that enforce the JWT's expiry drop
+#                                        # the session otherwise. Off only if yours ignores expiry.
 
 # MQTT Broker 2 - Let's Mesh Analyzer (EU)
 mqtt2_enabled = true

docs/packet-capture.md

@@ -94,6 +94,20 @@ mqtt2_username = user
 mqtt2_password = pass
 ```
 
+#### Connection tuning
+
+| Key | Default | What it does |
+|-----|---------|--------------|
+| `mqttN_keepalive` | `60` | Seconds between PINGREQs. Lower it to `30` for websockets through a proxy that drops idle connections. |
+| `mqttN_client_id` | generated | MQTT client ID. One is generated per broker; set this only if the broker requires a fixed value. |
+
+**Two brokers must never share a client ID.** A broker evicts an existing session
+when a second connection arrives under the same ID, so brokers that sit behind one
+cluster — `mqtt-a.example` and `mqtt-b.example` of the same service — will kick each
+other off in a loop, seconds apart, forever. The generated IDs already differ per
+broker; you only reintroduce the problem by setting `mqtt1_client_id` and
+`mqtt2_client_id` to the same string.
+
 #### Filtering by packet type
 
 You can limit which packet types are uploaded to each broker with `mqttN_upload_packet_types`. Use a comma-separated list of type numbers; if unset or empty, all packet types are uploaded.
@@ -140,6 +154,13 @@ Two separate settings:
 - **`jwt_ttl_seconds`** (global) / **`mqttN_jwt_ttl_seconds`** (per broker): lifetime of the JWT in the `exp` claim (`exp = iat + ttl`). Use this when the broker enforces a maximum token lifetime (e.g. 60 minutes → `3600`).
 - **`jwt_renewal_interval`** (global) / **`mqttN_jwt_renewal_interval`** (per broker): how often the bot refreshes the MQTT password for that broker. Set **less than** the TTL (e.g. TTL 3600s and renewal every 1800s) so the connection does not outlive the token.
 
+- **`mqttN_jwt_reconnect_on_renew`** (per broker, default `true`): reconnect right
+  after minting a new token. MQTT presents credentials once, at CONNECT, so a
+  renewed token does nothing for a session that is already open — brokers that
+  enforce the JWT's `exp` drop that session the moment the *original* token
+  expires. Reconnecting on renewal turns that eviction into one clean, scheduled
+  reconnect. Turn it off only if your broker ignores expiry on live sessions.
+
 Per-broker keys override the global values for that broker only. Omit them to inherit globals.
 
 ```ini
@@ -154,6 +175,13 @@ jwt_renewal_interval = 43200      # Default proactive refresh cadence (12 hours)
 # mqtt1_jwt_renewal_interval = 1800
 ```
 
+**Note**: When connecting to waev.app brokers the default settings will cause the connection not to authenticate properly. Please use the following settings on the MQTT connection for the waev.app brokers.
+
+```ini
+mqttN_jwt_ttl_seconds = 3600
+mqttN_jwt_renewal_interval = 3500
+```
+
 ---
 
 ## Packet Format
@@ -268,6 +296,27 @@ Common issues:
 3. **Check authentication** - Verify JWT token generation
 4. **Check logs** - Look for connection errors
 
+### MQTT Connecting and Disconnecting in a Loop
+
+Repeated `Disconnected from MQTT broker ... (rc=7: The connection was lost.)`
+followed immediately by `✓ Connected to MQTT broker`, over and over:
+
+1. **Check for a shared client ID** — if two brokers alternate (one connects as the
+   other drops, every few seconds), they are almost certainly one cluster behind two
+   hostnames, evicting each other's session. Give them distinct `mqttN_client_id`
+   values, or leave the key empty so one is generated per broker.
+2. **Check whether it lines up with your JWT TTL** — roughly one disconnect per
+   broker per `mqttN_jwt_ttl_seconds` means the broker is enforcing token expiry.
+   Leave `mqttN_jwt_reconnect_on_renew` on and set the renewal interval below the
+   TTL so the reconnect happens on your schedule instead of theirs.
+3. **Check for a second client on the same identity** — another capture tool running
+   against the same brokers with the same node public key can compete with the bot.
+4. **Lower `mqttN_keepalive`** to `30` if you are on websockets through a proxy.
+
+Note that `rc=` on a *disconnect* is a paho `MQTT_ERR_*` code, not a CONNACK code:
+`rc=7` is a lost connection and `rc=2` is a protocol error. They do not mean the same
+thing as the numbers in a `Failed to connect` line.
+
 ### No Packets Being Published
 
 1. **Verify MQTT connection** - Check logs for "Connected to MQTT broker"

modules/message_handler.py

@@ -1540,11 +1540,21 @@ class MessageHandler:
 
         # A channel message has no packet prefix or pubkey to match on, so everything
         # above lands on the most-recent-packet fallback. The decoded payload carries
-        # its own copies of fields the RF row also has, though, so the fallback can be
-        # checked rather than assumed.
+        # its own copies of fields the RF row also has, though, so the cache can be
+        # searched for this message's own packet rather than assuming the newest row.
         if recent_rf_data is not None and not rf_data_is_correlated(recent_rf_data):
-            if self._rf_data_matches_chan_payload(recent_rf_data, payload):
-                recent_rf_data = {**recent_rf_data, RF_MATCH_KEY: RF_MATCH_PAYLOAD}
+            verified = self._find_rf_row_matching_chan_payload(
+                payload, scope_eligible_only=scope_eligible_only
+            )
+            if verified is not None:
+                if verified.get("packet_prefix") != recent_rf_data.get("packet_prefix"):
+                    self.logger.debug(
+                        "Most recent RF row %s is not this message's packet (a later "
+                        "reception of another packet); the payload matches %s instead",
+                        (recent_rf_data.get("packet_prefix") or "?")[:16],
+                        (verified.get("packet_prefix") or "?")[:16],
+                    )
+                recent_rf_data = {**verified, RF_MATCH_KEY: RF_MATCH_PAYLOAD}
                 self.logger.debug(
                     "Verified RF row %s against the channel payload (GRP_TXT, path_len=%s, "
                     "SNR=%s); treating it as this message's packet",
@@ -1555,6 +1565,48 @@ class MessageHandler:
 
         return recent_rf_data
 
+    def _find_rf_row_matching_chan_payload(
+        self, payload: dict[str, Any] | None, *, scope_eligible_only: bool = False
+    ) -> dict[str, Any] | None:
+        """Return the cached RF row this channel message was received on, or None.
+
+        Checking only the newest row assumes the RF log row and the decoded CHAN
+        event for one reception arrive back to back with nothing in between. On a
+        dense mesh they do not: a repeater's echo of the very same packet is
+        routinely logged in the gap, so the newest row is that echo — a different
+        path length and a different measured SNR — and the message loses its route
+        even though its own row is sitting in the cache (#255). Search the cache
+        instead, and let _rf_data_matches_chan_payload decide which row is ours.
+
+        Two rows can only both match when they are receptions of the same packet
+        (same hash) that also agree on path length and SNR; anything else is
+        ambiguous and keeps the fallback tag, because a guess is what #80 cost.
+        """
+        if not payload:
+            return None
+
+        matches = [
+            row
+            for row in self.recent_rf_data
+            if self._rf_data_matches_chan_payload(row, payload)
+            and (not scope_eligible_only or self._is_rf_data_scope_eligible(row))
+        ]
+        if not matches:
+            return None
+
+        if len(matches) > 1:
+            hashes = {row.get("packet_hash") for row in matches}
+            if len(hashes) != 1 or not all(hashes):
+                self.logger.debug(
+                    "%d cached RF rows agree with the channel payload across %d packet(s); "
+                    "leaving the route unresolved rather than guessing",
+                    len(matches),
+                    len(hashes),
+                )
+                return None
+
+        return max(matches, key=lambda row: row["timestamp"])
+
     def _rf_data_matches_chan_payload(
         self, rf_data: dict[str, Any] | None, payload: dict[str, Any] | None
     ) -> bool:
@@ -1573,6 +1625,10 @@ class MessageHandler:
         a checked hypothesis. SNR is the discriminating one: it is the measured
         value of a single reception, quantised to 0.25 dB, so an unrelated packet
         matching all three is improbable rather than merely unlikely.
+
+        _find_rf_row_matching_chan_payload applies this to every cached row rather
+        than only the newest, so a repeater echo logged in between cannot displace
+        the message's own row (#255).
         """
         if not rf_data or not payload:
             return False

modules/service_plugins/packet_capture_service.py

@@ -275,6 +275,11 @@ class PacketCaptureService(BaseServicePlugin):
                  "help": "Authenticate with a signed JWT instead of username/password."},
                 {"key": "token_audience", "label": "Token audience", "type": "str", "default": "",
                  "help": "JWT audience claim (when using auth token)."},
+                {"key": "jwt_reconnect_on_renew", "label": "Reconnect on token renewal",
+                 "type": "bool", "default": True,
+                 "help": "Reconnect right after renewing the token, so the fresh one is in "
+                         "force. Brokers that enforce the JWT's expiry drop the session "
+                         "otherwise. Turn off only if your broker ignores expiry."},
                 {"key": "topic_status", "label": "Status topic", "type": "str", "default": ""},
                 {"key": "topic_packets", "label": "Packets topic", "type": "str", "default": ""},
                 {"key": "neighbors", "label": "Publish neighbours", "type": "bool", "default": True,
@@ -286,7 +291,13 @@ class PacketCaptureService(BaseServicePlugin):
                          "'neighbors', else <topic prefix>/neighbors."},
                 {"key": "websocket_path", "label": "WebSocket path", "type": "str", "default": "/mqtt",
                  "help": "Path when transport is websockets."},
-                {"key": "client_id", "label": "Client ID", "type": "str", "default": ""},
+                {"key": "client_id", "label": "Client ID", "type": "str", "default": "",
+                 "help": "Leave empty to generate one per broker. Set explicitly only if the "
+                         "broker requires a fixed ID; two brokers must never share one."},
+                {"key": "keepalive", "label": "Keepalive", "type": "int", "min": 5, "max": 3600,
+                 "default": 60,
+                 "help": "Seconds between PINGREQs. Lower it (30) for websockets through a proxy "
+                         "that drops idle connections."},
                 {"key": "upload_packet_types", "label": "Upload packet types", "type": "str", "default": "",
                  "help": "Comma-separated type numbers (e.g. 2,4). Empty = upload all."},
             ],
@@ -742,6 +753,7 @@ class PacketCaptureService(BaseServicePlugin):
                 ),
                 "broker_num": broker_num,
                 "websocket_path": config.get("PacketCapture", f"mqtt{broker_num}_websocket_path", fallback="/mqtt"),
+                "keepalive": config.getint("PacketCapture", f"mqtt{broker_num}_keepalive", fallback=60),
                 "client_id": config.get("PacketCapture", f"mqtt{broker_num}_client_id", fallback=None),
                 "upload_packet_types": upload_packet_types,
                 "include_decoded": config.getboolean(
@@ -751,6 +763,9 @@ class PacketCaptureService(BaseServicePlugin):
                 ),
                 "jwt_renewal_interval": jwt_renewal_interval,
                 "jwt_ttl_seconds": jwt_ttl_seconds,
+                "jwt_reconnect_on_renew": config.getboolean(
+                    "PacketCapture", f"mqtt{broker_num}_jwt_reconnect_on_renew", fallback=True
+                ),
             }
 
             # Set default topic_prefix if not set
@@ -839,6 +854,19 @@ class PacketCaptureService(BaseServicePlugin):
         """UTC ISO 8601 timestamp with Z suffix for broad consumer compatibility."""
         return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
 
+    @staticmethod
+    def _disconnect_reason(rc: int) -> str:
+        """Describe an on_disconnect return code.
+
+        paho reports MQTT_ERR_* here, not CONNACK codes, so a bare number reads
+        as the wrong thing entirely: rc=2 is a protocol error, not "client
+        identifier rejected".
+        """
+        try:
+            return f"rc={rc}: {mqtt.error_string(rc)}"
+        except Exception:
+            return f"rc={rc}"
+
     @staticmethod
     def _jwt_ttl_log_phrase(ttl_seconds: int) -> str:
         """Short TTL description for log lines."""
@@ -1632,7 +1660,11 @@ class PacketCaptureService(BaseServicePlugin):
                 if not client_id:
                     # Sanitize bot name for MQTT client ID (alphanumeric and hyphens only)
                     safe_name = "".join(c if c.isalnum() or c == "-" else "-" for c in bot_name)
-                    client_id = f"{safe_name}-packet-capture-{os.getpid()}"
+                    # The broker number keeps the ID distinct per broker. Two hosts that
+                    # share a session store — mqtt-a and mqtt-b of one cluster — evict
+                    # each other's session when both connect under the same ID, which
+                    # reads in the log as the two brokers flapping in lockstep.
+                    client_id = f"{safe_name}-packet-capture-{broker_config.get('broker_num', 1)}-{os.getpid()}"
 
                 # Create client based on transport type
                 transport = broker_config.get("transport", "tcp").lower()
@@ -1680,6 +1712,7 @@ class PacketCaptureService(BaseServicePlugin):
                 # Set username/password if provided
                 username = broker_config.get("username")
                 password = broker_config.get("password")
+                token_exp = None
 
                 if broker_config.get("use_auth_token"):
                     # Use auth token with audience if specified
@@ -1738,6 +1771,7 @@ class PacketCaptureService(BaseServicePlugin):
                         )
                         if token:
                             password = token
+                            token_exp = exp
                             ttl_phrase = self._jwt_ttl_log_phrase(ttl_used)
                             self.logger.debug(
                                 f"Created auth token for {broker_config['host']} "
@@ -1769,11 +1803,14 @@ class PacketCaptureService(BaseServicePlugin):
                         for mqtt_info in self.mqtt_clients:
                             if mqtt_info["client"] == client:
                                 mqtt_info["connected"] = True
+                                mqtt_info["down_since"] = None
+                                mqtt_info["stall_warned"] = False
                                 break
                         # Set global connected flag if any broker is connected
                         self.mqtt_connected = any(m.get("connected", False) for m in self.mqtt_clients)
                     else:
-                        # MQTT error codes: 0=success, 1=protocol, 2=client, 3=network, 4=transport, 5=auth
+                        # CONNACK return codes (MQTT 3.1.1 §3.2.2.3). These are NOT the
+                        # MQTT_ERR_* codes on_disconnect reports — see _disconnect_reason.
                         error_messages = {
                             1: "protocol version rejected",
                             2: "client identifier rejected",
@@ -1799,7 +1836,9 @@ class PacketCaptureService(BaseServicePlugin):
                             cfg = mqtt_info["config"]
                             host = cfg["host"]
                             if rc != 0:
-                                self.logger.warning(f"Disconnected from MQTT broker {host} (rc={rc})")
+                                self.logger.warning(
+                                    f"Disconnected from MQTT broker {host} ({self._disconnect_reason(rc)})"
+                                )
                             else:
                                 self.logger.debug(f"Disconnected from MQTT broker {host}")
                             break
@@ -1813,6 +1852,7 @@ class PacketCaptureService(BaseServicePlugin):
                 try:
                     host = broker_config["host"]
                     port = broker_config["port"]
+                    keepalive = int(broker_config.get("keepalive", 60))
 
                     # Validate hostname (basic check)
                     if not host or not host.strip():
@@ -1839,6 +1879,15 @@ class PacketCaptureService(BaseServicePlugin):
                             "client": client,
                             "config": broker_config,
                             "connected": False,  # Track connection status per broker
+                            # Expiry of the credential currently set on the client, so
+                            # the monitor can refresh it before paho's next retry.
+                            "token_exp": token_exp,
+                            # Set when the client first goes down; cleared on connect.
+                            "down_since": None,
+                            "stall_warned": False,
+                            # Serializes disconnect/loop_stop/reconnect/loop_start so
+                            # the renewal task and the monitor cannot overlap.
+                            "cycle_lock": asyncio.Lock(),
                         }
                     )
 
@@ -1852,7 +1901,7 @@ class PacketCaptureService(BaseServicePlugin):
                         # Run connect in executor to avoid blocking the event loop
                         loop = asyncio.get_event_loop()
                         try:
-                            await loop.run_in_executor(None, client.connect, host, port, 60)
+                            await loop.run_in_executor(None, client.connect, host, port, keepalive)
                         except Exception as connect_error:
                             # Connection failed, but don't block - let loop_start handle retries
                             self.logger.debug(f"Initial connect() call failed (non-blocking): {connect_error}")
@@ -1863,7 +1912,7 @@ class PacketCaptureService(BaseServicePlugin):
                         # Run connect in executor to avoid blocking the event loop
                         loop = asyncio.get_event_loop()
                         try:
-                            await loop.run_in_executor(None, client.connect, host, port, 60)
+                            await loop.run_in_executor(None, client.connect, host, port, keepalive)
                         except Exception as connect_error:
                             # Connection failed, but don't block - let loop_start handle retries
                             self.logger.debug(f"Initial connect() call failed (non-blocking): {connect_error}")
@@ -2895,14 +2944,21 @@ class PacketCaptureService(BaseServicePlugin):
             except Exception as e:
                 self.logger.error(f"Error publishing status to MQTT: {e}")
 
-    async def _renew_mqtt_auth_token(self, mqtt_client_info: dict[str, Any]) -> None:
-        """Mint a new auth token and apply it to one MQTT client (per-broker TTL)."""
+    async def _renew_mqtt_auth_token(self, mqtt_client_info: dict[str, Any]) -> bool:
+        """Mint a new auth token and apply it to one MQTT client (per-broker TTL).
+
+        Only updates the credentials used by the next CONNECT; it does not touch
+        the socket, so it is safe to call while paho's network thread is running.
+
+        Returns:
+            bool: True if a new token was minted and applied.
+        """
         config = mqtt_client_info["config"]
         client = mqtt_client_info["client"]
         broker_host = config.get("host", "unknown")
 
         if not config.get("use_auth_token"):
-            return
+            return False
 
         self.logger.debug(f"Renewing auth token for MQTT broker {broker_host}...")
 
@@ -2924,7 +2980,7 @@ class PacketCaptureService(BaseServicePlugin):
 
         if not device_public_key_hex:
             self.logger.warning(f"No device public key available for token renewal (broker: {broker_host})")
-            return
+            return False
 
         token_audience = config.get("token_audience") or broker_host
         username = f"v1_{device_public_key_hex.upper()}"
@@ -2950,12 +3006,14 @@ class PacketCaptureService(BaseServicePlugin):
 
             if token:
                 client.username_pw_set(username, token)
+                mqtt_client_info["token_exp"] = exp
                 ttl_phrase = self._jwt_ttl_log_phrase(ttl_used)
                 self.logger.info(f"✓ Renewed auth token for MQTT broker {broker_host} (TTL {ttl_phrase})")
-            else:
-                self.logger.warning(f"Failed to renew auth token for MQTT broker {broker_host}")
+                return True
+            self.logger.warning(f"Failed to renew auth token for MQTT broker {broker_host}")
         except Exception as e:
             self.logger.error(f"Error renewing token for MQTT broker {broker_host}: {e}")
+        return False
 
     async def jwt_renewal_scheduler_for_client(self, mqtt_client_info: dict[str, Any]) -> None:
         """Background task: renew JWT on one broker every config jwt_renewal_interval seconds."""
@@ -2972,7 +3030,17 @@ class PacketCaptureService(BaseServicePlugin):
                     break
                 if not config.get("use_auth_token"):
                     continue
-                await self._renew_mqtt_auth_token(mqtt_client_info)
+                renewed = await self._renew_mqtt_auth_token(mqtt_client_info)
+                # username_pw_set only affects the next CONNECT, so without this the
+                # session keeps running on the token it was opened with. Brokers that
+                # enforce the JWT's exp then evict us mid-stream (issue #248); cycling
+                # here turns that into one clean, scheduled reconnect instead.
+                if (
+                    renewed
+                    and config.get("jwt_reconnect_on_renew", True)
+                    and mqtt_client_info["client"].is_connected()
+                ):
+                    await self._cycle_mqtt_client(mqtt_client_info, "auth token renewed")
             except asyncio.CancelledError:
                 break
             except Exception as e:
@@ -3005,16 +3073,140 @@ class PacketCaptureService(BaseServicePlugin):
                 self.logger.error(f"Error in health check loop: {e}")
                 await asyncio.sleep(60)
 
-    async def mqtt_reconnection_monitor(self) -> None:
-        """Proactive MQTT reconnection monitor - checks and reconnects disconnected brokers.
+    async def _cycle_mqtt_client(self, mqtt_client_info: dict[str, Any], reason: str) -> bool:
+        """Take one MQTT client down and bring it back up, serialized.
 
-        Periodically checks connectivity of all configured MQTT brokers and attempts
-        reconnection if disconnected.
+        The steps must not overlap. disconnect() lets paho's network thread wind
+        down cleanly, loop_stop() joins it, and only then is it safe for this
+        thread to drive reconnect(). Calling reconnect() while that thread is
+        still running is what produces two sockets on one client.
+
+        Returns:
+            bool: True if the client is connected when this returns.
+        """
+        client = mqtt_client_info["client"]
+        broker_host = mqtt_client_info["config"].get("host", "unknown")
+        loop = asyncio.get_event_loop()
+
+        async with mqtt_client_info["cycle_lock"]:
+            self.logger.info(f"Cycling MQTT connection to {broker_host} ({reason})")
+            try:
+                await loop.run_in_executor(None, client.disconnect)
+            except Exception as e:
+                self.logger.debug(f"disconnect() during cycle of {broker_host}: {e}")
+            try:
+                await loop.run_in_executor(None, client.loop_stop)
+            except Exception as e:
+                self.logger.debug(f"loop_stop() during cycle of {broker_host}: {e}")
+
+            try:
+                await loop.run_in_executor(None, client.reconnect)
+            except Exception as e:
+                # Not fatal: loop_start() below hands the retries back to paho.
+                self.logger.debug(f"reconnect() during cycle of {broker_host} failed: {e}")
+
+            try:
+                client.loop_start()
+            except Exception as e:
+                self.logger.warning(f"Could not restart network loop for {broker_host}: {e}")
+                return False
+
+            # CONNACK arrives on the network thread; give it a moment to land.
+            await asyncio.sleep(2)
+            connected = client.is_connected()
+            mqtt_client_info["connected"] = connected
+            self.mqtt_connected = any(m.get("connected", False) for m in self.mqtt_clients)
+            if connected:
+                self.logger.info(f"✓ Reconnected to MQTT broker {broker_host}")
+            else:
+                self.logger.debug(f"Cycle of {broker_host} did not connect yet; paho will keep retrying")
+            return connected
+
+    @staticmethod
+    def _loop_thread_alive(client) -> bool:
+        """Whether paho's network thread is still running for this client.
+
+        While it lives, that thread owns reconnection and nothing else may drive
+        the socket. Reading the private attribute is deliberate: paho 1.x exposes
+        no public equivalent, and guessing wrong here is what caused the storm.
+        """
+        thread = getattr(client, "_thread", None)
+        if thread is None:
+            return False
+        try:
+            return bool(thread.is_alive())
+        except Exception:
+            return True
+
+    # How long a broker may stay down before the log says so at warning level.
+    MQTT_STALL_WARN_AFTER = 300
+    # Refresh the token this far ahead of its expiry, so paho's next retry carries
+    # a credential the broker will still accept.
+    MQTT_TOKEN_REFRESH_MARGIN = 120
+
+    async def _check_mqtt_client(self, mqtt_client_info: dict[str, Any], now: float) -> None:
+        """One watchdog pass over a single client.
+
+        Observes, keeps the auth token fresh for paho's next attempt, and
+        intervenes only when paho's network thread is gone. It must never call
+        connect()/reconnect() while that thread is alive.
+        """
+        client = mqtt_client_info["client"]
+        config = mqtt_client_info["config"]
+        broker_host = config.get("host", "unknown")
+
+        if client.is_connected():
+            mqtt_client_info["connected"] = True
+            mqtt_client_info["down_since"] = None
+            mqtt_client_info["stall_warned"] = False
+            return
+
+        mqtt_client_info["connected"] = False
+        down_since = mqtt_client_info.get("down_since")
+        if down_since is None:
+            down_since = now
+            mqtt_client_info["down_since"] = down_since
+        downtime = now - down_since
+
+        # Keep the credential valid so paho's retries can succeed. This only sets
+        # the fields used by the next CONNECT; it does not touch the socket, so it
+        # is safe alongside the network thread.
+        if config.get("use_auth_token"):
+            token_exp = mqtt_client_info.get("token_exp")
+            if token_exp is None or token_exp - now < self.MQTT_TOKEN_REFRESH_MARGIN:
+                await self._renew_mqtt_auth_token(mqtt_client_info)
+
+        if self._loop_thread_alive(client):
+            # paho is retrying with backoff. Say so once when the outage stops
+            # looking transient, then stay quiet.
+            if downtime >= self.MQTT_STALL_WARN_AFTER and not mqtt_client_info.get("stall_warned"):
+                self.logger.warning(
+                    f"MQTT broker {broker_host} has been disconnected for "
+                    f"{int(downtime // 60)}m; paho is still retrying"
+                )
+                mqtt_client_info["stall_warned"] = True
+            elif self.debug:
+                self.logger.debug(
+                    f"MQTT broker {broker_host} disconnected for {int(downtime)}s (paho retrying)"
+                )
+            return
+
+        # No network thread: nothing is retrying, so this one is ours to fix.
+        self.logger.info(f"MQTT network loop for {broker_host} is not running, restarting it")
+        await self._cycle_mqtt_client(mqtt_client_info, "network loop stopped")
+
+    async def mqtt_reconnection_monitor(self) -> None:
+        """Watch the MQTT clients and recover only what paho cannot recover itself.
+
+        paho owns reconnection here: reconnect_delay_set() plus loop_start() give
+        each client a network thread that retries with backoff. This loop must not
+        call connect()/reconnect() alongside that thread — two threads driving one
+        client's socket produce duplicate CONNACKs, spurious protocol errors, and a
+        fixed-interval retry that flattens paho's backoff into a storm (issue #248).
         """
         if not self.mqtt_enabled:
             return
 
-        # Reconnection check interval (check every 30 seconds)
         check_interval = 30
 
         while not self.should_exit:
@@ -3024,102 +3216,17 @@ class PacketCaptureService(BaseServicePlugin):
                 if not self.mqtt_clients:
                     continue
 
-                # Check each broker's connection status
+                now = time.time()
                 for mqtt_client_info in self.mqtt_clients:
-                    client = mqtt_client_info["client"]
-                    config = mqtt_client_info["config"]
-                    broker_host = config.get("host", "unknown")
+                    try:
+                        await self._check_mqtt_client(mqtt_client_info, now)
+                    except asyncio.CancelledError:
+                        raise
+                    except Exception as e:
+                        host = mqtt_client_info["config"].get("host", "unknown")
+                        self.logger.debug(f"Error checking MQTT broker {host}: {e}")
 
-                    # Check if client is connected
-                    if not client.is_connected():
-                        # Client is disconnected - attempt reconnection
-                        try:
-                            self.logger.info(f"MQTT broker {broker_host} is disconnected, attempting reconnection...")
-
-                            # If using auth tokens, try to renew the token before reconnecting
-                            if config.get("use_auth_token"):
-                                # Get device's public key for username
-                                device_public_key_hex = None
-                                if self.meshcore and hasattr(self.meshcore, "self_info"):
-                                    try:
-                                        self_info = self.meshcore.self_info
-                                        if isinstance(self_info, dict):
-                                            device_public_key_hex = self_info.get("public_key", "")
-                                        elif hasattr(self_info, "public_key"):
-                                            device_public_key_hex = self_info.public_key
-
-                                        # Convert to hex string if bytes
-                                        if isinstance(device_public_key_hex, bytes):
-                                            device_public_key_hex = device_public_key_hex.hex()
-                                        elif isinstance(device_public_key_hex, bytearray):
-                                            device_public_key_hex = bytes(device_public_key_hex).hex()
-                                    except Exception:
-                                        pass
-
-                                if device_public_key_hex:
-                                    # Create new auth token
-                                    token_audience = config.get("token_audience") or broker_host
-                                    username = f"v1_{device_public_key_hex.upper()}"
-
-                                    use_device = (
-                                        self.auth_token_method == "device"
-                                        and self.meshcore
-                                        and self.meshcore.is_connected
-                                    )
-                                    meshcore_for_key_fetch = (
-                                        self.meshcore if self.meshcore and self.meshcore.is_connected else None
-                                    )
-
-                                    try:
-                                        iat, exp = self._auth_token_iat_exp(config)
-                                        token = await create_auth_token_async(
-                                            meshcore_instance=meshcore_for_key_fetch,
-                                            public_key_hex=device_public_key_hex,
-                                            private_key_hex=self.private_key_hex,
-                                            iata=self.global_iata,
-                                            timestamp=iat,
-                                            audience=token_audience,
-                                            exp=exp,
-                                            owner_public_key=self.owner_public_key,
-                                            owner_email=self.owner_email,
-                                            use_device=use_device,
-                                        )
-                                        if token:
-                                            # Update credentials
-                                            client.username_pw_set(username, token)
-                                            self.logger.debug(
-                                                f"Renewed auth token for {broker_host} before reconnection"
-                                            )
-                                    except Exception as e:
-                                        self.logger.debug(f"Error renewing auth token for {broker_host}: {e}")
-
-                            # Attempt reconnection (non-blocking to avoid blocking event loop)
-                            config["host"]
-                            config["port"]
-                            loop = asyncio.get_event_loop()
-                            try:
-                                await loop.run_in_executor(None, client.reconnect)
-                            except Exception as reconnect_error:
-                                # Reconnection failed, but don't block - will retry on next cycle
-                                self.logger.debug(f"Reconnect() call failed (non-blocking): {reconnect_error}")
-
-                            # Give it a moment to connect
-                            await asyncio.sleep(2)
-
-                            # Check if reconnection succeeded
-                            if client.is_connected():
-                                self.logger.info(f"✓ Successfully reconnected to MQTT broker {broker_host}")
-                                mqtt_client_info["connected"] = True
-                                # Update global flag
-                                self.mqtt_connected = any(m.get("connected", False) for m in self.mqtt_clients)
-                            else:
-                                if self.debug:
-                                    self.logger.debug(
-                                        f"Reconnection attempt to {broker_host} still in progress or failed"
-                                    )
-
-                        except Exception as e:
-                            self.logger.debug(f"Error attempting MQTT reconnection to {broker_host}: {e}")
+                self.mqtt_connected = any(m.get("connected", False) for m in self.mqtt_clients)
 
             except asyncio.CancelledError:
                 break

modules/web_viewer/app.py

@@ -1256,11 +1256,16 @@ class BotDataViewer:
                 "style-src 'self' 'unsafe-inline' "
                 "https://cdn.jsdelivr.net https://cdnjs.cloudflare.com https://unpkg.com "
                 "https://fonts.googleapis.com; "
-                "img-src 'self' data: https://*.tile.openstreetmap.org "
-                "https://*.basemaps.cartocdn.com "
+                "img-src 'self' data: blob: https://*.tile.openstreetmap.org "
+                "https://tiles.openfreemap.org "
                 "https://unpkg.com https://cdn.jsdelivr.net https://cdnjs.cloudflare.com; "
-                "connect-src 'self' ws: wss: "
+                "connect-src 'self' ws: wss: https://tiles.openfreemap.org "
                 "https://cdn.jsdelivr.net https://cdnjs.cloudflare.com https://unpkg.com; "
+                # MapLibre GL runs its renderer in a worker spawned from a blob: URL.
+                # Without these it falls back to default-src 'self' and the dark
+                # basemap fails to start.
+                "worker-src 'self' blob:; "
+                "child-src 'self' blob:; "
                 "font-src 'self' https://cdn.jsdelivr.net https://cdnjs.cloudflare.com "
                 "https://fonts.gstatic.com"
             )

modules/web_viewer/templates/mesh.html

@@ -342,6 +342,8 @@
 
 <!-- Leaflet CSS -->
 <link rel="stylesheet" href="https://unpkg.com/leaflet@1.9.4/dist/leaflet.css" />
+<!-- MapLibre GL CSS (dark basemap is a vector style; see TILE_LAYERS) -->
+<link rel="stylesheet" href="https://unpkg.com/maplibre-gl@5.6.0/dist/maplibre-gl.css" />
 <!-- vis-network CSS -->
 <link rel="stylesheet" href="https://unpkg.com/vis-network@9.1.9/styles/vis-network.min.css" />
 
@@ -524,6 +526,16 @@
     .mesh-dark .leaflet-container {
         background: #1a1a1a;
     }
+    /* Raster fallback only (see setBaseTileLayer): invert the OSM tiles the way
+       openstreetmap.org itself did for dark mode.  Scoped to the tiles so node
+       markers, edges and popups keep their real colors.
+       This must target .leaflet-tile and not the enclosing .leaflet-tile-container:
+       the container is a 0x0 element that its absolutely-positioned tiles overflow,
+       so a filter there has an empty reference box and paints nothing.  Every
+       function here is pointwise, so filtering per tile leaves no visible seams. */
+    .mesh-basemap-fallback .leaflet-tile {
+        filter: invert(1) hue-rotate(180deg) brightness(0.95) contrast(0.9);
+    }
     .mesh-dark .leaflet-popup-content-wrapper,
     .mesh-dark .leaflet-popup-tip {
         background: #2d2d2d;
@@ -621,6 +633,10 @@
 
 <!-- Leaflet JS -->
 <script src="https://unpkg.com/leaflet@1.9.4/dist/leaflet.js"></script>
+<!-- MapLibre GL + Leaflet bridge, for the vector dark basemap. Must load after Leaflet:
+     the bridge binds against both globals. -->
+<script src="https://unpkg.com/maplibre-gl@5.6.0/dist/maplibre-gl.js"></script>
+<script src="https://unpkg.com/@maplibre/maplibre-gl-leaflet@0.1.4/leaflet-maplibre-gl.js"></script>
 <!-- vis-network JS -->
 <script src="https://unpkg.com/vis-network@9.1.9/standalone/umd/vis-network.min.js"></script>
 
@@ -785,25 +801,67 @@
         return meshThemeOverride || getSiteTheme();
     }
 
+    // Light is OSM raster.  Dark is OpenFreeMap's vector 'dark' style rendered through
+    // MapLibre GL: CARTO's raster dark_all now requires an API key and is being retired,
+    // and OSM itself hosts no dark tiles (its Standard layer is light only).
     const TILE_LAYERS = {
         light: {
             url: 'https://{s}.tile.openstreetmap.org/{z}/{x}/{y}.png',
             attribution: '© OpenStreetMap contributors'
         },
         dark: {
-            url: 'https://{s}.basemaps.cartocdn.com/dark_all/{z}/{x}/{y}{r}.png',
-            attribution: '© OpenStreetMap contributors © CARTO'
+            styleUrl: 'https://tiles.openfreemap.org/styles/dark',
+            // Used only by the raster fallback below; the vector style carries its own
+            // attribution, which the bridge reads off the style once it loads.
+            url: 'https://{s}.tile.openstreetmap.org/{z}/{x}/{y}.png',
+            attribution: '© OpenStreetMap contributors'
         }
     };
     let baseTileLayer = null;
 
+    // The vector basemap needs both the bridge and a working WebGL context.  Either can be
+    // missing (CDN blocked, software renderer, GPU blocklist), and a blank map is a worse
+    // failure than a filtered one, so fall back to inverted OSM raster tiles.
+    let vectorBasemapSupported = null; // cached: the probe allocates a GL context
+    function canUseVectorBasemap() {
+        if (vectorBasemapSupported !== null) return vectorBasemapSupported;
+        vectorBasemapSupported = false;
+        try {
+            if (typeof L.maplibreGL === 'function' && typeof maplibregl !== 'undefined') {
+                // MapLibre GL dropped maplibregl.supported() in v3 and requires WebGL 2
+                // as of v5, so probe for a webgl2 context directly.
+                vectorBasemapSupported = Boolean(
+                    document.createElement('canvas').getContext('webgl2')
+                );
+            }
+        } catch (error) {
+            console.debug('Vector basemap unavailable, using raster fallback:', error);
+        }
+        return vectorBasemapSupported;
+    }
+
     function setBaseTileLayer() {
         if (!map) return;
-        const cfg = TILE_LAYERS[getMeshTheme()];
+        const theme = getMeshTheme();
+        const cfg = TILE_LAYERS[theme];
         if (baseTileLayer) {
             map.removeLayer(baseTileLayer);
+            baseTileLayer = null;
         }
-        baseTileLayer = L.tileLayer(cfg.url, { attribution: cfg.attribution }).addTo(map);
+        const useVector = Boolean(cfg.styleUrl) && canUseVectorBasemap();
+        if (useVector) {
+            // The bridge renders into tilePane, so marker/overlay z-order is unchanged.
+            baseTileLayer = L.maplibreGL({ style: cfg.styleUrl });
+        } else {
+            baseTileLayer = L.tileLayer(cfg.url, { attribution: cfg.attribution });
+        }
+        // Drives the CSS inversion that stands in for dark cartography when we had to
+        // fall back to raster.  Never set when the vector style rendered.
+        const container = document.getElementById('mesh-graph-container');
+        if (container) {
+            container.classList.toggle('mesh-basemap-fallback', theme === 'dark' && !useVector);
+        }
+        baseTileLayer.addTo(map);
     }
 
     function applyMeshTheme() {
@@ -1963,7 +2021,7 @@
                     markerZoomAnimation: true // Enable zoom animation for markers
                 }).setView([0, 0], 2);
 
-                setBaseTileLayer(); // theme-aware basemap (light OSM / dark CARTO)
+                setBaseTileLayer(); // theme-aware basemap (light OSM raster / dark OpenFreeMap vector)
 
                 // Disable popup auto-pan globally to prevent bounding boxes
                 L.Popup.prototype.options.autoPan = false;

tests/test_message_handler.py

@@ -2426,3 +2426,95 @@ class TestChannelPayloadCorrelation:
         assert result[RF_MATCH_KEY] == RF_MATCH_PAYLOAD
         assert RF_MATCH_KEY not in row
         assert RF_MATCH_KEY not in handler.recent_rf_data[0]
+
+    def test_repeater_echo_does_not_displace_the_message_row(self, handler):
+        """#255: on a dense mesh a repeater's echo of the same packet is logged
+        between the reception and its CHAN event, so the newest row is the echo —
+        different path length, different measured SNR. The message's own row is
+        still in the cache and must be the one that is matched."""
+        heard = self._row(
+            timestamp=time.time() - 0.2,
+            packet_prefix="35e01500595cdf2fd7e580897cdae64a",
+            snr=13.25,
+            rssi=-32,
+            routing_info={"path_length": 0, "path_nodes": [], "packet_hash": "392926C85DCB87D0"},
+            packet_hash="392926C85DCB87D0",
+        )
+        echo = self._row(
+            packet_prefix="30f61501f0595cdf2fd7e580897cdae6",
+            snr=12.0,
+            rssi=-10,
+            routing_info={"path_length": 1, "path_nodes": ["F0"], "packet_hash": "392926C85DCB87D0"},
+            packet_hash="392926C85DCB87D0",
+        )
+        chan = {**self.CHAN, "SNR": 13.25, "path_len": 0}
+
+        handler.rf_data_timeout = 15.0
+        handler.message_timeout = 10.0
+        handler.enhanced_correlation = False
+        handler.recent_rf_data = [heard, echo]
+        result = asyncio.run(
+            handler._correlate_channel_message_rf_data(
+                None, "", chan, scope_eligible_only=False, extended_timeout=30.0
+            )
+        )
+
+        assert result[RF_MATCH_KEY] == RF_MATCH_PAYLOAD
+        assert rf_data_is_correlated(result) is True
+        assert result["packet_prefix"] == "35e01500595cdf2fd7e580897cdae64a"
+        # SNR/RSSI come from the message's own reception, not the echo's.
+        assert result["snr"] == 13.25
+        assert result["rssi"] == -32
+
+    def test_same_packet_heard_twice_alike_takes_the_newest(self, handler):
+        older = self._row(timestamp=time.time() - 0.2, packet_prefix="aa" * 16)
+        newer = self._row(packet_prefix="bb" * 16)
+        result = self._correlate(handler, [older, newer])
+        assert result[RF_MATCH_KEY] == RF_MATCH_PAYLOAD
+        assert result["packet_prefix"] == "bb" * 16
+
+    def test_two_packets_agreeing_is_ambiguous_and_stays_a_fallback(self, handler):
+        """Different packets that happen to agree on all three fields are a
+        coincidence, not evidence; #80 is what taking the guess cost."""
+        other = self._row(packet_hash="0123456789ABCDEF")
+        other["routing_info"] = {"path_length": 0, "packet_hash": "0123456789ABCDEF"}
+        result = self._correlate(handler, [self._row(timestamp=time.time() - 0.2), other])
+        assert result[RF_MATCH_KEY] == RF_MATCH_FALLBACK
+        assert rf_data_is_correlated(result) is False
+
+    def test_matching_rows_without_a_hash_are_ambiguous(self, handler):
+        rows = [
+            self._row(timestamp=time.time() - 0.2, packet_hash=None),
+            self._row(packet_hash=None),
+        ]
+        result = self._correlate(handler, rows)
+        assert result[RF_MATCH_KEY] == RF_MATCH_FALLBACK
+
+    def test_search_respects_scope_eligibility(self, handler):
+        """The scope correlation asks for TC_FLOOD rows usable for HMAC matching, so
+        the search must not hand back an ineligible row just because it matches."""
+        from modules.enums import RouteType
+
+        matching_but_ineligible = self._row(timestamp=time.time() - 0.2)
+        eligible_but_unmatched = self._row(
+            snr=-7.5,  # disagrees with the payload
+            packet_prefix="cc" * 16,
+            route_type_int=int(RouteType.TRANSPORT_FLOOD.value),
+            transport_code1=18583,
+            scope_payload_hex="ca37f40824e44f7c",
+        )
+        assert handler._is_rf_data_scope_eligible(eligible_but_unmatched) is True
+        assert handler._is_rf_data_scope_eligible(matching_but_ineligible) is False
+
+        handler.rf_data_timeout = 15.0
+        handler.message_timeout = 10.0
+        handler.enhanced_correlation = False
+        handler.recent_rf_data = [matching_but_ineligible, eligible_but_unmatched]
+        result = asyncio.run(
+            handler._correlate_channel_message_rf_data(
+                None, "", self.CHAN, scope_eligible_only=True, extended_timeout=30.0
+            )
+        )
+
+        assert result[RF_MATCH_KEY] == RF_MATCH_FALLBACK
+        assert result["packet_prefix"] == "cc" * 16

tests/unit/test_packet_capture_mqtt_reconnect.py

@@ -0,0 +1,286 @@
+"""PacketCapture MQTT reconnection behaviour (issue #248).
+
+paho owns reconnection once loop_start() is running. The watchdog must observe
+rather than drive the socket, brokers must not share a client ID, and a renewed
+token has to be put in force before the old one expires.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import configparser
+import logging
+from unittest.mock import MagicMock
+
+import pytest
+
+from modules.service_plugins.packet_capture_service import PacketCaptureService
+
+
+def _bot_from_ini(ini: str) -> MagicMock:
+    cp = configparser.ConfigParser()
+    cp.read_string(ini.strip())
+    bot = MagicMock()
+    bot.config = cp
+    return bot
+
+
+def _service() -> PacketCaptureService:
+    svc = object.__new__(PacketCaptureService)
+    svc.logger = logging.getLogger("test-packet-capture")
+    svc.debug = False
+    svc.mqtt_enabled = True
+    svc.mqtt_connected = False
+    svc.mqtt_clients = []
+    svc.should_exit = False
+    return svc
+
+
+def _client_info(svc, *, connected: bool, thread_alive: bool | None, **config) -> dict:
+    client = MagicMock()
+    client.is_connected.return_value = connected
+    if thread_alive is None:
+        client._thread = None
+    else:
+        thread = MagicMock()
+        thread.is_alive.return_value = thread_alive
+        client._thread = thread
+    info = {
+        "client": client,
+        "config": {"host": "broker.example", **config},
+        "connected": connected,
+        "token_exp": None,
+        "down_since": None,
+        "stall_warned": False,
+        "cycle_lock": asyncio.Lock(),
+    }
+    svc.mqtt_clients.append(info)
+    return info
+
+
+# --- client identity -------------------------------------------------------
+
+
+def test_generated_client_ids_are_distinct_per_broker(monkeypatch):
+    """Brokers behind one cluster evict each other when they share an ID."""
+    bot = _bot_from_ini(
+        """
+        [Bot]
+        bot_name = observer
+        [PacketCapture]
+        enabled = true
+        mqtt1_server = mqtt-a.waev.app
+        mqtt2_server = mqtt-b.waev.app
+        """
+    )
+    svc = _service()
+    svc.bot = bot
+    svc.mqtt_brokers = PacketCaptureService._parse_mqtt_brokers(svc, bot.config)
+
+    created_ids = []
+
+    def fake_client(client_id=None, transport=None):
+        created_ids.append(client_id)
+        client = MagicMock()
+        client.is_connected.return_value = True
+        return client
+
+    fake_mqtt = MagicMock()
+    fake_mqtt.Client.side_effect = fake_client
+    monkeypatch.setattr("modules.service_plugins.packet_capture_service.mqtt", fake_mqtt)
+
+    asyncio.run(PacketCaptureService.connect_mqtt_brokers(svc))
+
+    assert len(created_ids) == 2
+    assert created_ids[0] != created_ids[1]
+
+
+def test_explicit_client_id_is_respected(monkeypatch):
+    bot = _bot_from_ini(
+        """
+        [Bot]
+        bot_name = observer
+        [PacketCapture]
+        enabled = true
+        mqtt1_server = mqtt-a.waev.app
+        mqtt1_client_id = my-fixed-id
+        """
+    )
+    svc = _service()
+    svc.bot = bot
+    svc.mqtt_brokers = PacketCaptureService._parse_mqtt_brokers(svc, bot.config)
+
+    created_ids = []
+    fake_mqtt = MagicMock()
+    fake_mqtt.Client.side_effect = lambda client_id=None, transport=None: (
+        created_ids.append(client_id) or MagicMock()
+    )
+    monkeypatch.setattr("modules.service_plugins.packet_capture_service.mqtt", fake_mqtt)
+
+    asyncio.run(PacketCaptureService.connect_mqtt_brokers(svc))
+    assert created_ids == ["my-fixed-id"]
+
+
+# --- watchdog --------------------------------------------------------------
+
+
+def test_watchdog_does_not_reconnect_while_paho_retries():
+    """The storm in #248: reconnect() called under a live network thread."""
+    svc = _service()
+    info = _client_info(svc, connected=False, thread_alive=True)
+
+    asyncio.run(svc._check_mqtt_client(info, now=1000.0))
+
+    info["client"].reconnect.assert_not_called()
+    info["client"].connect.assert_not_called()
+    info["client"].disconnect.assert_not_called()
+    assert info["connected"] is False
+
+
+def test_watchdog_restarts_a_dead_network_loop():
+    svc = _service()
+    info = _client_info(svc, connected=False, thread_alive=False)
+
+    asyncio.run(svc._check_mqtt_client(info, now=1000.0))
+
+    info["client"].reconnect.assert_called_once()
+    info["client"].loop_start.assert_called_once()
+
+
+def test_watchdog_restarts_when_no_thread_was_ever_started():
+    svc = _service()
+    info = _client_info(svc, connected=False, thread_alive=None)
+
+    asyncio.run(svc._check_mqtt_client(info, now=1000.0))
+
+    info["client"].reconnect.assert_called_once()
+
+
+def test_watchdog_clears_downtime_once_connected():
+    svc = _service()
+    info = _client_info(svc, connected=False, thread_alive=True)
+    asyncio.run(svc._check_mqtt_client(info, now=1000.0))
+    assert info["down_since"] == 1000.0
+
+    info["client"].is_connected.return_value = True
+    asyncio.run(svc._check_mqtt_client(info, now=1030.0))
+
+    assert info["connected"] is True
+    assert info["down_since"] is None
+
+
+def test_watchdog_warns_once_when_an_outage_persists(caplog):
+    svc = _service()
+    info = _client_info(svc, connected=False, thread_alive=True)
+
+    asyncio.run(svc._check_mqtt_client(info, now=1000.0))
+    with caplog.at_level(logging.WARNING, logger="test-packet-capture"):
+        asyncio.run(svc._check_mqtt_client(info, now=1000.0 + svc.MQTT_STALL_WARN_AFTER))
+        asyncio.run(svc._check_mqtt_client(info, now=1000.0 + svc.MQTT_STALL_WARN_AFTER + 30))
+
+    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
+    assert len(warnings) == 1
+
+
+def test_watchdog_refreshes_an_expiring_token_without_touching_the_socket():
+    svc = _service()
+    info = _client_info(svc, connected=False, thread_alive=True, use_auth_token=True)
+    info["token_exp"] = 1000.0 + 10  # inside the refresh margin
+
+    renewed = []
+
+    async def fake_renew(client_info):
+        renewed.append(client_info)
+        return True
+
+    svc._renew_mqtt_auth_token = fake_renew
+    asyncio.run(svc._check_mqtt_client(info, now=1000.0))
+
+    assert renewed == [info]
+    info["client"].reconnect.assert_not_called()
+
+
+def test_watchdog_leaves_a_valid_token_alone():
+    svc = _service()
+    info = _client_info(svc, connected=False, thread_alive=True, use_auth_token=True)
+    info["token_exp"] = 1000.0 + 3600
+
+    renewed = []
+
+    async def fake_renew(client_info):
+        renewed.append(client_info)
+        return True
+
+    svc._renew_mqtt_auth_token = fake_renew
+    asyncio.run(svc._check_mqtt_client(info, now=1000.0))
+
+    assert renewed == []
+
+
+# --- connection cycling ----------------------------------------------------
+
+
+def test_cycle_serializes_teardown_before_reconnect():
+    """loop_stop() must join the network thread before this one drives the socket."""
+    svc = _service()
+    info = _client_info(svc, connected=True, thread_alive=True)
+    calls = []
+    for name in ("disconnect", "loop_stop", "reconnect", "loop_start"):
+        getattr(info["client"], name).side_effect = (
+            lambda *_a, _n=name, **_kw: calls.append(_n)
+        )
+
+    asyncio.run(svc._cycle_mqtt_client(info, "test"))
+
+    assert calls == ["disconnect", "loop_stop", "reconnect", "loop_start"]
+
+
+# --- config ----------------------------------------------------------------
+
+
+def test_keepalive_defaults_and_overrides():
+    bot = _bot_from_ini(
+        """
+        [PacketCapture]
+        enabled = false
+        mqtt1_server = a.example
+        mqtt2_server = b.example
+        mqtt2_keepalive = 30
+        """
+    )
+    svc = object.__new__(PacketCaptureService)
+    svc.bot = bot
+    brokers = PacketCaptureService._parse_mqtt_brokers(svc, bot.config)
+    assert brokers[0]["keepalive"] == 60
+    assert brokers[1]["keepalive"] == 30
+
+
+def test_jwt_reconnect_on_renew_defaults_on():
+    bot = _bot_from_ini(
+        """
+        [PacketCapture]
+        enabled = false
+        mqtt1_server = a.example
+        mqtt2_server = b.example
+        mqtt2_jwt_reconnect_on_renew = false
+        """
+    )
+    svc = object.__new__(PacketCaptureService)
+    svc.bot = bot
+    brokers = PacketCaptureService._parse_mqtt_brokers(svc, bot.config)
+    assert brokers[0]["jwt_reconnect_on_renew"] is True
+    assert brokers[1]["jwt_reconnect_on_renew"] is False
+
+
+# --- logging ---------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("rc", "fragment"),
+    [(7, "rc=7"), (2, "rc=2")],
+)
+def test_disconnect_reason_names_the_paho_error(rc, fragment):
+    reason = PacketCaptureService._disconnect_reason(rc)
+    assert fragment in reason
+    # A bare number reads as a CONNACK code; the description is the point.
+    assert reason != fragment