From ad4e0ed13aea1fb7945518eb7c9a16b98e9beafd Mon Sep 17 00:00:00 2001
From: Cort Buffington <n0mjs@me.com>
Date: Sun, 12 Jul 2026 10:00:33 -0500
Subject: [PATCH] reporting: mirror dashboard normalization to dmrlink3 (Step
 3)

Ports hblink3's dashboard-reporting overhaul to dmrlink3's IPSC path so the
two converge (plan: ~/development/dashboard-reporting-normalization.md):

- Granular peer events: a centralized reconciliation sweep in both the master
  and peer maintenance loops (report_maintenance) diffs currently-connected
  peers vs the last-reported set and emits send_peer connected/disconnected
  deltas only on change -- the dashboard reflects membership immediately
  instead of waiting for the periodic full push. (The single master link-state,
  in PEER mode, continues to ride the snapshot.)

- Demote the full push: reporting_loop now calls periodic_push, which re-pushes
  the full config+bridge snapshot only every REPORT_RESYNC_SECONDS (60s) as a
  reconciliation safety net and sends a tiny {'type':'ping'} heartbeat on the
  other ticks. Idle feed stops re-serializing full state every REPORT_INTERVAL.

- Ping-loss quality metric (PING_LOSS): detects a peer/master that STAYS
  CONNECTED but loses keepalives (lossy link -> choppy audio). One uniform
  gap-inference path serves both roles -- MASTER mode observes peers'
  MASTER_ALIVE_REQs, PEER mode observes the master's/peers' *_ALIVE_REPLYs to
  our requests. IPSC's cadence is known (ALIVE_TIMER) so no self-calibration is
  needed (unlike hblink3). A gap beyond the keepalive-timeout horizon is treated
  as a session lapse (reconnect), not loss. Sliding window pruned to
  PING_LOSS_WINDOW minutes; dashboard flags the row gold at >= PING_LOSS_WARN%.

Config: PING_LOSS_WINDOW / PING_LOSS_WARN added to GLOBAL (optional, defaults).
Dashboard: backend applies peer deltas + ignores the ping heartbeat; frontend
renders "N% ping loss" and golds the radio id. Tests: TestPeerEvents +
TestPingLoss (73 pass). Verified live over real sockets (peer delta fired,
heartbeat/resync split confirmed, PING_LOSS climbed to 30% under a lossy peer)
and deployed to the live dmrlink3 services clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
---
 config.py                       |   6 ++
 dashboard/server.py             |  26 ++++-
 dashboard/static/dashboard.html |  15 ++-
 dmrlink.py                      | 177 +++++++++++++++++++++++++++++---
 tests/helpers.py                |   3 +
 tests/test_protocol.py          | 147 ++++++++++++++++++++++++++
 6 files changed, 355 insertions(+), 19 deletions(-)

diff --git a/config.py b/config.py
index 44c0c46..4b84795 100644
--- a/config.py
+++ b/config.py
@@ -134,6 +134,12 @@ def build_config(_config_file):
                     'PATH':    config.get(section, 'PATH'),
                     'USE_ACL': config.get(section, 'USE_ACL'),
                     'SUB_ACL': config.get(section, 'SUB_ACL'),
+                    # Per-peer keepalive-loss quality metric (dashboard PING_LOSS):
+                    # window is the sliding span it is measured over (minutes,
+                    # clamped 1..60); warn is the % at/above which the dashboard
+                    # flags the peer/master gold. Optional -- default if absent.
+                    'PING_LOSS_WINDOW': config.getint(section, 'PING_LOSS_WINDOW', fallback=5),
+                    'PING_LOSS_WARN':   config.getint(section, 'PING_LOSS_WARN', fallback=5),
                 })
 
             elif section == 'REPORTS':
diff --git a/dashboard/server.py b/dashboard/server.py
index fd34d4d..842c23c 100644
--- a/dashboard/server.py
+++ b/dashboard/server.py
@@ -178,6 +178,7 @@ class State:
         self.log      = deque(maxlen=LOG_LINES)
         self.dmrlink  = False
         self.clients  = set()            # connected WebSockets
+        self.ping_loss_warn = 5          # PING_LOSS_WARN %: dashboard flags peer/master gold at/above this
 
 STATE = State()
 
@@ -231,8 +232,30 @@ async def handle_event(evt):
         interval = evt.get('report_interval')
         if interval:
             FEED_READ_TIMEOUT = max(float(interval) * 3, 30.0)
+        STATE.ping_loss_warn = evt.get('ping_loss_warn', STATE.ping_loss_warn)
         STATE.systems = enrich_systems(evt.get('systems', {}))
