feat(reliability): supervisor continuously enforces known-good UCI set

payload/user/remote_access/pager-webui/server.py

@@ -4761,14 +4761,28 @@ def health_check():
     # Re-park periodically — parking never runs wifi reload, so it is safe
     # to repeat.
     h['ticks'] = h.get('ticks', 0) + 1
-    if h['ticks'] % HEALTH_STA_PARK_INTERVAL == 0 and _sta_uplink_enabled():
+    if h['ticks'] % HEALTH_STA_PARK_INTERVAL == 0:
+        if _sta_uplink_enabled():
+            try:
+                _park_dummy_sta()
+                h['last_action'] = 'dummy STA re-parked'
+                import mk8_events
+                mk8_events.log_event('guard_fix',
+                                     msg='re-parked dummy_radio0 STA '
+                                         '(stock reconvergence)')
+            except Exception:
+                pass
+        # Continuous enforcement of the known-good set (boot-guard keys can
+        # be dropped or flipped by stock reconvergence/profile restores).
+        # clear_pool=False mid-run: never wipe collected SSIDs live.
         try:
-            _park_dummy_sta()
+            import mk8_guard
-            h['last_action'] = 'dummy STA re-parked'
+            rep = mk8_guard.reconcile(clear_pool=False)
-            import mk8_events
+            if rep['changed']:
-            mk8_events.log_event('guard_fix',
+                import mk8_events
-                                 msg='re-parked dummy_radio0 STA '
+                mk8_events.log_event(
-                                     '(stock reconvergence)')
+                    'guard_fix', sev='warn',
+                    msg='reconciler re-applied: %s' % ','.join(rep['changed']))
         except Exception:
             pass
     h['mem_percent'] = _mem_percent()

tests/test_health.py

@@ -155,20 +155,70 @@ class SupervisorExtrasTest(unittest.TestCase):
         server._health['ticks'] = server.HEALTH_STA_PARK_INTERVAL - 1
         with mock.patch.object(
                 server, '_sta_uplink_enabled', return_value=True), \
-             __import__('unittest').mock.patch.object(
+             mock.patch.object(
-                server, '_park_dummy_sta') as park:
+                server, '_park_dummy_sta') as park, \
+             mock.patch.object(
+                server, '_raise_monitors', return_value=[]), \
+             mock.patch('mk8_guard.reconcile',
+                        return_value={'changed': [], 'pool_cleared': False,
+                                      'monitors_raised': []}) as rec:
             h = server.health_check()
             park.assert_called_once()
+            rec.assert_called_once()
             self.assertEqual(server._health['ticks'],
                              server.HEALTH_STA_PARK_INTERVAL)
-        # not on interval ticks: no re-park
+        # not on interval ticks: no re-park, no reconcile
         server._health['ticks'] = 1
         with mock.patch.object(
                 server, '_sta_uplink_enabled', return_value=True), \
-             __import__('unittest').mock.patch.object(
+             mock.patch.object(
-                server, '_park_dummy_sta') as park:
+                server, '_park_dummy_sta') as park, \
+             mock.patch('mk8_guard.reconcile') as rec:
             server.health_check()
             park.assert_not_called()
+            rec.assert_not_called()
+
+    def test_health_reconcile_journals_changed_keys(self):
+        import mk8_events
+        events = []
+        old_log = mk8_events.log_event
+        old_run = server.device_run
+        old_iface = server._iface_up
+
+        def fake_run(args, timeout=20, input_data=None):
+            a = list(args)
+            if a[:2] == ['pidof', 'pineapd']:
+                return (0, '12345\n', '')
+            if a[:2] == ['ip', 'link', 'show']:
+                return (0, '4: wlan0mon: <UP> state unknown', '')
+            return (0, '', '')
+
+        mk8_events.log_event = lambda kind, **kw: events.append((kind, kw))
+        server.device_run = fake_run
+        server._iface_up = lambda name: True
+        old_ticks = server._health.get('ticks')
+        try:
+            server._health['ticks'] = server.HEALTH_STA_PARK_INTERVAL - 1
+            with mock.patch.object(server, '_sta_uplink_enabled',
+                                   return_value=False), \
+                 mock.patch.object(server, '_raise_monitors',
+                                   return_value=[]), \
+                 mock.patch('mk8_guard.reconcile',
+                            return_value={'changed':
+                                          ['pineapd.@pineapd[0].autossidpool'],
+                                          'pool_cleared': False,
+                                          'monitors_raised': []}):
+                server.health_check()
+            self.assertTrue(any(k == 'guard_fix' and 'autossidpool' in kw.get('msg', '')
+                                for k, kw in events))
+        finally:
+            mk8_events.log_event = old_log
+            server.device_run = old_run
+            server._iface_up = old_iface
+            if old_ticks is None:
+                server._health.pop('ticks', None)
+            else:
+                server._health['ticks'] = old_ticks
 
     def test_health_reports_events_and_counters(self):
         import mk8_events