-        await broadcast({'type': 'config', 'systems': STATE.systems})
+        await broadcast({'type': 'config', 'systems': STATE.systems,
+                         'ping_loss_warn': STATE.ping_loss_warn})
+
+    elif t == 'peer':
+        # Granular IPSC peer connect/disconnect delta. Apply it to the in-memory
+        # systems view and re-broadcast the (enriched) config so the browser
+        # renders it without waiting for the next slow resync. Ignored if the
+        # system isn't known yet -- the on-connect snapshot will carry it.
+        sysview = STATE.systems.get(evt.get('system'))
+        if sysview is not None:
+            peers = sysview.setdefault('PEERS', {})
+            rid = str(evt.get('radio_id'))
+            if evt.get('action') == 'connected' and evt.get('info') is not None:
+                peers[rid] = evt['info']
+            elif evt.get('action') == 'disconnected':
+                peers.pop(rid, None)
+            STATE.systems = enrich_systems(STATE.systems)
+            await broadcast({'type': 'config', 'systems': STATE.systems,
+                             'ping_loss_warn': STATE.ping_loss_warn})
+
+    elif t == 'ping':
+        pass   # liveness heartbeat; receiving it already reset the feed read timeout
 
     elif t == 'bridge':
         STATE.bridges = enrich_bridges(evt.get('bridges', {}))
@@ -380,6 +403,7 @@ def current_state():
         'bridges':     STATE.bridges,
         'active':      list(STATE.active.values()),
         'log':         list(STATE.log),
+        'ping_loss_warn': STATE.ping_loss_warn,
     }
 
 @app.websocket('/ws')
diff --git a/dashboard/static/dashboard.html b/dashboard/static/dashboard.html
index cc532c5..db3b685 100644
--- a/dashboard/static/dashboard.html
+++ b/dashboard/static/dashboard.html
@@ -90,7 +90,7 @@
 </div>
 
 <script>
-const ST = { systems:{}, bridges:{}, active:{}, tx_active:{}, hang:{}, trunk_streams:{}, log:[], dmrlink:false };
+const ST = { systems:{}, bridges:{}, active:{}, tx_active:{}, hang:{}, trunk_streams:{}, log:[], dmrlink:false, ping_loss_warn:5 };
 const STALE = 300000;   // backstop: drop an active call this many ms after START if no END arrives (well above amateur TOT)
 const LAST_HEARD_CFG   = '{{LAST_HEARD}}';                    // 'open' | 'closed' | 'off'
 const LAST_HEARD_COUNT = parseInt('{{LAST_HEARD_COUNT}}', 10) || 10;
@@ -257,8 +257,12 @@ function renderSystems(){
         const mUp  = mok && ms.connected_secs != null
           ? '<div class="sub">up <span class="js-since" data-base="'+ms.connected_secs+'" data-arr="'+Date.now()+'">'+dur(ms.connected_secs)+'</span></div>'
           : '';
+        // Keepalive-loss quality: PING_LOSS is the % lost over the last window.
+        const mLoss    = ms.PING_LOSS || 0;
+        const mridStyle = mLoss >= ST.ping_loss_warn ? ' style="color:var(--warn)"' : '';
+        const mLossTag  = mLoss > 0 ? '<div class="sub">· '+mLoss+'% ping loss</div>' : '';
         h += '<tr>'
-           + '<td class="mono">'+esc(mrid)+'</td>'
+           + '<td class="mono"'+mridStyle+'>'+esc(mrid)+mLossTag+'</td>'
            + '<td class="mono">'+esc(ms.IP||'—')+':'+esc(ms.PORT||'—')+mUp+'</td>'
            + '<td><span class="dot '+(mok?'ok':'bad')+'"></span>'
            + '<span class="'+(mok?'status-ok':'status-bad')+'">'+(mok?'Connected':'Disconnected')+'</span>'
@@ -291,8 +295,11 @@ function renderSystems(){
         const pUp = p.connected_secs != null
           ? '<div class="sub">up <span class="js-since" data-base="'+p.connected_secs+'" data-arr="'+Date.now()+'">'+dur(p.connected_secs)+'</span></div>'
           : '';
+        const pLoss     = p.PING_LOSS || 0;
+        const pridStyle = pLoss >= ST.ping_loss_warn ? ' style="color:var(--warn)"' : '';
+        const pLossTag  = pLoss > 0 ? '<div class="sub">· '+pLoss+'% ping loss</div>' : '';
         h += '<tr>'
-           + '<td class="mono">'+esc(rid)+'</td>'
+           + '<td class="mono"'+pridStyle+'>'+esc(rid)+pLossTag+'</td>'
            + '<td class="mono">'+esc(p.IP)+':'+esc(p.PORT)+pUp+'</td>'
            + '<td>'+statusHtml+'</td>'
            + '</tr>';
@@ -490,6 +497,7 @@ function connect(){
     const m = JSON.parse(ev.data);
     if(m.type==='initial'){
       setConn(m.dmrlink);
+      if(m.ping_loss_warn!=null) ST.ping_loss_warn = m.ping_loss_warn;
       ST.systems   = m.systems||{};
       ST.bridges   = m.bridges||{};
       ST.active    = {};
@@ -504,6 +512,7 @@ function connect(){
       if(!m.connected){ ST.systems={}; ST.bridges={}; ST.active={}; ST.tx_active={}; ST.hang={}; ST.trunk_streams={}; renderAll(); }
     }
     else if(m.type==='config'){
+      if(m.ping_loss_warn!=null) ST.ping_loss_warn = m.ping_loss_warn;
       ST.systems = m.systems||{};
       renderSystems();
     }
diff --git a/dmrlink.py b/dmrlink.py
index 48a4874..76d3efa 100644
--- a/dmrlink.py
+++ b/dmrlink.py
@@ -26,6 +26,7 @@ import signal
 
 from binascii import b2a_hex as ahex
 from binascii import a2b_hex as bhex
+from collections import deque
 from hashlib import sha1
 from hmac import new as hmac_new, compare_digest
 from socket import inet_ntoa as IPAddr
@@ -96,6 +97,26 @@ async def run_periodic(_interval, _func, _name, *_args):
 # Reporting server (NDJSON over TCP, replaces pickle/Netstring)
 # ---------------------------------------------------------------------------
 
+def _peer_snapshot(_pid, _p):
+    """JSON-serializable view of one IPSC peer. Shared by the full-state snapshot
+    and the granular send_peer delta (so both carry an identical shape, incl. the
+    PING_LOSS quality figure)."""
+    st = _p['STATUS']
+    return {
+        'RADIO_ID':                int_id(_pid),
+        'IP':                      _p['IP'],
+        'PORT':                    _p['PORT'],
+        'CONNECTED':               st['CONNECTED'],
+        'CONNECT_TIME':            st['CONNECT_TIME'],
+        'KEEP_ALIVES_SENT':        st['KEEP_ALIVES_SENT'],
+        'KEEP_ALIVES_RECEIVED':    st['KEEP_ALIVES_RECEIVED'],
+        'KEEP_ALIVES_OUTSTANDING': st['KEEP_ALIVES_OUTSTANDING'],
+        'KEEP_ALIVES_MISSED':      st['KEEP_ALIVES_MISSED'],
+        'KEEP_ALIVE_RX_TIME':      st['KEEP_ALIVE_RX_TIME'],
+        'PING_LOSS':               st.get('PING_LOSS', 0),
+    }
+
+
 def _systems_snapshot(_systems):
     """Produce a JSON-serializable view of the live SYSTEMS dict."""
     result = {}
@@ -138,20 +159,10 @@ def _systems_snapshot(_systems):
                 'KEEP_ALIVES_RECEIVED':   master['STATUS']['KEEP_ALIVES_RECEIVED'],
                 'KEEP_ALIVES_OUTSTANDING':master['STATUS']['KEEP_ALIVES_OUTSTANDING'],
                 'KEEP_ALIVES_MISSED':     master['STATUS']['KEEP_ALIVES_MISSED'],
+                'PING_LOSS':              master['STATUS'].get('PING_LOSS', 0),
             },
             'PEERS': {
-                str(int_id(pid)): {
-                    'RADIO_ID':                int_id(pid),
-                    'IP':                      p['IP'],
-                    'PORT':                    p['PORT'],
-                    'CONNECTED':               p['STATUS']['CONNECTED'],
-                    'CONNECT_TIME':            p['STATUS']['CONNECT_TIME'],
-                    'KEEP_ALIVES_SENT':        p['STATUS']['KEEP_ALIVES_SENT'],
-                    'KEEP_ALIVES_RECEIVED':    p['STATUS']['KEEP_ALIVES_RECEIVED'],
-                    'KEEP_ALIVES_OUTSTANDING': p['STATUS']['KEEP_ALIVES_OUTSTANDING'],
-                    'KEEP_ALIVES_MISSED':      p['STATUS']['KEEP_ALIVES_MISSED'],
-                    'KEEP_ALIVE_RX_TIME':      p['STATUS']['KEEP_ALIVE_RX_TIME'],
-                }
+                str(int_id(pid)): _peer_snapshot(pid, p)
                 for pid, p in peers.items()
             },
         }
@@ -183,6 +194,13 @@ def _enable_tcp_keepalive(_writer, _idle=60, _intvl=15, _cnt=4):
 # wedged dashboard can't grow dmrlink3 without limit.
 _REPORT_WRITE_BUFFER_LIMIT = 1 << 20   # 1 MiB
 
+# Real peer/bridge state changes now travel as granular delta events (send_peer,
+# send_bridge), so the full config+bridge snapshot no longer needs to go out every
+# REPORT_INTERVAL. periodic_push re-pushes the full state only this often (a
+# reconciliation safety net); the other ticks send a tiny heartbeat so the
+# dashboard's link-liveness read timeout stays fed. Mirrors hblink3.
+REPORT_RESYNC_SECONDS = 60
+
 
 class ReportServer:
     """Async TCP server that pushes NDJSON events to connected dashboard clients."""
@@ -270,9 +288,11 @@ class ReportServer:
     def send_config(self):
         # Advertise the push interval so the dashboard can size its "link dead"
         # read timeout relative to how often we actually send (this config push
-        # is the de-facto heartbeat).
+        # is the de-facto heartbeat / slow resync).
         self._send_json({'type': 'config',
                          'systems': _systems_snapshot(self._config['SYSTEMS']),
+                         # Dashboard flags a peer/master gold at/above this ping-loss %.
+                         'ping_loss_warn': self._config['GLOBAL'].get('PING_LOSS_WARN', 5),
                          'report_interval': self._config['REPORTS']['REPORT_INTERVAL']})
 
     def send_bridge(self):
@@ -281,9 +301,39 @@ class ReportServer:
     def send_bridge_event(self, _event):
         pass  # Overridden by BridgeReportServer in bridge.py
 
+    # Granular peer lifecycle delta (an IPSC peer registering or dropping) so the
+    # dashboard reflects membership changes immediately instead of waiting for the
+    # next slow resync. _info is a per-peer snapshot view on 'connected', omitted
+    # on 'disconnected'. Mirrors hblink3's send_peer.
+    def send_peer(self, _system, _radio_id, _action, _info=None):
+        evt = {'type': 'peer', 'system': _system, 'radio_id': _radio_id, 'action': _action}
+        if _info is not None:
+            evt['info'] = _info
+        self._send_json(evt)
+
     def send_rcm(self, _data):
         pass  # RCM monitoring not forwarded in dmrlink3
 
+    # True on the periodic ticks where a full state resync is due (a reconciliation
+    # safety net, ~REPORT_RESYNC_SECONDS apart); on the other ticks send only a
+    # tiny heartbeat so the dashboard's link-liveness timeout stays fed without
+    # re-pushing full state. Real state changes travel as granular events.
+    def _resync_due(self):
+        self._resync_tick = getattr(self, '_resync_tick', -1) + 1
+        interval = max(1, self._config['REPORTS'].get('REPORT_INTERVAL', 10))
+        every = max(1, round(REPORT_RESYNC_SECONDS / interval))
+        if self._resync_tick % every == 0:
+            return True
+        self._send_json({'type': 'ping'})
+        return False
+
+    def periodic_push(self):
+        # Full config+bridge only on the resync tick; heartbeat otherwise.
+        # send_bridge is a no-op in the base class (overridden by BridgeReportServer).
+        if self._resync_due():
+            self.send_config()
+            self.send_bridge()
+
 
 # ---------------------------------------------------------------------------
 # Reporting setup (called from async context)
@@ -304,8 +354,7 @@ def config_reports(_config, _factory):
     elif _config['REPORTS']['REPORT_NETWORKS'] == 'NETWORK':
         def reporting_loop(_server):
             logger.debug('(GLOBAL) Periodic reporting loop (NETWORK)')
-            _server.send_config()
-            _server.send_bridge()
+            _server.periodic_push()
         report_server = _factory(_config)
         loop.create_task(report_server.start(_config['REPORTS']['REPORT_PORT']))
         loop.create_task(run_periodic(_config['REPORTS']['REPORT_INTERVAL'],
@@ -512,6 +561,14 @@ class IPSC(asyncio.DatagramProtocol):
         self._master_sock = (self._master['IP'], self._master['PORT'])
         self._peers       = self._config['PEERS']
 
+        # Dashboard-reporting state (mirrors hblink3):
+        #   _reported_peers -- radio_ids last reported 'connected' via send_peer,
+        #     so the maintenance-loop reconciliation sweep emits deltas only on change.
+        #   _ping_q -- radio_id -> {'ev': deque[(ts, missed)], 'base': float, 'warm': list}
+        #     sliding window of keepalive observations feeding the PING_LOSS metric.
+        self._reported_peers = set()
+        self._ping_q = {}
+
         self.TS_FLAGS = self._local['MODE'] + self._local['FLAGS']
 
         self.MASTER_REG_REQ_PKT     = MASTER_REG_REQ   + self._local_id + self.TS_FLAGS + IPSC_VER
@@ -812,6 +869,7 @@ class IPSC(asyncio.DatagramProtocol):
                         'KEEP_ALIVES_OUTSTANDING': 0,
                         'KEEP_ALIVES_RECEIVED':    0,
                         'KEEP_ALIVE_RX_TIME':      0,
+                        'PING_LOSS':               0,
                     },
                 }
                 logger.debug('(%s) Peer Added: %s', self._system, int_id(_hex_radio_id))
@@ -918,6 +976,7 @@ class IPSC(asyncio.DatagramProtocol):
         self.reset_keep_alive(_peerid)
         self._peers[_peerid]['STATUS']['KEEP_ALIVES_RECEIVED'] += 1
         self._peers[_peerid]['STATUS']['KEEP_ALIVE_RX_TIME']   = int(time())
+        self._note_keepalive(_peerid)   # PEER mode: this peer's reply to our keepalive
         logger.debug('(%s) Keep-Alive Reply received from Peer %s',
                      self._system, int_id(_peerid))
 
@@ -933,6 +992,7 @@ class IPSC(asyncio.DatagramProtocol):
         self.reset_keep_alive(_peerid)
         self._master['STATUS']['KEEP_ALIVES_RECEIVED'] += 1
         self._master['STATUS']['KEEP_ALIVE_RX_TIME']   = int(time())
+        self._note_keepalive(_peerid)   # PEER mode: master's reply to our keepalive
         logger.debug('(%s) Keep-Alive Reply received from Master %s',
                      self._system, int_id(_peerid))
 
@@ -990,6 +1050,7 @@ class IPSC(asyncio.DatagramProtocol):
                     'KEEP_ALIVES_OUTSTANDING': 0,
                     'KEEP_ALIVES_RECEIVED':    0,
                     'KEEP_ALIVE_RX_TIME':      int(time()),
+                    'PING_LOSS':               0,
                 },
             }
         else:
@@ -1023,6 +1084,7 @@ class IPSC(asyncio.DatagramProtocol):
         if _peerid in self._peers:
             self._peers[_peerid]['STATUS']['KEEP_ALIVES_RECEIVED'] += 1
             self._peers[_peerid]['STATUS']['KEEP_ALIVE_RX_TIME']   = int(time())
+            self._note_keepalive(_peerid)   # MASTER mode: this peer's keepalive to us
             self.send_packet(self.MASTER_ALIVE_REPLY_PKT, (_host, _port))
             logger.debug('(%s) Master Keep-Alive Request from peer %s, %s:%s',
                          self._system, int_id(_peerid), _host, _port)
@@ -1038,6 +1100,88 @@ class IPSC(asyncio.DatagramProtocol):
             logger.warning('(%s) Peer List Request from *UNREGISTERED* peer %s',
                            self._system, int_id(_peerid))
 
+    # -----------------------------------------------------------------------
+    # Ping-loss quality metric (mirrors hblink3's per-repeater PING_LOSS).
+    # Detects a peer/master that STAYS CONNECTED but loses keepalives (lossy link
+    # -> choppy audio), which a simple up/down status never shows. hblink3 has to
+    # self-calibrate an unknown repeater ping cadence; IPSC's keepalive cadence is
+    # known -- ALIVE_TIMER -- so we use it directly as the expected interval. One
+    # uniform path serves BOTH roles: we note every keepalive-class packet OBSERVED
+    # from a radio and infer how many were lost in the gap since the previous one.
+    # MASTER mode observes peers' MASTER_ALIVE_REQs; PEER mode observes the master's
+    # / peers' *_ALIVE_REPLYs to our requests -- either way a lost keepalive widens
+    # the gap by (about) one interval. Window is a deque pruned to PING_LOSS_WINDOW
+    # minutes; loss% = missed / (received + missed).
+    # -----------------------------------------------------------------------
+    def _ping_loss_window_secs(self):
+        return min(60, max(1, self._CONFIG['GLOBAL'].get('PING_LOSS_WINDOW', 5))) * 60
+
+    def _prune_pings(self, _ev, _now):
+        cutoff = _now - self._ping_loss_window_secs()
+        while _ev and _ev[0][0] < cutoff:
+            _ev.popleft()
+
+    def _note_keepalive(self, _radio_id, _now=None):
+        now = time() if _now is None else _now
+        base = self._local['ALIVE_TIMER'] or 0
+        st = self._ping_q.get(_radio_id)
+        if st is None:
+            st = self._ping_q[_radio_id] = {'ev': deque(), 'last': 0.0}
+        gap = (now - st['last']) if st['last'] else 0.0
+        # Between base*1.5 (jitter guard) and the keepalive-timeout horizon, infer
+        # how many keepalives were lost in the gap. A gap beyond the horizon means
+        # the session lapsed (peer was down / re-registered) -- that is a disconnect,
+        # not a lossy link, so don't attribute the downtime as loss; start fresh.
+        horizon = base * (self._local.get('MAX_MISSED', 5) + 1)
+        if base > 0 and base * 1.5 < gap <= horizon:
+            missed = max(0, round(gap / base) - 1)
+        else:
+            missed = 0
+        st['ev'].append((now, missed))
+        st['last'] = now
+        self._prune_pings(st['ev'], now)
+
+    def _ping_loss_pct(self, _radio_id, _now):
+        st = self._ping_q.get(_radio_id)
+        if not st:
+            return 0
+        self._prune_pings(st['ev'], _now)
+        received = len(st['ev'])
+        missed = sum(m for _, m in st['ev'])
+        total = received + missed
+        return round(100.0 * missed / total) if total > 0 else 0
+
+    # Run once per maintenance sweep from either loop: refresh the PING_LOSS figure
+    # for each connected peer (and, in PEER mode, the master) so it rides the next
+    # snapshot / delta, prune dead radios' history, then emit granular peer
+    # connect/disconnect deltas for any change since the last sweep. The peer delta
+    # is a centralized reconciliation over self._peers -- it can't miss a scattered
+    # add/remove path and only emits on actual change (never a full-state flood).
+    # Mirrors hblink3.HBSYSTEM.report_peer_deltas. (The master link-state, a single
+    # well-known entity, rides the periodic snapshot rather than a delta.)
+    def report_maintenance(self):
+        now = time()
+        for pid, p in self._peers.items():
+            if p['STATUS']['CONNECTED']:
+                p['STATUS']['PING_LOSS'] = self._ping_loss_pct(pid, now)
+        if not self._local['MASTER_PEER']:
+            self._master_stat['PING_LOSS'] = (
+                self._ping_loss_pct(self._master['RADIO_ID'], now)
+                if self._master_stat['CONNECTED'] else 0)
+        for rid in list(self._ping_q):
+            self._prune_pings(self._ping_q[rid]['ev'], now)
+            if not self._ping_q[rid]['ev']:
+                del self._ping_q[rid]
+        if not self._report:
+            return
+        current = {pid for pid, p in self._peers.items() if p['STATUS']['CONNECTED']}
+        for pid in current - self._reported_peers:
+            self._report.send_peer(self._system, int_id(pid), 'connected',
+                                    _peer_snapshot(pid, self._peers[pid]))
+        for pid in self._reported_peers - current:
+            self._report.send_peer(self._system, int_id(pid), 'disconnected')
+        self._reported_peers = current
+
     # -----------------------------------------------------------------------
     # Connection maintenance loops (run_periodic callbacks)
     # -----------------------------------------------------------------------
@@ -1054,6 +1198,7 @@ class IPSC(asyncio.DatagramProtocol):
                 self.send_to_ipsc(self.PEER_LIST_REPLY_PKT + build_peer_list(self._peers))
                 logger.warning('(%s) Timeout exceeded for Peer %s — de-registering',
                                self._system, int_id(peer))
+        self.report_maintenance()
 
     def peer_maintenance_loop(self):
         logger.debug('(%s) PEER Maintenance Loop', self._system)
@@ -1132,6 +1277,8 @@ class IPSC(asyncio.DatagramProtocol):
                     self._peers[peer]['STATUS']['KEEP_ALIVES_SENT']        += 1
                     self._peers[peer]['STATUS']['KEEP_ALIVES_OUTSTANDING'] += 1
 
+        self.report_maintenance()
+
 
 # ---------------------------------------------------------------------------
 # __main__
diff --git a/tests/helpers.py b/tests/helpers.py
index dfe02df..0103a44 100644
--- a/tests/helpers.py
+++ b/tests/helpers.py
@@ -79,6 +79,7 @@ def make_system(radio_id_int, ip, port, master_peer=True,
             'PORT':         master_port if not master_peer else '',
             'STATUS': {
                 'CONNECTED':               False,
+                'CONNECT_TIME':            0,
                 'PEER_LIST':               False,
                 'KEEP_ALIVES_SENT':        0,
                 'KEEP_ALIVES_MISSED':      0,
@@ -161,11 +162,13 @@ def make_peer_entry(host='127.0.0.1', port=50200):
         'FLAGS_DECODE': {},
         'STATUS': {
             'CONNECTED':               True,
+            'CONNECT_TIME':            int(_time()),
             'KEEP_ALIVES_SENT':        0,
             'KEEP_ALIVES_MISSED':      0,
             'KEEP_ALIVES_OUTSTANDING': 0,
             'KEEP_ALIVES_RECEIVED':    0,
             'KEEP_ALIVE_RX_TIME':      int(_time()),
+            'PING_LOSS':               0,
         },
     }
 
diff --git a/tests/test_protocol.py b/tests/test_protocol.py
index b3e48ff..251ee67 100644
--- a/tests/test_protocol.py
+++ b/tests/test_protocol.py
@@ -260,5 +260,152 @@ class TestUnknownOpcodes(IPSCMasterBase):
         self.inject(pkt)
 
 
+# ---------------------------------------------------------------------------
+# Granular peer connect/disconnect reporting (dashboard delta events)
+# ---------------------------------------------------------------------------
+
+class _FakeReport:
+    """Captures send_peer() deltas emitted by the reconciliation sweep."""
+    def __init__(self):
+        self.peer_events = []
+
+    def send_peer(self, system, radio_id, action, info=None):
+        self.peer_events.append({'system': system, 'radio_id': radio_id,
+                                 'action': action, 'info': info})
+
+    def send_config(self):        pass
+    def send_bridge(self):        pass
+    def send_bridge_event(self, e): pass
+    def send_rcm(self, d):        pass
+
+
+class TestPeerEvents(unittest.TestCase):
+    """report_maintenance() emits a granular delta only when peer membership
+    actually changes -- the reconciliation-sweep contract mirrored from hblink3."""
+
+    SYSTEM_NAME = 'TEST-MASTER'
+
+    def setUp(self):
+        cfg = make_config({
+            self.SYSTEM_NAME: make_system(MASTER_ID, '127.0.0.1', 50100, master_peer=True),
+        })
+        self.report = _FakeReport()
+        self.proto = IPSC(self.SYSTEM_NAME, cfg, self.report)
+        self.proto.transport = MockDatagramTransport()
+
+    def register(self, peer_id_bytes, host='127.0.0.1', port=50200):
+        self.proto.datagram_received(make_reg_req(peer_id_bytes), (host, port))
+
+    def test_connect_emits_single_connected_delta(self):
+        self.register(PEER_A_ID_B)
+        self.proto.report_maintenance()
+        evts = self.report.peer_events
+        self.assertEqual(len(evts), 1)
+        self.assertEqual(evts[0]['action'], 'connected')
+        self.assertEqual(evts[0]['radio_id'], PEER_A_ID)
+        self.assertEqual(evts[0]['system'], self.SYSTEM_NAME)
+        self.assertIsNotNone(evts[0]['info'])          # connected carries a snapshot
+        self.assertEqual(evts[0]['info']['RADIO_ID'], PEER_A_ID)
+
+    def test_no_change_emits_nothing(self):
+        self.register(PEER_A_ID_B)
+        self.proto.report_maintenance()
+        self.report.peer_events.clear()
+        self.proto.report_maintenance()                 # nothing changed
+        self.assertEqual(self.report.peer_events, [])
+
+    def test_disconnect_emits_disconnected_delta_without_info(self):
+        self.register(PEER_A_ID_B)
+        self.proto.report_maintenance()
+        self.report.peer_events.clear()
+        # A de-registration (peer leaves) removes it from self._peers.
+        self.proto.datagram_received(make_dereg_req(PEER_A_ID_B), ('127.0.0.1', 50200))
+        self.proto.report_maintenance()
+        evts = self.report.peer_events
+        self.assertEqual(len(evts), 1)
+        self.assertEqual(evts[0]['action'], 'disconnected')
+        self.assertEqual(evts[0]['radio_id'], PEER_A_ID)
+        self.assertIsNone(evts[0]['info'])              # disconnected omits the snapshot
+
+    def test_two_peers_emit_two_connected_deltas(self):
+        self.register(PEER_A_ID_B)
+        self.register(PEER_B_ID_B, port=50201)
+        self.proto.report_maintenance()
+        actions = sorted((e['radio_id'], e['action']) for e in self.report.peer_events)
+        self.assertEqual(actions, [(PEER_A_ID, 'connected'), (PEER_B_ID, 'connected')])
+
+    def test_no_report_server_is_safe(self):
+        # Same sweep with reporting disabled (report=None) must not raise.
+        cfg = make_config({
+            self.SYSTEM_NAME: make_system(MASTER_ID, '127.0.0.1', 50100, master_peer=True),
+        })
+        proto = IPSC(self.SYSTEM_NAME, cfg, None)
+        proto.transport = MockDatagramTransport()
+        proto.datagram_received(make_reg_req(PEER_A_ID_B), ('127.0.0.1', 50200))
+        proto.report_maintenance()                      # must not raise
+
+
+# ---------------------------------------------------------------------------
+# Ping-loss quality metric (keepalive loss over a sliding window)
+# ---------------------------------------------------------------------------
+
+class TestPingLoss(unittest.TestCase):
+    """_note_keepalive/_ping_loss_pct infer keepalive loss from the gap between
+    observed keepalives vs the known ALIVE_TIMER cadence (5s in make_system)."""
+
+    def setUp(self):
+        cfg = make_config({
+            'S': make_system(MASTER_ID, '127.0.0.1', 50100, master_peer=True),
+        })
+        self.proto = IPSC('S', cfg, None)
+        self.proto.transport = MockDatagramTransport()
+        self.rid = PEER_A_ID_B
+
+    def feed(self, *times):
+        for t in times:
+            self.proto._note_keepalive(self.rid, _now=t)
+
+    def test_clean_cadence_is_zero_loss(self):
+        self.feed(0, 5, 10, 15, 20)                     # every 5s == ALIVE_TIMER
+        self.assertEqual(self.proto._ping_loss_pct(self.rid, 20), 0)
+
+    def test_first_observation_no_false_positive(self):
+        self.feed(1000)                                  # no prior -> gap 0 -> 0 missed
+        self.assertEqual(self.proto._ping_loss_pct(self.rid, 1000), 0)
+
+    def test_jitter_under_threshold_not_counted(self):
+        self.feed(0, 6)                                  # gap 6 < base*1.5 (7.5) -> not lossy
+        self.assertEqual(self.proto._ping_loss_pct(self.rid, 6), 0)
+
+    def test_gap_infers_missed_keepalives(self):
+        # 0,5 clean; then a 15s gap == two keepalives lost before the one at 20.
+        self.feed(0, 5, 20)
+        st = self.proto._ping_q[self.rid]
+        self.assertEqual([m for _, m in st['ev']], [0, 0, 2])
+        # received=3 events, missed=2 -> 2/(3+2) = 40%
+        self.assertEqual(self.proto._ping_loss_pct(self.rid, 20), 40)
+
+    def test_window_prunes_old_loss(self):
+        # PING_LOSS_WINDOW defaults to 5 min (300s). Loss at t~20 ages out by t~400.
+        self.feed(0, 5, 20)
+        self.assertGreater(self.proto._ping_loss_pct(self.rid, 20), 0)
+        self.feed(400, 405, 410)                         # fresh clean pings well past the window
+        self.assertEqual(self.proto._ping_loss_pct(self.rid, 410), 0)
+
+    def test_report_maintenance_sets_ping_loss_on_connected_peer(self):
+        cfg = make_config({'S': make_system(MASTER_ID, '127.0.0.1', 50100, master_peer=True)})
+        proto = IPSC('S', cfg, None)
+        proto.transport = MockDatagramTransport()
+        proto.datagram_received(make_reg_req(PEER_A_ID_B), ('127.0.0.1', 50200))
+        # Inject a lossy keepalive history near wall-clock now, so it stays inside
+        # the window when report_maintenance() refreshes against real time().
+        now = time()
+        proto._note_keepalive(PEER_A_ID_B, _now=now - 15)  # first: no prior gap
+        proto._note_keepalive(PEER_A_ID_B, _now=now - 10)  # +5s clean
+        proto._note_keepalive(PEER_A_ID_B, _now=now)       # +10s gap -> 1 missed
+        proto.report_maintenance()
+        self.assertGreater(proto._peers[PEER_A_ID_B]['STATUS']['PING_LOSS'], 0)
+
+
 if __name__ == '__main__':
     unittest.main()